gate: streaming — the model's events relayed as they come, the fringe as the last block

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 <noreply@anthropic.com>

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/<ref> 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

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()]

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} ->

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

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)}]