diff --git a/packages/adapter-utils/src/duplex-bridge-broker.ts b/packages/adapter-utils/src/duplex-bridge-broker.ts new file mode 100644 index 00000000000..fa615e3bae1 --- /dev/null +++ b/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; + 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; + +/** 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; + /** 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; + /** 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; + responseTimer: ReturnType; +} + +/** + * 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 | null = null; + let lossRecord: DuplexBrokerLossRecord | null = null; + let heartbeatTimer: ReturnType | 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(); + const pending = new Map(); + + 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 => { + 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 | undefined; + const timeout = new Promise((_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, + }; +} diff --git a/packages/adapter-utils/src/execution-target-sandbox.test.ts b/packages/adapter-utils/src/execution-target-sandbox.test.ts index 75756182a51..657bb76de4e 100644 --- a/packages/adapter-utils/src/execution-target-sandbox.test.ts +++ b/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 { + if (closeBehavior === "resolve") return Promise.resolve(); + if (closeBehavior === "reject") return Promise.reject(new Error("close rejected")); + return new Promise(() => {}); + }, + }; + + 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 & { 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 { + 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 => { + 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) => + fake.emitExit({ exitCode: 1 }), + }, + { + name: "protocol failure", + reason: "protocol_failure" as const, + trigger: (fake: ReturnType) => + fake.emitData("this is not json\n"), + }, + { + name: "stream failure", + reason: "stream_failure" as const, + trigger: (fake: ReturnType) => { + 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(); + }); +}); diff --git a/packages/adapter-utils/src/execution-target.ts b/packages/adapter-utils/src/execution-target.ts index 8c1aba36343..f6dc32a4f92 100644 --- a/packages/adapter-utils/src/execution-target.ts +++ b/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; 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;