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 e5174131f..9362bd07a 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); +}