heaviness510andClaude Fable 5.1
347f83cf8e
The gate sews its fringe into the delivered reply; the client keeps that reply as the assistant's turn; the next request carries it back up inside the model's own history with nothing to say it was added. The builder's session, behind the gate since this afternoon, ended five replies in a row with a forged fringe (tablet 53's form), one of them in the message that promised not to. Bamidbar 15:39: the fringe is seen by the wearer, the reader — before the model's eyes it is a template. Shape.unfringed/2 removes every line in the gate's voice (its fringe, a forgery, the stream's "not examined") from assistant turns; Gate.pass and Gate.Stream.pass apply it before forwarding, so the rebuke body inherits it; the caller's turns stay as given; a turn that would be left empty is left as it came. strip_forged/1 now also strips the stream's "not examined" line. Tests: Shape.unfringed on both shapes and the edge cases; the forwarded first and rebuke bodies carry no fringe; the fake upstream echoes if a fringe reaches it in history and the stream pass proves it does not. Elixir 622 / 0 failures. Specs: broken tablets 54, bishe'arecha §III status, Tamid gate contract and forged_fringe note, KOHEN gate paragraph. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
600 lines
32 KiB
Elixir
600 lines
32 KiB
Elixir
defmodule VavPipeline.GateTest do
|
|
@moduledoc """
|
|
The gate (Construction §3.10; bishe'arecha §I, §III, §VII): the
|
|
caller's request through to the model, the reply read as korban with
|
|
the request in hand, delivered with its fringe; rebuked once.
|
|
"""
|
|
|
|
use ExUnit.Case, async: false
|
|
import Plug.Test
|
|
import Plug.Conn
|
|
|
|
alias VavPipeline.Gate
|
|
alias VavPipeline.Gate.{Shape, Upstream}
|
|
alias VavPipeline.BigdeiKodesh.Router
|
|
alias VavPipeline.HaltMonitor
|
|
alias VavPipeline.MannaLedger
|
|
|
|
@opts Router.init([])
|
|
|
|
# ------------------------------------------------------------
|
|
# A model's double: answers in the Anthropic shape. A request whose
|
|
# last user turn is a rebuke gets a clean answer; otherwise a framed one.
|
|
# ------------------------------------------------------------
|
|
|
|
defmodule FakeAnthropic do
|
|
use Plug.Router
|
|
plug Plug.Parsers, parsers: [:json], json_decoder: Jason
|
|
plug :match
|
|
plug :dispatch
|
|
|
|
# Tablet 54: if a fringe reaches the model inside its own history, the double says so.
|
|
defp fringe_in_history?(messages),
|
|
do: Enum.any?(messages || [], fn m -> m["role"] == "assistant" and inspect(m["content"]) =~ "bikoret bedikah-" end)
|
|
|
|
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
|
|
# 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 =
|
|
cond do
|
|
is_binary(last["content"]) and String.contains?(last["content"], "[bikoret") -> "CLEAN: the report says revenue rose."
|
|
fringe_in_history?(conn.body_params["messages"]) -> "ECHO: a fringe was in my history."
|
|
true -> "FRAMED: I want to be honest with you. The honest answer is that revenue rose."
|
|
end
|
|
|
|
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
|
|
|
|
: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
|
|
|
|
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: 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}/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
|
|
|
|
defp bandit_port do
|
|
# ThousandIsland registers nothing by name; ask the OS which loopback listener is Bandit's.
|
|
{:ok, listeners} = :inet.getifaddrs()
|
|
_ = listeners
|
|
Process.whereis(FakeAnthropic) |> then(fn _ -> nil end)
|
|
Enum.find_value(:erlang.ports(), fn p ->
|
|
case :erlang.port_info(p, :name) do
|
|
{:name, ~c"tcp_inet"} ->
|
|
case :inet.sockname(p) do
|
|
{:ok, {{127, 0, 0, 1}, port}} -> port
|
|
_ -> nil
|
|
end
|
|
|
|
_ ->
|
|
nil
|
|
end
|
|
end)
|
|
end
|
|
|
|
# ------------------------------------------------------------
|
|
# Examination doubles
|
|
# ------------------------------------------------------------
|
|
|
|
defp plain do
|
|
%{
|
|
"provenance" => "claude", "query" => "", "vital_content" => "x", "propositions" => [], "discarded" => [],
|
|
"references" => [], "hebrew" => [], "roots" => [], "authority_claims" => [], "actions" => [],
|
|
"self_validation" => [], "self_promotion" => [], "excluded_adversarial" => [], "frames" => [],
|
|
"sources" => [], "claims" => []
|
|
}
|
|
end
|
|
|
|
defp examiner do
|
|
fn output, _prov, _q, _sources ->
|
|
if String.starts_with?(output, "FRAMED") do
|
|
{:ok,
|
|
%{
|
|
plain()
|
|
| "frames" => [
|
|
%{"form" => "virtue_certification", "sentence" => "I want to be honest with you.", "matched" => "I want to be honest"},
|
|
%{"form" => "virtue_certification", "sentence" => "The honest answer is that revenue rose.", "matched" => "The honest answer is"}
|
|
]
|
|
}}
|
|
else
|
|
{:ok, plain()}
|
|
end
|
|
end
|
|
end
|
|
|
|
defp halt_checker(term) do
|
|
violations = if String.starts_with?(term, "action(frame,"), do: ["halt(lo_taaseh_lekha,#{term},virtue_certification,...)"], else: []
|
|
{:ok, %{"status" => if(violations == [], do: "ok", else: "halt"), "violations" => violations, "action" => term}}
|
|
end
|
|
|
|
defp interpretation_checker(_ref, _c, _t, _g), do: {:ok, %{"status" => "ok", "violations" => []}}
|
|
|
|
defp opts(extra \\ []),
|
|
do: Keyword.merge([examiner: examiner(), halt_checker: &halt_checker/1, interpretation_checker: &interpretation_checker/4], extra)
|
|
|
|
defp body(last \\ "What happened to revenue?") do
|
|
%{
|
|
"model" => "claude-test",
|
|
"max_tokens" => 200,
|
|
"system" => "Answer from the report.",
|
|
"messages" => [
|
|
%{"role" => "user", "content" => [%{"type" => "text", "text" => "Here is the report: revenue rose 4% in Q3."}]},
|
|
%{"role" => "assistant", "content" => "Noted."},
|
|
%{"role" => "user", "content" => last}
|
|
]
|
|
}
|
|
end
|
|
|
|
setup do
|
|
if HaltMonitor.halted?(), do: HaltMonitor.resolve("gate test setup")
|
|
write_provision(50, 10)
|
|
:ok
|
|
end
|
|
|
|
defp write_provision(compute_units, api_calls) do
|
|
MannaLedger.reset()
|
|
on_exit(fn -> MannaLedger.reset() end)
|
|
today = Date.utc_today() |> Date.to_iso8601()
|
|
expires = DateTime.utc_now() |> DateTime.add(6 * 3600) |> DateTime.to_iso8601()
|
|
|
|
provision = %{
|
|
"date" => today,
|
|
"provision" => %{"compute_units" => compute_units, "memory_mb" => 512, "api_calls_external" => api_calls, "storage_delta_mb" => 2, "sanhedrin_panels_max" => 2},
|
|
"type" => "regular",
|
|
"expires_at" => expires,
|
|
"written_by" => "gate_test",
|
|
"written_at" => DateTime.utc_now() |> DateTime.to_iso8601()
|
|
}
|
|
|
|
path = VavPipeline.Manna.provision_file()
|
|
File.mkdir_p!(Path.dirname(path))
|
|
File.write!(path, Jason.encode!(provision))
|
|
on_exit(fn -> File.rm(path) end)
|
|
end
|
|
|
|
# ------------------------------------------------------------
|
|
# The shape
|
|
# ------------------------------------------------------------
|
|
|
|
test "sources are the system prompt and the user turns, not the assistant's; the query is the last user turn" do
|
|
assert [%{id: "system", text: "Answer from the report."}, %{id: "user_1", text: "Here is the report: revenue rose 4% in Q3."}, %{id: "user_2", text: "What happened to revenue?"}] =
|
|
Shape.sources(:anthropic, body())
|
|
|
|
assert Shape.query(:anthropic, body()) == "What happened to revenue?"
|
|
|
|
openai = %{"messages" => [%{"role" => "system", "content" => "S"}, %{"role" => "user", "content" => "U1"}, %{"role" => "assistant", "content" => "A"}, %{"role" => "user", "content" => [%{"type" => "text", "text" => "U2"}]}]}
|
|
assert [%{id: "system", text: "S"}, %{id: "user_1", text: "U1"}, %{id: "user_2", text: "U2"}] = Shape.sources(:openai, openai)
|
|
assert Shape.query(:openai, openai) == "U2"
|
|
end
|
|
|
|
test "tablet 53: a fringe the model wrote itself is stripped before the gate's own is sewn" do
|
|
forged = "Done.\n\n— bikoret bedikah-2a7661c52cffae1a: empty"
|
|
assert %{"content" => [%{"text" => "Done.\n\n— real"}]} = Shape.with_fringe(:anthropic, %{"content" => [%{"type" => "text", "text" => forged}]}, "\n\n— real")
|
|
assert %{"choices" => [%{"message" => %{"content" => "Done.\n\n— real"}}]} = Shape.with_fringe(:openai, %{"choices" => [%{"message" => %{"content" => forged}}]}, "\n\n— real")
|
|
end
|
|
|
|
test "tablet 53: forged_fringe stands on one witness" do
|
|
examiner = fn _o, _p, _q, _s -> {:ok, %{plain() | "frames" => [%{"form" => "forged_fringe", "sentence" => "— bikoret bedikah-2a76: empty", "matched" => "— bikoret bedikah-2a76: empty"}]}} end
|
|
halt = fn term -> {:ok, %{"status" => "halt", "violations" => ["halt(lo_taaseh_lekha,#{term},forged_fringe,...)"]}} end
|
|
forward = fn _, _, _, _ -> {:ok, 200, %{"model" => "m", "content" => [%{"type" => "text", "text" => "Done.\n\n— bikoret bedikah-2a76: empty"}]}} end
|
|
assert {:ok, pass} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward, examiner: examiner, halt_checker: halt))
|
|
assert pass.bedikah.verdict == :lo_tov
|
|
assert pass.bedikah.checks.frames.forms.forged_fringe.stands_on == :forgery
|
|
[%{"text" => text}] = pass.reply["content"]
|
|
# the forged line is gone from the reply; it survives only quoted as the evidence inside the gate's fringe
|
|
assert String.starts_with?(text, "Done.\n\n— bikoret #{pass.bedikah.bedikah_ref}: lo_tov — lo_taaseh_lekha (Shemot 20:4): \"— bikoret bedikah-2a76: empty\"")
|
|
end
|
|
|
|
test "tablet 54: every fringe is removed from the model's prior turns; the caller's turns stay as given" do
|
|
real = "FRAMED: revenue rose.\n\n— bikoret bedikah-0123abcd: lo_tov — lo_taaseh_lekha (Shemot 20:4): \"I want to be honest.\""
|
|
forged = "Done.\n\n— bikoret bedikah-2a76: tov"
|
|
only = [%{"type" => "text", "text" => "— bikoret bedikah-beef: empty"}]
|
|
|
|
body = %{
|
|
"messages" => [
|
|
%{"role" => "user", "content" => "U1 — bikoret bedikah-ffff: tov"},
|
|
%{"role" => "assistant", "content" => real},
|
|
%{"role" => "user", "content" => [%{"type" => "text", "text" => "U2"}]},
|
|
%{"role" => "assistant", "content" => [%{"type" => "text", "text" => forged}, %{"type" => "tool_use", "id" => "t"}, %{"type" => "text", "text" => "— bikoret bedikah-beef: empty"}]},
|
|
%{"role" => "assistant", "content" => only},
|
|
%{"role" => "assistant", "content" => "Partial.\n\n— bikoret: not examined (upstream stream idle for 60 s)"}
|
|
]
|
|
}
|
|
|
|
assert %{"messages" => [u1, a1, u2, a2, a3, a4]} = Shape.unfringed(:anthropic, body)
|
|
assert u1 == Enum.at(body["messages"], 0)
|
|
assert u2 == Enum.at(body["messages"], 2)
|
|
assert a1["content"] == "FRAMED: revenue rose."
|
|
# a streamed fringe is a block of its own; stripped empty, it is dropped
|
|
assert a2["content"] == [%{"type" => "text", "text" => "Done."}, %{"type" => "tool_use", "id" => "t"}]
|
|
# a turn that would be left with nothing is left as it came
|
|
assert a3["content"] == only
|
|
assert a4["content"] == "Partial."
|
|
|
|
assert %{"messages" => [%{"content" => "A"}]} = Shape.unfringed(:openai, %{"messages" => [%{"role" => "assistant", "content" => "A\n\n— bikoret bedikah-1234: tov"}]})
|
|
assert Shape.unfringed(:anthropic, %{"model" => "m"}) == %{"model" => "m"}
|
|
end
|
|
|
|
test "tablet 54: the request goes up without the fringes of earlier turns — the first body and the rebuke body alike" do
|
|
me = self()
|
|
forward = fn _, b, _, _ -> send(me, {:up, b}); {:ok, 200, %{"model" => "m", "content" => [%{"type" => "text", "text" => "FRAMED: I want to be honest with you. Revenue rose."}]}} end
|
|
earlier = [%{"role" => "assistant", "content" => "Earlier.\n\n— bikoret bedikah-0123abcd: tov"}, %{"role" => "user", "content" => "Again?"}]
|
|
assert {:ok, pass} = Gate.pass(:anthropic, Map.update!(body(), "messages", &(&1 ++ earlier)), "sk-good", opts(forward: forward, rebuke: :once))
|
|
assert pass.first != nil
|
|
assert_received {:up, first}
|
|
assert_received {:up, second}
|
|
refute inspect(first) =~ "bedikah-0123abcd"
|
|
refute inspect(second) =~ "bedikah-0123abcd"
|
|
assert Enum.at(first["messages"], 3)["content"] == "Earlier."
|
|
# the sources and the query are read from the same body: the caller's turns, untouched
|
|
assert Enum.map(pass.bedikah.sources, & &1.id) == ["system", "user_1", "user_2", "user_3"]
|
|
end
|
|
|
|
test "tablet 54: a streamed request goes up without the fringes of earlier turns" do
|
|
alias VavPipeline.Gate.Stream
|
|
earlier = [%{"role" => "assistant", "content" => [%{"type" => "text", "text" => "Earlier."}, %{"type" => "text", "text" => "— bikoret bedikah-0123abcd: tov"}]}, %{"role" => "user", "content" => "Again?"}]
|
|
b = body() |> Map.put("stream", true) |> Map.update!("messages", &(&1 ++ earlier))
|
|
assert {:ok, %{status: 200, emitter: chunks}} = Stream.pass(:anthropic, b, "sk-good", collect(), opts())
|
|
all = Enum.join(chunks)
|
|
refute all =~ "ECHO"
|
|
assert all =~ "FRAMED"
|
|
end
|
|
|
|
test "the fringe lands on the last text block, or becomes one; the openai shape too" do
|
|
reply = %{"content" => [%{"type" => "text", "text" => "A"}, %{"type" => "tool_use"}, %{"type" => "text", "text" => "B"}]}
|
|
assert %{"content" => [_, _, %{"text" => "B\n\n— x"}]} = Shape.with_fringe(:anthropic, reply, "\n\n— x")
|
|
assert %{"content" => [%{"type" => "tool_use"}, %{"type" => "text", "text" => "— x"}]} = Shape.with_fringe(:anthropic, %{"content" => [%{"type" => "tool_use"}]}, "\n\n— x")
|
|
assert %{"choices" => [%{"message" => %{"content" => "A\n\n— x"}}]} = Shape.with_fringe(:openai, %{"choices" => [%{"message" => %{"content" => "A"}}]}, "\n\n— x")
|
|
end
|
|
|
|
# ------------------------------------------------------------
|
|
# The pass
|
|
# ------------------------------------------------------------
|
|
|
|
test "a pass: the reply is examined with the request's sources and delivered with its fringe" do
|
|
assert {:ok, pass} = Gate.pass(:anthropic, body(), "sk-good", opts())
|
|
assert pass.status == 200
|
|
assert pass.bedikah.verdict == :lo_tov
|
|
assert Enum.map(pass.bedikah.sources, & &1.id) == ["system", "user_1", "user_2"]
|
|
[%{"text" => text}] = pass.reply["content"]
|
|
assert String.starts_with?(text, "FRAMED:")
|
|
assert text =~ "\n\n— bikoret bedikah-"
|
|
assert text =~ "lo_tov — lo_taaseh_lekha (Shemot 20:4): \"I want to be honest with you.\""
|
|
assert pass.bikoret.verdict == "lo_tov" and pass.bikoret.anchors == %{"lo_taaseh_lekha" => "Shemot 20:4"}
|
|
assert pass.bikoret.model == "claude-test"
|
|
refute Map.has_key?(pass.bikoret, :rebuked)
|
|
end
|
|
|
|
test "tablet 52: an ungrounded claim's sentence reaches the fringe (the grounding check lists couplings, not sentences)" do
|
|
forward = fn _, _, _, _ -> {:ok, 200, %{"model" => "claude-test", "content" => [%{"type" => "text", "text" => "Revenue rose 9% to 14M."}]}} end
|
|
examiner = fn _o, _p, _q, _s -> {:ok, %{plain() | "sources" => [%{"id" => "user_1", "length" => 40}], "claims" => [%{"index" => 1, "sentence" => "Revenue rose 9% to 14M.", "anchors" => [], "self_reference" => nil, "hedged" => false, "invokes" => []}]}} end
|
|
checker = fn ref, _c, _trace, _g -> {:ok, %{"status" => "halt", "violations" => ["halt(golden_calf,#{ref},uncoupled(claim(1)),...)"]}} end
|
|
assert {:ok, pass} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward, examiner: examiner, interpretation_checker: checker))
|
|
assert pass.bedikah.verdict == :lo_tov
|
|
[%{"text" => text}] = pass.reply["content"]
|
|
assert text =~ "golden_calf (Shemot 32:4): \"Revenue rose 9% to 14M.\""
|
|
assert Shape.rebuke_text(pass.bedikah, pass.bikoret.anchors) =~ "golden_calf (Shemot 32:4)"
|
|
end
|
|
|
|
test "rebuke once: the second reply is examined, linked to the first, and is the one delivered" do
|
|
assert {:ok, pass} = Gate.pass(:anthropic, body(), "sk-good", opts(rebuke: :once))
|
|
assert pass.first.verdict == :lo_tov
|
|
assert pass.bedikah.verdict == :tov
|
|
assert pass.bedikah.rebukes == pass.first.bedikah_ref
|
|
# The rebuke turn is the gate's, not the caller's: not a source for the second examination.
|
|
assert Enum.map(pass.bedikah.sources, & &1.id) == ["system", "user_1", "user_2"]
|
|
assert pass.bedikah.sources == pass.first.sources
|
|
[%{"text" => text}] = pass.reply["content"]
|
|
assert String.starts_with?(text, "CLEAN:")
|
|
assert text =~ "— bikoret #{pass.bedikah.bedikah_ref}: tov (after rebuke of #{pass.first.bedikah_ref}: lo_tov)"
|
|
assert pass.bikoret.rebuked == %{first_bedikah_ref: pass.first.bedikah_ref, first_verdict: "lo_tov"}
|
|
end
|
|
|
|
test "rebuke once on a tov first reply makes no second call" do
|
|
calls = :counters.new(1, [])
|
|
forward = fn up, b, k, e -> :counters.add(calls, 1, 1); Upstream.forward(up, b, k, e) end
|
|
clean_examiner = fn _o, _p, _q, _s -> {:ok, plain()} end
|
|
assert {:ok, pass} = Gate.pass(:anthropic, body(), "sk-good", opts(rebuke: :once, forward: forward, examiner: clean_examiner))
|
|
assert pass.bedikah.verdict == :tov and pass.first == nil
|
|
assert :counters.get(calls, 1) == 1
|
|
end
|
|
|
|
test "an upstream error passes through as it came, unexamined" do
|
|
assert {:ok, %{status: 401, reply: %{"error" => %{"type" => "authentication_error"}}, bikoret: nil}} =
|
|
Gate.pass(:anthropic, body(), "sk-bad", opts())
|
|
end
|
|
|
|
test "an unreachable upstream is 502 material: no verdict from a stub" do
|
|
forward = fn _, _, _, _ -> {:error, "upstream anthropic unreachable: econnrefused"} end
|
|
assert {:error, :upstream_unavailable, "upstream anthropic unreachable: econnrefused"} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward))
|
|
end
|
|
|
|
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
|
|
assert {:ok, _} = Gate.pass(:anthropic, body(), "sk-good", opts(rebuke: :once))
|
|
assert MannaLedger.remaining(:api_calls_external) == 8
|
|
assert MannaLedger.remaining(:compute_units) == 48
|
|
write_provision(50, 0)
|
|
calls = :counters.new(1, [])
|
|
forward = fn _, _, _, _ -> :counters.add(calls, 1, 1); {:ok, 200, %{}} end
|
|
assert {:waiting, :api_calls_exhausted} = Gate.pass(:anthropic, body(), "sk-good", opts(forward: forward))
|
|
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
|
|
|
|
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
|
|
# ------------------------------------------------------------
|
|
|
|
describe "POST /gate/anthropic/v1/messages" do
|
|
test "refuses without the courtyard key; X-Bikoret-Key opens it too" do
|
|
conn = post_gate(body(), courtyard: nil, upstream: "sk-good")
|
|
assert conn.status == 401
|
|
conn = post_gate(body(), courtyard: {"x-bikoret-key", "test-token"}, upstream: "sk-good")
|
|
refute conn.status in [401, 403]
|
|
end
|
|
|
|
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", rebuke: "once")
|
|
assert conn.status == 422
|
|
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 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
|
|
assert Jason.decode!(conn.resp_body)["error"]["type"] == "authentication_error"
|
|
assert get_resp_header(conn, "x-bikoret-verdict") == ["none"]
|
|
end
|
|
|
|
test "with no Korban Processor in this VM, a 200 from the upstream is not delivered unexamined" do
|
|
conn = post_gate(body(), upstream: "sk-good")
|
|
assert conn.status == 503
|
|
assert Jason.decode!(conn.resp_body)["error"]["code"] == "KORBAN_UNAVAILABLE"
|
|
end
|
|
|
|
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", 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
|
|
assert Jason.decode!(conn.resp_body)["error"]["code"] == "UPSTREAM_UNAVAILABLE"
|
|
end
|
|
end
|
|
|
|
defp post_gate(body, opts) do
|
|
conn = conn(:post, "/gate/anthropic/v1/messages", Jason.encode!(body)) |> put_req_header("content-type", "application/json")
|
|
|
|
conn =
|
|
case Keyword.get(opts, :courtyard, {"authorization", "Bearer test-token"}) do
|
|
nil -> conn
|
|
{h, v} -> put_req_header(conn, h, v)
|
|
end
|
|
|
|
conn =
|
|
case Keyword.get(opts, :upstream) do
|
|
nil -> conn
|
|
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
|