From 3564c9acc8ba2f4c76fee1169078ecf3d1cfd376 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 10:30:08 +0300 Subject: [PATCH 1/2] LOU-U7, LOU-U8, LOU-U9: durable execution gaps LOU-U7: an approval in the middle of a tool batch kept the transcript invalid - calls after the paused one never ran and got no result. The approval snapshot now records `remainingToolCalls`; resumeAfterApproval() records the paused call's result (or rejection) and AgentExecutor runs the remaining calls through the normal batch path, so they can run, fail validation or pause again (the resumed run defaults to the same approvalStore). Snapshots without the field resume as before (later calls do not run) but each unanswered call gets a "not run" error result. LOU-U8: execute() with a sessionId now honours new input. Checkpoints carry a `status`: an unfinished run resumes and new input is queued behind its pending tool results (a re-sent starting message is a retry, not appended); a finished run is kept (no longer deleted) and continues as a conversation; a run awaiting approval throws SessionAwaitingApprovalError. LOU-U9: the model's tool-call turn is checkpointed before any tool runs, and a resumed transcript whose last turn has unanswered calls runs exactly those calls without calling the model. Tools receive `toolCallId` in their execute options (non-sandbox path) as an idempotency key for the at-least-once re-run. Tests that encoded the old behaviour (checkpoint deleted on finish, input ignored on resume, later batch calls not run, empty execute options, checkpoint save counts) were updated. Docs: docs/durable-execution.md. Co-Authored-By: Claude Opus 5.5 --- README.md | 20 +- .../server/__tests__/runRegistry.test.ts | 19 +- apps/agent-forge/server/abortableProvider.ts | 9 +- apps/agent-forge/server/compileHooks.test.ts | 2 +- apps/agent-forge/server/runRegistry.ts | 32 +- docs/api-overview.md | 37 +- docs/durable-execution.md | 184 +++++++ llms-full.txt | 244 +++++++++- llms.txt | 1 + .../adapters/cloudflare.checkpoint.test.ts | 29 +- src/deploy/kvCheckpointStore.test.ts | 22 +- src/execution/AgentExecutor.test.ts | 47 +- src/execution/AgentExecutor.ts | 152 ++++-- src/execution/ApprovalGate.test.ts | 19 + src/execution/ApprovalGate.ts | 13 +- src/execution/agentRunState.ts | 106 ++-- src/execution/cancellation.test.ts | 4 +- src/execution/checkpoint.test.ts | 18 + src/execution/checkpoint.ts | 27 +- src/execution/durableExecution.test.ts | 457 ++++++++++++++++++ src/execution/errors.ts | 32 ++ src/execution/parallelToolCalls.test.ts | 15 +- src/execution/resume.test.ts | 31 +- src/execution/resume.ts | 41 +- src/execution/sandboxGuard.ts | 9 +- src/execution/toolCallExecution.ts | 3 +- src/execution/transcript.ts | 113 +++++ 27 files changed, 1479 insertions(+), 207 deletions(-) create mode 100644 docs/durable-execution.md create mode 100644 src/execution/durableExecution.test.ts create mode 100644 src/execution/transcript.ts diff --git a/README.md b/README.md index a1b25f4b6..7d292baeb 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,7 @@ console.log(result.usage.totalTokens, result.finishReason, result.steps); - [Configuration](docs/configuration.md) - agent spec fields, provider env vars, `AgentExecutor.execute()` options, CLI flags - [Deployment](docs/deployment.md) - `loushy build` targets: Node server, Docker, Cloudflare Workers - [API Overview](docs/api-overview.md) - the main exports; `npm run docs:build` generates the full TypeDoc reference +- [Durable execution](docs/durable-execution.md) - checkpoints, crash resume, multi-turn sessions, approvals mid-batch, at-least-once tools - [Tracing and observability](docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in - [Testing](docs/testing.md) - unit-test agents deterministically with the scripted `mockModel` - [Skills](docs/skills.md) - on-demand instructions: `defineSkill()`, `loadSkills()`, how they save context @@ -354,21 +355,26 @@ const result = await resumeAfterApproval( ### Durable execution / checkpoints -Pass a `sessionId` and a `checkpointStore`; `AgentExecutor` saves a -checkpoint after every tool result and rehydrates from it on the next call -with the same `sessionId` — so a crash mid-conversation resumes rather than -restarts: +Pass a `sessionId` and a `checkpointStore`; `AgentExecutor` checkpoints +after every model response and every tool result, on any host. The next +call with the same `sessionId` resumes an unfinished run (running only the +tool calls that have no result yet, without asking the model again), +continues a finished one as a multi-turn conversation, and refuses to +bypass a pending approval: ```typescript import { AgentExecutor, LocalStorageCheckpointStore } from '@loushy/build-ai-agent'; const checkpointStore = new LocalStorageCheckpointStore(storage); -await AgentExecutor.execute({ agent, input, provider, sessionId: 'session-123', checkpointStore }); -// ...process restarts... -await AgentExecutor.execute({ agent, input: 'continue', provider, sessionId: 'session-123', checkpointStore }); +await AgentExecutor.execute({ agent, input: 'Book a table for 2', provider, sessionId: 'session-123', checkpointStore }); +// ...process restarts, or the user simply replies... +await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'session-123', checkpointStore }); ``` +See [Durable execution](docs/durable-execution.md) for the exact guarantees +(including the at-least-once caveat for tools that were mid-flight). + ### Multi-agent delegation Wrap a child `AgentConfig` as a tool so a parent agent can delegate a task to diff --git a/apps/agent-forge/server/__tests__/runRegistry.test.ts b/apps/agent-forge/server/__tests__/runRegistry.test.ts index 536c0d04c..3fffed88a 100644 --- a/apps/agent-forge/server/__tests__/runRegistry.test.ts +++ b/apps/agent-forge/server/__tests__/runRegistry.test.ts @@ -99,24 +99,25 @@ describe('RunManager', () => { await runManager.run(agentId, 'please use current-date', SPEC); await waitForStatus(runManager, agentId, (s) => s.status === 'stopped'); - // The checkpoint from the aborted run must survive - AgentExecutor only - // deletes it on a successful terminal completion, never on an aborted - // run (see abortableProvider.ts's doc comment). + // The checkpoint from the aborted run must survive, still unfinished + // (see abortableProvider.ts's doc comment). const checkpoint = await checkpointStore.load(agentId); expect(checkpoint).not.toBeNull(); expect(checkpoint!.stepIndex).toBeGreaterThanOrEqual(1); // Running again picks the same checkpoint back up (AgentExecutor - // rehydrates from it and ignores the new `input`) rather than starting - // a brand new conversation from scratch. - await runManager.run(agentId, 'this input is ignored on resume', SPEC); + // rehydrates from it; LOU-U8 appends the new `input` after the resumed + // turn) rather than starting a brand new conversation from scratch. + await runManager.run(agentId, 'a follow-up appended on resume', SPEC); const resumed = await waitForStatus(runManager, agentId, (s) => s.status === 'stopped'); expect(resumed.resultText).toBe('This is a mock response.'); - // And the checkpoint is cleared once that resumed run completes - // successfully. + // And the checkpoint is kept, marked finished, once that resumed run + // completes successfully (LOU-U8: the next run continues the session). const afterResume = await checkpointStore.load(agentId); - expect(afterResume).toBeNull(); + expect(afterResume?.status).toBe('finished'); + expect(afterResume?.messages.map((m) => m.content)).toContain('please use current-date'); + expect(afterResume?.messages.map((m) => m.content)).toContain('a follow-up appended on resume'); }); it('emits structured log entries derived from the ExecutionEvent stream (O1)', async () => { diff --git a/apps/agent-forge/server/abortableProvider.ts b/apps/agent-forge/server/abortableProvider.ts index 13122e093..5cc2c96c7 100644 --- a/apps/agent-forge/server/abortableProvider.ts +++ b/apps/agent-forge/server/abortableProvider.ts @@ -22,11 +22,10 @@ * provider is itself abort-aware) throws, which propagates out of * AgentExecutor.execute()'s catch block (it only emits an 'error' event and * rethrows - it does not swallow) as a rejected promise. Crucially, - * AgentExecutor only clears the CheckpointStore entry on a *successful* - * terminal completion (see the bottom of runAgentLoop(), reached only after - * the while-loop `break`s) - never from the catch block - so an abort - * leaves the last-saved checkpoint (written after each completed tool - * result) intact. The next `run()` call for the same agent reuses the same + * AgentExecutor only marks the CheckpointStore entry 'finished' on a + * *successful* terminal completion - never from the catch block - so an + * abort leaves the last-saved 'in-progress' checkpoint (written after each + * model response and each completed tool result) intact. The next `run()` call for the same agent reuses the same * sessionId, so AgentExecutor.execute() rehydrates from exactly that * checkpoint instead of starting over. This is what makes Stop-then-Run * "resume from the last checkpoint" rather than "restart from scratch". diff --git a/apps/agent-forge/server/compileHooks.test.ts b/apps/agent-forge/server/compileHooks.test.ts index fe28e7b6f..950dde889 100644 --- a/apps/agent-forge/server/compileHooks.test.ts +++ b/apps/agent-forge/server/compileHooks.test.ts @@ -82,7 +82,7 @@ describe('compileHooksFromSpecPolicy', () => { maxSteps: 1, }); - expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, {}); + expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, { toolCallId: 'call-1' }); }); it('compiles a preGenerate hook that actually injects a message when the agent runs', async () => { diff --git a/apps/agent-forge/server/runRegistry.ts b/apps/agent-forge/server/runRegistry.ts index 75a90690b..b8afe5dd1 100644 --- a/apps/agent-forge/server/runRegistry.ts +++ b/apps/agent-forge/server/runRegistry.ts @@ -354,19 +354,16 @@ export class RunManager extends EventEmitter { * what makes Stop-then-Run resume from the last checkpoint (N3) instead * of starting over: AgentExecutor.execute() (see its doc comments) * rehydrates from `checkpointStore.load(sessionId)` when a checkpoint - * exists under that id, and only clears it on a *successful* terminal - * completion - never when the run was aborted (see abortableProvider.ts). - * So after a stop(), the checkpoint from the last completed tool result - * is still there, and the very next run() call for this agentId picks it + * exists under that id, and only marks it 'finished' on a *successful* + * terminal completion - never when the run was aborted (see + * abortableProvider.ts). So after a stop(), the unfinished checkpoint is + * still there, and the very next run() call for this agentId picks it * back up automatically. * - * One consequence worth calling out: when a checkpoint exists, - * AgentExecutor.execute() ignores `input` entirely and continues the - * rehydrated conversation - this mirrors resumeAfterApproval()'s - * documented behavior for the approval-gate case and is intentional: a - * resumed run is a continuation of the same paused conversation, not a - * new question. `input` is only actually used to seed a genuinely fresh - * run (no prior checkpoint). + * LOU-U8: `input` is appended to the rehydrated conversation as a new + * user turn (after the interrupted turn's pending tool results), unless + * it re-sends the message the interrupted run started from; after a + * finished run it continues the conversation. */ async run(agentId: string, input: string, spec?: AgentSpec): Promise { const existing = this.entries.get(agentId); @@ -402,15 +399,10 @@ export class RunManager extends EventEmitter { * string instead, so AgentExecutor's normal buildMessages() path builds * the system+user turn from scratch exactly like a Topbar "Run" click. * - * Known limitation (left for LOU-Q): if the agent was Stopped mid-flight - * (a checkpoint exists from an in-progress run, not a completed one), - * AgentExecutor.execute() ignores `input` entirely and resumes straight - * from that checkpoint (see run()'s original doc comment above) - so a - * chat message sent in that state is shown optimistically in the thread - * but is NOT actually incorporated into the resumed run's conversation. - * This mirrors existing Stop→Run checkpoint-resume behavior rather than - * inventing new semantics for it; a future epic may want a dedicated - * "discard checkpoint and continue with this new message instead" action. + * The checkpoint under this agent's sessionId already holds that same + * history; AgentExecutor.execute() (LOU-U8) recognises the re-sent prefix + * and appends only the new user turn - after a Stop mid-flight, once the + * interrupted turn's pending tool results are recorded. */ async sendMessage(agentId: string, text: string, spec?: AgentSpec): Promise { const existing = this.entries.get(agentId) ?? this.freshEntry(agentId); diff --git a/docs/api-overview.md b/docs/api-overview.md index cc73d3783..03f94f4c4 100644 --- a/docs/api-overview.md +++ b/docs/api-overview.md @@ -17,6 +17,7 @@ npm run docs:build # writes docs/api/index.html | `AgentExecutor.execute(opts)` | Static executor: runs an agent (LLM + tool-calling loop) and resolves to an `ExecutionResult`. | | `AgentType` | Agent type enum (e.g. `AgentType.SmartAssistant`). | | `resumeAfterApproval()` | Resume an execution paused for human approval. | +| `SessionAwaitingApprovalError` | Thrown by `execute()` when its `sessionId` is paused on an approval (see [Durable execution](./durable-execution.md)). | | `createDelegateTool()` | Wrap a child agent as a tool for multi-agent delegation. | ### Skills @@ -26,6 +27,18 @@ Pass `skills: [defineSkill({ name, description, content }), ...(await loadSkills the system prompt and the model loads bodies through an auto-registered `load_skill` tool. See [Skills](./skills.md). +### Durable execution + +`sessionId` + `checkpointStore` make a run crash-safe and a session +multi-turn: the run is checkpointed after every model response, every tool +result and every pause, and calling `execute()` again with the same +`sessionId` resumes an unfinished run (without re-calling the model for a +turn it already has), continues a finished conversation with the new +input, or throws `SessionAwaitingApprovalError` while an approval is +pending. Tools run at-least-once across a crash; `execute` receives the +call's `toolCallId` to use as an idempotency key. See +[Durable execution](./durable-execution.md) for the exact guarantees. + ### Cancellation Pass an `AbortSignal` to stop a run: `agent.send(input, { signal })`, @@ -65,9 +78,11 @@ How it behaves: - `onEvent` receives an `abort` event (its `abortReason` is the signal's `reason`), then `finish` with `finishReason: 'aborted'`. - With `sessionId` + `checkpointStore`, the state is checkpointed. Calling - `execute()` again with the same `sessionId` resumes where the run stopped. - Tool calls the run never reached get an `{ error }` result saying they were - cancelled, so the conversation stays valid for the provider. + `execute()` again with the same `sessionId` resumes where the run stopped; + new `input` is appended as the next user message (see + [Durable execution](./durable-execution.md)). Tool calls the run never + reached get an `{ error }` result saying they were cancelled, so the + conversation stays valid for the provider. - An already-aborted signal returns at once without calling the provider. ### Parallel tool calls @@ -110,10 +125,10 @@ Guarantees, whatever the limit: - **Approvals.** The first call that needs approval stops the batch: the calls before it run (concurrently) and their results are recorded, then the run pauses on that call (`finishReason: 'awaiting-approval'`). Calls after - it never start in this run. **Known gap:** after `resumeAfterApproval()` - those later calls are still not run and get no result (tracked as LOU-U7). - Until that lands, use `toolConcurrency: 1` and keep approval tools out of - multi-call turns if this matters to you. + it never start in this run. `resumeAfterApproval()` records the paused + call's result (or rejection) and then runs those later calls the same way, + so every call of the turn gets exactly one result - see + [Durable execution](./durable-execution.md#approvals-in-the-middle-of-a-tool-batch). - **Failures are isolated.** A tool that throws gets its own error result; its siblings carry on. A propagating error (`PropagatingToolError`, such as the delegation depth guard, or a throwing hook) stops new calls from @@ -123,9 +138,11 @@ Guarantees, whatever the limit: `finishReason: 'aborted'`. Calls that finished keep their results; the rest get a "cancelled" result. Running tools see the abort through their `abortSignal`. -- **Checkpoints.** With `sessionId` + `checkpointStore`, a checkpoint is - written each time the in-order run of finished calls grows (with `1`, after - every call, as before). A resumed run never runs a recorded call again. +- **Checkpoints.** With `sessionId` + `checkpointStore`, the model's turn is + checkpointed before any call starts, then again each time the in-order run + of finished calls grows (with `1`, after every call). A resumed run never + runs a recorded call again, and never asks the model again for a turn it + already checkpointed. ## Declarative specs diff --git a/docs/durable-execution.md b/docs/durable-execution.md new file mode 100644 index 000000000..5ae852f2a --- /dev/null +++ b/docs/durable-execution.md @@ -0,0 +1,184 @@ +# Durable execution + +Pass a `sessionId` and a `checkpointStore` to `AgentExecutor.execute()` and +the run survives a crash, a restart, an abort or a human approval that takes +days. Nothing here depends on a particular host: any `CheckpointStore` +(`LocalStorageCheckpointStore` on Node, the Cloudflare KV store the +`cloudflare-worker` build wires up, or your own three-method implementation) +and any `ApprovalStore` work the same way. + +```ts +import { AgentExecutor, LocalStorageCheckpointStore } from '@loushy/build-ai-agent'; + +const checkpoints = new LocalStorageCheckpointStore(storage); + +// First message of a conversation. +await AgentExecutor.execute({ agent, input: 'Book a table for 2 tonight', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); + +// Later - same process or another one - the next message of the same conversation. +await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); +``` + +## What is checkpointed, and when + +A checkpoint is the run's whole transcript plus its step count, usage, +`businessState` and a `status`. It is written under the `sessionId`: + +| Moment | `status` | +| ------ | -------- | +| Right after each model response that asks for tools, before any tool runs | `'in-progress'` | +| Each time a tool result is recorded (results are recorded in call order) | `'in-progress'` | +| When the run is aborted (`signal`) | `'in-progress'` | +| When the run pauses for an approval | `'awaiting-approval'` (with `approvalId`) | +| When the run finishes (the model stopped calling tools, or `maxSteps` ran out) | `'finished'` | + +A run that rejects (a provider error, a throwing hook, a +`PropagatingToolError`, a process crash) keeps its last `'in-progress'` +checkpoint. + +## What happens when you call `execute()` again with the same `sessionId` + +It depends on the stored `status`: + +- **Unfinished run** (`'in-progress'`: it crashed, was aborted, or a store + write failed). The run resumes. If its last model turn has tool calls + without a result, exactly those calls run first, through the normal tool + path (argument validation, hooks, `needsApproval`, `toolConcurrency`), + **without calling the model again**. Then the loop continues, with the + remaining `maxSteps` budget. + - New `input` is appended as a user message after those tool results. + It is never placed between a tool-call turn and its results, so the + transcript stays valid for every provider. This is what lets a user + interrupt a run (abort it) and redirect it with a new message. + - `input` that re-sends the message the run started from (the natural + "retry the same request after a crash") is treated as a retry and is + not appended again. `input: []` also just resumes. +- **Finished run** (`'finished'`). The session continues as a conversation: + the stored messages are kept and `input` becomes the next user turn. Pass + only the new message(s); if you re-send the stored history followed by + new messages, the history is recognised and not duplicated. `steps`, + `usage` and `toolCalls` on the result count from zero for the new run; + `messages` is the whole conversation. +- **Paused awaiting approval** (`'awaiting-approval'`). `execute()` throws + `SessionAwaitingApprovalError` (with `sessionId` and `approvalId`) and + does not call the model or any tool: new input must not bypass a pending + decision. Resolve it with `resumeAfterApproval()` first, passing the same + `checkpointStore`, then send the new message. + +System messages in `input` are ignored when it is added to a stored +session (the session already has its system prompt). To end a session and +start over under the same id, call `checkpointStore.delete(sessionId)`. + +```ts +import { AgentExecutor, SessionAwaitingApprovalError } from '@loushy/build-ai-agent'; + +try { + await AgentExecutor.execute({ agent, input: 'Are you done?', provider, sessionId: 'chat-42', checkpointStore }); +} catch (error) { + if (error instanceof SessionAwaitingApprovalError) { + console.log(`Waiting on approval ${error.approvalId} - resolve it with resumeAfterApproval() first.`); + } +} +``` + +## Approvals in the middle of a tool batch + +When one model turn asks for several tools and one of them needs approval, +the calls before it run and are recorded, and the run pauses on it. The +approval snapshot records the calls after it (`remainingToolCalls`). +`resumeAfterApproval()` then: + +1. records the paused call's result - the tool's result if approved, or a + structured `{ error, note }` rejection result if rejected; +2. runs the remaining calls through the same batch logic: they can run, + fail validation, or pause the run again on another approval (a new + `approvalId`; resolve it the same way, as many times as needed); +3. calls the model once every call of the turn has exactly one result. + +```ts +import { AgentExecutor, resumeAfterApproval } from '@loushy/build-ai-agent'; + +const paused = await AgentExecutor.execute({ + agent, input, provider, toolRegistry, approvalStore, sessionId: 'chat-42', checkpointStore, +}); + +if (paused.finishReason === 'awaiting-approval') { + // ...a human decides, possibly days later, in another process... + const result = await resumeAfterApproval( + { id: paused.approvalId!, approved: true }, + approvalStore, + toolRegistry, + provider, + {}, + checkpointStore, // clears the 'awaiting-approval' mark and keeps checkpointing + ); + console.log(result.finishReason); // 'stop', or 'awaiting-approval' again for the next gated call +} +``` + +The resumed run uses the same `approvalStore` for any later approval unless +you pass a different one in the options. + +## Guarantees + +- **Every tool call gets exactly one result, in call order**, after any + sequence of crashes, aborts, pauses and resumes. (An aborted call gets a + "cancelled" result; a rejected call gets a rejection result.) +- **A recorded result is never re-executed.** Results are checkpointed as + the in-order prefix of finished calls grows; a resumed run only runs calls + that have no recorded result. +- **A checkpointed model turn is never generated again.** If the process + dies after the model answered, the resumed run executes that answer's tool + calls instead of asking the model again (no extra cost, no different + plan). +- **The provider always receives a valid transcript**: tool results + directly follow their tool-call turn, and new user input comes after them. + +## At-least-once tools: make side effects idempotent + +A tool that was **running** when the process died has no recorded result, +so it runs again on resume. Tool execution is therefore at-least-once, not +exactly-once. For tools with side effects (charging a card, sending an +email), either gate them with `needsApproval` or make them idempotent. +Each call's `toolCallId` (the model's id for the call, unchanged when it is +re-run on resume) is passed to `execute`, so it works as an idempotency +key: + +```ts +import { defineTool } from '@loushy/build-ai-agent'; +import { z } from 'zod'; + +declare const payments: { charge(amount: number, opts: { idempotencyKey: string }): Promise<{ id: string }> }; + +const chargeCard = defineTool({ + name: 'charge_card', + description: 'Charge the customer', + input: z.object({ amount: z.number() }), + execute: async ({ amount }, { toolCallId }) => + payments.charge(amount, { idempotencyKey: toolCallId }), // a re-run cannot charge twice +}); +``` + +(`toolCallId` is set for tools run by `AgentExecutor` and by +`resumeAfterApproval()`, but not yet for tools that route through +`sandboxExecute`.) + +## Compatibility with stored data + +- Checkpoints written before `status` existed are treated as + `'in-progress'` (before this change, finished runs deleted their + checkpoint, so only unfinished runs were ever stored). +- Approval snapshots written before `remainingToolCalls` existed resume as + they used to: the calls after the paused one do not run. Each of them gets + an error result saying it was not run, so the transcript stays valid. + +## Limits + +- A `sessionId` identifies one logical conversation; running two + `execute()` calls on the same id at the same time is not supported. +- If you call `resumeAfterApproval()` without the `checkpointStore`, the + session keeps its `'awaiting-approval'` mark and later `execute()` calls + keep throwing `SessionAwaitingApprovalError`; pass the store (or delete + the checkpoint). +- Workers KV is eventually consistent across locations - see + [Deployment](./deployment.md) for the caveat. diff --git a/llms-full.txt b/llms-full.txt index b02923f09..40d4de748 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -210,6 +210,7 @@ console.log(result.usage.totalTokens, result.finishReason, result.steps); - [Configuration](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md) - agent spec fields, provider env vars, `AgentExecutor.execute()` options, CLI flags - [Deployment](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md) - `loushy build` targets: Node server, Docker, Cloudflare Workers - [API Overview](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/api-overview.md) - the main exports; `npm run docs:build` generates the full TypeDoc reference +- [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md) - checkpoints, crash resume, multi-turn sessions, approvals mid-batch, at-least-once tools - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in - [Testing](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/testing.md) - unit-test agents deterministically with the scripted `mockModel` - [Skills](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/skills.md) - on-demand instructions: `defineSkill()`, `loadSkills()`, how they save context @@ -352,21 +353,26 @@ const result = await resumeAfterApproval( ### Durable execution / checkpoints -Pass a `sessionId` and a `checkpointStore`; `AgentExecutor` saves a -checkpoint after every tool result and rehydrates from it on the next call -with the same `sessionId` — so a crash mid-conversation resumes rather than -restarts: +Pass a `sessionId` and a `checkpointStore`; `AgentExecutor` checkpoints +after every model response and every tool result, on any host. The next +call with the same `sessionId` resumes an unfinished run (running only the +tool calls that have no result yet, without asking the model again), +continues a finished one as a multi-turn conversation, and refuses to +bypass a pending approval: ```typescript import { AgentExecutor, LocalStorageCheckpointStore } from '@loushy/build-ai-agent'; const checkpointStore = new LocalStorageCheckpointStore(storage); -await AgentExecutor.execute({ agent, input, provider, sessionId: 'session-123', checkpointStore }); -// ...process restarts... -await AgentExecutor.execute({ agent, input: 'continue', provider, sessionId: 'session-123', checkpointStore }); +await AgentExecutor.execute({ agent, input: 'Book a table for 2', provider, sessionId: 'session-123', checkpointStore }); +// ...process restarts, or the user simply replies... +await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'session-123', checkpointStore }); ``` +See [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md) for the exact guarantees +(including the at-least-once caveat for tools that were mid-flight). + ### Multi-agent delegation Wrap a child `AgentConfig` as a tool so a parent agent can delegate a task to @@ -1033,6 +1039,7 @@ npm run docs:build # writes docs/api/index.html | `AgentExecutor.execute(opts)` | Static executor: runs an agent (LLM + tool-calling loop) and resolves to an `ExecutionResult`. | | `AgentType` | Agent type enum (e.g. `AgentType.SmartAssistant`). | | `resumeAfterApproval()` | Resume an execution paused for human approval. | +| `SessionAwaitingApprovalError` | Thrown by `execute()` when its `sessionId` is paused on an approval (see [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md)). | | `createDelegateTool()` | Wrap a child agent as a tool for multi-agent delegation. | ### Skills @@ -1042,6 +1049,18 @@ Pass `skills: [defineSkill({ name, description, content }), ...(await loadSkills the system prompt and the model loads bodies through an auto-registered `load_skill` tool. See [Skills](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/skills.md). +### Durable execution + +`sessionId` + `checkpointStore` make a run crash-safe and a session +multi-turn: the run is checkpointed after every model response, every tool +result and every pause, and calling `execute()` again with the same +`sessionId` resumes an unfinished run (without re-calling the model for a +turn it already has), continues a finished conversation with the new +input, or throws `SessionAwaitingApprovalError` while an approval is +pending. Tools run at-least-once across a crash; `execute` receives the +call's `toolCallId` to use as an idempotency key. See +[Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md) for the exact guarantees. + ### Cancellation Pass an `AbortSignal` to stop a run: `agent.send(input, { signal })`, @@ -1081,9 +1100,11 @@ How it behaves: - `onEvent` receives an `abort` event (its `abortReason` is the signal's `reason`), then `finish` with `finishReason: 'aborted'`. - With `sessionId` + `checkpointStore`, the state is checkpointed. Calling - `execute()` again with the same `sessionId` resumes where the run stopped. - Tool calls the run never reached get an `{ error }` result saying they were - cancelled, so the conversation stays valid for the provider. + `execute()` again with the same `sessionId` resumes where the run stopped; + new `input` is appended as the next user message (see + [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md)). Tool calls the run never + reached get an `{ error }` result saying they were cancelled, so the + conversation stays valid for the provider. - An already-aborted signal returns at once without calling the provider. ### Parallel tool calls @@ -1126,10 +1147,10 @@ Guarantees, whatever the limit: - **Approvals.** The first call that needs approval stops the batch: the calls before it run (concurrently) and their results are recorded, then the run pauses on that call (`finishReason: 'awaiting-approval'`). Calls after - it never start in this run. **Known gap:** after `resumeAfterApproval()` - those later calls are still not run and get no result (tracked as LOU-U7). - Until that lands, use `toolConcurrency: 1` and keep approval tools out of - multi-call turns if this matters to you. + it never start in this run. `resumeAfterApproval()` records the paused + call's result (or rejection) and then runs those later calls the same way, + so every call of the turn gets exactly one result - see + [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md#approvals-in-the-middle-of-a-tool-batch). - **Failures are isolated.** A tool that throws gets its own error result; its siblings carry on. A propagating error (`PropagatingToolError`, such as the delegation depth guard, or a throwing hook) stops new calls from @@ -1139,9 +1160,11 @@ Guarantees, whatever the limit: `finishReason: 'aborted'`. Calls that finished keep their results; the rest get a "cancelled" result. Running tools see the abort through their `abortSignal`. -- **Checkpoints.** With `sessionId` + `checkpointStore`, a checkpoint is - written each time the in-order run of finished calls grows (with `1`, after - every call, as before). A resumed run never runs a recorded call again. +- **Checkpoints.** With `sessionId` + `checkpointStore`, the model's turn is + checkpointed before any call starts, then again each time the in-order run + of finished calls grows (with `1`, after every call). A resumed run never + runs a recorded call again, and never asks the model again for a turn it + already checkpointed. ## Declarative specs @@ -2174,6 +2197,193 @@ Targets are `DeploymentAdapter` objects (`scaffold`, `build`, `describe`) registered by name with `registerAdapter()`; both are exported from the package root. +# Durable execution + +Source: docs/durable-execution.md + +Pass a `sessionId` and a `checkpointStore` to `AgentExecutor.execute()` and +the run survives a crash, a restart, an abort or a human approval that takes +days. Nothing here depends on a particular host: any `CheckpointStore` +(`LocalStorageCheckpointStore` on Node, the Cloudflare KV store the +`cloudflare-worker` build wires up, or your own three-method implementation) +and any `ApprovalStore` work the same way. + +```ts +import { AgentExecutor, LocalStorageCheckpointStore } from '@loushy/build-ai-agent'; + +const checkpoints = new LocalStorageCheckpointStore(storage); + +// First message of a conversation. +await AgentExecutor.execute({ agent, input: 'Book a table for 2 tonight', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); + +// Later - same process or another one - the next message of the same conversation. +await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'chat-42', checkpointStore: checkpoints }); +``` + +## What is checkpointed, and when + +A checkpoint is the run's whole transcript plus its step count, usage, +`businessState` and a `status`. It is written under the `sessionId`: + +| Moment | `status` | +| ------ | -------- | +| Right after each model response that asks for tools, before any tool runs | `'in-progress'` | +| Each time a tool result is recorded (results are recorded in call order) | `'in-progress'` | +| When the run is aborted (`signal`) | `'in-progress'` | +| When the run pauses for an approval | `'awaiting-approval'` (with `approvalId`) | +| When the run finishes (the model stopped calling tools, or `maxSteps` ran out) | `'finished'` | + +A run that rejects (a provider error, a throwing hook, a +`PropagatingToolError`, a process crash) keeps its last `'in-progress'` +checkpoint. + +## What happens when you call `execute()` again with the same `sessionId` + +It depends on the stored `status`: + +- **Unfinished run** (`'in-progress'`: it crashed, was aborted, or a store + write failed). The run resumes. If its last model turn has tool calls + without a result, exactly those calls run first, through the normal tool + path (argument validation, hooks, `needsApproval`, `toolConcurrency`), + **without calling the model again**. Then the loop continues, with the + remaining `maxSteps` budget. + - New `input` is appended as a user message after those tool results. + It is never placed between a tool-call turn and its results, so the + transcript stays valid for every provider. This is what lets a user + interrupt a run (abort it) and redirect it with a new message. + - `input` that re-sends the message the run started from (the natural + "retry the same request after a crash") is treated as a retry and is + not appended again. `input: []` also just resumes. +- **Finished run** (`'finished'`). The session continues as a conversation: + the stored messages are kept and `input` becomes the next user turn. Pass + only the new message(s); if you re-send the stored history followed by + new messages, the history is recognised and not duplicated. `steps`, + `usage` and `toolCalls` on the result count from zero for the new run; + `messages` is the whole conversation. +- **Paused awaiting approval** (`'awaiting-approval'`). `execute()` throws + `SessionAwaitingApprovalError` (with `sessionId` and `approvalId`) and + does not call the model or any tool: new input must not bypass a pending + decision. Resolve it with `resumeAfterApproval()` first, passing the same + `checkpointStore`, then send the new message. + +System messages in `input` are ignored when it is added to a stored +session (the session already has its system prompt). To end a session and +start over under the same id, call `checkpointStore.delete(sessionId)`. + +```ts +import { AgentExecutor, SessionAwaitingApprovalError } from '@loushy/build-ai-agent'; + +try { + await AgentExecutor.execute({ agent, input: 'Are you done?', provider, sessionId: 'chat-42', checkpointStore }); +} catch (error) { + if (error instanceof SessionAwaitingApprovalError) { + console.log(`Waiting on approval ${error.approvalId} - resolve it with resumeAfterApproval() first.`); + } +} +``` + +## Approvals in the middle of a tool batch + +When one model turn asks for several tools and one of them needs approval, +the calls before it run and are recorded, and the run pauses on it. The +approval snapshot records the calls after it (`remainingToolCalls`). +`resumeAfterApproval()` then: + +1. records the paused call's result - the tool's result if approved, or a + structured `{ error, note }` rejection result if rejected; +2. runs the remaining calls through the same batch logic: they can run, + fail validation, or pause the run again on another approval (a new + `approvalId`; resolve it the same way, as many times as needed); +3. calls the model once every call of the turn has exactly one result. + +```ts +import { AgentExecutor, resumeAfterApproval } from '@loushy/build-ai-agent'; + +const paused = await AgentExecutor.execute({ + agent, input, provider, toolRegistry, approvalStore, sessionId: 'chat-42', checkpointStore, +}); + +if (paused.finishReason === 'awaiting-approval') { + // ...a human decides, possibly days later, in another process... + const result = await resumeAfterApproval( + { id: paused.approvalId!, approved: true }, + approvalStore, + toolRegistry, + provider, + {}, + checkpointStore, // clears the 'awaiting-approval' mark and keeps checkpointing + ); + console.log(result.finishReason); // 'stop', or 'awaiting-approval' again for the next gated call +} +``` + +The resumed run uses the same `approvalStore` for any later approval unless +you pass a different one in the options. + +## Guarantees + +- **Every tool call gets exactly one result, in call order**, after any + sequence of crashes, aborts, pauses and resumes. (An aborted call gets a + "cancelled" result; a rejected call gets a rejection result.) +- **A recorded result is never re-executed.** Results are checkpointed as + the in-order prefix of finished calls grows; a resumed run only runs calls + that have no recorded result. +- **A checkpointed model turn is never generated again.** If the process + dies after the model answered, the resumed run executes that answer's tool + calls instead of asking the model again (no extra cost, no different + plan). +- **The provider always receives a valid transcript**: tool results + directly follow their tool-call turn, and new user input comes after them. + +## At-least-once tools: make side effects idempotent + +A tool that was **running** when the process died has no recorded result, +so it runs again on resume. Tool execution is therefore at-least-once, not +exactly-once. For tools with side effects (charging a card, sending an +email), either gate them with `needsApproval` or make them idempotent. +Each call's `toolCallId` (the model's id for the call, unchanged when it is +re-run on resume) is passed to `execute`, so it works as an idempotency +key: + +```ts +import { defineTool } from '@loushy/build-ai-agent'; +import { z } from 'zod'; + +declare const payments: { charge(amount: number, opts: { idempotencyKey: string }): Promise<{ id: string }> }; + +const chargeCard = defineTool({ + name: 'charge_card', + description: 'Charge the customer', + input: z.object({ amount: z.number() }), + execute: async ({ amount }, { toolCallId }) => + payments.charge(amount, { idempotencyKey: toolCallId }), // a re-run cannot charge twice +}); +``` + +(`toolCallId` is set for tools run by `AgentExecutor` and by +`resumeAfterApproval()`, but not yet for tools that route through +`sandboxExecute`.) + +## Compatibility with stored data + +- Checkpoints written before `status` existed are treated as + `'in-progress'` (before this change, finished runs deleted their + checkpoint, so only unfinished runs were ever stored). +- Approval snapshots written before `remainingToolCalls` existed resume as + they used to: the calls after the paused one do not run. Each of them gets + an error result saying it was not run, so the transcript stays valid. + +## Limits + +- A `sessionId` identifies one logical conversation; running two + `execute()` calls on the same id at the same time is not supported. +- If you call `resumeAfterApproval()` without the `checkpointStore`, the + session keeps its `'awaiting-approval'` mark and later `execute()` calls + keep throwing `SessionAwaitingApprovalError`; pass the store (or delete + the checkpoint). +- Workers KV is eventually consistent across locations - see + [Deployment](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/deployment.md) for the caveat. + # Tracing and observability Source: docs/observability.md diff --git a/llms.txt b/llms.txt index b34ea3137..6d87dbc86 100644 --- a/llms.txt +++ b/llms.txt @@ -9,6 +9,7 @@ - [Quick Start](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/quick-start.md): Every TypeScript snippet on this page is a complete, standalone ES module (`.mts`). - [API Overview](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/api-overview.md): The package root (`@loushy/build-ai-agent`) exports everything below. - [Configuration](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/configuration.md): A declarative agent is a YAML (`.yaml`/`.yml`) or JSON (`.json`) file, validated with zod by `loadSpec()` (`src/spec/schema.ts`). +- [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md): Pass a `sessionId` and a `checkpointStore` to `AgentExecutor.execute()` and the run survives a crash, a restart, an abort or a human approval that takes days. - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md): `AgentExecutor.execute()` and `FlowExecutor.execute()` emit spans through a `TraceExporter`. - [Skills](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/skills.md): A skill is a named chunk of instructions (a markdown document) that the model loads only when it needs it. - [Testing agents](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/testing.md): `@loushy/build-ai-agent/testing` ships `mockModel`: a scripted, deterministic `LLMProvider` for unit tests. diff --git a/src/deploy/adapters/cloudflare.checkpoint.test.ts b/src/deploy/adapters/cloudflare.checkpoint.test.ts index 44747f7ee..0d3e296c3 100644 --- a/src/deploy/adapters/cloudflare.checkpoint.test.ts +++ b/src/deploy/adapters/cloudflare.checkpoint.test.ts @@ -123,7 +123,7 @@ describe('cloudflare-worker: durable checkpointing (LOU-T2, built bundle)', () = expect(res.status).toBe(400); }); - it('a real run through the built bundle writes a checkpoint to KV mid-run and deletes it on completion', async () => { + it('a real run through the built bundle writes a checkpoint to KV and keeps it, marked finished, on completion (LOU-U8)', async () => { const kv = createMockKV(); const sessionId = 'writes-and-completes'; @@ -140,11 +140,11 @@ describe('cloudflare-worker: durable checkpointing (LOU-T2, built bundle)', () = expect(body.finishReason).toBe('stop'); // The run completed inside this single request/response cycle, so - // AgentExecutor's terminal-state cleanup (see AgentExecutor.ts) deletes - // the checkpoint before returning - proven here via the real built - // bundle + a real (mocked) KV binding, not a re-implementation of that - // logic. - expect(await kv.get(`checkpoints/${sessionId}`)).toBeNull(); + // AgentExecutor keeps its transcript marked 'finished' (LOU-U8) - the + // next request with this sessionId continues the conversation. Proven + // here via the real built bundle + a real (mocked) KV binding. + const stored = JSON.parse((await kv.get(`checkpoints/${sessionId}`))!) as Checkpoint; + expect(stored.status).toBe('finished'); }); it('pause -> resume: a checkpoint genuinely produced by the built bundle mid-run is rehydrated by a LATER, independent request', async () => { @@ -173,7 +173,9 @@ describe('cloudflare-worker: durable checkpointing (LOU-T2, built bundle)', () = // this is the real payload the built bundle's AgentExecutor produced, // not a hand-crafted fixture. expect(capturedPuts.length).toBeGreaterThan(0); - const midRunCheckpoint = capturedPuts[0]; + // LOU-U9: the first put is the model's tool-call turn (before the tool + // ran); take the one written after the tool result. + const midRunCheckpoint = capturedPuts.find((c) => c.messages.some((m) => m.role === 'tool'))!; expect(midRunCheckpoint.sessionId).toBe(captureSessionId); expect(midRunCheckpoint.messages.length).toBeGreaterThan(0); // A distinctive marker proving these ARE this run's real messages, not @@ -209,19 +211,20 @@ describe('cloudflare-worker: durable checkpointing (LOU-T2, built bundle)', () = expect(resumeBody.finishReason).toBe('stop'); // The resumed run's message history starts with EXACTLY the seeded - // checkpoint's pre-pause messages (rehydration ignores this request's - // own `message` entirely - see AgentExecutor.ts's rehydration branch), - // plus new messages appended to finish the run. + // checkpoint's pre-pause messages, then this request's own `message` as + // a new user turn (LOU-U8: new input on an unfinished run is appended + // after it), plus the messages that finish the run. expect(resumeBody.messages.length).toBeGreaterThan(midRunCheckpoint.messages.length); expect(resumeBody.messages.slice(0, midRunCheckpoint.messages.length)).toEqual(midRunCheckpoint.messages); expect( resumeBody.messages.some( (m: any) => typeof m.content === 'string' && m.content.includes('totally unrelated follow-up text') ) - ).toBe(false); + ).toBe(true); // Completed to a terminal state again, so the resumed session's - // checkpoint was cleaned up too. - expect(await resumeKv.get(`checkpoints/${resumeSessionId}`)).toBeNull(); + // checkpoint is marked finished too. + const resumed = JSON.parse((await resumeKv.get(`checkpoints/${resumeSessionId}`))!) as Checkpoint; + expect(resumed.status).toBe('finished'); }); }); diff --git a/src/deploy/kvCheckpointStore.test.ts b/src/deploy/kvCheckpointStore.test.ts index e26532c40..7e12a1037 100644 --- a/src/deploy/kvCheckpointStore.test.ts +++ b/src/deploy/kvCheckpointStore.test.ts @@ -74,6 +74,17 @@ describe('KVCheckpointStore', () => { expect(await store.load('session-1')).toEqual(checkpoint); }); + it('round-trips the LOU-U8 status and approvalId fields', async () => { + const store = new KVCheckpointStore(createMockKV()); + const paused = makeCheckpoint({ status: 'awaiting-approval', approvalId: 'approval-1' }); + + await store.save('session-1', paused); + expect(await store.load('session-1')).toEqual(paused); + + await store.save('session-1', makeCheckpoint({ status: 'finished' })); + expect((await store.load('session-1'))?.status).toBe('finished'); + }); + it('delete() removes the entry so a later load() misses', async () => { const kv = createMockKV(); const store = new KVCheckpointStore(kv); @@ -239,13 +250,10 @@ describe('KVCheckpointStore', () => { expect(chargeExecute).toHaveBeenCalledTimes(1); expect(lookupExecute).toHaveBeenCalledTimes(1); - // The run reached a terminal state, so AgentExecutor's own - // terminal-state cleanup deletes the checkpoint before returning - the - // real proof that a checkpoint genuinely existed in KV mid-run (not - // just left over) is the raw KV entry having been written and read at - // least once via the store's get/put, asserted below by inspecting the - // underlying mock KV's call history. + // The run reached a terminal state, so its checkpoint is kept, marked + // 'finished' (LOU-U8). The raw KV entry having been written via the + // store's put is asserted below from the mock KV's call history. expect(kv.putCalls.some(([key]) => key === `${DEFAULT_KV_KEY_PREFIX}${sessionId}`)).toBe(true); - expect(await checkpointStore.load(sessionId)).toBeNull(); + expect((await checkpointStore.load(sessionId))?.status).toBe('finished'); }); }); diff --git a/src/execution/AgentExecutor.test.ts b/src/execution/AgentExecutor.test.ts index a28298df4..0ed406f9a 100644 --- a/src/execution/AgentExecutor.test.ts +++ b/src/execution/AgentExecutor.test.ts @@ -273,7 +273,7 @@ describe('AgentExecutor', () => { toolRegistry ); - expect(execute).toHaveBeenCalledWith({ amount: 5 }, {}); + expect(execute).toHaveBeenCalledWith({ amount: 5 }, { toolCallId: 'call-2' }); expect(result.requiresApproval).toBeUndefined(); expect(result.result).toEqual({ ok: true }); }); @@ -421,10 +421,15 @@ describe('AgentExecutor', () => { checkpointStore, }); - expect(save).toHaveBeenCalledTimes(3); + // LOU-U9/U8: one save after each model turn (before its tool runs), + // one after each tool result, and a final 'finished' one. + expect(save).toHaveBeenCalledTimes(7); const lengths = save.mock.calls.map(([, checkpoint]) => checkpoint.messages.length); - expect(lengths[1]).toBeGreaterThan(lengths[0]); - expect(lengths[2]).toBeGreaterThan(lengths[1]); + for (let i = 1; i < lengths.length; i++) expect(lengths[i]).toBeGreaterThan(lengths[i - 1]); + expect(save.mock.calls.map(([, checkpoint]) => checkpoint.status)).toEqual([ + ...Array(6).fill('in-progress'), + 'finished', + ]); }); it('should write the businessState option into every checkpoint record (LOU-T1)', async () => { @@ -486,7 +491,7 @@ describe('AgentExecutor', () => { businessState: { orderId: 'ord_42', stage: 'processing' }, }); - expect(save).toHaveBeenCalledTimes(2); + expect(save).toHaveBeenCalledTimes(5); // 2 model turns + 2 tool results + finished for (const [, checkpoint] of save.mock.calls) { expect(checkpoint.businessState).toEqual({ orderId: 'ord_42', stage: 'processing' }); } @@ -539,8 +544,8 @@ describe('AgentExecutor', () => { }) ).resolves.toBeDefined(); - expect(save).toHaveBeenCalledTimes(1); - expect(save.mock.calls[0][1].businessState).toBeUndefined(); + expect(save).toHaveBeenCalledTimes(3); // model turn + tool result + finished + for (const [, checkpoint] of save.mock.calls) expect(checkpoint.businessState).toBeUndefined(); }); it('should resume from a checkpoint after a simulated crash + restart with zero message loss', async () => { @@ -832,7 +837,7 @@ describe('AgentExecutor', () => { expect(userMessage?.content).toBe('Hello'); }); - it('should delete the checkpoint once a run reaches a terminal finish reason', async () => { + it('should keep the checkpoint, marked finished, once a run reaches a terminal finish reason (LOU-U8)', async () => { const execute = vi.fn().mockResolvedValue({ ok: true }); toolRegistry.register('noop', { displayName: 'Noop', @@ -902,11 +907,13 @@ describe('AgentExecutor', () => { checkpointStore, }); - expect(checkpointStore.delete).toHaveBeenCalledWith('terminal-session'); - await expect(checkpointStore.load('terminal-session')).resolves.toBeNull(); + expect(checkpointStore.delete).not.toHaveBeenCalled(); + const finished = await checkpointStore.load('terminal-session'); + expect(finished.status).toBe('finished'); + expect(finished.messages.at(-1)).toEqual({ role: 'assistant', content: 'done' }); }); - it('should use fresh input (not stale stored messages) when execute() is called again with the same sessionId after completion', async () => { + it('should continue the conversation when execute() is called again with the same sessionId after completion (LOU-U8)', async () => { function makeCheckpointStore() { const records = new Map(); return { @@ -927,7 +934,7 @@ describe('AgentExecutor', () => { .build(); // First run: completes normally (no tool calls), reaching a terminal - // finish reason and clearing the checkpoint. + // finish reason; its checkpoint is kept, marked 'finished'. const firstProvider = createMockProvider({ name: 'mock', responses: ['first response'], @@ -941,9 +948,8 @@ describe('AgentExecutor', () => { checkpointStore, }); - // Second run: same sessionId, brand-new input. If the stale checkpoint - // were still around and rehydrated, this fresh input would be - // silently ignored. + // Second run: same sessionId, brand-new input - appended as the next + // user turn after the first exchange (never ignored). const secondProvider = createMockProvider({ name: 'mock', responses: ['second response'], @@ -957,9 +963,14 @@ describe('AgentExecutor', () => { checkpointStore, }); - const userMessage = secondResult.messages.find((m) => m.role === 'user'); - expect(userMessage?.content).toBe('second input, completely different'); + expect(secondResult.messages.map((m) => [m.role, m.content])).toEqual([ + ['user', 'first input'], + ['assistant', 'first response'], + ['user', 'second input, completely different'], + ['assistant', 'second response'], + ]); expect(secondResult.text).toBe('second response'); + expect(secondResult.steps).toBe(1); }); it('should pass temperature and maxTokens', async () => { @@ -1580,7 +1591,7 @@ describe('AgentExecutor', () => { await (AgentExecutor as any).executeToolCall(toolCall, agent, toolRegistry, undefined, undefined, undefined, hooks, undefined, []); - expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, {}); + expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, { toolCallId: 'call-1' }); }); it('a thrown hook error aborts the run and rejects execute(), without being swallowed', async () => { diff --git a/src/execution/AgentExecutor.ts b/src/execution/AgentExecutor.ts index 8885d7f5e..459c0234a 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -140,7 +140,37 @@ export interface ExecuteOptions { maxTokens?: number; onEvent?: (event: ExecutionEvent) => void; approvalStore?: ApprovalStore; + /** + * Durable execution: with `checkpointStore`, the run is checkpointed under + * this id after every model response, as tool results are recorded, and + * when it pauses, aborts or finishes. Calling `execute()` again with the + * same id (from any process) picks the session up (LOU-U8): + * + * - Unfinished run (crashed or aborted): resumes it. Tool calls of its last + * model turn that have no result run first, without calling the model. + * New `input` is appended as a user message after those results; input + * that re-sends the run's own starting message is treated as a retry + * and not appended again (`input: []` also just resumes). + * - Finished run: continues the conversation - the stored messages are + * kept and `input` is appended as the next user turn. Pass only the new + * message(s); a re-sent copy of the stored history is not duplicated. + * `steps`, `usage` and `toolCalls` count from zero for the new run. + * - Paused awaiting approval: throws `SessionAwaitingApprovalError`; + * resolve it with `resumeAfterApproval()` first. + * + * A tool that was running when the process died runs again on resume + * (at-least-once): make side-effecting tools idempotent - e.g. keyed on + * the `toolCallId` their `execute` receives - or approval-gated. + * Delete the session with `checkpointStore.delete(sessionId)`. + * + * @example + * ```ts + * await AgentExecutor.execute({ agent, input: 'Book a table for 2', provider, sessionId: 'chat-42', checkpointStore }); + * await AgentExecutor.execute({ agent, input: 'Make it 3 people', provider, sessionId: 'chat-42', checkpointStore }); + * ``` + */ sessionId?: string; + /** Where `sessionId` checkpoints are stored - see `sessionId` for the semantics. */ checkpointStore?: CheckpointStore; /** * When true, `input` is treated as a complete, ready-to-send message @@ -342,15 +372,17 @@ export interface ExecuteOptions { * - Approval: the first call that needs approval stops the batch. Calls * before it run (concurrently) and are recorded; the run then pauses * on that call (`finishReason: 'awaiting-approval'`); calls after it - * never start in this run. + * never start in this run. `resumeAfterApproval()` records the paused + * call's result and then runs the calls after it the same way (LOU-U7). * - A failing tool does not affect its siblings - each gets its own error * result. A propagating error (`PropagatingToolError`, a throwing hook) * stops new calls from starting, waits for the running ones, then * rejects `execute()` - no tool is left running detached. * - An abort (`signal`) mid-batch resolves with `finishReason: 'aborted'`: * finished calls keep their results, the rest get a "cancelled" result. - * - With checkpointing, results are checkpointed as the in-order prefix - * of finished calls grows, so a resumed run never re-runs a recorded call. + * - With checkpointing, the model's turn is checkpointed before any call + * starts, and results as the in-order prefix of finished calls grows, + * so a resumed run never re-runs a recorded call. * * @example * ```ts @@ -445,6 +477,18 @@ export class AgentExecutor { const state = await loadRunState(options); + // LOU-U7/U9: a resumed transcript may end with a model turn whose tool + // calls (some of them) have no result yet - finish those first, without + // calling the model again. + if (state.pendingToolCalls.length > 0) { + const resumed = await this.runStepOrAbort(options, state, () => + this.runPendingToolCalls(options, state, agentSpanId) + ); + if (resumed !== 'continue') { + return resumed; + } + } + // Execution loop with tool calling. LOU-V1: the signal is checked // before every model call (here) and every tool call (runToolCalls()). while (state.steps < maxSteps) { @@ -453,7 +497,9 @@ export class AgentExecutor { } state.steps++; - const outcome = await this.runStepOrAbort(options, state, tools, agentSpanId); + const outcome = await this.runStepOrAbort(options, state, () => + this.runStep(options, state, tools, agentSpanId) + ); if (outcome === 'stop') { return this.finishRun(options, state); } @@ -466,19 +512,19 @@ export class AgentExecutor { } /** - * runStep(), with a thrown error emitted as an `error` event and - * rethrown - unless the run's signal was aborted, in which case the - * rejection is the abort itself (e.g. the provider's AbortError) and the - * run ends as 'aborted' instead (LOU-V1). + * Runs one step (runStep(), or the resumed pending tool calls), with a + * thrown error emitted as an `error` event and rethrown - unless the + * run's signal was aborted, in which case the rejection is the abort + * itself (e.g. the provider's AbortError) and the run ends as 'aborted' + * instead (LOU-V1). */ - private static async runStepOrAbort( + private static async runStepOrAbort( options: ExecuteOptions, state: AgentRunState, - tools: ToolDefinition[], - agentSpanId: string - ): Promise<'continue' | 'stop' | ExecutionResult> { + step: () => Promise + ): Promise { try { - return await this.runStep(options, state, tools, agentSpanId); + return await step(); } catch (error) { if (options.signal?.aborted) { return this.abortRun(options, state); @@ -624,7 +670,6 @@ export class AgentExecutor { toolCalls: ToolCall[], agentSpanId: string ): Promise { - const { onEvent } = options; state.toolCalls.push(...toolCalls); // Add assistant message with tool calls @@ -634,6 +679,47 @@ export class AgentExecutor { toolCalls, }); + // LOU-U9: checkpoint the model's turn before any tool runs, so a crash + // from here on resumes by running the calls - never by asking the + // model again. + await saveStepCheckpoint(options, state); + + return this.runBatch(options, state, toolCalls, agentSpanId); + } + + /** + * LOU-U7/U9: runs the tool calls of the last model turn that have no + * result yet (left by a crash, or by an approval pause mid-batch), then + * appends any input queued behind them. Resolves to 'continue', or the + * ExecutionResult when the run pauses again or is aborted. + */ + private static async runPendingToolCalls( + options: ExecuteOptions, + state: AgentRunState, + agentSpanId: string + ): Promise<'continue' | ExecutionResult> { + const calls = state.pendingToolCalls; + state.pendingToolCalls = []; + const stopped = await this.runBatch(options, state, calls, agentSpanId); + if (stopped) { + return stopped; + } + state.messages.push(...state.queuedInput); + state.queuedInput = []; + return 'continue'; + } + + /** + * Runs one batch of tool calls through the LOU-V3 scheduler, recording + * results in call order and checkpointing as they are recorded. + */ + private static async runBatch( + options: ExecuteOptions, + state: AgentRunState, + toolCalls: ToolCall[], + agentSpanId: string + ): Promise { + const { onEvent } = options; const batch = await runToolBatch( toolCalls, options.toolConcurrency ?? 'unbounded', @@ -671,7 +757,10 @@ export class AgentExecutor { throw batch.failure.error; } if (batch.approval) { - return this.pauseForApproval(options, state, batch.approval.toolCall, batch.approval.outcome); + const { toolCall, outcome } = batch.approval; + const pausedAt = batch.unrecorded.findIndex((call) => call.toolCall === toolCall); + const remaining = batch.unrecorded.slice(pausedAt + 1).map((call) => call.toolCall); + return this.pauseForApproval(options, state, toolCall, outcome, remaining); } return undefined; } @@ -765,15 +854,17 @@ export class AgentExecutor { } /** - * Persists a pending approval (plus the snapshot resume.ts needs) and - * ends this execute() call with an 'awaiting-approval' result. The - * checkpoint is deliberately left in place so the run can be resumed. + * Persists a pending approval (plus the snapshot resume.ts needs, with + * the turn's not-yet-run calls - LOU-U7) and ends this execute() call + * with an 'awaiting-approval' result. The checkpoint is marked + * 'awaiting-approval' (LOU-U8) so new input cannot bypass the decision. */ private static async pauseForApproval( options: ExecuteOptions, state: AgentRunState, toolCall: ToolCall, - toolResult: ToolCallOutcome + toolResult: ToolCallOutcome, + remainingToolCalls: ToolCall[] ): Promise { const { agent, approvalStore, sessionId, onEvent } = options; if (!approvalStore) { @@ -792,13 +883,18 @@ export class AgentExecutor { }; const snapshot: ExecutionSnapshot = { agent, - currentMessages: state.messages, + // Queued input rides at the end; resume moves it behind the results. + currentMessages: [...state.messages, ...state.queuedInput], pendingToolCall: pending, steps: state.steps, sessionId, + remainingToolCalls, }; + // Approval first: a crash between the two writes then leaves a + // resumable 'in-progress' checkpoint, never one naming a lost approval. await approvalStore.save(pending, snapshot); + await saveStepCheckpoint(options, state, 'awaiting-approval', pending.id); this.emitEvent(onEvent, { type: 'finish', @@ -850,7 +946,7 @@ export class AgentExecutor { options: ExecuteOptions, state: AgentRunState ): Promise { - const { onEvent, sessionId, checkpointStore } = options; + const { onEvent } = options; // LOU-T4: `maxSteps` was exhausted, but the very last thing that // happened was a surfaced-to-the-model provider failure (not a genuine @@ -877,15 +973,11 @@ export class AgentExecutor { }); // The run has reached a terminal state (either the model stopped - // requesting tools, or maxSteps was exhausted) - as opposed to the - // 'awaiting-approval' early return in pauseForApproval(), which is a - // mid-flight pause where the checkpoint must stay in place so it can - // still be resumed. Clear the checkpoint here so a later execute() call - // reusing this sessionId builds fresh messages from its own `input` - // instead of silently resuming from this now-finished run. - if (sessionId && checkpointStore) { - await checkpointStore.delete(sessionId); - } + // requesting tools, or maxSteps was exhausted). LOU-U8: keep the + // transcript, marked 'finished', so a later execute() call reusing this + // sessionId continues the conversation with its new input instead of + // resuming this run (or forgetting it). + await saveStepCheckpoint(options, state, 'finished'); return toExecutionResult(state, state.finalText, state.finishReason); } diff --git a/src/execution/ApprovalGate.test.ts b/src/execution/ApprovalGate.test.ts index dbce3fd6b..f6065222f 100644 --- a/src/execution/ApprovalGate.test.ts +++ b/src/execution/ApprovalGate.test.ts @@ -141,6 +141,25 @@ describe('Execution - StorageServiceApprovalStore', () => { expect(second).toBeNull(); }); + it('should round-trip the LOU-U7 remainingToolCalls of a snapshot', async () => { + const store = createApprovalStore(); + const pending: PendingApproval = { + id: 'approval-mid-batch', + toolCallId: 'call-b', + toolName: 'chargeCard', + args: {}, + createdAt: new Date().toISOString(), + }; + const snapshot: ExecutionSnapshot = { + ...buildSnapshot(pending), + remainingToolCalls: [{ id: 'call-c', type: 'function', function: { name: 'sendEmail', arguments: '{"to":"a@b.c"}' } }], + }; + + await store.save(pending, snapshot); + + expect((await store.resolve('approval-mid-batch'))?.snapshot).toEqual(snapshot); + }); + it('should return null for an id that was never saved', async () => { const store = createApprovalStore(); const result = await store.resolve('never-saved'); diff --git a/src/execution/ApprovalGate.ts b/src/execution/ApprovalGate.ts index 9ba0ebd3d..1ee042f87 100644 --- a/src/execution/ApprovalGate.ts +++ b/src/execution/ApprovalGate.ts @@ -3,7 +3,7 @@ * Types and storage for human-in-the-loop tool approval. */ -import { Message } from '../providers'; +import { Message, ToolCall } from '../providers'; import { AgentConfig } from '../types'; import { StorageService } from '../storage'; @@ -48,6 +48,17 @@ export interface ExecutionSnapshot { * execution at all. */ sessionId?: string; + /** + * LOU-U7: the tool calls of the same model turn that come after the + * paused one and have not run yet, in call order. `resumeAfterApproval()` + * records the paused call's result (or rejection) and then runs these + * through the normal batch path - they can run, fail validation, or pause + * the run again on another approval. Absent on snapshots saved before + * this field existed, which means "no remaining calls": any call of that + * turn still without a result gets an error result saying it was not run, + * so the transcript stays valid for the provider. + */ + remainingToolCalls?: ToolCall[]; } /** diff --git a/src/execution/agentRunState.ts b/src/execution/agentRunState.ts index 18cad33f5..ac48cc889 100644 --- a/src/execution/agentRunState.ts +++ b/src/execution/agentRunState.ts @@ -7,8 +7,9 @@ import { Message, ToolCall } from '../providers'; import { AgentConfig } from '../types'; -import { Checkpoint } from './checkpoint'; -import { CompactedLLMProviderError } from './errors'; +import { Checkpoint, CheckpointStatus } from './checkpoint'; +import { CompactedLLMProviderError, SessionAwaitingApprovalError } from './errors'; +import { inputMessages, newSessionMessages, splitPendingTurn } from './transcript'; import type { ExecuteOptions, ExecutionResult } from './AgentExecutor'; import type { ToolCallOutcome } from './toolCallExecution'; import type { UnrecordedToolCall } from './toolBatch'; @@ -35,6 +36,20 @@ export interface AgentRunState { * read as if the model actually gave up on its own. */ lastSurfacedProviderError?: CompactedLLMProviderError; + /** + * LOU-U7/U9: tool calls of the last model turn that have no result yet + * (a run resumed after a crash, or after an approval mid-batch). They run + * through the normal batch path before the model is called again. + */ + pendingToolCalls: ToolCall[]; + /** + * LOU-U8: new input held back until `pendingToolCalls` all have results + * (a user message must never sit between a tool-call turn and its + * results). Persisted at the end of checkpoints and snapshots so it + * survives a crash or another approval pause; splitPendingTurn() moves it + * back behind the results on load. + */ + queuedInput: Message[]; } /** @@ -45,37 +60,27 @@ function buildMessages( input: string | Message[], skipSystemPromptInjection = false ): Message[] { - const messages: Message[] = []; - // Add system prompt, unless the caller has indicated `input` already // includes one (e.g. resume.ts rebuilding from an ExecutionSnapshot). - if (agent.prompt && !skipSystemPromptInjection) { - messages.push({ - role: 'system', - content: agent.prompt, - }); - } - - // Add input messages - if (typeof input === 'string') { - messages.push({ - role: 'user', - content: input, - }); - } else { - messages.push(...input); - } - - return messages; + const system: Message[] = + agent.prompt && !skipSystemPromptInjection ? [{ role: 'system', content: agent.prompt }] : []; + return [...system, ...inputMessages(input)]; } +const ZERO_USAGE: TokenUsage = { promptTokens: 0, completionTokens: 0, totalTokens: 0 }; + type InitialRunState = Pick< AgentRunState, 'messages' | 'toolCalls' | 'usage' | 'steps' | 'businessState' >; -/** Rehydrates run state from a checkpoint left by an earlier process. */ -function stateFromCheckpoint(checkpoint: Checkpoint, businessState: unknown): InitialRunState { +/** + * Rehydrates run state from a checkpoint left by an earlier call (LOU-U8): + * an unfinished run resumes where it stopped (new input, if any, is + * appended after it - see newSessionMessages()); a finished run continues + * as a conversation, with a fresh step budget, usage and tool-call list. + */ +function stateFromCheckpoint(checkpoint: Checkpoint, options: ExecuteOptions): InitialRunState { // LOU-T1: on a rehydrated run (e.g. a fresh process resuming after a // crash), the caller of this execute() call may have no way to know // what businessState a *previous* process attached - that's exactly @@ -85,11 +90,14 @@ function stateFromCheckpoint(checkpoint: Checkpoint, businessState: unknown): In // checkpoint rather than silently dropping it. An explicit // `businessState` passed to *this* call always wins (e.g. a caller // deliberately updating it as part of the resumed run). + const { businessState } = options; + const finished = checkpoint.status === 'finished'; + const added = newSessionMessages(checkpoint.messages, options.input, finished); return { - messages: [...checkpoint.messages], - toolCalls: [...(checkpoint.toolCalls as ToolCall[])], - usage: { ...checkpoint.usage }, - steps: checkpoint.stepIndex, + messages: [...checkpoint.messages, ...added], + toolCalls: finished ? [] : [...(checkpoint.toolCalls as ToolCall[])], + usage: finished ? { ...ZERO_USAGE } : { ...checkpoint.usage }, + steps: finished ? 0 : checkpoint.stepIndex, businessState: businessState === undefined ? checkpoint.businessState : businessState, }; } @@ -100,11 +108,7 @@ function freshState(options: ExecuteOptions): InitialRunState { return { messages: buildMessages(agent, input, skipSystemPromptInjection), toolCalls: [], - usage: { - promptTokens: 0, - completionTokens: 0, - totalTokens: 0, - }, + usage: { ...ZERO_USAGE }, steps: initialSteps ?? 0, businessState, }; @@ -112,7 +116,10 @@ function freshState(options: ExecuteOptions): InitialRunState { /** * If a checkpoint exists for this sessionId, rehydrate state from it - * instead of building messages from scratch. + * instead of building messages from scratch. Either way, unanswered tool + * calls of the last model turn become `pendingToolCalls` (LOU-U7/U9). + * Throws SessionAwaitingApprovalError when the session is paused on an + * approval (LOU-U8). */ export async function loadRunState(options: ExecuteOptions): Promise { const { sessionId, checkpointStore } = options; @@ -120,12 +127,21 @@ export async function loadRunState(options: ExecuteOptions): Promise { const { agent, sessionId, checkpointStore } = options; if (!sessionId || !checkpointStore) { @@ -149,11 +171,13 @@ export async function saveStepCheckpoint( agentId: agent.id || '', sessionId, stepIndex: state.steps, - messages: [...state.messages], + messages: [...state.messages, ...state.queuedInput], toolCalls: [...state.toolCalls], usage: state.usage, finishReason: state.finishReason, businessState: state.businessState, + status, + ...(approvalId !== undefined && { approvalId }), }; await checkpointStore.save(sessionId, checkpoint); } diff --git a/src/execution/cancellation.test.ts b/src/execution/cancellation.test.ts index 63aa69b5a..bb0f0b93d 100644 --- a/src/execution/cancellation.test.ts +++ b/src/execution/cancellation.test.ts @@ -281,7 +281,7 @@ describe('AgentExecutor cancellation (LOU-V1)', () => { }); const resumed = await AgentExecutor.execute({ agent: agentWithTools('fast'), - input: 'ignored - the checkpoint wins', + input: 'do the thing', // a retry of the same request: resumes, nothing appended (LOU-U8) provider: second.provider, toolRegistry: registryWith({ fast }), sessionId: 's1', @@ -292,7 +292,7 @@ describe('AgentExecutor cancellation (LOU-V1)', () => { expect(resumed.text).toBe('all done'); expect(resumed.steps).toBe(2); expect(sentRoles).toEqual(['system', 'user', 'assistant', 'tool']); - expect(checkpointStore.checkpoints.has('s1')).toBe(false); + expect(checkpointStore.checkpoints.get('s1')?.status).toBe('finished'); }); it('treats an abort-caused provider rejection as the abort: no compaction, no surfaced retry, no error event', async () => { diff --git a/src/execution/checkpoint.test.ts b/src/execution/checkpoint.test.ts index 2ca865ab4..75eaaeb76 100644 --- a/src/execution/checkpoint.test.ts +++ b/src/execution/checkpoint.test.ts @@ -130,6 +130,24 @@ describe('Execution - LocalStorageCheckpointStore', () => { }); }); + // LOU-U8 + it('should round-trip status, approvalId and a transcript with queued input behind a pending turn', async () => { + const store = createCheckpointStore(); + const checkpoint = buildCheckpoint({ + status: 'awaiting-approval', + approvalId: 'approval-9', + messages: [ + { role: 'user', content: 'Hi' }, + { role: 'assistant', content: '', toolCalls: [{ id: 'call-1', type: 'function', function: { name: 'noop', arguments: '{}' } }] }, + { role: 'user', content: 'queued' }, + ], + }); + + await store.save('session-1', checkpoint); + + expect(await store.load('session-1')).toEqual(checkpoint); + }); + it('should leave businessState absent (undefined) when never set - backward compatible', async () => { const store = createCheckpointStore(); const checkpoint = buildCheckpoint(); diff --git a/src/execution/checkpoint.ts b/src/execution/checkpoint.ts index 1ceda3490..185807179 100644 --- a/src/execution/checkpoint.ts +++ b/src/execution/checkpoint.ts @@ -7,8 +7,23 @@ import { Message } from '../providers'; import { StorageService } from '../storage'; /** - * A snapshot of an in-progress agent run, saved after each tool result so - * execution can resume from here (e.g. after a process restart). + * Where the run recorded in a {@link Checkpoint} stands (LOU-U8): + * + * - `'in-progress'`: the run has not finished (it crashed, was aborted, or + * is still running). `execute()` with the same `sessionId` resumes it. + * - `'awaiting-approval'`: paused on a tool call that needs a human + * decision. `execute()` with the same `sessionId` throws + * `SessionAwaitingApprovalError`; call `resumeAfterApproval()` instead. + * - `'finished'`: the run ended normally. `execute()` with the same + * `sessionId` continues the conversation: the new input becomes the next + * user turn after the stored messages. + */ +export type CheckpointStatus = 'in-progress' | 'awaiting-approval' | 'finished'; + +/** + * A snapshot of an agent run, saved after each model response, after each + * tool result and when the run pauses or finishes, so execution can resume + * from here (e.g. after a process restart). */ export interface Checkpoint { agentId: string; @@ -57,6 +72,14 @@ export interface Checkpoint { * "inherit". */ businessState?: unknown; + /** + * LOU-U8: where the run stands - see {@link CheckpointStatus}. Absent on + * checkpoints written before this field existed, which are treated as + * `'in-progress'` (the only kind that was ever kept). + */ + status?: CheckpointStatus; + /** LOU-U8: with `status: 'awaiting-approval'`, the id of the pending approval. */ + approvalId?: string; } /** diff --git a/src/execution/durableExecution.test.ts b/src/execution/durableExecution.test.ts new file mode 100644 index 000000000..9172b1d4b --- /dev/null +++ b/src/execution/durableExecution.test.ts @@ -0,0 +1,457 @@ +/** + * Durable execution gaps: + * - LOU-U7: an approval in the middle of a tool batch keeps the transcript valid. + * - LOU-U8: resuming a session with new input (unfinished / finished / awaiting approval). + * - LOU-U9: a checkpoint after every model turn, so a crash never re-calls the model. + * + * Every checkpoint goes through the real KVCheckpointStore (JSON round-trip) + * and every approval snapshot through a JSON round-trip, as a real store would. + */ + +import { describe, it, expect } from 'vitest'; +import { z } from 'zod'; +import { AgentExecutor, ExecuteOptions, ExecutionResult, PropagatingToolError } from './AgentExecutor'; +import type { ApprovalStore, ExecutionSnapshot, PendingApproval } from './ApprovalGate'; +import type { Checkpoint, CheckpointStore } from './checkpoint'; +import { SessionAwaitingApprovalError } from './errors'; +import { resumeAfterApproval } from './resume'; +import { newSessionMessages } from './transcript'; +import { KVCheckpointStore } from '../deploy/kvCheckpointStore'; +import { defineTool, DefinedTool } from '../tools/defineTool'; +import { ToolRegistry } from '../tools'; +import { mockModel, MockTurn } from '../testing'; +import { AgentConfig, AgentType } from '../types'; +import type { Message } from '../providers'; + +type Runs = Record; + +interface ToolOptions { + needsApproval?: boolean; + /** Throw a non-recoverable error (a simulated process death) on the first run. */ + crashOnce?: boolean; + input?: z.ZodObject; + seenIds?: string[]; +} + +function tool(name: string, runs: Runs, options: ToolOptions = {}): DefinedTool { + runs[name] = 0; + return defineTool({ + name, + description: name, + input: options.input ?? z.object({}), + needsApproval: options.needsApproval, + execute: async (_args, ctx) => { + runs[name]++; + options.seenIds?.push(ctx.toolCallId); + if (options.crashOnce && runs[name] === 1) { + throw new PropagatingToolError(`process died while running ${name}`); + } + return `${name} done`; + }, + }); +} + +function agentFor(tools: DefinedTool[]): AgentConfig { + const config: AgentConfig['tools'] = {}; + for (const t of tools) config[t.name] = { tool: t.name }; + return { id: 'agent-1', name: 'Agent', agentType: AgentType.SmartAssistant, prompt: 'p', tools: config }; +} + +function registryOf(tools: DefinedTool[]): ToolRegistry { + const registry = new ToolRegistry(); + registry.registerMany(tools); + return registry; +} + +function turnCalling(...names: string[]): MockTurn { + return { toolCalls: names.map((name) => ({ name, id: `call_${name}` })) }; +} + +function toolIds(messages: readonly Message[]): Array { + return messages.filter((m) => m.role === 'tool').map((m) => m.toolCallId); +} + +function roles(messages: readonly Message[]): string[] { + return messages.map((m) => m.role); +} + +/** A KVCheckpointStore over a Map, optionally failing the n-th save (1-based). */ +function checkpointStore(failSave?: number): KVCheckpointStore & { data: Map } { + const data = new Map(); + let saves = 0; + const store = new KVCheckpointStore({ + get: async (key) => data.get(key) ?? null, + put: async (key, value) => { + saves++; + if (saves === failSave) throw new Error('disk gone'); + data.set(key, value); + }, + delete: async (key) => { + data.delete(key); + }, + }); + return Object.assign(store, { data }); +} + +/** An ApprovalStore that JSON round-trips its records, like a real one. */ +function approvalStore(): ApprovalStore & { snapshots: ExecutionSnapshot[]; raw: Map } { + const raw = new Map(); + const snapshots: ExecutionSnapshot[] = []; + return { + raw, + snapshots, + async save(pending: PendingApproval, snapshot: ExecutionSnapshot) { + snapshots.push(JSON.parse(JSON.stringify(snapshot)) as ExecutionSnapshot); + raw.set(pending.id, JSON.stringify({ pending, snapshot })); + }, + async resolve(id: string) { + const record = raw.get(id); + raw.delete(id); + return record ? JSON.parse(record) : null; + }, + }; +} + +interface Harness { + tools: DefinedTool[]; + checkpoints?: CheckpointStore; + approvals?: ApprovalStore; +} + +function execute(h: Harness, script: MockTurn[], extra: Partial = {}) { + const model = mockModel(script); + const result = AgentExecutor.execute({ + agent: agentFor(h.tools), + input: 'go', + provider: model, + toolRegistry: registryOf(h.tools), + approvalStore: h.approvals, + checkpointStore: h.checkpoints, + sessionId: h.checkpoints ? 's1' : undefined, + ...extra, + }); + return { model, result }; +} + +async function resume(h: Harness, paused: ExecutionResult, script: MockTurn[], approved = true) { + const model = mockModel(script); + const result = await resumeAfterApproval( + { id: paused.approvalId!, approved, note: approved ? undefined : 'not today' }, + h.approvals!, + registryOf(h.tools), + model, + {}, + h.checkpoints + ); + return { model, result }; +} + +describe('LOU-U7: approval in the middle of a tool batch', () => { + it('approved: the paused call and the calls after it each run once, every call has one result in call order', async () => { + const runs: Runs = {}; + const h: Harness = { + tools: [tool('a', runs), tool('b', runs, { needsApproval: true }), tool('c', runs)], + approvals: approvalStore(), + }; + const paused = await execute(h, [turnCalling('a', 'b', 'c')]).result; + expect(paused.finishReason).toBe('awaiting-approval'); + expect(runs).toEqual({ a: 1, b: 0, c: 0 }); + const [snapshot] = (h.approvals as ReturnType).snapshots; + expect(snapshot.remainingToolCalls?.map((c) => c.id)).toEqual(['call_c']); + + const { model, result } = await resume(h, paused, ['all done']); + + expect(result.text).toBe('all done'); + expect(runs).toEqual({ a: 1, b: 1, c: 1 }); + expect(toolIds(result.messages)).toEqual(['call_a', 'call_b', 'call_c']); + // The model is called once, with a complete, provider-valid turn. + expect(model.calls).toHaveLength(1); + expect(toolIds(model.calls[0].messages as Message[])).toEqual(['call_a', 'call_b', 'call_c']); + }); + + it('rejected: the paused call gets a structured rejection result and the remaining calls still run', async () => { + const runs: Runs = {}; + const h: Harness = { + tools: [tool('a', runs), tool('b', runs, { needsApproval: true }), tool('c', runs)], + approvals: approvalStore(), + }; + const paused = await execute(h, [turnCalling('a', 'b', 'c')]).result; + + const { result } = await resume(h, paused, ['ok, skipped b'], false); + + expect(runs).toEqual({ a: 1, b: 0, c: 1 }); + const [, rejected, c] = result.messages.filter((m) => m.role === 'tool'); + expect(rejected).toMatchObject({ toolCallId: 'call_b', isError: true }); + expect(JSON.parse(rejected.content)).toEqual({ error: expect.stringContaining('rejected'), note: 'not today' }); + expect(c).toMatchObject({ toolCallId: 'call_c', content: '"c done"' }); + }); + + it('a remaining call that needs approval pauses again; the chain completes with no call run twice', async () => { + const runs: Runs = {}; + const approvals = approvalStore(); + const h: Harness = { + tools: [ + tool('a', runs), + tool('b', runs, { needsApproval: true }), + tool('c', runs, { needsApproval: true }), + tool('d', runs), + ], + approvals, + }; + const first = await execute(h, [turnCalling('a', 'b', 'c', 'd')]).result; + + const second = await resume(h, first, []); + expect(second.result.finishReason).toBe('awaiting-approval'); + expect(second.result.approvalId).not.toBe(first.approvalId); + expect(second.model.calls).toHaveLength(0); + expect(approvals.snapshots[1].pendingToolCall.toolCallId).toBe('call_c'); + expect(approvals.snapshots[1].remainingToolCalls?.map((c) => c.id)).toEqual(['call_d']); + expect(toolIds(approvals.snapshots[1].currentMessages)).toEqual(['call_a', 'call_b']); + + const third = await resume(h, second.result, ['finished']); + expect(third.result.text).toBe('finished'); + expect(runs).toEqual({ a: 1, b: 1, c: 1, d: 1 }); + expect(toolIds(third.result.messages)).toEqual(['call_a', 'call_b', 'call_c', 'call_d']); + }); + + it('a remaining call with invalid arguments gets its validation error result', async () => { + const runs: Runs = {}; + const h: Harness = { + tools: [tool('b', runs, { needsApproval: true }), tool('c', runs, { input: z.object({ n: z.number() }) })], + approvals: approvalStore(), + }; + const paused = await execute(h, [turnCalling('b', 'c')]).result; + + const { result } = await resume(h, paused, ['done']); + + expect(runs).toEqual({ b: 1, c: 0 }); + const c = result.messages.find((m) => m.toolCallId === 'call_c'); + expect(c?.isError).toBe(true); + expect(toolIds(result.messages)).toEqual(['call_b', 'call_c']); + }); + + it('a snapshot saved without remainingToolCalls (older SDK) resumes as before: later calls do not run, but get a result', async () => { + const runs: Runs = {}; + const approvals = approvalStore(); + const h: Harness = { + tools: [tool('a', runs), tool('b', runs, { needsApproval: true }), tool('c', runs)], + approvals, + }; + const paused = await execute(h, [turnCalling('a', 'b', 'c')]).result; + const record = JSON.parse(approvals.raw.get(paused.approvalId!)!); + delete record.snapshot.remainingToolCalls; + approvals.raw.set(paused.approvalId!, JSON.stringify(record)); + + const { result } = await resume(h, paused, ['done']); + + expect(runs).toEqual({ a: 1, b: 1, c: 0 }); + expect(toolIds(result.messages)).toEqual(['call_a', 'call_b', 'call_c']); + const c = result.messages.find((m) => m.toolCallId === 'call_c')!; + expect(c.isError).toBe(true); + expect(JSON.parse(c.content).error).toMatch(/not run/); + }); +}); + +describe('LOU-U9: checkpoint after every model turn', () => { + it("checkpoints the model's tool-call turn before any tool runs", async () => { + const checkpoints = checkpointStore(); + let atToolStart: Checkpoint | null = null; + const probe = defineTool({ + name: 'probe', + description: 'reads the checkpoint', + input: z.object({}), + execute: async () => { + atToolStart = await checkpoints.load('s1'); + return 'ok'; + }, + }); + + await execute({ tools: [probe], checkpoints }, [turnCalling('probe'), 'done']).result; + + expect(atToolStart!.status).toBe('in-progress'); + expect(atToolStart!.messages.at(-1)).toMatchObject({ role: 'assistant', toolCalls: [{ id: 'call_probe' }] }); + }); + + it('a crash mid-tool resumes by running exactly the missing calls, without calling the model first', async () => { + const runs: Runs = {}; + const seenIds: string[] = []; + const checkpoints = checkpointStore(); + const h: Harness = { tools: [tool('a', runs), tool('b', runs, { crashOnce: true, seenIds })], checkpoints }; + + const crashed = execute(h, [turnCalling('a', 'b'), 'never reached']); + await expect(crashed.result).rejects.toThrow('process died'); + expect(crashed.model.calls).toHaveLength(1); + + // A fresh execute() with the same sessionId and the same request (a retry). + const resumed = execute(h, ['all done']); + const result = await resumed.result; + + expect(result.text).toBe('all done'); + expect(resumed.model.calls).toHaveLength(1); + // `a`'s result was recorded before the crash, so it never runs again; + // `b` was mid-execution, so it runs again (at-least-once) with the same id. + expect(runs).toEqual({ a: 1, b: 2 }); + expect(seenIds).toEqual(['call_b', 'call_b']); + expect(toolIds(resumed.model.calls[0].messages as Message[])).toEqual(['call_a', 'call_b']); + expect(roles(result.messages).filter((r) => r === 'user')).toHaveLength(1); + }); + + it('a store failure right after the model responded resumes without re-calling the model', async () => { + const runs: Runs = {}; + const failing = checkpointStore(2); // save 1 = the model turn, save 2 = the first tool result + const h: Harness = { tools: [tool('a', runs)], checkpoints: failing }; + await expect(execute(h, [turnCalling('a'), 'never reached']).result).rejects.toThrow('disk gone'); + + const resumed = execute(h, ['all done']); + + expect((await resumed.result).text).toBe('all done'); + expect(resumed.model.calls).toHaveLength(1); + expect(runs.a).toBe(2); + }); +}); + +describe('LOU-U8: resume with new input', () => { + it('(a) unfinished run + new input: appended after the pending tool results, never between', async () => { + const runs: Runs = {}; + const h: Harness = { tools: [tool('a', runs), tool('b', runs, { crashOnce: true })], checkpoints: checkpointStore() }; + await expect(execute(h, [turnCalling('a', 'b')]).result).rejects.toThrow('process died'); + + const resumed = execute(h, ['done'], { input: 'also check the weather' }); + const result = await resumed.result; + + const sent = resumed.model.calls[0].messages as Message[]; + expect(roles(sent)).toEqual(['system', 'user', 'assistant', 'tool', 'tool', 'user']); + expect(sent.at(-1)?.content).toBe('also check the weather'); + expect(result.messages.at(-1)).toEqual({ role: 'assistant', content: 'done' }); + }); + + it('(a) aborted run + new input: appended at once (interrupt and redirect)', async () => { + const runs: Runs = {}; + const controller = new AbortController(); + const h: Harness = { tools: [tool('a', runs)], checkpoints: checkpointStore() }; + const aborted = await execute(h, [turnCalling('a')], { + signal: controller.signal, + onEvent: (event) => { + if (event.type === 'tool-result') controller.abort(); + }, + }).result; + expect(aborted.finishReason).toBe('aborted'); + + const resumed = execute(h, ['redirected'], { input: 'actually, summarize instead' }); + const result = await resumed.result; + + expect(result.text).toBe('redirected'); + expect(runs.a).toBe(1); + const sent = resumed.model.calls[0].messages as Message[]; + expect(roles(sent)).toEqual(['system', 'user', 'assistant', 'tool', 'user']); + expect(sent.at(-1)?.content).toBe('actually, summarize instead'); + expect(result.steps).toBe(2); + }); + + it('(b) finished run: the session continues as a conversation', async () => { + const h: Harness = { tools: [], checkpoints: checkpointStore() }; + await execute(h, ['hello'], { input: 'hi' }).result; + + const next = execute(h, ['fine, thanks'], { input: 'how are you?' }); + const result = await next.result; + + expect((next.model.calls[0].messages as Message[]).map((m) => [m.role, m.content])).toEqual([ + ['system', 'p'], + ['user', 'hi'], + ['assistant', 'hello'], + ['user', 'how are you?'], + ]); + expect(result.steps).toBe(1); + expect(result.messages).toHaveLength(5); + const stored = await h.checkpoints!.load('s1'); + expect(stored?.status).toBe('finished'); + expect(stored?.messages).toEqual(result.messages); + }); + + it('(b) a caller that re-sends the whole history does not duplicate it', async () => { + const h: Harness = { tools: [], checkpoints: checkpointStore() }; + const first = await execute(h, ['hello'], { input: 'hi' }).result; + + const next = execute(h, ['sure'], { input: [...first.messages, { role: 'user', content: 'next' }] }); + await next.result; + + expect(roles(next.model.calls[0].messages as Message[])).toEqual(['system', 'user', 'assistant', 'user']); + }); + + it('(c) awaiting approval: execute() throws SessionAwaitingApprovalError; after resumeAfterApproval the session continues', async () => { + const runs: Runs = {}; + const h: Harness = { + tools: [tool('b', runs, { needsApproval: true })], + checkpoints: checkpointStore(), + approvals: approvalStore(), + }; + const paused = await execute(h, [turnCalling('b')]).result; + + const blocked = execute(h, ['never'], { input: 'are you there?' }); + const error = await blocked.result.catch((e: unknown) => e); + expect(error).toBeInstanceOf(SessionAwaitingApprovalError); + expect(error).toMatchObject({ sessionId: 's1', approvalId: paused.approvalId }); + expect((error as Error).message).toMatch(/resumeAfterApproval/); + expect(blocked.model.calls).toHaveLength(0); + expect(runs.b).toBe(0); + + await resume(h, paused, ['charged']); + const next = execute(h, ['you are welcome'], { input: 'thanks' }); + const result = await next.result; + + expect(result.text).toBe('you are welcome'); + expect(roles(next.model.calls[0].messages as Message[])).toEqual([ + 'system', 'user', 'assistant', 'tool', 'assistant', 'user', + ]); + }); + + it('new input queued behind pending calls survives another approval pause', async () => { + const runs: Runs = {}; + const approvals = approvalStore(); + const h: Harness = { + tools: [tool('a', runs, { crashOnce: true }), tool('b', runs, { needsApproval: true })], + checkpoints: checkpointStore(), + approvals, + }; + await expect(execute(h, [turnCalling('a', 'b')], { toolConcurrency: 1 }).result).rejects.toThrow('process died'); + + const paused = await execute(h, [], { input: 'queued question' }).result; + expect(paused.finishReason).toBe('awaiting-approval'); + expect(paused.messages.some((m) => m.content === 'queued question')).toBe(false); + expect(approvals.snapshots[0].currentMessages.at(-1)).toEqual({ role: 'user', content: 'queued question' }); + + const { model } = await resume(h, paused, ['answered']); + + const sent = model.calls[0].messages as Message[]; + expect(roles(sent)).toEqual(['system', 'user', 'assistant', 'tool', 'tool', 'user']); + expect(sent.at(-1)?.content).toBe('queued question'); + expect(runs).toEqual({ a: 2, b: 1 }); + }); +}); + +describe('newSessionMessages (LOU-U8 input merging)', () => { + const stored: Message[] = [ + { role: 'system', content: 'p' }, + { role: 'user', content: 'hi' }, + { role: 'assistant', content: 'hello' }, + { role: 'user', content: '[provider-error] {"error":"rate limited"}' }, + ]; + + it('appends a repeated message to a finished conversation (it is a new turn)', () => { + expect(newSessionMessages(stored.slice(0, 3), 'hi', true)).toEqual([{ role: 'user', content: 'hi' }]); + }); + + it('treats the starting user turn as a retry of an unfinished run, skipping provider-error notes', () => { + expect(newSessionMessages(stored, 'hi', false)).toEqual([]); + expect(newSessionMessages(stored, [], false)).toEqual([]); + }); + + it('appends only what follows a lagging copy of the history to an unfinished run', () => { + const input: Message[] = [{ role: 'user', content: 'hi' }, { role: 'user', content: 'new' }]; + expect(newSessionMessages(stored, input, false)).toEqual([{ role: 'user', content: 'new' }]); + }); + + it('appends unrelated input to an unfinished run', () => { + expect(newSessionMessages(stored, 'something else', false)).toEqual([{ role: 'user', content: 'something else' }]); + }); +}); diff --git a/src/execution/errors.ts b/src/execution/errors.ts index daf5006a5..aeb7fe4e8 100644 --- a/src/execution/errors.ts +++ b/src/execution/errors.ts @@ -58,6 +58,38 @@ export class LLMProviderError extends SDKError { } } +/** + * LOU-U8: thrown by `AgentExecutor.execute()` when its `sessionId` names a + * run that is paused awaiting a human approval. New input must not bypass + * the pending decision: resolve it with `resumeAfterApproval()` first. + * + * @example + * ```ts + * try { + * await AgentExecutor.execute({ agent, input: 'hi', provider, sessionId, checkpointStore }); + * } catch (error) { + * if (error instanceof SessionAwaitingApprovalError) { + * console.log(`decide approval ${error.approvalId} first`); + * } + * } + * ``` + */ +export class SessionAwaitingApprovalError extends SDKError { + constructor( + public readonly sessionId: string, + public readonly approvalId: string | undefined + ) { + super( + `Session '${sessionId}' is paused awaiting approval${approvalId ? ` '${approvalId}'` : ''}, so execute() ` + + 'cannot add new input to it. Resolve the approval with resumeAfterApproval({ id: approvalId, approved: true }, ' + + 'approvalStore, toolRegistry, provider, options, checkpointStore) - passing the same checkpointStore so the ' + + 'session is marked as resumed - then call execute() again with your new input.', + 'SESSION_AWAITING_APPROVAL' + ); + this.name = 'SessionAwaitingApprovalError'; + } +} + /** * Flow execution error */ diff --git a/src/execution/parallelToolCalls.test.ts b/src/execution/parallelToolCalls.test.ts index 1b8034fd9..1b445f749 100644 --- a/src/execution/parallelToolCalls.test.ts +++ b/src/execution/parallelToolCalls.test.ts @@ -317,13 +317,17 @@ describe('parallel tool calls (LOU-V3)', () => { const { result } = run([a.tool, b.tool], ['a', 'b'], { sessionId: 's1', checkpointStore }); await Promise.all([a.started, b.started]); + // LOU-U9: the model's turn itself is checkpointed before any call starts. + expect(saved).toEqual([[]]); + b.release.resolve('B'); await tick(); - expect(saved).toEqual([]); // `b` finished, but `a` (before it) has not + expect(saved).toEqual([[]]); // `b` finished, but `a` (before it) has not a.release.resolve('A'); await result; - expect(saved).toEqual([['call_a', 'call_b']]); + // Then the in-order prefix once, then the 'finished' checkpoint (LOU-U8). + expect(saved).toEqual([[], ['call_a', 'call_b'], ['call_a', 'call_b']]); }); describe('approval inside a batch', () => { @@ -385,9 +389,10 @@ describe('parallel tool calls (LOU-V3)', () => { expect(resumed.finishReason).toBe('stop'); // The approved call ran once; calls before it were not re-executed. for (const name of [...before, names[position]]) expect(runs[name]).toBe(1); - // Calls after the approval call are not run on resume - today's - // behavior, tracked separately by LOU-U7. - for (const name of names.slice(position + 1)) expect(runs[name]).toBe(0); + // LOU-U7: calls after the approval call run on resume, exactly once, + // and every call of the turn ends up with one result, in call order. + for (const name of names.slice(position + 1)) expect(runs[name]).toBe(1); + expect(toolMessageIds(resumed.messages)).toEqual(names.map((n) => `call_${n}`)); }); }); diff --git a/src/execution/resume.test.ts b/src/execution/resume.test.ts index 72a4d474c..d028c8913 100644 --- a/src/execution/resume.test.ts +++ b/src/execution/resume.test.ts @@ -511,7 +511,11 @@ describe('Execution - resumeAfterApproval', () => { // point (it doesn't know about the pending 'chargeCard' call at all). const preResumeCheckpoint = await checkpointStore.load(sessionId); expect(preResumeCheckpoint).not.toBeNull(); - expect(preResumeCheckpoint!.stepIndex).toBe(1); + // LOU-U8/U9: the checkpoint now records the pause itself (the model's + // chargeCard turn was checkpointed before the tool gate ran). + expect(preResumeCheckpoint!.status).toBe('awaiting-approval'); + expect(preResumeCheckpoint!.approvalId).toBe(paused.approvalId); + expect(preResumeCheckpoint!.stepIndex).toBe(2); expect(preResumeCheckpoint!.messages.some((m) => m.toolName === 'chargeCard')).toBe(false); // The reviewer's repro: explicitly pass sessionId+checkpointStore @@ -556,10 +560,12 @@ describe('Execution - resumeAfterApproval', () => { // would yield 2 if the checkpoint were wrongly rehydrated instead). expect(resumed.steps).toBe(3); - // Defense in depth: the stale checkpoint must have been proactively - // cleared so a later, unrelated execute() call reusing this sessionId - // can't rehydrate this pre-pause state either. - expect(await checkpointStore.load(sessionId)).toBeNull(); + // Defense in depth: the pre-pause checkpoint was cleared before the + // resumed run started; what is stored now is that run's own 'finished' + // checkpoint (LOU-U8), which includes the deferred chargeCard result. + const finalCheckpoint = await checkpointStore.load(sessionId); + expect(finalCheckpoint?.status).toBe('finished'); + expect(finalCheckpoint?.messages).toEqual(resumed.messages); }); it('LOU-K5: writes a NEW checkpoint for a tool-call step taken after a successful resume, when a checkpointStore is supplied', async () => { @@ -684,10 +690,9 @@ describe('Execution - resumeAfterApproval', () => { expect(savedCheckpoint.messages.some((m: any) => m.toolName === 'chargeCard')).toBe(true); expect(savedCheckpoint.messages.some((m: any) => m.toolName === 'lookup')).toBe(true); - // The run reached a terminal state, so AgentExecutor's own - // "clear the checkpoint when finished" cleanup (unrelated to this fix) - // removes it afterwards - this is expected and not a regression. - expect(await checkpointStore.load(sessionId)).toBeNull(); + // The run reached a terminal state, so its checkpoint is kept, marked + // 'finished', for session continuation (LOU-U8). + expect((await checkpointStore.load(sessionId))?.status).toBe('finished'); }); it('LOU-T1: carries businessState across a pause-for-approval -> approve -> resume cycle without the resumed call re-passing it', async () => { @@ -819,8 +824,8 @@ describe('Execution - resumeAfterApproval', () => { expect(chargeExecute).toHaveBeenCalledTimes(1); expect(lookupExecute).toHaveBeenCalledTimes(2); - // The run reached a terminal state, so the checkpoint is deleted at the - // end - but every intermediate save the resumed run made along the way + // The run reached a terminal state, so the checkpoint is kept, marked + // 'finished' (LOU-U8) - and every save the resumed run made along the way // must have carried the businessState forward. expect(saveSpy).toHaveBeenCalled(); for (const [, checkpoint] of saveSpy.mock.calls) { @@ -829,7 +834,7 @@ describe('Execution - resumeAfterApproval', () => { stage: 'awaiting-approval', }); } - expect(await checkpointStore.load(sessionId)).toBeNull(); + expect((await checkpointStore.load(sessionId))?.status).toBe('finished'); }); it('LOU-T1: an explicit businessState passed to resumeAfterApproval() overrides the stale pre-pause checkpoint value', async () => { @@ -1113,7 +1118,7 @@ describe('Execution - resumeAfterApproval', () => { { hooks } ); - expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, {}); + expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, { toolCallId: 'call-1' }); }); it('a postToolCall hook that throws on the resume path propagates as a rejected promise, not a swallowed {error} tool-result', async () => { diff --git a/src/execution/resume.ts b/src/execution/resume.ts index 6d9ed28ad..8f907b907 100644 --- a/src/execution/resume.ts +++ b/src/execution/resume.ts @@ -3,7 +3,7 @@ * tool-call approval. */ -import { LLMProvider, Message } from '../providers'; +import { LLMProvider, Message, ToolCall } from '../providers'; import { ToolRegistry } from '../tools'; import { ToolDescriptor } from '../types'; import { @@ -18,6 +18,7 @@ import { NoopSandbox } from '../security/sandboxCore'; import { executeToolWithSandboxGuard } from './sandboxGuard'; import { HookRegistry } from './hooks'; import { toolErrorMessage } from './propagatingToolError'; +import { splitPendingTurn } from './transcript'; /** * Options passed through to the underlying AgentExecutor.execute() call @@ -120,12 +121,17 @@ export async function resumeAfterApproval( note: decision.note, }, true); messages.push(toolMessage); + closeUnlistedToolCalls(messages, snapshot.remainingToolCalls); + // LOU-U7: the turn's remaining calls still have no result here; + // AgentExecutor.execute() finds them in the transcript and runs them + // through its normal batch path before calling the model again. return continueResumedRun(snapshot, messages, { provider, toolRegistry, executeOptions, checkpointStore, + approvalStore, staleBusinessState, }); } @@ -174,6 +180,32 @@ async function clearStaleCheckpoint( return staleCheckpoint?.businessState; } +/** + * LOU-U7 backward compatibility: a snapshot saved before + * `remainingToolCalls` existed records no remaining calls, so - as before - + * no other call of the paused turn runs on resume. Each one still without a + * result gets an error result instead, so the transcript stays valid for + * the provider (every tool call answered exactly once). + */ +function closeUnlistedToolCalls(messages: Message[], remaining: ToolCall[] | undefined): void { + const keep = new Set((remaining ?? []).map((call) => call.id)); + for (const call of splitPendingTurn(messages).pendingToolCalls) { + if (keep.has(call.id)) { + continue; + } + messages.push({ + role: 'tool', + content: JSON.stringify({ + error: 'Tool call was not run: the run paused for approval before reaching it and this approval was saved without its remaining calls', + }), + name: call.function.name, + toolCallId: call.id, + toolName: call.function.name, + isError: true, + }); + } +} + /** The `tool` message carrying a resumed tool call's result (or rejection). */ function toolResultMessage(pending: PendingApproval, payload: unknown, isError = false): Message { return { @@ -285,7 +317,8 @@ async function executeApprovedTool( toolDesc, args, sandbox, - executeOptions.signal + executeOptions.signal, + pending.toolCallId ), }; } catch (error) { @@ -343,6 +376,7 @@ function continueResumedRun( toolRegistry: ToolRegistry; executeOptions: ResumeExecuteOptions; checkpointStore?: CheckpointStore; + approvalStore: ApprovalStore; staleBusinessState: unknown; } ): Promise { @@ -355,6 +389,9 @@ function continueResumedRun( toolRegistry: run.toolRegistry, sessionId: snapshot.sessionId, checkpointStore: run.checkpointStore, + // LOU-U7: a remaining call of the paused turn (or a later one) may need + // approval too - it pauses into the same store unless told otherwise. + approvalStore: executeOptions.approvalStore ?? run.approvalStore, // LOU-T1: carry the pre-pause checkpoint's businessState forward into // the resumed run's own checkpoint-writes by default, so a consumer's // domain state (order id, ticket id, workflow stage, ...) survives a diff --git a/src/execution/sandboxGuard.ts b/src/execution/sandboxGuard.ts index ff321aadc..485e79cb2 100644 --- a/src/execution/sandboxGuard.ts +++ b/src/execution/sandboxGuard.ts @@ -46,7 +46,8 @@ export async function executeToolWithSandboxGuard( toolDesc: ToolDescriptor, args: Record, sandbox: SandboxAdapter, - signal?: AbortSignal + signal?: AbortSignal, + toolCallId?: string ): Promise { if (toolDesc.requiresSandbox) { if (!toolDesc.sandboxExecute) { @@ -61,7 +62,9 @@ export async function executeToolWithSandboxGuard( } // The 'ai' SDK types toolCallId/messages as required, but tools invoked - // here are not part of an 'ai' SDK generation, so only abortSignal is set. - const executeOptions = { abortSignal: signal } as ToolExecutionOptions; + // here are not part of an 'ai' SDK generation, so `messages` is not set + // (LOU-U15). `toolCallId` (LOU-U9) is the model's id for this call - the + // same on a resumed re-run - so tools can use it as an idempotency key. + const executeOptions = { abortSignal: signal, toolCallId } as ToolExecutionOptions; return toolDesc.tool.execute ? toolDesc.tool.execute(args, executeOptions) : null; } diff --git a/src/execution/toolCallExecution.ts b/src/execution/toolCallExecution.ts index b0613e965..3fc653ac7 100644 --- a/src/execution/toolCallExecution.ts +++ b/src/execution/toolCallExecution.ts @@ -331,7 +331,8 @@ async function doExecuteToolCall( toolDesc, args, sandbox, - signal + signal, + toolCall.id ); return { diff --git a/src/execution/transcript.ts b/src/execution/transcript.ts new file mode 100644 index 000000000..28534c4e6 --- /dev/null +++ b/src/execution/transcript.ts @@ -0,0 +1,113 @@ +/** + * Durable-execution transcript helpers (LOU-U7, LOU-U8, LOU-U9): find the + * tool calls of an interrupted model turn that still have no result, and + * merge new input into a stored session transcript without duplicating it + * or breaking the "tool results directly follow their tool-call turn" rule + * every provider enforces. + */ + +import type { Message, ToolCall } from '../providers'; + +/** A transcript split around its last model turn's unanswered tool calls. */ +export interface PendingTurn { + /** The transcript up to the last assistant turn plus its recorded results, in call order. */ + messages: Message[]; + /** Tool calls of the last assistant turn that have no result yet, in call order. */ + pendingToolCalls: ToolCall[]; + /** + * Messages found after that turn which are not its results (e.g. a user + * message queued while the turn was incomplete). They belong after the + * turn's results, so they are held back until every call has one. + */ + queuedInput: Message[]; +} + +function lastIndexWhere(messages: readonly Message[], test: (m: Message) => boolean): number { + for (let i = messages.length - 1; i >= 0; i--) { + if (test(messages[i])) return i; + } + return -1; +} + +/** + * Splits off the unanswered tool calls of the transcript's last assistant + * turn, moving that turn's results (in call order) directly behind it and + * any other message found after it behind the results - held back as + * `queuedInput` while calls are still pending. A transcript already in + * that shape with every call answered comes back unchanged. + */ +export function splitPendingTurn(messages: Message[]): PendingTurn { + const turnIndex = lastIndexWhere(messages, (m) => m.role === 'assistant'); + const calls = turnIndex >= 0 ? (messages[turnIndex].toolCalls ?? []) : []; + if (calls.length === 0) { + return { messages, pendingToolCalls: [], queuedInput: [] }; + } + const order = new Map(calls.map((call, index) => [call.id, index])); + const after = messages.slice(turnIndex + 1); + const results = after.filter((m) => m.role === 'tool' && order.has(m.toolCallId ?? '')); + const others = after.filter((m) => !results.includes(m)); + const answered = new Set(results.map((m) => m.toolCallId)); + const pendingToolCalls = calls.filter((call) => !answered.has(call.id)); + const callIndex = (m: Message) => order.get(m.toolCallId ?? '') ?? 0; + const turn = [...messages.slice(0, turnIndex + 1), ...results.sort((a, b) => callIndex(a) - callIndex(b))]; + return pendingToolCalls.length === 0 + ? { messages: [...turn, ...others], pendingToolCalls, queuedInput: [] } + : { messages: turn, pendingToolCalls, queuedInput: others }; +} + +/** `input` as messages (a string is one user message). */ +export function inputMessages(input: string | Message[]): Message[] { + return typeof input === 'string' ? [{ role: 'user', content: input }] : [...input]; +} + +function messageKey(m: Message): string { + return JSON.stringify([m.role, m.content, m.toolCallId ?? null]); +} + +function sharedPrefixLength(known: readonly Message[], incoming: readonly Message[]): number { + let shared = 0; + while (shared < known.length && shared < incoming.length && messageKey(known[shared]) === messageKey(incoming[shared])) { + shared++; + } + return shared; +} + +/** True for a real user turn (not a LOU-T4 `[provider-error]` note). */ +function isUserTurn(m: Message): boolean { + return m.role === 'user' && !m.content.startsWith('[provider-error]'); +} + +/** True when `incoming` re-sends the user turn the stored, unfinished run started from. */ +function repeatsLatestUserTurn(known: readonly Message[], incoming: readonly Message[]): boolean { + const end = lastIndexWhere(known as Message[], isUserTurn); + if (end < 0 || incoming.length === 0 || incoming.length > end + 1) { + return false; + } + const candidate = known.slice(end + 1 - incoming.length, end + 1); + return candidate.every((m, i) => messageKey(m) === messageKey(incoming[i])); +} + +/** + * LOU-U8: the messages of `input` that are new to a stored session + * transcript - what a session resume or continuation appends. System + * messages in `input` are dropped (the session already has its own). + * + * - `input` that starts with the whole stored transcript (a caller that + * keeps the history itself and re-sends it) contributes only what follows. + * - For a finished run, anything else is appended as is. + * - For an unfinished run, `input` that re-sends the stored transcript's + * beginning, or the user turn the run started from, is a retry of that + * run and contributes nothing; a different message is appended. + */ +export function newSessionMessages(stored: readonly Message[], input: string | Message[], finished: boolean): Message[] { + const known = stored.filter((m) => m.role !== 'system'); + const incoming = inputMessages(input).filter((m) => m.role !== 'system'); + const shared = sharedPrefixLength(known, incoming); + if (shared === known.length || (!finished && shared > 0)) { + return incoming.slice(shared); + } + if (finished || !repeatsLatestUserTurn(known, incoming)) { + return incoming; + } + return []; +} From c9277852a6220303e7880fbb994791473b1b1ae0 Mon Sep 17 00:00:00 2001 From: Ali Mohammad Date: Thu, 1 Oct 2026 11:19:24 +0300 Subject: [PATCH 2/2] agent-forge: adapt to LOU-V5 RunUsage and tool execute options runRegistry's synthetic flow result and debugController.test's onLLMResponse call predate LOU-V5's RunUsage / three-argument onLLMResponse (main's agent-forge server typecheck failed on them); compileHooks.test now matches the tool options with objectContaining since they also carry onDelegatedUsage. Co-Authored-By: Claude Opus 5.5 --- .../server/__tests__/debugController.test.ts | 2 +- apps/agent-forge/server/compileHooks.test.ts | 2 +- apps/agent-forge/server/runRegistry.ts | 12 +++++++++++- 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/apps/agent-forge/server/__tests__/debugController.test.ts b/apps/agent-forge/server/__tests__/debugController.test.ts index a58e536e8..9ae2cf5a8 100644 --- a/apps/agent-forge/server/__tests__/debugController.test.ts +++ b/apps/agent-forge/server/__tests__/debugController.test.ts @@ -49,7 +49,7 @@ describe('DebugSession', () => { await hookPromise; // Step is one-shot: the next hook call does NOT pause again. - const secondPromise = session.hooks().onLLMResponse?.({} as any, 1); + const secondPromise = session.hooks().onLLMResponse?.({} as any, 1, {} as any); await secondPromise; expect(session.snapshot().paused).toBe(false); }); diff --git a/apps/agent-forge/server/compileHooks.test.ts b/apps/agent-forge/server/compileHooks.test.ts index 950dde889..56463784f 100644 --- a/apps/agent-forge/server/compileHooks.test.ts +++ b/apps/agent-forge/server/compileHooks.test.ts @@ -82,7 +82,7 @@ describe('compileHooksFromSpecPolicy', () => { maxSteps: 1, }); - expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, { toolCallId: 'call-1' }); + expect(execute).toHaveBeenCalledWith({ email: '[REDACTED]' }, expect.objectContaining({ toolCallId: 'call-1' })); }); it('compiles a preGenerate hook that actually injects a message when the agent runs', async () => { diff --git a/apps/agent-forge/server/runRegistry.ts b/apps/agent-forge/server/runRegistry.ts index b8afe5dd1..bf7c240a1 100644 --- a/apps/agent-forge/server/runRegistry.ts +++ b/apps/agent-forge/server/runRegistry.ts @@ -608,7 +608,17 @@ export class RunManager extends EventEmitter { text, messages: [{ role: 'assistant', content: text }], toolCalls: [], - usage: { promptTokens: 0, completionTokens: 0, totalTokens: 0 }, + usage: { + inputTokens: 0, + outputTokens: 0, + totalTokens: 0, + costUsd: 0, + modelCalls: 0, + estimated: false, + byModel: {}, + promptTokens: 0, + completionTokens: 0, + }, finishReason: 'stop', steps: flowResult.steps, };