diff --git a/CHANGELOG.md b/CHANGELOG.md index cbb3c7f0..15bbfc96 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `http_request` is no longer open to DNS rebinding (N13a). It resolved a host name once to check it against the private-address list and then let `fetch` resolve it again to connect, so a name that answered a public address to the check and `127.0.0.1` (or a cloud metadata or intranet address) to the connection reached the private address; the docs and a comment in the Worker runtime described the check as rebinding-safe. Now the one resolution is the check: the in-process path connects through an undici `Agent` whose `connect.lookup` checks every address and hands the socket the address it checked (each redirect hop included), and the sandboxed path (`sandboxExecute`, which agents use) resolves and checks the host in the agent's process and has the sandboxed process connect to that address (node:http/https with a fixed lookup; the host name still goes in `Host` and TLS SNI). The private-address list is now the shared one (below), which adds `0.0.0.0/8`, `100.64.0.0/10`, `192.0.0.0/24`, `198.18.0.0/15`, multicast and reserved `224.0.0.0/3`, `::`, `ff00::/8`, NAT64 `64:ff9b::/96` and 6to4 `2002::/16` to what `http_request` refused. Behavior changes: a host name that cannot be resolved now fails with the resolver's error instead of being passed to `fetch`; `undici` is loaded on the first `http_request` request (not only with `validateSSL: false`). New option `createHttpTool({ allowPrivate })` for hosts that may resolve to private addresses. Cloudflare Workers have no DNS hook, so neither tool exists in the Worker build; docs/deployment.md now says what a Worker can and cannot check. ### Changed +- BREAKING: `send()` and `AgentExecutor.execute()` with a listener (`createAgent({ onEvent })`, `onAgentEvent`, or the deprecated `onEvent` option) now stream model calls (M9, #232): a step's text arrives as several `text.delta` events, as in `stream()`, instead of one `text.delta` per step after the step finished. What a listener receives otherwise does not change: the same event types in the same order, exactly one `text.done` per step with the step's whole text, the same `step.done` usage and `run.done`, the same `ExecutionResult` (text, tool calls, usage, steps). A provider without `stream()`, or whose `supportsStreaming(model)` is `false`, still gives one `text.delta` per step. Output guardrails on `send()` still check only the final reply (a `stream()` you iterate checks every step's text, as before). `send()` without a listener still calls `provider.generate()`. Hooks' `ctx.emit` is set whenever the run has listeners (it was already; the docs said "only when streamed"). Migration: concatenate `text.delta`, or listen to `text.done` for whole steps; `execute({ streamModelCalls: false })` (new option, default `true`) restores whole steps. +- `MockLLMProvider` / `createMockProvider()`: `stream()` now streams the same step `generate()` returns (M9): the same tool calls (it streamed none, so a streamed mock run never called a tool), finish reason and usage, and text chunks that add up to exactly the generated text (the old chunks ended with an extra space: `'This is a mock response. '`). Runs on the mock provider with a listener (Agent Forge's, for one) therefore give the same results as before, and `stream()` runs on it now call tools as `send()` does. +- Agent Forge (M9): runs now stream their model calls, so the run WebSocket (`WS /agents/:id/stream`) carries several `text.delta` events per step; the chat, logs and trace views are unchanged (they use `text.done` and the reconciled messages). Stop is checked between streamed chunks and after the last one, as it was after each `generate()` call. - CI peer matrix (LOU-M8, #231): the `typecheck-ai7` and `typecheck-zod4` jobs are replaced by one `peers` job with five entries, each installed for real on top of the default install and run through `tsc`, `test:types`, both builds and `npx vitest run`: `ai4-zod4`, `ai6-zod3`, `ai6-zod4`, `ai7-zod3`, `ai7-zod4`. The two zod 4 entries on `ai` 6/7 also install `ollama-ai-provider-v2` (3.x / 4.x), and the new `src/providers/ollamaV2.contract.test.ts` runs `OllamaProvider` against the real package and a local fake Ollama server (`generate()`, `stream()`, a tool-call turn through `createAgent().send()` and `.stream()`). The `ai-v6` dev alias replaces the hand-made `ai` 6 stand-in in `aiMajorPeers.test.ts`. - `lousho init --provider ollama` now scaffolds `ai@^7.0.0` with `ollama-ai-provider-v2@^4.0.0` and `zod@^4.0.0` (it was `ai@^4.3.19` with `ollama-ai-provider@^1.2.0` and zod 3). Existing projects are not touched; to stay on the old pairing keep `ai@^4.3.19` and `ollama-ai-provider@^1.2.0`. - A run paused inside a sub-agent compares the sub-agent with its current definition on resume, under the lead's `onAgentDrift` (M10c). Behaviour change: with `onAgentDrift: 'error'`, `agent.approvals.resolve()` (and `resumeAfterApproval()`) now also rejects with `LOUSHO_AGENT_DRIFT` when a paused sub-agent's instructions, model or tools changed; before, only tool and provider default-model changes of a sub-agent were seen, and always as a warning. This holds at any depth (a sub-agent of a sub-agent uses the lead's mode). With `'warn'` (still the default) the `agent.drift` event carries `subagent`. A rejected resume puts the lead's approval record and its `'awaiting-approval'` checkpoint back, so fixing the sub-agent and resolving again finishes the run; before, a sub-agent's missing-tool error became an error result of the `task` call and the run continued. Migration: none for `'warn'` / `'ignore'`; with `'error'`, resolve approvals paused before a sub-agent changed with the old definition, or use `'warn'`. See docs/durable-execution.md#resuming-with-a-changed-agent. diff --git a/apps/agent-forge/server/__tests__/runRegistry.test.ts b/apps/agent-forge/server/__tests__/runRegistry.test.ts index 56faaa50..aa496962 100644 --- a/apps/agent-forge/server/__tests__/runRegistry.test.ts +++ b/apps/agent-forge/server/__tests__/runRegistry.test.ts @@ -2,8 +2,12 @@ import { describe, it, expect, beforeEach, afterEach } from 'vitest'; import * as fs from 'node:fs'; import * as path from 'node:path'; import * as os from 'node:os'; +import * as http from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { WebSocket } from 'ws'; import type { AgentSpec } from '@lousho/build-ai-agent'; import { RunManager } from '../runRegistry'; +import { attachWebSocketServer } from '../wsServer'; import { FileCheckpointStore } from '../checkpointStore'; import { FileApprovalStore } from '../approvalStore'; import { graphToSpec } from '../../src/graph/graphToSpec'; @@ -120,6 +124,49 @@ describe('RunManager', () => { expect(afterResume?.messages.map((m) => m.content)).toContain('a follow-up appended on resume'); }); + it('streams each model step over the WebSocket: several text.delta, one text.done per step (M9)', async () => { + const server = http.createServer(); + const wss = attachWebSocketServer(server, runManager); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const { port } = server.address() as AddressInfo; + const ws = new WebSocket(`ws://127.0.0.1:${port}/agents/agent-ws/stream`); + const events: any[] = []; + try { + ws.on('message', (data) => { + const message = JSON.parse(String(data)); + if (message.type === 'event') events.push(message.payload); + }); + await new Promise((resolve, reject) => { + ws.once('open', resolve); + ws.once('error', reject); + }); + + // A tool call, then the final reply: two model steps, each streamed by the mock provider. + await runManager.run('agent-ws', 'please use current-date', SPEC); + const final = await waitForStatus(runManager, 'agent-ws', (s) => s.status === 'stopped'); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(final.resultText).toBe('This is a mock response.'); + const steps: { deltas: string[]; done: string[] }[] = []; + for (const event of events) { + if (event.type === 'step.start') steps.push({ deltas: [], done: [] }); + if (event.type === 'text.delta') steps.at(-1)!.deltas.push(event.text); + if (event.type === 'text.done') steps.at(-1)!.done.push(event.text); + } + expect(steps).toHaveLength(2); + for (const step of steps) { + expect(step.deltas.length).toBeGreaterThan(1); + expect(step.done).toEqual(['This is a mock response.']); + expect(step.deltas.join('')).toBe(step.done[0]); + } + expect(events.filter((e) => e.type === 'tool.done').map((e) => e.toolName)).toEqual(['current-date']); + } finally { + ws.close(); + wss.close(); + await new Promise((resolve) => server.close(resolve)); + } + }); + it('emits structured log entries derived from the AgentEvent stream (O1)', async () => { const logs: any[] = []; runManager.on('log', (agentId: string, entry: any) => { diff --git a/apps/agent-forge/server/abortableProvider.test.ts b/apps/agent-forge/server/abortableProvider.test.ts new file mode 100644 index 00000000..d7aac573 --- /dev/null +++ b/apps/agent-forge/server/abortableProvider.test.ts @@ -0,0 +1,26 @@ +import { describe, expect, it } from 'vitest'; +import { MockLLMProvider } from '@lousho/build-ai-agent'; +import { RunAbortedError, withAbortSignal } from './abortableProvider'; + +const request = { model: 'mock-1', messages: [{ role: 'user' as const, content: 'hi' }] }; + +describe('withAbortSignal (M9: streamed calls)', () => { + it('passes a stream through chunk by chunk when not stopped', async () => { + const provider = withAbortSignal(new MockLLMProvider({ name: 'mock' }), new AbortController().signal); + const streamed = await provider.stream(request); + const deltas: string[] = []; + for await (const chunk of streamed.fullStream) if (chunk.type === 'text-delta') deltas.push(chunk.textDelta ?? ''); + expect(deltas.length).toBeGreaterThan(1); + expect(deltas.join('')).toBe(await streamed.text); + }); + + it('throws RunAbortedError from a stream stopped while it is read', async () => { + const controller = new AbortController(); + const provider = withAbortSignal(new MockLLMProvider({ name: 'mock' }), controller.signal); + const streamed = await provider.stream(request); + const read = async () => { + for await (const _chunk of streamed.fullStream) controller.abort(); + }; + await expect(read()).rejects.toBeInstanceOf(RunAbortedError); + }); +}); diff --git a/apps/agent-forge/server/abortableProvider.ts b/apps/agent-forge/server/abortableProvider.ts index f7160235..69eedc85 100644 --- a/apps/agent-forge/server/abortableProvider.ts +++ b/apps/agent-forge/server/abortableProvider.ts @@ -40,7 +40,7 @@ * cancellation; this wrapper's shape (signal checked, forwarded if the * real provider accepts one) is what that wiring would build on. */ -import type { LLMProvider, GenerateOptions, GenerateResult, StreamResult } from '@lousho/build-ai-agent'; +import type { LLMProvider, GenerateOptions, GenerateResult, StreamChunk, StreamResult } from '@lousho/build-ai-agent'; /** * Named `AbortError` (the fetch/AbortSignal convention) on purpose: the SDK's @@ -69,9 +69,27 @@ export function withAbortSignal(provider: LLMProvider, signal: AbortSignal): LLM checkAborted(); return result; }, + // M9: AgentExecutor.execute({ onAgentEvent }) streams model calls, so a + // stop() during a streamed call is checked between chunks and after the + // last one - the same point generate() checks after its call returns. async stream(options: GenerateOptions): Promise { checkAborted(); - return provider.stream(options); + const streamed = await provider.stream(options); + async function* fullStream(): AsyncGenerator { + for await (const chunk of streamed.fullStream) { + checkAborted(); + yield chunk; + } + checkAborted(); + } + return { + fullStream: fullStream(), + textStream: streamed.textStream, + text: streamed.text, + usage: streamed.usage, + finishReason: streamed.finishReason, + toolCalls: streamed.toolCalls, + }; }, supportsTools(model: string): boolean { return provider.supportsTools(model); diff --git a/apps/agent-forge/src/components/ChatPanel.tsx b/apps/agent-forge/src/components/ChatPanel.tsx index f000b5fa..001a7757 100644 --- a/apps/agent-forge/src/components/ChatPanel.tsx +++ b/apps/agent-forge/src/components/ChatPanel.tsx @@ -6,13 +6,13 @@ * ported from `.design-ref/agent-forge-mockup.html`'s bubble/avatar/ * timestamp layout. * - * Streaming granularity: MESSAGE-level, not token-level. AgentExecutor's - * execution loop (src/execution/AgentExecutor.ts) only ever calls - * `provider.generate()` - never `provider.stream()` - so there is no - * per-token event to render incrementally even for a provider (like - * MockLLMProvider) that DOES implement `.stream()`. A message appears in - * the thread once its run turn settles (completes, or pauses for - * approval); the "typing" indicator below fills the gap, driven by the + * Streaming granularity: MESSAGE-level, not token-level. Since M9 the + * server's `AgentExecutor.execute({ onAgentEvent })` streams each model + * call through `provider.stream()` when the provider can, so the + * WebSocket carries several `text.delta` events per step - but this panel + * renders the reconciled `Message[]` history, not the deltas. A message + * appears in the thread once its run turn settles (completes, or pauses + * for approval); the "typing" indicator below fills the gap, driven by the * real `status: 'running'` from LOU-N rather than a fake timeout. */ import { useEffect, useRef, useState } from 'react'; diff --git a/docs/compaction.md b/docs/compaction.md index d2f427d8..2a4f7478 100644 --- a/docs/compaction.md +++ b/docs/compaction.md @@ -58,8 +58,9 @@ strategy could not shrink anything, `tokensAfter` equals `tokensBefore`; when it failed (or the summarizer failed and the hook fell back to pruning), `error` is set and the run continues. `summary` is `true` when old turns were replaced by a summary (the text itself is not sent; use `onCompaction` for it). A -non-streaming `send()` emits no events. Hooks add their own events with -`ctx.emit?.(...)` on the `preGenerate` context, which exists only in streamed runs. +`send()` without a listener emits no events. Hooks add their own events with +`ctx.emit?.(...)` on the `preGenerate` context, which exists when the run has +listeners (`createAgent({ onEvent })`, `onAgentEvent`) or is streamed. ## Compacting a run diff --git a/docs/executor-api.md b/docs/executor-api.md index 588b1495..86b1c380 100644 --- a/docs/executor-api.md +++ b/docs/executor-api.md @@ -60,6 +60,7 @@ console.log(events); // includes 'run.start' and 'run.done' | `limits` | Token, cost, time and step budgets of the run; see [Budgets](./configuration.md#budgets). | | `temperature`, `maxTokens` | Generation parameters. | | `onAgentEvent` | Listener for the run's `AgentEvent`s (`run.start`, `tool.start`, `run.done`, ...); see [Streaming](./streaming.md#listening-without-iterating). | +| `streamModelCalls` | With a listener, stream each model call so text arrives as several `text.delta` events (default `true`); `false` generates whole steps, one `text.delta` each. Ignored by `stream()`. | | `approvalStore`, `sessionId` | Human-in-the-loop approvals (see `resumeAfterApproval()`). | | `checkpointStore` | Persist/resume execution checkpoints. | | `exporter` | A `TraceExporter` for tracing spans (OpenTelemetry GenAI conventions, see [observability](observability.md)). | diff --git a/docs/guardrails.md b/docs/guardrails.md index 33f610e8..ec576632 100644 --- a/docs/guardrails.md +++ b/docs/guardrails.md @@ -15,7 +15,7 @@ three points, each list in order: | List | Runs on | When | | ---- | ------- | ---- | | `input` | Each new user message (the user messages that end the transcript) | Before the first model call. A block makes no model call. | -| `output` | The final assistant text; in a streamed run, every step's text | Before it is emitted: before its `text.done`, and before `run.done`. | +| `output` | The final assistant text; in a `stream()` run (one you iterate), every step's text. `send()` with a listener checks the final text only. | Before it is emitted: before its `text.done`, and before `run.done`. | | `tools` | A tool call's arguments (`text` is them as JSON, `args` the object) | After the [permission rules](./approvals.md#permission-policies) (skipped when a rule denies the call) and before `needsApproval`. | A guardrail is `{ name, check(ctx) }`. `ctx` has `kind` (`'input'`, `'output'` diff --git a/docs/hooks.md b/docs/hooks.md index f9e5b942..6cc399af 100644 --- a/docs/hooks.md +++ b/docs/hooks.md @@ -53,7 +53,7 @@ Every method is optional: implement only the points you need. A hook that return | `preGenerate(ctx)` | Before each model call. | `GenerateHookContext`: `request` (the live `GenerateOptions` about to be sent) and `emit?`, plus the common fields. | Nothing; mutate `ctx.messages` or `ctx.request`, or throw. | | `postGenerate(ctx, result)` | After each model call resolves, with its `GenerateResult`. | `GenerateHookContext`. | Nothing; or throw. | -The common fields on every context are `agentId`, `agentName`, `sessionId`, `messages` (the live conversation history), `metadata` (a free-form bag) and `subagent` (set inside a sub-agent). `ctx.emit` exists only in streamed runs and adds `compaction.start` / `compaction.done` events to the stream. +The common fields on every context are `agentId`, `agentName`, `sessionId`, `messages` (the live conversation history), `metadata` (a free-form bag) and `subagent` (set inside a sub-agent). `ctx.emit` exists when the run has listeners (`createAgent({ onEvent })`, `onAgentEvent`) or is streamed, and adds `compaction.start` / `compaction.done` events to the run's events. ## Hook outcomes diff --git a/docs/streaming.md b/docs/streaming.md index 01c27211..fff2e800 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -77,10 +77,13 @@ To observe every run without iterating one, give the agent a listener: / `stream()` / `resumeAfterApproval()` options. It is called synchronously with each `AgentEvent` as it happens, on `send()` as on `stream()`, for session turns and for runs resumed after an approval. It gets the same events, -in the same order, as iterating the stream would, with one difference: -`send()` / `execute()` generate each model step whole, so a step's text comes -as a single `text.delta` (as it does on `stream()` with a provider that cannot -stream). Sub-agents' events arrive tagged with `subagent`. +in the same order, as iterating the stream would: with a listener, `send()` / +`execute()` stream each model call too, so a step's text arrives as several +`text.delta` events (one `text.delta` per step when the provider cannot +stream). Concatenate them, or use `text.done` for each step's whole text; +`AgentExecutor.execute({ streamModelCalls: false })` generates whole steps. +Output guardrails still check only the final reply of a `send()`, as without +a listener. Sub-agents' events arrive tagged with `subagent`. ```ts import { createAgent } from '@lousho/build-ai-agent'; @@ -234,7 +237,8 @@ provider has no `stream()`, or its `supportsStreaming(model)` returns 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()`. +`send()`. `send()` and `execute()` stream the same way when they have a +listener (`onEvent` / `onAgentEvent`); without one they 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 diff --git a/llms-full.txt b/llms-full.txt index 2df52b5b..6e05e17d 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -4106,8 +4106,9 @@ strategy could not shrink anything, `tokensAfter` equals `tokensBefore`; when it failed (or the summarizer failed and the hook fell back to pruning), `error` is set and the run continues. `summary` is `true` when old turns were replaced by a summary (the text itself is not sent; use `onCompaction` for it). A -non-streaming `send()` emits no events. Hooks add their own events with -`ctx.emit?.(...)` on the `preGenerate` context, which exists only in streamed runs. +`send()` without a listener emits no events. Hooks add their own events with +`ctx.emit?.(...)` on the `preGenerate` context, which exists when the run has +listeners (`createAgent({ onEvent })`, `onAgentEvent`) or is streamed. ## Compacting a run @@ -6592,6 +6593,7 @@ console.log(events); // includes 'run.start' and 'run.done' | `limits` | Token, cost, time and step budgets of the run; see [Budgets](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md#budgets). | | `temperature`, `maxTokens` | Generation parameters. | | `onAgentEvent` | Listener for the run's `AgentEvent`s (`run.start`, `tool.start`, `run.done`, ...); see [Streaming](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md#listening-without-iterating). | +| `streamModelCalls` | With a listener, stream each model call so text arrives as several `text.delta` events (default `true`); `false` generates whole steps, one `text.delta` each. Ignored by `stream()`. | | `approvalStore`, `sessionId` | Human-in-the-loop approvals (see `resumeAfterApproval()`). | | `checkpointStore` | Persist/resume execution checkpoints. | | `exporter` | A `TraceExporter` for tracing spans (OpenTelemetry GenAI conventions, see [observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md)). | @@ -6727,7 +6729,7 @@ three points, each list in order: | List | Runs on | When | | ---- | ------- | ---- | | `input` | Each new user message (the user messages that end the transcript) | Before the first model call. A block makes no model call. | -| `output` | The final assistant text; in a streamed run, every step's text | Before it is emitted: before its `text.done`, and before `run.done`. | +| `output` | The final assistant text; in a `stream()` run (one you iterate), every step's text. `send()` with a listener checks the final text only. | Before it is emitted: before its `text.done`, and before `run.done`. | | `tools` | A tool call's arguments (`text` is them as JSON, `args` the object) | After the [permission rules](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/approvals.md#permission-policies) (skipped when a rule denies the call) and before `needsApproval`. | A guardrail is `{ name, check(ctx) }`. `ctx` has `kind` (`'input'`, `'output'` @@ -6971,7 +6973,7 @@ Every method is optional: implement only the points you need. A hook that return | `preGenerate(ctx)` | Before each model call. | `GenerateHookContext`: `request` (the live `GenerateOptions` about to be sent) and `emit?`, plus the common fields. | Nothing; mutate `ctx.messages` or `ctx.request`, or throw. | | `postGenerate(ctx, result)` | After each model call resolves, with its `GenerateResult`. | `GenerateHookContext`. | Nothing; or throw. | -The common fields on every context are `agentId`, `agentName`, `sessionId`, `messages` (the live conversation history), `metadata` (a free-form bag) and `subagent` (set inside a sub-agent). `ctx.emit` exists only in streamed runs and adds `compaction.start` / `compaction.done` events to the stream. +The common fields on every context are `agentId`, `agentName`, `sessionId`, `messages` (the live conversation history), `metadata` (a free-form bag) and `subagent` (set inside a sub-agent). `ctx.emit` exists when the run has listeners (`createAgent({ onEvent })`, `onAgentEvent`) or is streamed, and adds `compaction.start` / `compaction.done` events to the run's events. ## Hook outcomes @@ -10257,10 +10259,13 @@ To observe every run without iterating one, give the agent a listener: / `stream()` / `resumeAfterApproval()` options. It is called synchronously with each `AgentEvent` as it happens, on `send()` as on `stream()`, for session turns and for runs resumed after an approval. It gets the same events, -in the same order, as iterating the stream would, with one difference: -`send()` / `execute()` generate each model step whole, so a step's text comes -as a single `text.delta` (as it does on `stream()` with a provider that cannot -stream). Sub-agents' events arrive tagged with `subagent`. +in the same order, as iterating the stream would: with a listener, `send()` / +`execute()` stream each model call too, so a step's text arrives as several +`text.delta` events (one `text.delta` per step when the provider cannot +stream). Concatenate them, or use `text.done` for each step's whole text; +`AgentExecutor.execute({ streamModelCalls: false })` generates whole steps. +Output guardrails still check only the final reply of a `send()`, as without +a listener. Sub-agents' events arrive tagged with `subagent`. ```ts import { createAgent } from '@lousho/build-ai-agent'; @@ -10414,7 +10419,8 @@ provider has no `stream()`, or its `supportsStreaming(model)` returns 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()`. +`send()`. `send()` and `execute()` stream the same way when they have a +listener (`onEvent` / `onAgentEvent`); without one they 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 diff --git a/src/context/compactionEvents.test.ts b/src/context/compactionEvents.test.ts index e6bf3dc4..cbea2032 100644 --- a/src/context/compactionEvents.test.ts +++ b/src/context/compactionEvents.test.ts @@ -207,24 +207,24 @@ describe('createAgent({ hooks, compaction }) options (LOU-W3.2)', () => { expect(seen.filter((s) => s === 'tool:fetch_page')).toHaveLength(4); }); - it('lets a hook add events with ctx.emit in streamed runs only', async () => { + it('lets a hook add events with ctx.emit when the run has listeners or is streamed', async () => { const hook: AgentHook = { name: 'emitter', preGenerate: (ctx) => ctx.emit?.({ type: 'compaction.start', strategy: 'custom', tokensBefore: 1, contextWindow: 2, thresholdTokens: 3 }), }; + const expected = ['run.start', 'step.start', 'compaction.start', 'text.delta', 'text.done', 'step.done', 'run.done']; const agent = createAgent({ provider: mockModel(['hi', 'hi']), hooks: [hook] }); const events = await collect(agent.stream('hello')); - expect(events.map((e) => e.type)).toEqual([ - 'run.start', - 'step.start', - 'compaction.start', - 'text.delta', - 'text.done', - 'step.done', - 'run.done', - ]); + expect(events.map((e) => e.type)).toEqual(expected); + // No listener: `ctx.emit` is not set, and the run is unaffected. expect((await agent.send('hello')).text).toBe('hi'); + + // M9: send() with a listener sets it too. + const heard: AgentEvent[] = []; + const listening = createAgent({ provider: mockModel(['hi']), hooks: [hook], onEvent: (event) => heard.push(event) }); + expect((await listening.send('hello')).text).toBe('hi'); + expect(heard.map((e) => e.type)).toEqual(expected); }); it('rejects a summarizer together with a strategy, and a threshold outside (0, 1]', () => { diff --git a/src/createAgent.ts b/src/createAgent.ts index 6b189d3e..2ae8d638 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -242,8 +242,10 @@ export interface CreateAgentBase { expect(streamed.status).toBe(200); expect(streamed.headers.get('content-type')).toContain('text/event-stream'); const raw = await streamed.text(); - expect(raw).toContain('"text":"This is a mock response. "'); + expect(raw).toContain('"text":"This is a mock response."'); expect(raw.endsWith('event: done\ndata: {}\n\n')).toBe(true); const transcript = await fetch(`${base}/chat/workerd-1`, { headers: { Authorization: 'Bearer dev-token' } }); diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index e685e201..94dfa335 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -258,9 +258,10 @@ export interface ExecuteOptions extends PermissionOptions { /** * LOU-D41: called with every {@link AgentEvent} of the run, synchronously * as it happens - the same events, in the same order, as `stream()` - * yields, on `execute()` too (which still generates each model step whole, - * so its text arrives as one `text.delta` per step). Sub-agents' events - * arrive tagged with `subagent`. See docs/streaming.md#listening-without-iterating. + * yields, on `execute()` too. M9: with a listener, `execute()` streams each + * model call when the provider can, so a step's text arrives as several + * `text.delta` events, as on `stream()` (see {@link ExecuteOptions.streamModelCalls}). + * Sub-agents' events arrive tagged with `subagent`. See docs/streaming.md#listening-without-iterating. * * @example * ```ts @@ -274,6 +275,20 @@ export interface ExecuteOptions extends PermissionOptions { * (a one-time `console.warn` says so). */ onEvent?: (event: ExecutionEvent) => void; + /** + * M9: whether a run with listeners (`onAgentEvent` / `onEvent`) streams its + * model calls through `provider.stream()` when the provider can, so each + * step's text reaches the listeners as several `text.delta` events. + * Default `true`; `false` generates each step whole (one `text.delta` per + * step). Ignored by `stream()`, which always streams, and by a run without + * listeners, which always generates. + * + * @example + * ```ts + * await AgentExecutor.execute({ agent, input: 'Hi', provider, onAgentEvent, streamModelCalls: false }); + * ``` + */ + streamModelCalls?: boolean; approvalStore?: ApprovalStore; /** * Durable execution: with `checkpointStore`, the run is checkpointed under @@ -624,7 +639,9 @@ export interface ExecutionResult { */ export class AgentExecutor { /** - * Execute agent without streaming + * Runs the agent to its result. With listeners (`onAgentEvent` / + * `onEvent`), its model calls are streamed to them (M9; see + * {@link ExecuteOptions.streamModelCalls}). */ static async execute(options: ExecuteOptions): Promise { // AgentExecutor is a static, instance-free API - there is no @@ -991,15 +1008,17 @@ export class AgentExecutor { /** * LOU-X4: the step's text after the output guardrails (they check the final - * reply, and every step's text when streamed, before it is emitted), or the - * blocked run's result. + * reply, and every step's text when the run is iterated - a `stream()` - + * before it is emitted), or the blocked run's result. `send()` / `execute()` + * with listeners stream their model calls (M9) but are not iterated: only + * the final reply is checked, as without listeners. */ private static async guardOutput( options: ExecuteOptions, state: AgentRunState, { text, toolCalls }: GenerateResult ): Promise { - if (!text || (toolCalls?.length && !runEventsOf(options)?.streamed)) return text; + if (!text || (toolCalls?.length && !runEventsOf(options)?.iterated)) return text; const checked = await checkOutputGuardrails(options, text, state.messages); return 'tripped' in checked ? this.stopForGuardrail(options, state, checked.tripped) : checked.text; } diff --git a/src/execution/agentRun.ts b/src/execution/agentRun.ts index 178f678d..cccc3ba5 100644 --- a/src/execution/agentRun.ts +++ b/src/execution/agentRun.ts @@ -7,8 +7,9 @@ * {@link RUN_EVENTS} key of its options), including how one model step is * obtained (streamed, with `text.delta` per chunk). The sink turns it into * AgentEvents for the run's listeners: an AgentRun's buffer, `onAgentEvent`, - * and the deprecated `onEvent` (through legacyEvents.ts). A non-streamed run - * with listeners gets a sink too (see {@link observeRun}). + * and the deprecated `onEvent` (through legacyEvents.ts). A run that is not + * iterated but has listeners (`send()`, `execute({ onAgentEvent })`) gets a + * sink too, and (M9) streams its model calls as well (see {@link observeRun}). * * Backpressure: none. The run never waits for the consumer; events are * buffered without loss until they are read. @@ -106,16 +107,21 @@ export interface AgentRun extends AsyncIterable { /** A tool call's outcome, as the loop reports it. */ export type ToolSettled = NonNullable; -/** The listeners a run's options can carry (LOU-D41). */ -export type RunListeners = Pick; +/** The listeners a run's options can carry (LOU-D41), and whether they get model calls streamed (M9). */ +export type RunListeners = Pick; /** * Everything AgentExecutor reports about a run, as AgentEvents. Internal: * reached through `options[RUN_EVENTS]`. */ export interface RunEventSink { - /** Whether model steps are streamed (a `stream()` run; `execute()` generates each step whole). */ - readonly streamed: boolean; + /** + * Whether the caller iterates the run (a `stream()` run). Output guardrails + * then check every step's text, not only the final reply. `send()` and + * `execute()` with listeners are not iterated, even though (M9) their + * model calls are streamed. + */ + readonly iterated: boolean; /** Adds `options`' listeners to the run's (each listener once). */ listen(options: RunListeners): void; /** A top-level run reports one `run.start`, however often it is (re)started. */ @@ -233,7 +239,13 @@ class RunEvents { private readonly legacy = new Set<(event: ExecutionEvent) => void>(); private readonly toolStarts = new Map(); - constructor(private readonly streamed: boolean) {} + /** + * `iterated`: the caller iterates a `stream()` (see {@link RunEventSink.iterated}). + * `streamModelCalls` (M9): each model call is streamed when the provider + * can, so its text arrives as several `text.delta`; otherwise each step is + * generated whole and its text is one `text.delta`. + */ + constructor(private readonly mode: { readonly iterated: boolean; readonly streamModelCalls: boolean }) {} listen({ onAgentEvent, onEvent }: RunListeners): void { if (onAgentEvent) this.listeners.add(onAgentEvent); @@ -320,7 +332,7 @@ class RunEvents { this.emit({ type: 'error', error: toEventError(error) }, subagent, { error }); }; return { - streamed: this.streamed, + iterated: this.mode.iterated, listen: (options) => this.listen(options), runStart: ({ id, name }) => this.emit({ type: 'run.start', agentName: name ?? '', ...(id !== undefined && { agentId: id }) }, subagent), textDone: (text, stepUsage) => this.emit({ type: 'text.done', text }, subagent, { stepUsage }), @@ -412,7 +424,7 @@ class RunEvents { onReasoning: (event) => this.emit(event, subagent), onOutput, }; - if (this.streamed && canStream(provider, request)) { + if (this.mode.streamModelCalls && canStream(provider, request)) { return generateViaStream(provider, request, sink); } const generated = await provider.generate(request); @@ -460,13 +472,18 @@ export function streamResumed( * stream's, or a new one when `options` has listeners - after adding * `options`' listeners to it, then reports how it ended (`run.done`, or * `error` and `run.done`). Without a sink or listeners, just runs it. + * + * M9: a new sink (listeners, no iteration) streams the run's model calls + * like `stream()` does, unless `options.streamModelCalls` is `false`; it is + * not `iterated`, so output guardrails behave as on a run without listeners. */ export async function observeRun( options: T, run: (options: T) => Promise ): Promise { - const { onAgentEvent, onEvent } = options; - const sink = runEventsOf(options) ?? (onAgentEvent || onEvent ? new RunEvents(false).sink() : undefined); + const { onAgentEvent, onEvent, streamModelCalls = true } = options; + const sink = + runEventsOf(options) ?? (onAgentEvent || onEvent ? new RunEvents({ iterated: false, streamModelCalls }).sink() : undefined); if (!sink) return run(options); sink.listen(options); try { @@ -481,7 +498,7 @@ export async function observeRun( class AgentRunImpl implements AgentRun { readonly result: Promise; - private readonly events = new RunEvents(true); + private readonly events = new RunEvents({ iterated: true, streamModelCalls: true }); private readonly controller = new AbortController(); private readonly queue: AgentEvent[] = []; private iterated = false; diff --git a/src/execution/cancellation.test.ts b/src/execution/cancellation.test.ts index fc5c9c3a..81b1e8cc 100644 --- a/src/execution/cancellation.test.ts +++ b/src/execution/cancellation.test.ts @@ -42,7 +42,7 @@ function scriptedProvider(generate: (options: GenerateOptions) => Promise true, - supportsStreaming: () => true, + supportsStreaming: () => false, // generate-only: `stream` is a stub getModels: async () => [], }; return { provider, generate: spy }; diff --git a/src/execution/generateStep.ts b/src/execution/generateStep.ts index ab23b517..1ac05940 100644 --- a/src/execution/generateStep.ts +++ b/src/execution/generateStep.ts @@ -174,8 +174,8 @@ export function generateInSpan( init.attributes, async (llmSpan) => { const llmStart = Date.now(); - // LOU-V2: a streaming run obtains the step through its sink (streamed - // when the provider can); everything around it is the same. + // LOU-V2: a run with a sink (a stream, or M9: listeners) obtains the + // step through it (streamed when the provider can); everything around it is the same. const runEvents = runEventsOf(options); const onOutput = () => options.inputQueue?.callOutput(); const generated = await abortable( diff --git a/src/execution/hooks.ts b/src/execution/hooks.ts index 1bb7ff11..8a9af701 100644 --- a/src/execution/hooks.ts +++ b/src/execution/hooks.ts @@ -163,7 +163,8 @@ export type HookEventPayload = Extract { it('gets the same AgentEvents from send() as stream() yields', async () => { const heard: AgentEvent[] = []; - // send() generates each step whole, so its text matches a stream only chunk for chunk with a non-streaming provider. - const provider = Object.assign(mockModel([...script(), ...script()]), { supportsStreaming: () => false }); + // M9: send() with a listener streams its model calls, so its text.delta events match a stream's chunk for chunk. + const provider = mockModel([...script(), ...script()]); const agent = createAgent({ instructions: 'Be brief.', provider, tools: [weather], onEvent: (e) => heard.push(e) }); await agent.send('go'); const sent = heard.splice(0); diff --git a/src/execution/sendStreams.live.test.ts b/src/execution/sendStreams.live.test.ts new file mode 100644 index 00000000..d214898e --- /dev/null +++ b/src/execution/sendStreams.live.test.ts @@ -0,0 +1,28 @@ +/** + * Live test (costs money, needs OPENROUTER_API_KEY; at most 0.05 USD): `send()` with a listener on + * `openrouter/openai/gpt-4o-mini` streams the model call, so the step's text arrives as several `text.delta` + * events (M9). Records `__fixtures__/cassettes/send-streams.json` for a replay test. Run with + * `npm run test:live -- src/execution/sendStreams`. + */ +import { describe, expect, it } from 'vitest'; +import { createAgent } from '../createAgent'; +import { resolveProvider } from '../providers/resolveProvider'; +import { recordReplay } from '../testing'; +import type { AgentEvent } from './agentEvents'; + +describe.skipIf(!process.env.OPENROUTER_API_KEY)('send() with a listener streams live (M9)', () => { + it('delivers the reply as several text.delta events that add up to result.text', async () => { + const provider = recordReplay(() => resolveProvider('openrouter/openai/gpt-4o-mini'), { + cassette: 'src/execution/__fixtures__/cassettes/send-streams.json', + mode: 'record', + }); + const heard: AgentEvent[] = []; + const agent = createAgent({ provider, maxSteps: 1, onEvent: (event) => heard.push(event) }); + + const result = await agent.send('Count from 1 to 20, comma separated.'); + + const deltas = heard.flatMap((event) => (event.type === 'text.delta' ? [event.text] : [])); + expect(deltas.length).toBeGreaterThan(1); + expect(deltas.join('')).toBe(result.text); + }); +}); diff --git a/src/execution/sendStreams.test.ts b/src/execution/sendStreams.test.ts new file mode 100644 index 00000000..3abc69b4 --- /dev/null +++ b/src/execution/sendStreams.test.ts @@ -0,0 +1,202 @@ +/** + * M9: `send()` and `AgentExecutor.execute()` with a listener stream their + * model calls like `stream()` does - a step's text reaches the listener as + * several `text.delta` events - without changing what the run returns. + */ + +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { APICallError } from 'ai'; +import { z } from 'zod'; +import { createAgent } from '../createAgent'; +import { defineTool } from '../tools/defineTool'; +import { ToolRegistry } from '../tools'; +import { LLMProviderRegistry, type LLMProvider } from '../providers/llm'; +import { mockModel, type MockModel, type MockTurn } from '../testing'; +import { AgentExecutor } from './AgentExecutor'; +import type { AgentEvent } from './agentEvents'; +import type { IoGuardrail } from './ioGuardrails'; + +const weather = defineTool({ + name: 'get_weather', + description: 'Weather for a city', + input: z.object({ city: z.string() }), + execute: async ({ city }) => ({ city, tempC: 21 }), +}); + +const script = (): MockTurn[] => [ + { + text: 'Let me check the weather.', + toolCalls: [{ id: 'call_rome', name: 'get_weather', args: { city: 'Rome' } }], + usage: { inputTokens: 12, outputTokens: 7 }, + }, + { text: 'It is warm in Rome today.', usage: { inputTokens: 30, outputTokens: 9 } }, +]; + +/** The text of each step's `text.delta` events, step by step. */ +function deltasPerStep(events: AgentEvent[]): string[][] { + const steps: string[][] = []; + for (const event of events) { + if (event.type === 'step.start') steps.push([]); + if (event.type === 'text.delta') steps.at(-1)?.push(event.text); + } + return steps; +} + +/** `model` with its `stream()` calls counted. */ +function counting(model: MockModel, name = model.name): LLMProvider & { streamed: number } { + const provider = { + streamed: 0, + name, + defaultModel: model.defaultModel, + generate: (call: Parameters[0]) => model.generate(call), + stream: (call: Parameters[0]) => { + provider.streamed += 1; + return model.stream(call); + }, + supportsTools: (id: string) => model.supportsTools(id), + supportsStreaming: (id: string) => model.supportsStreaming(id), + getModels: () => model.getModels(), + }; + return provider; +} + +describe('send() with a listener streams model calls (M9)', () => { + it('emits several text.delta per step that add up to the step text, with the usage of a generated run', async () => { + const heard: AgentEvent[] = []; + const provider = counting(mockModel(script())); + const agent = createAgent({ instructions: 'Be brief.', provider, tools: [weather], onEvent: (e) => heard.push(e) }); + + const result = await agent.send('Weather in Rome?'); + + expect(provider.streamed).toBe(2); + const steps = deltasPerStep(heard); + expect(steps).toHaveLength(2); + expect(steps.every((deltas) => deltas.length > 1)).toBe(true); + const done = heard.filter((e) => e.type === 'text.done').map((e) => e.text); + expect(steps.map((deltas) => deltas.join(''))).toEqual(done); + expect(done).toEqual(['Let me check the weather.', 'It is warm in Rome today.']); + expect(result.text).toBe('It is warm in Rome today.'); + + // The same script without a listener generates each step: same result, same usage (no double counting). + const plain = counting(mockModel(script())); + const generated = await createAgent({ instructions: 'Be brief.', provider: plain, tools: [weather] }).send('Weather in Rome?'); + expect(plain.streamed).toBe(0); + expect(result.usage).toEqual(generated.usage); + expect(result.toolCalls).toEqual(generated.toolCalls); + expect(result.steps).toBe(generated.steps); + expect(heard.filter((e) => e.type === 'step.done').map((e) => e.type === 'step.done' && e.usage?.totalTokens)).toEqual([ + 19, 39, + ]); + }); + + it('streams AgentExecutor.execute({ onAgentEvent }) too, and execute({ streamModelCalls: false }) generates whole steps', async () => { + const toolRegistry = new ToolRegistry(); + toolRegistry.registerMany([weather]); + const agent = { id: 'a', name: 'Agent', prompt: 'p', tools: { get_weather: { tool: 'get_weather' } } }; + + const streamedEvents: AgentEvent[] = []; + const streaming = counting(mockModel(script())); + await AgentExecutor.execute({ agent, provider: streaming, toolRegistry, input: 'go', onAgentEvent: (e) => streamedEvents.push(e) }); + expect(streaming.streamed).toBe(2); + expect(deltasPerStep(streamedEvents).map((deltas) => deltas.length > 1)).toEqual([true, true]); + + const wholeEvents: AgentEvent[] = []; + const whole = counting(mockModel(script())); + const result = await AgentExecutor.execute({ + agent, + provider: whole, + toolRegistry, + input: 'go', + onAgentEvent: (e) => wholeEvents.push(e), + streamModelCalls: false, + }); + expect(whole.streamed).toBe(0); + expect(deltasPerStep(wholeEvents)).toEqual([['Let me check the weather.'], ['It is warm in Rome today.']]); + expect(result.text).toBe('It is warm in Rome today.'); + }); + + it('generates whole steps with a provider that cannot stream (one text.delta per step)', async () => { + const model = mockModel(script()); + const { stream: _stream, ...generateOnly } = counting(model); + const heard: AgentEvent[] = []; + const agent = createAgent({ provider: generateOnly as unknown as LLMProvider, tools: [weather], onEvent: (e) => heard.push(e) }); + + const result = await agent.send('Weather in Rome?'); + + expect(deltasPerStep(heard)).toEqual([['Let me check the weather.'], ['It is warm in Rome today.']]); + expect(result.text).toBe('It is warm in Rome today.'); + + const unsupported: AgentEvent[] = []; + const notStreaming = Object.assign(counting(mockModel(script())), { supportsStreaming: () => false }); + await createAgent({ provider: notStreaming, tools: [weather], onEvent: (e) => unsupported.push(e) }).send('Weather in Rome?'); + expect(notStreaming.streamed).toBe(0); + expect(deltasPerStep(unsupported).map((deltas) => deltas.length)).toEqual([1, 1]); + }); + + it('runs output guardrails on the final reply only, as before, while stream() checks every step', async () => { + const checked: string[] = []; + const guardrail: IoGuardrail = { + name: 'no-check', + check: ({ text }) => { + checked.push(text); + return text.includes('check') ? { ok: false, reason: 'mentions check' } : { ok: true }; + }, + }; + + const heard: AgentEvent[] = []; + const listening = createAgent({ provider: mockModel(script()), tools: [weather], guardrails: { output: [guardrail] }, onEvent: (e) => heard.push(e) }); + const sent = await listening.send('Weather in Rome?'); + expect(sent.finishReason).toBe('stop'); + expect(checked).toEqual(['It is warm in Rome today.']); + expect(deltasPerStep(heard)[0].length).toBeGreaterThan(1); + + checked.length = 0; + const iterated = createAgent({ provider: mockModel(script()), tools: [weather], guardrails: { output: [guardrail] } }); + const run = iterated.stream('Weather in Rome?'); + for await (const _event of run) void _event; + expect((await run.result).finishReason).toBe('guardrail'); + expect(checked).toEqual(['Let me check the weather.']); + }); +}); + +describe('send() with a listener: retry and fallback cover streamed calls (M9)', () => { + const ENV_VARS = ['LOUSHO_MODEL', 'OPENAI_API_KEY', 'ANTHROPIC_API_KEY', 'OPENROUTER_API_KEY', 'OLLAMA_BASE_URL']; + const apiError = (statusCode: number) => + new APICallError({ message: `HTTP ${statusCode}`, url: 'https://api.example.com/v1/chat', requestBodyValues: {}, statusCode, isRetryable: true }); + + beforeEach(() => { + for (const name of ENV_VARS) vi.stubEnv(name, ''); + vi.stubEnv('OPENAI_API_KEY', 'sk-openai'); + vi.stubEnv('ANTHROPIC_API_KEY', 'sk-ant'); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + vi.restoreAllMocks(); + }); + + it('retries a failed streamed call, then falls back, reporting both to the listener', async () => { + const primary = counting(mockModel([{ error: apiError(503) }, { error: apiError(503) }]), 'openai'); + const fallback = counting(mockModel(['Hello from the fallback model.'], { defaultModel: 'claude-3-5-haiku-latest' }), 'anthropic'); + const providers: Record = { openai: primary, anthropic: fallback }; + vi.spyOn(LLMProviderRegistry, 'create').mockImplementation((name) => providers[name]); + const heard: AgentEvent[] = []; + const agent = createAgent({ + model: 'openai/gpt-4o-mini', + fallbackModels: ['anthropic/claude-3-5-haiku-latest'], + retry: { backoff: { initialMs: 1, jitter: false }, maxRetries: 1 }, + onEvent: (e) => heard.push(e), + }); + + const result = await agent.send('hi'); + + expect(result.text).toBe('Hello from the fallback model.'); + expect(primary.streamed).toBe(2); + expect(fallback.streamed).toBe(1); + const types = heard.map((e) => e.type); + expect(types).toContain('provider.retry'); + expect(types).toContain('provider.fallback'); + expect(deltasPerStep(heard)[0].length).toBeGreaterThan(1); + expect(deltasPerStep(heard)[0].join('')).toBe(result.text); + }); +}); diff --git a/src/providers/llm.test.ts b/src/providers/llm.test.ts index 71a682a8..c7e6ad1d 100644 --- a/src/providers/llm.test.ts +++ b/src/providers/llm.test.ts @@ -121,7 +121,8 @@ describe('LLM Providers', () => { } expect(chunks.length).toBeGreaterThan(0); - expect(chunks.join('')).toBe('Hello, world! '); + // M9: the chunks add up to exactly the text generate() returns (no trailing space). + expect(chunks.join('')).toBe('Hello, world!'); }); it('should stream with full chunks', async () => { @@ -140,6 +141,27 @@ describe('LLM Providers', () => { expect(chunks[chunks.length - 1].type).toBe('finish'); }); + it('streams the step generate() would return: text, tool calls, finish reason and usage (M9)', async () => { + const request = { + model: 'mock-model', + messages: [{ role: 'user' as const, content: 'please use current-date' }], + tools: [{ type: 'function' as const, function: { name: 'current-date', description: 'd', parameters: {} } }], + }; + const generated = await new MockLLMProvider({ name: 'mock' }).generate(request); + const streamed = await new MockLLMProvider({ name: 'mock' }).stream(request); + const chunks = []; + for await (const chunk of streamed.fullStream) chunks.push(chunk); + + const deltas = chunks.filter((c) => c.type === 'text-delta').map((c) => c.textDelta); + expect(deltas.length).toBeGreaterThan(1); + expect(deltas.join('')).toBe(generated.text); + expect(chunks.filter((c) => c.type === 'tool-call').map((c) => c.toolCall?.function)).toEqual( + generated.toolCalls?.map((call) => call.function) + ); + expect(chunks.at(-1)).toEqual({ type: 'finish', finishReason: 'tool_calls', usage: generated.usage }); + expect(await streamed.finishReason).toBe(generated.finishReason); + }); + it('should resolve text promise', async () => { const result = await provider.stream({ model: 'mock-model', diff --git a/src/providers/mock.ts b/src/providers/mock.ts index 046bb38d..cfc84830 100644 --- a/src/providers/mock.ts +++ b/src/providers/mock.ts @@ -74,6 +74,12 @@ export class MockLLMProvider implements LLMProvider { }; } + /** + * Streams the same step `generate()` would return - the same text (in + * word-sized `text-delta` chunks whose concatenation is that text), the + * same tool calls, finish reason and usage - so a run gets the same result + * whether it generates or streams its model calls (M9). + */ async stream(options: GenerateOptions): Promise { options.signal?.throwIfAborted(); if (this.simulateError) { @@ -81,53 +87,38 @@ export class MockLLMProvider implements LLMProvider { } const text = this.getNextResponse(); - const words = text.split(' '); - const chunks: StreamChunk[] = []; - - const fullStreamGenerator = async function* ( - this: MockLLMProvider - ): AsyncGenerator { - for (const word of words) { + const toolCalls = this.extractToolCalls(options); + const finishReason: GenerateResult['finishReason'] = toolCalls.length > 0 ? 'tool_calls' : 'stop'; + const promptTokens = this.countTokens(options.messages); + const completionTokens = this.countTokens([{ role: 'assistant', content: text }]); + const usage = { promptTokens, completionTokens, totalTokens: promptTokens + completionTokens }; + // Word-sized chunks ("This ", "is ", ..., "response.") that add up to `text`. + const deltas = text.match(/\S+\s*|\s+/g) ?? []; + + const fullStreamGenerator = async function* (this: MockLLMProvider): AsyncGenerator { + for (const textDelta of deltas) { if (this.delay > 0) { await abortableDelay(this.delay, options.signal); } - - const chunk: StreamChunk = { - type: 'text-delta', - textDelta: word + ' ', - }; - chunks.push(chunk); - yield chunk; + yield { type: 'text-delta', textDelta }; } - - yield { - type: 'finish', - finishReason: 'stop', - usage: { - promptTokens: this.countTokens(options.messages), - completionTokens: words.length, - totalTokens: this.countTokens(options.messages) + words.length, - }, - }; + for (const toolCall of toolCalls) { + yield { type: 'tool-call', toolCall }; + } + yield { type: 'finish', finishReason, usage }; }.bind(this); const textStreamGenerator = async function* (): AsyncGenerator { - for (const word of words) { - yield word + ' '; - } + yield* deltas; }; return { fullStream: fullStreamGenerator(), textStream: textStreamGenerator(), text: Promise.resolve(text), - usage: Promise.resolve({ - promptTokens: this.countTokens(options.messages), - completionTokens: words.length, - totalTokens: this.countTokens(options.messages) + words.length, - }), - finishReason: Promise.resolve('stop'), - toolCalls: Promise.resolve([]), + usage: Promise.resolve(usage), + finishReason: Promise.resolve(finishReason), + toolCalls: Promise.resolve(toolCalls), }; }