feat(sandbox): add the host duplex bridge broker with nested budgets

Add the host broker for the sandbox duplex channel. The broker reads
request frames from one persistent channel, forwards each request on the
existing API path, and writes one response frame back. It sends heartbeat
frames on an interval to prove liveness.

The broker keeps the forward path unchanged. The caller passes a forward
handler, and the broker calls it for each request. The handler owns the
route allowlist, the token replacement, and the run attribution, so those
rules stay in one place.

The broker runs nested timeout budgets: forward 30 s, response 32 s, and
gateway wait 35 s. Each inner budget is smaller than its outer budget. The
broker asserts this order at construction and rejects a configuration that
breaks it.

Loss is terminal. The broker detects loss through channel exit, a stream
write failure, a protocol failure, a heartbeat write failure, and a close
timeout. On loss the broker stops, records the loss for metrics only, and
dispatches nothing more. It never reconnects and never replays a request.

Wire the existing forward timeout in execution-target.ts to a new option
that defaults to the broker forward budget (30 s), so the current behavior
does not change.

Co-authored-by: Paperclip <noreply@paperclip.ing>

packages/adapter-utils/src/duplex-bridge-broker.ts

@@ -0,0 +1,487 @@
+/**
+ * Host broker for the sandbox duplex channel.
+ *
+ * The broker owns the host end of one persistent duplex channel to the sandbox
+ * gateway. It reads request frames from the channel, forwards each one on the
+ * existing Paperclip API path, and writes one response frame back. It sends a
+ * heartbeat frame on an interval to prove liveness.
+ *
+ * The broker does not hold the route allowlist, the token replacement, or the
+ * run attribution. The caller passes a forward handler, and the broker calls it
+ * for each request. The handler applies the real token and the signed run
+ * identifier, so those rules stay in one place on the existing forward path.
+ *
+ * The broker runs a set of nested timeout budgets:
+ *   - forward budget: the deadline for one forward call.
+ *   - response budget: the deadline for the broker to send one response frame.
+ *   - gateway wait budget: the deadline the in-sandbox gateway waits for the
+ *     response frame.
+ * Each inner budget is smaller than its outer budget, so the broker aborts and
+ * answers before the gateway gives up. The broker asserts this order at
+ * construction and fails a configuration that breaks it.
+ *
+ * Loss is terminal. The broker detects loss through channel exit, a stream
+ * write failure, a protocol failure, a heartbeat write failure, and a close
+ * timeout. On loss the broker stops the heartbeat, aborts every in-flight
+ * forward, records the loss, and dispatches nothing more. The broker never
+ * reconnects and never replays a request. The broker records the loss for
+ * metrics only and sends nothing about the loss to the sandbox.
+ */
+
+import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
+import {
+  DUPLEX_FRAME_VERSION,
+  DuplexFrameDecoder,
+  encodeDuplexFrame,
+  type DuplexFrame,
+  type DuplexRequestFrame,
+  type DuplexResponseFrame,
+  type DuplexResponseOutcome,
+} from "./duplex-frame-codec.js";
+
+/** The lifecycle states of the broker. The broker moves through them in order. */
+export type DuplexBrokerState = "opening" | "open" | "lost" | "closing" | "closed";
+
+/** The reason the broker classified a loss. The broker records it for metrics only. */
+export type DuplexBrokerLossReason =
+  | "channel_exit"
+  | "stream_failure"
+  | "protocol_failure"
+  | "heartbeat_write_failure"
+  | "close_timeout";
+
+/** The nested timeout budgets. Each inner budget is smaller than its outer budget. */
+export interface DuplexBrokerBudgets {
+  /** The deadline for one forward call, in milliseconds. */
+  forwardTimeoutMs: number;
+  /** The deadline for the broker to send one response frame, in milliseconds. */
+  responseBudgetMs: number;
+  /** The deadline the in-sandbox gateway waits for the response frame, in milliseconds. */
+  gatewayWaitMs: number;
+}
+
+/** The default nested budgets: forward 30 s, response 32 s, gateway wait 35 s. */
+export const DEFAULT_DUPLEX_BROKER_BUDGETS: DuplexBrokerBudgets = {
+  forwardTimeoutMs: 30_000,
+  responseBudgetMs: 32_000,
+  gatewayWaitMs: 35_000,
+};
+
+/** The default interval between two heartbeat frames, in milliseconds. */
+export const DEFAULT_DUPLEX_BROKER_HEARTBEAT_INTERVAL_MS = 5_000;
+
+/** The default deadline for an orderly channel close, in milliseconds. */
+export const DEFAULT_DUPLEX_BROKER_CLOSE_TIMEOUT_MS = 2_000;
+
+/** The result of one forward call. The broker turns it into one response frame. */
+export interface DuplexBrokerForwardResult {
+  status: number;
+  headers?: Record<string, string>;
+  body?: string;
+}
+
+/**
+ * The forward handler the broker calls for each request. The handler applies the
+ * real token and the run attribution, then forwards the request on the existing
+ * API path. The broker aborts `options.signal` when the forward budget ends or a
+ * loss happens, so a handler that threads the signal into its work stops early.
+ */
+export type DuplexBrokerForwardHandler = (
+  request: DuplexRequestFrame,
+  options: { signal: AbortSignal },
+) => Promise<DuplexBrokerForwardResult>;
+
+/** One request record. The broker captures the dispatch-start point for metrics only. */
+export interface DuplexBrokerRequestRecord {
+  id: string;
+  method: string;
+  path: string;
+  /** The point the broker started to dispatch the request, in milliseconds. */
+  dispatchStartMs: number;
+}
+
+/** One loss record. The broker reports it for metrics only. */
+export interface DuplexBrokerLossRecord {
+  reason: DuplexBrokerLossReason;
+  message: string;
+  /** The point the broker recorded the loss, in milliseconds. */
+  atMs: number;
+}
+
+/** The options for {@link createDuplexBridgeBroker}. */
+export interface DuplexBrokerOptions {
+  /** The duplex channel to the sandbox gateway. */
+  channel: CommandManagedDuplexChannel;
+  /** The forward handler the broker calls for each request. */
+  forwardRequest: DuplexBrokerForwardHandler;
+  /** The nested timeout budgets. The default is {@link DEFAULT_DUPLEX_BROKER_BUDGETS}. */
+  budgets?: Partial<DuplexBrokerBudgets>;
+  /** The interval between two heartbeat frames, in milliseconds. */
+  heartbeatIntervalMs?: number;
+  /** The deadline for an orderly channel close, in milliseconds. */
+  closeTimeoutMs?: number;
+  /** The maximum size of one inbound frame, in bytes. Forwarded to the decoder. */
+  maxFrameBytes?: number;
+  /** The clock the broker reads for the metric timestamps. The default is `Date.now`. */
+  now?: () => number;
+  /** The metrics sink for the per-request dispatch record. */
+  onRequestRecord?: (record: DuplexBrokerRequestRecord) => void;
+  /** The metrics sink for the terminal loss record. */
+  onLoss?: (record: DuplexBrokerLossRecord) => void;
+  /** The sink for a state change. The broker reports every transition. */
+  onStateChange?: (state: DuplexBrokerState) => void;
+  /** The sink for a diagnostic message. The broker never writes diagnostics to the channel. */
+  logger?: (message: string) => void;
+}
+
+/** The broker handle the factory returns. */
+export interface DuplexBridgeBroker {
+  /** The current state of the broker. */
+  readonly state: DuplexBrokerState;
+  /** The loss record, or `null` when the broker never lost the channel. */
+  readonly lossRecord: DuplexBrokerLossRecord | null;
+  /** Start the broker. It wires the channel listeners and moves to `open`. */
+  start(): void;
+  /** Close the channel cleanly. It moves through `closing` to `closed`. */
+  close(): Promise<void>;
+  /** Stop the sandbox child process. Safe to call more than one time. */
+  stop(): void;
+}
+
+/**
+ * Assert the nested budget order. Each inner budget must be smaller than its
+ * outer budget, so the broker answers before the gateway gives up. The function
+ * throws when the order breaks.
+ */
+export function assertNestedDuplexBrokerBudgets(budgets: DuplexBrokerBudgets): void {
+  if (!(budgets.forwardTimeoutMs < budgets.responseBudgetMs)) {
+    throw new Error(
+      `Duplex broker forward budget ${budgets.forwardTimeoutMs}ms must be smaller than the response budget ${budgets.responseBudgetMs}ms.`,
+    );
+  }
+  if (!(budgets.responseBudgetMs < budgets.gatewayWaitMs)) {
+    throw new Error(
+      `Duplex broker response budget ${budgets.responseBudgetMs}ms must be smaller than the gateway wait budget ${budgets.gatewayWaitMs}ms.`,
+    );
+  }
+}
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
+
+/** The internal bookkeeping for one in-flight request. */
+interface PendingRequest {
+  controller: AbortController;
+  responded: boolean;
+  forwardTimer: ReturnType<typeof setTimeout>;
+  responseTimer: ReturnType<typeof setTimeout>;
+}
+
+/**
+ * Create the host duplex bridge broker. The factory asserts the budget order and
+ * returns a handle. Call `start` to wire the channel and open the broker.
+ */
+export function createDuplexBridgeBroker(options: DuplexBrokerOptions): DuplexBridgeBroker {
+  const budgets: DuplexBrokerBudgets = {
+    ...DEFAULT_DUPLEX_BROKER_BUDGETS,
+    ...options.budgets,
+  };
+  assertNestedDuplexBrokerBudgets(budgets);
+
+  const channel = options.channel;
+  const forwardRequest = options.forwardRequest;
+  const heartbeatIntervalMs = options.heartbeatIntervalMs ?? DEFAULT_DUPLEX_BROKER_HEARTBEAT_INTERVAL_MS;
+  const closeTimeoutMs = options.closeTimeoutMs ?? DEFAULT_DUPLEX_BROKER_CLOSE_TIMEOUT_MS;
+  const now = options.now ?? (() => Date.now());
+  const decoder = new DuplexFrameDecoder(
+    options.maxFrameBytes ? { maxFrameBytes: options.maxFrameBytes } : undefined,
+  );
+
+  let state: DuplexBrokerState = "opening";
+  let stopped = false;
+  let started = false;
+  let closePromise: Promise<void> | null = null;
+  let lossRecord: DuplexBrokerLossRecord | null = null;
+  let heartbeatTimer: ReturnType<typeof setInterval> | null = null;
+  // The ids the broker already dispatched. The broker forwards one id one time,
+  // so a repeated frame never reaches the API twice.
+  const seenRequestIds = new Set<string>();
+  const pending = new Map<string, PendingRequest>();
+
+  const setState = (next: DuplexBrokerState): void => {
+    if (state === next) return;
+    state = next;
+    options.onStateChange?.(next);
+  };
+
+  const clearHeartbeat = (): void => {
+    if (heartbeatTimer !== null) {
+      clearInterval(heartbeatTimer);
+      heartbeatTimer = null;
+    }
+  };
+
+  const clearPending = (): void => {
+    for (const entry of pending.values()) {
+      clearTimeout(entry.forwardTimer);
+      clearTimeout(entry.responseTimer);
+      entry.controller.abort(new Error("Duplex broker stopped."));
+    }
+    pending.clear();
+  };
+
+  const recordLoss = (reason: DuplexBrokerLossReason, message: string): void => {
+    // Loss is terminal. Record it one time and stop every activity.
+    if (stopped) return;
+    stopped = true;
+    clearHeartbeat();
+    clearPending();
+    lossRecord = { reason, message, atMs: now() };
+    setState("lost");
+    options.logger?.(`Duplex broker lost the channel (${reason}): ${message}`);
+    options.onLoss?.(lossRecord);
+  };
+
+  const writeFrame = (frame: DuplexFrame): boolean => {
+    try {
+      channel.write(encodeDuplexFrame(frame));
+      return true;
+    } catch (error) {
+      recordLoss("stream_failure", errorMessage(error));
+      return false;
+    }
+  };
+
+  const respond = (
+    id: string,
+    result: DuplexBrokerForwardResult,
+    outcome: DuplexResponseOutcome,
+  ): void => {
+    const entry = pending.get(id);
+    if (!entry || entry.responded) return;
+    entry.responded = true;
+    clearTimeout(entry.forwardTimer);
+    clearTimeout(entry.responseTimer);
+    pending.delete(id);
+    // Do not write on a lost or closed channel. The gateway answers its own
+    // outstanding request on loss, so a late write would go to a dead channel.
+    if (state !== "open") return;
+    const frame: DuplexResponseFrame = {
+      version: DUPLEX_FRAME_VERSION,
+      type: "response",
+      id,
+      status: result.status,
+      headers: result.headers ?? {},
+      body: result.body ?? "",
+      outcome,
+    };
+    writeFrame(frame);
+  };
+
+  const dispatch = (frame: DuplexRequestFrame): void => {
+    // Dispatch only while open. After loss or close the broker forwards nothing.
+    if (state !== "open") return;
+    // Forward one id one time. A repeated id never reaches the API twice.
+    if (seenRequestIds.has(frame.id)) return;
+    seenRequestIds.add(frame.id);
+
+    const record: DuplexBrokerRequestRecord = {
+      id: frame.id,
+      method: frame.method,
+      path: frame.path,
+      dispatchStartMs: now(),
+    };
+    options.onRequestRecord?.(record);
+
+    const controller = new AbortController();
+    const forwardTimer = setTimeout(() => {
+      controller.abort(new Error("Duplex broker forward budget exceeded."));
+    }, budgets.forwardTimeoutMs);
+    const responseTimer = setTimeout(() => {
+      // Response-budget backstop. The forward rejection normally answers first,
+      // well before this deadline. This backstop answers a request whose forward
+      // rejection handling itself stalls, so the request never strands.
+      respond(
+        frame.id,
+        {
+          status: 504,
+          headers: {
+            "content-type": "application/json",
+            "x-paperclip-bridge-outcome": "indeterminate",
+          },
+          body: JSON.stringify({
+            error: "Duplex broker response budget exceeded.",
+            outcome: "indeterminate",
+            retryable: false,
+          }),
+        },
+        "indeterminate",
+      );
+    }, budgets.responseBudgetMs);
+    forwardTimer.unref?.();
+    responseTimer.unref?.();
+    pending.set(frame.id, { controller, responded: false, forwardTimer, responseTimer });
+
+    forwardRequest(frame, { signal: controller.signal }).then(
+      (result) => {
+        // Keep the outcome classification consistent with the file path. A
+        // possibly-committed mutation carries the indeterminate marker header, so
+        // map it to the indeterminate outcome. Any other result is completed.
+        const outcome: DuplexResponseOutcome =
+          result.headers?.["x-paperclip-bridge-outcome"] === "indeterminate"
+            ? "indeterminate"
+            : "completed";
+        respond(frame.id, result, outcome);
+      },
+      (error) => {
+        if (controller.signal.aborted) {
+          // The forward budget aborted the call after the request may have
+          // committed. Return a non-retryable 504 and mark the outcome
+          // indeterminate, so a caller does not retry a committed mutation.
+          respond(
+            frame.id,
+            {
+              status: 504,
+              headers: {
+                "content-type": "application/json",
+                "x-paperclip-bridge-outcome": "indeterminate",
+              },
+              body: JSON.stringify({
+                error: errorMessage(error),
+                outcome: "indeterminate",
+                retryable: false,
+              }),
+            },
+            "indeterminate",
+          );
+          return;
+        }
+        // A plain forward failure did not start a host mutation, so a retry is
+        // safe. Return a 502 with the completed outcome, so the gateway passes it
+        // through as a retryable status.
+        respond(
+          frame.id,
+          {
+            status: 502,
+            headers: { "content-type": "application/json" },
+            body: JSON.stringify({ error: errorMessage(error) }),
+          },
+          "completed",
+        );
+      },
+    );
+  };
+
+  const handleFrame = (frame: DuplexFrame): void => {
+    switch (frame.type) {
+      case "request":
+        dispatch(frame);
+        return;
+      case "close":
+        // The gateway asked for an orderly close. Close the channel.
+        void close();
+        return;
+      case "ready":
+      case "heartbeat":
+        // Liveness frames. The broker reads them and dispatches nothing.
+        return;
+      case "response":
+      case "error":
+        // The host never expects these on the read path. Ignore them.
+        return;
+      default:
+        return;
+    }
+  };
+
+  const onData = (chunk: string): void => {
+    if (stopped) return;
+    const results = decoder.push(chunk);
+    for (const result of results) {
+      if (stopped) return;
+      if (!result.ok) {
+        recordLoss("protocol_failure", result.error.message);
+        return;
+      }
+      handleFrame(result.frame);
+    }
+  };
+
+  const onExit = (): void => {
+    recordLoss("channel_exit", "The sandbox channel process exited.");
+  };
+
+  const sendHeartbeat = (): void => {
+    if (state !== "open") return;
+    try {
+      channel.write(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "heartbeat" }));
+    } catch (error) {
+      recordLoss("heartbeat_write_failure", errorMessage(error));
+    }
+  };
+
+  const close = (): Promise<void> => {
+    if (closePromise) return closePromise;
+    if (stopped) return Promise.resolve();
+    closePromise = (async () => {
+      setState("closing");
+      clearHeartbeat();
+      clearPending();
+      // Send an orderly close frame. Ignore a write failure here; the broker is
+      // already closing, so a dead channel needs no loss record.
+      try {
+        channel.write(encodeDuplexFrame({ version: DUPLEX_FRAME_VERSION, type: "close" }));
+      } catch (error) {
+        options.logger?.(`Duplex broker could not send the close frame: ${errorMessage(error)}`);
+      }
+      let closeTimer: ReturnType<typeof setTimeout> | undefined;
+      const timeout = new Promise<never>((_resolve, reject) => {
+        closeTimer = setTimeout(() => {
+          reject(new Error("Duplex broker channel close timed out."));
+        }, closeTimeoutMs);
+        closeTimer.unref?.();
+      });
+      try {
+        await Promise.race([channel.close(), timeout]);
+        if (closeTimer !== undefined) clearTimeout(closeTimer);
+        stopped = true;
+        setState("closed");
+      } catch (error) {
+        if (closeTimer !== undefined) clearTimeout(closeTimer);
+        recordLoss("close_timeout", errorMessage(error));
+      }
+    })();
+    return closePromise;
+  };
+
+  const start = (): void => {
+    if (started) return;
+    started = true;
+    channel.onData(onData);
+    channel.onExit(onExit);
+    heartbeatTimer = setInterval(sendHeartbeat, heartbeatIntervalMs);
+    heartbeatTimer.unref?.();
+    setState("open");
+  };
+
+  const stop = (): void => {
+    try {
+      channel.stop();
+    } catch (error) {
+      options.logger?.(`Duplex broker could not stop the channel: ${errorMessage(error)}`);
+    }
+  };
+
+  return {
+    get state() {
+      return state;
+    },
+    get lossRecord() {
+      return lossRecord;
+    },
+    start,
+    close,
+    stop,
+  };
+}

