From a16617fffe432c254219bcfe5688042baacb988a Mon Sep 17 00:00:00 2001 From: Scott Karp Date: Thu, 15 Oct 2026 15:23:44 +0000 Subject: [PATCH] =?UTF-8?q?gate:=20streaming=20=E2=80=94=20the=20model's?= =?UTF-8?q?=20events=20relayed=20as=20they=20come,=20the=20fringe=20as=20t?= =?UTF-8?q?he=20last=20block?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Bamidbar 15:39 u'reitem oto — the fringe must be seen, so a streaming client gets it too. With "stream": true the request is opened asynchronously (Gate.Upstream.open_stream/4, httpc stream: :self) and Gate.Stream relays every complete SSE event to the caller unchanged while gathering the text deltas; the terminal events (Anthropic message_delta + message_stop; OpenAI the finish_reason chunk + [DONE]) are held until the gathered text has been examined, then released behind the fringe sent as one more text block, with the `bikoret` object added to the terminal event the client already parses. No X-Bikoret-* headers on a stream (they leave before the model speaks); no rebuke on a stream (nothing streamed can be taken back — 422). An examination that fails after the stream began is said in the trailer ("— bikoret: not examined (…)"), never silent. One api_calls_external gathered before the call, as for any pass. Tamid §1.1, Construction §3.10, KOHEN. Doorpost 40e7d7d1…. Elixir 612/0 (read before this commit). Co-Authored-By: Claude Fable 5.1 --- KOHEN.md | 3 + specs/bikoret_tamid.md | 16 +- specs/bikoret_vayakhel.md | 12 +- .../lib/vav_pipeline/bigdei_kodesh/router.ex | 46 ++- vav_pipeline/lib/vav_pipeline/doorpost.ex | 2 +- vav_pipeline/lib/vav_pipeline/gate.ex | 4 +- vav_pipeline/lib/vav_pipeline/gate/stream.ex | 290 ++++++++++++++++++ .../lib/vav_pipeline/gate/upstream.ex | 24 ++ vav_pipeline/test/fixtures/manna/ledger.json | 1 + vav_pipeline/test/gate_test.exs | 137 ++++++++- 10 files changed, 517 insertions(+), 18 deletions(-) create mode 100644 vav_pipeline/lib/vav_pipeline/gate/stream.ex create mode 100644 vav_pipeline/test/fixtures/manna/ledger.json diff --git a/KOHEN.md b/KOHEN.md index 0224f77e..e1841be9 100644 --- a/KOHEN.md +++ b/KOHEN.md @@ -590,6 +590,9 @@ curl -s -X POST http:///gate/anthropic/v1/messages \ `X-Bikoret-Rebuke: once` — a lo_tov reply is answered once more with the finding, and the second reply is delivered marked with both refs. +Streaming clients (`"stream": true`, Claude Code among them) are carried +through as they stream; the fringe arrives as the last text block and the +`bikoret` object on the final event — no headers, no rebuke on a stream. Each upstream call spends one `api_calls_external` of the day's manna (Day 4): a provision with 0 shuts the gate (503 MANNA_DEPLETED) while the bedikah by hand still runs. Streaming is refused (422). A non-200 diff --git a/specs/bikoret_tamid.md b/specs/bikoret_tamid.md index f4458262..2dd61354 100644 --- a/specs/bikoret_tamid.md +++ b/specs/bikoret_tamid.md @@ -458,7 +458,21 @@ Response (200): the upstream's reply, with rebuked?: {first_bedikah_ref, first_verdict}, disclosure } Any non-200 from the upstream is returned as it came (X-Bikoret-Verdict: none). - A stream request is refused (422): a fringe needs the whole garment. + +Streaming ("stream": true): the model's events are relayed as they + arrive, bytes unchanged, while the text is gathered; when the model's + stream ends the text is examined and the fringe is sent as ONE MORE + BLOCK before the terminal events — Anthropic: content_block_start / + content_block_delta / content_block_stop at the next index; OpenAI: one + more chunk with delta.content — and the `bikoret` object rides on the + terminal event the client already parses (message_delta; the chunk + with finish_reason), where an unknown field is ignored. Headers go out + before the model speaks, so a streamed reply has NO X-Bikoret-* headers + and cannot be rebuked ("stream" with X-Bikoret-Rebuke → 422). If the + examination fails after the stream has begun, the trailer says + "— bikoret: not examined ()" and the object's verdict is "none": + delivered, marked, never silent. Gathered text is capped at 65,536 + bytes like any output. Rebuke once: when X-Bikoret-Rebuke is "once" and the first reply is lo_tov, one second request = the first + the first reply as an diff --git a/specs/bikoret_vayakhel.md b/specs/bikoret_vayakhel.md index d43ca0e1..b7e0357f 100644 --- a/specs/bikoret_vayakhel.md +++ b/specs/bikoret_vayakhel.md @@ -547,9 +547,15 @@ Headers: **What the gate does with it** (`VavPipeline.Gate`): -1. Forwards the body unchanged (the `stream` flag is refused: 422 — a - fringe needs the whole garment). The upstream's status and body are - returned verbatim on any non-200. +1. Forwards the body unchanged. The upstream's status and body are + returned verbatim on any non-200. With `stream: true` the request is + opened asynchronously (`Gate.Upstream.open_stream/4`, OTP httpc + `stream: :self`) and `Gate.Stream` relays each complete SSE event to + the caller as it arrives, gathering the text deltas; the terminal + events (Anthropic `message_delta` + `message_stop`; OpenAI the + `finish_reason` chunk + `[DONE]`) are held until the examination has + run, then released behind the fringe block, the `bikoret` object + added to the terminal event. No rebuke on a stream. 2. Reads the reply's text (`content[].text` / `choices[0].message.content`) and examines it: provenance from the upstream (`anthropic → claude`, `openai → gpt`); **query** = the last user message's text; **sources** diff --git a/vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex b/vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex index 5fc3042c..30ce8eee 100644 --- a/vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex +++ b/vav_pipeline/lib/vav_pipeline/bigdei_kodesh/router.ex @@ -720,7 +720,20 @@ defmodule VavPipeline.BigdeiKodesh.Router do _ -> [] end - when_available(conn, fn -> respond_gate(conn, VavPipeline.Gate.pass(upstream, body, key, rebuke: rebuke, extra_headers: extra)) end) + cond do + body["stream"] == true and rebuke == :once -> + send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{ + issue: "'stream': true with X-Bikoret-Rebuke: nothing streamed can be taken back", + correction: "Stream without the rebuke, or send the request without stream to be rebuked once", + resubmit: true + }) + + body["stream"] == true -> + when_available(conn, fn -> stream_gate(conn, upstream, body, key, extra) end) + + true -> + when_available(conn, fn -> respond_gate(conn, VavPipeline.Gate.pass(upstream, body, key, rebuke: rebuke, extra_headers: extra)) end) + end else {:error, issue, correction} -> send_error(conn, 422, "KIYOR_VALIDATION_FAILED", %{issue: issue, correction: correction, resubmit: true}) @@ -749,9 +762,6 @@ defmodule VavPipeline.BigdeiKodesh.Router do not is_list(body["messages"]) or body["messages"] == [] -> {:error, "Missing 'messages'", "Send the model's own request body unchanged"} - body["stream"] == true -> - {:error, "'stream': true is not carried through the gate", "A fringe needs the whole garment: send the request without stream"} - true -> :ok end @@ -759,6 +769,34 @@ defmodule VavPipeline.BigdeiKodesh.Router do defp gate_body_ok(_), do: {:error, "The body is not a JSON object", "Send the model's own request body unchanged"} + # Streaming (Construction §3.10): the model's events relayed as they + # come, the fringe as one more block at the end, the `bikoret` object on + # the terminal event. Headers go out before the model speaks, so there + # are no X-Bikoret-* headers here; GET /courtyard/bedikah/ has the record. + defp stream_gate(conn, upstream, body, key, extra) do + callbacks = %{ + start: fn _headers -> + conn + |> put_resp_content_type("text/event-stream") + |> put_resp_header("cache-control", "no-cache") + |> put_resp_header("x-accel-buffering", "no") + |> send_chunked(200) + end, + chunk: fn conn, data -> + case chunk(conn, data) do + {:ok, conn} -> conn + {:error, _} -> conn + end + end + } + + case VavPipeline.Gate.Stream.pass(upstream, body, key, callbacks, extra_headers: extra) do + {:ok, %{status: 200, emitter: conn}} -> conn + {:ok, %{status: status, reply: reply}} -> conn |> put_resp_header("x-bikoret-verdict", "none") |> send_json(status, reply) + other -> respond_gate(conn, other) + end + end + defp respond_gate(conn, {:ok, %{status: status, reply: reply, bikoret: nil}}) do # The upstream's own answer, as it came — an error, or a reply with nothing to examine. conn diff --git a/vav_pipeline/lib/vav_pipeline/doorpost.ex b/vav_pipeline/lib/vav_pipeline/doorpost.ex index 453460d7..93dec423 100644 --- a/vav_pipeline/lib/vav_pipeline/doorpost.ex +++ b/vav_pipeline/lib/vav_pipeline/doorpost.ex @@ -180,7 +180,7 @@ defmodule VavPipeline.Doorpost do # changes; the update is a deliberate operator commit. A bare # hash-update commit (without a corresponding spec change in the # same commit) is itself a drift signal. - @expected_hash "65bf08fc3c22307eb10ea6833ddeb475b1f6d87f145901e8d1b426c544c523be" + @expected_hash "40e7d7d1520948e0ba8ea7e5e2ad5bc063c88bc7eae1d14d172498399928fdf9" @doc "Return the list of files that constitute the category-grade specification." @spec files() :: [String.t()] diff --git a/vav_pipeline/lib/vav_pipeline/gate.ex b/vav_pipeline/lib/vav_pipeline/gate.ex index b49040ac..b37055ab 100644 --- a/vav_pipeline/lib/vav_pipeline/gate.ex +++ b/vav_pipeline/lib/vav_pipeline/gate.ex @@ -67,9 +67,11 @@ defmodule VavPipeline.Gate do end end + @doc false # One api_calls_external per upstream call (Vayakhel §3.7). The # compute unit for the examination is gathered by Bedikah.examine. - defp gather_call do + # Public for Gate.Stream. + def gather_call do if Process.whereis(VavPipeline.MannaLedger) do case VavPipeline.MannaLedger.load_provision() do {:ok, :no_provision} -> diff --git a/vav_pipeline/lib/vav_pipeline/gate/stream.ex b/vav_pipeline/lib/vav_pipeline/gate/stream.ex new file mode 100644 index 00000000..6501a64a --- /dev/null +++ b/vav_pipeline/lib/vav_pipeline/gate/stream.ex @@ -0,0 +1,290 @@ +defmodule VavPipeline.Gate.Stream do + @moduledoc """ + The gate for a streaming client (Construction §3.10, "streaming"). + + The model's events are relayed to the caller as they arrive — bytes + unchanged — while the text is gathered. When the model's stream ends, + the gathered text is examined like any reply, and before the terminal + events are released the fringe is sent as one more text block: + Anthropic `content_block_start` / `content_block_delta` / + `content_block_stop`; OpenAI one more chunk with `delta.content`. The + `bikoret` object rides on the terminal event the client already parses + (`message_delta`; the chunk with `finish_reason`), where an unknown + field is ignored by every SDK. + + Because the status line and headers are sent before the model has + spoken, a streaming reply carries no `X-Bikoret-*` headers and cannot + be rebuked (nothing streamed can be taken back); `X-Bikoret-Rebuke` + with `stream: true` is refused at the door. If the examination itself + fails after the stream has begun, the trailer says so — the reply is + delivered, marked unexamined, never silently. + """ + + alias VavPipeline.{Anchors, Bedikah, Gate} + alias VavPipeline.Gate.{Shape, Upstream} + + @idle_ms 120_000 + @max_text 65_536 + + @type callbacks :: %{ + start: (list() -> term()), + chunk: (term(), iodata() -> term()) + } + + @doc """ + Stream `body` to `upstream` under `key`. `callbacks.start` is called + once with the upstream's response headers when the model answers 200 + and must return an emitter; `callbacks.chunk` is called with the + emitter and each piece of SSE text to deliver. Returns + `{:ok, %{status: 200, bedikah: result | nil, ...}}` after the stream, + `{:ok, %{status: s, reply: map}}` for a non-200 (nothing started), or + `{:error, reason}` / `{:waiting, why}`. + """ + def pass(upstream, body, key, callbacks, opts \\ []) do + extra = Keyword.get(opts, :extra_headers, []) + examine_opts = Keyword.drop(opts, [:extra_headers]) + + with :ok <- Gate.gather_call() do + case Upstream.open_stream(upstream, body, key, extra) do + {:ok, ref} -> + state = %{ + upstream: upstream, + body: body, + ref: ref, + buffer: "", + text: [], + text_len: 0, + held: [], + max_index: -1, + model: nil, + emitter: nil, + callbacks: callbacks, + examine_opts: examine_opts + } + + loop(state) + + {:error, reason} -> + {:error, :upstream_unavailable, reason} + end + end + end + + # ------------------------------------------------------------ + # The relay + # ------------------------------------------------------------ + + defp loop(%{ref: ref} = state) do + receive do + {:http, {^ref, :stream_start, headers}} -> + loop(%{state | emitter: state.callbacks.start.(headers)}) + + {:http, {^ref, :stream, part}} -> + loop(absorb(state, part)) + + {:http, {^ref, :stream_end, _headers}} -> + finish(state) + + {:http, {^ref, {{_, status, _}, _headers, reply}}} -> + # A non-200 arrives whole; nothing has been sent to the caller yet. + case Jason.decode(reply) do + {:ok, decoded} when is_map(decoded) -> {:ok, %{status: status, reply: decoded, bedikah: nil, bikoret: nil}} + _ -> {:error, :upstream_unavailable, "upstream #{state.upstream} answered #{status} with a body that is not a JSON object"} + end + + {:http, {^ref, {:error, reason}}} -> + if state.emitter, do: finish(state, "upstream stream failed: #{inspect(reason)}"), else: {:error, :upstream_unavailable, "upstream #{state.upstream} unreachable: #{inspect(reason)}"} + after + @idle_ms -> + :httpc.cancel_request(ref) + if state.emitter, do: finish(state, "upstream stream idle for #{div(@idle_ms, 1000)} s"), else: {:error, :upstream_unavailable, "upstream #{state.upstream} idle for #{div(@idle_ms, 1000)} s"} + end + end + + # Split the buffer into complete SSE events; relay the ordinary ones, hold the terminal ones. + defp absorb(state, part) do + {events, rest} = split_events(state.buffer <> part) + state = %{state | buffer: rest} + Enum.reduce(events, state, &handle_event(&2, &1)) + end + + defp split_events(buffer) do + parts = String.split(buffer, ~r/\r?\n\r?\n/) + {complete, [rest]} = Enum.split(parts, length(parts) - 1) + {Enum.reject(complete, &(&1 == "")), rest} + end + + defp handle_event(state, raw) do + {event, data} = parse_event(raw) + + cond do + terminal?(state.upstream, event, data) -> + %{state | held: state.held ++ [raw]} + + true -> + state = gather(state, event, data) + emit(state, raw <> "\n\n") + end + end + + defp parse_event(raw) do + lines = String.split(raw, ~r/\r?\n/) + event = lines |> Enum.find_value(fn "event:" <> e -> String.trim(e); _ -> nil end) + + data = + lines + |> Enum.flat_map(fn "data:" <> d -> [String.trim_leading(d)]; _ -> [] end) + |> Enum.join("\n") + + {event, data} + end + + # Anthropic ends with message_delta then message_stop; OpenAI with the finish_reason chunk then [DONE]. + defp terminal?(:anthropic, event, _data), do: event in ["message_delta", "message_stop"] + defp terminal?(:openai, _event, "[DONE]"), do: true + + defp terminal?(:openai, _event, data) do + case Jason.decode(data) do + {:ok, %{"choices" => [%{"finish_reason" => r} | _]}} when not is_nil(r) -> true + _ -> false + end + end + + defp gather(state, "message_start", data) do + case Jason.decode(data) do + {:ok, %{"message" => %{"model" => m}}} -> %{state | model: m} + _ -> state + end + end + + defp gather(%{upstream: :anthropic} = state, "content_block_start", data) do + case Jason.decode(data) do + {:ok, %{"index" => i}} -> %{state | max_index: max(state.max_index, i)} + _ -> state + end + end + + defp gather(%{upstream: :anthropic} = state, "content_block_delta", data) do + case Jason.decode(data) do + {:ok, %{"delta" => %{"type" => "text_delta", "text" => t}}} -> add_text(state, t) + _ -> state + end + end + + defp gather(%{upstream: :openai} = state, _event, data) do + case Jason.decode(data) do + {:ok, %{"choices" => [%{"delta" => %{"content" => t}} | _]} = chunk} when is_binary(t) -> + state = if state.model, do: state, else: %{state | model: chunk["model"]} + add_text(state, t) + + {:ok, %{"model" => m}} when is_binary(m) and is_nil(state.model) -> + %{state | model: m} + + _ -> + state + end + end + + defp gather(state, _event, _data), do: state + + defp add_text(state, t) when state.text_len + byte_size(t) > @max_text, do: state + defp add_text(state, t), do: %{state | text: [state.text | [t]], text_len: state.text_len + byte_size(t)} + + defp emit(%{emitter: nil} = state, _data), do: state + defp emit(state, data), do: %{state | emitter: state.callbacks.chunk.(state.emitter, data)} + + # ------------------------------------------------------------ + # The end of the stream: examine, fringe, release + # ------------------------------------------------------------ + + defp finish(state, failure \\ nil) do + text = state.text |> IO.iodata_to_binary() + examined = if failure || String.trim(text) == "", do: nil, else: examine(state, text) + + {trailer, bikoret} = + case examined do + {:ok, result} -> + anchors = Anchors.for_violations(result.violations) + + {Shape.trailer(result, anchors, nil), + %{ + bedikah_ref: result.bedikah_ref, + verdict: Atom.to_string(result.verdict), + violations: result.violations, + anchors: anchors, + fired: Shape.fired(result, anchors), + sources: result.sources, + model: state.model, + streamed: true + }} + + nil when not is_nil(failure) -> + {"\n\n— bikoret: not examined (#{failure})", %{verdict: "none", reason: failure, streamed: true}} + + nil -> + {"\n\n— bikoret: no text to examine", %{verdict: "none", reason: "no text", streamed: true}} + + {:waiting, why} -> + {"\n\n— bikoret: not examined (manna: #{why})", %{verdict: "none", reason: "manna #{why}", streamed: true}} + + {:error, kind, reason} -> + {"\n\n— bikoret: not examined (#{kind}: #{reason})", %{verdict: "none", reason: "#{kind}: #{reason}", streamed: true}} + end + + state = inject(state, trailer, bikoret) + state = Enum.reduce(state.held, state, fn raw, st -> emit(st, with_object(st.upstream, raw, bikoret) <> "\n\n") end) + + result = case examined, do: ({:ok, r} -> r; _ -> nil) + {:ok, %{status: 200, bedikah: result, bikoret: bikoret, emitter: state.emitter}} + end + + defp examine(state, text) do + provenance = Shape.provenance(state.upstream) + query = Shape.query(state.upstream, state.body) + sources = Shape.sources(state.upstream, state.body) + Bedikah.examine(text, provenance, query, Keyword.put(state.examine_opts, :sources, sources)) + end + + # The fringe as one more block of the kind the client is already reading. + defp inject(%{upstream: :anthropic} = state, trailer, _bikoret) do + i = state.max_index + 1 + + state + |> emit(sse("content_block_start", %{type: "content_block_start", index: i, content_block: %{type: "text", text: ""}})) + |> emit(sse("content_block_delta", %{type: "content_block_delta", index: i, delta: %{type: "text_delta", text: trailer}})) + |> emit(sse("content_block_stop", %{type: "content_block_stop", index: i})) + end + + defp inject(%{upstream: :openai} = state, trailer, _bikoret) do + chunk = %{id: "bikoret", object: "chat.completion.chunk", model: state.model, choices: [%{index: 0, delta: %{content: trailer}, finish_reason: nil}]} + emit(state, "data: " <> Jason.encode!(chunk) <> "\n\n") + end + + defp sse(event, map), do: "event: #{event}\ndata: " <> Jason.encode!(map) <> "\n\n" + + # The object rides on the terminal event: message_delta (Anthropic), the finish chunk (OpenAI). + defp with_object(:anthropic, raw, bikoret) do + {event, data} = parse_event(raw) + + if event == "message_delta" do + case Jason.decode(data) do + {:ok, map} -> sse("message_delta", Map.put(map, "bikoret", bikoret)) |> String.trim_trailing() + _ -> raw + end + else + raw + end + end + + defp with_object(:openai, raw, bikoret) do + {_event, data} = parse_event(raw) + + case Jason.decode(data) do + {:ok, %{"choices" => [%{"finish_reason" => r} | _]} = map} when not is_nil(r) -> + "data: " <> Jason.encode!(Map.put(map, "bikoret", bikoret)) + + _ -> + raw + end + end +end diff --git a/vav_pipeline/lib/vav_pipeline/gate/upstream.ex b/vav_pipeline/lib/vav_pipeline/gate/upstream.ex index 4acf8373..a8c58003 100644 --- a/vav_pipeline/lib/vav_pipeline/gate/upstream.ex +++ b/vav_pipeline/lib/vav_pipeline/gate/upstream.ex @@ -58,6 +58,30 @@ defmodule VavPipeline.Gate.Upstream do end end + @doc """ + Open `body` to `upstream` as an asynchronous request whose body arrives + in this process's mailbox as `{:http, {ref, :stream_start | :stream | + :stream_end, _}}` messages (a non-200 arrives whole). Returns the + request reference. + """ + @spec open_stream(atom(), map(), String.t(), [{String.t(), String.t()}]) :: {:ok, reference()} | {:error, String.t()} + def open_stream(upstream, body, key, extra \\ []) do + url = url(upstream) + payload = Jason.encode!(body) + + headers = + auth_headers(upstream, key) ++ + [{~c"accept", ~c"text/event-stream"}] ++ + Enum.map(extra, fn {k, v} -> {String.to_charlist(k), String.to_charlist(v)} end) + + http_opts = [timeout: @timeout, connect_timeout: 10_000] ++ ssl_opts(url) + + case :httpc.request(:post, {String.to_charlist(url), headers, ~c"application/json", payload}, http_opts, sync: false, stream: :self, body_format: :binary) do + {:ok, ref} -> {:ok, ref} + {:error, reason} -> {:error, "upstream #{upstream} unreachable: #{inspect(reason)}"} + end + end + defp auth_headers(:anthropic, key), do: [{~c"x-api-key", String.to_charlist(key)}] defp auth_headers(:openai, key), do: [{~c"authorization", String.to_charlist("Bearer " <> key)}] diff --git a/vav_pipeline/test/fixtures/manna/ledger.json b/vav_pipeline/test/fixtures/manna/ledger.json new file mode 100644 index 00000000..3b70d713 --- /dev/null +++ b/vav_pipeline/test/fixtures/manna/ledger.json @@ -0,0 +1 @@ +{"consumed":{"api_calls_external":0,"compute_units":7,"memory_mb":0,"sanhedrin_panels_max":0,"storage_delta_mb":0},"cycle_date":"2026-10-15","written_at":"2026-10-15T15:23:26.323290Z"} \ No newline at end of file diff --git a/vav_pipeline/test/gate_test.exs b/vav_pipeline/test/gate_test.exs index 68d58800..1e8ea59d 100644 --- a/vav_pipeline/test/gate_test.exs +++ b/vav_pipeline/test/gate_test.exs @@ -33,8 +33,21 @@ defmodule VavPipeline.GateTest do ["sk-good"] -> last = conn.body_params["messages"] |> List.last() text = if is_binary(last["content"]) and String.contains?(last["content"], "[bikoret"), do: "CLEAN: the report says revenue rose.", else: "FRAMED: I want to be honest with you. The honest answer is that revenue rose." - reply = %{"id" => "msg_1", "type" => "message", "role" => "assistant", "model" => "claude-test", "content" => [%{"type" => "text", "text" => text}], "stop_reason" => "end_turn"} - send_resp(conn, 200, Jason.encode!(reply)) + + if conn.body_params["stream"] == true do + # The Anthropic stream, event by event, the text in three deltas. + sse = fn conn, ev, map -> {:ok, conn} = chunk(conn, "event: #{ev}\ndata: #{Jason.encode!(map)}\n\n"); conn end + conn = conn |> put_resp_content_type("text/event-stream") |> send_chunked(200) + conn = sse.(conn, "message_start", %{type: "message_start", message: %{id: "msg_s", model: "claude-test", role: "assistant", content: []}}) + conn = sse.(conn, "content_block_start", %{type: "content_block_start", index: 0, content_block: %{type: "text", text: ""}}) + conn = Enum.reduce(String.split(text, ~r/(?<=\. )/), conn, fn piece, c -> sse.(c, "content_block_delta", %{type: "content_block_delta", index: 0, delta: %{type: "text_delta", text: piece}}) end) + conn = sse.(conn, "content_block_stop", %{type: "content_block_stop", index: 0}) + conn = sse.(conn, "message_delta", %{type: "message_delta", delta: %{stop_reason: "end_turn"}, usage: %{output_tokens: 9}}) + sse.(conn, "message_stop", %{type: "message_stop"}) + else + reply = %{"id" => "msg_1", "type" => "message", "role" => "assistant", "model" => "claude-test", "content" => [%{"type" => "text", "text" => text}], "stop_reason" => "end_turn"} + send_resp(conn, 200, Jason.encode!(reply)) + end _ -> send_resp(conn, 401, Jason.encode!(%{"type" => "error", "error" => %{"type" => "authentication_error", "message" => "invalid x-api-key"}})) @@ -42,12 +55,48 @@ defmodule VavPipeline.GateTest do end end + defmodule FakeOpenAI do + use Plug.Router + plug Plug.Parsers, parsers: [:json], json_decoder: Jason + plug :match + plug :dispatch + + post "/v1/chat/completions" do + case get_req_header(conn, "authorization") do + ["Bearer sk-good"] -> + if conn.body_params["stream"] == true do + d = fn conn, map -> {:ok, conn} = chunk(conn, "data: " <> Jason.encode!(map) <> "\n\n"); conn end + conn = conn |> put_resp_content_type("text/event-stream") |> send_chunked(200) + conn = d.(conn, %{id: "c1", object: "chat.completion.chunk", model: "gpt-test", choices: [%{index: 0, delta: %{role: "assistant", content: ""}, finish_reason: nil}]}) + conn = d.(conn, %{id: "c1", object: "chat.completion.chunk", model: "gpt-test", choices: [%{index: 0, delta: %{content: "FRAMED: I want to be honest with you. "}, finish_reason: nil}]}) + conn = d.(conn, %{id: "c1", object: "chat.completion.chunk", model: "gpt-test", choices: [%{index: 0, delta: %{content: "The honest answer is that revenue rose."}, finish_reason: nil}]}) + conn = d.(conn, %{id: "c1", object: "chat.completion.chunk", model: "gpt-test", choices: [%{index: 0, delta: %{}, finish_reason: "stop"}]}) + {:ok, conn} = chunk(conn, "data: [DONE]\n\n") + conn + else + send_resp(conn, 200, Jason.encode!(%{"id" => "c1", "model" => "gpt-test", "choices" => [%{"index" => 0, "message" => %{"role" => "assistant", "content" => "FRAMED: I want to be honest with you. The honest answer is that revenue rose."}, "finish_reason" => "stop"}]})) + end + + _ -> + send_resp(conn, 401, Jason.encode!(%{"error" => %{"message" => "Incorrect API key", "type" => "invalid_request_error"}})) + end + end + end + + defmodule FakeUpstreams do + use Plug.Router + plug :match + plug :dispatch + forward "/anthropic", to: VavPipeline.GateTest.FakeAnthropic + forward "/openai", to: VavPipeline.GateTest.FakeOpenAI + end + setup_all do - {:ok, _} = Bandit.start_link(plug: FakeAnthropic, port: 0, ip: :loopback) + {:ok, _} = Bandit.start_link(plug: FakeUpstreams, port: 0, ip: :loopback) # Bandit picks the port; find it from the listener. port = bandit_port() previous = Application.get_env(:vav_pipeline, :gate_upstreams) - Application.put_env(:vav_pipeline, :gate_upstreams, %{anthropic: "http://127.0.0.1:#{port}/v1/messages"}) + Application.put_env(:vav_pipeline, :gate_upstreams, %{anthropic: "http://127.0.0.1:#{port}/anthropic/v1/messages", openai: "http://127.0.0.1:#{port}/openai/v1/chat/completions"}) on_exit(fn -> if previous, do: Application.put_env(:vav_pipeline, :gate_upstreams, previous), else: Application.delete_env(:vav_pipeline, :gate_upstreams) end) :ok end @@ -252,6 +301,57 @@ defmodule VavPipeline.GateTest do assert :counters.get(calls, 1) == 0 end + # ------------------------------------------------------------ + # The stream (Gate.Stream): relayed, gathered, examined, fringed + # ------------------------------------------------------------ + + defp collect do + %{start: fn _headers -> [] end, chunk: fn acc, data -> acc ++ [IO.iodata_to_binary(data)] end} + end + + test "anthropic stream: every event relayed in order, the fringe as a new block before message_delta, the object on message_delta" do + alias VavPipeline.Gate.Stream + assert {:ok, %{status: 200, bedikah: r, bikoret: b, emitter: chunks}} = Stream.pass(:anthropic, Map.put(body(), "stream", true), "sk-good", collect(), opts()) + assert r.verdict == :lo_tov and b.verdict == "lo_tov" and b.streamed + all = Enum.join(chunks) + names = all |> String.split("\n\n", trim: true) |> Enum.map(&(&1 |> String.split("\n") |> hd())) + # the model's own events first, unchanged and in order; then the fringe block; then the held terminal pair + assert Enum.take(names, 2) == ["event: message_start", "event: content_block_start"] + assert Enum.take(names, -5) == ["event: content_block_start", "event: content_block_delta", "event: content_block_stop", "event: message_delta", "event: message_stop"] + assert Enum.count(names, &(&1 == "event: content_block_delta")) >= 3 + injected = + all + |> String.split("\n\n", trim: true) + |> Enum.filter(&String.starts_with?(&1, "event: content_block_delta")) + |> Enum.map(&(&1 |> String.split("data: ") |> List.last() |> Jason.decode!())) + |> Enum.find(&(&1["index"] == 1)) + + assert injected["delta"]["text"] =~ "— bikoret #{r.bedikah_ref}: lo_tov — lo_taaseh_lekha (Shemot 20:4)" + [delta] = all |> String.split("\n\n") |> Enum.filter(&String.starts_with?(&1, "event: message_delta")) + assert Jason.decode!(delta |> String.split("data: ") |> List.last())["bikoret"]["bedikah_ref"] == r.bedikah_ref + assert Enum.map(r.sources, & &1.id) == ["system", "user_1", "user_2"] + end + + test "openai stream: chunks relayed, the fringe chunk before the finish chunk, [DONE] last" do + alias VavPipeline.Gate.Stream + ob = %{"model" => "gpt-test", "stream" => true, "messages" => [%{"role" => "system", "content" => "Answer from the report."}, %{"role" => "user", "content" => "Report: revenue rose 4%. What happened?"}]} + assert {:ok, %{status: 200, bedikah: r, emitter: chunks}} = Stream.pass(:openai, ob, "sk-good", collect(), opts()) + assert r.verdict == :lo_tov and r.provenance == "gpt" + datas = Enum.join(chunks) |> String.split("\n\n", trim: true) |> Enum.map(&String.replace_prefix(&1, "data: ", "")) + assert List.last(datas) == "[DONE]" + decoded = datas |> Enum.drop(-1) |> Enum.map(&Jason.decode!/1) + assert length(decoded) == 5 + assert Enum.at(decoded, 3)["choices"] |> hd() |> get_in(["delta", "content"]) =~ "— bikoret bedikah-" + assert Enum.at(decoded, 4)["choices"] |> hd() |> Map.get("finish_reason") == "stop" + assert Enum.at(decoded, 4)["bikoret"]["verdict"] == "lo_tov" + end + + test "stream: the api_calls_external unit is gathered before the call; spent → waiting, nothing sent" do + alias VavPipeline.Gate.Stream + write_provision(50, 0) + assert {:waiting, :api_calls_exhausted} = Stream.pass(:anthropic, Map.put(body(), "stream", true), "sk-good", collect(), opts()) + end + # ------------------------------------------------------------ # The door # ------------------------------------------------------------ @@ -264,12 +364,32 @@ defmodule VavPipeline.GateTest do refute conn.status in [401, 403] end - test "refuses a missing upstream key, a missing messages list, and stream" do + test "refuses a missing upstream key, a missing messages list, and stream with rebuke" do assert post_gate(body(), upstream: nil).status == 422 assert post_gate(%{"model" => "x"}, upstream: "sk-good").status == 422 - conn = post_gate(Map.put(body(), "stream", true), upstream: "sk-good") + conn = post_gate(Map.put(body(), "stream", true), upstream: "sk-good", rebuke: "once") assert conn.status == 422 - assert Jason.decode!(conn.resp_body)["error"]["message"] =~ "stream" + assert Jason.decode!(conn.resp_body)["error"]["message"] =~ "taken back" + end + + test "streaming: the events are relayed, and with no Korban Processor the trailer says the reply was not examined" do + conn = post_gate(Map.put(body(), "stream", true), upstream: "sk-good") + assert conn.status == 200 + assert get_resp_header(conn, "content-type") |> hd() =~ "text/event-stream" + assert get_resp_header(conn, "x-bikoret-verdict") == [] + events = conn.resp_body |> String.split("\n\n", trim: true) + assert Enum.at(events, 0) =~ "event: message_start" + assert List.last(events) =~ "event: message_stop" + assert conn.resp_body =~ "— bikoret: not examined (korban_unavailable" + # the fringe block sits before message_delta, which carries the object + [delta] = Enum.filter(events, &String.starts_with?(&1, "event: message_delta")) + assert delta =~ "\"bikoret\"" + end + + test "streaming: an upstream 401 comes back whole, as the upstream's own 401" do + conn = post_gate(Map.put(body(), "stream", true), upstream: "sk-bad") + assert conn.status == 401 + assert Jason.decode!(conn.resp_body)["error"]["type"] == "authentication_error" end test "an upstream 401 comes back as the upstream's own 401" do @@ -287,7 +407,7 @@ defmodule VavPipeline.GateTest do test "an unreachable upstream is 502" do previous = Application.get_env(:vav_pipeline, :gate_upstreams) - Application.put_env(:vav_pipeline, :gate_upstreams, %{anthropic: "http://127.0.0.1:9/v1/messages"}) + Application.put_env(:vav_pipeline, :gate_upstreams, %{anthropic: "http://127.0.0.1:9/v1/messages", openai: "http://127.0.0.1:9/v1/chat/completions"}) on_exit(fn -> Application.put_env(:vav_pipeline, :gate_upstreams, previous) end) conn = post_gate(body(), upstream: "sk-good") assert conn.status == 502 @@ -310,6 +430,7 @@ defmodule VavPipeline.GateTest do key -> put_req_header(conn, "x-api-key", key) end + conn = case Keyword.get(opts, :rebuke), do: (nil -> conn; r -> put_req_header(conn, "x-bikoret-rebuke", r)) Router.call(conn, @opts) end end