From b0272cf65da9b0912ab0dd8eb957653ef01a52b3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 16:23:52 +0000 Subject: [PATCH 1/2] [EPIC-D][LOU-D44] AgentExecutor.fork() creates a runnable session from a historical checkpoint AgentExecutor.fork({ sessionId, fromStep, newSessionId?, checkpointStore, patch? }) takes the newest history entry of fromStep, applies the patch (messages rewrite, businessState, a tool result replaced in place, appendInput queued as a user message) and saves it as the in-progress checkpoint of a new session (default .fork-), leaving the original untouched. execute({ sessionId, checkpointStore, input: [] }) or agent.resume() continues the fork. A missing step throws LOUSHY_CHECKPOINT_NOT_FOUND (new registry code, documented in docs/errors.md). createAgent() agents gain agent.fork(sessionId, options) over store.checkpoints. compareTrajectories(a, b) in src/evals/drift.ts compares two checkpoints or transcripts step by step (steps, first diverging step and the eval --drift diff) for Forge's side-by-side view. Docs: durable-execution.md "Fork and replay", llms-full.txt regenerated, CHANGELOG (Added). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_016bY29ibQgv3MDdjAbUT4YR --- CHANGELOG.md | 1 + docs/durable-execution.md | 69 ++++++++++ docs/errors.md | 12 ++ llms-full.txt | 81 ++++++++++++ src/createAgent.ts | 15 +++ src/evals/drift.test.ts | 30 ++++- src/evals/drift.ts | 65 ++++++++++ src/evals/index.ts | 2 + src/execution/AgentExecutor.ts | 22 +++- src/execution/checkpoint.ts | 39 ++++++ src/execution/fork.test.ts | 222 +++++++++++++++++++++++++++++++++ src/execution/fork.ts | 71 +++++++++++ src/utils/errorCodes.ts | 2 + 13 files changed, 629 insertions(+), 2 deletions(-) create mode 100644 src/execution/fork.test.ts create mode 100644 src/execution/fork.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index e0b2919ec..0e2e0aaf3 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 +- Fork and replay (LOU-D44): `AgentExecutor.fork({ sessionId, fromStep, newSessionId?, checkpointStore, patch? })` takes the newest entry of `fromStep` in the session's checkpoint history, applies `patch` (`messages(messages)`, `businessState`, `toolResult: { toolCallId, result }` replaced in place so the transcript stays provider-valid, `appendInput` queued as a user message) and saves it as the `'in-progress'` checkpoint of `newSessionId` (default `.fork-`), leaving the original session untouched; it returns `{ sessionId, step, checkpoint }`, and `execute({ sessionId, checkpointStore, input: [] })` continues the fork. A missing step throws the new `LOUSHY_CHECKPOINT_NOT_FOUND` code. `createAgent()` agents gain `agent.fork(sessionId, { fromStep, newSessionId?, patch? })` over `store.checkpoints` (resume with `agent.resume()`). `compareTrajectories(a, b)` compares two checkpoints or transcripts step by step: each run's steps (text, tool calls and results), the first diverging step and the `loushy eval --drift` diff. New types `ForkOptions`, `ForkPatch`, `ForkResult`, `TrajectoryComparison`, `TrajectoryStep` and `DriftEntry`. See docs/durable-execution.md#fork-and-replay. - Connect MCP servers from config (LOU-Z4): `connectMcp(servers, { lazy?, onError?, logger? })` (`@loushy/build-ai-agent/mcp` and the root) connects stdio (`command`/`args`/`env`) and streamable HTTP (`url`/`headers`) servers, lists their tools as `__` descriptors and returns `{ tools, close(), status() }`. Every server connects up front (listing tools needs a connection); `lazy` (default `true`) reconnects on the next tool call after `close()` or a dropped connection. `onError: 'skip'` leaves a failing server out with a warning instead of rejecting. `createAgent({ mcpServers })` connects them on the new `agent.ready()` or the first `send()` / `stream()` and adds their tools; the new `agent.close()` disconnects them (both are no-ops without `mcpServers`). `specToAgent()` now connects `spec.mcpServers` the same way (closes TODO(LOU-D20.2)); `agent.mcpServers` still lists them. `loadMcpTools()` accepts any `{ listTools, callTool }` (`McpClientLike`). See docs/configuration.md#connect-mcp-servers-mcpservers-connectmcp. - Error codes with fixes (LOU-D2): every `SDKError` now has a stable `code` (`LOUSHY__`, e.g. `LOUSHY_CONFIG_MISSING_PROVIDER`, `LOUSHY_PEER_MISSING`, `LOUSHY_SPEC_INVALID`, `LOUSHY_SESSION_AWAITING_APPROVAL`), a one-sentence `hint` and a `docs` link to its section of the new docs/errors.md; its message ends with a `[code] hint (docs)` line (`detail` is the message without it), and `toString()` always shows it. `ERROR_CODES` / `ErrorCode` (new exports) are the registry, and a test keeps it, the codes used in `src/` and docs/errors.md in sync. The plain `Error`s thrown by `createAgent()` (no model, `instructions` + `prompt`, `sessionId` without checkpoints), `resolveProvider()` (bad spec, unknown prefix, missing API key, missing peer), `AgentExecutor.execute()` / `stream()` (missing `provider` / `agent` / `input`, approval with no `approvalStore`), `resumeAfterApproval()` (unknown approval), sessions (invalid id, corrupt file, second iteration, no stream runner), `loadSpec()` and `specToAgent()` are now `ConfigurationError` / `ValidationError` / `SDKError` with codes; `MissingPeerDependencyError` extends `SDKError`. Existing message text is unchanged (the help line is appended). `loadSpec()` suggests the field a top-level typo meant (`unknown field 'promt' (did you mean 'prompt'?)`). See docs/errors.md. - Checkpoint history (LOU-D43): `memoryStore()`, `SqliteStore` and `LocalStorageCheckpointStore` keep a bounded history per session: every `save()` also appends the record to a ring (`historyLimit` option, default 50, `0` keeps none; the oldest are dropped), and the new optional `CheckpointStore.history(sessionId, { limit? })` returns `{ step, savedAt, status, checkpoint }` entries newest first. `CheckpointStore.delete(sessionId, { keepHistory: true })` keeps the history (by default `delete()` clears it too), `getCheckpointHistory(store, sessionId)` reads it and returns `undefined` for a store without `history()`, and `SqliteStore` gains the `checkpoint_history` table (added when an existing database file is opened; `prune()` removes old entries). New types `CheckpointHistoryEntry`, `CheckpointHistoryOptions`, `CheckpointDeleteOptions` and `MemoryStoreOptions`. `KVCheckpointStore` and Agent Forge's file store keep no history yet (LOU-D43.2). This is the first step of time-travel from a past step. See docs/durable-execution.md#checkpoint-history. diff --git a/docs/durable-execution.md b/docs/durable-execution.md index cbd8a5785..b440147aa 100644 --- a/docs/durable-execution.md +++ b/docs/durable-execution.md @@ -226,6 +226,75 @@ await checkpoints.delete('chat-42', { keepHistory: true }); an existing database file is opened; `prune()` also removes history entries older than its cutoff. +## Fork and replay + +`AgentExecutor.fork()` starts a new session from a step of another session's +history: "what if the tool had returned something else at step 1?" It takes the +newest history entry of `fromStep`, applies `patch`, saves the result as the +`'in-progress'` checkpoint of the new session and returns +`{ sessionId, step, checkpoint }`. Resume the fork like any unfinished run: + +```ts +import { AgentExecutor, compareTrajectories, memoryStore } from '@loushy/build-ai-agent'; + +const { checkpoints } = memoryStore(); +await AgentExecutor.execute({ agent, provider, toolRegistry, input: 'Plan my trip', sessionId: 'trip', checkpointStore: checkpoints }); + +const fork = await AgentExecutor.fork({ + sessionId: 'trip', + fromStep: 1, + checkpointStore: checkpoints, + patch: { toolResult: { toolCallId: 'call_weather', result: { forecast: 'rain' } } }, +}); +// fork.sessionId is 'trip.fork-1'; the 'trip' checkpoint and history are unchanged. +await AgentExecutor.execute({ agent, provider, toolRegistry, input: [], sessionId: fork.sessionId, checkpointStore: checkpoints }); + +const original = await checkpoints.load('trip'); +const replayed = await checkpoints.load(fork.sessionId); +if (original && replayed) { + const { divergedAt, a, b, drift } = compareTrajectories(original, replayed); + console.log(divergedAt, a.length, b.length, drift); // 1, then each run's steps and the tool-call diff +} +``` + +With `createAgent({ store })`, `agent.fork(sessionId, { fromStep, patch? })` +does the same over `store.checkpoints`, and `agent.resume(fork.sessionId)` +continues the fork: + +```ts +import { createAgent, memoryStore } from '@loushy/build-ai-agent'; + +const assistant = createAgent({ provider, store: memoryStore() }); +await assistant.send('Plan my trip', { sessionId: 'trip' }); +const fork = await assistant.fork('trip', { fromStep: 1, patch: { appendInput: 'It will rain, plan for that.' } }); +const result = await assistant.resume(fork.sessionId); +``` + +- `patch` is applied in this order: `messages(messages)` rewrites the + transcript; `businessState` replaces it; `toolResult: { toolCallId, result }` + replaces that call's result in place (same position, call id and tool name, + so the transcript stays valid for every provider), or records it when the + call has no result yet; `appendInput` queues a user message, which is sent + after any tool calls still pending. +- The fork keeps the source checkpoint's step count and usage, so the rest of + the `maxSteps` budget applies to it. Recorded tool results are not re-run; + tool calls without a result run first, as on any resume (so a fork of an + `'awaiting-approval'` entry goes through `needsApproval` again; the old + `approvalId` is not carried over). +- `newSessionId` names the fork; the default is `.fork-`, the + first `n` with no checkpoint. A `newSessionId` that already has a checkpoint + is refused. +- A step the history does not have (the session is unknown, the step never + ran, or it was dropped past `historyLimit`) throws an `SDKError` with code + [`LOUSHY_CHECKPOINT_NOT_FOUND`](./errors.md#loushy_checkpoint_not_found), + whose message lists the steps that are kept. A store without `history()` + cannot fork (`ConfigurationError`). +- `compareTrajectories(a, b)` takes two checkpoints or two transcripts and + returns each run's steps (one per assistant turn: its text and tool calls + with their results), `divergedAt` (the first step that differs, tool call + ids aside) and `drift`, the tool-order, argument, step-count and + finish-reason differences in the same shape as `loushy eval --drift`. + ## Compatibility with stored data - Checkpoints written before `status` existed are treated as diff --git a/docs/errors.md b/docs/errors.md index ff904938c..740a4b9fb 100644 --- a/docs/errors.md +++ b/docs/errors.md @@ -290,6 +290,18 @@ streaming runner. **Example:** `new AgentSession(run).stream('hi')`. +### LOUSHY_CHECKPOINT_NOT_FOUND + +**Means:** `AgentExecutor.fork()` or `agent.fork()` was asked for a step the +session's checkpoint history does not have: the session is unknown, the step +was never reached, or its entries were dropped past the store's `historyLimit`. +The message lists the steps that are kept. + +**Fix:** fork at one of the listed steps, or raise `historyLimit` on the store. +See [Durable execution](./durable-execution.md#fork-and-replay). + +**Example:** `agent.fork('job-1', { fromStep: 9 })` after a 3-step run. + ### LOUSHY_RUN_ALREADY_ITERATED **Means:** an `AgentRun` from `session.stream()` was iterated a second time. diff --git a/llms-full.txt b/llms-full.txt index 99718831b..7ca7c5d89 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -3049,6 +3049,75 @@ await checkpoints.delete('chat-42', { keepHistory: true }); an existing database file is opened; `prune()` also removes history entries older than its cutoff. +## Fork and replay + +`AgentExecutor.fork()` starts a new session from a step of another session's +history: "what if the tool had returned something else at step 1?" It takes the +newest history entry of `fromStep`, applies `patch`, saves the result as the +`'in-progress'` checkpoint of the new session and returns +`{ sessionId, step, checkpoint }`. Resume the fork like any unfinished run: + +```ts +import { AgentExecutor, compareTrajectories, memoryStore } from '@loushy/build-ai-agent'; + +const { checkpoints } = memoryStore(); +await AgentExecutor.execute({ agent, provider, toolRegistry, input: 'Plan my trip', sessionId: 'trip', checkpointStore: checkpoints }); + +const fork = await AgentExecutor.fork({ + sessionId: 'trip', + fromStep: 1, + checkpointStore: checkpoints, + patch: { toolResult: { toolCallId: 'call_weather', result: { forecast: 'rain' } } }, +}); +// fork.sessionId is 'trip.fork-1'; the 'trip' checkpoint and history are unchanged. +await AgentExecutor.execute({ agent, provider, toolRegistry, input: [], sessionId: fork.sessionId, checkpointStore: checkpoints }); + +const original = await checkpoints.load('trip'); +const replayed = await checkpoints.load(fork.sessionId); +if (original && replayed) { + const { divergedAt, a, b, drift } = compareTrajectories(original, replayed); + console.log(divergedAt, a.length, b.length, drift); // 1, then each run's steps and the tool-call diff +} +``` + +With `createAgent({ store })`, `agent.fork(sessionId, { fromStep, patch? })` +does the same over `store.checkpoints`, and `agent.resume(fork.sessionId)` +continues the fork: + +```ts +import { createAgent, memoryStore } from '@loushy/build-ai-agent'; + +const assistant = createAgent({ provider, store: memoryStore() }); +await assistant.send('Plan my trip', { sessionId: 'trip' }); +const fork = await assistant.fork('trip', { fromStep: 1, patch: { appendInput: 'It will rain, plan for that.' } }); +const result = await assistant.resume(fork.sessionId); +``` + +- `patch` is applied in this order: `messages(messages)` rewrites the + transcript; `businessState` replaces it; `toolResult: { toolCallId, result }` + replaces that call's result in place (same position, call id and tool name, + so the transcript stays valid for every provider), or records it when the + call has no result yet; `appendInput` queues a user message, which is sent + after any tool calls still pending. +- The fork keeps the source checkpoint's step count and usage, so the rest of + the `maxSteps` budget applies to it. Recorded tool results are not re-run; + tool calls without a result run first, as on any resume (so a fork of an + `'awaiting-approval'` entry goes through `needsApproval` again; the old + `approvalId` is not carried over). +- `newSessionId` names the fork; the default is `.fork-`, the + first `n` with no checkpoint. A `newSessionId` that already has a checkpoint + is refused. +- A step the history does not have (the session is unknown, the step never + ran, or it was dropped past `historyLimit`) throws an `SDKError` with code + [`LOUSHY_CHECKPOINT_NOT_FOUND`](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/errors.md#loushy_checkpoint_not_found), + whose message lists the steps that are kept. A store without `history()` + cannot fork (`ConfigurationError`). +- `compareTrajectories(a, b)` takes two checkpoints or two transcripts and + returns each run's steps (one per assistant turn: its text and tool calls + with their results), `divergedAt` (the first step that differs, tool call + ids aside) and `drift`, the tool-order, argument, step-count and + finish-reason differences in the same shape as `loushy eval --drift`. + ## Compatibility with stored data - Checkpoints written before `status` existed are treated as @@ -3365,6 +3434,18 @@ streaming runner. **Example:** `new AgentSession(run).stream('hi')`. +### LOUSHY_CHECKPOINT_NOT_FOUND + +**Means:** `AgentExecutor.fork()` or `agent.fork()` was asked for a step the +session's checkpoint history does not have: the session is unknown, the step +was never reached, or its entries were dropped past the store's `historyLimit`. +The message lists the steps that are kept. + +**Fix:** fork at one of the listed steps, or raise `historyLimit` on the store. +See [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#fork-and-replay). + +**Example:** `agent.fork('job-1', { fromStep: 9 })` after a 3-step run. + ### LOUSHY_RUN_ALREADY_ITERATED **Means:** an `AgentRun` from `session.stream()` was iterated a second time. diff --git a/src/createAgent.ts b/src/createAgent.ts index 9017034fc..d4865c39a 100644 --- a/src/createAgent.ts +++ b/src/createAgent.ts @@ -38,6 +38,7 @@ import type { SubagentSpec } from './execution/delegation'; import type { ApprovalStore } from './execution/ApprovalGate'; import { InMemoryApprovalStore } from './execution/InMemoryApprovalStore'; import { resumeAfterApproval } from './execution/resume'; +import type { ForkOptions, ForkResult } from './execution/checkpoint'; import { ConfigurationError } from './execution/errors'; import { createAgentApprovals, type AgentApprovals, type ApproveToolCall } from './createAgentApprovals'; import type { z } from 'zod'; @@ -363,6 +364,18 @@ export interface SimpleAgent { * ``` */ resume: (sessionId: string, options?: { signal?: AbortSignal }) => Promise | null>; + /** + * Forks the `sessionId` run at `fromStep` (LOU-D44): `AgentExecutor.fork()` + * over the agent's `store.checkpoints`, which must keep a history. Continue + * the fork with `agent.resume(fork.sessionId)`; the original is unchanged. + * + * @example + * ```ts + * const fork = await agent.fork('job-1', { fromStep: 1, patch: { appendInput: 'Use Celsius.' } }); + * const replayed = await agent.resume(fork.sessionId); + * ``` + */ + fork: (sessionId: string, options: Omit) => Promise; /** * Tool calls this agent is paused on, waiting for approval (LOU-D21). A * paused `send()` resolves with `finishReason: 'awaiting-approval'` and an @@ -507,6 +520,8 @@ export function createAgent( if (checkpoint.status === 'finished') return null; return approvals.settle(await run([], signal, durable(sessionId)), signal) as Promise>; }, + // durable() throws LOUSHY_CONFIG_MISSING_CHECKPOINT_STORE without `store.checkpoints`. + fork: async (sessionId, options) => AgentExecutor.fork({ ...options, ...(durable(sessionId) as SessionTurnCheckpoint) }), approvals: approvals.approvals, ready: mcp.ready, close: mcp.close, diff --git a/src/evals/drift.test.ts b/src/evals/drift.test.ts index ef1945637..1c21db46b 100644 --- a/src/evals/drift.test.ts +++ b/src/evals/drift.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect } from 'vitest'; import type { Cassette } from '../testing/cassette'; -import { diffTrajectories, trajectoryOf, type Trajectory } from './drift'; +import type { Message } from '../providers'; +import { compareTrajectories, diffTrajectories, trajectoryOf, type Trajectory } from './drift'; type Entry = Cassette['entries'][number]; @@ -77,3 +78,30 @@ describe('diffTrajectories', () => { expect(diffTrajectories(base, pricier, { usage: true })).toEqual([{ field: 'tokens', committed: '15', current: '905' }]); }); }); + +describe('compareTrajectories (LOU-D44)', () => { + const transcript = (callId: string, args: string, result: string): Message[] => [ + { role: 'user', content: 'refund order 42' }, + { role: 'assistant', content: '', toolCalls: [{ id: callId, type: 'function', function: { name: 'lookup', arguments: args } }] }, + { role: 'tool', content: result, toolCallId: callId, toolName: 'lookup' }, + { role: 'assistant', content: 'Done.' }, + ]; + + it('matches transcripts that differ only in call ids and argument key order', () => { + const diff = compareTrajectories(transcript('a', '{"id":42,"x":1}', 'ok'), transcript('b', '{"x":1,"id":42}', 'ok')); + expect(diff.divergedAt).toBeUndefined(); + expect(diff.drift).toEqual([]); + expect(diff.a).toEqual([ + { text: '', tools: [{ id: 'a', name: 'lookup', args: '{"id":42,"x":1}', result: 'ok' }] }, + { text: 'Done.', tools: [] }, + ]); + }); + + it('reports the first step whose tool result, arguments or text differ', () => { + expect(compareTrajectories(transcript('a', '{}', 'ok'), transcript('a', '{}', 'gone')).divergedAt).toBe(1); + const args = compareTrajectories(transcript('a', '{"id":1}', 'ok'), transcript('a', '{"id":2}', 'ok')); + expect(args.drift).toEqual([{ field: 'args', committed: 'lookup {"id":1}', current: 'lookup {"id":2}' }]); + const shorter = compareTrajectories(transcript('a', '{}', 'ok'), transcript('a', '{}', 'ok').slice(0, 3)); + expect(shorter).toMatchObject({ divergedAt: 2, drift: [{ field: 'steps', committed: '2', current: '1' }] }); + }); +}); diff --git a/src/evals/drift.ts b/src/evals/drift.ts index 6124f0339..4c1ee3629 100644 --- a/src/evals/drift.ts +++ b/src/evals/drift.ts @@ -5,6 +5,8 @@ */ import type { Cassette } from '../testing/cassette'; import { stableStringify } from '../testing/fingerprint'; +import type { Checkpoint } from '../execution/checkpoint'; +import type { Message } from '../providers'; /** What a recorded case did. */ export interface Trajectory { @@ -70,3 +72,66 @@ export function diffTrajectories(committed: Trajectory, current: Trajectory, opt if (options.usage) compare('tokens', String(committed.totalTokens ?? '-'), String(current.totalTokens ?? '-')); return drift; } + +/** One model turn of a transcript (LOU-D44). */ +export interface TrajectoryStep { + /** The assistant's text this turn (`''` for none). */ + text: string; + /** Its tool calls; `result` is the recorded result's content, absent while the call has none. */ + tools: { id: string; name: string; args: string; result?: string }[]; +} + +/** How two runs differ (LOU-D44): e.g. a session and a fork of it, for side-by-side rendering. */ +export interface TrajectoryComparison { + /** Each run's model turns, in order (step `n` is index `n - 1`). */ + a: TrajectoryStep[]; + b: TrajectoryStep[]; + /** The first step (from 1) whose text, tool calls or tool results differ (call ids aside); absent when none does. */ + divergedAt?: number; + /** Tool order, arguments, step count and finish reason differences, as `diffTrajectories()` (`a` as committed). */ + drift: DriftEntry[]; +} + +function stepsOf(run: Checkpoint | Message[]): TrajectoryStep[] { + const messages = Array.isArray(run) ? run : run.messages; + const results = new Map(messages.filter((m) => m.role === 'tool').map((m) => [m.toolCallId, m.content])); + return messages + .filter((m) => m.role === 'assistant') + .map((m) => ({ + text: m.content, + tools: (m.toolCalls ?? []).map((call) => ({ + id: call.id, + name: call.function.name, + args: normalizeArgs(call.function.arguments), + ...(results.has(call.id) ? { result: results.get(call.id) } : {}), + })), + })); +} + +function trajectoryOfSteps(steps: TrajectoryStep[], run: Checkpoint | Message[]): Trajectory { + const tools = steps.flatMap((step) => step.tools.map(({ name, args }) => ({ name, args }))); + return { tools, steps: steps.length, finishReason: Array.isArray(run) ? undefined : run.finishReason }; +} + +/** A step as compared: its call ids left out (each run gets its own). */ +function stepKey(step: TrajectoryStep | undefined): string { + return step ? stableStringify({ ...step, tools: step.tools.map(({ name, args, result }) => ({ name, args, result })) }) : '-'; +} + +/** + * Compares two runs (LOU-D44) - checkpoints, such as a session's and its + * `AgentExecutor.fork()`'s, or transcripts - turn by turn. A step is one + * assistant turn of the transcript. + */ +export function compareTrajectories(a: Checkpoint | Message[], b: Checkpoint | Message[]): TrajectoryComparison { + const left = stepsOf(a); + const right = stepsOf(b); + const length = Math.max(left.length, right.length); + const index = Array.from({ length }, (_, i) => i).find((i) => stepKey(left[i]) !== stepKey(right[i])); + return { + a: left, + b: right, + ...(index !== undefined ? { divergedAt: index + 1 } : {}), + drift: diffTrajectories(trajectoryOfSteps(left, a), trajectoryOfSteps(right, b)), + }; +} diff --git a/src/evals/index.ts b/src/evals/index.ts index e58ebb84a..33604abf9 100644 --- a/src/evals/index.ts +++ b/src/evals/index.ts @@ -9,4 +9,6 @@ export * from './llmJudge'; export * from './checks'; export * from './evalResult'; +export { compareTrajectories } from './drift'; +export type { TrajectoryComparison, TrajectoryStep, DriftEntry } from './drift'; export type { EvalJudgeConfig, EvalTestContext, CalledToolOptions, AgentSource } from './trajectory'; diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index 3c84da4d4..da0893a7a 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -15,7 +15,8 @@ import { AgentConfig } from '../types'; import { ToolRegistry } from '../tools'; import { SandboxAdapter, NoopSandbox } from '../security/sandboxCore'; import { ApprovalStore, ExecutionSnapshot, PendingApproval, SubagentSuspension } from './ApprovalGate'; -import { CheckpointStore } from './checkpoint'; +import { CheckpointStore, ForkOptions, ForkResult } from './checkpoint'; +import { forkSession } from './fork'; import type { CallUsage, RunUsage, StepUsage } from '../models/usage'; import { mergeDelegatedUsage } from './runUsage'; import { TraceExporter, withSpan } from './tracing'; @@ -622,6 +623,25 @@ export class AgentExecutor { }, options.signal); } + /** + * Forks a checkpointed run at an earlier step (LOU-D44): takes the newest + * entry of `fromStep` in the session's checkpoint history, applies `patch` + * (rewrite messages, replace a tool result, change `businessState`, queue + * a user message) and saves it as the `'in-progress'` checkpoint of a new + * session. The original session is not changed. Throws + * `LOUSHY_CHECKPOINT_NOT_FOUND` when the history has no such step. See + * docs/durable-execution.md#fork-and-replay. + * + * @example + * ```ts + * const fork = await AgentExecutor.fork({ sessionId: 'job-1', fromStep: 1, checkpointStore, patch: { appendInput: 'Use Celsius.' } }); + * await AgentExecutor.execute({ agent, provider, input: [], sessionId: fork.sessionId, checkpointStore }); + * ``` + */ + static fork(options: ForkOptions): Promise { + return forkSession(options); + } + /** * The actual execution loop, split out of execute() so the top-level * 'agent.run' span (LOU-E5) can wrap it via withSpan() while still diff --git a/src/execution/checkpoint.ts b/src/execution/checkpoint.ts index c93556b3e..0c5e27399 100644 --- a/src/execution/checkpoint.ts +++ b/src/execution/checkpoint.ts @@ -112,6 +112,45 @@ export interface CheckpointDeleteOptions { keepHistory?: boolean; } +/** Changes `AgentExecutor.fork()` applies to the forked checkpoint (LOU-D44), in this order. */ +export interface ForkPatch { + /** Rewrites the transcript (it gets a copy). */ + messages?: (messages: Message[]) => Message[]; + /** Replaces `businessState` (`null` clears it). */ + businessState?: unknown; + /** + * Replaces the result of the tool call `toolCallId` (the message keeps its + * place, call id and tool name, so the transcript stays valid), or records + * it when the call has no result yet. `result` is JSON-serialized, as tool + * results are. + */ + toolResult?: { toolCallId: string; result: unknown }; + /** Queues a user message, sent after any tool calls still pending. */ + appendInput?: string; +} + +/** Options for `AgentExecutor.fork()` (LOU-D44). */ +export interface ForkOptions { + /** The session to fork; it is not changed. */ + sessionId: string; + /** The step (`stepIndex`) to fork at; the newest history entry of that step is used. */ + fromStep: number; + /** The fork's session id. Default: `.fork-`, the first `n` with no checkpoint. */ + newSessionId?: string; + /** A store that keeps a history (`history()`); the fork is saved in it too. */ + checkpointStore: CheckpointStore; + patch?: ForkPatch; +} + +/** A fork made by `AgentExecutor.fork()`: continue it with `execute({ sessionId, checkpointStore, input: [] })`. */ +export interface ForkResult { + sessionId: string; + /** The step it was forked at. */ + step: number; + /** The fork's current checkpoint (`status: 'in-progress'`). */ + checkpoint: Checkpoint; +} + /** How many checkpoints a store's history keeps per session unless told otherwise. */ export const DEFAULT_CHECKPOINT_HISTORY_LIMIT = 50; diff --git a/src/execution/fork.test.ts b/src/execution/fork.test.ts new file mode 100644 index 000000000..608181521 --- /dev/null +++ b/src/execution/fork.test.ts @@ -0,0 +1,222 @@ +/** + * LOU-D44: AgentExecutor.fork() / agent.fork() start a new session from a + * step of another session's checkpoint history, optionally patched, and the + * fork resumes like any unfinished run while the original stays as it was. + */ +import { describe, it, expect } from 'vitest'; +import { z } from 'zod'; +import { AgentExecutor } from './AgentExecutor'; +import type { Checkpoint, CheckpointStore } from './checkpoint'; +import { ConfigurationError, SDKError } from './errors'; +import { memoryStore } from '../storage/agentStore'; +import { createAgent } from '../createAgent'; +import { compareTrajectories } from '../evals/drift'; +import { defineTool, type DefinedTool } from '../tools/defineTool';import { ToolRegistry } from '../tools'; +import { mockModel, type MockTurn } from '../testing'; +import type { AgentConfig } from '../types'; + +function tools(runs: Record): DefinedTool[] { + const counted = (name: string, result: unknown) => { + runs[name] = 0; + return defineTool({ + name, + description: name, + input: z.object({}), + execute: async () => { + runs[name]++; + return result; + }, + }); + }; + return [counted('weather', { forecast: 'sunny' }), counted('plan_picnic', 'picnic planned'), counted('buy_umbrella', 'umbrella bought')]; +} + +// Step 1 asks for the weather; step 2 acts on the forecast; step 3 reports the last tool result. +const askWeather: MockTurn = { toolCalls: [{ name: 'weather', id: 'call_weather' }] }; +const decide: MockTurn = (req) => { + const forecast = req.messages.find((m) => m.toolCallId === 'call_weather')?.content ?? ''; + const name = forecast.includes('rain') ? 'buy_umbrella' : 'plan_picnic'; + return { toolCalls: [{ name, id: `call_${name}` }] }; +}; +const report: MockTurn = (req) => `done: ${req.messages.at(-1)?.content}`; + +function setup() { + const runs: Record = {}; + const defined = tools(runs); + const registry = new ToolRegistry(); + registry.registerMany(defined); + const agent: AgentConfig = { + id: 'agent-1', + name: 'Planner', + prompt: 'Plan the trip.', + tools: Object.fromEntries(defined.map((t) => [t.name, { tool: t.name }])), + }; + const checkpointStore = memoryStore().checkpoints; + const execute = (script: MockTurn[], sessionId: string, input: string | [] = []) => { + const provider = mockModel(script); + return { provider, result: AgentExecutor.execute({ agent, provider, toolRegistry: registry, input, sessionId, checkpointStore }) }; + }; + return { runs, agent, registry, checkpointStore, execute }; +} + +async function runOriginal(env: ReturnType) { + const original = await env.execute([askWeather, decide, report], 'trip', 'Plan my trip').result; + expect(original.steps).toBe(3); + expect(original.text).toBe('done: "picnic planned"'); + const store = env.checkpointStore; + return { checkpoint: await store.load('trip'), history: await store.history?.('trip') }; +} + +describe('AgentExecutor.fork (LOU-D44)', () => { + it('forks at step 1 with a patched tool result, resumes into a different trajectory and leaves the original untouched', async () => { + const env = setup(); + const before = await runOriginal(env); + + const fork = await AgentExecutor.fork({ + sessionId: 'trip', + fromStep: 1, + checkpointStore: env.checkpointStore, + patch: { toolResult: { toolCallId: 'call_weather', result: { forecast: 'rain' } } }, + }); + expect(fork.sessionId).toBe('trip.fork-1'); + expect(fork.step).toBe(1); + expect(fork.checkpoint).toMatchObject({ sessionId: 'trip.fork-1', stepIndex: 1, status: 'in-progress' }); + // The result keeps its place, call id and tool name: the transcript stays provider-valid. + expect(fork.checkpoint.messages.map((m) => m.role)).toEqual(['system', 'user', 'assistant', 'tool']); + expect(fork.checkpoint.messages[3]).toMatchObject({ toolCallId: 'call_weather', toolName: 'weather', content: '{"forecast":"rain"}' }); + expect(await env.checkpointStore.load('trip.fork-1')).toEqual(fork.checkpoint); + + const { provider, result } = env.execute([decide, report], fork.sessionId); + const replayed = await result; + provider.assertExhausted(); + expect(replayed.text).toBe('done: "umbrella bought"'); + expect(replayed.steps).toBe(3); + // Step 1 is not replayed: the weather tool ran once, in the original. + expect(env.runs).toEqual({ weather: 1, plan_picnic: 1, buy_umbrella: 1 }); + + expect(await env.checkpointStore.load('trip')).toEqual(before.checkpoint); + expect(await env.checkpointStore.history?.('trip')).toEqual(before.history); + + const forked = await env.checkpointStore.load(fork.sessionId); + const diff = compareTrajectories(before.checkpoint as Checkpoint, forked as Checkpoint); + expect(diff.divergedAt).toBe(1); + expect(diff.a[0].tools[0]).toMatchObject({ name: 'weather', result: '{"forecast":"sunny"}' }); + expect(diff.b[0].tools[0]).toMatchObject({ name: 'weather', result: '{"forecast":"rain"}' }); + expect(diff.drift).toContainEqual({ field: 'tools', committed: 'weather > plan_picnic', current: 'weather > buy_umbrella' }); + + // A second fork gets the next id. + const again = await AgentExecutor.fork({ sessionId: 'trip', fromStep: 2, checkpointStore: env.checkpointStore }); + expect(again.sessionId).toBe('trip.fork-2'); + // Unpatched, it matches the original up to the step it was forked at. + const prefix = compareTrajectories(before.checkpoint as Checkpoint, again.checkpoint); + expect(prefix.divergedAt).toBe(3); + expect(prefix.b).toEqual(prefix.a.slice(0, 2)); + expect(prefix.drift).toContainEqual({ field: 'steps', committed: '3', current: '2' }); + }); + + it('appendInput queues a user message the resumed fork sends; messages and businessState patches apply', async () => { + const env = setup(); + await runOriginal(env); + + const fork = await AgentExecutor.fork({ + sessionId: 'trip', + fromStep: 1, + newSessionId: 'what-if', + checkpointStore: env.checkpointStore, + patch: { + messages: (messages) => messages.map((m) => (m.role === 'user' ? { ...m, content: 'Plan a beach day' } : m)), + businessState: { variant: 'b' }, + appendInput: 'It will rain, plan for that.', + }, + }); + expect(fork.sessionId).toBe('what-if'); + expect(fork.checkpoint.businessState).toEqual({ variant: 'b' }); + expect(fork.checkpoint.messages.at(-1)).toEqual({ role: 'user', content: 'It will rain, plan for that.' }); + + const rainy: MockTurn = (req) => { + expect(req.messages.map((m) => m.role)).toEqual(['system', 'user', 'assistant', 'tool', 'user']); + expect(req.messages[1].content).toBe('Plan a beach day'); + return { toolCalls: [{ name: 'buy_umbrella', id: 'call_umbrella' }] }; + }; + const { result } = env.execute([rainy, report], 'what-if'); + expect((await result).text).toBe('done: "umbrella bought"'); + }); + + it('records a result for a call that has none yet, and queues appendInput behind it', async () => { + const env = setup(); + await runOriginal(env); + // The oldest step-1 entry is the one saved right after the model asked for the weather. + const history = (await env.checkpointStore.history?.('trip')) ?? []; + const pending = history.filter((e) => e.step === 1).at(-1)?.checkpoint as Checkpoint; + const store = memoryStore().checkpoints; + await store.save('pending', pending); + + const fork = await AgentExecutor.fork({ + sessionId: 'pending', + fromStep: 1, + checkpointStore: store, + patch: { appendInput: 'Hurry.', toolResult: { toolCallId: 'call_weather', result: { forecast: 'rain' } } }, + }); + expect(fork.checkpoint.messages.slice(-2).map((m) => m.role)).toEqual(['tool', 'user']); + const replayed = await AgentExecutor.execute({ + agent: env.agent, + provider: mockModel([decide, report]), + toolRegistry: env.registry, + input: [], + sessionId: fork.sessionId, + checkpointStore: store, + }); + expect(replayed.text).toBe('done: "umbrella bought"'); + expect(env.runs.weather).toBe(1); // the recorded result is used, the tool does not run again + }); + + it('rejects a missing step, an unknown session, a store without history and a taken session id', async () => { + const env = setup(); + await runOriginal(env); + const fork = (options: { sessionId?: string; fromStep?: number; newSessionId?: string; checkpointStore?: CheckpointStore }) => + AgentExecutor.fork({ sessionId: 'trip', fromStep: 1, checkpointStore: env.checkpointStore, ...options }); + + const missing = await fork({ fromStep: 9 }).catch((error: unknown) => error); + expect(missing).toBeInstanceOf(SDKError); + expect(missing).toMatchObject({ code: 'LOUSHY_CHECKPOINT_NOT_FOUND' }); + expect((missing as SDKError).detail).toContain('steps kept: 1, 2, 3'); + await expect(fork({ sessionId: 'nope' })).rejects.toMatchObject({ code: 'LOUSHY_CHECKPOINT_NOT_FOUND' }); + + const store = env.checkpointStore; + const noHistory: CheckpointStore = { + save: (id, checkpoint) => store.save(id, checkpoint), + load: (id) => store.load(id), + delete: (id) => store.delete(id), + }; + await expect(fork({ checkpointStore: noHistory })).rejects.toBeInstanceOf(ConfigurationError); + await expect(fork({ newSessionId: 'trip' })).rejects.toBeInstanceOf(ConfigurationError); + await expect(fork({})).resolves.toMatchObject({ sessionId: 'trip.fork-1' }); + await expect(fork({ newSessionId: 'trip.fork-1' })).rejects.toThrow(/already has a checkpoint/); + await expect( + AgentExecutor.fork({ sessionId: 'trip', fromStep: 1, checkpointStore: env.checkpointStore, patch: { toolResult: { toolCallId: 'x', result: 1 } } }) + ).rejects.toThrow(/no tool call 'x'/); + }); +}); + +describe('agent.fork (LOU-D44)', () => { + it('forks a send() run in the agent store; agent.resume() continues the fork', async () => { + const runs: Record = {}; + const store = memoryStore(); + const provider = mockModel([askWeather, decide, report, decide, report]); + const agent = createAgent({ provider, instructions: 'Plan the trip.', tools: tools(runs), store }); + + expect((await agent.send('Plan my trip', { sessionId: 'trip' })).text).toBe('done: "picnic planned"'); + const fork = await agent.fork('trip', { fromStep: 1, patch: { toolResult: { toolCallId: 'call_weather', result: { forecast: 'rain' } } } }); + expect(fork.sessionId).toBe('trip.fork-1'); + + const replayed = await agent.resume(fork.sessionId); + expect(replayed?.text).toBe('done: "umbrella bought"'); + provider.assertExhausted(); + expect((await store.checkpoints.load('trip'))?.messages.at(-1)?.content).toBe('done: "picnic planned"'); + }); + + it('needs a checkpoint store', async () => { + const agent = createAgent({ provider: mockModel([]) }); + await expect(agent.fork('trip', { fromStep: 1 })).rejects.toMatchObject({ code: 'LOUSHY_CONFIG_MISSING_CHECKPOINT_STORE' }); + }); +}); diff --git a/src/execution/fork.ts b/src/execution/fork.ts new file mode 100644 index 000000000..4ed8b232a --- /dev/null +++ b/src/execution/fork.ts @@ -0,0 +1,71 @@ +/** + * Fork and replay (LOU-D44): a new session that starts from a checkpoint in + * another session's history, optionally patched, and resumes like any + * unfinished run. + */ + +import type { Message } from '../providers'; +import type { Checkpoint, CheckpointStore, ForkOptions, ForkPatch, ForkResult } from './checkpoint'; +import { ConfigurationError, SDKError } from './errors'; + +/** `.fork-` for the first `n` (from 1) the store has no checkpoint under. */ +async function nextForkId(store: CheckpointStore, sessionId: string): Promise { + for (let n = 1; ; n++) { + const id = `${sessionId}.fork-${n}`; + if (!(await store.load(id))) return id; + } +} + +/** `messages` with the result of `toolCallId` replaced in place, or appended when it has none yet. */ +function withToolResult(messages: Message[], { toolCallId, result }: NonNullable): Message[] { + const call = messages.flatMap((m) => (m.role === 'assistant' ? (m.toolCalls ?? []) : [])).find((c) => c.id === toolCallId); + if (!call) { + throw new ConfigurationError(`fork: the checkpoint has no tool call '${toolCallId}'.`, 'patch.toolResult'); + } + const name = call.function.name; + const message: Message = { role: 'tool', content: JSON.stringify(result), name, toolCallId, toolName: name }; + const index = messages.findIndex((m) => m.role === 'tool' && m.toolCallId === toolCallId); + // A pending call's result goes at the end; loading the checkpoint moves it behind its turn. + return index < 0 ? [...messages, message] : messages.map((m, i) => (i === index ? message : m)); +} + +function patchMessages(messages: Message[], patch: ForkPatch): Message[] { + let patched = patch.messages ? patch.messages(messages) : messages; + if (patch.toolResult) patched = withToolResult(patched, patch.toolResult); + if (patch.appendInput !== undefined) patched = [...patched, { role: 'user', content: patch.appendInput }]; + return patched; +} + +/** See `AgentExecutor.fork()`. */ +export async function forkSession(options: ForkOptions): Promise { + const { sessionId, fromStep, checkpointStore, patch = {} } = options; + if (!checkpointStore.history) { + throw new ConfigurationError( + `fork: the checkpoint store keeps no history, so session '${sessionId}' cannot be forked. ` + + 'Use memoryStore(), SqliteStore or LocalStorageCheckpointStore.', + 'checkpointStore' + ); + } + const history = await checkpointStore.history(sessionId); + const entry = history.find((e) => e.step === fromStep); + if (!entry) { + const kept = [...new Set(history.map((e) => e.step))].reverse().join(', ') || 'none'; + throw new SDKError(`fork: session '${sessionId}' has no checkpoint at step ${fromStep} (steps kept: ${kept}).`, 'LOUSHY_CHECKPOINT_NOT_FOUND'); + } + const newSessionId = options.newSessionId ?? (await nextForkId(checkpointStore, sessionId)); + if (newSessionId === sessionId || (await checkpointStore.load(newSessionId))) { + throw new ConfigurationError(`fork: '${newSessionId}' already has a checkpoint; pick a new session id.`, 'newSessionId'); + } + + const source = structuredClone(entry.checkpoint); + delete source.approvalId; + const checkpoint: Checkpoint = { + ...source, + sessionId: newSessionId, + messages: patchMessages(source.messages, patch), + ...('businessState' in patch && { businessState: patch.businessState }), + status: 'in-progress', + }; + await checkpointStore.save(newSessionId, checkpoint); + return { sessionId: newSessionId, step: entry.step, checkpoint }; +} diff --git a/src/utils/errorCodes.ts b/src/utils/errorCodes.ts index 7313cdb90..162ba81d7 100644 --- a/src/utils/errorCodes.ts +++ b/src/utils/errorCodes.ts @@ -38,6 +38,8 @@ export const ERROR_CODES = { LOUSHY_SESSION_ID_INVALID: "Use 1-128 characters from A-Z, a-z, 0-9, '_' and '-', or omit the id.", LOUSHY_SESSION_FILE_CORRUPT: 'Restore or delete the session file named in the message.', LOUSHY_SESSION_STREAM_UNSUPPORTED: 'Create the session with agent.session(), which can stream, or call send() instead.', + LOUSHY_CHECKPOINT_NOT_FOUND: + "Fork at a step the session's checkpoint history still keeps (checkpointStore.history() lists them), or raise the store's historyLimit.", LOUSHY_RUN_ALREADY_ITERATED: 'Iterate an AgentRun once; call stream() again for a new run.', LOUSHY_AGENT_EXECUTION_FAILED: 'Look at the `cause` for the underlying failure.', LOUSHY_FLOW_EXECUTION_FAILED: 'Look at the failing `step` and the `cause`.', From 1ba0656f8158cb188e4d617ea564328905bed4f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 1 Oct 2026 16:32:06 +0000 Subject: [PATCH 2/2] fix(evals): compareTrajectories reads message text through textOf() after multimodal parts (LOU-V11) Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_016bY29ibQgv3MDdjAbUT4YR --- src/evals/drift.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/evals/drift.ts b/src/evals/drift.ts index 4c1ee3629..d27da6207 100644 --- a/src/evals/drift.ts +++ b/src/evals/drift.ts @@ -7,6 +7,7 @@ import type { Cassette } from '../testing/cassette'; import { stableStringify } from '../testing/fingerprint'; import type { Checkpoint } from '../execution/checkpoint'; import type { Message } from '../providers'; +import { textOf } from '../providers/content'; /** What a recorded case did. */ export interface Trajectory { @@ -94,11 +95,11 @@ export interface TrajectoryComparison { function stepsOf(run: Checkpoint | Message[]): TrajectoryStep[] { const messages = Array.isArray(run) ? run : run.messages; - const results = new Map(messages.filter((m) => m.role === 'tool').map((m) => [m.toolCallId, m.content])); + const results = new Map(messages.filter((m) => m.role === 'tool').map((m) => [m.toolCallId, textOf(m)])); return messages .filter((m) => m.role === 'assistant') .map((m) => ({ - text: m.content, + text: textOf(m), tools: (m.toolCalls ?? []).map((call) => ({ id: call.id, name: call.function.name,