From 29161ef82d05e94dc41478a53a041fefae83f803 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 12:40:53 +0000 Subject: [PATCH] [EPIC-V][LOU-V8] AgentSession.stream() streams a session turn and persists it on done session.stream(input, { signal }) returns the same AgentRun as agent.stream(). It queues behind earlier calls, loads history, builds send()'s message list, streams the turn, and records it exactly as send() does before delivering run.done. Aborted or failed runs leave the transcript unchanged; an approval pause is saved up to the pause and resumes via approvals.resolve(). Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_016bY29ibQgv3MDdjAbUT4YR --- docs/sessions.md | 39 +++++ docs/streaming.md | 22 +++ llms-full.txt | 61 +++++++ src/createAgent.ts | 3 +- src/createAgentApprovals.ts | 22 ++- src/session/AgentSession.ts | 49 +++++- src/session/index.ts | 2 +- src/session/sessionStream.test.ts | 253 ++++++++++++++++++++++++++++++ src/session/sessionStream.ts | 109 +++++++++++++ 9 files changed, 554 insertions(+), 6 deletions(-) create mode 100644 src/session/sessionStream.test.ts create mode 100644 src/session/sessionStream.ts diff --git a/docs/sessions.md b/docs/sessions.md index e0e19de31..37508e276 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -22,6 +22,7 @@ console.log(session.id, session.messages.length); | ------ | ----------- | | `id` | The session id: generated (a UUID) unless you pass one. | | `send(input, { signal })` | Sends a user message with the whole conversation so far and resolves with the usual `ExecutionResult`. | +| `stream(input, { signal })` | Like `send()`, but returns an `AgentRun` that streams the turn as typed events - see [Streaming a session turn](#streaming-a-session-turn). | | `messages` | A read-only snapshot of the transcript (user, assistant and tool messages; no system prompt). Editing the snapshot does not change the session. | | `load()` | Reads the saved transcript from the store. `send()` does this for you; call it to show history before the first `send()` of a resumed session. | | `clear()` | Forgets the conversation, including in the store. | @@ -41,6 +42,44 @@ then continues the run as the session's next turn, so the tool call, its result and the final answer join the transcript (see [approval gates](../README.md#human-in-the-loop-approval-gates)). +## Streaming a session turn + +`session.stream(input, { signal })` is `agent.stream()` for a conversation: it +returns the same `AgentRun` (typed events plus a `result` promise, see +[Streaming](./streaming.md)), but the turn sees the transcript so far and joins +it. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const chat = createAgent({ instructions: 'Be brief.', provider }); +const session = chat.session(); + +for await (const event of session.stream('My name is Ali.')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +// The turn is already saved: this call sees it. +const { text } = await session.send('What is my name?'); +``` + +- It builds the same message list as `send()` and queues behind earlier calls + on the session (`send()` and `stream()` can be mixed). +- When the run ends, the new user message and the run's output are saved + exactly as `send()` saves them, and only then is `run.done` delivered. When + the `for await` loop ends, or `run.result` resolves, the transcript is + complete. The events carry the `runId` of the returned handle. +- An aborted run (`signal`, or breaking out of the loop early) or a failed one + leaves the transcript as it was before the call, like `send()`. A run that + was already finished when the loop was left is saved. A failed run ends the + stream with `error` and `run.done` (`finishReason: 'error'`), and + `run.result` rejects. This includes a store that fails to load or save: the + turn then has no `run.start`. +- A run that pauses on a `needsApproval` tool ends with `approval.requested` + and `run.done` (`'awaiting-approval'`), exactly as `send()` resolves, and the + transcript holds the turn up to the pause. `agent.approvals.resolve()` + continues it as the session's next turn. An `approve` callback does not apply + to streams, as with `agent.stream()`. + Sessions are a thin layer: the session owns the transcript and hands it to the executor on each turn. They do not use the checkpoint/resume mechanism (`sessionId` + `checkpointStore` on `AgentExecutor.execute()`). That lower-level diff --git a/docs/streaming.md b/docs/streaming.md index 8ab56aa22..e4e67035e 100644 --- a/docs/streaming.md +++ b/docs/streaming.md @@ -60,6 +60,28 @@ const result = await run.result; // no iteration needed console.log(result.finishReason, result.text); ``` +## Streaming a session turn + +`session.stream(input, { signal })` streams one turn of a multi-turn +[session](./sessions.md) and returns the same `AgentRun`. The run sees the +conversation so far, and when it ends its turn is saved to the session's +store, just as `session.send()` saves it. `run.done` is delivered after the +save, so the transcript is complete when the loop ends. An aborted or failed +run, or one you stop reading early, is not saved. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +const session = agent.session({ id: 'user-42' }); + +for await (const event of session.stream('My name is Ali.')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +const again = session.stream('What is my name?'); // streams with the first turn in its history +console.log((await again.result).text); +``` + ## Event schema (version 1) Every event has these fields: diff --git a/llms-full.txt b/llms-full.txt index 2202587c9..8ef54f5d1 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -3347,6 +3347,7 @@ console.log(session.id, session.messages.length); | ------ | ----------- | | `id` | The session id: generated (a UUID) unless you pass one. | | `send(input, { signal })` | Sends a user message with the whole conversation so far and resolves with the usual `ExecutionResult`. | +| `stream(input, { signal })` | Like `send()`, but returns an `AgentRun` that streams the turn as typed events - see [Streaming a session turn](#streaming-a-session-turn). | | `messages` | A read-only snapshot of the transcript (user, assistant and tool messages; no system prompt). Editing the snapshot does not change the session. | | `load()` | Reads the saved transcript from the store. `send()` does this for you; call it to show history before the first `send()` of a resumed session. | | `clear()` | Forgets the conversation, including in the store. | @@ -3366,6 +3367,44 @@ then continues the run as the session's next turn, so the tool call, its result and the final answer join the transcript (see [approval gates](https://github.com/LinuxDevil/agent-sdk/blob/main/README.md#human-in-the-loop-approval-gates)). +## Streaming a session turn + +`session.stream(input, { signal })` is `agent.stream()` for a conversation: it +returns the same `AgentRun` (typed events plus a `result` promise, see +[Streaming](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md)), but the turn sees the transcript so far and joins +it. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const chat = createAgent({ instructions: 'Be brief.', provider }); +const session = chat.session(); + +for await (const event of session.stream('My name is Ali.')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +// The turn is already saved: this call sees it. +const { text } = await session.send('What is my name?'); +``` + +- It builds the same message list as `send()` and queues behind earlier calls + on the session (`send()` and `stream()` can be mixed). +- When the run ends, the new user message and the run's output are saved + exactly as `send()` saves them, and only then is `run.done` delivered. When + the `for await` loop ends, or `run.result` resolves, the transcript is + complete. The events carry the `runId` of the returned handle. +- An aborted run (`signal`, or breaking out of the loop early) or a failed one + leaves the transcript as it was before the call, like `send()`. A run that + was already finished when the loop was left is saved. A failed run ends the + stream with `error` and `run.done` (`finishReason: 'error'`), and + `run.result` rejects. This includes a store that fails to load or save: the + turn then has no `run.start`. +- A run that pauses on a `needsApproval` tool ends with `approval.requested` + and `run.done` (`'awaiting-approval'`), exactly as `send()` resolves, and the + transcript holds the turn up to the pause. `agent.approvals.resolve()` + continues it as the session's next turn. An `approve` callback does not apply + to streams, as with `agent.stream()`. + Sessions are a thin layer: the session owns the transcript and hands it to the executor on each turn. They do not use the checkpoint/resume mechanism (`sessionId` + `checkpointStore` on `AgentExecutor.execute()`). That lower-level @@ -3633,6 +3672,28 @@ const result = await run.result; // no iteration needed console.log(result.finishReason, result.text); ``` +## Streaming a session turn + +`session.stream(input, { signal })` streams one turn of a multi-turn +[session](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/sessions.md) and returns the same `AgentRun`. The run sees the +conversation so far, and when it ends its turn is saved to the session's +store, just as `session.send()` saves it. `run.done` is delivered after the +save, so the transcript is complete when the loop ends. An aborted or failed +run, or one you stop reading early, is not saved. + +```ts +import { createAgent } from '@loushy/build-ai-agent'; + +const agent = createAgent({ model: 'openai/gpt-4o-mini' }); +const session = agent.session({ id: 'user-42' }); + +for await (const event of session.stream('My name is Ali.')) { + if (event.type === 'text.delta') process.stdout.write(event.text); +} +const again = session.stream('What is my name?'); // streams with the first turn in its history +console.log((await again.result).text); +``` + ## Event schema (version 1) Every event has these fields: diff --git a/src/createAgent.ts b/src/createAgent.ts index 82d218392..8e1b898fc 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -333,7 +333,8 @@ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { stream(message: string, options: SendOptions = {}): AgentRun { return AgentExecutor.stream(executeOptions(message, options.signal)); }, - session: (options?: SessionOptions) => approvals.session(run, options), + session: (options?: SessionOptions) => + approvals.session(run, (input, signal) => AgentExecutor.stream(executeOptions(input, signal)), options), approvals: approvals.approvals, }; registerSubagent(simpleAgent, { spec, description: config.description }); diff --git a/src/createAgentApprovals.ts b/src/createAgentApprovals.ts index 250e23717..91564bffd 100644 --- a/src/createAgentApprovals.ts +++ b/src/createAgentApprovals.ts @@ -8,7 +8,8 @@ import type { ApprovalDecision, ApprovalStore, PendingApproval } from './execution/ApprovalGate'; import type { ExecutionResult } from './execution/AgentExecutor'; -import { AgentSession, type SessionOptions, type SessionRunner } from './session/AgentSession'; +import type { AgentRun } from './execution/agentRun'; +import { AgentSession, type SessionOptions, type SessionRunner, type SessionStreamRunner } from './session/AgentSession'; /** * Decides a tool call that needs approval without pausing the run: `true` @@ -84,6 +85,20 @@ export function createAgentApprovals(options: { store: ApprovalStore; approve?: return result; } + /** `run`, with a pause it reports (as an event or in its result) bound to `session`. */ + function inSessionRun(session: ApprovalSession, run: AgentRun): AgentRun { + return { + runId: run.runId, + result: run.result.then((result) => inSession(session, result)), + async *[Symbol.asyncIterator]() { + for await (const event of run) { + if (event.type === 'approval.requested') sessions.set(event.approvalId, session); + yield event; + } + }, + }; + } + const approvals: AgentApprovals = { list: async () => [...pending.values()], resolve(decision, { signal } = {}) { @@ -99,10 +114,11 @@ export function createAgentApprovals(options: { store: ApprovalStore; approve?: store, approvals, settle, - session(run: SessionRunner, sessionOptions?: SessionOptions): AgentSession { + session(run: SessionRunner, stream: SessionStreamRunner, sessionOptions?: SessionOptions): AgentSession { const session: ApprovalSession = new ApprovalSession( async (input, signal) => inSession(session, await settle(await run(input, signal), signal)), - sessionOptions + sessionOptions, + (input, signal) => inSessionRun(session, stream(input, signal)) ); return session; }, diff --git a/src/session/AgentSession.ts b/src/session/AgentSession.ts index 8867fbec1..afb5c4c5d 100644 --- a/src/session/AgentSession.ts +++ b/src/session/AgentSession.ts @@ -7,6 +7,8 @@ import { randomUUID } from 'node:crypto'; import type { Message } from '../providers/llm'; import type { ExecutionResult } from '../execution/AgentExecutor'; +import type { AgentRun } from '../execution/agentRun'; +import { streamSessionTurn } from './sessionStream'; import { MemorySessionStore, assertSessionId, type SessionStore } from './sessionStore'; /** Options for `agent.session()`. */ @@ -26,6 +28,12 @@ export interface SessionOptions { */ export type SessionRunner = (input: Message[], signal?: AbortSignal) => Promise; +/** + * Streams one turn (LOU-V8): like {@link SessionRunner}, but returns the + * run's `AgentRun`. Supplied by `createAgent()`. + */ +export type SessionStreamRunner = (input: Message[], signal?: AbortSignal) => AgentRun; + /** * Longest prefix of `messages` that a provider accepts: every assistant * tool-call turn is followed by a result for each of its calls, and no tool @@ -68,15 +76,17 @@ export class AgentSession { readonly id: string; private readonly store: SessionStore; private readonly run: SessionRunner; + private readonly streamRun: SessionStreamRunner | undefined; private history: Message[] = []; private loaded = false; private tail: Promise = Promise.resolve(); - constructor(run: SessionRunner, options: SessionOptions = {}) { + constructor(run: SessionRunner, options: SessionOptions = {}, streamRun?: SessionStreamRunner) { if (options.id !== undefined) assertSessionId(options.id); this.id = options.id ?? randomUUID(); this.store = options.store ?? new MemorySessionStore(); this.run = run; + this.streamRun = streamRun; } /** @@ -110,6 +120,43 @@ export class AgentSession { return this.enqueue(() => this.turn(input, options.signal)); } + /** + * Like `send()`, but streams the turn as the `AgentRun` that `agent.stream()` + * returns (see docs/streaming.md). The turn waits for earlier calls, loads + * the history and runs like `send()` does; once the run ends, the new user + * message and the run's output are saved to the store exactly as `send()` + * saves them, and only then is `run.done` delivered, so the transcript is + * complete when the `for await` loop ends. `run.result` is `send()`'s result. + * + * Like `send()`, an aborted run (`signal`, or breaking out of the loop + * early, unless the run had already finished), or one that fails, leaves + * the transcript as it was. A run that + * pauses for approval is saved up to the pause and continues with + * `agent.approvals.resolve()`. If saving fails, `run.result` rejects and + * the stream ends with `error` and `run.done` (`finishReason: 'error'`). + * + * @example + * ```ts + * for await (const event of session.stream('And in Paris?')) { + * if (event.type === 'text.delta') process.stdout.write(event.text); + * } + * ``` + */ + stream(input: string, options: { signal?: AbortSignal } = {}): AgentRun { + const streamRun = this.streamRun; + if (!streamRun) throw new Error('This AgentSession was created without a streaming runner, so it cannot stream().'); + return streamSessionTurn( + (signal, started) => + this.enqueue(async () => { + await this.ensureLoaded(); + const run = streamRun([...this.history, { role: 'user', content: input }], signal); + started(run); + return this.record(await run.result); + }), + options.signal + ); + } + /** Forget the conversation (also deletes it from the store). */ clear(): Promise { return this.enqueue(async () => { diff --git a/src/session/index.ts b/src/session/index.ts index 27e917130..eeebd98cc 100644 --- a/src/session/index.ts +++ b/src/session/index.ts @@ -1,4 +1,4 @@ export { AgentSession } from './AgentSession'; -export type { SessionOptions } from './AgentSession'; +export type { SessionOptions, SessionRunner, SessionStreamRunner } from './AgentSession'; export { MemorySessionStore, FileSessionStore } from './sessionStore'; export type { SessionStore } from './sessionStore'; diff --git a/src/session/sessionStream.test.ts b/src/session/sessionStream.test.ts new file mode 100644 index 000000000..ad7372fbb --- /dev/null +++ b/src/session/sessionStream.test.ts @@ -0,0 +1,253 @@ +/** + * LOU-V8: `session.stream()` streams a session turn and persists it when the run is done. + */ +import { describe, it, expect, vi } from 'vitest'; +import { z } from 'zod'; +import { createAgent } from '../createAgent'; +import { defineTool } from '../tools/defineTool'; +import { mockModel } from '../testing'; +import type { AgentEvent } from '../execution/agentEvents'; +import type { AgentRun } from '../execution/agentRun'; +import type { SessionStore } from './index'; +import { MemorySessionStore } from './index'; +import type { Message } from '../providers/llm'; + +async function collect(run: AgentRun): Promise { + const events: AgentEvent[] = []; + for await (const event of run) events.push(event); + return events; +} + +const roles = (messages: readonly { role: string }[]): string[] => messages.map((m) => m.role); + +const emailTool = () => + defineTool({ + name: 'send_email', + description: 'Sends an email', + input: z.object({ to: z.string() }), + needsApproval: true, + execute: async ({ to }) => `sent to ${to}`, + }); +const callEmail = { toolCalls: [{ name: 'send_email', args: { to: 'sam@example.com' }, id: 'call_email' }] }; + +describe('AgentSession.stream()', () => { + it('streams the turn in order and has saved it when the run is done', async () => { + const store = new MemorySessionStore(); + const session = createAgent({ provider: mockModel(['Nice to meet you, Ali.']) }).session({ id: 's1', store }); + + const run = session.stream('My name is Ali.'); + const events: AgentEvent[] = []; + for await (const event of run) { + events.push(event); + // The transcript is complete by the time run.done is delivered. + if (event.type === 'run.done') expect(roles((await store.load('s1')) ?? [])).toEqual(['user', 'assistant']); + } + + expect(events.map((e) => e.type)).toEqual([ + 'run.start', + 'step.start', + ...events.filter((e) => e.type === 'text.delta').map(() => 'text.delta'), + 'text.done', + 'step.done', + 'run.done', + ]); + expect(events.filter((e) => e.type === 'text.delta').length).toBeGreaterThan(1); + expect(events.map((e) => e.seq)).toEqual(events.map((_, i) => i)); + expect(new Set(events.map((e) => e.runId))).toEqual(new Set([run.runId])); + const result = await run.result; + expect(result.text).toBe('Nice to meet you, Ali.'); + expect(events.at(-1)).toMatchObject({ type: 'run.done', finishReason: 'stop', text: 'Nice to meet you, Ali.' }); + expect(await store.load('s1')).toEqual([ + { role: 'user', content: 'My name is Ali.' }, + expect.objectContaining({ role: 'assistant', content: 'Nice to meet you, Ali.' }), + ]); + expect(session.messages).toEqual(await store.load('s1')); + }); + + it('shows a second stream() the first turn, and mixes with send()', async () => { + const model = mockModel(['Nice to meet you, Ali.', 'Your name is Ali.', 'Yes.']); + const session = createAgent({ prompt: 'Be brief.', provider: model }).session(); + + await collect(session.stream('My name is Ali.')); + await collect(session.stream('What is my name?')); + await session.send('Sure?'); + + expect(model.calls[1].messages.map((m) => `${m.role}:${m.content}`)).toEqual([ + 'system:Be brief.', + 'user:My name is Ali.', + 'assistant:Nice to meet you, Ali.', + 'user:What is my name?', + ]); + expect(roles(session.messages)).toEqual(['user', 'assistant', 'user', 'assistant', 'user', 'assistant']); + }); + + it('continues a stored conversation and queues behind an earlier send()', async () => { + const store = new MemorySessionStore(); + await createAgent({ provider: mockModel(['Hi Ali.']) }).session({ id: 'resume', store }).send('I am Ali.'); + const model = mockModel(['one', 'two']); + const session = createAgent({ provider: model }).session({ id: 'resume', store }); + + const first = session.send('first'); + const run = session.stream('second'); + await Promise.all([first, run.result]); + + expect(convoTexts(model.calls[1].messages)).toEqual(['I am Ali.', 'Hi Ali.', 'first', 'one', 'second']); + expect(roles(session.messages)).toHaveLength(6); + }); + + it('works without iterating, and an iterated run can only be read once', async () => { + const session = createAgent({ provider: mockModel(['a', 'b']) }).session(); + + const quiet = session.stream('one'); + expect((await quiet.result).text).toBe('a'); + expect(session.messages).toHaveLength(2); + + const run = session.stream('two'); + await collect(run); + await expect(collect(run)).rejects.toThrow(/only be iterated once/); + }); + + it('leaves the transcript unchanged when a stream is aborted by its signal', async () => { + const controller = new AbortController(); + const slow = defineTool({ + name: 'slow', + description: 'slow', + input: z.object({}), + execute: async () => { + controller.abort(); + return 'done'; + }, + }); + const model = mockModel(['hello', { toolCalls: [{ name: 'slow' }, { name: 'slow' }] }, 'next']); + const session = createAgent({ provider: model, tools: [slow] }).session(); + await session.send('hi'); + const before = session.messages; + + const run = session.stream('go', { signal: controller.signal }); + const events = await collect(run); + + expect((await run.result).finishReason).toBe('aborted'); + expect(events.at(-1)).toMatchObject({ type: 'run.done', finishReason: 'aborted' }); + expect(session.messages).toEqual(before); + await session.send('still there?'); + expect(roles(model.calls[2].messages.filter((m) => m.role !== 'system'))).toEqual(['user', 'assistant', 'user']); + }); + + it('leaves the transcript unchanged when the loop is exited early', async () => { + const wait = defineTool({ + name: 'wait', + description: 'waits', + input: z.object({}), + execute: () => new Promise((resolve) => setTimeout(() => resolve('waited'), 20)), + }); + const model = mockModel(['first reply', { toolCalls: [{ name: 'wait' }] }, 'never sent', 'ok']); + const session = createAgent({ provider: model, tools: [wait] }).session(); + await session.send('hi'); + const before = session.messages; + + const run = session.stream('wait for it'); + for await (const event of run) { + if (event.type === 'tool.start') break; + } + + expect((await run.result).finishReason).toBe('aborted'); + expect(session.messages).toEqual(before); + expect((await session.send('next')).text).toBe('never sent'); + }); + + it('leaves the transcript unchanged and reports a failed run', async () => { + const model = mockModel(['hello', { error: new Error('provider down') }, 'back']); + const session = createAgent({ provider: model }).session(); + await session.send('hi'); + const before = session.messages; + + const run = session.stream('again'); + const events = await collect(run); + + await expect(run.result).rejects.toThrow('provider down'); + expect(events.filter((e) => e.type === 'error')).toHaveLength(1); + expect(events.at(-1)).toMatchObject({ type: 'run.done', finishReason: 'error' }); + expect(session.messages).toEqual(before); + expect((await session.send('hello?')).text).toBe('back'); + }); + + it('ends with error and run.done when the history cannot be loaded or the turn cannot be saved', async () => { + const memory = new MemorySessionStore(); + const broken: SessionStore = { + load: () => Promise.reject(new Error('load failed')), + save: (id, messages) => memory.save(id, messages), + delete: (id) => memory.delete(id), + }; + const loadFails = createAgent({ provider: mockModel(['x']) }).session({ store: broken }); + const run = loadFails.stream('hi'); + const events = await collect(run); + await expect(run.result).rejects.toThrow('load failed'); + expect(events.map((e) => [e.type, e.seq])).toEqual([ + ['error', 0], + ['run.done', 1], + ]); + + const saveFails = createAgent({ provider: mockModel(['x']) }).session({ + store: { ...memory, load: (id) => memory.load(id), save: () => Promise.reject(new Error('disk full')), delete: (id) => memory.delete(id) }, + }); + const saving = saveFails.stream('hi'); + const savingEvents = await collect(saving); + await expect(saving.result).rejects.toThrow('disk full'); + expect(savingEvents.slice(-2).map((e) => e.type)).toEqual(['error', 'run.done']); + expect(savingEvents.at(-1)).toMatchObject({ finishReason: 'error' }); + expect(savingEvents.map((e) => e.seq)).toEqual(savingEvents.map((_, i) => i)); + }); + + it('saves a paused turn and resumes it with approvals.resolve()', async () => { + const execute = vi.fn(); + const tool = defineTool({ + name: 'send_email', + description: 'Sends an email', + input: z.object({ to: z.string() }), + needsApproval: true, + execute: async ({ to }) => { + execute(to); + return `sent to ${to}`; + }, + }); + const model = mockModel([callEmail, 'Email sent.', 'You asked me to email Sam.']); + const agent = createAgent({ provider: model, tools: [tool] }); + const session = agent.session(); + + const run = session.stream('Email Sam'); + const events = await collect(run); + const paused = await run.result; + + expect(paused.finishReason).toBe('awaiting-approval'); + expect(events.map((e) => e.type)).toContain('approval.requested'); + expect(events.at(-1)).toMatchObject({ type: 'run.done', finishReason: 'awaiting-approval' }); + expect(execute).not.toHaveBeenCalled(); + expect(roles(session.messages)).toEqual(['user']); + + await agent.approvals.resolve({ id: paused.approvalId!, approved: true }); + + expect(execute).toHaveBeenCalledWith('sam@example.com'); + expect(roles(session.messages)).toEqual(['user', 'assistant', 'tool', 'assistant']); + const followUp = await collect(session.stream('What did I ask?')); + expect(followUp.at(-1)).toMatchObject({ type: 'run.done', text: 'You asked me to email Sam.' }); + expect(model.calls[2].messages.map((m) => m.content)).toContain('Email sent.'); + }); + + it('binds a pause to its session as soon as approval.requested is seen', async () => { + const model = mockModel([callEmail, 'Email sent.']); + const agent = createAgent({ provider: model, tools: [emailTool()] }); + const session = agent.session(); + + let resolved: Promise | undefined; + for await (const event of session.stream('Email Sam')) { + if (event.type === 'approval.requested') resolved = agent.approvals.resolve({ id: event.approvalId, approved: true }); + } + await resolved; + + expect(roles(session.messages)).toEqual(['user', 'assistant', 'tool', 'assistant']); + }); +}); + +function convoTexts(messages: readonly Message[]): unknown[] { + return messages.filter((m) => m.role !== 'system').map((m) => m.content); +} diff --git a/src/session/sessionStream.ts b/src/session/sessionStream.ts new file mode 100644 index 000000000..51c410456 --- /dev/null +++ b/src/session/sessionStream.ts @@ -0,0 +1,109 @@ +/** + * LOU-V8: the `AgentRun` that `session.stream()` returns. A session has to + * queue behind earlier calls and load its history before it can start a turn, + * but a run handle must exist at once, so this wraps the run the session + * starts when its turn comes: same events, but under its own `runId`, and + * `run.done` is delivered only after the turn has been persisted. + */ + +import { nanoid } from 'nanoid'; +import type { ExecutionResult } from '../execution/AgentExecutor'; +import { AGENT_EVENT_SCHEMA_VERSION, type AgentEvent, type AgentEventPayload } from '../execution/agentEvents'; +import type { AgentRun } from '../execution/agentRun'; + +/** + * Runs one session turn. Call `started` with the run as soon as it exists; + * resolve with the turn's result once it is recorded (or reject). + */ +export type SessionTurn = (signal: AbortSignal, started: (run: AgentRun) => void) => Promise; + +class SessionRun implements AgentRun { + readonly runId = nanoid(); + readonly result: Promise; + private readonly controller = new AbortController(); + private readonly started: Promise; + private iterated = false; + + constructor(turn: SessionTurn, signal: AbortSignal | undefined) { + let onStarted!: (run: AgentRun) => void; + let onFailed!: (error: unknown) => void; + this.started = new Promise((resolve, reject) => { + onStarted = resolve; + onFailed = reject; + }); + this.started.catch(() => undefined); + const runSignal = signal ? AbortSignal.any([signal, this.controller.signal]) : this.controller.signal; + this.result = turn(runSignal, onStarted).catch((error: unknown) => { + onFailed(error); + throw 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 session.stream() again for a new run.' + ); + } + this.iterated = true; + let finished = false; + try { + let run: AgentRun; + try { + run = await this.started; + } catch (error) { + finished = true; + yield* this.failure(error, 0); + return; + } + for await (const event of run) { + if (event.type === 'run.done') { + finished = true; + try { + await this.result; + } catch (error) { + // An error that failed the run itself was already reported by its events. + if (event.finishReason !== 'error') { + yield* this.failure(error, event.seq); + return; + } + } + } + yield { ...event, runId: this.runId } as AgentEvent; + } + } finally { + if (!finished) { + this.controller.abort( + new DOMException('The AgentRun was not iterated to the end (the for-await loop exited early)', 'AbortError') + ); + } + } + } + + /** `error` then `run.done`, for a turn that failed outside the run's own events. */ + private *failure(error: unknown, seq: number): Generator { + const { name, message } = error as { name?: unknown; message?: unknown }; + yield this.event( + { + type: 'error', + error: { + name: typeof name === 'string' && name ? name : 'Error', + message: typeof message === 'string' ? message : String(error), + }, + }, + seq + ); + yield this.event({ type: 'run.done', finishReason: 'error', text: '' }, seq + 1); + } + + private event(payload: AgentEventPayload, seq: number): AgentEvent { + return { ...payload, runId: this.runId, seq, timestamp: new Date().toISOString(), v: AGENT_EVENT_SCHEMA_VERSION } as AgentEvent; + } +} + +/** Starts `turn` and returns its handle; `signal` (the caller's) and an early `break` abort the turn. */ +export function streamSessionTurn(turn: SessionTurn, signal?: AbortSignal): AgentRun { + return new SessionRun(turn, signal); +}