diff --git a/packages/trigger-sdk/src/v3/ai-shared.test.ts b/packages/trigger-sdk/src/v3/ai-shared.test.ts new file mode 100644 index 00000000000..89f058e3b3c --- /dev/null +++ b/packages/trigger-sdk/src/v3/ai-shared.test.ts @@ -0,0 +1,93 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { anySignal } from "./ai-shared.js"; + +describe("anySignal", () => { + it("returns a non-aborted signal when signals list is empty", () => { + const signal = anySignal([]); + expect(signal.aborted).toBe(false); + }); + + it("filters out null and undefined values", () => { + const controller = new AbortController(); + const signal = anySignal([null, undefined, controller.signal, null]); + expect(signal).toBe(controller.signal); + }); + + it("returns the single signal if only one valid signal is provided", () => { + const controller = new AbortController(); + const signal = anySignal([controller.signal]); + expect(signal).toBe(controller.signal); + }); + + describe("when AbortSignal.any is available (native)", () => { + it("aborts when any of the input signals aborts", () => { + const c1 = new AbortController(); + const c2 = new AbortController(); + const combined = anySignal([c1.signal, c2.signal]); + + expect(combined.aborted).toBe(false); + c2.abort("cancelled by c2"); + expect(combined.aborted).toBe(true); + expect(combined.reason).toBe("cancelled by c2"); + }); + }); + + describe("fallback when AbortSignal.any is undefined (Safari < 17.4, Node < 20.3)", () => { + const originalAny = AbortSignal.any; + + beforeEach(() => { + // @ts-expect-error - simulating environment without AbortSignal.any + delete AbortSignal.any; + }); + + afterEach(() => { + AbortSignal.any = originalAny; + }); + + it("returns an immediately aborted signal if an input signal is already aborted", () => { + const c1 = new AbortController(); + const c2 = new AbortController(); + c1.abort("already aborted reason"); + + const combined = anySignal([c1.signal, c2.signal]); + expect(combined.aborted).toBe(true); + expect(combined.reason).toBe("already aborted reason"); + }); + + it("aborts when the first signal is aborted", () => { + const c1 = new AbortController(); + const c2 = new AbortController(); + const combined = anySignal([c1.signal, c2.signal]); + + expect(combined.aborted).toBe(false); + c1.abort("first reason"); + expect(combined.aborted).toBe(true); + expect(combined.reason).toBe("first reason"); + }); + + it("aborts when the second signal is aborted", () => { + const c1 = new AbortController(); + const c2 = new AbortController(); + const combined = anySignal([c1.signal, c2.signal]); + + expect(combined.aborted).toBe(false); + c2.abort("second reason"); + expect(combined.aborted).toBe(true); + expect(combined.reason).toBe("second reason"); + }); + + it("cleans up event listeners on remaining signals when one signal aborts", () => { + const c1 = new AbortController(); + const c2 = new AbortController(); + + const c2RemoveSpy = vi.spyOn(c2.signal, "removeEventListener"); + const combined = anySignal([c1.signal, c2.signal]); + + expect(combined.aborted).toBe(false); + c1.abort("first aborted"); + + expect(combined.aborted).toBe(true); + expect(c2RemoveSpy).toHaveBeenCalledWith("abort", expect.any(Function)); + }); + }); +}); diff --git a/packages/trigger-sdk/src/v3/ai-shared.ts b/packages/trigger-sdk/src/v3/ai-shared.ts index d0434e24672..ff16ec45eb7 100644 --- a/packages/trigger-sdk/src/v3/ai-shared.ts +++ b/packages/trigger-sdk/src/v3/ai-shared.ts @@ -278,3 +278,50 @@ export function waitBeforeEofResubscribe(attempt: number, signal: AbortSignal): if (signal.aborted) done(); }); } + +/** + * Combines multiple AbortSignals into a single signal. + * + * Uses native `AbortSignal.any` when available, and falls back to an AbortController + * with listeners on environments where `AbortSignal.any` is not supported (such as + * Safari < 17.4 or Node < 20.3). + */ +export function anySignal(signals: (AbortSignal | undefined | null)[]): AbortSignal { + const validSignals = signals.filter((s): s is AbortSignal => s != null); + const first = validSignals[0]; + if (!first) { + return new AbortController().signal; + } + if (validSignals.length === 1) { + return first; + } + if (typeof AbortSignal !== "undefined" && typeof AbortSignal.any === "function") { + return AbortSignal.any(validSignals); + } + + const controller = new AbortController(); + const alreadyAborted = validSignals.find((s) => s.aborted); + if (alreadyAborted) { + controller.abort(alreadyAborted.reason); + return controller.signal; + } + + const onAbort = (e: Event) => { + const s = e.target as AbortSignal; + controller.abort(s?.reason); + cleanup(); + }; + + const cleanup = () => { + for (const s of validSignals) { + s.removeEventListener("abort", onAbort); + } + }; + + for (const signal of validSignals) { + signal.addEventListener("abort", onAbort, { once: true }); + } + + return controller.signal; +} + diff --git a/packages/trigger-sdk/src/v3/ai.ts b/packages/trigger-sdk/src/v3/ai.ts index 2020bc1013c..7cfa30e9360 100644 --- a/packages/trigger-sdk/src/v3/ai.ts +++ b/packages/trigger-sdk/src/v3/ai.ts @@ -136,6 +136,7 @@ import { type InferChatClientData, type InferChatUIMessage, type InferChatUIMessageFromTools, + anySignal, BACKGROUND_INJECTION_TYPE, PENDING_MESSAGE_INJECTED_TYPE, upsertIncomingMessage, @@ -11072,7 +11073,7 @@ function chatAgent< // Three signals for the user's run function const stopSignal = stopController.signal; const cancelSignal = runSignal; - const combinedSignal = AbortSignal.any([runSignal, stopController.signal]); + const combinedSignal = anySignal([runSignal, stopController.signal]); const pmConfig = locals.get(chatPendingMessagesKey); /** @@ -15859,7 +15860,7 @@ function createChatSession( return { done: true, value: undefined }; } - const combinedSignal = AbortSignal.any([runSignal, stop.signal]); + const combinedSignal = anySignal([runSignal, stop.signal]); const turnObj: ChatTurn = { number: turn, diff --git a/packages/trigger-sdk/src/v3/chat-client.ts b/packages/trigger-sdk/src/v3/chat-client.ts index 1a421af4990..abaf015a1f2 100644 --- a/packages/trigger-sdk/src/v3/chat-client.ts +++ b/packages/trigger-sdk/src/v3/chat-client.ts @@ -31,6 +31,7 @@ import { } from "@trigger.dev/core/v3"; import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js"; import { + anySignal, chatRunTags, MAX_EOF_RESUBSCRIBES, slimSubmitMessageForWire, @@ -815,7 +816,7 @@ export class AgentChat { const internalAbort = new AbortController(); const combinedSignal = abortSignal - ? AbortSignal.any([abortSignal, internalAbort.signal]) + ? anySignal([abortSignal, internalAbort.signal]) : internalAbort.signal; // Detached once the stream ends (see the `finally` below): a caller reusing diff --git a/packages/trigger-sdk/src/v3/chat-server.ts b/packages/trigger-sdk/src/v3/chat-server.ts index 22fd2e284e6..663d2e50a66 100644 --- a/packages/trigger-sdk/src/v3/chat-server.ts +++ b/packages/trigger-sdk/src/v3/chat-server.ts @@ -79,7 +79,7 @@ import type { UIMessageChunk, } from "ai"; import type { ChatInputChunk, ChatTaskWirePayload } from "./ai-shared.js"; -import { chatRunTags } from "./ai-shared.js"; +import { anySignal, chatRunTags } from "./ai-shared.js"; import { triggerConcurrencyBody } from "./concurrency-shared.js"; import { withResolvedExternalDeploymentId } from "./externalDeploymentId.js"; import type { SessionTriggerConfigInput } from "./sessions.js"; @@ -970,7 +970,7 @@ async function openHandoverSession(opts: { "out", { ...(customerLastEventId != null ? { lastEventId: customerLastEventId } : {}), - signal: AbortSignal.any([abortController.signal, subscriptionAbort.signal]), + signal: anySignal([abortController.signal, subscriptionAbort.signal]), onPart: (part) => { if (part.id) latestEventId = part.id; }, diff --git a/packages/trigger-sdk/src/v3/chat.test.ts b/packages/trigger-sdk/src/v3/chat.test.ts index a4d8b9841fa..20502b05595 100644 --- a/packages/trigger-sdk/src/v3/chat.test.ts +++ b/packages/trigger-sdk/src/v3/chat.test.ts @@ -2456,6 +2456,25 @@ describe("TriggerChatTransport", () => { vi.useRealTimers(); } }); + + it("safely reconnects and aborts when AbortSignal.any is not supported (Safari < 17.4)", async () => { + const originalAny = AbortSignal.any; + try { + // @ts-expect-error - simulating older browser/runtime without AbortSignal.any + delete AbortSignal.any; + + const { transport } = quietWatchTransport(); + const abort = new AbortController(); + const stream = await transport.reconnectToStream({ + chatId: "chat-own", + abortSignal: abort.signal, + }); + expect(stream).toBeDefined(); + abort.abort(); + } finally { + AbortSignal.any = originalAny; + } + }); }); describe("superseded stream teardown", () => { diff --git a/packages/trigger-sdk/src/v3/chat.ts b/packages/trigger-sdk/src/v3/chat.ts index ba74aff2cc2..4a2f756c8dd 100644 --- a/packages/trigger-sdk/src/v3/chat.ts +++ b/packages/trigger-sdk/src/v3/chat.ts @@ -46,6 +46,7 @@ function byteLength(body: string): number { } import { ChatTabCoordinator } from "./chat-tab-coordinator.js"; import { + anySignal, MAX_EOF_RESUBSCRIBES, slimSubmitMessageForWire, waitBeforeEofResubscribe, @@ -1355,7 +1356,7 @@ export class TriggerChatTransport implements ChatTransport { this.activeStreams.set(options.chatId, abortController); const abortSignal = options.abortSignal - ? AbortSignal.any([options.abortSignal, abortController.signal]) + ? anySignal([options.abortSignal, abortController.signal]) : abortController.signal; return this.subscribeToSessionStream(state, abortSignal, options.chatId, { @@ -2064,7 +2065,7 @@ export class TriggerChatTransport implements ChatTransport { let didSettle = false; this.activeStreams.set(chatId, internalAbort); const combinedSignal = abortSignal - ? AbortSignal.any([abortSignal, internalAbort.signal]) + ? anySignal([abortSignal, internalAbort.signal]) : internalAbort.signal; if (abortSignal) {