From ae7a8d0b4fe44a571cb70d465da2d48350876854 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 14:33:59 +0000 Subject: [PATCH] [LOU-W9] Sessions checkpoint every step and resume a crashed turn agent.session({ id, store, checkpointStore }) (or a { sessions, checkpoints } store such as SqliteStore) runs each turn with sessionId '.turn-' and that CheckpointStore, so the turn is checkpointed after every model response and tool result. session.resume() finishes an interrupted turn through the executor's resume path (input: []) without re-running recorded steps, adds it to the transcript as send() would and returns its result (null when nothing is pending); pending() reports an unfinished turn and discardPending() drops it. send()/stream() resume a pending turn first. A turn paused for approval stays in its checkpoint until it finishes; resume()/send() throw SessionAwaitingApprovalError and agent.approvals.resolve() continues the turn in the session, passing the session's checkpointStore to resumeAfterApproval(). Docs: "Durable sessions" in docs/sessions.md, cross-link from docs/durable-execution.md, shorter README SQLite example, CHANGELOG, llms. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016bY29ibQgv3MDdjAbUT4YR --- CHANGELOG.md | 1 + README.md | 23 +-- docs/durable-execution.md | 4 + docs/sessions.md | 59 +++++++- llms-full.txt | 86 ++++++++--- src/createAgent.ts | 20 ++- src/createAgentApprovals.ts | 41 ++++-- src/session/AgentSession.ts | 162 +++++++++++++++++++-- src/session/durableSession.test.ts | 226 +++++++++++++++++++++++++++++ src/session/index.ts | 9 +- 10 files changed, 559 insertions(+), 72 deletions(-) create mode 100644 src/session/durableSession.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index f5719e15f..03febc1c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] - 2026-09-28 ### Added +- Durable sessions (LOU-W9): `agent.session({ id, store, checkpointStore })` (or a `store` that carries both, `{ sessions, checkpoints }`, such as a `SqliteStore`) runs each turn with `sessionId: '.turn-'` and that `CheckpointStore`, so it is checkpointed after every model response and tool result. `session.resume()` finishes a turn interrupted by a crash, a failed checkpoint write or a `PropagatingToolError` (also in a new process) without re-running recorded tool calls or model responses, adds it to the transcript as `send()` would and returns its result (`null` when nothing is pending); `session.pending()` reports an unfinished turn and `session.discardPending()` drops it. `send()` / `stream()` resume a pending turn before sending the new message. A turn paused for approval stays in its checkpoint until it finishes: `resume()` / `send()` throw `SessionAwaitingApprovalError`, and `agent.approvals.resolve()` continues it in the session. Sessions without a checkpoint store are unchanged. See docs/sessions.md. - React hook (LOU-D15): new `@loushy/build-ai-agent/react` subpath with `useLoushyAgent(source, options?)`. `source` is `{ agent, sessionId? }` (in process, via `agent.stream()` or `agent.session({ id }).stream()`) or `{ url, headers?, fetch? }` (POSTs `{ input }` and reads the SSE or NDJSON event stream). It returns `messages` (text and tool calls with status, args and result), `status` (`idle`/`streaming`/`awaiting-approval`/`error`), `pendingApproval`, `error`, `usage`, `lastEvent`, `send()`, `stop()` (aborts; unmount aborts too), `approve(note?)` and `reject(note?)` (via `agent.approvals.resolve()` in process, or `POST ${approvalsUrl}/${id}` remotely). The framework-neutral `reduceAgentEvents()` reducer and `parseEventStream()` parser are exported from the same subpath. `react` (`^18 || ^19`) is an optional peer dependency. See docs/react.md. - Summarizing compaction (LOU-W3): `summarizeStrategy({ model, protectedTokens?, prompt?, maxSummaryTokens? })` replaces the turns before the protected tail with one `[Conversation summary]` user message written by `model` (an `LLMProvider` or a `"provider/model"` spec), keeping system and pinned messages and each assistant turn together with its tool results; if the summary call fails it falls back to pruning and reports `error`. `twoPhaseStrategy()` (the recommended strategy) prunes first and summarizes only when still above the threshold. `pinMessage()` / `isPinned()` and the new optional `Message.metadata` (never sent to providers) mark messages no built-in strategy compacts. `CompactionStrategy.compact()` may now be async, so `compactMessages()` returns a Promise; `CompactionResult` / `CompactionInfo` gain `summary` and `error`, and `CompactionInput` gains `thresholdTokens` and `signal`. A strategy that throws no longer fails the run: the hook reports the `error` to `onCompaction`. See docs/compaction.md. - `createAgent({ retry, fallbackModels })` (LOU-V7.2): `retry` (`withRetry()` options, or `false`) retries failed model calls and defaults to `{ maxRetries: 2 }` for models given as `provider/model` strings (or picked from the environment); a `provider` instance is wrapped only when `retry` is set. `fallbackModels` are `provider/model` strings tried in order once the primary model's retries are used up (`withFallback([withRetry(primary), withRetry(fallback), ...])`). `agent.stream()`, `session.stream()` and `AgentExecutor.stream()` report them as two new typed events, `provider.retry` (`attempt`, `maxRetries`, `delayMs`, `error: { message, category? }`, `provider`) and `provider.fallback` (`from`, `to`, `error: { message }`), for any `withRetry()` / `withFallback()` provider. Additive: the event schema version stays 1. diff --git a/README.md b/README.md index e9a153562..b981bb46b 100644 --- a/README.md +++ b/README.md @@ -145,24 +145,17 @@ if (run.finishReason === 'awaiting-approval') { ### Durable sessions with SQLite ```ts -import { AgentBuilder, AgentExecutor, createAgent, resolveProvider } from '@loushy/build-ai-agent'; +import { createAgent, resolveProvider } from '@loushy/build-ai-agent'; import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; const store = new SqliteStore('./.loushy/agent.db'); // sessions, checkpoints and approvals in one file -const provider = resolveProvider('openai/gpt-4o-mini'); - -// A conversation that survives restarts: same id + same file = same conversation. -const agent = createAgent({ provider, approvalStore: store.approvals }); -await agent.session({ id: 'user-42', store: store.sessions }).send('My name is Ali.'); -const { text } = await agent.session({ id: 'user-42', store: store.sessions }).send('What is my name?'); // "Ali" - -// A crash-safe run: checkpointed after every model response and tool result. -// Calling execute() again with the same sessionId resumes it instead of starting over. -const booker = AgentBuilder.create().setName('booker').setPrompt('You book restaurant tables.').build(); -await AgentExecutor.execute({ - agent: booker, input: 'Book a table for 2 tonight', provider, - sessionId: 'booking-7', checkpointStore: store.checkpoints, -}); +const agent = createAgent({ provider: resolveProvider('openai/gpt-4o-mini'), approvalStore: store.approvals }); + +// Same id + same file = same conversation, checkpointed after every model response and tool result. +await agent.session({ id: 'user-42', store }).send('My name is Ali.'); +const session = agent.session({ id: 'user-42', store }); +await session.resume(); // after a crash: finishes the interrupted turn without redoing finished steps +const { text } = await session.send('What is my name?'); // "Ali" ``` ## Documentation diff --git a/docs/durable-execution.md b/docs/durable-execution.md index 860c4a628..2e07fbdf4 100644 --- a/docs/durable-execution.md +++ b/docs/durable-execution.md @@ -19,6 +19,10 @@ await AgentExecutor.execute({ agent, input: 'Book a table for 2 tonight', provid await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); ``` +With `createAgent()`, a session does this for you: `agent.session({ id, store, checkpointStore })` +checkpoints every turn and `session.resume()` finishes an interrupted one - see +[Durable sessions](./sessions.md#durable-sessions). + ## What is checkpointed, and when A checkpoint is the run's whole transcript plus its step count, usage, diff --git a/docs/sessions.md b/docs/sessions.md index 0d07b1e96..d291b2d6a 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -26,6 +26,9 @@ console.log(session.id, session.messages.length); | `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. | +| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId? }`), or `null`. | +| `resume({ signal })` | In a durable session, finishes an interrupted turn and resolves with its result, or `null` when none is pending. | +| `discardPending()` | In a durable session, drops an unfinished turn without running it. | Concurrent `send()` calls on one session are queued and run one after another in call order, so the transcript never interleaves. @@ -81,11 +84,55 @@ const { text } = await session.send('What is my name?'); 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 -mechanism also continues a finished conversation, and additionally resumes a -run interrupted mid-turn (a crash, an abort, an approval pause) without -re-running finished tools - see [Durable execution](./durable-execution.md). +executor on each turn. Without a checkpoint store, a turn is saved only when it +ends, so a crash in the middle of a turn loses it; see the next section. + +## Durable sessions + +Give the session a `CheckpointStore` and every turn is checkpointed after each +model response and tool result, using the [durable execution](./durable-execution.md) +mechanism. A turn interrupted by a crash, a failed checkpoint write or a +`PropagatingToolError` can then be finished later, in another process: + +```ts +import { createAgent, FileSessionStore, type CheckpointStore } from '@loushy/build-ai-agent'; + +declare const checkpointStore: CheckpointStore; // e.g. LocalStorageCheckpointStore, or SqliteStore's `checkpoints` + +const agent = createAgent({ provider }); +const store = new FileSessionStore('./.loushy/sessions'); + +// After a restart: finish the turn that was running, if any. +const session = agent.session({ id: 'user-42', store, checkpointStore }); +const finished = await session.resume(); // ExecutionResult, or null when nothing was pending +console.log(finished?.text, await session.pending()); // pending() is null now +``` + +- Pass `checkpointStore`, or a `store` that carries both stores: any + `{ sessions, checkpoints }` object, such as a `SqliteStore` + (`agent.session({ id, store: sqliteStore })`). +- Each turn runs with `sessionId: '.turn-'` (`n` is the length + of the transcript when the turn started), so a new process finds the + interrupted turn without any extra bookkeeping. A finished turn joins the + transcript, exactly as a plain `send()` does, and its checkpoint is deleted. +- `resume()` continues the turn through the executor's resume path + (`input: []`): tool calls whose results were recorded do not run again, and + a recorded model response is not requested again. A tool that was running + when the process died does run again (at-least-once, see + [Durable execution](./durable-execution.md#at-least-once-tools-make-side-effects-idempotent)). +- `send()` and `stream()` resume a pending turn first, then send the new + message, so the model sees the finished turn (the resumed turn's events are + not streamed). Use `pending()` to check first, or `discardPending()` to drop + the unfinished turn instead. +- A turn that pauses on a `needsApproval` tool stays in its checkpoint, not in + the transcript, until it finishes. While it waits, `resume()`, `send()` and + `stream()` throw `SessionAwaitingApprovalError` (with its `approvalId`), and + `agent.approvals.resolve({ id, approved })` continues the turn in this + session. After a restart, open the session and call `resume()` (or `send()`) + once before resolving, so the agent knows which session the approval belongs + to; give both agents the same durable `approvalStore`. +- An aborted turn is dropped (its checkpoint is deleted), as without a + checkpoint store. `clear()` deletes a pending turn too. ## Stores @@ -146,7 +193,7 @@ import { createAgent, AgentExecutor } from '@loushy/build-ai-agent'; import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; const store = new SqliteStore('./.loushy/agent.db'); // or ':memory:' -const session = createAgent({ provider }).session({ id: 'user-42', store: store.sessions }); +const session = createAgent({ provider }).session({ id: 'user-42', store }); // transcript + per-step checkpoints await session.send('Hello'); // Same file, same store, for durable runs with approvals: diff --git a/llms-full.txt b/llms-full.txt index 368704a29..c237746c0 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -144,24 +144,17 @@ if (run.finishReason === 'awaiting-approval') { ### Durable sessions with SQLite ```ts -import { AgentBuilder, AgentExecutor, createAgent, resolveProvider } from '@loushy/build-ai-agent'; +import { createAgent, resolveProvider } from '@loushy/build-ai-agent'; import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; const store = new SqliteStore('./.loushy/agent.db'); // sessions, checkpoints and approvals in one file -const provider = resolveProvider('openai/gpt-4o-mini'); - -// A conversation that survives restarts: same id + same file = same conversation. -const agent = createAgent({ provider, approvalStore: store.approvals }); -await agent.session({ id: 'user-42', store: store.sessions }).send('My name is Ali.'); -const { text } = await agent.session({ id: 'user-42', store: store.sessions }).send('What is my name?'); // "Ali" - -// A crash-safe run: checkpointed after every model response and tool result. -// Calling execute() again with the same sessionId resumes it instead of starting over. -const booker = AgentBuilder.create().setName('booker').setPrompt('You book restaurant tables.').build(); -await AgentExecutor.execute({ - agent: booker, input: 'Book a table for 2 tonight', provider, - sessionId: 'booking-7', checkpointStore: store.checkpoints, -}); +const agent = createAgent({ provider: resolveProvider('openai/gpt-4o-mini'), approvalStore: store.approvals }); + +// Same id + same file = same conversation, checkpointed after every model response and tool result. +await agent.session({ id: 'user-42', store }).send('My name is Ali.'); +const session = agent.session({ id: 'user-42', store }); +await session.resume(); // after a crash: finishes the interrupted turn without redoing finished steps +const { text } = await session.send('What is my name?'); // "Ali" ``` ## Documentation @@ -2743,6 +2736,10 @@ await AgentExecutor.execute({ agent, input: 'Book a table for 2 tonight', provid await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); ``` +With `createAgent()`, a session does this for you: `agent.session({ id, store, checkpointStore })` +checkpoints every turn and `session.resume()` finishes an interrupted one - see +[Durable sessions](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/sessions.md#durable-sessions). + ## What is checkpointed, and when A checkpoint is the run's whole transcript plus its step count, usage, @@ -3793,6 +3790,9 @@ console.log(session.id, session.messages.length); | `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. | +| `pending()` | In a [durable session](#durable-sessions), the turn that has not finished (`{ status, approvalId? }`), or `null`. | +| `resume({ signal })` | In a durable session, finishes an interrupted turn and resolves with its result, or `null` when none is pending. | +| `discardPending()` | In a durable session, drops an unfinished turn without running it. | Concurrent `send()` calls on one session are queued and run one after another in call order, so the transcript never interleaves. @@ -3848,11 +3848,55 @@ const { text } = await session.send('What is my name?'); 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 -mechanism also continues a finished conversation, and additionally resumes a -run interrupted mid-turn (a crash, an abort, an approval pause) without -re-running finished tools - see [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md). +executor on each turn. Without a checkpoint store, a turn is saved only when it +ends, so a crash in the middle of a turn loses it; see the next section. + +## Durable sessions + +Give the session a `CheckpointStore` and every turn is checkpointed after each +model response and tool result, using the [durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md) +mechanism. A turn interrupted by a crash, a failed checkpoint write or a +`PropagatingToolError` can then be finished later, in another process: + +```ts +import { createAgent, FileSessionStore, type CheckpointStore } from '@loushy/build-ai-agent'; + +declare const checkpointStore: CheckpointStore; // e.g. LocalStorageCheckpointStore, or SqliteStore's `checkpoints` + +const agent = createAgent({ provider }); +const store = new FileSessionStore('./.loushy/sessions'); + +// After a restart: finish the turn that was running, if any. +const session = agent.session({ id: 'user-42', store, checkpointStore }); +const finished = await session.resume(); // ExecutionResult, or null when nothing was pending +console.log(finished?.text, await session.pending()); // pending() is null now +``` + +- Pass `checkpointStore`, or a `store` that carries both stores: any + `{ sessions, checkpoints }` object, such as a `SqliteStore` + (`agent.session({ id, store: sqliteStore })`). +- Each turn runs with `sessionId: '.turn-'` (`n` is the length + of the transcript when the turn started), so a new process finds the + interrupted turn without any extra bookkeeping. A finished turn joins the + transcript, exactly as a plain `send()` does, and its checkpoint is deleted. +- `resume()` continues the turn through the executor's resume path + (`input: []`): tool calls whose results were recorded do not run again, and + a recorded model response is not requested again. A tool that was running + when the process died does run again (at-least-once, see + [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#at-least-once-tools-make-side-effects-idempotent)). +- `send()` and `stream()` resume a pending turn first, then send the new + message, so the model sees the finished turn (the resumed turn's events are + not streamed). Use `pending()` to check first, or `discardPending()` to drop + the unfinished turn instead. +- A turn that pauses on a `needsApproval` tool stays in its checkpoint, not in + the transcript, until it finishes. While it waits, `resume()`, `send()` and + `stream()` throw `SessionAwaitingApprovalError` (with its `approvalId`), and + `agent.approvals.resolve({ id, approved })` continues the turn in this + session. After a restart, open the session and call `resume()` (or `send()`) + once before resolving, so the agent knows which session the approval belongs + to; give both agents the same durable `approvalStore`. +- An aborted turn is dropped (its checkpoint is deleted), as without a + checkpoint store. `clear()` deletes a pending turn too. ## Stores @@ -3913,7 +3957,7 @@ import { createAgent, AgentExecutor } from '@loushy/build-ai-agent'; import { SqliteStore } from '@loushy/build-ai-agent/sqlite'; const store = new SqliteStore('./.loushy/agent.db'); // or ':memory:' -const session = createAgent({ provider }).session({ id: 'user-42', store: store.sessions }); +const session = createAgent({ provider }).session({ id: 'user-42', store }); // transcript + per-step checkpoints await session.send('Hello'); // Same file, same store, for durable runs with approvals: diff --git a/src/createAgent.ts b/src/createAgent.ts index c4da18509..e12127701 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -336,12 +336,15 @@ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { const approvals = createAgentApprovals({ store: config.approvalStore ?? new InMemoryApprovalStore(), approve: config.approve, - resume: (approvalStore, decision, signal) => - resumeAfterApproval(decision, approvalStore, toolRegistry ?? new ToolRegistry(), provider, { - ...runOptions, + resume: (approvalStore, decision, signal, checkpointStore) => + resumeAfterApproval( + decision, approvalStore, - signal, - }), + toolRegistry ?? new ToolRegistry(), + provider, + { ...runOptions, approvalStore, signal }, + checkpointStore + ), }); const executeOptions = (input: string | Message[], signal?: AbortSignal): ExecuteOptions => ({ ...spec, @@ -359,8 +362,13 @@ export function createAgent(config: CreateAgentConfig = {}): SimpleAgent { stream(message: string, options: SendOptions = {}): AgentRun { return AgentExecutor.stream(executeOptions(message, options.signal)); }, + // LOU-W9: a checkpointed session's turn runs under its own sessionId + checkpointStore. session: (options?: SessionOptions) => - approvals.session(run, (input, signal) => AgentExecutor.stream(executeOptions(input, signal)), options), + approvals.session( + (input, signal, turn) => AgentExecutor.execute({ ...executeOptions(input, signal), ...turn }), + (input, signal, turn) => AgentExecutor.stream({ ...executeOptions(input, signal), ...turn }), + options + ), approvals: approvals.approvals, }; registerSubagent(simpleAgent, { spec, description: config.description }); diff --git a/src/createAgentApprovals.ts b/src/createAgentApprovals.ts index 91564bffd..22759dd69 100644 --- a/src/createAgentApprovals.ts +++ b/src/createAgentApprovals.ts @@ -9,6 +9,8 @@ import type { ApprovalDecision, ApprovalStore, PendingApproval } from './execution/ApprovalGate'; import type { ExecutionResult } from './execution/AgentExecutor'; import type { AgentRun } from './execution/agentRun'; +import type { CheckpointStore } from './execution/checkpoint'; +import { SessionAwaitingApprovalError } from './execution/errors'; import { AgentSession, type SessionOptions, type SessionRunner, type SessionStreamRunner } from './session/AgentSession'; /** @@ -41,13 +43,21 @@ export interface AgentApprovals { resolve(decision: ApprovalDecision, options?: { signal?: AbortSignal }): Promise; } -/** resumeAfterApproval() bound to an agent's registry, provider and options. */ -type ResumeRun = (store: ApprovalStore, decision: ApprovalDecision, signal?: AbortSignal) => Promise; +/** + * resumeAfterApproval() bound to an agent's registry, provider and options; + * `checkpointStore` is the paused session's, when its turns are checkpointed. + */ +type ResumeRun = ( + store: ApprovalStore, + decision: ApprovalDecision, + signal?: AbortSignal, + checkpointStore?: CheckpointStore +) => Promise; /** A session whose paused turn can be continued by `agent.approvals.resolve()`. */ class ApprovalSession extends AgentSession { - resolveWith(next: () => Promise): Promise { - return this.continueTurn(next); + resolveWith(next: (checkpointStore?: CheckpointStore) => Promise): Promise { + return this.continueTurn(() => next(this.checkpointStore)); } } @@ -69,12 +79,16 @@ export function createAgentApprovals(options: { store: ApprovalStore; approve?: }; /** With an `approve` callback, decides every pause until the run finishes. */ - async function settle(result: ExecutionResult, signal?: AbortSignal): Promise { + async function settle( + result: ExecutionResult, + signal?: AbortSignal, + checkpointStore?: CheckpointStore + ): Promise { let current = result; for (;;) { const request = current.approvalId ? pending.get(current.approvalId) : undefined; if (!approve || current.finishReason !== 'awaiting-approval' || !request) return current; - current = await resume(store, { id: request.id, approved: await approve(request) }, signal); + current = await resume(store, { id: request.id, approved: await approve(request) }, signal, checkpointStore); } } @@ -104,7 +118,8 @@ export function createAgentApprovals(options: { store: ApprovalStore; approve?: resolve(decision, { signal } = {}) { const session = sessions.get(decision.id); sessions.delete(decision.id); - const next = async () => inSession(session, await settle(await resume(store, decision, signal), signal)); + const next = async (checkpointStore?: CheckpointStore) => + inSession(session, await settle(await resume(store, decision, signal, checkpointStore), signal, checkpointStore)); return session ? session.resolveWith(next) : next(); }, }; @@ -116,9 +131,17 @@ export function createAgentApprovals(options: { store: ApprovalStore; approve?: settle, 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)), + async (input, signal, turn) => { + try { + return inSession(session, await settle(await run(input, signal, turn), signal, turn?.checkpointStore)); + } catch (error) { + // A checkpointed turn found paused (e.g. after a restart): resolving it continues this session. + if (error instanceof SessionAwaitingApprovalError && error.approvalId) sessions.set(error.approvalId, session); + throw error; + } + }, sessionOptions, - (input, signal) => inSessionRun(session, stream(input, signal)) + (input, signal, turn) => inSessionRun(session, stream(input, signal, turn)) ); return session; }, diff --git a/src/session/AgentSession.ts b/src/session/AgentSession.ts index afb5c4c5d..d5202d2ca 100644 --- a/src/session/AgentSession.ts +++ b/src/session/AgentSession.ts @@ -1,13 +1,16 @@ /** * AgentSession (LOU-W4): a multi-turn conversation on top of the stateless * executor. The session owns the transcript and passes it in on every - * `send()`; it does not use checkpoints or resume. + * `send()`. With a checkpoint store (LOU-W9), each turn also runs as a + * durable-execution session, so a turn interrupted mid-way can be resumed. */ import { randomUUID } from 'node:crypto'; import type { Message } from '../providers/llm'; import type { ExecutionResult } from '../execution/AgentExecutor'; import type { AgentRun } from '../execution/agentRun'; +import type { Checkpoint, CheckpointStore } from '../execution/checkpoint'; +import { SessionAwaitingApprovalError } from '../execution/errors'; import { streamSessionTurn } from './sessionStream'; import { MemorySessionStore, assertSessionId, type SessionStore } from './sessionStore'; @@ -18,21 +21,57 @@ export interface SessionOptions { * UUID. Passing an id that exists in `store` continues that conversation. */ id?: string; - /** Where the transcript is kept. Defaults to a new `MemorySessionStore`. */ - store?: SessionStore; + /** + * Where the transcript is kept. Defaults to a new `MemorySessionStore`. + * A `{ sessions, checkpoints }` object (such as a `SqliteStore`) keeps the + * transcript in `sessions` and checkpoints each turn in `checkpoints`. + */ + store?: SessionStore | SessionStores; + /** + * Checkpoints every turn after each model response and tool result + * (LOU-W9), so `resume()` can finish a turn that was interrupted (see + * docs/sessions.md, "Durable sessions"). Overrides `store.checkpoints`. + */ + checkpointStore?: CheckpointStore; +} + +/** A transcript store plus, optionally, a checkpoint store (e.g. a `SqliteStore`). */ +export interface SessionStores { + sessions: SessionStore; + checkpoints?: CheckpointStore; +} + +/** A checkpointed session's turn that has not finished (see `session.pending()`). */ +export interface PendingTurn { + /** `'in-progress'`: interrupted, `resume()` continues it. `'awaiting-approval'`: decide `approvalId` first. */ + status: 'in-progress' | 'awaiting-approval'; + approvalId?: string; +} + +/** Where a checkpointed session's turn is checkpointed: pass both to `AgentExecutor.execute()`. */ +export interface SessionTurnCheckpoint { + /** `.turn-`, `n` being the transcript length when the turn started. */ + sessionId: string; + checkpointStore: CheckpointStore; } /** * Runs one turn: `input` is the whole transcript so far, ending in the new - * user message. Supplied by `createAgent()`. + * user message (or `[]` to resume the checkpointed turn). With `checkpoint`, + * the turn must run with its `sessionId` and `checkpointStore`. Supplied by + * `createAgent()`. */ -export type SessionRunner = (input: Message[], signal?: AbortSignal) => Promise; +export type SessionRunner = ( + input: Message[], + signal?: AbortSignal, + checkpoint?: SessionTurnCheckpoint +) => 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; +export type SessionStreamRunner = (input: Message[], signal?: AbortSignal, checkpoint?: SessionTurnCheckpoint) => AgentRun; /** * Longest prefix of `messages` that a provider accepts: every assistant @@ -75,6 +114,8 @@ export function providerValidPrefix(messages: readonly Message[]): Message[] { export class AgentSession { readonly id: string; private readonly store: SessionStore; + /** Set when every turn is checkpointed (LOU-W9). */ + protected readonly checkpointStore: CheckpointStore | undefined; private readonly run: SessionRunner; private readonly streamRun: SessionStreamRunner | undefined; private history: Message[] = []; @@ -84,7 +125,10 @@ export class AgentSession { 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(); + const { store, checkpointStore } = options; + const stores = store && typeof (store as SessionStore).load !== 'function' ? (store as SessionStores) : undefined; + this.store = stores ? stores.sessions : ((store as SessionStore | undefined) ?? new MemorySessionStore()); + this.checkpointStore = checkpointStore ?? stores?.checkpoints; this.run = run; this.streamRun = streamRun; } @@ -110,6 +154,7 @@ export class AgentSession { * * A call that throws or is aborted leaves the transcript as it was before * the call (an aborted call resolves with `finishReason: 'aborted'`). + * In a checkpointed session, a pending turn is resumed first (see `resume()`). * * @example * ```ts @@ -148,8 +193,8 @@ export class AgentSession { return streamSessionTurn( (signal, started) => this.enqueue(async () => { - await this.ensureLoaded(); - const run = streamRun([...this.history, { role: 'user', content: input }], signal); + await this.beforeTurn(signal); + const run = streamRun([...this.history, { role: 'user', content: input }], signal, this.turnCheckpoint()); started(run); return this.record(await run.result); }), @@ -157,12 +202,56 @@ export class AgentSession { ); } - /** Forget the conversation (also deletes it from the store). */ + /** + * The checkpointed turn that has not finished, or `null` (always `null` + * without a checkpoint store). See docs/sessions.md, "Durable sessions". + */ + pending(): Promise { + return this.enqueue(async () => { + const checkpoint = await this.pendingCheckpoint(); + if (!checkpoint) return null; + const status = checkpoint.status === 'awaiting-approval' ? 'awaiting-approval' : 'in-progress'; + return { status, approvalId: checkpoint.approvalId }; + }); + } + + /** + * Finishes a checkpointed turn that was interrupted (a crash, a failed + * checkpoint write, a `PropagatingToolError`), without running recorded + * tool calls or model responses again. The turn joins the transcript as + * with `send()` and its result is returned; `null` when no turn is + * pending. Throws `SessionAwaitingApprovalError` when the turn waits on an + * approval: decide it with `agent.approvals.resolve()`. + * + * @example + * ```ts + * const session = agent.session({ id: 'user-42', store }); + * const finished = await session.resume(); // null when nothing was interrupted + * ``` + */ + resume(options: { signal?: AbortSignal } = {}): Promise { + return this.enqueue(() => this.resumePending(options.signal)); + } + + /** Drops a pending turn without finishing it; the transcript stays as it was before that turn. */ + discardPending(): Promise { + return this.enqueue(async () => { + if (await this.pendingCheckpoint()) await this.deleteCheckpoint(); + }); + } + + /** Forget the conversation (also deletes it, and a pending turn, from the store). */ clear(): Promise { return this.enqueue(async () => { + if (this.checkpointStore) { + await this.ensureLoaded(); + await this.deleteCheckpoint(); + } await this.store.delete(this.id); this.history = []; this.loaded = true; + // A first turn that finished just before a crash, so was never adopted. + await this.deleteCheckpoint(); }); } @@ -179,8 +268,42 @@ export class AgentSession { } private async turn(input: string, signal?: AbortSignal): Promise { + await this.beforeTurn(signal); + return this.record(await this.run([...this.history, { role: 'user', content: input }], signal, this.turnCheckpoint())); + } + + /** Loads the history and, in a checkpointed session, finishes a pending turn first. */ + private async beforeTurn(signal?: AbortSignal): Promise { + const resumed = await this.resumePending(signal); + if (resumed?.finishReason === 'awaiting-approval') { + throw new SessionAwaitingApprovalError(this.turnCheckpoint()?.sessionId ?? this.id, resumed.approvalId); + } + } + + /** Where the next turn is checkpointed: keyed by the transcript length, so it is found again after a restart. */ + private turnCheckpoint(): SessionTurnCheckpoint | undefined { + const { checkpointStore } = this; + return checkpointStore && { sessionId: `${this.id}.turn-${this.history.length}`, checkpointStore }; + } + + private async deleteCheckpoint(): Promise { + const turn = this.turnCheckpoint(); + await turn?.checkpointStore.delete(turn.sessionId); + } + + /** The pending turn's checkpoint. One that finished just before a crash is added to the transcript instead. */ + private async pendingCheckpoint(): Promise { await this.ensureLoaded(); - return this.record(await this.run([...this.history, { role: 'user', content: input }], signal)); + const turn = this.turnCheckpoint(); + const checkpoint = turn ? await turn.checkpointStore.load(turn.sessionId) : null; + if (checkpoint?.status !== 'finished') return checkpoint; + await this.commit(checkpoint.messages); + return null; + } + + private async resumePending(signal?: AbortSignal): Promise { + if (!(await this.pendingCheckpoint())) return null; + return this.record(await this.run([], signal, this.turnCheckpoint())); } /** @@ -196,11 +319,22 @@ export class AgentSession { } private async record(result: ExecutionResult): Promise { - if (result.finishReason === 'aborted') return result; - const withoutSystem = result.messages[0]?.role === 'system' ? result.messages.slice(1) : result.messages; + if (result.finishReason === 'aborted') { + await this.deleteCheckpoint(); + return result; + } + // A checkpointed turn joins the transcript once, when it finishes; until then its checkpoint holds it. + if (!this.checkpointStore || result.finishReason !== 'awaiting-approval') await this.commit(result.messages); + return result; + } + + /** Saves `messages` (minus the system prompt) as the transcript, then drops the finished turn's checkpoint. */ + private async commit(messages: readonly Message[]): Promise { + const withoutSystem = messages[0]?.role === 'system' ? messages.slice(1) : messages; const next = providerValidPrefix(withoutSystem); await this.store.save(this.id, next); + const turn = this.turnCheckpoint(); this.history = next; - return result; + if (turn) await turn.checkpointStore.delete(turn.sessionId); } } diff --git a/src/session/durableSession.test.ts b/src/session/durableSession.test.ts new file mode 100644 index 000000000..a08c30ca9 --- /dev/null +++ b/src/session/durableSession.test.ts @@ -0,0 +1,226 @@ +/** + * LOU-W9: checkpointed sessions. Every turn is checkpointed per step, so a + * turn interrupted mid-way (a crash, a failed checkpoint write) is finished + * by `session.resume()` in a fresh agent, without re-running finished steps. + */ +import { describe, it, expect } from 'vitest'; +import { z } from 'zod'; +import { createAgent } from '../createAgent'; +import { defineTool, type DefinedTool } from '../tools/defineTool'; +import { PropagatingToolError } from '../execution/AgentExecutor'; +import { SessionAwaitingApprovalError } from '../execution/errors'; +import { InMemoryApprovalStore } from '../execution/InMemoryApprovalStore'; +import { KVCheckpointStore } from '../deploy/kvCheckpointStore'; +import { SqliteStore } from '../storage/sqlite'; +import { mockModel, type MockTurn } from '../testing'; +import { MemorySessionStore } from './index'; +import type { AgentRun } from '../execution/agentRun'; +import type { Message } from '../providers/llm'; + +type Runs = Record; + +/** A tool that counts its runs and, with `crashOnce`, dies (PropagatingToolError) on its first run. */ +function tool(name: string, runs: Runs, options: { crashOnce?: boolean; needsApproval?: boolean } = {}): DefinedTool { + runs[name] = 0; + return defineTool({ + name, + description: name, + input: z.object({}), + needsApproval: options.needsApproval, + execute: async () => { + runs[name]++; + if (options.crashOnce && runs[name] === 1) throw new PropagatingToolError(`process died while running ${name}`); + return `${name} done`; + }, + }); +} + +/** A KVCheckpointStore over a Map (JSON round-trip), optionally failing the n-th save (1-based). */ +function checkpointStore(failSave?: number): KVCheckpointStore & { data: Map } { + const data = new Map(); + let saves = 0; + const store = new KVCheckpointStore({ + get: async (key) => data.get(key) ?? null, + put: async (key, value) => { + saves++; + if (saves === failSave) throw new Error('disk gone'); + data.set(key, value); + }, + delete: async (key) => { + data.delete(key); + }, + }); + return Object.assign(store, { data }); +} + +const calling = (name: string): MockTurn => ({ toolCalls: [{ name, id: `call_${name}` }] }); +const roles = (messages: readonly { role: string }[]): string[] => messages.map((m) => m.role); +const users = (messages: readonly Message[]): number => messages.filter((m) => m.role === 'user').length; +const fullTurn = ['user', 'assistant', 'tool', 'assistant', 'tool', 'assistant']; + +async function drain(run: AgentRun): Promise { + for await (const event of run) void event; +} + +describe('checkpointed sessions (LOU-W9)', () => { + it('resume() finishes a turn that crashed on its second step, in a fresh agent, without re-running finished steps', async () => { + const runs: Runs = {}; + const tools = [tool('a', runs), tool('b', runs, { crashOnce: true })]; + const sessions = new MemorySessionStore(); + const checkpoints = checkpointStore(); + + const first = mockModel([calling('a'), calling('b'), 'never reached']); + const crashed = createAgent({ provider: first, tools }).session({ id: 'chat', store: sessions, checkpointStore: checkpoints }); + await expect(crashed.send('do a then b')).rejects.toThrow('process died'); + expect(first.calls).toHaveLength(2); + expect(await sessions.load('chat')).toBeUndefined(); + expect([...checkpoints.data.keys()]).toEqual(['checkpoints/chat.turn-0']); + + const second = mockModel(['All done.']); + const session = createAgent({ provider: second, tools }).session({ id: 'chat', store: sessions, checkpointStore: checkpoints }); + expect(await session.pending()).toEqual({ status: 'in-progress', approvalId: undefined }); + + const result = await session.resume(); + + expect(result?.text).toBe('All done.'); + expect(second.calls).toHaveLength(1); + expect(runs).toEqual({ a: 1, b: 2 }); // `a` was recorded; `b` was running when the process died + const saved = (await sessions.load('chat'))!; + expect(roles(saved)).toEqual(fullTurn); + expect(users(saved)).toBe(1); + expect(session.messages).toEqual(saved); + expect(checkpoints.data.size).toBe(0); + expect(await session.pending()).toBeNull(); + expect(await session.resume()).toBeNull(); + }); + + it('a failed checkpoint write is resumed from a { sessions, checkpoints } store without calling the model again', async () => { + const runs: Runs = {}; + const tools = [tool('a', runs)]; + const store = { sessions: new MemorySessionStore(), checkpoints: checkpointStore(2) }; // save 2 = `a`'s result + + const first = mockModel([calling('a'), 'never reached']); + await expect(createAgent({ provider: first, tools }).session({ id: 'chat', store }).send('do a')).rejects.toThrow('disk gone'); + + const second = mockModel(['Done.']); + const result = await createAgent({ provider: second, tools }).session({ id: 'chat', store }).resume(); + + expect(result?.text).toBe('Done.'); + expect(second.calls).toHaveLength(1); + expect(runs.a).toBe(2); + expect(roles((await store.sessions.load('chat'))!)).toEqual(['user', 'assistant', 'tool', 'assistant']); + }); + + it('send() while a turn is pending finishes that turn first, then sends the new message', async () => { + const runs: Runs = {}; + const tools = [tool('a', runs), tool('b', runs, { crashOnce: true })]; + const store = { sessions: new MemorySessionStore(), checkpoints: checkpointStore() }; + await expect( + createAgent({ provider: mockModel([calling('a'), calling('b')]), tools }).session({ id: 'chat', store }).send('first') + ).rejects.toThrow('process died'); + + const model = mockModel(['First done.', 'Second done.']); + const session = createAgent({ provider: model, tools }).session({ id: 'chat', store }); + const result = await session.send('second'); + + expect(result.text).toBe('Second done.'); + expect(model.calls).toHaveLength(2); + expect(model.calls[1].messages.at(-1)).toMatchObject({ role: 'user', content: 'second' }); + expect(roles(session.messages)).toEqual([...fullTurn, 'user', 'assistant']); + expect(store.checkpoints.data.size).toBe(0); + }); + + it('stream() checkpoints its turn too, and a fresh session resumes it', async () => { + const runs: Runs = {}; + const tools = [tool('a', runs), tool('b', runs, { crashOnce: true })]; + const store = { sessions: new MemorySessionStore(), checkpoints: checkpointStore() }; + + const ok = createAgent({ provider: mockModel(['Hi.']) }).session({ id: 'chat', store }); + await drain(ok.stream('hello')); + expect(store.checkpoints.data.size).toBe(0); // a finished turn leaves no checkpoint + + const crashing = createAgent({ provider: mockModel([calling('a'), calling('b')]), tools }).session({ id: 'chat', store }); + const run = crashing.stream('do a then b'); + await drain(run); + await expect(run.result).rejects.toThrow('process died'); + expect([...store.checkpoints.data.keys()]).toEqual(['checkpoints/chat.turn-2']); + + const model = mockModel(['All done.']); + const result = await createAgent({ provider: model, tools }).session({ id: 'chat', store }).resume(); + + expect(result?.text).toBe('All done.'); + expect(model.calls).toHaveLength(1); + expect(roles((await store.sessions.load('chat'))!)).toEqual(['user', 'assistant', ...fullTurn]); + }); + + it('an approval pause survives a restart and resolves through agent.approvals.resolve()', async () => { + const runs: Runs = {}; + const tools = [tool('send_email', runs, { needsApproval: true })]; + const store = new SqliteStore(':memory:'); + const approvalStore = new InMemoryApprovalStore(); + + const before = createAgent({ provider: mockModel([calling('send_email')]), tools, approvalStore }).session({ id: 'chat', store }); + const paused = await before.send('Email Sam'); + expect(paused.finishReason).toBe('awaiting-approval'); + expect(before.messages).toEqual([]); // the paused turn lives in its checkpoint until it finishes + await expect(before.send('anything else?')).rejects.toBeInstanceOf(SessionAwaitingApprovalError); + + const model = mockModel(['Email sent.']); + const agent = createAgent({ provider: model, tools, approvalStore }); + const session = agent.session({ id: 'chat', store }); + expect(await session.pending()).toEqual({ status: 'awaiting-approval', approvalId: paused.approvalId }); + await expect(session.resume()).rejects.toMatchObject({ name: 'SessionAwaitingApprovalError', approvalId: paused.approvalId }); + + const result = await agent.approvals.resolve({ id: paused.approvalId!, approved: true }); + + expect(result.text).toBe('Email sent.'); + expect(runs.send_email).toBe(1); + expect(model.calls).toHaveLength(1); + expect(roles(session.messages)).toEqual(['user', 'assistant', 'tool', 'assistant']); + expect(roles((await store.sessions.load('chat'))!)).toEqual(['user', 'assistant', 'tool', 'assistant']); + expect(await session.pending()).toBeNull(); + store.close(); + }); + + it('a turn that finished just before its transcript was saved is added to the transcript, not run again', async () => { + const sessions = new MemorySessionStore(); + const checkpoints = checkpointStore(); + const failing = Object.assign(Object.create(sessions) as MemorySessionStore, { + save: async () => { + throw new Error('session store gone'); + }, + }); + const model = mockModel(['Hi.']); + await expect(createAgent({ provider: model }).session({ id: 'chat', store: failing, checkpointStore: checkpoints }).send('hello')).rejects.toThrow( + 'session store gone' + ); + + const session = createAgent({ provider: mockModel([]) }).session({ id: 'chat', store: sessions, checkpointStore: checkpoints }); + expect(await session.resume()).toBeNull(); + expect(roles(session.messages)).toEqual(['user', 'assistant']); + expect(roles((await sessions.load('chat'))!)).toEqual(['user', 'assistant']); + expect(checkpoints.data.size).toBe(0); + }); + + it('an aborted turn, discardPending() and clear() leave no pending turn', async () => { + const runs: Runs = {}; + const tools = [tool('a', runs), tool('b', runs, { crashOnce: true })]; + const store = { sessions: new MemorySessionStore(), checkpoints: checkpointStore() }; + const session = createAgent({ provider: mockModel([calling('a'), calling('b'), calling('a'), calling('b')]), tools }).session({ id: 'chat', store }); + + const aborted = await session.send('go', { signal: AbortSignal.abort() }); + expect(aborted.finishReason).toBe('aborted'); + expect(await session.pending()).toBeNull(); + + await expect(session.send('go')).rejects.toThrow('process died'); + expect(await session.pending()).toMatchObject({ status: 'in-progress' }); + await session.discardPending(); + expect(await session.pending()).toBeNull(); + expect(session.messages).toEqual([]); + + runs.b = 0; + await expect(session.send('go')).rejects.toThrow('process died'); + await session.clear(); + expect(store.checkpoints.data.size).toBe(0); + }); +}); diff --git a/src/session/index.ts b/src/session/index.ts index eeebd98cc..42c8c942c 100644 --- a/src/session/index.ts +++ b/src/session/index.ts @@ -1,4 +1,11 @@ export { AgentSession } from './AgentSession'; -export type { SessionOptions, SessionRunner, SessionStreamRunner } from './AgentSession'; +export type { + PendingTurn, + SessionOptions, + SessionRunner, + SessionStores, + SessionStreamRunner, + SessionTurnCheckpoint, +} from './AgentSession'; export { MemorySessionStore, FileSessionStore } from './sessionStore'; export type { SessionStore } from './sessionStore';