gate: Claude Code behind it — OAuth bearer, anthropic-beta, token counting, the gate's own watch

A coding agent's traffic differs from a hand-made request: with a
subscription Claude Code authenticates with an OAuth bearer and sends
anthropic-beta; it counts tokens at /v1/messages/count_tokens; it makes a
request per tool call; its tool-only turns have no text; its context runs
to megabytes. So: with the courtyard key in X-Bikoret-Key, the caller's
Authorization: Bearer is forwarded to Anthropic as its credential, and
anthropic-beta with it; count_tokens is carried through unexamined and
uncharged; the watch counts the gate at its own windows (120/min,
2,000/h per key — tier :gate); a reply with no text is delivered as it
came with bikoret.verdict "none" and no fringe block (stream and
non-stream); the sources keep the newest user turns whole and drop the
oldest past 1 MiB. Ingress body 32 MiB; ConfigMap manna amounts raised
to 5000/5000 (the kohen's lever). Tamid §1.1, KOHEN "Claude Code behind
the gate". Doorpost re-pinned. Elixir 617/0 read before this commit;
gate_test 25.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

vav_pipeline/test/gate_test.exs

@@ -28,9 +28,40 @@ defmodule VavPipeline.GateTest do
     plug :match
     plug :dispatch
 
+    post "/v1/messages/count_tokens" do
+      case {get_req_header(conn, "x-api-key"), get_req_header(conn, "authorization")} do
+        {["sk-good"], _} -> send_resp(conn, 200, Jason.encode!(%{"input_tokens" => 42}))
+        {_, ["Bearer sk-good-oauth"]} -> send_resp(conn, 200, Jason.encode!(%{"input_tokens" => 42}))
+        _ -> send_resp(conn, 401, Jason.encode!(%{"type" => "error", "error" => %{"type" => "authentication_error", "message" => "invalid"}}))
+      end
+    end
+
     post "/v1/messages" do
-      case get_req_header(conn, "x-api-key") do
-        ["sk-good"] ->
+      # Tool-only turn: a reply with no text block.
+      conn =
+        if conn.body_params["max_tokens"] == 7 and conn.body_params["stream"] == true do
+          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_t", model: "claude-test", role: "assistant", content: []}})
+          conn = sse.(conn, "content_block_start", %{type: "content_block_start", index: 0, content_block: %{type: "tool_use", id: "tu_1", name: "read", input: %{}}})
+          conn = sse.(conn, "content_block_delta", %{type: "content_block_delta", index: 0, delta: %{type: "input_json_delta", partial_json: "{\"path\":\"x\"}"}})
+          conn = sse.(conn, "content_block_stop", %{type: "content_block_stop", index: 0})
+          conn = sse.(conn, "message_delta", %{type: "message_delta", delta: %{stop_reason: "tool_use"}, usage: %{output_tokens: 3}})
+          sse.(conn, "message_stop", %{type: "message_stop"})
+        else
+          conn
+        end
+
+      if conn.state == :chunked do
+        conn
+      else
+      auth = case {get_req_header(conn, "x-api-key"), get_req_header(conn, "authorization")} do
+        {["sk-good"], _} -> :ok
+        {_, ["Bearer sk-good-oauth"]} -> if get_req_header(conn, "anthropic-beta") == ["oauth-2025-04-20"], do: :ok, else: :no_beta
+        _ -> :bad
+      end
+      case auth do
+        :ok ->
           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."
 
@@ -49,9 +80,13 @@ defmodule VavPipeline.GateTest do
             send_resp(conn, 200, Jason.encode!(reply))
           end
 
+        :no_beta ->
+          send_resp(conn, 400, Jason.encode!(%{"type" => "error", "error" => %{"type" => "invalid_request_error", "message" => "anthropic-beta required for oauth"}}))
+
         _ ->
           send_resp(conn, 401, Jason.encode!(%{"type" => "error", "error" => %{"type" => "authentication_error", "message" => "invalid x-api-key"}}))
       end
+      end
     end
   end
 
@@ -284,10 +319,10 @@ defmodule VavPipeline.GateTest do
     assert {:error, :upstream_unavailable, "upstream anthropic unreachable: econnrefused"} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward))
   end
 
