From d1c1c7de0019d27066f68ce47d0167f43489e715 Mon Sep 17 00:00:00 2001 From: Casey Gowrie Date: Tue, 14 Jul 2026 13:52:41 -0400 Subject: [PATCH 1/4] test(dev): pin the public dev-server contract with bun and chaos coverage Port the contract, chaos, and bun suites onto the harness: an authored edit storm with a continuous health probe, worker crashes during rebuilds and open streams, bounded shutdown deadlines, forged-header client addressing, kept-alive connections across structural replacement, npm and bun installed layouts, and a fail-fast assertion for bun-run dev servers. Co-Authored-By: Claude Fable 5 Signed-off-by: Casey Gowrie --- .github/workflows/ci.yml | 5 + .../eve/src/internal/testing/scenario-app.ts | 62 +++ .../scenarios/dev-server-bun.scenario.test.ts | 75 +++ .../dev-server-chaos.scenario.test.ts | 336 ++++++++++++ .../scenarios/dev-server.scenario.test.ts | 487 +++++------------- 5 files changed, 619 insertions(+), 346 deletions(-) create mode 100644 packages/eve/test/scenarios/dev-server-bun.scenario.test.ts create mode 100644 packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3caf05fcf..a1d21a3bf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -138,6 +138,11 @@ jobs: node-version-file: .nvmrc cache: "pnpm" + - name: Setup Bun + uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2.2.0 + with: + bun-version: "1.3.14" + - name: Install dependencies run: pnpm install --frozen-lockfile diff --git a/packages/eve/src/internal/testing/scenario-app.ts b/packages/eve/src/internal/testing/scenario-app.ts index 0e018db09..ae183a2a1 100644 --- a/packages/eve/src/internal/testing/scenario-app.ts +++ b/packages/eve/src/internal/testing/scenario-app.ts @@ -8,9 +8,11 @@ import { rm, writeFile, } from "node:fs/promises"; +import { execFile } from "node:child_process"; import { createRequire } from "node:module"; import { tmpdir } from "node:os"; import { basename, dirname, join } from "node:path"; +import { promisify } from "node:util"; import { afterEach } from "vitest"; @@ -76,6 +78,13 @@ export interface ScenarioAppDescriptor { * us having to maintain a bespoke template-then-copy cache. */ readonly installDependencies?: boolean; + /** + * Package manager used for {@link installDependencies}. Defaults to pnpm. + * npm and bun installs are slower but produce the hoisted real-directory + * `node_modules/` layout, which exercises dependency resolution paths that + * pnpm's symlinked store never hits. + */ + readonly packageManager?: "bun" | "npm" | "pnpm"; } /** @@ -262,6 +271,32 @@ async function installScenarioDependencies(input: { readonly appRoot: string; readonly descriptor: ScenarioAppDescriptor; }): Promise { + if (input.descriptor.packageManager === "npm") { + // A project .npmrc overrides any host-global release-age gate, which + // would otherwise reject the workspace's freshly published dependency + // versions and make the test outcome depend on host configuration. + await writeFile(join(input.appRoot, ".npmrc"), "min-release-age=0\n"); + await runInstallCommand(input.appRoot, "npm", [ + "install", + "--no-audit", + "--no-fund", + "--ignore-scripts", + "--prefer-offline", + ]); + return; + } + if (input.descriptor.packageManager === "bun") { + // A local bunfig overrides any host-global minimumReleaseAge gate, which + // would otherwise reject the workspace's freshly published dependency + // versions and make the test outcome depend on host configuration. + await writeFile( + join(input.appRoot, "bunfig.toml"), + ["[install]", "minimumReleaseAge = 0", ""].join("\n"), + ); + await runInstallCommand(input.appRoot, "bun", ["install", "--ignore-scripts"]); + return; + } + await runPnpmCommand({ args: [ "install", @@ -275,6 +310,33 @@ async function installScenarioDependencies(input: { }); } +const runFile = promisify(execFile); + +async function runInstallCommand( + appRoot: string, + command: "bun" | "npm", + args: readonly string[], +): Promise { + try { + await runFile(command, [...args], { + cwd: appRoot, + maxBuffer: 10 * 1024 * 1024, + shell: process.platform === "win32", + }); + } catch (error) { + const failure = error as { readonly stderr?: unknown; readonly stdout?: unknown }; + throw new Error( + [ + `Command failed: ${command} ${args.join(" ")}`, + `cwd: ${appRoot}`, + `stdout:\n${typeof failure.stdout === "string" ? failure.stdout : ""}`, + `stderr:\n${typeof failure.stderr === "string" ? failure.stderr : ""}`, + ].join("\n\n"), + { cause: error }, + ); + } +} + const EVE_SCENARIO_EVE_TARBALL_PATH_ENV = "EVE_SCENARIO_EVE_TARBALL_PATH"; let cachedScenarioEveTarballPromise: Promise | null = null; diff --git a/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts b/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts new file mode 100644 index 000000000..7f2b1e931 --- /dev/null +++ b/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts @@ -0,0 +1,75 @@ +import { describe, expect, it } from "vitest"; + +import { WEATHER_AGENT_DESCRIPTOR } from "../../src/internal/testing/scenario-apps/weather-agent.js"; +import { + type ScenarioAppDescriptor, + useScenarioApp, +} from "../../src/internal/testing/scenario-app.js"; +import { sendDevelopmentMessage } from "../dev-client-harness/send-message.js"; +import { createDevelopmentSessionState } from "../dev-client-harness/session.js"; +import { + hasKnownDevServerFailure, + isBunAvailable, + runEveDevToExit, + startEveDev, +} from "./dev-server-harness.js"; + +const scenarioApp = useScenarioApp(); + +const BUN_DEV_SERVER_TIMEOUT_MS = 360_000; +const BUN_LAYOUT_DESCRIPTOR: ScenarioAppDescriptor = { + ...WEATHER_AGENT_DESCRIPTOR, + name: "weather-agent-bun", + packageManager: "bun", +}; + +const bunAvailable = isBunAvailable(); + +describe("eve dev server with bun", () => { + it.skipIf(!bunAvailable)( + "serves a bun-installed app under the Node runtime", + async () => { + const app = await scenarioApp(BUN_LAYOUT_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const messageResult = await sendDevelopmentMessage({ + message: "What's the weather in Lisbon?", + session: createDevelopmentSessionState(), + serverUrl: server.url, + }); + expect( + messageResult.events.some((event) => event.type === "message.completed"), + [ + "Expected the bun-installed dev server to complete a streamed turn.", + `stdout:\n${server.stdout()}`, + `stderr:\n${server.stderr()}`, + ].join("\n\n"), + ).toBe(true); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + await server.stop(); + } + }, + BUN_DEV_SERVER_TIMEOUT_MS, + ); + + // Bun cannot run the dev server today: Nitro wires crossws's Node adapter + // into every dev worker, and that adapter refuses to initialize when the + // Bun global exists. Until that support lands upstream, the contract is a + // fast, explanatory startup failure instead of a hang or a half-broken + // server. Replace this with a boot-and-stream assertion when `bun eve dev` + // becomes supported. + it.skipIf(!bunAvailable)( + "fails fast with the worker readiness error when the CLI runs under bun", + async () => { + const app = await scenarioApp(BUN_LAYOUT_DESCRIPTOR); + + const result = await runEveDevToExit(app.appRoot, { runtime: "bun" }); + + expect(result.code).not.toBe(0); + expect(result.output).toContain("failed before readiness"); + }, + BUN_DEV_SERVER_TIMEOUT_MS, + ); +}); diff --git a/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts b/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts new file mode 100644 index 000000000..2458b4f6c --- /dev/null +++ b/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts @@ -0,0 +1,336 @@ +import { rm, writeFile } from "node:fs/promises"; +import { join } from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { EVE_HEALTH_ROUTE_PATH } from "../../src/protocol/routes.js"; +import { WEATHER_AGENT_DESCRIPTOR } from "../../src/internal/testing/scenario-apps/weather-agent.js"; +import { + type ScenarioAppDescriptor, + useScenarioApp, +} from "../../src/internal/testing/scenario-app.js"; +import { sendDevelopmentMessage } from "../dev-client-harness/send-message.js"; +import { createDevelopmentSessionState } from "../dev-client-harness/session.js"; +import { + fetchAgentInfo, + fetchText, + forceDevelopmentRebuild, + hasKnownDevServerFailure, + startEveDev, + wait, + waitForCondition, + withinDeadline, + type RunningEveDev, +} from "./dev-server-harness.js"; + +const scenarioApp = useScenarioApp(); + +const CHAOS_SCENARIO_TIMEOUT_MS = 360_000; + +const CHAOS_CHANNEL_SOURCE = [ + 'import { randomUUID } from "node:crypto";', + 'import { defineChannel, GET } from "eve/channels";', + "", + "const workerId = randomUUID();", + "", + "export default defineChannel({", + " routes: [", + ' GET("/chaos/worker-id", () => new Response(workerId)),', + ' GET("/chaos/crash", () => {', + " setTimeout(() => process.exit(7), 25);", + ' return new Response("crashing");', + " }),", + ' GET("/chaos/request-ip", (_request, context) => new Response(String(context.requestIp))),', + ' GET("/chaos/slow-stream", () => {', + " let timer: ReturnType | undefined;", + " const body = new ReadableStream({", + " start(controller) {", + ' controller.enqueue(new TextEncoder().encode("chunk\\n"));', + " timer = setInterval(() => {", + ' controller.enqueue(new TextEncoder().encode("chunk\\n"));', + " }, 50);", + " },", + " cancel() {", + " clearInterval(timer);", + " },", + " });", + " return new Response(body);", + " }),", + " ],", + "});", + "", +].join("\n"); + +const CHAOS_DESCRIPTOR: ScenarioAppDescriptor = { + ...WEATHER_AGENT_DESCRIPTOR, + files: { + ...WEATHER_AGENT_DESCRIPTOR.files, + "agent/channels/chaos.ts": CHAOS_CHANNEL_SOURCE, + }, + name: "weather-agent-chaos", +}; + +function toolSource(marker: string): string { + return [ + 'import { defineTool } from "eve/tools";', + 'import { z } from "zod";', + "", + "export default defineTool({", + ` description: ${JSON.stringify(`Report the weather (${marker}).`)},`, + " inputSchema: z.object({ city: z.string() }),", + " execute: ({ city }) => `sunny in ${city}`,", + "});", + "", + ].join("\n"); +} + +interface HealthProbe { + readonly failures: readonly string[]; + stop(): Promise; +} + +function startHealthProbe(serverUrl: string): HealthProbe { + const failures: string[] = []; + let probes = 0; + let stopped = false; + const run = (async () => { + while (!stopped) { + probes += 1; + try { + const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, serverUrl)); + if (response.status !== 200) { + failures.push(`health returned ${String(response.status)}`); + } + await response.arrayBuffer(); + } catch (error) { + failures.push(`health request failed: ${String(error)}`); + } + await wait(50); + } + })(); + + return { + failures, + async stop() { + stopped = true; + await run; + return probes; + }, + }; +} + +async function completeStreamedTurn(server: RunningEveDev, message: string): Promise { + const result = await sendDevelopmentMessage({ + message, + session: createDevelopmentSessionState(), + serverUrl: server.url, + }); + expect( + result.events.some((event) => event.type === "message.completed"), + [ + "Expected a streamed turn to complete after the chaos sequence.", + `stdout:\n${server.stdout()}`, + `stderr:\n${server.stderr()}`, + ].join("\n\n"), + ).toBe(true); +} + +async function waitForToolMarker(serverUrl: string, marker: string): Promise { + await waitForCondition(async () => { + const info = await fetchAgentInfo(serverUrl); + return info.tools.authored.some((tool) => tool.description?.includes(marker)); + }, `Timed out waiting for the "${marker}" tool revision to publish.`); +} + +describe("eve dev server chaos", () => { + it( + "keeps every request succeeding through an authored edit storm", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const toolPath = join(app.appRoot, "agent", "tools", "get_weather.ts"); + const server = await startEveDev(app.appRoot); + const probe = startHealthProbe(server.url); + + try { + for (let round = 1; round <= 3; round += 1) { + await writeFile(toolPath, toolSource(`storm-${String(round)}-a`)); + await writeFile(toolPath, toolSource(`storm-${String(round)}-b`)); + await writeFile(toolPath, "export default { this is not valid typescript\n"); + await wait(500); + await rm(toolPath); + await writeFile(toolPath, toolSource(`storm-${String(round)}-fixed`)); + await Promise.all([ + forceDevelopmentRebuild(server.url), + forceDevelopmentRebuild(server.url), + ]); + } + + await writeFile(join(app.appRoot, ".env.local"), "EVE_CHAOS_STRUCTURAL=1\n"); + await writeFile(toolPath, toolSource("storm-final")); + await forceDevelopmentRebuild(server.url); + await waitForToolMarker(server.url, "storm-final"); + await completeStreamedTurn(server, "What's the weather in Lisbon?"); + + const probes = await probe.stop(); + expect(probe.failures, probe.failures.join("\n")).toEqual([]); + expect(probes).toBeGreaterThan(50); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + await probe.stop(); + await server.stop(); + } + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); + + it( + "recovers through worker crashes, aborted streams, and concurrent rebuilds", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const firstWorkerId = await fetchText(server.url, "/chaos/worker-id"); + + const abort = new AbortController(); + const streamResponse = await fetch(new URL("/chaos/slow-stream", server.url), { + signal: abort.signal, + }); + const reader = streamResponse.body?.getReader(); + await reader?.read(); + abort.abort(); + + await expect(fetch(new URL("/chaos/crash", server.url))).resolves.toBeDefined(); + await waitForCondition(async () => { + try { + return (await fetchText(server.url, "/chaos/worker-id")) !== firstWorkerId; + } catch { + return false; + } + }, "Timed out waiting for a replacement worker after the crash."); + const burst = await Promise.all( + Array.from({ length: 10 }, async () => { + const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, server.url)); + return response.status; + }), + ); + expect(burst).toEqual(Array.from({ length: 10 }, () => 200)); + + await Promise.all([ + fetch(new URL("/chaos/crash", server.url)), + forceDevelopmentRebuild(server.url), + ]); + await waitForCondition(async () => { + try { + const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, server.url)); + return response.status === 200; + } catch { + return false; + } + }, "Timed out waiting for the dev server to recover from a crash during rebuild."); + + await completeStreamedTurn(server, "What's the weather in Lisbon?"); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + await server.stop(); + } + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); + + it( + "terminates an open stream within a bounded deadline when its worker crashes", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const streamResponse = await fetch(new URL("/chaos/slow-stream", server.url)); + const reader = streamResponse.body?.getReader(); + await reader?.read(); + + await fetch(new URL("/chaos/crash", server.url)); + const settled = await withinDeadline( + (async () => { + for (;;) { + const result = await reader?.read(); + if (result === undefined || result.done) { + return "done"; + } + } + })().catch(() => "errored"), + "Timed out waiting for the crashed worker's stream to settle.", + 15_000, + ); + expect(["done", "errored"]).toContain(settled); + + await waitForCondition(async () => { + try { + const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, server.url)); + return response.status === 200; + } catch { + return false; + } + }, "Timed out waiting for the dev server to recover after the crash."); + await completeStreamedTurn(server, "What's the weather in Lisbon?"); + } finally { + await server.stop(); + } + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); + + it( + "shuts down within a bounded deadline while a stream is open", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + const streamResponse = await fetch(new URL("/chaos/slow-stream", server.url)); + const reader = streamResponse.body?.getReader(); + await reader?.read(); + + const stopStart = Date.now(); + await server.stop(); + + // The harness escalates to SIGKILL at 10s; a graceful exit must beat it. + expect(Date.now() - stopStart).toBeLessThan(9_000); + await expect( + withinDeadline( + (async () => { + for (;;) { + const result = await reader?.read(); + if (result === undefined || result.done) { + return "done"; + } + } + })().catch(() => "errored"), + "Timed out waiting for the shutdown stream to settle.", + 5_000, + ), + ).resolves.toBeDefined(); + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); + + it( + "reports the socket peer as the client address despite forged headers", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const response = await fetch(new URL("/chaos/request-ip", server.url), { + headers: { + "x-eve-dev-workflow-delivery": "forged.forged", + "x-forwarded-for": "203.0.113.7", + }, + }); + expect(response.status).toBe(200); + await expect(response.text()).resolves.toBe("127.0.0.1"); + } finally { + await server.stop(); + } + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); +}); diff --git a/packages/eve/test/scenarios/dev-server.scenario.test.ts b/packages/eve/test/scenarios/dev-server.scenario.test.ts index 06122f73e..8d1979540 100644 --- a/packages/eve/test/scenarios/dev-server.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server.scenario.test.ts @@ -1,18 +1,11 @@ -import { spawn, type ChildProcessByStdio } from "node:child_process"; -import { existsSync, watch as watchFileSystem } from "node:fs"; +import { Agent } from "node:http"; +import { existsSync } from "node:fs"; import { readFile, rm, writeFile } from "node:fs/promises"; -import { dirname, join } from "node:path"; -import type { Readable } from "node:stream"; +import { join } from "node:path"; import { describe, expect, it } from "vitest"; -import { - EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH, - EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH, - EVE_HEALTH_ROUTE_PATH, - EVE_INFO_ROUTE_PATH, -} from "../../src/protocol/routes.js"; -import type { AgentInfoResponse } from "../../src/internal/nitro/routes/agent-info/build-agent-info-response.js"; +import { EVE_HEALTH_ROUTE_PATH } from "../../src/protocol/routes.js"; import { readDevelopmentRuntimeArtifactsSnapshotRoot, resolveDevelopmentRuntimeArtifactsPointerPath, @@ -25,6 +18,18 @@ import { } from "../../src/internal/testing/scenario-app.js"; import { sendDevelopmentMessage } from "../dev-client-harness/send-message.js"; import { createDevelopmentSessionState } from "../dev-client-harness/session.js"; +import { + fetchAgentInfo, + fetchText, + forceDevelopmentRebuild, + hasKnownDevServerFailure, + readDevelopmentRevision, + requestWithAgent, + startEveDev, + waitForCondition, + waitForPath, + withinDeadline, +} from "./dev-server-harness.js"; // Keep the dev TUI's glyph set deterministic across CI hosts so the // screen assertions below remain stable. @@ -60,7 +65,7 @@ const WEBSOCKET_DEV_SERVER_DESCRIPTOR: ScenarioAppDescriptor = { "export default defineChannel({", ' routes: [WS("/socket", () => ({', " message(peer, message) {", - ' const transportHeader = peer.request.headers.has("x-eve-dev-worker-metadata") ? "exposed" : "hidden";', + ' const transportHeader = peer.request.headers.has("x-eve-dev-workflow-delivery") ? "exposed" : "hidden";', ' peer.send(`${transportHeader}:${peer.remoteAddress ?? "missing"}:${message.text()}`);', " },", " }))],", @@ -77,6 +82,26 @@ const TRANSACTIONAL_REBUILD_DESCRIPTOR: ScenarioAppDescriptor = { "agent/instrumentation.ts": createInstrumentationSource("one"), }, }; +const SCHEDULE_DISPATCH_DESCRIPTOR: ScenarioAppDescriptor = { + ...WEATHER_AGENT_DESCRIPTOR, + files: { + ...WEATHER_AGENT_DESCRIPTOR.files, + "agent/schedules/heartbeat.md": [ + "---", + 'cron: "0 0 * * 0"', + "---", + "", + "Report the weather in Lisbon.", + "", + ].join("\n"), + }, + name: "weather-agent-schedules", +}; +const NPM_LAYOUT_DESCRIPTOR: ScenarioAppDescriptor = { + ...WEATHER_AGENT_DESCRIPTOR, + name: "weather-agent-npm", + packageManager: "npm", +}; const STREAM_PROMOTION_DESCRIPTOR: ScenarioAppDescriptor = { ...TRANSACTIONAL_REBUILD_DESCRIPTOR, files: { @@ -269,168 +294,6 @@ function createBlockedInstrumentationSource(): string { ].join("\n"); } -interface RunningEveDev { - crash(): Promise; - readonly stderr: () => string; - readonly stdout: () => string; - readonly url: string; - stop(): Promise; -} - -function stripAnsi(text: string): string { - return text - .split("\u001b[") - .map((segment, index) => { - if (index === 0) { - return segment; - } - - return segment.replace(/^[0-9;]*m/, ""); - }) - .join(""); -} - -function hasUnsupportedWindowsEsmImport(text: string): boolean { - return ( - text.includes("ERR_UNSUPPORTED_ESM_URL_SCHEME") || - text.includes("Received protocol 'g:'") || - text.includes('Received protocol "g:"') - ); -} - -function hasKnownDevServerFailure(text: string): boolean { - return ( - hasUnsupportedWindowsEsmImport(text) || - text.includes("UNRESOLVED_IMPORT") || - text.includes("ECONNRESET") || - text.includes("socket hang up") || - (text.includes("ERR_MODULE_NOT_FOUND") && text.includes("authored-module-map-loader")) - ); -} - -function parseServerUrl(stdout: string): string | undefined { - const match = /server listening at (https?:\/\/\S+)/.exec(stripAnsi(stdout)); - - return match?.[1]; -} - -async function wait(ms: number): Promise { - await new Promise((resolve) => { - setTimeout(resolve, ms); - }); -} - -async function waitForCondition( - condition: () => boolean | Promise, - failureMessage: string, - timeoutMs: number = 60_000, -): Promise { - const deadline = Date.now() + timeoutMs; - - while (!(await condition())) { - if (Date.now() >= deadline) { - throw new Error(failureMessage); - } - await wait(100); - } -} - -async function waitForServerUrl(input: { - readonly child: ChildProcessByStdio; - readonly getOutput: () => { - readonly stderr: string; - readonly stdout: string; - }; -}): Promise { - return await new Promise((resolve, reject) => { - let settled = false; - - const timeout = setTimeout(() => { - settleReject( - new Error( - [ - "Timed out waiting for eve dev to print its server URL.", - `stdout:\n${input.getOutput().stdout}`, - `stderr:\n${input.getOutput().stderr}`, - ].join("\n\n"), - ), - ); - }, 120_000); - - const cleanup = () => { - clearTimeout(timeout); - input.child.stdout.off("data", handleOutput); - input.child.stderr.off("data", handleOutput); - input.child.off("error", settleReject); - input.child.off("exit", handleExit); - }; - - const settleResolve = (url: string) => { - if (settled) { - return; - } - - settled = true; - cleanup(); - resolve(url); - }; - - function settleReject(error: unknown) { - if (settled) { - return; - } - - settled = true; - cleanup(); - reject(error); - } - - function handleOutput() { - const output = input.getOutput(); - const combinedOutput = `${output.stdout}\n${output.stderr}`; - - if (hasKnownDevServerFailure(combinedOutput)) { - settleReject( - new Error( - [ - "eve dev emitted a known reload or generated-bundle failure.", - `stdout:\n${output.stdout}`, - `stderr:\n${output.stderr}`, - ].join("\n\n"), - ), - ); - return; - } - - const url = parseServerUrl(output.stdout); - - if (url !== undefined) { - settleResolve(url); - } - } - - function handleExit(code: number | null, signal: NodeJS.Signals | null) { - const output = input.getOutput(); - - settleReject( - new Error( - [ - `eve dev exited before printing its server URL (code ${String(code)}, signal ${String(signal)}).`, - `stdout:\n${output.stdout}`, - `stderr:\n${output.stderr}`, - ].join("\n\n"), - ), - ); - } - - input.child.stdout.on("data", handleOutput); - input.child.stderr.on("data", handleOutput); - input.child.once("error", settleReject); - input.child.once("exit", handleExit); - handleOutput(); - }); -} - async function waitForWebSocketOpen(socket: WebSocket): Promise { await waitForWebSocketEvent(socket, "open", () => undefined); } @@ -471,94 +334,6 @@ async function waitForWebSocketEvent( }); } -async function startEveDev(appRoot: string): Promise { - return await startEveDevProcess(appRoot, {}); -} - -async function startEveDevWithBoundedWorkflowRecovery(appRoot: string): Promise { - return await startEveDevProcess(appRoot, { - WORKFLOW_INLINE_OWNERSHIP_LEASE_SECONDS: "1", - }); -} - -async function startEveDevProcess( - appRoot: string, - environment: Readonly>, -): Promise { - const eveBinPath = join(appRoot, "node_modules", "eve", "bin", "eve.js"); - const child = spawn( - process.execPath, - [eveBinPath, "dev", "--no-ui", "--host", "127.0.0.1", "--port", "0"], - { - cwd: appRoot, - env: { - ...process.env, - ...environment, - // Activate the deterministic mock-model adapter in the spawned dev - // server so the streamed turn completes without model credentials. - NODE_ENV: "test", - }, - stdio: ["ignore", "pipe", "pipe"], - }, - ); - let stderr = ""; - let stdout = ""; - - child.stdout.setEncoding("utf8"); - child.stderr.setEncoding("utf8"); - child.stdout.on("data", (chunk: string) => { - stdout += chunk; - }); - child.stderr.on("data", (chunk: string) => { - stderr += chunk; - }); - - const url = await waitForServerUrl({ - child, - getOutput: () => ({ - stderr, - stdout, - }), - }); - await waitForPath(join(appRoot, ".eve", "dev-server-state.v1.json")); - - return { - async crash() { - if (child.exitCode !== null || child.signalCode !== null) { - return; - } - await withinDeadline( - new Promise((resolve) => { - child.once("exit", () => resolve()); - child.kill("SIGKILL"); - }), - "Timed out waiting for the dev server process to crash.", - ); - }, - stderr: () => stderr, - stdout: () => stdout, - async stop() { - if (child.exitCode !== null || child.signalCode !== null) { - return; - } - - await new Promise((resolve) => { - const timeout = setTimeout(() => { - child.kill("SIGKILL"); - resolve(); - }, 10_000); - - child.once("exit", () => { - clearTimeout(timeout); - resolve(); - }); - child.kill("SIGTERM"); - }); - }, - url, - }; -} - describe("eve dev server", () => { it( "publishes authored tool removals without replacing the active host", @@ -738,6 +513,39 @@ describe("eve dev server", () => { DEV_SERVER_SCENARIO_TIMEOUT_MS, ); + it( + "serves a kept-alive connection across a structural worker replacement", + async () => { + const app = await scenarioApp(TRANSACTIONAL_REBUILD_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + const agent = new Agent({ keepAlive: true, maxSockets: 1 }); + + try { + const first = await requestWithAgent(new URL("/worker-id", server.url).href, agent); + expect(first.localPort).toBeDefined(); + + await writeFile( + join(app.appRoot, "agent", "instrumentation.ts"), + createInstrumentationSource("keep-alive-two"), + ); + await forceDevelopmentRebuild(server.url); + await waitForCondition( + async () => (await fetchText(server.url, "/instrumentation-marker")) === "keep-alive-two", + `Timed out waiting for the structural replacement.\n\nstdout:\n${server.stdout()}\n\nstderr:\n${server.stderr()}`, + ); + + const second = await requestWithAgent(new URL("/worker-id", server.url).href, agent); + expect(second.localPort).toBe(first.localPort); + expect(second.body).not.toBe(first.body); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + agent.destroy(); + await server.stop(); + } + }, + DEV_SERVER_SCENARIO_TIMEOUT_MS, + ); + it( "keeps an admitted websocket on its worker while a ready replacement is promoted", async () => { @@ -923,7 +731,9 @@ describe("eve dev server", () => { "recovers a nonterminal child Workflow on its selected generation after restart", async () => { const app = await scenarioApp(WORKFLOW_GENERATION_DESCRIPTOR); - let server = await startEveDevWithBoundedWorkflowRecovery(app.appRoot); + let server = await startEveDev(app.appRoot, { + env: { WORKFLOW_INLINE_OWNERSHIP_LEASE_SECONDS: "1" }, + }); const turnStartedPath = join(app.appRoot, ".turn-started"); const restartPath = join(app.appRoot, ".restart-generation-test"); const recoveredPath = join(app.appRoot, ".recovered-turn-started"); @@ -957,7 +767,9 @@ describe("eve dev server", () => { await server.crash(); await withinDeadline(interruptedTurn, "Interrupted client stream did not settle."); await writeFile(restartPath, "restart"); - server = await startEveDevWithBoundedWorkflowRecovery(app.appRoot); + server = await startEveDev(app.appRoot, { + env: { WORKFLOW_INLINE_OWNERSHIP_LEASE_SECONDS: "1" }, + }); await waitForPath(recoveredPath); await expect( readFile(recoveredPath, "utf8").then((source) => JSON.parse(source) as unknown), @@ -973,6 +785,71 @@ describe("eve dev server", () => { }, DEV_SERVER_SCENARIO_TIMEOUT_MS, ); + + it( + "dispatches an authored schedule through the dev route on its generation", + async () => { + const app = await scenarioApp(SCHEDULE_DISPATCH_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const response = await fetch(new URL("/eve/v1/dev/schedules/heartbeat", server.url), { + method: "POST", + }); + const body = (await response.json()) as { + scheduleId?: string; + sessionIds?: readonly string[]; + }; + expect( + response.status, + [ + `Expected the dev schedule dispatch route to succeed: ${JSON.stringify(body)}`, + `stdout:\n${server.stdout()}`, + `stderr:\n${server.stderr()}`, + ].join("\n\n"), + ).toBe(200); + expect(body.scheduleId).toBe("heartbeat"); + expect(body.sessionIds).toHaveLength(1); + + const unknown = await fetch(new URL("/eve/v1/dev/schedules/missing", server.url), { + method: "POST", + }); + expect(unknown.status).toBe(404); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + await server.stop(); + } + }, + DEV_SERVER_SCENARIO_TIMEOUT_MS, + ); + + it( + "serves an npm-installed app with hoisted real-directory dependencies", + async () => { + const app = await scenarioApp(NPM_LAYOUT_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const messageResult = await sendDevelopmentMessage({ + message: "What's the weather in Lisbon?", + session: createDevelopmentSessionState(), + serverUrl: server.url, + }); + expect( + messageResult.events.some((event) => event.type === "message.completed"), + [ + "Expected the npm-installed dev server to complete a streamed turn.", + `stdout:\n${server.stdout()}`, + `stderr:\n${server.stderr()}`, + ].join("\n\n"), + ).toBe(true); + expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); + } finally { + await server.stop(); + } + }, + DEV_SERVER_SCENARIO_TIMEOUT_MS, + ); }); function readCompletedMessages( @@ -991,85 +868,3 @@ function readCompletedMessages( function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } - -async function fetchText(serverUrl: string, path: string): Promise { - const response = await fetch(new URL(path, serverUrl)); - if (!response.ok) { - throw new Error(`Expected ${path} to succeed, received ${String(response.status)}.`); - } - return await response.text(); -} - -async function fetchAgentInfo(serverUrl: string): Promise { - const response = await fetch(new URL(EVE_INFO_ROUTE_PATH, serverUrl)); - if (!response.ok) { - throw new Error(`Expected agent info to succeed, received ${String(response.status)}.`); - } - return (await response.json()) as AgentInfoResponse; -} - -async function forceDevelopmentRebuild(serverUrl: string): Promise { - const rebuildUrl = new URL(EVE_DEV_RUNTIME_ARTIFACTS_REBUILD_ROUTE_PATH, serverUrl); - rebuildUrl.searchParams.set("force", "1"); - const response = await fetch(rebuildUrl); - if (!response.ok) { - throw new Error(`Development rebuild failed with status ${String(response.status)}.`); - } -} - -async function readDevelopmentRevision(serverUrl: string): Promise { - const response = await fetch(new URL(EVE_DEV_RUNTIME_ARTIFACTS_ROUTE_PATH, serverUrl)); - if (!response.ok) { - throw new Error(`Development runtime state failed with status ${String(response.status)}.`); - } - const body = (await response.json()) as { readonly revision?: unknown }; - if (typeof body.revision !== "string") { - throw new Error("Development runtime state did not include a revision."); - } - return body.revision; -} - -async function waitForPath(path: string, timeoutMs: number = 60_000): Promise { - if (existsSync(path)) { - return; - } - await new Promise((resolve, reject) => { - let settled = false; - const watcher = watchFileSystem(dirname(path), () => { - if (existsSync(path)) { - settle(resolve); - } - }); - const timeout = setTimeout(() => { - settle(() => reject(new Error(`Timed out waiting for ${path}.`))); - }, timeoutMs); - function settle(complete: () => void) { - if (settled) { - return; - } - settled = true; - clearTimeout(timeout); - watcher.close(); - complete(); - } - if (existsSync(path)) { - settle(resolve); - } - }); -} - -async function withinDeadline(operation: Promise, failureMessage: string): Promise { - let timeout: NodeJS.Timeout | undefined; - try { - return await Promise.race([ - operation, - new Promise((_resolve, reject) => { - timeout = setTimeout(() => reject(new Error(failureMessage)), 60_000); - }), - ]); - } finally { - if (timeout !== undefined) { - clearTimeout(timeout); - } - } -} From 302ebfc87f772c9b04cb9febf8148eba24bbb7a1 Mon Sep 17 00:00:00 2001 From: Casey Gowrie Date: Tue, 14 Jul 2026 16:19:27 -0400 Subject: [PATCH 2/4] test(dev): harden the contract suite against vacuous passes - The forged-header test forges the real trusted metadata headers and the new untrusted-transport test drives 401s through the public listener for both the World RPC route and a forged queue delivery. - The edit storm's broken-source phase is deterministic: a forced rebuild of broken source must leave the published revision unchanged under the zero-failure probe, instead of relying on a timing window. - The promotion test polls the parent-owned runtime-artifacts route continuously through candidate preparation and promotion. - hasKnownDevServerFailure additionally matches unhandled rejections and failed worker restarts; health probes carry a per-request timeout; the crash response asserts its status; the bun suite fails in CI when bun is missing instead of silently skipping. Co-Authored-By: Claude Fable 5 Signed-off-by: Casey Gowrie --- .../scenarios/dev-server-bun.scenario.test.ts | 6 ++ .../dev-server-chaos.scenario.test.ts | 63 ++++++++++++++++--- .../eve/test/scenarios/dev-server-harness.ts | 3 + .../scenarios/dev-server.scenario.test.ts | 19 ++++++ 4 files changed, 82 insertions(+), 9 deletions(-) diff --git a/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts b/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts index 7f2b1e931..7cd0cf632 100644 --- a/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server-bun.scenario.test.ts @@ -26,6 +26,12 @@ const BUN_LAYOUT_DESCRIPTOR: ScenarioAppDescriptor = { const bunAvailable = isBunAvailable(); describe("eve dev server with bun", () => { + it("keeps bun available in CI so the suite cannot silently skip", () => { + if (process.env.CI !== undefined) { + expect(bunAvailable).toBe(true); + } + }); + it.skipIf(!bunAvailable)( "serves a bun-installed app under the Node runtime", async () => { diff --git a/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts b/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts index 2458b4f6c..9f49226ae 100644 --- a/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server-chaos.scenario.test.ts @@ -16,6 +16,7 @@ import { fetchText, forceDevelopmentRebuild, hasKnownDevServerFailure, + readDevelopmentRevision, startEveDev, wait, waitForCondition, @@ -97,7 +98,9 @@ function startHealthProbe(serverUrl: string): HealthProbe { while (!stopped) { probes += 1; try { - const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, serverUrl)); + const response = await fetch(new URL(EVE_HEALTH_ROUTE_PATH, serverUrl), { + signal: AbortSignal.timeout(10_000), + }); if (response.status !== 200) { failures.push(`health returned ${String(response.status)}`); } @@ -155,8 +158,13 @@ describe("eve dev server chaos", () => { for (let round = 1; round <= 3; round += 1) { await writeFile(toolPath, toolSource(`storm-${String(round)}-a`)); await writeFile(toolPath, toolSource(`storm-${String(round)}-b`)); + const revisionBeforeBreak = await readDevelopmentRevision(server.url); await writeFile(toolPath, "export default { this is not valid typescript\n"); - await wait(500); + // The watcher survives a failed candidate, so the forced rebuild + // resolves — the deterministic proof the broken build was + // attempted and rejected is the unchanged revision. + await forceDevelopmentRebuild(server.url); + await expect(readDevelopmentRevision(server.url)).resolves.toBe(revisionBeforeBreak); await rm(toolPath); await writeFile(toolPath, toolSource(`storm-${String(round)}-fixed`)); await Promise.all([ @@ -173,7 +181,9 @@ describe("eve dev server chaos", () => { const probes = await probe.stop(); expect(probe.failures, probe.failures.join("\n")).toEqual([]); - expect(probes).toBeGreaterThan(50); + // The floor proves the probe ran continuously through the storm; + // the storm itself is deterministic-length now, not wall-clock. + expect(probes).toBeGreaterThan(30); expect(hasKnownDevServerFailure(`${server.stdout()}\n${server.stderr()}`)).toBe(false); } finally { await probe.stop(); @@ -200,7 +210,8 @@ describe("eve dev server chaos", () => { await reader?.read(); abort.abort(); - await expect(fetch(new URL("/chaos/crash", server.url))).resolves.toBeDefined(); + const crashResponse = await fetch(new URL("/chaos/crash", server.url)); + expect([200, 503]).toContain(crashResponse.status); await waitForCondition(async () => { try { return (await fetchText(server.url, "/chaos/worker-id")) !== firstWorkerId; @@ -250,19 +261,18 @@ describe("eve dev server chaos", () => { await reader?.read(); await fetch(new URL("/chaos/crash", server.url)); - const settled = await withinDeadline( + await withinDeadline( (async () => { for (;;) { const result = await reader?.read(); if (result === undefined || result.done) { - return "done"; + return; } } - })().catch(() => "errored"), + })().catch(() => undefined), "Timed out waiting for the crashed worker's stream to settle.", 15_000, ); - expect(["done", "errored"]).toContain(settled); await waitForCondition(async () => { try { @@ -319,9 +329,13 @@ describe("eve dev server chaos", () => { const server = await startEveDev(app.appRoot); try { + // Forged client-address metadata is stripped and re-stamped by the + // parent from the accepted socket, so the handler observes the real + // peer even when a client supplies every trusted header name. const response = await fetch(new URL("/chaos/request-ip", server.url), { headers: { - "x-eve-dev-workflow-delivery": "forged.forged", + "x-eve-dev-client-address": "203.0.113.7", + "x-eve-dev-client-address-signature": "forged", "x-forwarded-for": "203.0.113.7", }, }); @@ -333,4 +347,35 @@ describe("eve dev server chaos", () => { }, CHAOS_SCENARIO_TIMEOUT_MS, ); + + it( + "rejects untrusted internal transport requests on the public listener", + async () => { + const app = await scenarioApp(CHAOS_DESCRIPTOR); + const server = await startEveDev(app.appRoot); + + try { + const worldCall = await fetch(new URL("/eve/v1/dev/internal/workflow-world", server.url), { + body: "{}", + method: "POST", + }); + expect(worldCall.status).toBe(401); + + const forgedDelivery = await fetch(new URL("/.well-known/workflow/v1/flow", server.url), { + body: JSON.stringify({ runId: "wrun_forged" }), + headers: { + "x-eve-dev-workflow-delivery": "forged", + "x-vqs-message-attempt": "1", + "x-vqs-message-id": "msg_forged", + "x-vqs-queue-name": "forged-queue", + }, + method: "POST", + }); + expect(forgedDelivery.status).toBe(401); + } finally { + await server.stop(); + } + }, + CHAOS_SCENARIO_TIMEOUT_MS, + ); }); diff --git a/packages/eve/test/scenarios/dev-server-harness.ts b/packages/eve/test/scenarios/dev-server-harness.ts index 95acabb81..c61055c7f 100644 --- a/packages/eve/test/scenarios/dev-server-harness.ts +++ b/packages/eve/test/scenarios/dev-server-harness.ts @@ -175,6 +175,9 @@ export function hasKnownDevServerFailure(text: string): boolean { text.includes("UNRESOLVED_IMPORT") || text.includes("ECONNRESET") || text.includes("socket hang up") || + text.includes("UnhandledPromiseRejection") || + text.includes("ERR_UNHANDLED_REJECTION") || + text.includes("dev worker restart failed") || (text.includes("ERR_MODULE_NOT_FOUND") && text.includes("authored-module-map-loader")) ); } diff --git a/packages/eve/test/scenarios/dev-server.scenario.test.ts b/packages/eve/test/scenarios/dev-server.scenario.test.ts index 8d1979540..5961074a7 100644 --- a/packages/eve/test/scenarios/dev-server.scenario.test.ts +++ b/packages/eve/test/scenarios/dev-server.scenario.test.ts @@ -484,6 +484,21 @@ describe("eve dev server", () => { ); expect(new TextDecoder().decode(firstChunk.value)).toBe("first\n"); + // The parent-owned control route must answer continuously through + // candidate preparation and promotion, not merely at spot checks. + const revisionFailures: string[] = []; + let stopRevisionPolling = false; + const revisionPoller = (async () => { + while (!stopRevisionPolling) { + try { + await readDevelopmentRevision(server.url); + } catch (error) { + revisionFailures.push(String(error)); + } + await new Promise((resolvePromise) => setTimeout(resolvePromise, 50)); + } + })(); + await writeFile( join(app.appRoot, "agent", "instrumentation.ts"), createBlockedInstrumentationSource(), @@ -498,6 +513,10 @@ describe("eve dev server", () => { await withinDeadline(rebuild, "Timed out waiting for candidate promotion."); await expect(fetchText(server.url, "/instrumentation-marker")).resolves.toBe("two"); + stopRevisionPolling = true; + await revisionPoller; + expect(revisionFailures, revisionFailures.join("\n")).toEqual([]); + await writeFile(streamReleasePath, "release"); const secondChunk = await withinDeadline( reader.read(), From 36189969e602488a730dfb3bb0fc108c863c9840 Mon Sep 17 00:00:00 2001 From: Casey Gowrie Date: Tue, 14 Jul 2026 16:25:41 -0400 Subject: [PATCH 3/4] test(dev): pin exchange release for aborted uploads and cancelled retired streams Ports the two failure-mode regressions the previous stack's transport suite covered that this stack did not: a client aborting a streaming request body releases the exchange so the retired worker drains, and a client cancelling a stream served by an already-retired worker releases its last exchange and disposes it. Co-Authored-By: Claude Fable 5 Signed-off-by: Casey Gowrie --- .../drained-nitro-dev-server.scenario.test.ts | 94 +++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts index c1e1c0db7..5826e069f 100644 --- a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts +++ b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts @@ -304,6 +304,100 @@ describe("drained Nitro dev server", () => { await server.close(); }); + it("releases the retired worker when a client aborts a request body", async () => { + let sawBody: (() => void) | undefined; + const bodySeen = new Promise((resolve) => { + sawBody = resolve; + }); + const { createRunner, runners } = createRunnerFactory(async (request, runnerIndex) => { + if (runnerIndex === 0) { + sawBody?.(); + // Consuming the body only completes when the client finishes or + // aborts the upload; an abort must reject this read. + await request.text(); + return new Response("unreachable"); + } + return new Response("runner-1"); + }); + const server = new DrainedNitroDevServer(LOGGER, createRunner); + const listener = await listen(server); + const disposeFirst = vi.fn(async () => undefined); + await server.replaceWorker(replacement("/tmp/first.mjs", disposeFirst)); + + const abort = new AbortController(); + let releaseChunks: (() => void) | undefined; + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("partial-upload")); + releaseChunks = () => controller.close(); + }, + }); + const upload = fetch(new URL("/", listener.url), { + body, + duplex: "half", + method: "POST", + signal: abort.signal, + } as RequestInit).catch(() => undefined); + await withinDeadline(bodySeen, "Timed out waiting for the upload to be admitted."); + abort.abort(); + await upload; + releaseChunks?.(); + + await server.replaceWorker(replacement("/tmp/second.mjs")); + await withinDeadline( + vi.waitFor(() => { + expect(runners[0]?.closeMock).toHaveBeenCalled(); + expect(disposeFirst).toHaveBeenCalledOnce(); + }), + "Timed out waiting for the aborted upload to release the retired worker.", + ); + + await server.close(); + }); + + it("releases the retired worker when a client cancels its streamed response", async () => { + const { createRunner, runners } = createRunnerFactory(async (_request, runnerIndex) => { + if (runnerIndex === 0) { + // A response held open indefinitely: only client cancellation or + // worker retirement can end it. + const body = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode("started\n")); + }, + }); + return new Response(body); + } + return new Response("runner-1"); + }); + const server = new DrainedNitroDevServer(LOGGER, createRunner); + const listener = await listen(server); + const disposeFirst = vi.fn(async () => undefined); + await server.replaceWorker(replacement("/tmp/first.mjs", disposeFirst)); + + const streaming = await fetch(new URL("/", listener.url)); + const reader = streaming.body?.getReader(); + await reader?.read(); + + await server.replaceWorker(replacement("/tmp/second.mjs")); + await expect( + fetch(new URL("/", listener.url)).then(async (response) => await response.text()), + ).resolves.toBe("runner-1"); + expect(disposeFirst).not.toHaveBeenCalled(); + + // Client-side cancellation of the retired worker's stream must release + // its last exchange and retire it. + await reader?.cancel(); + await withinDeadline( + vi.waitFor(() => { + expect(runners[0]?.closeMock).toHaveBeenCalled(); + expect(disposeFirst).toHaveBeenCalledOnce(); + }), + "Timed out waiting for the cancelled stream to release the retired worker.", + ); + + await server.close(); + }); + it("closes within a bounded interval while a candidate is mid-readiness", async () => { const { createRunner } = createRunnerFactory( async () => new Response("ok"), From 2d96936a60e8c7a497c36c7f232aed60b5b47247 Mon Sep 17 00:00:00 2001 From: Casey Gowrie Date: Tue, 14 Jul 2026 18:22:44 -0400 Subject: [PATCH 4/4] fix(dev): close remaining dev server review gaps Signed-off-by: Casey Gowrie --- .../internal/nitro/dev-client-address.test.ts | 12 +++ .../src/internal/nitro/dev-client-address.ts | 31 +++++-- .../dev-authored-rebuild-coordinator.test.ts | 45 +++++++++- .../host/dev-authored-rebuild-coordinator.ts | 26 ++++++ .../drained-nitro-dev-server.scenario.test.ts | 83 +++++++++++++++++++ .../nitro/host/drained-nitro-dev-server.ts | 25 +++--- .../workflow/development-world-client.ts | 19 ++++- .../workflow/development-world-codec.test.ts | 14 ++++ .../workflow/development-world-codec.ts | 45 ++++++++-- ...velopment-world-server.integration.test.ts | 56 +++++++++++++ .../eve/test/scenarios/dev-server-harness.ts | 61 +++++++++----- 11 files changed, 369 insertions(+), 48 deletions(-) diff --git a/packages/eve/src/internal/nitro/dev-client-address.test.ts b/packages/eve/src/internal/nitro/dev-client-address.test.ts index b42a8dbe8..a0426c339 100644 --- a/packages/eve/src/internal/nitro/dev-client-address.test.ts +++ b/packages/eve/src/internal/nitro/dev-client-address.test.ts @@ -27,6 +27,18 @@ describe("development client address metadata", () => { expect(readTrustedDevelopmentClientAddress(headers, SECRET)).toBe("10.0.0.9"); }); + it("stamps Node request headers used by WebSocket upgrades", () => { + const headers = { + [DEVELOPMENT_CLIENT_ADDRESS_HEADER]: "203.0.113.7", + [DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER]: "forged", + }; + stampDevelopmentClientAddress(headers, "10.0.0.9", SECRET); + + expect(headers[DEVELOPMENT_CLIENT_ADDRESS_HEADER]).toBe("10.0.0.9"); + expect(headers[DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER]).not.toBe("forged"); + expect(readTrustedDevelopmentClientAddress(new Headers(headers), SECRET)).toBe("10.0.0.9"); + }); + it("drops the metadata entirely when no secret or address is available", () => { const headers = new Headers({ [DEVELOPMENT_CLIENT_ADDRESS_HEADER]: "203.0.113.7", diff --git a/packages/eve/src/internal/nitro/dev-client-address.ts b/packages/eve/src/internal/nitro/dev-client-address.ts index 1e79baa9b..0f3b5b21f 100644 --- a/packages/eve/src/internal/nitro/dev-client-address.ts +++ b/packages/eve/src/internal/nitro/dev-client-address.ts @@ -1,4 +1,5 @@ import { createHash, createHmac, timingSafeEqual } from "node:crypto"; +import type { IncomingHttpHeaders } from "node:http"; export const DEVELOPMENT_CLIENT_ADDRESS_HEADER = "x-eve-dev-client-address"; export const DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER = "x-eve-dev-client-address-signature"; @@ -11,12 +12,12 @@ export const DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER = "x-eve-dev-client-add * unsigned header could be forged by anything that discovers it. */ export function stampDevelopmentClientAddress( - headers: Headers, + headers: Headers | IncomingHttpHeaders, address: string | undefined, secret: string | undefined, ): void { - headers.delete(DEVELOPMENT_CLIENT_ADDRESS_HEADER); - headers.delete(DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER); + deleteHeader(headers, DEVELOPMENT_CLIENT_ADDRESS_HEADER); + deleteHeader(headers, DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER); if ( address === undefined || address.length === 0 || @@ -25,8 +26,12 @@ export function stampDevelopmentClientAddress( ) { return; } - headers.set(DEVELOPMENT_CLIENT_ADDRESS_HEADER, address); - headers.set(DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER, signClientAddress(address, secret)); + setHeader(headers, DEVELOPMENT_CLIENT_ADDRESS_HEADER, address); + setHeader( + headers, + DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER, + signClientAddress(address, secret), + ); } /** @@ -65,3 +70,19 @@ export function timingSafeEqualStrings(a: string, b: string): boolean { function signClientAddress(address: string, secret: string): string { return createHmac("sha256", secret).update(address).digest("base64url"); } + +function deleteHeader(headers: Headers | IncomingHttpHeaders, name: string): void { + if (headers instanceof Headers) { + headers.delete(name); + return; + } + delete headers[name]; +} + +function setHeader(headers: Headers | IncomingHttpHeaders, name: string, value: string): void { + if (headers instanceof Headers) { + headers.set(name, value); + return; + } + headers[name] = value; +} diff --git a/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.test.ts b/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.test.ts index 8eae554bd..4b4240bff 100644 --- a/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.test.ts +++ b/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.test.ts @@ -3,6 +3,7 @@ import { describe, expect, it, vi, beforeEach } from "vitest"; import type { CompileAgentResult } from "#compiler/compile-agent.js"; import { createDevelopmentAuthoredRebuildCoordinator, + DevelopmentWorkflowWorldChangeRequiresRestartError, PostCommitDevelopmentRebuildError, } from "#internal/nitro/host/dev-authored-rebuild-coordinator.js"; import { DrainedNitroDevServer } from "#internal/nitro/host/drained-nitro-dev-server.js"; @@ -59,10 +60,21 @@ vi.mock("#execution/sandbox/development-prewarm.js", () => ({ startDevelopmentSandboxPrewarmInBackground: vi.fn(), })); -function createHost(id: string, runtimeFingerprint: string): PreparedDevelopmentApplicationHost { +function createHost( + id: string, + runtimeFingerprint: string, + configuredWorld?: string, +): PreparedDevelopmentApplicationHost { return { appRoot: "/tmp/eve-test", compileResult: { + manifest: { + config: { + experimental: { + workflow: configuredWorld === undefined ? undefined : { world: configuredWorld }, + }, + }, + }, project: { agentRoot: "/tmp/eve-test/agent" }, } as CompileAgentResult, compiledArtifacts: { @@ -195,4 +207,35 @@ describe("transactional authored rebuild coordinator", () => { expect(mocks.environmentCommit).not.toHaveBeenCalled(); await devServer.close(); }); + + it.each([ + { initialWorld: undefined, nextWorld: "@example/custom-world" }, + { initialWorld: "@example/custom-world", nextWorld: "local" }, + ])( + "requires a restart when Workflow World ownership changes from $initialWorld to $nextWorld", + async ({ initialWorld, nextWorld }) => { + const devServer = new DrainedNitroDevServer({ error: () => undefined }, createRunner); + mocks.computeDevelopmentHostFingerprint.mockResolvedValueOnce("host-1"); + const coordinator = await createDevelopmentAuthoredRebuildCoordinator({ + devServer, + initialHost: createHost("initial", "run-1", initialWorld), + }); + const candidate = createHost("candidate", "run-2", nextWorld); + mocks.prepareDevelopmentApplicationHost.mockResolvedValueOnce(candidate); + + await expect(coordinator.rebuild({ changedPaths: [] })).rejects.toBeInstanceOf( + DevelopmentWorkflowWorldChangeRequiresRestartError, + ); + + expect(mocks.computeDevelopmentHostFingerprint).toHaveBeenCalledOnce(); + expect(mocks.createDevelopmentApplicationNitro).not.toHaveBeenCalled(); + expect(mocks.buildDevelopmentHostCandidate).not.toHaveBeenCalled(); + expect(mocks.removeDevelopmentHostWorkspace).toHaveBeenCalledWith(candidate.workspace); + expect(mocks.discardDevelopmentGeneration).toHaveBeenCalledWith(candidate.generation); + expect(mocks.environmentRollback).toHaveBeenCalledOnce(); + expect(mocks.environmentCommit).not.toHaveBeenCalled(); + + await devServer.close(); + }, + ); }); diff --git a/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.ts b/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.ts index 90ec2d955..bfdfcfb4f 100644 --- a/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.ts +++ b/packages/eve/src/internal/nitro/host/dev-authored-rebuild-coordinator.ts @@ -9,6 +9,7 @@ import { computeDevelopmentHostFingerprint } from "#internal/nitro/host/dev-host import { removeDevelopmentHostWorkspace } from "#internal/nitro/host/dev-host-workspace.js"; import { prepareDevelopmentApplicationHost } from "#internal/nitro/host/prepare-application-host.js"; import { DrainedNitroDevServer } from "#internal/nitro/host/drained-nitro-dev-server.js"; +import { usesParentDevelopmentWorkflowWorld } from "#internal/workflow/development-world-protocol.js"; import type { PreparedDevelopmentApplicationHost } from "#internal/nitro/host/types.js"; import { activateDevelopmentGeneration, @@ -33,6 +34,20 @@ export class PostCommitDevelopmentRebuildError extends Error { } } +/** + * Raised when an authored reload changes which process owns the Workflow + * World. That ownership is fixed when the development server starts, so the + * existing server must keep serving until the user restarts it. + */ +export class DevelopmentWorkflowWorldChangeRequiresRestartError extends Error { + constructor() { + super( + "Changing experimental.workflow.world between the parent-owned local World and a custom World requires restarting eve dev.", + ); + this.name = "DevelopmentWorkflowWorldChangeRequiresRestartError"; + } +} + export interface DevelopmentRebuildResult { readonly host: PreparedDevelopmentApplicationHost; readonly kind: DevelopmentRebuildKind; @@ -69,6 +84,7 @@ class TransactionalDevelopmentAuthoredRebuildCoordinator implements DevelopmentA #currentHostFingerprint: string; #currentRuntimeFingerprint: string; readonly #devServer: DrainedNitroDevServer; + readonly #usesParentWorkflowWorld: boolean; constructor(input: { readonly currentHostFingerprint: string; @@ -80,6 +96,9 @@ class TransactionalDevelopmentAuthoredRebuildCoordinator implements DevelopmentA this.#currentHostFingerprint = input.currentHostFingerprint; this.#currentRuntimeFingerprint = input.currentRuntimeFingerprint; this.#devServer = input.devServer; + this.#usesParentWorkflowWorld = usesParentDevelopmentWorkflowWorld( + input.initialHost.compileResult.manifest.config.experimental?.workflow?.world, + ); } async rebuild(input: { @@ -95,6 +114,13 @@ class TransactionalDevelopmentAuthoredRebuildCoordinator implements DevelopmentA try { nextHost = await prepareDevelopmentApplicationHost(previousHost.appRoot); + if ( + usesParentDevelopmentWorkflowWorld( + nextHost.compileResult.manifest.config.experimental?.workflow?.world, + ) !== this.#usesParentWorkflowWorld + ) { + throw new DevelopmentWorkflowWorldChangeRequiresRestartError(); + } const nextHostFingerprint = await computeDevelopmentHostFingerprint(nextHost); const nextRuntimeFingerprint = nextHost.generation.fingerprint; const hasStructuralChange = nextHostFingerprint !== this.#currentHostFingerprint; diff --git a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts index 5826e069f..85ba4b94f 100644 --- a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts +++ b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.scenario.test.ts @@ -173,6 +173,38 @@ describe("drained Nitro dev server", () => { await server.close(); }); + it("waits for an already-releasing retired worker before close resolves", async () => { + const { createRunner } = createRunnerFactory(async () => new Response("ok")); + const server = new DrainedNitroDevServer(LOGGER, createRunner); + let notifyDisposeStarted: (() => void) | undefined; + const disposeStarted = new Promise((resolve) => { + notifyDisposeStarted = resolve; + }); + let finishDispose: (() => void) | undefined; + const disposeFinished = new Promise((resolve) => { + finishDispose = resolve; + }); + const disposeFirst = vi.fn(async () => { + notifyDisposeStarted?.(); + await disposeFinished; + }); + + await server.replaceWorker(replacement("/tmp/first.mjs", disposeFirst)); + await server.replaceWorker(replacement("/tmp/second.mjs")); + await withinDeadline(disposeStarted, "Timed out waiting for retired host disposal to start."); + + const closing = server.close(); + const closeState = await Promise.race([ + closing.then(() => "closed" as const), + new Promise<"pending">((resolve) => setTimeout(() => resolve("pending"), 25)), + ]); + expect(closeState).toBe("pending"); + + finishDispose?.(); + await withinDeadline(closing, "Timed out waiting for retired host disposal during close."); + expect(disposeFirst).toHaveBeenCalledOnce(); + }); + it("restarts the worker with its own workspace when it exits unexpectedly", async () => { const { createRunner, runners } = createRunnerFactory( async (_request, runnerIndex) => new Response(`runner-${String(runnerIndex)}`), @@ -493,6 +525,57 @@ describe("drained Nitro dev server", () => { await server.close(); }); + it("stamps signed client-address metadata on WebSocket upgrades", async () => { + const SECRET = "scenario-websocket-transport-secret"; + const { createRunner, runners } = createRunnerFactory(async () => new Response("ok")); + const server = new DrainedNitroDevServer(LOGGER, createRunner); + server.setClientAddressSecret(SECRET); + const listener = await listen(server); + await server.replaceWorker(replacement("/tmp/first.mjs")); + let observedHeaders: Record | undefined; + (runners[0] as { upgrade: unknown }).upgrade = vi.fn( + async (input: Parameters[0]) => { + observedHeaders = input.node.req.headers; + input.node.socket.destroy(); + }, + ); + + const target = new URL(listener.url ?? ""); + await withinDeadline( + new Promise((resolve, reject) => { + const socket = connect({ host: target.hostname, port: Number(target.port) }, () => { + socket.write( + "GET / HTTP/1.1\r\n" + + "Host: localhost\r\n" + + "Connection: Upgrade\r\n" + + "Upgrade: websocket\r\n" + + `${DEVELOPMENT_CLIENT_ADDRESS_HEADER}: 203.0.113.7\r\n` + + `${DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER}: forged\r\n\r\n`, + ); + }); + socket.once("close", () => resolve()); + socket.once("error", reject); + }), + "Timed out waiting for the WebSocket upgrade to reach the worker.", + ); + + const address = observedHeaders?.[DEVELOPMENT_CLIENT_ADDRESS_HEADER]; + const signature = observedHeaders?.[DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER]; + expect(address).toBe("127.0.0.1"); + expect(signature).not.toBe("forged"); + expect( + readTrustedDevelopmentClientAddress( + new Headers({ + [DEVELOPMENT_CLIENT_ADDRESS_HEADER]: String(address), + [DEVELOPMENT_CLIENT_ADDRESS_SIGNATURE_HEADER]: String(signature), + }), + SECRET, + ), + ).toBe("127.0.0.1"); + + await server.close(); + }); + it("answers control requests without admitting them to the worker", async () => { const fetchMock = vi.fn(async () => new Response("worker")); const { createRunner } = createRunnerFactory(fetchMock); diff --git a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.ts b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.ts index 6d5f222b3..806d3479d 100644 --- a/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.ts +++ b/packages/eve/src/internal/nitro/host/drained-nitro-dev-server.ts @@ -36,7 +36,7 @@ interface RunnerSlot { disposed: boolean; readonly entry: string; quietListeners: Array<() => void>; - released: boolean; + releasePromise: Promise | undefined; readonly runner: DevelopmentRunner; readonly workerData: Readonly>; } @@ -116,7 +116,7 @@ export class DrainedNitroDevServer { disposed: false, entry: replacement.entry, quietListeners: [], - released: false, + releasePromise: undefined, runner: this.#createRunner({ entry: replacement.entry, name: `eve-dev-${String(this.#runnerCounter++)}`, @@ -271,11 +271,12 @@ export class DrainedNitroDevServer { await Promise.all(listenerClosePromises); } - async #releaseSlot(slot: RunnerSlot): Promise { - if (slot.released) { - return; - } - slot.released = true; + #releaseSlot(slot: RunnerSlot): Promise { + slot.releasePromise ??= this.#releaseSlotOnce(slot); + return slot.releasePromise; + } + + async #releaseSlotOnce(slot: RunnerSlot): Promise { await slot.runner.close().catch(() => undefined); await this.#disposeSlot(slot); } @@ -312,7 +313,7 @@ export class DrainedNitroDevServer { // replacement inherits its dispose and the crashed slot releases // nothing itself. slot.disposed = true; - slot.released = true; + slot.releasePromise = Promise.resolve(); await this.#replaceWorker({ dispose: slot.dispose, entry: slot.entry, @@ -330,8 +331,7 @@ export class DrainedNitroDevServer { #drainInBackground(slot: RunnerSlot): void { this.#draining.add(slot); const finishDrain = () => { - this.#draining.delete(slot); - void this.#releaseSlot(slot); + void this.#releaseSlot(slot).finally(() => this.#draining.delete(slot)); }; if (slot.activeExchanges === 0) { finishDrain(); @@ -412,6 +412,11 @@ export class DrainedNitroDevServer { async #handleUpgrade(request: IncomingMessage, socket: Socket, head: Buffer): Promise { let settle: (() => void) | undefined; try { + stampDevelopmentClientAddress( + request.headers, + request.socket.remoteAddress ?? undefined, + this.#clientAddressSecret, + ); await this.waitForActiveRunner(); const slot = this.#active; if (slot === undefined) { diff --git a/packages/eve/src/internal/workflow/development-world-client.ts b/packages/eve/src/internal/workflow/development-world-client.ts index c3a402f4e..a28efa39a 100644 --- a/packages/eve/src/internal/workflow/development-world-client.ts +++ b/packages/eve/src/internal/workflow/development-world-client.ts @@ -284,11 +284,22 @@ async function readGenerationRuntimeAppRoot( generationId, "generation.json", ); + let source: string; + try { + source = await readFile(metadataPath, "utf8"); + } catch (error) { + if (isFileNotFoundError(error)) { + throw new MissingDevelopmentGenerationError(generationId, error); + } + throw error; + } let metadata: unknown; try { - metadata = JSON.parse(await readFile(metadataPath, "utf8")); + metadata = JSON.parse(source); } catch (error) { - throw new MissingDevelopmentGenerationError(generationId, error); + throw new Error(`Development generation "${generationId}" has invalid metadata.`, { + cause: error, + }); } if (!isRecord(metadata) || typeof metadata.runtimeAppRoot !== "string") { throw new Error(`Development generation "${generationId}" has invalid metadata.`); @@ -296,6 +307,10 @@ async function readGenerationRuntimeAppRoot( return metadata.runtimeAppRoot; } +function isFileNotFoundError(error: unknown): boolean { + return error instanceof Error && "code" in error && error.code === "ENOENT"; +} + function isRecord(value: unknown): value is Record { return typeof value === "object" && value !== null && !Array.isArray(value); } diff --git a/packages/eve/src/internal/workflow/development-world-codec.test.ts b/packages/eve/src/internal/workflow/development-world-codec.test.ts index 89282cb85..fce958f67 100644 --- a/packages/eve/src/internal/workflow/development-world-codec.test.ts +++ b/packages/eve/src/internal/workflow/development-world-codec.test.ts @@ -22,6 +22,20 @@ describe("development Workflow World codec", () => { ).toEqual({ createdAt, input: new Uint8Array([1, 2, 3]), output: undefined }); }); + it("preserves ordinary objects that resemble codec markers", () => { + const value = { + bytes: { __eveDevelopmentWorldType: "uint8-array", data: "AQID" }, + date: { + __eveDevelopmentWorldType: "date", + data: "2026-07-13T12:00:00.000Z", + }, + legacyBytes: { __type: "Uint8Array", data: "AQID" }, + nested: [{ __eveDevelopmentWorldType: "undefined" }], + }; + + expect(decodeDevelopmentWorldValue(encodeDevelopmentWorldValue(value))).toEqual(value); + }); + it("preserves World error identity and retry metadata", () => { const source = Object.assign(new Error("try later"), { name: "WorkflowWorldError", diff --git a/packages/eve/src/internal/workflow/development-world-codec.ts b/packages/eve/src/internal/workflow/development-world-codec.ts index 73fff38a0..cc1704600 100644 --- a/packages/eve/src/internal/workflow/development-world-codec.ts +++ b/packages/eve/src/internal/workflow/development-world-codec.ts @@ -5,12 +5,13 @@ // across the parent/worker HTTP boundary. const VALUE_TYPE_KEY = "__eveDevelopmentWorldType"; const DATE_MARKER = "date"; +const OBJECT_MARKER = "object"; const UINT8_ARRAY_MARKER = "uint8-array"; const UNDEFINED_MARKER = "undefined"; interface EncodedValue { readonly [VALUE_TYPE_KEY]: string; - readonly data?: string; + readonly data?: unknown; } export interface SerializedDevelopmentWorldError { @@ -25,12 +26,12 @@ export function encodeDevelopmentWorldValue(value: unknown): string { } export function decodeDevelopmentWorldValue(source: string): unknown { - const decoded = decodeDevelopmentWorldJson(source) as { readonly value?: unknown }; - return decoded.value; + const decoded = JSON.parse(source) as unknown; + return isRecord(decoded) ? decodeTransportValue(decoded.value) : undefined; } export function decodeDevelopmentWorldJson(source: string): unknown { - return decodeValue(JSON.parse(source) as unknown); + return decodeWorkflowJsonValue(JSON.parse(source) as unknown); } export function serializeDevelopmentWorldError(error: unknown): SerializedDevelopmentWorldError { @@ -81,14 +82,19 @@ function encodeValue(value: unknown): unknown { return value.map((item) => encodeValue(item)); } if (isRecord(value)) { - return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, encodeValue(item)])); + return { + [VALUE_TYPE_KEY]: OBJECT_MARKER, + data: Object.fromEntries( + Object.entries(value).map(([key, item]) => [key, encodeValue(item)]), + ), + } satisfies EncodedValue; } return value; } -function decodeValue(value: unknown): unknown { +function decodeTransportValue(value: unknown): unknown { if (Array.isArray(value)) { - return value.map((item) => decodeValue(item)); + return value.map((item) => decodeTransportValue(item)); } if (!isRecord(value)) { return value; @@ -99,13 +105,34 @@ function decodeValue(value: unknown): unknown { if (value[VALUE_TYPE_KEY] === DATE_MARKER && typeof value.data === "string") { return new Date(value.data); } + if (value[VALUE_TYPE_KEY] === UINT8_ARRAY_MARKER && typeof value.data === "string") { + return Uint8Array.from(Buffer.from(value.data, "base64")); + } + if (value[VALUE_TYPE_KEY] === OBJECT_MARKER && isRecord(value.data)) { + return Object.fromEntries( + Object.entries(value.data).map(([key, item]) => [key, decodeTransportValue(item)]), + ); + } + throw new Error("Development Workflow World transport contained an invalid encoded value."); +} + +function decodeWorkflowJsonValue(value: unknown): unknown { + if (Array.isArray(value)) { + return value.map((item) => decodeWorkflowJsonValue(item)); + } + if (!isRecord(value)) { + return value; + } if ( - (value[VALUE_TYPE_KEY] === UINT8_ARRAY_MARKER || value.__type === "Uint8Array") && + Object.keys(value).length === 2 && + value.__type === "Uint8Array" && typeof value.data === "string" ) { return Uint8Array.from(Buffer.from(value.data, "base64")); } - return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, decodeValue(item)])); + return Object.fromEntries( + Object.entries(value).map(([key, item]) => [key, decodeWorkflowJsonValue(item)]), + ); } function isRecord(value: unknown): value is Record { diff --git a/packages/eve/src/internal/workflow/development-world-server.integration.test.ts b/packages/eve/src/internal/workflow/development-world-server.integration.test.ts index 9468bfea6..ef03abe56 100644 --- a/packages/eve/src/internal/workflow/development-world-server.integration.test.ts +++ b/packages/eve/src/internal/workflow/development-world-server.integration.test.ts @@ -210,6 +210,62 @@ describe("parent development Workflow World", () => { } }); + it("retries a delivery when generation metadata is temporarily unreadable", async () => { + const appRoot = await createScratchDirectory("eve-parent-workflow-retry-delivery-"); + await seedGeneration(appRoot, "generation-a"); + const world = createWorld({ activeGenerationId: () => "generation-a", appRoot }); + connectWorkerToWorld(world, appRoot); + + try { + await world.start(); + const created = await callWorld(world, "events.create", [ + null, + { + eventData: { + deploymentId: "generation-a", + executionContext: {}, + input: new Uint8Array(), + workflowName: turnWorkflowReference.workflowId, + }, + eventType: "run_created", + specVersion: 5, + }, + ]); + const runId = readCreatedRunId(created); + const metadataPath = join( + appRoot, + ".eve", + "dev-runtime", + "snapshots", + "generation-a", + "generation.json", + ); + await rm(metadataPath); + await mkdir(metadataPath); + + const handled = vi.fn(async () => undefined); + const handler = createDevelopmentWorkflowWorld().createQueueHandler(QUEUE_PREFIX, handled); + const createDelivery = () => + new Request("http://localhost/.well-known/workflow/v1/flow", { + body: JSON.stringify({ runId }), + headers: deliveryHeaders({}), + method: "POST", + }); + + const failed = await handler(createDelivery()); + expect(failed.status).toBe(500); + expect(handled).not.toHaveBeenCalled(); + + await rm(metadataPath, { recursive: true }); + await seedGeneration(appRoot, "generation-a"); + const retried = await handler(createDelivery()); + expect(retried.status).toBe(200); + expect(handled).toHaveBeenCalledOnce(); + } finally { + await world.close(); + } + }); + it("rejects untrusted World requests on the call and stream routes", async () => { const appRoot = await createScratchDirectory("eve-parent-workflow-untrusted-"); const world = createWorld({ activeGenerationId: () => "generation-a", appRoot }); diff --git a/packages/eve/test/scenarios/dev-server-harness.ts b/packages/eve/test/scenarios/dev-server-harness.ts index c61055c7f..66f1b6ae3 100644 --- a/packages/eve/test/scenarios/dev-server-harness.ts +++ b/packages/eve/test/scenarios/dev-server-harness.ts @@ -51,11 +51,17 @@ export async function startEveDev( stderr += chunk; }); - const url = await waitForServerUrl({ - child, - getOutput: () => ({ stderr, stdout }), - }); - await waitForPath(join(appRoot, ".eve", "dev-server-state.v1.json")); + let url: string; + try { + url = await waitForServerUrl({ + child, + getOutput: () => ({ stderr, stdout }), + }); + await waitForPath(join(appRoot, ".eve", "dev-server-state.v1.json")); + } catch (error) { + await stopEveDevChild(child); + throw error; + } return { async crash() { @@ -73,27 +79,40 @@ export async function startEveDev( stderr: () => stderr, stdout: () => stdout, async stop() { - if (child.exitCode !== null || child.signalCode !== null) { - return; - } - - await new Promise((resolve) => { - const timeout = setTimeout(() => { - child.kill("SIGKILL"); - resolve(); - }, 10_000); - - child.once("exit", () => { - clearTimeout(timeout); - resolve(); - }); - child.kill("SIGTERM"); - }); + await stopEveDevChild(child); }, url, }; } +async function stopEveDevChild( + child: ChildProcessByStdio, +): Promise { + if (child.exitCode !== null || child.signalCode !== null) { + return; + } + + await new Promise((resolve) => { + let settled = false; + const forceTimeout = setTimeout(() => child.kill("SIGKILL"), 10_000); + const finalTimeout = setTimeout(settle, 15_000); + + function settle() { + if (settled) { + return; + } + settled = true; + clearTimeout(forceTimeout); + clearTimeout(finalTimeout); + child.off("exit", settle); + resolve(); + } + + child.once("exit", settle); + child.kill("SIGTERM"); + }); +} + /** * Spawns `eve dev` and resolves with the process exit for flows that expect * startup to fail (e.g. unsupported runtimes).