packages/adapter-utils/src/execution-target-sandbox.test.ts

@@ -40,6 +40,22 @@ import {
 import { createSandboxRunLogTailFactory } from "./sandbox-run-log-stream.js";
 import { runChildProcess } from "./server-utils.js";
 import { shellQuote } from "./ssh.js";
+import type { CommandManagedDuplexChannel } from "./command-managed-runtime.js";
+import {
+  DUPLEX_FRAME_VERSION,
+  decodeDuplexLine,
+  encodeDuplexFrame,
+  type DuplexRequestFrame,
+  type DuplexResponseFrame,
+} from "./duplex-frame-codec.js";
+import {
+  assertNestedDuplexBrokerBudgets,
+  createDuplexBridgeBroker,
+  type DuplexBrokerForwardResult,
+  type DuplexBrokerLossRecord,
+  type DuplexBrokerRequestRecord,
+  type DuplexBrokerState,
+} from "./duplex-bridge-broker.js";
 
 const execFileAsync = promisify(execFile);
 
@@ -3275,3 +3291,306 @@ describe("sandbox duplex gateway", () => {
     await gateway.stop();
   }, 20000);
 });
+
+/**
+ * A scripted fake duplex channel. The test drives the read path with
+ * `emitData`/`emitExit`, and reads the frames the broker wrote through `written`.
+ * The `writeError` and `closeBehavior` hooks let a test force a stream failure
+ * and a close timeout.
+ */
+function createFakeDuplexChannel(): {
+  channel: CommandManagedDuplexChannel;
+  emitData: (chunk: string) => void;
+  emitExit: (exit: { exitCode: number | null }) => void;
+  written: DuplexResponseFrame[];
+  writtenTypes: string[];
+  stopped: () => number;
+  setWriteError: (error: Error | null) => void;
+  setCloseBehavior: (behavior: "resolve" | "hang" | "reject") => void;
+} {
+  let dataListener: ((chunk: string) => void) | null = null;
+  let exitListener: ((exit: { exitCode: number | null }) => void) | null = null;
+  let writeError: Error | null = null;
+  let closeBehavior: "resolve" | "hang" | "reject" = "resolve";
+  let stopCount = 0;
+  const written: DuplexResponseFrame[] = [];
+  const writtenTypes: string[] = [];
+
+  const channel: CommandManagedDuplexChannel = {
+    write(data: string): void {
+      if (writeError) throw writeError;
+      const decoded = decodeDuplexLine(data.replace(/\n$/, ""));
+      if (decoded.ok) {
+        writtenTypes.push(decoded.frame.type);
+        if (decoded.frame.type === "response") written.push(decoded.frame);
+      }
+    },
+    onData(listener: (chunk: string) => void): void {
+      dataListener = listener;
+    },
+    onExit(listener: (exit: { exitCode: number | null }) => void): void {
+      exitListener = listener;
+    },
+    stop(): void {
+      stopCount += 1;
+    },
+    close(): Promise<void> {
+      if (closeBehavior === "resolve") return Promise.resolve();
+      if (closeBehavior === "reject") return Promise.reject(new Error("close rejected"));
+      return new Promise<void>(() => {});
+    },
+  };
+
+  return {
+    channel,
+    emitData: (chunk: string) => dataListener?.(chunk),
+    emitExit: (exit: { exitCode: number | null }) => exitListener?.(exit),
+    written,
+    writtenTypes,
+    stopped: () => stopCount,
+    setWriteError: (error: Error | null) => {
+      writeError = error;
+    },
+    setCloseBehavior: (behavior: "resolve" | "hang" | "reject") => {
+      closeBehavior = behavior;
+    },
+  };
+}
+
+/** Build one request frame line the fake channel can emit. */
+function requestFrameLine(overrides: Partial<DuplexRequestFrame> & { id: string }): string {
+  const frame: DuplexRequestFrame = {
+    version: DUPLEX_FRAME_VERSION,
+    type: "request",
+    id: overrides.id,
+    method: overrides.method ?? "POST",
+    path: overrides.path ?? "/api/issues/PAP-1/comments",
+    query: overrides.query ?? "",
+    headers: overrides.headers ?? { authorization: "Bearer bridge-token" },
+    body: overrides.body ?? JSON.stringify({ body: "hello" }),
+  };
+  return encodeDuplexFrame(frame);
+}
+
+/** Wait for the pending microtasks and macrotasks to settle. */
+function flushMacrotasks(): Promise<void> {
+  return new Promise((resolve) => setImmediate(resolve));
+}
+
+describe("createDuplexBridgeBroker", () => {
+  it("forwards a decoded request to the handler and writes the handler response back", async () => {
+    const fake = createFakeDuplexChannel();
+    const received: DuplexRequestFrame[] = [];
+    // The forward handler owns the token replacement and the run attribution. The
+    // broker passes the decoded request straight through, so the sandbox request
+    // still carries only the bridge token here, and the handler applies the real
+    // token and the signed run identifier on the existing forward path.
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async (request): Promise<DuplexBrokerForwardResult> => {
+        received.push(request);
+        expect(request.headers.authorization).toBe("Bearer bridge-token");
+        expect(request.headers["x-paperclip-run-id"]).toBeUndefined();
+        return {
+          status: 201,
+          headers: { "content-type": "application/json" },
+          body: JSON.stringify({ ok: true }),
+        };
+      },
+    });
+
+    broker.start();
+    fake.emitData(requestFrameLine({ id: "req-1" }));
+    await flushMacrotasks();
+
+    expect(received).toHaveLength(1);
+    expect(received[0].id).toBe("req-1");
+    expect(fake.written).toHaveLength(1);
+    expect(fake.written[0]).toMatchObject({
+      type: "response",
+      id: "req-1",
+      status: 201,
+      outcome: "completed",
+    });
+    expect(JSON.parse(fake.written[0].body)).toEqual({ ok: true });
+
+    await broker.close();
+  });
+
+  it("moves through opening, open, closing, closed in order", async () => {
+    const fake = createFakeDuplexChannel();
+    const states: DuplexBrokerState[] = [];
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async () => ({ status: 200 }),
+      onStateChange: (state) => states.push(state),
+    });
+
+    expect(broker.state).toBe("opening");
+    broker.start();
+    expect(broker.state).toBe("open");
+    await broker.close();
+    expect(broker.state).toBe("closed");
+    expect(states).toEqual(["open", "closing", "closed"]);
+    expect(fake.writtenTypes).toContain("close");
+  });
+
+  it.each([
+    {
+      name: "channel exit",
+      reason: "channel_exit" as const,
+      trigger: (fake: ReturnType<typeof createFakeDuplexChannel>) =>
+        fake.emitExit({ exitCode: 1 }),
+    },
+    {
+      name: "protocol failure",
+      reason: "protocol_failure" as const,
+      trigger: (fake: ReturnType<typeof createFakeDuplexChannel>) =>
+        fake.emitData("this is not json\n"),
+    },
+    {
+      name: "stream failure",
+      reason: "stream_failure" as const,
+      trigger: (fake: ReturnType<typeof createFakeDuplexChannel>) => {
+        fake.setWriteError(new Error("broken pipe"));
+        fake.emitData(requestFrameLine({ id: "req-stream" }));
+      },
+    },
+  ])("enters lost on $name", async ({ reason, trigger }) => {
+    const fake = createFakeDuplexChannel();
+    const losses: DuplexBrokerLossRecord[] = [];
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async () => ({ status: 200 }),
+      onLoss: (record) => losses.push(record),
+    });
+    broker.start();
+
+    trigger(fake);
+    await flushMacrotasks();
+
+    expect(broker.state).toBe("lost");
+    expect(broker.lossRecord?.reason).toBe(reason);
+    expect(losses).toHaveLength(1);
+  });
+
+  it("enters lost on a close timeout", async () => {
+    const fake = createFakeDuplexChannel();
+    fake.setCloseBehavior("hang");
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async () => ({ status: 200 }),
+      closeTimeoutMs: 20,
+    });
+    broker.start();
+
+    await broker.close();
+
+    expect(broker.state).toBe("lost");
+    expect(broker.lossRecord?.reason).toBe("close_timeout");
+  });
+
+  it("stops the heartbeat, marks the run bridge ended, and dispatches nothing after loss", async () => {
+    const fake = createFakeDuplexChannel();
+    const forwarded: string[] = [];
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async (request) => {
+        forwarded.push(request.id);
+        return { status: 200 };
+      },
+    });
+    broker.start();
+
+    fake.emitExit({ exitCode: 1 });
+    await flushMacrotasks();
+    expect(broker.state).toBe("lost");
+    expect(broker.lossRecord).not.toBeNull();
+
+    // A request that arrives after loss reaches nothing. The broker never
+    // reconnects and never replays a request.
+    fake.emitData(requestFrameLine({ id: "after-loss" }));
+    await flushMacrotasks();
+    expect(forwarded).toEqual([]);
+  });
+
+  it("forwards one request id one time, so a repeated frame never reaches the API twice", async () => {
+    const fake = createFakeDuplexChannel();
+    const forwarded: string[] = [];
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async (request) => {
+        forwarded.push(request.id);
+        return { status: 200, body: "ok" };
+      },
+    });
+    broker.start();
+
+    fake.emitData(requestFrameLine({ id: "dup" }));
+    await flushMacrotasks();
+    fake.emitData(requestFrameLine({ id: "dup" }));
+    await flushMacrotasks();
+
+    expect(forwarded).toEqual(["dup"]);
+    expect(fake.written).toHaveLength(1);
+
+    await broker.close();
+  });
+
+  it("captures the dispatch-start point per request for metrics only", async () => {
+    const fake = createFakeDuplexChannel();
+    const records: DuplexBrokerRequestRecord[] = [];
+    let clock = 1000;
+    const broker = createDuplexBridgeBroker({
+      channel: fake.channel,
+      forwardRequest: async () => ({ status: 200 }),
+      now: () => clock,
+      onRequestRecord: (record) => records.push(record),
+    });
+    broker.start();
+
+    clock = 2500;
+    fake.emitData(requestFrameLine({ id: "metric-1", method: "GET", path: "/api/agents/me" }));
+    await flushMacrotasks();
+
+    expect(records).toEqual([
+      { id: "metric-1", method: "GET", path: "/api/agents/me", dispatchStartMs: 2500 },
+    ]);
+    // The record never reaches the channel. The broker writes only a response.
+    expect(fake.writtenTypes).not.toContain("request");
+
+    await broker.close();
+  });
+
+  it("rejects a configuration where an inner budget is not smaller than its outer budget", () => {
+    expect(() =>
+      assertNestedDuplexBrokerBudgets({
+        forwardTimeoutMs: 32_000,
+        responseBudgetMs: 32_000,
+        gatewayWaitMs: 35_000,
+      }),
+    ).toThrow(/forward budget/);
+    expect(() =>
+      assertNestedDuplexBrokerBudgets({
+        forwardTimeoutMs: 30_000,
+        responseBudgetMs: 35_000,
+        gatewayWaitMs: 35_000,
+      }),
+    ).toThrow(/response budget/);
+    expect(() =>
+      createDuplexBridgeBroker({
+        channel: createFakeDuplexChannel().channel,
+        forwardRequest: async () => ({ status: 200 }),
+        budgets: { forwardTimeoutMs: 40_000 },
+      }),
+    ).toThrow(/forward budget/);
+    // The default budget set holds the nested order.
+    expect(() =>
+      assertNestedDuplexBrokerBudgets({
+        forwardTimeoutMs: 30_000,
+        responseBudgetMs: 32_000,
+        gatewayWaitMs: 35_000,
+      }),
+    ).not.toThrow();
+  });
+});

