Bump h1 0.6.2 + livery 0.3.0 + hackney 4.3.0; deferred streaming status; httpc->hackney

apps/barrel_inference_cli/rebar.config

@@ -6,7 +6,7 @@
 %% its only dependency; it needs no Erlang dependency on the server app.
 
 {deps, [
-    {livery, "0.2.7"}
+    {livery, "~> 0.3.0"}
 ]}.
 
 {elvis, [

apps/barrel_inference_server/rebar.config

@@ -4,18 +4,18 @@
 %% ruleset, and its Hex/ex_doc publishing metadata.
 
 {deps, [
-    {instrument, "1.0.0"},
+    {instrument, "~> 1.1.3"},
     %% Single inbound + outbound HTTP stack. Hackney remains as a
     %% transitive dep of livery's default adapter; we no longer call
     %% it directly.
-    {livery, "0.2.7"},
+    {livery, "~> 0.3.0"},
     %% Sibling umbrella app (apps/barrel_inference). rebar3 resolves it
     %% locally as a project app during umbrella builds, shadowing this
     %% Hex constraint; the constraint is what the published
     %% barrel_inference_server package declares.
-    {barrel_inference, "0.8.0"},
+    {barrel_inference, "~> 0.8.0"},
     %% MCP client library (used for the server-side MCP tool bridge).
-    {barrel_mcp, "2.2.2"}
+    {barrel_mcp, "~> 2.2.2"}
 ]}.
 
 {ct_opts, [{verbose, true}]}.

apps/barrel_inference_server/src/barrel_inference_server_h_chat.erl

@@ -4,8 +4,9 @@
 %%% livery migration. Mirrors the request/response shape exactly; only
 %%% the framework call surface differs.
 %%%
-%%% Uses `livery_resp:stream/3' with manual SSE framing for full
-%%% control over comments, multi-frame finishes, and `[DONE]'.
+%%% Uses livery's native `{sse, _}' decision: each Emit call takes a
+%%% `#{data => Iodata}' map (or `#{event => Bin, data => Iodata}') and
+%%% livery handles the wire framing.
 
 -module(barrel_inference_server_h_chat).
 
@@ -21,8 +22,10 @@
         maybe_end_session/1,
         record_session_committed/2,
         release_slot/1,
-        drive_stream/4,
-        drive_buffered/3
+        drive_stream_post_admit/2,
+        drive_buffered/3,
+        resolve_stream/3,
+        pre_admit_loop/1
     ]}
 ).
 
@@ -129,27 +132,92 @@ start_inference(R0, Api) ->
     R1 = R0#barrel_inference_request{model_id = Real},
     case R1#barrel_inference_request.stream of
         true ->
