diff --git a/packages/ai/src/chatTransport.test.ts b/packages/ai/src/chatTransport.test.ts index ba5f956d4..750ff7110 100644 --- a/packages/ai/src/chatTransport.test.ts +++ b/packages/ai/src/chatTransport.test.ts @@ -1359,6 +1359,58 @@ describe("TriggerChatTransport", function () { expect(errors[0]?.error.message).toBe("stream subscribe root cause"); }); + it("preserves stream subscribe failures when cleanup run-store delete throws", async function () { + const errors: TriggerChatTransportError[] = []; + const runStore = new FailingCleanupDeleteRunStore(1); + + const server = await startServer(function (req, res) { + if (req.method === "POST" && req.url === "/api/v1/tasks/chat-task/trigger") { + res.writeHead(200, { + "content-type": "application/json", + "x-trigger-jwt": "pk_stream_subscribe_cleanup_delete_failure", + }); + res.end(JSON.stringify({ id: "run_stream_subscribe_cleanup_delete_failure" })); + return; + } + + res.writeHead(404); + res.end(); + }); + + const transport = new TriggerChatTransport({ + task: "chat-task", + stream: "chat-stream", + accessToken: "pk_trigger", + baseURL: server.url, + runStore, + onError: function onError(error) { + errors.push(error); + }, + }); + + (transport as any).fetchRunStream = async function fetchRunStream() { + throw new Error("stream subscribe root cause"); + }; + + await expect( + transport.sendMessages({ + trigger: "submit-message", + chatId: "chat-stream-subscribe-cleanup-delete-failure", + messageId: undefined, + messages: [], + abortSignal: undefined, + }) + ).rejects.toThrowError("stream subscribe root cause"); + + expect(errors).toHaveLength(1); + expect(errors[0]).toMatchObject({ + phase: "streamSubscribe", + chatId: "chat-stream-subscribe-cleanup-delete-failure", + runId: "run_stream_subscribe_cleanup_delete_failure", + }); + expect(errors[0]?.error.message).toBe("stream subscribe root cause"); + }); + it("cleans up async run-store state when stream subscription fails", async function () { const runStore = new AsyncTrackedRunStore(); @@ -1992,6 +2044,46 @@ describe("TriggerChatTransport", function () { expect(errors[0]?.error.message).toBe("reconnect root cause"); }); + it("preserves reconnect failures when cleanup run-store delete throws", async function () { + const errors: TriggerChatTransportError[] = []; + const runStore = new FailingCleanupDeleteRunStore(1); + runStore.set({ + chatId: "chat-reconnect-cleanup-delete-failure", + runId: "run_reconnect_cleanup_delete_failure", + publicAccessToken: "pk_reconnect_cleanup_delete_failure", + streamKey: "chat-stream", + lastEventId: "100-0", + isActive: true, + }); + + const transport = new TriggerChatTransport({ + task: "chat-task", + stream: "chat-stream", + accessToken: "pk_trigger", + runStore, + onError: function onError(error) { + errors.push(error); + }, + }); + + (transport as any).fetchRunStream = async function fetchRunStream() { + throw new Error("reconnect root cause"); + }; + + const stream = await transport.reconnectToStream({ + chatId: "chat-reconnect-cleanup-delete-failure", + }); + + expect(stream).toBeNull(); + expect(errors).toHaveLength(1); + expect(errors[0]).toMatchObject({ + phase: "reconnect", + chatId: "chat-reconnect-cleanup-delete-failure", + runId: "run_reconnect_cleanup_delete_failure", + }); + expect(errors[0]?.error.message).toBe("reconnect root cause"); + }); + it("normalizes non-Error reconnect failures before reporting onError", async function () { const errors: TriggerChatTransportError[] = []; const runStore = new InMemoryTriggerChatRunStore(); @@ -2486,6 +2578,23 @@ class FailingCleanupSetRunStore extends InMemoryTriggerChatRunStore { } } +class FailingCleanupDeleteRunStore extends InMemoryTriggerChatRunStore { + private deleteCalls = 0; + + constructor(private readonly failOnDeleteCall: number) { + super(); + } + + public delete(chatId: string): void { + this.deleteCalls += 1; + if (this.deleteCalls === this.failOnDeleteCall) { + throw new Error("cleanup delete failed"); + } + + super.delete(chatId); + } +} + class AsyncTrackedRunStore implements TriggerChatRunStore { private readonly runs = new Map(); public readonly getCalls: string[] = [];