diff --git a/.fallowrc.json b/.fallowrc.json index 9aa2bd107..a6ddd41ca 100644 --- a/.fallowrc.json +++ b/.fallowrc.json @@ -37,7 +37,8 @@ "members": [ "save", "load", - "delete" + "delete", + "history" ] }, { diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e7a77118..a958e832c 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 +- Agent Forge time travel API (LOU-D45): Agent Forge's `FileCheckpointStore` keeps the bounded per-session checkpoint history (`history()`, `historyLimit` option, `delete({ keepHistory })`) with the same semantics as `LocalStorageCheckpointStore`, checked by the SDK's history contract suite. The runtime control server gains `GET /runs/:id/history` (each step's status, save time, finish reason, tool calls, tokens and cost), `POST /runs/:id/fork` (`{ fromStep, patch?: { toolResult?, appendInput?, businessState? } }`, forks with `AgentExecutor.fork()` and starts the fork as a run of its own) and `GET /runs/compare?a=&b=` (`compareTrajectories()`). Fixed on the way: Agent Forge's chat transcript compiles against multimodal `Message.content` (LOU-V11) again, keeping each message's text. See docs/agent-forge.md#time-travel. - 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. - Multimodal message parts (LOU-V11): `Message.content` is now `string | ContentPart[]`, where a `ContentPart` is `{ type: 'text', text }`, `{ type: 'image', image, mimeType? }` (an `http(s)` URL, a `data:` URL or a `Uint8Array`) or `{ type: 'file', data, mimeType, filename? }` (new types `ContentPart`, `TextContentPart`, `ImageContentPart`, `FileContentPart`). The built-in providers map the parts of `user` messages to the `ai` v4 user-content parts: images go to the model with all four (the `ai` SDK downloads an image URL first for Anthropic and Ollama); file parts become a `[file () not sent]` text note with a one-time `console.warn`, since the pinned peers (`@ai-sdk/*` 0.0.x, `ollama-ai-provider`) cannot send files (a subclass sets `acceptsFileParts = true` to send them). `system`, `assistant` and `tool` messages are sent as their text. The new `textOf(message)` returns a message's text (the string, or its text parts joined), and is what `estimateTokens()` (each image or file part counts 1,000 tokens), compaction, `recordReplay()` fingerprints (cassette messages gain an optional `attachments` list: each image or file as its URL or a digest of its bytes), `mockModel()`, `MockLLMProvider`, OpenTelemetry content attributes and the React reducer (`ui.send` takes parts too) now read. `FileSessionStore` saves `Uint8Array`s as `{ "$bytes": "" }` and loads them back. Pass multimodal messages as `AgentExecutor.execute({ input: Message[] })`; `agent.send()` / `session.send()` taking parts is LOU-V12. See docs/providers.md#multimodal-input. - Declarative permission policies (LOU-X2): `permissions?: PermissionRule[]` on `createAgent()`, `ExecuteOptions` and `resumeAfterApproval()` options. A rule is `{ tool: string | string[] | RegExp | '*', when?: (args, { toolName, toolCallId, sessionId }) => boolean | Promise, action: 'allow' | 'deny' | 'ask', reason? }`; rules are checked in order for each tool call (after argument validation and `preToolCall` hooks, before `needsApproval`) and the first match decides: `allow` runs the call without approval even when `needsApproval` would ask, `deny` gives the model a tool error with the new `kind: 'denied'` (`ToolDeniedError`, plus `reason`) and streams `tool.error`, `ask` pauses for approval (or asks `approve`). No match keeps today's behaviour. New helpers `allow(tools)`, `deny(tools, reason?)`, `ask(tools)` and types `PermissionRule`, `PermissionAction`, `PermissionToolMatcher`, `PermissionContext`, `PermissionDecision`, `PermissionDecisionEntry`, `PermissionOptions`. Audit log: `onPermissionDecision(entry)` gets `{ toolName, toolCallId, decision: 'allow' | 'deny' | 'ask' | 'default', rule?: { index, reason? }, args?, at }` per call (`args` left out under `redactContent`), and streams get the same entry as the new typed `permission.decision` event (`PermissionDecisionEvent`; additive, the event schema version stays 1). Neither is produced unless `permissions` or `onPermissionDecision` is set. Sub-agents inherit the lead's rules (checked before their own) and its `onPermissionDecision`. See docs/approvals.md#permission-policies. diff --git a/apps/agent-forge/server/__tests__/chatReconcile.test.ts b/apps/agent-forge/server/__tests__/chatReconcile.test.ts index 5185c165c..5bccc0c71 100644 --- a/apps/agent-forge/server/__tests__/chatReconcile.test.ts +++ b/apps/agent-forge/server/__tests__/chatReconcile.test.ts @@ -6,7 +6,7 @@ import type { ChatMessage } from '../../shared/wireTypes'; const T1 = '2026-01-01T00:00:00.000Z'; const T2 = '2026-01-01T00:00:05.000Z'; -function chatMsg(m: Message, id: string, timestamp = T1): ChatMessage { +function chatMsg(m: Message & { content: string }, id: string, timestamp = T1): ChatMessage { return { ...m, id, timestamp }; } diff --git a/apps/agent-forge/server/__tests__/checkpointStore.test.ts b/apps/agent-forge/server/__tests__/checkpointStore.test.ts new file mode 100644 index 000000000..e77a5d707 --- /dev/null +++ b/apps/agent-forge/server/__tests__/checkpointStore.test.ts @@ -0,0 +1,21 @@ +/** + * LOU-D45: Forge's FileCheckpointStore keeps the same bounded history ring + * as the SDK's stores - the SDK's own history contract suite, run against it. + */ +import { afterAll } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import { FileCheckpointStore } from '../checkpointStore'; +import { describeCheckpointHistoryContract } from '../../../../src/execution/__fixtures__/checkpointHistoryContract'; + +const dirs: string[] = []; +afterAll(() => { + for (const dir of dirs) fs.rmSync(dir, { recursive: true, force: true }); +}); + +describeCheckpointHistoryContract('FileCheckpointStore (Agent Forge)', (options) => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-d45-store-')); + dirs.push(dir); + return new FileCheckpointStore(dir, options); +}); diff --git a/apps/agent-forge/server/__tests__/timeTravel.test.ts b/apps/agent-forge/server/__tests__/timeTravel.test.ts new file mode 100644 index 000000000..c7f3958fe --- /dev/null +++ b/apps/agent-forge/server/__tests__/timeTravel.test.ts @@ -0,0 +1,152 @@ +/** + * LOU-D45: the time-travel routes - `GET /runs/:id/history`, + * `POST /runs/:id/fork` and `GET /runs/compare` - against real runs of the + * mock provider, through the same RunManager lifecycle as every other run. + */ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import * as fs from 'node:fs'; +import * as path from 'node:path'; +import * as os from 'node:os'; +import request from 'supertest'; +import type { AgentSpec } from '@loushy/build-ai-agent'; +import { createApp } from '../app'; +import { RunManager } from '../runRegistry'; +import { FileCheckpointStore } from '../checkpointStore'; +import { FileApprovalStore } from '../approvalStore'; +import { createFsAgentStore } from '../../src/persistence/fsAgentStore'; +import type { AgentRunStatusPayload, RunComparisonPayload, RunHistoryPayload } from '../../shared/wireTypes'; + +const SPEC: AgentSpec = { + name: 'time-travel-agent', + prompt: 'You are a helpful agent.', + provider: { type: 'mock', model: 'mock-1' }, +}; +const GATED_SPEC: AgentSpec = { ...SPEC, tools: ['demo-approval'] }; + +function settled(runManager: RunManager, runId: string, status: AgentRunStatusPayload['status']): Promise { + if (runManager.status(runId).status === status) return Promise.resolve(runManager.status(runId)); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => reject(new Error(`Timed out waiting for ${runId} to be ${status}`)), 3000); + const onStatus = (payload: AgentRunStatusPayload) => { + if (payload.agentId !== runId || payload.status !== status) return; + clearTimeout(timer); + runManager.off('status', onStatus); + resolve(payload); + }; + runManager.on('status', onStatus); + }); +} + +describe('LOU-D45 time-travel routes', () => { + let baseDir: string; + let runManager: RunManager; + let app: ReturnType; + + beforeEach(() => { + baseDir = fs.mkdtempSync(path.join(os.tmpdir(), 'lou-d45-app-test-')); + const agentStore = createFsAgentStore(baseDir); + runManager = new RunManager({ + baseDir, + checkpointStore: new FileCheckpointStore(baseDir), + approvalStore: new FileApprovalStore(baseDir), + loadSpec: (id) => agentStore.load(id), + saveSpec: (id, spec) => agentStore.save(id, spec), + }); + app = createApp({ agentStore, runManager, baseDir }); + }); + + afterEach(() => { + fs.rmSync(baseDir, { recursive: true, force: true }); + }); + + async function finishedRun(runId: string): Promise { + await request(app).post(`/agents/${runId}/run`).send({ input: 'What is the weather?', spec: SPEC }).expect(202); + await settled(runManager, runId, 'stopped'); + } + + it('GET /runs/:id/history lists the steps with status, finish reason, tool calls and tokens', async () => { + await finishedRun('weather'); + const res = await request(app).get('/runs/weather/history').expect(200); + const body = res.body as RunHistoryPayload; + expect(body.runId).toBe('weather'); + expect(body.steps).toHaveLength(1); + expect(body.steps[0]).toMatchObject({ step: 1, status: 'finished', finishReason: 'stop', toolCalls: [] }); + expect(body.steps[0].tokens).toBeGreaterThan(0); + expect(new Date(body.steps[0].savedAt).toISOString()).toBe(body.steps[0].savedAt); + }); + + it('GET /runs/:id/history is 404 for a run without checkpoints and 400 for an invalid id', async () => { + await request(app).get('/runs/never-ran/history').expect(404); + await request(app).get(`/runs/${encodeURIComponent('../x')}/history`).expect(400); + }); + + it('POST /runs/:id/fork with appendInput starts the fork through the registry; compare shows where it diverged', async () => { + await finishedRun('weather'); + const fork = await request(app) + .post('/runs/weather/fork') + .send({ fromStep: 1, patch: { appendInput: 'And in Celsius?' } }) + .expect(202); + expect(fork.body).toMatchObject({ runId: 'weather.fork-1', fromStep: 1 }); + expect(['running', 'stopped']).toContain(fork.body.status.status); + + const done = await settled(runManager, 'weather.fork-1', 'stopped'); + expect(done.resultText).toBeTruthy(); + expect((await request(app).get('/runs/weather.fork-1/history').expect(200)).body.steps.at(-1).step).toBe(2); + + const compare = await request(app).get('/runs/compare').query({ a: 'weather', b: 'weather.fork-1' }).expect(200); + const comparison = compare.body as RunComparisonPayload; + expect(comparison.a).toHaveLength(1); + expect(comparison.b).toHaveLength(2); + expect(comparison.divergedAt).toBe(2); + expect(comparison.drift).toContainEqual({ field: 'steps', committed: '1', current: '2' }); + }); + + it('POST /runs/:id/fork with toolResult answers a call paused for approval; history shows the call', async () => { + await request(app).post('/agents/gated/run').send({ input: 'please use demo-approval', spec: GATED_SPEC }).expect(202); + await settled(runManager, 'gated', 'paused'); + const [step] = (await request(app).get('/runs/gated/history').expect(200)).body.steps as RunHistoryPayload['steps']; + expect(step).toMatchObject({ step: 1, status: 'awaiting-approval', finishReason: 'tool_calls' }); + expect(step.toolCalls).toEqual([expect.objectContaining({ name: 'demo-approval' })]); + expect(step.toolCalls[0].result).toBeUndefined(); + + const toolCallId = step.toolCalls[0].id; + const fork = await request(app) + .post('/runs/gated/fork') + .send({ fromStep: 1, patch: { toolResult: { toolCallId, result: { done: false, note: 'edited' } } } }) + .expect(202); + await settled(runManager, fork.body.runId, 'stopped'); + expect(runManager.status('gated').status).toBe('paused'); // the source run is untouched + + const comparison = (await request(app).get(`/runs/compare?a=gated&b=${fork.body.runId}`).expect(200)).body as RunComparisonPayload; + expect(comparison.divergedAt).toBe(1); + expect(comparison.b[0].tools[0].result).toBe(JSON.stringify({ done: false, note: 'edited' })); + }); + + it('POST /runs/:id/fork validates its body and maps fork errors', async () => { + await finishedRun('weather'); + const fork = (body: object) => request(app).post('/runs/weather/fork').send(body); + await fork({}).expect(400); + await fork({ fromStep: -1 }).expect(400); + await fork({ fromStep: 1.5 }).expect(400); + await fork({ fromStep: 1, patch: [] }).expect(400); + await fork({ fromStep: 1, patch: { messages: [] } }).expect(400); + await fork({ fromStep: 1, patch: { appendInput: ' ' } }).expect(400); + await fork({ fromStep: 1, patch: { toolResult: 'x' } }).expect(400); + await fork({ fromStep: 1, patch: { toolResult: { toolCallId: 'c1' } } }).expect(400); + const unknownCall = await fork({ fromStep: 1, patch: { toolResult: { toolCallId: 'nope', result: 1 } } }).expect(400); + expect(unknownCall.body.error).toMatch(/no tool call 'nope'/); + const noStep = await fork({ fromStep: 9 }).expect(404); + expect(noStep.body.error).toMatch(/no checkpoint at step 9/); + await request(app).post('/runs/unknown/fork').send({ fromStep: 1 }).expect(404); + }); + + it('GET /runs/compare validates its query and is 404 when a run has no checkpoint', async () => { + await finishedRun('weather'); + await request(app).get('/runs/compare?a=weather').expect(400); + await request(app).get(`/runs/compare?a=weather&b=${encodeURIComponent('../x')}`).expect(400); + await request(app).get('/runs/compare?a=weather&b=missing').expect(404); + const same = (await request(app).get('/runs/compare?a=weather&b=weather').expect(200)).body as RunComparisonPayload; + expect(same.divergedAt).toBeUndefined(); + expect(same.drift).toEqual([]); + }); +}); diff --git a/apps/agent-forge/server/app.ts b/apps/agent-forge/server/app.ts index 52ca0e78a..5d0b0472b 100644 --- a/apps/agent-forge/server/app.ts +++ b/apps/agent-forge/server/app.ts @@ -10,6 +10,9 @@ * POST /agents/:id/stop -> abort the in-flight run, if any * GET /agents/:id/status -> AgentRunStatusPayload * POST /agents/:id/approve -> body: { approvalId, approved, note? } + * GET /runs/:id/history -> RunHistoryPayload (LOU-D45 time travel) + * POST /runs/:id/fork -> body: ForkRunRequest; starts the fork, 202 ForkRunResponse + * GET /runs/compare?a=&b= -> RunComparisonPayload (compareTrajectories) * * `WS /agents/:id/stream` is wired separately in wsServer.ts (attached to * the same underlying http.Server, since `ws` needs the raw HTTP upgrade @@ -27,7 +30,7 @@ import * as fs from 'node:fs'; import * as path from 'node:path'; import express, { type Express, type Request, type Response, type NextFunction } from 'express'; import cors from 'cors'; -import type { AgentSpec } from '@loushy/build-ai-agent'; +import { ConfigurationError, SDKError, type AgentSpec, type ForkPatch } from '@loushy/build-ai-agent'; import { TriggerRegistry } from '@loushy/build-ai-agent/triggers'; import type { AgentStore } from '../src/persistence/AgentStore'; import { @@ -41,7 +44,7 @@ import { ChatTriggerAdapter } from './chatTriggerAdapter'; import { isValidAgentId } from './types'; import { SecretsStore, isSecretProvider } from './secretsStore'; import { SettingsStore } from './settingsStore'; -import type { SettingsProfile } from '../shared/wireTypes'; +import type { ForkRunRequest, ForkRunResponse, SettingsProfile } from '../shared/wireTypes'; import { DEPLOY_ADAPTERS, isDeployAdapter, runDeploy } from './deployRunner'; export interface CreateAppOptions { @@ -190,6 +193,82 @@ function registerRunRoutes(app: Express, runManager: RunManager): void { ); } +/** LOU-D45: the `ForkPatch` fields a fork request may carry, each with its check. */ +const FORK_PATCH_CHECKS = new Map boolean>([ + ['toolResult', (v) => isObject(v) && typeof v.toolCallId === 'string' && 'result' in v], + ['appendInput', (v) => typeof v === 'string' && v.trim() !== ''], + ['businessState', () => true], +]); + +function isObject(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +/** LOU-D45: a validated `POST /runs/:id/fork` body, or the 400 message. */ +function parseForkRequest(body: Partial | undefined): { fromStep: number; patch: ForkPatch } | string { + const { fromStep, patch = {} } = body ?? {}; + if (!Number.isInteger(fromStep) || (fromStep as number) < 0) return "'fromStep' must be a non-negative integer"; + if (!isObject(patch)) return "'patch' must be an object"; + const invalid = Object.entries(patch).find(([key, value]) => !FORK_PATCH_CHECKS.get(key)?.(value)); + if (invalid) { + return `'patch.${invalid[0]}' is invalid: 'patch' takes toolResult { toolCallId: string, result }, appendInput (a non-empty string) and businessState`; + } + return { fromStep: fromStep as number, patch }; +} + +/** LOU-D45 time travel: a run's step history, forking it from a step, and comparing two runs. */ +function registerTimeTravelRoutes(app: Express, runManager: RunManager): void { + app.get( + '/runs/compare', + asyncRoute(async (req, res) => { + const { a, b } = req.query; + if (typeof a !== 'string' || typeof b !== 'string' || !isValidAgentId(a) || !isValidAgentId(b)) { + res.status(400).json({ error: "Query must include valid run ids 'a' and 'b'" }); + return; + } + const comparison = await runManager.compare(a, b); + if (!comparison) { + res.status(404).json({ error: `No checkpoint for run '${a}' or '${b}'` }); + return; + } + res.json(comparison); + }) + ); + + app.get( + '/runs/:id/history', + asyncRoute(async (req, res) => { + const steps = await runManager.history(paramId(req)); + if (!steps) { + res.status(404).json({ error: `No checkpoint history for run '${paramId(req)}'` }); + return; + } + res.json({ runId: paramId(req), steps }); + }) + ); + + app.post( + '/runs/:id/fork', + asyncRoute(async (req, res) => { + const parsed = parseForkRequest(req.body as Partial | undefined); + if (typeof parsed === 'string') { + res.status(400).json({ error: parsed }); + return; + } + try { + const runId = await runManager.fork(paramId(req), parsed.fromStep, parsed.patch); + res.status(202).json({ runId, fromStep: parsed.fromStep, status: runManager.status(runId) } satisfies ForkRunResponse); + } catch (error) { + respondWithMappedError(res, error, [ + [AgentNotFoundError, 404], + [ConfigurationError, 400], + [SDKError, 404], // LOUSHY_CHECKPOINT_NOT_FOUND: no checkpoint at that step + ]); + } + }) + ); +} + /** P1/P3 chat transport and multi-session history. */ function registerChatRoutes(app: Express, runManager: RunManager, triggerRegistry: TriggerRegistry): void { // P1: chat transport. `POST /agents/:id/message` appends a user message @@ -424,7 +503,7 @@ function registerStaticClient(app: Express, staticDir: string | undefined): void // `App.tsx` view state, but still refresh-safe) - serve `index.html` and // let the client app take over, the same as `vite preview`/most SPA // hosts do. - app.get(/^(?!\/health|\/agents).*/, (_req, res) => { + app.get(/^(?!\/health|\/agents|\/runs).*/, (_req, res) => { res.sendFile(indexHtml); }); } @@ -461,7 +540,7 @@ export function createApp({ // trusting each store to sanitize it - closes off path traversal via a // percent-encoded `..%2F..%2F...` id, which Express happily hands to // `req.params.id` as a decoded string containing `/`/`..`. - app.use('/agents/:id', (req, res, next) => { + app.use(['/agents/:id', '/runs/:id'], (req, res, next) => { if (!isValidAgentId(paramId(req))) { res.status(400).json({ error: 'Invalid agent id' }); return; @@ -473,6 +552,7 @@ export function createApp({ registerRunRoutes(app, runManager); registerChatRoutes(app, runManager, triggerRegistry); registerDebugRoutes(app, runManager); + registerTimeTravelRoutes(app, runManager); registerProviderKeyRoutes(app, secrets); registerProfileRoutes(app, settings); registerDeployRoutes(app, agentStore, baseDir); diff --git a/apps/agent-forge/server/chatReconcile.ts b/apps/agent-forge/server/chatReconcile.ts index e1da8b44b..d1025ab6a 100644 --- a/apps/agent-forge/server/chatReconcile.ts +++ b/apps/agent-forge/server/chatReconcile.ts @@ -27,13 +27,18 @@ * fresh ones for genuinely new messages. */ import { randomUUID } from 'node:crypto'; -import type { Message } from '@loushy/build-ai-agent'; +import { textOf, type Message } from '@loushy/build-ai-agent'; import type { ChatMessage } from '../shared/wireTypes'; -function sameMessage(a: Message, b: Message): boolean { +/** The transcript keeps text only (LOU-V11 content parts are flattened with `textOf()`). */ +function toChatMessage(m: Message, id: string, timestamp: string): ChatMessage { + return { ...m, content: textOf(m), id, timestamp }; +} + +function sameMessage(a: ChatMessage, b: Message): boolean { return ( a.role === b.role && - a.content === b.content && + a.content === textOf(b) && a.toolCallId === b.toolCallId && a.toolName === b.toolName && JSON.stringify(a.toolCalls ?? null) === JSON.stringify(b.toolCalls ?? null) @@ -67,14 +72,12 @@ export function reconcileChatMessages( const offset = offset1 > offset0 ? 1 : 0; const matched = offset === 1 ? offset1 : offset0; - const leading: ChatMessage[] = next.slice(0, offset).map((m) => ({ ...m, id: randomUUID(), timestamp: settledAt })); - const matchedPart: ChatMessage[] = prev.slice(0, matched).map((old, idx) => ({ - ...next[offset + idx], - id: old.id, - timestamp: old.timestamp, - })); + const leading: ChatMessage[] = next.slice(0, offset).map((m) => toChatMessage(m, randomUUID(), settledAt)); + const matchedPart: ChatMessage[] = prev + .slice(0, matched) + .map((old, idx) => toChatMessage(next[offset + idx], old.id, old.timestamp)); const trailing: ChatMessage[] = next .slice(offset + matched) - .map((m) => ({ ...m, id: randomUUID(), timestamp: settledAt })); + .map((m) => toChatMessage(m, randomUUID(), settledAt)); return [...leading, ...matchedPart, ...trailing]; } diff --git a/apps/agent-forge/server/checkpointStore.ts b/apps/agent-forge/server/checkpointStore.ts index a30f19438..2c6e274d8 100644 --- a/apps/agent-forge/server/checkpointStore.ts +++ b/apps/agent-forge/server/checkpointStore.ts @@ -16,20 +16,53 @@ * verbatim via `JSON.stringify`/`JSON.parse`, so any JSON-serializable * `businessState` a consumer attaches round-trips through this store for * free, exactly like every other field. + * + * LOU-D45: also keeps the bounded per-session `history()` ring the SDK's + * `LocalStorageCheckpointStore` keeps (same helpers, same semantics), under + * `.loushy/agents//checkpoint-history/.json`, so the + * time-travel panel can list a run's steps and `AgentExecutor.fork()` can + * fork from any of them. */ import * as fs from 'node:fs'; import * as path from 'node:path'; -import type { Checkpoint, CheckpointStore } from '@loushy/build-ai-agent'; +import { + appendToRing, + newestFirst, + resolveHistoryLimit, + toHistoryEntry, + type Checkpoint, + type CheckpointDeleteOptions, + type CheckpointHistoryEntry, + type CheckpointHistoryOptions, + type CheckpointStore, +} from '@loushy/build-ai-agent'; export class FileCheckpointStore implements CheckpointStore { - constructor(private readonly baseDir: string) {} + private readonly historyLimit: number; + + /** @param options.historyLimit checkpoints kept per session in `history()` (default 50, `0` keeps none) */ + constructor( + private readonly baseDir: string, + options: { historyLimit?: number } = {} + ) { + this.historyLimit = resolveHistoryLimit(options.historyLimit); + } + + private dir(agentId: string, kind = 'checkpoints'): string { + return path.join(this.baseDir, '.loushy', 'agents', agentId, kind); + } - private dir(agentId: string): string { - return path.join(this.baseDir, '.loushy', 'agents', agentId, 'checkpoints'); + private filePath(agentId: string, sessionId: string, kind?: string): string { + return path.join(this.dir(agentId, kind), `${sessionId}.json`); } - private filePath(agentId: string, sessionId: string): string { - return path.join(this.dir(agentId), `${sessionId}.json`); + private historyPath(sessionId: string): string { + return this.filePath(this.agentIdFromSessionId(sessionId), sessionId, 'checkpoint-history'); + } + + private readRing(sessionId: string): CheckpointHistoryEntry[] { + const file = this.historyPath(sessionId); + return fs.existsSync(file) ? (JSON.parse(fs.readFileSync(file, 'utf8')) as CheckpointHistoryEntry[]) : []; } /** @@ -50,6 +83,11 @@ export class FileCheckpointStore implements CheckpointStore { const agentId = this.agentIdFromSessionId(sessionId); fs.mkdirSync(this.dir(agentId), { recursive: true }); fs.writeFileSync(this.filePath(agentId, sessionId), JSON.stringify(checkpoint), 'utf8'); + if (this.historyLimit > 0) { + const ring = appendToRing(this.readRing(sessionId), toHistoryEntry(checkpoint), this.historyLimit); + fs.mkdirSync(this.dir(agentId, 'checkpoint-history'), { recursive: true }); + fs.writeFileSync(this.historyPath(sessionId), JSON.stringify(ring), 'utf8'); + } } async load(sessionId: string): Promise { @@ -59,9 +97,13 @@ export class FileCheckpointStore implements CheckpointStore { return JSON.parse(fs.readFileSync(file, 'utf8')) as Checkpoint; } - async delete(sessionId: string): Promise { + async delete(sessionId: string, options: CheckpointDeleteOptions = {}): Promise { const agentId = this.agentIdFromSessionId(sessionId); - const file = this.filePath(agentId, sessionId); - if (fs.existsSync(file)) fs.unlinkSync(file); + fs.rmSync(this.filePath(agentId, sessionId), { force: true }); + if (!options.keepHistory) fs.rmSync(this.historyPath(sessionId), { force: true }); + } + + async history(sessionId: string, options?: CheckpointHistoryOptions): Promise { + return newestFirst(this.readRing(sessionId), options); } } diff --git a/apps/agent-forge/server/runRegistry.ts b/apps/agent-forge/server/runRegistry.ts index 875551da2..305d30c12 100644 --- a/apps/agent-forge/server/runRegistry.ts +++ b/apps/agent-forge/server/runRegistry.ts @@ -22,7 +22,12 @@ import { type Message, resumeAfterApproval, emptyRunUsage, + compareTrajectories, + getCheckpointHistory, type AgentSpec, + type CheckpointHistoryEntry, + type ForkPatch, + type TrajectoryComparison, } from '@loushy/build-ai-agent'; import { buildAgentFromSpec, extractFlowFromSpec } from './buildAgent'; import { withAbortSignal, RunAbortedError } from './abortableProvider'; @@ -43,6 +48,7 @@ import type { ChatStatePayload, ChatSessionMeta, ChatSessionRecord, + RunHistoryStep, } from '../shared/wireTypes'; interface RunEntry { @@ -149,6 +155,22 @@ function toSdkMessage(m: ChatMessage): Message { }; } +/** LOU-D45: a history entry as the History panel lists it. */ +function toHistoryStep({ step, status, savedAt, checkpoint }: CheckpointHistoryEntry): RunHistoryStep { + const call = checkpoint.stepUsage?.filter((usage) => usage.step === step).at(-1); + // The step's model turn is the transcript's last assistant message. + const toolCalls = compareTrajectories(checkpoint, []).a.at(-1)?.tools ?? []; + return { + step, + status, + savedAt, + // checkpoint.finishReason is the previous step's while this one's tool calls run. + finishReason: toolCalls.length > 0 ? 'tool_calls' : checkpoint.finishReason, + toolCalls, + ...(call && { tokens: call.usage.totalTokens, costUsd: call.costUsd }), + }; +} + export class RunManager extends EventEmitter { private readonly entries = new Map(); /** @@ -466,7 +488,7 @@ export class RunManager extends EventEmitter { private async launch( agentId: string, input: string | Message[], - spec: AgentSpec | undefined, + spec?: AgentSpec, options: { skipSystemPromptInjection?: boolean } = {} ): Promise { const existing = this.entries.get(agentId); @@ -747,6 +769,46 @@ export class RunManager extends EventEmitter { .catch((error: unknown) => this.handleRunFailed(agentId, error)); } + /** + * LOU-D45: the steps of run `runId`'s latest execution (a run id is a + * checkpoint session id: the agent id, or a fork's id), oldest first, each + * the newest checkpoint saved at that step - the one fork() forks from. + * Undefined when the run has no history. + */ + async history(runId: string): Promise { + const entries = (await getCheckpointHistory(this.opts.checkpointStore, runId)) ?? []; + const rows: CheckpointHistoryEntry[] = []; + for (const entry of entries) { + const last = rows.at(-1); + if (last && entry.step > last.step) break; // an earlier execution of the conversation + if (last?.step !== entry.step) rows.push(entry); + } + return rows.length > 0 ? rows.reverse().map(toHistoryStep) : undefined; + } + + /** + * LOU-D45: forks run `runId` at `fromStep` (`AgentExecutor.fork()`, with + * `patch`) and starts the fork as a run of its own, under the returned id, + * with the source run's spec - same lifecycle, status and WS stream as any + * other run (`/agents//status`, `WS /agents//stream`). + */ + async fork(runId: string, fromStep: number, patch: ForkPatch): Promise { + const spec = this.entries.get(runId)?.lastSpec ?? (await this.opts.loadSpec(runId)); + if (!spec) throw new AgentNotFoundError(`No agent spec for run '${runId}'`); + const { checkpointStore } = this.opts; + const fork = await AgentExecutor.fork({ sessionId: runId, fromStep, checkpointStore, patch }); + // Not via launch()'s `spec` argument, which would save the fork as an agent. + this.setEntry(fork.sessionId, { lastSpec: spec }); + await this.launch(fork.sessionId, []); + return fork.sessionId; + } + + /** LOU-D45: `compareTrajectories()` of two runs' latest checkpoints, or undefined when either has none. */ + async compare(a: string, b: string): Promise { + const [left, right] = await Promise.all([this.opts.checkpointStore.load(a), this.opts.checkpointStore.load(b)]); + return left && right ? compareTrajectories(left, right) : undefined; + } + /** * O3: replaces the breakpoint set for `agentId`, persisted across runs. * If a run is currently in flight, its live DebugSession is updated too diff --git a/apps/agent-forge/server/tsconfig.json b/apps/agent-forge/server/tsconfig.json index 05b8915d7..0c0e99217 100644 --- a/apps/agent-forge/server/tsconfig.json +++ b/apps/agent-forge/server/tsconfig.json @@ -1,7 +1,9 @@ { "compilerOptions": { "target": "ES2022", - "lib": ["ES2022"], + // DOM as in the SDK's own tsconfig: server tests import SDK sources + // (e.g. the shared checkpoint-history contract) that use fetch's DOM types. + "lib": ["ES2022", "DOM"], "module": "ESNext", "moduleResolution": "bundler", "declaration": false, diff --git a/apps/agent-forge/shared/wireTypes.ts b/apps/agent-forge/shared/wireTypes.ts index 17a12f57d..c4a2057c2 100644 --- a/apps/agent-forge/shared/wireTypes.ts +++ b/apps/agent-forge/shared/wireTypes.ts @@ -10,6 +10,7 @@ * defines the wire shape of the status/events the server reports back to * the app over `GET /agents/:id/status` and `WS /agents/:id/stream`. */ +import type { CheckpointStatus, TrajectoryComparison, TrajectoryStep } from '@loushy/build-ai-agent'; /** * Status pill states the app's agent list / topbar render (LOU-L/M's @@ -221,3 +222,42 @@ export interface DeployResult { stderr: string; command: string; } + +/** + * LOU-D45: one step of a run's checkpoint history (`GET /runs/:id/history`, + * where a run id is the session id: the agent id, or a fork's id) - the + * newest checkpoint saved at that step of the latest execution. + */ +export interface RunHistoryStep { + step: number; + status: CheckpointStatus; + savedAt: string; + /** The model's finish reason for this step: `tool_calls` when it called tools, else as recorded. */ + finishReason?: string; + /** The step's tool calls; `result` is absent while a call has none. */ + toolCalls: TrajectoryStep['tools']; + /** This step's model call, when its usage (and price) is known. */ + tokens?: number; + costUsd?: number; +} + +export interface RunHistoryPayload { + runId: string; + steps: RunHistoryStep[]; +} + +/** LOU-D45: `POST /runs/:id/fork` body - see `ForkPatch` (src/execution/checkpoint.ts). */ +export interface ForkRunRequest { + fromStep: number; + patch?: { toolResult?: { toolCallId: string; result: unknown }; appendInput?: string; businessState?: unknown }; +} + +/** LOU-D45: the started fork; its live status streams on `WS /agents//stream`. */ +export interface ForkRunResponse { + runId: string; + fromStep: number; + status: AgentRunStatusPayload; +} + +/** LOU-D45: `GET /runs/compare?a=&b=` - `compareTrajectories(a, b)`. */ +export type RunComparisonPayload = TrajectoryComparison; diff --git a/docs/agent-forge.md b/docs/agent-forge.md index 5d776c836..3e4ee7dfb 100644 --- a/docs/agent-forge.md +++ b/docs/agent-forge.md @@ -8,8 +8,9 @@ reading and writing the exact same `AgentSpec` YAML that `loushy dev` and ships as part of this SDK's npm package (LOU-S). This doc covers: installation, the `loushy studio` quickstart, a -first-agent walkthrough, hook authoring, and how settings/secrets/deploy -wiring works (and doesn't yet). +first-agent walkthrough, hook authoring, time travel (replaying a run from +a past step), and how settings/secrets/deploy wiring works (and doesn't +yet). ## Installation @@ -162,6 +163,45 @@ the server compiles enabled hooks into a real `HookRegistry` for the run (`server/compileHooks.ts`), routed through the same `SandboxAdapter` a sandboxed tool uses. +## Time travel + +Agent Forge keeps every run's checkpoints, not just the latest: its file +store (`server/checkpointStore.ts`) keeps the same bounded history as the +SDK's `LocalStorageCheckpointStore` (the newest 50 saves per run, under +`.loushy/agents//checkpoint-history/`; see +[Checkpoint history](./durable-execution.md#checkpoint-history)). From +that history you can fork a run at any step, change what happened there, +and replay it next to the original, with +[`AgentExecutor.fork()`](./durable-execution.md#fork-and-replay) and +`compareTrajectories()` doing the work. + +A run id is the run's checkpoint session id: the agent id for the agent's +own runs, `.fork-` for a fork of run ``. The runtime control +server has three routes for it: + +| Route | What it does | +| --- | --- | +| `GET /runs/:id/history` | `{ runId, steps }`: one entry per step of the run's latest execution (the newest checkpoint saved at that step), oldest first, with `status`, `savedAt`, `finishReason`, `toolCalls` (`{ id, name, args, result? }`), and `tokens` / `costUsd` of the step's model call when known. 404 when the run has no checkpoints. | +| `POST /runs/:id/fork` | Body `{ fromStep, patch? }`, where `patch` is `{ toolResult?: { toolCallId, result }, appendInput?, businessState? }`. Forks the run at `fromStep`, then starts the fork through the run registry with the original run's spec. Returns 202 `{ runId, fromStep, status }`; the fork streams on `WS /agents//stream` and reports on `GET /agents//status` like any run. 400 for a bad body or an unknown `toolCallId`, 404 for an unknown run or a step with no checkpoint. | +| `GET /runs/compare?a=&b=` | `compareTrajectories(a, b)` of two runs' latest checkpoints: each run's model turns, `divergedAt` (the first turn that differs) and the `drift` entries. 404 when either run has no checkpoint. | + +The flow, for an agent `weather` that has run once: + +```bash +curl localhost:4750/runs/weather/history +# {"runId":"weather","steps":[{"step":1,"status":"finished","finishReason":"stop","toolCalls":[],"tokens":...}]} + +curl -X POST localhost:4750/runs/weather/fork -H 'content-type: application/json' \ + -d '{"fromStep":1,"patch":{"appendInput":"And in Celsius?"}}' +# {"runId":"weather.fork-1","fromStep":1,"status":{"status":"running",...}} + +curl 'localhost:4750/runs/compare?a=weather&b=weather.fork-1' +# {"a":[...],"b":[...],"divergedAt":2,"drift":[{"field":"steps","committed":"1","current":"2"}]} +``` + +The fork is a run of its own: the original run's checkpoints, status and +chat are unchanged. + ## Settings, secrets and deploy wiring The bottom drawer has a **Settings** tab, but provider API keys, secrets and diff --git a/llms-full.txt b/llms-full.txt index 709d3db4b..da4e40b3a 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -1598,8 +1598,9 @@ reading and writing the exact same `AgentSpec` YAML that `loushy dev` and ships as part of this SDK's npm package (LOU-S). This doc covers: installation, the `loushy studio` quickstart, a -first-agent walkthrough, hook authoring, and how settings/secrets/deploy -wiring works (and doesn't yet). +first-agent walkthrough, hook authoring, time travel (replaying a run from +a past step), and how settings/secrets/deploy wiring works (and doesn't +yet). ## Installation @@ -1752,6 +1753,45 @@ the server compiles enabled hooks into a real `HookRegistry` for the run (`server/compileHooks.ts`), routed through the same `SandboxAdapter` a sandboxed tool uses. +## Time travel + +Agent Forge keeps every run's checkpoints, not just the latest: its file +store (`server/checkpointStore.ts`) keeps the same bounded history as the +SDK's `LocalStorageCheckpointStore` (the newest 50 saves per run, under +`.loushy/agents//checkpoint-history/`; see +[Checkpoint history](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#checkpoint-history)). From +that history you can fork a run at any step, change what happened there, +and replay it next to the original, with +[`AgentExecutor.fork()`](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#fork-and-replay) and +`compareTrajectories()` doing the work. + +A run id is the run's checkpoint session id: the agent id for the agent's +own runs, `.fork-` for a fork of run ``. The runtime control +server has three routes for it: + +| Route | What it does | +| --- | --- | +| `GET /runs/:id/history` | `{ runId, steps }`: one entry per step of the run's latest execution (the newest checkpoint saved at that step), oldest first, with `status`, `savedAt`, `finishReason`, `toolCalls` (`{ id, name, args, result? }`), and `tokens` / `costUsd` of the step's model call when known. 404 when the run has no checkpoints. | +| `POST /runs/:id/fork` | Body `{ fromStep, patch? }`, where `patch` is `{ toolResult?: { toolCallId, result }, appendInput?, businessState? }`. Forks the run at `fromStep`, then starts the fork through the run registry with the original run's spec. Returns 202 `{ runId, fromStep, status }`; the fork streams on `WS /agents//stream` and reports on `GET /agents//status` like any run. 400 for a bad body or an unknown `toolCallId`, 404 for an unknown run or a step with no checkpoint. | +| `GET /runs/compare?a=&b=` | `compareTrajectories(a, b)` of two runs' latest checkpoints: each run's model turns, `divergedAt` (the first turn that differs) and the `drift` entries. 404 when either run has no checkpoint. | + +The flow, for an agent `weather` that has run once: + +```bash +curl localhost:4750/runs/weather/history +# {"runId":"weather","steps":[{"step":1,"status":"finished","finishReason":"stop","toolCalls":[],"tokens":...}]} + +curl -X POST localhost:4750/runs/weather/fork -H 'content-type: application/json' \ + -d '{"fromStep":1,"patch":{"appendInput":"And in Celsius?"}}' +# {"runId":"weather.fork-1","fromStep":1,"status":{"status":"running",...}} + +curl 'localhost:4750/runs/compare?a=weather&b=weather.fork-1' +# {"a":[...],"b":[...],"divergedAt":2,"drift":[{"field":"steps","committed":"1","current":"2"}]} +``` + +The fork is a run of its own: the original run's checkpoints, status and +chat are unchanged. + ## Settings, secrets and deploy wiring The bottom drawer has a **Settings** tab, but provider API keys, secrets and diff --git a/src/execution/__fixtures__/checkpointHistoryContract.ts b/src/execution/__fixtures__/checkpointHistoryContract.ts new file mode 100644 index 000000000..492c897a0 --- /dev/null +++ b/src/execution/__fixtures__/checkpointHistoryContract.ts @@ -0,0 +1,116 @@ +/** + * LOU-D43 `CheckpointStore.history()` contract suite, shared so every store + * that keeps a history runs the same assertions: the SDK's own stores + * (checkpointHistory.contract.test.ts) and Agent Forge's `FileCheckpointStore` + * (apps/agent-forge/server/__tests__/checkpointStore.test.ts, LOU-D45). + */ +import { describe, it, expect } from 'vitest'; +import { DEFAULT_CHECKPOINT_HISTORY_LIMIT, type CheckpointStore } from '../checkpoint'; +import { makeCheckpoint } from '../../storage/sqlite/__fixtures__/storeContracts'; + +/** Returns a fresh, empty store keeping at most `historyLimit` entries per session. */ +export type CheckpointHistoryStoreFactory = (options?: { historyLimit?: number }) => CheckpointStore; + +const save = (store: CheckpointStore, sessionId: string, stepIndex: number, extra = {}) => + store.save(sessionId, makeCheckpoint({ sessionId, stepIndex, ...extra })); + +export function describeCheckpointHistoryContract(name: string, factory: CheckpointHistoryStoreFactory): void { + describe(`CheckpointStore.history contract: ${name}`, () => { + it('is empty for a session that was never saved', async () => { + expect(await factory().history?.('nope')).toEqual([]); + }); + + it('lists every save newest first with step, savedAt, status and the checkpoint', async () => { + const store = factory(); + await save(store, 's', 1); + await save(store, 's', 2, { status: 'awaiting-approval', approvalId: 'a1' }); + await save(store, 's', 3, { status: 'finished', businessState: { orderId: 7 } }); + + const history = (await store.history?.('s')) ?? []; + expect(history.map((entry) => [entry.step, entry.status])).toEqual([ + [3, 'finished'], + [2, 'awaiting-approval'], + [1, 'in-progress'], // a checkpoint without a status counts as in-progress + ]); + expect(history[0].checkpoint).toEqual(makeCheckpoint({ sessionId: 's', stepIndex: 3, status: 'finished', businessState: { orderId: 7 } })); + expect(history[1].checkpoint.approvalId).toBe('a1'); + for (const entry of history) expect(new Date(entry.savedAt).toISOString()).toBe(entry.savedAt); + expect(history[0].savedAt >= history[2].savedAt).toBe(true); + expect((await store.load('s'))?.stepIndex).toBe(3); // load still returns the latest only + }); + + it('keeps two saves of the same step as two entries', async () => { + const store = factory(); + await save(store, 's', 4); + await save(store, 's', 4, { status: 'finished' }); + expect((await store.history?.('s'))?.map((entry) => entry.status)).toEqual(['finished', 'in-progress']); + }); + + it('bounds the ring: the oldest entries are dropped past historyLimit (default 50)', async () => { + const small = factory({ historyLimit: 3 }); + for (let step = 0; step < 5; step++) await save(small, 's', step); + expect((await small.history?.('s'))?.map((entry) => entry.step)).toEqual([4, 3, 2]); + + const standard = factory(); + for (let step = 0; step < DEFAULT_CHECKPOINT_HISTORY_LIMIT + 5; step++) await save(standard, 's', step); + const kept = (await standard.history?.('s')) ?? []; + expect(kept).toHaveLength(DEFAULT_CHECKPOINT_HISTORY_LIMIT); + expect([kept[0].step, kept[kept.length - 1].step]).toEqual([DEFAULT_CHECKPOINT_HISTORY_LIMIT + 4, 5]); + }); + + it('limit returns the newest N entries', async () => { + const store = factory(); + for (let step = 0; step < 5; step++) await save(store, 's', step); + expect((await store.history?.('s', { limit: 2 }))?.map((entry) => entry.step)).toEqual([4, 3]); + expect(await store.history?.('s', { limit: 0 })).toEqual([]); + expect(await store.history?.('s', { limit: 99 })).toHaveLength(5); + }); + + it('keeps sessions apart', async () => { + const store = factory(); + await save(store, 'a', 1); + await save(store, 'b', 2); + await save(store, 'b', 3); + expect((await store.history?.('a'))?.map((entry) => entry.step)).toEqual([1]); + expect((await store.history?.('b'))?.map((entry) => entry.step)).toEqual([3, 2]); + }); + + it('delete() clears the history with the checkpoint', async () => { + const store = factory(); + await save(store, 's', 1); + await save(store, 'other', 1); + await store.delete('s'); + expect(await store.load('s')).toBeNull(); + expect(await store.history?.('s')).toEqual([]); + expect(await store.history?.('other')).toHaveLength(1); + }); + + it('delete({ keepHistory: true }) drops the checkpoint but keeps the history, which keeps growing', async () => { + const store = factory(); + await save(store, 's', 1); + await save(store, 's', 2); + await store.delete('s', { keepHistory: true }); + expect(await store.load('s')).toBeNull(); + expect((await store.history?.('s'))?.map((entry) => entry.step)).toEqual([2, 1]); + await save(store, 's', 3); + expect((await store.history?.('s'))?.map((entry) => entry.step)).toEqual([3, 2, 1]); + }); + + it('returns copies: changing a returned entry changes nothing stored', async () => { + const store = factory(); + await save(store, 's', 1); + const [entry] = (await store.history?.('s')) ?? []; + entry.checkpoint.messages.length = 0; + expect((await store.history?.('s'))?.[0].checkpoint.messages).toHaveLength(2); + }); + + it('historyLimit 0 keeps no history; a negative or fractional one is refused', async () => { + const off = factory({ historyLimit: 0 }); + await save(off, 's', 1); + expect(await off.history?.('s')).toEqual([]); + expect((await off.load('s'))?.stepIndex).toBe(1); + expect(() => factory({ historyLimit: -1 })).toThrow(/historyLimit/); + expect(() => factory({ historyLimit: 1.5 })).toThrow(/historyLimit/); + }); + }); +} diff --git a/src/execution/checkpointHistory.contract.test.ts b/src/execution/checkpointHistory.contract.test.ts index ea9d87003..ccf10263a 100644 --- a/src/execution/checkpointHistory.contract.test.ts +++ b/src/execution/checkpointHistory.contract.test.ts @@ -7,7 +7,6 @@ import { describe, it, expect, afterEach } from 'vitest'; import { getCheckpointHistory, - DEFAULT_CHECKPOINT_HISTORY_LIMIT, LocalStorageCheckpointStore, type CheckpointStore, } from './checkpoint'; @@ -17,15 +16,14 @@ import { SqliteStore } from '../storage/sqlite'; import { KVCheckpointStore } from '../deploy/kvCheckpointStore'; import { createFakeFs } from './__fixtures__/fakeFs'; import { makeCheckpoint } from '../storage/sqlite/__fixtures__/storeContracts'; - -type Factory = (options?: { historyLimit?: number }) => CheckpointStore; +import { describeCheckpointHistoryContract, type CheckpointHistoryStoreFactory } from './__fixtures__/checkpointHistoryContract'; const sqliteStores: SqliteStore[] = []; afterEach(() => { while (sqliteStores.length) sqliteStores.pop()?.close(); }); -const implementations: Array<[string, Factory]> = [ +const implementations: Array<[string, CheckpointHistoryStoreFactory]> = [ ['memoryStore()', (options) => memoryStore(options).checkpoints], [ 'SqliteStore', @@ -44,107 +42,10 @@ const implementations: Array<[string, Factory]> = [ ], ]; -const save = (store: CheckpointStore, sessionId: string, stepIndex: number, extra = {}) => - store.save(sessionId, makeCheckpoint({ sessionId, stepIndex, ...extra })); - -describe.each(implementations)('CheckpointStore.history contract: %s', (_name, factory) => { - it('is empty for a session that was never saved', async () => { - expect(await factory().history?.('nope')).toEqual([]); - }); - - it('lists every save newest first with step, savedAt, status and the checkpoint', async () => { - const store = factory(); - await save(store, 's', 1); - await save(store, 's', 2, { status: 'awaiting-approval', approvalId: 'a1' }); - await save(store, 's', 3, { status: 'finished', businessState: { orderId: 7 } }); - - const history = (await store.history?.('s')) ?? []; - expect(history.map((entry) => [entry.step, entry.status])).toEqual([ - [3, 'finished'], - [2, 'awaiting-approval'], - [1, 'in-progress'], // a checkpoint without a status counts as in-progress - ]); - expect(history[0].checkpoint).toEqual(makeCheckpoint({ sessionId: 's', stepIndex: 3, status: 'finished', businessState: { orderId: 7 } })); - expect(history[1].checkpoint.approvalId).toBe('a1'); - for (const entry of history) expect(new Date(entry.savedAt).toISOString()).toBe(entry.savedAt); - expect(history[0].savedAt >= history[2].savedAt).toBe(true); - expect((await store.load('s'))?.stepIndex).toBe(3); // load still returns the latest only - }); - - it('keeps two saves of the same step as two entries', async () => { - const store = factory(); - await save(store, 's', 4); - await save(store, 's', 4, { status: 'finished' }); - expect((await store.history?.('s'))?.map((entry) => entry.status)).toEqual(['finished', 'in-progress']); - }); - - it('bounds the ring: the oldest entries are dropped past historyLimit (default 50)', async () => { - const small = factory({ historyLimit: 3 }); - for (let step = 0; step < 5; step++) await save(small, 's', step); - expect((await small.history?.('s'))?.map((entry) => entry.step)).toEqual([4, 3, 2]); - - const standard = factory(); - for (let step = 0; step < DEFAULT_CHECKPOINT_HISTORY_LIMIT + 5; step++) await save(standard, 's', step); - const kept = (await standard.history?.('s')) ?? []; - expect(kept).toHaveLength(DEFAULT_CHECKPOINT_HISTORY_LIMIT); - expect([kept[0].step, kept[kept.length - 1].step]).toEqual([DEFAULT_CHECKPOINT_HISTORY_LIMIT + 4, 5]); - }); - - it('limit returns the newest N entries', async () => { - const store = factory(); - for (let step = 0; step < 5; step++) await save(store, 's', step); - expect((await store.history?.('s', { limit: 2 }))?.map((entry) => entry.step)).toEqual([4, 3]); - expect(await store.history?.('s', { limit: 0 })).toEqual([]); - expect(await store.history?.('s', { limit: 99 })).toHaveLength(5); - }); - - it('keeps sessions apart', async () => { - const store = factory(); - await save(store, 'a', 1); - await save(store, 'b', 2); - await save(store, 'b', 3); - expect((await store.history?.('a'))?.map((entry) => entry.step)).toEqual([1]); - expect((await store.history?.('b'))?.map((entry) => entry.step)).toEqual([3, 2]); - }); - - it('delete() clears the history with the checkpoint', async () => { - const store = factory(); - await save(store, 's', 1); - await save(store, 'other', 1); - await store.delete('s'); - expect(await store.load('s')).toBeNull(); - expect(await store.history?.('s')).toEqual([]); - expect(await store.history?.('other')).toHaveLength(1); - }); - - it('delete({ keepHistory: true }) drops the checkpoint but keeps the history, which keeps growing', async () => { - const store = factory(); - await save(store, 's', 1); - await save(store, 's', 2); - await store.delete('s', { keepHistory: true }); - expect(await store.load('s')).toBeNull(); - expect((await store.history?.('s'))?.map((entry) => entry.step)).toEqual([2, 1]); - await save(store, 's', 3); - expect((await store.history?.('s'))?.map((entry) => entry.step)).toEqual([3, 2, 1]); - }); - - it('returns copies: changing a returned entry changes nothing stored', async () => { - const store = factory(); - await save(store, 's', 1); - const [entry] = (await store.history?.('s')) ?? []; - entry.checkpoint.messages.length = 0; - expect((await store.history?.('s'))?.[0].checkpoint.messages).toHaveLength(2); - }); +for (const [name, factory] of implementations) describeCheckpointHistoryContract(name, factory); - it('historyLimit 0 keeps no history; a negative or fractional one is refused', async () => { - const off = factory({ historyLimit: 0 }); - await save(off, 's', 1); - expect(await off.history?.('s')).toEqual([]); - expect((await off.load('s'))?.stepIndex).toBe(1); - expect(() => factory({ historyLimit: -1 })).toThrow(/historyLimit/); - expect(() => factory({ historyLimit: 1.5 })).toThrow(/historyLimit/); - }); -}); +const save = (store: CheckpointStore, sessionId: string, stepIndex: number) => + store.save(sessionId, makeCheckpoint({ sessionId, stepIndex })); describe('getCheckpointHistory()', () => { it('reads the history of a store that keeps one', async () => {