From 04100aa9efd3ef4332fec9ed9052686fdf3bbeb6 Mon Sep 17 00:00:00 2001 From: Amruth Pillai Date: Sun, 9 Aug 2026 15:46:31 +0200 Subject: [PATCH] fix(agent): abort stopped runs with an AbortError, not a bare string Stopping (or archiving) an agent run called controller.abort("USER_STOPPED") with a plain string reason. The AI SDK only recognizes a cancellation when the reason is an AbortError (err.name === "AbortError" / isAbortError); a bare string is treated as a real stream error, and its rejection escaped the background resumable-stream pump and crashed the whole server process with ERR_UNHANDLED_REJECTION on every user Stop. Abort with a DOMException named AbortError (label preserved as the message) so the SDK cancels the run gracefully. Same fix for the USER_ARCHIVED path. Claude-Session: https://claude.ai/code/session_01ULhhLQ24DvnYwzP4afDuye --- .../api/src/features/agent/service.test.ts | 84 +++++++++++++++++++ packages/api/src/features/agent/service.ts | 10 ++- 2 files changed, 92 insertions(+), 2 deletions(-) diff --git a/packages/api/src/features/agent/service.test.ts b/packages/api/src/features/agent/service.test.ts index 633e1aae4..f269d17db 100644 --- a/packages/api/src/features/agent/service.test.ts +++ b/packages/api/src/features/agent/service.test.ts @@ -1038,6 +1038,90 @@ describe("agentService.messages.send", () => { }); }); +describe("agentService.messages.stop", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + // Regression: stop() must abort the run with an AbortError. A bare-string abort reason is not + // recognized by the AI SDK as a cancellation, so its rejection escapes the background stream + // pump and crashes the whole process with ERR_UNHANDLED_REJECTION. + it("aborts the active run with an AbortError the AI SDK recognizes as a cancellation", async () => { + const activeThread = buildActiveThread(); + const persistedMessage = { + id: "message-1", + userId: "user-1", + threadId: "thread-1", + role: "user", + status: "completed", + sequence: 0, + uiMessage: { id: "ui-message-1", role: "user", parts: [{ type: "text", text: "hi" }] }, + }; + + dbMock.select + // send(): getThread, next sequence, message count, thread messages + .mockImplementationOnce(() => selectLimitResult([activeThread])) + .mockImplementationOnce(() => selectWhereResult([{ maxSequence: -1 }])) + .mockImplementationOnce(() => selectWhereResult([{ total: 1 }])) + .mockImplementationOnce(() => selectOrderByResult([persistedMessage])) + // stop(): getThread now reports the active run registered by send() (generateId() -> "test-id") + .mockImplementationOnce(() => + selectLimitResult([buildActiveThread({ activeRunId: "test-id", activeStreamId: "test-id" })]), + ); + + dbMock.insert.mockReturnValue({ + values: vi.fn(() => ({ returning: vi.fn(async () => [persistedMessage]) })), + }); + dbMock.update.mockReturnValue({ set: vi.fn(() => ({ where: vi.fn(async () => undefined) })) }); + + claimActiveAgentRunMock.mockResolvedValue(true); + clearActiveAgentRunIfCurrentMock.mockResolvedValue(undefined); + aiProvidersServiceMock.getRunnableById.mockResolvedValue({ + id: "provider-1", + provider: "openai", + model: "gpt-5", + apiKey: "secret", + baseURL: null, + }); + aiProvidersServiceMock.markUsed.mockResolvedValue(undefined); + + const [{ convertToModelMessages, ToolLoopAgent }, { agentStreamLifecycle }, { streamToEventIterator }] = + await Promise.all([import("ai"), import("./streams"), import("@orpc/server")]); + vi.mocked(convertToModelMessages).mockResolvedValue([{ role: "user", content: [{ type: "text", text: "hi" }] }]); + + let capturedSignal: AbortSignal | undefined; + class MockToolLoopAgent { + stream = vi.fn(({ abortSignal }: { abortSignal: AbortSignal }) => { + capturedSignal = abortSignal; + return { toUIMessageStream: vi.fn(() => new ReadableStream()) }; + }); + } + vi.mocked(ToolLoopAgent).mockImplementation(MockToolLoopAgent as never); + vi.mocked(agentStreamLifecycle.create).mockResolvedValue(new ReadableStream()); + vi.mocked(streamToEventIterator).mockReturnValue("iterator" as never); + + const { agentService } = await import("./service"); + + await agentService.messages.send({ + threadId: "thread-1", + userId: "user-1", + // biome-ignore lint/suspicious/noExplicitAny: minimal fixture for unit test + message: { id: "ui-message-1", role: "user", parts: [{ type: "text", text: "hi" }] } as any, + }); + + expect(capturedSignal).toBeDefined(); + expect(capturedSignal?.aborted).toBe(false); + + await agentService.messages.stop({ userId: "user-1", threadId: "thread-1" }); + + expect(capturedSignal?.aborted).toBe(true); + const reason = capturedSignal?.reason as Error; + expect(reason).toBeInstanceOf(Error); + expect(reason.name).toBe("AbortError"); + expect(reason.message).toBe("USER_STOPPED"); + }); +}); + describe("agentService.threads.archive", () => { beforeEach(() => { vi.clearAllMocks(); diff --git a/packages/api/src/features/agent/service.ts b/packages/api/src/features/agent/service.ts index 6e0b35537..fccf35818 100644 --- a/packages/api/src/features/agent/service.ts +++ b/packages/api/src/features/agent/service.ts @@ -41,6 +41,12 @@ const ROLLED_BACK_MESSAGE = "This patch was rolled back when the resume was rest const activeRunControllers = new Map(); const canceledRunsWithPersistedPartial = new Set(); +// Abort reasons MUST be an AbortError: the AI SDK only treats `err.name === "AbortError"` +// (via isAbortError) as a cancellation. A bare-string reason is treated as a genuine stream +// error whose rejection escapes the background (resumable-stream) pump and takes down the whole +// process with ERR_UNHANDLED_REJECTION. The label is preserved as the DOMException message. +const abortReason = (label: string) => new DOMException(label, "AbortError"); + type AgentThreadRecord = typeof schema.agentThread.$inferSelect; type AgentMessageRecord = typeof schema.agentMessage.$inferSelect; type AgentActionRecord = typeof schema.agentAction.$inferSelect; @@ -965,7 +971,7 @@ export const agentService = { const activeStreamId = thread.activeStreamId; if (activeRunId) { - activeRunControllers.get(activeRunId)?.abort("USER_ARCHIVED"); + activeRunControllers.get(activeRunId)?.abort(abortReason("USER_ARCHIVED")); activeRunControllers.delete(activeRunId); try { await clearActiveAgentRunIfCurrent({ @@ -1195,7 +1201,7 @@ export const agentService = { persistError = error; } finally { if (activeRunId) { - activeRunControllers.get(activeRunId)?.abort("USER_STOPPED"); + activeRunControllers.get(activeRunId)?.abort(abortReason("USER_STOPPED")); activeRunControllers.delete(activeRunId); try { await clearActiveAgentRunIfCurrent({