diff --git a/app/server.test.ts b/app/server.test.ts index 5477679..6db7155 100644 --- a/app/server.test.ts +++ b/app/server.test.ts @@ -3,6 +3,7 @@ import type { RequestListener } from "node:http"; import { describe, expect, it, vi } from "vitest"; import type { ChannelsControl } from "@copilotkit/runtime/v2"; import { + installUnhandledRejectionBackstop, startOpenTagServer, type HttpServerLike, type RuntimeListener, @@ -134,6 +135,34 @@ describe("startOpenTagServer", () => { }); }); +describe("installUnhandledRejectionBackstop", () => { + it("logs every leaked rejection, including a non-Error reason", () => { + const target = new EventEmitter(); + const consoleError = vi + .spyOn(console, "error") + .mockImplementation(() => undefined); + const leaked = new TypeError("terminated"); + + installUnhandledRejectionBackstop(target); + target.emit("unhandledRejection", leaked); + target.emit("unhandledRejection", "second"); + + expect(consoleError).toHaveBeenCalledTimes(2); + expect(consoleError.mock.calls[0]?.[1]).toBe(leaked); + expect(consoleError.mock.calls[1]?.[1]).toBe("second"); + consoleError.mockRestore(); + }); + + it("subscribes the live process by default, which is what the entrypoint relies on", () => { + const on = vi.spyOn(process, "on").mockReturnValue(process); + + installUnhandledRejectionBackstop(); + + expect(on).toHaveBeenCalledWith("unhandledRejection", expect.any(Function)); + on.mockRestore(); + }); +}); + describe("createOpenTagApplication", () => { it("declares one adapter-free managed Channel", () => { const environment: AppEnvironment = { diff --git a/server.ts b/server.ts index 0d019e5..797f18e 100644 --- a/server.ts +++ b/server.ts @@ -26,6 +26,13 @@ export interface SignalTarget { off(signal: NodeJS.Signals, listener: () => void): unknown; } +export interface RejectionTarget { + on( + event: "unhandledRejection", + listener: (reason: unknown) => void, + ): unknown; +} + export interface RunningOpenTagServer { server: HttpServerLike; shutdown(): Promise; @@ -156,6 +163,22 @@ export async function startOpenTagServer( return { server: startedServer, shutdown }; } +/** + * Node terminates the process on an unhandled rejection. When a fetch to the + * agent fails, the transport leaks one *in addition to* rejecting the promise + * the caller awaits. Every call site already catches that awaited rejection, so + * no application-level try/catch can reach the leaked copy, and a transient + * agent hiccup takes the whole runtime down with it. Logging the leak keeps the + * Channel session alive. + */ +export function installUnhandledRejectionBackstop( + target: RejectionTarget = process, +): void { + target.on("unhandledRejection", (reason) => { + console.error("[opentag] unhandled rejection; runtime kept alive", reason); + }); +} + export async function main(): Promise { const application = createOpenTagApplication(); const running = await startOpenTagServer({ @@ -177,6 +200,7 @@ const isMain = import.meta.url === pathToFileURL(process.argv[1]).href; if (isMain) { + installUnhandledRejectionBackstop(); try { const { Agent, setGlobalDispatcher } = await import("undici"); setGlobalDispatcher(