From 74bf7e3178efdcb48b1081598bd2438d0ab1f06d Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 12:59:28 -0700 Subject: [PATCH 1/9] Add e2e coverage and a cookbook recipe for background hook subscribers and merged hook inboxes Two userland patterns an agent session needs from hooks: a `for await` over a hook that the workflow body never awaits, pushing payloads into a local inbox that each turn drains for steering; and several hooks merged into one async iterator, including a hook added to the merge after the run started (a Slack thread whose id is only known after the first reply). Both are plain JavaScript, so what the new e2e tests establish is that hook delivery stays in event-log order relative to step results: the inbox a turn drains on the live run is the inbox every replay drains at that turn, checked by echoing the drained messages back through the step's recorded arguments. Six tests cover a fixed-turn loop under paced bursts, a run that returns while its subscriber is still parked on the hook, a drain-then-wait session loop, a three-hook merge with round-robin sends (exact log order), concurrent senders with a step per message (per-hook order), and a hook added mid-run. `E2E_INBOX_SCALE` multiplies the message counts for soaks. The cookbook page (v5 only) ships the subscriber, the `mergeAsyncIterables` helper, and the `subscribeInbox` drain/wait wrapper, with a steering example and the Slack-thread shape. Co-Authored-By: Claude Fable 5.1 --- .changeset/hook-inbox-e2e.md | 5 + .../v5/cookbook/agent-patterns/hook-inbox.mdx | 314 ++++++++++++++ .../docs/v5/cookbook/agent-patterns/meta.json | 7 +- docs/content/docs/v5/cookbook/index.mdx | 1 + docs/lib/cookbook-tree.ts | 9 + packages/core/e2e/e2e.test.ts | 389 ++++++++++++++++++ workbench/example/workflows/99_e2e.ts | 323 +++++++++++++++ 7 files changed, 1047 insertions(+), 1 deletion(-) create mode 100644 .changeset/hook-inbox-e2e.md create mode 100644 docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx diff --git a/.changeset/hook-inbox-e2e.md b/.changeset/hook-inbox-e2e.md new file mode 100644 index 0000000000..7f6563efe3 --- /dev/null +++ b/.changeset/hook-inbox-e2e.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +Add e2e coverage for background hook subscribers and merged hook inboxes: a `for await` over a hook that the workflow body never awaits, several hooks merged into one async iterator (including a hook added mid-run), and a drain-then-wait session loop, each checked for event-log-ordered delivery across replays under dozens to hundreds of payloads. diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx new file mode 100644 index 0000000000..6d84bd280b --- /dev/null +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -0,0 +1,314 @@ +--- +title: Hook Inbox & Steering +description: Buffer hook payloads in the background while an agent loop runs, drain them at turn boundaries to steer the agent, and merge several hooks into one ordered inbox. +type: guide +summary: Subscribe to a hook with a `for await` loop that the workflow body never awaits, so payloads land in a local inbox as they arrive and each agent turn can drain them. Merge multiple hooks into a single async iterator, and add hooks to the merge mid-run, while the runtime keeps delivery in event-log order across replays. +related: + - /docs/foundations/hooks + - /docs/api-reference/workflow/create-hook + - /docs/api-reference/workflow-api/resume-hook + - /cookbook/agent-patterns/agent-cancellation +--- + + + +An agent loop runs one turn at a time: call the model, run tools, repeat. Messages from the user, or from other systems, keep arriving while a turn is in flight. Awaiting the hook directly would block the loop, and racing the hook against every turn would drop messages that arrive between races. Instead, subscribe to the hook in the background and let each turn drain what arrived since the last one. + +## When to use this + +- **Steering**: the user sends follow-up instructions while the agent is working, and the next turn should take them into account +- **Cancellation and interruption**: an `end` or `stop` message should be noticed at the next turn boundary without a separate hook +- **Multiple channels**: a session receives messages on more than one token, such as an identity token plus a Slack thread that only exists after the first reply + +## Pattern: background subscriber with a local inbox + +The subscriber is an async function that iterates the hook and pushes into an array. The workflow body starts it and moves on. Each turn drains the array and hands the drained messages to the step that runs the turn. + +```typescript lineNumbers +import { createHook, getWorkflowMetadata } from "workflow"; + +export type InboxMessage = + | { type: "steer"; text: string } + | { type: "end" }; + +declare function runTurn( + turn: number, + steering: string[] +): Promise<{ finished: boolean }>; // @setup + +export async function agentSession(prompt: string) { + "use workflow"; + + const { workflowRunId } = getWorkflowMetadata(); + using inboxHook = createHook({ + token: `session:${workflowRunId}`, // [!code highlight] + }); + + const inbox: InboxMessage[] = []; + let ended = false; + + // Background subscriber. Intentionally NOT awaited: it keeps running while + // the turn loop below is busy, so payloads land in `inbox` as they arrive. + const subscription = (async () => { // [!code highlight] + for await (const message of inboxHook) { // [!code highlight] + if (message.type === "end") { // [!code highlight] + ended = true; // [!code highlight] + break; // [!code highlight] + } // [!code highlight] + inbox.push(message); // [!code highlight] + } // [!code highlight] + })(); // [!code highlight] + + let steering: string[] = [prompt]; + for (let turn = 0; !ended; turn++) { + const result = await runTurn(turn, steering); + + // Drain everything that arrived during this turn; it steers the next one. + steering = inbox.splice(0).map((m) => (m.type === "steer" ? m.text : "")); // [!code highlight] + + if (result.finished && steering.length === 0) { + break; + } + } + + // `using` disposes the hook here, releasing the token. The subscriber's + // pending `for await` is abandoned with it; that is fine, the run completes. + return { turnsWithSteering: steering.length }; +} +``` + +### Step that runs a turn + +The step receives the drained messages as an argument, so they are recorded in the event log with the turn that used them. + +```typescript lineNumbers +export async function runTurn(turn: number, steering: string[]) { + "use step"; + // Call the model with the conversation so far plus `steering` as new user + // messages, run any tool calls, and report whether the agent is done. + return { finished: steering.length === 0 && turn > 0 }; +} +``` + +### API route to send a message + +Any process can steer the session, since the token is derived from the run ID. + +```typescript lineNumbers +import { resumeHook } from "workflow/api"; + +export async function POST( + request: Request, + { params }: { params: Promise<{ runId: string }> } +) { + const { runId } = await params; + const message = await request.json(); + await resumeHook(`session:${runId}`, message); // [!code highlight] + return Response.json({ ok: true }); +} +``` + +## Pattern: merging several hooks into one inbox + +A session may listen on more than one token: its own identity token, a Slack thread, an auth callback. Merging their async iterators gives the workflow one ordered stream to consume. The helper below is plain JavaScript and runs unchanged inside a workflow. Its `add()` lets you merge in a hook after consumption has started. + +```typescript lineNumbers +/** One item from the merged stream: the value plus which source produced it. */ +export type MergedItem = { index: number; value: T }; + +/** + * Merges async iterables into one, yielding each item as soon as its source + * produces it. `add()` merges in another source while the result is being + * consumed. + */ +export function mergeAsyncIterables(initial: AsyncIterable[] = []) { + type Slot = { index: number; result: IteratorResult }; + const iterators = new Map>(); + const pending = new Map>(); + let nextIndex = 0; + // Resolved whenever a source is added, so a consumer blocked in + // `Promise.race` over the current sources re-races with the new one. + let wake!: () => void; + let woken = new Promise((resolve) => { + wake = () => resolve(null); + }); + + const pull = (index: number) => { + const iterator = iterators.get(index); + if (!iterator) return; + pending.set( + index, + iterator.next().then((result) => ({ index, result })) + ); + }; + + const add = (source: AsyncIterable): number => { + const index = nextIndex++; + iterators.set(index, source[Symbol.asyncIterator]()); + pull(index); + const previousWake = wake; + woken = new Promise((resolve) => { + wake = () => resolve(null); + }); + previousWake(); + return index; + }; + + for (const source of initial) add(source); + + const merged: AsyncIterable> & { + add: (source: AsyncIterable) => number; + } = { + add, + async *[Symbol.asyncIterator]() { + try { + while (pending.size > 0) { + const winner = await Promise.race([...pending.values(), woken]); + if (winner === null) continue; // a source was added: re-race + const { index, result } = winner; + if (result.done) { + pending.delete(index); + iterators.delete(index); + continue; + } + pull(index); + yield { index, value: result.value }; + } + } finally { + // A disposed hook never settles its pending `next()`, so this is + // fire-and-forget cleanup. + for (const iterator of iterators.values()) { + void iterator.return?.().catch(() => {}); + } + } + }, + }; + return merged; +} +``` + +### Wrapping the subscriber as an inbox + +For a session that runs until told to stop, the loop needs to wait for the next message without spinning. Wrap the subscriber so it exposes `drain()` and `wait()`. While nothing is buffered, the only pending promise in the run is the subscriber's `await hook`, so `await inbox.wait()` suspends the workflow durably until a payload arrives. + +```typescript lineNumbers +/** + * Subscribes to `source` in the background and buffers what it yields. + * `wait()` resolves once a message has been buffered or the source ended. + */ +export function subscribeInbox( + source: AsyncIterable, + isEnd: (value: T) => boolean +) { + const buffered: T[] = []; + let ended = false; + let notify = () => {}; + let changed = new Promise((resolve) => { + notify = resolve; + }); + const bump = () => { + const previous = notify; + changed = new Promise((resolve) => { + notify = resolve; + }); + previous(); + }; + const done = (async () => { + for await (const value of source) { + if (isEnd(value)) { + ended = true; + bump(); + break; + } + buffered.push(value); + bump(); + } + })(); + return { + done, + drain: () => buffered.splice(0), + get ended() { + return ended; + }, + async wait() { + if (buffered.length === 0 && !ended) await changed; + }, + }; +} +``` + +### Adding a hook mid-session + +A Slack bot only learns the thread's id after it posts the first reply. Start with the identity hook, subscribe to the merged inbox, and add the thread's hook once the id is known. Messages sent to either token land in the same inbox in the order they were received, and each turn drains whatever arrived since the last one. + +```typescript lineNumbers +import { createHook } from "workflow"; + +type InboxMessage = { text: string; end?: boolean }; + +declare function mergeAsyncIterables( + initial?: AsyncIterable[] +): AsyncIterable<{ index: number; value: T }> & { + add: (source: AsyncIterable) => number; +}; // @setup +declare function subscribeInbox( + source: AsyncIterable, + isEnd: (value: T) => boolean +): { + done: Promise; + drain: () => T[]; + readonly ended: boolean; + wait: () => Promise; +}; // @setup +declare function postFirstReply(channel: string): Promise; // @setup +declare function runTurn(steering: string[]): Promise; // @setup + +export async function slackSession(sessionId: string, channel: string) { + "use workflow"; + + const identity = createHook({ token: `session:${sessionId}` }); + const merged = mergeAsyncIterables([identity]); // [!code highlight] + const inbox = subscribeInbox(merged, ({ value }) => value.end === true); + + // First turn: post the reply, learn the thread id, subscribe to the thread. + const threadId = await postFirstReply(channel); + const thread = createHook({ token: `slack:${threadId}` }); + merged.add(thread); // [!code highlight] + + while (true) { + await inbox.wait(); // suspends durably until a message arrives // [!code highlight] + if (inbox.ended) break; + const steering = inbox.drain().map(({ value }) => value.text); + await runTurn(steering); + } + + identity.dispose(); + thread.dispose(); + return { threadId }; +} +``` + +## How it works + +1. **The subscriber is ordinary workflow code.** `for await (const message of hook)` awaits the hook once per payload. Nothing about it is special to the runtime; it is just a promise chain the body never awaits. +2. **Payloads are delivered in event-log order.** Every `resumeHook()` appends a `hook_received` event, and every step result appends a `step_completed` event. The runtime delivers a hook payload to the workflow before or after a step result according to their positions in the log, on the live run and on every replay. So the inbox a turn drains is the same inbox every replay drains at that turn. +3. **The drained messages are recorded.** Passing `steering` to the step stores it in the log as that step's arguments. A replay reconstructs the inbox from `hook_received` events and calls the step with the same arguments, so the two agree. +4. **Merging is `Promise.race` over `hook.next()` calls.** Because each payload resolves in log order, the merged stream is the log order of all participating hooks. `add()` re-races with the new hook included. +5. **A pending subscriber does not block completion.** When the body returns, the run completes even if the subscriber is still awaiting the hook. Dispose hooks with `using` or `hook.dispose()` so the token is released for the next session. + +## Adapting this + +- **Message kinds**: put `type` on the payload (`steer`, `cancel`, `end`) and branch in the turn loop. This replaces one hook per kind with one hook per session. +- **Coalescing**: if several steering messages arrive during one turn, combine them into one user message before the next model call. +- **Idle timeout**: race the subscription against `sleep("30m")` and end the session when nothing arrives. +- **Cancellation**: pair an `end` message with an `AbortController` so the in-flight turn stops early. See [Agent Cancellation](/cookbook/agent-patterns/agent-cancellation). +- **Deterministic tokens**: derive tokens from ids the sender already has (run ID, session ID, Slack thread ts) so senders never need a lookup. See [Token design](/docs/foundations/hooks#token-design). + +## Key APIs + +- [`createHook()`](/docs/api-reference/workflow/create-hook): creates the inbox hook; hooks are `AsyncIterable`, which is what the subscriber and the merge helper consume. +- [`resumeHook()`](/docs/api-reference/workflow-api/resume-hook): sends a payload to a hook by token from any server-side code. +- [`getWorkflowMetadata()`](/docs/api-reference/workflow/get-workflow-metadata): provides the run ID for a deterministic per-session token. +- [Hooks](/docs/foundations/hooks): token design, `for await` iteration, and disposal semantics. diff --git a/docs/content/docs/v5/cookbook/agent-patterns/meta.json b/docs/content/docs/v5/cookbook/agent-patterns/meta.json index 3530caec34..30db658997 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/meta.json +++ b/docs/content/docs/v5/cookbook/agent-patterns/meta.json @@ -1,4 +1,9 @@ { "title": "Agent Patterns", - "pages": ["durable-agent", "human-in-the-loop", "agent-cancellation"] + "pages": [ + "durable-agent", + "human-in-the-loop", + "agent-cancellation", + "hook-inbox" + ] } diff --git a/docs/content/docs/v5/cookbook/index.mdx b/docs/content/docs/v5/cookbook/index.mdx index dc2ed6dae4..799faef118 100644 --- a/docs/content/docs/v5/cookbook/index.mdx +++ b/docs/content/docs/v5/cookbook/index.mdx @@ -11,6 +11,7 @@ Use these workflow patterns and copy-paste code examples to implement common use - [**WorkflowAgent**](/cookbook/agent-patterns/durable-agent): Build durable, resumable AI agents with AI SDK's WorkflowAgent - [**Human-in-the-Loop**](/cookbook/agent-patterns/human-in-the-loop): Pause an agent for human approval, then resume based on the decision - [**Agent Cancellation**](/cookbook/agent-patterns/agent-cancellation): Stop a running agent immediately via `run.cancel()` or gracefully via a hook + `Promise.race` +- [**Hook Inbox & Steering**](/docs/cookbook/agent-patterns/hook-inbox): Buffer hook payloads in a background subscriber to steer an agent loop, and merge several hooks into one ordered inbox ## Common patterns diff --git a/docs/lib/cookbook-tree.ts b/docs/lib/cookbook-tree.ts index 833372489b..38992d10a1 100644 --- a/docs/lib/cookbook-tree.ts +++ b/docs/lib/cookbook-tree.ts @@ -50,6 +50,7 @@ export const slugToCategory: Record = { 'durable-agent': 'agent-patterns', 'human-in-the-loop': 'agent-patterns', 'agent-cancellation': 'agent-patterns', + 'hook-inbox': 'agent-patterns', // Integrations 'ai-sdk': 'integrations', @@ -153,6 +154,14 @@ export const recipes: Record = { 'Cancel a running agent from the outside using AbortSignal — a stop hook fires controller.abort(), the agent step bails out of the model stream, and the client gets a clean stop notification.', category: 'agent-patterns', }, + 'hook-inbox': { + slug: 'hook-inbox', + title: 'Hook Inbox & Steering', + description: + 'Buffer hook payloads in a background subscriber to steer an agent loop at turn boundaries, and merge several hooks into one ordered inbox.', + category: 'agent-patterns', + skipVersions: ['v4'], + }, // Integrations 'ai-sdk': { diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index ab2abe2b10..e006503894 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1382,6 +1382,395 @@ describe.concurrent('e2e', () => { expect([...returnValue].sort((a, b) => a - b)).toEqual([0, 1, 2, 3, 4]); }); + // ==================== HOOK INBOX PATTERNS ==================== + // Background hook subscribers and merged hook inboxes (see the matching + // section in workflows/99_e2e.ts). These stress two userland patterns: + // + // 1. A `for await` over a hook that the workflow body never awaits, pushing + // payloads into a local inbox that the body drains at step boundaries + // (steering an agent loop). + // 2. Several hooks merged into one async iterator, including a hook added + // to the merge after the workflow has started. + // + // Both are plain JavaScript; what the tests establish is that hook delivery + // stays in event-log order relative to step results, so a replay observes + // the same inbox contents at every decision point as the live run did, even + // with dozens of payloads landing mid-step. + describe('hook inbox patterns', () => { + const range = (n: number) => Array.from({ length: n }, (_, i) => i); + // Soak knob: multiplies every per-hook message count. The default is a + // per-PR regression check; set e.g. `E2E_INBOX_SCALE=4` to push several + // hundred payloads through each pattern when hunting for ordering flakes. + const SCALE = Math.max(1, Number(process.env.E2E_INBOX_SCALE ?? 1) || 1); + + /** Every event of `runId`, ascending, fetched as pages of EVENT_POLL_PAGE_SIZE. */ + async function listAllRunEvents(runId: string): Promise { + const world = await getWorld(); + const events: WorkflowEvent[] = []; + let cursor: string | undefined; + while (true) { + const page = await world.events.list({ + runId, + resolveData: 'none', + pagination: { limit: EVENT_POLL_PAGE_SIZE, sortOrder: 'asc', cursor }, + }); + events.push(...page.data); + if (!page.hasMore || !page.cursor) break; + cursor = page.cursor ?? undefined; + } + return events; + } + + /** + * Asserts the run's event log holds exactly the expected number of + * `hook_received` and `step_completed` events and no `hook_conflict`. + * Duplicated or dropped deliveries would already fail the return-value + * assertions; this pins the log itself so a run that happened to return + * the right answer over a wrong log still fails. + */ + async function expectInboxEventLog( + runId: string, + expected: { hookReceived: number; stepCompleted: number } + ) { + const events = await listAllRunEvents(runId); + const count = (type: string) => + events.filter((e) => e.eventType === type).length; + expect(count('hook_conflict')).toBe(0); + expect(count('hook_received')).toBe(expected.hookReceived); + expect(count('step_completed')).toBe(expected.stepCompleted); + } + + test( + 'backgroundInboxWorkflow - background subscriber steers turns deterministically under load', + { timeout: 240_000 }, + async () => { + const token = `inbox-${Math.random().toString(36).slice(2)}`; + const TURNS = 10; + // Bursts are paced by the run's own progress (one burst per completed + // turn) so payloads land while a step is executing, i.e. between a + // turn's `step_created` and its `step_completed` in the log. + const BURSTS = TURNS - 2; + const BURST_SIZE = 8 * SCALE; + const MESSAGES = BURSTS * BURST_SIZE; + + const run = await start(await e2e('backgroundInboxWorkflow'), [ + token, + TURNS, + ]); + await waitForHook(token, { runId: run.runId }); + + let seq = 0; + for (let burst = 1; burst <= BURSTS; burst++) { + await waitForRunEvents( + run.runId, + (event) => event.eventType === 'step_completed', + { + minCount: burst, + timeoutMs: 120_000, + description: `step_completed #${burst} before burst ${burst}`, + } + ); + for (let i = 0; i < BURST_SIZE; i++) { + await resumeHook(token, { seq: seq++ }); + } + } + await resumeHook(token, { seq, done: true }); + + const result = await run.returnValue; + expect(result.turns).toHaveLength(TURNS); + expect(result.subscribed).toBe(MESSAGES); + + // What the workflow observed in its inbox at each turn boundary must + // match what the event log recorded as that turn's step arguments. + // The live run wrote the arguments; the final replay produced + // `steeringSeen`. A replay that delivered a payload on the other side + // of a step boundary would make them disagree. + for (const turn of result.turns) { + expect(turn.steeringApplied, `turn ${turn.turn}`).toEqual( + turn.steeringSeen + ); + } + + // Every payload was delivered exactly once, in send order, across the + // turn boundaries and the final drain. + const delivered = [ + ...result.turns.flatMap((t) => t.steeringSeen), + ...result.drained, + ]; + expect(delivered).toEqual(range(MESSAGES)); + + // The subscriber really ran in the background: some payloads were + // drained at a turn boundary rather than all at the end. Bursts are + // sent after turn k completes and before turn k+2 completes, so at + // least the last burst is observed by a later turn. + expect( + result.turns.filter((t) => t.steeringSeen.length > 0).length + ).toBeGreaterThan(0); + + await expectInboxEventLog(run.runId, { + hookReceived: MESSAGES + 1, + stepCompleted: TURNS, + }); + } + ); + + test( + 'backgroundInboxWorkflow - run completes while the subscriber is still awaiting the hook', + { timeout: 240_000 }, + async () => { + const token = `inbox-open-${Math.random().toString(36).slice(2)}`; + const TURNS = 4; + const MESSAGES = 12 * SCALE; + + const run = await start(await e2e('backgroundInboxWorkflow'), [ + token, + TURNS, + false, + ]); + await waitForHook(token, { runId: run.runId }); + // Send everything after the first turn so at least one turn boundary + // sees a non-empty inbox, then never send an end marker: the workflow + // must complete with the subscriber still parked on the hook. + await waitForRunEvents( + run.runId, + (event) => event.eventType === 'step_completed', + { timeoutMs: 120_000, description: 'first step_completed' } + ); + for (let seq = 0; seq < MESSAGES; seq++) { + await resumeHook(token, { seq }); + } + + const result = await run.returnValue; + expect(result.subscribed).toBeNull(); + expect(result.turns).toHaveLength(TURNS); + for (const turn of result.turns) { + expect(turn.steeringApplied, `turn ${turn.turn}`).toEqual( + turn.steeringSeen + ); + } + const delivered = [ + ...result.turns.flatMap((t) => t.steeringSeen), + ...result.drained, + ]; + expect(delivered).toEqual(range(MESSAGES)); + + // `using` released the token even though the subscriber never saw an + // end marker: a new hook on the same token must not conflict. + const { json } = await cliInspectJson(`runs ${run.runId}`); + expect(json.status).toBe('completed'); + await expectInboxEventLog(run.runId, { + hookReceived: MESSAGES, + stepCompleted: TURNS, + }); + } + ); + + test( + 'inboxWaitLoopWorkflow - drain-then-wait loop groups payloads identically on replay', + { timeout: 240_000 }, + async () => { + const token = `inbox-wait-${Math.random().toString(36).slice(2)}`; + const BURSTS = 6; + const BURST_SIZE = 5 * SCALE; + const MESSAGES = BURSTS * BURST_SIZE; + + const run = await start(await e2e('inboxWaitLoopWorkflow'), [token]); + await waitForHook(token, { runId: run.runId }); + + // Each burst is sent after the previous burst's turn has completed, so + // the live run drains bursts as groups while the payloads inside a + // burst arrive back-to-back and are mostly buffered during the step. + let seq = 0; + for (let burst = 0; burst < BURSTS; burst++) { + if (burst > 0) { + await waitForRunEvents( + run.runId, + (event) => event.eventType === 'step_completed', + { + minCount: burst, + timeoutMs: 120_000, + description: `step_completed #${burst} before burst ${burst}`, + } + ); + } + for (let i = 0; i < BURST_SIZE; i++) { + await resumeHook(token, { seq: seq++ }); + } + } + await waitForRunEvents( + run.runId, + (event) => event.eventType === 'step_completed', + { + minCount: BURSTS, + timeoutMs: 120_000, + description: 'step_completed for the last burst', + } + ); + await resumeHook(token, { seq, done: true }); + + const result = await run.returnValue; + // Every turn drained at least one message (the loop only runs a turn + // after `wait()` resolves), and each turn's step was called with + // exactly what the workflow drained. + expect(result.turns.length).toBeGreaterThanOrEqual(BURSTS); + expect(result.turns.length).toBeLessThanOrEqual(MESSAGES); + for (const turn of result.turns) { + expect(turn.steeringSeen.length).toBeGreaterThan(0); + expect(turn.steeringApplied, `turn ${turn.turn}`).toEqual( + turn.steeringSeen + ); + } + const delivered = [ + ...result.turns.flatMap((t) => t.steeringSeen), + ...result.drained, + ]; + expect(delivered).toEqual(range(MESSAGES)); + expect(result.drained).toEqual([]); + + await expectInboxEventLog(run.runId, { + hookReceived: MESSAGES + 1, + stepCompleted: result.turns.length, + }); + } + ); + + test( + 'mergedHooksWorkflow - merged iterator preserves event-log order across hooks', + { timeout: 240_000 }, + async () => { + const id = Math.random().toString(36).slice(2); + const tokens = range(3).map((i) => `merge-${id}-${i}`); + const PER_HOOK = 40 * SCALE; + + const run = await start(await e2e('mergedHooksWorkflow'), [ + tokens, + false, + ]); + await Promise.all( + tokens.map((token) => waitForHook(token, { runId: run.runId })) + ); + + // Round-robin, strictly sequential sends: each `hook_received` is + // committed before the next is sent, so the log order IS this order, + // and the merged iterator must reproduce it exactly, not just per + // hook. + const expected: { source: number; seq: number }[] = []; + for (let seq = 0; seq < PER_HOOK; seq++) { + for (let source = 0; source < tokens.length; source++) { + await resumeHook(tokens[source], { seq }); + expected.push({ source, seq }); + } + } + for (const token of tokens) { + await resumeHook(token, { seq: PER_HOOK, done: true }); + } + + const received = await run.returnValue; + expect(received).toEqual(expected); + + await expectInboxEventLog(run.runId, { + hookReceived: tokens.length * (PER_HOOK + 1), + stepCompleted: 0, + }); + } + ); + + test( + 'mergedHooksWorkflow - concurrent senders with a step per message keep per-hook order', + { timeout: 300_000 }, + async () => { + const id = Math.random().toString(36).slice(2); + const tokens = range(3).map((i) => `merge-step-${id}-${i}`); + const PER_HOOK = 25 * SCALE; + + const run = await start(await e2e('mergedHooksWorkflow'), [ + tokens, + true, + ]); + await Promise.all( + tokens.map((token) => waitForHook(token, { runId: run.runId })) + ); + + // Three independent senders race each other; only each sender's own + // order is defined. Every message is processed through a step, so + // each payload forces a replay over a log that grows by a hook + // payload and a step per message. + await Promise.all( + tokens.map(async (token) => { + for (let seq = 0; seq < PER_HOOK; seq++) { + await resumeHook(token, { seq }); + } + await resumeHook(token, { seq: PER_HOOK, done: true }); + }) + ); + + const received = await run.returnValue; + expect(received).toHaveLength(tokens.length * PER_HOOK); + for (let source = 0; source < tokens.length; source++) { + expect( + received.filter((r) => r.source === source).map((r) => r.seq), + `hook ${source}` + ).toEqual(range(PER_HOOK)); + } + + await expectInboxEventLog(run.runId, { + hookReceived: tokens.length * (PER_HOOK + 1), + stepCompleted: tokens.length * PER_HOOK, + }); + } + ); + + test( + 'dynamicInboxWorkflow - a hook added to a merged inbox mid-run joins the same ordered stream', + { timeout: 240_000 }, + async () => { + const sessionId = Math.random().toString(36).slice(2); + const identityToken = `identity-${sessionId}`; + const threadToken = `thread:thread-${sessionId}`; + const PHASE_ONE = 15 * SCALE; + const PHASE_TWO = 30 * SCALE; + + const run = await start(await e2e('dynamicInboxWorkflow'), [ + identityToken, + sessionId, + ]); + await waitForHook(identityToken, { runId: run.runId }); + + const expected: { source: number; seq: number }[] = []; + let seq = 0; + + // Phase 1: only the identity hook exists. + for (let i = 0; i < PHASE_ONE; i++) { + await resumeHook(identityToken, { seq }); + expected.push({ source: 0, seq: seq++ }); + } + + // Phase 2: the thread hook is added after the first reply is posted. + // Alternate between the two hooks; the merged inbox must interleave + // them in exactly this (log) order. + await waitForHook(threadToken, { runId: run.runId }); + for (let i = 0; i < PHASE_TWO; i++) { + const source = i % 2; + await resumeHook(source === 0 ? identityToken : threadToken, { + seq, + }); + expected.push({ source, seq: seq++ }); + } + await resumeHook(identityToken, { seq, done: true }); + + const result = await run.returnValue; + expect(result.threadId).toBe(`thread-${sessionId}`); + expect(result.received).toEqual(expected); + + await expectInboxEventLog(run.runId, { + hookReceived: PHASE_ONE + PHASE_TWO + 1, + stepCompleted: 1, + }); + } + ); + }); + // ==================== END HOOK INBOX PATTERNS ==================== + // ==================== ERROR HANDLING TESTS ==================== describe('error handling', () => { describe('error propagation', () => { diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 587c200215..435405ed76 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -3894,3 +3894,326 @@ export async function crossRegionStreamWorkflow(chunkCount: number) { await closeCrossRegionStream(writable); return 'done'; } + +////////////////////////////////////////////////////////// +// Background hook subscribers and merged hook inboxes +////////////////////////////////////////////////////////// + +/** Payload shape shared by the inbox workflows below. */ +type InboxMessage = { seq: number; done?: boolean }; + +/** One item yielded by {@link mergeAsyncIterables}: the value plus which source it came from. */ +type MergedItem = { index: number; value: T }; + +/** + * Merges several async iterables into one, yielding each item as soon as its + * source produces it, tagged with the source's index. Sources can be added + * while the merge is being consumed via `add()`, so a workflow can start with + * one hook (its "identity" inbox) and merge in more later, e.g. a Slack-thread + * hook whose token is only known after the first reply is posted. + * + * This is plain JavaScript (no workflow primitives), so it runs unchanged in + * the workflow sandbox. Inside a workflow, each source's `next()` is an `await + * hook`, and the runtime delivers hook payloads in event-log order, so the + * merged order is deterministic across replays. + */ +function mergeAsyncIterables(initial: AsyncIterable[] = []) { + type Slot = { index: number; result: IteratorResult }; + const iterators = new Map>(); + const pending = new Map>(); + let nextIndex = 0; + // Resolved whenever a source is added, so a consumer blocked in + // `Promise.race` over the current sources re-races with the new one. + let wake!: () => void; + let woken = new Promise((resolve) => { + wake = () => resolve(null); + }); + + const pull = (index: number) => { + const iterator = iterators.get(index); + if (!iterator) return; + pending.set( + index, + iterator.next().then((result) => ({ index, result })) + ); + }; + + const add = (source: AsyncIterable): number => { + const index = nextIndex++; + iterators.set(index, source[Symbol.asyncIterator]()); + pull(index); + const previousWake = wake; + woken = new Promise((resolve) => { + wake = () => resolve(null); + }); + previousWake(); + return index; + }; + + for (const source of initial) add(source); + + const merged: AsyncIterable> & { + add: (source: AsyncIterable) => number; + } = { + add, + async *[Symbol.asyncIterator]() { + try { + while (pending.size > 0) { + const winner = await Promise.race([...pending.values(), woken]); + if (winner === null) continue; // a source was added: re-race + const { index, result } = winner; + if (result.done) { + pending.delete(index); + iterators.delete(index); + continue; + } + pull(index); + yield { index, value: result.value }; + } + } finally { + // Let sources clean up. A disposed hook's iterator never settles a + // pending `next()`, so this is fire-and-forget. + for (const iterator of iterators.values()) { + void iterator.return?.().catch(() => {}); + } + } + }, + }; + return merged; +} + +/** + * Stands in for one agent turn (a model call plus tool calls). The `steering` + * argument is the inbox contents the workflow saw when it started the turn, and + * it is echoed back so the test can compare what the workflow *observed* with + * what was *recorded* in the step's arguments in the event log: a replay that + * observed the inbox differently from the live run would call this step with + * different arguments than the log holds, and the runtime would hand back the + * recorded return value, making the two disagree. + */ +async function inboxTurnStep(turn: number, steering: number[]) { + 'use step'; + await new Promise((resolve) => setTimeout(resolve, 300)); + return { turn, steering }; +} + +/** + * Pattern 1: a "background" hook subscriber. The `for await` over the hook is + * never awaited by the main body: it runs concurrently, appending every payload + * to a local inbox as it arrives. The turn loop drains the inbox at each turn + * boundary and passes the drained messages to the step as steering. The + * workflow finishes when it has run `turns` turns and the subscriber has seen + * the `done` marker. + * + * Determinism requirement under test: the set of messages drained at each turn + * boundary must be the same on every replay, i.e. each `hook_received` is + * delivered to the subscriber before or after a given `step_completed` + * consistently with the event log. + * + * With `waitForDone: false` the workflow returns as soon as the turns are + * over, while the subscriber is still parked on `await hook`: the common + * agent shape where the loop ends and nobody sends an explicit end marker. + * The run must still complete, and `using` must still dispose the hook. + */ +export async function backgroundInboxWorkflow( + token: string, + turns: number, + waitForDone = true +) { + 'use workflow'; + + const inbox: InboxMessage[] = []; + using hook = createHook({ token }); + + // Background subscriber: intentionally NOT awaited here. + const subscriber = (async () => { + let count = 0; + for await (const message of hook) { + if (message.done) break; + inbox.push(message); + count++; + } + return count; + })(); + + const turnLog: { + turn: number; + steeringSeen: number[]; + steeringApplied: number[]; + }[] = []; + for (let turn = 0; turn < turns; turn++) { + const steering = inbox.splice(0).map((m) => m.seq); + const result = await inboxTurnStep(turn, steering); + turnLog.push({ + turn, + steeringSeen: steering, + steeringApplied: result.steering, + }); + } + + const subscribed = waitForDone ? await subscriber : null; + const drained = inbox.splice(0).map((m) => m.seq); + return { turns: turnLog, drained, subscribed }; +} + +/** + * A background subscriber wrapped as an inbox with `drain()` and `wait()`. + * `wait()` resolves once a message has been pushed (or the source ended), so a + * loop can `await inbox.wait()` instead of spinning; while nothing is buffered + * the only pending promise is the subscriber's `await hook`, so the run + * suspends durably. + */ +function subscribeInbox(source: AsyncIterable, isEnd: (v: T) => boolean) { + const buffered: T[] = []; + let ended = false; + let notify = () => {}; + let changed = new Promise((resolve) => { + notify = resolve; + }); + const bump = () => { + const previous = notify; + changed = new Promise((resolve) => { + notify = resolve; + }); + previous(); + }; + const done = (async () => { + for await (const value of source) { + if (isEnd(value)) { + ended = true; + bump(); + break; + } + buffered.push(value); + bump(); + } + })(); + return { + done, + drain: () => buffered.splice(0), + get ended() { + return ended; + }, + async wait() { + if (buffered.length === 0 && !ended) await changed; + }, + }; +} + +/** + * Pattern 1 in its "chat session" shape: instead of a fixed number of turns, + * the loop drains the inbox, runs a turn over what it drained, and then waits + * for the next message. What each turn drains depends on how many payloads + * the subscriber pushed between the previous turn's step result and the + * `wait()` resolving: a replay must reproduce the live run's grouping exactly, + * which is what the echoed `steering` argument checks. + */ +export async function inboxWaitLoopWorkflow(token: string) { + 'use workflow'; + + using hook = createHook({ token }); + const inbox = subscribeInbox(hook, (m) => m.done === true); + + const turnLog: { + turn: number; + steeringSeen: number[]; + steeringApplied: number[]; + }[] = []; + let turn = 0; + while (true) { + await inbox.wait(); + if (inbox.ended) break; + const steering = inbox.drain().map((m) => m.seq); + const result = await inboxTurnStep(turn, steering); + turnLog.push({ + turn, + steeringSeen: steering, + steeringApplied: result.steering, + }); + turn++; + } + await inbox.done; + return { turns: turnLog, drained: inbox.drain().map((m) => m.seq) }; +} + +async function recordMergedMessage(source: number, seq: number) { + 'use step'; + return { source, seq }; +} + +/** + * Pattern 2: several hooks merged into one async iterator. Each hook is its own + * token (e.g. an identity inbox, a Slack thread, an auth callback), and the + * workflow consumes a single ordered stream. With `stepPerMessage`, every + * message is also processed through a step so each payload forces a replay + * over a growing log. Each hook is closed by its own `done` marker; the + * workflow returns once every hook has been closed. + */ +export async function mergedHooksWorkflow( + tokens: string[], + stepPerMessage: boolean +) { + 'use workflow'; + + const hooks = tokens.map((token) => createHook({ token })); + const received: { source: number; seq: number }[] = []; + let open = hooks.length; + + for await (const { index, value } of mergeAsyncIterables(hooks)) { + if (value.done) { + open--; + if (open === 0) break; + continue; + } + if (stepPerMessage) { + received.push(await recordMergedMessage(index, value.seq)); + } else { + received.push({ source: index, seq: value.seq }); + } + } + + for (const hook of hooks) hook.dispose(); + return received; +} + +async function postFirstReplyStep(sessionId: string) { + 'use step'; + // A real app would post to Slack here and get the thread timestamp back. + return `thread-${sessionId}`; +} + +/** + * Pattern 1 + 2 together, shaped like a chat session: the run starts with only + * its identity hook, a background subscriber drains the merged inbox, and after + * the first reply is posted the thread's hook is added to the same merged + * inbox. Messages sent to either token land in one ordered inbox. The identity + * hook's `done` marker ends the session. + */ +export async function dynamicInboxWorkflow( + identityToken: string, + sessionId: string +) { + 'use workflow'; + + const identity = createHook({ token: identityToken }); + const inbox = mergeAsyncIterables([identity]); + const received: { source: number; seq: number }[] = []; + + // Background subscriber over the merged inbox; runs for the whole session. + const subscriber = (async () => { + for await (const { index, value } of inbox) { + if (value.done) break; + received.push({ source: index, seq: value.seq }); + } + })(); + + // First turn: post a reply, learn the thread id, subscribe to the thread. + const threadId = await postFirstReplyStep(sessionId); + const thread = createHook({ token: `thread:${threadId}` }); + inbox.add(thread); + + await subscriber; + identity.dispose(); + thread.dispose(); + return { threadId, received }; +} From d827bb59a62260f120117ba4c3d79a39fe745d3a Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 13:14:26 -0700 Subject: [PATCH 2/9] Stop the early-return inbox test racing slow resumes; drain before ending the wait loop On every Vercel lane the early-return test failed with HookNotFoundError: each resume is slow enough there that the run finished its four turns and `using` disposed the hook while the test was still sending. The workflow now takes a `minMessages` floor and keeps turning until its subscriber has seen that many payloads, which is as replay-safe as reading the inbox. The drain-then-wait loop checked `ended` before draining, so payloads pushed between the last drain and the end marker were left over; it cost one retry on some Vercel lanes. The workflow and the cookbook example now drain first. The dynamic-inbox test's thread-hook lookup also gets the 120s budget the event waits use, since it depends on a step completing first. Co-Authored-By: Claude Fable 5.1 --- .../v5/cookbook/agent-patterns/hook-inbox.mdx | 6 ++++- packages/core/e2e/e2e.test.ts | 18 +++++++++++---- workbench/example/workflows/99_e2e.ts | 23 ++++++++++++++----- 3 files changed, 36 insertions(+), 11 deletions(-) diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx index 6d84bd280b..1894771849 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -279,8 +279,12 @@ export async function slackSession(sessionId: string, channel: string) { while (true) { await inbox.wait(); // suspends durably until a message arrives // [!code highlight] - if (inbox.ended) break; + // Drain before checking for the end marker so nothing is left behind. const steering = inbox.drain().map(({ value }) => value.text); + if (steering.length === 0) { + if (inbox.ended) break; + continue; + } await runTurn(steering); } diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index e006503894..737d121301 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1522,10 +1522,14 @@ describe.concurrent('e2e', () => { const TURNS = 4; const MESSAGES = 12 * SCALE; + // `minMessages` keeps the run turning until it has seen every message, + // so a slow lane cannot finish the run (and dispose the hook) while + // this test is still sending. const run = await start(await e2e('backgroundInboxWorkflow'), [ token, TURNS, false, + MESSAGES, ]); await waitForHook(token, { runId: run.runId }); // Send everything after the first turn so at least one turn boundary @@ -1542,7 +1546,7 @@ describe.concurrent('e2e', () => { const result = await run.returnValue; expect(result.subscribed).toBeNull(); - expect(result.turns).toHaveLength(TURNS); + expect(result.turns.length).toBeGreaterThanOrEqual(TURNS); for (const turn of result.turns) { expect(turn.steeringApplied, `turn ${turn.turn}`).toEqual( turn.steeringSeen @@ -1560,7 +1564,7 @@ describe.concurrent('e2e', () => { expect(json.status).toBe('completed'); await expectInboxEventLog(run.runId, { hookReceived: MESSAGES, - stepCompleted: TURNS, + stepCompleted: result.turns.length, }); } ); @@ -1747,8 +1751,14 @@ describe.concurrent('e2e', () => { // Phase 2: the thread hook is added after the first reply is posted. // Alternate between the two hooks; the merged inbox must interleave - // them in exactly this (log) order. - await waitForHook(threadToken, { runId: run.runId }); + // them in exactly this (log) order. Unlike the identity hook, this + // one only registers after the run's first replay has completed a + // step, so under suite load it gets the same budget as the event + // waits rather than waitForHook's default. + await waitForHook(threadToken, { + runId: run.runId, + timeoutMs: 120_000, + }); for (let i = 0; i < PHASE_TWO; i++) { const source = i % 2; await resumeHook(source === 0 ? identityToken : threadToken, { diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 435405ed76..d91d64d73d 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -4014,11 +4014,17 @@ async function inboxTurnStep(turn: number, steering: number[]) { * over, while the subscriber is still parked on `await hook`: the common * agent shape where the loop ends and nobody sends an explicit end marker. * The run must still complete, and `using` must still dispose the hook. + * `minMessages` keeps the turn loop going until the subscriber has seen that + * many payloads, so a test can send a fixed number of messages without racing + * the run to completion (a resume against the disposed hook would fail with + * HookNotFoundError). Reading the count at a turn boundary is as replay-safe + * as reading the inbox itself. */ export async function backgroundInboxWorkflow( token: string, turns: number, - waitForDone = true + waitForDone = true, + minMessages = 0 ) { 'use workflow'; @@ -4026,14 +4032,14 @@ export async function backgroundInboxWorkflow( using hook = createHook({ token }); // Background subscriber: intentionally NOT awaited here. + let received = 0; const subscriber = (async () => { - let count = 0; for await (const message of hook) { if (message.done) break; inbox.push(message); - count++; + received++; } - return count; + return received; })(); const turnLog: { @@ -4041,7 +4047,7 @@ export async function backgroundInboxWorkflow( steeringSeen: number[]; steeringApplied: number[]; }[] = []; - for (let turn = 0; turn < turns; turn++) { + for (let turn = 0; turn < turns || received < minMessages; turn++) { const steering = inbox.splice(0).map((m) => m.seq); const result = await inboxTurnStep(turn, steering); turnLog.push({ @@ -4122,8 +4128,13 @@ export async function inboxWaitLoopWorkflow(token: string) { let turn = 0; while (true) { await inbox.wait(); - if (inbox.ended) break; + // Drain before checking for the end marker: payloads pushed between the + // previous drain and the end marker still get a turn. const steering = inbox.drain().map((m) => m.seq); + if (steering.length === 0) { + if (inbox.ended) break; + continue; + } const result = await inboxTurnStep(turn, steering); turnLog.push({ turn, From 1b6cd3ab799a345f260a0fe38e7b1e4e71c0f10a Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 13:28:33 -0700 Subject: [PATCH 3/9] Split the cookbook into a general Background Hook Subscriber recipe; end the inbox in a finally Review found that `subscribeInbox().wait()` could hang forever when the source iterable completed or threw without matching `isEnd`, since `ended` was only set on the end-marker branch. The subscriber now sets `ended` and wakes waiters in a `finally`, in the e2e workflow and in the docs. The background subscriber is a general pattern, not an agent one, so it gets its own Common Patterns page: subscribe-then-read-at-safe-points over a long import, the `subscribeInbox` drain/wait wrapper, how a subscription ends, and why buffer reads after an await are replay-safe. The Agent Patterns page now covers steering at turn boundaries and merged inboxes and links to it for the shared helper. Both v5-only pages are linked through the version-aware `/docs/cookbook/...` hrefs the docs lint requires. Co-Authored-By: Claude Fable 5.1 --- .../v5/cookbook/agent-patterns/hook-inbox.mdx | 69 +----- .../background-hook-subscriber.mdx | 219 ++++++++++++++++++ .../v5/cookbook/common-patterns/meta.json | 3 +- docs/content/docs/v5/cookbook/index.mdx | 3 +- docs/lib/cookbook-tree.ts | 11 +- workbench/example/workflows/99_e2e.ts | 13 +- 6 files changed, 253 insertions(+), 65 deletions(-) create mode 100644 docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx index 1894771849..fa8ab50be0 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -1,9 +1,10 @@ --- title: Hook Inbox & Steering -description: Buffer hook payloads in the background while an agent loop runs, drain them at turn boundaries to steer the agent, and merge several hooks into one ordered inbox. +description: Steer an agent loop with messages that arrive mid-turn, and merge several hooks into one ordered inbox, including hooks added after the session started. type: guide -summary: Subscribe to a hook with a `for await` loop that the workflow body never awaits, so payloads land in a local inbox as they arrive and each agent turn can drain them. Merge multiple hooks into a single async iterator, and add hooks to the merge mid-run, while the runtime keeps delivery in event-log order across replays. +summary: Apply the background hook subscriber to an agent loop. Messages arriving during a turn are drained at the next turn boundary and passed to the model as steering. Merge multiple hooks into a single async iterator, and add hooks to the merge mid-run, while the runtime keeps delivery in event-log order across replays. related: + - /docs/cookbook/common-patterns/background-hook-subscriber - /docs/foundations/hooks - /docs/api-reference/workflow/create-hook - /docs/api-reference/workflow-api/resume-hook @@ -14,7 +15,7 @@ related: text="Add a steerable inbox to this durable agent loop. In the "use workflow" function, create one hook per session with `createHook()` from `workflow` using a deterministic token such as `session:${workflowRunId}`. Start a background subscriber: an async IIFE that runs `for await (const message of hook)` and pushes each payload into a local `inbox` array, and do NOT await it before the turn loop. In the turn loop, `inbox.splice(0)` at the start of each turn and pass the drained messages to the "use step" function that runs the turn. Dispose the hook with `using` or `hook.dispose()` before returning. If messages can arrive on more than one token (for example a Slack thread whose id is only known after the first reply), merge the hooks with an async-iterable merge helper that supports `add()` and subscribe to the merged iterator instead. Expose a server route that calls `resumeHook(token, payload)` from `workflow/api`. Verify that messages sent mid-turn are applied on the next turn, that replays observe the same inbox at every turn, and that messages from every token arrive in send order." /> -An agent loop runs one turn at a time: call the model, run tools, repeat. Messages from the user, or from other systems, keep arriving while a turn is in flight. Awaiting the hook directly would block the loop, and racing the hook against every turn would drop messages that arrive between races. Instead, subscribe to the hook in the background and let each turn drain what arrived since the last one. +An agent loop runs one turn at a time: call the model, run tools, repeat. Messages from the user, or from other systems, keep arriving while a turn is in flight. Awaiting the hook directly would block the loop, and racing the hook against every turn would drop messages that arrive between races. Instead, apply the [Background Hook Subscriber](/docs/cookbook/common-patterns/background-hook-subscriber) pattern: iterate the hook in the background and let each turn drain what arrived since the last one. This recipe covers the agent-specific parts: steering at turn boundaries, and one inbox fed by several hooks. ## When to use this @@ -22,7 +23,7 @@ An agent loop runs one turn at a time: call the model, run tools, repeat. Messag - **Cancellation and interruption**: an `end` or `stop` message should be noticed at the next turn boundary without a separate hook - **Multiple channels**: a session receives messages on more than one token, such as an identity token plus a Slack thread that only exists after the first reply -## Pattern: background subscriber with a local inbox +## Pattern: steering at turn boundaries The subscriber is an async function that iterates the hook and pushes into an array. The workflow body starts it and moves on. Each turn drains the array and hands the drained messages to the step that runs the turn. @@ -189,55 +190,9 @@ export function mergeAsyncIterables(initial: AsyncIterable[] = []) { } ``` -### Wrapping the subscriber as an inbox +### Waiting for the next message -For a session that runs until told to stop, the loop needs to wait for the next message without spinning. Wrap the subscriber so it exposes `drain()` and `wait()`. While nothing is buffered, the only pending promise in the run is the subscriber's `await hook`, so `await inbox.wait()` suspends the workflow durably until a payload arrives. - -```typescript lineNumbers -/** - * Subscribes to `source` in the background and buffers what it yields. - * `wait()` resolves once a message has been buffered or the source ended. - */ -export function subscribeInbox( - source: AsyncIterable, - isEnd: (value: T) => boolean -) { - const buffered: T[] = []; - let ended = false; - let notify = () => {}; - let changed = new Promise((resolve) => { - notify = resolve; - }); - const bump = () => { - const previous = notify; - changed = new Promise((resolve) => { - notify = resolve; - }); - previous(); - }; - const done = (async () => { - for await (const value of source) { - if (isEnd(value)) { - ended = true; - bump(); - break; - } - buffered.push(value); - bump(); - } - })(); - return { - done, - drain: () => buffered.splice(0), - get ended() { - return ended; - }, - async wait() { - if (buffered.length === 0 && !ended) await changed; - }, - }; -} -``` +For a session that runs until told to stop, the loop needs to block between messages without spinning. The `subscribeInbox()` wrapper from [Background Hook Subscriber](/docs/cookbook/common-patterns/background-hook-subscriber#pattern-the-subscriber-as-an-inbox-with-drain-and-wait) gives the subscriber `drain()` and a durable `wait()`; the example below uses it over the merged stream. ### Adding a hook mid-session @@ -253,6 +208,7 @@ declare function mergeAsyncIterables( ): AsyncIterable<{ index: number; value: T }> & { add: (source: AsyncIterable) => number; }; // @setup +// subscribeInbox() is defined in the Background Hook Subscriber recipe. declare function subscribeInbox( source: AsyncIterable, isEnd: (value: T) => boolean @@ -296,11 +252,9 @@ export async function slackSession(sessionId: string, channel: string) { ## How it works -1. **The subscriber is ordinary workflow code.** `for await (const message of hook)` awaits the hook once per payload. Nothing about it is special to the runtime; it is just a promise chain the body never awaits. -2. **Payloads are delivered in event-log order.** Every `resumeHook()` appends a `hook_received` event, and every step result appends a `step_completed` event. The runtime delivers a hook payload to the workflow before or after a step result according to their positions in the log, on the live run and on every replay. So the inbox a turn drains is the same inbox every replay drains at that turn. -3. **The drained messages are recorded.** Passing `steering` to the step stores it in the log as that step's arguments. A replay reconstructs the inbox from `hook_received` events and calls the step with the same arguments, so the two agree. -4. **Merging is `Promise.race` over `hook.next()` calls.** Because each payload resolves in log order, the merged stream is the log order of all participating hooks. `add()` re-races with the new hook included. -5. **A pending subscriber does not block completion.** When the body returns, the run completes even if the subscriber is still awaiting the hook. Dispose hooks with `using` or `hook.dispose()` so the token is released for the next session. +1. **Steering is a buffer read at a turn boundary.** The subscriber pushes payloads as they arrive; the loop reads the buffer right after a step result. The runtime delivers hook payloads and step results in event-log order, so every replay reads the same buffer at the same turn. Passing the drained messages to the step records them as its arguments, so the recorded turn and the replayed inbox always agree. The general mechanics are in [Background Hook Subscriber](/docs/cookbook/common-patterns/background-hook-subscriber#how-it-works). +2. **Merging is `Promise.race` over `hook.next()` calls.** Because each payload resolves in log order, the merged stream is the log order of all participating hooks. `add()` re-races with the new hook included, so a hook created after a step joins the same ordered stream. +3. **A pending subscriber does not block completion.** When the body returns, the run completes even if the subscriber is still awaiting the hook. Dispose hooks with `using` or `hook.dispose()` so the tokens are released for the next session. ## Adapting this @@ -315,4 +269,5 @@ export async function slackSession(sessionId: string, channel: string) { - [`createHook()`](/docs/api-reference/workflow/create-hook): creates the inbox hook; hooks are `AsyncIterable`, which is what the subscriber and the merge helper consume. - [`resumeHook()`](/docs/api-reference/workflow-api/resume-hook): sends a payload to a hook by token from any server-side code. - [`getWorkflowMetadata()`](/docs/api-reference/workflow/get-workflow-metadata): provides the run ID for a deterministic per-session token. +- [Background Hook Subscriber](/docs/cookbook/common-patterns/background-hook-subscriber): the general form of the subscriber and the `subscribeInbox()` wrapper used here. - [Hooks](/docs/foundations/hooks): token design, `for await` iteration, and disposal semantics. diff --git a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx new file mode 100644 index 0000000000..7d4cc94eb4 --- /dev/null +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -0,0 +1,219 @@ +--- +title: Background Hook Subscriber +description: React to hook payloads as they arrive while the workflow keeps doing other work, by iterating the hook in a subscriber the workflow body never awaits. +type: guide +summary: Start a `for await` loop over a hook and do not await it. Payloads land in a local buffer as they arrive; the main flow reads the buffer at the points where it can act on them. Delivery follows event-log order relative to step results, so every replay observes the same buffer at the same points. A `subscribeInbox` wrapper adds `drain()` and a durable `wait()`. +related: + - /docs/foundations/hooks + - /docs/api-reference/workflow/create-hook + - /docs/api-reference/workflow-api/resume-hook + - /cookbook/common-patterns/webhooks + - /docs/cookbook/agent-patterns/hook-inbox +--- + + + +Awaiting a hook pauses the workflow until a payload arrives. That is the right shape when the payload is the next thing the workflow needs. It is the wrong shape when the workflow has other work to do and payloads are side information: a pause request during a long import, a priority change while a queue drains, progress acknowledgements from an external system. For those, iterate the hook in the background and let the main flow read what has arrived whenever it reaches a point where it can act. + +## When to use this + +- **Control signals during long work**: pause, cancel, reprioritize, or reconfigure a loop that is busy running steps +- **Collecting callbacks while continuing**: gather acknowledgements or partial results from an external system without stopping to wait for each one +- **Steering an agent**: feed user messages into an agent loop at turn boundaries. See [Hook Inbox & Steering](/docs/cookbook/agent-patterns/hook-inbox) for that specialization + +## Pattern: subscribe, then read at safe points + +The subscriber is an async function that iterates the hook and pushes into an array. The body starts it and moves on. At each iteration of its own loop the body reads the array, acts on it, and hands what it read to the step so the decision is recorded. + +```typescript lineNumbers +import { createHook } from "workflow"; + +type Control = { type: "skip"; ids: string[] } | { type: "stop" }; + +declare function importBatch( + batch: string[], + skipped: string[] +): Promise<{ imported: number }>; // @setup + +export async function importJob(jobId: string, batches: string[][]) { + "use workflow"; + + using control = createHook({ token: `import:${jobId}` }); // [!code highlight] + + const pending: Control[] = []; + let stopped = false; + + // Background subscriber. Intentionally NOT awaited: it keeps running while + // the batches import, so control messages land in `pending` as they arrive. + const subscription = (async () => { // [!code highlight] + for await (const message of control) { // [!code highlight] + if (message.type === "stop") { // [!code highlight] + stopped = true; // [!code highlight] + break; // [!code highlight] + } // [!code highlight] + pending.push(message); // [!code highlight] + } // [!code highlight] + })(); // [!code highlight] + + let imported = 0; + const skipped = new Set(); + + for (const batch of batches) { + // Read everything that arrived during the previous batch. + for (const message of pending.splice(0)) { // [!code highlight] + if (message.type === "skip") for (const id of message.ids) skipped.add(id); + } + if (stopped) break; + + // Pass what the workflow read into the step so it is recorded with it. + const result = await importBatch(batch, [...skipped]); // [!code highlight] + imported += result.imported; + } + + // `using` disposes the hook here. The subscriber may still be parked on + // `await hook`; the run completes anyway. + return { imported, stopped }; +} +``` + +### Sending a control message + +Any process that knows the job id can steer the import, because the token is deterministic. + +```typescript lineNumbers +import { resumeHook } from "workflow/api"; + +export async function POST( + request: Request, + { params }: { params: Promise<{ jobId: string }> } +) { + const { jobId } = await params; + const message = await request.json(); + await resumeHook(`import:${jobId}`, message); // [!code highlight] + return Response.json({ ok: true }); +} +``` + +## Pattern: the subscriber as an inbox with `drain()` and `wait()` + +When the main flow sometimes has nothing to do until the next payload, it needs to block without spinning. Wrap the subscriber so it exposes `drain()` and `wait()`. While the buffer is empty, the only pending promise in the run is the subscriber's `await hook`, so `await inbox.wait()` suspends the workflow durably until a payload arrives. + +```typescript lineNumbers +/** + * Subscribes to `source` in the background and buffers what it yields. + * `wait()` resolves once a message has been buffered or the source ended. + */ +export function subscribeInbox( + source: AsyncIterable, + isEnd: (value: T) => boolean +) { + const buffered: T[] = []; + let ended = false; + let notify = () => {}; + let changed = new Promise((resolve) => { + notify = resolve; + }); + const bump = () => { + const previous = notify; + changed = new Promise((resolve) => { + notify = resolve; + }); + previous(); + }; + const done = (async () => { + try { + for await (const value of source) { + if (isEnd(value)) break; + buffered.push(value); + bump(); + } + } finally { + // Runs on the end marker, when the source completes on its own, and + // when it throws, so `wait()` always unblocks once the source stops. + ended = true; + bump(); + } + })(); + return { + done, + drain: () => buffered.splice(0), + get ended() { + return ended; + }, + async wait() { + if (buffered.length === 0 && !ended) await changed; + }, + }; +} +``` + +### A loop that blocks between payloads + +Drain before checking for the end, so payloads pushed between the last drain and the end marker are still processed. + +```typescript lineNumbers +import { createHook } from "workflow"; + +type Job = { id: string; end?: boolean }; + +declare function subscribeInbox( + source: AsyncIterable, + isEnd: (value: T) => boolean +): { + done: Promise; + drain: () => T[]; + readonly ended: boolean; + wait: () => Promise; +}; // @setup +declare function processJobs(jobs: Job[]): Promise; // @setup + +export async function jobConsumer(queueId: string) { + "use workflow"; + + using hook = createHook({ token: `queue:${queueId}` }); + const inbox = subscribeInbox(hook, (job) => job.end === true); // [!code highlight] + + while (true) { + await inbox.wait(); // suspends durably while nothing is buffered // [!code highlight] + const jobs = inbox.drain(); // [!code highlight] + if (jobs.length === 0) { + if (inbox.ended) break; + continue; + } + // Everything that arrived since the last drain is processed together. + await processJobs(jobs); + } +} +``` + +## Ending the subscription + +There are two ways a subscription ends, and both are fine: + +- **An end marker.** The subscriber `break`s when it sees a payload it recognizes as the end (`stop`, `end: true`). Use this when the workflow should wait for the sender to say it is done, and `await` the subscription before returning. +- **The body returns first.** The subscriber is still parked on `await hook`. The run completes regardless; the pending await is abandoned with the run. Use `using` (or call `hook.dispose()`) so the token is released for the next run. + +What you should not do is read the buffer from code that has no durable position, such as a timer callback: only reads that happen right after an `await` on a step, a sleep, or the inbox's `wait()` are anchored to the event log. + +## How it works + +1. **The subscriber is ordinary workflow code.** `for await (const m of hook)` awaits the hook once per payload. The runtime does not treat it specially; it is a promise chain the body chose not to await. +2. **Deliveries are ordered by the event log.** Each `resumeHook()` appends a `hook_received` event and each step result appends a `step_completed` event. The runtime hands a hook payload to the workflow before or after a step result according to their positions in the log, on the live run and on every replay. A buffer read right after a step therefore sees the same payloads on every replay. +3. **What you read gets recorded.** Passing the read values to the step stores them in the log as that step's arguments. A replay rebuilds the buffer from the `hook_received` events and calls the step with the same arguments. +4. **`wait()` suspends durably.** With the buffer empty, the subscriber's pending `await hook` is the only open work, so the runtime suspends the run until the next payload and replays it from the log when one arrives. + +## Adapting this + +- **Coalesce**: if many payloads can arrive between reads, reduce them to one decision (latest wins, or a set union as with `skip` above) before acting. +- **Timeouts**: race `inbox.wait()` against `sleep()` to bound how long the flow blocks between payloads. +- **Multiple sources**: merge several hooks into one async iterable and subscribe to the merged stream. [Hook Inbox & Steering](/docs/cookbook/agent-patterns/hook-inbox) shows a merge helper that supports adding hooks mid-run. +- **Webhooks**: the same subscriber works over `createWebhook()`, whose iterator yields `Request` objects. + +## Key APIs + +- [`createHook()`](/docs/api-reference/workflow/create-hook): creates the hook; hooks are `AsyncIterable`, which is what the subscriber consumes. +- [`resumeHook()`](/docs/api-reference/workflow-api/resume-hook): sends a payload to a hook by token from server-side code. +- [`createWebhook()`](/docs/api-reference/workflow/create-webhook): the HTTP-callback variant of a hook, iterable the same way. +- [Hooks](/docs/foundations/hooks): token design, iteration, and disposal semantics. diff --git a/docs/content/docs/v5/cookbook/common-patterns/meta.json b/docs/content/docs/v5/cookbook/common-patterns/meta.json index 3c1ed7e585..69bb820ab7 100644 --- a/docs/content/docs/v5/cookbook/common-patterns/meta.json +++ b/docs/content/docs/v5/cookbook/common-patterns/meta.json @@ -10,6 +10,7 @@ "scheduling", "timeouts", "idempotency", - "webhooks" + "webhooks", + "background-hook-subscriber" ] } diff --git a/docs/content/docs/v5/cookbook/index.mdx b/docs/content/docs/v5/cookbook/index.mdx index 799faef118..938e5ccfc0 100644 --- a/docs/content/docs/v5/cookbook/index.mdx +++ b/docs/content/docs/v5/cookbook/index.mdx @@ -11,7 +11,7 @@ Use these workflow patterns and copy-paste code examples to implement common use - [**WorkflowAgent**](/cookbook/agent-patterns/durable-agent): Build durable, resumable AI agents with AI SDK's WorkflowAgent - [**Human-in-the-Loop**](/cookbook/agent-patterns/human-in-the-loop): Pause an agent for human approval, then resume based on the decision - [**Agent Cancellation**](/cookbook/agent-patterns/agent-cancellation): Stop a running agent immediately via `run.cancel()` or gracefully via a hook + `Promise.race` -- [**Hook Inbox & Steering**](/docs/cookbook/agent-patterns/hook-inbox): Buffer hook payloads in a background subscriber to steer an agent loop, and merge several hooks into one ordered inbox +- [**Hook Inbox & Steering**](/docs/cookbook/agent-patterns/hook-inbox): Steer an agent loop with messages that arrive mid-turn, and merge several hooks into one ordered inbox ## Common patterns @@ -24,6 +24,7 @@ Use these workflow patterns and copy-paste code examples to implement common use - [**Timeouts**](/cookbook/common-patterns/timeouts): Add deadlines to slow steps, hooks, and webhooks by racing them against a durable sleep - [**Idempotency**](/cookbook/common-patterns/idempotency): Ensure side effects and duplicate starts are safe to retry - [**Webhooks**](/cookbook/common-patterns/webhooks): Receive HTTP callbacks from external services and process them durably +- [**Background Hook Subscriber**](/docs/cookbook/common-patterns/background-hook-subscriber): React to hook payloads while the workflow keeps working, by iterating the hook in a subscriber the body never awaits ## Integrations diff --git a/docs/lib/cookbook-tree.ts b/docs/lib/cookbook-tree.ts index 38992d10a1..e024eb9ac0 100644 --- a/docs/lib/cookbook-tree.ts +++ b/docs/lib/cookbook-tree.ts @@ -45,6 +45,7 @@ export const slugToCategory: Record = { timeouts: 'common-patterns', idempotency: 'common-patterns', webhooks: 'common-patterns', + 'background-hook-subscriber': 'common-patterns', // Agent Patterns 'durable-agent': 'agent-patterns', @@ -131,6 +132,14 @@ export const recipes: Record = { 'Receive HTTP callbacks from external services, process them durably, and respond inline.', category: 'common-patterns', }, + 'background-hook-subscriber': { + slug: 'background-hook-subscriber', + title: 'Background Hook Subscriber', + description: + 'React to hook payloads as they arrive while the workflow keeps doing other work, by iterating the hook in a subscriber the body never awaits.', + category: 'common-patterns', + skipVersions: ['v4'], + }, // Agent Patterns 'durable-agent': { @@ -158,7 +167,7 @@ export const recipes: Record = { slug: 'hook-inbox', title: 'Hook Inbox & Steering', description: - 'Buffer hook payloads in a background subscriber to steer an agent loop at turn boundaries, and merge several hooks into one ordered inbox.', + 'Steer an agent loop with messages that arrive mid-turn, and merge several hooks into one ordered inbox, including hooks added after the session started.', category: 'agent-patterns', skipVersions: ['v4'], }, diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index d91d64d73d..e4e996ef4f 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -4084,13 +4084,16 @@ function subscribeInbox(source: AsyncIterable, isEnd: (v: T) => boolean) { previous(); }; const done = (async () => { - for await (const value of source) { - if (isEnd(value)) { - ended = true; + try { + for await (const value of source) { + if (isEnd(value)) break; + buffered.push(value); bump(); - break; } - buffered.push(value); + } finally { + // Runs on the end marker, when the source completes on its own, and + // when it throws, so `wait()` always unblocks once the source stops. + ended = true; bump(); } })(); From de9f6710245bb99a580abb503fae0f04eae9ae0b Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 13:31:40 -0700 Subject: [PATCH 4/9] Highlight the key lines of the mergeAsyncIterables recipe Co-Authored-By: Claude Fable 5.1 --- .../v5/cookbook/agent-patterns/hook-inbox.mdx | 26 +++++++++---------- 1 file changed, 13 insertions(+), 13 deletions(-) diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx index fa8ab50be0..934f61930b 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -127,7 +127,7 @@ export type MergedItem = { index: number; value: T }; export function mergeAsyncIterables(initial: AsyncIterable[] = []) { type Slot = { index: number; result: IteratorResult }; const iterators = new Map>(); - const pending = new Map>(); + const pending = new Map>(); // [!code highlight] let nextIndex = 0; // Resolved whenever a source is added, so a consumer blocked in // `Promise.race` over the current sources re-races with the new one. @@ -139,21 +139,21 @@ export function mergeAsyncIterables(initial: AsyncIterable[] = []) { const pull = (index: number) => { const iterator = iterators.get(index); if (!iterator) return; - pending.set( - index, - iterator.next().then((result) => ({ index, result })) - ); + pending.set( // [!code highlight] + index, // [!code highlight] + iterator.next().then((result) => ({ index, result })) // [!code highlight] + ); // [!code highlight] }; - const add = (source: AsyncIterable): number => { + const add = (source: AsyncIterable): number => { // [!code highlight] const index = nextIndex++; - iterators.set(index, source[Symbol.asyncIterator]()); - pull(index); + iterators.set(index, source[Symbol.asyncIterator]()); // [!code highlight] + pull(index); // [!code highlight] const previousWake = wake; woken = new Promise((resolve) => { wake = () => resolve(null); }); - previousWake(); + previousWake(); // [!code highlight] return index; }; @@ -166,16 +166,16 @@ export function mergeAsyncIterables(initial: AsyncIterable[] = []) { async *[Symbol.asyncIterator]() { try { while (pending.size > 0) { - const winner = await Promise.race([...pending.values(), woken]); - if (winner === null) continue; // a source was added: re-race + const winner = await Promise.race([...pending.values(), woken]); // [!code highlight] + if (winner === null) continue; // a source was added: re-race // [!code highlight] const { index, result } = winner; if (result.done) { pending.delete(index); iterators.delete(index); continue; } - pull(index); - yield { index, value: result.value }; + pull(index); // [!code highlight] + yield { index, value: result.value }; // [!code highlight] } } finally { // A disposed hook never settles its pending `next()`, so this is From 4609514c968e2cdd0f0ac0468df4cc1e93ff96ac Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 13:33:55 -0700 Subject: [PATCH 5/9] Highlight the drain and wait paths in the Background Hook Subscriber recipe Co-Authored-By: Claude Fable 5.1 --- .../background-hook-subscriber.mdx | 28 +++++++++---------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx index 7d4cc94eb4..4c05957f47 100644 --- a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -109,13 +109,13 @@ export function subscribeInbox( source: AsyncIterable, isEnd: (value: T) => boolean ) { - const buffered: T[] = []; + const buffered: T[] = []; // [!code highlight] let ended = false; let notify = () => {}; let changed = new Promise((resolve) => { notify = resolve; }); - const bump = () => { + const bump = () => { // [!code highlight] const previous = notify; changed = new Promise((resolve) => { notify = resolve; @@ -124,26 +124,26 @@ export function subscribeInbox( }; const done = (async () => { try { - for await (const value of source) { + for await (const value of source) { // [!code highlight] if (isEnd(value)) break; - buffered.push(value); - bump(); + buffered.push(value); // [!code highlight] + bump(); // [!code highlight] } - } finally { + } finally { // [!code highlight] // Runs on the end marker, when the source completes on its own, and // when it throws, so `wait()` always unblocks once the source stops. - ended = true; - bump(); + ended = true; // [!code highlight] + bump(); // [!code highlight] } })(); return { done, - drain: () => buffered.splice(0), + drain: () => buffered.splice(0), // [!code highlight] get ended() { return ended; }, - async wait() { - if (buffered.length === 0 && !ended) await changed; + async wait() { // [!code highlight] + if (buffered.length === 0 && !ended) await changed; // [!code highlight] }, }; } @@ -178,12 +178,12 @@ export async function jobConsumer(queueId: string) { while (true) { await inbox.wait(); // suspends durably while nothing is buffered // [!code highlight] const jobs = inbox.drain(); // [!code highlight] - if (jobs.length === 0) { - if (inbox.ended) break; + if (jobs.length === 0) { // [!code highlight] + if (inbox.ended) break; // [!code highlight] continue; } // Everything that arrived since the last drain is processed together. - await processJobs(jobs); + await processJobs(jobs); // [!code highlight] } } ``` From 3fbf42fc2bdabd6309b8dbd74fc97c9a139cd9f3 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 14:33:09 -0700 Subject: [PATCH 6/9] Strip @setup lines at build time; document and test the hand-off ending `// @setup` marks type-check-only declarations in docs code samples (#846), originally hidden client-side by a custom CodeBlock that the geistdocs migration (#2222) dropped, so every marked line has rendered since. A remark plugin now removes them once at build time, whole multi-line declarations included, covering the rendered page and the processed-markdown exports. The Background Hook Subscriber recipe gains the hand-off ending: when the loop exits on its own, release the hook, commit the release with a step, drain once more, and start a successor run with anything left. The new `handoffInboxWorkflow` e2e test drives it with a sender that runs until the parent completes and follows the successor chain. That test found a runtime gap the pattern cannot close: `resumeHook` is accepted until `hook_disposed` commits at the next suspension, but `hook.dispose()` stops the in-memory iterator immediately, so a payload that lands in between is buffered with no consumer and lost (2 of 7 postgres runs, 3 of 13 world-local runs, always exactly one payload sitting between the last turn's `step_completed` and `hook_disposed`). The test asserts today's guarantee and bounds the loss by counting those log positions; the docs say so in a callout. Closing it means `dispose()` keeping delivery of payloads that precede the disposal event, which changes its documented semantics and is left as a proposal. Co-Authored-By: Claude Fable 5.1 --- .../v5/cookbook/agent-patterns/hook-inbox.mdx | 2 +- .../background-hook-subscriber.mdx | 48 ++++- docs/lib/remark-strip-setup-lines.ts | 86 +++++++++ docs/source.config.ts | 9 +- packages/core/e2e/e2e.test.ts | 173 ++++++++++++++++++ workbench/example/workflows/99_e2e.ts | 79 ++++++++ 6 files changed, 394 insertions(+), 3 deletions(-) create mode 100644 docs/lib/remark-strip-setup-lines.ts diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx index 934f61930b..f00058a495 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -208,7 +208,6 @@ declare function mergeAsyncIterables( ): AsyncIterable<{ index: number; value: T }> & { add: (source: AsyncIterable) => number; }; // @setup -// subscribeInbox() is defined in the Background Hook Subscriber recipe. declare function subscribeInbox( source: AsyncIterable, isEnd: (value: T) => boolean @@ -262,6 +261,7 @@ export async function slackSession(sessionId: string, channel: string) { - **Coalescing**: if several steering messages arrive during one turn, combine them into one user message before the next model call. - **Idle timeout**: race the subscription against `sleep("30m")` and end the session when nothing arrives. - **Cancellation**: pair an `end` message with an `AbortController` so the in-flight turn stops early. See [Agent Cancellation](/cookbook/agent-patterns/agent-cancellation). +- **Ending without dropping messages**: when the loop exits on its own, release the hook, commit the release with a step, drain once more, and hand anything left to a fresh run. See [Handing off late arrivals](/docs/cookbook/common-patterns/background-hook-subscriber#handing-off-late-arrivals-to-a-new-run). - **Deterministic tokens**: derive tokens from ids the sender already has (run ID, session ID, Slack thread ts) so senders never need a lookup. See [Token design](/docs/foundations/hooks#token-design). ## Key APIs diff --git a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx index 4c05957f47..462aa16cb6 100644 --- a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -193,10 +193,56 @@ export async function jobConsumer(queueId: string) { There are two ways a subscription ends, and both are fine: - **An end marker.** The subscriber `break`s when it sees a payload it recognizes as the end (`stop`, `end: true`). Use this when the workflow should wait for the sender to say it is done, and `await` the subscription before returning. -- **The body returns first.** The subscriber is still parked on `await hook`. The run completes regardless; the pending await is abandoned with the run. Use `using` (or call `hook.dispose()`) so the token is released for the next run. +- **The body returns first.** The subscriber is still parked on `await hook`. The run completes regardless; the pending await is abandoned with the run. Use `using` (or call `hook.dispose()`) so the token is released for the next run. If payloads can arrive right up to the end and none may be lost, hand them off as shown next. What you should not do is read the buffer from code that has no durable position, such as a timer callback: only reads that happen right after an `await` on a step, a sleep, or the inbox's `wait()` are anchored to the event log. +### Handing off late arrivals to a new run + +When the body returns first, payloads that arrived after its last read but before the hook was released would finish with the run. If those must not be lost, release the hook explicitly, commit the release with a step, drain once more, and start a fresh run with whatever was left: + +```typescript lineNumbers +import { createHook } from "workflow"; + +type Message = { text: string }; + +declare function runTurn(steering: string[]): Promise<{ finished: boolean }>; // @setup +declare function commitRelease(): Promise; // @setup +declare function startSuccessor(sessionId: string, carried: string[]): Promise; // @setup + +export async function session(sessionId: string, carried: string[]) { + "use workflow"; + + const inbox: Message[] = []; + const hook = createHook({ token: `session:${sessionId}` }); + void (async () => { + for await (const message of hook) inbox.push(message); + })(); + + let steering = carried; + while (true) { + const result = await runTurn(steering); + steering = inbox.splice(0).map((m) => m.text); + if (result.finished && steering.length === 0) break; + } + + hook.dispose(); // senders now get HookNotFoundError for this token // [!code highlight] + await commitRelease(); // any step: commits the release and anchors the drain below // [!code highlight] + const late = [...steering, ...inbox.splice(0).map((m) => m.text)]; // [!code highlight] + if (late.length > 0) { + // Whatever beat the release is handed to a new run instead of dropped. + return { successor: await startSuccessor(sessionId, late) }; // [!code highlight] + } + return { successor: null }; +} +``` + +Two details matter here. The drain happens *after* an awaited step, not right after `dispose()`, so it is anchored to the event log like every other read. And the successor is started from a step (`startSuccessor` wraps `start()` from `workflow/api`), so it happens once, on the live run. A sender whose `resumeHook()` lands after the release is committed sees `HookNotFoundError` rather than a silent drop, and can start or look up the successor itself. + + +There is a window this pattern cannot close today. `dispose()` stops the subscriber's iterator immediately, but the release only becomes durable at the run's next suspension, and `resumeHook()` keeps being accepted until then. A payload acknowledged in that window is in the event log ahead of the release but has no consumer left to claim it, so it is neither processed nor handed off. In the e2e suite this window is a few milliseconds wide and catches roughly one payload in five runs under a 40 ms send cadence. Closing it needs the runtime to keep delivering payloads that precede `hook_disposed`; until then, treat the hand-off as best effort for messages sent in the last instant before the run ends. + + ## How it works 1. **The subscriber is ordinary workflow code.** `for await (const m of hook)` awaits the hook once per payload. The runtime does not treat it specially; it is a promise chain the body chose not to await. diff --git a/docs/lib/remark-strip-setup-lines.ts b/docs/lib/remark-strip-setup-lines.ts new file mode 100644 index 0000000000..5e8bdbf64d --- /dev/null +++ b/docs/lib/remark-strip-setup-lines.ts @@ -0,0 +1,86 @@ +/** + * Remove `// @setup` lines from fenced code blocks. + * + * Docs samples are type-checked as written (packages/docs-typecheck), so an + * example that calls a step it does not define declares it inline: + * + * declare function processRecord(record: Record): Promise; // @setup + * + * Those declarations exist for the type checker, not the reader. This plugin + * drops every line carrying the marker (and any blank lines it leaves at the + * top of the block) before rendering, so neither the highlighted page nor the + * processed-markdown export (llms.txt, `.md` routes, copy-page) shows them. + * + * The marker was originally hidden client-side in a custom CodeBlock + * (#846); that component did not survive the geistdocs migration (#2222), + * which is why this runs in the remark stage instead: one pass, server-side, + * covering every consumer of the processed content. + */ +const SETUP_LINE = /\/\/\s*@setup\b/; +// A line that begins a statement. Used to find the start of a multi-line +// declaration whose marker sits on its last line: +// +// declare function subscribeInbox( +// source: AsyncIterable +// ): { drain: () => T[] }; // @setup +const STATEMENT_START = + /^(declare|type|interface|import|export|const|let|var|function|class|enum|abstract)\b/; + +function stripSetupLines(value: string): string { + if (!SETUP_LINE.test(value)) return value; + const lines = value.split('\n'); + const drop = new Array(lines.length).fill(false); + for (let i = 0; i < lines.length; i++) { + if (!SETUP_LINE.test(lines[i])) continue; + drop[i] = true; + // Walk back to the line that opened this statement, so a marker on the + // closing line of a multi-line declaration removes all of it. Stop at a + // blank line or another dropped line: the statement cannot span those. + let j = i; + while ( + j > 0 && + !STATEMENT_START.test(lines[j].trim()) && + lines[j - 1].trim() !== '' && + !drop[j - 1] + ) { + j--; + drop[j] = true; + } + } + const kept = lines.filter((_, i) => !drop[i]); + // Drop blank lines left at the top of the block (typically the separator + // between the declarations and the example proper), and collapse any run + // of blank lines a removed declaration left in the middle. + let start = 0; + while (start < kept.length && kept[start].trim() === '') start++; + const out: string[] = []; + for (const line of kept.slice(start)) { + if ( + line.trim() === '' && + out.length > 0 && + out[out.length - 1].trim() === '' + ) { + continue; + } + out.push(line); + } + return out.join('\n'); +} + +/** Minimal structural view of an mdast tree; avoids depending on @types/mdast. */ +type Node = { type: string; value?: string; children?: Node[] }; + +function walk(node: Node): void { + if (node.type === 'code' && typeof node.value === 'string') { + node.value = stripSetupLines(node.value); + } + if (node.children) for (const child of node.children) walk(child); +} + +export function remarkStripSetupLines() { + return (tree: Node) => { + walk(tree); + }; +} + +export { stripSetupLines }; diff --git a/docs/source.config.ts b/docs/source.config.ts index d1245665da..cb8a5629c5 100644 --- a/docs/source.config.ts +++ b/docs/source.config.ts @@ -5,6 +5,7 @@ import { } from '@vercel/geistdocs/source-config'; import { defineDocs } from 'fumadocs-mdx/config'; import { z } from 'zod'; +import { remarkStripSetupLines } from './lib/remark-strip-setup-lines'; // You can customise Zod schemas for frontmatter and `meta.json` here // see https://fumadocs.dev/docs/mdx/collections @@ -71,4 +72,10 @@ export const worldsV5Docs = defineDocs({ }, }); -export default defineGeistdocsSourceConfig(); +export default defineGeistdocsSourceConfig({ + mdxOptions: { + // Drop `// @setup` type-check-only lines from code samples before they + // are rendered or exported as markdown. See lib/remark-strip-setup-lines.ts. + remarkPlugins: [remarkStripSetupLines], + }, +}); diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 737d121301..6d363e68af 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1638,6 +1638,179 @@ describe.concurrent('e2e', () => { } ); + test( + 'handoffInboxWorkflow - late arrivals after the loop exits are handed to a new run; loss is bounded to the disposal window', + { timeout: 300_000 }, + async () => { + const token = `inbox-handoff-${Math.random().toString(36).slice(2)}`; + const TURNS = 4; + + const parent = await start(await e2e('handoffInboxWorkflow'), [ + token, + TURNS, + [], + 0, + ]); + await waitForHook(token, { runId: parent.runId }); + + // Send steadily until the parent run completes. A send that lands in + // the disposal window is rejected with HookNotFoundError; that is the + // contract (the sender knows, and can start or find the next run), so + // count it and retry the same seq against the successor's hook. + // Bounded so a starved lane cannot turn this sender into a load + // generator: the parent runs TURNS * 300 ms of steps, so the cap is + // far above what a healthy run can absorb before it completes. + const MAX_MESSAGES = 60 * SCALE; + const accepted: number[] = []; + let rejections = 0; + let parentDone = false; + const parentResult = parent.returnValue.then((r) => { + parentDone = true; + return r; + }); + // Once the last turn has started, send a burst without pausing so + // some payloads land during that turn and there is something to hand + // off. Sends stay strictly sequential so log order is send order. + let lastTurnStarted = false; + void waitForRunEvents( + parent.runId, + (event) => event.eventType === 'step_completed', + { minCount: TURNS - 1, timeoutMs: 240_000 } + ).then( + () => { + lastTurnStarted = true; + }, + () => {} + ); + let burst = 0; + let seq = 0; + while (!parentDone && seq < MAX_MESSAGES) { + try { + await resumeHook(token, { seq }); + accepted.push(seq++); + } catch (err) { + if (!HookNotFoundError.is(err)) throw err; + rejections++; + } + if (lastTurnStarted && burst < 5) { + burst++; + continue; + } + await sleep(40); + } + + // Follow the chain of hand-offs. The sender stopped when the parent + // finished, so the chain is short, but every generation is checked. + const generations = [await parentResult]; + let next = generations[0].childRunId; + while (next) { + const child = await getRun<(typeof generations)[0]>(next).returnValue; + generations.push(child); + next = child.childRunId; + } + + // Always print the shape: on a failure this is what says which + // generation lost or duplicated a payload. + console.log( + `[handoff] accepted=${JSON.stringify(accepted)} rejectedInWindow=${rejections} generations=${JSON.stringify( + generations.map((g) => ({ + generation: g.generation, + runTurns: g.turns.map((t) => t.steeringSeen), + late: g.late, + })) + )}` + ); + + // When there was something to hand off, the child started with + // exactly the parent's late drain. (Whether anything arrived in the + // last turn is timing; the burst above makes it the common case, and + // the log line records when it did not happen.) + if (generations[0].late.length > 0) { + expect(generations).toHaveLength(2); + expect(generations[1].turns[0].steeringSeen).toEqual( + generations[0].late + ); + } else { + expect(generations).toHaveLength(1); + console.warn( + '[handoff] no late arrivals; hand-off path not exercised' + ); + } + + // Every generation: what the workflow observed at each turn matches + // what the log recorded as that turn's step arguments. + for (const g of generations) { + expect(g.turns).toHaveLength(TURNS); + for (const turn of g.turns) { + expect( + turn.steeringApplied, + `gen ${g.generation} turn ${turn.turn}` + ).toEqual(turn.steeringSeen); + } + } + + // Accounting. Every processed payload was acknowledged, each exactly + // once, in send order, and nothing was left in the final run. + const processed = generations.flatMap((g) => + g.turns.flatMap((t) => t.steeringSeen) + ); + const last = generations[generations.length - 1]; + expect(last.late).toEqual([]); + expect(new Set(processed).size).toBe(processed.length); + expect(processed).toEqual( + accepted.filter((seq) => processed.includes(seq)) + ); + + // KNOWN GAP (runtime, not this pattern): `hook.dispose()` stops the + // in-memory iterator at once, but the world keeps accepting resumes + // until `hook_disposed` commits at the run's next suspension. A payload + // acknowledged in that window sits in the log before `hook_disposed` + // with no consumer left to claim it, and is lost. Until the runtime + // delivers pre-disposal payloads, the only payloads that may go missing + // are those, so bound the loss by counting them in the parent's log: + // `hook_received` events after the last turn's `step_completed` and + // before `hook_disposed`. + const parentEvents = await listAllRunEvents(parent.runId); + let turnsCompleted = 0; + let inWindow = 0; + for (const event of parentEvents) { + if (event.eventType === 'step_completed') turnsCompleted++; + else if (event.eventType === 'hook_disposed') break; + else if ( + event.eventType === 'hook_received' && + turnsCompleted === TURNS + ) { + inWindow++; + } + } + const dropped = accepted.filter((seq) => !processed.includes(seq)); + console.log( + `[handoff] dropped=${JSON.stringify(dropped)} disposalWindowPayloads=${inWindow}` + ); + expect(dropped.length).toBeLessThanOrEqual(inWindow); + // Anything dropped sits exactly at the hand-off seam: newer than + // everything the parent processed or handed off, older than everything + // the successor processed. Never a hole inside either generation. + const parentNewest = Math.max( + -1, + ...generations[0].turns.flatMap((t) => t.steeringSeen), + ...generations[0].late + ); + // A successor's first turn is the carried set (the parent's late + // drain), so its own payloads start at turn 1. + const childOldest = Math.min( + Number.POSITIVE_INFINITY, + ...generations + .slice(1) + .flatMap((g) => g.turns.slice(1).flatMap((t) => t.steeringSeen)) + ); + for (const seq of dropped) { + expect(seq).toBeGreaterThan(parentNewest); + expect(seq).toBeLessThan(childOldest); + } + } + ); + test( 'mergedHooksWorkflow - merged iterator preserves event-log order across hooks', { timeout: 240_000 }, diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index e4e996ef4f..13ab143eeb 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -4231,3 +4231,82 @@ export async function dynamicInboxWorkflow( thread.dispose(); return { threadId, received }; } + +async function commitInboxDisposalStep(generation: number) { + 'use step'; + return generation; +} + +async function startHandoffRunStep( + token: string, + turns: number, + carried: number[], + generation: number +): Promise { + 'use step'; + const run = await start(handoffInboxWorkflow, [ + token, + turns, + carried, + generation, + ]); + return run.runId; +} + +/** + * Pattern 1 with the "hand off" ending: the turn loop exits on its own (no + * end marker), the hook is disposed so the token is released, a step commits + * the disposal, and only then is the inbox drained one last time. Anything + * that landed between the last turn and the disposal is handed to a fresh run + * that starts with those messages as its first steering, instead of being + * dropped with the finished run. + * + * Two things are under test. Ordering: the late drain must be anchored after + * the committing step so a replay sees the same late set (a drain right after + * `dispose()` can race the subscriber's own delivery chain). Accounting: every + * payload the sender got an acknowledgement for is processed by exactly one + * generation, and a sender that hits the disposal window sees + * `HookNotFoundError` rather than a silent drop. + */ +export async function handoffInboxWorkflow( + token: string, + turns: number, + carried: number[], + generation: number +) { + 'use workflow'; + + const inbox: InboxMessage[] = []; + const hook = createHook({ token }); + void (async () => { + for await (const message of hook) inbox.push(message); + })(); + + const turnLog: { + turn: number; + steeringSeen: number[]; + steeringApplied: number[]; + }[] = []; + let steering = carried; + for (let turn = 0; turn < turns; turn++) { + const result = await inboxTurnStep(turn, steering); + turnLog.push({ + turn, + steeringSeen: steering, + steeringApplied: result.steering, + }); + steering = inbox.splice(0).map((m) => m.seq); + } + + // Done with this run: release the token, commit the release, then look at + // what arrived in the meantime. + hook.dispose(); + await commitInboxDisposalStep(generation); + const late = [...steering, ...inbox.splice(0).map((m) => m.seq)]; + + let childRunId: string | null = null; + if (late.length > 0) { + childRunId = await startHandoffRunStep(token, turns, late, generation + 1); + } + return { generation, turns: turnLog, late, childRunId }; +} From 8fe31b3604bcfdd7e7bba26fed0241c998d6d35e Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 14:45:30 -0700 Subject: [PATCH 7/9] Follow hand-off chains of any length and bound the late/carried divergence On Vercel the return value is polled every 5s, so the sender outlives the parent and the successor hands off again; the test now follows the chain instead of asserting two generations. The lanes also showed the disposal window from the other side: the parent returned late=[2,3,4] while its successor was started with [2,3], because the successor's arguments were recorded on the live retained-VM run and the return value came from a fresh replay that did deliver the window payload. The test asserts the carried set is a prefix of the parent's late and bounds the excess, like the drops, by the window payloads in each generation's log. The docs callout says late is not an exact record of what the successor received until dispose() keeps delivering pre-disposal payloads. Co-Authored-By: Claude Fable 5.1 --- .../background-hook-subscriber.mdx | 2 +- packages/core/e2e/e2e.test.ts | 115 ++++++++++-------- 2 files changed, 63 insertions(+), 54 deletions(-) diff --git a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx index 462aa16cb6..81ce60fd55 100644 --- a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -240,7 +240,7 @@ export async function session(sessionId: string, carried: string[]) { Two details matter here. The drain happens *after* an awaited step, not right after `dispose()`, so it is anchored to the event log like every other read. And the successor is started from a step (`startSuccessor` wraps `start()` from `workflow/api`), so it happens once, on the live run. A sender whose `resumeHook()` lands after the release is committed sees `HookNotFoundError` rather than a silent drop, and can start or look up the successor itself. -There is a window this pattern cannot close today. `dispose()` stops the subscriber's iterator immediately, but the release only becomes durable at the run's next suspension, and `resumeHook()` keeps being accepted until then. A payload acknowledged in that window is in the event log ahead of the release but has no consumer left to claim it, so it is neither processed nor handed off. In the e2e suite this window is a few milliseconds wide and catches roughly one payload in five runs under a 40 ms send cadence. Closing it needs the runtime to keep delivering payloads that precede `hook_disposed`; until then, treat the hand-off as best effort for messages sent in the last instant before the run ends. +There is a window this pattern cannot close today. `dispose()` stops the subscriber's iterator immediately, but the release only becomes durable at the run's next suspension, and `resumeHook()` keeps being accepted until then. A payload acknowledged in that window is in the event log ahead of the release but has no consumer left to claim it, so on the live run it is neither processed nor handed off, while a fresh replay of the same log does deliver it and reports it in `late`. In the e2e suite this window is a few milliseconds wide and catches roughly one payload in five runs under a 40 ms send cadence. Closing it needs the runtime to keep delivering payloads that precede `hook_disposed` after `dispose()`; until then, treat the hand-off as best effort for messages sent in the last instant before the run ends, and do not read `late` back as an exact record of what the successor received. ## How it works diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 6d363e68af..22ad49dca9 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1721,22 +1721,6 @@ describe.concurrent('e2e', () => { )}` ); - // When there was something to hand off, the child started with - // exactly the parent's late drain. (Whether anything arrived in the - // last turn is timing; the burst above makes it the common case, and - // the log line records when it did not happen.) - if (generations[0].late.length > 0) { - expect(generations).toHaveLength(2); - expect(generations[1].turns[0].steeringSeen).toEqual( - generations[0].late - ); - } else { - expect(generations).toHaveLength(1); - console.warn( - '[handoff] no late arrivals; hand-off path not exercised' - ); - } - // Every generation: what the workflow observed at each turn matches // what the log recorded as that turn's step arguments. for (const g of generations) { @@ -1749,13 +1733,34 @@ describe.concurrent('e2e', () => { } } + // Chain continuity. A successor's first turn is the set its parent + // handed off, as recorded in the step that started it. The parent's + // RETURNED `late` may be longer than that by payloads that fell in the + // disposal window (see below): the successor was started on the live + // run, while the return value comes from whichever replay committed + // it, and only a fresh replay delivers window payloads. So `carried` + // must be a prefix of `late`, and the excess is counted as divergence. + let divergence = 0; + for (let i = 0; i + 1 < generations.length; i++) { + const late = generations[i].late; + const carried = generations[i + 1].turns[0].steeringSeen; + expect(carried.length).toBeGreaterThan(0); + expect(late.slice(0, carried.length)).toEqual(carried); + divergence += late.length - carried.length; + } + const last = generations[generations.length - 1]; + expect(last.childRunId).toBeNull(); + if (generations.length === 1) { + console.warn( + '[handoff] no late arrivals; hand-off path not exercised' + ); + } + // Accounting. Every processed payload was acknowledged, each exactly - // once, in send order, and nothing was left in the final run. + // once, in send order. const processed = generations.flatMap((g) => g.turns.flatMap((t) => t.steeringSeen) ); - const last = generations[generations.length - 1]; - expect(last.late).toEqual([]); expect(new Set(processed).size).toBe(processed.length); expect(processed).toEqual( accepted.filter((seq) => processed.includes(seq)) @@ -1765,48 +1770,52 @@ describe.concurrent('e2e', () => { // in-memory iterator at once, but the world keeps accepting resumes // until `hook_disposed` commits at the run's next suspension. A payload // acknowledged in that window sits in the log before `hook_disposed` - // with no consumer left to claim it, and is lost. Until the runtime - // delivers pre-disposal payloads, the only payloads that may go missing - // are those, so bound the loss by counting them in the parent's log: - // `hook_received` events after the last turn's `step_completed` and + // with no consumer left to claim it. On the retained-VM path it is + // lost; a fresh replay delivers it, which is the divergence counted + // above. Until the runtime delivers pre-disposal payloads, bound both + // by the number of such payloads in each generation's log: + // `hook_received` events after the TURNS-th `step_completed` and // before `hook_disposed`. - const parentEvents = await listAllRunEvents(parent.runId); - let turnsCompleted = 0; let inWindow = 0; - for (const event of parentEvents) { - if (event.eventType === 'step_completed') turnsCompleted++; - else if (event.eventType === 'hook_disposed') break; - else if ( - event.eventType === 'hook_received' && - turnsCompleted === TURNS - ) { - inWindow++; + const runIds = [ + parent.runId, + ...generations.map((g) => g.childRunId).filter(Boolean), + ] as string[]; + for (const runId of runIds) { + let turnsCompleted = 0; + for (const event of await listAllRunEvents(runId)) { + if (event.eventType === 'step_completed') turnsCompleted++; + else if (event.eventType === 'hook_disposed') break; + else if ( + event.eventType === 'hook_received' && + turnsCompleted === TURNS + ) { + inWindow++; + } } } const dropped = accepted.filter((seq) => !processed.includes(seq)); console.log( - `[handoff] dropped=${JSON.stringify(dropped)} disposalWindowPayloads=${inWindow}` + `[handoff] dropped=${JSON.stringify(dropped)} divergence=${divergence} disposalWindowPayloads=${inWindow}` ); expect(dropped.length).toBeLessThanOrEqual(inWindow); - // Anything dropped sits exactly at the hand-off seam: newer than - // everything the parent processed or handed off, older than everything - // the successor processed. Never a hole inside either generation. - const parentNewest = Math.max( - -1, - ...generations[0].turns.flatMap((t) => t.steeringSeen), - ...generations[0].late - ); - // A successor's first turn is the carried set (the parent's late - // drain), so its own payloads start at turn 1. - const childOldest = Math.min( - Number.POSITIVE_INFINITY, - ...generations - .slice(1) - .flatMap((g) => g.turns.slice(1).flatMap((t) => t.steeringSeen)) - ); - for (const seq of dropped) { - expect(seq).toBeGreaterThan(parentNewest); - expect(seq).toBeLessThan(childOldest); + expect(divergence).toBeLessThanOrEqual(inWindow); + // A dropped payload sits at a hand-off seam or the tail, never inside + // the span a single generation processed. + for (const g of generations) { + // A successor's turn 0 is its parent's hand-off; its own span + // starts at turn 1, so the seam between them is not "inside". + const own = g.generation === 0 ? g.turns : g.turns.slice(1); + const span = own.flatMap((t) => t.steeringSeen); + if (span.length === 0) continue; + const lo = Math.min(...span); + const hi = Math.max(...span); + for (const seq of dropped) { + expect( + seq > lo && seq < hi, + `seq ${seq} inside gen ${g.generation}` + ).toBe(false); + } } } ); From cfbb8672313a5cdfa0c038cfa6e1b7e0cfa9f5f3 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 15:05:41 -0700 Subject: [PATCH 8/9] Deliver payloads that precede hook_disposed after dispose(), in both engines `resumeHook` is accepted until `hook_disposed` commits at the run's next suspension, but `hook.dispose()` stopped the in-memory iterator immediately. A payload landing in between sat in the log ahead of the disposal with no consumer left to claim it: lost on the retained-VM path, delivered by a fresh replay, so the workflow's view was replay-inconsistent too. The hand-off e2e test hit both (2/7 postgres, 3/13 world-local runs; `late=[2,3,4]` returned against a successor started with `[2,3]` on Vercel). Disposal is now the durable event it always was on the world side. After `dispose()`, awaiters and the `for await` iterator keep receiving payloads that precede `hook_disposed`, and iteration ends when that event is observed. node:vm: iterator waiters are tracked with their payload resolvers so the disposal event ends only the idle ones (a waiter already receiving a payload ends on its next pull), settled through promiseQueue for log order; `disposeHook` no longer drops awaiters or schedules a suspension, since a waiter parked at end of log already does. QuickJS: the host marks the hook and resolves a parked iterator with a sentinel when it processes `hook_disposed`; `next()` returns done only once the buffer is drained and the flag is set, and a plain await stays pending on the sentinel. Unit test pins the shape; the hand-off e2e test asserts zero loss and `late == carried` again and logs how many payloads fell in the window. dispose() docs updated in the JSDoc, API reference, foundations page, and the cookbook recipe. Co-Authored-By: Claude Fable 5.1 --- .changeset/hook-inbox-e2e.md | 2 +- .../v5/api-reference/workflow/create-hook.mdx | 2 +- .../background-hook-subscriber.mdx | 6 +- docs/content/docs/v5/foundations/hooks.mdx | 2 +- packages/core/e2e/e2e.test.ts | 71 +++-------- packages/core/src/create-hook.ts | 6 +- packages/core/src/runtime/quickjs-runtime.ts | 50 ++++++-- packages/core/src/workflow.test.ts | 86 +++++++++++++ packages/core/src/workflow/hook.ts | 113 ++++++++++++++++-- 9 files changed, 257 insertions(+), 81 deletions(-) diff --git a/.changeset/hook-inbox-e2e.md b/.changeset/hook-inbox-e2e.md index 7f6563efe3..e434130d8a 100644 --- a/.changeset/hook-inbox-e2e.md +++ b/.changeset/hook-inbox-e2e.md @@ -2,4 +2,4 @@ '@workflow/core': patch --- -Add e2e coverage for background hook subscribers and merged hook inboxes: a `for await` over a hook that the workflow body never awaits, several hooks merged into one async iterator (including a hook added mid-run), and a drain-then-wait session loop, each checked for event-log-ordered delivery across replays under dozens to hundreds of payloads. +`hook.dispose()` no longer stops delivery on the spot. The release becomes durable at the run's next suspension, and `resumeHook()` is accepted until then, so a payload accepted in between sits in the event log ahead of `hook_disposed`; it is now delivered to awaiters and `for await` consumers, whose iteration ends once the disposal event is observed. Previously such a payload was acknowledged and then lost on the retained-VM path (and made the workflow's view replay-inconsistent). Also adds e2e coverage for background hook subscribers, merged hook inboxes (including a hook added mid-run), the drain-then-wait session loop, and the dispose-and-hand-off ending. diff --git a/docs/content/docs/v5/api-reference/workflow/create-hook.mdx b/docs/content/docs/v5/api-reference/workflow/create-hook.mdx index 8a914d2d09..ba820ea077 100644 --- a/docs/content/docs/v5/api-reference/workflow/create-hook.mdx +++ b/docs/content/docs/v5/api-reference/workflow/create-hook.mdx @@ -240,7 +240,7 @@ export async function handoffWorkflow(channelId: string) { } ``` -After calling `dispose()`, the hook will no longer receive events and its token becomes available for other workflows to use. +The release is committed at the workflow's next suspension, after which the token becomes available for other workflows to use. Payloads accepted before the release is committed are still delivered, to awaiters and to `for await` iteration, which ends once the release is durable, so nothing sent between the `dispose()` call and the release taking effect is dropped. ### Automatic disposal with `using` diff --git a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx index 81ce60fd55..e3aac03bc0 100644 --- a/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -237,10 +237,10 @@ export async function session(sessionId: string, carried: string[]) { } ``` -Two details matter here. The drain happens *after* an awaited step, not right after `dispose()`, so it is anchored to the event log like every other read. And the successor is started from a step (`startSuccessor` wraps `start()` from `workflow/api`), so it happens once, on the live run. A sender whose `resumeHook()` lands after the release is committed sees `HookNotFoundError` rather than a silent drop, and can start or look up the successor itself. +Two details matter here. The drain happens *after* an awaited step, not right after `dispose()`, so it is anchored to the event log like every other read and sees every payload that preceded the release. And the successor is started from a step (`startSuccessor` wraps `start()` from `workflow/api`), so it happens once, on the live run. A sender whose `resumeHook()` lands after the release is committed sees `HookNotFoundError` rather than a silent drop, and can start or look up the successor itself. - -There is a window this pattern cannot close today. `dispose()` stops the subscriber's iterator immediately, but the release only becomes durable at the run's next suspension, and `resumeHook()` keeps being accepted until then. A payload acknowledged in that window is in the event log ahead of the release but has no consumer left to claim it, so on the live run it is neither processed nor handed off, while a fresh replay of the same log does deliver it and reports it in `late`. In the e2e suite this window is a few milliseconds wide and catches roughly one payload in five runs under a 40 ms send cadence. Closing it needs the runtime to keep delivering payloads that precede `hook_disposed` after `dispose()`; until then, treat the hand-off as best effort for messages sent in the last instant before the run ends, and do not read `late` back as an exact record of what the successor received. + +`dispose()` does not stop delivery on the spot. The release becomes durable at the run's next suspension, and `resumeHook()` keeps being accepted until then; every payload accepted before that is in the event log ahead of the release and is still delivered to the subscriber, whose `for await` ends only once the release is durable. That is why the drain after `commitRelease()` is complete: it holds everything that beat the release, on the live run and on every replay. ## How it works diff --git a/docs/content/docs/v5/foundations/hooks.mdx b/docs/content/docs/v5/foundations/hooks.mdx index 61e4d1e54f..cfe4476801 100644 --- a/docs/content/docs/v5/foundations/hooks.mdx +++ b/docs/content/docs/v5/foundations/hooks.mdx @@ -237,7 +237,7 @@ hook.dispose(); // Manually release the token ``` -After disposal, the hook will no longer receive events and the async iterator will stop yielding values. +The release is committed at the workflow's next suspension. Payloads accepted before that are still delivered, and the async iterator stops yielding once the release is durable, so a `for await` loop drains everything that beat the release and then ends on its own. ## Understanding webhooks diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index 22ad49dca9..f93467451b 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1639,7 +1639,7 @@ describe.concurrent('e2e', () => { ); test( - 'handoffInboxWorkflow - late arrivals after the loop exits are handed to a new run; loss is bounded to the disposal window', + 'handoffInboxWorkflow - late arrivals after the loop exits are handed to a new run, none dropped', { timeout: 300_000 }, async () => { const token = `inbox-handoff-${Math.random().toString(36).slice(2)}`; @@ -1733,49 +1733,35 @@ describe.concurrent('e2e', () => { } } - // Chain continuity. A successor's first turn is the set its parent - // handed off, as recorded in the step that started it. The parent's - // RETURNED `late` may be longer than that by payloads that fell in the - // disposal window (see below): the successor was started on the live - // run, while the return value comes from whichever replay committed - // it, and only a fresh replay delivers window payloads. So `carried` - // must be a prefix of `late`, and the excess is counted as divergence. - let divergence = 0; + // Chain continuity: a successor's first turn is exactly the set its + // parent handed off. The parent's returned `late` is what it passed to + // the step that started the successor, so the two agree on every + // replay. for (let i = 0; i + 1 < generations.length; i++) { - const late = generations[i].late; - const carried = generations[i + 1].turns[0].steeringSeen; - expect(carried.length).toBeGreaterThan(0); - expect(late.slice(0, carried.length)).toEqual(carried); - divergence += late.length - carried.length; + expect(generations[i + 1].turns[0].steeringSeen).toEqual( + generations[i].late + ); + expect(generations[i].late.length).toBeGreaterThan(0); } const last = generations[generations.length - 1]; expect(last.childRunId).toBeNull(); + expect(last.late).toEqual([]); if (generations.length === 1) { console.warn( '[handoff] no late arrivals; hand-off path not exercised' ); } - // Accounting. Every processed payload was acknowledged, each exactly - // once, in send order. + // Accounting: every acknowledged payload was processed by exactly one + // generation, in send order. Payloads accepted between the last + // turn's `step_completed` and `hook_disposed` are the ones that used + // to be lost: `dispose()` now keeps delivering everything that + // precedes the disposal event, so they reach the late drain and the + // successor. The count of such payloads is logged so a run that + // exercised the window is recognizable. const processed = generations.flatMap((g) => g.turns.flatMap((t) => t.steeringSeen) ); - expect(new Set(processed).size).toBe(processed.length); - expect(processed).toEqual( - accepted.filter((seq) => processed.includes(seq)) - ); - - // KNOWN GAP (runtime, not this pattern): `hook.dispose()` stops the - // in-memory iterator at once, but the world keeps accepting resumes - // until `hook_disposed` commits at the run's next suspension. A payload - // acknowledged in that window sits in the log before `hook_disposed` - // with no consumer left to claim it. On the retained-VM path it is - // lost; a fresh replay delivers it, which is the divergence counted - // above. Until the runtime delivers pre-disposal payloads, bound both - // by the number of such payloads in each generation's log: - // `hook_received` events after the TURNS-th `step_completed` and - // before `hook_disposed`. let inWindow = 0; const runIds = [ parent.runId, @@ -1794,29 +1780,10 @@ describe.concurrent('e2e', () => { } } } - const dropped = accepted.filter((seq) => !processed.includes(seq)); console.log( - `[handoff] dropped=${JSON.stringify(dropped)} divergence=${divergence} disposalWindowPayloads=${inWindow}` + `[handoff] disposalWindowPayloads=${inWindow} rejectedInWindow=${rejections}` ); - expect(dropped.length).toBeLessThanOrEqual(inWindow); - expect(divergence).toBeLessThanOrEqual(inWindow); - // A dropped payload sits at a hand-off seam or the tail, never inside - // the span a single generation processed. - for (const g of generations) { - // A successor's turn 0 is its parent's hand-off; its own span - // starts at turn 1, so the seam between them is not "inside". - const own = g.generation === 0 ? g.turns : g.turns.slice(1); - const span = own.flatMap((t) => t.steeringSeen); - if (span.length === 0) continue; - const lo = Math.min(...span); - const hi = Math.max(...span); - for (const seq of dropped) { - expect( - seq > lo && seq < hi, - `seq ${seq} inside gen ${g.generation}` - ).toBe(false); - } - } + expect(processed).toEqual(accepted); } ); diff --git a/packages/core/src/create-hook.ts b/packages/core/src/create-hook.ts index 358101c704..eba3965f6d 100644 --- a/packages/core/src/create-hook.ts +++ b/packages/core/src/create-hook.ts @@ -59,7 +59,11 @@ export interface Hook extends AsyncIterable, Thenable { /** * Disposes the hook, releasing its token for reuse by other workflows. * - * After calling `dispose()`, the hook will no longer receive any events. + * The release is committed at the workflow's next suspension, and + * `resumeHook()` keeps being accepted until then. Payloads accepted before + * the release is committed are still delivered to awaiters and to + * `for await` iteration, which ends once the release is durable; nothing + * is dropped between calling `dispose()` and the release taking effect. * This is useful when you want to explicitly release a hook token before * the workflow completes, allowing another workflow to register a hook * with the same token. diff --git a/packages/core/src/runtime/quickjs-runtime.ts b/packages/core/src/runtime/quickjs-runtime.ts index 4cf2700b3d..17e2ea8bdb 100644 --- a/packages/core/src/runtime/quickjs-runtime.ts +++ b/packages/core/src/runtime/quickjs-runtime.ts @@ -282,6 +282,9 @@ globalThis.__workflowError = undefined; // Keyed by correlationId → array of payloads (preserves delivery order). // This mirrors the event-replay runtime's payloadsQueue in hook.ts. globalThis.__hookPayloadBuffer = {}; +// Delivered to a hook's pending resolver by the host when the hook_disposed +// event is processed, so a parked iterator ends. Never a payload value. +globalThis.__HOOK_DISPOSED = { __hookDisposed: true }; // Buffer for step/wait/attr terminal outcomes that arrive before this VM // has constructed the corresponding awaiting promise. In fresh-VM replay @@ -711,6 +714,11 @@ globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")] = function(options) { token: token, created: false, conflict: null, + // Set by the host when the hook_disposed event is processed. Disposal is + // durable, not an in-memory switch: payloads accepted before it commits + // precede it in the log and are still delivered, and iteration ends only + // once this flag is set (mirrors the node:vm engine's hook.ts). + disposed: false, getConflictResolvers: [], }; @@ -751,11 +759,9 @@ globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")] = function(options) { hasCreatedEvent: false, }); } - // If there's a pending resolver, resolve it with undefined to break the iterator - if (globalThis.__resolvers[correlationId]) { - globalThis.__resolvers[correlationId].resolve(undefined); - delete globalThis.__resolvers[correlationId]; - } + // A pending awaiter stays parked: a payload that beat the disposal + // resolves it, and the hook_disposed event ends the iterator (the host + // resolves the resolver with __HOOK_DISPOSED when it processes it). } function getConflict() { @@ -781,7 +787,12 @@ globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")] = function(options) { var hook = { token: token, then: function(onFulfilled, onRejected) { - return createHookPromise().then(onFulfilled, onRejected); + return createHookPromise().then(function(value) { + // The disposal marker ends iterators, not plain awaits: a payload + // await on a disposed hook stays pending, as it always has. + if (value === globalThis.__HOOK_DISPOSED) return new Promise(function() {}); + return value; + }).then(onFulfilled, onRejected); }, getConflict: getConflict, dispose: disposeHook, @@ -790,16 +801,18 @@ globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")] = function(options) { // Symbol.dispose for explicit resource management hook[Symbol.dispose] = disposeHook; - // AsyncIterable — yields payloads until disposed + // AsyncIterable — yields every payload that precedes hook_disposed in the + // log, then ends once that event has been processed. hook[Symbol.asyncIterator] = function() { return { next: function() { - if (isDisposed) { + var buf = globalThis.__hookPayloadBuffer[correlationId]; + var state = globalThis.__hooks[correlationId]; + if ((!buf || buf.length === 0) && state && state.disposed) { return Promise.resolve({ done: true, value: undefined }); } return createHookPromise().then(function(value) { - // If disposed while waiting, signal done - if (isDisposed) return { done: true, value: undefined }; + if (value === globalThis.__HOOK_DISPOSED) return { done: true, value: undefined }; return { done: false, value: value }; }); }, @@ -2429,6 +2442,23 @@ async function processEvents( break; } case 'hook_disposed': { + // The disposal is durable: mark the hook so a later iterator `next()` + // ends, and end a parked iterator now. Payloads that precede this + // event in the log were delivered by the hook_received cases above, + // in log order; nothing can follow it. + vm.evalCode( + `if (globalThis.__hooks && globalThis.__hooks[${cidJs}]) globalThis.__hooks[${cidJs}].disposed = true;` + + `if (globalThis.__resolvers[${cidJs}]) {` + + `globalThis.__resolvers[${cidJs}].resolve(globalThis.__HOOK_DISPOSED);` + + `delete globalThis.__resolvers[${cidJs}];}` + ).dispose(); + { + resolved = true; + let b: number; + do { + b = vm.executePendingJobs(); + } while (b > 0); + } // Disambiguate from the `hook` pending op with the same // correlationId: we want to mark the `hook_dispose` entry. markCreated(vm, cidJs, 'hook_dispose'); diff --git a/packages/core/src/workflow.test.ts b/packages/core/src/workflow.test.ts index aa7acb780a..244acbe89a 100644 --- a/packages/core/src/workflow.test.ts +++ b/packages/core/src/workflow.test.ts @@ -5000,6 +5000,92 @@ describe('runWorkflow', () => { } }); + it('delivers payloads that precede hook_disposed to an iterator after dispose()', async () => { + // `dispose()` releases the token at the NEXT suspension, and resumes are + // accepted until then. A payload that lands in that window sits in the + // log before `hook_disposed`; it must reach a `for await` consumer, and + // iteration must end only when the disposal event is observed. Ending + // iteration at the `dispose()` call instead orphaned such payloads. + const ops: Promise[] = []; + const workflowRun: WorkflowRun = { + runId: 'test-run-123', + workflowName: 'workflow', + status: 'running', + input: await dehydrateWorkflowArguments( + [], + 'wrun_123', + noEncryptionKey, + ops + ), + createdAt: new Date('2024-01-01T00:00:00.000Z'), + updatedAt: new Date('2024-01-01T00:00:00.000Z'), + startedAt: new Date('2024-01-01T00:00:00.000Z'), + deploymentId: 'test-deployment', + }; + const correlationId = 'hook_01HK153X00VFKAJV9XFN9JXXRS'; + const payload = async (message: string) => + dehydrateStepReturnValue({ message }, 'wrun_123', noEncryptionKey, ops); + const events: Event[] = [ + { + eventId: 'event-0', + runId: workflowRun.runId, + eventType: 'hook_created' as const, + correlationId, + eventData: {}, + createdAt: new Date(), + }, + { + eventId: 'event-1', + runId: workflowRun.runId, + eventType: 'hook_received', + correlationId, + eventData: { token: 'test-token', payload: await payload('one') }, + createdAt: new Date(), + }, + { + eventId: 'event-2', + runId: workflowRun.runId, + eventType: 'hook_received', + correlationId, + eventData: { token: 'test-token', payload: await payload('two') }, + createdAt: new Date(), + }, + { + eventId: 'event-3', + runId: workflowRun.runId, + eventType: 'hook_disposed', + correlationId, + eventData: { token: 'test-token' }, + createdAt: new Date(), + }, + ]; + + const result = await runWorkflow( + `const createHook = globalThis[Symbol.for("WORKFLOW_CREATE_HOOK")]; + async function workflow() { + const hook = createHook({ token: 'test-token' }); + const inbox = []; + const subscription = (async () => { + for await (const p of hook) inbox.push(p.message); + })(); + hook.dispose(); + await subscription; + return inbox; + }${getWorkflowTransformCode('workflow')}`, + workflowRun, + events, + noEncryptionKey + ); + expect( + await hydrateWorkflowReturnValue( + result as any, + 'wrun_123', + noEncryptionKey, + ops + ) + ).toEqual(['one', 'two']); + }); + it('should not warn when queue is empty on completion', async () => { const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {}); try { diff --git a/packages/core/src/workflow/hook.ts b/packages/core/src/workflow/hook.ts index 047f940231..448e8b1d10 100644 --- a/packages/core/src/workflow/hook.ts +++ b/packages/core/src/workflow/hook.ts @@ -124,6 +124,17 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { // Queue of promises that resolve to the next hook payload const promises: PromiseWithResolvers[] = []; + // Iterator awaiters parked with no payload buffered. `resolvers` is the + // entry registered in `promises` for the next payload; `waiter` is what + // the iterator awaits. The `hook_disposed` event ends the waiters whose + // `resolvers` are still in `promises` (no payload committed to them); + // one already shifted out is receiving a payload and ends on its next + // pull instead. + const iteratorWaiters: { + resolvers: PromiseWithResolvers; + waiter: PromiseWithResolvers>; + }[] = []; + // Queue of promises that resolve once hook registration is confirmed // (with `null`) or a token conflict is detected (with the conflicting // `Run`). These back the `hook.getConflict()` getter. @@ -452,6 +463,24 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { ctx.invocationsQueue.delete(correlationId); // Mark that the event log confirms disposal happened hasDisposedEvent = true; + // The disposal is durable: no payload can follow it in the log, so + // end iteration for every waiter with no payload committed to it. A + // waiter whose resolvers were already shifted out of `promises` is + // receiving an earlier-in-log payload and ends on its next pull. + // Settled through promiseQueue so the end is ordered after the + // deliveries queued before this event. + const ending = iteratorWaiters.filter((entry) => + promises.includes(entry.resolvers) + ); + for (const entry of ending) { + iteratorWaiters.splice(iteratorWaiters.indexOf(entry), 1); + promises.splice(promises.indexOf(entry.resolvers), 1); + } + ctx.promiseQueue = ctx.promiseQueue.then(() => { + for (const entry of ending) { + entry.waiter.resolve({ done: true, value: undefined }); + } + }); // We're done processing any more events for this hook return EventConsumerResult.Finished; } @@ -545,7 +574,18 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { return resolvers.promise; } - // Helper function to dispose the hook + // Helper function to dispose the hook. + // + // Disposal is a durable event, not an in-memory switch. `resumeHook()` + // keeps being accepted until `hook_disposed` commits at the run's next + // suspension, and every payload accepted before that sits in the log + // ahead of the disposal. Those payloads are still delivered: awaiters and + // the async iterator keep receiving `hook_received` events until the + // `hook_disposed` event is observed, which is what ends iteration. Ending + // it here instead would orphan any payload in that window (buffered with + // no consumer left to claim it) on the retained-VM path while a fresh + // replay of the same log delivered it: a silent drop and a replay + // divergence from one call. function disposeHook(): void { if (isDisposed) { return; // Already disposed, nothing to do @@ -563,18 +603,63 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { queueItem.disposed = true; } - // Drain any pending promises that are waiting for payloads. - // Without this, promises created by `await hook` or the async iterator's - // `yield await this` would hang forever since the event consumer will - // never deliver another hook_received after disposal. - if (promises.length > 0) { - promises.length = 0; - scheduleWorkflowSuspension(ctx); - } + // Awaiters parked on this hook are settled by the log, not by this + // call: a payload that beat the disposal resolves them, and the + // `hook_disposed` event ends the iterator. Both need the disposal + // written, which happens at the next suspension. No suspension is + // scheduled here: an awaiter still parked when the log runs out + // already schedules one (see the null-event branch of the consumer), + // and scheduling from here could fire before a `hook_disposed` that is + // in the log has been consumed. webhookLogger.debug('Hook disposed', { correlationId, token }); } + // Next iterator result: a buffered or future payload, or `done` once the + // disposal event has been observed and no payload precedes it. Fast paths + // settle through `ctx.promiseQueue` so resolution order matches log order. + function nextIteratorResult(): Promise> { + // A buffered payload or a conflict settles through the same path a + // plain `await hook` takes. + if (payloadsQueue.length > 0 || (hasConflict && conflictErrorRef)) { + return createHookPromise().then((value) => ({ done: false, value })); + } + if (hasDisposedEvent) { + const done = withResolvers>(); + ctx.promiseQueue = ctx.promiseQueue.then(() => { + done.resolve({ done: true, value: undefined }); + }); + return done.promise; + } + // Park for the next payload, exactly as `createHookPromise` would, + // but keep hold of the resolvers so `hook_disposed` can tell an idle + // waiter from one already receiving a payload. + const entry = { + resolvers: withResolvers(), + waiter: withResolvers>(), + }; + iteratorWaiters.push(entry); + const drop = () => { + const index = iteratorWaiters.indexOf(entry); + if (index !== -1) iteratorWaiters.splice(index, 1); + }; + entry.resolvers.promise.then( + (value) => { + drop(); + entry.waiter.resolve({ done: false, value }); + }, + (error) => { + drop(); + entry.waiter.reject(error); + } + ); + if (eventLogEmpty) { + scheduleWorkflowSuspension(ctx); + } + promises.push(entry.resolvers); + return entry.waiter.promise; + } + const hook: Hook = { token, @@ -590,10 +675,14 @@ export function createCreateHook(ctx: WorkflowOrchestratorContext) { return createHookPromise().then(onfulfilled, onrejected); }, - // Support `for await (const payload of hook) { … }` syntax + // Support `for await (const payload of hook) { … }` syntax. Iteration + // ends when the `hook_disposed` event is observed, after every payload + // that precedes it in the log has been yielded; see `disposeHook`. async *[Symbol.asyncIterator]() { - while (!isDisposed) { - yield await this; + while (true) { + const result = await nextIteratorResult(); + if (result.done) return; + yield result.value; } }, From bff32adcc0b00c5b35c61998eb073b5a61627c91 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 4 Sep 2026 15:16:16 -0700 Subject: [PATCH 9/9] Merge hooks in resolution order with one next() outstanding per source; add ordering stress tests Two new e2e tests probe the merged inbox where the round-robin test could not: a seeded, bursty cross-hook send order (A,A,A,B,C,C,A,...) delivered while each merged item costs a 400 ms step, so payloads buffer across hooks and a replay must order them at claim time, plus a hook that only ever receives its done and another that closes early. Both compare the replay's merge order with the arguments the live run recorded per step. They failed against the Promise.race helper: replay-deterministic, but draining one hook at a time. Two causes, both in the helper. A race over already-settled promises picks by source order, and pulling a source only when its item is consumed means a payload buffered for hook A cannot be ordered against hook B's until A's previous item is yielded, which degrades to round-robin. The helper now records each source's next() result in a FIFO as it resolves and keeps exactly one next() outstanding per live source, so the runtime's delivery barriers, which resolve pending hook awaits in log order, define the merged order. Same change in the cookbook, with the reasoning. Co-Authored-By: Claude Fable 5.1 --- .../v5/cookbook/agent-patterns/hook-inbox.mdx | 78 +++++++--- packages/core/e2e/e2e.test.ts | 146 ++++++++++++++++++ workbench/example/workflows/99_e2e.ts | 126 ++++++++++++--- 3 files changed, 303 insertions(+), 47 deletions(-) diff --git a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx index f00058a495..5bea91972e 100644 --- a/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -115,6 +115,8 @@ export async function POST( A session may listen on more than one token: its own identity token, a Slack thread, an auth callback. Merging their async iterators gives the workflow one ordered stream to consume. The helper below is plain JavaScript and runs unchanged inside a workflow. Its `add()` lets you merge in a hook after consumption has started. +One detail matters for ordering. The runtime resolves each hook's `await` in event-log order, so the helper records the order in which the sources' `next()` promises **resolve** and yields from that queue. A `Promise.race` over the pending promises would not preserve log order: while the consumer is busy in a step, several payloads settle, and a race over already-settled promises picks by the order the sources were listed, draining one hook before the next. + ```typescript lineNumbers /** One item from the merged stream: the value plus which source produced it. */ export type MergedItem = { index: number; value: T }; @@ -125,35 +127,57 @@ export type MergedItem = { index: number; value: T }; * consumed. */ export function mergeAsyncIterables(initial: AsyncIterable[] = []) { - type Slot = { index: number; result: IteratorResult }; const iterators = new Map>(); - const pending = new Map>(); // [!code highlight] + // Items in the order their source's `next()` RESOLVED. Every source always + // has exactly one `next()` outstanding (see `pull`), so inside a workflow, + // where the runtime resolves pending hook awaits in event-log order, this + // queue fills in log order across hooks. Two things break that and are + // avoided here: `Promise.race` over the pending promises (while the consumer + // is away in a step several settle, and a race over settled promises picks + // by source order), and pulling a source only when its item is consumed (a + // payload buffered for hook A then cannot be ordered against hook B's until + // A's previous item is yielded, which degrades to round-robin). + const ready: { index: number; result: IteratorResult }[] = []; // [!code highlight] + let active = 0; let nextIndex = 0; - // Resolved whenever a source is added, so a consumer blocked in - // `Promise.race` over the current sources re-races with the new one. + // Resolved whenever `ready` gains an item or a source is added, so a + // consumer waiting on an empty queue wakes up. let wake!: () => void; - let woken = new Promise((resolve) => { - wake = () => resolve(null); + let woken = new Promise((resolve) => { + wake = resolve; }); + const signal = () => { + const previous = wake; + woken = new Promise((resolve) => { + wake = resolve; + }); + previous(); + }; const pull = (index: number) => { const iterator = iterators.get(index); if (!iterator) return; - pending.set( // [!code highlight] - index, // [!code highlight] - iterator.next().then((result) => ({ index, result })) // [!code highlight] - ); // [!code highlight] + iterator.next().then( + (result) => { + ready.push({ index, result }); // [!code highlight] + // Keep one `next()` outstanding per live source, whether or not the + // consumer has caught up: that is what lets the runtime order this + // source's next payload against the other sources' by log position. + if (!result.done) pull(index); // [!code highlight] + signal(); // [!code highlight] + }, + (error) => { + ready.push({ index, result: Promise.reject(error) as never }); + signal(); // [!code highlight] + } + ); }; const add = (source: AsyncIterable): number => { // [!code highlight] const index = nextIndex++; - iterators.set(index, source[Symbol.asyncIterator]()); // [!code highlight] + iterators.set(index, source[Symbol.asyncIterator]()); + active++; pull(index); // [!code highlight] - const previousWake = wake; - woken = new Promise((resolve) => { - wake = () => resolve(null); - }); - previousWake(); // [!code highlight] return index; }; @@ -165,21 +189,25 @@ export function mergeAsyncIterables(initial: AsyncIterable[] = []) { add, async *[Symbol.asyncIterator]() { try { - while (pending.size > 0) { - const winner = await Promise.race([...pending.values(), woken]); // [!code highlight] - if (winner === null) continue; // a source was added: re-race // [!code highlight] - const { index, result } = winner; + while (active > 0 || ready.length > 0) { + if (ready.length === 0) { + await woken; + continue; + } + const { index, result } = ready.shift()!; // [!code highlight] + if (result instanceof Promise) { + await result; // rethrows the source's error + } if (result.done) { - pending.delete(index); + active--; iterators.delete(index); continue; } - pull(index); // [!code highlight] yield { index, value: result.value }; // [!code highlight] } } finally { - // A disposed hook never settles its pending `next()`, so this is - // fire-and-forget cleanup. + // Let sources clean up. A disposed hook's iterator never settles a + // pending `next()`, so this is fire-and-forget. for (const iterator of iterators.values()) { void iterator.return?.().catch(() => {}); } @@ -252,7 +280,7 @@ export async function slackSession(sessionId: string, channel: string) { ## How it works 1. **Steering is a buffer read at a turn boundary.** The subscriber pushes payloads as they arrive; the loop reads the buffer right after a step result. The runtime delivers hook payloads and step results in event-log order, so every replay reads the same buffer at the same turn. Passing the drained messages to the step records them as its arguments, so the recorded turn and the replayed inbox always agree. The general mechanics are in [Background Hook Subscriber](/docs/cookbook/common-patterns/background-hook-subscriber#how-it-works). -2. **Merging is `Promise.race` over `hook.next()` calls.** Because each payload resolves in log order, the merged stream is the log order of all participating hooks. `add()` re-races with the new hook included, so a hook created after a step joins the same ordered stream. +2. **Merging is a queue of resolutions.** Each source's `next()` pushes into a FIFO when it resolves, and the runtime resolves hook payloads in log order, so the merged stream is the log order of all participating hooks. `add()` starts pulling from the new hook into the same queue, so a hook created after a step joins the same ordered stream. 3. **A pending subscriber does not block completion.** When the body returns, the run completes even if the subscriber is still awaiting the hook. Dispose hooks with `using` or `hook.dispose()` so the tokens are released for the next session. ## Adapting this diff --git a/packages/core/e2e/e2e.test.ts b/packages/core/e2e/e2e.test.ts index f93467451b..c67daff372 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1873,6 +1873,152 @@ describe.concurrent('e2e', () => { } ); + /** Deterministic PRNG so a failing interleaving can be replayed from its seed. */ + function seededRandom(seed: number) { + let state = seed >>> 0; + return () => { + state = (state * 1664525 + 1013904223) >>> 0; + return state / 0x1_0000_0000; + }; + } + + test( + 'mergedHooksReplayCheckWorkflow - irregular cross-hook interleaving buffered under a slow step keeps log order on replay', + { timeout: 300_000 }, + async () => { + const id = Math.random().toString(36).slice(2); + const tokens = range(3).map((i) => `merge-replay-${id}-${i}`); + const PER_HOOK = 12 * SCALE; + const STEP_DELAY_MS = 400; + // Bursty, irregular source order (A,A,A,B,C,C,A,...), reproducible + // from the seed printed on failure. Sends are strictly sequential so + // the event log records exactly this order. + const seed = Math.floor(Math.random() * 0xffff_ffff); + const random = seededRandom(seed); + const remaining = tokens.map(() => PER_HOOK); + const nextSeq = tokens.map(() => 0); + const order: { source: number; seq: number }[] = []; + while (remaining.some((n) => n > 0)) { + const candidates = remaining.flatMap((n, i) => (n > 0 ? [i] : [])); + const source = candidates[Math.floor(random() * candidates.length)]; + // Runs of the same source are what stress claim order. + const run = 1 + Math.floor(random() * Math.min(4, remaining[source])); + for (let k = 0; k < run; k++) { + order.push({ source, seq: nextSeq[source]++ }); + remaining[source]--; + } + } + + const run = await start(await e2e('mergedHooksReplayCheckWorkflow'), [ + tokens, + STEP_DELAY_MS, + ]); + await Promise.all( + tokens.map((token) => waitForHook(token, { runId: run.runId })) + ); + + // Each merged item costs the run a 400 ms step, while a send takes a + // few ms, so the bulk of these land while the run is parked in a step + // and are consumed as buffered payloads by the next replay. + for (const { source, seq } of order) { + await resumeHook(tokens[source], { seq }); + } + for (const token of tokens) { + await resumeHook(token, { seq: PER_HOOK, done: true }); + } + + const result = await run.returnValue; + const context = `seed=${seed}`; + const fmt = (xs: { source: number; seq: number }[]) => + xs.map((x) => `${'ABC'[x.source]}${x.seq}`).join(' '); + console.log( + `[merge-replay] ${context}\n sent: ${fmt(order)}\n observed: ${fmt(result.items.map((i) => i.observed))}\n recorded: ${fmt(result.items.map((i) => i.recorded))}` + ); + // The merge yielded exactly the log order, not the hook order it + // happened to poll in. + expect( + result.items.map((i) => i.observed), + context + ).toEqual(order); + // Replay agreement: the merge order the final replay observed is the + // order the live run recorded in each step's arguments. + for (const item of result.items) { + expect(item.recorded, context).toEqual(item.observed); + } + // Per-hook order is a consequence, but assert it explicitly too. + for (let source = 0; source < tokens.length; source++) { + expect( + result.items + .filter((i) => i.observed.source === source) + .map((i) => i.observed.seq), + `${context} hook ${source}` + ).toEqual(range(PER_HOOK)); + } + await expectInboxEventLog(run.runId, { + hookReceived: tokens.length * (PER_HOOK + 1), + stepCompleted: tokens.length * PER_HOOK, + }); + } + ); + + test( + 'mergedHooksReplayCheckWorkflow - a silent hook and an early done do not disturb the other sources', + { timeout: 300_000 }, + async () => { + const id = Math.random().toString(36).slice(2); + // Hook 3 never receives a payload, only its `done` at the very end. + const tokens = range(4).map((i) => `merge-silent-${id}-${i}`); + const PER_HOOK = 10 * SCALE; + const EARLY = 0; // closed after a third of its messages + + const run = await start(await e2e('mergedHooksReplayCheckWorkflow'), [ + tokens, + 50, + ]); + await Promise.all( + tokens.map((token) => waitForHook(token, { runId: run.runId })) + ); + + const order: { source: number; seq: number }[] = []; + const earlyCount = Math.floor(PER_HOOK / 3); + for (let seq = 0; seq < PER_HOOK; seq++) { + for (const source of [0, 1, 2]) { + if (source === EARLY && seq >= earlyCount) continue; + await resumeHook(tokens[source], { seq }); + order.push({ source, seq }); + } + if (seq === earlyCount - 1) { + // Hook 0 closes while hooks 1 and 2 keep flowing. + await resumeHook(tokens[EARLY], { seq: earlyCount, done: true }); + } + } + for (const source of [1, 2]) { + await resumeHook(tokens[source], { seq: PER_HOOK, done: true }); + } + // The silent hook is closed last; the run cannot finish before this. + await resumeHook(tokens[3], { seq: 0, done: true }); + + const result = await run.returnValue; + const fmt = (xs: { source: number; seq: number }[]) => + xs.map((x) => `${'ABCD'[x.source]}${x.seq}`).join(' '); + console.log( + `[merge-silent]\n sent: ${fmt(order)}\n observed: ${fmt(result.items.map((i) => i.observed))}\n recorded: ${fmt(result.items.map((i) => i.recorded))}\n closed: ${result.closedInOrder.join(',')}` + ); + expect(result.items.map((i) => i.observed)).toEqual(order); + for (const item of result.items) { + expect(item.recorded).toEqual(item.observed); + } + // Closures arrive in log order: the early hook first, the silent one last. + expect(result.closedInOrder[0]).toBe(EARLY); + expect(result.closedInOrder[result.closedInOrder.length - 1]).toBe(3); + expect(result.closedInOrder).toHaveLength(tokens.length); + await expectInboxEventLog(run.runId, { + hookReceived: order.length + tokens.length, + stepCompleted: order.length, + }); + } + ); + test( 'dynamicInboxWorkflow - a hook added to a merged inbox mid-run joins the same ordered stream', { timeout: 240_000 }, diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 13ab143eeb..a2d9130239 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -3914,39 +3914,62 @@ type MergedItem = { index: number; value: T }; * * This is plain JavaScript (no workflow primitives), so it runs unchanged in * the workflow sandbox. Inside a workflow, each source's `next()` is an `await - * hook`, and the runtime delivers hook payloads in event-log order, so the - * merged order is deterministic across replays. + * hook`, and the runtime resolves hook payloads in event-log order; the merge + * yields items in the order those awaits resolved, so the merged order is the + * log order and is deterministic across replays. */ function mergeAsyncIterables(initial: AsyncIterable[] = []) { - type Slot = { index: number; result: IteratorResult }; const iterators = new Map>(); - const pending = new Map>(); + // Items in the order their source's `next()` RESOLVED. Every source always + // has exactly one `next()` outstanding (see `pull`), so inside a workflow, + // where the runtime resolves pending hook awaits in event-log order, this + // queue fills in log order across hooks. Two things break that and are + // avoided here: `Promise.race` over the pending promises (while the consumer + // is away in a step several settle, and a race over settled promises picks + // by source order), and pulling a source only when its item is consumed (a + // payload buffered for hook A then cannot be ordered against hook B's until + // A's previous item is yielded, which degrades to round-robin). + const ready: { index: number; result: IteratorResult }[] = []; + let active = 0; let nextIndex = 0; - // Resolved whenever a source is added, so a consumer blocked in - // `Promise.race` over the current sources re-races with the new one. + // Resolved whenever `ready` gains an item or a source is added, so a + // consumer waiting on an empty queue wakes up. let wake!: () => void; - let woken = new Promise((resolve) => { - wake = () => resolve(null); + let woken = new Promise((resolve) => { + wake = resolve; }); + const signal = () => { + const previous = wake; + woken = new Promise((resolve) => { + wake = resolve; + }); + previous(); + }; const pull = (index: number) => { const iterator = iterators.get(index); if (!iterator) return; - pending.set( - index, - iterator.next().then((result) => ({ index, result })) + iterator.next().then( + (result) => { + ready.push({ index, result }); + // Keep one `next()` outstanding per live source, whether or not the + // consumer has caught up: that is what lets the runtime order this + // source's next payload against the other sources' by log position. + if (!result.done) pull(index); + signal(); + }, + (error) => { + ready.push({ index, result: Promise.reject(error) as never }); + signal(); + } ); }; const add = (source: AsyncIterable): number => { const index = nextIndex++; iterators.set(index, source[Symbol.asyncIterator]()); + active++; pull(index); - const previousWake = wake; - woken = new Promise((resolve) => { - wake = () => resolve(null); - }); - previousWake(); return index; }; @@ -3958,16 +3981,20 @@ function mergeAsyncIterables(initial: AsyncIterable[] = []) { add, async *[Symbol.asyncIterator]() { try { - while (pending.size > 0) { - const winner = await Promise.race([...pending.values(), woken]); - if (winner === null) continue; // a source was added: re-race - const { index, result } = winner; + while (active > 0 || ready.length > 0) { + if (ready.length === 0) { + await woken; + continue; + } + const { index, result } = ready.shift()!; + if (result instanceof Promise) { + await result; // rethrows the source's error + } if (result.done) { - pending.delete(index); + active--; iterators.delete(index); continue; } - pull(index); yield { index, value: result.value }; } } finally { @@ -4190,6 +4217,61 @@ export async function mergedHooksWorkflow( return received; } +async function slowRecordMergedMessage( + source: number, + seq: number, + delayMs: number +) { + 'use step'; + await new Promise((resolve) => setTimeout(resolve, delayMs)); + return { source, seq }; +} + +/** + * Pattern 2 under buffering pressure, with a replay-agreement check. Every + * merged item goes through a step that sleeps `stepDelayMs`, so a sender that + * keeps going during the step piles payloads up across several hooks; the + * next replay consumes all of them as *buffered* payloads and the merge has to + * order them by log position at claim time, not by which hook it asked first. + * The step echoes its arguments, so `observed` (the replay's view of what the + * merge yielded) can be compared with `recorded` (what the live run passed to + * the step). Each hook is closed by its own `done` marker, whenever that + * arrives; the run returns once every hook is closed. + */ +export async function mergedHooksReplayCheckWorkflow( + tokens: string[], + stepDelayMs: number +) { + 'use workflow'; + + const hooks = tokens.map((token) => createHook({ token })); + const items: { + observed: { source: number; seq: number }; + recorded: { source: number; seq: number }; + }[] = []; + const closedInOrder: number[] = []; + let open = hooks.length; + + for await (const { index, value } of mergeAsyncIterables(hooks)) { + if (value.done) { + closedInOrder.push(index); + open--; + if (open === 0) break; + continue; + } + const observed = { source: index, seq: value.seq }; + const recorded = await slowRecordMergedMessage( + index, + value.seq, + stepDelayMs + ); + items.push({ observed, recorded }); + } + + for (const hook of hooks) hook.dispose(); + return { items, closedInOrder }; +} + async function postFirstReplyStep(sessionId: string) { 'use step'; // A real app would post to Slack here and get the thread timestamp back.