From ae41d89dd533d2c9e76e2a62eae0748c8833b4e8 Mon Sep 17 00:00:00 2001
From: Bit Cloud <dev@warsuit.io>
Date: Wed, 2 Sep 2026 16:15:55 +0000
Subject: [PATCH 1/4] feat: add the BitMind gateway - service auth,
 attestation, and a governed relay

The doorway BitMind's run plane talks through, as its own process
(server/src/bitmind/main.ts) because the server itself cannot yet boot
without the Intelligence contract and the enclave it runs in has none.

- POST /bitmind/v1/run: Bearer service-token auth (timing-safe), body
  validated against the pinned @ag-ui/core@0.0.57 RunAgentInputSchema,
  then relayed whole to the configured downstream AG-UI agent
  (agent-langgraph by default) with its managed-agent token. BitMind's
  identity rides in forwardedProps untouched.
- GET /bitmind/v1/attestation: what stands behind the door, honestly -
  isolated_computers: false until enclave computers exist, which keeps
  BitMind's activation gate closed and its worker off.
- Admission control (429 past the concurrency ceiling, staging default 2
  per the enclave note), idempotency-key conflict detection (409 while
  the same claimed attempt is already streaming), a whole-run relay
  ceiling, redirect refusal, and loopback bind by default.
- @ag-ui/core pinned exactly at 0.0.57 in server deps; a test asserts
  the attested protocol version equals the installed package so the pin
  cannot drift silently.