packages/adapter-utils/src/execution-target.ts

@@ -36,6 +36,7 @@ import {
   createSandboxRunLogTailFactory,
   type SandboxRunLogTailFactory,
 } from "./sandbox-run-log-stream.js";
+import { DEFAULT_DUPLEX_BROKER_BUDGETS } from "./duplex-bridge-broker.js";
 import { createSshCommandManagedRuntimeRunner, parseSshRemoteExecutionSpec, runSshCommand, shellQuote } from "./ssh.js";
 import {
   ensureCommandResolvable,
@@ -2194,6 +2195,11 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
   hostApiUrl?: string | null;
   onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
   maxBodyBytes?: number | null;
+  // The deadline for one forward call, in milliseconds. This is the inner budget
+  // of the duplex broker's nested budget set. The default is the forward budget
+  // in `DEFAULT_DUPLEX_BROKER_BUDGETS` (30 s), so the current behavior does not
+  // change when the caller sets no option.
+  forwardTimeoutMs?: number | null;
   // Return the current-run parent-context token. The factory threads it into the
   // callback bridge worker, which reads it per request so each request
   // `sandbox.exec` span parents to the live run span. When it is absent, the
@@ -2218,6 +2224,10 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
   if (hostApiToken.length === 0) {
     throw new Error("Sandbox bridge mode requires a host-side Paperclip API token.");
   }
+  // The forward budget for one relayed request. It stays at the broker's default
+  // forward budget (30 s) when the caller sets no option, so current behavior
+  // does not change.
+  const forwardTimeoutMs = input.forwardTimeoutMs ?? DEFAULT_DUPLEX_BROKER_BUDGETS.forwardTimeoutMs;
 
   const runtimeRootDir =
     input.runtimeRootDir?.trim().length
@@ -2300,10 +2310,10 @@ export async function startAdapterExecutionTargetPaperclipBridge(input: {
         headers.set("authorization", `Bearer ${hostApiToken}`);
         headers.set("x-paperclip-run-id", input.runId);
         // Abort the forward when the worker aborts the request (its per-iteration
-        // timeout or watchdog fired), or after the 30s ceiling, whichever comes
+        // timeout or watchdog fired), or after the forward budget, whichever comes
         // first. The worker abort lets the bridge fail a hung forward fast
-        // instead of stranding the request until the 30s ceiling.
-        const timeoutSignal = AbortSignal.timeout(30_000);
+        // instead of stranding the request until the forward budget.
+        const timeoutSignal = AbortSignal.timeout(forwardTimeoutMs);
         const forwardSignal = options?.signal
           ? AbortSignal.any([options.signal, timeoutSignal])
           : timeoutSignal;