mirror of
https://github.com/AmruthPillai/Reactive-Resume.git
synced 2026-08-23 23:02:17 +10:00
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
This commit is contained in:
@@ -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", () => {
|
describe("agentService.threads.archive", () => {
|
||||||
beforeEach(() => {
|
beforeEach(() => {
|
||||||
vi.clearAllMocks();
|
vi.clearAllMocks();
|
||||||
|
|||||||
@@ -41,6 +41,12 @@ const ROLLED_BACK_MESSAGE = "This patch was rolled back when the resume was rest
|
|||||||
const activeRunControllers = new Map<string, AbortController>();
|
const activeRunControllers = new Map<string, AbortController>();
|
||||||
const canceledRunsWithPersistedPartial = new Set<string>();
|
const canceledRunsWithPersistedPartial = new Set<string>();
|
||||||
|
|
||||||
|
// 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 AgentThreadRecord = typeof schema.agentThread.$inferSelect;
|
||||||
type AgentMessageRecord = typeof schema.agentMessage.$inferSelect;
|
type AgentMessageRecord = typeof schema.agentMessage.$inferSelect;
|
||||||
type AgentActionRecord = typeof schema.agentAction.$inferSelect;
|
type AgentActionRecord = typeof schema.agentAction.$inferSelect;
|
||||||
@@ -965,7 +971,7 @@ export const agentService = {
|
|||||||
const activeStreamId = thread.activeStreamId;
|
const activeStreamId = thread.activeStreamId;
|
||||||
|
|
||||||
if (activeRunId) {
|
if (activeRunId) {
|
||||||
activeRunControllers.get(activeRunId)?.abort("USER_ARCHIVED");
|
activeRunControllers.get(activeRunId)?.abort(abortReason("USER_ARCHIVED"));
|
||||||
activeRunControllers.delete(activeRunId);
|
activeRunControllers.delete(activeRunId);
|
||||||
try {
|
try {
|
||||||
await clearActiveAgentRunIfCurrent({
|
await clearActiveAgentRunIfCurrent({
|
||||||
@@ -1195,7 +1201,7 @@ export const agentService = {
|
|||||||
persistError = error;
|
persistError = error;
|
||||||
} finally {
|
} finally {
|
||||||
if (activeRunId) {
|
if (activeRunId) {
|
||||||
activeRunControllers.get(activeRunId)?.abort("USER_STOPPED");
|
activeRunControllers.get(activeRunId)?.abort(abortReason("USER_STOPPED"));
|
||||||
activeRunControllers.delete(activeRunId);
|
activeRunControllers.delete(activeRunId);
|
||||||
try {
|
try {
|
||||||
await clearActiveAgentRunIfCurrent({
|
await clearActiveAgentRunIfCurrent({
|
||||||
|
|||||||
Reference in New Issue
Block a user