---
 .env.example                         |  19 ++
 bun.lock                             |   1 +
 docs/bitmind-gateway.md              |  60 ++++++
 server/package.json                  |   5 +-
 server/src/bitmind/config.ts         | 111 ++++++++++
 server/src/bitmind/gateway.ts        | 202 ++++++++++++++++++
 server/src/bitmind/main.ts           |  31 +++
 server/tests/bitmind-gateway.test.ts | 300 +++++++++++++++++++++++++++
 8 files changed, 728 insertions(+), 1 deletion(-)
 create mode 100644 docs/bitmind-gateway.md
 create mode 100644 server/src/bitmind/config.ts
 create mode 100644 server/src/bitmind/gateway.ts
 create mode 100644 server/src/bitmind/main.ts
 create mode 100644 server/tests/bitmind-gateway.test.ts

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/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..6824a3f4a
--- /dev/null
+++ b/server/src/bitmind/config.ts
@@ -0,0 +1,111 @@
+/**
+ * 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;
+  const value = Number.parseInt(raw, 10);
+  if (!Number.isInteger(value) || value < minimum || value > maximum) {
+    throw new Error(
+      `${name}=${raw} must be an integer 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..b1afaa109
--- /dev/null
+++ b/server/src/bitmind/gateway.ts
@@ -0,0 +1,202 @@
+import { RunAgentInputSchema } from "@ag-ui/core";
+import { matchesToken } from "../../../shared/agent-authorisation";
+import { AG_UI_PROTOCOL_VERSION, type BitmindGatewayConfig } from "./config";
+
+/**
+ * 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<string, AbortController>();
+
+  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<Response> {
+    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 key =
+      request.headers.get("idempotency-key")?.trim() || input.data.runId;
+    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);
+    const release = () => {
+      clearTimeout(timeout);
+      active.delete(key);
+    };
+    request.signal.addEventListener("abort", () => {
+      controller.abort(request.signal.reason as Error | undefined);
+    });
+    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) {
+      release();
+      return Response.json(
+        { error: "The execution backend refused the run." },
+        { status: 502 },
+      );
+    }
+
+    const relayed = downstream.body.pipeThrough(
+      new TransformStream<Uint8Array, Uint8Array>({
+        flush() {
+          release();
+        },
+      }),
+    );
+    controller.signal.addEventListener("abort", release, { once: true });
+
+    return new Response(relayed, {
+      headers: {
+        "content-type":
+          downstream.headers.get("content-type") ?? "text/event-stream",
+        "cache-control": "no-cache",
+      },
+    });
+  }
+
+  return {
+    async fetch(request: Request): Promise<Response> {
+      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..27b18c95f
--- /dev/null
+++ b/server/tests/bitmind-gateway.test.ts
@@ -0,0 +1,300 @@
+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<Parameters<typeof createBitmindGateway>[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,
+      fencing_token: 1,
+    },
+  };
+}
+
+function runRequest(body: unknown, headers: Record<string, string> = {}) {
+  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<string, unknown>;
+    };
+    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: "<script>alert(1)</script>" }),
+    );
+    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<Uint8Array>({
+            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<Uint8Array>({
+            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"), { "idempotency-key": "a" }),
+      ),
+      await gateway.fetch(
+        runRequest(runInput("run-2"), { "idempotency-key": "b" }),
+      ),
+    ];
+    expect(responses.map((r) => r.status)).toEqual([200, 200]);
+
+    const refused = await gateway.fetch(
+      runRequest(runInput("run-3"), { "idempotency-key": "c" }),
+    );
+    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/);
+  });
+});

From 77095ce346990c2fc716d941d795c8a1a8f2656e Mon Sep 17 00:00:00 2001
From: Bit Cloud <dev@warsuit.io>
Date: Wed, 2 Sep 2026 17:19:43 +0000
Subject: [PATCH 2/4] fix: address review on the BitMind gateway

- The relay is pumped by hand instead of piped through a transform: the
  admission slot comes back on every ending - clean close, source error,
  consumer cancellation, timeout - where flush() only ran for the first.
  Consumer cancellation also aborts the downstream fetch, so the far side
  is told to stop rather than abandoned.
- A 2xx with a body is not a run: only text/event-stream may be reserved
  and relayed; a JSON login page or error is a 502 with the slot released.
- BitMind's identity statement is now required in full (strict
  forwardedProps schema incl. message_id), must agree with the envelope
  (forwardedProps.run_id === runId), and the idempotency-key header must
  equal run_id:fencing_token - the key is bound to the attempt, never
  caller-selectable.
- The gateway holds its own attestation: while tools: false and
  interrupts: false, non-empty tools and resume inputs are refused before
  they can reach the downstream, rather than trusting it to.
- agent-langgraph propagates cancellation: the response stream's cancel()
  and the request's own signal both abort the model invocation
  (streamEvents receives the signal), so a disconnected consumer stops
  the work instead of leaving it running beyond the attested ceiling.
  Response lifecycle factored into respond.ts so tests reach it without
  binding a port.
- Gateway limits parse the whole string: 2workers, 1000ms and 2.5 are
  refused at startup instead of silently becoming ceilings nobody set.
---
 agent-langgraph/src/index.ts          |  58 +++-------
 agent-langgraph/src/respond.ts        |  77 +++++++++++++
 agent-langgraph/src/stream.ts         |   5 +
 agent-langgraph/tests/respond.test.ts | 107 ++++++++++++++++++
 server/src/bitmind/config.ts          |  15 ++-
 server/src/bitmind/gateway.ts         | 124 +++++++++++++++++++--
 server/tests/bitmind-gateway.test.ts  | 150 ++++++++++++++++++++++++--
 7 files changed, 473 insertions(+), 63 deletions(-)
 create mode 100644 agent-langgraph/src/respond.ts
 create mode 100644 agent-langgraph/tests/respond.test.ts

diff --git a/agent-langgraph/src/index.ts b/agent-langgraph/src/index.ts
index fc7ed28f2..8caa476b8 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,7 @@ 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";
 
 /**
  * The same Bot, on a framework.
@@ -384,43 +383,20 @@ function buildGraph(input: RunAgentInput) {
     .compile();
 }
 
-async function runAgent(input: RunAgentInput): Promise<Response> {
-  const encoder = new EventEncoder();
-  const stream = new ReadableStream<Uint8Array>({
-    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 +420,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<AsyncIterable<RunStreamEvent>>,
+  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<Uint8Array>({
+    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<AsyncIterable<RunStreamEvent>>,
   input: Pick<RunAgentInput, "runId" | "threadId">,
   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<void> {
   /*
    * 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/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<AsyncIterable<RunStreamEvent>> => {
+    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/server/src/bitmind/config.ts b/server/src/bitmind/config.ts
index 6824a3f4a..f5040558a 100644
--- a/server/src/bitmind/config.ts
+++ b/server/src/bitmind/config.ts
@@ -41,10 +41,19 @@ function integer(
 ): number {
   const raw = environment[name]?.trim();
   if (!raw) return fallback;
-  const value = Number.parseInt(raw, 10);
-  if (!Number.isInteger(value) || value < minimum || value > maximum) {
+  // 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 an integer between ${minimum} and ${maximum}.`,
+      `${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;
diff --git a/server/src/bitmind/gateway.ts b/server/src/bitmind/gateway.ts
index b1afaa109..5a19b7c57 100644
--- a/server/src/bitmind/gateway.ts
+++ b/server/src/bitmind/gateway.ts
@@ -1,7 +1,26 @@
 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
@@ -90,9 +109,49 @@ export function createBitmindGateway(
         { 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 key =
-      request.headers.get("idempotency-key")?.trim() || input.data.runId;
+    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." },
@@ -150,20 +209,65 @@ export function createBitmindGateway(
         { 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") ?? "";
+    if (!contentType.toLowerCase().startsWith("text/event-stream")) {
+      controller.abort(
+        new Error("backend did not answer with an event stream"),
+      );
+      release();
+      return Response.json(
+        { error: "The execution backend did not answer with an event stream." },
+        { status: 502 },
+      );
+    }
 
-    const relayed = downstream.body.pipeThrough(
-      new TransformStream<Uint8Array, Uint8Array>({
-        flush() {
+    // 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<Uint8Array>({
+      async pull(streamController) {
+        let result: Awaited<ReturnType<typeof reader.read>>;
+        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 },
     );
-    controller.signal.addEventListener("abort", release, { once: true });
 
     return new Response(relayed, {
       headers: {
-        "content-type":
-          downstream.headers.get("content-type") ?? "text/event-stream",
+        "content-type": contentType,
         "cache-control": "no-cache",
       },
     });
diff --git a/server/tests/bitmind-gateway.test.ts b/server/tests/bitmind-gateway.test.ts
index 27b18c95f..b4e3e617f 100644
--- a/server/tests/bitmind-gateway.test.ts
+++ b/server/tests/bitmind-gateway.test.ts
@@ -34,6 +34,7 @@ function runInput(runId = "run-1") {
       workspace_id: "workspace-1",
       agent_id: "agent-1",
       run_id: runId,
+      message_id: "message-1",
       fencing_token: 1,
     },
   };
@@ -232,18 +233,12 @@ describe("run relay", () => {
       hanging,
     );
     const responses = [
-      await gateway.fetch(
-        runRequest(runInput("run-1"), { "idempotency-key": "a" }),
-      ),
-      await gateway.fetch(
-        runRequest(runInput("run-2"), { "idempotency-key": "b" }),
-      ),
+      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"), { "idempotency-key": "c" }),
-    );
+    const refused = await gateway.fetch(runRequest(runInput("run-3")));
     expect(refused.status).toBe(429);
     expect(refused.headers.get("retry-after")).toBe("5");
 
@@ -298,3 +293,140 @@ describe("configuration", () => {
     ).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<string, unknown> };
+    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<Uint8Array>({
+            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<Uint8Array>({
+            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/);
+    },
+  );
+});

From ed08c539137203e11dbbf323939efac6ea237d2c Mon Sep 17 00:00:00 2001
From: Bit Cloud <dev@warsuit.io>
Date: Wed, 2 Sep 2026 18:04:58 +0000
Subject: [PATCH 3/4] fix: close the cancellation and media-type gaps the
 re-review found

- A refused (non-2xx) response can arrive over a body that keeps streaming:
  the fetch is now aborted and the body cancelled before admission is
  released, matching what the non-SSE path already did - and that path now
  cancels the body too.
- The media type is compared exactly (segment before the first ';',
  trimmed, case-insensitive): text/event-streaming and
  text/event-stream+json are refused, while charset parameters survive
  both the check and the forwarded header.
- Abort events are not replayed: a caller that hung up while the body was
  being read or validated is now caught by an explicit aborted check after
  the listener registers, before a slot is reserved or the agent reached.
- A cancelled run now stops an in-flight governed tool call, not only the
  model stream: the call moved to tools.ts with the signal passed into its
  fetch, the graph's nodes take the node config (streamEvents propagates
  the run signal through it) and hand it to both the model invocation and
  the tool call. Tested with a fetch that blocks until aborted.
---
 agent-langgraph/src/index.ts         | 150 +++++++++++----------------
 agent-langgraph/src/tools.ts         |  68 ++++++++++++
 agent-langgraph/tests/tools.test.ts  |  89 ++++++++++++++++
 server/src/bitmind/gateway.ts        |  23 +++-
 server/tests/bitmind-gateway.test.ts |  92 ++++++++++++++++
 5 files changed, 330 insertions(+), 92 deletions(-)
 create mode 100644 agent-langgraph/src/tools.ts
 create mode 100644 agent-langgraph/tests/tools.test.ts

diff --git a/agent-langgraph/src/index.ts b/agent-langgraph/src/index.ts
index 8caa476b8..87c3f9bb5 100644
--- a/agent-langgraph/src/index.ts
+++ b/agent-langgraph/src/index.ts
@@ -14,6 +14,7 @@ import { hasManagedAgentToken } from "../../shared/agent-authorisation";
 import { toLangChainMessages } from "./history";
 import { readReasoningEffort } from "./model-options";
 import { respondWithRun } from "./respond";
+import { callDeploymentTool } from "./tools";
 
 /**
  * The same Bot, on a framework.
@@ -226,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<string, unknown>,
-): Promise<string> {
-  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.
  *
@@ -329,58 +290,65 @@ 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<string, unknown>,
+                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<string, unknown>,
-            );
-            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()
+  );
 }
 
 function runAgent(input: RunAgentInput, clientSignal?: AbortSignal): Response {
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<string, unknown>,
+  signal?: AbortSignal,
+  fetchImplementation: typeof fetch = fetch,
+): Promise<string> {
+  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/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/server/src/bitmind/gateway.ts b/server/src/bitmind/gateway.ts
index 5a19b7c57..353d82419 100644
--- a/server/src/bitmind/gateway.ts
+++ b/server/src/bitmind/gateway.ts
@@ -178,6 +178,17 @@ export function createBitmindGateway(
     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;
@@ -203,6 +214,11 @@ export function createBitmindGateway(
       );
     }
     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." },
@@ -213,10 +229,15 @@ export function createBitmindGateway(
     // 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") ?? "";
-    if (!contentType.toLowerCase().startsWith("text/event-stream")) {
+    // 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." },
diff --git a/server/tests/bitmind-gateway.test.ts b/server/tests/bitmind-gateway.test.ts
index b4e3e617f..f1c4d222e 100644
--- a/server/tests/bitmind-gateway.test.ts
+++ b/server/tests/bitmind-gateway.test.ts
@@ -430,3 +430,95 @@ describe("strict limits", () => {
     },
   );
 });
+
+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<Uint8Array>({
+            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);
+  });
+});

From 0a179b30dec6d728507adf2792736797316f1127 Mon Sep 17 00:00:00 2001
From: Bit Cloud <dev@warsuit.io>
Date: Wed, 2 Sep 2026 19:57:15 +0000
Subject: [PATCH 4/4] fix: let only the slot's owner release it
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A cancelled relay ends through several paths at once — the abort listener,
the stream's cancel(), and the pending read resolving — and each called
release(), which deleted the idempotency key unconditionally. A same-key
retry admitted between two of those endings had its own entry deleted while
its stream was still live, so the concurrency accounting lost the run and a
third POST with the same key was admitted instead of refused with 409.

release() now gives the slot back only when this relay still holds it, which
also makes every ending safe to run more than once.
---
 server/src/bitmind/gateway.ts        | 15 +++++++-
 server/tests/bitmind-gateway.test.ts | 51 ++++++++++++++++++++++++++++
 2 files changed, 65 insertions(+), 1 deletion(-)

diff --git a/server/src/bitmind/gateway.ts b/server/src/bitmind/gateway.ts
index 353d82419..5a6bf6d81 100644
--- a/server/src/bitmind/gateway.ts
+++ b/server/src/bitmind/gateway.ts
@@ -171,9 +171,22 @@ export function createBitmindGateway(
     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);
-      active.delete(key);
+      if (active.get(key) === controller) {
+        active.delete(key);
+      }
     };
     request.signal.addEventListener("abort", () => {
       controller.abort(request.signal.reason as Error | undefined);
diff --git a/server/tests/bitmind-gateway.test.ts b/server/tests/bitmind-gateway.test.ts
index f1c4d222e..554d324dc 100644
--- a/server/tests/bitmind-gateway.test.ts
+++ b/server/tests/bitmind-gateway.test.ts
@@ -522,3 +522,54 @@ describe("follow-up findings", () => {
     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<Uint8Array>({
+            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);
+  });
+});