From a8bff0285530a9be71e288ab330099f9bc82a515 Mon Sep 17 00:00:00 2001
From: Nathan Rajlich <n@n8.io>
Date: Tue, 11 Aug 2026 00:57:17 -0700
Subject: [PATCH 1/2] core: quickjs engine divergence-detection and
 write-fencing parity

---
 .changeset/quickjs-divergence-parity.md       |   5 +
 .../src/runtime/quickjs-divergence.test.ts    | 353 ++++++++++++++++++
 .../core/src/runtime/quickjs-divergence.ts    | 244 ++++++++++++
 .../quickjs-entrypoint.fencing.test.ts        | 235 ++++++++++++
 .../core/src/runtime/quickjs-entrypoint.ts    | 321 +++++++++++-----
 .../core/src/runtime/quickjs-runtime.test.ts  | 177 +++++++++
 packages/core/src/runtime/quickjs-runtime.ts  |  80 ++++
 7 files changed, 1324 insertions(+), 91 deletions(-)
 create mode 100644 .changeset/quickjs-divergence-parity.md
 create mode 100644 packages/core/src/runtime/quickjs-divergence.test.ts
 create mode 100644 packages/core/src/runtime/quickjs-divergence.ts
 create mode 100644 packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts

diff --git a/.changeset/quickjs-divergence-parity.md b/.changeset/quickjs-divergence-parity.md
new file mode 100644
index 0000000000..18f8eadc4a
--- /dev/null
+++ b/.changeset/quickjs-divergence-parity.md
@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+QuickJS engine: divergence-detection and write-fencing parity with the node:vm engine. Replays now arbitrate the event log at each fixed point (orphaned events, stepName/token/resumeAt mismatches) and escalate `ReplayDivergenceError` through the existing recovery machinery instead of silently delivering wrong payloads or surfacing corruption as `USER_ERROR`; all replay-context event writes now carry the optimistic-concurrency precondition snapshot (closing the documented KNOWN GAP), with `run_failed` left unfenced to match the node engine.
diff --git a/packages/core/src/runtime/quickjs-divergence.test.ts b/packages/core/src/runtime/quickjs-divergence.test.ts
new file mode 100644
index 0000000000..20b6ad7c79
--- /dev/null
+++ b/packages/core/src/runtime/quickjs-divergence.test.ts
@@ -0,0 +1,353 @@
+import type { Event } from '@workflow/world';
+import { describe, expect, it } from 'vitest';
+import {
+  findReplayDivergence,
+  type VmKnownOp,
+  type VmReplayView,
+} from './quickjs-divergence.js';
+
+/**
+ * Unit coverage for the QuickJS engine's replay-divergence arbitration —
+ * the pure core behind the fixed-point sweep in quickjs-runtime.ts. The
+ * semantics mirror the node:vm engine's EventsConsumer checks: every
+ * non-structural event must be claimed by an operation the replay drew,
+ * with matching identity fields.
+ */
+
+function makeEvent(overrides: Partial<Record<string, unknown>>): Event {
+  return {
+    eventId: 'evnt_01TEST',
+    runId: 'wrun_test',
+    eventType: 'step_created',
+    correlationId: 'step_01AAAA',
+    eventData: {},
+    createdAt: new Date('2025-01-01T00:00:00Z'),
+    ...overrides,
+  } as unknown as Event;
+}
+
+function view(ops: VmKnownOp[], extra?: Partial<VmReplayView>): VmReplayView {
+  return { ops, ...extra };
+}
+
+describe('findReplayDivergence', () => {
+  describe('structural events (never require a claim)', () => {
+    it.each([
+      ['run_created', {}],
+      ['run_started', {}],
+      ['run_completed', {}],
+      ['run_failed', {}],
+    ])('skips %s', (eventType) => {
+      const events = [makeEvent({ eventType, correlationId: 'wrun_test' })];
+      expect(findReplayDivergence(events, view([]))).toBeNull();
+    });
+
+    it('skips events without a correlationId', () => {
+      const events = [
+        makeEvent({ eventType: 'step_created', correlationId: undefined }),
+      ];
+      expect(findReplayDivergence(events, view([]))).toBeNull();
+    });
+
+    it('skips attr_set events written by a step', () => {
+      const events = [
+        makeEvent({
+          eventType: 'attr_set',
+          correlationId: 'attr_01AAAA',
+          eventData: { changes: {}, writer: { type: 'step' } },
+        }),
+      ];
+      expect(findReplayDivergence(events, view([]))).toBeNull();
+    });
+
+    it('skips event types outside the tracked families', () => {
+      const events = [
+        makeEvent({ eventType: 'some_future_event', correlationId: 'x_1' }),
+      ];
+      expect(findReplayDivergence(events, view([]))).toBeNull();
+    });
+  });
+
+  describe('orphaned events', () => {
+    it('reports a step event for a correlation id the replay never drew', () => {
+      const events = [
+        makeEvent({
+          eventType: 'step_created',
+          correlationId: 'step_01ORPHAN',
+          eventId: 'evnt_ORPHAN',
+          eventData: { stepName: 'step//test//other' },
+        }),
+      ];
+      const err = findReplayDivergence(
+        events,
+        view([{ correlationId: 'step_01AAAA', type: 'step' }])
+      );
+      expect(err).not.toBeNull();
+      expect(err?.name).toBe('ReplayDivergenceError');
+      expect(err?.message).toContain('Replay could not consume event');
+      expect(err?.message).toContain('step_01ORPHAN');
+      expect(err?.eventId).toBe('evnt_ORPHAN');
+    });
+
+    it('reports an attr_set written by the workflow for an unknown id', () => {
+      const events = [
+        makeEvent({
+          eventType: 'attr_set',
+          correlationId: 'attr_01ORPHAN',
+          eventData: { changes: {}, writer: { type: 'workflow' } },
+        }),
+      ];
+      expect(findReplayDivergence(events, view([]))).not.toBeNull();
+    });
+
+    it('accepts a hook event whose id is only known to the hook machinery', () => {
+      const events = [
+        makeEvent({
+          eventType: 'hook_received',
+          correlationId: 'hook_01AAAA',
+          eventData: {},
+        }),
+      ];
+      expect(
+        findReplayDivergence(events, view([], { hookCids: ['hook_01AAAA'] }))
+      ).toBeNull();
+      expect(
+        findReplayDivergence(events, view([], { abortCids: ['hook_01AAAA'] }))
+      ).toBeNull();
+    });
+
+    it('returns the FIRST divergence in log order', () => {
+      const events = [
+        makeEvent({
+          eventType: 'wait_created',
+          correlationId: 'wait_01FIRST',
+          eventId: 'evnt_FIRST',
+        }),
+        makeEvent({
+          eventType: 'step_created',
+          correlationId: 'step_01SECOND',
+          eventId: 'evnt_SECOND',
+        }),
+      ];
+      const err = findReplayDivergence(events, view([]));
+      expect(err?.eventId).toBe('evnt_FIRST');
+    });
+  });
+
+  describe('family mismatches', () => {
+    it('reports a step event for an id the replay drew as a wait', () => {
+      const events = [
+        makeEvent({
+          eventType: 'step_completed',
+          correlationId: 'wait_01AAAA',
+          eventData: { result: 1 },
+        }),
+      ];
+      const err = findReplayDivergence(
+        events,
+        view([{ correlationId: 'wait_01AAAA', type: 'wait' }])
+      );
+      expect(err?.message).toContain('does not match');
+    });
+
+    it('accepts hook_disposed claimed by a hook_dispose op sharing the hook id', () => {
+      const events = [
+        makeEvent({
+          eventType: 'hook_disposed',
+          correlationId: 'hook_01AAAA',
+        }),
+      ];
+      expect(
+        findReplayDivergence(
+          events,
+          view([
+            { correlationId: 'hook_01AAAA', type: 'hook', token: 't' },
+            { correlationId: 'hook_01AAAA', type: 'hook_dispose' },
+          ])
+        )
+      ).toBeNull();
+    });
+  });
+
+  describe('identity mismatches', () => {
+    const stepOp: VmKnownOp = {
+      correlationId: 'step_01AAAA',
+      type: 'step',
+      stepId: 'step//test//releaseStep',
+    };
+
+    it('reports a step event recorded for a different step function', () => {
+      const events = [
+        makeEvent({
+          eventType: 'step_created',
+          correlationId: 'step_01AAAA',
+          eventData: { stepName: 'step//test//recoverStep' },
+        }),
+      ];
+      const err = findReplayDivergence(events, view([stepOp]));
+      expect(err?.message).toContain('belongs to "step//test//recoverStep"');
+      expect(err?.message).toContain(
+        'current step consumer is "step//test//releaseStep"'
+      );
+    });
+
+    it('accepts a step event with the matching stepName', () => {
+      const events = [
+        makeEvent({
+          eventType: 'step_created',
+          correlationId: 'step_01AAAA',
+          eventData: { stepName: 'step//test//releaseStep' },
+        }),
+      ];
+      expect(findReplayDivergence(events, view([stepOp]))).toBeNull();
+    });
+
+    it('accepts a step event that does not carry a stepName', () => {
+      const events = [
+        makeEvent({
+          eventType: 'step_completed',
+          correlationId: 'step_01AAAA',
+          eventData: { result: 42 },
+        }),
+      ];
+      expect(findReplayDivergence(events, view([stepOp]))).toBeNull();
+    });
+
+    it('reports a hook event recorded under a different token', () => {
+      const events = [
+        makeEvent({
+          eventType: 'hook_created',
+          correlationId: 'hook_01AAAA',
+          eventData: { token: 'other-token' },
+        }),
+      ];
+      const err = findReplayDivergence(
+        events,
+        view([{ correlationId: 'hook_01AAAA', type: 'hook', token: 'mine' }])
+      );
+      expect(err?.message).toContain('belongs to token "other-token"');
+    });
+
+    it('accepts a hook event with the matching token', () => {
+      const events = [
+        makeEvent({
+          eventType: 'hook_created',
+          correlationId: 'hook_01AAAA',
+          eventData: { token: 'mine' },
+        }),
+      ];
+      expect(
+        findReplayDivergence(
+          events,
+          view([{ correlationId: 'hook_01AAAA', type: 'hook', token: 'mine' }])
+        )
+      ).toBeNull();
+    });
+
+    it('reports a wait_completed with a different resumeAt', () => {
+      const events = [
+        makeEvent({
+          eventType: 'wait_completed',
+          correlationId: 'wait_01AAAA',
+          eventData: { resumeAt: '2025-01-01T00:01:00.000Z' },
+        }),
+      ];
+      const err = findReplayDivergence(
+        events,
+        view([
+          {
+            correlationId: 'wait_01AAAA',
+            type: 'wait',
+            resumeAt: '2025-01-01T00:02:00.000Z',
+          },
+        ])
+      );
+      expect(err?.message).toContain('resumeAt');
+    });
+
+    it('accepts a wait_completed with the matching resumeAt (and one without eventData)', () => {
+      const ops = [
+        {
+          correlationId: 'wait_01AAAA',
+          type: 'wait',
+          resumeAt: '2025-01-01T00:01:00.000Z',
+        },
+      ];
+      expect(
+        findReplayDivergence(
+          [
+            makeEvent({
+              eventType: 'wait_completed',
+              correlationId: 'wait_01AAAA',
+              eventData: { resumeAt: '2025-01-01T00:01:00.000Z' },
+            }),
+          ],
+          view(ops)
+        )
+      ).toBeNull();
+      // The entrypoint's elapsed-wait pass writes wait_completed without
+      // eventData — nothing to validate, must be accepted.
+      expect(
+        findReplayDivergence(
+          [
+            makeEvent({
+              eventType: 'wait_completed',
+              correlationId: 'wait_01AAAA',
+              eventData: undefined,
+            }),
+          ],
+          view(ops)
+        )
+      ).toBeNull();
+    });
+  });
+
+  it('accepts a fully reproduced log', () => {
+    const events = [
+      makeEvent({ eventType: 'run_created', correlationId: 'wrun_test' }),
+      makeEvent({ eventType: 'run_started', correlationId: 'wrun_test' }),
+      makeEvent({
+        eventType: 'hook_created',
+        correlationId: 'hook_01AAAA',
+        eventData: { token: 't1' },
+      }),
+      makeEvent({
+        eventType: 'step_created',
+        correlationId: 'step_01AAAA',
+        eventData: { stepName: 'step//test//add' },
+      }),
+      makeEvent({
+        eventType: 'step_completed',
+        correlationId: 'step_01AAAA',
+        eventData: { result: 17 },
+      }),
+      makeEvent({
+        eventType: 'wait_created',
+        correlationId: 'wait_01AAAA',
+      }),
+      makeEvent({
+        eventType: 'hook_received',
+        correlationId: 'hook_01AAAA',
+        eventData: { payload: 'x' },
+      }),
+    ];
+    expect(
+      findReplayDivergence(
+        events,
+        view([
+          { correlationId: 'hook_01AAAA', type: 'hook', token: 't1' },
+          {
+            correlationId: 'step_01AAAA',
+            type: 'step',
+            stepId: 'step//test//add',
+          },
+          {
+            correlationId: 'wait_01AAAA',
+            type: 'wait',
+            resumeAt: '2025-01-01T00:05:00.000Z',
+          },
+        ])
+      )
+    ).toBeNull();
+  });
+});
diff --git a/packages/core/src/runtime/quickjs-divergence.ts b/packages/core/src/runtime/quickjs-divergence.ts
new file mode 100644
index 0000000000..a459d4f500
--- /dev/null
+++ b/packages/core/src/runtime/quickjs-divergence.ts
@@ -0,0 +1,244 @@
+import { ReplayDivergenceError } from '@workflow/errors';
+import type { Event } from '@workflow/world';
+
+/**
+ * Replay-divergence arbitration for the QuickJS engine.
+ *
+ * The node:vm engine detects a corrupted / diverged event log through its
+ * `EventsConsumer`: every event must be *claimed* by a consumer registered by
+ * the replaying workflow code, consumers validate identity fields
+ * (`stepName`, hook `token`, wait `resumeAt`), and an event nobody claims is
+ * escalated as `ReplayDivergenceError` — which `runtime.ts` turns into
+ * bounded recovery replays and, past the budget, a terminal
+ * `CorruptedEventLogError`.
+ *
+ * The QuickJS engine has no consumer registry: the host feeds events into VM
+ * heap structures keyed by correlation id. Before this module existed, a log
+ * the replay did not reproduce was absorbed silently — a `step_completed`
+ * for a different step's ordinal resolved the wrong call with the wrong
+ * payload (no `stepName` check), and events for correlation ids the replay
+ * never drew sat in dead buffers while `run_completed` was written over the
+ * unreproduced log. The same corruption class the node engine reports as
+ * `CORRUPTED_EVENT_LOG` therefore surfaced on QuickJS as `USER_ERROR`
+ * (e.g. a self-`HookConflictError`) or as a silently wrong completion.
+ *
+ * `findReplayDivergence` closes that gap. It runs against the replay's
+ * *fixed point* — the moment the VM's job queue is fully drained and no more
+ * progress is possible — which is a stronger position than the node engine
+ * ever gets to check from: there is no in-flight cross-realm microtask work
+ * to wait out, so no grace window or delivery-idle heuristic is needed. The
+ * caller dumps the VM's known operations (every correlation id the replay
+ * drew, with identity fields) and this function arbitrates the observed
+ * event log against them.
+ */
+
+/** One operation the replaying VM has drawn a correlation id for. */
+export interface VmKnownOp {
+  correlationId: string;
+  /** `step` | `wait` | `hook` | `hook_dispose` | `attribute` (open set). */
+  type: string;
+  /** Step operations: the full step id (`step//<module>//<name>`). */
+  stepId?: string;
+  /** Hook operations: the user-supplied (or system-derived) token. */
+  token?: string;
+  /** Wait operations: the ISO resume timestamp the VM computed. */
+  resumeAt?: string;
+}
+
+/**
+ * The replay's view of its own draws at a fixed point, dumped from the VM.
+ */
+export interface VmReplayView {
+  ops: VmKnownOp[];
+  /** Correlation ids registered in `globalThis.__hooks` (hook machinery). */
+  hookCids?: string[];
+  /** Correlation ids registered in `globalThis.__abortSignals`. */
+  abortCids?: string[];
+}
+
+/** The operation family an event type must be claimed by. */
+function expectedFamilies(eventType: string): string[] | null {
+  if (eventType.startsWith('step_')) return ['step'];
+  if (eventType.startsWith('wait_')) return ['wait'];
+  if (eventType.startsWith('hook_')) return ['hook', 'hook_dispose'];
+  if (eventType === 'attr_set') return ['attribute'];
+  return null;
+}
+
+function eventDataOf(event: Event): Record<string, unknown> | undefined {
+  return 'eventData' in event && event.eventData
+    ? (event.eventData as Record<string, unknown>)
+    : undefined;
+}
+
+/**
+ * Events that need no claim from workflow code (parity with the node
+ * engine's structural lifecycle consumer in workflow.ts): run lifecycle
+ * events, attribute writes not performed by the workflow body, and anything
+ * without a correlation id or outside the families the engines track.
+ */
+function isStructural(event: Event): boolean {
+  if (!event.correlationId) return true;
+  if (event.eventType.startsWith('run_')) return true;
+  if (event.eventType === 'attr_set') {
+    const writer = eventDataOf(event)?.writer as { type?: string } | undefined;
+    if (writer?.type !== 'workflow') return true;
+  }
+  return expectedFamilies(event.eventType) === null;
+}
+
+/**
+ * Arbitrate the observed event log against the replay's fixed-point view.
+ * Returns the first divergence in log order, or `null` when the replay
+ * reproduced the log. Divergences, in the order they are checked per event:
+ *
+ *  1. **Orphaned event** — the log contains a correlation id this replay
+ *     never drew. Mirrors the node engine's unconsumed-event check
+ *     (`onUnconsumedEvent` → workflow.ts).
+ *  2. **Family mismatch** — the correlation id exists but belongs to a
+ *     different kind of operation (a `step_*` event for a cid the replay
+ *     drew as a wait, etc.). Mirrors the node consumers' unexpected-event
+ *     checks.
+ *  3. **Identity mismatch** — the correlation id and family match but the
+ *     recorded identity differs from what this replay derived: a step
+ *     event's `stepName` (step.ts), a hook event's `token` (hook.ts), or a
+ *     `wait_completed`'s `resumeAt` (sleep.ts).
+ */
+export function findReplayDivergence(
+  events: readonly Event[],
+  view: VmReplayView
+): ReplayDivergenceError | null {
+  const opsByCid = new Map<string, VmKnownOp[]>();
+  for (const op of view.ops) {
+    const list = opsByCid.get(op.correlationId);
+    if (list) {
+      list.push(op);
+    } else {
+      opsByCid.set(op.correlationId, [op]);
+    }
+  }
+  const auxCids = new Set<string>([
+    ...(view.hookCids ?? []),
+    ...(view.abortCids ?? []),
+  ]);
+
+  for (const event of events) {
+    if (isStructural(event)) continue;
+    const divergence = arbitrateEvent(event, opsByCid, auxCids);
+    if (divergence) return divergence;
+  }
+
+  return null;
+}
+
+function arbitrateEvent(
+  event: Event,
+  opsByCid: Map<string, VmKnownOp[]>,
+  auxCids: Set<string>
+): ReplayDivergenceError | null {
+  const cid = event.correlationId as string;
+  const families = expectedFamilies(event.eventType) as string[];
+  const ops = opsByCid.get(cid);
+
+  if (ops === undefined || ops.length === 0) {
+    // Hook machinery can know a cid the pending list does not (defensive;
+    // the bootstrap pushes a pending op for every draw today).
+    if (families.includes('hook') && auxCids.has(cid)) return null;
+    return new ReplayDivergenceError(
+      `Replay could not consume event: eventType=${event.eventType}, correlationId=${cid}, eventId=${event.eventId}.`,
+      { eventId: event.eventId }
+    );
+  }
+
+  const familyOps = ops.filter((op) => families.includes(op.type));
+  if (familyOps.length === 0) {
+    return new ReplayDivergenceError(
+      `Replay divergence: event ${event.eventType} for ${cid} does not match the "${ops[0].type}" operation the replay drew for that correlation id`,
+      { eventId: event.eventId }
+    );
+  }
+
+  return findIdentityMismatch(event, families, familyOps);
+}
+
+/**
+ * Identity validation for an event whose correlation id and family both
+ * matched a VM operation: the recorded identity fields must equal what this
+ * replay derived. Mirrors the node consumers' checks in step.ts, hook.ts
+ * and sleep.ts.
+ */
+function findIdentityMismatch(
+  event: Event,
+  families: string[],
+  familyOps: VmKnownOp[]
+): ReplayDivergenceError | null {
+  if (families.includes('step')) {
+    return findStepNameMismatch(event, familyOps[0]);
+  }
+  if (families.includes('hook')) {
+    return findHookTokenMismatch(event, familyOps);
+  }
+  if (event.eventType === 'wait_completed') {
+    return findResumeAtMismatch(event, familyOps[0]);
+  }
+  return null;
+}
+
+function findStepNameMismatch(
+  event: Event,
+  op: VmKnownOp
+): ReplayDivergenceError | null {
+  const eventStepName = eventDataOf(event)?.stepName;
+  if (
+    typeof eventStepName === 'string' &&
+    typeof op.stepId === 'string' &&
+    eventStepName !== op.stepId
+  ) {
+    return new ReplayDivergenceError(
+      `Replay divergence: step event ${event.eventType} for ${event.correlationId} belongs to "${eventStepName}", but the current step consumer is "${op.stepId}"`,
+      { eventId: event.eventId }
+    );
+  }
+  return null;
+}
+
+function findHookTokenMismatch(
+  event: Event,
+  familyOps: VmKnownOp[]
+): ReplayDivergenceError | null {
+  const eventToken = eventDataOf(event)?.token;
+  const op = familyOps.find((o) => typeof o.token === 'string');
+  if (
+    typeof eventToken === 'string' &&
+    op !== undefined &&
+    eventToken !== op.token
+  ) {
+    return new ReplayDivergenceError(
+      `Replay divergence: hook event ${event.eventType} for ${event.correlationId} belongs to token "${eventToken}", but the current hook consumer expects "${op.token}"`,
+      { eventId: event.eventId }
+    );
+  }
+  return null;
+}
+
+function findResumeAtMismatch(
+  event: Event,
+  op: VmKnownOp
+): ReplayDivergenceError | null {
+  const eventResumeAt = eventDataOf(event)?.resumeAt;
+  if (eventResumeAt === undefined || typeof op.resumeAt !== 'string') {
+    return null;
+  }
+  const eventMs = new Date(eventResumeAt as string | Date).getTime();
+  const expectedMs = new Date(op.resumeAt).getTime();
+  if (eventMs !== expectedMs) {
+    const eventForMessage = Number.isFinite(eventMs)
+      ? new Date(eventMs).toISOString()
+      : String(eventResumeAt);
+    return new ReplayDivergenceError(
+      `Replay divergence: wait_completed event for ${event.correlationId} has resumeAt "${eventForMessage}", but the current wait consumer expects "${new Date(op.resumeAt).toISOString()}"`,
+      { eventId: event.eventId }
+    );
+  }
+  return null;
+}
diff --git a/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts
new file mode 100644
index 0000000000..c7c4672db2
--- /dev/null
+++ b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts
@@ -0,0 +1,235 @@
+/**
+ * Pins the QuickJS engine's optimistic-concurrency precondition guard:
+ * every replay-context event write must carry the view snapshot
+ * (`stateUpdatedAt` / `stateEventCount`) describing the event log the
+ * invocation derived the write from, so a supporting World can reject a
+ * stale writer with 412 — parity with the node:vm engine's `createGuarded`
+ * (suspension-handler.ts) and guarded `run_completed`. The terminal
+ * `run_failed` is deliberately unfenced (same asymmetry as the node
+ * engine: a failing run must be able to terminate from a stale view).
+ *
+ * The QuickJS VM itself is mocked — the guard lives entirely in the
+ * entrypoint's write paths.
+ */
+import {
+  type CreateEventRequest,
+  type Event,
+  SPEC_VERSION_CURRENT,
+  type WorkflowRun,
+  type World,
+} from '@workflow/world';
+import { monotonicFactory } from 'ulid';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { dehydrateStepReturnValue } from '../serialization.js';
+import { latestEventStateUpdatedAt } from './helpers.js';
+import { setWorld } from './world.js';
+
+vi.mock('@vercel/functions', () => ({ waitUntil: vi.fn() }));
+vi.mock('./get-port-lazy.js', () => ({
+  getPortLazy: vi.fn().mockResolvedValue(3000),
+}));
+
+const startQuickJSWorkflow = vi.fn();
+vi.mock('./quickjs-runtime.js', () => ({
+  startQuickJSWorkflow: (...args: unknown[]) => startQuickJSWorkflow(...args),
+}));
+
+const ulid = monotonicFactory();
+
+function makeEvent(
+  eventType: string,
+  overrides: Partial<Record<string, unknown>> = {}
+): Event {
+  return {
+    eventId: `evnt_${ulid()}`,
+    runId: 'wrun_quickjs_fencing',
+    eventType,
+    eventData: {},
+    createdAt: new Date('2026-05-19T12:00:00.000Z'),
+    ...overrides,
+  } as unknown as Event;
+}
+
+async function runScenario(options: {
+  events: Event[];
+  vmResult: Record<string, unknown>;
+}) {
+  const runId = 'wrun_quickjs_fencing';
+  const startedAt = new Date('2026-05-19T12:00:00.000Z');
+  const workflowRun: WorkflowRun = {
+    runId,
+    workflowName: 'workflow',
+    status: 'running',
+    input: [],
+    deploymentId: 'dpl_quickjs_fencing',
+    specVersion: SPEC_VERSION_CURRENT,
+    startedAt,
+    createdAt: startedAt,
+    updatedAt: startedAt,
+  };
+
+  const created: {
+    request: CreateEventRequest;
+    params: Record<string, unknown> | undefined;
+  }[] = [];
+  let listCallCount = 0;
+  const listEvents = vi.fn(async () => {
+    listCallCount++;
+    if (listCallCount === 1) {
+      return {
+        data: [...options.events],
+        cursor: options.events.at(-1)?.eventId ?? null,
+        hasMore: false,
+      };
+    }
+    return { data: [], cursor: null, hasMore: false };
+  });
+  const createEvent = vi.fn(
+    async (
+      _runId: string,
+      request: CreateEventRequest,
+      params?: Record<string, unknown>
+    ) => {
+      created.push({ request, params });
+      return { event: { ...request, runId, eventId: `evnt_${ulid()}` } };
+    }
+  );
+
+  setWorld({
+    specVersion: SPEC_VERSION_CURRENT,
+    capabilities: { preconditionGuard: true },
+    events: { list: listEvents, create: createEvent },
+    runs: { get: vi.fn(async () => workflowRun) },
+    queue: vi.fn().mockResolvedValue({ messageId: 'msg_quickjs' }),
+    getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+  } as unknown as World);
+
+  startQuickJSWorkflow.mockResolvedValue({
+    result: options.vmResult,
+    continueWithEvents: vi.fn(),
+    dispose: vi.fn(),
+  });
+
+  const { runWorkflowWithQuickJS } = await import('./quickjs-entrypoint.js');
+  await runWorkflowWithQuickJS({
+    workflowCode: '// mocked VM',
+    workflowName: 'workflow',
+    workflowRun,
+    preloadedEvents: options.events,
+    preloadedEventsComplete: true,
+  });
+
+  return { created };
+}
+
+afterEach(() => {
+  vi.clearAllMocks();
+  vi.unstubAllEnvs();
+});
+
+describe('QuickJS entrypoint precondition guard', () => {
+  const baseLog = () => [
+    makeEvent('run_created', { eventData: { input: undefined } }),
+    makeEvent('run_started'),
+  ];
+
+  it('fences run_completed with the view snapshot', async () => {
+    const events = baseLog();
+    const { created } = await runScenario({
+      events,
+      vmResult: {
+        completed: {
+          result: await dehydrateStepReturnValue(
+            'done',
+            'wrun_quickjs_fencing',
+            undefined
+          ),
+        },
+      },
+    });
+
+    const runCompleted = created.find(
+      (c) => c.request.eventType === 'run_completed'
+    );
+    expect(runCompleted).toBeDefined();
+    expect(runCompleted?.params).toMatchObject({
+      stateUpdatedAt: latestEventStateUpdatedAt(events),
+      stateEventCount: events.length,
+    });
+  });
+
+  it('fences suspension writes (hook_created, wait_created) with the view snapshot', async () => {
+    const events = baseLog();
+    const { created } = await runScenario({
+      events,
+      vmResult: {
+        suspended: {
+          pendingOperations: [
+            {
+              type: 'hook',
+              correlationId: 'hook_01FENCE',
+              token: 'fence-token',
+              isWebhook: false,
+              hasCreatedEvent: false,
+            },
+            {
+              type: 'wait',
+              correlationId: 'wait_01FENCE',
+              // Far future so the loop schedules a continuation instead of
+              // completing the wait.
+              resumeAt: '2027-01-01T00:00:00.000Z',
+              hasCreatedEvent: false,
+            },
+          ],
+        },
+      },
+    });
+
+    const expected = {
+      stateUpdatedAt: latestEventStateUpdatedAt(events),
+      stateEventCount: events.length,
+    };
+    const hookCreated = created.find(
+      (c) => c.request.eventType === 'hook_created'
+    );
+    const waitCreated = created.find(
+      (c) => c.request.eventType === 'wait_created'
+    );
+    expect(hookCreated?.params).toMatchObject(expected);
+    expect(waitCreated?.params).toMatchObject(expected);
+  });
+
+  it('leaves run_failed unfenced (terminal-failure asymmetry)', async () => {
+    const { created } = await runScenario({
+      events: baseLog(),
+      vmResult: {
+        failed: { message: 'boom', name: 'Error' },
+      },
+    });
+
+    const runFailed = created.find((c) => c.request.eventType === 'run_failed');
+    expect(runFailed).toBeDefined();
+    expect(runFailed?.params).toBeUndefined();
+  });
+
+  it('sends no snapshot when the guard is disabled', async () => {
+    vi.stubEnv('WORKFLOW_PRECONDITION_GUARD', '0');
+    const { created } = await runScenario({
+      events: baseLog(),
+      vmResult: {
+        completed: {
+          result: await dehydrateStepReturnValue(
+            'done',
+            'wrun_quickjs_fencing',
+            undefined
+          ),
+        },
+      },
+    });
+
+    const runCompleted = created.find(
+      (c) => c.request.eventType === 'run_completed'
+    );
+    expect(runCompleted?.params ?? {}).not.toHaveProperty('stateUpdatedAt');
+  });
+});
diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts
index f725cf9ee3..cf4ea07e36 100644
--- a/packages/core/src/runtime/quickjs-entrypoint.ts
+++ b/packages/core/src/runtime/quickjs-entrypoint.ts
@@ -49,7 +49,13 @@ import {
   getMaxInlineSteps,
 } from './constants.js';
 import { getPortLazy } from './get-port-lazy.js';
