From ccd836b543033a629e1bac6e33be413a542c9708 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 10:17:13 +0300 Subject: [PATCH 1/2] LOU-V2: typed, versioned event stream (agent.stream()) WIP before merging main. Co-Authored-By: Claude Opus 5.5 --- README.md | 15 +- docs/streaming.md | 286 ++++++++++++++++++ src/createAgent.ts | 44 ++- src/execution/AgentExecutor.ts | 62 +++- src/execution/agentEvents.test-d.ts | 88 ++++++ src/execution/agentEvents.ts | 235 +++++++++++++++ src/execution/agentRun.ts | 309 +++++++++++++++++++ src/execution/agentStream.aiSdk.test.ts | 71 +++++ src/execution/agentStream.test.ts | 378 ++++++++++++++++++++++++ src/execution/generateStep.ts | 8 +- src/execution/index.ts | 27 ++ src/execution/streamStep.ts | 92 ++++++ 12 files changed, 1596 insertions(+), 19 deletions(-) create mode 100644 docs/streaming.md create mode 100644 src/execution/agentEvents.test-d.ts create mode 100644 src/execution/agentEvents.ts create mode 100644 src/execution/agentRun.ts create mode 100644 src/execution/agentStream.aiSdk.test.ts create mode 100644 src/execution/agentStream.test.ts create mode 100644 src/execution/streamStep.ts diff --git a/README.md b/README.md index a1b25f4b6..b1a749762 100644 --- a/README.md +++ b/README.md @@ -22,6 +22,7 @@ dashboard — any provider, any deploy target, no lock-in. - **Human-in-the-loop** — flag a tool `needsApproval` and pause execution until a human approves or rejects it, then `resumeAfterApproval()` from any process - **Durable execution** — pass a `sessionId` + `checkpointStore` and a crash mid-conversation resumes instead of restarting - **Cancellation** — pass an `AbortSignal` (`agent.send(input, { signal })`) to stop a run; it resolves with `finishReason: 'aborted'` and the transcript so far, and the signal reaches the provider, tools and delegated agents +- **Streaming events** — `for await (const event of agent.stream(input))` yields a typed, versioned, JSON-serializable event stream (`text.delta` tokens as they arrive, `tool.start`/`tool.done`, steps, approvals, a final `run.done`) ready to forward over SSE or WebSockets; `await run.result` gives the same result as `send()` ([docs/streaming.md](docs/streaming.md)) - **Parallel tool calls** — when the model asks for several tools in one turn they run concurrently (cap it with `toolConcurrency`, or `1` for sequential), and results still reach the transcript in the model's call order - **Skills** — `defineSkill()` / `loadSkills('./skills')`: only each skill's name and description sit in the system prompt; the model loads the full markdown on demand through an auto-registered `load_skill` tool - **Multi-agent delegation** — wrap a child agent as a tool with `createDelegateTool()`, with a `maxDepth` guard against delegation loops @@ -52,6 +53,17 @@ console.log(text); `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `OPENROUTER_API_KEY`, `OLLAMA_BASE_URL` that is present. +Stream the reply token by token instead (see [docs/streaming.md](docs/streaming.md) for every event): + +```typescript +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +for await (const event of agent.stream('Hello!')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +``` + When you need a custom provider (your own `LLMProvider`, a mock, extra config), pass the instance instead: @@ -212,6 +224,7 @@ console.log(result.usage.totalTokens, result.finishReason, result.steps); - [Configuration](docs/configuration.md) - agent spec fields, provider env vars, `AgentExecutor.execute()` options, CLI flags - [Deployment](docs/deployment.md) - `loushy build` targets: Node server, Docker, Cloudflare Workers - [API Overview](docs/api-overview.md) - the main exports; `npm run docs:build` generates the full TypeDoc reference +- [Streaming](docs/streaming.md) - `agent.stream()`: the typed event schema, terminal and SSE examples - [Tracing and observability](docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in - [Testing](docs/testing.md) - unit-test agents deterministically with the scripted `mockModel` - [Skills](docs/skills.md) - on-demand instructions: `defineSkill()`, `loadSkills()`, how they save context @@ -575,7 +588,7 @@ npm run pipeline:demo:trigger # POSTs a synthetic error to kick it off ### Core -- **`createAgent()`** - zero-config `{ send }` agent +- **`createAgent()`** - zero-config `{ send, stream }` agent - **`AgentBuilder`** - fluent `AgentConfig` builder - **`AgentExecutor`** - static executor (`execute()`, approvals, checkpoints, tracing) - **`ToolRegistry`** - manage available tools diff --git a/docs/streaming.md b/docs/streaming.md new file mode 100644 index 000000000..458937f09 --- /dev/null +++ b/docs/streaming.md @@ -0,0 +1,286 @@ +# Streaming + +`agent.stream()` runs an agent exactly like `agent.send()` and reports the run +as a stream of typed events: model text as it arrives, every tool call, +step boundaries, approval requests, and a final `run.done`. Every event is a +plain JSON object, so you can forward it to a browser over Server-Sent Events +or a WebSocket without converting it. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +for await (const event of agent.stream('Weather in Paris?')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +``` + +The full API has the same method: `AgentExecutor.stream(options)` takes the +same options as `AgentExecutor.execute()` (approval store, checkpoints, hooks, +tracing, `toolConcurrency`, ...). + +## The `AgentRun` handle + +`stream()` returns an `AgentRun`: + +```ts no-verify +interface AgentRun extends AsyncIterable { + readonly runId: string; // the runId on every event + readonly result: Promise; // what send() / execute() would return +} +``` + +- **The run starts immediately.** You do not have to iterate: `await run.result` + on its own drives the run to completion. +- **`result`** resolves with the same `ExecutionResult` that `send()` returns, + and rejects with the same error `send()` would reject with. An aborted run + resolves with `finishReason: 'aborted'`; a run paused for approval resolves + with `finishReason: 'awaiting-approval'` and `approvalId`. Leaving `result` + unawaited never causes an unhandled rejection. +- **Breaking out of the `for await` loop early aborts the run** (through the + same path as `signal`, see [Cancellation](#cancellation)). `result` then + resolves with `finishReason: 'aborted'`. +- **No backpressure; nothing is dropped.** The run never waits for the + consumer. Events are buffered until you read them, so a slow consumer sees + every event, in order. Iterating after the run finished still yields all of + its events. +- **One consumer.** An `AgentRun` can be iterated once; a second `for await` + throws. To fan out, collect the events yourself. +- Invalid options (no `provider`, `agent` or `input`, a bad `toolConcurrency`) + throw synchronously from `stream()`. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +const run = agent.stream('Summarize the report', { signal: AbortSignal.timeout(30_000) }); +const result = await run.result; // no iteration needed +console.log(result.finishReason, result.text); +``` + +## Event schema (version 1) + +Every event has these fields: + +| Field | Type | Meaning | +| ----------- | -------- | ------- | +| `type` | string | The event type (table below). Narrow on it in TypeScript. | +| `runId` | string | Identifies the run. The same on every event of one `stream()` call. | +| `seq` | number | `0` for the first event, then `+1` per event, with no gaps. | +| `timestamp` | string | When the event was emitted, ISO 8601 (`2026-10-01T09:30:00.000Z`). | +| `v` | `1` | Schema version, exported as `AGENT_EVENT_SCHEMA_VERSION`. | + +The event types and their extra fields: + +| `type` | Extra fields | Emitted | +| -------------------- | ------------ | ------- | +| `run.start` | `agentName: string`, `agentId?: string` | First event of every run. | +| `step.start` | `step: number` | A model step begins: one model call plus the tool calls it asks for. `step` counts from 1 (a run resumed from a checkpoint continues the count). | +| `text.delta` | `text: string` | A chunk of model text, as it arrives. | +| `text.done` | `text: string` | The step's complete text: the concatenation of its `text.delta` events. Only for steps with text. | +| `tool.start` | `toolCallId: string`, `toolName: string`, `args: Record` | A tool call starts. `args` are the model's arguments parsed from JSON (`{}` when they are not valid JSON). | +| `tool.done` | `toolCallId`, `toolName`, `result: unknown`, `durationMs: number` | A tool call returned. `result` is the value as it would be JSON-encoded (`undefined` becomes `null`, a `Date` becomes a string). `durationMs` counts from its `tool.start`. | +| `tool.error` | `toolCallId`, `toolName`, `error: { name: string, message: string }`, `durationMs: number` | A tool call failed: it threw, its arguments did not match its schema (`name: 'ToolArgumentsValidationError'`), or the tool does not exist. The model gets the error as the call's result and the run continues. | +| `approval.requested` | `approvalId: string`, `toolCallId`, `toolName`, `args: Record` | A tool call needs a human decision. The run then stops; resume it with `resumeAfterApproval()` (see the README). | +| `step.done` | `step: number`, `finishReason: string`, `usage?: { promptTokens, completionTokens, totalTokens }` | A step ends. `finishReason` is the model's (`'stop'`, `'tool_calls'`, `'length'`, ...), or `'awaiting-approval'`, `'aborted'` or `'error'` when the step ended that way. `usage` is this step's model call, absent when the call produced no response. | +| `error` | `error: { name: string, message: string }` | An error. If it ends the run, `run.done` with `finishReason: 'error'` follows. A provider error retried under `surfaceRetryableProviderErrors` is followed by further steps instead. | +| `run.done` | `finishReason: string`, `text: string`, `usage?: { promptTokens, completionTokens, totalTokens }` | Last event of every run, exactly once, including aborted, failed and awaiting-approval runs. `finishReason` and `text` match `run.result`; a failed run has `finishReason: 'error'`, `text: ''` and no `usage`. | + +Optional fields are left out when they have no value. They are never +`undefined`, so `JSON.parse(JSON.stringify(event))` returns an equal object. + +### Ordering guarantees + +- `run.start` is first and `run.done` is last, each exactly once. +- Each `step.start` is followed by exactly one `step.done` with the same + `step`, before the next `step.start`. Everything a step does happens between + the two. +- Inside a step: `text.delta` events, then `text.done`, then the tool events. +- Tool calls of one step run in parallel (see `toolConcurrency`): + `tool.start` events come in the model's call order and `tool.done` / + `tool.error` events in completion order. Match them by `toolCallId`. +- When a call needs approval, the calls before it run and report, then + `approval.requested`, `step.done` (`'awaiting-approval'`) and `run.done` + (`'awaiting-approval'`) follow. Calls after it never start. + +A text-only run: + +``` +run.start → step.start → text.delta × n → text.done → step.done → run.done +``` + +A run with one tool call: + +``` +run.start +step.start(1) → tool.start → tool.done → step.done(1, 'tool_calls') +step.start(2) → text.delta × n → text.done → step.done(2, 'stop') +run.done('stop') +``` + +### TypeScript + +`AgentEvent` is a discriminated union: narrowing on `event.type` gives the +payload. Each event type is exported too (`TextDeltaEvent`, `ToolDoneEvent`, +`RunDoneEvent`, ...), and `AgentEventOf<'tool.done'>` picks one by name. + +```ts +import { isAgentEvent, isToolEvent, type AgentEvent } from '@loushy/build-ai-agent'; + +function render(event: AgentEvent): string { + switch (event.type) { + case 'text.delta': + return event.text; + case 'tool.start': + return `\n[${event.toolName}] ${JSON.stringify(event.args)}\n`; + case 'tool.error': + return `\n[${event.toolName} failed: ${event.error.message}]\n`; + case 'run.done': + return `\n(${event.finishReason})\n`; + default: + return ''; + } +} + +// isToolEvent / isTextEvent / isStepEvent narrow to an event family: +const toolNames = (events: AgentEvent[]) => events.filter(isToolEvent).map((e) => e.toolName); + +// isAgentEvent validates an event received over the wire: +const received: unknown = JSON.parse('{"type":"run.start","runId":"r","seq":0,"timestamp":"","v":1,"agentName":"a"}'); +if (isAgentEvent(received)) console.log(render(received), toolNames([received])); +``` + +### Versioning + +`v` changes only when an existing event changes incompatibly (a field +removed, renamed or retyped). New event types and new optional fields can be +added without changing `v`, so ignore event types you do not know. + +## Token streaming and providers + +When the provider implements `stream()` (all built-in providers and +`mockModel` do), each model step is streamed and `text.delta` events arrive +as the model produces text. Tool calls are assembled from the stream. When a +provider has no `stream()`, or its `supportsStreaming(model)` returns +`false`, the step falls back to `generate()` and its text arrives as a single +`text.delta` followed by `text.done`. + +Streaming changes only how one model step is obtained. Hooks, argument +validation, approvals, parallel tool calls, checkpoints, cancellation, +tracing spans and the other `execute()` callbacks behave exactly as in +`send()`. `send()` and `execute()` themselves still use `generate()`. + +A custom provider's `stream()` should yield `text-delta` chunks and a final +`finish` chunk with `finishReason` and `usage`. Tool calls can be yielded as +`tool-call` chunks, or resolved on the `toolCalls` promise of the +`StreamResult`. An `error` chunk fails the step. + +## Cancellation + +Pass `signal` to stop a run from outside, or break out of the loop. Both end +the run the same way as an aborted `send()`: the signal is checked between +stream chunks, before every model call and every tool call, and reaches the +provider and the tools. The stream ends with `step.done` (`'aborted'`, when a +step was running) and `run.done` (`'aborted'`), and `result` resolves with +`finishReason: 'aborted'`. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +const run = agent.stream('Write a long story'); + +let chars = 0; +for await (const event of run) { + if (event.type === 'text.delta') chars += event.text.length; + if (chars > 500) break; // aborts the run +} +console.log((await run.result).finishReason); // 'aborted' +``` + +## Example: terminal + +```ts +import { createAgent, defineTool } from '@loushy/build-ai-agent'; +import { z } from 'zod'; + +const getWeather = defineTool({ + name: 'get_weather', + description: 'Current weather for a city', + input: z.object({ city: z.string() }), + execute: async ({ city }) => ({ city, tempC: 21 }), +}); + +const agent = createAgent({ model: 'openai/gpt-4o-mini', tools: [getWeather] }); + +for await (const event of agent.stream('Weather in Paris?')) { + switch (event.type) { + case 'text.delta': + process.stdout.write(event.text); + break; + case 'tool.start': + console.log(`\n→ ${event.toolName}(${JSON.stringify(event.args)})`); + break; + case 'tool.done': + console.log(`← ${event.toolName} in ${event.durationMs}ms`); + break; + case 'run.done': + console.log(`\n[${event.finishReason}, ${event.usage?.totalTokens ?? 0} tokens]`); + break; + } +} +``` + +## Example: Server-Sent Events + +Events are JSON-serializable, so an SSE endpoint is one `res.write` per +event. Abort the run when the client disconnects. + +```ts +import { createServer } from 'node:http'; +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +createServer(async (req, res) => { + const question = new URL(req.url ?? '/', 'http://localhost').searchParams.get('q') ?? 'Hello!'; + const controller = new AbortController(); + res.on('close', () => controller.abort()); + + res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' }); + for await (const event of agent.stream(question, { signal: controller.signal })) { + res.write(`data: ${JSON.stringify(event)}\n\n`); + } + res.end(); +}).listen(3000); +``` + +In the browser: + +```ts no-verify +const source = new EventSource('/chat?q=Weather%20in%20Paris%3F'); +source.onmessage = (message) => { + const event = JSON.parse(message.data); + if (event.type === 'text.delta') output.textContent += event.text; + if (event.type === 'run.done') source.close(); +}; +``` + +## Example: the full API + +`AgentExecutor.stream()` accepts every `execute()` option. Here a tool that +needs approval ends the stream with `approval.requested`: + +```ts +import { AgentExecutor } from '@loushy/build-ai-agent'; + +const run = AgentExecutor.stream({ agent, input, provider, toolRegistry, approvalStore }); +for await (const event of run) { + if (event.type === 'approval.requested') { + console.log(`Approve ${event.toolName}(${JSON.stringify(event.args)})? id=${event.approvalId}`); + } +} +const paused = await run.result; // paused.finishReason === 'awaiting-approval' +``` diff --git a/src/createAgent.ts b/src/createAgent.ts index c3cdcb67c..0ad674b5e 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -17,7 +17,8 @@ import { AgentBuilder } from './core/AgentBuilder'; import { AgentType } from './types'; -import { AgentExecutor, ExecutionResult } from './execution/AgentExecutor'; +import { AgentExecutor, ExecuteOptions, ExecutionResult } from './execution/AgentExecutor'; +import type { AgentRun } from './execution/agentRun'; import { LLMProvider } from './providers/llm'; import { ToolRegistry } from './tools/ToolRegistry'; import { ToolDescriptor } from './types'; @@ -147,6 +148,22 @@ export interface SendOptions { export interface SimpleAgent { /** Send a single user message and get back the full execution result text. */ send: (message: string, options?: SendOptions) => Promise; + /** + * Send a single user message and stream the run as typed events (LOU-V2): + * `text.delta` chunks as the model writes, `tool.start`/`tool.done`, + * step boundaries, and a final `run.done`. Iterate the returned + * `AgentRun`, or await its `result` (the same `ExecutionResult` `send()` + * returns). Breaking out of the loop early aborts the run. See + * docs/streaming.md. + * + * @example + * ```ts + * for await (const event of agent.stream('Weather in Paris?')) { + * if (event.type === 'text.delta') process.stdout.write(event.text); + * } + * ``` + */ + stream: (message: string, options?: SendOptions) => AgentRun; } /** @@ -179,18 +196,23 @@ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { if (config.provider && config.model) builder.setSettings({ model: config.model }); const agent = builder.build(); + const executeOptions = (message: string, options: SendOptions): ExecuteOptions => ({ + agent, + input: message, + provider, + toolRegistry, + skills: config.skills, + maxSteps: config.maxSteps, + toolConcurrency: config.toolConcurrency, + signal: options.signal, + }); + return { async send(message: string, options: SendOptions = {}): Promise { - return AgentExecutor.execute({ - agent, - input: message, - provider, - toolRegistry, - skills: config.skills, - maxSteps: config.maxSteps, - toolConcurrency: config.toolConcurrency, - signal: options.signal, - }); + return AgentExecutor.execute(executeOptions(message, options)); + }, + stream(message: string, options: SendOptions = {}): AgentRun { + return AgentExecutor.stream(executeOptions(message, options)); }, }; } diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index 8885d7f5e..d423d5267 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -51,6 +51,7 @@ import { saveStepCheckpoint, toExecutionResult, } from './agentRunState'; +import { AgentRun, RUN_EVENTS, StreamingExecuteOptions, runEventsOf, startAgentRun } from './agentRun'; export { PropagatingToolError } from './propagatingToolError'; @@ -421,6 +422,45 @@ export class AgentExecutor { ); } + /** + * Runs the agent like {@link AgentExecutor.execute} - same options, same + * loop, same `ExecutionResult` - and streams the run as typed + * `AgentEvent`s (LOU-V2). Model output is streamed through + * `provider.stream()` when the provider has it (one `text.delta` per + * chunk); otherwise each step uses `generate()` and emits its text as a + * single `text.delta`. `onEvent` and the other callbacks still fire. + * + * The run starts immediately. Iterate the returned {@link AgentRun} for + * the events, or await `run.result`; breaking out of the `for await` + * early aborts the run (`result` then resolves with + * `finishReason: 'aborted'`). Invalid options throw synchronously. + * See docs/streaming.md for the event schema. + * + * @example + * ```ts + * const run = AgentExecutor.stream({ agent, input: 'Weather in Paris?', provider, toolRegistry }); + * for await (const event of run) { + * if (event.type === 'text.delta') process.stdout.write(event.text); + * } + * const { finishReason } = await run.result; + * ``` + */ + static stream(options: ExecuteOptions): AgentRun { + this.validateExecuteOptions(options, 'AgentExecutor.stream'); + return startAgentRun(({ signal, onEvent, sink }) => { + const streaming: StreamingExecuteOptions = { + ...options, + signal, + onEvent: (event) => { + options.onEvent?.(event); + onEvent(event); + }, + [RUN_EVENTS]: sink, + }; + return this.execute(streaming); + }, options.signal); + } + /** * The actual execution loop, split out of execute() so the top-level * 'agent.run' span (LOU-E5) can wrap it via withSpan() while still @@ -477,10 +517,15 @@ export class AgentExecutor { tools: ToolDefinition[], agentSpanId: string ): Promise<'continue' | 'stop' | ExecutionResult> { + const runEvents = runEventsOf(options); + runEvents?.stepStart(state.steps); try { - return await this.runStep(options, state, tools, agentSpanId); + const outcome = await this.runStep(options, state, tools, agentSpanId); + runEvents?.stepDone(state.steps, typeof outcome === 'string' ? undefined : outcome.finishReason); + return outcome; } catch (error) { if (options.signal?.aborted) { + runEvents?.stepDone(state.steps, 'aborted'); return this.abortRun(options, state); } this.emitEvent(options.onEvent, { @@ -488,6 +533,7 @@ export class AgentExecutor { timestamp: new Date(), error: error as Error, }); + runEvents?.stepDone(state.steps, 'error'); throw error; } } @@ -799,6 +845,7 @@ export class AgentExecutor { }; await approvalStore.save(pending, snapshot); + runEventsOf(options)?.approvalRequested(pending); this.emitEvent(onEvent, { type: 'finish', @@ -932,26 +979,29 @@ export class AgentExecutor { * runAgentLoop() (e.g. `provider.generate(...)` when `provider` is * undefined). */ - private static validateExecuteOptions(options: ExecuteOptions): void { + private static validateExecuteOptions( + options: ExecuteOptions, + caller = 'AgentExecutor.execute' + ): void { if (!options || !options.provider) { throw new Error( - "AgentExecutor.execute: 'provider' is required. " + + `${caller}: 'provider' is required. ` + "Example: AgentExecutor.execute({ agent, input, provider: myProvider })" ); } if (!options.agent) { throw new Error( - "AgentExecutor.execute: 'agent' is required. " + + `${caller}: 'agent' is required. ` + 'Example: AgentExecutor.execute({ agent: AgentBuilder.create()...build(), input, provider })' ); } if (options.input === undefined || options.input === null) { throw new Error( - "AgentExecutor.execute: 'input' is required. " + + `${caller}: 'input' is required. ` + "Example: AgentExecutor.execute({ agent, input: 'hello', provider })" ); } - assertToolConcurrency(options.toolConcurrency, 'AgentExecutor.execute'); + assertToolConcurrency(options.toolConcurrency, caller); } /** diff --git a/src/execution/agentEvents.test-d.ts b/src/execution/agentEvents.test-d.ts new file mode 100644 index 000000000..94e2a0dd9 --- /dev/null +++ b/src/execution/agentEvents.test-d.ts @@ -0,0 +1,88 @@ +import { describe, it, expectTypeOf } from 'vitest'; +import { createAgent } from '../createAgent'; +import { createMockProvider } from '../providers/mock'; +import type { ExecutionResult } from './AgentExecutor'; +import type { AgentRun } from './agentRun'; +import { + AGENT_EVENT_SCHEMA_VERSION, + isToolEvent, + type AgentEvent, + type AgentEventError, + type AgentEventOf, + type AgentEventUsage, +} from './agentEvents'; + +const event = { + type: 'text.delta', + text: 'hi', + runId: 'r1', + seq: 0, + timestamp: new Date(0).toISOString(), + v: AGENT_EVENT_SCHEMA_VERSION, +} as AgentEvent; + +describe('AgentEvent types', () => { + it('narrows the payload on event.type', () => { + switch (event.type) { + case 'text.delta': + case 'text.done': + expectTypeOf(event.text).toBeString(); + break; + case 'step.start': + expectTypeOf(event.step).toBeNumber(); + // @ts-expect-error - step.start has no finishReason + void event.finishReason; + break; + case 'tool.start': + expectTypeOf(event.args).toEqualTypeOf>(); + expectTypeOf(event.toolCallId).toBeString(); + break; + case 'tool.done': + expectTypeOf(event.result).toBeUnknown(); + expectTypeOf(event.durationMs).toBeNumber(); + break; + case 'tool.error': + case 'error': + expectTypeOf(event.error).toEqualTypeOf(); + break; + case 'approval.requested': + expectTypeOf(event.approvalId).toBeString(); + break; + case 'step.done': + expectTypeOf(event.usage).toEqualTypeOf(); + break; + case 'run.done': + expectTypeOf(event.text).toBeString(); + expectTypeOf(event.finishReason).toEqualTypeOf(); + break; + case 'run.start': + expectTypeOf(event.agentName).toBeString(); + break; + default: + expectTypeOf(event).toBeNever(); + } + }); + + it('carries the common fields on every event', () => { + expectTypeOf(event.seq).toBeNumber(); + expectTypeOf(event.runId).toBeString(); + expectTypeOf(event.timestamp).toBeString(); + expectTypeOf(event.v).toEqualTypeOf(); + expectTypeOf(event.v).toEqualTypeOf<1>(); + }); + + it('resolves AgentEventOf and the family guards', () => { + expectTypeOf['toolName']>().toBeString(); + if (isToolEvent(event)) { + expectTypeOf(event.type).toEqualTypeOf<'tool.start' | 'tool.done' | 'tool.error'>(); + } + }); + + it('agent.stream() returns an AgentRun: async iterable of events plus result', () => { + const agent = createAgent({ provider: createMockProvider() }); + const run = agent.stream('hi', { signal: AbortSignal.timeout(1000) }); + expectTypeOf(run).toEqualTypeOf(); + expectTypeOf(run).toMatchTypeOf>(); + expectTypeOf(run.result).toEqualTypeOf>(); + }); +}); diff --git a/src/execution/agentEvents.ts b/src/execution/agentEvents.ts new file mode 100644 index 000000000..edee099c6 --- /dev/null +++ b/src/execution/agentEvents.ts @@ -0,0 +1,235 @@ +/** + * LOU-V2: the typed, versioned event schema of `agent.stream()` / + * `AgentExecutor.stream()`. Every event is a plain JSON value (no `Error` + * instances, no functions, no `undefined` fields), so it can be sent over + * SSE or a WebSocket unchanged and parsed back to the same object. + * + * See docs/streaming.md for the full table - this file is the public schema. + */ + +import type { ExecutionFinishReason } from './AgentExecutor'; + +/** + * Version of the {@link AgentEvent} schema, carried on every event as `v`. + * It changes only on a breaking change to an existing event (a field removed, + * renamed or retyped); new event types and new optional fields keep it. + */ +export const AGENT_EVENT_SCHEMA_VERSION = 1 as const; + +/** Token usage, as reported on `step.done` and `run.done`. */ +export interface AgentEventUsage { + promptTokens: number; + completionTokens: number; + totalTokens: number; +} + +/** A JSON-safe error: the `name` and `message` of the original error. */ +export interface AgentEventError { + name: string; + message: string; +} + +/** Fields every {@link AgentEvent} carries. */ +export interface AgentEventBase { + /** Discriminant: narrow on it to get the event's payload. */ + type: TType; + /** Identifies the run; the same on every event of one `stream()` call. */ + runId: string; + /** 0 for the first event of the run, then +1 per event, with no gaps. */ + seq: number; + /** When the event was emitted, as an ISO-8601 string. */ + timestamp: string; + /** Schema version, always {@link AGENT_EVENT_SCHEMA_VERSION}. */ + v: typeof AGENT_EVENT_SCHEMA_VERSION; +} + +/** First event of every run. */ +export interface RunStartEvent extends AgentEventBase<'run.start'> { + agentName: string; + /** The agent's id, when it has one. */ + agentId?: string; +} + +/** A model step (one model call plus the tool calls it asked for) begins. `step` counts from 1. */ +export interface StepStartEvent extends AgentEventBase<'step.start'> { + step: number; +} + +/** A chunk of model text, emitted as it arrives. */ +export interface TextDeltaEvent extends AgentEventBase<'text.delta'> { + text: string; +} + +/** The complete text of the current step (the concatenation of its `text.delta`s). */ +export interface TextDoneEvent extends AgentEventBase<'text.done'> { + text: string; +} + +/** A tool call starts. Emitted in the model's call order. */ +export interface ToolStartEvent extends AgentEventBase<'tool.start'> { + toolCallId: string; + toolName: string; + /** The arguments the model sent, parsed from JSON (`{}` when they are not valid JSON). */ + args: Record; +} + +/** A tool call returned. Emitted in completion order. */ +export interface ToolDoneEvent extends AgentEventBase<'tool.done'> { + toolCallId: string; + toolName: string; + /** The tool's result as it would be JSON-encoded (`undefined` becomes `null`). */ + result: unknown; + /** Milliseconds since this call's `tool.start`. */ + durationMs: number; +} + +/** + * A tool call failed (it threw, its arguments were invalid, or the tool does + * not exist). The model receives the error as the call's result and the run + * continues. + */ +export interface ToolErrorEvent extends AgentEventBase<'tool.error'> { + toolCallId: string; + toolName: string; + error: AgentEventError; + /** Milliseconds since this call's `tool.start`. */ + durationMs: number; +} + +/** + * A tool call needs a human decision. The run then ends with + * `run.done { finishReason: 'awaiting-approval' }`; resume it with + * `resumeAfterApproval(approvalId, ...)`. + */ +export interface ApprovalRequestedEvent extends AgentEventBase<'approval.requested'> { + approvalId: string; + toolCallId: string; + toolName: string; + args: Record; +} + +/** A step ends. Every `step.start` is followed by exactly one `step.done`. */ +export interface StepDoneEvent extends AgentEventBase<'step.done'> { + step: number; + /** + * The model's finish reason for this step (`'stop'`, `'tool_calls'`, ...), + * or `'awaiting-approval'`, `'aborted'` or `'error'` when the step ended that way. + */ + finishReason: ExecutionFinishReason; + /** Tokens used by this step's model call, when it produced a response. */ + usage?: AgentEventUsage; +} + +/** + * An error. When it ends the run, `run.done { finishReason: 'error' }` + * follows; a provider error retried under `surfaceRetryableProviderErrors` + * is followed by more steps instead. + */ +export interface AgentErrorEvent extends AgentEventBase<'error'> { + error: AgentEventError; +} + +/** + * Last event of every run, emitted exactly once - also for aborted, failed + * and awaiting-approval runs. + */ +export interface RunDoneEvent extends AgentEventBase<'run.done'> { + /** `ExecutionResult.finishReason`, or `'error'` when the run failed. */ + finishReason: ExecutionFinishReason; + /** `ExecutionResult.text` (`''` when the run failed). */ + text: string; + /** Total tokens of the run; absent when the run failed. */ + usage?: AgentEventUsage; +} + +/** + * Every event `agent.stream()` yields, discriminated by `type`. + * + * @example + * ```ts + * for await (const event of agent.stream('Weather in Paris?')) { + * if (event.type === 'text.delta') process.stdout.write(event.text); + * if (event.type === 'tool.start') console.log(`calling ${event.toolName}`, event.args); + * } + * ``` + */ +export type AgentEvent = + | RunStartEvent + | StepStartEvent + | TextDeltaEvent + | TextDoneEvent + | ToolStartEvent + | ToolDoneEvent + | ToolErrorEvent + | ApprovalRequestedEvent + | StepDoneEvent + | AgentErrorEvent + | RunDoneEvent; + +/** The `type` of an {@link AgentEvent}. */ +export type AgentEventType = AgentEvent['type']; + +/** + * The event with the given `type`. + * + * @example + * ```ts + * const onTool = (e: AgentEventOf<'tool.done'>) => console.log(e.toolName, e.durationMs); + * ``` + */ +export type AgentEventOf = Extract; + +/** An event without the fields the run fills in (`runId`, `seq`, `timestamp`, `v`). */ +export type AgentEventPayload = { + [K in AgentEventType]: Omit, keyof AgentEventBase> & { type: K }; +}[AgentEventType]; + +const EVENT_TYPES: ReadonlySet = new Set([ + 'run.start', + 'step.start', + 'text.delta', + 'text.done', + 'tool.start', + 'tool.done', + 'tool.error', + 'approval.requested', + 'step.done', + 'error', + 'run.done', +]); + +/** + * Whether `value` looks like an {@link AgentEvent} of this schema version - + * for a client parsing events received over SSE/WebSocket. + * + * @example + * ```ts + * const data: unknown = JSON.parse('{"type":"run.start"}'); + * if (isAgentEvent(data) && data.type === 'text.delta') console.log(data.text); + * ``` + */ +export function isAgentEvent(value: unknown): value is AgentEvent { + if (typeof value !== 'object' || value === null) return false; + const event = value as Partial>; + return ( + event.v === AGENT_EVENT_SCHEMA_VERSION && + typeof event.type === 'string' && + EVENT_TYPES.has(event.type) && + typeof event.seq === 'number' + ); +} + +/** `tool.start`, `tool.done` or `tool.error` - they all carry `toolCallId` and `toolName`. */ +export function isToolEvent(event: AgentEvent): event is ToolStartEvent | ToolDoneEvent | ToolErrorEvent { + return event.type.startsWith('tool.'); +} + +/** `text.delta` or `text.done` - both carry `text`. */ +export function isTextEvent(event: AgentEvent): event is TextDeltaEvent | TextDoneEvent { + return event.type.startsWith('text.'); +} + +/** `step.start` or `step.done` - both carry `step`. */ +export function isStepEvent(event: AgentEvent): event is StepStartEvent | StepDoneEvent { + return event.type.startsWith('step.'); +} diff --git a/src/execution/agentRun.ts b/src/execution/agentRun.ts new file mode 100644 index 000000000..b7875980c --- /dev/null +++ b/src/execution/agentRun.ts @@ -0,0 +1,309 @@ +/** + * LOU-V2: `AgentRun`, the handle `agent.stream()` / `AgentExecutor.stream()` + * return - an async iterable of {@link AgentEvent}s plus a `result` promise. + * + * The run itself is the ordinary AgentExecutor loop. This module only + * (1) translates the loop's existing `onEvent` callbacks into AgentEvents, + * (2) hands the loop a {@link RunEventSink} (under the {@link RUN_EVENTS} + * key of its options) for what those callbacks do not cover - step + * boundaries, approval requests and how one model step is obtained + * (streamed, with `text.delta` per chunk) - and (3) buffers the events + * for the consumer. + * + * Backpressure: none. The run never waits for the consumer; events are + * buffered without loss until they are read. + */ + +import { nanoid } from 'nanoid'; +import type { GenerateOptions, GenerateResult, LLMProvider } from '../providers'; +import type { ExecuteOptions, ExecutionEvent, ExecutionResult } from './AgentExecutor'; +import type { PendingApproval } from './ApprovalGate'; +import { + AGENT_EVENT_SCHEMA_VERSION, + AgentEvent, + AgentEventError, + AgentEventPayload, + AgentEventUsage, +} from './agentEvents'; +import { parseToolArguments } from './toolCallExecution'; +import { canStream, generateViaStream } from './streamStep'; + +/** + * The handle returned by `agent.stream()` and `AgentExecutor.stream()`. + * + * Iterate it with `for await` to get the run's {@link AgentEvent}s (the run + * starts right away and events are buffered until you read them), or just + * await `result`. Breaking out of the loop early aborts the run. It can be + * iterated once. + * + * @example + * ```ts + * const run = agent.stream('Weather in Paris?'); + * for await (const event of run) { + * if (event.type === 'text.delta') process.stdout.write(event.text); + * } + * const result = await run.result; // the same ExecutionResult send() returns + * ``` + */ +export interface AgentRun extends AsyncIterable { + /** The `runId` every event of this run carries. */ + readonly runId: string; + /** + * Settles when the run ends, whether or not the events are read: resolves + * with the ExecutionResult (`finishReason: 'aborted'` after an abort or an + * early `break`), or rejects with the error that failed the run - the same + * outcomes as `send()`/`execute()`. Not awaiting it never causes an + * unhandled rejection. + */ + readonly result: Promise; +} + +/** + * The parts of a run AgentExecutor reports to an AgentRun directly, beyond + * its `onEvent` callbacks. Internal: reached through `options[RUN_EVENTS]`. + */ +export interface RunEventSink { + stepStart(step: number): void; + /** `finishReason` overrides the step's own (the model's) finish reason. */ + stepDone(step: number, finishReason?: string): void; + approvalRequested(pending: PendingApproval): void; + /** Obtains one model step - streamed when the provider can. */ + generate(provider: LLMProvider, request: GenerateOptions): Promise; +} + +/** Options key under which a streaming run hands the loop its {@link RunEventSink}. */ +export const RUN_EVENTS: unique symbol = Symbol('loushy.agentRunEvents'); + +/** ExecuteOptions as passed through the loop of a streaming run. */ +export type StreamingExecuteOptions = ExecuteOptions & { [RUN_EVENTS]?: RunEventSink }; + +/** The run's event sink, when `options` belong to a streaming run. */ +export function runEventsOf(options: ExecuteOptions): RunEventSink | undefined { + return (options as StreamingExecuteOptions)[RUN_EVENTS]; +} + +/** Starts the run with the composed signal, `onEvent` and sink. */ +export type RunStarter = (wiring: { + signal: AbortSignal; + onEvent: (event: ExecutionEvent) => void; + sink: RunEventSink; +}) => Promise; + +function toEventError(error: unknown): AgentEventError { + const err = error as { name?: unknown; message?: unknown } | null | undefined; + return { + name: typeof err?.name === 'string' && err.name ? err.name : 'Error', + message: typeof err?.message === 'string' ? err.message : String(error), + }; +} + +function toEventUsage(usage: AgentEventUsage): AgentEventUsage { + return { + promptTokens: usage.promptTokens, + completionTokens: usage.completionTokens, + totalTokens: usage.totalTokens, + }; +} + +/** `value` as it survives a JSON round trip (`undefined` -> `null`; unserializable -> its string form). */ +function toJsonValue(value: unknown): unknown { + if (value === undefined) return null; + try { + return JSON.parse(JSON.stringify(value)) ?? null; + } catch { + return String(value); + } +} + +class AgentRunImpl implements AgentRun { + readonly runId = nanoid(); + readonly result: Promise; + private readonly controller = new AbortController(); + private readonly queue: AgentEvent[] = []; + private readonly toolStarts = new Map(); + private seq = 0; + private closed = false; + private iterated = false; + private wake: (() => void) | undefined; + private stepResult: GenerateResult | undefined; + private lastError: unknown; + + constructor(start: RunStarter, signal: AbortSignal | undefined) { + const runSignal = signal ? AbortSignal.any([signal, this.controller.signal]) : this.controller.signal; + this.result = start({ + signal: runSignal, + onEvent: (event) => this.translate(event), + sink: this.sink(), + }).then( + (result) => this.finish(result), + (error: unknown) => this.fail(error) + ); + this.result.catch(() => undefined); + } + + async *[Symbol.asyncIterator](): AsyncIterator { + if (this.iterated) { + throw new Error( + 'AgentRun can only be iterated once. Collect the events in the first for-await loop, ' + + 'or call agent.stream() again for a new run.' + ); + } + this.iterated = true; + try { + yield* this.drain(); + } finally { + if (!this.closed) { + this.controller.abort(new DOMException('The AgentRun was not iterated to the end (the for-await loop exited early)', 'AbortError')); + } + } + } + + private async *drain(): AsyncGenerator { + for (;;) { + const next = this.queue.shift(); + if (next) { + yield next; + } else if (this.closed) { + return; + } else { + await new Promise((resolve) => (this.wake = resolve)); + } + } + } + + private emit(payload: AgentEventPayload): void { + if (this.closed) return; + const event = { + ...payload, + runId: this.runId, + seq: this.seq++, + timestamp: new Date().toISOString(), + v: AGENT_EVENT_SCHEMA_VERSION, + } as AgentEvent; + this.queue.push(event); + if (payload.type === 'run.done') this.closed = true; + this.wake?.(); + this.wake = undefined; + } + + private finish(result: ExecutionResult): ExecutionResult { + this.emit({ + type: 'run.done', + finishReason: result.finishReason, + text: result.text, + usage: toEventUsage(result.usage), + }); + return result; + } + + private fail(error: unknown): never { + if (error !== this.lastError) { + this.emit({ type: 'error', error: toEventError(error) }); + } + this.emit({ type: 'run.done', finishReason: 'error', text: '' }); + throw error; + } + + /** Maps the loop's `onEvent` callbacks to AgentEvents (`abort`/`finish` are covered by `run.done`). */ + private translate(event: ExecutionEvent): void { + switch (event.type) { + case 'start': + this.emit({ + type: 'run.start', + agentName: event.agentName ?? '', + ...(event.agentId !== undefined && { agentId: event.agentId }), + }); + break; + case 'text-complete': + this.emit({ type: 'text.done', text: event.text ?? '' }); + break; + case 'tool-call': + if (event.toolCall) this.toolStarted(event.toolCall); + break; + case 'tool-result': + if (event.toolResult) this.toolSettled(event.toolResult); + break; + case 'error': + this.lastError = event.error; + this.emit({ type: 'error', error: toEventError(event.error) }); + break; + } + } + + private toolStarted(toolCall: NonNullable): void { + this.toolStarts.set(toolCall.id, Date.now()); + const args = parseToolArguments(toolCall, {}); + this.emit({ + type: 'tool.start', + toolCallId: toolCall.id, + toolName: toolCall.function.name, + args: (typeof args === 'object' && args !== null ? args : {}) as Record, + }); + } + + private toolSettled(outcome: NonNullable): void { + const { toolCallId, toolName } = outcome; + const durationMs = Date.now() - (this.toolStarts.get(toolCallId) ?? Date.now()); + if (outcome.error === undefined) { + this.emit({ type: 'tool.done', toolCallId, toolName, result: toJsonValue(outcome.result), durationMs }); + return; + } + const name = (outcome.result as { error?: unknown } | null)?.error; + this.emit({ + type: 'tool.error', + toolCallId, + toolName, + error: { name: typeof name === 'string' ? name : 'Error', message: outcome.error }, + durationMs, + }); + } + + private sink(): RunEventSink { + return { + stepStart: (step) => { + this.stepResult = undefined; + this.emit({ type: 'step.start', step }); + }, + stepDone: (step, finishReason) => { + const usage = this.stepResult?.usage; + this.emit({ + type: 'step.done', + step, + finishReason: finishReason ?? this.stepResult?.finishReason ?? 'error', + ...(usage && { usage: toEventUsage(usage) }), + }); + }, + approvalRequested: (pending) => + this.emit({ + type: 'approval.requested', + approvalId: pending.id, + toolCallId: pending.toolCallId, + toolName: pending.toolName, + args: toJsonValue(pending.args) as Record, + }), + generate: async (provider, request) => { + this.stepResult = await this.generateStep(provider, request); + return this.stepResult; + }, + }; + } + + /** Streams the step when the provider can; otherwise one `generate()` and a single `text.delta`. */ + private async generateStep(provider: LLMProvider, request: GenerateOptions): Promise { + const onTextDelta = (text: string) => this.emit({ type: 'text.delta', text }); + if (canStream(provider, request)) { + return generateViaStream(provider, request, onTextDelta); + } + const generated = await provider.generate(request); + if (generated.text) onTextDelta(generated.text); + return generated; + } +} + +/** + * Starts a run and returns its {@link AgentRun} handle. `signal` (the + * caller's) and an early `break` both abort the run. + */ +export function startAgentRun(start: RunStarter, signal?: AbortSignal): AgentRun { + return new AgentRunImpl(start, signal); +} diff --git a/src/execution/agentStream.aiSdk.test.ts b/src/execution/agentStream.aiSdk.test.ts new file mode 100644 index 000000000..c4c76a845 --- /dev/null +++ b/src/execution/agentStream.aiSdk.test.ts @@ -0,0 +1,71 @@ +/** + * LOU-V2: AgentExecutor.stream() through a real 'ai'-SDK-backed provider. + * Its stream reports tool calls only on the `toolCalls` promise (not as + * chunks) and uses the SDK's `'tool-calls'` finish reason spelling. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { z } from 'zod'; + +const streamTextMock = vi.fn(); +vi.mock('ai', async () => { + const actual = await vi.importActual('ai'); + return { ...actual, streamText: (...args: unknown[]) => streamTextMock(...args) }; +}); + +import { createAgent } from '../createAgent'; +import { OpenAIProvider } from '../providers/OpenAIProvider'; +import { defineTool } from '../tools/defineTool'; +import type { AgentEvent } from './agentEvents'; + +const usage = { promptTokens: 4, completionTokens: 2, totalTokens: 6 }; + +function sdkStream(deltas: string[], finishReason: string, toolCalls: unknown[] = []) { + return { + textStream: (async function* () { + yield* deltas; + })(), + text: Promise.resolve(deltas.join('')), + usage: Promise.resolve(usage), + finishReason: Promise.resolve(finishReason), + toolCalls: Promise.resolve(toolCalls), + }; +} + +describe('AgentExecutor.stream() with an ai-SDK provider', () => { + it('assembles tool calls from the stream result and streams the reply text', async () => { + streamTextMock + .mockReturnValueOnce(sdkStream([], 'tool-calls', [{ toolCallId: 'c1', toolName: 'lookup', args: { q: 'x' } }])) + .mockReturnValueOnce(sdkStream(['Found ', 'it.'], 'stop')); + const lookup = defineTool({ + name: 'lookup', + description: 'Look something up', + input: z.object({ q: z.string() }), + execute: async ({ q }) => `result for ${q}`, + }); + const agent = createAgent({ provider: new OpenAIProvider({ name: 'openai', apiKey: 'k' }), tools: [lookup] }); + + const run = agent.stream('find x'); + const events: AgentEvent[] = []; + for await (const event of run) events.push(event); + + expect(events.map((e) => e.type)).toEqual([ + 'run.start', + 'step.start', + 'tool.start', + 'tool.done', + 'step.done', + 'step.start', + 'text.delta', + 'text.delta', + 'text.done', + 'step.done', + 'run.done', + ]); + expect(events.find((e) => e.type === 'tool.done')).toMatchObject({ toolCallId: 'c1', result: 'result for x' }); + expect(events.find((e) => e.type === 'step.done')).toMatchObject({ finishReason: 'tool_calls', usage }); + const result = await run.result; + expect(result.text).toBe('Found it.'); + expect(result.usage).toEqual({ promptTokens: 8, completionTokens: 4, totalTokens: 12 }); + }); +}); diff --git a/src/execution/agentStream.test.ts b/src/execution/agentStream.test.ts new file mode 100644 index 000000000..c05f767ca --- /dev/null +++ b/src/execution/agentStream.test.ts @@ -0,0 +1,378 @@ +/** + * LOU-V2: `agent.stream()` / `AgentExecutor.stream()` - the typed, versioned + * event stream over the ordinary AgentExecutor loop. + */ + +import { describe, it, expect, vi } from 'vitest'; +import { z } from 'zod'; +import { AgentExecutor, ExecutionEvent } from './AgentExecutor'; +import type { ApprovalStore } from './ApprovalGate'; +import type { CheckpointStore } from './checkpoint'; +import { AGENT_EVENT_SCHEMA_VERSION, AgentEvent, AgentEventOf, AgentEventType, isAgentEvent, isStepEvent, isTextEvent, isToolEvent } from './agentEvents'; +import type { AgentRun } from './agentRun'; +import { createAgent } from '../createAgent'; +import { defineTool, DefinedTool } from '../tools/defineTool'; +import { ToolRegistry } from '../tools'; +import { mockModel, MockModel, MockTurn } from '../testing'; +import { AgentConfig, AgentType } from '../types'; +import type { LLMProvider, StreamChunk } from '../providers'; + +/** Reads every event, then checks the invariants every run must satisfy. */ +async function collect(run: AgentRun): Promise { + const events: AgentEvent[] = []; + for await (const event of run) events.push(event); + expectWellFormed(events, run.runId); + return events; +} + +function expectWellFormed(events: AgentEvent[], runId: string): void { + events.forEach((event, index) => { + expect(event.seq).toBe(index); + expect(event.runId).toBe(runId); + expect(event.v).toBe(AGENT_EVENT_SCHEMA_VERSION); + expect(new Date(event.timestamp).toISOString()).toBe(event.timestamp); + expect(JSON.parse(JSON.stringify(event))).toStrictEqual(event); + expect(isAgentEvent(event)).toBe(true); + }); + expect(events[0].type).toBe('run.start'); + expect(events.filter((e) => e.type === 'run.done')).toHaveLength(1); + expect(events.at(-1)?.type).toBe('run.done'); + const starts = events.filter((e) => e.type === 'step.start').map((e) => e.step); + const dones = events.filter((e) => e.type === 'step.done').map((e) => e.step); + expect(dones).toEqual(starts); +} + +const typesOf = (events: AgentEvent[]): AgentEventType[] => events.map((e) => e.type); + +function only(events: AgentEvent[], type: T): AgentEventOf[] { + return events.filter((e): e is AgentEventOf => e.type === type); +} + +function agentWith(provider: LLMProvider, tools: DefinedTool[] = []) { + return createAgent({ instructions: 'Be brief.', provider, tools }); +} + +/** AgentExecutor.stream() options for an agent with `tools` (for options createAgent does not expose). */ +function executorOptions(provider: LLMProvider, tools: DefinedTool[]) { + const toolConfig: AgentConfig['tools'] = {}; + for (const t of tools) toolConfig[t.name] = { tool: t.name }; + const agent: AgentConfig = { id: 'agent-1', name: 'Agent', agentType: AgentType.SmartAssistant, prompt: 'p', tools: toolConfig }; + const toolRegistry = new ToolRegistry(); + toolRegistry.registerMany(tools); + return { agent, provider, toolRegistry, input: 'go' }; +} + +const weather = defineTool({ + name: 'get_weather', + description: 'Weather for a city', + input: z.object({ city: z.string() }), + execute: async ({ city }) => ({ city, tempC: 21 }), +}); + +/** Wraps a mockModel so its stream pauses after the first chunk until `release()`. */ +function pausingAfterFirstChunk(model: MockModel): { provider: MockModel; firstChunk: Promise; release: () => void } { + let release!: () => void; + let reached!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const firstChunk = new Promise((resolve) => (reached = resolve)); + const stream = model.stream.bind(model); + model.stream = async (options) => { + const streamed = await stream(options); + const source = streamed.fullStream; + const fullStream = (async function* (): AsyncGenerator { + let first = true; + for await (const chunk of source) { + yield chunk; + if (first) { + first = false; + reached(); + await gate; + } + } + })(); + return { ...streamed, fullStream }; + }; + return { provider: model, firstChunk, release }; +} + +describe('agent.stream()', () => { + it('streams a text-only run as word chunks, bracketed by run and step events', async () => { + const run = agentWith(mockModel([{ text: 'Hello there world', usage: { inputTokens: 3, outputTokens: 2 } }])).stream('hi'); + const events = await collect(run); + + expect(typesOf(events)).toEqual([ + 'run.start', + 'step.start', + 'text.delta', + 'text.delta', + 'text.delta', + 'text.done', + 'step.done', + 'run.done', + ]); + expect(only(events, 'text.delta').map((e) => e.text)).toEqual(['Hello ', 'there ', 'world']); + expect(only(events, 'text.done')[0].text).toBe('Hello there world'); + const usage = { promptTokens: 3, completionTokens: 2, totalTokens: 5 }; + expect(only(events, 'step.done')[0]).toMatchObject({ step: 1, finishReason: 'stop', usage }); + expect(only(events, 'run.done')[0]).toMatchObject({ finishReason: 'stop', text: 'Hello there world', usage }); + expect(only(events, 'run.start')[0].agentName).toBe('agent'); + + const result = await run.result; + expect(result.text).toBe('Hello there world'); + expect(result.messages.at(-1)).toEqual({ role: 'assistant', content: 'Hello there world' }); + }); + + it('reports a tool run step by step and resolves the same result send() returns', async () => { + const turns: MockTurn[] = [{ toolCalls: [{ name: 'get_weather', args: { city: 'Paris' } }] }, 'Sunny, 21C.']; + const run = agentWith(mockModel(turns), [weather]).stream('Weather in Paris?'); + const events = await collect(run); + + expect(typesOf(events)).toEqual([ + 'run.start', + 'step.start', + 'tool.start', + 'tool.done', + 'step.done', + 'step.start', + 'text.delta', + 'text.delta', + 'text.done', + 'step.done', + 'run.done', + ]); + expect(only(events, 'tool.start')[0]).toMatchObject({ toolCallId: 'call_1', toolName: 'get_weather', args: { city: 'Paris' } }); + const done = only(events, 'tool.done')[0]; + expect(done).toMatchObject({ toolCallId: 'call_1', toolName: 'get_weather', result: { city: 'Paris', tempC: 21 } }); + expect(done.durationMs).toBeGreaterThanOrEqual(0); + expect(only(events, 'step.done').map((e) => [e.step, e.finishReason])).toEqual([ + [1, 'tool_calls'], + [2, 'stop'], + ]); + + const sent = await agentWith(mockModel(turns), [weather]).send('Weather in Paris?'); + const streamed = await run.result; + expect(streamed).toEqual(sent); + }); + + it('starts parallel tool calls in call order and reports them done in completion order', async () => { + let fastDone!: () => void; + const fastFinished = new Promise((resolve) => (fastDone = resolve)); + const slow = defineTool({ + name: 'slow', + description: 'slow', + input: z.object({}), + execute: async () => { + await fastFinished; + await new Promise((resolve) => setTimeout(resolve, 5)); + return 'slow done'; + }, + }); + const fast = defineTool({ + name: 'fast', + description: 'fast', + input: z.object({}), + execute: async () => { + fastDone(); + return 'fast done'; + }, + }); + const model = mockModel([{ toolCalls: [{ name: 'slow' }, { name: 'fast' }] }, 'ok']); + const events = await collect(agentWith(model, [slow, fast]).stream('go')); + + const toolEvents = events.filter(isToolEvent).map((e) => `${e.type}:${e.toolName}`); + expect(toolEvents).toEqual(['tool.start:slow', 'tool.start:fast', 'tool.done:fast', 'tool.done:slow']); + }); + + it('reports a failing tool as tool.error with a JSON-safe error and carries on', async () => { + class WeatherDown extends Error { + override name = 'WeatherDown'; + } + const broken = defineTool({ + name: 'get_weather', + description: 'Weather', + input: z.object({ city: z.string() }), + execute: async () => { + throw new WeatherDown('service unavailable'); + }, + }); + const model = mockModel([{ toolCalls: [{ name: 'get_weather', args: { city: 'Oslo' } }] }, 'Sorry, no data.']); + const run = agentWith(model, [broken]).stream('Weather in Oslo?'); + const events = await collect(run); + + const [failure] = only(events, 'tool.error'); + expect(failure).toMatchObject({ + toolCallId: 'call_1', + toolName: 'get_weather', + error: { name: 'WeatherDown', message: 'service unavailable' }, + }); + expect(failure.durationMs).toBeGreaterThanOrEqual(0); + expect(only(events, 'tool.done')).toHaveLength(0); + expect((await run.result).text).toBe('Sorry, no data.'); + }); + + it('ends with approval.requested and run.done awaiting-approval when a tool needs approval', async () => { + const refund = defineTool({ + name: 'refund', + description: 'Refund an order', + input: z.object({ orderId: z.string() }), + needsApproval: true, + execute: async () => 'refunded', + }); + const saved: string[] = []; + const approvalStore: ApprovalStore = { + save: async (pending) => void saved.push(pending.id), + resolve: async () => null, + }; + const model = mockModel([{ toolCalls: [{ name: 'refund', args: { orderId: 'A1' } }] }]); + const run = AgentExecutor.stream({ ...executorOptions(model, [refund]), approvalStore }); + const events = await collect(run); + + expect(typesOf(events)).toEqual(['run.start', 'step.start', 'tool.start', 'approval.requested', 'step.done', 'run.done']); + const result = await run.result; + expect(only(events, 'approval.requested')[0]).toMatchObject({ + approvalId: result.approvalId, + toolCallId: 'call_1', + toolName: 'refund', + args: { orderId: 'A1' }, + }); + expect(saved).toEqual([result.approvalId]); + expect(only(events, 'step.done')[0].finishReason).toBe('awaiting-approval'); + expect(only(events, 'run.done')[0].finishReason).toBe('awaiting-approval'); + }); + + it('stops mid-stream when the signal is aborted, ending with run.done aborted', async () => { + const { provider, firstChunk, release } = pausingAfterFirstChunk(mockModel(['one two three four'])); + const controller = new AbortController(); + const run = agentWith(provider).stream('count', { signal: controller.signal }); + void firstChunk.then(() => { + controller.abort(); + release(); + }); + const events = await collect(run); + + expect(typesOf(events)).toEqual(['run.start', 'step.start', 'text.delta', 'step.done', 'run.done']); + expect(only(events, 'step.done')[0].finishReason).toBe('aborted'); + expect(only(events, 'run.done')[0].finishReason).toBe('aborted'); + expect((await run.result).finishReason).toBe('aborted'); + }); + + it('returns at once with run.start and run.done for an already-aborted signal', async () => { + const model = mockModel(['never']); + const run = agentWith(model).stream('hi', { signal: AbortSignal.abort() }); + expect(typesOf(await collect(run))).toEqual(['run.start', 'run.done']); + expect((await run.result).finishReason).toBe('aborted'); + expect(model.calls).toHaveLength(0); + }); + + it('aborts the run when the consumer breaks out of the loop early', async () => { + let toolSignal: AbortSignal | undefined; + const hang = defineTool({ + name: 'hang', + description: 'Waits until cancelled', + input: z.object({}), + execute: (_args, { abortSignal }) => { + toolSignal = abortSignal; + return new Promise((_resolve, reject) => { + if (abortSignal?.aborted) reject(abortSignal.reason); + abortSignal?.addEventListener('abort', () => reject(abortSignal.reason)); + }); + }, + }); + const run = agentWith(mockModel([{ toolCalls: [{ name: 'hang' }] }, 'unused']), [hang]).stream('go'); + + const seen: AgentEventType[] = []; + for await (const event of run) { + seen.push(event.type); + if (event.type === 'tool.start') break; + } + + const result = await run.result; + expect(seen).toEqual(['run.start', 'step.start', 'tool.start']); + expect(result.finishReason).toBe('aborted'); + expect(toolSignal?.aborted).toBe(true); + }); + + it('drives the run to completion when only result is awaited, keeping every event for a later read', async () => { + const onEvent = vi.fn<(event: ExecutionEvent) => void>(); + const run = AgentExecutor.stream({ ...executorOptions(mockModel(['Done here']), []), onEvent }); + + const result = await run.result; + expect(result.text).toBe('Done here'); + expect(onEvent.mock.calls.map(([e]) => e.type)).toEqual(['start', 'text-complete', 'finish']); + + const events = await collect(run); + expect(only(events, 'text.delta').map((e) => e.text)).toEqual(['Done ', 'here']); + }); + + it('falls back to generate() with a single text.delta when the provider has no stream()', async () => { + const model = mockModel(['Hi there']); + const generateOnly = Object.assign(Object.create(model) as MockModel, { stream: undefined }); + const events = await collect(agentWith(generateOnly).stream('hi')); + + expect(typesOf(events)).toEqual(['run.start', 'step.start', 'text.delta', 'text.done', 'step.done', 'run.done']); + expect(only(events, 'text.delta')[0].text).toBe('Hi there'); + expect(model.calls).toHaveLength(1); + }); + + it('falls back to generate() when the provider does not support streaming the model', async () => { + const model = mockModel(['Hi there']); + model.supportsStreaming = () => false; + const stream = vi.spyOn(model, 'stream'); + const events = await collect(agentWith(model).stream('hi')); + + expect(only(events, 'text.delta').map((e) => e.text)).toEqual(['Hi there']); + expect(stream).not.toHaveBeenCalled(); + }); + + it('reports a provider failure as error then run.done error, and result rejects', async () => { + const run = agentWith(mockModel([{ error: new Error('model exploded') }])).stream('hi'); + const events = await collect(run); + + expect(typesOf(events)).toEqual(['run.start', 'step.start', 'error', 'step.done', 'run.done']); + expect(only(events, 'error')[0].error.message).toContain('model exploded'); + expect(only(events, 'step.done')[0].finishReason).toBe('error'); + expect(only(events, 'run.done')[0]).toStrictEqual(expect.objectContaining({ finishReason: 'error', text: '' })); + expect(only(events, 'run.done')[0]).not.toHaveProperty('usage'); + await expect(run.result).rejects.toThrow(/model exploded/); + }); + + it('emits an error event for a failure outside any step (and never an unhandled rejection)', async () => { + const checkpointStore: CheckpointStore = { + load: async () => { + throw new Error('checkpoint store offline'); + }, + save: async () => undefined, + delete: async () => undefined, + }; + const run = AgentExecutor.stream({ ...executorOptions(mockModel([]), []), sessionId: 's1', checkpointStore }); + const events = await collect(run); + + expect(typesOf(events)).toEqual(['run.start', 'error', 'run.done']); + expect(only(events, 'error')[0].error).toEqual({ name: 'Error', message: 'checkpoint store offline' }); + }); + + it('streams model text that comes with tool calls before the tools run', async () => { + const model = mockModel([{ text: 'Checking.', toolCalls: [{ name: 'get_weather', args: { city: 'Rome' } }] }, 'Warm.']); + const events = await collect(agentWith(model, [weather]).stream('Rome?')); + expect(typesOf(events).slice(0, 5)).toEqual(['run.start', 'step.start', 'text.delta', 'text.done', 'tool.start']); + }); + + it('can be iterated only once', async () => { + const run = agentWith(mockModel(['x'])).stream('hi'); + await collect(run); + await expect(collect(run)).rejects.toThrow(/only be iterated once/); + }); + + it('throws synchronously on invalid options, naming stream()', () => { + const { agent } = executorOptions(mockModel([]), []); + expect(() => AgentExecutor.stream({ agent, input: 'x' } as never)).toThrow(/AgentExecutor.stream: 'provider' is required/); + }); + + it('family type guards narrow by prefix', async () => { + const events = await collect(agentWith(mockModel(['a b'])).stream('hi')); + expect(events.filter(isTextEvent).map((e) => e.type)).toEqual(['text.delta', 'text.delta', 'text.done']); + expect(events.filter(isStepEvent).map((e) => e.step)).toEqual([1, 1]); + expect(isAgentEvent({ type: 'text.delta', v: 2, seq: 0 })).toBe(false); + expect(isAgentEvent('text.delta')).toBe(false); + }); +}); diff --git a/src/execution/generateStep.ts b/src/execution/generateStep.ts index e17266e60..27966aee0 100644 --- a/src/execution/generateStep.ts +++ b/src/execution/generateStep.ts @@ -18,6 +18,7 @@ import { isModelActionableProviderErrorCategory, } from './errors'; import type { ExecuteOptions } from './AgentExecutor'; +import { runEventsOf } from './agentRun'; /** * Build tools from agent and registry @@ -124,7 +125,12 @@ export function generateInSpan( init.attributes, async (llmSpan) => { const llmStart = Date.now(); - const generated = await provider.generate(generateRequest); + // LOU-V2: a streaming run obtains the step through its sink (streamed + // when the provider can); everything around it is the same. + const runEvents = runEventsOf(options); + const generated = runEvents + ? await runEvents.generate(provider, generateRequest) + : await provider.generate(generateRequest); const llmLatencyMs = Date.now() - llmStart; recordLlmResult(llmSpan, generated, captureContent); diff --git a/src/execution/index.ts b/src/execution/index.ts index d1332aef1..c4df09bf8 100644 --- a/src/execution/index.ts +++ b/src/execution/index.ts @@ -20,3 +20,30 @@ export * from './hooks'; export { ToolArgumentsValidationError } from './toolArgsValidation'; export type { ToolArgumentIssue } from './toolArgsValidation'; export type { ToolConcurrency } from './toolBatch'; +export { + AGENT_EVENT_SCHEMA_VERSION, + isAgentEvent, + isToolEvent, + isTextEvent, + isStepEvent, +} from './agentEvents'; +export type { + AgentEvent, + AgentEventType, + AgentEventOf, + AgentEventBase, + AgentEventUsage, + AgentEventError, + RunStartEvent, + StepStartEvent, + TextDeltaEvent, + TextDoneEvent, + ToolStartEvent, + ToolDoneEvent, + ToolErrorEvent, + ApprovalRequestedEvent, + StepDoneEvent, + AgentErrorEvent, + RunDoneEvent, +} from './agentEvents'; +export type { AgentRun } from './agentRun'; diff --git a/src/execution/streamStep.ts b/src/execution/streamStep.ts new file mode 100644 index 000000000..27b768378 --- /dev/null +++ b/src/execution/streamStep.ts @@ -0,0 +1,92 @@ +/** + * LOU-V2: obtains one model step through `provider.stream()` instead of + * `provider.generate()`, reporting text chunks as they arrive and assembling + * the same GenerateResult the rest of the AgentExecutor loop works with. + */ + +import type { GenerateOptions, GenerateResult, LLMProvider, StreamChunk, StreamResult, ToolCall } from '../providers'; + +const FINISH_REASONS: ReadonlySet = new Set([ + 'stop', + 'length', + 'tool_calls', + 'content_filter', + 'error', +]); + +/** + * A stream's finish reason as a GenerateResult one. Accepts both our + * spelling and the 'ai' SDK's (`'tool-calls'`); anything else is `'error'`, + * as in `generate()`. + */ +function toFinishReason(reason: string | undefined): GenerateResult['finishReason'] { + const normalized = (reason ?? 'stop').replace(/-/g, '_'); + return FINISH_REASONS.has(normalized) ? (normalized as GenerateResult['finishReason']) : 'error'; +} + +/** Whether to stream this request: the provider has `stream()` and supports streaming its model. */ +export function canStream(provider: LLMProvider, request: GenerateOptions): boolean { + const candidate = provider as Partial; + if (typeof candidate.stream !== 'function') return false; + return typeof candidate.supportsStreaming !== 'function' || candidate.supportsStreaming.call(provider, request.model ?? ''); +} + +interface StreamedParts { + text: string; + toolCalls: ToolCall[]; + finish?: StreamChunk; +} + +/** Folds one chunk into the parts collected so far; throws on an `error` chunk. */ +function applyChunk(parts: StreamedParts, chunk: StreamChunk, onTextDelta: (text: string) => void): void { + if (chunk.type === 'text-delta' && chunk.textDelta) { + parts.text += chunk.textDelta; + onTextDelta(chunk.textDelta); + } else if (chunk.type === 'tool-call' && chunk.toolCall) { + parts.toolCalls.push(chunk.toolCall); + } else if (chunk.type === 'finish') { + parts.finish = chunk; + } else if (chunk.type === 'error') { + throw chunk.error ?? new Error('The model stream reported an error without details'); + } +} + +/** + * The stream's final-value promises we may not read must never surface as + * unhandled rejections when the stream itself fails. + */ +function silenceFinalValues(streamed: StreamResult): void { + for (const value of [streamed.text, streamed.usage, streamed.finishReason, streamed.toolCalls]) { + Promise.resolve(value).catch(() => undefined); + } +} + +/** + * Runs one model step through `provider.stream()`. Text comes from the + * `text-delta` chunks; tool calls from `tool-call` chunks, else from the + * stream's `toolCalls` promise (the 'ai' SDK adapters only report them + * there); finish reason and usage from the `finish` chunk, else from the + * stream's promises. The signal is checked between chunks. + */ +export async function generateViaStream( + provider: LLMProvider, + request: GenerateOptions, + onTextDelta: (text: string) => void +): Promise { + const streamed = await provider.stream(request); + silenceFinalValues(streamed); + const parts: StreamedParts = { text: '', toolCalls: [] }; + for await (const chunk of streamed.fullStream) { + request.signal?.throwIfAborted(); + applyChunk(parts, chunk, onTextDelta); + } + request.signal?.throwIfAborted(); + + const toolCalls = parts.toolCalls.length > 0 ? parts.toolCalls : ((await streamed.toolCalls) ?? []); + return { + text: parts.text, + finishReason: toFinishReason(parts.finish?.finishReason ?? (await streamed.finishReason)), + usage: parts.finish?.usage ?? (await streamed.usage), + ...(toolCalls.length > 0 ? { toolCalls } : {}), + }; +} From 7e0d502ba965f04516c05deffb4d8d81b6e0c2b1 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 10:23:33 +0300 Subject: [PATCH 2/2] LOU-V2: split type test for fallow; extra stream error coverage; regenerate llms Co-Authored-By: Claude Opus 5.5 --- llms-full.txt | 304 +++++++++++++++++++++++++++- llms.txt | 1 + src/execution/agentEvents.test-d.ts | 86 ++++---- src/execution/agentStream.test.ts | 19 ++ 4 files changed, 371 insertions(+), 39 deletions(-) diff --git a/llms-full.txt b/llms-full.txt index cc3654646..7db972d50 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -20,6 +20,7 @@ dashboard — any provider, any deploy target, no lock-in. - **Human-in-the-loop** — flag a tool `needsApproval` and pause execution until a human approves or rejects it, then `resumeAfterApproval()` from any process - **Durable execution** — pass a `sessionId` + `checkpointStore` and a crash mid-conversation resumes instead of restarting - **Cancellation** — pass an `AbortSignal` (`agent.send(input, { signal })`) to stop a run; it resolves with `finishReason: 'aborted'` and the transcript so far, and the signal reaches the provider, tools and delegated agents +- **Streaming events** — `for await (const event of agent.stream(input))` yields a typed, versioned, JSON-serializable event stream (`text.delta` tokens as they arrive, `tool.start`/`tool.done`, steps, approvals, a final `run.done`) ready to forward over SSE or WebSockets; `await run.result` gives the same result as `send()` ([docs/streaming.md](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md)) - **Parallel tool calls** — when the model asks for several tools in one turn they run concurrently (cap it with `toolConcurrency`, or `1` for sequential), and results still reach the transcript in the model's call order - **Skills** — `defineSkill()` / `loadSkills('./skills')`: only each skill's name and description sit in the system prompt; the model loads the full markdown on demand through an auto-registered `load_skill` tool - **Multi-agent delegation** — wrap a child agent as a tool with `createDelegateTool()`, with a `maxDepth` guard against delegation loops @@ -51,6 +52,17 @@ console.log(text); `OPENAI_API_KEY`, `ANTHROPIC_API_KEY`, `OPENROUTER_API_KEY`, `OLLAMA_BASE_URL` that is present. +Stream the reply token by token instead (see [docs/streaming.md](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md) for every event): + +```typescript +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +for await (const event of agent.stream('Hello!')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +``` + When you need a custom provider (your own `LLMProvider`, a mock, extra config), pass the instance instead: @@ -211,6 +223,7 @@ console.log(result.usage.totalTokens, result.finishReason, result.steps); - [Configuration](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md) - agent spec fields, provider env vars, `AgentExecutor.execute()` options, CLI flags - [Deployment](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md) - `loushy build` targets: Node server, Docker, Cloudflare Workers - [API Overview](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/api-overview.md) - the main exports; `npm run docs:build` generates the full TypeDoc reference +- [Streaming](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md) - `agent.stream()`: the typed event schema, terminal and SSE examples - [Workspace tools](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/workspace-tools.md) - file system and shell tools for coding agents, and their security model - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in - [Testing](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/testing.md) - unit-test agents deterministically with the scripted `mockModel` @@ -575,7 +588,7 @@ npm run pipeline:demo:trigger # POSTs a synthetic error to kick it off ### Core -- **`createAgent()`** - zero-config `{ send }` agent +- **`createAgent()`** - zero-config `{ send, stream }` agent - **`AgentBuilder`** - fluent `AgentConfig` builder - **`AgentExecutor`** - static executor (`execute()`, approvals, checkpoints, tracing) - **`ToolRegistry`** - manage available tools @@ -2470,6 +2483,295 @@ Use [`mockModel`](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/testing bodies until it loads one: the first request's system prompt lists the skills, and the request after a `load_skill` call contains the body in the tool result. +# Streaming + +Source: docs/streaming.md + +`agent.stream()` runs an agent exactly like `agent.send()` and reports the run +as a stream of typed events: model text as it arrives, every tool call, +step boundaries, approval requests, and a final `run.done`. Every event is a +plain JSON object, so you can forward it to a browser over Server-Sent Events +or a WebSocket without converting it. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +for await (const event of agent.stream('Weather in Paris?')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +``` + +The full API has the same method: `AgentExecutor.stream(options)` takes the +same options as `AgentExecutor.execute()` (approval store, checkpoints, hooks, +tracing, `toolConcurrency`, ...). + +## The `AgentRun` handle + +`stream()` returns an `AgentRun`: + +```ts no-verify +interface AgentRun extends AsyncIterable { + readonly runId: string; // the runId on every event + readonly result: Promise; // what send() / execute() would return +} +``` + +- **The run starts immediately.** You do not have to iterate: `await run.result` + on its own drives the run to completion. +- **`result`** resolves with the same `ExecutionResult` that `send()` returns, + and rejects with the same error `send()` would reject with. An aborted run + resolves with `finishReason: 'aborted'`; a run paused for approval resolves + with `finishReason: 'awaiting-approval'` and `approvalId`. Leaving `result` + unawaited never causes an unhandled rejection. +- **Breaking out of the `for await` loop early aborts the run** (through the + same path as `signal`, see [Cancellation](#cancellation)). `result` then + resolves with `finishReason: 'aborted'`. +- **No backpressure; nothing is dropped.** The run never waits for the + consumer. Events are buffered until you read them, so a slow consumer sees + every event, in order. Iterating after the run finished still yields all of + its events. +- **One consumer.** An `AgentRun` can be iterated once; a second `for await` + throws. To fan out, collect the events yourself. +- Invalid options (no `provider`, `agent` or `input`, a bad `toolConcurrency`) + throw synchronously from `stream()`. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +const run = agent.stream('Summarize the report', { signal: AbortSignal.timeout(30_000) }); +const result = await run.result; // no iteration needed +console.log(result.finishReason, result.text); +``` + +## Event schema (version 1) + +Every event has these fields: + +| Field | Type | Meaning | +| ----------- | -------- | ------- | +| `type` | string | The event type (table below). Narrow on it in TypeScript. | +| `runId` | string | Identifies the run. The same on every event of one `stream()` call. | +| `seq` | number | `0` for the first event, then `+1` per event, with no gaps. | +| `timestamp` | string | When the event was emitted, ISO 8601 (`2026-10-01T09:30:00.000Z`). | +| `v` | `1` | Schema version, exported as `AGENT_EVENT_SCHEMA_VERSION`. | + +The event types and their extra fields: + +| `type` | Extra fields | Emitted | +| -------------------- | ------------ | ------- | +| `run.start` | `agentName: string`, `agentId?: string` | First event of every run. | +| `step.start` | `step: number` | A model step begins: one model call plus the tool calls it asks for. `step` counts from 1 (a run resumed from a checkpoint continues the count). | +| `text.delta` | `text: string` | A chunk of model text, as it arrives. | +| `text.done` | `text: string` | The step's complete text: the concatenation of its `text.delta` events. Only for steps with text. | +| `tool.start` | `toolCallId: string`, `toolName: string`, `args: Record` | A tool call starts. `args` are the model's arguments parsed from JSON (`{}` when they are not valid JSON). | +| `tool.done` | `toolCallId`, `toolName`, `result: unknown`, `durationMs: number` | A tool call returned. `result` is the value as it would be JSON-encoded (`undefined` becomes `null`, a `Date` becomes a string). `durationMs` counts from its `tool.start`. | +| `tool.error` | `toolCallId`, `toolName`, `error: { name: string, message: string }`, `durationMs: number` | A tool call failed: it threw, its arguments did not match its schema (`name: 'ToolArgumentsValidationError'`), or the tool does not exist. The model gets the error as the call's result and the run continues. | +| `approval.requested` | `approvalId: string`, `toolCallId`, `toolName`, `args: Record` | A tool call needs a human decision. The run then stops; resume it with `resumeAfterApproval()` (see the README). | +| `step.done` | `step: number`, `finishReason: string`, `usage?: { promptTokens, completionTokens, totalTokens }` | A step ends. `finishReason` is the model's (`'stop'`, `'tool_calls'`, `'length'`, ...), or `'awaiting-approval'`, `'aborted'` or `'error'` when the step ended that way. `usage` is this step's model call, absent when the call produced no response. | +| `error` | `error: { name: string, message: string }` | An error. If it ends the run, `run.done` with `finishReason: 'error'` follows. A provider error retried under `surfaceRetryableProviderErrors` is followed by further steps instead. | +| `run.done` | `finishReason: string`, `text: string`, `usage?: { promptTokens, completionTokens, totalTokens }` | Last event of every run, exactly once, including aborted, failed and awaiting-approval runs. `finishReason` and `text` match `run.result`; a failed run has `finishReason: 'error'`, `text: ''` and no `usage`. | + +Optional fields are left out when they have no value. They are never +`undefined`, so `JSON.parse(JSON.stringify(event))` returns an equal object. + +### Ordering guarantees + +- `run.start` is first and `run.done` is last, each exactly once. +- Each `step.start` is followed by exactly one `step.done` with the same + `step`, before the next `step.start`. Everything a step does happens between + the two. +- Inside a step: `text.delta` events, then `text.done`, then the tool events. +- Tool calls of one step run in parallel (see `toolConcurrency`): + `tool.start` events come in the model's call order and `tool.done` / + `tool.error` events in completion order. Match them by `toolCallId`. +- When a call needs approval, the calls before it run and report, then + `approval.requested`, `step.done` (`'awaiting-approval'`) and `run.done` + (`'awaiting-approval'`) follow. Calls after it never start. + +A text-only run: + +``` +run.start → step.start → text.delta × n → text.done → step.done → run.done +``` + +A run with one tool call: + +``` +run.start +step.start(1) → tool.start → tool.done → step.done(1, 'tool_calls') +step.start(2) → text.delta × n → text.done → step.done(2, 'stop') +run.done('stop') +``` + +### TypeScript + +`AgentEvent` is a discriminated union: narrowing on `event.type` gives the +payload. Each event type is exported too (`TextDeltaEvent`, `ToolDoneEvent`, +`RunDoneEvent`, ...), and `AgentEventOf<'tool.done'>` picks one by name. + +```ts +import { isAgentEvent, isToolEvent, type AgentEvent } from '@loushy/build-ai-agent'; + +function render(event: AgentEvent): string { + switch (event.type) { + case 'text.delta': + return event.text; + case 'tool.start': + return `\n[${event.toolName}] ${JSON.stringify(event.args)}\n`; + case 'tool.error': + return `\n[${event.toolName} failed: ${event.error.message}]\n`; + case 'run.done': + return `\n(${event.finishReason})\n`; + default: + return ''; + } +} + +// isToolEvent / isTextEvent / isStepEvent narrow to an event family: +const toolNames = (events: AgentEvent[]) => events.filter(isToolEvent).map((e) => e.toolName); + +// isAgentEvent validates an event received over the wire: +const received: unknown = JSON.parse('{"type":"run.start","runId":"r","seq":0,"timestamp":"","v":1,"agentName":"a"}'); +if (isAgentEvent(received)) console.log(render(received), toolNames([received])); +``` + +### Versioning + +`v` changes only when an existing event changes incompatibly (a field +removed, renamed or retyped). New event types and new optional fields can be +added without changing `v`, so ignore event types you do not know. + +## Token streaming and providers + +When the provider implements `stream()` (all built-in providers and +`mockModel` do), each model step is streamed and `text.delta` events arrive +as the model produces text. Tool calls are assembled from the stream. When a +provider has no `stream()`, or its `supportsStreaming(model)` returns +`false`, the step falls back to `generate()` and its text arrives as a single +`text.delta` followed by `text.done`. + +Streaming changes only how one model step is obtained. Hooks, argument +validation, approvals, parallel tool calls, checkpoints, cancellation, +tracing spans and the other `execute()` callbacks behave exactly as in +`send()`. `send()` and `execute()` themselves still use `generate()`. + +A custom provider's `stream()` should yield `text-delta` chunks and a final +`finish` chunk with `finishReason` and `usage`. Tool calls can be yielded as +`tool-call` chunks, or resolved on the `toolCalls` promise of the +`StreamResult`. An `error` chunk fails the step. + +## Cancellation + +Pass `signal` to stop a run from outside, or break out of the loop. Both end +the run the same way as an aborted `send()`: the signal is checked between +stream chunks, before every model call and every tool call, and reaches the +provider and the tools. The stream ends with `step.done` (`'aborted'`, when a +step was running) and `run.done` (`'aborted'`), and `result` resolves with +`finishReason: 'aborted'`. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +const run = agent.stream('Write a long story'); + +let chars = 0; +for await (const event of run) { + if (event.type === 'text.delta') chars += event.text.length; + if (chars > 500) break; // aborts the run +} +console.log((await run.result).finishReason); // 'aborted' +``` + +## Example: terminal + +```ts +import { createAgent, defineTool } from '@loushy/build-ai-agent'; +import { z } from 'zod'; + +const getWeather = defineTool({ + name: 'get_weather', + description: 'Current weather for a city', + input: z.object({ city: z.string() }), + execute: async ({ city }) => ({ city, tempC: 21 }), +}); + +const agent = createAgent({ model: 'openai/gpt-4o-mini', tools: [getWeather] }); + +for await (const event of agent.stream('Weather in Paris?')) { + switch (event.type) { + case 'text.delta': + process.stdout.write(event.text); + break; + case 'tool.start': + console.log(`\n→ ${event.toolName}(${JSON.stringify(event.args)})`); + break; + case 'tool.done': + console.log(`← ${event.toolName} in ${event.durationMs}ms`); + break; + case 'run.done': + console.log(`\n[${event.finishReason}, ${event.usage?.totalTokens ?? 0} tokens]`); + break; + } +} +``` + +## Example: Server-Sent Events + +Events are JSON-serializable, so an SSE endpoint is one `res.write` per +event. Abort the run when the client disconnects. + +```ts +import { createServer } from 'node:http'; +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); + +createServer(async (req, res) => { + const question = new URL(req.url ?? '/', 'http://localhost').searchParams.get('q') ?? 'Hello!'; + const controller = new AbortController(); + res.on('close', () => controller.abort()); + + res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache' }); + for await (const event of agent.stream(question, { signal: controller.signal })) { + res.write(`data: ${JSON.stringify(event)}\n\n`); + } + res.end(); +}).listen(3000); +``` + +In the browser: + +```ts no-verify +const source = new EventSource('/chat?q=Weather%20in%20Paris%3F'); +source.onmessage = (message) => { + const event = JSON.parse(message.data); + if (event.type === 'text.delta') output.textContent += event.text; + if (event.type === 'run.done') source.close(); +}; +``` + +## Example: the full API + +`AgentExecutor.stream()` accepts every `execute()` option. Here a tool that +needs approval ends the stream with `approval.requested`: + +```ts +import { AgentExecutor } from '@loushy/build-ai-agent'; + +const run = AgentExecutor.stream({ agent, input, provider, toolRegistry, approvalStore }); +for await (const event of run) { + if (event.type === 'approval.requested') { + console.log(`Approve ${event.toolName}(${JSON.stringify(event.args)})? id=${event.approvalId}`); + } +} +const paused = await run.result; // paused.finishReason === 'awaiting-approval' +``` + # Testing agents Source: docs/testing.md diff --git a/llms.txt b/llms.txt index 0993d06be..8e95d4cfa 100644 --- a/llms.txt +++ b/llms.txt @@ -11,6 +11,7 @@ - [Configuration](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md): A declarative agent is a YAML (`.yaml`/`.yml`) or JSON (`.json`) file, validated with zod by `loadSpec()` (`src/spec/schema.ts`). - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md): `AgentExecutor.execute()` and `FlowExecutor.execute()` emit spans through a `TraceExporter`. - [Skills](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/skills.md): A skill is a named chunk of instructions (a markdown document) that the model loads only when it needs it. +- [Streaming](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md): `agent.stream()` runs an agent exactly like `agent.send()` and reports the run as a stream of typed events: model text as it arrives, every tool call, step boundaries, approval requests, and a final `run.done`. - [Testing agents](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/testing.md): `@loushy/build-ai-agent/testing` ships `mockModel`: a scripted, deterministic `LLMProvider` for unit tests. - [Workspace tools](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/workspace-tools.md): Workspace tools give an agent a file system and a shell, so you can build coding agents in the style of Claude Code. diff --git a/src/execution/agentEvents.test-d.ts b/src/execution/agentEvents.test-d.ts index 94e2a0dd9..6d4f63e35 100644 --- a/src/execution/agentEvents.test-d.ts +++ b/src/execution/agentEvents.test-d.ts @@ -9,6 +9,7 @@ import { type AgentEvent, type AgentEventError, type AgentEventOf, + type AgentEventType, type AgentEventUsage, } from './agentEvents'; @@ -22,45 +23,54 @@ const event = { } as AgentEvent; describe('AgentEvent types', () => { - it('narrows the payload on event.type', () => { - switch (event.type) { - case 'text.delta': - case 'text.done': - expectTypeOf(event.text).toBeString(); - break; - case 'step.start': - expectTypeOf(event.step).toBeNumber(); - // @ts-expect-error - step.start has no finishReason - void event.finishReason; - break; - case 'tool.start': - expectTypeOf(event.args).toEqualTypeOf>(); - expectTypeOf(event.toolCallId).toBeString(); - break; - case 'tool.done': - expectTypeOf(event.result).toBeUnknown(); - expectTypeOf(event.durationMs).toBeNumber(); - break; - case 'tool.error': - case 'error': - expectTypeOf(event.error).toEqualTypeOf(); - break; - case 'approval.requested': - expectTypeOf(event.approvalId).toBeString(); - break; - case 'step.done': - expectTypeOf(event.usage).toEqualTypeOf(); - break; - case 'run.done': - expectTypeOf(event.text).toBeString(); - expectTypeOf(event.finishReason).toEqualTypeOf(); - break; - case 'run.start': - expectTypeOf(event.agentName).toBeString(); - break; - default: - expectTypeOf(event).toBeNever(); + it('narrows text events on event.type', () => { + if (event.type === 'text.delta') expectTypeOf(event.text).toBeString(); + if (event.type === 'text.done') expectTypeOf(event.text).toBeString(); + }); + + it('narrows step events on event.type', () => { + if (event.type === 'step.start') { + expectTypeOf(event.step).toBeNumber(); + // @ts-expect-error - step.start has no finishReason + void event.finishReason; } + if (event.type === 'step.done') expectTypeOf(event.usage).toEqualTypeOf(); + }); + + it('narrows tool events on event.type', () => { + if (event.type === 'tool.start') expectTypeOf(event.args).toEqualTypeOf>(); + if (event.type === 'tool.done') expectTypeOf(event.durationMs).toBeNumber(); + if (event.type === 'tool.error') expectTypeOf(event.error).toEqualTypeOf(); + }); + + it('narrows tool results and approvals on event.type', () => { + if (event.type === 'tool.done') expectTypeOf(event.result).toBeUnknown(); + if (event.type === 'approval.requested') expectTypeOf(event.approvalId).toBeString(); + }); + + it('narrows run and error events on event.type', () => { + if (event.type === 'run.start') expectTypeOf(event.agentName).toBeString(); + if (event.type === 'error') expectTypeOf(event.error).toEqualTypeOf(); + if (event.type === 'run.done') { + expectTypeOf(event.text).toBeString(); + expectTypeOf(event.finishReason).toEqualTypeOf(); + } + }); + + it('covers every event type', () => { + expectTypeOf().toEqualTypeOf< + | 'run.start' + | 'step.start' + | 'text.delta' + | 'text.done' + | 'tool.start' + | 'tool.done' + | 'tool.error' + | 'approval.requested' + | 'step.done' + | 'error' + | 'run.done' + >(); }); it('carries the common fields on every event', () => { diff --git a/src/execution/agentStream.test.ts b/src/execution/agentStream.test.ts index c05f767ca..b9d5d467b 100644 --- a/src/execution/agentStream.test.ts +++ b/src/execution/agentStream.test.ts @@ -357,6 +357,25 @@ describe('agent.stream()', () => { expect(typesOf(events).slice(0, 5)).toEqual(['run.start', 'step.start', 'text.delta', 'text.done', 'tool.start']); }); + it('fails the step on an error chunk from the provider stream', async () => { + const model = mockModel(['never shown']); + model.stream = async () => ({ + fullStream: (async function* (): AsyncGenerator { + yield { type: 'error', error: new Error('stream broke') }; + })(), + textStream: (async function* () {})(), + text: Promise.resolve(''), + usage: Promise.reject(new Error('stream broke')), + finishReason: Promise.reject(new Error('stream broke')), + toolCalls: Promise.reject(new Error('stream broke')), + }); + const run = agentWith(model).stream('hi'); + const events = await collect(run); + + expect(only(events, 'error')[0].error.message).toContain('stream broke'); + await expect(run.result).rejects.toThrow(/stream broke/); + }); + it('can be iterated only once', async () => { const run = agentWith(mockModel(['x'])).stream('hi'); await collect(run);