[EPIC-V][LOU-V8] AgentSession.stream() streams a session turn and persists it on done

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

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:

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:

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 });

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;
     },

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<ExecutionResult>;
 
+/**
+ * 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<unknown> = 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<void> {
     return this.enqueue(async () => {

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';