diff --git a/README.md b/README.md index 039ced5ac..19eeab3c8 100644 --- a/README.md +++ b/README.md @@ -250,6 +250,7 @@ console.log(session.id, session.messages.length); - [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 - [Streaming](docs/streaming.md) - `agent.stream()`: the typed event schema, terminal and SSE examples - [Workspace tools](docs/workspace-tools.md) - file system and shell tools for coding agents, and their security model - [Tracing and observability](docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in @@ -398,21 +399,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). + ### Sub-agents and delegation Give a lead agent named sub-agents; it delegates with one auto-registered 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/runRegistry.ts b/apps/agent-forge/server/runRegistry.ts index f090d8111..875551da2 100644 --- a/apps/agent-forge/server/runRegistry.ts +++ b/apps/agent-forge/server/runRegistry.ts @@ -355,19 +355,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); @@ -403,15 +400,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 a7b4272b2..9676c94ce 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. | ### Sub-agents @@ -36,6 +37,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 })`, @@ -75,9 +88,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 @@ -120,10 +135,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 @@ -133,9 +148,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..c82fa2ee5 --- /dev/null +++ b/docs/durable-execution.md @@ -0,0 +1,185 @@ +# 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; `usage` continues from the checkpointed + totals (the same holds for `resumeAfterApproval()`, from the snapshot). + - 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/docs/sessions.md b/docs/sessions.md index e61f07800..9bf464571 100644 --- a/docs/sessions.md +++ b/docs/sessions.md @@ -37,8 +37,10 @@ contains an assistant tool-call turn without the matching tool results. Sessions are a thin layer: the session owns the transcript and hands it to the executor on each turn. They do not use the checkpoint/resume mechanism -(`sessionId` + `checkpointStore` on `AgentExecutor.execute()`), which is for -resuming one interrupted run, not for continuing a conversation. +(`sessionId` + `checkpointStore` on `AgentExecutor.execute()`). That lower-level +mechanism also continues a finished conversation, and additionally resumes a +run interrupted mid-turn (a crash, an abort, an approval pause) without +re-running finished tools - see [Durable execution](./durable-execution.md). ## Stores diff --git a/llms-full.txt b/llms-full.txt index 88d5d9d41..513f0e2aa 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -248,6 +248,7 @@ console.log(session.id, session.messages.length); - [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 - [Streaming](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/streaming.md) - `agent.stream()`: the typed event schema, terminal and SSE examples - [Workspace tools](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/workspace-tools.md) - file system and shell tools for coding agents, and their security model - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md) - OpenTelemetry GenAI spans, attribute table, content opt-in @@ -396,21 +397,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). + ### Sub-agents and delegation Give a lead agent named sub-agents; it delegates with one auto-registered @@ -1181,6 +1187,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. | ### Sub-agents @@ -1200,6 +1207,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 })`, @@ -1239,9 +1258,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 @@ -1284,10 +1305,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 @@ -1297,9 +1318,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 @@ -2559,6 +2582,194 @@ 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; `usage` continues from the checkpointed + totals (the same holds for `resumeAfterApproval()`, from the snapshot). + - 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. + # Evals Source: docs/evals.md @@ -3017,8 +3228,10 @@ contains an assistant tool-call turn without the matching tool results. Sessions are a thin layer: the session owns the transcript and hands it to the executor on each turn. They do not use the checkpoint/resume mechanism -(`sessionId` + `checkpointStore` on `AgentExecutor.execute()`), which is for -resuming one interrupted run, not for continuing a conversation. +(`sessionId` + `checkpointStore` on `AgentExecutor.execute()`). That lower-level +mechanism also continues a finished conversation, and additionally resumes a +run interrupted mid-turn (a crash, an abort, an approval pause) without +re-running finished tools - see [Durable execution](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/durable-execution.md). ## Stores diff --git a/llms.txt b/llms.txt index a62e121ae..b633e3c40 100644 --- a/llms.txt +++ b/llms.txt @@ -10,6 +10,7 @@ - [API Overview](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/api-overview.md): The package root (`@loushy/build-ai-agent`) exports everything below. - [Agent directories](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/agent-directories.md): An agent can be defined as a directory. - [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. - [Evals](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/evals.md): Evals are regression tests for agent *behaviour*: which tools the agent called, in what order, with which arguments, how many steps it took, and what it said. - [Tracing and observability](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/observability.md): `AgentExecutor.execute()` and `FlowExecutor.execute()` emit spans through a `TraceExporter`. - [Sessions](https://github.com/LinuxDevil/agent-sdk/blob/main/docs/sessions.md): `agent.send(text)` is single-turn: every call starts from an empty history. 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 b8041e247..beee7edca 100644 --- a/src/execution/AgentExecutor.ts +++ b/src/execution/AgentExecutor.ts @@ -187,7 +187,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 @@ -400,15 +430,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 @@ -552,6 +584,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; + } + } + return this.runSteps(options, state, tools, agentSpanId); } @@ -572,7 +616,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); } @@ -585,21 +631,22 @@ 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 of the + * current step - LOU-U7/U9), bracketed by the stream's step events, 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 { const runEvents = runEventsOf(options); runEvents?.stepStart(state.steps); try { - const outcome = await this.runStep(options, state, tools, agentSpanId); + const outcome = await step(); runEvents?.stepDone(state.steps, typeof outcome === 'string' ? undefined : outcome.finishReason); return outcome; } catch (error) { @@ -751,10 +798,7 @@ export class AgentExecutor { toolCalls: ToolCall[], agentSpanId: string ): Promise { - const { onEvent } = options; state.toolCalls.push(...toolCalls); - // LOU-Y1: tool calls whose sub-agent paused for approval, in call order. - const suspensions: SubagentSuspension[] = []; // Add assistant message with tool calls state.messages.push({ @@ -763,6 +807,49 @@ 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; + // LOU-Y1: tool calls whose sub-agent paused for approval, in call order. + const suspensions: SubagentSuspension[] = []; const batch = await runToolBatch( toolCalls, options.toolConcurrency ?? 'unbounded', @@ -805,7 +892,10 @@ export class AgentExecutor { } const suspension = settleSuspensions(state.messages, suspensions, Boolean(batch.approval)); 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); } if (suspension) { return this.savePause(options, state, suspensionRecord(options, state, suspension)); @@ -912,15 +1002,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 } = options; if (!approvalStore) { @@ -940,10 +1032,12 @@ export class AgentExecutor { const snapshot: ExecutionSnapshot = { // The agent as configured: resume re-applies skills and sub-agents. agent: baseAgentOf(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, usage: structuredClone(state.usage), }; return this.savePause(options, state, { pending, snapshot }); @@ -951,7 +1045,8 @@ export class AgentExecutor { /** * Saves an approval record and ends this execute() call with an - * 'awaiting-approval' result. + * 'awaiting-approval' result. The checkpoint is marked + * 'awaiting-approval' (LOU-U8) so new input cannot bypass the decision. */ private static async savePause( options: ExecuteOptions, @@ -959,7 +1054,10 @@ export class AgentExecutor { { pending, snapshot }: { pending: PendingApproval; snapshot: ExecutionSnapshot } ): Promise { const { approvalStore, onEvent } = options; + // 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); runEventsOf(options)?.approvalRequested(pending); this.emitEvent(onEvent, { @@ -1012,7 +1110,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 @@ -1039,15 +1137,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 dda4ce44f..422e97745 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'; import type { RunUsage } from '../models/usage'; @@ -55,6 +55,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[]; /** Usage the paused run had spent (LOU-V5), so the resumed run continues its totals. */ usage?: RunUsage; /** diff --git a/src/execution/agentRunState.ts b/src/execution/agentRunState.ts index ffe4bd829..568675796 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 { CallUsage, RunUsage, StepUsage } from '../models/usage'; import { emptyRunUsage, recordStepUsage, restoreRunUsage } from './runUsage'; import type { ExecuteOptions, ExecutionResult } from './AgentExecutor'; @@ -37,6 +38,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[]; } /** @@ -47,28 +62,11 @@ 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)]; } type InitialRunState = Pick< @@ -76,8 +74,13 @@ type InitialRunState = Pick< 'messages' | 'toolCalls' | 'usage' | 'stepUsage' | '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 @@ -87,13 +90,17 @@ 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[])], - // LOU-V5: continue from the checkpointed totals (older checkpoints: token counts only). - usage: restoreRunUsage(checkpoint.usage), - stepUsage: [...(checkpoint.stepUsage ?? [])], - steps: checkpoint.stepIndex, + messages: [...checkpoint.messages, ...added], + toolCalls: finished ? [] : [...(checkpoint.toolCalls as ToolCall[])], + // LOU-V5: an unfinished run continues from the checkpointed totals (older + // checkpoints: token counts only); a new turn of a finished one starts at zero. + usage: finished ? emptyRunUsage() : restoreRunUsage(checkpoint.usage), + stepUsage: finished ? [] : [...(checkpoint.stepUsage ?? [])], + steps: finished ? 0 : checkpoint.stepIndex, businessState: businessState === undefined ? checkpoint.businessState : businessState, }; } @@ -113,7 +120,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; @@ -121,12 +131,21 @@ export async function loadRunState(options: ExecuteOptions): Promise { const { agent, sessionId, checkpointStore } = options; if (!sessionId || !checkpointStore) { @@ -151,12 +176,14 @@ export async function saveStepCheckpoint( agentId: agent.id || '', sessionId, stepIndex: state.steps, - messages: [...state.messages], + messages: [...state.messages, ...state.queuedInput], toolCalls: [...state.toolCalls], usage: structuredClone(state.usage), stepUsage: [...state.stepUsage], 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 26457286e..fffc13c80 100644 --- a/src/execution/checkpoint.ts +++ b/src/execution/checkpoint.ts @@ -9,8 +9,23 @@ import type { StepUsage } from '../models/usage'; import type { CheckpointUsage } from './runUsage'; /** - * 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; @@ -62,6 +77,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..c4d61e128 --- /dev/null +++ b/src/execution/durableExecution.test.ts @@ -0,0 +1,518 @@ +/** + * 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('usage continuity across resumes (LOU-V5 x LOU-U7/U9)', () => { + const callTurn = (...names: string[]): MockTurn => ({ ...(turnCalling(...names) as object), usage: { inputTokens: 100, outputTokens: 10 } }); + const finalTurn: MockTurn = { text: 'done', usage: { inputTokens: 200, outputTokens: 20 } }; + + it('a crash-resume with pending tool calls continues the checkpointed totals', async () => { + const runs: Runs = {}; + const h: Harness = { tools: [tool('a', runs), tool('b', runs, { crashOnce: true })], checkpoints: checkpointStore() }; + await expect(execute(h, [callTurn('a', 'b')]).result).rejects.toThrow('process died'); + + const result = await execute(h, [finalTurn]).result; + + expect(result.usage).toMatchObject({ inputTokens: 300, outputTokens: 30, totalTokens: 330, modelCalls: 2 }); + expect(result.stepUsage).toHaveLength(2); + }); + + it('an approval pause -> resume with remaining calls continues the snapshotted totals', async () => { + const runs: Runs = {}; + const h: Harness = { + tools: [tool('a', runs), tool('b', runs, { needsApproval: true }), tool('c', runs)], + approvals: approvalStore(), + checkpoints: checkpointStore(), + }; + const paused = await execute(h, [callTurn('a', 'b', 'c')]).result; + expect(paused.usage).toMatchObject({ inputTokens: 100, outputTokens: 10, modelCalls: 1 }); + + const { result } = await resume(h, paused, [finalTurn]); + + expect(runs).toEqual({ a: 1, b: 1, c: 1 }); + expect(result.usage).toMatchObject({ inputTokens: 300, outputTokens: 30, totalTokens: 330, modelCalls: 2 }); + }); + + it('tool execute options carry toolCallId, abortSignal and onDelegatedUsage together', async () => { + let seen: Record = {}; + const probe = defineTool({ + name: 'probe', + description: 'records its options', + input: z.object({}), + execute: async (_args, ctx) => { + seen = { ...ctx }; + return 'ok'; + }, + }); + const controller = new AbortController(); + + await execute({ tools: [probe] }, [turnCalling('probe'), 'done'], { signal: controller.signal }).result; + + expect(seen.toolCallId).toBe('call_probe'); + expect(seen.abortSignal).toBe(controller.signal); + expect(typeof seen.onDelegatedUsage).toBe('function'); + }); + + it('a new turn of a finished session counts its own usage from zero', async () => { + const h: Harness = { tools: [], checkpoints: checkpointStore() }; + await execute(h, [finalTurn], { input: 'hi' }).result; + + const next = await execute(h, [{ text: 'again', usage: { inputTokens: 50, outputTokens: 5 } }], { input: 'more' }).result; + + expect(next.usage).toMatchObject({ inputTokens: 50, outputTokens: 5, modelCalls: 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 97787a1e0..49d02f9b7 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'; import { replaceToolResult, type ToolCallScope } from './subagentRuntime'; import type { RunUsage } from '../models/usage'; import { emptyRunUsage, mergeDelegatedUsage, restoreRunUsage } from './runUsage'; @@ -135,12 +136,17 @@ export async function resumeAfterApproval( } // LOU-Y1: a call whose sub-agent paused already has a placeholder result. replaceToolResult(messages, step.message); + 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, usage: ctx.usage, }); @@ -220,6 +226,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 { @@ -334,7 +366,7 @@ async function executeApprovedTool( args, sandbox, executeOptions.signal, - undefined, + { toolCallId: pending.toolCallId }, scope ), }; @@ -393,6 +425,7 @@ function continueResumedRun( toolRegistry: ToolRegistry; executeOptions: ResumeExecuteOptions; checkpointStore?: CheckpointStore; + approvalStore: ApprovalStore; staleBusinessState: unknown; usage: RunUsage; } @@ -406,6 +439,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 ac372fe1d..abe026887 100644 --- a/src/execution/sandboxGuard.ts +++ b/src/execution/sandboxGuard.ts @@ -28,6 +28,12 @@ import type { RunUsage } from '../models/usage'; export interface ToolRunContext { /** Called by the delegate tool with a finished child run's usage, so the parent run adds it to its totals. */ onDelegatedUsage?: (usage: RunUsage) => void; + /** + * LOU-U9: the model's id for this tool call - unchanged when a call that + * was running when the process died is re-run on resume, so tools can use + * it as an idempotency key. + */ + toolCallId?: string; } /** @@ -71,7 +77,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. + // here are not part of an 'ai' SDK generation, so `messages` is not set + // (LOU-U15). `runContext` carries `toolCallId` (LOU-U9) and + // `onDelegatedUsage` (LOU-V5). const executeOptions = { abortSignal: signal, ...runContext } as ToolExecutionOptions; // LOU-Y1: lets a delegate/`task` tool's sub-agent inherit from this run. bindToolCallScope(executeOptions, scope); diff --git a/src/execution/subagentRuntime.ts b/src/execution/subagentRuntime.ts index cd30ea572..009ba0748 100644 --- a/src/execution/subagentRuntime.ts +++ b/src/execution/subagentRuntime.ts @@ -154,7 +154,7 @@ function pendingForSuspension(suspension: SubagentSuspension): PendingApproval { /** The approval record that pauses a parent run on a suspended sub-agent. */ export function suspensionRecord( run: { agent: AgentConfig; sessionId?: string }, - state: { messages: Message[]; steps: number; usage: RunUsage }, + state: { messages: Message[]; steps: number; usage: RunUsage; queuedInput?: Message[] }, suspension: SubagentSuspension ): { pending: PendingApproval; snapshot: ExecutionSnapshot } { const pending = pendingForSuspension(suspension); @@ -162,7 +162,8 @@ export function suspensionRecord( pending, snapshot: { agent: baseAgentOf(run.agent), - currentMessages: state.messages, + // LOU-U8: queued input rides at the end; resume moves it behind the results. + currentMessages: [...state.messages, ...(state.queuedInput ?? [])], pendingToolCall: pending, steps: state.steps, sessionId: run.sessionId, diff --git a/src/execution/toolCallExecution.ts b/src/execution/toolCallExecution.ts index 40d53ec2c..fdc4cedea 100644 --- a/src/execution/toolCallExecution.ts +++ b/src/execution/toolCallExecution.ts @@ -343,7 +343,8 @@ async function doExecuteToolCall( args, ctx.sandbox, ctx.signal, - { onDelegatedUsage: ctx.onDelegatedUsage }, + // LOU-U9: `toolCallId` is the tool's idempotency key on a re-run. + { onDelegatedUsage: ctx.onDelegatedUsage, toolCallId: toolCall.id }, ctx.scope ); 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 []; +}