[core] Complete a due wait even when its run is already terminal

.changeset/complete-due-waits-on-terminal-runs.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Record a wait's `wait_completed` when its queue message comes due even if the run has already finished, instead of acknowledging the delivery with the wait left open in the log

.changeset/wait-completed-after-terminal-reap-local.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-local': patch
+---
+
+Accept `wait_completed` for a wait whose entity the run's terminal transition reaped, recording the event without resurrecting the entity

.changeset/wait-completed-after-terminal-reap-postgres.md

@@ -0,0 +1,5 @@
+---
+'@workflow/world-postgres': patch
+---
+
+Accept `wait_completed` for a wait whose row the run's terminal transition deleted, recording the event without recreating the row and deduplicating against the event log

docs/content/docs/v5/how-it-works/event-sourcing.mdx

@@ -199,6 +199,8 @@ Events are categorized by the entity type they affect. Each event contains metad
 | `wait_created` | Creates a new wait in `waiting` state. Contains the timestamp when the wait should complete. |
 | `wait_completed` | Transitions the wait to `completed` state when the delay period has elapsed. |
 
+A wait's `wait_completed` is recorded when the wait comes due, even if the run finished first — a `Promise.race([sleep('1h'), hook])` whose hook won still records the sleep's completion an hour later. Only the log entry is written in that case: the run is terminal, so nothing is replayed and the wait entity itself stays released. Without it a `wait_created` with no matching `wait_completed` would read as a run that ended while still sleeping.
+
 ## Terminal States
 
 Terminal states represent the end of an entity's lifecycle. Once an entity reaches a terminal state, no further events can transition it to another state.

packages/core/src/runtime.ts

@@ -27,6 +27,7 @@ import {
   getQueueTopicPrefix,
   isLegacySpecVersion,
   isTerminalRunEventType,
+  isTerminalWorkflowRunStatus,
   ROOT_RUN_ID_ATTRIBUTE,
   type RunInput,
   resolveQueueNamespace,
@@ -67,6 +68,7 @@ import {
   guardDeploymentAffinity,
   type ReenqueueArgs,
 } from './runtime/deployment-guard.js';
+import { completeDueWaits } from './runtime/due-waits.js';
 import {
   absorbSkippedSlotReport,
   appendUniqueEvents,
@@ -2070,6 +2072,17 @@ export function workflowEntrypoint(
                               eventType: terminalEvent.eventType,
                             }
                           );
+                          // The preload is the complete log (that is what
+                          // `usableReplayPreload` attests), so a due wait is
+                          // visible right here — record its completion before
+                          // consuming the delivery. See runtime/due-waits.ts.
+                          await completeDueWaits({
+                            world,
+                            runId,
+                            events: result.events,
+                            specVersion: result.run.specVersion,
+                            requestId,
+                          });
                           return;
                         }
                         workflowRun = result.run;
