fix(core): pre-check deployment affinity before the lazy resume write

.changeset/lazy-resume-deployment-affinity.md

@@ -0,0 +1,6 @@
+---
+'@workflow/world': minor
+'@workflow/core': patch
+---
+
+Carry the run's pinned deployment on lazy resume messages so misrouted deliveries re-route before the `hook_received` write.

packages/core/src/runtime.test.ts

@@ -297,10 +297,12 @@ describe('workflowEntrypoint replay guards', () => {
 
   it('re-routes a misrouted lazy hook resume with its payload intact', async () => {
     // The lazy-resume producer parallelizes the `hook_received` write with this
-    // queue publish, so `hookInput` may be the only copy of the payload. The
-    // guard runs ahead of the re-ensure, so nothing is written here — which
-    // means the re-routed message has to carry `hookInput` or the resume is
-    // lost when the producer's direct write had not landed.
+    // queue publish, so `hookInput` may be the only copy of the payload. A
+    // modern message carries `hookInput.deploymentId`, so the cheap pre-check
+    // detects the mismatch BEFORE the fast path's hook_received write —
+    // zero event writes before re-routing — and the re-routed message has to
+    // carry the complete `hookInput` or the resume is lost when the
+    // producer's direct write had not landed.
     const workflowRun = await misroutedRun();
     const queueCalls: QueueCall[] = [];
     const hookInput = {
@@ -309,6 +311,7 @@ describe('workflowEntrypoint replay guards', () => {
       token: 'tok_1',
       payload: { serialized: true },
       payloadDigest: 'sha256:abc',
+      deploymentId: 'dpl_origin',
     };
 
     const createdEvents = await runWorkflowHandlerWithEvents(
@@ -320,13 +323,19 @@ describe('workflowEntrypoint replay guards', () => {
 
     expect(queueCalls).toHaveLength(1);
     expect(queueCalls[0].opts).toMatchObject({ deploymentId: 'dpl_origin' });
+    // The complete hookInput — payload included — survives re-routing.
     expect(queueCalls[0].message).toMatchObject({
       deploymentMismatchRetryCount: 1,
       hookInput,
     });
+    // Zero event writes before re-routing: neither the fast path's
+    // hook_received nor the generic setup's run_started ran.
     expect(createdEvents).not.toContainEqual(
       expect.objectContaining({ eventType: 'hook_received' })
     );
+    expect(createdEvents).not.toContainEqual(
+      expect.objectContaining({ eventType: 'run_started' })
+    );
   });
 
   it('fails a misrouted run once the re-route budget is spent', async () => {

packages/core/src/runtime.ts

@@ -1553,6 +1553,63 @@ export function workflowEntrypoint(
                     }
                   }
 
+                  // Deployment-affinity pre-check for the lazy hook fast
+                  // path below. New lazy-resume messages carry the run's
+                  // pinned deployment (`hookInput.deploymentId`), so a
+                  // misrouted delivery is detectable with a cheap ambient
+                  // deployment-id comparison BEFORE the fast path's
+                  // hook_received write — the correctly-routed common case
+                  // pays no run fetch and no latency. Only a detected
+                  // mismatch fetches the authoritative run and hands it to
+                  // the existing guard (which owns re-route/fail policy);
+                  // the re-routed message keeps the complete `hookInput`,
+                  // since it may hold the only copy of the resume payload.
+                  // When the pinned or ambient id is unavailable (older
+                  // producer message, a world without deployment affinity,
+                  // or a getDeploymentId failure), behavior is unchanged:
+                  // the authoritative guard after run setup — which remains
+                  // the protection before replay and step execution — still
+                  // covers the delivery; the fast path's write is idempotent
+                  // per (runId, resumeId), so a pre-guard write from an
+                  // older message stays convergent.
+                  if (
+                    !workflowRun &&
+                    hookInput?.deploymentId !== undefined &&
+                    // Same eligibility as guardDeploymentAffinity: without
+                    // atomic, immutable deployments the ids legitimately
+                    // differ (e.g. dpl_local@<version>) and comparing them
+                    // would trigger spurious run fetches.
+                    world.capabilities?.deploymentAffinity === true
+                  ) {
+                    const pinnedDeploymentId = hookInput.deploymentId;
+                    let currentDeploymentId: string | undefined;
+                    try {
+                      currentDeploymentId = await world.getDeploymentId();
+                    } catch {
+                      currentDeploymentId = undefined;
+                    }
+                    if (
+                      currentDeploymentId !== undefined &&
+                      currentDeploymentId !== pinnedDeploymentId
+                    ) {
+                      const run = await world.runs.get(runId, {
+                        resolveData: 'none',
+                      });
+                      if (
+                        (await guardDeployment(
+                          run,
+                          async () => ({
+                            ...(await replayMessage()),
+                            hookInput,
+                          }),
+                          awaitRunReady
+                        )) !== 'continue'
+                      ) {
+                        return;
+                      }
+                    }
+                  }
+
                   // --- Lazy hook resume fast path ---
                   // A lazy hook delivery (resumeId + digest on the queue
                   // message) hoists the consumer's idempotent hook_received
@@ -1980,17 +2037,24 @@ export function workflowEntrypoint(
 
                   // Covers every flow replay — initial start, step completions,
                   // hook resumptions, wait completions — and stops before any
-                  // workflow code, inline step, or event write happens when the
-                  // run is not pinned here (see `guardDeploymentAffinity`).
-                  // `awaitRunReady` orders turbo's `run_started` before either
-                  // stopping action.
+                  // workflow code or inline step executes when the run is not
+                  // pinned here (see `guardDeploymentAffinity`). This remains
+                  // the AUTHORITATIVE protection before replay and step
+                  // execution. `awaitRunReady` orders turbo's `run_started`
+                  // before either stopping action.
                   //
-                  // Runs ahead of the lazy-hook re-ensure below, so a misrouted
-                  // resume writes nothing here — which means the re-routed
-                  // message must carry `hookInput`, or a resume whose producer
-                  // write had not yet landed would be lost (that re-ensure is
-                  // idempotent per `resumeId`, so the pinned deployment still
-                  // converges on one event).
+                  // Lazy hook resumes are additionally pre-checked above the
+                  // fast path: a new message's `hookInput.deploymentId` is
+                  // compared against the ambient deployment id for free, and
+                  // only a detected mismatch fetches the authoritative run —
+                  // so a misrouted modern resume re-routes with zero event
+                  // writes. Older messages (no `hookInput.deploymentId`) reach
+                  // the fast path's idempotent hook_received write before this
+                  // guard; that write converges per (runId, resumeId) on the
+                  // pinned deployment, so it is safe, just not free. Either
+                  // way the re-routed message must carry `hookInput`, or a
+                  // resume whose producer write had not yet landed would be
+                  // lost.
                   if (
                     (await guardDeployment(
                       workflowRun,

packages/core/src/runtime/resume-hook.consumer-preload.test.ts

@@ -138,6 +138,13 @@ async function runResumeConsumerScenario(options: {
    * classification (consume the message vs rethrow for redelivery).
    */
   reEnsureRejection?: Error;
+  /**
+   * When set, the queue message's hookInput carries the producer-stamped
+   * pinned deployment id, activating the consumer's cheap pre-write
+   * affinity check. Omitted by default so most scenarios double as
+   * older-message fixtures (no deploymentId still parses and runs).
+   */
+  hookDeploymentId?: string;
 }) {
   const hookPreload = options.hookPreload ?? 'event-only';
   const runId = 'wrun_resume_consumer_preload';
@@ -325,14 +332,20 @@ async function runResumeConsumerScenario(options: {
     | ((message: unknown, metadata: unknown) => Promise<unknown>)
     | undefined;
   const queue = vi.fn().mockResolvedValue({ messageId: 'msg_resume' });
+  const runsGet = vi.fn(async () => workflowRun);
   setWorld({
     specVersion: SPEC_VERSION_CURRENT,
+    // Atomic, immutable deployments (as world-vercel declares), with the
+    // ambient id matching the run's pin — so the affinity pre-check and the
+    // authoritative guard both see a correctly routed delivery.
+    capabilities: { deploymentAffinity: true },
+    getDeploymentId: vi.fn(async () => deploymentId),
     createQueueHandler: vi.fn((_prefix, handler) => {
       capturedHandler = handler;
       return vi.fn();
     }),
     events: { list: listEvents, create: createEvent },
-    runs: { get: vi.fn(async () => workflowRun) },
+    runs: { get: runsGet },
     queue,
     getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
   } as unknown as World);
@@ -356,6 +369,9 @@ async function runResumeConsumerScenario(options: {
           token: hookToken,
           payload: payloadBytes,
           payloadDigest,
+          ...(options.hookDeploymentId !== undefined
+            ? { deploymentId: options.hookDeploymentId }
+            : {}),
         },
       },
       {
@@ -388,6 +404,7 @@ async function runResumeConsumerScenario(options: {
     runCompletedCreates,
     listEvents,
     createEvent,
+    runsGet,
     handlerError,
   };
 }
@@ -428,6 +445,33 @@ describe('lazy hook resume consumer preload', () => {
     expect(runCompletedCreates).toHaveLength(1);
   });
 
+  it('adds no run fetch for a correctly routed modern message (matching hookInput.deploymentId)', async () => {
+    // The pre-write affinity check is a pure ambient-id comparison on the
+    // matching path: the run is never fetched, and the streamed replay path
+    // is selected exactly as without the field.
+    const {
+      hookReceivedCreates,
+      runStartedCreates,
+      runCompletedCreates,
+      listEvents,
+      runsGet,
+      handlerError,
+    } = await runResumeConsumerScenario({
+      preloadHasHookReceived: false,
+      hookPreload: 'complete',
+      hookDeploymentId: 'dpl_resume_consumer_preload',
+    });
+
+    expect(handlerError).toBeUndefined();
+    expect(runsGet).not.toHaveBeenCalled();
+    // The streamed replay path remains selected: one setup request, no
+    // run_started, no events.list.
+    expect(hookReceivedCreates).toHaveLength(1);
+    expect(runStartedCreates).toHaveLength(0);
+    expect(listEvents).not.toHaveBeenCalled();
+    expect(runCompletedCreates).toHaveLength(1);
+  });
+
   it('initializes replay from the producer-won canonical event in the preload', async () => {
     const {
       hookReceivedCreates,

packages/core/src/runtime/resume-hook.parallel.test.ts

@@ -116,6 +116,9 @@ describe('resumeHook (parallel fast path)', () => {
       token: hook.token,
       payload: PAYLOAD_BYTES,
       payloadDigest: digest,
+      // The run's pinned deployment from the resume context, for the
+      // consumer's cheap pre-write affinity check.
+      deploymentId: 'deployment_par',
     });
   });
 

packages/core/src/runtime/resume-hook.ts

@@ -593,6 +593,10 @@ async function resumeHookImpl<T = any>(
                 token: hook.token,
                 payload: dehydratedPayload,
                 payloadDigest,
+                // Deployment affinity for the consumer's cheap pre-write
+                // check: lets a misrouted delivery re-route before its
+                // hoisted hook_received write instead of after.
+                deploymentId: resumeContext.deploymentId,
               },
             } satisfies WorkflowInvokePayload,
             queueOptions

packages/world/src/queue.ts

@@ -169,6 +169,16 @@ export const HookResumeInputSchema = z.object({
    * content-stable server-side.
    */
   payloadDigest: z.string(),
+  /**
+   * The deployment the run is pinned to, from the producer's resume context.
+   * Lets the consumer detect a misrouted delivery with a cheap ambient
+   * deployment-id comparison BEFORE its hoisted `hook_received` replay-preload
+   * write — only a detected mismatch pays for the authoritative run fetch and
+   * the deployment-affinity guard. Optional for queued-message compatibility:
+   * messages from older producers omit it and simply skip the pre-write
+   * check (the authoritative guard before replay still protects them).
+   */
+  deploymentId: z.string().optional(),
 });
 export type HookResumeInput = z.infer<typeof HookResumeInputSchema>;