diff --git a/src/agent/runtime/index.ts b/src/agent/runtime/index.ts index 56b1551cc5..af85a7320e 100644 --- a/src/agent/runtime/index.ts +++ b/src/agent/runtime/index.ts @@ -849,6 +849,14 @@ export class AgentRuntime { }; const chain = new MiddlewareChain(this.config.middleware); + // Hold the in-flight agent-loop promise so stream cancellation can detach a + // no-op rejection handler. When the client cancels, we abort the shared + // signal; the loop (model fetch / tool execution) then rejects with an + // AbortError. The `start` body awaits it, but cancellation can land after + // that await settles, leaving the rejection without a consumer — fatal as + // an unhandled rejection under Deno (#2334). + let inFlight: Promise | undefined; + return new ReadableStream({ start: async (controller) => { try { @@ -870,7 +878,7 @@ export class AgentRuntime { model: effectiveModel, }, }); - const response = await chain.execute( + inFlight = chain.execute( agentContext, () => this.executeAgentLoopStreaming( @@ -890,6 +898,7 @@ export class AgentRuntime { streamAbortSignal, ), ); + const response = await inFlight; throwIfAborted(streamAbortSignal); callbacks?.onFinish?.(response); throwIfAborted(streamAbortSignal); @@ -914,7 +923,18 @@ export class AgentRuntime { } }, cancel(reason) { - streamAbortController.abort(reason); + // The client disconnected (e.g. the Chat Stop button). Treat this as a + // clean stop: detach a no-op handler from the in-flight loop so the + // AbortError it throws when we abort the shared signal cannot surface as + // an unhandled rejection, then abort. Guard the abort itself so a + // synchronous signal-abort rejection can never escape here (#2334). + inFlight?.catch(() => {}); + try { + streamAbortController.abort(reason); + } catch { + // Aborting an already-aborted controller, or a synchronous reject + // from a signal consumer, is a no-op for cancellation purposes. + } }, }); } diff --git a/src/agent/runtime/runtime-stream-cancel.test.ts b/src/agent/runtime/runtime-stream-cancel.test.ts new file mode 100644 index 0000000000..02e77f2d68 --- /dev/null +++ b/src/agent/runtime/runtime-stream-cancel.test.ts @@ -0,0 +1,164 @@ +import "#veryfront/schemas/_test-setup.ts"; +import { assert } from "#veryfront/testing/assert.ts"; +import { describe, it } from "#veryfront/testing/bdd.ts"; +import { type ModelRuntime } from "#veryfront/provider"; +import { tool } from "#veryfront/tool"; +import { defineSchema } from "#veryfront/schemas/index.ts"; +import { agent } from "../index.ts"; + +/** + * Regression coverage for #2334: cancelling an in-flight agent run must be + * treated as a clean stop, not surface as an uncaught `AbortError`. + * + * The reproduction cancels the response body's reader (exactly what Deno's HTTP + * server does when the client disconnects / hits the Chat "Stop" button) while + * the model stream — and a tool execution — are still in flight. Before the fix + * the runtime's stream `cancel` aborted the shared signal with the client's + * foreign reason, and the resulting rejection propagated with no handler, + * crashing the process under Deno. Deno's test runner fails on any unhandled + * rejection, so these tests fail loudly if the regression returns. + */ + +function flushMicrotasks(): Promise { + return new Promise((resolve) => setTimeout(resolve, 20)); +} + +/** A model stream that stays open until the run is aborted, then rejects its + * pending read with the abort reason — mirroring a real provider fetch body. */ +function createPendingModelStream(abortSignal: AbortSignal | undefined): ReadableStream { + return new ReadableStream({ + start(controller) { + controller.enqueue({ type: "text-start", id: "t" }); + controller.enqueue({ type: "text-delta", id: "t", delta: "thinking" }); + + if (!abortSignal) { + return; + } + if (abortSignal.aborted) { + controller.error(abortSignal.reason); + return; + } + abortSignal.addEventListener("abort", () => { + controller.error(abortSignal.reason); + }, { once: true }); + }, + }); +} + +describe("agent runtime stream cancellation (#2334)", () => { + it("cancelling a model-streaming run does not raise an unhandled AbortError", async () => { + const model: ModelRuntime = { + provider: "hosted", + modelId: "hosted/cancel-crash-model", + async doGenerate() { + return { + content: [], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream(options: unknown) { + const abortSignal = (options as { abortSignal?: AbortSignal }).abortSignal; + return { stream: createPendingModelStream(abortSignal) }; + }, + }; + + const assistant = agent({ + model: "hosted/cancel-crash-model", + system: "cancel crash test", + maxSteps: 1, + resolveModelTransport: async () => ({ model }), + }); + + const response = (await assistant.stream({ input: "hi" })).toDataStreamResponse(); + const body = response.body; + assert(body !== null, "expected a streaming response body"); + + const reader = body.getReader(); + // Pull the opening frames so the run is genuinely mid-stream. + await reader.read(); + // The client disconnects: cancel with a foreign AbortError reason, exactly + // as Deno hands to the stream's cancel algorithm. + await reader.cancel(new DOMException("client disconnected", "AbortError")); + + await flushMicrotasks(); + assert(true, "cancellation completed without an unhandled rejection"); + }); + + it("cancelling while a tool is executing does not raise an unhandled AbortError", async () => { + let releaseTool: (() => void) | undefined; + const toolStarted = Promise.withResolvers(); + + const slowTool = tool({ + id: "slow_tool", + description: "A tool that stays in flight until the run is cancelled", + inputSchema: defineSchema((v) => v.object({}))(), + execute: async (_input, context) => { + toolStarted.resolve(); + const abortSignal = (context as { abortSignal?: AbortSignal })?.abortSignal; + await new Promise((resolve) => { + releaseTool = resolve; + abortSignal?.addEventListener("abort", () => resolve(), { once: true }); + }); + return { ok: true }; + }, + }); + + let call = 0; + const model: ModelRuntime = { + provider: "hosted", + modelId: "hosted/cancel-crash-tool", + async doGenerate() { + return { + content: [], + finishReason: "stop", + usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 }, + }; + }, + async doStream(options: unknown) { + call += 1; + const abortSignal = (options as { abortSignal?: AbortSignal }).abortSignal; + if (call === 1) { + // First step: emit a tool call so a tool execution opens. + return { + stream: new ReadableStream({ + start(controller) { + controller.enqueue({ + type: "tool-call", + toolCallId: "slow-1", + toolName: "slow_tool", + input: "{}", + }); + controller.enqueue({ type: "finish", finishReason: "tool-calls" }); + controller.close(); + }, + }), + }; + } + // Any later step stays open until aborted. + return { stream: createPendingModelStream(abortSignal) }; + }, + }; + + const assistant = agent({ + model: "hosted/cancel-crash-tool", + system: "cancel crash tool test", + tools: { slow_tool: slowTool }, + maxSteps: 3, + resolveModelTransport: async () => ({ model }), + }); + + const response = (await assistant.stream({ input: "run the tool" })).toDataStreamResponse(); + const body = response.body; + assert(body !== null, "expected a streaming response body"); + + const reader = body.getReader(); + await reader.read(); + await toolStarted.promise; + await reader.cancel(new DOMException("client disconnected", "AbortError")); + releaseTool?.(); + + await flushMicrotasks(); + assert(true, "cancellation during tool execution completed cleanly"); + }); +}); diff --git a/src/agent/streaming/tool-execution-data-event-bridge.test.ts b/src/agent/streaming/tool-execution-data-event-bridge.test.ts index 1173e433b6..a4bfda26fc 100644 --- a/src/agent/streaming/tool-execution-data-event-bridge.test.ts +++ b/src/agent/streaming/tool-execution-data-event-bridge.test.ts @@ -93,4 +93,51 @@ describe("createToolExecutionDataEventBridgeStream", () => { controller.close(); await reader.cancel(); }); + + it("cancel resolves cleanly when the base reader cancel rejects (#2334)", async () => { + // Mirrors the production crash: the upstream agent runtime's stream cancel + // aborts an in-flight signal, and the rejection propagates back through the + // base reader's cancel. The bridge must absorb it so cancellation does not + // escape as an unhandled rejection (fatal under Deno). + const stream = createToolExecutionDataEventBridgeStream({ + baseStream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"type":"message-start"}\n\n')); + }, + cancel() { + throw new DOMException("The signal has been aborted", "AbortError"); + }, + }), + installPublisher() {}, + }); + + const reader = stream.getReader(); + await reader.read(); + + // Must not reject — before the fix this surfaced the base reader's + // AbortError to the (often un-awaiting) consumer. + await reader.cancel(new DOMException("client disconnected", "AbortError")); + }); + + it("cancel still forwards the reason to the base reader on the happy path", async () => { + let cancelledWith: unknown = "unset"; + const stream = createToolExecutionDataEventBridgeStream({ + baseStream: new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('data: {"type":"message-start"}\n\n')); + }, + cancel(reason) { + cancelledWith = reason; + }, + }), + installPublisher() {}, + }); + + const reader = stream.getReader(); + await reader.read(); + const reason = new DOMException("client disconnected", "AbortError"); + await reader.cancel(reason); + + assertEquals(cancelledWith, reason); + }); }); diff --git a/src/agent/streaming/tool-execution-data-event-bridge.ts b/src/agent/streaming/tool-execution-data-event-bridge.ts index 7e6c2dd9d5..39a5455a49 100644 --- a/src/agent/streaming/tool-execution-data-event-bridge.ts +++ b/src/agent/streaming/tool-execution-data-event-bridge.ts @@ -83,7 +83,16 @@ export function createToolExecutionDataEventBridgeStream( })(); }, async cancel(reason) { - await baseReader?.cancel(reason); + // Cancellation is best-effort teardown (the client disconnected / hit + // Stop). Forwarding the cancel to the base reader can reject — e.g. the + // upstream agent runtime aborts an in-flight signal whose rejection + // surfaces through the cancel chain. Swallow it so it does not escape as + // an unhandled rejection, which is fatal under Deno (#2334). + try { + await baseReader?.cancel(reason); + } catch { + // Stream is being torn down; a failed cancel is a clean stop here. + } }, }); }