diff --git a/.env.example b/.env.example index 201282873..22025ea92 100644 --- a/.env.example +++ b/.env.example @@ -330,3 +330,22 @@ AGENT_TOOL_TOKEN= # Do not accept a default in production. WORKER_SHARED_SECRET= + +# --- BitMind gateway (server/src/bitmind/main.ts) --------------------------------- +# +# The doorway BitMind's run plane talks through, run as its own process inside the +# execution enclave. All three are required for it to start; the tokens have no +# defaults on purpose. Generate: openssl rand -hex 32 +# +# BITMIND_SERVICE_TOKEN authenticates BitMind's worker to this gateway. +# BITMIND_AGENT_TOKEN is the managed-agent token of the downstream AG-UI agent +# (agent-langgraph's MANAGED_AGENT_TOKEN) the gateway relays runs to. +BITMIND_SERVICE_TOKEN= +BITMIND_AGENT_TOKEN= +# Where runs are relayed. Loopback in the enclave; defaults to agent-langgraph. +BITMIND_AGENT_URL=http://localhost:4201/ag-ui +# Loopback bind and admission ceilings. The enclave note starts staging at two. +BITMIND_GATEWAY_HOST=127.0.0.1 +BITMIND_GATEWAY_PORT=4310 +BITMIND_MAX_CONCURRENT_RUNS=2 +BITMIND_RUN_TIMEOUT_MS=900000 diff --git a/agent-langgraph/src/index.ts b/agent-langgraph/src/index.ts index fc7ed28f2..87c3f9bb5 100644 --- a/agent-langgraph/src/index.ts +++ b/agent-langgraph/src/index.ts @@ -1,5 +1,4 @@ -import type { BaseEvent, RunAgentInput } from "@ag-ui/core"; -import { EventEncoder } from "@ag-ui/encoder"; +import type { RunAgentInput } from "@ag-ui/core"; import { ChatAnthropic } from "@langchain/anthropic"; import { type AIMessage, ToolMessage } from "@langchain/core/messages"; import { ChatGoogleGenerativeAI } from "@langchain/google-genai"; @@ -14,7 +13,8 @@ import { serve } from "bun"; import { hasManagedAgentToken } from "../../shared/agent-authorisation"; import { toLangChainMessages } from "./history"; import { readReasoningEffort } from "./model-options"; -import { streamRun } from "./stream"; +import { respondWithRun } from "./respond"; +import { callDeploymentTool } from "./tools"; /** * The same Bot, on a framework. @@ -227,52 +227,12 @@ function buildModel() { * Not the vendor: this deployment. A Bot that called an MCP server directly would be a Bot that * walked around the grant, the policy and the audit row, and those are the product. So the loop runs * here, in this process, and every call it makes goes back through the deployment that granted it. + * The call itself lives in `tools.ts`, where a test can cancel it mid-flight. */ const TOOL_URL = process.env.OPENBOT_TOOL_URL ?? "http://localhost:3001/api/agent-tools/call"; const TOOL_TOKEN = process.env.AGENT_TOOL_TOKEN ?? ""; -async function callTool( - run: string, - name: string, - args: Record, -): Promise { - if (!TOOL_TOKEN) { - return "Refused. This Bot has no credential for calling tools back through its deployment."; - } - if (!run) { - /* - * No statement from the deployment about whose run this is, so there is nothing to act on behalf - * of. Reported as a result rather than thrown: the run continues and says what it could not do. - */ - return "Refused. This run carried no signed statement of which Bot and person it is for."; - } - try { - const response = await fetch(TOOL_URL, { - method: "POST", - headers: { - "content-type": "application/json", - "x-openbot-agent-token": TOOL_TOKEN, - }, - /* - * The deployment's own statement, handed straight back. - * - * The Bot and the actor used to be sent from here, which meant this process asserted who it was - * acting for. It is not in a position to know, and anything holding the token could claim - * anything, so the deployment says it and this only carries the note. - */ - body: JSON.stringify({ name, args, run }), - }); - const body = (await response.json()) as { text?: string }; - return body.text ?? "The tool returned nothing."; - } catch (error) { - // Reported to the model as a result rather than thrown: the run continues and says what broke. - return `That tool could not be called: ${ - error instanceof Error ? error.message : "unknown error" - }`; - } -} - /** * The deployment's signed statement of what this run is. * @@ -330,97 +290,81 @@ function buildGraph(input: RunAgentInput) { const bound = tools.length > 0 ? model.bindTools(tools) : model; const ours = deploymentToolsOf(input); - return new StateGraph(MessagesAnnotation) - .addNode("answer", async (state) => ({ - messages: [await bound.invoke(state.messages)], - })) - .addNode("tools", async (state) => { - const last = state.messages.at(-1) as AIMessage; - const results = await Promise.all( + return ( + new StateGraph(MessagesAnnotation) + // The node config carries the run's signal (streamEvents propagates it), so a + // cancelled run stops the model invocation and any tool call in flight — not + // just the reading of the stream. + .addNode("answer", async (state, nodeConfig) => ({ + messages: [await bound.invoke(state.messages, nodeConfig)], + })) + .addNode("tools", async (state, nodeConfig) => { + const last = state.messages.at(-1) as AIMessage; + const results = await Promise.all( + /* + * Only this deployment's own tools. A component is drawn by the surface, and a decision is + * answered there by a person, so neither is executed here and neither gets a result invented + * here. The run ends instead, and the surface starts the next one carrying what it produced. + */ + (last.tool_calls ?? []) + .filter((call) => ours.has(call.name)) + .map(async (call) => { + const text = await callDeploymentTool( + { url: TOOL_URL, token: TOOL_TOKEN }, + run, + call.name, + (call.args ?? {}) as Record, + nodeConfig?.signal, + ); + return new ToolMessage({ + content: text, + tool_call_id: call.id ?? call.name, + name: call.name, + }); + }), + ); + return { messages: results }; + }) + .addEdge(START, "answer") + .addConditionalEdges("answer", (state) => { + const last = state.messages.at(-1) as AIMessage | undefined; + const calls = last?.tool_calls ?? []; + if (calls.length === 0) return END; /* - * Only this deployment's own tools. A component is drawn by the surface, and a decision is - * answered there by a person, so neither is executed here and neither gets a result invented - * here. The run ends instead, and the surface starts the next one carrying what it produced. + * A call the surface owns ends the run. + * + * This is how a tool that lives in the browser is supposed to work: the Bot asks for it, the + * run finishes, the surface draws it or puts the question to a person, and the surface begins + * the next run with the answer in hand. Running the loop through it here instead invents a + * result: the Bot apologises for a chart the person is looking at, and an approval card that + * has already been answered on its behalf sits waiting for a click that can never land. + * + * A turn that asks for both kinds at once ends too, and the model asks again for what it still + * has no answer to. That is the rarer case and the safe way round: the alternative runs a + * governed tool whose result nobody is waiting for. */ - (last.tool_calls ?? []) - .filter((call) => ours.has(call.name)) - .map(async (call) => { - const text = await callTool( - run, - call.name, - (call.args ?? {}) as Record, - ); - return new ToolMessage({ - content: text, - tool_call_id: call.id ?? call.name, - name: call.name, - }); - }), - ); - return { messages: results }; - }) - .addEdge(START, "answer") - .addConditionalEdges("answer", (state) => { - const last = state.messages.at(-1) as AIMessage | undefined; - const calls = last?.tool_calls ?? []; - if (calls.length === 0) return END; - /* - * A call the surface owns ends the run. - * - * This is how a tool that lives in the browser is supposed to work: the Bot asks for it, the - * run finishes, the surface draws it or puts the question to a person, and the surface begins - * the next run with the answer in hand. Running the loop through it here instead invents a - * result: the Bot apologises for a chart the person is looking at, and an approval card that - * has already been answered on its behalf sits waiting for a click that can never land. - * - * A turn that asks for both kinds at once ends too, and the model asks again for what it still - * has no answer to. That is the rarer case and the safe way round: the alternative runs a - * governed tool whose result nobody is waiting for. - */ - if (callsTheSurface(calls, ours)) return END; - return "tools"; - }) - .addEdge("tools", "answer") - .compile(); + if (callsTheSurface(calls, ours)) return END; + return "tools"; + }) + .addEdge("tools", "answer") + .compile() + ); } -async function runAgent(input: RunAgentInput): Promise { - const encoder = new EventEncoder(); - const stream = new ReadableStream({ - async start(controller) { - const utf8 = new TextEncoder(); - const send = (event: BaseEvent) => - controller.enqueue(utf8.encode(encoder.encodeSSE(event))); - - send({ - type: "RUN_STARTED", - threadId: input.threadId, - runId: input.runId, - } as BaseEvent); - - // The graph is built and its event stream opened inside `streamRun`, so a failure doing either - // is reported as RUN_ERROR through the same path as a failure mid-stream. - await streamRun( - async () => - buildGraph(input).streamEvents( - { messages: toLangChainMessages(input) }, - { version: "v2" }, - ), - input, - send, - ); - - controller.close(); - }, - }); - - return new Response(stream, { - headers: { - "content-type": encoder.getContentType(), - "cache-control": "no-cache", - connection: "keep-alive", - }, - }); +function runAgent(input: RunAgentInput, clientSignal?: AbortSignal): Response { + // The graph is built and its event stream opened inside `streamRun`, so a failure + // doing either is reported as RUN_ERROR through the same path as a failure + // mid-stream. The signal reaches the framework itself: an aborted run stops the + // model call, not just the reading of it. + return respondWithRun( + input, + async (signal) => + buildGraph(input).streamEvents( + { messages: toLangChainMessages(input) }, + { version: "v2", signal }, + ), + clientSignal, + ); } serve({ @@ -444,7 +388,7 @@ serve({ return Response.json({ error: "Unauthorized." }, { status: 401 }); } const input = (await request.json()) as RunAgentInput; - return runAgent(input); + return runAgent(input, request.signal); } return Response.json({ error: "Not found." }, { status: 404 }); diff --git a/agent-langgraph/src/respond.ts b/agent-langgraph/src/respond.ts new file mode 100644 index 000000000..1fddde9b4 --- /dev/null +++ b/agent-langgraph/src/respond.ts @@ -0,0 +1,77 @@ +import type { BaseEvent, RunAgentInput } from "@ag-ui/core"; +import { EventEncoder } from "@ag-ui/encoder"; +import { type RunStreamEvent, streamRun } from "./stream"; + +/** + * One run, answered as an AG-UI SSE response — with a way to make it stop. + * + * Its own module for the reason `stream.ts` is: `index.ts` calls `serve()` at module + * scope, so the response lifecycle — and above all its cancellation — has to live + * where a test can reach it without binding a port. + * + * Cancellation has two doors and both lead to the same abort. The caller's signal + * (the HTTP request's own) fires when the client disconnects; the stream's `cancel()` + * fires when the consumer lets go of the body. Either way the model invocation is + * aborted through the signal handed to `makeEvents`, because a consumer that hung up + * does not stop the model on its own — the tokens keep costing money and the process + * keeps holding capacity for a reply nobody will read. + */ +export function respondWithRun( + input: RunAgentInput, + makeEvents: (signal: AbortSignal) => Promise>, + clientSignal?: AbortSignal, +): Response { + const encoder = new EventEncoder(); + const halt = new AbortController(); + if (clientSignal?.aborted) halt.abort(clientSignal.reason); + clientSignal?.addEventListener( + "abort", + () => { + halt.abort(clientSignal.reason); + }, + { once: true }, + ); + + const stream = new ReadableStream({ + async start(controller) { + const utf8 = new TextEncoder(); + const send = (event: BaseEvent) => { + try { + controller.enqueue(utf8.encode(encoder.encodeSSE(event))); + } catch { + // The consumer is gone; there is nowhere to say anything. The abort below + // is what stops the work itself. + } + }; + + send({ + type: "RUN_STARTED", + threadId: input.threadId, + runId: input.runId, + } as BaseEvent); + + await streamRun(() => makeEvents(halt.signal), input, send, halt.signal); + + try { + controller.close(); + } catch { + // Already cancelled by the consumer. + } + }, + cancel(reason) { + halt.abort( + reason instanceof Error + ? reason + : new Error("run cancelled by its consumer"), + ); + }, + }); + + return new Response(stream, { + headers: { + "content-type": encoder.getContentType(), + "cache-control": "no-cache", + connection: "keep-alive", + }, + }); +} diff --git a/agent-langgraph/src/stream.ts b/agent-langgraph/src/stream.ts index dbc9eb7da..a174f9bd7 100644 --- a/agent-langgraph/src/stream.ts +++ b/agent-langgraph/src/stream.ts @@ -49,6 +49,10 @@ export async function streamRun( makeEvents: () => Promise>, input: Pick, send: (event: BaseEvent) => void, + /** Aborted when the consumer hung up or the run was cancelled. The framework's own + * stream gets the same signal and ends itself; this check is the belt for an + * iterable that ignores it, so a cancelled run never keeps reading regardless. */ + signal?: AbortSignal, ): Promise { /* * One message id per stretch of prose. @@ -92,6 +96,7 @@ export async function streamRun( >(); for await (const event of events) { + if (signal?.aborted) break; if (event.event === "on_chat_model_stream") { /* * Both content shapes, because the API decides which one arrives. diff --git a/agent-langgraph/src/tools.ts b/agent-langgraph/src/tools.ts new file mode 100644 index 000000000..122c2afa8 --- /dev/null +++ b/agent-langgraph/src/tools.ts @@ -0,0 +1,68 @@ +/** + * A governed tool call, made back through the deployment — abortable. + * + * Its own module for the reason `stream.ts` and `respond.ts` are: `index.ts` binds a + * port at import time, and the part of a tool call worth testing — that a cancelled + * run stops the request in flight rather than letting the governed action finish for + * nobody — needs driving without a server. + * + * The signal matters more here than anywhere else in this process. A model stream cut + * off mid-token wastes tokens; a governed action that keeps executing after its run + * was cancelled is work happening on somebody's computer with no one entitled to it. + * Aborting the fetch closes the socket, which is the deployment's cue to stop the + * action it was running on this run's behalf. + */ + +export interface DeploymentToolSettings { + url: string; + token: string; +} + +export async function callDeploymentTool( + settings: DeploymentToolSettings, + run: string, + name: string, + args: Record, + signal?: AbortSignal, + fetchImplementation: typeof fetch = fetch, +): Promise { + if (!settings.token) { + return "Refused. This Bot has no credential for calling tools back through its deployment."; + } + if (!run) { + /* + * No statement from the deployment about whose run this is, so there is nothing + * to act on behalf of. Reported as a result rather than thrown: the run + * continues and says what it could not do. + */ + return "Refused. This run carried no signed statement of which Bot and person it is for."; + } + try { + const response = await fetchImplementation(settings.url, { + method: "POST", + ...(signal ? { signal } : {}), + headers: { + "content-type": "application/json", + "x-openbot-agent-token": settings.token, + }, + /* + * The deployment's own statement, handed straight back. + * + * The Bot and the actor used to be sent from here, which meant this process + * asserted who it was acting for. It is not in a position to know, and + * anything holding the token could claim anything, so the deployment says it + * and this only carries the note. + */ + body: JSON.stringify({ name, args, run }), + }); + const body = (await response.json()) as { text?: string }; + return body.text ?? "The tool returned nothing."; + } catch (error) { + // Reported to the model as a result rather than thrown: the run continues and + // says what broke. A cancelled run's graph is being torn down anyway; the + // sentence is for the transcript, and the closed socket is for the deployment. + return `That tool could not be called: ${ + error instanceof Error ? error.message : "unknown error" + }`; + } +} diff --git a/agent-langgraph/tests/respond.test.ts b/agent-langgraph/tests/respond.test.ts new file mode 100644 index 000000000..94286dd48 --- /dev/null +++ b/agent-langgraph/tests/respond.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, test } from "bun:test"; +import type { RunAgentInput } from "@ag-ui/core"; +import { respondWithRun } from "../src/respond"; +import type { RunStreamEvent } from "../src/stream"; + +const input = { + threadId: "thread-1", + runId: "run-1", + messages: [], + tools: [], + context: [], + state: {}, + forwardedProps: {}, +} as unknown as RunAgentInput; + +/** A model stream that keeps talking until its signal says stop. */ +function endlessEvents(observed: { aborted: boolean; yielded: number }) { + return async ( + signal: AbortSignal, + ): Promise> => { + signal.addEventListener("abort", () => { + observed.aborted = true; + }); + return (async function* stream() { + while (!signal.aborted) { + observed.yielded += 1; + yield { + event: "on_chat_model_stream", + data: { chunk: { content: "word " } }, + } satisfies RunStreamEvent; + await new Promise((resolve) => setTimeout(resolve, 5)); + } + })(); + }; +} + +describe("a run that is cancelled", () => { + test("a consumer letting go of the body aborts the model work", async () => { + const observed = { aborted: false, yielded: 0 }; + const response = respondWithRun(input, endlessEvents(observed)); + expect(response.headers.get("content-type")).toContain("text/event-stream"); + + const reader = response.body?.getReader(); + expect(reader).toBeDefined(); + await reader?.read(); + await reader?.cancel(new Error("consumer hung up")); + + // The abort must reach the signal handed to the model stream — a consumer that + // hung up does not stop the model on its own. + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(observed.aborted).toBe(true); + const yieldedAtCancel = observed.yielded; + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(observed.yielded).toBe(yieldedAtCancel); + }); + + test("the request's own signal aborts the model work on disconnect", async () => { + const observed = { aborted: false, yielded: 0 }; + const client = new AbortController(); + const response = respondWithRun( + input, + endlessEvents(observed), + client.signal, + ); + + const reader = response.body?.getReader(); + await reader?.read(); + client.abort(new Error("client disconnected")); + + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(observed.aborted).toBe(true); + }); + + test("a signal already aborted before the run starts never yields at all", async () => { + const observed = { aborted: false, yielded: 0 }; + const client = new AbortController(); + client.abort(new Error("gone before it began")); + const response = respondWithRun( + input, + endlessEvents(observed), + client.signal, + ); + await response.text(); + expect(observed.yielded).toBe(0); + }); +}); + +describe("a run that completes", () => { + test("streams RUN_STARTED first and ends the stream", async () => { + const events: RunStreamEvent[] = [ + { event: "on_chat_model_stream", data: { chunk: { content: "Hello." } } }, + ]; + const response = respondWithRun(input, () => + Promise.resolve( + (async function* stream() { + for (const event of events) yield event; + })(), + ), + ); + const text = await response.text(); + expect(text.indexOf('"RUN_STARTED"')).toBeGreaterThanOrEqual(0); + expect(text.indexOf('"RUN_STARTED"')).toBeLessThan( + text.indexOf('"TEXT_MESSAGE_CONTENT"'), + ); + expect(text).toContain('"RUN_FINISHED"'); + }); +}); diff --git a/agent-langgraph/tests/tools.test.ts b/agent-langgraph/tests/tools.test.ts new file mode 100644 index 000000000..6f61c2c71 --- /dev/null +++ b/agent-langgraph/tests/tools.test.ts @@ -0,0 +1,89 @@ +import { describe, expect, test } from "bun:test"; +import { callDeploymentTool } from "../src/tools"; + +const settings = { + url: "http://localhost:3001/api/agent-tools/call", + token: "tool-token", +}; + +describe("a governed tool call", () => { + test("carries the run's signal into the request itself", async () => { + let seenSignal: AbortSignal | undefined; + const fakeFetch: typeof fetch = (_url, init) => { + seenSignal = init?.signal ?? undefined; + return Promise.resolve(Response.json({ text: "done" })); + }; + const controller = new AbortController(); + const text = await callDeploymentTool( + settings, + "signed-run", + "browse", + { url: "https://example.com" }, + controller.signal, + fakeFetch, + ); + expect(text).toBe("done"); + expect(seenSignal).toBe(controller.signal); + }); + + test("cancellation mid-request stops the call instead of waiting out the action", async () => { + // The case the model-stream tests cannot cover: the run is cancelled while the + // tool's HTTP request is in flight. The fetch must abort — a governed action + // continuing after its run was cancelled is work nobody is entitled to any more. + const hangingFetch: typeof fetch = (_url, init) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener("abort", () => { + reject( + init.signal?.reason instanceof Error + ? init.signal.reason + : new Error("aborted"), + ); + }); + // Never resolves on its own: only the abort ends it. + }); + const controller = new AbortController(); + const call = callDeploymentTool( + settings, + "signed-run", + "browse", + {}, + controller.signal, + hangingFetch, + ); + setTimeout(() => { + controller.abort(new Error("run cancelled")); + }, 20); + const started = Date.now(); + const text = await call; + // Promptly, and as a spoken result for the transcript rather than a throw. + expect(Date.now() - started).toBeLessThan(1_000); + expect(text).toContain("could not be called"); + expect(text).toContain("run cancelled"); + }); + + test("refusals still answer without touching the network", async () => { + const untouched: typeof fetch = () => { + throw new Error("must not fetch"); + }; + expect( + await callDeploymentTool( + { ...settings, token: "" }, + "run", + "browse", + {}, + undefined, + untouched, + ), + ).toContain("no credential"); + expect( + await callDeploymentTool( + settings, + "", + "browse", + {}, + undefined, + untouched, + ), + ).toContain("no signed statement"); + }); +}); diff --git a/bun.lock b/bun.lock index 415b9ce62..071cf2c60 100644 --- a/bun.lock +++ b/bun.lock @@ -63,6 +63,7 @@ "version": "0.0.0", "dependencies": { "@ag-ui/client": "0.0.57", + "@ag-ui/core": "0.0.57", "@better-auth/drizzle-adapter": "^1.7.1", "@better-auth/sso": "^1.7.1", "@copilotkit/runtime": "1.69.0", diff --git a/docs/bitmind-gateway.md b/docs/bitmind-gateway.md new file mode 100644 index 000000000..8e09c13d8 --- /dev/null +++ b/docs/bitmind-gateway.md @@ -0,0 +1,60 @@ +# BitMind gateway + +The surface BitMind's run plane talks to, run as its own process +(`server/src/bitmind/main.ts`) inside the execution enclave. It exists so the two +sides can meet on the real protocol — AG-UI at the pinned `@ag-ui/core@0.0.57` +(bit-mind ADR-0002) — before the full server learns to boot without the Intelligence +contract. + +## Surface + +| Route | Method | Auth | Purpose | +| --- | --- | --- | --- | +| `/health` | GET | none | Liveness for a supervisor. Reports nothing else. | +| `/bitmind/v1/attestation` | GET | Bearer `BITMIND_SERVICE_TOKEN` | What is behind the door, honestly. | +| `/bitmind/v1/run` | POST | Bearer `BITMIND_SERVICE_TOKEN` | One `RunAgentInput` in, the AG-UI SSE event stream out. | + +A run is validated against the pinned `RunAgentInputSchema` and relayed whole to the +configured downstream agent (`BITMIND_AGENT_URL`, by default `agent-langgraph` on +loopback), authenticated with that agent's managed-agent token. BitMind's identity +statement rides in `forwardedProps` — `workspace_id`, `agent_id`, `run_id`, +`message_id`, `fencing_token` — and is forwarded untouched. + +## What the gateway owns + +- **Service authentication**, timing-safe, on everything but `/health`. +- **Input validation** against the pinned protocol schemas; invalid bodies are + refused without being echoed. +- **Admission control**: at most `BITMIND_MAX_CONCURRENT_RUNS` relays (staging starts + at two, matching the enclave note); past the ceiling a run is refused with 429 and + `retry-after` rather than degrading the host. +- **Idempotency**: BitMind sends `idempotency-key: run_id:fencing_token`; while a + relay under that key is live, a second POST is refused with 409. BitMind's bounded + retry and event dedupe absorb the rest. +- **A run ceiling** (`BITMIND_RUN_TIMEOUT_MS`) so an abandoned stream cannot hold a + slot forever. + +## What it deliberately does not do yet + +- **No computers, no tools, no interrupts.** The relayed agent is prose-only; nothing + consequential can happen through this door. That is why the attestation reports + `isolated_computers: false` — BitMind's activation gate (bit-mind + `docs/operations/openbot-single-host-enclave.md`) requires `true` before its worker + may be enabled, and no isolation exists here to attest. The flag turns true when + enclave-managed agent computers actually stand behind the gateway, not before. +- **No policy or audit surface.** Those live in the server's governed gateway; they + join this path when tool execution does. +- **Interrupt-driven approvals** (`RUN_FINISHED` with an interrupt outcome, answered + via `RunAgentInput.resume`) are the contract BitMind's approval gate is built + against; this gateway relays them faithfully when the downstream emits them, but + the shipped `agent-langgraph` does not yet. + +## Running it + +``` +BITMIND_SERVICE_TOKEN=… BITMIND_AGENT_TOKEN=… bun run --filter server bitmind:start +``` + +It refuses to start without both tokens, binds loopback by default, and never holds +BitMind database credentials, OIDC secrets, or the Docker socket — per the enclave +boundary. diff --git a/server/package.json b/server/package.json index 4b6fa4c13..970307c8d 100644 --- a/server/package.json +++ b/server/package.json @@ -9,10 +9,13 @@ "db:generate": "bun --env-file=../.env drizzle-kit generate --config=drizzle.config.ts", "db:migrate": "bun --env-file=../.env drizzle-kit migrate --config=drizzle.config.ts", "dev": "bun --env-file=../.env --watch src/index.ts", - "typecheck": "tsc --noEmit" + "typecheck": "tsc --noEmit", + "bitmind:dev": "bun --env-file=../.env --watch src/bitmind/main.ts", + "bitmind:start": "bun src/bitmind/main.ts" }, "dependencies": { "@ag-ui/client": "0.0.57", + "@ag-ui/core": "0.0.57", "@better-auth/drizzle-adapter": "^1.7.1", "@better-auth/sso": "^1.7.1", "@copilotkit/runtime": "1.69.0", diff --git a/server/src/bitmind/config.ts b/server/src/bitmind/config.ts new file mode 100644 index 000000000..f5040558a --- /dev/null +++ b/server/src/bitmind/config.ts @@ -0,0 +1,120 @@ +/** + * Configuration for the BitMind gateway, read once at startup. + * + * Its own module and its own environment surface, deliberately apart from the server's + * `config.ts`: that file requires the Intelligence contract to resolve at all, and the + * point of this gateway is to run where Intelligence is not configured. The two grow + * together again when the server itself learns a standalone mode; until then this reads + * exactly what the gateway needs and nothing the enclave must not hold. + */ + +/** The AG-UI protocol version this deployment is pinned to, as ADR-0002 records it. + * A contract test asserts this matches the installed `@ag-ui/core`, so a dependency + * bump cannot silently move the protocol underneath either side. */ +export const AG_UI_PROTOCOL_VERSION = "0.0.57"; + +export interface BitmindGatewayConfig { + /** Bearer token BitMind authenticates with. Never logged, never echoed. */ + serviceToken: string; + /** The AG-UI agent endpoint runs are relayed to, loopback in the enclave. */ + agentUrl: string; + /** The managed-agent token the downstream agent requires. */ + agentToken: string; + /** Admission ceiling: new runs are refused before host pressure builds. */ + maxConcurrentRuns: number; + /** Whole-run ceiling on the relay, so an abandoned stream cannot hold a slot. */ + runTimeoutMs: number; +} + +/** Where the gateway listens. Loopback by default: the enclave boundary requires it. */ +export interface BitmindGatewayListen { + host: string; + port: number; +} + +function integer( + environment: NodeJS.ProcessEnv, + name: string, + fallback: number, + minimum: number, + maximum: number, +): number { + const raw = environment[name]?.trim(); + if (!raw) return fallback; + // The whole string or nothing: parseInt's numeric-prefix tolerance turns + // "2workers" into 2 and "1000ms" into 1000, which is a ceiling somebody believes + // they set and did not. A limit is a safety number; a malformed one must fail in + // front of whoever deployed it. + if (!/^\d+$/.test(raw)) { + throw new Error( + `${name}=${raw} must be a whole number between ${minimum} and ${maximum}.`, + ); + } + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < minimum || value > maximum) { + throw new Error( + `${name}=${raw} must be a whole number between ${minimum} and ${maximum}.`, + ); + } + return value; +} + +export function bitmindGatewayConfig( + environment: NodeJS.ProcessEnv, +): BitmindGatewayConfig { + const serviceToken = environment.BITMIND_SERVICE_TOKEN?.trim(); + if (!serviceToken) { + throw new Error( + "BITMIND_SERVICE_TOKEN is not set. The gateway authenticates every BitMind call and will not start without it. Generate one: openssl rand -hex 32", + ); + } + const agentToken = environment.BITMIND_AGENT_TOKEN?.trim(); + if (!agentToken) { + throw new Error( + "BITMIND_AGENT_TOKEN is not set. The downstream agent requires its managed-agent token; without it every relayed run would be refused.", + ); + } + const agentUrl = + environment.BITMIND_AGENT_URL?.trim() || "http://localhost:4201/ag-ui"; + const parsed = new URL(agentUrl); + if (parsed.username || parsed.password) { + throw new Error("BITMIND_AGENT_URL must not carry credentials."); + } + if (parsed.protocol !== "http:" && parsed.protocol !== "https:") { + throw new Error( + `BITMIND_AGENT_URL must be http or https, not ${parsed.protocol}`, + ); + } + return { + serviceToken, + agentToken, + agentUrl, + // The enclave note starts staging at two concurrent agent computers; the same + // ceiling applies to runs until computers exist at all. + maxConcurrentRuns: integer( + environment, + "BITMIND_MAX_CONCURRENT_RUNS", + 2, + 1, + 64, + ), + runTimeoutMs: integer( + environment, + "BITMIND_RUN_TIMEOUT_MS", + 900_000, + 1_000, + 3_600_000, + ), + }; +} + +export function bitmindGatewayListen( + environment: NodeJS.ProcessEnv, +): BitmindGatewayListen { + return { + // Loopback unless somebody deliberately says otherwise. The enclave exposes this + // gateway to BitMind over the private path only; a public bind is a mistake. + host: environment.BITMIND_GATEWAY_HOST?.trim() || "127.0.0.1", + port: integer(environment, "BITMIND_GATEWAY_PORT", 4310, 1, 65_535), + }; +} diff --git a/server/src/bitmind/gateway.ts b/server/src/bitmind/gateway.ts new file mode 100644 index 000000000..5a6bf6d81 --- /dev/null +++ b/server/src/bitmind/gateway.ts @@ -0,0 +1,340 @@ +import { RunAgentInputSchema } from "@ag-ui/core"; +import { z } from "zod"; +import { matchesToken } from "../../../shared/agent-authorisation"; +import { AG_UI_PROTOCOL_VERSION, type BitmindGatewayConfig } from "./config"; + +/** + * BitMind's identity statement, required in full. + * + * The generic AG-UI envelope says nothing about who a run belongs to; this gateway + * refuses an attempt whose identity is missing or self-contradictory rather than + * relaying it and letting the far side guess. `fencing_token` is what makes a retry + * of the same claimed attempt recognisable, so it must be a whole number, and the + * `idempotency-key` header — when sent — must agree with it: a caller-selectable key + * detached from the run would defeat the double-start protection it exists for. + */ +const BitmindForwardedPropsSchema = z.object({ + workspace_id: z.string().min(1), + agent_id: z.string().min(1).nullable(), + run_id: z.string().min(1), + message_id: z.string().min(1), + fencing_token: z.number().int().nonnegative(), +}); + +/** + * The doorway BitMind talks through: one POST per run, answered with the AG-UI event + * stream, plus the attestation its activation gate reads before it will enable a + * worker at all. + * + * This is deliberately a relay and not a runtime. The downstream agent owns the model + * conversation; this process owns what an enclave boundary needs owned on its edge — + * service authentication, input validation against the pinned protocol schemas, + * admission control, and an honest statement of what is and is not behind the door. + * Nothing consequential can happen through it yet: the relayed agent is prose-only + * (BitMind sends no tools and no computer exists here), which is exactly why + * `isolated_computers` attests false and BitMind's worker stays disabled. + * + * Kept as a handler factory rather than a bound server, the way `agent-langgraph` + * splits its logic from `serve()`, so tests drive it with plain Requests. + */ + +/** Attestation for the enclave activation gate. Every field is a statement BitMind + * may act on, so nothing here is aspirational: capabilities appear when they exist. */ +export interface BitmindAttestation { + service: "openbot-bitmind-gateway"; + protocol: { ag_ui: string }; + isolated_computers: boolean; + execution: { backend: "relay"; tools: boolean; interrupts: boolean }; + limits: { max_concurrent_runs: number; run_timeout_ms: number }; + active_runs: number; +} + +const JSON_HEADERS = { "content-type": "application/json" }; + +function unauthorized(): Response { + return Response.json({ error: "Unauthorized." }, { status: 401 }); +} + +function bearerToken(request: Request): string { + const header = request.headers.get("authorization")?.trim() ?? ""; + return header.toLowerCase().startsWith("bearer ") + ? header.slice("bearer ".length).trim() + : ""; +} + +export function createBitmindGateway( + config: BitmindGatewayConfig, + fetchImplementation: typeof fetch = fetch, +) { + /** + * Runs currently relayed, by idempotency key. + * + * The key is BitMind's `idempotency-key` header (`run_id:fencing_token`), so a retry + * of the same claimed attempt cannot start a second stream while the first is live — + * it is refused with 409 and BitMind's bounded retry tries again once the first + * relay has ended. Falls back to the run id so an unkeyed caller still cannot + * double-start a run. + */ + const active = new Map(); + + function attestation(): BitmindAttestation { + return { + service: "openbot-bitmind-gateway", + protocol: { ag_ui: AG_UI_PROTOCOL_VERSION }, + // No agent computers exist behind this gateway yet, so no isolation exists to + // attest. BitMind's activation gate requires true; false keeps its worker off, + // which is the correct state until the enclave provides real computers. + isolated_computers: false, + execution: { backend: "relay", tools: false, interrupts: false }, + limits: { + max_concurrent_runs: config.maxConcurrentRuns, + run_timeout_ms: config.runTimeoutMs, + }, + active_runs: active.size, + }; + } + + async function relayRun(request: Request): Promise { + let body: unknown; + try { + body = await request.json(); + } catch { + return Response.json({ error: "Body must be JSON." }, { status: 400 }); + } + const input = RunAgentInputSchema.safeParse(body); + if (!input.success) { + // Generic on purpose: echoing the parse failure would reflect caller input. + return Response.json( + { error: "Body is not a valid RunAgentInput." }, + { status: 400 }, + ); + } + const identity = BitmindForwardedPropsSchema.safeParse( + input.data.forwardedProps, + ); + if (!identity.success) { + return Response.json( + { error: "forwardedProps must carry the BitMind identity statement." }, + { status: 400 }, + ); + } + if (identity.data.run_id !== input.data.runId) { + return Response.json( + { error: "forwardedProps.run_id must match runId." }, + { status: 400 }, + ); + } + // While the attestation says tools: false and interrupts: false, the gateway + // holds that boundary itself rather than trusting the far side to. A run + // carrying tools or resume answers is asking for capabilities nobody attested. + if (input.data.tools.length > 0) { + return Response.json( + { error: "This gateway attests tools: false and relays none." }, + { status: 400 }, + ); + } + if ((input.data.resume?.length ?? 0) > 0) { + return Response.json( + { + error: + "This gateway attests interrupts: false and accepts no resume.", + }, + { status: 400 }, + ); + } + + const expectedKey = `${identity.data.run_id}:${String(identity.data.fencing_token)}`; + const offeredKey = request.headers.get("idempotency-key")?.trim(); + if (offeredKey && offeredKey !== expectedKey) { + return Response.json( + { error: "idempotency-key must be run_id:fencing_token." }, + { status: 400 }, + ); + } + const key = expectedKey; + if (active.has(key)) { + return Response.json( + { error: "A run with this idempotency key is already streaming." }, + { status: 409 }, + ); + } + if (active.size >= config.maxConcurrentRuns) { + return Response.json( + { error: "The gateway is at its concurrency ceiling." }, + { status: 429, headers: { "retry-after": "5" } }, + ); + } + + // The relay ends when BitMind hangs up, when the ceiling passes, or when the + // stream finishes — whichever comes first releases the slot. + const controller = new AbortController(); + const timeout = setTimeout(() => { + controller.abort(new Error("run relay timed out")); + }, config.runTimeoutMs); + /** + * Give the slot back — but only if this relay still holds it. + * + * A cancelled relay ends through several paths at once (the abort listener, the + * stream's `cancel()`, and the pending `reader.read()` resolving), and a same-key + * retry can legitimately be admitted between two of them. An unconditional delete + * would then evict the *new* relay's entry while its stream is live, taking both + * the concurrency accounting and the 409 double-start refusal with it. Ownership + * is the guard: whoever is in the map is the only one who can leave it, which also + * makes every one of those endings safe to run more than once. + */ + const release = () => { + clearTimeout(timeout); + if (active.get(key) === controller) { + active.delete(key); + } + }; + request.signal.addEventListener("abort", () => { + controller.abort(request.signal.reason as Error | undefined); + }); + // Abort events are not replayed: a caller that hung up while the body was still + // being read or validated has already fired its signal, and the listener above + // heard nothing. Checked AFTER registering, so a signal firing between the check + // and the listener cannot slip through either way — one of the two catches it. + if (request.signal.aborted) { + clearTimeout(timeout); + return Response.json( + { error: "The caller aborted before the run was relayed." }, + { status: 400 }, + ); + } + active.set(key, controller); + + let downstream: Response; + try { + downstream = await fetchImplementation(config.agentUrl, { + method: "POST", + signal: controller.signal, + redirect: "error", + headers: { + "content-type": "application/json", + accept: "text/event-stream", + "x-openbot-agent-token": config.agentToken, + }, + // The validated input, forwarded whole. `forwardedProps` carries BitMind's + // identity statement (workspace, agent, run, fencing token) untouched. + body: JSON.stringify(input.data), + }); + } catch { + release(); + return Response.json( + { error: "The execution backend is unreachable." }, + { status: 502, headers: JSON_HEADERS }, + ); + } + if (!downstream.ok || !downstream.body) { + // A refusal can arrive as headers over a body that keeps streaming. The slot + // must not come back while that request is still live, so the fetch is + // aborted and the body ended before admission is released. + controller.abort(new Error("backend refused the run")); + await downstream.body?.cancel().catch(() => undefined); + release(); + return Response.json( + { error: "The execution backend refused the run." }, + { status: 502 }, + ); + } + // A 2xx with a body is not yet an agent: a proxy login page or a JSON error + // answers exactly that way. Only the AG-UI stream content type may be reserved + // and reported to BitMind as a run in progress. + const contentType = downstream.headers.get("content-type") ?? ""; + // The media type itself, exactly: startsWith would admit text/event-streaming + // and text/event-stream+json, neither of which is the AG-UI wire. Parameters + // after the first ";" (charset) are legitimate and preserved when forwarding. + const mediaType = (contentType.split(";", 1)[0] ?? "").trim().toLowerCase(); + if (mediaType !== "text/event-stream") { + controller.abort( + new Error("backend did not answer with an event stream"), + ); + await downstream.body.cancel().catch(() => undefined); + release(); + return Response.json( + { error: "The execution backend did not answer with an event stream." }, + { status: 502 }, + ); + } + + // Pumped by hand rather than piped: the slot must come back on EVERY ending — + // clean close, source error, consumer cancellation, timeout — and a transform's + // flush() only runs for the first of those. The reader loop's finally is the one + // place all four paths pass through. + const reader = downstream.body.getReader(); + const relayed = new ReadableStream({ + async pull(streamController) { + let result: Awaited>; + try { + result = await reader.read(); + } catch (error) { + release(); + streamController.error(error); + return; + } + if (result.done) { + release(); + streamController.close(); + return; + } + streamController.enqueue(result.value); + }, + async cancel(reason) { + // BitMind hung up. The far side is told, not just abandoned: the model must + // stop working, and the slot comes back only alongside that abort. + controller.abort( + reason instanceof Error ? reason : new Error("relay cancelled"), + ); + await reader.cancel(reason).catch(() => undefined); + release(); + }, + }); + controller.signal.addEventListener( + "abort", + () => { + void reader.cancel(controller.signal.reason).catch(() => undefined); + release(); + }, + { once: true }, + ); + + return new Response(relayed, { + headers: { + "content-type": contentType, + "cache-control": "no-cache", + }, + }); + } + + return { + async fetch(request: Request): Promise { + const url = new URL(request.url); + + // Liveness for a supervisor. No auth and no information beyond being up. + if (url.pathname === "/health") { + return Response.json({ status: "ok" }); + } + + // Everything else is BitMind's surface and authenticates first. + if (!matchesToken(config.serviceToken, bearerToken(request))) { + return unauthorized(); + } + + if ( + url.pathname === "/bitmind/v1/attestation" && + request.method === "GET" + ) { + return Response.json(attestation()); + } + if (url.pathname === "/bitmind/v1/run" && request.method === "POST") { + return relayRun(request); + } + return Response.json({ error: "Not found." }, { status: 404 }); + }, + /** Exposed for tests: how many relays are live right now. */ + activeRuns(): number { + return active.size; + }, + }; +} diff --git a/server/src/bitmind/main.ts b/server/src/bitmind/main.ts new file mode 100644 index 000000000..8e00c7496 --- /dev/null +++ b/server/src/bitmind/main.ts @@ -0,0 +1,31 @@ +import { serve } from "bun"; +import { bitmindGatewayConfig, bitmindGatewayListen } from "./config"; +import { createBitmindGateway } from "./gateway"; + +/** + * The BitMind gateway as its own process. + * + * A separate entry rather than a mount inside `src/index.ts`, because the server + * refuses to start without the Intelligence contract and the enclave this runs in has + * none. When the server grows a standalone mode, `createBitmindGateway` mounts there + * and this file retires; until then the enclave supervises this process directly. + * + * Configuration is validated before the port binds: a gateway that cannot + * authenticate its caller or reach its agent must fail in front of whoever deployed + * it, not in front of the first run. + */ +const config = bitmindGatewayConfig(process.env); +const listen = bitmindGatewayListen(process.env); +const gateway = createBitmindGateway(config); + +serve({ + hostname: listen.host, + port: listen.port, + // Idle SSE relays are kept open well past Bun's default while a model thinks. + idleTimeout: 120, + fetch: (request) => gateway.fetch(request), +}); + +console.info( + `bitmind-gateway listening on http://${listen.host}:${String(listen.port)}/bitmind/v1 (agent: ${config.agentUrl})`, +); diff --git a/server/tests/bitmind-gateway.test.ts b/server/tests/bitmind-gateway.test.ts new file mode 100644 index 000000000..554d324dc --- /dev/null +++ b/server/tests/bitmind-gateway.test.ts @@ -0,0 +1,575 @@ +import { describe, expect, test } from "bun:test"; +import { + AG_UI_PROTOCOL_VERSION, + bitmindGatewayConfig, +} from "../src/bitmind/config"; +import { createBitmindGateway } from "../src/bitmind/gateway"; +import type { BitmindAttestation } from "../src/bitmind/gateway"; + +const SERVICE_TOKEN = "service-token-for-tests-0000000000000000"; +const AGENT_TOKEN = "managed-agent-token-for-tests-00000000"; + +function config( + overrides: Partial[0]> = {}, +) { + return { + serviceToken: SERVICE_TOKEN, + agentUrl: "http://localhost:4201/ag-ui", + agentToken: AGENT_TOKEN, + maxConcurrentRuns: 2, + runTimeoutMs: 30_000, + ...overrides, + }; +} + +function runInput(runId = "run-1") { + return { + threadId: "conversation-1", + runId, + messages: [{ id: "m1", role: "user", content: "Find me sources" }], + tools: [], + context: [], + state: {}, + forwardedProps: { + workspace_id: "workspace-1", + agent_id: "agent-1", + run_id: runId, + message_id: "message-1", + fencing_token: 1, + }, + }; +} + +function runRequest(body: unknown, headers: Record = {}) { + return new Request("http://gateway/bitmind/v1/run", { + method: "POST", + headers: { + authorization: `Bearer ${SERVICE_TOKEN}`, + "content-type": "application/json", + ...headers, + }, + body: JSON.stringify(body), + }); +} + +function sseEvents(events: object[]): string { + return events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join(""); +} + +/** A downstream agent that answers with a canned stream and remembers the request. */ +function fakeAgent(events: object[]) { + const seen: { url?: string; init?: RequestInit } = {}; + const agentFetch: typeof fetch = (url, init) => { + seen.url = String(url); + seen.init = init; + return Promise.resolve( + new Response(sseEvents(events), { + status: 200, + headers: { "content-type": "text/event-stream" }, + }), + ); + }; + return { seen, agentFetch }; +} + +describe("authentication", () => { + test("everything but /health requires the service token", async () => { + const gateway = createBitmindGateway(config()); + const health = await gateway.fetch(new Request("http://gateway/health")); + expect(health.status).toBe(200); + + for (const [path, method] of [ + ["/bitmind/v1/attestation", "GET"], + ["/bitmind/v1/run", "POST"], + ] as const) { + const bare = await gateway.fetch( + new Request(`http://gateway${path}`, { method }), + ); + expect(bare.status).toBe(401); + const wrong = await gateway.fetch( + new Request(`http://gateway${path}`, { + method, + headers: { authorization: "Bearer not-the-token" }, + }), + ); + expect(wrong.status).toBe(401); + } + }); +}); + +describe("attestation", () => { + test("states the pin, the honest capability set, and the ceilings", async () => { + const gateway = createBitmindGateway(config()); + const response = await gateway.fetch( + new Request("http://gateway/bitmind/v1/attestation", { + headers: { authorization: `Bearer ${SERVICE_TOKEN}` }, + }), + ); + expect(response.status).toBe(200); + const body = (await response.json()) as BitmindAttestation; + expect(body.protocol.ag_ui).toBe(AG_UI_PROTOCOL_VERSION); + // No enclave computers exist behind this gateway yet: attesting true here would + // switch BitMind's worker on against isolation that does not exist. + expect(body.isolated_computers).toBe(false); + expect(body.execution).toEqual({ + backend: "relay", + tools: false, + interrupts: false, + }); + expect(body.limits.max_concurrent_runs).toBe(2); + expect(body.active_runs).toBe(0); + }); + + test("the declared protocol version is the installed one", async () => { + // The drift fence: a dependency bump that moves @ag-ui/core must be a deliberate + // pin move, not a lockfile accident this attestation then lies about. + const manifest = (await import("@ag-ui/core/package.json")) as { + version: string; + }; + expect(manifest.version).toBe(AG_UI_PROTOCOL_VERSION); + }); +}); + +describe("run relay", () => { + test("a valid run is forwarded whole with the managed-agent token", async () => { + const { seen, agentFetch } = fakeAgent([ + { type: "RUN_STARTED", threadId: "conversation-1", runId: "run-1" }, + { type: "RUN_FINISHED", threadId: "conversation-1", runId: "run-1" }, + ]); + const gateway = createBitmindGateway(config(), agentFetch); + const response = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toBe("text/event-stream"); + const text = await response.text(); + expect(text).toContain('"RUN_STARTED"'); + expect(text).toContain('"RUN_FINISHED"'); + + expect(seen.url).toBe("http://localhost:4201/ag-ui"); + const headers = new Headers(seen.init?.headers); + expect(headers.get("x-openbot-agent-token")).toBe(AGENT_TOKEN); + // The service token authenticates BitMind to this gateway and goes no further. + expect(headers.get("authorization")).toBeNull(); + const forwarded = JSON.parse(String(seen.init?.body)) as { + forwardedProps: Record; + }; + expect(forwarded.forwardedProps).toEqual(runInput().forwardedProps); + // The stream ended, so the slot is free again. + expect(gateway.activeRuns()).toBe(0); + }); + + test("a body that is not a RunAgentInput is refused without echoing it", async () => { + const gateway = createBitmindGateway(config(), () => { + throw new Error("must not reach the agent"); + }); + const invalid = await gateway.fetch( + runRequest({ runId: "run-1", messages: "" }), + ); + expect(invalid.status).toBe(400); + expect(await invalid.text()).not.toContain("script"); + + const notJson = await gateway.fetch( + new Request("http://gateway/bitmind/v1/run", { + method: "POST", + headers: { authorization: `Bearer ${SERVICE_TOKEN}` }, + body: "not json", + }), + ); + expect(notJson.status).toBe(400); + }); + + test("the same idempotency key cannot stream twice at once", async () => { + // A downstream that never finishes, so the first relay stays live. + let releaseFirst = () => {}; + const hanging: typeof fetch = () => + Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: {}\n\n")); + releaseFirst = () => { + controller.close(); + }; + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ), + ); + const gateway = createBitmindGateway(config(), hanging); + const first = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(first.status).toBe(200); + expect(gateway.activeRuns()).toBe(1); + + const replay = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(replay.status).toBe(409); + + releaseFirst(); + await first.text(); + expect(gateway.activeRuns()).toBe(0); + }); + + test("runs past the ceiling are refused with retry-after", async () => { + const holds: (() => void)[] = []; + const hanging: typeof fetch = () => + Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + holds.push(() => { + controller.close(); + }); + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ), + ); + const gateway = createBitmindGateway( + config({ maxConcurrentRuns: 2 }), + hanging, + ); + const responses = [ + await gateway.fetch(runRequest(runInput("run-1"))), + await gateway.fetch(runRequest(runInput("run-2"))), + ]; + expect(responses.map((r) => r.status)).toEqual([200, 200]); + + const refused = await gateway.fetch(runRequest(runInput("run-3"))); + expect(refused.status).toBe(429); + expect(refused.headers.get("retry-after")).toBe("5"); + + for (const release of holds) release(); + await Promise.all(responses.map((r) => r.text())); + expect(gateway.activeRuns()).toBe(0); + }); + + test("an unreachable or refusing backend is a 502, never a hung slot", async () => { + const unreachable = createBitmindGateway(config(), () => + Promise.reject(new Error("connect ECONNREFUSED")), + ); + const down = await unreachable.fetch(runRequest(runInput())); + expect(down.status).toBe(502); + expect(unreachable.activeRuns()).toBe(0); + + const refusing = createBitmindGateway(config(), () => + Promise.resolve( + Response.json({ error: "Unauthorized." }, { status: 401 }), + ), + ); + const refused = await refusing.fetch(runRequest(runInput())); + expect(refused.status).toBe(502); + expect(refusing.activeRuns()).toBe(0); + }); +}); + +describe("configuration", () => { + test("refuses to start without its tokens, and never invents them", () => { + expect(() => bitmindGatewayConfig({})).toThrow(/BITMIND_SERVICE_TOKEN/); + expect(() => + bitmindGatewayConfig({ BITMIND_SERVICE_TOKEN: "token" }), + ).toThrow(/BITMIND_AGENT_TOKEN/); + }); + + test("refuses an agent URL that carries credentials or an odd scheme", () => { + const base = { + BITMIND_SERVICE_TOKEN: "token", + BITMIND_AGENT_TOKEN: "token", + }; + expect(() => + bitmindGatewayConfig({ + ...base, + BITMIND_AGENT_URL: "http://user:pw@host/ag-ui", + }), + ).toThrow(/credentials/); + expect(() => + bitmindGatewayConfig({ + ...base, + BITMIND_AGENT_URL: "file:///etc/passwd", + }), + ).toThrow(/http or https/); + }); +}); + +describe("identity enforcement", () => { + test("an attempt without its full identity statement is refused", async () => { + const gateway = createBitmindGateway(config(), () => { + throw new Error("must not reach the agent"); + }); + const input = runInput() as { forwardedProps: Record }; + delete input.forwardedProps.message_id; + const response = await gateway.fetch(runRequest(input)); + expect(response.status).toBe(400); + }); + + test("a run whose envelope and identity disagree is refused", async () => { + const gateway = createBitmindGateway(config(), () => { + throw new Error("must not reach the agent"); + }); + const input = runInput("run-1") as { forwardedProps: { run_id: string } }; + input.forwardedProps.run_id = "run-somebody-else"; + const response = await gateway.fetch(runRequest(input)); + expect(response.status).toBe(400); + }); + + test("the idempotency key is bound to the attempt, never caller-selectable", async () => { + const gateway = createBitmindGateway(config(), () => { + throw new Error("must not reach the agent"); + }); + const response = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "whatever-i-like" }), + ); + expect(response.status).toBe(400); + }); +}); + +describe("capability enforcement", () => { + test("tools are refused while the attestation says tools: false", async () => { + let reached = false; + const gateway = createBitmindGateway(config(), () => { + reached = true; + throw new Error("must not reach the agent"); + }); + const input = runInput() as { tools: unknown[] }; + input.tools = [ + { name: "browser_navigate", description: "go", parameters: {} }, + ]; + const response = await gateway.fetch(runRequest(input)); + expect(response.status).toBe(400); + expect(reached).toBe(false); + }); + + test("resume answers are refused while the attestation says interrupts: false", async () => { + let reached = false; + const gateway = createBitmindGateway(config(), () => { + reached = true; + throw new Error("must not reach the agent"); + }); + const input = runInput() as { resume?: unknown[] }; + input.resume = [{ interruptId: "int-1", status: "resolved" }]; + const response = await gateway.fetch(runRequest(input)); + expect(response.status).toBe(400); + expect(reached).toBe(false); + }); +}); + +describe("stream lifecycle", () => { + test("a backend that answers JSON is a 502, not a relayed run", async () => { + const gateway = createBitmindGateway(config(), () => + Promise.resolve( + Response.json({ error: "please sign in" }, { status: 200 }), + ), + ); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(502); + expect(gateway.activeRuns()).toBe(0); + }); + + test("a source error mid-stream releases the slot", async () => { + const failing: typeof fetch = () => + Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: {}\n\n")); + controller.error(new Error("backend fell over")); + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ), + ); + const gateway = createBitmindGateway(config(), failing); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(200); + await expect(response.text()).rejects.toThrow(); + expect(gateway.activeRuns()).toBe(0); + }); + + test("a consumer that hangs up releases the slot and stops the backend", async () => { + let downstreamAborted = false; + const hanging: typeof fetch = (_url, init) => { + init?.signal?.addEventListener("abort", () => { + downstreamAborted = true; + }); + return Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: {}\n\n")); + // Never closes: the consumer is the one who ends this. + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ), + ); + }; + const gateway = createBitmindGateway(config(), hanging); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(200); + expect(gateway.activeRuns()).toBe(1); + await response.body?.cancel(new Error("BitMind hung up")); + expect(gateway.activeRuns()).toBe(0); + expect(downstreamAborted).toBe(true); + }); +}); + +describe("strict limits", () => { + test.each(["2workers", "2.5", "1000ms", "-1", "1e3"])( + "a malformed ceiling like %s is refused, never coerced", + (value) => { + expect(() => + bitmindGatewayConfig({ + BITMIND_SERVICE_TOKEN: "token", + BITMIND_AGENT_TOKEN: "token", + BITMIND_MAX_CONCURRENT_RUNS: value, + }), + ).toThrow(/whole number/); + }, + ); +}); + +describe("follow-up findings", () => { + test("a refusal over a still-streaming body ends the request before admitting more", async () => { + let downstreamAborted = false; + let bodyCancelled = false; + const refusing: typeof fetch = (_url, init) => { + init?.signal?.addEventListener("abort", () => { + downstreamAborted = true; + }); + return Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("still talking\n")); + // Never closes on its own. + }, + cancel() { + bodyCancelled = true; + }, + }), + { status: 503, headers: { "content-type": "text/event-stream" } }, + ), + ); + }; + const gateway = createBitmindGateway(config(), refusing); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(502); + expect(gateway.activeRuns()).toBe(0); + expect(downstreamAborted).toBe(true); + expect(bodyCancelled).toBe(true); + }); + + test.each(["text/event-streaming", "text/event-stream+json"])( + "a media type like %s is not the AG-UI wire", + async (contentType) => { + const gateway = createBitmindGateway(config(), () => + Promise.resolve( + new Response("data: {}\n\n", { + status: 200, + headers: { "content-type": contentType }, + }), + ), + ); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(502); + expect(gateway.activeRuns()).toBe(0); + }, + ); + + test("parameters on the real media type survive the check and the forward", async () => { + const gateway = createBitmindGateway(config(), () => + Promise.resolve( + new Response("data: {}\n\n", { + status: 200, + headers: { "content-type": "text/event-stream; charset=utf-8" }, + }), + ), + ); + const response = await gateway.fetch(runRequest(runInput())); + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toBe( + "text/event-stream; charset=utf-8", + ); + await response.text(); + expect(gateway.activeRuns()).toBe(0); + }); + + test("a caller that already hung up never reserves a slot or reaches the agent", async () => { + let reached = false; + const gateway = createBitmindGateway(config(), () => { + reached = true; + throw new Error("must not reach the agent"); + }); + // Abort events are not replayed: this signal fired before the gateway could + // listen, which is exactly the case a listener alone misses. + const controller = new AbortController(); + const request = new Request("http://gateway/bitmind/v1/run", { + method: "POST", + headers: { + authorization: `Bearer ${SERVICE_TOKEN}`, + "content-type": "application/json", + }, + body: JSON.stringify(runInput()), + signal: controller.signal, + }); + controller.abort(new Error("caller gone")); + const response = await gateway.fetch(request); + expect(response.status).toBe(400); + expect(reached).toBe(false); + expect(gateway.activeRuns()).toBe(0); + }); +}); + +describe("cancellation and the slot's owner", () => { + test("a cancelled relay does not release the retry that replaced it", async () => { + let calls = 0; + const hanging: typeof fetch = () => { + calls += 1; + return Promise.resolve( + new Response( + new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("data: {}\n\n")); + // Never closes: only the consumer or the gateway ends this. + }, + }), + { status: 200, headers: { "content-type": "text/event-stream" } }, + ), + ); + }; + const gateway = createBitmindGateway(config(), hanging); + const first = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(first.status).toBe(200); + expect(gateway.activeRuns()).toBe(1); + + // BitMind hangs up and immediately retries the same attempt. The cancelled + // relay ends through several paths (abort listener, stream cancel, the pending + // read resolving); the retry is admitted in the middle of them, and none of + // those late endings may take the retry's slot with them. + const cancelling = first.body?.cancel(new Error("BitMind hung up")); + const retry = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(retry.status).toBe(200); + expect(gateway.activeRuns()).toBe(1); + await cancelling; + expect(gateway.activeRuns()).toBe(1); + + // The retry is streaming, so the idempotency refusal must still hold and the + // backend must not be asked a third time. + const third = await gateway.fetch( + runRequest(runInput(), { "idempotency-key": "run-1:1" }), + ); + expect(third.status).toBe(409); + expect(gateway.activeRuns()).toBe(1); + expect(calls).toBe(2); + + await retry.body?.cancel(new Error("done")); + expect(gateway.activeRuns()).toBe(0); + }); +});