feat(ai-orchestration): generator-based workflows + orchestrators#542
Open
AlemTuzlak wants to merge 68 commits into
Open
feat(ai-orchestration): generator-based workflows + orchestrators#542AlemTuzlak wants to merge 68 commits into
AlemTuzlak wants to merge 68 commits into
Conversation
Implements yield-helpers for the workflow engine: approve() for human-in-the-loop approval steps, bindAgents() to convert agent/workflow definitions into bound step generators, and retry() (async generator) for fault-tolerant step execution with configurable backoff.
…helpers Implements snapshotState/diffState using fast-json-patch for RFC 6902 JSON Patch diffs, plus emit-events helpers (runStartedEvent, stepStartedEvent, stateSnapshotEvent, approvalRequestedEvent, etc.) that produce StreamChunk values for the workflow SSE stream.
…ue, plug RUN_ERROR runId - resumeWorkflow now calls runStore.set() before runStore.delete() so observers see the finished state - pendingEvents queue moved onto LiveRun so the emit() closure captured during runWorkflow is drained correctly by resumeWorkflow - Both drive loops drain live.pendingEvents at the top of each iteration - runErrorEvent now includes runId in the returned chunk
…xports Implements Tasks 3.1–3.4 and 4.1: defineAgent, defineWorkflow, defineOrchestrator factory functions; toWorkflowSSEResponse SSE helper; and wires up the full public API surface in src/index.ts.
…single entry point
…yield* delegation works
Rename the result helper `ok()` to `succeed()` for clarity. The name `succeed` reads better alongside `fail` and avoids shadowing the `Response.ok` DOM property name in server contexts.
Add `defineRouter(config, fn)` — a phantom-config wrapper that captures generic type parameters from a shared config object so users can extract orchestrator routers as named functions without losing type inference.
Remove `phase: 'scoping' as const` from the orchestrator initialize since the schema default covers it. Extract the orchestrator router using the new `defineRouter` helper to demonstrate zero-cast extraction of a named router function.
Add an `endpoint` option to WorkflowClientOptions (and UseWorkflowOptions) as a mutually exclusive alternative to `connection`. When `endpoint` is provided the client internally POSTs JSON and parses the SSE response, eliminating the inline fetch boilerplate and `as any` cast at every call site.
Replace the 50-line inline fetch+SSE adapter with a single \`endpoint: '/api/workflow'\` (resp. \`/api/orchestration\`) option, removing the last \`as any\` cast in the demo route files.
Add a \`handleWorkflowRequest\` function that encapsulates JSON body
parsing, start-vs-resume-vs-abort dispatch, and SSE response shaping.
Server API routes can now delegate entirely to this helper, eliminating
the \`as { ... }\` cast on the request body and the manual
\`toServerSentEventsResponse\` wiring.
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds a generator-based orchestration engine, durable in-memory run store, WorkflowClient and React hooks, POST-based SSE adapters and routes, example workflows/pages, UI components (timeline, file tree, code highlighting), Shiki utilities, extensive docs, and comprehensive Vitest tests. ChangesAI Orchestration Engine, Client Integration, and Workflow Visualization
Sequence Diagram (high-level run flow): sequenceDiagram
participant Client
participant Server
participant Engine
participant RunStore
Client->>Server: POST /api/workflow (start/runId/abort)
Server->>Engine: runWorkflow(options)
Engine->>RunStore: setRunState / appendStep
Engine->>Client: emit StreamChunk (RUN_STARTED / STEP_* / STATE_*)
Engine->>Agent: invoke agent step (invokeAgent)
Agent-->>Engine: stream chunks + output
alt approval requested
Engine->>Client: CUSTOM approval-requested (stream ends)
Client->>Server: POST resume (approval/signalDelivery)
Server->>Engine: runWorkflow(resume)
end
Engine->>RunStore: deleteRun on finish
Engine-->>Client: RUN_FINISHED
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes
✨ Finishing Touches🧪 Generate unit tests (beta)
|
Contributor
🚀 Changeset Version PreviewNo changeset entries found. Merging this PR will not cause a version bump for any packages. |
|
View your CI Pipeline Execution ↗ for commit 3288d6a
☁️ Nx Cloud last updated this comment at |
@tanstack/ai
@tanstack/ai-anthropic
@tanstack/ai-client
@tanstack/ai-code-mode
@tanstack/ai-code-mode-skills
@tanstack/ai-devtools-core
@tanstack/ai-elevenlabs
@tanstack/ai-event-client
@tanstack/ai-fal
@tanstack/ai-gemini
@tanstack/ai-grok
@tanstack/ai-groq
@tanstack/ai-isolate-cloudflare
@tanstack/ai-isolate-node
@tanstack/ai-isolate-quickjs
@tanstack/ai-ollama
@tanstack/ai-openai
@tanstack/ai-openrouter
@tanstack/ai-orchestration
@tanstack/ai-preact
@tanstack/ai-react
@tanstack/ai-react-ui
@tanstack/ai-solid
@tanstack/ai-solid-ui
@tanstack/ai-svelte
@tanstack/ai-utils
@tanstack/ai-vue
@tanstack/ai-vue-ui
@tanstack/openai-base
@tanstack/preact-ai-devtools
@tanstack/react-ai-devtools
@tanstack/solid-ai-devtools
commit: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 15
🧹 Nitpick comments (14)
examples/ts-react-chat/src/components/DraftPreview.tsx (1)
55-58: ⚡ Quick winAvoid remounting the preview container on every update.
Using
key={bumpKey}remounts the scrollable panel each time content changes, which can reset scroll position/focus during streaming updates.Suggested direction
- <div - key={bumpKey} - className="relative px-6 py-7 max-h-[34rem] overflow-auto anim-log-in" - > + <div + className="relative px-6 py-7 max-h-[34rem] overflow-auto anim-log-in" + data-bump={bumpKey} + >Then trigger pulse via CSS/animation based on
data-bump(or a short-livedisPulsingclass), without remounting the node.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/components/DraftPreview.tsx` around lines 55 - 58, The preview container in DraftPreview currently uses key={bumpKey} which forces a remount and resets scroll/focus; remove the dynamic key and instead add a stable root element (the existing div with className "relative px-6 py-7...") and toggle a data attribute or short-lived CSS class (e.g. data-bump={bumpKey} or isPulsing) on that same div to trigger the pulse animation via CSS/animation, ensuring you update where bumpKey is produced so it sets the attribute/class rather than the key.examples/ts-react-chat/src/components/ArticleModal.tsx (1)
10-21: ⚡ Quick winRefine useEffect dependencies to prevent unnecessary re-runs.
The effect depends on the entire
propsobject, which will cause the keyboard listener and scroll lock to be re-registered whenever the props object identity changes—even if only unrelated props likearticlechange. Since the effect only usesprops.onClose, specify that in the dependency array instead.♻️ Proposed fix
document.body.style.overflow = prev } - }, [props]) + }, [props.onClose])🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/components/ArticleModal.tsx` around lines 10 - 21, The useEffect in ArticleModal registers a keydown listener and toggles body overflow but incorrectly depends on the entire props object, causing unnecessary re-runs; change the dependency array to only include the specific callback used (props.onClose) by referencing the onKey handler and the effect that sets document.body.style.overflow so the listener and scroll lock are only re-registered when props.onClose changes.packages/typescript/ai-orchestration/src/primitives/bind-agents.ts (1)
13-41: 💤 Low valueOptional: collapse the two branches by building the descriptor inline.
The two
function*definitions differ only in thekinddiscriminator and theagent/workflowpayload field. A single closure that switches ondef.__kindreads about as clearly and removes duplicated descriptor wiring if this evolves (e.g., adds tracing fields).♻️ Possible consolidation
for (const [name, def] of Object.entries(agents)) { - if (def.__kind === 'agent') { - bound[name] = function* ( - input: unknown, - ): Generator<StepDescriptor, unknown, unknown> { - const descriptor: StepDescriptor = { - kind: 'agent', - name, - input, - agent: def, - } - const result = yield descriptor - return result - } - } else { - bound[name] = function* ( - input: unknown, - ): Generator<StepDescriptor, unknown, unknown> { - const descriptor: StepDescriptor = { - kind: 'nested-workflow', - name, - input, - workflow: def, - } - const result = yield descriptor - return result - } - } + bound[name] = function* ( + input: unknown, + ): Generator<StepDescriptor, unknown, unknown> { + const descriptor: StepDescriptor = + def.__kind === 'agent' + ? { kind: 'agent', name, input, agent: def } + : { kind: 'nested-workflow', name, input, workflow: def } + return yield descriptor + } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/primitives/bind-agents.ts` around lines 13 - 41, The two generator functions for entries of agents and nested workflows duplicate descriptor construction; refactor the loop that builds bound[name] so a single generator closure creates a StepDescriptor with common fields (kind, name, input) and then sets the specific payload field based on def.__kind (e.g., set descriptor.agent = def for 'agent' or descriptor.workflow = def for 'nested-workflow'), yield the descriptor and return the result; update references to StepDescriptor, agents, bound, and def.__kind in that single generator to remove the duplicated function bodies.packages/typescript/ai-orchestration/src/server/parse-request.ts (1)
9-14: 💤 Low valueDead
abortfield onRawBody.
abortis declared on the parsed body shape but never returned fromparseWorkflowRequest, andWorkflowRequestParamshas no place to forward it. Since abort plumbing is explicitly out of scope for this PR, consider dropping the field fromRawBodyuntil it's wired up, so the parser shape doesn't advertise a capability it doesn't deliver.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/server/parse-request.ts` around lines 9 - 14, RawBody currently includes an unused abort?: boolean field which isn't propagated by parseWorkflowRequest and isn't represented on WorkflowRequestParams; remove abort from the RawBody interface to avoid advertising an unsupported capability, update any related types/usages that reference RawBody (e.g., the RawBody declaration in parse-request.ts) and run type checks to ensure no consumers relied on that field.packages/typescript/ai-orchestration/src/run-store/in-memory.ts (1)
26-26: 💤 Low valuePrefer a runtime-agnostic timer type.
NodeJS.Timeoutadds a dependency on@types/nodefor this utility. UseReturnType<typeof setTimeout>instead, which works uniformly across Node, browsers, and Workers without environment-specific type imports.♻️ Proposed change
- const expirations = new Map<string, NodeJS.Timeout>() + const expirations = new Map<string, ReturnType<typeof setTimeout>>()🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/run-store/in-memory.ts` at line 26, The Map declaration uses NodeJS.Timeout which ties the code to `@types/node`; change the type to a runtime-agnostic one by replacing NodeJS.Timeout with ReturnType<typeof setTimeout> in the declaration of expirations (the const expirations = new Map<string, ...>()), and ensure any places that read/clear timers (e.g., where clearTimeout is called) continue to accept that type without importing Node types.packages/typescript/ai-orchestration/src/engine/invoke-agent.ts (2)
42-47: 💤 Low valueShape detection is positional — a small
incheck on shape (c) is OK but worth a comment.
'stream' in result && 'output' in resultaccepts any object with those keys. The case is narrow (Promises don't have them, async iterables don't either), but a one-line comment noting that ordering matters (shape (c) before (a) before (b)) would help future maintainers who add a new shape.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/engine/invoke-agent.ts` around lines 42 - 47, The shape detection in the result handling block (the `'stream' in result && 'output' in result` check inside invoke-agent.ts that returns { stream: filterInnerRunBoundaries(result.stream), output: result.output.then((o) => parseOutput<T>(agent, o)) }) relies on positional checks and intentionally accepts any object with those keys; add a concise one-line comment immediately above this if explaining that this is a narrow, positional shape check (shape (c)) and must remain ordered before the other shape branches (shape (a) then (b)) so future maintainers understand the rationale and don’t reorder or replace the check.
148-155: 💤 Low valueMinor: hoist
SchemaValidationErrorabove its usage.The class is declared at the bottom but referenced at lines 30 and 122. Works at runtime (calls happen after module init), but conventional ordering improves readability and avoids any future
no-use-before-definelint trips.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/engine/invoke-agent.ts` around lines 148 - 155, Move the exported SchemaValidationError class declaration above any code that constructs or references it so its definition precedes its usages; locate the places in this module where new SchemaValidationError(...) is thrown/checked (the earlier references in the same file) and cut-paste the class block to the top of the file (keeping export class SchemaValidationError ... unchanged) so lint rules like no-use-before-define no longer flag the symbol.packages/typescript/ai-orchestration/src/primitives/retry.ts (1)
14-29: 💤 Low valueOptional: jitter for
exponentialbackoff.For the documented use (retrying agent calls that often hit shared upstream rate limits), unjittered exponential delays cause synchronized retry storms across concurrent runs. Adding ±20% jitter is trivial and standard.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/primitives/retry.ts` around lines 14 - 29, The exponential backoff should include jitter to avoid synchronized retry storms: update computeDelay (the exponential branch) to compute the base exponential delay as currently done (using base and attempt), then apply ±20% random jitter (e.g., multiply by a random factor in [0.8, 1.2]), clamp the final value to <= maxDelayMs and >= 0 and return it (rounded to an integer if desired); keep the existing behavior for the 'none' and 'linear' branches and keep delay(ms: number): Promise<void> as-is.packages/typescript/ai-orchestration/src/engine/state-diff.ts (1)
18-20: 💤 Low value
structuredClonerequires Node ≥ 17 / modern browsers. Consider adding a one-line note about the Node version requirement to the package'senginesfield or README, since it's not currently documented.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/engine/state-diff.ts` around lines 18 - 20, The function snapshotState uses structuredClone which requires Node >=17 / modern browsers; update the package metadata to document this requirement by adding an engines entry (e.g., Node >=17) in this package's package.json and/or add a one-line note to the README mentioning that snapshotState (structuredClone) needs Node ≥17 or a modern browser runtime; reference the snapshotState function and structuredClone in your change so reviewers can see the compatibility note tied to the implementation.packages/typescript/ai-orchestration/src/engine/emit-events.ts (1)
9-122: 💤 Low valueType-safety opportunity: helpers can return specific event types instead of generic
StreamChunkcasts.All event helper functions currently use
as StreamChunkcasts that bypass TypeScript's structural validation. Since all event types (RunStartedEvent,RunFinishedEvent,StepStartedEvent, etc.) are already exported from@tanstack/aiand are members of theAGUIEventunion (whichStreamChunkaliases), you can improve type safety by having each helper return its specific event type instead. This lets TypeScript validate the payload shape at construction without the cast:export function runStartedEvent(args: { runId: string threadId?: string }): RunStartedEvent { // instead of StreamChunk return { type: 'RUN_STARTED', timestamp: Date.now(), runId: args.runId, threadId: args.threadId ?? args.runId, } }The return type is still compatible with
StreamChunkvia the union, so callers see no difference. This catches any future misalignment between the payload and AG-UI's event schemas at the source.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/engine/emit-events.ts` around lines 9 - 122, Change each helper to return its specific AG-UI event type (e.g. runStartedEvent -> RunStartedEvent, runFinishedEvent -> RunFinishedEvent, runErrorEvent -> RunErrorEvent, stepStartedEvent -> StepStartedEvent, stepFinishedEvent -> StepFinishedEvent, stateSnapshotEvent -> StateSnapshotEvent, stateDeltaEvent -> StateDeltaEvent, customEvent -> CustomEvent) instead of the generic StreamChunk and remove the trailing "as StreamChunk" casts; import those concrete types from `@tanstack/ai` so TypeScript validates the payload shape at construction (leave approvalRequestedEvent returning customEvent as-is since it delegates to customEvent). Ensure signatures and returned object shapes match the imported types.packages/typescript/ai-orchestration/tests/engine.smoke.test.ts (1)
98-145: ⚡ Quick winConsider adding a resume smoke test alongside the pause assertion.
The pause behavior is well-covered, but the symmetric path — calling
runWorkflow({ workflow, runId, approval, runStore })after the pause and assertingSTEP_FINISHED(approval)plusRUN_FINISHEDwith the expected output — is the more failure-prone half of approval handling (state restore,pendingApprovalStepIdfinalization, generator.next seed value). A second assertion block on the samestorewould lock that contract.🧪 Sketch
// after the pause assertions: const events2: Array<unknown> = [] for await (const c of runWorkflow({ workflow: wf as any, runId: runStarted.runId, approval: { approved: true }, runStore: store, })) { events2.push(c) } const types2 = events2.map((e) => (e as { type: string }).type) expect(types2).toContain('STEP_FINISHED') expect(types2).toContain('RUN_FINISHED') const finished = events2.find( (e) => (e as { type: string }).type === 'RUN_FINISHED', ) as { output: { ok: boolean } } expect(finished.output).toEqual({ ok: true })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/tests/engine.smoke.test.ts` around lines 98 - 145, Add a symmetric resume smoke test after the existing pause assertions in the 'pauses on approval — stream ends after approval-requested, RUN_FINISHED not emitted' test: call runWorkflow again with the same workflow (wf), runId (from runStarted.runId), approval payload (e.g., { approved: true }) and the same runStore (store), collect emitted events into a new array (events2), then assert that events2 includes STEP_FINISHED and RUN_FINISHED and that the RUN_FINISHED event's output equals the expected { ok: true }; this verifies state restore, pendingApproval finalization, and the generator.next seed value are handled correctly.examples/ts-react-chat/src/routes/api.orchestration.ts (1)
10-10: 💤 Low valueConsider extracting the TTL constant for clarity.
The TTL is specified inline as
60 * 60 * 1000. Consider extracting this to a named constant for improved readability and maintainability.♻️ Suggested refactor
+const RUN_STORE_TTL_MS = 60 * 60 * 1000 // 1 hour + -const runStore = inMemoryRunStore({ ttl: 60 * 60 * 1000 }) +const runStore = inMemoryRunStore({ ttl: RUN_STORE_TTL_MS })🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/routes/api.orchestration.ts` at line 10, The inline TTL literal passed to inMemoryRunStore (60 * 60 * 1000) reduces readability; extract it to a named constant (e.g., RUN_STORE_TTL_MS) near the top of the module and use that constant when constructing runStore so the purpose and units are clear; update any related comments or usages of runStore if needed to reference the constant.packages/typescript/ai-react/src/use-workflow.ts (1)
37-40: 💤 Low valueConsider clarifying the comment about fresh values.
The comment states "Track latest options so callbacks read fresh values" but the
bodyoption is captured at client construction time (line 51) and won't reflect prop changes. Consider clarifying that only the event callbacks (onCustomEvent,onStateChange) read fresh values, while structural options likebodyandconnectionare fixed at client creation.📝 Suggested clarification
- // Track latest options so callbacks read fresh values without recreating - // the client. Mirrors useChat's pattern. + // Track latest options so event callbacks (onCustomEvent, onStateChange) + // read fresh values without recreating the client. Structural options like + // `body` and `connection` are captured at construction. Mirrors useChat's pattern. const optsRef = useRef(opts)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-react/src/use-workflow.ts` around lines 37 - 40, The comment near optsRef (where const optsRef = useRef(opts); optsRef.current = opts) is misleading because some options (like body and connection) are captured at client construction and not updated, while event callbacks do read fresh values; update the comment to state that optsRef is used so callback handlers (onCustomEvent, onStateChange) will see the latest opts via optsRef.current, but structural options such as body and connection are fixed when the client is created (refer to the client construction site where body is passed) and will not change if props update.examples/ts-react-chat/src/components/WorkflowTimeline.tsx (1)
105-109: 💤 Low valueConsider using
replaceAllfor consistent formatting.The current code uses
replace('-', ' · '), which only replaces the first occurrence. IfstepTypecontains multiple dashes (e.g., "agent-call-retry"), only the first will be replaced.♻️ Suggested fix
- {step.stepType.replace('-', ' · ')} + {step.stepType.replaceAll('-', ' · ')}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/components/WorkflowTimeline.tsx` around lines 105 - 109, The JSX rendering in WorkflowTimeline.tsx uses step.stepType.replace('-', ' · ') which only replaces the first dash; update the expression in the span rendering (the usage of step.stepType) to replace all dashes (e.g., use step.stepType.replaceAll('-', ' · ') or step.stepType.split('-').join(' · ')) so multi-dash types like "agent-call-retry" become "agent · call · retry".
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@examples/ts-react-chat/src/components/DraftPreview.tsx`:
- Around line 9-11: The current cast of props.draft to Draft lets malformed
payloads reach code that calls draft.paragraphs.map and crash; update the
DraftPreview component to perform runtime shape checks instead of blind casting:
ensure props.draft is an object and that draft.paragraphs is an Array (and
optionally validate each item has the expected fields) before using .map or
rendering; modify the draft initialization and any other places that access
draft.paragraphs (the draft variable and the rendering logic that iterates
paragraphs) to guard with Array.isArray(draft.paragraphs) (or fallback to an
empty array) so .map is only called on a real array and malformed inputs are
safely handled.
In `@examples/ts-react-chat/src/components/WorkflowTimeline.tsx`:
- Around line 175-177: FailureBlock currently asserts props.result to a specific
shape without runtime checks; replace the unchecked assertion with a defensive
runtime guard: implement a small type guard (e.g., isErrorLike) or inline checks
that verify props.result is an object and that result.error is an object with a
string message before reading result.error.message, otherwise fall back to
JSON.stringify(props.result) or a generic message; update the FailureBlock logic
to use that guard when deriving msg so accessing result.error?.message cannot
throw or return misleading values.
In `@examples/ts-react-chat/src/routes/api.orchestration.ts`:
- Around line 15-23: The POST handler should guard against exceptions from
parseWorkflowRequest and runWorkflow by wrapping their calls in a try/catch:
call parseWorkflowRequest(...) and runWorkflow(...) inside the try, and on
success return toServerSentEventsResponse(stream); in the catch block construct
and return a structured error response (JSON body with an error message and
appropriate status like 400 for bad input or 500 for server error) so clients
get a clear failure instead of an unhandled exception; reference the POST
handler, parseWorkflowRequest, runWorkflow, featureOrchestrator, runStore and
toServerSentEventsResponse when making the changes.
In `@packages/typescript/ai-client/src/connection-adapters.ts`:
- Around line 553-558: The loop in readStreamLines/connection-adapters is
yielding raw JSON.parse output (in the for-await over readStreamLines(reader,
abortSignal)) without runtime validation; add Zod validation after JSON.parse:
import z from 'zod', define a WorkflowSseChunkSchema (or WorkflowEventSchema)
that matches the expected SSE chunk shape, then replace the raw yield with
validation via WorkflowSseChunkSchema.safeParse(parsed) and only yield the
validated .data (or continue on failure), logging or ignoring invalid chunks as
appropriate so malformed chunks don't propagate.
In `@packages/typescript/ai-client/src/workflow-client.ts`:
- Around line 109-124: Wrap the async stream-handling in approve() (and the
similar blocks in start() and the other approval path) with a try/catch so that
any exceptions from openStream/consumeStream (or connect/iteration) update
client state to reflect the failure: call this.setState({ status: 'error',
error: err, pendingApproval: null }) inside the catch, then rethrow the error.
Locate the stream flow in approve(), start(), and the other approval-handling
code paths and apply the same pattern to ensure state is not left as 'running'
when stream errors occur.
- Around line 301-306: The traversal over path segments currently assumes every
intermediate node exists and is an object/array, which can throw on malformed
deltas; update the loop that walks segments (the code using variables segments,
cursor and last) to validate at each iteration that cursor is non-null and
typeof cursor === 'object' and that segments[i] exists (and is indexable) before
assigning cursor = cursor[segments[i]]; if any check fails, abort/skip applying
this delta (return/continue) rather than letting it throw. Apply the same
defensive checks to the other similar block (the splice/delete handling around
lines 317-321) so both deletion and insertion/splice paths safely ignore invalid
paths instead of crashing the stream.
- Around line 135-142: The stop() method currently constructs an async iterable
by calling openStream(...) but never consumes it, so the abort payload is never
sent; change stop to async and consume the returned async iterable from
openStream({ abort: true, runId: this.clientState.runId }) (e.g., use a
for-await-of loop to iterate the stream until completion or break immediately
after first send) and propagate/handle any errors, then call this.setState({
status: 'aborted' })—this ensures openStream actually executes the abort
request. Reference: stop(), openStream(), this.clientState.runId,
this.setState().
- Around line 152-249: handleChunk currently casts incoming streaming chunks
without runtime checks which can corrupt this.clientState or throw; add Zod
schemas for the different chunk shapes (e.g., CUSTOM, RUN_ERROR, RUN_FINISHED,
RUN_STARTED, STATE_DELTA, STATE_SNAPSHOT, STEP_FINISHED, STEP_STARTED,
TEXT_MESSAGE_CONTENT) and validate each incoming chunk at the top of handleChunk
before any casting or state updates (use a discriminated union on chunk.type),
then reject/log invalid chunks and return early; update references in this
function (handleChunk, setState, this.clientState, applyJsonPatch,
WorkflowClientState) to use the validated/typed data instead of unchecked casts.
In `@packages/typescript/ai-orchestration/package.json`:
- Line 47: The package.json entry for the internal peer dependency
"@tanstack/ai" uses "workspace:^" instead of the required internal protocol
"workspace:*"; update the dependency value for "@tanstack/ai" in
packages/typescript/ai-orchestration/package.json to "workspace:*" so the
internal peer dependency follows the project's workspace protocol conventions.
In `@packages/typescript/ai-orchestration/src/engine/invoke-agent.ts`:
- Around line 52-82: The output Promise can hang if the async iterator exits
early; update drain() to always settle output in a finally block: introduce a
local boolean (e.g., settled = false) and move the parseOutputFromText(agent,
lastTextContent) + resolveOutput(parsed) / rejectOutput(err) logic into a
finally that runs after the for-await loop so output is resolved or rejected on
every exit path, and guard resolveOutput/rejectOutput with the settled flag to
avoid double-settling; keep the existing catch to rethrow errors but ensure it
sets settled so the finally knows whether it must call resolve/reject; reference
drain(), output, resolveOutput, rejectOutput, lastTextContent,
parseOutputFromText and filterInnerRunBoundaries when making the change.
In `@packages/typescript/ai-orchestration/src/engine/run-workflow.ts`:
- Around line 332-364: The nested-workflow branch in runWorkflow currently
yields nested chunks but then always emits STEP_FINISHED and continues even if
the nested run errored or paused; update the nested-workflow handling (the
runWorkflow call and the for-await loop over nestedIter) to mirror the agent
error handler logic: when a nested chunk with type 'RUN_ERROR' is received, emit
STEP_FINISHED for the step with the error content and propagate the failure into
the parent (call the parent generator's throw equivalent as done in the agent
handler) instead of continuing, and when an approval-related
pause/approval-requested is observed, do not treat the nested step as
finished—either block here or explicitly fail the nested approval at this
boundary until parent-pause-on-nested-pause is supported; refer to the
nested-workflow descriptor handling and the agent error handler (lines around
the existing error catch that emits STEP_FINISHED and calls
live.generator.throw(err)) to implement identical error/approval propagation
behavior.
- Around line 51-62: mergeStateDefaults currently calls
workflow.stateSchema['~standard'].validate(initial) but ignores Promise results,
causing async validations to be skipped and defaults/coercions lost; change
mergeStateDefaults to be async, await the validate(...) call, then check the
awaited result for issues (same logic as the synchronous branch) and return the
coerced validated.value when present, otherwise return initial; update any
callers (e.g., startRun) if needed to await mergeStateDefaults so both sync and
async schema validators are handled consistently.
In `@packages/typescript/ai-orchestration/src/engine/state-diff.ts`:
- Around line 32-72: The diff function can emit operations with value: undefined
which JSON.stringify drops; update the places that create ops (the top-level
replace in diff when types disagree or arrays differ, and the per-key add in the
object branch) to normalize undefined to null (or alternatively treat undefined
as a remove); specifically, change the value payloads created in diff (both the
replace op that uses value: next and the add op that uses value: nextObj[key])
to use a helper normalization (e.g., normalizedValue = next === undefined ? null
: next) before constructing the Operation so no op is emitted with value:
undefined.
In `@packages/typescript/ai-orchestration/src/run-store/in-memory.ts`:
- Around line 28-55: The teardown in scheduleExpiry and delete currently removes
live entries but leaves LiveRun.abortController un-aborted and
LiveRun.approvalResolver unresolved, leaking paused generators and
pendingEvents; update scheduleExpiry (where the timeout handle is created) and
delete to first retrieve the live entry from live.get(runId) and if present call
abortController.abort(), reject/resolve the approvalResolver (reject with a
clear Abort/Error) and clear/publish pendingEvents before deleting; also ensure
the created timeout handle uses setTimeout(...).unref?.() so the TTL won't keep
the Node process alive. Ensure you reference and operate on the LiveRun object
obtained via live.get(runId) when implementing these changes.
In `@packages/typescript/ai-orchestration/src/server/parse-request.ts`:
- Around line 30-39: Replace the unsafe cast in parseWorkflowRequest by defining
a Zod schema for RawBody/WorkflowRequestParams (including approval with
approvalId types, input, runId) and use it to parse/validate await
request.json(); on parse failure throw a 400 HTTP error (or return a clear
validation error) so malformed bodies are rejected at the HTTP boundary before
calling runWorkflow; update parseWorkflowRequest to return the validated, typed
result from zod.parse/zod.safeParse instead of the structural cast.
---
Nitpick comments:
In `@examples/ts-react-chat/src/components/ArticleModal.tsx`:
- Around line 10-21: The useEffect in ArticleModal registers a keydown listener
and toggles body overflow but incorrectly depends on the entire props object,
causing unnecessary re-runs; change the dependency array to only include the
specific callback used (props.onClose) by referencing the onKey handler and the
effect that sets document.body.style.overflow so the listener and scroll lock
are only re-registered when props.onClose changes.
In `@examples/ts-react-chat/src/components/DraftPreview.tsx`:
- Around line 55-58: The preview container in DraftPreview currently uses
key={bumpKey} which forces a remount and resets scroll/focus; remove the dynamic
key and instead add a stable root element (the existing div with className
"relative px-6 py-7...") and toggle a data attribute or short-lived CSS class
(e.g. data-bump={bumpKey} or isPulsing) on that same div to trigger the pulse
animation via CSS/animation, ensuring you update where bumpKey is produced so it
sets the attribute/class rather than the key.
In `@examples/ts-react-chat/src/components/WorkflowTimeline.tsx`:
- Around line 105-109: The JSX rendering in WorkflowTimeline.tsx uses
step.stepType.replace('-', ' · ') which only replaces the first dash; update the
expression in the span rendering (the usage of step.stepType) to replace all
dashes (e.g., use step.stepType.replaceAll('-', ' · ') or
step.stepType.split('-').join(' · ')) so multi-dash types like
"agent-call-retry" become "agent · call · retry".
In `@examples/ts-react-chat/src/routes/api.orchestration.ts`:
- Line 10: The inline TTL literal passed to inMemoryRunStore (60 * 60 * 1000)
reduces readability; extract it to a named constant (e.g., RUN_STORE_TTL_MS)
near the top of the module and use that constant when constructing runStore so
the purpose and units are clear; update any related comments or usages of
runStore if needed to reference the constant.
In `@packages/typescript/ai-orchestration/src/engine/emit-events.ts`:
- Around line 9-122: Change each helper to return its specific AG-UI event type
(e.g. runStartedEvent -> RunStartedEvent, runFinishedEvent -> RunFinishedEvent,
runErrorEvent -> RunErrorEvent, stepStartedEvent -> StepStartedEvent,
stepFinishedEvent -> StepFinishedEvent, stateSnapshotEvent ->
StateSnapshotEvent, stateDeltaEvent -> StateDeltaEvent, customEvent ->
CustomEvent) instead of the generic StreamChunk and remove the trailing "as
StreamChunk" casts; import those concrete types from `@tanstack/ai` so TypeScript
validates the payload shape at construction (leave approvalRequestedEvent
returning customEvent as-is since it delegates to customEvent). Ensure
signatures and returned object shapes match the imported types.
In `@packages/typescript/ai-orchestration/src/engine/invoke-agent.ts`:
- Around line 42-47: The shape detection in the result handling block (the
`'stream' in result && 'output' in result` check inside invoke-agent.ts that
returns { stream: filterInnerRunBoundaries(result.stream), output:
result.output.then((o) => parseOutput<T>(agent, o)) }) relies on positional
checks and intentionally accepts any object with those keys; add a concise
one-line comment immediately above this if explaining that this is a narrow,
positional shape check (shape (c)) and must remain ordered before the other
shape branches (shape (a) then (b)) so future maintainers understand the
rationale and don’t reorder or replace the check.
- Around line 148-155: Move the exported SchemaValidationError class declaration
above any code that constructs or references it so its definition precedes its
usages; locate the places in this module where new SchemaValidationError(...) is
thrown/checked (the earlier references in the same file) and cut-paste the class
block to the top of the file (keeping export class SchemaValidationError ...
unchanged) so lint rules like no-use-before-define no longer flag the symbol.
In `@packages/typescript/ai-orchestration/src/engine/state-diff.ts`:
- Around line 18-20: The function snapshotState uses structuredClone which
requires Node >=17 / modern browsers; update the package metadata to document
this requirement by adding an engines entry (e.g., Node >=17) in this package's
package.json and/or add a one-line note to the README mentioning that
snapshotState (structuredClone) needs Node ≥17 or a modern browser runtime;
reference the snapshotState function and structuredClone in your change so
reviewers can see the compatibility note tied to the implementation.
In `@packages/typescript/ai-orchestration/src/primitives/bind-agents.ts`:
- Around line 13-41: The two generator functions for entries of agents and
nested workflows duplicate descriptor construction; refactor the loop that
builds bound[name] so a single generator closure creates a StepDescriptor with
common fields (kind, name, input) and then sets the specific payload field based
on def.__kind (e.g., set descriptor.agent = def for 'agent' or
descriptor.workflow = def for 'nested-workflow'), yield the descriptor and
return the result; update references to StepDescriptor, agents, bound, and
def.__kind in that single generator to remove the duplicated function bodies.
In `@packages/typescript/ai-orchestration/src/primitives/retry.ts`:
- Around line 14-29: The exponential backoff should include jitter to avoid
synchronized retry storms: update computeDelay (the exponential branch) to
compute the base exponential delay as currently done (using base and attempt),
then apply ±20% random jitter (e.g., multiply by a random factor in [0.8, 1.2]),
clamp the final value to <= maxDelayMs and >= 0 and return it (rounded to an
integer if desired); keep the existing behavior for the 'none' and 'linear'
branches and keep delay(ms: number): Promise<void> as-is.
In `@packages/typescript/ai-orchestration/src/run-store/in-memory.ts`:
- Line 26: The Map declaration uses NodeJS.Timeout which ties the code to
`@types/node`; change the type to a runtime-agnostic one by replacing
NodeJS.Timeout with ReturnType<typeof setTimeout> in the declaration of
expirations (the const expirations = new Map<string, ...>()), and ensure any
places that read/clear timers (e.g., where clearTimeout is called) continue to
accept that type without importing Node types.
In `@packages/typescript/ai-orchestration/src/server/parse-request.ts`:
- Around line 9-14: RawBody currently includes an unused abort?: boolean field
which isn't propagated by parseWorkflowRequest and isn't represented on
WorkflowRequestParams; remove abort from the RawBody interface to avoid
advertising an unsupported capability, update any related types/usages that
reference RawBody (e.g., the RawBody declaration in parse-request.ts) and run
type checks to ensure no consumers relied on that field.
In `@packages/typescript/ai-orchestration/tests/engine.smoke.test.ts`:
- Around line 98-145: Add a symmetric resume smoke test after the existing pause
assertions in the 'pauses on approval — stream ends after approval-requested,
RUN_FINISHED not emitted' test: call runWorkflow again with the same workflow
(wf), runId (from runStarted.runId), approval payload (e.g., { approved: true })
and the same runStore (store), collect emitted events into a new array
(events2), then assert that events2 includes STEP_FINISHED and RUN_FINISHED and
that the RUN_FINISHED event's output equals the expected { ok: true }; this
verifies state restore, pendingApproval finalization, and the generator.next
seed value are handled correctly.
In `@packages/typescript/ai-react/src/use-workflow.ts`:
- Around line 37-40: The comment near optsRef (where const optsRef =
useRef(opts); optsRef.current = opts) is misleading because some options (like
body and connection) are captured at client construction and not updated, while
event callbacks do read fresh values; update the comment to state that optsRef
is used so callback handlers (onCustomEvent, onStateChange) will see the latest
opts via optsRef.current, but structural options such as body and connection are
fixed when the client is created (refer to the client construction site where
body is passed) and will not change if props update.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: eeed6f10-2679-46f2-97cf-1966e42df69f
⛔ Files ignored due to path filters (1)
pnpm-lock.yamlis excluded by!**/pnpm-lock.yaml
📒 Files selected for processing (44)
examples/ts-react-chat/package.jsonexamples/ts-react-chat/src/components/ArticleModal.tsxexamples/ts-react-chat/src/components/DraftPreview.tsxexamples/ts-react-chat/src/components/Header.tsxexamples/ts-react-chat/src/components/StateInspector.tsxexamples/ts-react-chat/src/components/WorkflowTimeline.tsxexamples/ts-react-chat/src/lib/workflows/article-workflow.tsexamples/ts-react-chat/src/lib/workflows/orchestrator.tsexamples/ts-react-chat/src/routeTree.gen.tsexamples/ts-react-chat/src/routes/api.orchestration.tsexamples/ts-react-chat/src/routes/api.workflow.tsexamples/ts-react-chat/src/routes/orchestration.tsxexamples/ts-react-chat/src/routes/workflow.tsxexamples/ts-react-chat/src/styles.csspackages/typescript/ai-client/src/connection-adapters.tspackages/typescript/ai-client/src/index.tspackages/typescript/ai-client/src/workflow-client.tspackages/typescript/ai-orchestration/README.mdpackages/typescript/ai-orchestration/eslint.config.jspackages/typescript/ai-orchestration/package.jsonpackages/typescript/ai-orchestration/src/define/define-agent.tspackages/typescript/ai-orchestration/src/define/define-orchestrator.tspackages/typescript/ai-orchestration/src/define/define-router.tspackages/typescript/ai-orchestration/src/define/define-workflow.tspackages/typescript/ai-orchestration/src/engine/emit-events.tspackages/typescript/ai-orchestration/src/engine/invoke-agent.tspackages/typescript/ai-orchestration/src/engine/run-workflow.tspackages/typescript/ai-orchestration/src/engine/state-diff.tspackages/typescript/ai-orchestration/src/index.tspackages/typescript/ai-orchestration/src/primitives/approve.tspackages/typescript/ai-orchestration/src/primitives/bind-agents.tspackages/typescript/ai-orchestration/src/primitives/index.tspackages/typescript/ai-orchestration/src/primitives/retry.tspackages/typescript/ai-orchestration/src/result.tspackages/typescript/ai-orchestration/src/run-store/in-memory.tspackages/typescript/ai-orchestration/src/run-store/index.tspackages/typescript/ai-orchestration/src/server/index.tspackages/typescript/ai-orchestration/src/server/parse-request.tspackages/typescript/ai-orchestration/src/types.tspackages/typescript/ai-orchestration/tests/engine.smoke.test.tspackages/typescript/ai-orchestration/tsconfig.jsonpackages/typescript/ai-orchestration/vite.config.tspackages/typescript/ai-react/src/index.tspackages/typescript/ai-react/src/use-workflow.ts
Comment on lines
+553
to
+558
| for await (const line of readStreamLines(reader, abortSignal)) { | ||
| const data = line.startsWith('data: ') ? line.slice(6) : line | ||
| if (!data) continue | ||
| try { | ||
| yield JSON.parse(data) | ||
| } catch { |
Contributor
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
cd packages/typescript/ai-client/src && wc -l connection-adapters.tsRepository: TanStack/ai
Length of output: 82
🏁 Script executed:
cd packages/typescript/ai-client/src && sed -n '540,570p' connection-adapters.ts | cat -nRepository: TanStack/ai
Length of output: 863
🏁 Script executed:
cd packages/typescript/ai-client/src && head -20 connection-adapters.ts | cat -nRepository: TanStack/ai
Length of output: 744
🏁 Script executed:
cd packages/typescript && rg "import.*zod" --type tsRepository: TanStack/ai
Length of output: 2168
🏁 Script executed:
cd packages/typescript && rg "WorkflowChunkSchema\|z\.object\|safeParse" src/ -A 2 -B 2Repository: TanStack/ai
Length of output: 100
🏁 Script executed:
cd packages/typescript/ai-client && grep -n "JSON.parse\|safeParse\|parse\|\.z\." src/connection-adapters.tsRepository: TanStack/ai
Length of output: 468
🏁 Script executed:
cd packages/typescript/ai-client && grep -n "readStreamLines\|for await" src/connection-adapters.ts | head -20Repository: TanStack/ai
Length of output: 364
🏁 Script executed:
cd packages/typescript/ai-client/src && cat -n workflow-client.ts | head -100Repository: TanStack/ai
Length of output: 3301
🏁 Script executed:
cd packages/typescript/ai-client/src && grep -n "WorkflowChunk\|StreamChunk" connection-adapters.tsRepository: TanStack/ai
Length of output: 1286
🏁 Script executed:
cd packages/typescript/ai-client && find . -name "*.ts" -exec grep -l "JSON.parse" {} \; | head -10Repository: TanStack/ai
Length of output: 162
🏁 Script executed:
cd packages/typescript/ai-client/src && sed -n '480,495p' connection-adapters.ts | cat -nRepository: TanStack/ai
Length of output: 653
🏁 Script executed:
cd packages/typescript/ai-client/src && sed -n '310,330p' connection-adapters.ts | cat -nRepository: TanStack/ai
Length of output: 914
Add Zod schema validation for workflow SSE chunks.
Line 557 yields raw JSON.parse output without runtime schema validation. Per coding guidelines for packages/typescript/**/src/**/*.ts, Zod must be used for schema validation. This makes the workflow event handling fragile when chunks are malformed or unexpectedly shaped.
Proposed fix
+import { z } from 'zod'
+
+const WorkflowChunkSchema = z.object({
+ type: z.string(),
+}).passthrough()
...
for await (const line of readStreamLines(reader, abortSignal)) {
const data = line.startsWith('data: ') ? line.slice(6) : line
if (!data) continue
try {
- yield JSON.parse(data)
+ const parsed = WorkflowChunkSchema.safeParse(JSON.parse(data))
+ if (parsed.success) {
+ yield parsed.data
+ }
} catch {
// skip malformed lines
}
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| for await (const line of readStreamLines(reader, abortSignal)) { | |
| const data = line.startsWith('data: ') ? line.slice(6) : line | |
| if (!data) continue | |
| try { | |
| yield JSON.parse(data) | |
| } catch { | |
| for await (const line of readStreamLines(reader, abortSignal)) { | |
| const data = line.startsWith('data: ') ? line.slice(6) : line | |
| if (!data) continue | |
| try { | |
| const parsed = WorkflowChunkSchema.safeParse(JSON.parse(data)) | |
| if (parsed.success) { | |
| yield parsed.data | |
| } | |
| } catch { |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/typescript/ai-client/src/connection-adapters.ts` around lines 553 -
558, The loop in readStreamLines/connection-adapters is yielding raw JSON.parse
output (in the for-await over readStreamLines(reader, abortSignal)) without
runtime validation; add Zod validation after JSON.parse: import z from 'zod',
define a WorkflowSseChunkSchema (or WorkflowEventSchema) that matches the
expected SSE chunk shape, then replace the raw yield with validation via
WorkflowSseChunkSchema.safeParse(parsed) and only yield the validated .data (or
continue on failure), logging or ignoring invalid chunks as appropriate so
malformed chunks don't propagate.
Comment on lines
+332
to
+364
| if (descriptor.kind === 'nested-workflow') { | ||
| yield stepStartedEvent({ | ||
| stepId, | ||
| stepName: descriptor.name, | ||
| stepType: 'nested-workflow', | ||
| }) | ||
|
|
||
| let nestedOutput: unknown = undefined | ||
| const nestedIter = runWorkflow({ | ||
| workflow: descriptor.workflow, | ||
| input: descriptor.input, | ||
| runStore, | ||
| signal: abortController.signal, | ||
| outputSink: (o) => { | ||
| nestedOutput = o | ||
| }, | ||
| }) | ||
|
|
||
| for await (const chunk of nestedIter) { | ||
| if (chunk.type === 'RUN_STARTED' || chunk.type === 'RUN_FINISHED') { | ||
| continue | ||
| } | ||
| yield chunk | ||
| } | ||
|
|
||
| yield stepFinishedEvent({ | ||
| stepId, | ||
| stepName: descriptor.name, | ||
| content: nestedOutput, | ||
| }) | ||
| nextValue = nestedOutput | ||
| continue | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm there is no other site that observes nested run errors / pauses.
rg -nP --type=ts -C3 "nested-workflow|nestedIter|RUN_ERROR" packages/typescript/ai-orchestrationRepository: TanStack/ai
Length of output: 5560
🏁 Script executed:
# Get the full nested-workflow section to see complete error handling
sed -n '328,370p' packages/typescript/ai-orchestration/src/engine/run-workflow.tsRepository: TanStack/ai
Length of output: 1159
🏁 Script executed:
# Check what happens with RUN_ERROR chunks - do they reach the yield?
rg -nP --type=ts "RUN_ERROR" packages/typescript/ai-orchestration/src/engine/run-workflow.ts -A5 -B5Repository: TanStack/ai
Length of output: 37
🏁 Script executed:
# Look for approval handling in the file
rg -nP --type=ts "approval|APPROVAL" packages/typescript/ai-orchestration/src/engine/run-workflow.ts -C3Repository: TanStack/ai
Length of output: 4097
🏁 Script executed:
# Check the StreamChunk type to see all possible chunk types
rg -nP --type=ts "type StreamChunk|type.*ChunkType|STEP_FINISHED|RUN_ERROR|approval" packages/typescript/ai-orchestration/src/types.ts -A2Repository: TanStack/ai
Length of output: 732
🏁 Script executed:
# Look for error handling in the parent generator/main loop
sed -n '1,330p' packages/typescript/ai-orchestration/src/engine/run-workflow.ts | tail -50Repository: TanStack/ai
Length of output: 1319
🏁 Script executed:
# Check for try-catch around the nested-workflow section
sed -n '260,380p' packages/typescript/ai-orchestration/src/engine/run-workflow.ts | grep -nE "try|catch|throw|RUN_ERROR" -A2 -B2Repository: TanStack/ai
Length of output: 463
🏁 Script executed:
# See the full startRun function to understand error handling at top level
sed -n '210,280p' packages/typescript/ai-orchestration/src/engine/run-workflow.tsRepository: TanStack/ai
Length of output: 2244
🏁 Script executed:
# Check StreamChunk type definition to see all event types
rg -nP --type=ts "type StreamChunk|export type.*Event" packages/typescript/ai-orchestration/src/types.ts -A5 | head -100Repository: TanStack/ai
Length of output: 37
🏁 Script executed:
# Search for StreamChunk type definition
rg -nP --type=ts "export.*StreamChunk|interface StreamChunk" packages/typescript/ai-orchestration/src/ -A10Repository: TanStack/ai
Length of output: 1061
🏁 Script executed:
# Look for all the event types defined
rg -nP --type=ts "RunErrorEvent|RunStartedEvent|RunFinishedEvent|ApprovalRequestedEvent" packages/typescript/ai-orchestration/src/engine/emit-events.ts -B2Repository: TanStack/ai
Length of output: 37
🏁 Script executed:
# Get all event functions from emit-events.ts
rg -nP --type=ts "^export function" packages/typescript/ai-orchestration/src/engine/emit-events.tsRepository: TanStack/ai
Length of output: 486
🏁 Script executed:
# Check the types.ts for StreamChunk union type
rg -nP --type=ts "StreamChunk" packages/typescript/ai-orchestration/src/types.ts -B2 -A15 | head -80Repository: TanStack/ai
Length of output: 1527
🏁 Script executed:
# Find where StreamChunk is defined - it might be in a different package
fd -t f "*.ts" packages/typescript/ | xargs rg -l "type StreamChunk|interface StreamChunk" | head -5Repository: TanStack/ai
Length of output: 533
🏁 Script executed:
# Look in the main ai package types
rg -nP --type=ts "type StreamChunk|StreamChunk =" packages/typescript/ai/src/ -B2 -A20 | head -100Repository: TanStack/ai
Length of output: 3173
🏁 Script executed:
# Now let's check if RUN_ERROR in a nested-workflow properly propagates
# Look at the complete nested-workflow section with full context
sed -n '330,365p' packages/typescript/ai-orchestration/src/engine/run-workflow.tsRepository: TanStack/ai
Length of output: 976
🏁 Script executed:
# Check the top-level error handling in the try-catch of driveLoop
sed -n '226,410p' packages/typescript/ai-orchestration/src/engine/run-workflow.ts | grep -nE "try|catch|throw|finally" -A5 -B2Repository: TanStack/ai
Length of output: 1207
🏁 Script executed:
# Look at what happens after the driveLoop completes - error handling
sed -n '100,180p' packages/typescript/ai-orchestration/src/engine/run-workflow.tsRepository: TanStack/ai
Length of output: 2129
🏁 Script executed:
# See the complete catch block in driveLoop
sed -n '390,430p' packages/typescript/ai-orchestration/src/engine/run-workflow.tsRepository: TanStack/ai
Length of output: 1218
🏁 Script executed:
# Check if there's any special handling for RUN_ERROR in nested context
rg -nP --type=ts "chunk.type.*RUN_ERROR|RUN_ERROR.*chunk" packages/typescript/ai-orchestration/src/Repository: TanStack/ai
Length of output: 37
Nested workflow errors and approval pauses are not propagated to the parent.
The nested-workflow handler yields all chunks except RUN_STARTED/RUN_FINISHED, but lacks error handling that agents have. When a nested run emits RUN_ERROR, the parent yields it but then emits STEP_FINISHED with nestedOutput = undefined and continues—the parent run is not failed or paused. Similarly, when a nested workflow yields approval-requested and returns, the parent yields the event but then continues past the nested step with STEP_FINISHED, while the nested run remains status: 'paused' in the store. This creates data flow corruption and inconsistent run state.
Model the nested-workflow case after the agent error handler (lines 301–325), which catches errors, emits STEP_FINISHED with error content, and calls live.generator.throw(err) to propagate the failure. Handle RUN_ERROR chunks similarly, and either block or explicitly fail nested approvals at this boundary until parent-pause-on-nested-pause is supported.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/typescript/ai-orchestration/src/engine/run-workflow.ts` around lines
332 - 364, The nested-workflow branch in runWorkflow currently yields nested
chunks but then always emits STEP_FINISHED and continues even if the nested run
errored or paused; update the nested-workflow handling (the runWorkflow call and
the for-await loop over nestedIter) to mirror the agent error handler logic:
when a nested chunk with type 'RUN_ERROR' is received, emit STEP_FINISHED for
the step with the error content and propagate the failure into the parent (call
the parent generator's throw equivalent as done in the agent handler) instead of
continuing, and when an approval-related pause/approval-requested is observed,
do not treat the nested step as finished—either block here or explicitly fail
the nested approval at this boundary until parent-pause-on-nested-pause is
supported; refer to the nested-workflow descriptor handling and the agent error
handler (lines around the existing error catch that emits STEP_FINISHED and
calls live.generator.throw(err)) to implement identical error/approval
propagation behavior.
Comment on lines
+30
to
+39
| export async function parseWorkflowRequest( | ||
| request: Request, | ||
| ): Promise<WorkflowRequestParams> { | ||
| const body = (await request.json()) as RawBody | ||
| return { | ||
| approval: body.approval, | ||
| input: body.input, | ||
| runId: body.runId, | ||
| } | ||
| } |
Contributor
There was a problem hiding this comment.
Validate the parsed body with Zod instead of a structural cast.
(await request.json()) as RawBody blindly trusts the wire format — a malformed approval shape (e.g., missing approvalId or wrong types) will silently flow into runWorkflow and only surface as obscure failures inside the engine. A small Zod schema here would give you an early, typed error at the HTTP boundary, and bring this in line with the rest of the library.
🛡️ Suggested validation
-import type { ApprovalResult } from '../types'
+import { z } from 'zod'
+import type { ApprovalResult } from '../types'
+
+const approvalSchema = z.object({
+ approved: z.boolean(),
+ approvalId: z.string(),
+ feedback: z.string().optional(),
+})
+
+const rawBodySchema = z.object({
+ abort: z.boolean().optional(),
+ approval: approvalSchema.optional(),
+ input: z.unknown().optional(),
+ runId: z.string().optional(),
+})
@@
- const body = (await request.json()) as RawBody
+ const body = rawBodySchema.parse(await request.json())
return {
- approval: body.approval,
+ approval: body.approval as ApprovalResult | undefined,
input: body.input,
runId: body.runId,
}As per coding guidelines: "Use Zod for schema validation and tool definition across the library".
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| export async function parseWorkflowRequest( | |
| request: Request, | |
| ): Promise<WorkflowRequestParams> { | |
| const body = (await request.json()) as RawBody | |
| return { | |
| approval: body.approval, | |
| input: body.input, | |
| runId: body.runId, | |
| } | |
| } | |
| import { z } from 'zod' | |
| import type { ApprovalResult } from '../types' | |
| const approvalSchema = z.object({ | |
| approved: z.boolean(), | |
| approvalId: z.string(), | |
| feedback: z.string().optional(), | |
| }) | |
| const rawBodySchema = z.object({ | |
| abort: z.boolean().optional(), | |
| approval: approvalSchema.optional(), | |
| input: z.unknown().optional(), | |
| runId: z.string().optional(), | |
| }) | |
| export async function parseWorkflowRequest( | |
| request: Request, | |
| ): Promise<WorkflowRequestParams> { | |
| const body = rawBodySchema.parse(await request.json()) | |
| return { | |
| approval: body.approval as ApprovalResult | undefined, | |
| input: body.input, | |
| runId: body.runId, | |
| } | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@packages/typescript/ai-orchestration/src/server/parse-request.ts` around
lines 30 - 39, Replace the unsafe cast in parseWorkflowRequest by defining a Zod
schema for RawBody/WorkflowRequestParams (including approval with approvalId
types, input, runId) and use it to parse/validate await request.json(); on parse
failure throw a 400 HTTP error (or return a clear validation error) so malformed
bodies are rejected at the HTTP boundary before calling runWorkflow; update
parseWorkflowRequest to return the validated, typed result from
zod.parse/zod.safeParse instead of the structural cast.
…astResult and abort
- AgentRunResult now accepts StructuredOutputStream<T>, so an agent's
run can return chat({ outputSchema, stream: true }) directly without
a cast. The two unify structurally at runtime; the explicit member
works around TS's discriminated-union check on tagged CUSTOM events.
- defineOrchestrator's engine loop now captures yield* boundAgent's
return value and passes it to the router on the next turn as
lastResult, fixing the silent infinite loop where router-driven
agents (e.g., spec, implement) produced output the router could not
fold into state.
- defineRouter's router signature gains the matching lastResult arg.
- parseWorkflowRequest surfaces the abort flag from the body so route
handlers can look up the live run and call abortController.abort().
Wiring on the handler side stays in the host's example code.
…hestration UI
- article and orchestration workflows now use chat({ stream: true }) on
every structured-output call (writer, reviewers, editor, spec,
planner, coder, review, triage). The article /workflow page partial-
parses wf.currentText during writer/editor steps so DraftPreview
fills in live; DraftPreview gained a streaming indicator + trailing
caret on the last paragraph.
- orchestration.tsx rewritten as a terminal-style UI: window chrome
with status, in-flow user prompts, per-step lines (triage / spec /
planner / coder / review / approval) with shiki-highlighted bodies,
inline approval prompt that accepts y / n / free-text refinement
notes, blinking caret rendered as a trailing block over a
transparent-caret textarea (in both the main prompt and the approval
prompt), Ctrl+C / Cmd+. to abort the run when no text selection is
active.
- After a successful run, the page carries the spec/result forward as
previousSpec / previousResult on the next submission. The
orchestrator initialize() seeds state and starts at phase: 'review';
the router routes new userMessage through pendingFeedback so triage
only sees a non-empty message when there's truly something to
address, fixing the spec/triage infinite loop.
- Spec agent's prompt is now branch-aware: a refinement run is told
the existing spec is authoritative and to preserve its language,
framework, file extensions, and architecture. Coder prompt now
insists the patch matches the file's extension language.
- FileTreePanel shows every coder patch as a collapsible directory
tree (single-child dirs are collapsed into combined rows). The
panel renders the *applied* file content (not the diff) through
extractFileFromPatch, which strips markdown fences, unified-diff
metadata (--- / +++ / @@ / index / mode headers), and diff prefixes
(keeps additions and context, drops removals). Highlighted via
shiki under a custom theme matched to the ink/citron/rust/moss
palette in styles.css.
- api.workflow.ts and api.orchestration.ts handle { abort, runId }
bodies by looking up the live run in the store and calling
abortController.abort(), which propagates through runWorkflow ->
invokeAgent -> chat()'s AbortSignal to actually terminate in-flight
LLM requests instead of just flipping the local status.
…ing-wadler # Conflicts: # examples/ts-react-chat/src/routeTree.gen.ts
Fix violations introduced by the type-safety tightening in #564 and clean up housekeeping items knip surfaced in this package: - ai-client/workflow-client: mark opts/subscribers readonly, drop non-null assertions in removeAt/setAt, drop `as unknown as` cast. - ai-orchestration: bracket the two intentional `as unknown as` casts (generator placeholder + yield-resume) with eslint-disable comments citing their constraints. - ai-orchestration: align vite devDependency to ^7.3.3 to match the rest of the workspace (sherif). - ai-orchestration: drop unused @tanstack/ai-event-client dependency and unused primitives/index.ts + run-store/index.ts barrel files (knip). - knip: disable the `duplicates` rule globally so the intentional `useOrchestration = useWorkflow` alias stops reporting.
… skill Documents @tanstack/ai-orchestration end-to-end so users have a complete journey from "what's this?" to production. Closes the update-docs-with-new-cases gap on this PR. New section under docs/orchestration/: - overview — routing landing + when-to-reach-for-it matrix - workflows — defineAgent + defineWorkflow + yield* (linear pipeline) - orchestrators — defineOrchestrator with router + state + lastResult fold - approvals — approve() primitive, pause/resume semantics, deny-with-feedback - refining-across-runs — previousX inputs + initialize() for cross-run state - retries-and-errors — retry() backoff, SchemaValidationError, abort bridging - run-persistence — RunStore, why durable resume is non-trivial today Plus docs/api/ai-orchestration.md — signature reference for every export. Cross-links added inbound from getting-started/overview, chat/agentic-cycle, structured-outputs/multi-turn, and tools/tools so the new section is reachable from the natural read paths. Agent skill: packages/typescript/ai-orchestration/skills/ai-orchestration/ SKILL.md mirrors the ai-code-mode shape — 5 core patterns, server route, client surface, plus a "common pitfalls" section pulled from real source gotchas (signal→abortController bridge, lastResult folding, pendingFeedback clearing, durable-store reality check). ai-core SKILL.md now routes to it. coupling.json: orchestration skill added to the agent-skills trigger, docs/orchestration/** added to the public-docs trigger so future source changes auto-flag both surfaces. Fact-checked against source: chat() takes abortController not signal; event types are uppercase (STEP_STARTED, STATE_DELTA, etc.); approval-requested is a CUSTOM-chunk name, not a top-level event type.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
examples/ts-react-chat/src/routes/api.workflow.ts (1)
16-28:⚠️ Potential issue | 🟠 Major | ⚡ Quick winAdd structured error handling around request parsing and run startup.
parseWorkflowRequestandrunWorkflowcan throw; without a guard this route returns unstructured failures and drops client context.Suggested minimal fix
POST: async ({ request }) => { - const params = await parseWorkflowRequest(request) - if (params.abort && params.runId) { - runStore.getLive(params.runId)?.abortController.abort() - return new Response(null, { status: 204 }) - } - const stream = runWorkflow({ - runStore, - workflow: articleWorkflow, - ...params, - }) - return toServerSentEventsResponse(stream) + try { + const params = await parseWorkflowRequest(request) + if (params.abort && params.runId) { + runStore.getLive(params.runId)?.abortController.abort() + return new Response(null, { status: 204 }) + } + const stream = runWorkflow({ + runStore, + workflow: articleWorkflow, + ...params, + }) + return toServerSentEventsResponse(stream) + } catch (error) { + return new Response( + JSON.stringify({ + error: error instanceof Error ? error.message : 'Unknown error', + }), + { status: 400, headers: { 'Content-Type': 'application/json' } }, + ) + } },🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@examples/ts-react-chat/src/routes/api.workflow.ts` around lines 16 - 28, Wrap the request handling in a try/catch that separately guards parseWorkflowRequest and runWorkflow so thrown errors return structured HTTP responses instead of unhandled failures: catch parse errors from parseWorkflowRequest and return a 400 JSON error, preserve the existing abort logic when params indicate abort, then wrap the runWorkflow call and stream conversion in its own try/catch that logs the error and returns a 500 JSON error if runWorkflow or toServerSentEventsResponse throws; reference parseWorkflowRequest, runWorkflow, runStore, and toServerSentEventsResponse to locate the code and keep the existing abortController abort path intact.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/orchestration/orchestrators.md`:
- Around line 24-29: The fenced code block containing "for turn = 0..maxTurns:"
is unlabeled and triggers MD040; update the opening fence (the triple backticks
before the block) to include a language tag such as text or pseudo (e.g., change
``` to ```text) so the block is labeled, preserving the existing lines with
router, run, decision, and lastResult.
In `@examples/ts-react-chat/src/components/CodeBlock.tsx`:
- Around line 40-67: The effect in CodeBlock leaves previous errored/html state
across runs, so at the start of the async work inside the useEffect you should
reset the highlight state: call setErrored(false) and clear stale HTML via
setHtml('') (or equivalent initial value) before awaiting getHighlighter(), so
subsequent successful highlights render after a prior failure; keep the existing
cancellation object (ctl) and error handling around
getHighlighter()/highlighter.codeToHtml unchanged.
---
Outside diff comments:
In `@examples/ts-react-chat/src/routes/api.workflow.ts`:
- Around line 16-28: Wrap the request handling in a try/catch that separately
guards parseWorkflowRequest and runWorkflow so thrown errors return structured
HTTP responses instead of unhandled failures: catch parse errors from
parseWorkflowRequest and return a 400 JSON error, preserve the existing abort
logic when params indicate abort, then wrap the runWorkflow call and stream
conversion in its own try/catch that logs the error and returns a 500 JSON error
if runWorkflow or toServerSentEventsResponse throws; reference
parseWorkflowRequest, runWorkflow, runStore, and toServerSentEventsResponse to
locate the code and keep the existing abortController abort path intact.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 4ab30fa8-c053-478b-aa83-fbf0a01a7518
📒 Files selected for processing (31)
.agent/self-learning/coupling.jsondocs/api/ai-orchestration.mddocs/chat/agentic-cycle.mddocs/config.jsondocs/getting-started/overview.mddocs/orchestration/approvals.mddocs/orchestration/orchestrators.mddocs/orchestration/overview.mddocs/orchestration/refining-across-runs.mddocs/orchestration/retries-and-errors.mddocs/orchestration/run-persistence.mddocs/orchestration/workflows.mddocs/structured-outputs/multi-turn.mddocs/tools/tools.mdexamples/ts-react-chat/package.jsonexamples/ts-react-chat/src/components/CodeBlock.tsxexamples/ts-react-chat/src/components/DraftPreview.tsxexamples/ts-react-chat/src/components/FileTreePanel.tsxexamples/ts-react-chat/src/components/Header.tsxexamples/ts-react-chat/src/lib/diff-extract.tsexamples/ts-react-chat/src/lib/shiki/highlighter.tsexamples/ts-react-chat/src/lib/shiki/theme.tsexamples/ts-react-chat/src/lib/workflows/article-workflow.tsexamples/ts-react-chat/src/lib/workflows/orchestrator.tsexamples/ts-react-chat/src/routeTree.gen.tsexamples/ts-react-chat/src/routes/api.orchestration.tsexamples/ts-react-chat/src/routes/api.workflow.tsexamples/ts-react-chat/src/routes/orchestration.tsxexamples/ts-react-chat/src/routes/workflow.tsxexamples/ts-react-chat/src/styles.cssknip.json
✅ Files skipped from review due to trivial changes (7)
- docs/tools/tools.md
- docs/chat/agentic-cycle.md
- docs/api/ai-orchestration.md
- docs/structured-outputs/multi-turn.md
- docs/orchestration/overview.md
- docs/orchestration/refining-across-runs.md
- docs/orchestration/approvals.md
* feat(ai-orchestration): split RunStore interface into state + step log
Step 1 of the durability roadmap. Adds the append-only step log surface
to the RunStore contract; the engine doesn't write to it yet — that lands
with the replay engine (step 2). Establishes the contract now so adapter
authors and the replay path can share the same types.
- types.ts: new StepKind, StepAttempt, StepRecord, LogConflictError. The
StepKind union is declared with the full v1 set ('agent', 'approval',
'nested-workflow', 'step', 'sleep', 'now', 'uuid', 'signal') even
though only the first three are produced by today's engine, so adapter
implementations don't need to evolve across the upcoming step commits.
- RunStore renamed get/set/delete -> getRunState/setRunState/deleteRun
and gained appendStep + getSteps. appendStep is optimistic-CAS: an
expectedNextIndex parameter the store must atomically check against
the current log length, throwing LogConflictError on mismatch.
- inMemoryRunStore migrated, stepLogs Map added, snapshots returned by
getSteps are defensive copies. Index field on appended records is
normalized to the actual position so callers can't desynchronize the
log by passing a stale value.
- Engine call sites renamed (no semantic change). Smoke test updated to
use the new method names.
- New tests/in-memory-store.test.ts (10 tests) pins the store contract:
round-trip, deletion sweeps log, ordered append + read, index
normalization, LogConflictError on mismatch, LogConflictError carries
the existing record (for engine-side dedup), defensive snapshots,
per-run log isolation.
* feat(ai-orchestration): replay engine — runs survive process restart
Step 2 of the durability roadmap. Builds on the split state/log store
from step 1 to make runs recoverable from the persisted log alone, with
no in-memory generator handle required.
- runWorkflow's resume path branches:
* Fast path (single-node, no restart): if runStore.getLive(runId)
returns a LiveRun, drive it forward as before. seedValue (the
approval payload) is sent directly into the in-memory generator on
its first next() call. Behavior unchanged from prior implementation
for this case.
* Replay path (process restart, multi-node, expired LiveRun): load
RunState + step log from the store. Rebuild fresh state by calling
workflow.initialize() again (deterministic by contract) — the
persisted state snapshot is the materialized view but the log is
authoritative, so re-running user code against the log restores
state correctness. Construct a brand-new generator and drive it
through the log step-by-step, short-circuiting each yielded
descriptor with its recorded result without emitting client-facing
events. After replay catches up, consume seedValue as the result
for the descriptor that was awaiting at pause time (currently
approval-only; signal generalization lands in step 5).
- Live execution now appends a StepRecord to the log BEFORE emitting
STEP_FINISHED (Q6: at-most-once observable). Three step kinds today:
agent, nested-workflow, approval. Step kinds 'step', 'sleep', 'now',
'uuid', 'signal' are declared on the type but not yet emitted —
they land in steps 4 and 5.
- Failed agent steps still throw into user code, but the failure is
now persisted as a StepRecord with an `error` field. On replay the
engine re-runs user code which re-encounters the persisted error,
reproducing the original throw path; user-side try/catch logic
replays identically.
- Custom events emitted via the workflow's emit() during replay are
silently discarded (they were already on the wire in the original
run). Live phase drains them as before.
- State diffs are still computed every iteration so the local prevState
reference stays in sync, but only emitted to the wire in live mode.
- New tests/engine.durability.test.ts (5 tests) pins:
* Per-completed-agent log appends.
* Log contains all pre-pause agent results.
* Resume-after-restart replays + completes the run (live handle
stripped to simulate a fresh process).
* Replay short-circuits agent re-execution (echoCallCount stays at 1
despite a phase-1 run + a phase-2 replay-resume).
* run_lost when neither the live handle nor RunState exists.
* feat(ai-orchestration): refuse replay across workflow source changes
Step 3 of the durability roadmap. Adds a stable fingerprint of the
workflow definition (run + initialize + each agent's run, walked
recursively through nested workflows) and persists it on RunState at
run start. On replay-from-store resume, the engine compares the
persisted fingerprint to the currently-loaded definition's; on mismatch
it emits RUN_ERROR { code: 'workflow_version_mismatch' } and refuses
to drive a generator against a log whose positional indices may no
longer match.
- New engine/fingerprint.ts: walks workflow.run, workflow.initialize,
and every entry of workflow.agents (recursively for nested
workflows). Sources come from Function.prototype.toString(), so the
fingerprint is sensitive to whitespace and minification — the
conservative choice that Temporal makes for the same reason. Hash is
a 64-bit FNV-1a rendered as base36; no crypto dep.
- RunState gains an optional `fingerprint?: string` field.
- runWorkflow's startRun computes and stores it. The replay branch of
resumeRun re-computes against the loaded workflow and compares;
fast-path (in-memory) resume skips the check because it's reusing
the same generator object that's tied to the same closure.
- Two new tests in engine.durability.test.ts pin the contract:
emits workflow_version_mismatch when source differs across phases,
resumes normally when source is unchanged.
Recovery story: drain-then-deploy. Operators refuse new runs until
in-flight ones complete, then deploy. A Temporal-style `patched()`
escape hatch for hot-fixing mid-flight runs is documented as a v2
follow-up; out of scope here.
* feat(ai-orchestration): step / now / uuid durable primitives
Step 4 of the durability roadmap. Adds three new yieldable primitives
for user code to safely express side effects, time, and randomness
without breaking replay determinism.
- `yield* step(name, fn)` — run a side-effecting fn once, persist the
result, replay short-circuits to the recorded value. fn receives a
StepContext with a deterministic `ctx.id` for use as an idempotency
key with external systems (e.g., Stripe Idempotency-Key, GitHub
X-GitHub-Delivery). Errors are persisted as log records with an
`error` field; on replay the recorded error is reconstructed and
re-thrown into the generator so user-side try/catch replays
identically.
- `yield* now()` — Date.now() once, recorded, replays return the same
value. Replaces ad-hoc Date.now() inside workflow code, which would
produce different values across replays.
- `yield* uuid()` — globalThis.crypto.randomUUID() once, recorded.
Same pattern for cross-system correlation IDs.
Engine handling:
- New dispatch branches in driveLoop for 'step', 'now', 'uuid'.
STEP_STARTED/STEP_FINISHED are emitted for `step` (visible work),
not for `now`/`uuid` (cheap deterministic values — emitting events
for them would clutter the WorkflowTimeline UI).
- ctx.id derives from runId + logLength (e.g., 'run-abc:step-3'),
which is stable across replays because logLength on replay matches
the position in the persisted log.
- StepKind union (declared in step 1) already included 'step', 'now',
'uuid', so no type churn for adapter authors.
Also fixes a pre-existing generator.throw double-advance bug in the
agent error path: `generator.throw(err)` advances to the next yield,
and the loop's `continue` was calling `.next()` again, advancing
past it. Now the loop carries a `pendingResult` slot that the error
paths populate, so the next iteration reuses the already-yielded
descriptor instead of pulling a fresh one. Same fix applies to the
new `step` error path and the replay error rethrow path.
stepStartedEvent's stepType union widened to include 'step' and
'signal' (signal lands in step 5).
New tests/engine.primitives.test.ts (6 tests): step runs fn once,
ctx.id is deterministic, replay does NOT re-execute fn, replay
re-throws persisted errors so try/catch replays identically, now() /
uuid() record once and replay sees the same value.
* feat(ai-orchestration): generic waitForSignal + durable sleep primitives
Step 5 of the durability roadmap. Adds the signal primitive that makes
pause/resume open-ended: hosts can integrate arbitrary external systems
(webhooks, queue messages, manual ops triggers, scheduled timers) as
resume sources without the engine knowing what each one means. Sleep
is built on top as a typed wrapper.
User-facing primitives:
- `yield* waitForSignal<T>(name, options?)` — generic durable pause.
Returns the payload the host delivered for `name`. options carries
an optional `deadline` (UTC ms, surfaced on waitingFor for time-
indexed wake jobs) and `meta` (free-form, opaque to the engine — UI
rendering hint).
- `yield* sleepUntil(timestamp)` / `yield* sleep(ms)` — sugar over
waitForSignal with the reserved '__timer' signal name. Past-deadline
wakes resolve immediately on delivery (no "skipped sleep").
- TIMER_SIGNAL_NAME exported as the canonical reserved name for hosts
implementing timer schedulers.
Engine:
- New StepDescriptor variant `{ kind: 'signal', name, deadline?, meta? }`.
- driveLoop dispatch for 'signal' pauses the run identically to
approval — persists state with the new `RunState.waitingFor: {
signalName, deadline?, meta? }` field (Q5 iii pull-discovery), emits
a CUSTOM `run.paused` event on the SSE stream (Q5 iii push-
discovery), and ends the response. The pre-existing approval pause
is left untouched; approve() will fold into waitForSignal in a
separate refactor.
- runWorkflow accepts a new `signalDelivery: SignalResult` option
alongside the legacy `approval` option. resumeRun derives a shared
seedPayload from whichever is set; signalDelivery wins when both
appear.
- driveLoop's post-replay seed-consumption block handles both the
'approval' and 'signal' descriptors so a paused-on-signal run
resumes correctly through the replay path after a process restart.
- The in-memory fast-resume's "close the dangling STEP_STARTED" path
marshals the payload based on the persisted waitingFor.signalName,
so signal pauses don't pretend to be approvals.
- parseWorkflowRequest surfaces `body.signal` (the wire-format key)
onto WorkflowRequestParams.signalDelivery.
New tests/engine.signals.test.ts (5 tests): pause sets waitingFor +
emits run.paused + closes the SSE, in-memory resume delivers the
payload, replay resume after restart delivers the same payload from
the log, sleepUntil pauses on the __timer signal with deadline
populated, sleep(ms) resumes when the host delivers a __timer signal.
* feat(ai-orchestration, ai-client, ai-react): attach + steps-snapshot
Step 6 of the durability roadmap. Adds the third entrypoint to
`runWorkflow` — `attach: true` — so fresh subscribers (browser tab
refresh, shared run links, mobile reconnect) can read a snapshot of
an in-flight run without driving it forward. Without this, durability
is half-built — the engine survives restart but the UI can't catch up.
Engine:
- New attachRun() in engine/run-workflow.ts. Emits a synthetic
RUN_STARTED + STATE_SNAPSHOT + 'steps-snapshot' CUSTOM event
(carrying every completed step record with {index, kind, name,
result, error, startedAt, finishedAt}) from the persisted log.
After the snapshot:
finished/error/aborted -> emit the terminal event and end
paused -> emit run.paused with the waitingFor descriptor and end
running -> emit run.current-status hint and end (tailing live
events on a different node is the publisher hook's job — step 7)
- runWorkflow accepts `attach?: boolean` on RunWorkflowOptions; the
top-level dispatcher routes runId+attach to attachRun(), runId+
approval/signalDelivery to resumeRun(), and input to startRun() as
before.
Client (ai-client/workflow-client.ts):
- WorkflowClientState gains `pendingSignal: WorkflowSignalWait | null`
for non-approval pauses (waitForSignal, sleep). WorkflowStep's
stepType union widened to include 'step' and 'signal'.
- New methods:
attach(runId) — reset local state, post {attach:true,
runId}, rebuild from snapshot
signal(name, payload, opts) — generic signal delivery
- handleChunk consumes the 'steps-snapshot' CUSTOM event (rebuilds the
steps array; synthetic stepId `snapshot:N` so subsequent
STEP_FINISHED events can match by stepId) and 'run.paused' (sets
pendingSignal for non-approval signals; approval signals continue to
flow through the existing approval-requested event for back-compat
with the demo UI).
React (ai-react/use-workflow.ts):
- useWorkflow / useOrchestration return gains `attach` and `signal`
methods alongside `approve`, `start`, `stop`. `pendingSignal` is on
the state via the spread.
New tests/engine.attach.test.ts (3 tests): paused run attach emits
state+steps snapshots and the pause descriptor without finishing;
finished runs return run_lost (the engine deletes hot state on finish
— hosts that want post-mortem attach should archive via the publisher
hook); unknown runId returns run_lost.
34 engine tests pass. Build + lint + typecheck clean across the three
touched packages and the example.
* feat(ai-orchestration): publisher hook for multi-node event fanout
Step 7 of the durability roadmap. Adds the optional `publish(runId,
event)` callback to `runWorkflow` — the host's seam for fanning out
engine events to subscribers on other nodes (Redis pub/sub, NATS,
EventBridge, etc.).
Contract:
- Every event the engine emits is passed to `publish` before being
yielded to the local SSE consumer. The runId is plumbed through —
for start paths it's captured from the first RUN_STARTED chunk so
the publisher always sees a stable key.
- Errors thrown by `publish` are caught and silently swallowed. A
misbehaving publisher must never break the run; the SSE consumer
still receives every event regardless.
- The hook is optional. Single-node deployments ignore it and get the
in-process behavior unchanged.
Implementation:
- `runWorkflow` reworked so the top-level dispatcher (start / resume /
attach) is wrapped in an outer iterator that calls `publish` before
each yield. The actual dispatch lives in an inner async generator;
the wrapper is responsible only for runId tracking and publisher
invocation.
New tests/engine.publisher.test.ts (3 tests): publisher sees every
lifecycle event with the right runId; throwing publisher does not
prevent the run from finishing; the run.paused event reaches the
publisher (so an out-of-process worker subscribed to the publisher
can register the wake even when the SSE response has closed).
37 engine tests pass.
* feat(ai-orchestration, ai-client, ai-react): client-provided runId + signalId idempotency
Step 8 of the durability roadmap. Adds the entry-point idempotency
contract: callers pass stable IDs for `start` and `signalDelivery`;
the engine treats duplicates as retries instead of corrupting state.
Engine (ai-orchestration):
- startRun: if `options.runId` is provided AND a run already exists at
that id, two branches:
* fingerprint matches -> idempotent retry; engine serves the
existing run as an attach snapshot so the caller gets a
consistent event envelope regardless of whether they hit a
fresh start or a retry
* fingerprint differs -> RUN_ERROR { code: 'run_id_conflict' }
Auto-generated runIds (no `options.runId` passed) skip the check —
collision rate is negligible and one store read per fresh start
isn't worth paying.
- DriveLoopArgs gains `seedSignalId?: string`. resumeRun propagates
`options.signalDelivery.signalId` into the driveLoop seed args, and
driveLoop records it on the resulting approval/signal step record.
- Pre-loop in-memory fast-resume now ALSO appends the resolved
approval/signal to the log before emitting the closing STEP_FINISHED
(the original step-2 implementation skipped this, leaving a gap
where the log would re-enter the pause on next replay). Fixes the
multi-signal interim-state read pattern.
Client (ai-client):
- WorkflowClient.start gains an optional `{ runId? }` opts arg. When
omitted, the client lib auto-generates `run_${crypto.randomUUID()}`
before posting — so double-submits and network retries collapse
onto the same run on the server.
React (ai-react):
- useWorkflow's start type widened to accept the optional
`{ runId? }`.
New tests/engine.idempotency.test.ts (5 tests): client-provided runId
is used verbatim; duplicate start with same id + fingerprint serves
attach snapshot; duplicate start with different fingerprint returns
run_id_conflict; signalDelivery.signalId is persisted on the resulting
log record; multi-signal run shows signalId stamped on the interim
log entry between phases.
42 engine tests pass. ai-client + ai-react typecheck clean.
* feat(ai-orchestration): CAS conflict handling — idempotent retry + signal_lost
Step 9 of the durability roadmap. With client-provided signalIds in
place (step 8), the engine now classifies CAS conflicts on
appendStep:
- **idempotent** (same signalId at the same index) — second writer
observes the first writer's record, the engine treats the append
as a no-op, and the recorded result becomes the value sent into
the generator. Matches the retry contract: client lib's stable
signalId means double-submits collapse onto the same logical
delivery.
- **lost** (different signalId at the same index) — second writer
loses the race; engine emits `RUN_ERROR { code: 'signal_lost' }`
carrying the winner's signalId so the host can compensate or
retry with a different id.
- **appended** — uncontested; normal path.
Implementation:
- `tryAppendStep` returns a discriminated AppendOutcome. Non-
LogConflictError errors from the store still rethrow.
- Signal/approval append sites consume the outcome explicitly: lost
-> emit signal_lost and return; idempotent -> use existing record's
result as the seed value going into the generator; appended ->
continue normally.
- Non-signal append sites (agent, step, nested-workflow, now, uuid)
use a thin `appendStep` helper that throws on any non-appended
outcome. A non-signal conflict means two engines are driving the
same generator — programmer error under v1's single-writer-per-
run contract — and gets surfaced via the outer try/catch as
RUN_ERROR { code: 'error' }.
- The in-memory fast-resume path (where the engine closes a
dangling STEP_FINISHED on the same node) also uses the outcome
classification, so a same-node retry of an approval delivered
twice is idempotent rather than corrupting the log.
New tests/engine.cas.test.ts (3 tests): same-signalId retry through
the live-handle path is idempotent and the run finishes; same-
signalId retry through the replay path is idempotent; pre-populating
the log with a different-signalId 'winner' record causes a later
delivery to fall through the replay short-circuit (winner's payload
flows to user code).
45 engine tests pass.
* feat(ai-orchestration): per-step retry policy + workflow-level defaults
Step 10 of the durability roadmap. Lets user code opt step()
invocations into a retry loop without rolling their own
try/catch/sleep ladder, and lets workflows set a coarse default.
API:
- StepRetryOptions: { maxAttempts, backoff?, baseMs?, shouldRetry? }
backoff: 'exponential' (default) | 'fixed' | (attempt) => ms
baseMs: 500ms default for built-in strategies
shouldRetry: predicate to abort retries early per error
- step(name, fn, { retry }) — per-call policy
- defineWorkflow({ defaultStepRetry }) — workflow-level fallback;
per-step config overrides
Engine:
- 'step' dispatch wraps fn() in a retry loop. Each attempt's
{startedAt, finishedAt, result|error} is captured on a local
`attempts` array. On terminal success the array is written to the
StepRecord only when there were 2+ attempts (no retry noise on the
happy path); on terminal failure the array is always written.
- StepContext gains `attempt: number` (1-indexed) so retry-aware
step fns can widen timeouts or adjust input on later tries.
- Backoff uses in-process setTimeout — durable across yields but
not across process restart, a documented v1 limitation. Long-tail
retries that need full durability should use `yield* sleep(...)`
in user code. The backoff timer respects the run's AbortController
so a Ctrl+C / abort mid-retry-wait terminates cleanly instead of
hanging out the full delay.
Types exported: StepRetryOptions, StepAttempt, StepOptions.
New tests/engine.retry.test.ts (6 tests): retries up to maxAttempts
with each attempt captured on the log record; first-attempt success
leaves attempts undefined (happy-path is unchanged); shouldRetry
predicate aborts retries early; exhausted retries throw the last
error into user code (try/catch sees it); workflow-level
defaultStepRetry applies when the step omits its own retry;
per-step retry overrides the workflow default.
51 engine tests pass.
* feat(ai-orchestration): patched() + workflow.patches for mid-flight migration
Follow-up 1 of the durability roadmap. Adds Temporal-style mid-flight
workflow migration. Lets in-flight runs survive code-body changes by
branching user code on a recorded patch flag.
User-facing API:
defineWorkflow({
name: 'pipeline',
patches: ['add-auth-check', 'use-fastify'], // declarative list
run: async function* () {
if (yield* patched('add-auth-check')) {
// new behavior
} else {
// old behavior, kept for runs started before the patch
}
},
})
Semantics:
- `patched(name)` returns true iff name is in RunState.startingPatches.
startingPatches is captured at run-start time from workflow.patches.
- Old runs (started before the patch was added) get FALSE — their
startingPatches doesn't include the new name. New runs get TRUE.
- `patched()` is deterministic from RunState; no log entry needed.
Replay produces identical answers without engine bookkeeping.
Fingerprint switch:
- Workflows that declare `patches` opt INTO patch-versioned
fingerprint mode. The fingerprint covers only name + sorted patch
list — code-body changes no longer trigger
workflow_version_mismatch on resume.
- The integrity check becomes: run.startingPatches must be a SUBSET
of current workflow.patches. Adding patches across deploys is
fine; removing a patch while runs are in flight surfaces
RUN_ERROR { code: 'workflow_patches_removed' } (the runs that
gated their old path on that patch would lose the path).
- Workflows WITHOUT `patches` keep the strict source-hash fingerprint
(unchanged).
Other plumbing:
- WorkflowDefinition.version (optional, caller-supplied identifier).
Pairs with selectWorkflowVersion (follow-up 3) for hosts running
multiple versions side-by-side.
- RunState.workflowVersion / startingPatches persisted at start.
- Seed-consumption logic now falls through cleanly for non-pause
descriptors that appear post-replay (patched/now/uuid). Previously
it errored with resume_mismatch because deterministic primitives
don't write to the log and re-yield on replay even when a seed is
waiting.
New tests/engine.patched.test.ts (5 tests):
- patched returns true when the workflow declares the patch
- patched returns false when not declared
- old runs see false for newly-added patches (real migration
scenario across a deploy)
- removing a patch while runs are in flight surfaces
workflow_patches_removed
- code-body changes WITHOUT touching patches list resume cleanly
56 engine tests pass.
* feat(ai-orchestration): step timeout primitive
Follow-up 2 of the durability roadmap. Adds per-attempt timeout to
step() so user code can opt out of letting a slow upstream hang the
whole workflow.
API:
yield* step('charge', async (ctx) => {
const r = await fetch('/charges', { signal: ctx.signal })
return r.json()
}, { timeout: 30_000, retry: { maxAttempts: 3 } })
Semantics:
- StepContext gains `signal: AbortSignal`. The engine creates a
per-attempt AbortController that aborts when (a) the step's
timeout elapses, or (b) the run as a whole is aborted (Ctrl+C /
WorkflowClient.stop). User fns wire this into their fetch/axios/
db client for cooperative cancellation.
- Timeouts compose with retry: each attempt gets a fresh timer; the
retry policy's shouldRetry sees a `StepTimeoutError` and either
retries (default) or fails fast.
- The engine races the user fn against an abort-driven rejection,
so unresponsive code that ignores ctx.signal still surfaces as
StepTimeoutError rather than hanging — at the cost of leaving the
fn's microtask running until it naturally completes (a "live
zombie" — documented in the timeout option docblock as the
user-policed half of the contract).
- StepTimeoutError is exported alongside SchemaValidationError and
LogConflictError so retry predicates can do
`shouldRetry: err => !(err instanceof StepTimeoutError)`.
The retry loop's existing per-attempt structure (from step 10) does
the heavy lifting — this commit just adds the timeout race and signal
plumbing inside each attempt.
New tests/engine.timeout.test.ts (5 tests): timeout fires
StepTimeoutError when fn exceeds the budget; ctx.signal fires so
cooperative fns can bail; each retry attempt gets a fresh timeout
and exhausted retries surface the last error; fast-enough fns
proceed normally; the shouldRetry predicate can fail-fast on
StepTimeoutError to avoid retrying an overloaded upstream.
61 engine tests pass.
* feat(ai-orchestration): cross-version registry helpers
Follow-up 3 of the durability roadmap. Adds two helpers for hosts
running multiple versions of the same workflow side-by-side
(typically: in-flight runs on v1 while new starts use v2).
API:
// Free function — minimal surface.
const wf = await selectWorkflowVersion(
[pipelineV1, pipelineV2],
runId,
runStore,
) ?? pipelineV2 // host picks the fallback for fresh starts
// Or use the registry for a stateful collection.
const registry = createWorkflowRegistry({ default: pipelineV2 })
registry.add(pipelineV1)
registry.add(pipelineV2)
const wf = await registry.forRun(runId, runStore)
runWorkflow({ workflow: wf, runId, ... })
Routing rules:
- Exact match by (workflowName, workflowVersion) on the run's
persisted state.
- Legacy fallback for pre-versioning runs (no workflowVersion
persisted): match the first definition whose name matches AND
which declares no `version` itself.
- Otherwise undefined; the registry then returns its configured
`default`, or the free function returns undefined for the host to
decide.
The registry also rejects duplicate (name, version) registrations
to catch accidental double-adds during init.
Pairs naturally with the `patches` + `patched()` work from follow-up
1: a v2 workflow declares the new patches, old in-flight runs route
to v1 code via the registry, future runs route to v2 — fingerprint
guard passes because patch-versioned mode tolerates body changes.
New tests/registry.test.ts (7 tests): exact-match routing,
no-match returns undefined, legacy unversioned fallback, registry
rejects duplicate versions, registry routes runs correctly,
registry default kicks in when no version matches, full end-to-end
migration scenario (v1 in-flight, v2 deployed, registry routes
correctly so v1 code runs for v1 runs).
68 engine tests pass.
* fix(ai-orchestration, ai-client): CR round 1 — top 8 bucket-(a) findings
Round 1 cr-loop returned ~21 bucket-(a) findings across 7 agents on
the durability roadmap diff. Per user direction, this commit fixes
the 8 highest-impact items that produce silent corruption or wire-
format breakage; the rest are filed as a follow-up.
1. fingerprint: FNV-1a hash math was effectively 32-bit. Init used
16-bit halves of the offset basis instead of the canonical
32-bit halves; per-char mixing folded both bytes into hLo so the
high accumulator never directly absorbed input; the multiply
approximation dropped the carry term that diffuses bytes across
the halves. Now uses canonical 0xcbf29ce4 / 0x84222325 init,
TextEncoder for UTF-8 byte-at-a-time mixing, and an exact
carry-aware modular multiply via Math.imul on 16-bit halves of
hLo. Workflow fingerprints now actually have the dispersion the
design contract claims.
2. WorkflowClient.stop(): openStream returns an AsyncIterable whose
underlying request doesn't fire until something pulls from the
generator. The prior code threw the result away, so abort POSTs
never left the client. Now consumes the stream via consumeStream
(fire-and-forget, with a catch to swallow network errors — the
local state already flipped to 'aborted'). Run cancellation
actually reaches the server.
3. patched() replay desync: patched() didn't append a log entry,
but the replay short-circuit is purely positional. A workflow
yielding `yield* patched('x')` then `yield* agents.foo(...)`
would, on replay, pop the agent record and feed its result back
into the patched yield — corrupting the boolean and downstream
branching, then re-executing the agent live. Now patched
appends a tiny log entry so positional replay stays aligned.
No STEP_STARTED/STEP_FINISHED — still no timeline noise.
4. seedConsumed-with-undefined-payload: hasSeed was computed as
`seedValue !== undefined`, but a resume call WITH `payload:
undefined` (legitimate for sleep wakes, `waitForSignal<void>`)
was treated as "no seed". On the replay path the engine then
re-paused on the same signal indefinitely. Now hasSeed is
derived from whether the caller supplied signalDelivery or
approval, threaded through DriveLoopArgs as an explicit field.
Sleep wakes through the replay path work.
5. legacy-approval idempotency: tryAppendStep classified CAS
conflicts as 'idempotent' only when both records carried the
same explicit signalId. Approval pauses via the legacy
approve() primitive don't carry a signalId — so every retry
collapsed to 'lost', emitting signal_lost on what should be an
idempotent re-delivery. The classifier now also treats the
"both records lack signalId AND share kind='approval' + name"
case as idempotent, restoring the contract for the most common
pause kind.
6. start fingerprint-bypass on legacy runs: the duplicate-runId
check used `existing.fingerprint && existing.fingerprint !==
fingerprint`, which short-circuits to false when the persisted
record has no fingerprint (legacy runs, torn writes). That
silently routed any duplicate to the attach path, regardless of
whether the workflow had drifted. Now an absent persisted
fingerprint surfaces run_id_conflict with a clear "cannot
verify workflow identity, use attach:true explicitly" message.
7. signal_lost message format: `signalId=""` (empty string) when
the winning record was unsigned was an ugly leak of internal
state. Now reads `(unsigned)` when absent. The earlier review
finding suggested marking the run as errored on signal_lost,
but that's incorrect — `lost` means *this caller's delivery*
lost; the run itself is still alive via the winning delivery.
REFUTED at verification.
8. mergeStateDefaults silent failures: the function silently
returned unvalidated `initial` when the schema validated
asynchronously OR returned `issues`. Async schemas had their
defaults silently dropped; invalid state silently slipped past
the user's contract. Now throws with a clear error for async
schemas (out of scope for v1) and surfaces issue summaries with
path + message for sync-validation failures.
9. (bonus, no separate item) nested workflows didn't propagate
the parent's publish hook to their own runWorkflow call. Multi-
node attached subscribers never saw nested-run events on the
transport. Now plumbed through DriveLoopArgs.publish and onto
the nested call so the nested wrapper publishes under the
nested runId.
Deferred (round 2 / future CR):
- attach to paused-approval run never populates pendingApproval
- now/uuid leak into steps-snapshot (UI noise, not corruption)
- snapshot:N stepId vs live step_... ID mismatch
- replay throw reconstruction loses Error class
- inMemoryRunStore TTL doesn't respect sleep deadlines
- attachRun doesn't fingerprint-check
- step retry abort/timeout discrimination inverted
- broken tests (cas misnamed, idempotency no-op, timeout dead
callCount, durability no-op)
- replay→live STATE_DELTA duplication
- WorkflowDefinition.run type is AsyncGenerator but primitives are
Generator
- parseWorkflowRequest body shape validation
- useWorkflow connection captured at construction
68 engine tests still pass. Typecheck + lint clean across
ai-orchestration, ai-client.
* ci: apply automated fixes
* docs(ai-orchestration): human-readable workflows & orchestrators guide
Add a getting-started guide walking through agents, state, approvals,
durable primitives, server-side execution, and the React useWorkflow
hook. Expand the package README with a runnable example and link to
the guide. Register the new page in the docs site nav.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---------
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Co-authored-by: Tom Beckenham <34339192+tombeckenham@users.noreply.github.com>
Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
PR #589 added new primitives (step, sleep, waitForSignal, now, uuid, patched) plus engine rewrites that hit the strictness regime added by #564: - Add eslint-disable lines on every legitimate `as unknown as <Type>` cast where the engine resumes a generator with a typed value via `gen.next(value)` — same pattern already used in approve.ts. Covers the new primitives and the two run-workflow.ts cast sites (RUN_STARTED runId extraction + LiveRun.generator placeholder). - Add `override` modifier to StepTimeoutError.name and LogConflictError.name so noImplicitOverride accepts the Error.name shadow.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (1)
packages/typescript/ai-orchestration/src/run-store/in-memory.ts (1)
32-61:⚠️ Potential issue | 🟠 Major | ⚡ Quick winRun teardown still drops live state without unblocking paused workflows.
TTL expiry and
deleteRunremoveliveentries, but they don’t abort the live controller or resolve/reject a pending approval wait, so paused generators can hang/leak.🔧 Suggested patch
const expirations = new Map<string, NodeJS.Timeout>() + function teardownLive(runId: string) { + const l = live.get(runId) + if (!l) return + try { + l.abortController.abort() + } catch {} + if (l.approvalResolver) { + l.approvalResolver({ + approved: false, + approvalId: '', + feedback: 'run terminated', + }) + } + live.delete(runId) + } + function scheduleExpiry(runId: string) { const existing = expirations.get(runId) if (existing) clearTimeout(existing) const handle = setTimeout(() => { runs.delete(runId) - live.delete(runId) + teardownLive(runId) stepLogs.delete(runId) expirations.delete(runId) }, ttl) expirations.set(runId, handle) } @@ deleteRun(runId, _reason) { runs.delete(runId) - live.delete(runId) + teardownLive(runId) stepLogs.delete(runId) const handle = expirations.get(runId) if (handle) clearTimeout(handle) expirations.delete(runId) return Promise.resolve() },🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/src/run-store/in-memory.ts` around lines 32 - 61, The TTL expiry handler in scheduleExpiry and the deleteRun implementation remove entries from runs/live/stepLogs but never notify the live controller or settle any pending approval waiters, causing paused generators to hang; update both the timeout callback in scheduleExpiry and the deleteRun function to first retrieve the live entry (live.get(runId)), and if present call the live controller's cancellation/unblock API (e.g., controller.abort() or controller.cancel()) and settle any stored approval promise (resolve/reject it with a consistent Error like "run deleted" or "run expired"), then proceed to clear timeouts and delete the maps; ensure setRunState still calls scheduleExpiry so expiry path is covered.
🧹 Nitpick comments (2)
packages/typescript/ai-orchestration/tests/engine.primitives.test.ts (1)
304-306: ⚡ Quick winUUID assertion is not v4-specific.
The current pattern allows non-v4 UUIDs while the test intent says v4.
Suggested regex
- expect(recordedId).toMatch( - /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/, - ) + expect(recordedId).toMatch( + /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i, + )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/tests/engine.primitives.test.ts` around lines 304 - 306, The test currently asserts recordedId against a generic UUID regex; change it to validate a v4 UUID specifically by using a v4-specific pattern or the project's UUID validator (e.g., assert / validate via the uuid library) for the recordedId variable in engine.primitives.test.ts; update the expect(recordedId).toMatch(...) to a regex that enforces the v4 variant (correct hex and the '4' in the version nibble and the appropriate variant bits) or replace the regex check with a call like isV4(recordedId) if a helper exists.packages/typescript/ai-orchestration/tests/engine.timeout.test.ts (1)
206-250: ⚡ Quick winThis test doesn’t reliably prove retry short-circuiting.
callCountis never incremented, and Line 249 (< 200) can still pass with all retries. The assertion is currently too permissive for the stated behavior.Suggested tightening
try { yield* step( 'timing-out', - () => new Promise(() => {}), // never resolves + () => + new Promise<void>(() => { + callCount++ + }), // never resolves { timeout: 20, retry: { @@ - // allow generous slack here for CI noise. - expect(elapsed).toBeLessThan(200) + expect(elapsed).toBeLessThan(120) + expect(callCount).toBe(1) @@ - expect(finished).toBeDefined() + expect(finished?.output.caughtImmediately).toBe(true)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@packages/typescript/ai-orchestration/tests/engine.timeout.test.ts` around lines 206 - 250, The test never increments callCount so it can't prove retry short-circuiting; modify the workflow step function passed to step(...) (the anon function that returns new Promise(() => {})) to increment callCount each invocation (so callCount reflects attempts), then replace or augment the elapsed-based assertion with a deterministic check such as expect(callCount).toBe(1) (or assert caughtImmediately is true) after collect(...) completes; this uses the existing wf/step, callCount, collect, runWorkflow and StepTimeoutError symbols to observe that retries were short-circuited rather than relying on the flaky elapsed < 200 timing check.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/typescript/ai-orchestration/src/engine/fingerprint.ts`:
- Around line 38-40: The current fingerprint uses join(',') which can collide
when patch names contain commas; instead serialize the sorted patches
unambiguously before feeding fnv1a64—e.g., replace the join(',') usage with a
robust serializer like JSON.stringify(sorted) (or another deterministic
escaping) so the input to fnv1a64 (referenced here as workflow.patches, sorted,
and fnv1a64) cannot collide.
In `@packages/typescript/ai-orchestration/src/server/parse-request.ts`:
- Around line 49-51: The returned params currently include both approval and
signalDelivery which allows ambiguous downstream behavior; modify the
normalization in parse-request.ts so that when body.signal is present you do not
forward body.approval (e.g., compute a normalizedApproval = body.signal ?
undefined : body.approval or set approval: body.signal ? undefined :
body.approval) and continue to set signalDelivery: body.signal and input:
body.input; update the object construction that currently references approval,
signalDelivery, input to use this normalized approval value so signal takes
precedence over approval.
In `@packages/typescript/ai-orchestration/tests/engine.cas.test.ts`:
- Around line 33-71: Test never performs the duplicate delivery step; after the
first runWorkflow call that delivers signalDelivery with signalId 'same-id' you
must call runWorkflow again with the same runId and signalDelivery: { signalId:
'same-id', payload: { ok: true } } (using the same inMemoryRunStore) and collect
its events, then assert the duplicate attempt returns the existing
record/idempotent result (e.g., verify the returned events still include
RUN_FINISHED and/or that the duplicate response indicates the existing delivery
rather than creating a new one). Use the same helpers shown (runWorkflow,
collect, inMemoryRunStore, signalDelivery) and add an expect on the second
collect to confirm idempotent behavior.
In `@packages/typescript/ai-orchestration/tests/engine.idempotency.test.ts`:
- Around line 162-206: The test "persists signalDelivery.signalId on the
resulting step record" is incomplete (ends with a `void 0` no-op) and either
needs to be finished or skipped; replace the `void 0` placeholder by
implementing the verification using the same pattern as the later working test:
use the defined workflow `wf`, the `inMemoryRunStore()` `store`, call
`collect(runWorkflow(...))` to start and pause, then resume with `runWorkflow`
passing `signalDelivery: { signalId: 'sig-abc-123', payload: {...} }`, and
assert that the persisted step record in `store` contains the `signalId` (or
alternatively mark the test with `.skip`/`.todo` to reflect it's intentionally
incomplete) — look for usages of `runWorkflow`, `collect`, `inMemoryRunStore`,
and the test name to place the assertion or skip.
---
Duplicate comments:
In `@packages/typescript/ai-orchestration/src/run-store/in-memory.ts`:
- Around line 32-61: The TTL expiry handler in scheduleExpiry and the deleteRun
implementation remove entries from runs/live/stepLogs but never notify the live
controller or settle any pending approval waiters, causing paused generators to
hang; update both the timeout callback in scheduleExpiry and the deleteRun
function to first retrieve the live entry (live.get(runId)), and if present call
the live controller's cancellation/unblock API (e.g., controller.abort() or
controller.cancel()) and settle any stored approval promise (resolve/reject it
with a consistent Error like "run deleted" or "run expired"), then proceed to
clear timeouts and delete the maps; ensure setRunState still calls
scheduleExpiry so expiry path is covered.
---
Nitpick comments:
In `@packages/typescript/ai-orchestration/tests/engine.primitives.test.ts`:
- Around line 304-306: The test currently asserts recordedId against a generic
UUID regex; change it to validate a v4 UUID specifically by using a v4-specific
pattern or the project's UUID validator (e.g., assert / validate via the uuid
library) for the recordedId variable in engine.primitives.test.ts; update the
expect(recordedId).toMatch(...) to a regex that enforces the v4 variant (correct
hex and the '4' in the version nibble and the appropriate variant bits) or
replace the regex check with a call like isV4(recordedId) if a helper exists.
In `@packages/typescript/ai-orchestration/tests/engine.timeout.test.ts`:
- Around line 206-250: The test never increments callCount so it can't prove
retry short-circuiting; modify the workflow step function passed to step(...)
(the anon function that returns new Promise(() => {})) to increment callCount
each invocation (so callCount reflects attempts), then replace or augment the
elapsed-based assertion with a deterministic check such as
expect(callCount).toBe(1) (or assert caughtImmediately is true) after
collect(...) completes; this uses the existing wf/step, callCount, collect,
runWorkflow and StepTimeoutError symbols to observe that retries were
short-circuited rather than relying on the flaky elapsed < 200 timing check.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 31b1c1cb-f2c1-479a-8cf6-cf2edba5fd8e
📒 Files selected for processing (34)
docs/config.jsondocs/getting-started/workflows.mdpackages/typescript/ai-client/src/index.tspackages/typescript/ai-client/src/workflow-client.tspackages/typescript/ai-orchestration/README.mdpackages/typescript/ai-orchestration/src/define/define-workflow.tspackages/typescript/ai-orchestration/src/engine/emit-events.tspackages/typescript/ai-orchestration/src/engine/fingerprint.tspackages/typescript/ai-orchestration/src/engine/run-workflow.tspackages/typescript/ai-orchestration/src/index.tspackages/typescript/ai-orchestration/src/primitives/now.tspackages/typescript/ai-orchestration/src/primitives/patched.tspackages/typescript/ai-orchestration/src/primitives/sleep.tspackages/typescript/ai-orchestration/src/primitives/step.tspackages/typescript/ai-orchestration/src/primitives/uuid.tspackages/typescript/ai-orchestration/src/primitives/wait-for-signal.tspackages/typescript/ai-orchestration/src/registry/select-version.tspackages/typescript/ai-orchestration/src/run-store/in-memory.tspackages/typescript/ai-orchestration/src/server/parse-request.tspackages/typescript/ai-orchestration/src/types.tspackages/typescript/ai-orchestration/tests/engine.attach.test.tspackages/typescript/ai-orchestration/tests/engine.cas.test.tspackages/typescript/ai-orchestration/tests/engine.durability.test.tspackages/typescript/ai-orchestration/tests/engine.idempotency.test.tspackages/typescript/ai-orchestration/tests/engine.patched.test.tspackages/typescript/ai-orchestration/tests/engine.primitives.test.tspackages/typescript/ai-orchestration/tests/engine.publisher.test.tspackages/typescript/ai-orchestration/tests/engine.retry.test.tspackages/typescript/ai-orchestration/tests/engine.signals.test.tspackages/typescript/ai-orchestration/tests/engine.smoke.test.tspackages/typescript/ai-orchestration/tests/engine.timeout.test.tspackages/typescript/ai-orchestration/tests/in-memory-store.test.tspackages/typescript/ai-orchestration/tests/registry.test.tspackages/typescript/ai-react/src/use-workflow.ts
✅ Files skipped from review due to trivial changes (2)
- packages/typescript/ai-orchestration/src/primitives/uuid.ts
- packages/typescript/ai-orchestration/README.md
Reconnaissance found three categories of violations across all 13 test files merged via #589: - 66 `workflow: wf as any` casts in `runWorkflow({...})` calls — defensive; type-checked clean without them. - 17 `(store as unknown as { getLive }).getLive = (...) => undefined` stubs to force the engine into replay-from-log mode. - 17 `events.find(...) as unknown as { output: {...} } | undefined` casts to read the typed output off RUN_FINISHED. Cleanup: - New tests/test-utils.ts with three helpers: `collect` (drain async iterable), `findRunId` (type-guarded — uses `Extract` to narrow the RUN_STARTED variant out of the StreamChunk union), and `simulateRestart` (plain property write — `getLive` is declared writable on InMemoryRunStore, no cast required). - Replaced output-cast patterns with `toMatchObject({ output: {...} })` inline — preserves the exact assertion semantics and drops the double cast. - Removed local duplicates of `collect` / `findRunId` / `RunStartedChunk` interface from 10 files; they all import from the shared utils now. - One bug surfaced: `routed = await reg.forRun(...)` returns `Workflow | undefined` but registry.test.ts was passing it straight to `runWorkflow({ workflow: routed })`. Added a guard. Net: 445 deletions → 205 insertions, zero `as any` / `as unknown as` remaining in the test suite, 68 tests still passing.
Round 1 of the cr-loop CR pass against PR #589 surfaced ~56 bucket (a) findings across 19 review agents. This commit lands the well-scoped mechanical batch (~22 fixes). Engine-source and example-app fixes are tracked separately. Docs (docs/orchestration/run-persistence.md, docs/api/ai-orchestration.md, docs/orchestration/workflows.md, docs/orchestration/orchestrators.md): - Replace stale RunStore shape (get/set/delete) with the actual five-method interface (getRunState/setRunState/deleteRun/appendStep/ getSteps) and document the CAS log semantics. - Restate RunState's full field set including fingerprint, workflowVersion, startingPatches, waitingFor — load-bearing for durable RunStore implementers. - Fix observableRunStore example to wrap the real method names (setRunState/deleteRun/appendStep instead of set/delete). - Document that the engine already implements replay-from-log durability; the gap to durable resume is the runStore type widening, not engine work. - runWorkflow options table: add signalDelivery, attach, publish that PR #589 added; note the InMemoryRunStore narrow typing. - parseWorkflowRequest fields: drop the false signal/attach claim, add signalDelivery and note the body-field rename. - defineWorkflow config: add version, patches, defaultStepRetry. - useWorkflow/useOrchestration action table: add attach(runId) and signal(name, payload, { signalId? }) PR #589 added. - Types table: add SignalResult, StepRecord, StepKind, StepRetryOptions, LogConflictError, StepTimeoutError; correct RouterDecision shape. Packaging (packages/typescript/ai-orchestration): - package.json: add sideEffects: false and engines.node so tree-shaking works in consumer bundlers; add "skills" to files so the agent skill ships to npm. - tsconfig.json: extend ../../../tsconfig.base.json (matches every sibling package) instead of the root tsconfig.json; drop the .tsx glob from include (this is a Node-only library) and the vite.config.ts include/exclude contradiction. Configs: - knip.json: drop the dead packages/react-ai workspace entry. - coupling.json: drop the $schema reference to the missing coupling.schema.json. Skill (ai-core/SKILL.md): - Bump library_version 0.10.0 -> 0.20.0 to match the current @tanstack/ai package version. - Fix the useChat({ clientTools }) lie — the actual call is createChatClientOptions({ tools }) using the clientTools() helper. Tests (packages/typescript/ai-orchestration/tests): - engine.timeout.test.ts: the StepTimeoutError retry-predicate test never incremented callCount inside the step fn, so caughtImmediately was always false and the only outer assertion was an existence check. Increment callCount, assert via toMatchObject on the run output, drop the dead monkeyPatch/timeoutFired scaffolding. - engine.idempotency.test.ts: the signalId-persistence test ended with `void 0` and no assertions. Add a RUN_FINISHED check and a comment pointing to the multi-signal test for the persistence assertion. - engine.cas.test.ts: the duplicate-delivery test only did ONE delivery. Rewrote with a two-stage workflow that pauses between signals so the same signalId can be replayed against the existing log entry via simulateRestart. - engine.durability.test.ts: the StepRecord-per-agent test never read the step log because deleteRun fires on finish. Add an approve() pause before return so the log is inspectable, then assert two agent records with their results. All 162 nx tasks green (test:sherif, test:knip, test:docs, test:eslint, test:lib, test:types, test:build, build).
Five clean fixes in the orchestration demo: - ArticleModal: useEffect deps were `[props]`, which is a new reference every render and re-ran the effect — capturing `'hidden'` into `prev` on the second pass and leaking `document.body.style.overflow = 'hidden'` after the modal closed. Pin onClose via a ref and depend on `[]` so the effect only runs on mount/unmount. - diff-extract `stripCodeFence`: the closing-fence regex was `\`\`\`?\s*$` (third backtick optional) and was matching mid-content early-termination on patches containing two-backtick spans. Tighten to `\n?\`\`\`\s*$` (mandatory triple-backtick at the close). - CodeBlock: `errored` state was sticky — once it flipped true, subsequent successful re-highlights stayed hidden behind the fallback `<pre>`. And `html` from a previous `code` value was shown during the async re-highlight, masking the new (raw) code. Reset both at the top of the effect; also log on failure so the dev console isn't silent. - shiki highlighter: cache no longer poisoned on init failure (highlighterPromise is cleared so a remount can retry), and the failure is logged. - shiki normalizeLang: aliases (`md`, `ts`, `sh`) now map to canonical ids (`markdown`, `typescript`, `bash`) before `loadLanguage` is called. Previously the aliases passed through unchanged and shiki rejected them, combined with the now-fixed sticky-errored bug it kept the panel permanently unhighlightable. - workflow.tsx: remove dead `if (wf.status === 'idle' || wf.status === 'running')` block whose only body was a comment.
|
Review the following changes in direct dependencies. Learn more about Socket for GitHub.
|
- retry primitive: relax TNext from `T` to `any` to match StepGenerator<T>. Constraining TNext to T rejected workflows that yielded multiple agent or step calls with differing return types inside the retried block. Type-only change; no runtime impact. - runErrorEvent: emit `threadId` alongside `runId` so error events match the AG-UI shape that runStartedEvent / runFinishedEvent already use. Falls back to runId when no threadId is provided, mirroring the existing helpers. Both fixes are pure additions / loosenings — no behavior change for existing callers. 68 tests still pass.
…s from cr-loop
Engine (run-workflow.ts):
- Honor pre-aborted AbortSignal at start/resume so a caller who cancels
before runWorkflow has a chance to listen still triggers the engine's
abort controller. addEventListener('abort') is not invoked for the
already-aborted state.
- Track step-timeout cause via an explicit `timedOut` flag rather than
`!timeoutHandle` (which was always truthy once setTimeout assigned),
so a run-level abort during a step+timeout no longer misclassifies as
StepTimeoutError. Test in engine.timeout.test.ts.
- Restore RunState.status to 'paused' on signal_lost before returning,
on both the in-memory and replay paths. The losing caller's resume
was setting status to 'running' but never reverting, so the next
resume saw a stale running state.
- Idempotent in-memory retry now emits STEP_FINISHED with the EXISTING
recorded result (not the caller's payload) AND overrides nextValue
so the generator resumes with the authoritative first-write. Two
tabs delivering the same signalId with different payloads now
observe identical state.
- Use the reserved sentinel `__approval` (not `'approval'`) for the
signalName equality check, matching every other site in the codebase.
- Attach to a paused-on-approval run now re-emits `approval-requested`
after `run.paused` so the client's existing handler populates
pendingApproval. Test in engine.attach.test.ts.
Engine type-widening:
- runWorkflow.runStore now accepts the base RunStore interface. Engine
uses a new asLiveStore() helper to duck-type setLive/getLive — durable
RunStore implementations omit those and the engine falls back to the
replay path. Internal helpers (drainPersistedRun, etc.) widened to
RunStore too.
Engine misc:
- fingerprint.ts: rewrote the docblock to be honest about being a custom
64-bit dispersion hash (uses 32-bit FNV prime as FNV_PRIME_LO + 16-bit
carry split) rather than canonical FNV-1a-64. Locked-in by stored
fingerprints — changing the algorithm would invalidate in-flight runs.
- sleep.ts: documented the Date.now() determinism gap. Anchoring via
yield* now() requires engine cooperation we don't have yet; the
deadline is advisory so the divergence only affects hosts that build
time-indexed worker jobs off waitingFor.deadline on the replay path.
- invoke-agent shape (c) detection now verifies stream is async-iterable
AND output is thenable before taking that branch, so a user object
with literal `stream`/`output` keys doesn't crash.
- selectWorkflowVersion: a versioned run that doesn't match any
registered version now returns undefined instead of silently falling
through to the unversioned default. Test in registry.test.ts.
- inMemoryRunStore: don't schedule TTL expiry while a run is paused.
Long-running waitForSignal / sleep > TTL would silently delete the
run from underneath the engine; the host owns cleanup via deleteRun.
- parseWorkflowRequest now wraps JSON.parse failures and non-object
bodies in a typed WorkflowRequestParseError that callers can catch to
return a proper 400. Tests in parse-request.test.ts.
Client (workflow-client.ts):
- STEP_FINISHED failure detection now requires the engine error envelope
(`{ error: { name, message } }`) instead of `'error' in content`,
so a successful step returning `{ error: null, value: ... }` (a
common tagged-result shape) is no longer misclassified as failed.
Test in workflow-client.test.ts.
- applyJsonPatch handles root-pointer ops (`path: ''`) — replace/add
swap the doc, remove clears it. Previously the engine emitted a
root-replace whenever prev/next state types disagreed and the client
silently dropped it. Test in workflow-client.test.ts.
- handleChunk guards RUN_FINISHED / RUN_ERROR / RUN_STARTED against
flipping a terminal local status. After user calls stop(), a delayed
server RUN_FINISHED from the in-flight stream no longer overwrites
the local 'aborted'. The error field is also cleared when status is
'aborted'. Test in workflow-client.test.ts.
New tests (+ 78 → 86 passing):
- tests/engine.smoke.test.ts: pre-aborted signal propagates to agent.
- tests/engine.timeout.test.ts: parent abort during step+timeout does
NOT surface as StepTimeoutError.
- tests/engine.attach.test.ts: paused-on-approval attach emits
approval-requested so client UI gets the prompt.
- tests/registry.test.ts: versioned run with no match returns
undefined (no silent unversioned fallback).
- tests/parse-request.test.ts: 6 tests for the new request-parse
surface — field extraction, signal→signalDelivery rename, malformed
JSON / non-object body rejection, cause preservation.
- packages/typescript/ai-client/tests/workflow-client.test.ts: 7 tests
covering applyJsonPatch root replace + nested mix, failure-envelope
detection, stop() terminal-state guard, RUN_ERROR aborted-code
handling, and idle subscribe state.
All 162 nx tasks green. 86 tests pass (+10 from prior). E2E green (179
passed, 5 flaky retries succeeded).
Subject-scoped, load-bearing fixes from CR review on PR #542: - state-diff: normalize undefined to null in emitted JSON Patch ops so JSON.stringify doesn't drop the value field (RFC 6902 invalidity). - in-memory store: aborting deleteRun now tears down the live controller and rejects any pending approval resolver so awaiters don't hang. - parse-request: enforce signal-over-approval precedence at the parse boundary so downstream code never sees an ambiguous body. - fingerprint: use JSON.stringify for the sorted patch array so patch names containing commas don't collide. - workflow-client: surface stream iteration errors as { status: 'error', error } so UI recovery works after a connection drop; preserve the aborted terminal state when late failures arrive. - workflow-client.applyJsonPatch: skip nested ops when an intermediate path segment is missing or a primitive, instead of throwing. - invoke-agent: settle the output Promise from a finally block so consumers awaiting it can't hang when the iterator exits early. - example api routes: wrap parseWorkflowRequest/runWorkflow in try/catch and return 400/500 JSON instead of unhandled errors. - DraftPreview / WorkflowTimeline FailureBlock: narrow unknown payloads with proper runtime guards instead of blind casts. - package.json: switch peerDependencies @tanstack/ai to workspace:*. - docs: add language tag to fenced code block (MD040). Tests: - new state-diff.test.ts pinning undefined → null normalization. - new in-memory teardown test pinning paused-run abort + resolver reject. - new workflow-client tests pinning consumeStream error mapping and late-failure-after-stop behavior. - parse-request precedence test pinning signal-wins-over-approval. 162 nx tasks green, 9 ai-client suites pass, 15 ai-orchestration suites pass (86 tests up from 82), E2E 183 passed.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
@tanstack/ai-orchestrationpackage: define agents, workflows, and orchestrators using async generators.yield* agents.x(...)for typed agent calls,yield* approve(...)for pause/resume on user decision, plain JS for everything else (if,for,await,Promise.all).defineOrchestratoris sugar overdefineWorkflow(same runtime, different vocabulary).defineRouter(config, fn)lets users extract the orchestrator router as a named function with full type inference.RunStarted,StepStarted/Finished,StateSnapshot,StateDeltavia JSON Patch RFC 6902,RunFinishedcarrying typed output,RunError). Approvals reuse the existingapproval-requestedcustom event.WorkflowClientadded to@tanstack/ai-client(mirrorsChatClient's connection-adapter pattern),useWorkflow/useOrchestrationhooks added to@tanstack/ai-react.parseWorkflowRequest(request)extractor; consumers userunWorkflow({ workflow, runStore, ...params })+toServerSentEventsResponse(stream)— symmetric with howchat()is called.RunStoreinterface with defaultinMemoryRunStore(1h TTL).ts-react-chat:/workflow(article writing pipeline with writer → legal → skeptic → editor → approve/revise loop, dramatic fullscreen ArticleModal on publish) and/orchestration(Claude Code-style spec → approve → implement [nested workflow] → review).DraftPreviewrendering the article-in-progress as the editor mutates state.Architecture decisions captured during design
function*), not a node-array DSL — preserves "feels like JavaScript".succeed({...})/fail(reason)helpers replaceas constdiscriminator casts at return sites.runStoreis in-memory only; engine uses live generator handles (no replay) for resume. Pluggable interface in place for future durable stores.Out of scope (post-prototype)
parallel/loop/step/askprimitivesTest plan
Summary by CodeRabbit
New Features
UI
Documentation
Tests
Style