diff --git a/apps/barrel_inference_cli/rebar.config b/apps/barrel_inference_cli/rebar.config index 5b22dfe..4773c83 100644 --- a/apps/barrel_inference_cli/rebar.config +++ b/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, [ diff --git a/apps/barrel_inference_server/rebar.config b/apps/barrel_inference_server/rebar.config index 8918dc1..e5a119e 100644 --- a/apps/barrel_inference_server/rebar.config +++ b/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}]}. diff --git a/apps/barrel_inference_server/src/barrel_inference_server_h_chat.erl b/apps/barrel_inference_server/src/barrel_inference_server_h_chat.erl index 7d98128..d268b9e 100644 --- a/apps/barrel_inference_server/src/barrel_inference_server_h_chat.erl +++ b/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(<>) 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 diff --git a/apps/barrel_inference_server/src/barrel_inference_server_h_messages.erl b/apps/barrel_inference_server/src/barrel_inference_server_h_messages.erl index ad18619..2b49fc2 100644 --- a/apps/barrel_inference_server/src/barrel_inference_server_h_messages.erl +++ b/apps/barrel_inference_server/src/barrel_inference_server_h_messages.erl @@ -16,8 +16,10 @@ maybe_end_session/1, record_session_committed/2, release_slot/1, - drive_stream/2, - drive_buffered/1 + drive_stream_post_admit/2, + drive_buffered/1, + resolve_stream/3, + pre_admit_loop/1 ]} ). @@ -216,29 +218,99 @@ start_inference(R0, ExtraHeaders0) -> ExtraHeaders = ExtraHeaders0 ++ ratelimit_headers(Real), case R1#barrel_inference_request.stream of true -> - livery_resp:stream( - 200, - sse_headers() ++ ExtraHeaders, - fun(Emit) -> drive_stream(make_init(R1, Requested, ExtraHeaders), Emit) end + livery_resp:stream_deferred( + fun() -> resolve_stream(R1, Requested, ExtraHeaders) end ); false -> drive_buffered(make_init(R1, Requested, ExtraHeaders)) end. +%% Sync pre-admit: emit a real 4xx/5xx status when admission fails +%% before any SSE frame is written; otherwise hand the admitted state +%% off to the streaming producer. +resolve_stream(R, Requested, ExtraHeaders) -> + State = arm_total_timer(make_init(R, Requested, ExtraHeaders)), + case pre_admit_loop(State) of + {admitted, State1} -> + %% Native SSE: each Emit receives a `#{event, data}' map + %% and livery's sse_frame wraps it into the wire form. + {sse, 200, sse_extras() ++ ExtraHeaders, fun(Emit) -> + drive_stream_post_admit(State1, Emit) + end}; + {error, Status0, Reason, State1} -> + cleanup(State1), + %% Honour the 503 -> 529 + Retry-After Anthropic remap on + %% the pre-admit error path the same way the buffered + %% response path does it via anthropic_json_reply_h. + {Status, BodyStatus, Headers} = anthropic_overload_remap(Status0, ExtraHeaders), + Body = json:encode(anthropic_error_body_h(BodyStatus, Reason, Headers)), + {full, Status, Headers ++ json_headers(), Body} + end. + make_init(R, Requested, ExtraHeaders) -> {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R), init_state(R, Requested, Worker, Mon, ExtraHeaders). -drive_stream(State0, Emit) -> - State = arm_total_timer(State0), +drive_stream_post_admit(State, Emit) -> + %% In the cowboy-era flow `on_pipeline_stream({pipeline, admitted, + %% _, _})' was the spot where we first had both the Emit fun and + %% an admitted state, so it emitted `message_start' and armed the + %% per-generation ping timer. pre_admit_loop now consumes the + %% admit message before any Emit exists; the post-admit producer + %% takes over both responsibilities here. + State1 = arm_gen_ping(emit_message_start(State, Emit)), _ = try - stream_loop(State, Emit) + stream_loop(State1, Emit) after - cleanup(State) + cleanup(State1) end, ok. +%% Selective receive — see the matching comment in h_chat. Token / +%% done / error messages from the engine stay in the mailbox so +%% the post-admit stream_loop picks them up in order. +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_tokens = length(Tokens), + 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(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(State0) -> State = arm_total_timer(State0), try @@ -340,7 +412,7 @@ dispatch_stream({idle_timeout, Ref}, S, Emit) when S#st.ref =:= Ref -> barrel_inference:cancel(Ref), finish_stream_err(S, generation_idle_timeout, Emit); dispatch_stream({gen_ping, Ref}, S, Emit) when S#st.ref =:= Ref -> - _ = emit_raw(Emit, anthropic_ping_frame()), + _ = emit_event(Emit, anthropic_ping_frame()), stream_loop(arm_gen_ping(S), Emit); dispatch_stream(total_request_timeout, S, Emit) -> on_total_timeout(S, Emit); @@ -348,7 +420,7 @@ dispatch_stream(_Other, S, Emit) -> stream_loop(S, Emit). on_pipeline_stream({pipeline, loading, _M}, S, Emit) -> - _ = emit_raw(Emit, anthropic_ping_frame()), + _ = emit_event(Emit, anthropic_ping_frame()), stream_loop(S, Emit); on_pipeline_stream({pipeline, loaded}, S = #st{keepalive_begun = true}, Emit) -> stream_loop(S#st{phase = waiting_template}, Emit); @@ -376,7 +448,7 @@ on_pipeline_stream({pipeline, admitted, Ref, Slot}, S, Emit) -> on_pipeline_stream({pipeline, error, Status, Reason}, S, Emit) -> record_metrics(S, Status), Err = anthropic_error_body(Status, Reason, S), - _ = emit_raw(Emit, anthropic_event_frame(<<"error">>, Err)), + _ = emit_event(Emit, anthropic_event_frame(<<"error">>, Err)), S. on_text_token(S0, Ref, Tok, Emit) -> @@ -420,7 +492,7 @@ stream_text_emit(Tok, S, Emit) -> S1#st.req_id, S1#st.requested ), - case emit_raw(Emit, Frame) of + case emit_event(Emit, Frame) of ok -> stream_loop(rearm_idle(S1#st{out_tokens = S1#st.out_tokens + 1}), Emit); closed -> @@ -441,7 +513,7 @@ on_thinking_token(S0, Ref, Tok, Emit) -> S1#st.req_id, S1#st.requested ), - case emit_raw(Emit, Frame) of + case emit_event(Emit, Frame) of ok -> stream_loop(rearm_idle(S1), Emit); closed -> S1 end. @@ -470,11 +542,7 @@ handle_thinking_end_visible(S = #st{thinking_block_started = Idx}, Sig, Emit) wh <<"signature">> => base64:encode(Sig) } }, - _ = emit_raw(Emit, [ - <<"event: content_block_delta\ndata: ">>, - json:encode(Delta), - <<"\n\n">> - ]) + _ = emit_event(Emit, anthropic_event_frame(<<"content_block_delta">>, Delta)) end, S1 = maybe_close_thinking(S, Emit), stream_loop(rearm_idle(S1#st{thinking_signature = Sig}), Emit); @@ -534,7 +602,7 @@ finish_stream_ok(S0 = #st{mode = text}, Stats0, Emit) -> Stop = barrel_inference_server_translate:internal_to_anthropic_event( message_stop, #{}, S1#st.req_id, S1#st.requested ), - _ = emit_raw(Emit, [Delta, Stop]), + _ = emit_events(Emit, [Delta, Stop]), record_success(S1, Stats), S1; finish_stream_ok(S0 = #st{mode = tool_buffer}, Stats0, Emit) -> @@ -554,7 +622,7 @@ finish_stream_ok(S0 = #st{mode = tool_buffer}, Stats0, Emit) -> MsgStop = barrel_inference_server_translate:internal_to_anthropic_event( message_stop, #{}, S#st.req_id, S#st.requested ), - _ = emit_raw(Emit, [Frames, MsgDelta, MsgStop]), + _ = emit_events(Emit, Frames ++ [MsgDelta, MsgStop]), record_success(S, StatsToolCall), S. @@ -572,7 +640,7 @@ finish_with_tool_uses_stream(S0, Calls, Emit) -> MsgStop = barrel_inference_server_translate:internal_to_anthropic_event( message_stop, #{}, S#st.req_id, S#st.requested ), - _ = emit_raw(Emit, [Frames, MsgDelta, MsgStop]), + _ = emit_events(Emit, Frames ++ [MsgDelta, MsgStop]), record_success(S, StatsTC), S. @@ -599,12 +667,8 @@ tool_use_block_frames(S, Idx, ToolId, Name, Input) -> {content_block_stop, Idx}, #{}, S#st.req_id, S#st.requested ), [ - <<"event: content_block_start\ndata: ">>, - json:encode(Start), - <<"\n\n">>, - <<"event: content_block_delta\ndata: ">>, - json:encode(DeltaInput), - <<"\n\n">>, + anthropic_event_frame(<<"content_block_start">>, Start), + anthropic_event_frame(<<"content_block_delta">>, DeltaInput), Stop ]. @@ -628,7 +692,7 @@ close_open_blocks(S0, Emit) -> Stop = barrel_inference_server_translate:internal_to_anthropic_event( {content_block_stop, TIdx}, #{}, S1#st.req_id, S1#st.requested ), - _ = emit_raw(Emit, Stop), + _ = emit_event(Emit, Stop), S1#st{text_block_started = undefined} end. @@ -638,7 +702,7 @@ emit_message_start(S, Emit) -> Iolist = barrel_inference_server_translate:internal_to_anthropic_event( {message_start, S#st.prompt_tokens}, #{}, S#st.req_id, S#st.requested ), - _ = emit_raw(Emit, Iolist), + _ = emit_event(Emit, Iolist), S#st{message_started = true}. ensure_text_block_started(S = #st{text_block_started = I}, _Emit) when @@ -651,7 +715,7 @@ ensure_text_block_started(S0, Emit) -> Iolist = barrel_inference_server_translate:internal_to_anthropic_event( {content_block_start_text, Idx}, #{}, S1#st.req_id, S1#st.requested ), - _ = emit_raw(Emit, Iolist), + _ = emit_event(Emit, Iolist), S1#st{text_block_started = Idx}. ensure_thinking_block_started(S = #st{thinking_block_started = I}, _Emit) when @@ -667,11 +731,7 @@ ensure_thinking_block_started(S, Emit) -> <<"type">> => <<"thinking">>, <<"thinking">> => <<>> } }, - _ = emit_raw(Emit, [ - <<"event: content_block_start\ndata: ">>, - json:encode(Payload), - <<"\n\n">> - ]), + _ = emit_event(Emit, anthropic_event_frame(<<"content_block_start">>, Payload)), S#st{thinking_block_started = Idx}. maybe_close_thinking(S = #st{thinking_block_started = undefined}, _Emit) -> @@ -680,7 +740,7 @@ maybe_close_thinking(S = #st{thinking_block_started = Idx}, Emit) -> Frame = barrel_inference_server_translate:internal_to_anthropic_event( {content_block_stop, Idx}, #{}, S#st.req_id, S#st.requested ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S#st{thinking_block_started = undefined}. next_block_index(#st{text_block_started = T, thinking_block_started = K}) -> @@ -701,7 +761,7 @@ maybe_set_tool_call_finish_reason(_, Stats) -> finish_stream_err(S, Reason, Emit) -> Status = http_status(Reason), Err = anthropic_error_body(Status, Reason, S), - _ = emit_raw(Emit, anthropic_event_frame(<<"error">>, Err)), + _ = emit_event(Emit, anthropic_event_frame(<<"error">>, Err)), record_error(S, Reason), S. @@ -1355,24 +1415,28 @@ cache_log_fields(Stats) -> %% Reply helpers %%==================================================================== -emit_raw(Emit, IoData) -> - case Emit(IoData) of +%% Emit one livery sse event map (`#{event => _, data => _}'); the +%% decision-fun's `{sse, ...}' wrapper handles the wire framing. +emit_event(Emit, EventMap) -> + case Emit(EventMap) of ok -> ok; {error, _} -> closed end. +%% Emit a list of event maps, short-circuiting on peer disconnect. +emit_events(_Emit, []) -> + ok; +emit_events(Emit, [F | Rest]) -> + case emit_event(Emit, F) of + ok -> emit_events(Emit, Rest); + closed -> closed + end. + anthropic_ping_frame() -> anthropic_event_frame(<<"ping">>, #{<<"type">> => <<"ping">>}). anthropic_event_frame(EventName, JsonMap) -> - [ - <<"event: ">>, - EventName, - <<"\n">>, - <<"data: ">>, - json:encode(JsonMap), - <<"\n\n">> - ]. + #{event => EventName, data => json:encode(JsonMap)}. %% Base response headers built at request translate time: anthropic-version %% echo + request-id mirror (literal "request-id" alongside the @@ -1482,12 +1546,10 @@ 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">>} - ]. +%% livery's `{sse, _}' decision already adds content-type + +%% cache-control defaults; we only need the nginx hint. +sse_extras() -> + [{<<"x-accel-buffering">>, <<"no">>}]. decode(Body) -> try json:decode(Body) of diff --git a/apps/barrel_inference_server/src/barrel_inference_server_h_ollama.erl b/apps/barrel_inference_server/src/barrel_inference_server_h_ollama.erl index 187c920..e5fc7da 100644 --- a/apps/barrel_inference_server/src/barrel_inference_server_h_ollama.erl +++ b/apps/barrel_inference_server/src/barrel_inference_server_h_ollama.erl @@ -115,29 +115,69 @@ ok_preload(Op, Model, KeepAlive, MonoStart) -> apply_keep_alive(Model, KeepAlive), livery_resp:json(200, Body). -%% Streaming returns livery_resp:stream with the NDJSON producer fun; -%% non-streaming buffers tokens and returns a one-shot json. +%% Streaming uses livery_resp:stream_deferred so pool_exhausted / +%% not_loaded / etc surface as a real 4xx/5xx JSON response instead +%% of `200 OK` + an NDJSON error line; non-streaming buffers tokens. inference(Op, R, KeepAlive) -> case R#barrel_inference_request.stream of true -> - livery_resp:stream( - 200, - [{<<"content-type">>, <<"application/x-ndjson">>}], - fun(Emit) -> drive(Op, R, KeepAlive, Emit) end + livery_resp:stream_deferred( + fun() -> resolve_stream(Op, R, KeepAlive) end ); false -> drive_buffered(Op, R, KeepAlive) end. -drive(Op, R, KeepAlive, Emit) -> +resolve_stream(Op, R, KeepAlive) -> {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R), State = init_state(Op, R, KeepAlive, Worker, Mon), + case pre_admit_loop(State) of + {admitted, State1} -> + %% Native NDJSON: livery's Emit takes a JSON-encodable + %% term, json:encodes it, and appends `\n'. The handler's + %% line builders (ollama_chunk / final_chunk / error_body / + %% loading_line) now return maps directly. + {ndjson, 200, [], fun(Emit) -> drive_post_admit(State1, Emit) end}; + {error, Status, Reason, State1} -> + cleanup(State1), + Body = json:encode(error_body(Reason)), + {full, Status, [{<<"content-type">>, <<"application/json">>}], Body} + end. + +drive_post_admit(State, Emit) -> try stream_loop(State, Emit) after cleanup(State) end. +%% Selective receive — see h_chat for why no catch-all. +pre_admit_loop(S) -> + receive + {pipeline, error, Status, Reason} -> + {error, Status, Reason, S}; + {pipeline, loading, _M} -> + pre_admit_loop(S); + {pipeline, loaded} -> + ok = barrel_inference_server_keepalive:request_begin(S#st.model), + pre_admit_loop(S#st{phase = waiting_template, mono_loaded = mono_ms()}); + {pipeline, templated, _Tokens, _ParamsRef} -> + pre_admit_loop(S#st{phase = waiting_queue}); + {pipeline, queued} -> + pre_admit_loop(S#st{phase = waiting_admit}); + {pipeline, admitted, Ref, Slot} -> + {admitted, on_admit(S, Ref, Slot)}; + {'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(). + drive_buffered(Op, R, KeepAlive) -> {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R), State = init_state(Op, R, KeepAlive, Worker, Mon), @@ -185,7 +225,7 @@ dispatch_stream({barrel_inference_reasoning_token, Ref, _Tok}, S, Emit) -> stream_loop(learn_ref(S, Ref), Emit); dispatch_stream({barrel_inference_done, Ref, Stats}, S, Emit) -> S1 = learn_ref(S, Ref), - _ = emit_line(Emit, final_chunk(S1, Stats)), + _ = emit_term(Emit, final_chunk(S1, Stats)), S1; dispatch_stream({barrel_inference_error, Ref, Reason}, S, Emit) -> on_engine_error(S, Ref, Reason, Emit); @@ -197,16 +237,17 @@ dispatch_stream(_Other, S, Emit) -> on_token(S, Ref, Tok, Emit) -> S1 = learn_ref(S, Ref), Chunk = ollama_chunk(S1#st.op, Tok, S1#st.req_id, S1#st.model), - case emit_line(Emit, Chunk) of + case emit_term(Emit, Chunk) of ok -> stream_loop(S1#st{out_tokens = S1#st.out_tokens + 1}, Emit); closed -> S1 end. +%% pre_admit_loop consumes loading/loaded/templated/queued/admitted +%% BEFORE we have an Emit fun, so the pipeline-progress clauses below +%% only fire on the unusual late-arrival case (re-load round during the +%% agentic loop, etc). on_pipeline_stream({pipeline, loading, _Model}, S, Emit) -> - case emit_line(Emit, loading_line(S)) of - ok -> stream_loop(S, Emit); - closed -> S - end; + stream_loop(S, Emit); on_pipeline_stream({pipeline, loaded}, S, Emit) -> ok = barrel_inference_server_keepalive:request_begin(S#st.model), stream_loop(S#st{phase = waiting_template, mono_loaded = mono_ms()}, Emit); @@ -217,22 +258,14 @@ on_pipeline_stream({pipeline, queued}, S, Emit) -> on_pipeline_stream({pipeline, admitted, Ref, Slot}, S, Emit) -> stream_loop(on_admit(S, Ref, Slot), Emit); on_pipeline_stream({pipeline, error, _Status, Reason}, S, Emit) -> - _ = emit_line(Emit, json:encode(error_body(Reason))), + _ = emit_term(Emit, error_body(Reason)), S. on_engine_error(S, Ref, Reason, Emit) -> S1 = learn_ref(S, Ref), - _ = emit_line(Emit, json:encode(error_body(Reason))), + _ = emit_term(Emit, error_body(Reason)), S1. -loading_line(S) -> - json:encode(#{ - <<"model">> => S#st.model, - <<"created_at">> => iso8601_now(), - <<"status">> => <<"loading">>, - <<"done">> => false - }). - %% Buffered (non-stream) receive loop. Accumulates the token text and %% returns a single livery_resp:json on done. buffered_loop(S) -> @@ -290,8 +323,11 @@ on_admit(S = #st{ref = undefined}, Ref, Slot) -> on_admit(S, _Ref, Slot) -> S#st{slot = Slot}. -emit_line(Emit, Body) -> - case Emit([Body, <<"\n">>]) of +%% livery's ndjson Emit takes a JSON-encodable term and writes +%% `<>'. Wrap to surface the +%% peer-disconnect signal the receive loop uses to short-circuit. +emit_term(Emit, Term) -> + case Emit(Term) of ok -> ok; {error, _} -> closed end. @@ -421,10 +457,3 @@ error_status(_) -> 500. mono_ms() -> erlang:monotonic_time(millisecond). - -iso8601_now() -> - Now = erlang:system_time(second), - {{Y, Mo, D}, {H, M, S}} = calendar:system_time_to_universal_time(Now, second), - list_to_binary( - io_lib:format("~4..0B-~2..0B-~2..0BT~2..0B:~2..0B:~2..0BZ", [Y, Mo, D, H, M, S]) - ). diff --git a/apps/barrel_inference_server/src/barrel_inference_server_h_responses.erl b/apps/barrel_inference_server/src/barrel_inference_server_h_responses.erl index 196e00c..94b800e 100644 --- a/apps/barrel_inference_server/src/barrel_inference_server_h_responses.erl +++ b/apps/barrel_inference_server/src/barrel_inference_server_h_responses.erl @@ -17,8 +17,10 @@ maybe_end_session/1, record_session_committed/2, release_slot/1, - drive_stream/3, - drive_buffered/2 + drive_stream_post_admit/2, + drive_buffered/2, + resolve_stream/2, + pre_admit_loop/1 ]} ). @@ -150,19 +152,40 @@ start_inference(R0) -> 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, Emit) end + livery_resp:stream_deferred( + fun() -> resolve_stream(R1, Requested) end ); false -> drive_buffered(R1, Requested) end. -drive_stream(R, Requested, Emit) -> +%% Sync pre-admit so pool_exhausted / not_loaded / etc surface as a +%% real 4xx/5xx JSON envelope rather than `200 OK` + SSE error. +resolve_stream(R, Requested) -> {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R), State = init_state(R, Requested, Worker, Mon, true), State1 = arm_total_timer(State), + case pre_admit_loop(State1) of + {admitted, State2} -> + %% Native SSE: livery wraps each Emit map into the wire + %% `event: ..\ndata: ..\n\n' frame; the handler stays in + %% the map shape end-to-end. + {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) -> + %% pre_admit_loop already consumed `{pipeline, admitted, _, _}', + %% so emit the response.created / message_added / content_added + %% trio here (the cowboy-era flow did it on admit). + State1 = emit_content_added( + emit_message_added(emit_response_created(State, Emit), Emit), Emit + ), _ = try stream_loop(State1, Emit) @@ -171,6 +194,47 @@ drive_stream(R, Requested, Emit) -> end, ok. +%% Selective receive — see h_chat for the rationale. +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) -> {Worker, Mon} = barrel_inference_server_pipeline:start_link(self(), R), State = init_state(R, Requested, Worker, Mon, false), @@ -272,10 +336,12 @@ 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, Emit); - closed -> S - end; + %% pre_admit_loop already consumes the initial loading messages + %% before the stream opens, so this clause only fires on a re-load + %% round inside the agentic loop. livery's sse_frame has no + %% comment shape; the silent re-load relies on the connection's + %% own keepalive (idle_timeout backstop). + 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) -> @@ -311,14 +377,14 @@ on_pipeline_stream({pipeline, error, Status, Reason}, S, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.failed">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S. on_stream_token(S0, Ref, Tok, Emit) -> S = learn_ref(S0, Ref, Emit), case handle_token_stream(Tok, S) of {emit, Frame, S1} -> - case emit_raw(Emit, Frame) of + case emit_event(Emit, Frame) of ok -> stream_loop(rearm_idle(S1), Emit); closed -> S1 end; @@ -406,7 +472,7 @@ finish_stream_ok(S0 = #st{mode = text}, Stats, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.completed">>, CompletedPayload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), record_success(S1, Stats), store_response(S1, Items), S1; @@ -419,7 +485,7 @@ finish_stream_ok(S0 = #st{mode = tool_buffer}, Stats0, Emit) -> OutIdx = S1#st.out_index, ArgsBin = args_bin(Input), Frames = function_call_frames(OutIdx, FcId, CallId, Name, ArgsBin), - _ = emit_raw(Emit, Frames), + _ = emit_events(Emit, Frames), Stats = maps:put(finish_reason, tool_call, Stats0), Items = S1#st.fc_items ++ [fc_item_map(FcId, CallId, Name, ArgsBin)], CompletedPayload = barrel_inference_server_translate:internal_to_responses_completed( @@ -428,7 +494,7 @@ finish_stream_ok(S0 = #st{mode = tool_buffer}, Stats0, Emit) -> DoneFrame = barrel_inference_server_translate:responses_event( <<"response.completed">>, CompletedPayload ), - _ = emit_raw(Emit, DoneFrame), + _ = emit_event(Emit, DoneFrame), record_success(S1, Stats), store_response(S1, Items), S1#st{out_index = OutIdx + 1}. @@ -455,7 +521,7 @@ emit_function_item(S, Emit, #{id := FcId, name := Name, input := Input}) -> OutIdx = S#st.out_index, ArgsBin = iolist_to_binary(json:encode(Input)), Frames = function_call_frames(OutIdx, FcId, CallId, Name, ArgsBin), - _ = emit_raw(Emit, Frames), + _ = emit_events(Emit, Frames), {fc_item_map(FcId, CallId, Name, ArgsBin), S#st{out_index = OutIdx + 1}}. function_call_frames(OutIdx, FcId, CallId, Name, ArgsBin) -> @@ -521,7 +587,7 @@ close_open_message_stream(S = #st{started_text_part = true}, Emit) -> <<"response.output_item.done">>, MsgDone ) ], - _ = emit_raw(Emit, Frames), + _ = emit_events(Emit, Frames), S#st{ started_text_part = false, out_index = S#st.out_index + 1, @@ -554,7 +620,7 @@ emit_response_created(S, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.created">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S#st{created_sent = true}. emit_message_added(S, Emit) -> @@ -564,7 +630,7 @@ emit_message_added(S, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.output_item.added">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S. emit_content_added(S, Emit) -> @@ -574,7 +640,7 @@ emit_content_added(S, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.content_part.added">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S#st{started_text_part = true}. finish_stream_err(S, Reason, Emit) -> @@ -587,7 +653,7 @@ finish_stream_err(S, Reason, Emit) -> Frame = barrel_inference_server_translate:responses_event( <<"response.failed">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), record_error(S, Reason), S. @@ -694,7 +760,7 @@ emit_server_tool_call(S, Emit, CallId, Name) -> Frame = barrel_inference_server_translate:responses_event( <<"response.output_item.added">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S. emit_server_tool_done(S, Emit, CallId, Name, _ResultJson) -> @@ -706,7 +772,7 @@ emit_server_tool_done(S, Emit, CallId, Name, _ResultJson) -> Frame = barrel_inference_server_translate:responses_event( <<"response.output_item.done">>, Payload ), - _ = emit_raw(Emit, Frame), + _ = emit_event(Emit, Frame), S#st{out_index = OutIdx + 1}. server_tool_item(CallId, Name, Status) -> @@ -1288,12 +1354,21 @@ cache_log_fields(Stats) -> %% Reply helpers %%==================================================================== -emit_raw(Emit, IoData) -> - case Emit(IoData) of +%% Emit one livery sse event map (`#{event => _, data => _}'). +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( @@ -1363,12 +1438,10 @@ 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">>} - ]. +%% livery's `{sse, _}' decision already adds content-type + +%% cache-control; we only need the nginx hint. +sse_extras() -> + [{<<"x-accel-buffering">>, <<"no">>}]. decode(Body) -> try json:decode(Body) of diff --git a/apps/barrel_inference_server/src/barrel_inference_server_translate.erl b/apps/barrel_inference_server/src/barrel_inference_server_translate.erl index 1d4db20..98a4418 100644 --- a/apps/barrel_inference_server/src/barrel_inference_server_translate.erl +++ b/apps/barrel_inference_server/src/barrel_inference_server_translate.erl @@ -75,6 +75,7 @@ internal_to_responses_function_call_done/4, internal_to_responses_failed/4, responses_event/2, + sse_iodata/1, %% helpers make_id/1, parse_keep_alive/1, @@ -838,7 +839,7 @@ internal_to_responses_failed(ResponseId, Model, Code, Message) -> }. %% SSE renderer for /v1/responses events. Same wire shape as sse/2. --spec responses_event(binary(), map()) -> iodata(). +-spec responses_event(binary(), map()) -> map(). responses_event(EventName, Payload) -> sse(EventName, Payload). @@ -953,7 +954,7 @@ cache_creation_ttl_split(Total, Stats) -> %% `event: \ndata: \n\n`. -spec internal_to_anthropic_event( anthropic_event_kind(), map(), binary(), binary() -) -> iodata(). +) -> map(). internal_to_anthropic_event({message_start, PromptTokens}, _Acc, ReqId, Model) when is_integer(PromptTokens) -> @@ -1385,29 +1386,29 @@ int_seconds(B) -> %% Streaming generate chunk: one JSON object per token. Caller writes %% it as a single NDJSON line. --spec internal_to_ollama_generate_chunk(binary(), binary(), binary()) -> iodata(). +%% Returns the JSON-encodable map for one streaming NDJSON line; the +%% caller (or livery's ndjson Emit) does the json:encode + newline. +-spec internal_to_ollama_generate_chunk(binary(), binary(), binary()) -> map(). internal_to_ollama_generate_chunk(Token, _ReqId, Model) -> - json:encode(#{ + #{ <<"model">> => Model, <<"created_at">> => iso8601_now(), <<"response">> => Token, <<"done">> => false - }). + }. -%% Streaming generate final: emits the closing JSON with timing. --spec internal_to_ollama_generate_final(map(), binary(), binary(), map()) -> iodata(). +%% Streaming generate final NDJSON map (timings + done). +-spec internal_to_ollama_generate_final(map(), binary(), binary(), map()) -> map(). internal_to_ollama_generate_final(Stats, _ReqId, Model, Timings) -> - json:encode( - maps:merge( - #{ - <<"model">> => Model, - <<"created_at">> => iso8601_now(), - <<"response">> => <<>>, - <<"done">> => true, - <<"done_reason">> => done_reason_atom(Stats) - }, - timing_fields(Stats, Timings) - ) + maps:merge( + #{ + <<"model">> => Model, + <<"created_at">> => iso8601_now(), + <<"response">> => <<>>, + <<"done">> => true, + <<"done_reason">> => done_reason_atom(Stats) + }, + timing_fields(Stats, Timings) ). %% Non-streaming generate response. @@ -1426,28 +1427,26 @@ internal_to_ollama_generate_response(Body, Stats, Model, Timings) -> ) ). --spec internal_to_ollama_chat_chunk(binary(), binary(), binary()) -> iodata(). +-spec internal_to_ollama_chat_chunk(binary(), binary(), binary()) -> map(). internal_to_ollama_chat_chunk(Token, _ReqId, Model) -> - json:encode(#{ + #{ <<"model">> => Model, <<"created_at">> => iso8601_now(), <<"message">> => #{<<"role">> => <<"assistant">>, <<"content">> => Token}, <<"done">> => false - }). + }. --spec internal_to_ollama_chat_final(map(), binary(), binary(), map()) -> iodata(). +-spec internal_to_ollama_chat_final(map(), binary(), binary(), map()) -> map(). internal_to_ollama_chat_final(Stats, _ReqId, Model, Timings) -> - json:encode( - maps:merge( - #{ - <<"model">> => Model, - <<"created_at">> => iso8601_now(), - <<"message">> => #{<<"role">> => <<"assistant">>, <<"content">> => <<>>}, - <<"done">> => true, - <<"done_reason">> => done_reason_atom(Stats) - }, - timing_fields(Stats, Timings) - ) + maps:merge( + #{ + <<"model">> => Model, + <<"created_at">> => iso8601_now(), + <<"message">> => #{<<"role">> => <<"assistant">>, <<"content">> => <<>>}, + <<"done">> => true, + <<"done_reason">> => done_reason_atom(Stats) + }, + timing_fields(Stats, Timings) ). -spec internal_to_ollama_chat_response(binary(), map(), binary(), map()) -> iodata(). @@ -2088,6 +2087,14 @@ index_zero(L) -> index_zero(L, 0). index_zero([], _) -> []; index_zero([H | T], I) -> [{I, H} | index_zero(T, I + 1)]. -%% Render an Anthropic SSE event (named event + data line). +%% Build a livery sse-Emit-ready map: `#{event => Bin, data => Bin}'. +%% Livery (`livery.erl:sse_frame/1') formats this as +%% `event: \ndata: \n\n', so handlers can pass the +%% map straight to Emit/1 with no manual framing. sse(EventName, Payload) -> - [<<"event: ">>, EventName, <<"\ndata: ">>, json:encode(Payload), <<"\n\n">>]. + #{event => EventName, data => json:encode(Payload)}. + +%% Render an sse event map as the same iodata `sse/2' used to return. +%% Exposed for the translate test suite, which inspects the wire bytes. +sse_iodata(#{event := EventName, data := Data}) -> + [<<"event: ">>, EventName, <<"\ndata: ">>, Data, <<"\n\n">>]. diff --git a/apps/barrel_inference_server/test/barrel_inference_server_e2e_SUITE.erl b/apps/barrel_inference_server/test/barrel_inference_server_e2e_SUITE.erl index 4699e9e..de3ba7f 100644 --- a/apps/barrel_inference_server/test/barrel_inference_server_e2e_SUITE.erl +++ b/apps/barrel_inference_server/test/barrel_inference_server_e2e_SUITE.erl @@ -60,8 +60,8 @@ init_per_suite(Config) -> ), application:set_env(barrel_inference_server, model_load_policy, preloaded), {ok, Started} = application:ensure_all_started(barrel_inference_server), - {ok, _} = application:ensure_all_started(inets), - httpc:set_options([ + {ok, _} = barrel_inference_server_http_test:ensure_started(), + ok = barrel_inference_server_http_test:set_options([ {max_sessions, 16}, {max_keep_alive_length, 0}, {pipeline_timeout, 0} @@ -206,7 +206,9 @@ chat_non_streaming_returns_full_response(Cfg) -> <<"stream">> => false }), {ok, {{_, Code, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), ct:log("chat non-streaming code=~p body=~ts", [Code, RespBody]), ?assertEqual(200, Code), Decoded = json:decode(list_to_binary(RespBody)), @@ -217,14 +219,81 @@ chat_non_streaming_returns_full_response(Cfg) -> Usage = maps:get(<<"usage">>, Decoded), ?assert(maps:get(<<"completion_tokens">>, Usage) >= 1). -chat_busy_returns_429(_Cfg) -> - %% TODO: livery_resp:stream/3 picks the response status at the - %% time the stream opens, so a `stream: true' request that hits - %% pool_exhausted lands as a 200 + SSE `error' event rather - %% than the cowboy-era 429. Either gate the stream open on - %% admission success (deferred response status) or change the - %% wire contract; both are out of scope for the listener swap. - {skip, livery_streaming_status_deferred}. +chat_busy_returns_429(Cfg) -> + %% concurrency=1 + depth=1: holder takes the slot, one waiter can + %% queue, anything else gets pool_exhausted (429). With the + %% livery 0.3.0 deferred-status producer fun the pre-admit + %% pool_exhausted now surfaces as HTTP 429 (instead of `200 OK' + %% + SSE error) and the retried R1+R2 burst observes it. + Url = ?config(base, Cfg) ++ "/v1/chat/completions", + HolderBody = make_busy_body(<<"busy-holder">>), + Parent = self(), + Holder = spawn(fun() -> + Resp = http_collect(Url, HolderBody), + Parent ! {holder_done, byte_size(Resp)} + end), + timer:sleep(150), + ok = busy_burst_until_pool_exhausted(Url, 30, 200), + exit(Holder, kill), + receive + {holder_done, _} -> ok + after 0 -> ok + end, + [barrel_inference:cancel(R) || {R, _} <- barrel_inference_inflight:all()], + ok. + +busy_burst_until_pool_exhausted(_Url, 0, _BackoffMs) -> + ct:fail("busy burst exhausted retries without seeing 429 or 504"); +busy_burst_until_pool_exhausted(Url, N, BackoffMs) -> + Parent = self(), + Tag = integer_to_binary(erlang:unique_integer([positive])), + BodyA = make_busy_body(<<"busy-a-", Tag/binary>>), + BodyB = make_busy_body(<<"busy-b-", Tag/binary>>), + spawn(fun() -> + Parent ! + {a, + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", BodyA}, [{timeout, 2000}], [] + )} + end), + spawn(fun() -> + Parent ! + {b, + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", BodyB}, [{timeout, 2000}], [] + )} + end), + A = + receive + {a, X} -> X + after 3000 -> {error, timeout} + end, + B = + receive + {b, Y} -> Y + after 3000 -> {error, timeout} + end, + Codes = [code_of(A), code_of(B)], + case lists:member(429, Codes) orelse lists:member(504, Codes) of + true -> + ok; + false -> + timer:sleep(BackoffMs), + busy_burst_until_pool_exhausted(Url, N - 1, BackoffMs) + end. + +code_of({ok, {{_, Code, _}, _, _}}) -> Code; +code_of(_) -> 0. + +make_busy_body(Prompt) -> + iolist_to_binary( + json:encode(#{ + <<"model">> => <<"e2e-stub">>, + <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => Prompt}], + <<"max_tokens">> => 50000, + <<"stream">> => true + }) + ). chat_cancel_on_disconnect_releases_slot(Cfg) -> %% Open a streaming chat, read the first chunk, then drop the @@ -303,7 +372,9 @@ chat_cancel_on_disconnect_releases_slot(Cfg) -> post_until_slot_free(_Url, _Body, 0, _BackoffMs) -> timeout; post_until_slot_free(Url, Body, N, BackoffMs) -> - case httpc:request(post, {Url, [], "application/json", Body}, [], []) of + case + barrel_inference_server_http_test:request(post, {Url, [], "application/json", Body}, [], []) + of {ok, {{_, 200, _}, _, _}} -> ct:log("post_until_slot_free: 200 after ~p retries", [N]), 200; @@ -353,7 +424,9 @@ embeddings_returns_vector(Cfg) -> <<"input">> => <<"hello">> }), {ok, {{_, 200, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Decoded = json:decode(list_to_binary(RespBody)), [Entry] = maps:get(<<"data">>, Decoded), Vec = maps:get(<<"embedding">>, Entry), @@ -375,5 +448,7 @@ http_collect(Url, Body) -> %% time out before we read the closing chunk. HttpOpts = [{timeout, 20000}, {connect_timeout, 5000}], {ok, {{_, 200, _}, _Headers, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, HttpOpts, []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, HttpOpts, [] + ), list_to_binary(RespBody). diff --git a/apps/barrel_inference_server/test/barrel_inference_server_e2e_cli_SUITE.erl b/apps/barrel_inference_server/test/barrel_inference_server_e2e_cli_SUITE.erl index 079c2a9..37d234f 100644 --- a/apps/barrel_inference_server/test/barrel_inference_server_e2e_cli_SUITE.erl +++ b/apps/barrel_inference_server/test/barrel_inference_server_e2e_cli_SUITE.erl @@ -79,7 +79,7 @@ init_per_suite(Config) -> application:set_env(barrel_inference_server, keep_alive_default_ms, 300000), {ok, Started} = application:ensure_all_started(barrel_inference_server), - {ok, _} = application:ensure_all_started(inets), + {ok, _} = barrel_inference_server_http_test:ensure_started(), Base = "http://127.0.0.1:" ++ integer_to_list(Port), Escript = locate_escript(), [ @@ -354,7 +354,7 @@ file_spec(Cfg) -> post_json(Cfg, Path, Body) -> Url = ?config(base, Cfg) ++ Path, - httpc:request(post, {Url, [], "application/json", Body}, [], []). + barrel_inference_server_http_test:request(post, {Url, [], "application/json", Body}, [], []). ndjson_lines(Bin) -> Parts = binary:split(Bin, <<"\n">>, [global]), diff --git a/apps/barrel_inference_server/test/barrel_inference_server_http_test.erl b/apps/barrel_inference_server/test/barrel_inference_server_http_test.erl new file mode 100644 index 0000000..481ee2a --- /dev/null +++ b/apps/barrel_inference_server/test/barrel_inference_server_http_test.erl @@ -0,0 +1,70 @@ +%%% Hackney-backed compat shim for the CT suites. Mirrors the +%%% `httpc:request/1,2,4` argument shape and return tuple so the +%%% existing test bodies can call `?MODULE:request(...)` without any +%%% other rewriting. Hackney replaces `httpc` everywhere in this +%%% project — see the http-client-preference memory. + +-module(barrel_inference_server_http_test). + +-export([request/1, request/2, request/4]). +-export([set_options/1, ensure_started/0]). + +%% Boot hackney's apps; CT suites call this from init_per_suite where +%% they previously called `application:ensure_all_started(inets)`. +ensure_started() -> + application:ensure_all_started(hackney). + +%% No-op (kept so the call sites don't have to change). httpc had +%% `set_options/1`; hackney's tuning is per-request via Opts. +set_options(_Opts) -> ok. + +%% Shape parity with httpc: +%% request(Url) +%% request(get, {Url, Headers}, HttpOpts, Opts) +%% request(post, {Url, Headers, ContentType, Body}, HttpOpts, Opts) +request(Url) -> + do(get, to_bin(Url), [], <<>>, []). + +request(get, {Url, Headers}) -> + do(get, to_bin(Url), normalise_req_headers(Headers), <<>>, []). + +request(Method, {Url, Headers}, HttpOpts, _Opts) -> + do(Method, to_bin(Url), normalise_req_headers(Headers), <<>>, hackney_opts(HttpOpts)); +request(Method, {Url, Headers, ContentType, Body}, HttpOpts, _Opts) -> + AllHeaders = [ + {<<"content-type">>, to_bin(ContentType)} + | normalise_req_headers(Headers) + ], + do(Method, to_bin(Url), AllHeaders, to_bin(Body), hackney_opts(HttpOpts)). + +%% --- internals --- + +do(Method, Url, Headers, Body, Opts) -> + case hackney:request(Method, Url, Headers, Body, [with_body | Opts]) of + {ok, Status, RespHeaders, RespBody} -> + HeaderList = [ + {string:to_lower(binary_to_list(K)), binary_to_list(V)} + || {K, V} <- RespHeaders + ], + {ok, {{"HTTP/1.1", Status, ""}, HeaderList, binary_to_list(RespBody)}}; + {error, _} = E -> + E + end. + +normalise_req_headers(L) -> + [{to_bin(K), to_bin(V)} || {K, V} <- L]. + +hackney_opts(HttpOpts) -> + %% Translate the few httpc options the suites pass. + lists:flatmap( + fun + ({timeout, T}) -> [{recv_timeout, T}, {connect_timeout, T}]; + ({connect_timeout, T}) -> [{connect_timeout, T}]; + (_) -> [] + end, + HttpOpts + ). + +to_bin(B) when is_binary(B) -> B; +to_bin(L) when is_list(L) -> list_to_binary(L); +to_bin(A) when is_atom(A) -> atom_to_binary(A, utf8). diff --git a/apps/barrel_inference_server/test/barrel_inference_server_ollama_translate_tests.erl b/apps/barrel_inference_server/test/barrel_inference_server_ollama_translate_tests.erl index 802111b..62c80b4 100644 --- a/apps/barrel_inference_server/test/barrel_inference_server_ollama_translate_tests.erl +++ b/apps/barrel_inference_server/test/barrel_inference_server_ollama_translate_tests.erl @@ -152,27 +152,21 @@ keep_alive_parser_test_() -> %% ============================================================================= generate_chunk_shape_test() -> - Bin = iolist_to_binary( - barrel_inference_server_translate:internal_to_ollama_generate_chunk( - <<"Hello">>, <<"req-1">>, <<"llama3">> - ) + M = barrel_inference_server_translate:internal_to_ollama_generate_chunk( + <<"Hello">>, <<"req-1">>, <<"llama3">> ), - M = json:decode(Bin), ?assertEqual(<<"llama3">>, maps:get(<<"model">>, M)), ?assertEqual(<<"Hello">>, maps:get(<<"response">>, M)), ?assertEqual(false, maps:get(<<"done">>, M)), ?assert(is_binary(maps:get(<<"created_at">>, M))). generate_final_shape_test() -> - Bin = iolist_to_binary( - barrel_inference_server_translate:internal_to_ollama_generate_final( - #{finish_reason => stop, prompt_tokens => 5, completion_tokens => 3}, - <<"req-1">>, - <<"llama3">>, - #{total_duration_ns => 1000000, load_duration_ns => 500000} - ) + M = barrel_inference_server_translate:internal_to_ollama_generate_final( + #{finish_reason => stop, prompt_tokens => 5, completion_tokens => 3}, + <<"req-1">>, + <<"llama3">>, + #{total_duration_ns => 1000000, load_duration_ns => 500000} ), - M = json:decode(Bin), ?assertEqual(true, maps:get(<<"done">>, M)), ?assertEqual(<<"stop">>, maps:get(<<"done_reason">>, M)), ?assertEqual(1000000, maps:get(<<"total_duration">>, M)), diff --git a/apps/barrel_inference_server/test/barrel_inference_server_smoke_SUITE.erl b/apps/barrel_inference_server/test/barrel_inference_server_smoke_SUITE.erl index 0f9bf68..fd7d08b 100644 --- a/apps/barrel_inference_server/test/barrel_inference_server_smoke_SUITE.erl +++ b/apps/barrel_inference_server/test/barrel_inference_server_smoke_SUITE.erl @@ -168,7 +168,7 @@ init_per_suite(Config) -> } ), {ok, Started} = application:ensure_all_started(barrel_inference_server), - {ok, _} = application:ensure_all_started(inets), + {ok, _} = barrel_inference_server_http_test:ensure_started(), Url = io_lib:format("http://127.0.0.1:~p", [chosen_port()]), [ {base, lists:flatten(Url)}, @@ -198,13 +198,13 @@ chosen_port() -> health_returns_200(Cfg) -> Url = ?config(base, Cfg) ++ "/health", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assertEqual(<<"ok">>, maps:get(<<"status">>, Decoded)). health_ready_returns_503_with_no_models(Cfg) -> Url = ?config(base, Cfg) ++ "/health/ready", - {ok, {{_, Code, _}, _, Body}} = httpc:request(Url), + {ok, {{_, Code, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assertEqual(503, Code), ?assertEqual(<<"not_ready">>, maps:get(<<"status">>, Decoded)), @@ -212,14 +212,14 @@ health_ready_returns_503_with_no_models(Cfg) -> models_returns_openai_list_shape(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/models", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assertEqual(<<"list">>, maps:get(<<"object">>, Decoded)), ?assertMatch([_ | _], maps:get(<<"data">>, Decoded)). models_returns_aliases(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/models", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), Ids = [maps:get(<<"id">>, M) || M <- maps:get(<<"data">>, Decoded)], ?assert(lists:member(<<"alias-a">>, Ids)), @@ -227,14 +227,14 @@ models_returns_aliases(Cfg) -> models_unknown_returns_404(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/models/no-such-model", - {ok, {{_, 404, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 404, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), Err = maps:get(<<"error">>, Decoded), ?assertEqual(<<"model_not_found">>, maps:get(<<"code">>, Err)). metrics_returns_prometheus_text(Cfg) -> Url = ?config(base, Cfg) ++ "/metrics", - {ok, {{_, 200, _}, Headers, _Body}} = httpc:request(Url), + {ok, {{_, 200, _}, Headers, _Body}} = barrel_inference_server_http_test:request(Url), %% Content-type starts with text/plain. Body may be empty if no %% counter has been observed yet (the suite order is not %% guaranteed across runs). @@ -245,7 +245,9 @@ embeddings_unknown_model_returns_404(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/embeddings", Body = json:encode(#{<<"model">> => <<"nope">>, <<"input">> => <<"hi">>}), {ok, {{_, Code, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), %% No model is loaded; resolve_model is identity, ensure_loaded %% sees not_found from barrel_inference:model_info/1 (not loaded path). %% In practice this returns 503 from ensure_loaded {error, not_loaded} @@ -256,7 +258,7 @@ embeddings_unknown_model_returns_404(Cfg) -> embeddings_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/embeddings", {ok, {{_, 400, _}, _, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [], "application/json", "{not json"}, [], @@ -266,7 +268,7 @@ embeddings_invalid_json_returns_400(Cfg) -> chat_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/chat/completions", {ok, {{_, 400, _}, _, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [], "application/json", "{nope"}, [], @@ -275,8 +277,11 @@ chat_invalid_json_returns_400(Cfg) -> %% Pre-stream errors on a streaming /v1/messages request must surface %% as an Anthropic SSE `event: error` frame, not a JSON envelope. -%% Anthropic SDKs read the streaming body as SSE; a JSON response -%% decodes as a transport error rather than a proper error event. +%% Pre-admit failures (e.g. unknown model) now return a real 4xx JSON +%% envelope before any SSE frame is opened — livery_resp:stream_deferred +%% lets the producer choose the response shape after admission. Tests +%% that previously expected `200 + event: error` get the same Anthropic +%% error envelope, just on a 4xx response. messages_streaming_unknown_model_emits_event_error(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/messages", Body = json:encode(#{ @@ -286,16 +291,11 @@ messages_streaming_unknown_model_emits_event_error(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, Status, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Bin = list_to_binary(RespBody), - %% Cowboy streams with HTTP 200; the error is conveyed via the - %% SSE error event rather than an HTTP status. (200 is what - %% Anthropic itself returns for the streaming-error path.) - ?assertEqual(200, Status), - ?assert(binary:match(Bin, <<"event: error">>) =/= nomatch), - ?assert(binary:match(Bin, <<"\"type\":\"error\"">>) =/= nomatch), - %% The inner error.type must be one of Anthropic's enum values, not - %% a freeform string like the pre-fix \"server_error\". + ?assertEqual(404, Status), ?assert(binary:match(Bin, <<"\"type\":\"not_found_error\"">>) =/= nomatch). %% /v1/messages/count_tokens with no loaded model surfaces as 529 @@ -311,7 +311,9 @@ count_tokens_unknown_model_returns_529(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"hi">>}] }), {ok, {{_, Status, _}, Headers, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), ?assertEqual(529, Status), {value, {_, Retry}} = lists:keysearch("retry-after", 1, Headers), ?assert(list_to_integer(Retry) >= 1), @@ -324,7 +326,7 @@ count_tokens_unknown_model_returns_529(Cfg) -> count_tokens_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/messages/count_tokens", {ok, {{_, 400, _}, _, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [], "application/json", "{not json"}, [], @@ -341,7 +343,7 @@ anthropic_version_header_echoed(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, _, _}, Headers, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [{"anthropic-version", "2024-12-01"}], "application/json", Body}, [], @@ -358,7 +360,7 @@ accepts_body_above_one_mb(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/messages", Big = binary:copy(<<"x">>, 5 * 1024 * 1024), {ok, {{_, Status, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", Big}, [], []), + barrel_inference_server_http_test:request(post, {Url, [], "application/json", Big}, [], []), ?assertEqual(400, Status). %% Cowboy's default per-read `length' is 8 MB; the handler must loop @@ -366,11 +368,13 @@ accepts_body_above_one_mb(Cfg) -> %% treating the first non-final chunk as 413. A 10 MB body is enough %% to force at least two reads even on a fast localhost socket. accepts_body_above_cowboy_default_length(_Cfg) -> - %% TODO: livery's h1 adapter RSTs the connection on early - %% response (it does not drain the remaining upload), so httpc - %% sees socket_closed instead of the 400/413 the server actually - %% sent. Skip until livery grows graceful body-drain. - {skip, livery_h1_no_body_drain}. + %% TODO: livery 0.3.2 / h1 0.6.2 race on multi-MiB POST. + %% Server returns 413/400 but the response wire-send lands on a + %% h1 connection process that's already gone (noproc), and + %% hackney sees socket_closed before reading the response. + %% Needs upstream livery to guard the emit on the h1 process + %% still being alive (or hackney recv-during-send). + {skip, livery_h1_response_after_lingering_close}. %% Anthropic SDKs read `request-id` (no x- prefix) into %% message._request_id; the existing middleware stamps `x-request-id`. @@ -384,7 +388,9 @@ messages_emits_request_id_header(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, _, _}, Headers, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), {value, {_, AnthrId}} = lists:keysearch("request-id", 1, Headers), {value, {_, XReqId}} = lists:keysearch("x-request-id", 1, Headers), ?assertEqual(XReqId, AnthrId), @@ -401,7 +407,7 @@ messages_no_allowlist_accepts_any_key(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, Status, _}, _, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [{"x-api-key", "not-used"}], "application/json", Body}, [], @@ -432,7 +438,9 @@ messages_logs_user_id_in_event(Cfg) -> #{<<"role">> => <<"user">>, <<"content">> => <<"x">>} ] }), - {ok, _} = httpc:request(post, {Url, [], "application/json", Body}, [], []), + {ok, _} = barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), receive {log_event, Report} -> ?assertEqual(<<"u-test">>, maps:get(user_id, Report)), @@ -478,7 +486,9 @@ messages_emits_ratelimit_headers(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, _, _}, Headers, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), {value, {_, Limit}} = lists:keysearch("anthropic-ratelimit-requests-limit", 1, Headers), {value, {_, Remaining}} = @@ -501,7 +511,9 @@ messages_error_body_carries_request_id(Cfg) -> <<"messages">> => [#{<<"role">> => <<"user">>, <<"content">> => <<"x">>}] }), {ok, {{_, 404, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Decoded = json:decode(list_to_binary(RespBody)), ?assertMatch(#{<<"request_id">> := <<"req_", _/binary>>}, Decoded). @@ -509,18 +521,16 @@ messages_error_body_carries_request_id(Cfg) -> %% not the catch-all `api_error`. SDKs match on the type to decide %% retry behaviour. messages_413_returns_request_too_large_type(_Cfg) -> - %% TODO: same as accepts_body_above_cowboy_default_length: - %% livery's h1 RSTs on early-response, so httpc sees - %% socket_closed before reading the 413 body. The server-side - %% 413 fires correctly (verified in the listener log on a CI - %% run). Skip until livery drains the body before close. - {skip, livery_h1_no_body_drain}. + %% TODO: same livery/h1 race as accepts_body_above_cowboy_default_length. + {skip, livery_h1_response_after_lingering_close}. chat_missing_model_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/chat/completions", Body = json:encode(#{<<"messages">> => []}), {ok, {{_, 400, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Decoded = json:decode(list_to_binary(RespBody)), ?assertMatch(#{<<"error">> := _}, Decoded). @@ -533,7 +543,9 @@ chat_too_many_messages_returns_400(Cfg) -> ], Body = json:encode(#{<<"model">> => <<"x">>, <<"messages">> => Many}), {ok, {{_, 400, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Decoded = json:decode(list_to_binary(RespBody)), ?assertMatch( #{<<"error">> := #{<<"code">> := <<"too_many_messages">>}}, @@ -542,14 +554,16 @@ chat_too_many_messages_returns_400(Cfg) -> request_id_minted_when_absent(Cfg) -> Url = ?config(base, Cfg) ++ "/health", - {ok, {{_, 200, _}, Headers, _}} = httpc:request(Url), + {ok, {{_, 200, _}, Headers, _}} = barrel_inference_server_http_test:request(Url), {value, {_, Id}} = lists:keysearch("x-request-id", 1, Headers), ?assert(string:prefix(Id, "req_") =/= nomatch). request_id_echoed_when_present(Cfg) -> Url = ?config(base, Cfg) ++ "/health", {ok, {{_, 200, _}, Headers, _}} = - httpc:request(get, {Url, [{"x-request-id", "abc-123"}]}, [], []), + barrel_inference_server_http_test:request( + get, {Url, [{"x-request-id", "abc-123"}]}, [], [] + ), {value, {_, Id}} = lists:keysearch("x-request-id", 1, Headers), ?assertEqual("abc-123", Id). @@ -557,7 +571,7 @@ cors_disabled_by_default(Cfg) -> %% This suite enables CORS, so this test verifies that no Origin %% header skips the CORS branch entirely. Url = ?config(base, Cfg) ++ "/health", - {ok, {{_, 200, _}, Headers, _}} = httpc:request(Url), + {ok, {{_, 200, _}, Headers, _}} = barrel_inference_server_http_test:request(Url), %% No Origin header on the request -> no Access-Control-* on the %% response. ?assertEqual(false, lists:keymember("access-control-allow-origin", 1, Headers)). @@ -570,7 +584,7 @@ cors_preflight_returns_204(Cfg) -> {"access-control-request-headers", "content-type"} ], {ok, {{_, 204, _}, RespHeaders, _}} = - httpc:request(options, {Url, Headers}, [], []), + barrel_inference_server_http_test:request(options, {Url, Headers}, [], []), {value, {_, AllowOrigin}} = lists:keysearch("access-control-allow-origin", 1, RespHeaders), ?assertEqual("*", AllowOrigin), @@ -580,7 +594,9 @@ cors_preflight_returns_204(Cfg) -> cors_headers_present_on_response(Cfg) -> Url = ?config(base, Cfg) ++ "/health", {ok, {{_, 200, _}, Headers, _}} = - httpc:request(get, {Url, [{"origin", "http://example.com"}]}, [], []), + barrel_inference_server_http_test:request( + get, {Url, [{"origin", "http://example.com"}]}, [], [] + ), ?assert(lists:keymember("access-control-allow-origin", 1, Headers)). %%==================================================================== @@ -590,7 +606,7 @@ cors_headers_present_on_response(Cfg) -> responses_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/responses", {ok, {{_, 400, _}, _, _}} = - httpc:request( + barrel_inference_server_http_test:request( post, {Url, [], "application/json", "{not json"}, [], @@ -601,7 +617,9 @@ responses_missing_model_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/responses", Body = json:encode(#{}), {ok, {{_, 400, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Decoded = json:decode(list_to_binary(RespBody)), ?assertMatch(#{<<"error">> := _}, Decoded). @@ -613,7 +631,9 @@ responses_string_input_unknown_model_returns_404(Cfg) -> <<"max_output_tokens">> => 4 }), {ok, {{_, Status, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), ?assertEqual(404, Status). %% Non-streaming responses include a `resp_` id. We can't actually @@ -629,7 +649,9 @@ responses_emits_response_id_header(Cfg) -> <<"max_output_tokens">> => 4 }), {ok, {{_, Status, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), %% Endpoint is reachable; 404 (model resolution) or another non-405 %% is acceptable. The point is that we don't route-miss. ?assert(Status =/= 405). @@ -643,7 +665,9 @@ responses_streaming_unknown_model_emits_event_error(Cfg) -> <<"max_output_tokens">> => 8 }), {ok, {{_, Status, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), %% On the streaming path the handler opens an SSE stream on the %% pipeline-error and emits a `response.failed` event. HTTP 200. Bin = list_to_binary(RespBody), @@ -659,8 +683,8 @@ responses_streaming_unknown_model_emits_event_error(Cfg) -> end. responses_413_returns_request_too_large_type(_Cfg) -> - %% TODO: same livery h1 early-response RST as the other 413 tests. - {skip, livery_h1_no_body_drain}. + %% TODO: same livery/h1 race as accepts_body_above_cowboy_default_length. + {skip, livery_h1_response_after_lingering_close}. %% Codex sends the Responses request as an `input` array of message %% items (content as `input_text` parts) plus a top-level @@ -685,7 +709,9 @@ responses_codex_envelope_accepted(Cfg) -> <<"max_output_tokens">> => 8 }), {ok, {{_, Status, _}, _, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Bin = list_to_binary(RespBody), %% Not a 400: the array-input + instructions envelope parsed. An %% unknown model then fails at load, which on the streaming path @@ -714,7 +740,9 @@ request_id_custom_header_name_echoed(Cfg) -> try Url = ?config(base, Cfg) ++ "/health", {ok, {{_, 200, _}, Headers, _}} = - httpc:request(get, {Url, [{"x-trace-id", "trace-abc"}]}, [], []), + barrel_inference_server_http_test:request( + get, {Url, [{"x-trace-id", "trace-abc"}]}, [], [] + ), {value, {_, Id}} = lists:keysearch("x-trace-id", 1, Headers), ?assertEqual("trace-abc", Id), ?assertEqual(false, lists:keymember("x-request-id", 1, Headers)) @@ -761,12 +789,16 @@ method_not_allowed_sweep(Cfg) -> check_method_405(Base, Path, get) -> Url = Base ++ Path, - {ok, {{_, Status, _}, _, _}} = httpc:request(get, {Url, []}, [], []), + {ok, {{_, Status, _}, _, _}} = barrel_inference_server_http_test:request( + get, {Url, []}, [], [] + ), ?assertEqual({Path, 405}, {Path, Status}); check_method_405(Base, Path, post) -> Url = Base ++ Path, {ok, {{_, Status, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", <<"{}">>}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", <<"{}">>}, [], [] + ), ?assertEqual({Path, 405}, {Path, Status}). %% OpenAI legacy completions (POST /v1/completions) is routed via the @@ -775,13 +807,17 @@ check_method_405(Base, Path, post) -> completions_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/completions", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). completions_missing_model_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/v1/completions", Body = json:encode(#{<<"prompt">> => <<"hello">>}), {ok, {{_, Status, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), ?assertEqual(400, Status). %% Ollama-shaped non-streaming GET endpoints. /api/version reports the @@ -790,20 +826,20 @@ completions_missing_model_returns_400(Cfg) -> %% should all return 200 with a JSON body in the smoke baseline. api_version_returns_200(Cfg) -> Url = ?config(base, Cfg) ++ "/api/version", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(is_map(Decoded)), ?assert(maps:is_key(<<"version">>, Decoded)). api_tags_returns_200(Cfg) -> Url = ?config(base, Cfg) ++ "/api/tags", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(maps:is_key(<<"models">>, Decoded)). api_ps_returns_200(Cfg) -> Url = ?config(base, Cfg) ++ "/api/ps", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(maps:is_key(<<"models">>, Decoded)). @@ -813,7 +849,9 @@ api_ps_returns_200(Cfg) -> api_search_invalid_json_returns_400(Cfg) -> Url = ?config(base, Cfg) ++ "/api/search", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). %% POST /api/pull is NDJSON-streaming. Even on a fast-failing spec %% (unknown model) the body must consist of `\n'-terminated parseable @@ -823,7 +861,9 @@ api_pull_emits_ndjson_lines(Cfg) -> Url = ?config(base, Cfg) ++ "/api/pull", Body = json:encode(#{<<"model">> => <<"no-such-model:notag">>, <<"stream">> => true}), {ok, {{_, _Status, _}, Headers, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Bin = list_to_binary(RespBody), %% The Content-Type is application/x-ndjson on success and %% application/json on a synchronous error envelope; either is fine @@ -858,13 +898,13 @@ api_pull_emits_ndjson_lines(Cfg) -> livery_listener_serves_health(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/health", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assertEqual(<<"ok">>, maps:get(<<"status">>, Decoded)). livery_listener_serves_metrics(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/metrics", - {ok, {{_, 200, _}, Headers, _Body}} = httpc:request(Url), + {ok, {{_, 200, _}, Headers, _Body}} = barrel_inference_server_http_test:request(Url), {value, {_, CT}} = lists:keysearch("content-type", 1, Headers), %% Status + content-type prove the route is wired. The actual body %% bytes are exercised by the cowboy metrics_returns_prometheus_text @@ -875,7 +915,7 @@ livery_listener_serves_metrics(Cfg) -> livery_listener_serves_models_list(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/v1/models", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assertEqual(<<"list">>, maps:get(<<"object">>, Decoded)), ?assert(is_list(maps:get(<<"data">>, Decoded))). @@ -891,7 +931,9 @@ livery_listener_echoes_request_id_header(Cfg) -> try Url = ?config(livery_base, Cfg) ++ "/health", {ok, {{_, 200, _}, Headers, _}} = - httpc:request(get, {Url, [{"x-trace-id", "abc-trace"}]}, [], []), + barrel_inference_server_http_test:request( + get, {Url, [{"x-trace-id", "abc-trace"}]}, [], [] + ), {value, {_, Id}} = lists:keysearch("x-trace-id", 1, Headers), ?assertEqual("abc-trace", Id) after @@ -907,12 +949,16 @@ livery_listener_echoes_request_id_header(Cfg) -> livery_ollama_generate_invalid_json_returns_400(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/generate", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). livery_ollama_chat_invalid_json_returns_400(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/chat", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). %% A streaming Ollama generate against an unknown model is the %% live-fire path: the handler enters the inference branch, the @@ -922,28 +968,32 @@ livery_ollama_chat_invalid_json_returns_400(Cfg) -> livery_embeddings_openai_invalid_json_returns_400(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/v1/embeddings", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). livery_embeddings_ollama_invalid_json_returns_400(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/embed", {ok, {{_, 400, _}, _, _}} = - httpc:request(post, {Url, [], "application/json", "{not json"}, [], []). + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", "{not json"}, [], [] + ). livery_api_tags_returns_200(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/tags", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(maps:is_key(<<"models">>, Decoded)). livery_api_version_returns_200(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/version", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(maps:is_key(<<"version">>, Decoded)). livery_api_ps_returns_200(Cfg) -> Url = ?config(livery_base, Cfg) ++ "/api/ps", - {ok, {{_, 200, _}, _, Body}} = httpc:request(Url), + {ok, {{_, 200, _}, _, Body}} = barrel_inference_server_http_test:request(Url), Decoded = json:decode(list_to_binary(Body)), ?assert(maps:is_key(<<"models">>, Decoded)). @@ -955,7 +1005,9 @@ livery_ollama_generate_unknown_model_streams_ndjson(Cfg) -> <<"stream">> => true }), {ok, {{_, _Status, _}, Headers, RespBody}} = - httpc:request(post, {Url, [], "application/json", Body}, [], []), + barrel_inference_server_http_test:request( + post, {Url, [], "application/json", Body}, [], [] + ), Bin = list_to_binary(RespBody), {value, {_, CT}} = lists:keysearch("content-type", 1, Headers), ?assert(string:str(CT, "x-ndjson") =/= 0 orelse string:str(CT, "json") =/= 0), diff --git a/apps/barrel_inference_server/test/barrel_inference_server_translate_SUITE.erl b/apps/barrel_inference_server/test/barrel_inference_server_translate_SUITE.erl index 4a29fe5..47bd219 100644 --- a/apps/barrel_inference_server/test/barrel_inference_server_translate_SUITE.erl +++ b/apps/barrel_inference_server/test/barrel_inference_server_translate_SUITE.erl @@ -1186,10 +1186,7 @@ anthropic_event_message_delta_emits_stop_sequence(_Cfg) -> finish_reason => stop, stop_sequence => <<"END">> }, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"\"stop_reason\":\"stop_sequence\"">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"stop_sequence\":\"END\"">>) =/= nomatch). @@ -1228,19 +1225,13 @@ anthropic_response_thinking_then_text_blocks(_Cfg) -> ?assertEqual(<<"answer">>, maps:get(<<"text">>, B2)). anthropic_event_message_start(_Cfg) -> - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_start, 42}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_start, 42}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"event: message_start">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"id\":\"msg_1\"">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"input_tokens\":42">>) =/= nomatch). anthropic_event_text_delta(_Cfg) -> - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {text_delta, <<"hello">>, 0}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({text_delta, <<"hello">>, 0}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"event: content_block_delta">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"text\":\"hello\"">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"index\":0">>) =/= nomatch). @@ -1249,26 +1240,10 @@ anthropic_event_text_delta(_Cfg) -> %% so consecutive blocks must carry distinct indices. The emitter must %% honour whatever index the handler supplies. anthropic_event_content_block_index_threads_through(_Cfg) -> - Start = iolist_to_binary( - barrel_inference_server_translate:internal_to_anthropic_event( - {content_block_start_text, 3}, #{}, <<"msg_1">>, <<"claude">> - ) - ), - Delta = iolist_to_binary( - barrel_inference_server_translate:internal_to_anthropic_event( - {text_delta, <<"t">>, 3}, #{}, <<"msg_1">>, <<"claude">> - ) - ), - Thinking = iolist_to_binary( - barrel_inference_server_translate:internal_to_anthropic_event( - {thinking_delta, <<"r">>, 2}, #{}, <<"msg_1">>, <<"claude">> - ) - ), - Stop = iolist_to_binary( - barrel_inference_server_translate:internal_to_anthropic_event( - {content_block_stop, 3}, #{}, <<"msg_1">>, <<"claude">> - ) - ), + Start = event_bin({content_block_start_text, 3}, #{}, <<"msg_1">>, <<"claude">>), + Delta = event_bin({text_delta, <<"t">>, 3}, #{}, <<"msg_1">>, <<"claude">>), + Thinking = event_bin({thinking_delta, <<"r">>, 2}, #{}, <<"msg_1">>, <<"claude">>), + Stop = event_bin({content_block_stop, 3}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Start, <<"\"index\":3">>) =/= nomatch), ?assert(binary:match(Delta, <<"\"index\":3">>) =/= nomatch), ?assert(binary:match(Thinking, <<"\"index\":2">>) =/= nomatch), @@ -1276,10 +1251,7 @@ anthropic_event_content_block_index_threads_through(_Cfg) -> anthropic_event_message_delta(_Cfg) -> Stats = #{completion_tokens => 4, finish_reason => length}, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"event: message_delta">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"stop_reason\":\"max_tokens\"">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"output_tokens\":4">>) =/= nomatch). @@ -1294,10 +1266,7 @@ anthropic_event_message_delta_emits_cache_read_on_exact_hit(_Cfg) -> cache_delta => #{read => 128, created => 0}, finish_reason => stop }, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"\"cache_read_input_tokens\":128">>) =/= nomatch), ?assertEqual(nomatch, binary:match(Bin, <<"\"cache_creation_input_tokens\"">>)). @@ -1308,10 +1277,7 @@ anthropic_event_message_delta_emits_cache_creation_on_cold(_Cfg) -> cache_delta => #{read => 0, created => 128}, finish_reason => stop }, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"\"cache_creation_input_tokens\":128">>) =/= nomatch), ?assertEqual(nomatch, binary:match(Bin, <<"\"cache_read_input_tokens\"">>)). @@ -1327,10 +1293,7 @@ anthropic_event_message_delta_emits_cache_creation_nested_5m(_Cfg) -> finish_reason => stop, cache_hints => [#{kind => system, hash => <<"h">>, ttl => <<"5m">>}] }, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"\"ephemeral_5m_input_tokens\":64">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"ephemeral_1h_input_tokens\":0">>) =/= nomatch). @@ -1342,10 +1305,7 @@ anthropic_event_message_delta_emits_cache_creation_nested_1h(_Cfg) -> finish_reason => stop, cache_hints => [#{kind => system, hash => <<"h">>, ttl => <<"1h">>}] }, - Iolist = barrel_inference_server_translate:internal_to_anthropic_event( - {message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">> - ), - Bin = iolist_to_binary(Iolist), + Bin = event_bin({message_delta, Stats}, #{}, <<"msg_1">>, <<"claude">>), ?assert(binary:match(Bin, <<"\"ephemeral_5m_input_tokens\":0">>) =/= nomatch), ?assert(binary:match(Bin, <<"\"ephemeral_1h_input_tokens\":64">>) =/= nomatch). @@ -1729,3 +1689,12 @@ base_chat() -> text_block(Text) -> #{<<"type">> => <<"text">>, <<"text">> => Text}. + +%% Anthropic event helper: render the SSE frame as a binary so the +%% existing `binary:match' assertions still work after +%% `internal_to_anthropic_event/4' moved to returning a map. +event_bin(Kind, Acc, ReqId, Model) -> + Frame = barrel_inference_server_translate:internal_to_anthropic_event( + Kind, Acc, ReqId, Model + ), + iolist_to_binary(barrel_inference_server_translate:sse_iodata(Frame)). diff --git a/rebar.config b/rebar.config index 0f5d3ae..be4c8d4 100644 --- a/rebar.config +++ b/rebar.config @@ -19,14 +19,11 @@ {deps, []}. {project_plugins, [ - {rebar3_proper, "~> 0.12"}, - {rebar3_hex, "~> 7.0"}, - {rebar3_ex_doc, "~> 0.2"}, - %% Runtime used ~> 5.0, server pinned 4.1.1. Unified to ~> 5.0; - %% the server's elvis ruleset needs a shakeout against the newer - %% elvis_core (rule names may have changed). - {rebar3_lint, "~> 5.0"}, - {erlfmt, "~> 1.7"} + {rebar3_proper, "~> 0.12.1"}, + {rebar3_hex, "~> 7.1.0"}, + {rebar3_ex_doc, "~> 0.3.0"}, + {rebar3_lint, "~> 5.0.4"}, + {erlfmt, "~> 1.8.0"} ]}. {shell, [ @@ -81,7 +78,7 @@ ]}, {test, [ {erl_opts, [debug_info, nowarn_export_all]}, - {deps, [{proper, "1.4.0"}]} + {deps, [{proper, "~> 1.4.0"}]} ]} ]}. diff --git a/rebar.lock b/rebar.lock index f350d13..1c82356 100644 --- a/rebar.lock +++ b/rebar.lock @@ -1,47 +1,47 @@ {"1.2.0", -[{<<"barrel_mcp">>,{pkg,<<"barrel_mcp">>,<<"2.2.2">>},0}, - {<<"certifi">>,{pkg,<<"certifi">>,<<"2.16.0">>},2}, - {<<"h1">>,{pkg,<<"erlang_h1">>,<<"0.6.1">>},1}, - {<<"h2">>,{pkg,<<"h2">>,<<"0.6.0">>},1}, - {<<"hackney">>,{pkg,<<"hackney">>,<<"4.0.0">>},1}, +[{<<"barrel_mcp">>,{pkg,<<"barrel_mcp">>,<<"2.2.3">>},0}, + {<<"certifi">>,{pkg,<<"certifi">>,<<"2.17.0">>},2}, + {<<"h1">>,{pkg,<<"erlang_h1">>,<<"0.6.2">>},1}, + {<<"h2">>,{pkg,<<"h2">>,<<"0.10.1">>},1}, + {<<"hackney">>,{pkg,<<"hackney">>,<<"4.4.0">>},1}, {<<"idna">>,{pkg,<<"idna">>,<<"7.1.0">>},2}, - {<<"instrument">>,{pkg,<<"instrument">>,<<"1.0.0">>},0}, - {<<"livery">>,{pkg,<<"livery">>,<<"0.2.7">>},0}, + {<<"instrument">>,{pkg,<<"instrument">>,<<"1.1.4">>},0}, + {<<"livery">>,{pkg,<<"livery">>,<<"0.3.2">>},0}, {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.5.0">>},1}, - {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.4.1">>},2}, - {<<"quic">>,{pkg,<<"quic">>,<<"1.0.0">>},1}, + {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.4.2">>},2}, + {<<"quic">>,{pkg,<<"quic">>,<<"1.6.5">>},1}, {<<"ssl_verify_fun">>,{pkg,<<"ssl_verify_fun">>,<<"1.1.7">>},2}, - {<<"webtransport">>,{pkg,<<"webtransport">>,<<"0.3.2">>},1}, + {<<"webtransport">>,{pkg,<<"webtransport">>,<<"0.4.1">>},1}, {<<"ws">>,{pkg,<<"erlang_ws">>,<<"0.3.0">>},1}]}. [ {pkg_hash,[ - {<<"barrel_mcp">>, <<"EBB5A3A15A55ECB36D0857B05B2080C376B5806242822208FFFEDFC561F535DC">>}, - {<<"certifi">>, <<"A4EDFC1D2DA3424D478A3271133BF28E0EC5E6FD8C009AAB5A4AE980CB165CE9">>}, - {<<"h1">>, <<"4F16011A82B759395D9DF151D69C262B3F8A03EDAD6692A9CE1812565AB4B8BE">>}, - {<<"h2">>, <<"32B519F7A5A16FFB699C5EF9784AD279F3D1AA45B563E3328B996C8DB64C27D4">>}, - {<<"hackney">>, <<"DFC90299EDC31553898CD32567F1D46FDB477AFED9ECA694B9C879DDADFA3D62">>}, + {<<"barrel_mcp">>, <<"52197EF7AE48DAD1FE8BD8EFC5DC994A7FBA9FCF6A46622D4B72490D0A09F903">>}, + {<<"certifi">>, <<"835748414307E15E05B17D0E518190228CE648B08D569A5CC93A85A40F3E5C9B">>}, + {<<"h1">>, <<"35401E355F0B3068CF41E1A5633CB7105955716A15F2F0D1D5DB66DAFB45D58C">>}, + {<<"h2">>, <<"925E7A0967E15E2D12544CF719CF657B32D4DBB9E5602DC6F06066B6309E0020">>}, + {<<"hackney">>, <<"AC34AE5C06A7927B671EECC0AB3806E3E4D481A613B657F6E2CA19F0CE3DD56B">>}, {<<"idna">>, <<"1067A13043538129602D2F2CE6899D8713125C7D19734AA557CE2E3EA55BD4F1">>}, - {<<"instrument">>, <<"D9027BEC640DECD7CC27C34F57422DDE6106FDACCF51E2BBE0EB99ABA2165976">>}, - {<<"livery">>, <<"6C9C7B5F3FBF04F66A39B9C7BCDCBF813E7B1A20B6B548EE2C19E76C57EB531B">>}, + {<<"instrument">>, <<"1B9B5C232BDF9840C1E2D617A1912660CFEECD84A012BDCEC5907C712DFFA717">>}, + {<<"livery">>, <<"DEBD5BCF3777A6A0E7853E7FD3A889CB82AE5F7C32F9944CBABEC83C4A3CBCC4">>}, {<<"mimerl">>, <<"F35ACA6F23242339B3666E0AC0702379E362B469D0AEA167F6CC713547E777ED">>}, - {<<"parse_trans">>, <<"6E6AA8167CB44CC8F39441D05193BE6E6F4E7C2946CB2759F015F8C56B76E5FF">>}, - {<<"quic">>, <<"2FE86FEEDAB19A2CD4E1355553FB7F30CD7968C66D79E45E331A8CA60FDF4661">>}, + {<<"parse_trans">>, <<"C352DDC1A0D5E54F9B1654D45F9C432EEF76F9CEA371C55DDFF769EF688FDB74">>}, + {<<"quic">>, <<"28B7D49C1732B2DE3861701BB03AE7798537240552370AC49F0D139D2EA8F621">>}, {<<"ssl_verify_fun">>, <<"354C321CF377240C7B8716899E182CE4890C5938111A1296ADD3EC74CF1715DF">>}, - {<<"webtransport">>, <<"1A2976420A7AFA6675A7E19FE034956174966A6BA8D9F68EEF785056AE0E1E58">>}, + {<<"webtransport">>, <<"60EF6CD99282B5964C8172D674519239E31E357D934BDE028BEC70DA34636FE5">>}, {<<"ws">>, <<"6778E82A478FF6D205C14E5F17883447CF8BF4919E2C4F811FE3BE263635E37D">>}]}, {pkg_hash_ext,[ - {<<"barrel_mcp">>, <<"03F1E6DECE689BEF7E31F6DAB23D6362F2CE7E5DDCB776FD04C3C0D2C29588F9">>}, - {<<"certifi">>, <<"8A64F6669D85E9CC0E5086FCF29A5B13DE57A13EFA23D3582874B9A19303F184">>}, - {<<"h1">>, <<"759E345D19E49DE97606348F2B6600F1EB52F6847DC34D8060504A46B4019F6A">>}, - {<<"h2">>, <<"5AC1C05675EFECACBB8014D3923B3A19B1F6C4672F2C103AC24D563169338888">>}, - {<<"hackney">>, <<"56EEA3F252B8A672756986B1BEE827D75C933F9C335D3E0C54F57F48345F272C">>}, + {<<"barrel_mcp">>, <<"96717F4669257F3E39121270537A3AA00A3B89E22C9054D4269B315D9D56FC6F">>}, + {<<"certifi">>, <<"8122798A17F0293C80DAADA25D0F81C7F4D708C73FEF782C7C9B1950E26E4D21">>}, + {<<"h1">>, <<"F2055964E01E5A1F8E31601DA6EF91C052C56B14C364CC7C3763A1237221D365">>}, + {<<"h2">>, <<"686794F441A5AE0F03D1AB40E1713BD1C5130D1D43FB803E79C856C0A63FA8D1">>}, + {<<"hackney">>, <<"F3C59429485C6E773F2B8D1BF687EEC90645471C354F341E27374BDCE61A9115">>}, {<<"idna">>, <<"6AE959A025BF36DF61A8CAB8508D9654891B5426A84C44D82DEAFFD6DDF8C71F">>}, - {<<"instrument">>, <<"23E705B35D92DDC13E05BC5306B9D9A7795F40501EFD078248F0CADDC725AEDB">>}, - {<<"livery">>, <<"710A212C9409E3ACACC58178E35514FC5FFE8761E707B03BC78EC5BBCDB651B7">>}, + {<<"instrument">>, <<"BBC0153D92EBCB1BF655A8B78C3CBA0C2FE3AB0BE060630CCEA11B6B1397D554">>}, + {<<"livery">>, <<"54236AC1DDFB06ABA4F6FE519C09BFD4A01081074C854C6A3134BAD10BCE7367">>}, {<<"mimerl">>, <<"DB648CE065BAE14EA84CA8B5DD123F42F49417CEF693541110BF6F9E9BE9ECC4">>}, - {<<"parse_trans">>, <<"620A406CE75DADA827B82E453C19CF06776BE266F5A67CFF34E1EF2CBB60E49A">>}, - {<<"quic">>, <<"52F59E0BBA92ECEE80A7E685F92B8F498E7ED01F6DD05A682808AFB70279AFCC">>}, + {<<"parse_trans">>, <<"4C25347DE3B7C35732D32E69AB43D1CEEE0BEAE3F3B3ADE1B59CBD3DD224D9CA">>}, + {<<"quic">>, <<"DE1A88972C33201A50D1A17C8C4A14528BD1D8F25EF705897D680AE312D0AA78">>}, {<<"ssl_verify_fun">>, <<"FE4C190E8F37401D30167C8C405EDA19469F34577987C76DDE613E838BBC67F8">>}, - {<<"webtransport">>, <<"9BC90C2DAB6C09FB6532A27BCAF10A2ACE5749A7DB1C583329462974DE7DD9B1">>}, + {<<"webtransport">>, <<"006E4E52A8F03B69201D4637C85B424A4DDCCC1909F32C5742FED8D495A75174">>}, {<<"ws">>, <<"4F6FA9DFBDB49D27B6407F7ED5EA34F09499892079A9E5BED7389E352FFC80C5">>}]} ].