feat(ai-orchestration): generator-based workflows + orchestrators#542
Open
AlemTuzlak wants to merge 47 commits intomainfrom
Open
feat(ai-orchestration): generator-based workflows + orchestrators#542AlemTuzlak wants to merge 47 commits intomainfrom
AlemTuzlak wants to merge 47 commits intomainfrom
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.
Replace the manual body-cast, dispatch logic, and toServerSentEventsResponse wiring with a single \`handleWorkflowRequest\` call. Both api.workflow and api.orchestration routes now read like plain configuration.
Remove the endpoint discriminated union from WorkflowClientOptions and WorkflowClient.openStream. Connection adapter is now the only way to connect; the endpoint shortcut moves to the public fetchWorkflowEvents helper.
Mirror fetchServerSentEvents but typed for workflow body shape (input / runId / approval). Reuses the shared readStreamLines SSE parser already in connection-adapters.ts.
…Options Mirrors the existing fetchServerSentEvents re-export so the workflow import pattern is symmetric with the chat pattern.
…orkflowRequest Delete the magic do-everything handleWorkflowRequest wrapper. Replace with parseWorkflowRequest(request) which returns params to spread into runWorkflow, mirroring how chat routes pull messages + data out of JSON.
Client pages now use connection: fetchWorkflowEvents('/api/...') instead
of endpoint shortcut. API routes use the 3-line parse->run->respond
pattern that mirrors the existing chat route structure exactly.
…FINISHED event Include the workflow's typed return value in the RUN_FINISHED chunk and capture it in WorkflowClientState.output so the client can render the published article from wf.output instead of the last step result.
…al response Extend ApprovalResult with optional feedback field, thread it through the approve() method on WorkflowClient and the useWorkflow hook so the UI can request targeted editor revisions instead of binary approve/deny.
…-way approval UI Restructure the article workflow to edit-first-then-approve: after legal and skeptic reviews the editor runs immediately, then a revision loop (max 4 rounds) asks Publish/Revise/Discard. Revise re-runs the editor with the textarea feedback. The UI gains a textarea and a third button; the published article now renders from wf.output (RUN_FINISHED output).
Contributor
📝 WalkthroughWalkthroughThis PR introduces ChangesOrchestration Engine & Core Infrastructure
Example Workflows & UI
Packaging & Build
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
✨ Finishing Touches📝 Generate docstrings
🧪 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 16354b7
☁️ 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-vue
@tanstack/ai-vue-ui
@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
+9
to
+11
| const draft = ( | ||
| props.draft && typeof props.draft === 'object' ? props.draft : null | ||
| ) as Draft | null |
Contributor
There was a problem hiding this comment.
Harden draft shape validation before using paragraphs.map.
draft is only object-checked/cast, so malformed payloads (e.g. paragraphs: "text" or {}) can reach Line 77 and crash when .map is invoked.
Suggested fix
interface Draft {
title?: string
paragraphs?: Array<string>
}
export function DraftPreview(props: { draft: unknown; phase?: string }) {
- const draft = (
- props.draft && typeof props.draft === 'object' ? props.draft : null
- ) as Draft | null
+ const raw = props.draft
+ const draft: Draft | null =
+ raw && typeof raw === 'object'
+ ? {
+ title: typeof (raw as { title?: unknown }).title === 'string'
+ ? (raw as { title: string }).title
+ : undefined,
+ paragraphs: Array.isArray((raw as { paragraphs?: unknown }).paragraphs)
+ ? (raw as { paragraphs: unknown[] }).paragraphs.filter(
+ (p): p is string => typeof p === 'string',
+ )
+ : undefined,
+ }
+ : nullAlso applies to: 24-26, 77-92
🤖 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 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.
Comment on lines
+175
to
+177
| function FailureBlock(props: { result: unknown }) { | ||
| const result = props.result as { error?: { message?: string } } | ||
| const msg = result.error?.message ?? JSON.stringify(props.result) |
Contributor
There was a problem hiding this comment.
Type assertion lacks runtime validation.
The code asserts props.result as { error?: { message?: string } } without validation. If the actual error structure differs, accessing result.error?.message might fail silently or produce unexpected output. Consider adding runtime validation or a more defensive check.
🛡️ Suggested fix
function FailureBlock(props: { result: unknown }) {
- const result = props.result as { error?: { message?: string } }
- const msg = result.error?.message ?? JSON.stringify(props.result)
+ const msg =
+ typeof props.result === 'object' &&
+ props.result !== null &&
+ 'error' in props.result &&
+ typeof props.result.error === 'object' &&
+ props.result.error !== null &&
+ 'message' in props.result.error &&
+ typeof props.result.error.message === 'string'
+ ? props.result.error.message
+ : JSON.stringify(props.result)
return (🤖 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 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.
Comment on lines
+15
to
+23
| POST: async ({ request }) => { | ||
| const params = await parseWorkflowRequest(request) | ||
| const stream = runWorkflow({ | ||
| runStore, | ||
| workflow: featureOrchestrator, | ||
| ...params, | ||
| }) | ||
| return toServerSentEventsResponse(stream) | ||
| }, |
Contributor
There was a problem hiding this comment.
Add error handling for the workflow execution.
The POST handler does not wrap parseWorkflowRequest or runWorkflow in try/catch blocks. If either throws (e.g., malformed request body, workflow initialization failure), the server will return an unhandled error response. Consider adding error handling to return structured error responses to the client.
🛡️ Proposed fix
POST: async ({ request }) => {
+ try {
const params = await parseWorkflowRequest(request)
const stream = runWorkflow({
runStore,
workflow: featureOrchestrator,
...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.orchestration.ts` around lines 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.
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
+109
to
+124
| async approve(approved: boolean, feedback?: string): Promise<void> { | ||
| if (!this.clientState.pendingApproval || !this.clientState.runId) { | ||
| throw new Error('No pending approval') | ||
| } | ||
| const approvalId = this.clientState.pendingApproval.approvalId | ||
| const runId = this.clientState.runId | ||
| this.setState({ | ||
| pendingApproval: null, | ||
| status: 'running', | ||
| }) | ||
| const workflowStream = this.openStream({ | ||
| approval: { approvalId, approved, feedback }, | ||
| runId, | ||
| }) | ||
| await this.consumeStream(workflowStream) | ||
| } |
Contributor
There was a problem hiding this comment.
Stream exceptions are not mapped into client state.
If connect()/iteration throws, start() and approve() reject but state can remain running with no error, which breaks UI recovery.
Proposed fix
private async consumeStream(stream: AsyncIterable<unknown>): Promise<void> {
- for await (const raw of stream) {
- this.handleChunk(raw as Record<string, unknown>)
+ try {
+ for await (const raw of stream) {
+ this.handleChunk(raw as Record<string, unknown>)
+ }
+ } catch (err) {
+ this.setState({
+ error: {
+ message: err instanceof Error ? err.message : 'Unknown workflow stream error',
+ },
+ status: 'error',
+ })
+ throw err
}
}Also applies to: 126-133, 146-150
🤖 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/workflow-client.ts` around lines 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.
Comment on lines
+51
to
+62
| function mergeStateDefaults( | ||
| workflow: AnyWorkflowDefinition, | ||
| initial: Record<string, unknown>, | ||
| ): Record<string, unknown> { | ||
| if (workflow.stateSchema) { | ||
| const validated = workflow.stateSchema['~standard'].validate(initial) | ||
| if (!(validated instanceof Promise) && !validated.issues) { | ||
| return validated.value as Record<string, unknown> | ||
| } | ||
| } | ||
| return initial | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🌐 Web query:
Standard Schema spec ~standard.validate sync vs async return type
💡 Result:
In the Standard Schema v1 specification, the ~standard.validate function has a dual return type: Result<Output> | Promise<Result<Output>>. It can return either a synchronous Result object (for success with value and undefined issues, or failure with issues array) or a Promise resolving to that Result if async operations are involved [1][2][3][4]. Libraries are encouraged to prefer synchronous validation when possible, only returning Promises when necessary [1][2]. Consumers expecting only sync validation can check if the result is a Promise and throw an error if so [1][5][2]. Some libraries like Zod attempt sync first then fallback to async, potentially causing multiple calls for async schemas [6].
Citations:
- 1: https://standardschema.dev/schema
- 2: https://github.com/standard-schema/standard-schema/tree/24384111d696b8539bc8d6ac9d131c71305f7c11
- 3: https://registry.npmjs.org/@standard-schema/spec
- 4: https://github.com/standard-schema/standard-schema
- 5: Explicit sync/async schemas standard-schema/standard-schema#22
- 6: [v4] async .check() function gets called twice when validating with StandardSchema ["~standard"].validate(...) colinhacks/zod#5137
🏁 Script executed:
# First, let's locate and examine the actual file and function
find . -type f -name "run-workflow.ts" | head -20Repository: TanStack/ai
Length of output: 121
🏁 Script executed:
# Check the full context of the file around the mergeStateDefaults function
if [ -f "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" ]; then
wc -l "packages/typescript/ai-orchestration/src/engine/run-workflow.ts"
fiRepository: TanStack/ai
Length of output: 123
🏁 Script executed:
# Search for all occurrences of mergeStateDefaults to find call sites
rg "mergeStateDefaults" --type ts --type tsx -A 3 -B 3Repository: TanStack/ai
Length of output: 83
🏁 Script executed:
# Check the complete function and surrounding context
if [ -f "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" ]; then
cat -n "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" | head -100
fiRepository: TanStack/ai
Length of output: 3825
🏁 Script executed:
# Search for mergeStateDefaults usage (fixing the ts/tsx issue)
rg "mergeStateDefaults" --type ts -A 3 -B 3Repository: TanStack/ai
Length of output: 1397
🏁 Script executed:
# Get more of the file to see how mergeStateDefaults is called
cat -n "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" | sed -n '100,200p'Repository: TanStack/ai
Length of output: 3359
🏁 Script executed:
# Check if there are any other call sites of mergeStateDefaults
rg "mergeStateDefaults" --type ts -B 5 -A 5Repository: TanStack/ai
Length of output: 2097
🏁 Script executed:
# Check if the function is exported or used anywhere else
rg "mergeStateDefaults" --type tsRepository: TanStack/ai
Length of output: 248
🏁 Script executed:
# Let's also check the schema validation behavior by looking at types
rg "stateSchema" --type ts -B 2 -A 2 | head -50Repository: TanStack/ai
Length of output: 1607
🏁 Script executed:
# Let's see the complete context of how state is used after mergeStateDefaults
cat -n "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" | sed -n '117,145p'Repository: TanStack/ai
Length of output: 1035
🏁 Script executed:
# Check if there are any type constraints that would prevent making mergeStateDefaults async
rg "RunState" --type ts -A 5 -B 2 | head -40Repository: TanStack/ai
Length of output: 3542
🏁 Script executed:
# Let's verify the Standard Schema import and usage to understand the validation contract
rg "~standard" --type ts -B 3 -A 3Repository: TanStack/ai
Length of output: 8437
🏁 Script executed:
# Check how stateSchema is typed
rg "stateSchema" --type ts -B 3 -A 3 | head -60Repository: TanStack/ai
Length of output: 2135
🏁 Script executed:
# Let's look at the invoke-agent error handling more closely
cat -n "packages/typescript/ai-orchestration/src/engine/invoke-agent.ts" | sed -n '1,100p'Repository: TanStack/ai
Length of output: 3925
🏁 Script executed:
# Check the exact error handling pattern for async validation in invoke-agent
rg "Async.*schema validation not supported" --type ts -B 5 -A 2Repository: TanStack/ai
Length of output: 1719
🏁 Script executed:
# Let's check if there are any tests for mergeStateDefaults behavior
rg "mergeStateDefaults" --type tsRepository: TanStack/ai
Length of output: 248
🏁 Script executed:
# Check the full parseOutput function to see how agent output handles async
cat -n "packages/typescript/ai-orchestration/src/engine/invoke-agent.ts" | sed -n '115,145p'Repository: TanStack/ai
Length of output: 1238
🏁 Script executed:
# Let's check if there's any documentation or comments about why async validation is rejected
rg "Async.*schema validation not supported" --type ts -B 10 -A 5 | head -80Repository: TanStack/ai
Length of output: 3248
🏁 Script executed:
# Check if there are any other schema validation patterns we should be aware of
rg "\.validate\(" --type ts -B 2 -A 4 | head -100Repository: TanStack/ai
Length of output: 3562
🏁 Script executed:
# Check the schema-converter pattern for context - how it handles the Promise case
cat -n "packages/typescript/ai/src/activities/chat/tools/schema-converter.ts" | sed -n '70,130p'Repository: TanStack/ai
Length of output: 2311
🏁 Script executed:
# Let's find the validateSync and async validation patterns in schema-converter
rg "validateSync|validate\(" "packages/typescript/ai/src/activities/chat/tools/schema-converter.ts" -B 5 -A 8Repository: TanStack/ai
Length of output: 894
🏁 Script executed:
# Let's check what the standard practice is - look for any comments or documentation about stateSchema
rg "stateSchema" --type ts -B 5 -A 5 | grep -A 10 -B 10 "comment\|//\|validation"Repository: TanStack/ai
Length of output: 37
🏁 Script executed:
# Final check: see the complete behavior of mergeStateDefaults one more time with context
cat -n "packages/typescript/ai-orchestration/src/engine/run-workflow.ts" | sed -n '45,175p'Repository: TanStack/ai
Length of output: 4683
mergeStateDefaults silently loses schema defaults on async validation.
When workflow.stateSchema['~standard'].validate(initial) returns a Promise (which the Standard Schema spec allows for any conforming schema), the early-return branch is skipped and the function returns the unparsed initial instead of the schema-coerced value. The run then proceeds with state missing schema-applied defaults and coercions, silently diverging from the typed state.
The codebase establishes a clear pattern: either explicitly reject async validation (as in invoke-agent.ts) or await it (as in schema-converter.ts). mergeStateDefaults does neither. Since the call site in startRun is already async, make this function async with await to properly handle both sync and async validation paths.
🛡️ Suggested fix
-function mergeStateDefaults(
+async function mergeStateDefaults(
workflow: AnyWorkflowDefinition,
initial: Record<string, unknown>,
-): Record<string, unknown> {
+): Promise<Record<string, unknown>> {
if (workflow.stateSchema) {
- const validated = workflow.stateSchema['~standard'].validate(initial)
- if (!(validated instanceof Promise) && !validated.issues) {
- return validated.value as Record<string, unknown>
- }
+ const validated = await workflow.stateSchema['~standard'].validate(initial)
+ if (!validated.issues) {
+ return validated.value as Record<string, unknown>
+ }
}
return initial
}🤖 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
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.
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
+32
to
+72
| function diff(prev: unknown, next: unknown, path: string): Array<Operation> { | ||
| if (Object.is(prev, next)) return [] | ||
|
|
||
| const prevIsObj = isObject(prev) | ||
| const nextIsObj = isObject(next) | ||
|
|
||
| // One is a primitive (or null), or types disagree — replace whole node. | ||
| if (!prevIsObj || !nextIsObj || Array.isArray(prev) !== Array.isArray(next)) { | ||
| return [{ op: 'replace', path: path || '', value: next }] | ||
| } | ||
|
|
||
| if (Array.isArray(prev) && Array.isArray(next)) { | ||
| // Length mismatch → replace the array. Same length → diff element-wise. | ||
| if (prev.length !== next.length) { | ||
| return [{ op: 'replace', path: path || '', value: next }] | ||
| } | ||
| const ops: Array<Operation> = [] | ||
| for (let i = 0; i < prev.length; i++) { | ||
| ops.push(...diff(prev[i], next[i], `${path}/${i}`)) | ||
| } | ||
| return ops | ||
| } | ||
|
|
||
| // Both are plain objects. | ||
| const prevObj = prev as Record<string, unknown> | ||
| const nextObj = next as Record<string, unknown> | ||
| const ops: Array<Operation> = [] | ||
| const allKeys = new Set([...Object.keys(prevObj), ...Object.keys(nextObj)]) | ||
|
|
||
| for (const key of allKeys) { | ||
| const subPath = `${path}/${escapeJsonPointer(key)}` | ||
| const prevHas = Object.prototype.hasOwnProperty.call(prevObj, key) | ||
| const nextHas = Object.prototype.hasOwnProperty.call(nextObj, key) | ||
|
|
||
| if (prevHas && nextHas) { | ||
| ops.push(...diff(prevObj[key], nextObj[key], subPath)) | ||
| } else if (nextHas) { | ||
| ops.push({ op: 'add', path: subPath, value: nextObj[key] }) | ||
| } else { | ||
| ops.push({ op: 'remove', path: subPath }) | ||
| } |
Contributor
There was a problem hiding this comment.
undefined values produce malformed JSON Patch ops on the wire.
When next (or a child value) is undefined, both the top-level replace (Line 40/46) and the per-key add (Line 69) emit { op, path, value: undefined }. JSON.stringify drops undefined properties, so the SSE consumer sees { "op": "add", "path": "..." } with no value — RFC 6902 invalid and the client applier will silently set the key to undefined (or throw, depending on the implementation).
If state values can ever be undefined, normalize to null (or omit the key entirely and emit remove instead).
🤖 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
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.
Comment on lines
+28
to
+55
| function scheduleExpiry(runId: string) { | ||
| const existing = expirations.get(runId) | ||
| if (existing) clearTimeout(existing) | ||
| const handle = setTimeout(() => { | ||
| runs.delete(runId) | ||
| live.delete(runId) | ||
| expirations.delete(runId) | ||
| }, ttl) | ||
| expirations.set(runId, handle) | ||
| } | ||
|
|
||
| return { | ||
| get(runId) { | ||
| return Promise.resolve(runs.get(runId)) | ||
| }, | ||
| set(runId, state) { | ||
| runs.set(runId, state) | ||
| scheduleExpiry(runId) | ||
| return Promise.resolve() | ||
| }, | ||
| delete(runId, _reason) { | ||
| runs.delete(runId) | ||
| live.delete(runId) | ||
| const handle = expirations.get(runId) | ||
| if (handle) clearTimeout(handle) | ||
| expirations.delete(runId) | ||
| return Promise.resolve() | ||
| }, |
Contributor
There was a problem hiding this comment.
Cleanup on delete / TTL expiry can leave pending approvals hanging.
When a run is paused at an approval step, LiveRun.approvalResolver is a promise resolver waiting for gen.next(approvalResult). If the TTL fires (Lines 31–35) or delete (Lines 48–55) is called while paused, the live entry is dropped but the resolver is never invoked and abortController is not aborted — any code awaiting that resolver, plus the underlying generator, will silently hang and leak. The pendingEvents queue is also discarded without notice.
Consider aborting the controller and rejecting the approval resolver as part of teardown so callers observe a clean failure rather than a stuck promise.
🛡️ Suggested teardown
+ function teardownLive(runId: string, reason: DeleteReason) {
+ const l = live.get(runId)
+ if (!l) return
+ try {
+ l.abortController.abort(new Error(`run ${reason}`))
+ } catch {}
+ if (l.approvalResolver) {
+ // Resolve as not-approved so awaiters unblock; engine can interpret.
+ l.approvalResolver({ approved: false, approvalId: '', feedback: `run ${reason}` })
+ }
+ live.delete(runId)
+ }
+
function scheduleExpiry(runId: string) {
const existing = expirations.get(runId)
if (existing) clearTimeout(existing)
const handle = setTimeout(() => {
+ teardownLive(runId, 'aborted')
runs.delete(runId)
- live.delete(runId)
expirations.delete(runId)
}, ttl)
expirations.set(runId, handle)
}
@@
- delete(runId, _reason) {
+ delete(runId, reason) {
+ teardownLive(runId, reason)
runs.delete(runId)
- live.delete(runId)
const handle = expirations.get(runId)
if (handle) clearTimeout(handle)
expirations.delete(runId)
return Promise.resolve()
},You may also want setTimeout(...).unref?.() so a long TTL doesn't keep a Node process alive after the request finishes.
🤖 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
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.
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.
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