diff --git a/.changeset/quiet-agents-work.md b/.changeset/quiet-agents-work.md new file mode 100644 index 000000000..7c5abcf8d --- /dev/null +++ b/.changeset/quiet-agents-work.md @@ -0,0 +1,8 @@ +--- +'@tanstack/ai-orchestration': minor +--- + +Add the first Workflow-backed AI orchestration package. It provides typed agent +definitions, a `ctx.ai.agent()` Workflow middleware, and adapters that project +Workflow events into TanStack AI AG-UI streams without introducing another +workflow runtime or store. diff --git a/packages/ai-orchestration/README.md b/packages/ai-orchestration/README.md new file mode 100644 index 000000000..6e7c0659b --- /dev/null +++ b/packages/ai-orchestration/README.md @@ -0,0 +1,60 @@ +# @tanstack/ai-orchestration + +Durable TanStack AI agent calls powered by TanStack Workflow. + +```ts +import { chat } from '@tanstack/ai' +import { + agentMiddleware, + defineAgent, + toAIStream, +} from '@tanstack/ai-orchestration' +import { openaiText } from '@tanstack/ai-openai' +import { + createWorkflow, + inMemoryRunStore, + runWorkflow, +} from '@tanstack/workflow-core' +import { z } from 'zod' + +const writer = defineAgent({ + name: 'writer', + input: z.object({ topic: z.string() }), + output: z.object({ article: z.string() }), + run: ({ input, abortController }) => + chat({ + adapter: openaiText('gpt-5.5'), + messages: [{ role: 'user', content: `Write about ${input.topic}` }], + outputSchema: z.object({ article: z.string() }), + stream: true, + abortController, + }), +}) + +const article = createWorkflow({ + id: 'article', + input: z.object({ topic: z.string() }), +}) + .middleware([agentMiddleware()]) + .handler(async (ctx) => + ctx.ai.agent('draft', writer, { topic: ctx.input.topic }), + ) + +const workflowEvents = runWorkflow({ + workflow: article, + runStore: inMemoryRunStore(), + input: { topic: 'durable execution' }, +}) + +for await (const chunk of toAIStream(workflowEvents)) { + // Send AG-UI chunks to the client or a durable delivery stream. +} +``` + +Workflow owns execution, replay, retries, approvals, signals, deadlines, stores, +and leases. This package only defines agent-step ergonomics and maps Workflow +events to TanStack AI's AG-UI stream. + +AI tool approvals inside an agent call are intentionally unsupported for now. +Use `ctx.approve()` between agent calls so Workflow can persist and resume the +pause. diff --git a/packages/ai-orchestration/package.json b/packages/ai-orchestration/package.json new file mode 100644 index 000000000..68e57ee50 --- /dev/null +++ b/packages/ai-orchestration/package.json @@ -0,0 +1,68 @@ +{ + "name": "@tanstack/ai-orchestration", + "version": "0.0.0", + "description": "Durable TanStack AI agent orchestration powered by TanStack Workflow.", + "author": "", + "license": "MIT", + "homepage": "https://tanstack.com/ai", + "repository": { + "type": "git", + "url": "git+https://github.com/TanStack/ai.git", + "directory": "packages/ai-orchestration" + }, + "bugs": { + "url": "https://github.com/TanStack/ai/issues" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + }, + "type": "module", + "module": "./dist/esm/index.js", + "types": "./dist/esm/index.d.ts", + "exports": { + ".": { + "types": "./dist/esm/index.d.ts", + "import": "./dist/esm/index.js" + } + }, + "sideEffects": false, + "engines": { + "node": ">=18" + }, + "files": [ + "dist", + "src" + ], + "scripts": { + "build": "vite build", + "clean": "premove ./build ./dist", + "lint:fix": "oxlint src --type-aware --fix", + "test:build": "publint --strict", + "test:coverage": "vitest run --coverage", + "test:oxlint": "oxlint src --type-aware", + "test:lib": "vitest run", + "test:lib:dev": "pnpm test:lib --watch", + "test:types": "tsc" + }, + "keywords": [ + "ai", + "tanstack", + "workflow", + "durable-execution", + "orchestration", + "agents" + ], + "dependencies": { + "@standard-schema/spec": "^1.1.0", + "@tanstack/workflow-core": "^0.0.4" + }, + "peerDependencies": { + "@tanstack/ai": "workspace:^" + }, + "devDependencies": { + "@tanstack/ai": "workspace:*", + "@vitest/coverage-v8": "4.0.14", + "zod": "^4.2.0" + } +} diff --git a/packages/ai-orchestration/src/agent.ts b/packages/ai-orchestration/src/agent.ts new file mode 100644 index 000000000..cf710879c --- /dev/null +++ b/packages/ai-orchestration/src/agent.ts @@ -0,0 +1,83 @@ +import type { StandardSchemaV1 } from '@standard-schema/spec' +import type { StreamChunk } from '@tanstack/ai' +import type { + AgentDefinition, + AgentRunResult, + AgentStreamResult, +} from './types' + +export interface DefineAgentConfig { + name: TName + description?: string + input?: StandardSchemaV1 + output?: StandardSchemaV1 + run: (context: { + input: TInput + signal: AbortSignal + abortController: AbortController + }) => AgentRunResult +} + +export function defineAgent( + config: DefineAgentConfig & { + output: StandardSchemaV1 + }, +): AgentDefinition +export function defineAgent< + TInput = unknown, + const TName extends string = string, +>( + config: DefineAgentConfig & { output?: undefined }, +): AgentDefinition +export function defineAgent( + config: DefineAgentConfig, +): AgentDefinition { + return { + kind: 'agent', + name: config.name, + description: config.description, + inputSchema: config.input, + outputSchema: config.output, + run: config.run, + } +} + +export function agentStream( + stream: AsyncIterable, + output: TOutput | Promise, +): AgentStreamResult { + return { kind: 'agent-stream', stream, output } +} + +export class AgentValidationError extends Error { + override name = 'AgentValidationError' + + constructor( + message: string, + readonly issues: ReadonlyArray, + ) { + super(message) + } +} + +export class AgentApprovalUnsupportedError extends Error { + override name = 'AgentApprovalUnsupportedError' + + constructor(agentName: string) { + super( + `Agent "${agentName}" paused for an AI tool approval inside a durable step. Use Workflow ctx.approve() between agent calls until approval resumption is defined.`, + ) + } +} + +export class AgentStreamError extends Error { + override name = 'AgentStreamError' + + constructor( + agentName: string, + message: string, + readonly code?: string, + ) { + super(`Agent "${agentName}" failed: ${message}`) + } +} diff --git a/packages/ai-orchestration/src/constants.ts b/packages/ai-orchestration/src/constants.ts new file mode 100644 index 000000000..3fed21e36 --- /dev/null +++ b/packages/ai-orchestration/src/constants.ts @@ -0,0 +1,8 @@ +export const AI_AGENT_META_KEY = 'tanstack.ai.agent' +export const AI_STREAM_CHUNK_EVENT = 'tanstack-ai.stream-chunk' + +export const WORKFLOW_APPROVAL_REQUESTED_EVENT = 'workflow.approval.requested' +export const WORKFLOW_APPROVAL_RESOLVED_EVENT = 'workflow.approval.resolved' +export const WORKFLOW_SIGNAL_AWAITED_EVENT = 'workflow.signal.awaited' +export const WORKFLOW_SIGNAL_RESOLVED_EVENT = 'workflow.signal.resolved' +export const WORKFLOW_STEP_FAILED_EVENT = 'workflow.step.failed' diff --git a/packages/ai-orchestration/src/events.ts b/packages/ai-orchestration/src/events.ts new file mode 100644 index 000000000..8d4db84b4 --- /dev/null +++ b/packages/ai-orchestration/src/events.ts @@ -0,0 +1,254 @@ +import { + AI_AGENT_META_KEY, + AI_STREAM_CHUNK_EVENT, + WORKFLOW_APPROVAL_REQUESTED_EVENT, + WORKFLOW_APPROVAL_RESOLVED_EVENT, + WORKFLOW_SIGNAL_AWAITED_EVENT, + WORKFLOW_SIGNAL_RESOLVED_EVENT, + WORKFLOW_STEP_FAILED_EVENT, +} from './constants' +import { EventType } from '@tanstack/ai' +import type { StreamChunk } from '@tanstack/ai' +import type { WorkflowEvent } from '@tanstack/workflow-core' + +export interface WorkflowEventMapperOptions { + /** AG-UI thread ID. Defaults to the Workflow run ID. */ + threadId?: string + /** AG-UI run ID or a resolver from Workflow run ID. Defaults to the Workflow run ID. */ + runId?: string | ((workflowRunId: string) => string) + /** Project Workflow state patches as AG-UI state patches. Disabled by default. */ + includeState?: boolean +} + +export interface WorkflowEventMappingContext extends WorkflowEventMapperOptions { + workflowRunId?: string +} + +export type WorkflowEventMapper = ( + event: WorkflowEvent, + workflowRunId?: string, +) => Array + +export function workflowEventToStreamChunks( + event: WorkflowEvent, + context: WorkflowEventMappingContext = {}, +): Array { + const workflowRunId = + event.type === 'RUN_STARTED' || + event.type === 'RUN_FINISHED' || + event.type === 'RUN_ERRORED' + ? event.runId + : context.workflowRunId + const runId = workflowRunId + ? resolveRunId(workflowRunId, context.runId) + : undefined + const threadId = context.threadId ?? workflowRunId + + switch (event.type) { + case 'RUN_STARTED': + return [ + { + type: EventType.RUN_STARTED, + timestamp: event.ts, + runId: runId ?? event.runId, + threadId: threadId ?? event.runId, + }, + ] + case 'RUN_FINISHED': + return [ + { + type: EventType.RUN_FINISHED, + timestamp: event.ts, + runId: runId ?? event.runId, + threadId: threadId ?? event.runId, + result: event.output, + }, + ] + case 'RUN_ERRORED': + return [ + { + type: EventType.RUN_ERROR, + timestamp: event.ts, + message: event.error.message, + code: event.code, + }, + ] + case 'STEP_STARTED': { + const agentName = readAgentName(event.meta) + return [ + { + type: EventType.STEP_STARTED, + timestamp: event.ts, + stepName: agentName ?? event.stepId, + stepId: event.stepId, + stepType: agentName ? 'agent' : 'workflow', + }, + ] + } + case 'STEP_FINISHED': { + const agentName = readAgentName(event.meta) + return [ + { + type: EventType.STEP_FINISHED, + timestamp: event.ts, + stepName: agentName ?? event.stepId, + stepId: event.stepId, + }, + ] + } + case 'STEP_FAILED': { + const agentName = readAgentName(event.meta) + return [ + { + type: EventType.STEP_FINISHED, + timestamp: event.ts, + stepName: agentName ?? event.stepId, + stepId: event.stepId, + }, + customChunk(event.ts, WORKFLOW_STEP_FAILED_EVENT, { + stepId: event.stepId, + error: event.error, + }), + ] + } + case 'APPROVAL_REQUESTED': + return [ + customChunk(event.ts, WORKFLOW_APPROVAL_REQUESTED_EVENT, { + stepId: event.stepId, + approvalId: event.approvalId, + title: event.title, + ...(event.description === undefined + ? {} + : { description: event.description }), + ...(event.meta === undefined ? {} : { meta: event.meta }), + }), + ] + case 'APPROVAL_RESOLVED': + return [ + customChunk(event.ts, WORKFLOW_APPROVAL_RESOLVED_EVENT, { + stepId: event.stepId, + approvalId: event.approvalId, + approved: event.approved, + ...(event.feedback === undefined ? {} : { feedback: event.feedback }), + ...(event.meta === undefined ? {} : { meta: event.meta }), + }), + ] + case 'SIGNAL_AWAITED': + return [ + customChunk(event.ts, WORKFLOW_SIGNAL_AWAITED_EVENT, { + stepId: event.stepId, + name: event.name, + ...(event.deadline === undefined ? {} : { deadline: event.deadline }), + ...(event.meta === undefined ? {} : { meta: event.meta }), + }), + ] + case 'SIGNAL_RESOLVED': + return [ + customChunk(event.ts, WORKFLOW_SIGNAL_RESOLVED_EVENT, { + stepId: event.stepId, + name: event.name, + ...(event.signalId === undefined ? {} : { signalId: event.signalId }), + ...(event.meta === undefined ? {} : { meta: event.meta }), + }), + ] + case 'STATE_DELTA': + return context.includeState + ? [ + { + type: EventType.STATE_DELTA, + timestamp: event.ts, + delta: [...event.delta], + }, + ] + : [] + case 'CUSTOM': + if (event.name === AI_STREAM_CHUNK_EVENT) { + const chunk = readStreamChunk(event.value.chunk) + return chunk ? [chunk] : [] + } + return [customChunk(event.ts, event.name, event.value, runId, threadId)] + case 'NOW_RECORDED': + case 'UUID_RECORDED': + return [] + } +} + +export function createWorkflowEventMapper( + options: WorkflowEventMapperOptions = {}, +): WorkflowEventMapper { + let currentWorkflowRunId: string | undefined + + return (event, workflowRunId) => { + if (event.type === 'RUN_STARTED') currentWorkflowRunId = event.runId + return workflowEventToStreamChunks(event, { + ...options, + workflowRunId: workflowRunId ?? currentWorkflowRunId, + }) + } +} + +export async function* toAIStream( + events: AsyncIterable, + options: WorkflowEventMapperOptions = {}, +): AsyncIterable { + const map = createWorkflowEventMapper(options) + for await (const event of events) { + for (const chunk of map(event)) yield chunk + } +} + +export function createAIEventPublisher(options: { + publish: (runId: string, chunk: StreamChunk) => void | Promise + mapping?: WorkflowEventMapperOptions +}): (runId: string, event: WorkflowEvent) => Promise { + const map = createWorkflowEventMapper(options.mapping) + + return async (workflowRunId, event) => { + const runId = resolveRunId(workflowRunId, options.mapping?.runId) + for (const chunk of map(event, workflowRunId)) { + await options.publish(runId, chunk) + } + } +} + +function resolveRunId( + workflowRunId: string, + configured: WorkflowEventMapperOptions['runId'], +): string { + if (typeof configured === 'function') return configured(workflowRunId) + return configured ?? workflowRunId +} + +function readAgentName(meta: Record | undefined) { + const value = meta?.[AI_AGENT_META_KEY] + return typeof value === 'string' ? value : undefined +} + +function customChunk( + timestamp: number, + name: string, + value: Record, + runId?: string, + threadId?: string, +): StreamChunk { + return { + type: 'CUSTOM', + timestamp, + name, + value, + ...(runId === undefined ? {} : { runId }), + ...(threadId === undefined ? {} : { threadId }), + } +} + +function readStreamChunk(value: unknown): StreamChunk | undefined { + if ( + typeof value !== 'object' || + value === null || + !('type' in value) || + typeof value.type !== 'string' + ) { + return undefined + } + return value as StreamChunk +} diff --git a/packages/ai-orchestration/src/execute-agent.ts b/packages/ai-orchestration/src/execute-agent.ts new file mode 100644 index 000000000..796a33ba8 --- /dev/null +++ b/packages/ai-orchestration/src/execute-agent.ts @@ -0,0 +1,163 @@ +import { + AgentApprovalUnsupportedError, + AgentStreamError, + AgentValidationError, +} from './agent' +import type { StandardSchemaV1 } from '@standard-schema/spec' +import type { StreamChunk } from '@tanstack/ai' +import type { AgentDefinition, AgentRunValue, AgentStreamResult } from './types' + +const NO_STRUCTURED_OUTPUT = Symbol('no-structured-output') + +export async function executeAgent(args: { + definition: AgentDefinition + input: TInput + signal: AbortSignal + emit: (chunk: StreamChunk) => void +}): Promise { + const { definition } = args + const input = definition.inputSchema + ? await validateSchema( + definition.inputSchema, + args.input, + `Input validation failed for agent "${definition.name}"`, + ) + : args.input + const { abortController, cleanup } = linkedAbortController(args.signal) + + try { + const result = await definition.run({ + input, + signal: abortController.signal, + abortController, + }) + + if (isAgentStreamResult(result)) { + // `output` is already in flight, so observe it before draining can throw. + // Otherwise a failing stream leaves it rejected and unhandled, which + // terminates the worker under Node's default unhandled-rejection policy. + const output = Promise.resolve(result.output) + output.catch(() => undefined) + + await drainAgentStream(definition.name, result.stream, args.emit) + return validateOutput(definition, await output) + } + + if (isAsyncIterable(result)) { + const output = await drainAgentStream(definition.name, result, args.emit) + return validateOutput(definition, output) + } + + return validateOutput(definition, result) + } finally { + cleanup() + } +} + +async function drainAgentStream( + agentName: string, + stream: AsyncIterable, + emit: (chunk: StreamChunk) => void, +): Promise { + let text = '' + let structuredOutput: unknown | typeof NO_STRUCTURED_OUTPUT = + NO_STRUCTURED_OUTPUT + + for await (const chunk of stream) { + if (chunk.type === 'RUN_STARTED' || chunk.type === 'RUN_FINISHED') { + continue + } + if (chunk.type === 'RUN_ERROR') { + throw new AgentStreamError(agentName, chunk.message, chunk.code) + } + if (chunk.type === 'CUSTOM' && chunk.name === 'approval-requested') { + throw new AgentApprovalUnsupportedError(agentName) + } + + emit(chunk) + + if (chunk.type === 'TEXT_MESSAGE_CONTENT') { + text += chunk.delta + } else if ( + chunk.type === 'CUSTOM' && + chunk.name === 'structured-output.complete' && + isRecord(chunk.value) && + 'object' in chunk.value + ) { + structuredOutput = chunk.value.object + } + } + + return structuredOutput === NO_STRUCTURED_OUTPUT ? text : structuredOutput +} + +async function validateOutput( + definition: AgentDefinition, + rawOutput: unknown, +): Promise { + if (!definition.outputSchema) return rawOutput as TOutput + + let candidate = rawOutput + if (typeof candidate === 'string') { + try { + candidate = JSON.parse(candidate) + } catch { + // Let the schema report the useful validation issue for plain text. + } + } + + return validateSchema( + definition.outputSchema, + candidate, + `Output validation failed for agent "${definition.name}"`, + ) +} + +async function validateSchema( + schema: StandardSchemaV1, + value: unknown, + message: string, +): Promise { + const result = await schema['~standard'].validate(value) + if (result.issues) { + throw new AgentValidationError(message, result.issues) + } + return result.value +} + +function linkedAbortController(signal: AbortSignal): { + abortController: AbortController + cleanup: () => void +} { + const abortController = new AbortController() + const abort = () => abortController.abort(signal.reason) + + if (signal.aborted) { + abort() + return { abortController, cleanup: () => undefined } + } + + signal.addEventListener('abort', abort, { once: true }) + return { + abortController, + cleanup: () => signal.removeEventListener('abort', abort), + } +} + +function isAgentStreamResult( + value: AgentRunValue, +): value is AgentStreamResult { + return isRecord(value) && value.kind === 'agent-stream' +} + +function isAsyncIterable(value: unknown): value is AsyncIterable { + return ( + (typeof value === 'object' || typeof value === 'function') && + value !== null && + Symbol.asyncIterator in value + ) +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null +} diff --git a/packages/ai-orchestration/src/index.ts b/packages/ai-orchestration/src/index.ts new file mode 100644 index 000000000..d3b36171f --- /dev/null +++ b/packages/ai-orchestration/src/index.ts @@ -0,0 +1,42 @@ +export { + AgentApprovalUnsupportedError, + AgentStreamError, + AgentValidationError, + agentStream, + defineAgent, +} from './agent' +export { agentMiddleware } from './middleware' +export { + createAIEventPublisher, + createWorkflowEventMapper, + toAIStream, + workflowEventToStreamChunks, +} from './events' +export { + AI_AGENT_META_KEY, + AI_STREAM_CHUNK_EVENT, + WORKFLOW_APPROVAL_REQUESTED_EVENT, + WORKFLOW_APPROVAL_RESOLVED_EVENT, + WORKFLOW_SIGNAL_AWAITED_EVENT, + WORKFLOW_SIGNAL_RESOLVED_EVENT, + WORKFLOW_STEP_FAILED_EVENT, +} from './constants' +export type { + WorkflowEventMapper, + WorkflowEventMapperOptions, + WorkflowEventMappingContext, +} from './events' +export type { + AIWorkflowContext, + AgentDefinition, + AgentInput, + AgentOptions, + AgentOutput, + AgentRunContext, + AgentRunResult, + AgentRunValue, + AgentStreamResult, + AnyAgentDefinition, + InferSchema, + SchemaInput, +} from './types' diff --git a/packages/ai-orchestration/src/middleware.ts b/packages/ai-orchestration/src/middleware.ts new file mode 100644 index 000000000..6cd92614e --- /dev/null +++ b/packages/ai-orchestration/src/middleware.ts @@ -0,0 +1,56 @@ +import { createMiddleware } from '@tanstack/workflow-core' +import { AI_AGENT_META_KEY, AI_STREAM_CHUNK_EVENT } from './constants' +import { executeAgent } from './execute-agent' +import type { StreamChunk } from '@tanstack/ai' +import type { Middleware, WorkflowCtx } from '@tanstack/workflow-core' +import type { + AIWorkflowContext, + AgentInput, + AgentOptions, + AgentOutput, + AnyAgentDefinition, +} from './types' + +/** Adds the package-owned `ctx.ai` namespace to a Workflow handler. */ +export function agentMiddleware(): Middleware< + WorkflowCtx, + { ai: AIWorkflowContext } +> { + return createMiddleware().server<{ ai: AIWorkflowContext }>( + async ({ ctx, next }) => { + const ai: AIWorkflowContext = { + agent: ( + id: string, + definition: TAgent, + input: AgentInput, + options?: AgentOptions, + ): Promise> => { + const meta = { + ...options?.meta, + [AI_AGENT_META_KEY]: definition.name, + } + + return ctx.step( + id, + (step) => + executeAgent, AgentOutput>({ + definition, + input, + signal: step.signal, + emit: (chunk: StreamChunk) => { + ctx.emit(AI_STREAM_CHUNK_EVENT, { + stepId: step.id, + attempt: step.attempt, + chunk, + }) + }, + }), + { ...options, meta }, + ) + }, + } + + return next({ context: { ai } }) + }, + ) +} diff --git a/packages/ai-orchestration/src/types.ts b/packages/ai-orchestration/src/types.ts new file mode 100644 index 000000000..24e99d372 --- /dev/null +++ b/packages/ai-orchestration/src/types.ts @@ -0,0 +1,69 @@ +import type { StandardSchemaV1 } from '@standard-schema/spec' +import type { StreamChunk } from '@tanstack/ai' +import type { StepOptions } from '@tanstack/workflow-core' + +export type SchemaInput = StandardSchemaV1 +export type InferSchema = + TSchema extends StandardSchemaV1 ? Output : never + +export interface AgentRunContext { + input: TInput + /** Aborts with the enclosing Workflow step, including step timeouts. */ + signal: AbortSignal + /** Linked controller for TanStack AI APIs that accept an AbortController. */ + abortController: AbortController +} + +export interface AgentStreamResult { + readonly kind: 'agent-stream' + stream: AsyncIterable + output: TOutput | Promise +} + +export type AgentRunValue = + | TOutput + | AsyncIterable + | AgentStreamResult + +export type AgentRunResult = + | AgentRunValue + | Promise> + +export interface AgentDefinition< + TInput = unknown, + TOutput = unknown, + TName extends string = string, +> { + readonly kind: 'agent' + readonly name: TName + readonly description?: string + readonly inputSchema?: StandardSchemaV1 + readonly outputSchema?: StandardSchemaV1 + readonly run: (context: AgentRunContext) => AgentRunResult +} + +// `any` is intentional at this erased registry boundary. The conditional +// helpers below recover each concrete definition's input and output types. +// oxlint-disable-next-line typescript/no-explicit-any +export type AnyAgentDefinition = AgentDefinition + +export type AgentInput = + TAgent extends AgentDefinition + ? Input + : never + +export type AgentOutput = + TAgent extends AgentDefinition + ? Output + : never + +export type AgentOptions = StepOptions + +export interface AIWorkflowContext { + agent: ( + id: string, + definition: TAgent, + input: AgentInput, + options?: AgentOptions, + ) => Promise> +} diff --git a/packages/ai-orchestration/tests/orchestration.test-d.ts b/packages/ai-orchestration/tests/orchestration.test-d.ts new file mode 100644 index 000000000..653dc4946 --- /dev/null +++ b/packages/ai-orchestration/tests/orchestration.test-d.ts @@ -0,0 +1,33 @@ +import { expectTypeOf } from 'vitest' +import { createWorkflow } from '@tanstack/workflow-core' +import { z } from 'zod' +import { agentMiddleware, defineAgent } from '../src' +import type { AgentOutput } from '../src' + +const writer = defineAgent({ + name: 'writer', + input: z.object({ topic: z.string() }), + output: z.object({ article: z.string() }), + run: async ({ input }) => ({ article: input.topic }), +}) + +const workflow = createWorkflow({ id: 'typed' }) + .middleware([agentMiddleware()]) + .handler(async (ctx) => { + const result = await ctx.ai.agent('draft', writer, { topic: 'Workflow' }) + expectTypeOf(result).toEqualTypeOf<{ article: string }>() + + // @ts-expect-error the agent input schema requires topic + await ctx.ai.agent('invalid', writer, { subject: 'Workflow' }) + + return result + }) + +expectTypeOf(workflow.id).toEqualTypeOf() + +const textAgent = defineAgent({ + name: 'text', + run: async () => 'done', +}) + +expectTypeOf>().toEqualTypeOf() diff --git a/packages/ai-orchestration/tests/orchestration.test.ts b/packages/ai-orchestration/tests/orchestration.test.ts new file mode 100644 index 000000000..c556e4c1a --- /dev/null +++ b/packages/ai-orchestration/tests/orchestration.test.ts @@ -0,0 +1,358 @@ +import { + createWorkflow, + inMemoryRunStore, + runWorkflow, +} from '@tanstack/workflow-core' +import { describe, expect, it } from 'vitest' +import { z } from 'zod' +import { EventType } from '@tanstack/ai' +import { + AgentApprovalUnsupportedError, + AgentStreamError, + agentMiddleware, + agentStream, + createAIEventPublisher, + defineAgent, + toAIStream, +} from '../src' +import type { StreamChunk } from '@tanstack/ai' +import type { WorkflowEvent } from '@tanstack/workflow-core' + +describe('Workflow-backed agent orchestration', () => { + it('runs a streamed agent as a Workflow step and projects AG-UI events', async () => { + let calls = 0 + const writer = defineAgent({ + name: 'writer', + input: z.object({ topic: z.string() }), + output: z.object({ article: z.string() }), + run: ({ input }) => { + calls += 1 + return streamText(JSON.stringify({ article: `About ${input.topic}` })) + }, + }) + const workflow = createWorkflow({ + id: 'article', + input: z.object({ topic: z.string() }), + }) + .middleware([agentMiddleware()]) + .handler((ctx) => + ctx.ai.agent('draft', writer, { topic: ctx.input.topic }), + ) + + const chunks = await collect( + toAIStream( + runWorkflow({ + workflow, + runStore: inMemoryRunStore(), + input: { topic: 'workflows' }, + }), + ), + ) + + expect(calls).toBe(1) + expect(chunks.filter((chunk) => chunk.type === 'RUN_STARTED')).toHaveLength( + 1, + ) + expect( + chunks.filter((chunk) => chunk.type === 'RUN_FINISHED'), + ).toHaveLength(1) + expect(chunks).toContainEqual( + expect.objectContaining({ + type: 'STEP_STARTED', + stepName: 'writer', + stepId: 'draft', + stepType: 'agent', + }), + ) + expect(chunks).toContainEqual( + expect.objectContaining({ + type: 'TEXT_MESSAGE_CONTENT', + delta: '{"article":"About workflows"}', + }), + ) + expect(chunks.at(-1)).toEqual( + expect.objectContaining({ + type: 'RUN_FINISHED', + result: { article: 'About workflows' }, + }), + ) + }) + + it('replays a completed agent step when an approval resumes the workflow', async () => { + let calls = 0 + const researcher = defineAgent({ + name: 'researcher', + input: z.object({ query: z.string() }), + output: z.object({ finding: z.string() }), + run: async ({ input }) => { + calls += 1 + return { finding: input.query.toUpperCase() } + }, + }) + const workflow = createWorkflow({ + id: 'research', + input: z.object({ query: z.string() }), + }) + .middleware([agentMiddleware()]) + .handler(async (ctx) => { + const result = await ctx.ai.agent('research', researcher, ctx.input) + const approval = await ctx.approve({ title: 'Publish?' }) + return { ...result, approved: approval.approved } + }) + const runStore = inMemoryRunStore() + + const firstEvents = await collect( + runWorkflow({ + workflow, + runStore, + input: { query: 'durability' }, + }), + ) + const runId = findRunId(firstEvents) + const approval = firstEvents.find( + ( + event, + ): event is Extract => + event.type === 'APPROVAL_REQUESTED', + ) + + expect(approval).toBeDefined() + expect((await runStore.getRunState(runId))?.status).toBe('paused') + + const resumedEvents = await collect( + runWorkflow({ + workflow, + runStore, + runId, + approval: { + approvalId: approval!.approvalId, + approved: true, + }, + }), + ) + + expect(calls).toBe(1) + expect(resumedEvents.at(-1)).toEqual( + expect.objectContaining({ + type: 'RUN_FINISHED', + output: { finding: 'DURABILITY', approved: true }, + }), + ) + expect((await runStore.getRunState(runId))?.status).toBe('finished') + }) + + it('inherits automatic deadline yielding before starting an agent step', async () => { + let calls = 0 + const worker = defineAgent({ + name: 'worker', + run: async () => { + calls += 1 + return 'done' + }, + }) + const workflow = createWorkflow({ id: 'deadline' }) + .middleware([agentMiddleware()]) + .handler((ctx) => ctx.ai.agent('work', worker, undefined)) + const runStore = inMemoryRunStore() + + const yieldedEvents = await collect( + runWorkflow({ + workflow, + runStore, + input: {}, + deadline: Date.now() + 10, + minYieldRemainingMs: 1_000, + }), + ) + const runId = findRunId(yieldedEvents) + + expect(calls).toBe(0) + expect(yieldedEvents).toContainEqual( + expect.objectContaining({ type: 'SIGNAL_AWAITED', name: '__timer' }), + ) + expect((await runStore.getRunState(runId))?.status).toBe('paused') + + const resumedEvents = await collect( + runWorkflow({ + workflow, + runStore, + runId, + signalDelivery: { + signalId: 'deadline-resume', + name: '__timer', + payload: undefined, + }, + deadline: Date.now() + 60_000, + }), + ) + + expect(calls).toBe(1) + expect(resumedEvents.at(-1)?.type).toBe('RUN_FINISHED') + }) + + it('does not checkpoint an AI tool approval as completed agent output', async () => { + const interactive = defineAgent({ + name: 'interactive', + run: () => approvalStream(), + }) + const workflow = createWorkflow({ id: 'interactive' }) + .middleware([agentMiddleware()]) + .handler((ctx) => ctx.ai.agent('interactive', interactive, undefined)) + const events = await collect( + runWorkflow({ + workflow, + runStore: inMemoryRunStore(), + input: {}, + }), + ) + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'STEP_FAILED', + error: expect.objectContaining({ + name: AgentApprovalUnsupportedError.name, + }), + }), + ) + expect(events.some((event) => event.type === 'RUN_FINISHED')).toBe(false) + }) + + it('fails the step without an unhandled rejection when an agentStream errors', async () => { + const unhandled: Array = [] + const onUnhandled = (reason: unknown) => unhandled.push(reason) + process.on('unhandledRejection', onUnhandled) + + const broken = defineAgent({ + name: 'broken', + run: () => + agentStream( + errorStream('model exploded'), + Promise.reject(new Error('output never settles')), + ), + }) + const workflow = createWorkflow({ id: 'broken' }) + .middleware([agentMiddleware()]) + .handler((ctx) => ctx.ai.agent('broken', broken, undefined)) + + const events = await collect( + runWorkflow({ + workflow, + runStore: inMemoryRunStore(), + input: {}, + }), + ) + // Let a rejection scheduled during the run reach the process handler. + await new Promise((resolve) => setTimeout(resolve, 0)) + process.off('unhandledRejection', onUnhandled) + + expect(unhandled).toEqual([]) + expect(events).toContainEqual( + expect.objectContaining({ + type: 'STEP_FAILED', + error: expect.objectContaining({ name: AgentStreamError.name }), + }), + ) + expect(events.some((event) => event.type === 'RUN_FINISHED')).toBe(false) + }) + + it('adapts the Workflow publish hook without changing execution IDs', async () => { + const published: Array<{ runId: string; chunk: StreamChunk }> = [] + const publisher = createAIEventPublisher({ + mapping: { runId: (workflowRunId) => `ai:${workflowRunId}` }, + publish: (runId, chunk) => { + published.push({ runId, chunk }) + }, + }) + const workflow = createWorkflow({ id: 'publisher' }).handler(async () => + Promise.resolve('done'), + ) + + const events = await collect( + runWorkflow({ + workflow, + runStore: inMemoryRunStore(), + input: {}, + publish: publisher, + }), + ) + const workflowRunId = findRunId(events) + + expect(published.map((entry) => entry.runId)).toEqual([ + `ai:${workflowRunId}`, + `ai:${workflowRunId}`, + ]) + expect(published.map((entry) => entry.chunk.type)).toEqual([ + 'RUN_STARTED', + 'RUN_FINISHED', + ]) + }) +}) + +async function* streamText(text: string): AsyncIterable { + yield { + type: EventType.RUN_STARTED, + runId: 'inner-run', + threadId: 'inner-thread', + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: 'message-1', + role: 'assistant', + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'message-1', + delta: text, + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_END, + messageId: 'message-1', + timestamp: Date.now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: 'inner-run', + threadId: 'inner-thread', + timestamp: Date.now(), + } +} + +async function* errorStream(message: string): AsyncIterable { + yield { + type: EventType.RUN_ERROR, + message, + timestamp: Date.now(), + } +} + +async function* approvalStream(): AsyncIterable { + yield { + type: 'CUSTOM', + name: 'approval-requested', + value: { + toolCallId: 'tool-call-1', + toolName: 'write_file', + input: {}, + approval: { id: 'approval-1', needsApproval: true }, + }, + timestamp: Date.now(), + } +} + +async function collect(iterable: AsyncIterable): Promise> { + const values: Array = [] + for await (const value of iterable) values.push(value) + return values +} + +function findRunId(events: ReadonlyArray): string { + const started = events.find( + (event): event is Extract => + event.type === 'RUN_STARTED', + ) + if (!started) throw new Error('RUN_STARTED not found') + return started.runId +} diff --git a/packages/ai-orchestration/tsconfig.json b/packages/ai-orchestration/tsconfig.json new file mode 100644 index 000000000..c38689f4e --- /dev/null +++ b/packages/ai-orchestration/tsconfig.json @@ -0,0 +1,8 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "outDir": "dist" + }, + "include": ["src", "tests"], + "exclude": ["node_modules", "dist"] +} diff --git a/packages/ai-orchestration/vite.config.ts b/packages/ai-orchestration/vite.config.ts new file mode 100644 index 000000000..77bcc2e60 --- /dev/null +++ b/packages/ai-orchestration/vite.config.ts @@ -0,0 +1,36 @@ +import { defineConfig, mergeConfig } from 'vitest/config' +import { tanstackViteConfig } from '@tanstack/vite-config' +import packageJson from './package.json' + +const config = defineConfig({ + test: { + name: packageJson.name, + dir: './', + watch: false, + globals: true, + environment: 'node', + include: ['tests/**/*.test.ts'], + coverage: { + provider: 'v8', + reporter: ['text', 'json', 'html', 'lcov'], + exclude: [ + 'node_modules/', + 'dist/', + 'tests/', + '**/*.test.ts', + '**/*.config.ts', + '**/types.ts', + ], + include: ['src/**/*.ts'], + }, + }, +}) + +export default mergeConfig( + config, + tanstackViteConfig({ + entry: ['./src/index.ts'], + srcDir: './src', + cjs: false, + }), +) diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9aa43432e..77fd7eb72 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -2043,6 +2043,25 @@ importers: specifier: ^4.2.0 version: 4.3.6 + packages/ai-orchestration: + dependencies: + '@standard-schema/spec': + specifier: ^1.1.0 + version: 1.1.0 + '@tanstack/workflow-core': + specifier: ^0.0.4 + version: 0.0.4(@opentelemetry/api@1.9.1) + devDependencies: + '@tanstack/ai': + specifier: workspace:* + version: link:../ai + '@vitest/coverage-v8': + specifier: 4.0.14 + version: 4.0.14(supports-color@7.2.0)(vitest@4.1.10) + zod: + specifier: ^4.2.0 + version: 4.3.6 + packages/ai-preact: dependencies: '@tanstack/ai-client': @@ -2596,6 +2615,9 @@ importers: '@tanstack/ai-openrouter': specifier: workspace:* version: link:../../packages/ai-openrouter + '@tanstack/ai-orchestration': + specifier: workspace:* + version: link:../../packages/ai-orchestration '@tanstack/ai-react': specifier: workspace:* version: link:../../packages/ai-react @@ -2617,6 +2639,9 @@ importers: '@tanstack/react-start': specifier: ^1.159.0 version: 1.159.5(crossws@0.4.6(srvx@0.11.17))(react-dom@19.2.3(react@19.2.3))(react@19.2.3)(vite-plugin-solid@2.11.10(solid-js@1.9.10)(vite@8.1.4(@types/node@24.10.3)(esbuild@0.28.1)(jiti@2.7.0)(less@4.6.6)(sass@1.101.0)(terser@5.44.1)(tsx@4.21.0)(yaml@2.9.0)))(vite@8.1.4(@types/node@24.10.3)(esbuild@0.28.1)(jiti@2.7.0)(less@4.6.6)(sass@1.101.0)(terser@5.44.1)(tsx@4.21.0)(yaml@2.9.0)) + '@tanstack/workflow-core': + specifier: ^0.0.4 + version: 0.0.4(@opentelemetry/api@1.9.1) arktype: specifier: ^2.1.28 version: 2.2.0 @@ -8989,6 +9014,12 @@ packages: peerDependencies: vite: ^8.0.0 + '@tanstack/workflow-core@0.0.4': + resolution: {integrity: sha512-Ytl8uF1v+xKDCez1P7Qcfoe6OZ2HskQzfey8Kci5h83UoEOt8azi/KmfrkrE28ev46QPSqkOn4tTLvILgCi67w==} + engines: {node: '>=18'} + peerDependencies: + '@opentelemetry/api': ^1.9.0 + '@tanstack/zod-adapter@1.166.9': resolution: {integrity: sha512-HHllQ/CKGi8YBbftv6OmzojtHM6Rk4UszAFICAgUMbwiqtKqjlIZQ/7mv2IPNxBb8YlOQgzyQ4jz2UTEXIi6YA==} engines: {node: '>=20.19'} @@ -23288,6 +23319,11 @@ snapshots: - typescript - webpack + '@tanstack/workflow-core@0.0.4(@opentelemetry/api@1.9.1)': + dependencies: + '@opentelemetry/api': 1.9.1 + '@standard-schema/spec': 1.1.0 + '@tanstack/zod-adapter@1.166.9(@tanstack/react-router@1.159.5(react-dom@19.2.3(react@19.2.3))(react@19.2.3))(zod@4.3.6)': dependencies: '@tanstack/react-router': 1.159.5(react-dom@19.2.3(react@19.2.3))(react@19.2.3) diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml index d0616db5d..dd2bed434 100644 --- a/pnpm-workspace.yaml +++ b/pnpm-workspace.yaml @@ -7,6 +7,8 @@ minimumReleaseAge: 1440 minimumReleaseAgeExclude: - 'kiira@0.6.0' # root dev dependency (TS7-native engine); freshly published - 'kiira-core@0.6.0' # transitive dependency of kiira@0.6.0 + # First-party runtime required by the Workflow-backed orchestration package. + - '@tanstack/workflow-core@0.0.4' # Working closely with the aimock team; the e2e suite often needs fresh # releases of this test-only mock as it tracks provider SDK changes. - '@copilotkit/aimock' diff --git a/rfcs/0001-workflow-backed-orchestration.md b/rfcs/0001-workflow-backed-orchestration.md new file mode 100644 index 000000000..77b3dd13b --- /dev/null +++ b/rfcs/0001-workflow-backed-orchestration.md @@ -0,0 +1,278 @@ +# RFC 0001: Workflow-backed AI orchestration + +- Status: Proposed +- Owners: TanStack AI and TanStack Workflow +- Target package: `@tanstack/ai-orchestration` + +## Summary + +TanStack AI should use TanStack Workflow as the durable execution layer for +multi-step AI work. TanStack AI should not own a second workflow engine, replay +log, scheduler, lease model, or durable primitive set. + +The first package is intentionally small: + +- `defineAgent()` describes a typed AI unit. +- `agentMiddleware()` adds `ctx.ai.agent()` to a Workflow handler. +- Agent calls execute as Workflow steps. +- Workflow events can be projected into TanStack AI `StreamChunk` events. +- Hosts can publish that projection to Durable Streams or another delivery log. + +The existing `chat()` agent loop remains useful inside one agent call. Workflow +coordinates work across calls, waits, approvals, process restarts, and runtime +drives. + +## Motivation + +TanStack AI has several adjacent concerns that must stay separate: + +1. Model and tool execution. +2. Durable business-process execution. +3. Message, thread, artifact, and interrupt persistence. +4. Reconnectable delivery of live events. + +The old orchestration prototype mixed these concerns. It wrapped a Workflow +engine with its own workflow definitions, generator descriptors, store aliases, +request parser, nested workflow behavior, and client API. That duplicated APIs +without resolving ownership. + +TanStack Workflow now has the required execution foundation: closure handlers, +durable steps, replay, signals, approvals, timers, retries, step timeouts, +version routing, runtime deadlines, cooperative yielding, leases, sweeping, and +pluggable stores. TanStack AI already has AG-UI events and resumable delivery +streams. The missing layer is a narrow adapter between them. + +## Ownership + +| Concern | Owner | +| ---------------------------------------------------- | --------------------------- | +| Workflow handler and durable state | TanStack Workflow | +| Step replay, retries, timeouts, deterministic values | TanStack Workflow | +| Signals, durable waits, and workflow approvals | TanStack Workflow | +| Runtime deadlines, yielding, leases, sweep, recovery | TanStack Workflow | +| Agent definition and model-facing ergonomics | TanStack AI orchestration | +| `chat()` loop and server/client tool execution | TanStack AI | +| AG-UI event projection | TanStack AI orchestration | +| Message, thread, artifact, interrupt records | TanStack AI persistence | +| Event delivery, reconnect, and cursor replay | TanStack AI durable streams | + +Durable execution and durable event delivery are different guarantees. Workflow +decides what work has completed. Durable Streams decides which emitted events a +client can reconnect to. + +## Proposed API + +```ts +import { chat } from '@tanstack/ai' +import { agentMiddleware, defineAgent } from '@tanstack/ai-orchestration' +import { openaiText } from '@tanstack/ai-openai' +import { createWorkflow } from '@tanstack/workflow-core' +import { z } from 'zod' + +const writer = defineAgent({ + name: 'writer', + input: z.object({ topic: z.string() }), + output: z.object({ article: z.string() }), + run: ({ input, abortController }) => + chat({ + adapter: openaiText('gpt-5.5'), + messages: [{ role: 'user', content: `Write about ${input.topic}` }], + outputSchema: z.object({ article: z.string() }), + stream: true, + abortController, + }), +}) + +export const article = createWorkflow({ + id: 'article', + input: z.object({ topic: z.string() }), +}) + .middleware([agentMiddleware()]) + .handler(async (ctx) => { + const draft = await ctx.ai.agent('draft', writer, { + topic: ctx.input.topic, + }) + + const review = await ctx.approve({ title: 'Publish this article?' }) + return { draft, approved: review.approved } + }) +``` + +The async handler and context API is the only workflow authoring model. TanStack +AI will not expose a generator equivalent for these operations. Each durable +primitive has one canonical call shape, which keeps documentation, types, and +generated code aligned. The term "yield" remains reserved for Workflow's +cooperative runtime scheduling behavior. + +`ctx.ai` is the package-owned middleware namespace. Workflow core does not need +to reserve every official integration name. A conflicting `ai` extension is a +composition error visible in TypeScript, while Workflow's own fields remain +reserved by core. + +The first argument to `ctx.ai.agent()` is the stable Workflow step ID. It is not +an arbitrary batch position. Reordering code across deployed workflow versions +still follows Workflow's normal versioning rules. + +## Agent execution semantics + +An agent call is one Workflow step. + +- Input is validated before the model call. +- The Workflow step signal is linked to the AbortController accepted by + TanStack AI. +- Stream chunks are emitted live as non-checkpoint Workflow custom events. +- The validated final output becomes the Workflow step checkpoint. +- Replay returns the checkpoint without calling the model again. +- If a worker dies before the checkpoint, recovery may call the model again. + +The final point is unavoidable without provider idempotency. This API therefore +offers at-least-once agent execution around a durable checkpoint, not exactly +once model execution. Agent tools with external side effects must continue to +use idempotency keys or separate Workflow steps. + +Workflow retries are opt-in through the existing step options. Runtime deadline +handling is automatic because `ctx.ai.agent()` delegates to `ctx.step()`. +Workflow checks the drive budget before starting fresh work. Step timeouts stay +independent and abort the active agent call. + +The whole `chat()` loop inside that call is currently one atomic durable unit. +Workflow does not checkpoint between model turns or tools owned by that loop, +and it cannot cooperatively yield until the call returns. This is appropriate +for bounded calls. Long autonomous loops need a later integration that exposes +model turns and side-effecting tools as Workflow operations; the first package +does not claim that guarantee. + +## Streaming and replay + +Model token chunks do not belong in Workflow's replay checkpoint log. They are +large, transient, and do not make model execution deterministic. + +The event path is: + +```text +chat() -> ctx.emit() -> Workflow publish hook -> AG-UI projection -> delivery log +``` + +The package filters inner model `RUN_STARTED` and `RUN_FINISHED` events. The +Workflow run owns the outer AG-UI lifecycle. `createAIEventPublisher()` adapts +Workflow's publish hook for a host transport. `toAIStream()` adapts the direct +`runWorkflow()` iterable. + +For a request-bound run, hosts can pass `toAIStream()` through TanStack AI's +durable response helpers, which append before forwarding. Workflow's generic +publish hook is best-effort fan-out rather than a transaction with the step +checkpoint. A background host therefore cannot treat successful token publish +as part of Workflow correctness. The persistence RFC must define reconciliation +and publisher-failure behavior before background token delivery is considered +durable end to end. + +A host may assign an AG-UI run ID distinct from the Workflow run ID. The mapper +accepts a stable ID resolver. The Workflow run ID remains the execution key; +the AI run ID remains the client protocol key. Hosts must persist that +correlation when the IDs differ. + +Workflow state patches are not projected by default. AI message and artifact +state should come from the AI persistence layer, not from arbitrary Workflow +state. Hosts may opt into Workflow state projection for applications that use +the AG-UI state channel intentionally. + +## Approvals and interrupts + +Workflow approvals and AI tool approvals are not currently the same object. + +For the first release: + +- Durable human gates use `ctx.approve()` between agent calls. +- Workflow approval events use namespaced custom AG-UI events. +- An `approval-requested` event emitted inside a `chat()` call fails the agent + step with a targeted error instead of checkpointing partial output. + +The AI RFC follow-up must define whether an AI interrupt is: + +1. A persisted AI-domain object resumed through a Workflow signal. +2. A direct projection of a Workflow approval. +3. A protocol envelope that can represent both. + +Until then, silently translating one approval model into the other would create +an API that cannot resume reliably. + +## Persistence + +Workflow stores execution state and checkpoints. It should not become the +canonical message database. + +TanStack AI persistence should own: + +- threads and messages, +- tool-call and artifact projections, +- interrupt records, +- user-visible run metadata, +- the mapping between AI run IDs and Workflow run IDs. + +That persistence layer can consume the same AG-UI projection used by clients. +It must tolerate repeated delivery because transports and recovered steps can +produce duplicate attempts. + +## Child workflows + +True child workflows are deferred. Wrapping a child run inside `ctx.step()` is +not sufficient: cancellation, parent-child status, signals, leases, and replay +all need first-class runtime semantics. + +Composition in the first release uses normal helper functions and agent calls +inside one Workflow definition. + +## Alternatives rejected + +### Add a generator DSL + +Rejected, not deferred. `yield* agents.writer()` would duplicate Workflow's +async context API and every durable primitive without adding a new guarantee. +Maintaining equivalent generator and async forms would split documentation, +examples, type behavior, and generated code while encouraging invalid mixtures +of `await`, `yield`, and `yield*`. It would also overload "yield" when Workflow +already uses that term for cooperative runtime scheduling. + +Generator syntax is not planned and will not be added as optional sugar. Any +future proposal would have to demonstrate a required semantic that the async +context API cannot express and replace, rather than duplicate, the canonical +authoring model. + +### Let TanStack AI own the workflow store + +Rejected. It would fork durability semantics and force every Workflow store +fix to be reimplemented in AI. + +### Persist every token in Workflow + +Rejected. Token delivery durability belongs in a delivery log. Workflow keeps +only execution checkpoints and coordination records. + +### Treat process death as normal retry behavior + +Rejected as a complete strategy. Recovery is required, but intentionally +driving into a host timeout increases duplicate model calls, leaves stale +leases until recovery, and loses the chance to stop before starting fresh work. +Workflow deadlines and cooperative yielding reduce that ambiguity. + +## Rollout + +1. Land Workflow runtime event publishing and stale-lease recovery. +2. Land `@tanstack/ai-orchestration` as experimental. +3. Build one server example using a real Workflow runtime adapter and Durable + Streams. +4. Define the AI persistence and interrupt contracts in a separate RFC. +5. Add framework client helpers only after the persisted projection is stable. + +## Decisions still required + +The following do not block the experimental package: + +- The public AI persistence interface and database ownership. +- The interrupt envelope and approval mapping. +- Whether AI run IDs are always distinct or only distinct when supplied by a + host. +- Attempt/reset semantics for token streams after a worker dies mid-agent. +- Backpressure and failure semantics for background event publishers. +- Durable boundaries inside long `chat()` model/tool loops. +- First-class child workflow semantics. diff --git a/testing/e2e/package.json b/testing/e2e/package.json index 78ffbf406..9be444a1c 100644 --- a/testing/e2e/package.json +++ b/testing/e2e/package.json @@ -31,6 +31,7 @@ "@tanstack/ai-ollama": "workspace:*", "@tanstack/ai-openai": "workspace:*", "@tanstack/ai-openrouter": "workspace:*", + "@tanstack/ai-orchestration": "workspace:*", "@tanstack/ai-react": "workspace:*", "@tanstack/ai-react-ui": "workspace:*", "@tanstack/devtools-event-bus": "^0.4.1", @@ -38,6 +39,7 @@ "@tanstack/react-ai-devtools": "workspace:*", "@tanstack/react-router": "^1.158.4", "@tanstack/react-start": "^1.159.0", + "@tanstack/workflow-core": "^0.0.4", "arktype": "^2.1.28", "react": "^19.2.3", "react-dom": "^19.2.3", diff --git a/testing/e2e/src/routeTree.gen.ts b/testing/e2e/src/routeTree.gen.ts index 13ad39d5f..4b806416f 100644 --- a/testing/e2e/src/routeTree.gen.ts +++ b/testing/e2e/src/routeTree.gen.ts @@ -21,6 +21,7 @@ import { Route as DevtoolsChatRouteImport } from './routes/devtools-chat' import { Route as ChatClientDefaultBridgeRouteImport } from './routes/chat-client-default-bridge' import { Route as IndexRouteImport } from './routes/index' import { Route as ProviderIndexRouteImport } from './routes/$provider/index' +import { Route as ApiWorkflowOrchestrationRouteImport } from './routes/api.workflow-orchestration' import { Route as ApiVideoRouteImport } from './routes/api.video' import { Route as ApiTtsRouteImport } from './routes/api.tts' import { Route as ApiTranscriptionRouteImport } from './routes/api.transcription' @@ -120,6 +121,12 @@ const ProviderIndexRoute = ProviderIndexRouteImport.update({ path: '/$provider/', getParentRoute: () => rootRouteImport, } as any) +const ApiWorkflowOrchestrationRoute = + ApiWorkflowOrchestrationRouteImport.update({ + id: '/api/workflow-orchestration', + path: '/api/workflow-orchestration', + getParentRoute: () => rootRouteImport, + } as any) const ApiVideoRoute = ApiVideoRouteImport.update({ id: '/api/video', path: '/api/video', @@ -361,6 +368,7 @@ export interface FileRoutesByFullPath { '/api/transcription': typeof ApiTranscriptionRouteWithChildren '/api/tts': typeof ApiTtsRouteWithChildren '/api/video': typeof ApiVideoRouteWithChildren + '/api/workflow-orchestration': typeof ApiWorkflowOrchestrationRoute '/$provider/': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute '/api/image/stream': typeof ApiImageStreamRoute @@ -413,6 +421,7 @@ export interface FileRoutesByTo { '/api/transcription': typeof ApiTranscriptionRouteWithChildren '/api/tts': typeof ApiTtsRouteWithChildren '/api/video': typeof ApiVideoRouteWithChildren + '/api/workflow-orchestration': typeof ApiWorkflowOrchestrationRoute '/$provider': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute '/api/image/stream': typeof ApiImageStreamRoute @@ -466,6 +475,7 @@ export interface FileRoutesById { '/api/transcription': typeof ApiTranscriptionRouteWithChildren '/api/tts': typeof ApiTtsRouteWithChildren '/api/video': typeof ApiVideoRouteWithChildren + '/api/workflow-orchestration': typeof ApiWorkflowOrchestrationRoute '/$provider/': typeof ProviderIndexRoute '/api/audio/stream': typeof ApiAudioStreamRoute '/api/image/stream': typeof ApiImageStreamRoute @@ -520,6 +530,7 @@ export interface FileRouteTypes { | '/api/transcription' | '/api/tts' | '/api/video' + | '/api/workflow-orchestration' | '/$provider/' | '/api/audio/stream' | '/api/image/stream' @@ -572,6 +583,7 @@ export interface FileRouteTypes { | '/api/transcription' | '/api/tts' | '/api/video' + | '/api/workflow-orchestration' | '/$provider' | '/api/audio/stream' | '/api/image/stream' @@ -624,6 +636,7 @@ export interface FileRouteTypes { | '/api/transcription' | '/api/tts' | '/api/video' + | '/api/workflow-orchestration' | '/$provider/' | '/api/audio/stream' | '/api/image/stream' @@ -677,6 +690,7 @@ export interface RootRouteChildren { ApiTranscriptionRoute: typeof ApiTranscriptionRouteWithChildren ApiTtsRoute: typeof ApiTtsRouteWithChildren ApiVideoRoute: typeof ApiVideoRouteWithChildren + ApiWorkflowOrchestrationRoute: typeof ApiWorkflowOrchestrationRoute ProviderIndexRoute: typeof ProviderIndexRoute } @@ -766,6 +780,13 @@ declare module '@tanstack/react-router' { preLoaderRoute: typeof ProviderIndexRouteImport parentRoute: typeof rootRouteImport } + '/api/workflow-orchestration': { + id: '/api/workflow-orchestration' + path: '/api/workflow-orchestration' + fullPath: '/api/workflow-orchestration' + preLoaderRoute: typeof ApiWorkflowOrchestrationRouteImport + parentRoute: typeof rootRouteImport + } '/api/video': { id: '/api/video' path: '/api/video' @@ -1138,6 +1159,7 @@ const rootRouteChildren: RootRouteChildren = { ApiTranscriptionRoute: ApiTranscriptionRouteWithChildren, ApiTtsRoute: ApiTtsRouteWithChildren, ApiVideoRoute: ApiVideoRouteWithChildren, + ApiWorkflowOrchestrationRoute: ApiWorkflowOrchestrationRoute, ProviderIndexRoute: ProviderIndexRoute, } export const routeTree = rootRouteImport diff --git a/testing/e2e/src/routes/api.workflow-orchestration.ts b/testing/e2e/src/routes/api.workflow-orchestration.ts new file mode 100644 index 000000000..7d3ff179d --- /dev/null +++ b/testing/e2e/src/routes/api.workflow-orchestration.ts @@ -0,0 +1,71 @@ +import { createFileRoute } from '@tanstack/react-router' +import { EventType, toServerSentEventsResponse } from '@tanstack/ai' +import { + agentMiddleware, + defineAgent, + toAIStream, +} from '@tanstack/ai-orchestration' +import { + createWorkflow, + inMemoryRunStore, + runWorkflow, +} from '@tanstack/workflow-core' +import type { StreamChunk } from '@tanstack/ai' + +const writer = defineAgent({ + name: 'writer', + run: () => fixedAgentStream(), +}) + +const workflow = createWorkflow({ id: 'e2e-orchestration' }) + .middleware([agentMiddleware()]) + .handler((ctx) => ctx.ai.agent('draft', writer, undefined)) + +async function* fixedAgentStream(): AsyncIterable { + yield { + type: EventType.RUN_STARTED, + runId: 'inner-model-run', + threadId: 'inner-model-thread', + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_START, + messageId: 'workflow-message', + role: 'assistant', + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_CONTENT, + messageId: 'workflow-message', + delta: 'Workflow-backed response', + timestamp: Date.now(), + } + yield { + type: EventType.TEXT_MESSAGE_END, + messageId: 'workflow-message', + timestamp: Date.now(), + } + yield { + type: EventType.RUN_FINISHED, + runId: 'inner-model-run', + threadId: 'inner-model-thread', + timestamp: Date.now(), + } +} + +export const Route = createFileRoute('/api/workflow-orchestration')({ + server: { + handlers: { + POST: () => + toServerSentEventsResponse( + toAIStream( + runWorkflow({ + workflow, + runStore: inMemoryRunStore(), + input: {}, + }), + ), + ), + }, + }, +}) diff --git a/testing/e2e/tests/workflow-orchestration.spec.ts b/testing/e2e/tests/workflow-orchestration.spec.ts new file mode 100644 index 000000000..9ef5ffeb0 --- /dev/null +++ b/testing/e2e/tests/workflow-orchestration.spec.ts @@ -0,0 +1,46 @@ +import { expect, test } from '@playwright/test' + +/** + * Provider-free Workflow integration harness. The route uses a fixed agent + * stream, so no model HTTP request reaches aimock. + */ +test('Workflow-backed orchestration projects one outer AG-UI run', async ({ + request, +}) => { + const response = await request.post('/api/workflow-orchestration') + expect(response.ok()).toBeTruthy() + + const chunks = parseSse(await response.text()) + const types = chunks.map((chunk) => chunk.type) + + expect(types).toEqual([ + 'RUN_STARTED', + 'STEP_STARTED', + 'TEXT_MESSAGE_START', + 'TEXT_MESSAGE_CONTENT', + 'TEXT_MESSAGE_END', + 'STEP_FINISHED', + 'RUN_FINISHED', + ]) + expect(chunks[1]).toMatchObject({ + stepName: 'writer', + stepId: 'draft', + stepType: 'agent', + }) + expect(chunks[3]).toMatchObject({ delta: 'Workflow-backed response' }) + expect(chunks.at(-1)).toMatchObject({ + result: 'Workflow-backed response', + }) + expect(chunks.some((chunk) => chunk.runId === 'inner-model-run')).toBe(false) +}) + +function parseSse(body: string): Array> { + return body + .split('\n\n') + .filter((block) => block.trim().length > 0) + .map((block) => { + const data = block.split('\n').find((line) => line.startsWith('data:')) + if (!data) throw new Error('SSE event missing data') + return JSON.parse(data.slice(data.indexOf(':') + 1).trim()) + }) +}