From abfab29ff9e70648b833f44ccb666091e45e5e0c Mon Sep 17 00:00:00 2001 From: Varun S G Date: Sat, 5 Sep 2026 00:13:19 +0530 Subject: [PATCH 1/2] fix(agent): prevent unhandled rejection when aborting stream (#113) --- .../unhandled-rejection-stream-abort.md | 5 ++ packages/agent/src/lib/model-result.ts | 3 + .../agent/tests/unit/run-cancellation.test.ts | 60 +++++++++++++++++++ 3 files changed, 68 insertions(+) create mode 100644 .changeset/unhandled-rejection-stream-abort.md diff --git a/.changeset/unhandled-rejection-stream-abort.md b/.changeset/unhandled-rejection-stream-abort.md new file mode 100644 index 00000000..70dba835 --- /dev/null +++ b/.changeset/unhandled-rejection-stream-abort.md @@ -0,0 +1,5 @@ +--- +'@openrouter/agent': patch +--- + +Prevent unhandled promise rejection crash when aborting a tool-enabled stream. diff --git a/packages/agent/src/lib/model-result.ts b/packages/agent/src/lib/model-result.ts index bbc4f047..f57b181f 100644 --- a/packages/agent/src/lib/model-result.ts +++ b/packages/agent/src/lib/model-result.ts @@ -1098,6 +1098,9 @@ export class ModelResult< } broadcaster.complete(); }); + // Attach an immediate rejection handler so early consumer exit (such as an abort) + // does not leave executionPromise as an unhandled promise rejection. + void executionPromise.catch(() => {}); return { consumer, executionPromise, diff --git a/packages/agent/tests/unit/run-cancellation.test.ts b/packages/agent/tests/unit/run-cancellation.test.ts index 1f5ca21b..0f8b3d34 100644 --- a/packages/agent/tests/unit/run-cancellation.test.ts +++ b/packages/agent/tests/unit/run-cancellation.test.ts @@ -389,4 +389,64 @@ describe('per-request timeoutMs composition', () => { expect(text).toBe('done'); expect(mockBetaResponsesSend).toHaveBeenCalledTimes(2); }); + + it('does not emit unhandledRejection when aborting a tool-enabled stream', async () => { + const unhandledRejections: unknown[] = []; + const onUnhandled = (err: unknown) => { + unhandledRejections.push(err); + }; + process.on('unhandledRejection', onUnhandled); + + try { + const controller = new AbortController(); + let streamController!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(c) { + streamController = c; + }, + }); + + mockBetaResponsesSend.mockImplementation(async (_client, _req, options) => { + options.signal?.addEventListener('abort', () => { + streamController.error(options.signal?.reason ?? new Error('Aborted')); + }); + return { + ok: true, + value: stream, + }; + }); + + const result = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + echoTool, + ], + signal: controller.signal, + }); + + let streamError: unknown; + try { + streamController.enqueue({ + type: 'response.output_text.delta', + delta: 'chunk', + sequenceNumber: 0, + } as models.StreamEvents); + + for await (const _event of result.getFullResponsesStream()) { + controller.abort(new DOMException('This operation was aborted', 'AbortError')); + } + } catch (err) { + streamError = err; + } + + expect(streamError).toBeDefined(); + + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(unhandledRejections).toHaveLength(0); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); }); From 5ce0c6754135207be12a25ccbcb3272285ab45ae Mon Sep 17 00:00:00 2001 From: Luke Parke <5702154+LukasParke@users.noreply.github.com> Date: Wed, 21 Oct 2026 17:57:11 +0000 Subject: [PATCH 2/2] test(agent): abort matrix for unhandled rejections across stream getters (#113) Covers every stream getter, abort during client tool execution, early break then abort, parallel runs sharing one signal, and getText with and without tools. 7 of the 11 cases fail without the #122 fix; #122's own test covers only getFullResponsesStream. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../stream-abort-unhandled-rejection.test.ts | 329 ++++++++++++++++++ 1 file changed, 329 insertions(+) create mode 100644 packages/agent/tests/unit/stream-abort-unhandled-rejection.test.ts diff --git a/packages/agent/tests/unit/stream-abort-unhandled-rejection.test.ts b/packages/agent/tests/unit/stream-abort-unhandled-rejection.test.ts new file mode 100644 index 00000000..3573a047 --- /dev/null +++ b/packages/agent/tests/unit/stream-abort-unhandled-rejection.test.ts @@ -0,0 +1,329 @@ +/** + * Abort matrix for #113: aborting a run through any consumer surface must + * reject that consumer and must NOT leave a dangling rejected promise behind + * (which surfaces as a process-level `unhandledRejection`). + * + * The per-getter cases matter because every stream getter shares the same + * broadcaster/execution path; a regression there shows up on all of them, + * not only on `getFullResponsesStream`. + */ +import type { OpenRouterCore } from '@openrouter/sdk/core'; +import type * as models from '@openrouter/sdk/models'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { z } from 'zod/v4'; + +const mockBetaResponsesSend = vi.hoisted(() => vi.fn()); + +vi.mock('@openrouter/sdk/funcs/betaResponsesSend', () => ({ + betaResponsesSend: mockBetaResponsesSend, +})); + +import { callModel } from '../../src/inner-loop/call-model.js'; +import { tool } from '../../src/lib/tool.js'; + +const client = { + _options: {}, +} as OpenRouterCore; + +const echoTool = tool({ + name: 'echo', + inputSchema: z.object({ + value: z.string(), + }), + execute: async ({ value }) => ({ + echoed: value, + }), +}); + +const textDelta = { + type: 'response.output_text.delta', + delta: 'chunk', + sequenceNumber: 0, +} as models.StreamEvents; + +function abortError(): DOMException { + return new DOMException('This operation was aborted', 'AbortError'); +} + +/** A stream that emits `events` and then errors with the signal's reason on abort. */ +function abortableStream( + signal: AbortSignal | undefined, + events: models.StreamEvents[], +): ReadableStream { + return new ReadableStream({ + start(c) { + for (const e of events) { + c.enqueue(e); + } + signal?.addEventListener('abort', () => { + c.error(signal.reason ?? new Error('Aborted')); + }); + }, + }); +} + +function mockAbortableSend( + events: models.StreamEvents[] = [ + textDelta, + ], +): void { + mockBetaResponsesSend.mockImplementation( + async ( + _client: unknown, + _req: unknown, + options: { + signal?: AbortSignal; + }, + ) => ({ + ok: true, + value: abortableStream(options.signal, events), + }), + ); +} + +function completedWithToolCall(name: string): models.StreamEvents { + return { + type: 'response.completed', + sequenceNumber: 1, + response: { + id: 'resp_tool', + object: 'response', + createdAt: 0, + model: 'test-model', + status: 'completed', + completedAt: 0, + output: [ + { + type: 'function_call', + id: 'fc_1', + callId: 'call_1', + name, + arguments: '{}', + status: 'completed', + }, + ], + error: null, + incompleteDetails: null, + temperature: null, + topP: null, + presencePenalty: null, + frequencyPenalty: null, + metadata: null, + instructions: null, + tools: [], + toolChoice: 'auto', + parallelToolCalls: false, + }, + } as unknown as models.StreamEvents; +} + +/** Let any dangling rejections reach the process-level handler. */ +const settle = () => new Promise((resolve) => setTimeout(resolve, 50)); + +let unhandledRejections: unknown[] = []; +const onUnhandled = (err: unknown) => { + unhandledRejections.push(err); +}; + +beforeEach(() => { + mockBetaResponsesSend.mockReset(); + unhandledRejections = []; + process.on('unhandledRejection', onUnhandled); +}); + +afterEach(() => { + process.off('unhandledRejection', onUnhandled); +}); + +const streamGetters = [ + 'getFullResponsesStream', + 'getTextStream', + 'getItemsStream', + 'getReasoningStream', + 'getToolStream', + 'getToolCallsStream', + 'getNewMessagesStream', +] as const; + +describe('aborting a tool-enabled run does not leak unhandled rejections (#113)', () => { + for (const getter of streamGetters) { + it(`${getter}: abort mid-stream rejects the consumer only`, async () => { + const controller = new AbortController(); + mockAbortableSend(); + // Abort once the request has been dispatched and the stream is live. + mockBetaResponsesSend.mockImplementationOnce( + async ( + _client: unknown, + _req: unknown, + options: { + signal?: AbortSignal; + }, + ) => { + setTimeout(() => controller.abort(abortError()), 0); + return { + ok: true, + value: abortableStream(options.signal, [ + textDelta, + ]), + }; + }, + ); + + const result = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + echoTool, + ], + signal: controller.signal, + }); + + let streamError: unknown; + try { + const stream: AsyncIterable = result[getter](); + for await (const _event of stream) { + // drain + } + } catch (err) { + streamError = err; + } + await settle(); + + expect(streamError).toBeDefined(); + expect(unhandledRejections).toHaveLength(0); + }); + } + + it('getItemsStream: abort while a client tool is executing', async () => { + const controller = new AbortController(); + const slowTool = tool({ + name: 'slow', + inputSchema: z.object({}), + execute: async () => { + controller.abort(abortError()); + await new Promise((resolve) => setTimeout(resolve, 20)); + return { + ok: true, + }; + }, + }); + mockAbortableSend([ + textDelta, + completedWithToolCall('slow'), + ]); + + const result = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + slowTool, + ], + signal: controller.signal, + }); + + let streamError: unknown; + try { + for await (const _item of result.getItemsStream()) { + // drain + } + } catch (err) { + streamError = err; + } + await settle(); + + expect(streamError).toBeDefined(); + expect(unhandledRejections).toHaveLength(0); + }); + + it('getFullResponsesStream: early break, then abort', async () => { + const controller = new AbortController(); + mockAbortableSend(); + + const result = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + echoTool, + ], + signal: controller.signal, + }); + + for await (const _event of result.getFullResponsesStream()) { + break; + } + controller.abort(abortError()); + await settle(); + + expect(unhandledRejections).toHaveLength(0); + }); + + it('parallel runs sharing one signal: every consumer rejects, nothing leaks', async () => { + const controller = new AbortController(); + mockAbortableSend(); + + const runOne = async (): Promise => { + const result = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + echoTool, + ], + signal: controller.signal, + }); + try { + for await (const _delta of result.getTextStream()) { + // drain + } + } catch (err) { + return err; + } + return undefined; + }; + + const runs = [ + runOne(), + runOne(), + runOne(), + ]; + await vi.waitFor(() => expect(mockBetaResponsesSend).toHaveBeenCalledTimes(3)); + controller.abort(abortError()); + const errors = await Promise.all(runs); + await settle(); + + expect(errors.every((e) => e !== undefined)).toBe(true); + expect(unhandledRejections).toHaveLength(0); + }); + + it('getText: abort rejects with and without tools', async () => { + const controller = new AbortController(); + mockAbortableSend(); + + const withTools = callModel(client, { + model: 'test-model', + input: 'hello', + tools: [ + echoTool, + ], + signal: controller.signal, + }).getText(); + const withoutTools = callModel(client, { + model: 'test-model', + input: 'hello', + signal: controller.signal, + }).getText(); + + await vi.waitFor(() => expect(mockBetaResponsesSend).toHaveBeenCalledTimes(2)); + controller.abort(abortError()); + const settled = await Promise.allSettled([ + withTools, + withoutTools, + ]); + await settle(); + + expect(settled.map((s) => s.status)).toEqual([ + 'rejected', + 'rejected', + ]); + expect(unhandledRejections).toHaveLength(0); + }); +});