-            livery_resp:stream(
-                200,
-                sse_headers(),
-                fun(Emit) -> drive_stream(R1, Requested, Api, Emit) end
+            livery_resp:stream_deferred(
+                fun() -> resolve_stream(R1, Requested, Api) end
             );
         false ->
             drive_buffered(R1, Requested, Api)
     end.
 
-drive_stream(R, Requested, Api, Emit) ->
+%% Run the sync pre-admit phase, then either resolve to an SSE stream
+%% (admit succeeded; producer drives the live receive loop) or to a
+%% one-shot JSON error envelope (pre-admit failure surfaces as a real
+%% 4xx/5xx status instead of a 200 + SSE error frame).
+resolve_stream(R, Requested, Api) ->
     {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R),
     State = init_state(R, Requested, Api, Worker, Mon, true),
     State1 = arm_total_timer(State),
+    case pre_admit_loop(State1) of
+        {admitted, State2} ->
+            {sse, 200, sse_extras(), fun(Emit) ->
+                drive_stream_post_admit(State2, Emit)
+            end};
+        {error, Status, Reason, State2} ->
+            cleanup(State2),
+            Body = json:encode(error_body(Reason, Status)),
+            {full, Status, json_headers(), Body}
+    end.
+
+drive_stream_post_admit(State, Emit) ->
     _ =
         try
-            stream_loop(State1, Emit)
+            stream_loop(State, Emit)
         after
-            cleanup(State1)
+            cleanup(State)
         end,
     ok.
 
+%% Consume pipeline messages up to admission. Errors (pool_exhausted,
+%% not_loaded, model_not_found, ...) come back as
+%% {error, Status, Reason} so the caller can return a JSON response.
+%%
+%% Critically, this is a SELECTIVE receive: only the pre-admit
+%% pipeline messages match. Token / done / error messages from the
+%% engine (which can race ahead of the pipeline's admit when the
+%% engine decodes fast) stay in the mailbox, and stream_loop picks
+%% them up in order after admit fires. Adding a catch-all here
+%% would silently drop those races and the post-admit loop would
+%% hang on the SSE stream.
+pre_admit_loop(S) ->
+    receive
+        {pipeline, error, Status, Reason} ->
+            {error, Status, Reason, S};
+        {pipeline, loading, _M} ->
+            pre_admit_loop(S);
+        {pipeline, loaded} when S#st.keepalive_begun ->
+            pre_admit_loop(S#st{phase = waiting_template});
+        {pipeline, loaded} ->
+            ok = barrel_inference_server_keepalive:request_begin(S#st.model),
+            pre_admit_loop(S#st{phase = waiting_template, keepalive_begun = true});
+        {pipeline, templated, Tokens, ParamsRef} ->
+            pre_admit_loop(S#st{
+                phase = waiting_queue,
+                prompt_token_ids = Tokens,
+                chat_params_ref = ParamsRef
+            });
+        {pipeline, queued} ->
+            pre_admit_loop(S#st{phase = waiting_admit});
+        {pipeline, admitted, Ref, Slot} ->
+            S1 = monitor_engine(
+                arm_prefill_timer(S#st{phase = running, ref = Ref, slot = Slot})
+            ),
+            {admitted, S1};
+        {'DOWN', Mon, process, _Pid, _Reason} when Mon =:= S#st.engine_mon ->
+            {error, 500, model_crashed, S#st{engine_mon = undefined}};
+        {'DOWN', Mon, process, _Pid, normal} when Mon =:= S#st.worker_mon ->
+            pre_admit_loop(S#st{worker = undefined, worker_mon = undefined});
+        {'DOWN', Mon, process, _Pid, _Reason} when Mon =:= S#st.worker_mon ->
+            {error, 500, pipeline_crashed, S#st{worker = undefined, worker_mon = undefined}}
+    after pre_admit_timeout() ->
+        {error, 504, prefill_timeout, S}
+    end.
+
+pre_admit_timeout() ->
+    barrel_inference_server_config:prefill_ms().
+
+json_headers() ->
+    [{<<"content-type">>, <<"application/json">>}].
+
 drive_buffered(R, Requested, Api) ->
     {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R),
     State = init_state(R, Requested, Api, Worker, Mon, false),
@@ -249,10 +317,7 @@ dispatch_stream(_Other, S, Emit) ->
     stream_loop(S, Emit).
 
 on_pipeline_stream({pipeline, loading, _M}, S, Emit) ->
-    case emit_raw(Emit, [<<": loading\n\n">>]) of
-        ok -> stream_loop(S#st{stream_started = true}, Emit);
-        closed -> S
-    end;
+    stream_loop(S, Emit);
 on_pipeline_stream({pipeline, loaded}, S = #st{keepalive_begun = true}, Emit) ->
     stream_loop(S#st{phase = waiting_template}, Emit);
 on_pipeline_stream({pipeline, loaded}, S, Emit) ->
@@ -271,14 +336,14 @@ on_pipeline_stream({pipeline, admitted, Ref, Slot}, S, Emit) ->
 on_pipeline_stream({pipeline, error, Status, Reason}, S, Emit) ->
     record_metrics(S, Status),
     Err = json:encode(error_body(Reason, Status)),
-    _ = emit_raw(Emit, [<<"data: ">>, Err, <<"\n\n">>, <<"data: [DONE]\n\n">>]),
+    _ = emit_events(Emit, [#{data => Err}, done_frame()]),
     S.
 
 on_stream_token(S0, Ref, Tok, Emit) ->
     S = learn_ref(S0, Ref),
     case handle_token_stream(Tok, S) of
-        {emit, Out, S1} ->
-            case emit_raw(Emit, Out) of
+        {emit, Frame, S1} ->
+            case emit_event(Emit, Frame) of
                 ok -> stream_loop(rearm_idle(S1), Emit);
                 closed -> S1
             end;
@@ -311,14 +376,14 @@ stream_text_more(Tok, S) ->
     Iolist = barrel_inference_server_translate:internal_to_openai_chat_chunk(
         Tok, S#st.req_id, S#st.requested
     ),
-    {emit, [<<"data: ">>, Iolist, <<"\n\n">>], S#st{out_tokens = S#st.out_tokens + 1}}.
+    {emit, #{data => Iolist}, S#st{out_tokens = S#st.out_tokens + 1}}.
 
 on_stream_reasoning(S0, Ref, Tok, Emit) ->
     S = learn_ref(S0, Ref),
     Iolist = barrel_inference_server_translate:internal_to_openai_reasoning_chunk(
         Tok, S#st.req_id, S#st.requested
     ),
-    case emit_raw(Emit, [<<"data: ">>, Iolist, <<"\n\n">>]) of
+    case emit_event(Emit, #{data => Iolist}) of
         ok -> stream_loop(rearm_idle(S), Emit);
         closed -> S
     end.
@@ -363,14 +428,8 @@ finish_stream_ok(S = #st{mode = text}, Stats, Emit) ->
     Final = barrel_inference_server_translate:internal_to_openai_chat_final(
         Stats, S#st.req_id, S#st.requested
     ),
-    Frames = [
-        <<"data: ">>,
-        Final,
-        <<"\n\n">>,
-        usage_chunk(S, Stats),
-        <<"data: [DONE]\n\n">>
-    ],
-    _ = emit_raw(Emit, Frames),
+    Frames = [#{data => Final}] ++ usage_frames(S, Stats) ++ [done_frame()],
+    _ = emit_events(Emit, Frames),
     record_success(S, Stats),
     S;
 finish_stream_ok(S = #st{mode = tool_buffer}, Stats, Emit) ->
@@ -379,17 +438,9 @@ finish_stream_ok(S = #st{mode = tool_buffer}, Stats, Emit) ->
     Stop = barrel_inference_server_translate:internal_to_openai_chat_final(
         ToolStats, S#st.req_id, S#st.requested
     ),
-    Frames = [
-        <<"data: ">>,
-        First,
-        <<"\n\n">>,
-        <<"data: ">>,
-        Stop,
-        <<"\n\n">>,
-        usage_chunk(S, ToolStats),
-        <<"data: [DONE]\n\n">>
-    ],
-    _ = emit_raw(Emit, Frames),
+    Frames =
+        [#{data => First}, #{data => Stop}] ++ usage_frames(S, ToolStats) ++ [done_frame()],
+    _ = emit_events(Emit, Frames),
     record_success(S, Stats),
     S.
 
@@ -399,24 +450,16 @@ finish_with_tool_calls_stream(S, Calls, Emit) ->
     Stop = barrel_inference_server_translate:internal_to_openai_chat_final(
         ToolStats, S#st.req_id, S#st.requested
     ),
-    Frames = [
-        <<"data: ">>,
-        First,
-        <<"\n\n">>,
-        <<"data: ">>,
-        Stop,
-        <<"\n\n">>,
-        usage_chunk(S, ToolStats),
-        <<"data: [DONE]\n\n">>
-    ],
-    _ = emit_raw(Emit, Frames),
+    Frames =
+        [#{data => First}, #{data => Stop}] ++ usage_frames(S, ToolStats) ++ [done_frame()],
+    _ = emit_events(Emit, Frames),
     record_success(S, ToolStats),
     S.
 
 finish_stream_err(S, Reason, Emit) ->
     Status = http_status(Reason),
     Err = json:encode(error_body(Reason, Status)),
-    _ = emit_raw(Emit, [<<"data: ">>, Err, <<"\n\n">>, <<"data: [DONE]\n\n">>]),
+    _ = emit_events(Emit, [#{data => Err}, done_frame()]),
     record_error(S, Reason),
     S.
 
@@ -978,13 +1021,15 @@ is_tool_first_byte(<<C, _/binary>>) when C =:= $\s; C =:= $\t; C =:= $\r; C =:=
 is_tool_first_byte(<<${, _/binary>>) -> true;
 is_tool_first_byte(_) -> false.
 
-usage_chunk(#st{include_usage = false}, _Stats) ->
+usage_frames(#st{include_usage = false}, _Stats) ->
     [];
-usage_chunk(S = #st{include_usage = true}, Stats) ->
+usage_frames(S = #st{include_usage = true}, Stats) ->
     Chunk = barrel_inference_server_translate:internal_to_openai_usage_chunk(
         Stats, S#st.req_id, S#st.requested
     ),
-    [<<"data: ">>, Chunk, <<"\n\n">>].
+    [#{data => Chunk}].
+
+done_frame() -> #{data => <<"[DONE]">>}.
 
 %%====================================================================
 %% Metrics
@@ -1054,12 +1099,20 @@ cache_log_fields(Stats) ->
 %% Reply helpers
 %%====================================================================
 
-emit_raw(Emit, IoData) ->
-    case Emit(IoData) of
+emit_event(Emit, EventMap) ->
+    case Emit(EventMap) of
         ok -> ok;
         {error, _} -> closed
     end.
 
+emit_events(_Emit, []) ->
+    ok;
+emit_events(Emit, [F | Rest]) ->
+    case emit_event(Emit, F) of
+        ok -> emit_events(Emit, Rest);
+        closed -> closed
+    end.
+
 error_body({context_overflow, Tokens, Ctx}, _Status) ->
     Msg = iolist_to_binary(
         io_lib:format("prompt is too long: ~B tokens > ~B maximum", [Tokens, Ctx])
@@ -1106,12 +1159,8 @@ http_status({error, {decode_failed, _}}) -> 503;
 http_status({decode_failed, _}) -> 503;
 http_status(_) -> 500.
 
-sse_headers() ->
-    [
-        {<<"content-type">>, <<"text/event-stream">>},
-        {<<"cache-control">>, <<"no-cache">>},
-        {<<"x-accel-buffering">>, <<"no">>}
-    ].
+sse_extras() ->
+    [{<<"x-accel-buffering">>, <<"no">>}].
 
 decode(Body) ->
     try