-  test "a reply with no text cannot be examined" do
-    forward = fn _, _, _, _ -> {:ok, 200, %{"content" => [%{"type" => "tool_use"}]}} end
-    assert {:error, :upstream_unavailable, reason} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward))
-    assert reason =~ "no text"
+  test "a reply with no text (tool use only) is delivered as it came; the object says no text" do
+    forward = fn _, _, _, _ -> {:ok, 200, %{"model" => "claude-test", "content" => [%{"type" => "tool_use"}]}} end
+    assert {:ok, %{status: 200, bedikah: nil, reply: %{"content" => [%{"type" => "tool_use"}]}, bikoret: %{verdict: "none", reason: "no text"}}} =
+             Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward))
   end
 
   test "manna: one external call per upstream request; spent calls wait before any call is made" do
@@ -352,6 +387,22 @@ defmodule VavPipeline.GateTest do
     assert {:waiting, :api_calls_exhausted} = Stream.pass(:anthropic, Map.put(body(), "stream", true), "sk-good", collect(), opts())
   end
 
+  test "a tool-only streamed turn gets no fringe block; the object says no text" do
+    alias VavPipeline.Gate.Stream
+    assert {:ok, %{status: 200, bedikah: nil, bikoret: %{verdict: "none", reason: "no text"}, emitter: chunks}} =
+             Stream.pass(:anthropic, body() |> Map.put("stream", true) |> Map.put("max_tokens", 7), "sk-good", collect(), opts())
+    names = Enum.join(chunks) |> String.split("\n\n", trim: true) |> Enum.map(&(&1 |> String.split("\n") |> hd()))
+    assert names == ["event: message_start", "event: content_block_start", "event: content_block_delta", "event: content_block_stop", "event: message_delta", "event: message_stop"]
+    assert Enum.join(chunks) =~ "\"bikoret\":{"
+  end
+
+  test "sources keep the newest turns whole and drop the oldest past 1 MiB" do
+    big = String.duplicate("x", 700_000)
+    b = %{"messages" => [%{"role" => "user", "content" => big <> " one"}, %{"role" => "user", "content" => big <> " two"}, %{"role" => "user", "content" => "last"}]}
+    ids = Shape.sources(:anthropic, b) |> Enum.map(& &1.id)
+    assert ids == ["user_2", "user_3"]
+  end
+
   # ------------------------------------------------------------
   # The door
   # ------------------------------------------------------------
@@ -392,6 +443,36 @@ defmodule VavPipeline.GateTest do
       assert Jason.decode!(conn.resp_body)["error"]["type"] == "authentication_error"
     end
 
+    test "an OAuth bearer with the courtyard key in X-Bikoret-Key is forwarded, anthropic-beta with it" do
+      conn =
+        conn(:post, "/gate/anthropic/v1/messages", Jason.encode!(body()))
+        |> put_req_header("content-type", "application/json")
+        |> put_req_header("x-bikoret-key", "test-token")
+        |> put_req_header("authorization", "Bearer sk-good-oauth")
+        |> put_req_header("anthropic-beta", "oauth-2025-04-20")
+        |> Router.call(@opts)
+
+      # reached the model (which demanded the beta header) — then, with no Korban in this VM, 503
+      assert conn.status == 503
+      assert Jason.decode!(conn.resp_body)["error"]["code"] == "KORBAN_UNAVAILABLE"
+    end
+
+    test "token counting is carried through under the caller's credential, unexamined" do
+      conn =
+        conn(:post, "/gate/anthropic/v1/messages/count_tokens", Jason.encode!(%{"model" => "x", "messages" => [%{"role" => "user", "content" => "hi"}]}))
+        |> put_req_header("content-type", "application/json")
+        |> put_req_header("authorization", "Bearer test-token")
+        |> put_req_header("x-api-key", "sk-good")
+        |> Router.call(@opts)
+
+      assert conn.status == 200
+      assert Jason.decode!(conn.resp_body) == %{"input_tokens" => 42}
+    end
+
+    test "the gate counts knocks at its own windows: 120 a minute, not the courtyard's 20" do
+      assert VavPipeline.BigdeiKodesh.Mishmar.limits(:gate) == [minute: 120, hour: 2_000]
+    end
+
     test "an upstream 401 comes back as the upstream's own 401" do
       conn = post_gate(body(), upstream: "sk-bad")
       assert conn.status == 401