diff --git a/.changeset/clean-workflow-world-shutdown.md b/.changeset/clean-workflow-world-shutdown.md new file mode 100644 index 000000000..c077d77e4 --- /dev/null +++ b/.changeset/clean-workflow-world-shutdown.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Self-hosted eve servers now await Workflow world and sandbox cleanup during SIGINT or SIGTERM, and local development asks worker-owned Workflow worlds to close before retiring a worker during reload or shutdown. Configured Worlds that release queue workers from `close()` can therefore resume durable sessions after restart without waiting for stale locks to expire. diff --git a/docs/guides/deployment/self-hosting.md b/docs/guides/deployment/self-hosting.md index c470f57ee..a190b7727 100644 --- a/docs/guides/deployment/self-hosting.md +++ b/docs/guides/deployment/self-hosting.md @@ -18,6 +18,8 @@ The build writes the Nitro server under `.output/`. `eve start` serves that outp Run this process under the same process manager or container platform you use for other Node web services. Configure Transport Layer Security (TLS), scaling, restarts, and log collection in that platform. +On `SIGINT` or `SIGTERM`, eve runs Nitro's coordinated close lifecycle before exiting. That lifecycle awaits the configured Workflow world's `close()` implementation and tracked sandbox cleanup, so a replacement process can immediately reclaim queue work and resume durable sessions. Give the process enough termination grace to finish those adapters' shutdown paths. + ## Configure model access and route auth Set `AI_GATEWAY_API_KEY` to use a string model ID through the Vercel AI Gateway from a non-Vercel host. To call a provider directly, install its [AI SDK provider package](https://ai-sdk.dev/docs/foundations/providers-and-models). Then pass its model object in `agent.ts` and set its API key. See [Agent configuration](../../agent-config#set-the-model) for examples. diff --git a/docs/reference/cli.md b/docs/reference/cli.md index f99ca05e9..a9a4a65f8 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -220,6 +220,8 @@ Local callback-based connection authorization requires a persistent server. Run Local dev records the last ready URL per resolved app root in `.eve/dev-server-state.v1.json`. A second interactive `eve dev` reconnects only when that URL is loopback and healthy; each terminal UI creates a fresh client session while sharing the server process. A stale or malformed record is replaced when eve starts a new server. Passing `--host`, `--port`, or a `PORT` environment value skips reconnection and reports a healthy recorded server instead. +When a rebuild retires a development worker, or when `eve dev` shuts down, eve gives the worker up to 300 ms to finish its Nitro close lifecycle before terminating it. That bounded grace lets a configured Workflow world's `close()` release queue work and connections without allowing an unresponsive worker to exceed the CLI's shutdown budget. + Local dev keeps immutable runtime source snapshots under `.eve/dev-runtime/snapshots/` so in-flight turns hold a consistent code revision while new turns pick up rebuilds. The terminal REPL keeps its logical session across successful rebuilds, so the next turn continues the conversation on the latest generation; `/new` terminally retires that session before clearing the transcript, and the next prompt starts a fresh session with a new session-scoped sandbox on first sandbox use. After a generation is superseded, `eve dev` retains it for at least 30 minutes and also retains the five most recently superseded generations, regardless of the configured Workflow World. The active generation is never pruned. Old runtime snapshots and local sandbox templates are pruned in the background. For manual cleanup, stop `eve dev` before deleting `.eve/dev-runtime/snapshots/` or `.eve/sandbox-cache/local/templates/`. A turn that remains unfinished beyond the automatic retention window can no longer resume after its generation is pruned. When no authored `agent/instrumentation.ts` exists, local dev also records traces under `.eve/traces/`, and bounds that store by age, size, and a keep-newest floor. Configure it with `EVE_TRACES*` in `.env.local`; see [local trace retention](../guides/instrumentation#local-trace-retention) for the rules and defaults. diff --git a/packages/eve/src/internal/application/compiled-artifacts.test.ts b/packages/eve/src/internal/application/compiled-artifacts.test.ts index 3dff1e8d6..aa8af642e 100644 --- a/packages/eve/src/internal/application/compiled-artifacts.test.ts +++ b/packages/eve/src/internal/application/compiled-artifacts.test.ts @@ -26,6 +26,9 @@ describe("createWorkflowWorldPluginSource", () => { expect(source).toContain("setWorld(workflowWorld);"); expect(source).toContain("await getWorld();"); expect(source).toContain("await workflowWorld.start?.();"); + expect(source).toContain("workflowWorldClosePromise ??= Promise.resolve()"); + expect(source).toContain("await workflowWorld.close?.();"); + expect(source).toContain('nitroApp.hooks.hook("close", closeWorkflowWorld);'); }); it("configures the vendored local World with eve's app-local data resolver", () => { diff --git a/packages/eve/src/internal/application/compiled-artifacts.ts b/packages/eve/src/internal/application/compiled-artifacts.ts index 81114d440..cec826a31 100644 --- a/packages/eve/src/internal/application/compiled-artifacts.ts +++ b/packages/eve/src/internal/application/compiled-artifacts.ts @@ -336,7 +336,22 @@ export function createWorkflowWorldPluginSource(input: { "await getWorld();", "await workflowWorld.start?.();", "", - "export default function installWorkflowWorldPlugin() {}", + "let workflowWorldClosePromise;", + "", + "async function closeWorkflowWorld() {", + " workflowWorldClosePromise ??= Promise.resolve()", + " .then(async () => {", + " await workflowWorld.close?.();", + " })", + " .catch((error) => {", + ' console.error("[eve] Failed to close Workflow world:", error);', + " });", + " await workflowWorldClosePromise;", + "}", + "", + "export default function installWorkflowWorldPlugin(nitroApp) {", + ' nitroApp.hooks.hook("close", closeWorkflowWorld);', + "}", "", ].join("\n"); } diff --git a/packages/eve/src/internal/nitro/host/create-application-nitro.scenario.test.ts b/packages/eve/src/internal/nitro/host/create-application-nitro.scenario.test.ts index 919138520..e137379b7 100644 --- a/packages/eve/src/internal/nitro/host/create-application-nitro.scenario.test.ts +++ b/packages/eve/src/internal/nitro/host/create-application-nitro.scenario.test.ts @@ -627,7 +627,7 @@ describe("application Nitro creation", () => { ); }); - it("includes the sandbox shutdown plugin only for production builds", async () => { + it("includes coordinated shutdown plugins only for production builds", async () => { const productionNitroStub = createNitroStub(); const devNitroStub = createNitroStub(); createNitroMock.mockResolvedValueOnce(productionNitroStub.nitro); @@ -644,10 +644,16 @@ describe("application Nitro creation", () => { const devPlugins = createNitroMock.mock.calls[1]?.[0].plugins as string[]; expect(productionPlugins).toEqual( - expect.arrayContaining([expect.stringContaining("sandbox-shutdown-plugin.ts")]), + expect.arrayContaining([ + expect.stringContaining("sandbox-shutdown-plugin.ts"), + expect.stringContaining("server-shutdown-plugin.ts"), + ]), ); expect(devPlugins).not.toEqual( - expect.arrayContaining([expect.stringContaining("sandbox-shutdown-plugin.ts")]), + expect.arrayContaining([ + expect.stringContaining("sandbox-shutdown-plugin.ts"), + expect.stringContaining("server-shutdown-plugin.ts"), + ]), ); }); diff --git a/packages/eve/src/internal/nitro/host/create-application-nitro.ts b/packages/eve/src/internal/nitro/host/create-application-nitro.ts index 89e78ca19..6f1b3bf1e 100644 --- a/packages/eve/src/internal/nitro/host/create-application-nitro.ts +++ b/packages/eve/src/internal/nitro/host/create-application-nitro.ts @@ -813,6 +813,7 @@ export async function createProductionApplicationNitro( const nitroPlugins = createApplicationNitroPlugins(preparedHost); nitroPlugins.push( resolvePackageSourceFilePath("src/internal/nitro/host/sandbox-shutdown-plugin.ts"), + resolvePackageSourceFilePath("src/internal/nitro/host/server-shutdown-plugin.ts"), ); await prepareEveVersionedCacheDirectory(options.buildDir); diff --git a/packages/eve/src/internal/nitro/host/dev-runner.scenario.test.ts b/packages/eve/src/internal/nitro/host/dev-runner.scenario.test.ts new file mode 100644 index 000000000..848cc5879 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/dev-runner.scenario.test.ts @@ -0,0 +1,88 @@ +import { readFile, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { createNodeDevelopmentRunner } from "#internal/nitro/host/dev-runner.js"; +import { useTemporaryDirectories } from "#internal/testing/use-temporary-app-roots.js"; + +const createScratchDirectory = useTemporaryDirectories(); + +describe("Node development runner shutdown", () => { + it("awaits worker-owned close hooks exactly once", async () => { + const scratchRoot = await createScratchDirectory("eve-dev-runner-close-"); + const eventsPath = join(scratchRoot, "events.log"); + const entryPath = join(scratchRoot, "entry.mjs"); + + await writeFile( + entryPath, + [ + 'import { appendFile } from "node:fs/promises";', + `const eventsPath = ${JSON.stringify(eventsPath)};`, + "export default {", + ' fetch() { return new Response("ok"); },', + " ipc: {", + " async onClose() {", + ' await appendFile(eventsPath, "close:start\\n");', + " await new Promise((resolve) => setTimeout(resolve, 25));", + ' await appendFile(eventsPath, "close:end\\n");', + " },", + " },", + "};", + "", + ].join("\n"), + "utf8", + ); + + const runner = createNodeDevelopmentRunner({ + entry: entryPath, + name: "close-probe", + workerData: {}, + }); + await runner.waitForReady(5_000); + + await runner.close(); + await runner.close(); + + await expect(readFile(eventsPath, "utf8")).resolves.toBe("close:start\nclose:end\n"); + }); + + it("force-reaps a worker whose close hook never settles", async () => { + const scratchRoot = await createScratchDirectory("eve-dev-runner-stuck-close-"); + const eventsPath = join(scratchRoot, "events.log"); + const entryPath = join(scratchRoot, "entry.mjs"); + + await writeFile( + entryPath, + [ + 'import { appendFile } from "node:fs/promises";', + `const eventsPath = ${JSON.stringify(eventsPath)};`, + "export default {", + ' fetch() { return new Response("ok"); },', + " ipc: {", + " async onClose() {", + ' await appendFile(eventsPath, "close\\n");', + " await new Promise(() => undefined);", + " },", + " },", + "};", + "", + ].join("\n"), + "utf8", + ); + + const runner = createNodeDevelopmentRunner({ + entry: entryPath, + name: "stuck-close-probe", + workerData: {}, + }); + await runner.waitForReady(5_000); + + const startedAt = Date.now(); + await runner.close(); + await runner.close(); + + expect(Date.now() - startedAt).toBeLessThan(1_000); + await expect(readFile(eventsPath, "utf8")).resolves.toBe("close\n"); + }); +}); diff --git a/packages/eve/src/internal/nitro/host/dev-runner.ts b/packages/eve/src/internal/nitro/host/dev-runner.ts index a30fd169d..f20e184fc 100644 --- a/packages/eve/src/internal/nitro/host/dev-runner.ts +++ b/packages/eve/src/internal/nitro/host/dev-runner.ts @@ -4,6 +4,11 @@ import { Worker } from "node:worker_threads"; import { BaseEnvRunner } from "#compiled/env-runner/index.js"; import { resolvePackageCompiledFilePath } from "#internal/application/package.js"; +// The outer dev-server process grants its child 550ms to close over IPC before +// escalating to process signals. Keep worker cleanup inside that window while +// still giving Nitro hooks enough time to release worker-owned resources. +const DEVELOPMENT_WORKER_SHUTDOWN_GRACE_MS = 300; + export interface DevelopmentRunner { readonly closed: boolean; close(cause?: unknown): Promise; @@ -98,8 +103,15 @@ class NodeDevelopmentRunner extends BaseEnvRunner implements DevelopmentRunner { } this.#worker = undefined; + await requestWorkerShutdown(worker); worker.removeAllListeners(); - await worker.terminate(); + const ignoreTerminationError = () => undefined; + worker.on("error", ignoreTerminationError); + try { + await worker.terminate(); + } finally { + worker.off("error", ignoreTerminationError); + } } protected override _handleMessage(message: unknown): void { @@ -139,6 +151,56 @@ class NodeDevelopmentRunner extends BaseEnvRunner implements DevelopmentRunner { export const createNodeDevelopmentRunner: DevelopmentRunnerFactory = (input) => new NodeDevelopmentRunner(input); +async function requestWorkerShutdown(worker: Worker): Promise { + await new Promise((resolve) => { + let settled = false; + let timeout: NodeJS.Timeout | undefined; + + const settle = () => { + if (settled) { + return; + } + settled = true; + if (timeout !== undefined) { + clearTimeout(timeout); + } + worker.off("error", settle); + worker.off("exit", settle); + worker.off("message", onMessage); + resolve(); + }; + const onMessage = (message: unknown) => { + if (isWorkerExitMessage(message)) { + settle(); + } + }; + + worker.once("error", settle); + worker.once("exit", settle); + worker.on("message", onMessage); + timeout = setTimeout(settle, DEVELOPMENT_WORKER_SHUTDOWN_GRACE_MS); + timeout.unref(); + + if (worker.threadId === -1) { + settle(); + return; + } + + try { + worker.postMessage({ event: "shutdown" }); + } catch { + settle(); + } + }); +} + +function isWorkerExitMessage(value: unknown): value is { readonly event: "exit" } { + if (typeof value !== "object" || value === null) { + return false; + } + return (value as Record).event === "exit"; +} + function isWorkerInitializationError( value: unknown, ): value is { readonly error: string; readonly event: "init-error" } { diff --git a/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.test.ts b/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.test.ts index 6ebcd7332..7faf9ebe1 100644 --- a/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.test.ts +++ b/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.test.ts @@ -5,91 +5,16 @@ import { trackActiveSandboxHandle, } from "#execution/sandbox/active-handles.js"; import { - installSandboxShutdownHandlers, + installSandboxShutdownHook, runSandboxShutdown, - shouldInstallSandboxShutdown, } from "#internal/nitro/host/sandbox-shutdown-plugin.js"; -type SignalListener = () => void; - -function createFakeProcess(env: Record = {}) { - const listeners = new Map(); - const exit = vi.fn<(code?: number) => void>(); - return { - emit(event: string): void { - listeners.get(event)?.(); - }, - env, - exit, - listeners, - once(event: "SIGINT" | "SIGTERM", listener: SignalListener): unknown { - listeners.set(event, listener); - return this; - }, - }; -} - afterEach(() => { clearActiveSandboxHandlesForTest(); vi.unstubAllEnvs(); }); -describe("shouldInstallSandboxShutdown", () => { - it("installs on a plain production server", () => { - expect(shouldInstallSandboxShutdown({})).toBe(true); - }); - - it("skips eve dev processes", () => { - vi.stubEnv("EVE_DEV", "1"); - expect(shouldInstallSandboxShutdown({})).toBe(false); - }); - - it("skips dev sandbox run workers", () => { - expect(shouldInstallSandboxShutdown({ EVE_DEVELOPMENT_SANDBOX_RUN_ID: "dev-run" })).toBe(false); - }); - - it("skips Vercel serverless instances", () => { - expect(shouldInstallSandboxShutdown({ VERCEL: "1" })).toBe(false); - }); -}); - -describe("installSandboxShutdownHandlers", () => { - it("registers no handlers when shutdown ownership is elsewhere", () => { - const fakeProcess = createFakeProcess({ VERCEL: "1" }); - - installSandboxShutdownHandlers({ log: () => {}, process: fakeProcess }); - - expect(fakeProcess.listeners.size).toBe(0); - }); - - it("stops tracked sandboxes and exits 143 on SIGTERM", async () => { - const handle = { shutdown: vi.fn(async () => {}) }; - trackActiveSandboxHandle({ backendName: "docker", handle, sessionKey: "session-1" }); - const fakeProcess = createFakeProcess(); - - installSandboxShutdownHandlers({ log: () => {}, process: fakeProcess }); - fakeProcess.emit("SIGTERM"); - - await vi.waitFor(() => { - expect(fakeProcess.exit).toHaveBeenCalledWith(143); - }); - expect(handle.shutdown).toHaveBeenCalledTimes(1); - }); - - it("stops tracked sandboxes and exits 130 on SIGINT", async () => { - const handle = { shutdown: vi.fn(async () => {}) }; - trackActiveSandboxHandle({ backendName: "docker", handle, sessionKey: "session-1" }); - const fakeProcess = createFakeProcess(); - - installSandboxShutdownHandlers({ log: () => {}, process: fakeProcess }); - fakeProcess.emit("SIGINT"); - - await vi.waitFor(() => { - expect(fakeProcess.exit).toHaveBeenCalledWith(130); - }); - expect(handle.shutdown).toHaveBeenCalledTimes(1); - }); - +describe("installSandboxShutdownHook", () => { it("stops tracked sandboxes through the nitro close hook", async () => { const handle = { shutdown: vi.fn(async () => {}) }; trackActiveSandboxHandle({ backendName: "docker", handle, sessionKey: "session-1" }); @@ -105,15 +30,27 @@ describe("installSandboxShutdownHandlers", () => { }, }; - installSandboxShutdownHandlers({ + installSandboxShutdownHook({ + env: {}, log: () => {}, nitroApp, - process: createFakeProcess(), }); await closeHandler?.(); expect(handle.shutdown).toHaveBeenCalledTimes(1); }); + + it("registers no close hook when shutdown ownership is elsewhere", () => { + const hook = vi.fn(); + + installSandboxShutdownHook({ + env: { VERCEL: "1" }, + log: () => {}, + nitroApp: { hooks: { hook } }, + }); + + expect(hook).not.toHaveBeenCalled(); + }); }); describe("runSandboxShutdown", () => { diff --git a/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.ts b/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.ts index 716a33743..96a8b1c4a 100644 --- a/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.ts +++ b/packages/eve/src/internal/nitro/host/sandbox-shutdown-plugin.ts @@ -1,8 +1,5 @@ import { shutdownActiveSandboxHandles } from "#execution/sandbox/active-handles.js"; -import { isEveDevEnvironment } from "#internal/application/optional-package-install.js"; - -const SHUTDOWN_SIGNALS = ["SIGINT", "SIGTERM"] as const; -type ShutdownSignal = (typeof SHUTDOWN_SIGNALS)[number]; +import { shouldInstallServerShutdown } from "#internal/nitro/host/server-shutdown-plugin.js"; /** * Bounds sandbox shutdown so a wedged provider cannot keep the server @@ -13,40 +10,12 @@ const SANDBOX_SHUTDOWN_TIMEOUT_MS = 15_000; let installed = false; -interface SandboxShutdownProcess { - readonly env: Record; - exit(code?: number): void; - once(event: ShutdownSignal, listener: () => void): unknown; -} - interface NitroAppLike { readonly hooks?: { hook(name: "close", handler: () => Promise): unknown; }; } -/** - * Reports whether this server process owns sandbox shutdown. - * - * - `eve dev` workers are excluded: the dev CLI parent already stops - * dev-tagged sandboxes when the dev server closes. - * - Vercel serverless instances are excluded: instance recycling is not - * a server stop, and persistent session sandboxes must keep serving - * later invocations. - */ -export function shouldInstallSandboxShutdown(env: Record): boolean { - if (isEveDevEnvironment()) { - return false; - } - if (env.EVE_DEVELOPMENT_SANDBOX_RUN_ID !== undefined) { - return false; - } - if (env.VERCEL !== undefined) { - return false; - } - return true; -} - /** * Stops all tracked sandboxes, bounded by * {@link SANDBOX_SHUTDOWN_TIMEOUT_MS}. Never throws. @@ -68,36 +37,21 @@ export async function runSandboxShutdown(log: (message: string) => void): Promis } } -function exitCodeForSignal(signal: ShutdownSignal): number { - return signal === "SIGINT" ? 130 : 143; -} - /** - * Wires sandbox shutdown into the server lifecycle: the nitro `close` - * hook plus SIGINT/SIGTERM, since the node-server preset installs no - * signal handling of its own. Exposed for tests; the plugin default - * export applies it to the real `process`. + * Wires tracked sandbox cleanup into Nitro's coordinated close lifecycle. */ -export function installSandboxShutdownHandlers(input: { +export function installSandboxShutdownHook(input: { + readonly env: Record; readonly log: (message: string) => void; readonly nitroApp?: NitroAppLike; - readonly process: SandboxShutdownProcess; }): void { - if (!shouldInstallSandboxShutdown(input.process.env)) { + if (!shouldInstallServerShutdown(input.env)) { return; } input.nitroApp?.hooks?.hook("close", async () => { await runSandboxShutdown(input.log); }); - - for (const signal of SHUTDOWN_SIGNALS) { - input.process.once(signal, () => { - void runSandboxShutdown(input.log).finally(() => { - input.process.exit(exitCodeForSignal(signal)); - }); - }); - } } export default function sandboxShutdownPlugin(nitroApp?: NitroAppLike): void { @@ -105,9 +59,9 @@ export default function sandboxShutdownPlugin(nitroApp?: NitroAppLike): void { return; } installed = true; - installSandboxShutdownHandlers({ + installSandboxShutdownHook({ + env: process.env, log: (message) => console.error(message), nitroApp, - process, }); } diff --git a/packages/eve/src/internal/nitro/host/server-shutdown-plugin.test.ts b/packages/eve/src/internal/nitro/host/server-shutdown-plugin.test.ts new file mode 100644 index 000000000..930c9965f --- /dev/null +++ b/packages/eve/src/internal/nitro/host/server-shutdown-plugin.test.ts @@ -0,0 +1,127 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; + +import { + installServerShutdownHandlers, + shouldInstallServerShutdown, +} from "#internal/nitro/host/server-shutdown-plugin.js"; + +type SignalListener = () => void; + +function createFakeProcess(env: Record = {}) { + const listeners = new Map(); + return { + emit(event: string): void { + listeners.get(event)?.(); + }, + env, + exitCode: undefined as number | undefined, + listeners, + once(event: "SIGINT" | "SIGTERM", listener: SignalListener): unknown { + listeners.set(event, listener); + return this; + }, + }; +} + +afterEach(() => { + vi.unstubAllEnvs(); +}); + +describe("shouldInstallServerShutdown", () => { + it("installs on a plain production server", () => { + expect(shouldInstallServerShutdown({})).toBe(true); + }); + + it("skips eve dev processes", () => { + vi.stubEnv("EVE_DEV", "1"); + expect(shouldInstallServerShutdown({})).toBe(false); + }); + + it("skips dev sandbox run workers", () => { + expect(shouldInstallServerShutdown({ EVE_DEVELOPMENT_SANDBOX_RUN_ID: "dev-run" })).toBe(false); + }); + + it("skips Vercel serverless instances", () => { + expect(shouldInstallServerShutdown({ VERCEL: "1" })).toBe(false); + }); +}); + +describe("installServerShutdownHandlers", () => { + it("awaits Nitro close before setting the SIGTERM exit code", async () => { + const close = Promise.withResolvers(); + const callHook = vi.fn(() => close.promise); + const fakeProcess = createFakeProcess(); + + installServerShutdownHandlers({ + log: () => {}, + nitroApp: { hooks: { callHook } }, + process: fakeProcess, + }); + fakeProcess.emit("SIGTERM"); + + await vi.waitFor(() => { + expect(callHook).toHaveBeenCalledWith("close"); + }); + expect(fakeProcess.exitCode).toBeUndefined(); + + close.resolve(); + await vi.waitFor(() => { + expect(fakeProcess.exitCode).toBe(143); + }); + }); + + it("runs the coordinated close path only once across competing signals", async () => { + const close = Promise.withResolvers(); + const callHook = vi.fn(() => close.promise); + const fakeProcess = createFakeProcess(); + + installServerShutdownHandlers({ + log: () => {}, + nitroApp: { hooks: { callHook } }, + process: fakeProcess, + }); + fakeProcess.emit("SIGTERM"); + fakeProcess.emit("SIGINT"); + close.resolve(); + + await vi.waitFor(() => { + expect(fakeProcess.exitCode).toBe(143); + }); + expect(callHook).toHaveBeenCalledTimes(1); + }); + + it("logs close failures before setting the exit code", async () => { + const log = vi.fn(); + const fakeProcess = createFakeProcess(); + + installServerShutdownHandlers({ + log, + nitroApp: { + hooks: { + callHook: vi.fn(async () => { + throw new Error("close failed"); + }), + }, + }, + process: fakeProcess, + }); + fakeProcess.emit("SIGINT"); + + await vi.waitFor(() => { + expect(fakeProcess.exitCode).toBe(130); + }); + expect(log).toHaveBeenCalledWith(expect.stringContaining("close failed")); + }); + + it("registers no signal handlers when shutdown ownership is elsewhere", () => { + const fakeProcess = createFakeProcess({ VERCEL: "1" }); + + installServerShutdownHandlers({ + log: () => {}, + nitroApp: { hooks: { callHook: vi.fn(async () => {}) } }, + process: fakeProcess, + }); + + expect(fakeProcess.listeners.size).toBe(0); + }); +}); diff --git a/packages/eve/src/internal/nitro/host/server-shutdown-plugin.ts b/packages/eve/src/internal/nitro/host/server-shutdown-plugin.ts new file mode 100644 index 000000000..f11eb8125 --- /dev/null +++ b/packages/eve/src/internal/nitro/host/server-shutdown-plugin.ts @@ -0,0 +1,93 @@ +import { isEveDevEnvironment } from "#internal/application/optional-package-install.js"; +import { toErrorMessage } from "#shared/errors.js"; + +const SHUTDOWN_SIGNALS = ["SIGINT", "SIGTERM"] as const; +type ShutdownSignal = (typeof SHUTDOWN_SIGNALS)[number]; + +let installed = false; + +interface ServerShutdownProcess { + readonly env: Record; + exitCode?: string | number | null; + once(event: ShutdownSignal, listener: () => void): unknown; +} + +interface NitroAppLike { + readonly hooks?: { + callHook(name: "close"): Promise; + }; +} + +/** + * Reports whether this process owns the self-hosted server shutdown lifecycle. + * + * Development workers delegate cleanup to the CLI parent. Vercel instances + * may be recycled without retiring persistent session resources. + */ +export function shouldInstallServerShutdown(env: Record): boolean { + if (isEveDevEnvironment()) { + return false; + } + if (env.EVE_DEVELOPMENT_SANDBOX_RUN_ID !== undefined) { + return false; + } + if (env.VERCEL !== undefined) { + return false; + } + return true; +} + +function exitCodeForSignal(signal: ShutdownSignal): number { + return signal === "SIGINT" ? 130 : 143; +} + +/** + * Routes process signals through Nitro's close lifecycle exactly once. + * + * srvx owns the HTTP server's graceful drain for these same signals. Setting + * `exitCode` instead of calling `process.exit()` lets both independent cleanup + * paths settle before Node exits naturally. + */ +export function installServerShutdownHandlers(input: { + readonly log: (message: string) => void; + readonly nitroApp?: NitroAppLike; + readonly process: ServerShutdownProcess; +}): void { + const closeHooks = input.nitroApp?.hooks; + if (!shouldInstallServerShutdown(input.process.env) || closeHooks === undefined) { + return; + } + + let shutdownRequested = false; + for (const signal of SHUTDOWN_SIGNALS) { + input.process.once(signal, () => { + if (shutdownRequested) { + return; + } + shutdownRequested = true; + + void Promise.resolve() + .then(async () => { + await closeHooks.callHook("close"); + }) + .catch((error: unknown) => { + input.log(`eve: server shutdown failed: ${toErrorMessage(error)}`); + }) + .finally(() => { + input.process.exitCode = exitCodeForSignal(signal); + }); + }); + } +} + +export default function serverShutdownPlugin(nitroApp?: NitroAppLike): void { + if (installed || nitroApp?.hooks === undefined) { + return; + } + installed = true; + installServerShutdownHandlers({ + log: (message) => console.error(message), + nitroApp, + process, + }); +} diff --git a/packages/eve/test/scenarios/cli.scenario.test.ts b/packages/eve/test/scenarios/cli.scenario.test.ts index 292f3b649..3af395fc7 100644 --- a/packages/eve/test/scenarios/cli.scenario.test.ts +++ b/packages/eve/test/scenarios/cli.scenario.test.ts @@ -1,5 +1,5 @@ import { spawn, type ChildProcess } from "node:child_process"; -import { access, mkdir, realpath, writeFile } from "node:fs/promises"; +import { access, mkdir, readFile, realpath, writeFile } from "node:fs/promises"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; @@ -74,6 +74,107 @@ async function createMinimalAppRoot(prefix: string): Promise { return appRoot; } +async function createWorkflowShutdownProbeAppRoot(): Promise<{ + readonly appRoot: string; + readonly eventsPath: string; + readonly lockPath: string; +}> { + const appRoot = await createScratchDirectory("eve-cli-workflow-shutdown-"); + const eventsPath = join(appRoot, "workflow-world-events.log"); + const lockPath = join(appRoot, "workflow-world.lock"); + const workflowWorldPath = join(appRoot, "workflow-shutdown-world.mjs"); + + await mkdir(join(appRoot, "agent"), { + recursive: true, + }); + await writeFile( + join(appRoot, "package.json"), + `${JSON.stringify( + { + name: "eve-cli-workflow-shutdown-test", + private: true, + type: "module", + }, + null, + 2, + )}\n`, + ); + await writeFile( + join(appRoot, "agent", "agent.mjs"), + [ + "export default {", + ' model: "openai/gpt-5.4",', + " experimental: {", + ` workflow: { world: ${JSON.stringify(workflowWorldPath)} },`, + " },", + "};", + "", + ].join("\n"), + ); + await writeFile(join(appRoot, "agent", "instructions.md"), "You are a precise assistant.\n"); + await mkdir(join(appRoot, "agent", "channels"), { + recursive: true, + }); + await writeFile( + join(appRoot, "agent", "channels", "shutdown.mjs"), + [ + "export default {", + ' __kind: "eve:channel",', + ' adapter: { kind: "channel" },', + " routes: [", + " {", + ' method: "GET",', + ' path: "/shutdown/slow-stream",', + " handler() {", + " const body = new ReadableStream({", + " start(controller) {", + ' controller.enqueue(new TextEncoder().encode("start\\n"));', + " setTimeout(() => {", + ' controller.enqueue(new TextEncoder().encode("finish\\n"));', + " controller.close();", + " }, 500);", + " },", + " });", + " return new Response(body);", + " },", + " },", + " ],", + "};", + "", + ].join("\n"), + ); + await writeFile( + workflowWorldPath, + [ + 'import { appendFile, rm, writeFile } from "node:fs/promises";', + "", + `const eventsPath = ${JSON.stringify(eventsPath)};`, + `const lockPath = ${JSON.stringify(lockPath)};`, + "", + "export function createWorld() {", + " return {", + " specVersion: 5,", + " events: {},", + " createQueueHandler() {", + " return async () => Response.json({ ok: true });", + " },", + " async start() {", + ' await writeFile(lockPath, String(process.pid), { flag: "wx" });', + ' await appendFile(eventsPath, "start\\n");', + " },", + " async close() {", + ' await appendFile(eventsPath, "close\\n");', + " await rm(lockPath);", + " },", + " };", + "}", + "", + ].join("\n"), + ); + + return { appRoot, eventsPath, lockPath }; +} + async function startPackagedEveStart(appRoot: string): Promise { const child = spawn( process.execPath, @@ -425,6 +526,56 @@ describe("runCli", () => { } }, 120_000); + it("closes a configured Workflow world before an immediate restart", async () => { + const { buildApplication } = await import("../../src/internal/nitro/host.js"); + const { appRoot, eventsPath, lockPath } = await createWorkflowShutdownProbeAppRoot(); + + await buildApplication(appRoot, { + skipVercelSandboxPrewarm: false, + }); + + const firstServer = await startPackagedEveStart(appRoot); + try { + await expect(fetch(new URL(EVE_HEALTH_ROUTE_PATH, firstServer.url))).resolves.toMatchObject({ + status: 200, + }); + + const response = await fetch(new URL("/shutdown/slow-stream", firstServer.url)); + expect(response.status).toBe(200); + expect(response.body).not.toBeNull(); + if (response.body === null) { + throw new Error("Expected the shutdown probe route to return a streaming body."); + } + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + const firstChunk = await reader.read(); + expect(decoder.decode(firstChunk.value)).toBe("start\n"); + + const stopping = firstServer.stop(); + const secondChunk = await reader.read(); + expect(decoder.decode(secondChunk.value)).toBe("finish\n"); + await expect(reader.read()).resolves.toMatchObject({ done: true }); + await stopping; + } finally { + await firstServer.stop(); + } + + await expect(access(lockPath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(readFile(eventsPath, "utf8")).resolves.toBe("start\nclose\n"); + + const secondServer = await startPackagedEveStart(appRoot); + try { + await expect(fetch(new URL(EVE_HEALTH_ROUTE_PATH, secondServer.url))).resolves.toMatchObject({ + status: 200, + }); + } finally { + await secondServer.stop(); + } + + await expect(access(lockPath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(readFile(eventsPath, "utf8")).resolves.toBe("start\nclose\nstart\nclose\n"); + }, 120_000); + it("surfaces discovery diagnostics in build failures", async () => { const workspaceRoot = await createScratchDirectory("eve-cli-build-diagnostics-"); const resolvedWorkspaceRoot = await realpath(workspaceRoot); diff --git a/packages/eve/test/scenarios/dev-server-shutdown.scenario.test.ts b/packages/eve/test/scenarios/dev-server-shutdown.scenario.test.ts index af2663c17..c592be999 100644 --- a/packages/eve/test/scenarios/dev-server-shutdown.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server-shutdown.scenario.test.ts @@ -1,10 +1,113 @@ +import { readFile, readdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + import { describe, expect, it } from "vitest"; import { WEATHER_AGENT_DESCRIPTOR } from "../../src/internal/testing/scenario-apps/weather-agent.js"; import { useScenarioApp } from "../../src/internal/testing/scenario-app.js"; -import { signalEveDevDuringStartup, startEveDev } from "./dev-server-harness.js"; +import type { RunningEveDev } from "./dev-server-harness.js"; +import { + forceDevelopmentRebuild, + signalEveDevDuringStartup, + startEveDev, + waitForCondition, +} from "./dev-server-harness.js"; const scenarioApp = useScenarioApp(); +const WORKFLOW_SHUTDOWN_DESCRIPTOR = { + files: { + "agent/agent.mjs": 'export default { model: "openai/gpt-5.4" };\n', + "agent/instrumentation.mjs": 'globalThis.__EVE_SHUTDOWN_PROBE__ = "one";\nexport default {};\n', + "agent/instructions.md": "Exercise development worker shutdown.\n", + }, + installDependencies: true, + name: "dev-workflow-shutdown", +} as const; + +interface WorkflowShutdownProbe { + readonly appRoot: string; + readonly eventsPath: string; + readonly lockDirectoryPath: string; +} + +async function createWorkflowShutdownProbe(): Promise { + const app = await scenarioApp(WORKFLOW_SHUTDOWN_DESCRIPTOR); + const eventsPath = join(app.appRoot, "workflow-world-events.log"); + const lockDirectoryPath = join(app.appRoot, "workflow-world-locks"); + const workflowWorldPath = join(app.appRoot, "workflow-shutdown-world.mjs"); + + await writeFile( + join(app.appRoot, "agent", "agent.mjs"), + [ + "export default {", + ' model: "openai/gpt-5.4",', + " experimental: {", + ` workflow: { world: ${JSON.stringify(workflowWorldPath)} },`, + " },", + "};", + "", + ].join("\n"), + "utf8", + ); + await writeFile( + workflowWorldPath, + [ + 'import { appendFile, mkdir, readdir, rm, writeFile } from "node:fs/promises";', + 'import { join } from "node:path";', + 'import { threadId } from "node:worker_threads";', + "", + `const eventsPath = ${JSON.stringify(eventsPath)};`, + `const lockDirectoryPath = ${JSON.stringify(lockDirectoryPath)};`, + "const instanceId = `${process.pid}:${threadId}`;", + "const instanceLockPath = join(lockDirectoryPath, `${process.pid}-${threadId}.lock`);", + "", + "export function createWorld() {", + " return {", + " specVersion: 5,", + " events: {},", + " createQueueHandler() {", + " return async () => Response.json({ ok: true });", + " },", + " async start() {", + " await mkdir(lockDirectoryPath, { recursive: true });", + " const existingLocks = await readdir(lockDirectoryPath);", + " const foreignLock = existingLocks.find(", + " (name) => !name.startsWith(`${process.pid}-`),", + " );", + " if (foreignLock !== undefined) {", + " throw new Error(`Workflow world lock is still held by ${foreignLock}.`);", + " }", + ' await writeFile(instanceLockPath, "locked", { flag: "wx" });', + " await appendFile(eventsPath, `start:${instanceId}\\n`);", + " },", + " async close() {", + " await appendFile(eventsPath, `close:${instanceId}\\n`);", + " await rm(instanceLockPath);", + " },", + " };", + "}", + "", + ].join("\n"), + "utf8", + ); + + return { + appRoot: app.appRoot, + eventsPath, + lockDirectoryPath, + }; +} + +async function readEventLines(eventsPath: string): Promise { + try { + return (await readFile(eventsPath, "utf8")).trim().split("\n").filter(Boolean); + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") { + return []; + } + throw error; + } +} describe("eve dev shutdown", () => { for (const signal of ["SIGINT", "SIGTERM"] as const) { @@ -29,4 +132,68 @@ describe("eve dev shutdown", () => { expect(exit.forcedKill).toBe(false); expect(exit.durationMs).toBeLessThan(1_000); }, 120_000); + + it("closes configured Workflow worlds on reload and before an immediate restart", async () => { + const probe = await createWorkflowShutdownProbe(); + let firstServer: RunningEveDev | undefined; + let secondServer: RunningEveDev | undefined; + + try { + firstServer = await startEveDev(probe.appRoot); + await waitForCondition( + async () => + (await readEventLines(probe.eventsPath)).filter((line) => line.startsWith("start:")) + .length === 1, + "Timed out waiting for the first Workflow world to start.", + ); + + await writeFile( + join(probe.appRoot, "agent", "instrumentation.mjs"), + 'globalThis.__EVE_SHUTDOWN_PROBE__ = "two";\nexport default {};\n', + "utf8", + ); + await forceDevelopmentRebuild(firstServer.url); + await waitForCondition(async () => { + const events = await readEventLines(probe.eventsPath); + return ( + events.filter((line) => line.startsWith("start:")).length >= 2 && + events.some((line) => line.startsWith("close:")) + ); + }, "Timed out waiting for the retired development worker to close its Workflow world."); + + const firstExit = await firstServer.signalAndAwaitExit("SIGTERM"); + expect(firstExit.forcedKill).toBe(false); + expect(firstExit.durationMs).toBeLessThan(1_000); + expect(firstExit.code).toBe(0); + expect(await readdir(probe.lockDirectoryPath)).toEqual([]); + + secondServer = await startEveDev(probe.appRoot); + await waitForCondition( + async () => + (await readEventLines(probe.eventsPath)).filter((line) => line.startsWith("start:")) + .length >= 3, + "Timed out waiting for the immediate restart to acquire its Workflow world lock.", + ); + + const secondExit = await secondServer.signalAndAwaitExit("SIGTERM"); + expect(secondExit.forcedKill).toBe(false); + expect(secondExit.durationMs).toBeLessThan(1_000); + expect(secondExit.code).toBe(0); + expect(await readdir(probe.lockDirectoryPath)).toEqual([]); + + const events = await readEventLines(probe.eventsPath); + const starts = events + .filter((line) => line.startsWith("start:")) + .map((line) => line.slice("start:".length)); + const closes = events + .filter((line) => line.startsWith("close:")) + .map((line) => line.slice("close:".length)); + + expect(starts.length).toBeGreaterThanOrEqual(3); + expect(closes.sort()).toEqual(starts.sort()); + } finally { + await firstServer?.stop(); + await secondServer?.stop(); + } + }, 120_000); });