diff --git a/.changeset/pin-durable-session-turns.md b/.changeset/pin-durable-session-turns.md new file mode 100644 index 0000000000..cad733d0f6 --- /dev/null +++ b/.changeset/pin-durable-session-turns.md @@ -0,0 +1,5 @@ +--- +"eve": patch +--- + +Keep turns for parked sessions on the deployment that started their durable driver, preventing newer deployments from writing incompatible session state. diff --git a/packages/eve/src/execution/dispatch-turn-step.ts b/packages/eve/src/execution/dispatch-turn-step.ts index 1cd87aeed7..39a8011a52 100644 --- a/packages/eve/src/execution/dispatch-turn-step.ts +++ b/packages/eve/src/execution/dispatch-turn-step.ts @@ -3,7 +3,8 @@ import { type TurnWorkflowDispatchInput, } from "#execution/durable-session-migrations/turn-workflow.js"; import { buildTurnAttributes, readRootSessionId } from "#execution/eve-workflow-attributes.js"; -import { startWorkflowPreferLatest, turnWorkflowReference } from "#execution/workflow-runtime.js"; +import { turnWorkflowReference } from "#execution/workflow-runtime.js"; +import { start } from "#internal/workflow/runtime.js"; import { normalizeEveAttributes } from "#runtime/attributes/normalize.js"; /** Starts a per-turn child workflow for the current driver session. */ @@ -12,20 +13,16 @@ export async function dispatchTurnStep( ): Promise<{ readonly runId: string }> { "use step"; - const run = await startWorkflowPreferLatest( - turnWorkflowReference, - [createTurnWorkflowInput(input)], - { - allowReservedAttributes: true, - attributes: normalizeEveAttributes( - buildTurnAttributes({ - parentSessionId: input.sessionState.sessionId, - requestId: input.delivery.kind === "deliver" ? input.delivery.requestId : undefined, - rootSessionId: readRootSessionId(input.serializedContext) ?? input.sessionState.sessionId, - }), - ), - }, - ); + const run = await start(turnWorkflowReference, [createTurnWorkflowInput(input)], { + allowReservedAttributes: true, + attributes: normalizeEveAttributes( + buildTurnAttributes({ + parentSessionId: input.sessionState.sessionId, + requestId: input.delivery.kind === "deliver" ? input.delivery.requestId : undefined, + rootSessionId: readRootSessionId(input.serializedContext) ?? input.sessionState.sessionId, + }), + ), + }); return { runId: run.runId }; } diff --git a/packages/eve/src/execution/durable-session-store.ts b/packages/eve/src/execution/durable-session-store.ts index 08ca4345ca..d4409c3749 100644 --- a/packages/eve/src/execution/durable-session-store.ts +++ b/packages/eve/src/execution/durable-session-store.ts @@ -7,13 +7,12 @@ * `"eve.session"` stream remains as a fallback for old in-flight * sessions that only carry a small state handle. * - * The driver workflow run is pinned to the deployment that called - * `start()`; child turn workflows run on latest. Both - * {@link DurableSessionState} and {@link DurableSessionSnapshot} carry - * a `version` so a pinned driver can ferry shapes written by newer - * steps. Adding optional fields is forward-compatible (devalue - * preserves unknown POJO fields); shape-breaking changes bump - * `version` and add a migrator. + * The driver workflow run and its child turn workflows stay pinned to + * the same deployment. Both {@link DurableSessionState} and + * {@link DurableSessionSnapshot} carry a `version` for schema + * evolution within that deployment. Adding optional fields is + * forward-compatible (devalue preserves unknown POJO fields); + * shape-breaking changes bump `version` and add a migrator. */ import type { ModelMessage } from "ai"; diff --git a/packages/eve/src/execution/next-driver-action.ts b/packages/eve/src/execution/next-driver-action.ts index 5671ceca31..b235ffd2a1 100644 --- a/packages/eve/src/execution/next-driver-action.ts +++ b/packages/eve/src/execution/next-driver-action.ts @@ -1,7 +1,7 @@ /** * Closed-contract dispatch surface between session-mutating step - * bodies (latest deployment) and the durable driver workflow (pinned - * to whichever deployment called `start()`). + * bodies and the durable driver workflow. Both run on the deployment + * that started the session. * * The driver matches on `kind` and follows a fixed playbook per arm. * Adding a new arm is breaking (pinned drivers can't dispatch an diff --git a/packages/eve/src/execution/workflow-runtime.ts b/packages/eve/src/execution/workflow-runtime.ts index adb74ef2e2..a4edb89226 100644 --- a/packages/eve/src/execution/workflow-runtime.ts +++ b/packages/eve/src/execution/workflow-runtime.ts @@ -101,9 +101,8 @@ export const workflowEntryReference = { /** * Stable workflow reference used by the driver to dispatch per-turn * child workflow runs. The id omits the package version stamp so - * `start(turnWorkflowReference, args, { deploymentId: "latest" })` - * routes to the latest deployment's turn workflow even when the eve - * version differs from the caller's deployment. + * explicitly version-routed callers can find the same workflow on a + * newer deployment even when the eve version differs. */ export const turnWorkflowReference = { workflowId: `workflow//${STABLE_ID_BASE}//${TURN_WORKFLOW_NAME}`, diff --git a/packages/eve/src/execution/workflow-steps.test.ts b/packages/eve/src/execution/workflow-steps.test.ts index f41de0ed6c..767620a30f 100644 --- a/packages/eve/src/execution/workflow-steps.test.ts +++ b/packages/eve/src/execution/workflow-steps.test.ts @@ -51,11 +51,7 @@ import { emitTerminalSessionFailureStep } from "#execution/terminal-session-fail import { resolveEffectiveOutputSchema } from "#execution/effective-output-schema.js"; import { turnStep } from "#execution/workflow-steps.js"; import { routeProxiedDeliverStep } from "#execution/proxied-deliver-step.js"; -import { - LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE, - turnWorkflowReference, - workflowEntryReference, -} from "#execution/workflow-runtime.js"; +import { turnWorkflowReference, workflowEntryReference } from "#execution/workflow-runtime.js"; vi.mock("./durable-session-store.js", async (importOriginal) => { const actual = await importOriginal(); @@ -630,7 +626,7 @@ describe("dispatchTurnStep", () => { }; } - it("starts turn workflows on the latest deployment in Vercel production", async () => { + it("keeps a parked session's turn workflow on its driver deployment in Vercel production", async () => { vi.stubEnv("VERCEL_ENV", "production"); const input = createTurnInput(); startMock.mockResolvedValue({ runId: "turn-run" }); @@ -648,12 +644,11 @@ describe("dispatchTurnStep", () => { "$eve.root": "sess-test", "$eve.type": "turn", }, - deploymentId: "latest", }, ); }); - it("starts turn workflows on the latest promoted generation in local development", async () => { + it("keeps a parked session's turn workflow on its driver generation in local development", async () => { vi.stubEnv("EVE_DEV", "1"); const input = createTurnInput(); startMock.mockResolvedValue({ runId: "turn-run" }); @@ -663,7 +658,7 @@ describe("dispatchTurnStep", () => { expect(startMock).toHaveBeenCalledWith( turnWorkflowReference, [createTurnWorkflowInput(input)], - expect.objectContaining({ deploymentId: "latest" }), + expect.not.objectContaining({ deploymentId: "latest" }), ); }); @@ -689,37 +684,6 @@ describe("dispatchTurnStep", () => { }, ); }); - - it("falls back to the current deployment when latest is unsupported", async () => { - vi.stubEnv("VERCEL_ENV", "production"); - const input = createTurnInput(); - startMock - .mockRejectedValueOnce(new Error(LATEST_DEPLOYMENT_UNSUPPORTED_MESSAGE)) - .mockResolvedValueOnce({ runId: "turn-run" }); - - await expect(dispatchTurnStep(input)).resolves.toEqual({ runId: "turn-run" }); - - const wireInput = createTurnWorkflowInput(input); - expect(startMock).toHaveBeenNthCalledWith(1, turnWorkflowReference, [wireInput], { - allowReservedAttributes: true, - attributes: { - "$eve.channel_request_id": "req_turn", - "$eve.parent": "sess-test", - "$eve.root": "sess-test", - "$eve.type": "turn", - }, - deploymentId: "latest", - }); - expect(startMock).toHaveBeenNthCalledWith(2, turnWorkflowReference, [wireInput], { - allowReservedAttributes: true, - attributes: { - "$eve.channel_request_id": "req_turn", - "$eve.parent": "sess-test", - "$eve.root": "sess-test", - "$eve.type": "turn", - }, - }); - }); }); describe("dispatchRuntimeActionsStep", () => {