-import { getWorkflowQueueName, queueMessage } from './helpers.js';
+import {
+  getWorkflowQueueName,
+  isPreconditionGuardEnabled,
+  latestEventStateUpdatedAt,
+  type PreconditionSnapshotParams,
+  queueMessage,
+} from './helpers.js';
 import {
   BASELINE_BUNDLE_FILENAME,
   type PendingAttribute,
@@ -205,6 +211,15 @@ async function dispatchPendingOps(params: {
    * queues.
    */
   nextTraceCarrier: () => Promise<Record<string, string>>;
+  /**
+   * Optimistic-concurrency snapshot describing the event log this batch of
+   * writes was derived from (see `preconditionSnapshotParams`). Attached to
+   * every event create so a supporting World rejects writes from a stale
+   * view with 412 — the rejection propagates to runtime.ts, which restarts
+   * the replay over a corrected log. Mirrors the node:vm engine's
+   * `createGuarded` in suspension-handler.ts.
+   */
+  precondition?: PreconditionSnapshotParams;
   wfdiag: (checkpoint: string, fields: Record<string, unknown>) => void;
 }): Promise<{
   createdAttributeEvent: boolean;
@@ -218,6 +233,7 @@ async function dispatchPendingOps(params: {
     pendingOperations,
     namespace,
     nextTraceCarrier,
+    precondition,
   } = params;
   const skipStepCreation = params.skipStepCreation;
   const wfdiag = params.wfdiag;
@@ -261,26 +277,30 @@ async function dispatchPendingOps(params: {
           typeof hook.metadata === 'undefined'
             ? undefined
             : await encryptSerializedData(hook.metadata, encryptionKey);
-        const result = await world.events.create(runId, {
-          eventType: 'hook_created',
-          specVersion: SPEC_VERSION_CURRENT,
-          correlationId: hook.correlationId,
-          eventData: {
-            token: hook.token,
-            tokenRetentionUntil:
-              hook.tokenRetentionUntil === undefined
-                ? undefined
-                : new Date(hook.tokenRetentionUntil),
-            metadata: encryptedMetadata,
-            // Always include isWebhook explicitly. Worlds default it to
-            // `true` when absent, which would break the public webhook
-            // endpoint's 404 guard for hooks created via createHook().
-            isWebhook: hook.isWebhook,
-            // System hooks (AbortController) are exempt from user
-            // token namespace conflict checks.
-            ...(hook.isSystem ? { isSystem: true } : {}),
-          } as any,
-        });
+        const result = await world.events.create(
+          runId,
+          {
+            eventType: 'hook_created',
+            specVersion: SPEC_VERSION_CURRENT,
+            correlationId: hook.correlationId,
+            eventData: {
+              token: hook.token,
+              tokenRetentionUntil:
+                hook.tokenRetentionUntil === undefined
+                  ? undefined
+                  : new Date(hook.tokenRetentionUntil),
+              metadata: encryptedMetadata,
+              // Always include isWebhook explicitly. Worlds default it to
+              // `true` when absent, which would break the public webhook
+              // endpoint's 404 guard for hooks created via createHook().
+              isWebhook: hook.isWebhook,
+              // System hooks (AbortController) are exempt from user
+              // token namespace conflict checks.
+              ...(hook.isSystem ? { isSystem: true } : {}),
+            } as any,
+          },
+          precondition
+        );
 
         // If storage detected a real token conflict with another
         // workflow's hook, re-queue so the workflow handler can
@@ -321,15 +341,19 @@ async function dispatchPendingOps(params: {
             )) as Uint8Array)
           : undefined;
       try {
-        await world.events.create(runId, {
-          eventType: 'hook_received',
-          specVersion: SPEC_VERSION_CURRENT,
-          correlationId: hook.correlationId,
-          eventData: {
-            token: hook.token,
-            payload: abortPayload,
-          } as any,
-        });
+        await world.events.create(
+          runId,
+          {
+            eventType: 'hook_received',
+            specVersion: SPEC_VERSION_CURRENT,
+            correlationId: hook.correlationId,
+            eventData: {
+              token: hook.token,
+              payload: abortPayload,
+            } as any,
+          },
+          precondition
+        );
       } catch (err) {
         if (!EntityConflictError.is(err)) throw err;
       }
@@ -363,11 +387,15 @@ async function dispatchPendingOps(params: {
     op: PendingHookDispose
   ): Promise<void> => {
     try {
-      await world.events.create(runId, {
-        eventType: 'hook_disposed',
-        specVersion: SPEC_VERSION_CURRENT,
-        correlationId: op.correlationId,
-      });
+      await world.events.create(
+        runId,
+        {
+          eventType: 'hook_disposed',
+          specVersion: SPEC_VERSION_CURRENT,
+          correlationId: op.correlationId,
+        },
+        precondition
+      );
     } catch (err) {
       if (EntityConflictError.is(err)) return;
       // Disposing a hook whose entity no longer (or never) exists is an
@@ -443,15 +471,19 @@ async function dispatchPendingOps(params: {
           // on the host side — matching what
           // `dehydrateStepArguments` does in the node:vm engine.
           try {
-            await world.events.create(runId, {
-              eventType: 'step_created',
-              specVersion: SPEC_VERSION_CURRENT,
-              correlationId: step.correlationId,
-              eventData: {
-                stepName: step.stepId,
-                input: await encryptSerializedData(step.input, encryptionKey),
+            await world.events.create(
+              runId,
+              {
+                eventType: 'step_created',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: step.correlationId,
+                eventData: {
+                  stepName: step.stepId,
+                  input: await encryptSerializedData(step.input, encryptionKey),
+                },
               },
-            });
+              precondition
+            );
           } catch (err) {
             if (EntityConflictError.is(err)) return;
             throw err;
@@ -467,18 +499,22 @@ async function dispatchPendingOps(params: {
       opsPromises.push(
         (async () => {
           try {
-            await world.events.create(runId, {
-              eventType: 'attr_set',
-              specVersion: SPEC_VERSION_CURRENT,
-              correlationId: attr.correlationId,
-              eventData: {
-                changes: attr.changes,
-                writer: { type: 'workflow' },
-                ...(attr.allowReservedAttributes
-                  ? { allowReservedAttributes: true }
-                  : {}),
-              } as any,
-            });
+            await world.events.create(
+              runId,
+              {
+                eventType: 'attr_set',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: attr.correlationId,
+                eventData: {
+                  changes: attr.changes,
+                  writer: { type: 'workflow' },
+                  ...(attr.allowReservedAttributes
+                    ? { allowReservedAttributes: true }
+                    : {}),
+                } as any,
+              },
+              precondition
+            );
             createdAttributeEvent = true;
           } catch (err) {
             if (EntityConflictError.is(err)) {
@@ -496,14 +532,18 @@ async function dispatchPendingOps(params: {
       opsPromises.push(
         (async () => {
           try {
-            await world.events.create(runId, {
-              eventType: 'wait_created',
-              specVersion: SPEC_VERSION_CURRENT,
-              correlationId: wait.correlationId,
-              eventData: {
-                resumeAt: new Date(wait.resumeAt),
+            await world.events.create(
+              runId,
+              {
+                eventType: 'wait_created',
+                specVersion: SPEC_VERSION_CURRENT,
+                correlationId: wait.correlationId,
+                eventData: {
+                  resumeAt: new Date(wait.resumeAt),
+                },
               },
-            });
+              precondition
+            );
           } catch (err) {
             if (EntityConflictError.is(err)) return;
             throw err;
@@ -525,15 +565,16 @@ async function dispatchPendingOps(params: {
  * This replaces the `node:vm` replay path (runWorkflow + EventsConsumer)
  * with a QuickJS VM invocation that performs the same full event replay.
  *
- * KNOWN GAP — precondition guard: unlike the node:vm path, no event write
- * in this file participates in the optimistic-concurrency precondition
- * guard (`withPreconditionRetry` + `stateUpdatedAtForCreate`), which
- * protects a writer holding a stale event-log snapshot from clobbering a
- * concurrent one. The engine currently relies on per-(runId,
- * correlationId) event uniqueness (EntityConflictError dedup) alone. This
- * is a deliberate simplification while the engine is experimental — wiring
- * the guard is tracked follow-up work; anyone adding new write paths here
- * should not assume parity with the node engine on this axis.
+ * Precondition guard: every replay-context event write in this file carries
+ * the optimistic-concurrency snapshot (`preconditionSnapshotParams`
+ * semantics — see the view tracker in `runWorkflowWithQuickJS`), so a
+ * supporting World rejects writes derived from a stale event-log view with
+ * 412. The rejection propagates out of this entrypoint to the replay loop's
+ * catch in runtime.ts, which restarts the replay over a corrected log —
+ * the same recovery the node:vm engine uses. `run_failed` is deliberately
+ * unfenced, matching the node engine's asymmetry (a terminal failure must
+ * be recordable even from a stale view). When adding a new write path
+ * here, thread the snapshot through it.
  */
 export async function runWorkflowWithQuickJS(params: {
   workflowCode: string;
@@ -687,6 +728,7 @@ export async function runWorkflowWithQuickJS(params: {
   // preload (lazy hook fast path) is trusted the same way.
   let events: Event[];
   let eventsFetchedPages = 0;
+  let initialFetchCursor: string | null = null;
   const usePreloaded =
     (preloadedEventsComplete === true &&
       Array.isArray(preloadedEvents) &&
@@ -720,8 +762,60 @@ export async function runWorkflowWithQuickJS(params: {
     }
 
     events = allEvents;
+    initialFetchCursor = cursor;
   }
 
+  // ---- Optimistic-concurrency view tracker ----
+  // The precondition snapshot attached to every replay-context write below
+  // describes the event-log view this invocation derived the write from:
+  // the maximum event-id ULID time over every event it has observed, the
+  // count of those events, and the listing cursor. Self-written events are
+  // observed either inline (when the create returns the event and it is
+  // spliced into the local log) or on read-back via fetchUnseenEvents —
+  // watermark and count always describe the same consistent set, which is
+  // the invariant the backend's marker/count comparison relies on. See
+  // `preconditionSnapshotParams` in helpers.ts for the field semantics.
+  //
+  // Also derives the open-hook state used to suppress optimistic inline
+  // starts (see the executeStep call in the inline loop): while a hook is
+  // open, an out-of-band hook_received can make this view stale at any
+  // moment, so a fenced claim must settle before the step body runs.
+  let viewMaxEvent: Event | undefined;
+  let viewEventCount = 0;
+  let viewCursor: string | null = initialFetchCursor;
+  const openHookCids = new Set<string>();
+  const observeView = (observed: Event[], cursor?: string | null): void => {
+    for (const e of observed) {
+      if (!e.eventId) continue;
+      viewEventCount++;
+      if (viewMaxEvent === undefined || e.eventId > viewMaxEvent.eventId) {
+        viewMaxEvent = e;
+      }
+      if (e.correlationId) {
+        if (e.eventType === 'hook_created') {
+          openHookCids.add(e.correlationId);
+        } else if (e.eventType === 'hook_disposed') {
+          openHookCids.delete(e.correlationId);
+        }
+      }
+    }
+    if (cursor) viewCursor = cursor;
+  };
+  observeView(events);
+  const preconditionSnapshot = (): PreconditionSnapshotParams => {
+    if (!isPreconditionGuardEnabled() || viewMaxEvent === undefined) return {};
+    const stateUpdatedAt = latestEventStateUpdatedAt([viewMaxEvent]);
+    if (stateUpdatedAt === undefined) return {};
+    return {
+      stateUpdatedAt,
+      stateEventCount: viewEventCount,
+      ...(viewCursor ? { stateCursor: viewCursor } : {}),
+    };
+  };
+  const guardEnforced =
+    isPreconditionGuardEnabled() &&
+    world.capabilities?.preconditionGuard === true;
+
   // Event-limit guard: fail a runaway run once its log reaches the
   // server-supplied ceiling — same enforcement point as the node:vm
   // engine's replay loop.
@@ -765,12 +859,21 @@ export async function runWorkflowWithQuickJS(params: {
       const resumeAt = eventData?.resumeAt;
       if (resumeAt && now >= new Date(resumeAt as string).getTime()) {
         try {
-          const result = await world.events.create(runId, {
-            eventType: 'wait_completed',
-            specVersion: SPEC_VERSION_CURRENT,
-            correlationId: event.correlationId,
-          });
-          if (result.event) events.push(result.event);
+          const result = await world.events.create(
+            runId,
+            {
+              eventType: 'wait_completed',
+              specVersion: SPEC_VERSION_CURRENT,
+              correlationId: event.correlationId,
+            },
+            preconditionSnapshot()
+          );
+          if (result.event) {
+            events.push(result.event);
+            // Spliced into the local log ahead of the seen-set init, so a
+            // read-back never observes it — count it into the view here.
+            observeView([result.event]);
+          }
         } catch (err) {
           if (EntityConflictError.is(err)) continue;
           throw err;
@@ -959,6 +1062,7 @@ export async function runWorkflowWithQuickJS(params: {
       hasMore = response.data.length > 0 && response.cursor != null;
     }
     observeEventsForOwnership(unseen);
+    observeView(unseen, cursor);
     return unseen;
   };
 
@@ -1021,6 +1125,7 @@ export async function runWorkflowWithQuickJS(params: {
         nextTraceCarrier,
         pendingOperations: opsToDispatch,
         skipStepCreation: inlineClaimCids,
+        precondition: preconditionSnapshot(),
         wfdiag,
       });
       if (
@@ -1066,11 +1171,15 @@ export async function runWorkflowWithQuickJS(params: {
         waitCompletePromises.push(
           (async () => {
             try {
-              await world.events.create(runId, {
-                eventType: 'wait_completed',
-                specVersion: SPEC_VERSION_CURRENT,
-                correlationId: wait.correlationId,
-              });
+              await world.events.create(
+                runId,
+                {
+                  eventType: 'wait_completed',
+                  specVersion: SPEC_VERSION_CURRENT,
+                  correlationId: wait.correlationId,
+                },
+                preconditionSnapshot()
+              );
             } catch (err) {
               if (EntityConflictError.is(err)) return;
               throw err;
@@ -1261,6 +1370,19 @@ export async function runWorkflowWithQuickJS(params: {
       // own, matching the node:vm engine, where a long sequential
       // workflow likewise runs step-by-step until the platform reclaims
       // the invocation and a redelivery resumes from the log.
+      // Guard the inline claims: the lazy step_started (which creates the
+      // step) carries the same view snapshot as the durable writes above,
+      // so a stale replay can never commit a step body's result — the
+      // node engine's `inlineClaimSnapshot`. While a hook is open, an
+      // out-of-band hook_received can stale this view at any moment;
+      // on guard-enforcing Worlds, await the claim before running the
+      // body so a 412-fenced step never executes user code (mirrors the
+      // node engine's suppressOptimisticStart).
+      const inlineClaimSnapshot = preconditionSnapshot();
+      const suppressOptimisticStart =
+        guardEnforced &&
+        (openHookCids.size > 0 ||
+          pendingOperations.some((op) => op.type === 'hook'));
       budget.pause();
       let outcomes: StepExecutionResult[];
       try {
@@ -1296,6 +1418,10 @@ export async function runWorkflowWithQuickJS(params: {
                   // A lazy step is brand-new by construction — first
                   // attempt.
                   authoritativeAttempt: 1,
+                  // Fence the claim against stale views — see
+                  // inlineClaimSnapshot above.
+                  preconditionSnapshot: inlineClaimSnapshot,
+                  suppressOptimisticStart,
                 }))()
             )
           )
@@ -1405,6 +1531,7 @@ export async function runWorkflowWithQuickJS(params: {
           namespace,
           nextTraceCarrier,
           pendingOperations: result.completed.drainOperations,
+          precondition: preconditionSnapshot(),
           wfdiag,
         });
       } catch (err) {
@@ -1422,16 +1549,24 @@ export async function runWorkflowWithQuickJS(params: {
     // events have the same `encr`-prefixed payload shape that the node:vm
     // engine's `dehydrateWorkflowReturnValue` produces.
     try {
-      await world.events.create(runId, {
-        eventType: 'run_completed',
-        specVersion: SPEC_VERSION_CURRENT,
-        eventData: {
-          output: await encryptSerializedData(
-            result.completed.result,
-            encryptionKey
-          ),
+      await world.events.create(
+        runId,
+        {
+          eventType: 'run_completed',
+          specVersion: SPEC_VERSION_CURRENT,
+          eventData: {
+            output: await encryptSerializedData(
+              result.completed.result,
+              encryptionKey
+            ),
+          },
         },
-      });
+        // Fenced: a completion derived from a stale view must not land —
+        // the 412 propagates to runtime.ts, which restarts the replay
+        // over the corrected log (same as the node engine's guarded
+        // run_completed).
+        preconditionSnapshot()
+      );
       wfdiag('exit_completed', { result: 'run_completed_written' });
     } catch (err) {
       if (EntityConflictError.is(err) || RunExpiredError.is(err)) {
@@ -1645,6 +1780,7 @@ export async function runWorkflowWithQuickJS(params: {
           namespace,
           nextTraceCarrier,
           pendingOperations: result.failed.drainOperations,
+          precondition: preconditionSnapshot(),
           wfdiag,
         });
       } catch (err) {
@@ -1769,6 +1905,9 @@ export async function runWorkflowWithQuickJS(params: {
       }
     }
     try {
+      // Deliberately UNFENCED (no precondition snapshot), matching the
+      // node engine's asymmetry: a terminal run_failed must be recordable
+      // even from a stale view, or a failing run could never terminate.
       await world.events.create(runId, {
         eventType: 'run_failed',
         specVersion: SPEC_VERSION_CURRENT,
diff --git a/packages/core/src/runtime/quickjs-runtime.test.ts b/packages/core/src/runtime/quickjs-runtime.test.ts
index a5232d1e0f..36da372a60 100644
--- a/packages/core/src/runtime/quickjs-runtime.test.ts
+++ b/packages/core/src/runtime/quickjs-runtime.test.ts
@@ -5,6 +5,7 @@ import {
   __peekBaselineEntryForTests,
   BASELINE_BUNDLE_FILENAME,
   runQuickJSWorkflow,
+  startQuickJSWorkflow,
 } from './quickjs-runtime.js';
 
 /** Helper to deserialize the format-prefixed result bytes */
@@ -1499,3 +1500,179 @@ describe('baseline snapshot startup optimization', () => {
     __clearBaselineSnapshotCacheForTests();
   });
 });
+
+describe('replay-divergence arbitration', () => {
+  // Parity with the node:vm engine's EventsConsumer: a log the replay does
+  // not reproduce must escalate as ReplayDivergenceError (which runtime.ts
+  // turns into bounded recovery replays and, past the budget, a terminal
+  // CORRUPTED_EVENT_LOG) — never be absorbed into a wrong completion,
+  // a wrong-payload delivery, or a spurious user error.
+
+  const stepWorkflow = `
+    var add = globalThis[Symbol.for("WORKFLOW_USE_STEP")]("step//test//add");
+    async function workflow() { return await add(10, 7); }
+    workflow.workflowId = "workflow//test//workflow";
+    globalThis.__private_workflows.set("workflow//test//workflow", workflow);
+  `;
+
+  it('rejects a replay whose log contains a correlation id it never drew', async () => {
+    const run = makeRun();
+    await expect(
+      runQuickJSWorkflow({
+        workflowCode: stepWorkflow,
+        workflowId: 'workflow//test//workflow',
+        workflowRun: run,
+        events: [
+          runCreatedEvent(run),
+          {
+            eventId: 'evnt_orphan',
+            runId: run.runId,
+            eventType: 'step_created' as const,
+            correlationId: 'step_01JUNKJUNKJUNKJUNKJUNKJUNK',
+            eventData: { stepName: 'step//test//other' },
+            createdAt: new Date('2025-01-01T00:00:01Z'),
+          },
+        ],
+      })
+    ).rejects.toMatchObject({
+      name: 'ReplayDivergenceError',
+      eventId: 'evnt_orphan',
+    });
+  });
+
+  it('rejects a replay when a recorded step belongs to a different step function', async () => {
+    const run = makeRun();
+    // Learn the deterministic correlation id the workflow draws.
+    const first = await runQuickJSWorkflow({
+      workflowCode: stepWorkflow,
+      workflowId: 'workflow//test//workflow',
+      workflowRun: run,
+      events: [],
+    });
+    const stepCid = first.suspended!.pendingOperations[0].correlationId;
+
+    // Same ordinal, different step function — the racing-writer shape that
+    // used to resolve the wrong call with the wrong payload silently.
+    await expect(
+      runQuickJSWorkflow({
+        workflowCode: stepWorkflow,
+        workflowId: 'workflow//test//workflow',
+        workflowRun: run,
+        events: [
+          runCreatedEvent(run),
+          {
+            eventId: 'evnt_wrong_step',
+            runId: run.runId,
+            eventType: 'step_created' as const,
+            correlationId: stepCid,
+            eventData: { stepName: 'step//test//DIFFERENT' },
+            createdAt: new Date('2025-01-01T00:00:01Z'),
+          },
+          {
+            eventId: 'evnt_wrong_step_done',
+            runId: run.runId,
+            eventType: 'step_completed' as const,
+            correlationId: stepCid,
+            eventData: { result: 999 },
+            createdAt: new Date('2025-01-01T00:00:02Z'),
+          },
+        ],
+      })
+    ).rejects.toMatchObject({ name: 'ReplayDivergenceError' });
+  });
+
+  it('accepts a healthy replay of a fully reproduced log', async () => {
+    const run = makeRun();
+    const first = await runQuickJSWorkflow({
+      workflowCode: stepWorkflow,
+      workflowId: 'workflow//test//workflow',
+      workflowRun: run,
+      events: [],
+    });
+    const stepCid = first.suspended!.pendingOperations[0].correlationId;
+
+    const replayed = await runQuickJSWorkflow({
+      workflowCode: stepWorkflow,
+      workflowId: 'workflow//test//workflow',
+      workflowRun: run,
+      events: [
+        runCreatedEvent(run),
+        {
+          eventId: 'evnt_step_created',
+          runId: run.runId,
+          eventType: 'step_created' as const,
+          correlationId: stepCid,
+          eventData: { stepName: 'step//test//add' },
+          createdAt: new Date('2025-01-01T00:00:01Z'),
+        },
+        {
+          eventId: 'evnt_step_done',
+          runId: run.runId,
+          eventType: 'step_completed' as const,
+          correlationId: stepCid,
+          eventData: { result: 17 },
+          createdAt: new Date('2025-01-01T00:00:02Z'),
+        },
+      ],
+    });
+    expect(replayed.completed).toBeDefined();
+    expect(unwrapResult(replayed.completed!.result)).toBe(17);
+  });
+
+  it('records a genuine user failure instead of arbitrating the log', async () => {
+    // The workflow throws before ever drawing the orphaned id. A genuine
+    // user failure must be recorded as such — divergence detection only
+    // arbitrates logs the replay claims to have reproduced.
+    const run = makeRun();
+    const throwingWorkflow = `
+      async function workflow() { throw new Error("user boom"); }
+      workflow.workflowId = "workflow//test//workflow";
+      globalThis.__private_workflows.set("workflow//test//workflow", workflow);
+    `;
+    const result = await runQuickJSWorkflow({
+      workflowCode: throwingWorkflow,
+      workflowId: 'workflow//test//workflow',
+      workflowRun: run,
+      events: [
+        runCreatedEvent(run),
+        {
+          eventId: 'evnt_orphan',
+          runId: run.runId,
+          eventType: 'step_created' as const,
+          correlationId: 'step_01JUNKJUNKJUNKJUNKJUNKJUNK',
+          eventData: { stepName: 'step//test//other' },
+          createdAt: new Date('2025-01-01T00:00:01Z'),
+        },
+      ],
+    });
+    expect(result.failed?.message).toBe('user boom');
+  });
+
+  it('rejects a live continuation fed events the session cannot reproduce', async () => {
+    const run = makeRun();
+    const session = await startQuickJSWorkflow({
+      workflowCode: stepWorkflow,
+      workflowId: 'workflow//test//workflow',
+      workflowRun: run,
+      events: [runCreatedEvent(run)],
+    });
+    expect(session.result.suspended).toBeDefined();
+    const stepCid =
+      session.result.suspended!.pendingOperations[0].correlationId;
+
+    await expect(
+      session.continueWithEvents([
+        {
+          eventId: 'evnt_wrong_step',
+          runId: run.runId,
+          eventType: 'step_created' as const,
+          correlationId: stepCid,
+          eventData: { stepName: 'step//test//DIFFERENT' },
+          createdAt: new Date('2025-01-01T00:00:01Z'),
+        } as any,
+      ])
+    ).rejects.toMatchObject({ name: 'ReplayDivergenceError' });
+    // The session disposed itself on divergence; dispose() must be a no-op.
+    session.dispose();
+  });
+});
diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts
index cf4f02a9c2..9468d35fb1 100644
--- a/packages/core/src/runtime/quickjs-runtime.ts
+++ b/packages/core/src/runtime/quickjs-runtime.ts
@@ -29,6 +29,7 @@
  * `node:vm` engine's replay determinism.
  */
 
+import type { ReplayDivergenceError } from '@workflow/errors';
 import type {
   Event,
   RunInput,
@@ -54,6 +55,10 @@ import {
   isQuickJSBaselineSnapshotEnabled,
 } from './constants.js';
 import { quickjsExtensions, quickjsWasm } from './quickjs-assets.generated.js';
+import {
+  findReplayDivergence,
+  type VmReplayView,
+} from './quickjs-divergence.js';
 import {
   adoptSerdeRoot,
   captureSerdeRoot,
@@ -1696,12 +1701,30 @@ export async function startQuickJSWorkflow(
       }
     }
 
+    // ---- Replay-divergence arbitration ----
+    // The replay is at a fixed point: every deliverable event has been
+    // delivered and the VM's job queue is drained. Arbitrate the log
+    // against the VM's draws before evaluating the workflow state — a
+    // divergence must escalate through runtime.ts's recovery machinery
+    // (bounded recovery replays → CorruptedEventLogError), never be
+    // absorbed into a wrong completion/suspension. Parity with the
+    // node:vm engine's EventsConsumer checks.
+    const observedEvents: Event[] = [...events];
+    {
+      const divergence = sweepForReplayDivergence(vm, observedEvents);
+      if (divergence) {
+        vm.dispose();
+        throw divergence;
+      }
+    }
+
     // ---- Check result ----
     return makeLiveSession(
       vm,
       serde,
       interruptBudget,
       advanceClock,
+      observedEvents,
       options.encryptionKey
     );
   }
@@ -1732,6 +1755,7 @@ function makeLiveSession(
   serde: QuickJSSerde,
   interruptBudget: InterruptBudget,
   advanceClock: (ms: number) => void,
+  observedEvents: Event[],
   encryptionKey?: DecryptionKey
 ): QuickJSWorkflowSession {
   const result = checkWorkflowState(vm, serde, { keepAliveOnSuspend: true });
@@ -1768,6 +1792,19 @@ function makeLiveSession(
         } while (batch > 0);
       } while (madeProgress && --maxIterations > 0);
 
+      // Fixed point for this burst — arbitrate the full observed log
+      // (initial replay + every fed delta) against the VM's draws before
+      // evaluating state. See the sweep in startQuickJSWorkflow.
+      observedEvents.push(...newEvents);
+      {
+        const divergence = sweepForReplayDivergence(vm, observedEvents);
+        if (divergence) {
+          alive = false;
+          vm.dispose();
+          throw divergence;
+        }
+      }
+
       const next = checkWorkflowState(vm, serde, {
         keepAliveOnSuspend: true,
       });
@@ -2406,6 +2443,49 @@ function markCreated(vm: QuickJS, cidJs: string, opType?: string): void {
   ).dispose();
 }
 
+// ---- Replay-Divergence Arbitration ----
+
+/**
+ * Arbitrate the observed event log against the VM's fixed-point view of its
+ * own draws. Returns the divergence to escalate, or `null` when the replay
+ * reproduced the log (or when the workflow already failed — a genuine user
+ * failure is recorded as such; divergence detection only arbitrates logs the
+ * replay claims to have reproduced).
+ *
+ * Must only be called at a fixed point (event drain loop converged, VM job
+ * queue empty): that is what makes an unclaimed event evidence of divergence
+ * rather than of work still in flight. The node:vm engine needs a grace
+ * window and a delivery-idle gate to reach the same certainty; here the
+ * host drains the VM's microtask queue synchronously, so the fixed point is
+ * exact.
+ */
+function sweepForReplayDivergence(
+  vm: QuickJS,
+  observedEvents: readonly Event[]
+): ReplayDivergenceError | null {
+  {
+    using failed = vm.evalCode('globalThis.__workflowError !== undefined');
+    if (vm.dump(failed)) return null;
+  }
+  using viewHandle = vm.evalCode(`(function(){
+    return {
+      ops: globalThis.__pending.map(function(p){
+        return {
+          correlationId: p.correlationId,
+          type: p.type,
+          stepId: p.stepId,
+          token: p.token,
+          resumeAt: p.resumeAt,
+        };
+      }),
+      hookCids: globalThis.__hooks ? Object.keys(globalThis.__hooks) : [],
+      abortCids: globalThis.__abortSignals ? Object.keys(globalThis.__abortSignals) : [],
+    };
+  })()`);
+  const view = vm.dump(viewHandle) as VmReplayView;
+  return findReplayDivergence(observedEvents, view);
+}
+
 // ---- State Checking ----
 
 /**

From 5bed21a520d6f8af750c163b501235aaf4b85972 Mon Sep 17 00:00:00 2001
From: Nathan Rajlich <n@n8.io>
Date: Tue, 11 Aug 2026 01:39:59 -0700
Subject: [PATCH 2/2] Address review: arbitrate cid-less tracked events, carry
 resumeAt on quickjs wait_completed

---
 .../src/runtime/quickjs-divergence.test.ts    | 21 ++++++++-
 .../core/src/runtime/quickjs-divergence.ts    | 17 ++++++--
 .../quickjs-entrypoint.fencing.test.ts        | 43 +++++++++++++++++++
 .../core/src/runtime/quickjs-entrypoint.ts    | 12 ++++++
 4 files changed, 88 insertions(+), 5 deletions(-)

diff --git a/packages/core/src/runtime/quickjs-divergence.test.ts b/packages/core/src/runtime/quickjs-divergence.test.ts
index 20b6ad7c79..e86448a892 100644
--- a/packages/core/src/runtime/quickjs-divergence.test.ts
+++ b/packages/core/src/runtime/quickjs-divergence.test.ts
@@ -42,13 +42,30 @@ describe('findReplayDivergence', () => {
       expect(findReplayDivergence(events, view([]))).toBeNull();
     });
 
-    it('skips events without a correlationId', () => {
+    it('skips untracked events without a correlationId', () => {
       const events = [
-        makeEvent({ eventType: 'step_created', correlationId: undefined }),
+        makeEvent({ eventType: 'some_future_event', correlationId: undefined }),
       ];
       expect(findReplayDivergence(events, view([]))).toBeNull();
     });
 
+    it('reports a tracked-family event missing its correlationId as divergence', () => {
+      // A step/wait/hook event without a correlation id is a malformed log
+      // entry — no replay can ever claim it, so skipping it would declare a
+      // log reproduced that was not.
+      const events = [
+        makeEvent({
+          eventType: 'step_created',
+          correlationId: undefined,
+          eventId: 'evnt_NOCID',
+        }),
+      ];
+      const err = findReplayDivergence(events, view([]));
+      expect(err).not.toBeNull();
+      expect(err?.message).toContain('missing a correlationId');
+      expect(err?.eventId).toBe('evnt_NOCID');
+    });
+
     it('skips attr_set events written by a step', () => {
       const events = [
         makeEvent({
diff --git a/packages/core/src/runtime/quickjs-divergence.ts b/packages/core/src/runtime/quickjs-divergence.ts
index a459d4f500..fdf4c0d1ef 100644
--- a/packages/core/src/runtime/quickjs-divergence.ts
+++ b/packages/core/src/runtime/quickjs-divergence.ts
@@ -75,10 +75,13 @@ function eventDataOf(event: Event): Record<string, unknown> | undefined {
  * Events that need no claim from workflow code (parity with the node
  * engine's structural lifecycle consumer in workflow.ts): run lifecycle
  * events, attribute writes not performed by the workflow body, and anything
- * without a correlation id or outside the families the engines track.
+ * outside the families the engines track. A tracked-family event is NOT
+ * structural even when its correlation id is missing — that is a malformed
+ * log entry and must be arbitrated as divergence, exactly as the node
+ * engine's consumers (which all filter on correlation-id equality) would
+ * leave it unclaimed.
  */
 function isStructural(event: Event): boolean {
-  if (!event.correlationId) return true;
   if (event.eventType.startsWith('run_')) return true;
   if (event.eventType === 'attr_set') {
     const writer = eventDataOf(event)?.writer as { type?: string } | undefined;
@@ -136,8 +139,16 @@ function arbitrateEvent(
   opsByCid: Map<string, VmKnownOp[]>,
   auxCids: Set<string>
 ): ReplayDivergenceError | null {
-  const cid = event.correlationId as string;
+  const cid = event.correlationId;
   const families = expectedFamilies(event.eventType) as string[];
+  if (!cid) {
+    // A tracked-family event without a correlation id is a malformed log
+    // entry: no replay can ever claim it (every draw has an id).
+    return new ReplayDivergenceError(
+      `Replay divergence: event ${event.eventType} (eventId=${event.eventId}) is missing a correlationId, so no replay can consume it`,
+      { eventId: event.eventId }
+    );
+  }
   const ops = opsByCid.get(cid);
 
   if (ops === undefined || ops.length === 0) {
diff --git a/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts
index c7c4672db2..297e885eb9 100644
--- a/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts
+++ b/packages/core/src/runtime/quickjs-entrypoint.fencing.test.ts
@@ -199,6 +199,49 @@ describe('QuickJS entrypoint precondition guard', () => {
     expect(waitCreated?.params).toMatchObject(expected);
   });
 
+  it('fences the elapsed-wait completion and carries the wait resumeAt', async () => {
+    // The wait elapsed before this invocation, so the pre-VM pass writes
+    // its wait_completed. The event must carry resumeAt (shape parity with
+    // the node engine's elapsed-wait completion, and the input to the
+    // replay-divergence resumeAt identity check) and the view snapshot.
+    const resumeAt = '2026-05-19T12:00:01.000Z';
+    const events = [
+      ...baseLog(),
+      makeEvent('wait_created', {
+        correlationId: 'wait_01ELAPSED',
+        eventData: { resumeAt },
+      }),
+    ];
+    // The write's snapshot describes the log BEFORE the self-written
+    // wait_completed (which the entrypoint splices into `events`) — capture
+    // the expectation up front.
+    const expectedSnapshot = {
+      stateUpdatedAt: latestEventStateUpdatedAt(events),
+      stateEventCount: events.length,
+    };
+    const { created } = await runScenario({
+      events,
+      vmResult: {
+        completed: {
+          result: await dehydrateStepReturnValue(
+            'done',
+            'wrun_quickjs_fencing',
+            undefined
+          ),
+        },
+      },
+    });
+
+    const waitCompleted = created.find(
+      (c) => c.request.eventType === 'wait_completed'
+    );
+    expect(waitCompleted).toBeDefined();
+    expect(
+      (waitCompleted?.request.eventData as { resumeAt?: Date })?.resumeAt
+    ).toEqual(new Date(resumeAt));
+    expect(waitCompleted?.params).toMatchObject(expectedSnapshot);
+  });
+
   it('leaves run_failed unfenced (terminal-failure asymmetry)', async () => {
     const { created } = await runScenario({
       events: baseLog(),
diff --git a/packages/core/src/runtime/quickjs-entrypoint.ts b/packages/core/src/runtime/quickjs-entrypoint.ts
index cf4ea07e36..7281e41804 100644
--- a/packages/core/src/runtime/quickjs-entrypoint.ts
+++ b/packages/core/src/runtime/quickjs-entrypoint.ts
@@ -865,6 +865,13 @@ export async function runWorkflowWithQuickJS(params: {
               eventType: 'wait_completed',
               specVersion: SPEC_VERSION_CURRENT,
               correlationId: event.correlationId,
+              // Carry the wait's resumeAt like the node engine's elapsed-wait
+              // completion does (runtime.ts), so the event shapes match and
+              // the replay-divergence resumeAt identity check has something
+              // to validate.
+              eventData: {
+                resumeAt: new Date(resumeAt as string | Date),
+              },
             },
             preconditionSnapshot()
           );
@@ -1177,6 +1184,11 @@ export async function runWorkflowWithQuickJS(params: {
                   eventType: 'wait_completed',
                   specVersion: SPEC_VERSION_CURRENT,
                   correlationId: wait.correlationId,
+                  // resumeAt parity with the node engine's elapsed-wait
+                  // completion — see the pre-VM pass above.
+                  eventData: {
+                    resumeAt: new Date(wait.resumeAt),
+                  },
                 },
                 preconditionSnapshot()
               );