@@ -2300,6 +2313,22 @@ export function workflowEntrypoint(
                           // so that we actually exit here without replaying the workflow at all, in the case
                           // the replaying the workflow is itself failing.
 
+                          // Terminal: no replay, no suspension dispatch — but
+                          // a wait this delivery may have been scheduled for
+                          // still gets its completion recorded before the
+                          // message is acknowledged. See runtime/due-waits.ts.
+                          if (isTerminalWorkflowRunStatus(result.run.status)) {
+                            await completeDueWaits({
+                              world,
+                              runId,
+                              events:
+                                eventLog.type === 'ready'
+                                  ? eventLog.events
+                                  : undefined,
+                              specVersion: result.run.specVersion,
+                              requestId,
+                            });
+                          }
                           return;
                         }
                       } catch (err) {
@@ -2316,6 +2345,17 @@ export function workflowEntrypoint(
                             'Run already finished during setup, skipping',
                             { workflowRunId: runId, message: err.message }
                           );
+                          // Same as the non-running branch above: the run is
+                          // done, but a due wait's completion is this
+                          // delivery's job and is not skipped with the rest.
+                          // The log has to be read here — the rejection
+                          // carried no preload.
+                          await completeDueWaits({
+                            world,
+                            runId,
+                            specVersion: workflowRun?.specVersion,
+                            requestId,
+                          });
                           return;
                         } else {
                           const errorCode = getWorkflowSetupErrorCode(err);
@@ -2735,7 +2775,20 @@ export function workflowEntrypoint(
                       // derived from these events, so checking the log here
                       // gives us the same signal as a runs.get() round-trip
                       // without the extra request per loop iteration.
+                      //
+                      // A wait that has come due is still completed first: the
+                      // delivery may BE that wait's continuation, and the
+                      // completion belongs in the log whether or not there is
+                      // anything left to replay (see runtime/due-waits.ts).
+                      // Only then is the message acknowledged.
                       if (hasRecordedTerminalRunEvent(eventLog.events, runId)) {
+                        await completeDueWaits({
+                          world,
+                          runId,
+                          events: eventLog.events,
+                          specVersion: workflowRun?.specVersion,
+                          requestId,
+                        });
                         return;
                       }
 

packages/core/src/runtime/due-waits.test.ts

@@ -0,0 +1,272 @@
+import { EntityConflictError, WorkflowWorldError } from '@workflow/errors';
+import {
+  type CreateEventRequest,
+  type Event,
+  SPEC_VERSION_CURRENT,
+  SPEC_VERSION_LEGACY,
+  slotToEventId,
+  type World,
+} from '@workflow/world';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { completeDueWaits, findDueWaits } from './due-waits.js';
+import { setWorld } from './world.js';
+
+const runId = 'wrun_due_waits';
+const now = new Date('2026-05-19T12:00:00.000Z');
+
+let eventIndex = 0;
+function event(data: CreateEventRequest): Event {
+  eventIndex += 1;
+  return {
+    ...data,
+    specVersion: data.specVersion ?? SPEC_VERSION_CURRENT,
+    runId,
+    eventId: slotToEventId(eventIndex),
+    createdAt: now,
+  } as Event;
+}
+
+function waitCreated(correlationId: string, resumeAt: Date): Event {
+  return event({
+    eventType: 'wait_created',
+    specVersion: SPEC_VERSION_CURRENT,
+    correlationId,
+    eventData: { resumeAt },
+  });
+}
+
+function waitCompleted(correlationId: string): Event {
+  return event({
+    eventType: 'wait_completed',
+    specVersion: SPEC_VERSION_CURRENT,
+    correlationId,
+  });
+}
+
+/**
+ * A World that records `events.create` calls, optionally rejecting
+ * `wait_completed` writes with `reject`.
+ */
+function fakeWorld(reject?: (correlationId: string) => unknown) {
+  const create = vi.fn(async (_runId: string, data: CreateEventRequest) => {
+    if (data.eventType === 'wait_completed' && reject) {
+      const err = reject(data.correlationId);
+      if (err) throw err;
+    }
+    return { event: event(data) };
+  });
+  const list = vi.fn();
+  const world = {
+    specVersion: SPEC_VERSION_CURRENT,
+    events: { create, list },
+  } as unknown as World;
+  return { world, create, list };
+}
+
+describe('findDueWaits', () => {
+  it('returns waits whose resumeAt has passed and that have no completion', () => {
+    const due = findDueWaits(
+      [
+        waitCreated('wait_past', new Date(+now - 1_000)),
+        waitCreated('wait_exactly_now', now),
+        waitCreated('wait_future', new Date(+now + 1_000)),
+        waitCreated('wait_done', new Date(+now - 1_000)),
+        waitCompleted('wait_done'),
+      ],
+      +now
+    );
+
+    expect(due.map((w) => w.correlationId)).toEqual([
+      'wait_past',
+      'wait_exactly_now',
+    ]);
+    expect(due[0]?.resumeAt).toEqual(new Date(+now - 1_000));
+  });
+
+  it('accepts a resumeAt that arrived as a string', () => {
+    const created = waitCreated('wait_str', new Date(+now - 1_000));
+    (created as { eventData: { resumeAt: unknown } }).eventData.resumeAt =
+      new Date(+now - 1_000).toISOString();
+
+    expect(findDueWaits([created], +now).map((w) => w.correlationId)).toEqual([
+      'wait_str',
+    ]);
+  });
+
+  it('leaves a wait alone when its deadline cannot be read', () => {
+    const created = waitCreated('wait_bad', new Date(+now - 1_000));
+    (created as { eventData: { resumeAt: unknown } }).eventData.resumeAt =
+      'not-a-date';
+
+    expect(findDueWaits([created], +now)).toEqual([]);
+  });
+});
+
+describe('completeDueWaits', () => {
+  afterEach(() => {
+    setWorld(undefined);
+    vi.restoreAllMocks();
+  });
+
+  it('records a wait_completed carrying the wait_created resumeAt', async () => {
+    const { world, create } = fakeWorld();
+    const resumeAt = new Date(+now - 5_000);
+
+    const summary = await completeDueWaits({
+      world,
+      runId,
+      events: [waitCreated('wait_1', resumeAt)],
+      requestId: 'req_1',
+      now: +now,
+    });
+
+    expect(summary).toEqual({
+      completed: ['wait_1'],
+      alreadyCompleted: [],
+      unrecordable: [],
+    });
+    expect(create).toHaveBeenCalledTimes(1);
+    expect(create).toHaveBeenCalledWith(
+      runId,
+      {
+        eventType: 'wait_completed',
+        specVersion: SPEC_VERSION_CURRENT,
+        correlationId: 'wait_1',
+        eventData: { resumeAt },
+      },
+      { requestId: 'req_1' }
+    );
+  });
+
+  it('writes nothing when no wait is due', async () => {
+    const { world, create } = fakeWorld();
+
+    const summary = await completeDueWaits({
+      world,
+      runId,
+      events: [
+        waitCreated('wait_future', new Date(+now + 60_000)),
+        waitCreated('wait_done', new Date(+now - 60_000)),
+        waitCompleted('wait_done'),
+      ],
+      now: +now,
+    });
+
+    expect(summary.completed).toEqual([]);
+    expect(create).not.toHaveBeenCalled();
+  });
+
+  it('loads the event log when the caller has none', async () => {
+    const { world, create, list } = fakeWorld();
+    list.mockResolvedValue({
+      data: [waitCreated('wait_loaded', new Date(+now - 1_000))],
+      hasMore: false,
+      cursor: null,
+    });
+    // loadWorkflowRunEvents reads the ambient World, not the argument.
+    setWorld(world);
+
+    const summary = await completeDueWaits({ world, runId, now: +now });
+
+    expect(list).toHaveBeenCalledTimes(1);
+    expect(summary.completed).toEqual(['wait_loaded']);
+    expect(create).toHaveBeenCalledTimes(1);
+  });
+
+  it('treats a concurrent completion as done, not as a failure', async () => {
+    const { world } = fakeWorld(
+      () => new EntityConflictError('Wait "wait_1" already completed')
+    );
+
+    const summary = await completeDueWaits({
+      world,
+      runId,
+      events: [waitCreated('wait_1', new Date(+now - 1_000))],
+      now: +now,
+    });
+
+    expect(summary).toEqual({
+      completed: [],
+      alreadyCompleted: ['wait_1'],
+      unrecordable: [],
+    });
+  });
+
+  it('reports a wait an older World cannot record, without failing the delivery', async () => {
+    // A backend that drops a terminal run's waits outright has nowhere to put
+    // the completion. Redelivery would reach the same verdict, so the caller
+    // must still be free to acknowledge.
+    const { world } = fakeWorld(
+      (correlationId) =>
+        new WorkflowWorldError(`Wait "${correlationId}" not found`)
+    );
+
+    const summary = await completeDueWaits({
+      world,
+      runId,
+      events: [waitCreated('wait_1', new Date(+now - 1_000))],
+      now: +now,
+    });
+
+    expect(summary).toEqual({
+      completed: [],
+      alreadyCompleted: [],
+      unrecordable: ['wait_1'],
+    });
+  });
+
+  it('rethrows a retryable World failure so the message is redelivered', async () => {
+    const { world } = fakeWorld(
+      () => new WorkflowWorldError('upstream unavailable', { status: 503 })
+    );
+
+    await expect(
+      completeDueWaits({
+        world,
+        runId,
+        events: [waitCreated('wait_1', new Date(+now - 1_000))],
+        now: +now,
+      })
+    ).rejects.toThrow('upstream unavailable');
+  });
+
+  it('completes every due wait even when one of them cannot be recorded', async () => {
+    const { world, create } = fakeWorld((correlationId) =>
+      correlationId === 'wait_1'
+        ? new WorkflowWorldError('Wait "wait_1" not found')
+        : undefined
+    );
+
+    const summary = await completeDueWaits({
+      world,
+      runId,
+      events: [
+        waitCreated('wait_1', new Date(+now - 2_000)),
+        waitCreated('wait_2', new Date(+now - 1_000)),
+      ],
+      now: +now,
+    });
+
+    expect(summary.unrecordable).toEqual(['wait_1']);
+    expect(summary.completed).toEqual(['wait_2']);
+    expect(create).toHaveBeenCalledTimes(2);
+  });
+
+  it('uses the legacy write shape for a legacy run', async () => {
+    const { world, create } = fakeWorld();
+
+    await completeDueWaits({
+      world,
+      runId,
+      events: [waitCreated('wait_1', new Date(+now - 1_000))],
+      specVersion: SPEC_VERSION_LEGACY,
+      now: +now,
+    });
+
+    expect(create).toHaveBeenCalledWith(
+      runId,
+      { eventType: 'wait_completed', correlationId: 'wait_1' },
+      { requestId: undefined, v1Compat: true }
+    );
+  });
+});