diff --git a/.changeset/calm-local-turns.md b/.changeset/calm-local-turns.md new file mode 100644 index 0000000000..3d15afbf6d --- /dev/null +++ b/.changeset/calm-local-turns.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Prevent long local Workflow deliveries from timing out and replaying an in-flight turn. Explicit local delivery timeout overrides continue to take precedence. diff --git a/packages/eve/src/internal/application/compiled-artifacts.test.ts b/packages/eve/src/internal/application/compiled-artifacts.test.ts index 3dff1e8d6c..16693195e4 100644 --- a/packages/eve/src/internal/application/compiled-artifacts.test.ts +++ b/packages/eve/src/internal/application/compiled-artifacts.test.ts @@ -37,6 +37,10 @@ describe("createWorkflowWorldPluginSource", () => { expect(source).toContain("/compiled/@workflow/world-local/index.js"); expect(source).toContain("resolveLocalWorkflowWorldDataDirectory(process.cwd())"); + expect(source).toContain("applyLocalWorkflowWorldDeliveryTimeoutDefaults"); + expect(source.indexOf("applyLocalWorkflowWorldDeliveryTimeoutDefaults();")).toBeLessThan( + source.indexOf("workflowWorldModule.createWorld"), + ); expect(source).not.toContain("createWorldFromModule(workflowWorldModule)"); }); diff --git a/packages/eve/src/internal/application/compiled-artifacts.ts b/packages/eve/src/internal/application/compiled-artifacts.ts index fe60097b7b..bf52cb37ba 100644 --- a/packages/eve/src/internal/application/compiled-artifacts.ts +++ b/packages/eve/src/internal/application/compiled-artifacts.ts @@ -304,12 +304,17 @@ function resolveWorkflowWorldWiring(packageName: string): WorkflowWorldWiring { const dataDirectoryImportSpecifier = stringifyEsmImportSpecifier( resolvePackageSourceFilePath("src/internal/workflow/local-world-data-directory.ts"), ); + const deliveryTimeoutsImportSpecifier = stringifyEsmImportSpecifier( + resolvePackageSourceFilePath("src/internal/workflow/local-world-delivery-timeouts.ts"), + ); const moduleImportSpecifier = resolvePackageCompiledFilePath( `src/compiled/${packageName}/index.js`, ); const importLines = ` +import { applyLocalWorkflowWorldDeliveryTimeoutDefaults } from ${deliveryTimeoutsImportSpecifier}; import { resolveLocalWorkflowWorldDataDirectory } from ${dataDirectoryImportSpecifier};`.trimStart(); const createWorldSource = ` +applyLocalWorkflowWorldDeliveryTimeoutDefaults(); const workflowWorld = await workflowWorldModule.createWorld({ dataDir: resolveLocalWorkflowWorldDataDirectory(process.cwd()), });`.trimStart(); 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 05620eb7e4..b426a71de4 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 @@ -34,6 +34,13 @@ const SECRET = "workflow-transport-secret"; const RUN_ID = "wrun_01J00000000000000000000000"; const AGENT_NAME = "workflow-world-test"; const QUEUE_PREFIX = deriveEveWorkflowQueuePrefix(AGENT_NAME); +const LOCAL_DELIVERY_TIMEOUT_ENV_NAMES = [ + "WORKFLOW_LOCAL_BODY_TIMEOUT_MS", + "WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS", +] as const; +const originalLocalDeliveryTimeoutEnv = new Map( + LOCAL_DELIVERY_TIMEOUT_ENV_NAMES.map((name) => [name, process.env[name]]), +); const originalFetch = globalThis.fetch; @@ -42,9 +49,46 @@ afterEach(() => { delete process.env[DEVELOPMENT_WORKFLOW_SECRET_ENV]; delete process.env[DEVELOPMENT_WORKER_APP_ROOT_ENV]; delete process.env.WORKFLOW_LOCAL_BASE_URL; + for (const name of LOCAL_DELIVERY_TIMEOUT_ENV_NAMES) { + const originalValue = originalLocalDeliveryTimeoutEnv.get(name); + if (originalValue === undefined) { + delete process.env[name]; + } else { + process.env[name] = originalValue; + } + } }); describe("parent development Workflow World", () => { + it("defaults local delivery timeouts to unbounded", async () => { + for (const name of LOCAL_DELIVERY_TIMEOUT_ENV_NAMES) { + delete process.env[name]; + } + const appRoot = await createScratchDirectory("eve-parent-workflow-timeouts-"); + const world = createWorld({ activeGenerationId: () => "generation-a", appRoot }); + + try { + expect(process.env.WORKFLOW_LOCAL_BODY_TIMEOUT_MS).toBe("0"); + expect(process.env.WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS).toBe("0"); + } finally { + await world.close(); + } + }); + + it("preserves explicit local delivery timeouts", async () => { + process.env.WORKFLOW_LOCAL_BODY_TIMEOUT_MS = "123"; + process.env.WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS = "456"; + const appRoot = await createScratchDirectory("eve-parent-workflow-explicit-timeouts-"); + const world = createWorld({ activeGenerationId: () => "generation-a", appRoot }); + + try { + expect(process.env.WORKFLOW_LOCAL_BODY_TIMEOUT_MS).toBe("123"); + expect(process.env.WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS).toBe("456"); + } finally { + await world.close(); + } + }); + it("stores local Workflow state under .eve/.workflow-data", async () => { const appRoot = await createScratchDirectory("eve-parent-workflow-data-dir-"); const world = createWorld({ activeGenerationId: () => "generation-a", appRoot }); diff --git a/packages/eve/src/internal/workflow/development-world-server.ts b/packages/eve/src/internal/workflow/development-world-server.ts index 0ce9fc980a..9144cbd979 100644 --- a/packages/eve/src/internal/workflow/development-world-server.ts +++ b/packages/eve/src/internal/workflow/development-world-server.ts @@ -9,6 +9,7 @@ import { LOCAL_WORKFLOW_WORLD_DATA_DIRECTORY_RELATIVE_PATH, resolveLocalWorkflowWorldDataDirectory, } from "#internal/workflow/local-world-data-directory.js"; +import { applyLocalWorkflowWorldDeliveryTimeoutDefaults } from "#internal/workflow/local-world-delivery-timeouts.js"; import { decodeDevelopmentWorldValue, encodeDevelopmentWorldValue, @@ -77,6 +78,7 @@ class LocalParentDevelopmentWorkflowWorld implements ParentDevelopmentWorkflowWo this.#appRoot = input.appRoot; this.#resolveActiveGenerationId = input.resolveActiveGenerationId; this.#transportSecret = input.transportSecret; + applyLocalWorkflowWorldDeliveryTimeoutDefaults(); this.#world = createWorld({ dataDir: resolveLocalWorkflowWorldDataDirectory(input.appRoot), recoverActiveRuns: false, diff --git a/packages/eve/src/internal/workflow/local-world-delivery-timeouts.test.ts b/packages/eve/src/internal/workflow/local-world-delivery-timeouts.test.ts new file mode 100644 index 0000000000..1f6ee1a18e --- /dev/null +++ b/packages/eve/src/internal/workflow/local-world-delivery-timeouts.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from "vitest"; + +import { applyLocalWorkflowWorldDeliveryTimeoutDefaults } from "#internal/workflow/local-world-delivery-timeouts.js"; + +describe("applyLocalWorkflowWorldDeliveryTimeoutDefaults", () => { + it("defaults unset and empty local delivery timeouts to unbounded", () => { + const env: Record = { + WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS: "", + }; + + applyLocalWorkflowWorldDeliveryTimeoutDefaults(env); + + expect(env.WORKFLOW_LOCAL_BODY_TIMEOUT_MS).toBe("0"); + expect(env.WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS).toBe("0"); + }); + + it("preserves explicit local delivery timeouts", () => { + const env: Record = { + WORKFLOW_LOCAL_BODY_TIMEOUT_MS: "123", + WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS: "456", + }; + + applyLocalWorkflowWorldDeliveryTimeoutDefaults(env); + + expect(env.WORKFLOW_LOCAL_BODY_TIMEOUT_MS).toBe("123"); + expect(env.WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS).toBe("456"); + }); +}); diff --git a/packages/eve/src/internal/workflow/local-world-delivery-timeouts.ts b/packages/eve/src/internal/workflow/local-world-delivery-timeouts.ts new file mode 100644 index 0000000000..22e001d7fc --- /dev/null +++ b/packages/eve/src/internal/workflow/local-world-delivery-timeouts.ts @@ -0,0 +1,20 @@ +const LOCAL_WORKFLOW_DELIVERY_TIMEOUT_ENV_NAMES = [ + "WORKFLOW_LOCAL_BODY_TIMEOUT_MS", + "WORKFLOW_LOCAL_HEADERS_TIMEOUT_MS", +] as const; + +/** + * Disables world-local's delivery deadlines unless the operator supplied an + * override. world-local snapshots these values when `createWorld()` constructs + * its queue and supports `0` as an unbounded timeout. Without this default, a + * long inline turn can be retried while its original provider call is active. + */ +export function applyLocalWorkflowWorldDeliveryTimeoutDefaults( + env: Record = process.env as Record, +): void { + for (const name of LOCAL_WORKFLOW_DELIVERY_TIMEOUT_ENV_NAMES) { + if (env[name] === undefined || env[name] === "") { + env[name] = "0"; + } + } +}