[core] Detect when out-of-band events can change a replay's path

.changeset/quiet-hooks-observe.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Record on the invocation trace whether out-of-band events could affect each inline step boundary

packages/core/src/global.ts

@@ -89,6 +89,16 @@ export class WorkflowSuspension extends Error {
   attributeCount: number;
   hookDisposedCount: number;
   abortCount: number;
+  /**
+   * Correlation ids of the open hooks that workflow code was waiting on a
+   * payload for when the VM went idle, i.e. the hooks whose `hook_received`
+   * could change what this suspension leads to. A hook absent from the set has
+   * no waiting consumer, so a payload landing for it is buffered and inert
+   * until the first `await`. `undefined` when the suspension was built without
+   * the workflow context that tracks awaiters, which callers must read as
+   * "every open hook is observed". See `runtime/out-of-band-observation.ts`.
+   */
+  observedHookIds?: ReadonlySet<string>;
 
   constructor(itemsInput: Map<string, QueueItem>, global: typeof globalThis) {
     // Convert Map to array for iteration and storage

packages/core/src/private.ts

@@ -150,6 +150,17 @@ export interface WorkflowOrchestratorContext {
    * Using Map instead of Array for O(1) lookup/delete operations.
    */
   invocationsQueue: Map<string, QueueItem>;
+  /**
+   * Per open hook, a probe for whether workflow code is currently waiting on
+   * its next payload (an `await hook`, a pending `for await` iteration, or a
+   * `.then` on it). Registered by `createHook` and dropped once the hook can
+   * receive nothing more. Snapshotted into
+   * {@link WorkflowSuspension.observedHookIds} when a suspension is raised.
+   *
+   * Optional so older/out-of-tree contexts (and lightweight test harnesses)
+   * degrade to "every open hook is observed".
+   */
+  hookPayloadAwaiters?: Map<string, () => boolean>;
   onWorkflowError: (error: Error) => void;
   /**
    * Mints the ULID body of a correlation id. Every entity a replay creates
@@ -1022,8 +1033,19 @@ export function scheduleWorkflowSuspension(
   const generation = ctx.suspensionGeneration;
   scheduleWhenIdle(ctx, () => {
     if (generation !== ctx.suspensionGeneration) return;
-    ctx.onWorkflowError(
-      new WorkflowSuspension(ctx.invocationsQueue, ctx.globalThis)
+    const suspension = new WorkflowSuspension(
+      ctx.invocationsQueue,
+      ctx.globalThis
     );
+    // Taken at idle, the same instant the queue is: the awaiters that are
+    // pending now are exactly the ones this suspension's outcome depends on.
+    if (ctx.hookPayloadAwaiters) {
+      const observed = new Set<string>();
+      for (const [correlationId, isAwaited] of ctx.hookPayloadAwaiters) {
+        if (isAwaited()) observed.add(correlationId);
+      }
+      suspension.observedHookIds = observed;
+    }
+    ctx.onWorkflowError(suspension);
   });
 }

packages/core/src/runtime.ts

@@ -109,6 +109,10 @@ import {
   hasOpenWaitDueBy,
   openHookAndWaitState,
 } from './runtime/open-hook-wait-state.js';
+import {
+  type OutOfBandObservation,
+  observeOutOfBandWriters,
+} from './runtime/out-of-band-observation.js';
 import {
   handleReplayBudgetExhausted,
   ReplayBudget,
@@ -1107,6 +1111,19 @@ export function workflowEntrypoint(
                   // completion that already exists is acted on now rather
                   // than when the wait's own timer would have fired.
                   let eventLogFromInlineDelta = false;
+                  // Inline-step boundaries this invocation scheduled, split by
+                  // whether an out-of-band event could have changed what they
+                  // led to. Recorded on the span so the share of boundaries a
+                  // future run-ahead could pipeline is measurable. See
+                  // runtime/out-of-band-observation.ts.
+                  const outOfBandBoundaries = {
+                    inert: 0,
+                    observedHook: 0,
+                    unknownHook: 0,
+                    abortSignal: 0,
+                    externalStep: 0,
+                    waitDue: 0,
+                  };
                   let loopIteration = 0;
                   // Hooks whose force-claim victim wake this invocation has
                   // already sent (its own forced creations, and the replay's
@@ -4852,6 +4869,71 @@ export function workflowEntrypoint(
                           });
                         }
 
+                        // Whether any event an out-of-band writer could append
+                        // now is able to change the decisions this batch leads
+                        // to. Nothing branches on it yet: it is the gate a
+                        // run-ahead of the inline batch must consult before
+                        // continuing past this boundary without the batch's
+                        // terminal writes, and the module documents the de-opt
+                        // and pipeline-depth rules that gate implies.
+                        // `undefined` when the batch runs no step inline:
+                        // only those boundaries can be pipelined.
+                        const outOfBandObservation:
+                          | OutOfBandObservation
+                          | undefined =
+                          lazyInlineSteps.length > 0
+                            ? observeOutOfBandWriters({
+                                items: err.items,
+                                observedHookIds: err.observedHookIds,
+                                selfExecutedStepIds: new Set([
+                                  ...inlineCorrelationIds,
+                                  ...ownedRecoverySteps.map(
+                                    (s) => s.correlationId
+                                  ),
+                                ]),
+                                waitDue: waitDueThisInvocation,
+                              })
+                            : undefined;
+                        if (outOfBandObservation) {
+                          if (outOfBandObservation.inert) {
+                            outOfBandBoundaries.inert++;
+                          } else {
+                            if (outOfBandObservation.observedHookCount > 0) {
+                              outOfBandBoundaries.observedHook++;
+                            }
+                            // Kept apart from observed hooks: a suspension
+                            // without awaiter tracking (QuickJS) blocks every
+                            // boundary with an open hook, which would
+                            // otherwise read as workflow code awaiting it.
+                            if (outOfBandObservation.unknownHookCount > 0) {
+                              outOfBandBoundaries.unknownHook++;
+                            }
+                            if (outOfBandObservation.abortSignalHookCount > 0) {
+                              outOfBandBoundaries.abortSignal++;
+                            }
+                            if (outOfBandObservation.externalStepCount > 0) {
+                              outOfBandBoundaries.externalStep++;
+                            }
+                            if (outOfBandObservation.waitDue) {
+                              outOfBandBoundaries.waitDue++;
+                            }
+                          }
+                          span?.setAttributes({
+                            'workflow.out_of_band.inert_boundaries':
+                              outOfBandBoundaries.inert,
+                            'workflow.out_of_band.observed_hook_boundaries':
+                              outOfBandBoundaries.observedHook,
+                            'workflow.out_of_band.unknown_hook_boundaries':
+                              outOfBandBoundaries.unknownHook,
+                            'workflow.out_of_band.abort_signal_boundaries':
+                              outOfBandBoundaries.abortSignal,
+                            'workflow.out_of_band.external_step_boundaries':
+                              outOfBandBoundaries.externalStep,
+                            'workflow.out_of_band.wait_due_boundaries':
+                              outOfBandBoundaries.waitDue,
+                          });
+                        }
+
                         // Stale-sensitive batch: a hook is open in the run (or
                         // was created by this suspension, so its hook_received
                         // can land any moment), so an out-of-band event can make