fix(core): premature suspension when hooks have buffered payloads with concurrent pending entities (#1294)

* test: reproduce hook+sleep promiseQueue regression

Add failing tests that reproduce a regression from #1246 where the
sleep's WorkflowSuspension fires before all hook payloads are delivered
when a hook and sleep run concurrently.

Root cause: when the null event fires, the sleep queues a suspension
through promiseQueue. After the first hook payload resolves, subsequent
hook payload resolutions are queued AFTER the sleep suspension, causing
the workflow to terminate prematurely via Promise.race.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test: expand tests to isolate bug to hook + pending entity pattern

Add control tests proving sequential steps are NOT affected by the
promiseQueue regression, isolating the bug to hooks specifically:

Failing (hook-based):
- hook + sleep: all 3 payloads → step invocation
- hook + sleep: 2 payloads → return
- hook + incomplete step: 2 payloads → return

Passing (step-based controls):
- sleep + sequential steps: both step events exist
- sleep + sequential steps: only 1st step completed
- incomplete step + sequential steps: all step events exist
- hook only (no concurrent entity): payloads + step

The bug is: any entity that queues a suspension through promiseQueue
at null-event time preempts hook payload delivery, because hooks
buffer payloads in payloadsQueue and only resolve them one-at-a-time
as the workflow code iterates. Steps are unaffected because each step
has its own events consumed before null fires.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* fix(core): move WorkflowSuspension calls back to setTimeout(0) from promiseQueue

The promiseQueue refactor (#1246) moved ALL promise resolutions through
the queue, including WorkflowSuspension calls. This caused premature
termination when hooks had buffered payloads and a concurrent entity
(sleep or incomplete step) was pending:

1. Null event fires → all subscribers run
2. Sleep/step queues WorkflowSuspension via promiseQueue.then()
3. Hook's next payload resolution is also queued via promiseQueue.then()
4. Suspension fires first → Promise.race terminates workflow
5. Hook payload never delivered → infinite retry loop

Fix: move WorkflowSuspension calls back to setTimeout(0) (macrotask).
Suspensions fire AFTER all microtask-based deliveries (promiseQueue
resolve/reject for step results and hook payloads) have completed.
Non-suspension resolve/reject calls remain on promiseQueue for
deterministic ordering of data delivery.

Affected paths:
- step.ts: null event handler
- sleep.ts: null event handler
- hook.ts: null event handler, eventLogEmpty suspension, dispose suspension

* fix(core): use pendingDeliveries counter + scheduleWhenIdle for suspensions

Replace all nested setTimeout/promiseQueue suspension patterns with a
clean idle-polling mechanism:

1. pendingDeliveries counter: incremented before async hydration (step
   results, hook payloads), decremented in finally block after delivery
2. scheduleWhenIdle(ctx, fn): polls via setTimeout(0) → check counter →
   if > 0, wait for promiseQueue.then() → repeat. Only fires fn when
   pendingDeliveries reaches 0.

This correctly handles:
- Sync deserialization (no encryption): counter is 0, fires immediately
  after first setTimeout(0)
- Async deserialization (with encryption): waits for decryption to
  complete before firing
- Multi-round hook payload delivery: each createHookPromise() call
  increments the counter, preventing premature suspension between
  delivery rounds

Also adds payloadsQueue.length check to hook null handler — don't
trigger suspension if buffered payloads can satisfy pending awaits.

Tests now run in both sync and async modes (14 total = 7 scenarios x 2).
All 164 tests pass.

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Nathan Rajlich <n@n8.io>

Author: 3 people

.changeset/fix-hook-sleep-suspension.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Fix premature workflow suspension when hooks have buffered payloads and a concurrent sleep or incomplete step is pending

packages/core/e2e/e2e.test.ts

@@ -1968,4 +1968,76 @@ describe('e2e', () => {
       expect(returnValue.endTime - returnValue.startTime).toBeGreaterThan(9999);
     });
   });
+
+  test(
+    'hookWithSleepWorkflow - hook payloads delivered correctly with concurrent sleep',
+    { timeout: 90_000 },
+    async () => {
+      // Regression test: when a hook and sleep run concurrently, multiple
+      // hook_received events should all be processed even though the sleep
+      // has no wait_completed event. Previously, the sleep's WorkflowSuspension
+      // would terminate the workflow before all hook payloads were delivered.
+      const token = Math.random().toString(36).slice(2);
+
+      const run = await start(await e2e('hookWithSleepWorkflow'), [token]);
+
+      // Wait for the hook to be registered
+      await new Promise((resolve) => setTimeout(resolve, 5_000));
+
+      // Send 3 payloads: two normal ones, then one with done=true
+      let hook = await getHookByToken(token);
+      expect(hook.runId).toBe(run.runId);
+      await resumeHook(hook, { type: 'subscribe', id: 1 });
+
+      // Wait for the first payload to be processed (step must complete)
+      await new Promise((resolve) => setTimeout(resolve, 3_000));
+
+      hook = await getHookByToken(token);
+      await resumeHook(hook, { type: 'subscribe', id: 2 });
+
+      await new Promise((resolve) => setTimeout(resolve, 3_000));
+
+      hook = await getHookByToken(token);
+      await resumeHook(hook, { type: 'done', done: true });
+
+      const returnValue = await run.returnValue;
+      expect(returnValue).toBeInstanceOf(Array);
+      expect(returnValue).toHaveLength(3);
+      expect(returnValue[0]).toMatchObject({
+        processed: true,
+        type: 'subscribe',
+        id: 1,
+      });
+      expect(returnValue[1]).toMatchObject({
+        processed: true,
+        type: 'subscribe',
+        id: 2,
+      });
+      expect(returnValue[2]).toMatchObject({ processed: true, type: 'done' });
+
+      const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+      expect(runData.status).toBe('completed');
+    }
+  );
+
+  test(
+    'sleepWithSequentialStepsWorkflow - sequential steps work with concurrent sleep (control)',
+    { timeout: 60_000 },
+    async () => {
+      // Control test: proves that void sleep('1d').then() does NOT break
+      // sequential step execution. Steps have per-event consumption so the
+      // sleep's pending suspension doesn't interfere. This contrasts with
+      // hookWithSleepWorkflow where the bug manifests.
+      const run = await start(
+        await e2e('sleepWithSequentialStepsWorkflow'),
+        []
+      );
+
+      const returnValue = await run.returnValue;
+      expect(returnValue).toEqual({ a: 3, b: 6, c: 10, shouldCancel: false });
+
+      const { json: runData } = await cliInspectJson(`runs ${run.runId}`);
+      expect(runData.status).toBe('completed');
+    }
+  );
 });

packages/core/src/async-deserialization-ordering.test.ts

@@ -43,6 +43,7 @@ function setupWorkflowContext(events: Event[]): WorkflowOrchestratorContext {
     ),
     onWorkflowError: vi.fn(),
     promiseQueue: Promise.resolve(),
+    pendingDeliveries: 0,
   };
 }