fix(core): fail runs delivered to the wrong deployment

.changeset/fail-cross-deployment-runs.md

@@ -0,0 +1,6 @@
+---
+"@workflow/core": patch
+"@workflow/errors": patch
+---
+
+Fail a run with the new `DEPLOYMENT_MISMATCH` error code when queue delivery reaches a deployment other than the one that created it, instead of exhausting retries on an undecryptable step. The failure is recorded without the origin deployment's key so it works even after that deployment is gone.

packages/core/src/classify-error.test.ts

@@ -6,6 +6,7 @@ import {
   RuntimeDecryptionError,
   ThrottleError,
   TooEarlyError,
+  WorkflowDeploymentMismatchError,
   WorkflowNotRegisteredError,
   WorkflowRuntimeError,
   WorkflowWorldError,
@@ -42,6 +43,18 @@ describe('classifyRunError', () => {
     );
   });
 
+  it('classifies WorkflowDeploymentMismatchError as DEPLOYMENT_MISMATCH', () => {
+    expect(
+      classifyRunError(
+        new WorkflowDeploymentMismatchError(
+          'wrun_test',
+          'dpl_expected',
+          'dpl_actual'
+        )
+      )
+    ).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH);
+  });
+
   it('classifies plain Error as USER_ERROR', () => {
     expect(classifyRunError(new Error('user code broke'))).toBe(
       RUN_ERROR_CODES.USER_ERROR

packages/core/src/classify-error.ts

@@ -6,6 +6,7 @@ import {
   RuntimeDecryptionError,
   StepNotRegisteredError,
   ThrottleError,
+  WorkflowDeploymentMismatchError,
   WorkflowNotRegisteredError,
   WorkflowRuntimeError,
   WorkflowWorldError,
@@ -116,6 +117,12 @@ export function classifyRunError(err: unknown): RunErrorCode {
     return RUN_ERROR_CODES.CORRUPTED_EVENT_LOG;
   }
 
+  // A run delivered to the wrong deployment gets its own code so dashboards
+  // and the UI can distinguish code-skew misrouting from generic runtime bugs.
+  if (WorkflowDeploymentMismatchError.is(err)) {
+    return RUN_ERROR_CODES.DEPLOYMENT_MISMATCH;
+  }
+
   // World-layer faults — both a malformed response (contract violation) and a
   // transient infrastructure failure (throttle / 5xx / transport / timeout,
   // e.g. a firewall challenge) — are the backend's fault, not the user's.

packages/core/src/describe-error.test.ts

@@ -3,6 +3,7 @@ import {
   RUN_ERROR_CODES,
   SerializationError,
   StepNotRegisteredError,
+  WorkflowDeploymentMismatchError,
   WorkflowRuntimeError,
 } from '@workflow/errors';
 import { describe, expect, test } from 'vitest';
@@ -108,6 +109,17 @@ describe('describeError', () => {
     expect(result.hint).toContain('SDK contract');
   });
 
+  test('WorkflowDeploymentMismatchError is attributed to the SDK with the deployment hint', () => {
+    const result = describeError(
+      new WorkflowDeploymentMismatchError('wrun', 'dpl_a', 'dpl_b')
+    );
+    expect(result.attribution).toBe('sdk');
+    expect(result.errorCode).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH);
+    expect(result.hint).toContain(
+      'deployment other than the one that created it'
+    );
+  });
+
   test('precomputed errorCode wins over classifyRunError when both are provided', () => {
     // A plain Error would classify as USER_ERROR, but passing REPLAY_TIMEOUT
     // explicitly overrides that — useful for callers that know the failure
@@ -201,6 +213,18 @@ describe('describeRunError', () => {
     expect(result.hint).toContain('SDK contract');
   });
 
+  test('DEPLOYMENT_MISMATCH errorCode is attributed to the SDK', () => {
+    const result = describeRunError({
+      errorCode: RUN_ERROR_CODES.DEPLOYMENT_MISMATCH,
+      errorName: 'WorkflowDeploymentMismatchError',
+    });
+    expect(result.attribution).toBe('sdk');
+    expect(result.errorCode).toBe(RUN_ERROR_CODES.DEPLOYMENT_MISMATCH);
+    expect(result.hint).toContain(
+      'deployment other than the one that created it'
+    );
+  });
+
   test('RUNTIME_ERROR code without errorName still lands as SDK', () => {
     const result = describeRunError({
       errorCode: RUN_ERROR_CODES.RUNTIME_ERROR,

packages/core/src/describe-error.ts

@@ -85,6 +85,8 @@ const MAX_DELIVERIES_HINT =
   'The workflow queue exceeded its max-delivery budget. This usually indicates a persistent runtime failure — check the most recent stack traces for the underlying cause.';
 const WORLD_CONTRACT_HINT =
   'The workflow backend returned data that violated the SDK contract. This is not retryable; please report it with the stack trace and runId.';
+const DEPLOYMENT_MISMATCH_HINT =
+  'The run was delivered to a deployment other than the one that created it, and was stopped to protect against code-skew errors. This usually happens on preview branches after a new deployment supersedes the one a run started on. Verify that the original deployment is still available and that queue callbacks route to it.';
 
 function normalizeErrorCode(code: string | undefined): RunErrorCode {
   // Values read back from persisted events are `string | undefined` — we
@@ -144,6 +146,9 @@ export function describeRunError(
       hint: WORLD_CONTRACT_HINT,
     };
   }
+  if (errorCode === RUN_ERROR_CODES.DEPLOYMENT_MISMATCH) {
+    return { attribution: 'sdk', errorCode, hint: DEPLOYMENT_MISMATCH_HINT };
+  }
   if (name === 'WorkflowRuntimeError' || name === 'StepNotRegisteredError') {
     return { attribution: 'sdk', errorCode, hint: RUNTIME_ERROR_HINT };
   }
@@ -204,6 +209,16 @@ export function describeError(
     };
   }
 
+  // Check DEPLOYMENT_MISMATCH before the generic WorkflowRuntimeError branch —
+  // WorkflowDeploymentMismatchError subclasses it, but has its own code + hint.
+  if (effectiveCode === RUN_ERROR_CODES.DEPLOYMENT_MISMATCH) {
+    return {
+      attribution: 'sdk',
+      errorCode: effectiveCode,
+      hint: DEPLOYMENT_MISMATCH_HINT,
+    };
+  }
+
   if (err instanceof WorkflowRuntimeError) {
     return {
       attribution: 'sdk',

packages/core/src/runtime-trace-mode.test.ts

@@ -125,6 +125,7 @@ async function driveHandler(opts: {
 
   setWorld({
     specVersion: SPEC_VERSION_CURRENT,
+    getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
     createQueueHandler: vi.fn(
       (
         _prefix: string,
@@ -352,6 +353,7 @@ describe('workflowEntrypoint trace modes', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,

packages/core/src/runtime.test.ts

@@ -17,6 +17,7 @@ import { workflowEntrypoint } from './runtime.js';
 import {
   dehydrateStepReturnValue,
   dehydrateWorkflowArguments,
+  hydrateRunError,
 } from './serialization.js';
 
 // Capture every promise handed to `waitUntil` so tests can assert that
@@ -62,6 +63,7 @@ async function runWorkflowHandlerWithEvents(
      * pin the log's contents keep full control.
      */
     dynamicEventLog?: boolean;
+    currentDeploymentId?: string;
   } = {}
 ) {
   const createdEvents = options.createdEvents ?? [];
@@ -89,6 +91,9 @@ async function runWorkflowHandlerWithEvents(
 
   setWorld({
     specVersion: SPEC_VERSION_CURRENT,
+    getDeploymentId: vi.fn(
+      async () => options.currentDeploymentId ?? workflowRun.deploymentId
+    ),
     createQueueHandler: vi.fn(
       (
         _prefix: string,
@@ -146,6 +151,57 @@ describe('workflowEntrypoint replay guards', () => {
     `;globalThis.__private_workflows = new Map();
     globalThis.__private_workflows.set(${JSON.stringify(workflowName)}, ${workflowName});`;
 
+  it('fails a run delivered to a different deployment before executing it', async () => {
+    // A run may only execute on the deployment that created it. On delivery
+    // elsewhere the run is failed before any workflow code or step runs, and
+    // the failure is recorded with the DEPLOYMENT_MISMATCH code without
+    // resolving the (possibly-gone) origin deployment's encryption key.
+    const workflowRun: WorkflowRun = {
+      runId: 'wrun_wrong_deployment',
+      workflowName: 'workflow',
+      status: 'running',
+      input: await dehydrateWorkflowArguments(
+        [],
+        'wrun_wrong_deployment',
+        undefined,
+        []
+      ),
+      createdAt: new Date('2024-01-01T00:00:00.000Z'),
+      updatedAt: new Date('2024-01-01T00:00:00.000Z'),
+      startedAt: new Date('2024-01-01T00:00:00.000Z'),
+      deploymentId: 'dpl_origin',
+    };
+
+    const createdEvents = await runWorkflowHandlerWithEvents(
+      `async function workflow() {
+        throw new Error('workflow code must not execute');
+      }${getWorkflowTransformCode('workflow')}`,
+      workflowRun,
+      [],
+      { currentDeploymentId: 'dpl_current' }
+    );
+
+    const failedEvent = createdEvents.find(
+      (event: any) => event.eventType === 'run_failed'
+    ) as any;
+    expect(failedEvent).toBeDefined();
+    expect(failedEvent.eventData.errorCode).toBe(
+      RUN_ERROR_CODES.DEPLOYMENT_MISMATCH
+    );
+    const error = await hydrateRunError(
+      failedEvent.eventData.error,
+      workflowRun.runId,
+      undefined
+    );
+    expect(error).toMatchObject({ name: 'WorkflowDeploymentMismatchError' });
+    expect((error as Error).message).toContain(
+      'belongs to deployment "dpl_origin", but was received by deployment "dpl_current"'
+    );
+    expect(createdEvents).not.toContainEqual(
+      expect.objectContaining({ eventType: 'run_completed' })
+    );
+  });
+
   it('records run_failed when run_started response schema validation fails', async () => {
     const createdEvents: unknown[] = [];
     const schemaError = new WorkflowWorldError(
@@ -173,6 +229,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => 'test-deployment'),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -271,6 +328,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => 'test-deployment'),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -368,6 +426,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -447,6 +506,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -808,6 +868,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -917,6 +978,7 @@ describe('workflowEntrypoint replay guards', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -1138,6 +1200,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (
           _prefix: string,
@@ -1383,6 +1446,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => {
     });
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => workflowRun.deploymentId),
       createQueueHandler: vi.fn(
         (_p: string, handler: (m: unknown, md: unknown) => Promise<unknown>) =>
           async () => {
@@ -1558,6 +1622,7 @@ describe('workflowEntrypoint turbo mode', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => 'test-deployment'),
       createQueueHandler: vi.fn(
         (_p: string, handler: (m: unknown, md: unknown) => Promise<unknown>) =>
           async () => {
@@ -1826,6 +1891,7 @@ describe('workflowEntrypoint latency telemetry (ttfs / stso)', () => {
 
     setWorld({
       specVersion: SPEC_VERSION_CURRENT,
+      getDeploymentId: vi.fn(async () => 'test-deployment'),
       createQueueHandler: vi.fn(
         (_p: string, handler: (m: unknown, md: unknown) => Promise<unknown>) =>
           async () => {

packages/core/src/runtime.ts

@@ -39,6 +39,7 @@ import {
   isInlineOwnershipEnabled,
   isTurboEnabled,
 } from './runtime/constants.js';
+import { failRunIfDeploymentMismatch } from './runtime/deployment-guard.js';
 import {
   getQueueOverhead,
   getWorkflowQueueName,
@@ -702,6 +703,15 @@ export function workflowEntrypoint(
                         );
                         return;
                       }
+                      if (
+                        await failRunIfDeploymentMismatch({
+                          world,
+                          run: bgRun,
+                          requestId,
+                        })
+                      ) {
+                        return;
+                      }
                       const bgStartedAt = bgRun.startedAt
                         ? +bgRun.startedAt
                         : Date.now();
@@ -1065,6 +1075,23 @@ export function workflowEntrypoint(
                     } // end else (non-turbo run_started)
                   } // end if (!workflowRun)
 
+                  // Stop before replaying (or executing inline steps) if this
+                  // run reached a deployment other than the one that owns it.
+                  // Continuing risks code skew, and any step it dispatches would
+                  // resolve the wrong encryption key. `awaitRunReady` ensures a
+                  // turbo-backgrounded run_started has landed before we record
+                  // the failure.
+                  if (
+                    await failRunIfDeploymentMismatch({
+                      world,
+                      run: workflowRun,
+                      requestId,
+                      beforeFail: awaitRunReady,
+                    })
+                  ) {
+                    return;
+                  }
+
                   // Resolve the encryption key for this run's deployment.
                   // Used eagerly here since both runWorkflow (input
                   // hydration / hook payload decryption) and the run_failed