diff --git a/src/adapters/openai-chat.ts b/src/adapters/openai-chat.ts index a637747cef..80eacb0c1e 100644 --- a/src/adapters/openai-chat.ts +++ b/src/adapters/openai-chat.ts @@ -1743,6 +1743,21 @@ export function createOpenAIChatAdapter(provider: OcxProviderConfig): ProviderAd if (idDelta && !call.id) call.id = idDelta; if (typeof rawName === "string" && rawName && !call.name) call.name = rawName; if (typeof rawArguments === "string") call.sawArgumentsString = true; + // Tool-call deltas are BUFFERED until a terminal signal, so this adapter can + // consume upstream frames for a long time while yielding nothing. The Responses + // bridge reads adapter activity, not socket activity, so a model that streams a + // large argument payload looks identical to a hung upstream and the stall + // watchdog can abort a turn that was progressing normally. + // + // Found while investigating #2156, but it is NOT that bug: a stall abort emits + // `response.incomplete` with `upstream_stall_timeout` from the bridge, whereas + // that report shows the adapter's own end-of-stream error after `reader.read()` + // returned EOF with tool calls still pending. Different path, different frame. + // + // A heartbeat is invisible downstream — the bridge consumes it to re-arm the + // watchdog and emits nothing — which is the same remedy the Cursor, Anthropic, + // Google, and Kiro adapters already use for their own silent phases. + yield { type: "heartbeat" }; if (typeof rawArguments === "string" && rawArguments) { const previousBytes = call.argsBytes; const nextBytes = previousBytes + budgetEncoder.encode(rawArguments).byteLength; diff --git a/src/server/responses/terminal-guard.ts b/src/server/responses/terminal-guard.ts index 22865c7274..347a489f19 100644 --- a/src/server/responses/terminal-guard.ts +++ b/src/server/responses/terminal-guard.ts @@ -193,6 +193,16 @@ export async function* guardTerminalEventStream(options: GuardedEventStreamOptio const seen: AdapterEvent[] = []; let terminalSeen = false; for await (const event of source) { + // A heartbeat is adapter liveness, not turn content: it exists so the bridge watchdog + // can tell a buffering adapter from a hung one. Retaining it here would put an + // unbounded number of empty markers into `seen`, which feeds both the continuation + // analysis and the rebuilt request — and the openai-chat adapter now emits one per + // tool-call delta, so a long argument payload alone could grow this array without + // limit. The empty-completion guard already passes them through unretained; match it. + if (event.type === "heartbeat") { + yield event; + continue; + } if (event.type === "done") { terminalSeen = true; const analysis = (options.adapterName === "anthropic" || options.adapterName === "openai-chat") diff --git a/tests/openai-chat-eof.test.ts b/tests/openai-chat-eof.test.ts index f87ff4c89f..66eb211b71 100644 --- a/tests/openai-chat-eof.test.ts +++ b/tests/openai-chat-eof.test.ts @@ -11,7 +11,10 @@ const provider = { adapter: "openai-chat", baseUrl: "https://example.test/v1", a async function collect(gen: AsyncGenerator): Promise { const out: AdapterEvent[] = []; - for await (const e of gen) out.push(e); + // Heartbeats are invisible downstream: the bridge consumes them to re-arm its stall + // watchdog and emits nothing. Dropping them here keeps these assertions about the wire + // the client actually sees (#2156). + for await (const e of gen) if (e.type !== "heartbeat") out.push(e); return out; } diff --git a/tests/openai-chat-hardening.test.ts b/tests/openai-chat-hardening.test.ts index 81671b0968..e3ead6e28d 100644 --- a/tests/openai-chat-hardening.test.ts +++ b/tests/openai-chat-hardening.test.ts @@ -40,7 +40,10 @@ function provider(overrides: Partial = {}): OcxProviderConfig async function collect(stream: AsyncGenerator): Promise { const events: AdapterEvent[] = []; - for await (const event of stream) events.push(event); + // Heartbeats are invisible downstream: the bridge consumes them to re-arm its stall + // watchdog and emits nothing. Dropping them here keeps these assertions about the wire + // the client actually sees. + for await (const event of stream) if (event.type !== "heartbeat") events.push(event); return events; } @@ -920,4 +923,33 @@ describe("openai-chat response_format emission", () => { .toEqual({ type: "json_object" }); }); }); + +// Tool-call deltas are BUFFERED until a terminal signal, so this adapter can consume upstream +// frames for a long time while yielding nothing downstream. The Responses bridge arms its +// stall watchdog on ADAPTER activity, not socket activity, so a model streaming a large +// argument payload was indistinguishable from a hung upstream. +// +// Found while investigating #2156 but deliberately NOT claimed as its fix: a stall abort +// emits `response.incomplete` with `upstream_stall_timeout`, while that report shows the +// adapter's own EOF error with tool calls still pending. This pins the mechanism only. +test("tool-call deltas emit heartbeats so a long buffering phase is not read as a stall", async () => { + const adapter = createOpenAIChatAdapter(provider()); + const frames = ['data: ' + JSON.stringify({ choices: [{ delta: { tool_calls: [{ index: 0, id: "call_a", function: { name: "shell", arguments: "" } }] } }] }) + '\n\n']; + // Many argument chunks and nothing else: exactly the shape that looked like silence. + for (let i = 0; i < 12; i += 1) { + frames.push('data: ' + JSON.stringify({ choices: [{ delta: { tool_calls: [{ index: 0, function: { arguments: '"x"' } }] } }] }) + '\n\n'); + } + frames.push('data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}]}\n\n', "data: [DONE]\n\n"); + + const raw: AdapterEvent[] = []; + for await (const event of adapter.parseStream(new Response(frames.join("")))) raw.push(event); + + // One per consumed tool-call delta: the watchdog sees activity for the whole phase. + expect(raw.filter(e => e.type === "heartbeat").length).toBeGreaterThanOrEqual(12); + // And the client-visible wire is unchanged -- a heartbeat is consumed by the bridge. + const visible = raw.filter(e => e.type !== "heartbeat"); + expect(visible.some(e => e.type === "error")).toBe(false); + expect(visible).toContainEqual({ type: "tool_call_start", id: "call_a", name: "shell" }); + expect(visible.at(-1)).toMatchObject({ type: "done" }); +}); }); diff --git a/tests/terminal-guard.test.ts b/tests/terminal-guard.test.ts index 9fe5b68d4d..906c726997 100644 --- a/tests/terminal-guard.test.ts +++ b/tests/terminal-guard.test.ts @@ -188,6 +188,54 @@ describe("terminal guard", () => { expect(actual.at(-1)).toMatchObject({ usage: { inputTokens: 30, outputTokens: 5, totalTokens: 35 } }); }); + + // A heartbeat is adapter liveness, not turn content. The openai-chat adapter emits one per + // tool-call delta while it buffers, so retaining them here would let a single large argument + // payload grow `seen` without bound — and `seen` is what both the continuation analysis and + // the rebuilt request read. Passing them through unretained is what the empty-completion + // guard already does. + // A heartbeat is adapter liveness, not turn content. The openai-chat adapter emits one per + // tool-call delta while it buffers, so retaining them would grow the guard's record without + // bound on a large argument payload. `analyzeTerminalTurn` and `buildContinuationRequest` + // both read that record, so pin the contract on the pure functions that consume it plus the + // observable passthrough. + test("a retained heartbeat would corrupt the continuation record", () => { + const clean: AdapterEvent[] = [ + { type: "text_delta", text: "我接下来会修改相关文件。" }, + ]; + const padded: AdapterEvent[] = [ + { type: "text_delta", text: "我接下来会修改相关文件。" }, + ...Array.from({ length: 50 }, () => ({ type: "heartbeat" }) as AdapterEvent), + ]; + const request = parsed("继续检查"); + // The guard must not let liveness markers change what the continuation decides or sends. + expect(analyzeTerminalTurn(request, padded).assistantText) + .toBe(analyzeTerminalTurn(request, clean).assistantText); + expect(JSON.stringify(buildContinuationRequest(request, padded).context.messages)) + .toBe(JSON.stringify(buildContinuationRequest(request, clean).context.messages)); + }); + + test("heartbeats reach the consumer so the bridge watchdog stays armed", async () => { + const actual: AdapterEvent[] = []; + for await (const event of guardTerminalEventStream({ + parsed: parsed("继续检查"), + firstEvents: (async function* () { + yield { type: "text_delta", text: "我接下来会修改相关文件。" } as AdapterEvent; + for (let i = 0; i < 50; i++) yield { type: "heartbeat" } as AdapterEvent; + yield { type: "tool_call_start", id: "call_1", name: "exec_command" } as AdapterEvent; + yield { type: "tool_call_end" } as AdapterEvent; + yield { type: "done", usage: { inputTokens: 10, outputTokens: 2 } } as AdapterEvent; + })(), + continuation: () => (async function* () { + yield { type: "done" } as AdapterEvent; + })(), + adapterName: "openai-chat", + })) actual.push(event); + + expect(actual.filter(event => event.type === "heartbeat")).toHaveLength(50); + expect(actual.filter(event => event.type === "done")).toHaveLength(1); + }); + test("stops after the configured continuation bound", async () => { let continuations = 0; const actual: AdapterEvent[] = [];