diff --git a/.changeset/hook-inbox-e2e.md b/.changeset/hook-inbox-e2e.md new file mode 100644 index 0000000000..e434130d8a --- /dev/null +++ b/.changeset/hook-inbox-e2e.md @@ -0,0 +1,5 @@ +--- +'@workflow/core': patch +--- + +`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/agent-patterns/hook-inbox.mdx b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx new file mode 100644 index 0000000000..5bea91972e --- /dev/null +++ b/docs/content/docs/v5/cookbook/agent-patterns/hook-inbox.mdx @@ -0,0 +1,301 @@ +--- +title: Hook Inbox & Steering +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: 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 + - /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, 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 + +- **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: 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. + +```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. + +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 }; + +/** + * 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[] = []) { + const iterators = 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 }[] = []; // [!code highlight] + let active = 0; + let nextIndex = 0; + // 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; + }); + const signal = () => { + const previous = wake; + woken = new Promise((resolve) => { + wake = resolve; + }); + previous(); + }; + + const pull = (index: number) => { + const iterator = iterators.get(index); + if (!iterator) return; + 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]()); + active++; + pull(index); // [!code highlight] + return index; + }; + + for (const source of initial) add(source); + + const merged: AsyncIterable> & { + add: (source: AsyncIterable) => number; + } = { + add, + async *[Symbol.asyncIterator]() { + try { + 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) { + active--; + iterators.delete(index); + continue; + } + yield { index, value: result.value }; // [!code highlight] + } + } 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; +} +``` + +### Waiting for the next message + +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 + +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] + // 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); + } + + identity.dispose(); + thread.dispose(); + return { threadId }; +} +``` + +## 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 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 + +- **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). +- **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 + +- [`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/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/common-patterns/background-hook-subscriber.mdx b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx new file mode 100644 index 0000000000..e3aac03bc0 --- /dev/null +++ b/docs/content/docs/v5/cookbook/common-patterns/background-hook-subscriber.mdx @@ -0,0 +1,265 @@ +--- +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[] = []; // [!code highlight] + let ended = false; + let notify = () => {}; + let changed = new Promise((resolve) => { + notify = resolve; + }); + const bump = () => { // [!code highlight] + const previous = notify; + changed = new Promise((resolve) => { + notify = resolve; + }); + previous(); + }; + const done = (async () => { + try { + for await (const value of source) { // [!code highlight] + if (isEnd(value)) break; + buffered.push(value); // [!code highlight] + bump(); // [!code highlight] + } + } 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; // [!code highlight] + bump(); // [!code highlight] + } + })(); + return { + done, + drain: () => buffered.splice(0), // [!code highlight] + get ended() { + return ended; + }, + async wait() { // [!code highlight] + if (buffered.length === 0 && !ended) await changed; // [!code highlight] + }, + }; +} +``` + +### 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) { // [!code highlight] + if (inbox.ended) break; // [!code highlight] + continue; + } + // Everything that arrived since the last drain is processed together. + await processJobs(jobs); // [!code highlight] + } +} +``` + +## 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. 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 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. + + +`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 + +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 dc2ed6dae4..938e5ccfc0 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): Steer an agent loop with messages that arrive mid-turn, and merge several hooks into one ordered inbox ## Common patterns @@ -23,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/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/docs/lib/cookbook-tree.ts b/docs/lib/cookbook-tree.ts index 833372489b..e024eb9ac0 100644 --- a/docs/lib/cookbook-tree.ts +++ b/docs/lib/cookbook-tree.ts @@ -45,11 +45,13 @@ export const slugToCategory: Record = { timeouts: 'common-patterns', idempotency: 'common-patterns', webhooks: 'common-patterns', + 'background-hook-subscriber': 'common-patterns', // Agent Patterns 'durable-agent': 'agent-patterns', 'human-in-the-loop': 'agent-patterns', 'agent-cancellation': 'agent-patterns', + 'hook-inbox': 'agent-patterns', // Integrations 'ai-sdk': 'integrations', @@ -130,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': { @@ -153,6 +163,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: + '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'], + }, // Integrations 'ai-sdk': { 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 ab2abe2b10..c67daff372 100644 --- a/packages/core/e2e/e2e.test.ts +++ b/packages/core/e2e/e2e.test.ts @@ -1382,6 +1382,700 @@ 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; + + // `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 + // 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.length).toBeGreaterThanOrEqual(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: result.turns.length, + }); + } + ); + + 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( + '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)}`; + 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, + })) + )}` + ); + + // 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); + } + } + + // 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++) { + 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 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) + ); + let inWindow = 0; + 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++; + } + } + } + console.log( + `[handoff] disposalWindowPayloads=${inWindow} rejectedInWindow=${rejections}` + ); + expect(processed).toEqual(accepted); + } + ); + + 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, + }); + } + ); + + /** 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 }, + 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. 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, { + 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/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; } }, diff --git a/workbench/example/workflows/99_e2e.ts b/workbench/example/workflows/99_e2e.ts index 587c200215..a2d9130239 100644 --- a/workbench/example/workflows/99_e2e.ts +++ b/workbench/example/workflows/99_e2e.ts @@ -3894,3 +3894,501 @@ 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 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[] = []) { + const iterators = 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 `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; + }); + const signal = () => { + const previous = wake; + woken = new Promise((resolve) => { + wake = resolve; + }); + previous(); + }; + + const pull = (index: number) => { + const iterator = iterators.get(index); + if (!iterator) return; + 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); + return index; + }; + + for (const source of initial) add(source); + + const merged: AsyncIterable> & { + add: (source: AsyncIterable) => number; + } = { + add, + async *[Symbol.asyncIterator]() { + try { + 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) { + active--; + iterators.delete(index); + continue; + } + 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. + * `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, + minMessages = 0 +) { + 'use workflow'; + + const inbox: InboxMessage[] = []; + using hook = createHook({ token }); + + // Background subscriber: intentionally NOT awaited here. + let received = 0; + const subscriber = (async () => { + for await (const message of hook) { + if (message.done) break; + inbox.push(message); + received++; + } + return received; + })(); + + const turnLog: { + turn: number; + steeringSeen: number[]; + steeringApplied: number[]; + }[] = []; + 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({ + 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 () => { + 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; + }, + }; +} + +/** + * 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(); + // 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, + 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 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. + 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 }; +} + +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 }; +}