From e26afb63a46c374f4f482f4808317336790abb7a Mon Sep 17 00:00:00 2001 From: Mario Zechner Date: Wed, 2 Sep 2026 17:53:25 +0200 Subject: [PATCH] fix(agent): retain settled tools until placement --- packages/agent/docs/harness.md | 29 ++- packages/agent/docs/post-wp05-roadmap.md | 1 + packages/agent/docs/tool-durability.md | 26 +-- .../09-lane-snapshot-settled-tools.md | 93 +++----- packages/agent/src/harness/agent-harness.ts | 24 ++- packages/agent/src/harness/execution/tools.ts | 14 ++ .../agent/src/harness/runtime/drive/tools.ts | 203 ++++++++++-------- packages/agent/src/harness/runtime/lane.ts | 38 +++- packages/agent/src/harness/runtime/reducer.ts | 44 +++- .../test/harness/runtime/drive-tools.test.ts | 119 ++++++++-- .../test/harness/runtime/reducer.test.ts | 90 ++++++++ .../agent/test/harness/runtime/watch.test.ts | 171 +++++++++++++-- packages/agent/test/harness/types.test.ts | 2 + .../src/experimental/client-tui-chat.ts | 11 +- .../src/experimental/mini/tui/view.ts | 11 +- 15 files changed, 643 insertions(+), 233 deletions(-) diff --git a/packages/agent/docs/harness.md b/packages/agent/docs/harness.md index c7c18c747..9cafbfa12 100644 --- a/packages/agent/docs/harness.md +++ b/packages/agent/docs/harness.md @@ -781,13 +781,15 @@ Tool execution separates effect completion from source-ordered tree placement: |---|---|---|---| | call _i_ `planned` | clearance passed (`before_tool`, lookup, arg validation) | `TX[ upsert pi.op.tool_args/O:{stepId}:{i} = effective args, S(call i = effect_pending, replay) ]` | dispatch | | call _i_ `effect_pending` | tool calls `onUpdate(partial, { checkpoint:true })` | `TX[ upsert pi.pending.tool_output/O:{resultEntryId} = partial ]` after invocation fencing; state unchanged | `effect_pending` | -| call _i_ `effect_pending` | effect settled; latest update delivery and latest checkpoint write awaited; `after_tool` applied; `tool_end` emitted | `TX[ upsert pi.pending.entry/{resultEntryId} = finalized result, delete pi.pending.tool_output/O:{resultEntryId}, delete pi.op.tool_memo/O:{resultEntryId}:*, S(call i = outcome_ready, terminate) ]` | `outcome_ready` | -| call _i_ `planned` | unknown tool / invalid args / `before_tool` blocks or throws / control cancelled | stage the complete synthetic result in `pi.pending.entry` and write `outcome_ready` without effect intent | `outcome_ready` | +| call _i_ `effect_pending` | effect settled; latest update delivery and latest checkpoint write awaited; `after_tool` applied | `TX[ upsert pi.pending.entry/{resultEntryId} = finalized result, delete pi.pending.tool_output/O:{resultEntryId}, delete pi.op.tool_memo/O:{resultEntryId}:*, S(call i = outcome_ready, terminate) ]`, with post-commit `tool_end` | `outcome_ready` | +| call _i_ `planned` | unknown tool / invalid args / `before_tool` blocks or throws / control cancelled | `TX[ upsert pi.pending.entry/{resultEntryId} = complete synthetic result, S(call i = outcome_ready, terminate) ]`, with post-commit `tool_start` followed by `tool_end`; no effect intent | `outcome_ready` | | source-ready prefix | first non-completed calls are `outcome_ready` | `TX[ insert result entries in source order, delete their pi.pending.entry values, insert reported usage, upsert pi.branch.tip, S(calls = completed / next checkpoint) ]` | `completed` or checkpoint | **Updates and checkpoints.** Every `onUpdate` is a process-local `tool_update` observation: the synchronous callback emits the event and retains the latest delivery promise internally; tools neither receive nor await it. `checkpoint:true` additionally requests replacement of the invocation's bounded durable progress snapshot: each such call synchronously enqueues one invocation-fenced value replacement on the mutation line, attaches the ordinary harness-fault observer, and replaces only the process-local latest checkpoint-write promise reference. No checkpoint write is dropped or coalesced; Session FIFO preserves request order, and each mutation verifies the same call is still `effect_pending` when it executes. The tool alone controls cadence, duplicate suppression, and bounding — requesting checkpoints faster than storage commits queues memory under the trusted-tool contract, and the API imposes no generic byte cap or truncation. When the tool promise settles, the harness stops accepting updates and closes checkpoint admission; a late request returns without committing. Before `after_tool`, the procedure awaits the latest update-delivery promise **and** the latest checkpoint-write promise — each implies completion of everything earlier in its queue. Checkpoint writes order before outcome staging, and staging deletes the value; a failed checkpoint commit follows the ordinary storage-fault path and prevents staging. -**Staging.** Outcome staging is the point after which the tool can never replay. After `after_tool`, the procedure constructs the complete canonical final result — bounded independently of progress snapshots — emits and awaits `tool_end` for a real execution, then stages its `ToolResultMessage`; the state carries only `terminate` and the reserved id. Tool-reported usage stays inside the staged message until materialization, where its ledger row commits atomically with the entry; added tool names likewise become active from the materialized transcript point, never from invisible staging. +**Staging.** Outcome staging is the point after which the tool can never replay. After `after_tool`, the procedure constructs the complete canonical final result — bounded independently of progress snapshots — and stages its `ToolResultMessage`; the state carries only `terminate` and the reserved id. The staging commit publishes `tool_end` after committed state is installed, so that event is durable evidence that the call is `outcome_ready`. For a fresh synthetic call, the same staging commit publishes `tool_start` followed by `tool_end`; it never crosses the external tool-effect boundary or runs `after_tool`. Tool-reported usage stays inside the staged message until materialization, where its ledger row commits atomically with the entry; added tool names likewise become active from the materialized transcript point, never from invisible staging. + +`tool_start`/`tool_end` bracket public processing of a fresh call and finalized-result availability, not necessarily an external effect. Historical events are not replayed: an unsafe restored `effect_pending` call is represented as running by the initial snapshot and may emit only a recovery-tagged `tool_end` when its interruption result stages. A safe replay's checkpoint-clear commit publishes its recovery-tagged `tool_start`; its staging commit later publishes `tool_end`. After any outcome stages, the procedure materializes the contiguous `outcome_ready` prefix from the first non-completed source position; several results may enter the tree in one transaction, each parented to the previous inserted result. When the final call materializes, the same transaction deletes the addresses from `scanValues(operationToolArgsPrefix(O, stepId))` and chooses: **every** completed call set `terminate: true` → `checkpoint{may_finish, includeFinalAssistant: false}`; otherwise `checkpoint{need_assistant(overflowRecoveryUsed: false)}`. `terminate` lets a tool end the run without another provider turn (a "submit final result" tool in place of structured output); the result record still embeds no message payload. @@ -795,7 +797,7 @@ Modes: **sequential** — clear → intent → execute → finalize → stage Blocked and invalid calls skip intent/execution but still stage a synthetic outcome. A missing tool implementation is the ordinary unknown-tool case: stage an `isError:true` `ToolResultMessage` saying the named tool is unavailable, then continue the batch and later assistant turn; the harness constructs that message directly, omits `details`, and must not invent a value for the tool's typed details contract. A crash before staging reruns ordinary clearance, including `before_tool` under its replay contract; a crash after staging never reruns the hook or tool. -Calls are tracked internally by `sourceIndex` (position in the assistant message's complete content array); hooks and events see provider `toolCallId` and tool name. `AgentHarnessToolInvocation.invocationId` equals the reserved `resultEntryId`, is stable across safe replay, and scopes durable memos under `operationToolMemo(O, invocationId, name)`. Memo names must be non-empty with no `:`; `setMemo(name, undefined)` deletes. Memo operations synchronously enqueue on the mutation line before returning their promises, and tools must await writes; each job verifies the same effect-pending invocation when it executes, so a queued memo write cannot outlive staging. A pre-return write is FIFO-ordered before staging and then deleted by it; a post-return call rejects after capability expiry; no separate write drain exists. Flue-style named effect memoization (`step.do(name, effect)`) awaits these operations: a committed value returns on replay, while a crash before its memo commit may rerun the effect. There is no nested per-step replay state and no exactly-once external-effect promise. +Calls are tracked internally by `sourceIndex` (position in the assistant message's complete content array); hooks and events see provider `toolCallId` and tool name. A provider `toolCallId` is unique only within its tool-call batch and may be reused by a later assistant message. `AgentHarnessToolInvocation.invocationId` equals the reserved session-unique `resultEntryId`, is stable across safe replay, and scopes durable memos under `operationToolMemo(O, invocationId, name)`. Memo names must be non-empty with no `:`; `setMemo(name, undefined)` deletes. Memo operations synchronously enqueue on the mutation line before returning their promises, and tools must await writes; each job verifies the same effect-pending invocation when it executes, so a queued memo write cannot outlive staging. A pre-return write is FIFO-ordered before staging and then deleted by it; a post-return call rejects after capability expiry; no separate write drain exists. Flue-style named effect memoization (`step.do(name, effect)`) awaits these operations: a committed value returns on replay, while a crash before its memo commit may rerun the effect. There is no nested per-step replay state and no exactly-once external-effect promise. ## 3.9 Summary generation — compaction and navigation summaries @@ -1112,8 +1114,12 @@ interface LaneSnapshot { retry?: { attempt: number; maxAttempts: number; nextAttemptAt: number }; deferred?: { handle: DeferredHandle; poll: number }; streamingMessage?: AssistantMessage; - runningTools: Array<{ toolCallId: string; toolName: string; args: unknown; - partialResult?: AgentToolResult }>; + runningTools: Array< + | { status: "running"; toolCallId: string; toolName: string; args: unknown; + result?: AgentToolResult } + | { status: "settled"; toolCallId: string; toolName: string; args: unknown; + result: AgentToolResult; isError: boolean } + >; }; queues: LaneQueuedItem[]; faulted: boolean; @@ -1129,7 +1135,7 @@ interface WatchHandle { `OperationStatus` includes `"running" | "open" | "aborting"`, but current snapshot and reducer paths produce only `"open"` and `"aborting"`; `"running"` has no defined producer and is tracked as contract cleanup (§0.9, roadmap). -`watch(context)` captures one coherent presentation snapshot on the Session mutation line, then exposes events serialized after it. Capture performs one compaction-bounded transcript read, the newest result lookup named by `lastOperationId`, current stats, and exact state-directed reads for inbox payloads, frames, deferred source, and running tool progress. Required missing references fault capture; optional frame/checkpoint absence is legal; results remain unrelated to recovery. `queues` is the one global ordered tagged inbox including pending writes; `configuration`, `stats`, and `faulted` make the initial snapshot self-sufficient before any event arrives. First snapshot, reconnect capture, and `resnapshot()` share one path. +`watch(context)` captures one coherent presentation snapshot on the Session mutation line, then exposes events serialized after it. Capture performs one compaction-bounded transcript read, the newest result lookup named by `lastOperationId`, current stats, and exact state-directed reads for inbox payloads, frames, deferred source, effect-pending tool progress, and outcome-ready staged results. A running tool's optional `result` is its latest complete progress snapshot; a settled tool's required `result` is final and remains in `runningTools` until its own `entry_added` moves presentation to the transcript. Required missing references fault capture; optional frame/checkpoint absence is legal; results remain unrelated to recovery. `queues` is the one global ordered tagged inbox including pending writes; `configuration`, `stats`, and `faulted` make the initial snapshot self-sufficient before any event arrives. First snapshot, reconnect capture, and `resnapshot()` share one path. `reduceLaneSnapshot(snapshot, event)` is the normative client fold: for non-navigation histories, folding a snapshot over its own events produces the next snapshot. It returns `{ rebase: true }` for `navigation_end`; the client calls `handle.resnapshot(context)` without tearing down or resubscribing. Resnapshot marks a barrier on the event-bus delivery tail while the mutation line still holds the captured boundary — queued pre-boundary watcher deliveries are invalidated and post-boundary events held until the fresh snapshot installs — so calling it from inside the listener neither deadlocks nor refolds stale queue/usage state. The reducer ignores other lanes' events, applies session-wide usage totals, and clones its input rather than mutating caller state. @@ -1151,7 +1157,7 @@ Events are passive committed-state/lifecycle observations: they never drive exec `queue_update` carries the complete ordered `LaneQueuedItem[]` after every inbox change and is the sole authoritative queue event; there is no `write_pending`. Lane configuration updates carry `previous` and `value`; global data-bearing configuration updates do the same, while tools/resources remain notification-only because code registries are not replicated. Usage events carry authoritative committed totals from `CommitResult`/storage stats. -Acceptance publishes after its transaction: the start event, message lifecycle plus `entry_added` for placed queued/request entries, then `queue_update` when capture changed the inbox. Standalone structural starts publish before `accept` resolves. Provider/tool live events may precede the transaction persisting their final content; `entry_added` always means the entry is queryable. +Acceptance publishes after its transaction: the start event, message lifecycle plus `entry_added` for placed queued/request entries, then `queue_update` when capture changed the inbox. Standalone structural starts publish before `accept` resolves. Provider streaming and `tool_update` observations may precede the transaction persisting final content; `tool_start` is emitted from the commit that establishes fresh effect intent or synthetic outcome readiness, `tool_end` is emitted only after its finalized result stages, and `entry_added` always means the immutable entry is queryable. `tool_start` carries effective arguments for an intended effect and source arguments for an immediate synthetic result; `tool_end` carries the finalized result but does not repeat arguments. Clients depend on the terminal taxonomy: `run_end` closes a run; `navigation_end` closes navigation and requires snapshot rebase; standalone-compaction `compaction_end` closes compaction; in-run `compaction_start`/`compaction_end` are nested segment brackets that do not clear the run; `run_suspend` keeps the operation open. Every structural start has one matching end, including `aborted`. `compaction_end.status` is `completed | declined | failed | aborted` (success carries `entryId`); `run_end` is `completed | failed | aborted`; `navigation_end` additionally permits `declined`. @@ -1213,9 +1219,9 @@ The request function, not the block, owns registry dispatch, auth, and admission ### Tool phases -`tools.ts` exposes phases at the exact durable boundaries of §3.8 — `prepareToolCall`, `applyBeforeToolDecision`, `executeToolCall`, `finalizeToolCall`, `createToolResultMessage` (shapes in source). Hooks remain separate gated invocations and commits remain explicit operation-procedure statements; neither hides behind a callback bag. The batch procedure composes: prepare (lookup, `prepareArguments`, initial validation) → `before_tool` → apply decision (block or validate replacement arguments) → commit `pi.op.tool_args` + effect-pending intent → execute (effect + live updates + checkpoint requests) → stop updates, expire memo capability, close checkpoint admission → await latest `tool_update` delivery and latest checkpoint write → `after_tool` → finalize → emit and await `tool_end` → commit `pi.pending.entry` + `outcome_ready` + invocation cleanup → materialize source-ready outcomes as entries + usage. +`tools.ts` exposes phases at the exact durable boundaries of §3.8 — `prepareToolCall`, `applyBeforeToolDecision`, `executeToolCall`, `finalizeToolCall`, `createToolResultMessage` (shapes in source). Hooks remain separate gated invocations and commits remain explicit operation-procedure statements; neither hides behind a callback bag. The batch procedure composes: prepare (lookup, `prepareArguments`, initial validation) → `before_tool` → apply decision (block or validate replacement arguments) → commit `pi.op.tool_args` + effect-pending intent with post-commit `tool_start` → execute (effect + live updates + checkpoint requests) → stop updates, expire memo capability, close checkpoint admission → await latest `tool_update` delivery and latest checkpoint write → `after_tool` → finalize → commit `pi.pending.entry` + `outcome_ready` + invocation cleanup with post-commit `tool_end` → materialize source-ready outcomes as entries + usage. -Unknown tools, `prepareArguments` failures, invalid initial/replacement arguments, and blocked calls produce an immediate raw error `AgentToolResult` with `isError: true` and no invented `details`; `createToolResultMessage` constructs the canonical synthetic message before staging `outcome_ready`. The old inline declarations instead put a `ToolResultMessage` directly in the immediate outcome. `AgentHarnessTool.prepareArguments` is deterministic/idempotent computation and may repeat before intent; effectful policy belongs in `before_tool`. At `tool.execute` admission, `executeToolCall(call, gate, onUpdate, toolContext, invocation, context)` derives `withAbortSignal(gate.signal, context)` and invokes `AgentHarnessTool.execute` directly through `gate.admit(...)`, with the admitted Context trailing; there is no neutral `AgentTool` adapter. The old four-argument declaration and adapter description predate this source shape. The block converts expected tool throws to an error result and stops accepting updates when the tool promise settles; the declared raw tool-effect span is not emitted until T1 (§5.8); update/checkpoint promise retention and the await-both rule follow §3.8. `finalizeToolCall` applies the field-by-field patch before `tool_end` and staging. +Unknown tools, `prepareArguments` failures, invalid initial/replacement arguments, and blocked calls produce an immediate raw error `AgentToolResult` with `isError: true` and no invented `details`; `createToolResultMessage` constructs the canonical synthetic message before staging `outcome_ready`. Their outcome-staging commit emits `tool_start` followed by `tool_end`; they still invoke no tool effect or `after_tool`. The old inline declarations instead put a `ToolResultMessage` directly in the immediate outcome. `AgentHarnessTool.prepareArguments` is deterministic/idempotent computation and may repeat before intent; effectful policy belongs in `before_tool`. At `tool.execute` admission, `executeToolCall(call, gate, onUpdate, toolContext, invocation, context)` derives `withAbortSignal(gate.signal, context)` and invokes `AgentHarnessTool.execute` directly through `gate.admit(...)`, with the admitted Context trailing; there is no neutral `AgentTool` adapter. The old four-argument declaration and adapter description predate this source shape. The block converts expected tool throws to an error result and stops accepting updates when the tool promise settles; the declared raw tool-effect span is not emitted until T1 (§5.8); update/checkpoint promise retention and the await-both rule follow §3.8. `finalizeToolCall` applies the field-by-field patch before outcome staging and post-commit `tool_end`. Before starting any call in a live batch, the procedure resolves `toolContext` once and filters the current `AgentHarnessTool` registry to the complete captured active-name set, retaining that procedure-local snapshot. The `executeToolCall` call site supplies each call's stable invocation (`invocationId: resultEntryId`, `operationId`, `turnId`, memos), update callback, tool context, and current invocation Context. An absent implementation — or a provider call outside the captured active names — becomes the §3.8 synthetic unknown-tool result and does not suspend the batch. Every call observes the same application context and its own stable invocation identity. Safe replay creates a new code/context snapshot but passes the same invocation id and memos after deleting the stale progress checkpoint. `AgentHarnessTool.replay` defaults to `"never"`. @@ -1290,6 +1296,7 @@ Workflow: keep a future package's row here until actionable; move exact files/te | WP06 | complete | Separated Session, Branch, AgentLane, AgentHarness; one keyless Session mutation line. | [Session, Branch, Lane separation](work-packages/06-session-branch-lane-separation.md) | | WP07 | complete | Removed SQLite storage-layer ownership; added live read-only source forks, no-create opens, deletion reservation, physical/path safety, and all-settled close. | [SQLite host ownership and live forks](work-packages/07-sqlite-host-ownership-live-forks.md) | | WP08 | in progress — Slice A | Replace implicit-main forks with named-branch/tree semantics and bounded-memory backend copies. | [Named-branch and tree forks with streaming copies](work-packages/08-named-branch-streaming-forks.md) | +| WP09 | complete | Project effect-pending and settled-but-unplaced tool calls continuously through snapshots and lifecycle events until transcript placement. | [LaneSnapshot settled-but-unplaced tools](work-packages/09-lane-snapshot-settled-tools.md) | WP05 subsumed the former R2–R12 execution rows; their implemented contract is in Parts 0–5 and the completed handoff. @@ -1389,7 +1396,7 @@ Each durable mutation race has exactly two durable histories. Matching callers i For each recovery prefix: close, reopen, drive, and compare against uninterrupted recovery — invoking recovery twice from the initial prefix is **not** sufficient. Every operation kind also covers accept → close before first drive → reopen → drive. At every test-controlled committed lane boundary, compare the published `Lane.state` with a fresh `restoreLaneState` result; divergence is an implementation defect, never silently healed by the next transition. One corruption assertion constructs an `aborted` response with running control directly and requires the consuming transition to reject it as an invariant defect; provider conformance separately proves implementations emit `aborted` only for the supplied signal. -**Tier B — writer conformance.** Run the public harness against the instrumented-storage decorator (a spy wrapping `Storage.commit()` recording every transaction's writes in order); assert exact write order and content against the Part 3 transaction tables and §5.5 ordering, with faux provider/tool/hook spies interleaving starts/events with commits. It catches: effects before intent; missing awaits of latest update delivery or checkpoint write before `after_tool`; per-frame storage awaits in the provider loop; frame appends out of provider-event order or persisted for `done`/`error`; settlements missing their frame-list delete; `tool_end` after rather than before staging; missing response/usage settlement; checkpoint or frame writes after their child state settled; outcomes not staged before replay becomes impossible; out-of-order tree placement; late result-id reservation; memos or staged/checkpoint/frame values leaked by outcome/terminal cleanup. +**Tier B — writer conformance.** Run the public harness against the instrumented-storage decorator (a spy wrapping `Storage.commit()` recording every transaction's writes in order); assert exact write order and content against the Part 3 transaction tables and §5.5 ordering, with faux provider/tool/hook spies interleaving starts/events with commits. It catches: effects before intent; missing awaits of latest update delivery or checkpoint write before `after_tool`; per-frame storage awaits in the provider loop; frame appends out of provider-event order or persisted for `done`/`error`; settlements missing their frame-list delete; `tool_end` before rather than after staging; missing response/usage settlement; checkpoint or frame writes after their child state settled; outcomes not staged before replay becomes impossible; out-of-order tree placement; late result-id reservation; memos or staged/checkpoint/frame values leaked by outcome/terminal cleanup. **Tier C — deterministic interleavings.** Every race in §9.2, in both orders, with test-only gated commits and controlled hooks, providers, tools, and timers. diff --git a/packages/agent/docs/post-wp05-roadmap.md b/packages/agent/docs/post-wp05-roadmap.md index 9ccd7df5e..1c8f6fc9d 100644 --- a/packages/agent/docs/post-wp05-roadmap.md +++ b/packages/agent/docs/post-wp05-roadmap.md @@ -201,6 +201,7 @@ Completed by this audit: Remaining cleanup: +- **Harness-owned DTO boundary:** the durable Harness still imports legacy agent-loop DTOs from `src/types.ts`: `AgentMessage`, `AgentToolResult`, `AgentToolCall`, `AgentTool`, `QueueMode`, and `ThinkingLevel`. Make a later one-shot cut to independent `HarnessMessage`, `CustomHarnessMessages`, `HarnessToolResult`, `HarnessToolCall`, `HarnessQueueMode`, and `HarnessThinkingLevel` definitions; keep `AgentHarnessTool` as the executable Harness configuration type. Update Harness internals, root exports, declaration-merging tests, documentation, and experimental coding-agent consumers together. Do not alias the new message/tool DTOs back to legacy types, because that preserves the coupling this work is intended to remove. - Preserve shared-container support; do not remove it incidentally. - Remove or use unused `insertEntryRow()` and `insertUsageLedgerRow()`. - Consolidate duplicated SQLite branch payload/structure scan plumbing only after correctness tests pin both paths. diff --git a/packages/agent/docs/tool-durability.md b/packages/agent/docs/tool-durability.md index ec32a6320..f4a1e1f38 100644 --- a/packages/agent/docs/tool-durability.md +++ b/packages/agent/docs/tool-durability.md @@ -57,7 +57,7 @@ Each call already reserves its result entry ID before execution. Use that as the invocationId = resultEntryId ``` -It is session-unique, survives safe replay, and is distinct from the provider's `toolCallId`. +It is session-unique, survives safe replay, and is distinct from the provider's batch-local `toolCallId`, which a later assistant message may reuse. The surrounding operation state continues to provide: @@ -233,6 +233,7 @@ planned set pi.op.tool_args, set call = effect_pending(replay) ] +→ post-commit tool_start → admit tool execution ``` @@ -260,8 +261,8 @@ When the tool promise settles: 2. await the latest tracked `tool_update` delivery and latest checkpoint-write promise; each implies completion of its preceding queue; 3. run `after_tool` when this is a real fresh or safely replayed result and cancellation did not prevent the hook; 4. construct the complete final `ToolResultMessage`; -5. emit and await `tool_end` for a real execution; -6. commit the result as `outcome_ready`. +5. commit the result as `outcome_ready`; +6. emit and await `tool_end` from the committed staging transition. `setMemo()` returns a promise and tools must await it; `step.do` always does. An unawaited pre-return mutation is still enqueued before staging and is deleted by staging. Calls begun after capability expiry reject. No separate invocation-write drain exists. @@ -280,7 +281,7 @@ TX[ ] ``` -The transaction is the linearization point after which the invocation can never replay. +The transaction is the linearization point after which the invocation can never replay. Its post-commit `tool_end` is therefore durable evidence that the finalized outcome is ready; it is no longer a pre-commit effect observation. The staged message contains the final: @@ -566,22 +567,23 @@ A reconnecting client may see: - live process-local progress newer than the latest durable checkpoint before disconnect; - after process replacement, only the latest committed bounded checkpoint; -- `outcome_ready` calls as no longer running but not yet visible in the transcript until source-ordered materialization; +- `outcome_ready` calls as settled rows in `runningTools` until source-ordered materialization; - completed calls in the transcript. -In `LaneSnapshot`, an effect-pending tool's `partialResult` prefers its latest process-local update and falls back to the durable checkpoint after reopen. An `outcome_ready` call is settled and does not appear in `runningTools`. +`LaneSnapshot.operation.runningTools` is a discriminated union. An effect-pending tool has `status: "running"` and an optional `result` containing the latest complete progress snapshot, falling back to the durable checkpoint after reopen. An outcome-ready call has `status: "settled"`, its required complete final `result`, and `isError`; it remains there until its immutable result entry's `entry_added` removes the row and places the same presentation in the transcript. Planned and completed calls are omitted. ## Events and hooks -- `tool_start` remains a live effect event. +- `tool_start` begins public processing presentation for a fresh call; it is emitted from the commit that establishes effect intent or a synthetic staged outcome and does not by itself prove an external effect started. It carries effective arguments for an intended effect and source arguments for an immediate synthetic result. - live progress events and durable progress checkpoints do not prove completion. - The harness awaits the latest `tool_update` delivery before `after_tool`, preserving the existing listener ordering without making `onUpdate` async. -- `tool_end` reports a finalized real result before its `outcome_ready` staging commit, in completion order. It is observation, not proof of durability. -- synthetic blocked and unsafe-recovery outcomes emit no tool-effect lifecycle and run no post-effect tool hook. -- message lifecycle and `entry_added` occur when the staged result materializes, not when it first becomes `outcome_ready`. +- `tool_end` carries the complete finalized result after its `outcome_ready` staging commit, in completion order. It is durable settlement evidence and does not repeat the arguments from `tool_start`. +- for a fresh blocked, invalid, truncated, or planned-cancellation synthetic outcome, the staging commit emits `tool_start` followed by `tool_end`; these paths still run no tool effect or post-effect hook. Cancellation after effect intent uses the earlier intent-bound start and a staging-bound end. +- an unsafe restored effect is already represented as running by the initial snapshot and may emit only a recovery-tagged `tool_end` when interruption synthesis stages. +- message lifecycle and `entry_added` occur when the staged result materializes, not when it first becomes `outcome_ready`; `entry_added` removes only that settled row. - passive listeners cannot mutate invocation state reentrantly. -Instrumented-storage tests assert this ordering. A crash after `tool_end` but before outcome staging may recover or safely replay the still-uncertain call; only staging prevents replay. Historical events are not replayed, though a safely replayed execution emits its own recovery-tagged lifecycle. +Instrumented-storage tests assert `intent commit → tool_start → tool_update* → outcome staging → tool_end → source-ordered placement` for execution and `outcome staging → tool_start → tool_end → source-ordered placement` for fresh synthetic results. Historical events are not replayed; a safely replayed execution emits recovery `tool_start` from its checkpoint-clear commit and `tool_end` from outcome staging. ## Races @@ -670,7 +672,7 @@ Instrumented-storage tests assert this ordering. A crash after `tool_end` but be ### Atomicity and instrumentation -- exact intent, synchronous update acceptance, asynchronous update-delivery, `after_tool`, `tool_end`, outcome-ready, source-ordered message lifecycle, and materialization order; +- exact intent, synchronous update acceptance, asynchronous update-delivery, `after_tool`, outcome-ready staging, post-commit `tool_end`, source-ordered message lifecycle, and materialization order; - outcome staging is atomic with memo/output cleanup; - materialization is atomic with pending deletion, usage, tip, and state; - crash at every boundary; diff --git a/packages/agent/docs/work-packages/09-lane-snapshot-settled-tools.md b/packages/agent/docs/work-packages/09-lane-snapshot-settled-tools.md index 7e54f6f6d..c9106b2f9 100644 --- a/packages/agent/docs/work-packages/09-lane-snapshot-settled-tools.md +++ b/packages/agent/docs/work-packages/09-lane-snapshot-settled-tools.md @@ -6,7 +6,7 @@ - Branch at handoff creation: `dev` - Baseline commit: `d14d6b22327d545d6a253f932165b63e48d7f9c8` - The user reported the worktree clean immediately before this handoff. -- This document is a handoff for implementation after conversation compaction. It is not itself the normative harness specification. `packages/agent/docs/harness.md` remains normative and must be updated as part of the work. +- This document began as an implementation handoff after conversation compaction and now records the implemented design. It is not itself the normative harness specification; `packages/agent/docs/harness.md` remains normative. ## Goal @@ -23,14 +23,14 @@ completed → transcript toolResult entry For each call after it becomes presentation-active, `runningTools` and placed transcript entries must not have a gap or overlap. Placement is a source-prefix flush, not an all-tools barrier. Remove a call from `runningTools` on its own `entry_added`, never on `turn_end`. -## Current bug +## Original bug -A call disappears between real-effect completion and source-ordered tree placement: +A call disappeared between real-effect completion and source-ordered tree placement: -1. `packages/agent/src/harness/runtime/reducer.ts`: `tool_end` splices the call out of `runningTools`. -2. `packages/agent/src/harness/runtime/lane.ts`: `captureLaneSnapshot()`, `case "tools"`, projects only `effect_pending` calls and skips `outcome_ready` calls. -3. The finalized result is already durable at `pendingEntry(resultEntryId)`, but a fresh/reconnected snapshot cannot display it. -4. It reappears only after `entry_added` places the immutable `toolResult` entry. +1. `packages/agent/src/harness/runtime/reducer.ts`: `tool_end` spliced the call out of `runningTools`. +2. `packages/agent/src/harness/runtime/lane.ts`: `captureLaneSnapshot()`, `case "tools"`, projected only `effect_pending` calls and skipped `outcome_ready` calls. +3. The finalized result was already durable at `pendingEntry(resultEntryId)`, but a fresh/reconnected snapshot could not display it. +4. It reappeared only after `entry_added` placed the immutable `toolResult` entry. For a parallel batch `[A, B, C]`, if B settles while A remains pending, B may stay `outcome_ready` until A is ready. If A is already placed, B can place without waiting for C. Therefore clearing everything at `turn_end` is wrong: early-placed results would temporarily exist in both `transcript` and `runningTools`. @@ -43,24 +43,23 @@ Mini does **not** replicate structural object deltas. - Reconnect/rebase fetches another complete snapshot. - `tool_update` currently carries a complete replacement progress result, not a nested diff. -Current durable tool flow in `packages/agent/src/harness/runtime/drive/tools.ts`: +Implemented durable tool flow in `packages/agent/src/harness/runtime/drive/tools.ts`: ```text prepare → before_tool -→ intent commit (effect_pending + effective args) +→ intent commit (effect_pending + effective args), then tool_start → execute/update/checkpoint → after_tool → finalize -→ emit tool_end CURRENT: before staging, real effects only -→ publishToolOutcome staging commit (pendingEntry + outcome_ready) +→ publishToolOutcome staging commit (pendingEntry + outcome_ready), then tool_end → materializeReady prefix placement → entry_added ``` -All real, immediate synthetic, cancellation, and recovery outcomes eventually converge through `publishToolOutcome()`. +All real, immediate synthetic, cancellation, and recovery outcomes converge through `publishToolOutcome()`. -`Lane.settleOperation()` supports commit-bound events. It commits, publishes process-local state, constructs the event batch, and the public operation awaits delivery. In parallel execution, `materializeReady()` is scheduled only after the outcome-completion promise resolves. This permits `tool_end` to be moved into the outcome staging transition: it will be delivered after staging and before placement. +`Lane.settleOperation()` supports commit-bound events. It commits, publishes process-local state, constructs the event batch, and the public operation awaits delivery. In parallel execution, `materializeReady()` is scheduled only after the outcome-completion promise resolves. Consequently, `tool_end` is delivered after staging and before placement. ## Agreed event contract change @@ -84,13 +83,15 @@ intent commit ### Fresh synthetic call ```text -tool_start -→ TX[pendingEntry + outcome_ready] +TX[pendingEntry + outcome_ready] +→ tool_start → tool_end → source-ordered placement → entry_added ``` +The staging transaction's post-commit event batch contains `tool_start` followed by `tool_end`, so a watcher cannot observe either lifecycle event without the authoritative staged state. + Fresh synthetic calls include: - unknown tool; @@ -104,7 +105,7 @@ Fresh synthetic calls include: Historical lifecycle events are not replayed. - A restored `effect_pending` call is already represented by the initial snapshot. -- Safe replay may emit a recovery-tagged `tool_start`/`tool_end` around the replayed effect. +- Safe replay emits recovery-tagged `tool_start` from the checkpoint-clear commit and `tool_end` from the later outcome-staging commit. - Unsafe interruption synthesis may emit a recovery-tagged `tool_end` without a newly emitted `tool_start`; the initial snapshot supplied the running row. - A call already restored as `outcome_ready` appears as settled in the initial snapshot and needs no replayed end event before placement. @@ -173,32 +174,16 @@ Upsert is required because a watch may capture durable `effect_pending` state be ### `tool_end` - Resolve with `matchingOperation(snapshot, event.runId)`. -- Upsert/replace the matching row as: - -```ts -{ - status: "settled", - toolCallId, - toolName, - args, - result: event.result, - isError: event.isError, -} -``` - +- Find the existing row by batch-local `toolCallId`; only one tool batch is presentation-active at a time. +- Replace it with `status: "settled"`, preserving its arguments and using `event.result` and `event.isError`. - This naturally removes the provisional interpretation of the old `result`; there is no separate `partialResult` to delete. - The finalized result remains displayed until placement. -A robust end upsert requires `args` on `tool_end`. Add `args: unknown` to the public event shape and populate it. This handles the race where a watcher registers after synthetic `tool_start` but before staging: its durable capture sees `planned`, misses the historical start, then receives post-commit `tool_end`. +`tool_end` does not carry arguments and cannot create a row. Fresh synthetic `tool_start` and `tool_end` are emitted together after the staging commit, eliminating the old capture/event gap. Unsafe recovery relies on the initial snapshot's running row. ### `entry_added` -Before the existing transcript dedupe guard: - -- If `event.entry` is a message whose role is `toolResult`, remove the matching `toolCallId` from `snapshot.operation?.runningTools`. -- Then run the existing duplicate-entry guard and transcript logic. - -Removal before dedupe lets a duplicate/replayed placement event heal a stale row without duplicating transcript content. +If `event.entry` is a message whose role is `toolResult`, remove the matching batch-local `toolCallId` from `snapshot.operation?.runningTools`, then apply the transcript update. Harness events are serialized, trusted, emitted exactly once, and not historically replayed, so neither duplicate-entry handling nor cross-batch identity is needed. Do not clear tool rows on `turn_end`. @@ -271,36 +256,27 @@ Extend/refactor it so post-commit event production has the complete canonical fi Synthetic helpers currently return `ToolResultMessage` directly. Refactor carefully so synthetic outcomes also carry the canonical result data. Do not invent `details` in the transcript: existing unknown/invalid synthetic results deliberately omit message details. -### Universal `tool_start` +### Commit-bound `tool_start` -Fresh synthetic paths must emit `tool_start` before staging. +For fresh execution, `publishToolIntent()` attaches `tool_start` to the commit that persists effective arguments and changes the call to `effect_pending`. The public operation awaits delivery before admitting `executeToolCall()`, preserving `tool_start → tool_update*` without requiring each update callback to await delivery. -A useful implementation shape is a shared `emitToolStart(...)` helper used from: +For a fresh synthetic call that never writes effect intent, `publishToolOutcome()` attaches `tool_start` before `tool_end` in the outcome-staging commit's event batch. It reports the source block arguments. -- real execution after effective argument intent is durable and before effect updates/end; -- immediate synthetic outcome paths, using source `block.arguments` when no effective arguments exist; -- planned cancellation synthesis. - -For actual execution, start must be queued before `executeToolCall()` can synchronously fail or publish an update. Event-bus serialization must preserve `tool_start → tool_update*` without requiring each update callback to await delivery. - -For executable calls, report persisted/effective validated arguments. For synthetic calls that never wrote effective arguments, report the source block arguments. - -Do not emit a fresh start merely to replay historical lifecycle for an already-restored unsafe `effect_pending` call; its initial snapshot is the baseline. +For safe recovery, the checkpoint-clear commit emits recovery-tagged `tool_start` using persisted effective arguments. Do not emit a fresh start for an already-restored unsafe `effect_pending` call; its initial snapshot is the baseline. ### Post-commit `tool_end` Remove the current pre-staging `tool_end` emission from `performToolInvocation()`. -Make `publishToolOutcome()` attach `tool_end` to the same staging command's `events` callback. The event must carry: +`publishToolOutcome()` attaches `tool_end` to the same staging command's `events` callback. The event carries: - `runId`, `turnId`, `toolCallId`, `toolName`; -- `args` (persisted effective args when present, otherwise source arguments); - canonical final `result`; - `isError`; - cancellation-normalized durable `terminate`; - `recovery: true` where applicable. -The staging planner has a `SessionReader`. It can read persisted `operationToolArgs`; pass source arguments as the legal fallback for planned synthetic outcomes. Event data must describe the state actually committed, especially cancellation forcing `terminate: false`. +Arguments belong to `tool_start` and are not repeated on `tool_end`. Event data describes the state actually committed, especially cancellation forcing `terminate: false`. Because `Lane.command()` awaits retained event delivery and `runParallel()` schedules materialization from the outcome-completion promise, the required order is: @@ -314,7 +290,7 @@ staging commit ### Call sites to audit -Every `publishToolOutcome()` call must supply the source-argument fallback and recovery context correctly: +Every `publishToolOutcome()` call must supply the source tool call and recovery context correctly: - immediate outcome in `startToolInvocation()`; - cancellation after intent but before execution; @@ -377,8 +353,7 @@ Add focused coverage for: - `tool_update` from a stale/wrong `runId` does not mutate the current operation; - `tool_start` upserts rather than duplicates a row captured from durable intent; -- `tool_end` can upsert a missing row using its `args` payload; -- duplicate `entry_added` removes a stale matching row before transcript dedupe. +- `tool_end` settles the existing batch-local row. ### Capture/watch tests @@ -401,15 +376,15 @@ File: `packages/agent/test/harness/runtime/drive-tools.test.ts` Update/add ordering assertions proving: -- real: `tool_start < tool_update* < staging commit < tool_end < entry_added`; -- immediate synthetic: `tool_start < staging commit < tool_end < entry_added`, no tool effect and no `after_tool`; +- real: `intent commit < tool_start < tool_update* < staging commit < tool_end < entry_added`; +- immediate synthetic: `staging commit < tool_start < tool_end < entry_added`, no tool effect and no `after_tool`; - planned cancellation gets coherent start/end; - unsafe recovery uses initial snapshot plus recovery-tagged end without replaying effects; - B can emit post-commit end and remain settled while A blocks placement; - source-order placement remains unchanged; - a crash after staging cannot replay the call. -The current normative tests/documentation explicitly require `tool_end` before staging. Reverse those expectations deliberately. +The baseline normative tests/documentation required `tool_end` before staging; the implementation reverses those expectations deliberately. ### Type/event catalog tests @@ -418,7 +393,7 @@ Audit: - `packages/agent/test/harness/types.test.ts` - `packages/agent/src/harness/telemetry.ts` -No new event name is planned, but `tool_end` gains `args` and semantics change. +No new event name is added. `tool_end` omits arguments and its semantics change. ### Mini regression @@ -439,7 +414,7 @@ Read both documents completely before editing: - `packages/agent/docs/harness.md` (normative) - `packages/agent/docs/tool-durability.md` -Update all statements that currently require: +The implementation updates baseline statements that required: - `tool_end` before staging; - `tool_start`/`tool_end` only for real effects; diff --git a/packages/agent/src/harness/agent-harness.ts b/packages/agent/src/harness/agent-harness.ts index 0b09eb860..55b853e13 100644 --- a/packages/agent/src/harness/agent-harness.ts +++ b/packages/agent/src/harness/agent-harness.ts @@ -191,6 +191,23 @@ export interface LaneInfo { operation: CurrentOperationInfo | null; } +export type LaneSnapshotTool = + | { + status: "running"; + toolCallId: string; + toolName: string; + args: unknown; + result?: AgentToolResult; + } + | { + status: "settled"; + toolCallId: string; + toolName: string; + args: unknown; + result: AgentToolResult; + isError: boolean; + }; + export interface OpenOperation { lane: string; operationId: string; @@ -224,12 +241,7 @@ export interface LaneSnapshot { retry?: { attempt: number; maxAttempts: number; nextAttemptAt: number }; deferred?: { handle: DeferredHandle; poll: number }; streamingMessage?: AssistantMessage; - runningTools: { - toolCallId: string; - toolName: string; - args: unknown; - partialResult?: AgentToolResult; - }[]; + runningTools: LaneSnapshotTool[]; }; queues: LaneQueuedItem[]; faulted: boolean; diff --git a/packages/agent/src/harness/execution/tools.ts b/packages/agent/src/harness/execution/tools.ts index 57371c1be..09c383f57 100644 --- a/packages/agent/src/harness/execution/tools.ts +++ b/packages/agent/src/harness/execution/tools.ts @@ -180,6 +180,20 @@ export function finalizeToolCall( }; } +/** Reconstruct the canonical tool result represented by a staged transcript message. */ +export function toolResultFromMessage( + message: ToolResultMessage, + terminate: boolean, +): AgentToolResult { + return { + content: message.content, + details: message.details, + ...(message.usage === undefined ? {} : { usage: message.usage }), + ...(message.addedToolNames === undefined ? {} : { addedToolNames: message.addedToolNames }), + ...(terminate ? { terminate: true } : {}), + }; +} + /** Convert finalized tool output to the provider-facing transcript message. */ export function createToolResultMessage(call: FinalizedToolCall): ToolResultMessage { return { diff --git a/packages/agent/src/harness/runtime/drive/tools.ts b/packages/agent/src/harness/runtime/drive/tools.ts index f55f1ca02..ccf16d3c9 100644 --- a/packages/agent/src/harness/runtime/drive/tools.ts +++ b/packages/agent/src/harness/runtime/drive/tools.ts @@ -10,6 +10,7 @@ import { type FinalizedToolCall, finalizeToolCall, prepareToolCall, + toolResultFromMessage, } from "../../execution/tools.ts"; import { SessionInvariantError } from "../../session/session.ts"; import type { JsonValue, ToolBatch, ToolCall, ToolsOperation } from "../../session/types.ts"; @@ -35,7 +36,7 @@ import { } from "./tool-placement.ts"; type ToolCallTask = { completion: Promise }; -type ToolOutcome = { message: ToolResultMessage; terminate: boolean }; +type ToolOutcome = { toolCall: AgentToolCall; message: ToolResultMessage; terminate: boolean }; type PreparedToolInvocation = | { kind: "ready"; cleared: ClearedToolCall } | { kind: "outcome"; outcome: ToolOutcome }; @@ -146,32 +147,41 @@ function syntheticMessage( }; } -function abortedMessage(toolCall: AgentToolCall): ToolResultMessage { - return syntheticMessage(toolCall, [{ type: "text", text: "Tool execution was cancelled before completion." }]); -} - -function interruptedMessage( - toolCall: AgentToolCall, - checkpoint: AgentToolResult | undefined, -): ToolResultMessage { - return syntheticMessage( +function abortedOutcome(toolCall: AgentToolCall): ToolOutcome { + return { toolCall, - [...(checkpoint?.content ?? []), { type: "text", text: INTERRUPTION_MARKER }], - checkpoint === undefined ? {} : { details: checkpoint.details, usage: checkpoint.usage }, - ); + message: syntheticMessage(toolCall, [{ type: "text", text: "Tool execution was cancelled before completion." }]), + terminate: false, + }; } -function truncatedMessage(toolCall: AgentToolCall): ToolResultMessage { - return syntheticMessage(toolCall, [ - { - type: "text", - text: `Tool call ${JSON.stringify(toolCall.name)} was not executed because the assistant response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments.`, - }, - ]); +function interruptedOutcome(toolCall: AgentToolCall, checkpoint: AgentToolResult | undefined): ToolOutcome { + return { + toolCall, + message: syntheticMessage( + toolCall, + [...(checkpoint?.content ?? []), { type: "text", text: INTERRUPTION_MARKER }], + checkpoint === undefined ? {} : { details: checkpoint.details, usage: checkpoint.usage }, + ), + terminate: false, + }; +} + +function truncatedOutcome(toolCall: AgentToolCall): ToolOutcome { + return { + toolCall, + message: syntheticMessage(toolCall, [ + { + type: "text", + text: `Tool call ${JSON.stringify(toolCall.name)} was not executed because the assistant response hit the output token limit, so its arguments may be truncated. Re-issue the tool call with complete arguments.`, + }, + ]), + terminate: false, + }; } function outcomeFromFinalizedCall(finalized: FinalizedToolCall): ToolOutcome { - return { message: createToolResultMessage(finalized), terminate: finalized.terminate }; + return { toolCall: finalized.toolCall, message: createToolResultMessage(finalized), terminate: finalized.terminate }; } async function publishToolIntent( @@ -179,8 +189,10 @@ async function publishToolIntent( drive: Drive, run: ToolsOperation, planned: Extract, + toolCall: AgentToolCall, args: Record, replay: "never" | "safe", + recovery: boolean, ): Promise>> { return lane.continueOperation( run, @@ -196,6 +208,18 @@ async function publishToolIntent( writes: [setValue(operationToolArgs(drive.operationId, run.batch.turnId, planned.sourceIndex), args)], operationState: withToolBatch(run, replaceCall(run.batch, effectPending)), materialize: () => effectPending, + events: () => [ + { + type: "tool_start", + lane: lane.name, + runId: drive.operationId, + turnId: run.batch.turnId, + toolCallId: toolCall.id, + toolName: toolCall.name, + args, + ...(recovery ? { recovery: true as const } : {}), + }, + ], }; }, drive.context, @@ -206,32 +230,62 @@ async function publishToolOutcome( lane: Lane, drive: Drive, capability: ToolsOperation, - call: ToolCall, - message: ToolResultMessage, - terminate: boolean, + call: Extract, + finalized: ToolOutcome, + recovery: boolean, ): Promise { await lane.settleOperation( capability, async (_state, run, _meta, reader) => { + const { toolCall } = finalized; const memos = await reader.scanValues( operationToolMemoPrefix(drive.operationId, call.resultEntryId), drive.context, ); + const durableTerminate = run.control.status === "running" && finalized.terminate; const outcome: Extract = { status: "outcome_ready", sourceIndex: call.sourceIndex, resultEntryId: call.resultEntryId, - terminate: run.control.status === "running" && terminate, + terminate: durableTerminate, }; return { kind: "commit", writes: [ - setValue(pendingEntry(call.resultEntryId), { type: "message", payload: message }), + setValue(pendingEntry(call.resultEntryId), { type: "message", payload: finalized.message }), deleteValue(pendingToolOutput(drive.operationId, call.resultEntryId)), ...memos.map(({ address }) => deleteValue(address)), ], operationState: withToolBatch(run, replaceCall(run.batch, outcome)), materialize: () => undefined, + events: () => [ + ...(call.status === "planned" + ? [ + { + type: "tool_start" as const, + lane: lane.name, + runId: drive.operationId, + turnId: run.batch.turnId, + toolCallId: toolCall.id, + toolName: toolCall.name, + args: toolCall.arguments, + ...(recovery ? { recovery: true as const } : {}), + }, + ] + : []), + { + type: "tool_end", + lane: lane.name, + runId: drive.operationId, + turnId: run.batch.turnId, + toolCallId: toolCall.id, + toolName: toolCall.name, + result: toolResultFromMessage(finalized.message, durableTerminate), + isError: finalized.message.isError, + terminate: durableTerminate, + ...(recovery ? { recovery: true as const } : {}), + }, + ], }; }, drive.context, @@ -243,6 +297,7 @@ async function clearReplayCheckpoint( drive: Drive, batch: ToolBatch, call: Extract, + toolCall: AgentToolCall, ): Promise> { return lane.command(async (state, reader) => { const stored = await reader.getValue( @@ -257,6 +312,18 @@ async function clearReplayCheckpoint( writes: [deleteValue(pendingToolOutput(drive.operationId, call.resultEntryId))], next: state, materialize: () => stored.value, + events: () => [ + { + type: "tool_start", + lane: lane.name, + runId: drive.operationId, + turnId: batch.turnId, + toolCallId: toolCall.id, + toolName: toolCall.name, + args: stored.value, + recovery: true, + }, + ], }; }, drive.context); } @@ -292,9 +359,6 @@ async function performToolInvocation( const capability = invocationCapability(lane, drive, batch, call); const progress = openToolProgress(lane, drive, batch.turnId, call.sourceIndex, call.resultEntryId); let latestUpdateDelivery: Promise = Promise.resolve(); - let startDelivered: Promise = Promise.resolve(); - let startQueued = false; - const bufferedUpdates: AgentToolResult[] = []; const publishUpdate = (partial: AgentToolResult): void => { latestUpdateDelivery = lane.emitBatch( [ @@ -320,48 +384,26 @@ async function performToolInvocation( cleared, drive.gate, (partial, options) => { - if (startQueued) publishUpdate(partial); - else bufferedUpdates.push(partial); + publishUpdate(partial); if (options?.checkpoint === true) progress.write(partial); }, toolContext, capability.invocation, drive.context, ); - startDelivered = lane.emitBatch( - [ - { - type: "tool_start", - lane: lane.name, - runId: drive.operationId, - turnId: batch.turnId, - toolCallId: cleared.toolCall.id, - toolName: cleared.toolCall.name, - args: cleared.args, - ...(recovery ? { recovery: true as const } : {}), - }, - ], - drive.context, - ); - startQueued = true; - for (const update of bufferedUpdates) publishUpdate(update); } catch (error) { capability.expire(); progress.seal(); await progress.drain(); if (!(error instanceof AbortRequested)) throw error; await error.cancellation; - return { - message: recovery ? interruptedMessage(cleared.toolCall, undefined) : abortedMessage(cleared.toolCall), - terminate: false, - }; + return recovery ? interruptedOutcome(cleared.toolCall, undefined) : abortedOutcome(cleared.toolCall); } const executed = await execution.finally(() => { capability.expire(); progress.seal(); }); - await startDelivered; await latestUpdateDelivery; await progress.drain(); @@ -387,26 +429,8 @@ async function performToolInvocation( if (!(error instanceof AbortRequested)) throw error; patch = undefined; } - const cancelled = lane.state.operation!.state.control.status === "cancel_requested"; - const finalized = finalizeToolCall(cleared, executed, cancelled ? { ...patch, terminate: false } : patch); - await lane.emitBatch( - [ - { - type: "tool_end", - lane: lane.name, - runId: drive.operationId, - turnId: batch.turnId, - toolCallId: cleared.toolCall.id, - toolName: cleared.toolCall.name, - result: finalized.result, - isError: finalized.isError, - terminate: cancelled ? false : finalized.terminate, - ...(recovery ? { recovery: true as const } : {}), - }, - ], - drive.context, - ); - return { message: createToolResultMessage(finalized), terminate: cancelled ? false : finalized.terminate }; + const finalized = finalizeToolCall(cleared, executed, patch); + return { toolCall: finalized.toolCall, message: createToolResultMessage(finalized), terminate: finalized.terminate }; } async function prepareToolInvocation( @@ -418,7 +442,7 @@ async function prepareToolInvocation( ): Promise> { const toolCall = toolCallFor(sources, call); if (sources.assistant.stopReason === "length") { - return { kind: "outcome", outcome: { message: truncatedMessage(toolCall), terminate: false } }; + return { kind: "outcome", outcome: truncatedOutcome(toolCall) }; } const prepared = prepareToolCall(toolCall, tools); if ("kind" in prepared) return { kind: "outcome", outcome: outcomeFromFinalizedCall(prepared) }; @@ -440,7 +464,7 @@ async function prepareToolInvocation( } catch (error) { if (!(error instanceof AbortRequested)) throw error; await error.cancellation; - return { kind: "outcome", outcome: { message: abortedMessage(toolCall), terminate: false } }; + return { kind: "outcome", outcome: abortedOutcome(toolCall) }; } const cleared = applyBeforeToolDecision(prepared, decision); return "kind" in cleared @@ -460,22 +484,22 @@ async function startToolInvocation( ): Promise { const prepared = await prepareToolInvocation(lane, drive, sources, call, tools); if (prepared.kind === "outcome") { - return { - completion: publishToolOutcome(lane, drive, run, call, prepared.outcome.message, prepared.outcome.terminate), - }; + return { completion: publishToolOutcome(lane, drive, run, call, prepared.outcome, recovery) }; } const effectPending = await publishToolIntent( lane, drive, run, call, + prepared.cleared.toolCall, prepared.cleared.args, prepared.cleared.tool.replay ?? "never", + recovery, ); return { completion: effectPending.kind === "cancel_requested" - ? publishToolOutcome(lane, drive, run, call, abortedMessage(prepared.cleared.toolCall), false) + ? publishToolOutcome(lane, drive, run, call, abortedOutcome(prepared.cleared.toolCall), recovery) : performToolInvocation( lane, drive, @@ -484,9 +508,7 @@ async function startToolInvocation( prepared.cleared, toolContext, recovery, - ).then((outcome) => - publishToolOutcome(lane, drive, run, effectPending.value, outcome.message, outcome.terminate), - ), + ).then((outcome) => publishToolOutcome(lane, drive, run, effectPending.value, outcome, recovery)), }; } @@ -503,17 +525,17 @@ async function recoverToolInvocation( const toolCall = toolCallFor(sources, call); const tool = toolsByName.get(toolCall.name); if (!cancelled && call.replay === "safe" && tool?.replay === "safe") { - const args = await clearReplayCheckpoint(lane, drive, run.batch, call); + const args = await clearReplayCheckpoint(lane, drive, run.batch, call, toolCall); const cleared: ClearedToolCall = { toolCall, tool, args }; return { completion: performToolInvocation(lane, drive, run.batch, call, cleared, toolContext, true).then((outcome) => - publishToolOutcome(lane, drive, run, call, outcome.message, outcome.terminate), + publishToolOutcome(lane, drive, run, call, outcome, true), ), }; } const checkpoint = await readCheckpoint(lane, drive, call); return { - completion: publishToolOutcome(lane, drive, run, call, interruptedMessage(toolCall, checkpoint), false), + completion: publishToolOutcome(lane, drive, run, call, interruptedOutcome(toolCall, checkpoint), true), }; } @@ -543,10 +565,17 @@ async function runSequential( if (current.run.control.status === "cancel_requested") { const toolCall = toolCallFor(sources, call); if (call.status === "planned") { - await publishToolOutcome(lane, drive, current.run, call, abortedMessage(toolCall), false); + await publishToolOutcome(lane, drive, current.run, call, abortedOutcome(toolCall), recovery); } else { const checkpoint = await readCheckpoint(lane, drive, call); - await publishToolOutcome(lane, drive, current.run, call, interruptedMessage(toolCall, checkpoint), false); + await publishToolOutcome( + lane, + drive, + current.run, + call, + interruptedOutcome(toolCall, checkpoint), + recovery, + ); } continue; } diff --git a/packages/agent/src/harness/runtime/lane.ts b/packages/agent/src/harness/runtime/lane.ts index 219701e28..108ded398 100644 --- a/packages/agent/src/harness/runtime/lane.ts +++ b/packages/agent/src/harness/runtime/lane.ts @@ -34,6 +34,7 @@ import type { import { type BranchPreparation, prepareBranchEntries } from "../compaction/branch-summarization.ts"; import { prepareCompaction } from "../compaction/compaction.ts"; import { awaitWithContext, type Context } from "../context.ts"; +import { toolResultFromMessage } from "../execution/tools.ts"; import type { HookRegistry } from "../hooks.ts"; import { formatPromptTemplateInvocation } from "../prompt-templates.ts"; import { @@ -1796,7 +1797,7 @@ export class Lane implements AgentLane { throw new SessionInvariantError("Tool batch assistant entry is invalid"); } for (const call of batch.calls) { - if (call.status !== "effect_pending") continue; + if (call.status === "planned" || call.status === "completed") continue; const block = assistant.message.content[call.sourceIndex]; if (block?.type !== "toolCall") { throw new SessionInvariantError( @@ -1807,18 +1808,37 @@ export class Lane implements AgentLane { operationToolArgs(operation.meta.operationId, batch.turnId, call.sourceIndex), context, ); - if (args === undefined) { - throw new SessionInvariantError(`Tool call ${block.id} is missing persisted arguments`); + if (call.status === "effect_pending") { + if (args === undefined) { + throw new SessionInvariantError(`Tool call ${block.id} is missing persisted arguments`); + } + const checkpoint = await reader.getValue( + pendingToolOutput(operation.meta.operationId, call.resultEntryId), + context, + ); + runningTools.push({ + status: "running", + toolCallId: block.id, + toolName: block.name, + args: args.value, + ...(checkpoint === undefined ? {} : { result: checkpoint.value }), + }); + continue; + } + const staged = await reader.getValue(pendingEntry(call.resultEntryId), context); + if (staged?.value.type !== "message" || staged.value.payload.role !== "toolResult") { + throw new SessionInvariantError(`Tool call ${call.resultEntryId} is missing its staged result`); + } + if (staged.value.payload.toolCallId !== block.id || staged.value.payload.toolName !== block.name) { + throw new SessionInvariantError(`Tool call ${call.resultEntryId} has a mismatched staged result`); } - const checkpoint = await reader.getValue( - pendingToolOutput(operation.meta.operationId, call.resultEntryId), - context, - ); runningTools.push({ + status: "settled", toolCallId: block.id, toolName: block.name, - args: args.value, - ...(checkpoint === undefined ? {} : { partialResult: checkpoint.value }), + args: args?.value ?? block.arguments, + result: toolResultFromMessage(staged.value.payload, call.terminate), + isError: staged.value.payload.isError, }); } break; diff --git a/packages/agent/src/harness/runtime/reducer.ts b/packages/agent/src/harness/runtime/reducer.ts index d4764472b..16574d1a3 100644 --- a/packages/agent/src/harness/runtime/reducer.ts +++ b/packages/agent/src/harness/runtime/reducer.ts @@ -3,6 +3,14 @@ import type { OperationResultRecord } from "../session/types.ts"; export type LaneSnapshotReduction = "rebase" | undefined; +type LaneOperationSnapshot = NonNullable; + +function upsertTool(operation: LaneOperationSnapshot, tool: LaneOperationSnapshot["runningTools"][number]): void { + const index = operation.runningTools.findIndex((candidate) => candidate.toolCallId === tool.toolCallId); + if (index === -1) operation.runningTools.push(tool); + else operation.runningTools[index] = tool; +} + function matchingOperation( snapshot: LaneSnapshot, operationId: string, @@ -108,7 +116,9 @@ export function reduceLaneSnapshot( } case "tool_start": { const operation = matchingOperation(snapshot, event.runId); - operation?.runningTools.push({ + if (operation === undefined) return; + upsertTool(operation, { + status: "running", toolCallId: event.toolCallId, toolName: event.toolName, args: event.args, @@ -116,24 +126,42 @@ export function reduceLaneSnapshot( return; } case "tool_update": { - const tool = snapshot.operation?.runningTools.find((candidate) => candidate.toolCallId === event.toolCallId); - if (tool !== undefined) tool.partialResult = event.partialResult; + const operation = matchingOperation(snapshot, event.runId); + const tool = operation?.runningTools.find((candidate) => candidate.toolCallId === event.toolCallId); + if (tool?.status === "running") tool.result = event.partialResult; return; } case "tool_end": { const operation = matchingOperation(snapshot, event.runId); - const index = - operation?.runningTools.findIndex((candidate) => candidate.toolCallId === event.toolCallId) ?? -1; - if (operation !== undefined && index !== -1) operation.runningTools.splice(index, 1); + if (operation === undefined) return; + const index = operation.runningTools.findIndex((candidate) => candidate.toolCallId === event.toolCallId); + const current = operation.runningTools[index]; + if (current === undefined) return; + operation.runningTools[index] = { + status: "settled", + toolCallId: event.toolCallId, + toolName: event.toolName, + args: current.args, + result: event.result, + isError: event.isError, + }; return; } - case "entry_added": - if (snapshot.transcript.some((entry) => entry.id === event.entry.id)) return; + case "entry_added": { + if (event.entry.type === "message" && event.entry.message.role === "toolResult") { + const operation = snapshot.operation; + if (operation !== null) { + const toolCallId = event.entry.message.toolCallId; + const index = operation.runningTools.findIndex((candidate) => candidate.toolCallId === toolCallId); + if (index !== -1) operation.runningTools.splice(index, 1); + } + } if (event.entry.type === "compaction") snapshot.transcript.splice(0, snapshot.transcript.length, event.entry); else snapshot.transcript.push(event.entry); snapshot.tipId = event.entry.id; if (event.entry.type === "message") snapshot.stats.messageCount += 1; return; + } case "queue_update": snapshot.queues = event.queues; return; diff --git a/packages/agent/test/harness/runtime/drive-tools.test.ts b/packages/agent/test/harness/runtime/drive-tools.test.ts index 5b2f3dec7..8c1b84880 100644 --- a/packages/agent/test/harness/runtime/drive-tools.test.ts +++ b/packages/agent/test/harness/runtime/drive-tools.test.ts @@ -9,7 +9,7 @@ import { Type } from "typebox"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { HarnessEvent, WatchHandle } from "../../../src/harness/agent-harness.ts"; import { DEFAULT_COMPACTION_SETTINGS } from "../../../src/harness/compaction/compaction.ts"; -import { BACKGROUND_CONTEXT } from "../../../src/harness/context.ts"; +import { BACKGROUND_CONTEXT, type Context } from "../../../src/harness/context.ts"; import { HookRegistry } from "../../../src/harness/hooks.ts"; import { runTools } from "../../../src/harness/runtime/drive/tools.ts"; import { Lane } from "../../../src/harness/runtime/lane.ts"; @@ -45,7 +45,7 @@ interface Fixture { assistantEntryId: string; resultEntryIds: string[]; operationId: string; - config: Config; + observations: string[]; } interface FixtureOptions { @@ -69,6 +69,32 @@ function unusedWatch(): WatchHandle { throw new Error("watch is not used by tool procedure tests"); } +class ObservedMemoryStorage extends MemoryStorage { + readonly observations: string[] = []; + + override async commit(writes: Write[], context: Context) { + const result = await super.commit(writes, context); + if ( + writes.some((write) => write.kind === "value" && write.op === "set" && write.namespace === "pi.op.tool_args") + ) { + this.observations.push("intent_commit"); + } + const stagesOutcome = writes.some( + (write) => write.kind === "value" && write.op === "set" && write.namespace === "pi.pending.entry", + ); + if (stagesOutcome) this.observations.push("outcome_commit"); + if ( + !stagesOutcome && + writes.some( + (write) => write.kind === "value" && write.op === "delete" && write.namespace === "pi.pending.tool_output", + ) + ) { + this.observations.push("replay_commit"); + } + return result; + } +} + function deferred(): { promise: Promise; resolve(value: T): void } { let resolvePromise: ((value: T) => void) | undefined; const promise = new Promise((resolve) => { @@ -113,7 +139,7 @@ function tool( } async function createFixture(options: FixtureOptions): Promise { - const storage = new MemoryStorage({ now: () => 100 }); + const storage = new ObservedMemoryStorage({ now: () => 100 }); const session = new StorageBackedSession( { id: `drive-tools-${sessions.length}`, createdAt: 1, storageVersion: 1 }, storage, @@ -130,7 +156,7 @@ async function createFixture(options: FixtureOptions): Promise { activeToolNames: options.calls.map(({ name }) => name), }; const assistant = fauxAssistantMessage( - options.calls.map(({ name, value = name }) => fauxToolCall(name, { value })), + options.calls.map(({ name, value = name }, index) => fauxToolCall(name, { value }, { id: `call-${index}` })), { stopReason: options.stopReason ?? "toolUse", timestamp: 20 }, ); const calls = @@ -203,6 +229,7 @@ async function createFixture(options: FixtureOptions): Promise { (cause) => (cause instanceof Error ? cause : new Error(String(cause))), async (batch) => { events.push(...structuredClone(batch)); + storage.observations.push(...batch.map((event) => event.type)); await options.onEmit?.(batch); }, unusedWatch, @@ -210,7 +237,17 @@ async function createFixture(options: FixtureOptions): Promise { ); const drive = new Drive({ operationId }, BACKGROUND_CONTEXT); lane.activeDrive = drive; - return { session, lane, drive, hooks, events, assistantEntryId, resultEntryIds, operationId, config }; + return { + session, + lane, + drive, + hooks, + events, + assistantEntryId, + resultEntryIds, + operationId, + observations: storage.observations, + }; } function currentRun(fixture: Fixture): OperationState { @@ -325,6 +362,17 @@ describe("durable tool batch", () => { await expect(lateInvocation?.setMemo("late", true)).rejects.toThrow("no longer owns"); const eventTypes = fixture.events.map(({ type }) => type); expect(eventTypes.indexOf("tool_update")).toBeLessThan(eventTypes.indexOf("tool_end")); + const firstIntent = fixture.observations.indexOf("intent_commit"); + const firstStart = fixture.observations.indexOf("tool_start"); + const firstUpdate = fixture.observations.indexOf("tool_update"); + const firstCommit = fixture.observations.indexOf("outcome_commit"); + const firstEnd = fixture.observations.indexOf("tool_end"); + const firstPlacement = fixture.observations.indexOf("entry_added"); + expect(firstIntent).toBeLessThan(firstStart); + expect(firstStart).toBeLessThan(firstUpdate); + expect(firstUpdate).toBeLessThan(firstCommit); + expect(firstCommit).toBeLessThan(firstEnd); + expect(firstEnd).toBeLessThan(firstPlacement); expect(fixture.events.filter(({ type }) => type === "entry_added")).toHaveLength(2); expect(fixture.events.filter(({ type }) => type === "usage")).toHaveLength(1); expect(fixture.events.at(-1)?.type).toBe("turn_end"); @@ -352,6 +400,12 @@ describe("durable tool batch", () => { await waitFor(() => currentCalls(fixture)[1]?.status === "outcome_ready"); expect(currentCalls(fixture).map(({ status }) => status)).toEqual(["effect_pending", "outcome_ready"]); expect(await fixture.session.getEntry(fixture.resultEntryIds[1]!, BACKGROUND_CONTEXT)).toBeUndefined(); + expect(fixture.events.find((event) => event.type === "tool_start" && event.toolName === "b")).toMatchObject({ + args: { value: "b" }, + }); + expect(fixture.events.find((event) => event.type === "tool_end" && event.toolName === "b")).toMatchObject({ + result: { content: [{ type: "text", text: "b" }] }, + }); finishA.resolve(); await running; @@ -402,6 +456,14 @@ describe("durable tool batch", () => { expect(safeExecute).toHaveBeenCalledOnce(); expect(safeExecute.mock.calls[0]?.[0]).toBe("persisted"); expect(unsafeExecute).not.toHaveBeenCalled(); + expect( + fixture.events.find((event) => event.type === "tool_start" && event.toolCallId === "call-0"), + ).toMatchObject({ recovery: true, args: { value: "persisted" } }); + expect(fixture.observations.indexOf("replay_commit")).toBeLessThan(fixture.observations.indexOf("tool_start")); + expect(fixture.events.find((event) => event.type === "tool_end" && event.toolCallId === "call-0")).toMatchObject({ + recovery: true, + isError: false, + }); const unsafeEntry = await fixture.session.getEntry(fixture.resultEntryIds[1]!, BACKGROUND_CONTEXT); expect(unsafeEntry?.type === "message" ? unsafeEntry.message : undefined).toMatchObject({ role: "toolResult", @@ -413,6 +475,11 @@ describe("durable tool batch", () => { ? unsafeEntry.message.content.at(-1) : undefined, ).toMatchObject({ type: "text", text: expect.stringContaining("external outcome is unknown") }); + expect(fixture.events.some((event) => event.type === "tool_start" && event.toolCallId === "call-1")).toBe(false); + expect(fixture.events.find((event) => event.type === "tool_end" && event.toolCallId === "call-1")).toMatchObject({ + recovery: true, + isError: true, + }); }); it("reconciles a restored cancelled batch without hooks, context, or effects", async () => { @@ -456,6 +523,12 @@ describe("durable tool batch", () => { ? pending.message.content.at(-1) : undefined, ).toMatchObject({ type: "text", text: expect.stringContaining("external outcome is unknown") }); + const starts = fixture.events.filter((event) => event.type === "tool_start"); + const ends = fixture.events.filter((event) => event.type === "tool_end"); + expect(starts).toHaveLength(1); + expect(starts[0]).toMatchObject({ toolName: "planned", args: { value: "planned" } }); + expect(ends).toHaveLength(2); + expect(ends.find((event) => event.toolCallId === starts[0]?.toolCallId)).toMatchObject({ isError: true }); }); it("materializes outcome-ready state without resolving tools or tool context", async () => { @@ -463,7 +536,7 @@ describe("durable tool batch", () => { const context = vi.fn(() => undefined); const staged: ToolResultMessage = { role: "toolResult", - toolCallId: "call-1", + toolCallId: "call-0", toolName: "ready", content: [{ type: "text", text: "already done" }], addedToolNames: ["later"], @@ -478,19 +551,10 @@ describe("durable tool batch", () => { extraWrites: ({ resultEntryIds }) => [ storedValues.setValue(storedValues.pendingEntry(resultEntryIds[0]!), { type: "message", - payload: { ...staged, toolCallId: "call-1" }, + payload: staged, }), ], }); - const assistant = await fixture.session.getEntry(fixture.assistantEntryId, BACKGROUND_CONTEXT); - if (assistant?.type !== "message" || assistant.message.role !== "assistant") throw new Error("missing assistant"); - const block = assistant.message.content[0]; - if (block?.type !== "toolCall") throw new Error("missing call"); - await fixture.session.setValue( - storedValues.pendingEntry(fixture.resultEntryIds[0]!), - { type: "message", payload: { ...staged, toolCallId: block.id } }, - BACKGROUND_CONTEXT, - ); await driveTools(fixture); expect(execute).not.toHaveBeenCalled(); @@ -511,6 +575,8 @@ describe("durable tool batch", () => { tools: [tool("present", execute)], stopReason: "length", }); + const truncatedAfterTool = vi.fn(() => undefined); + truncated.hooks.on("after_tool", truncatedAfterTool); await driveTools(truncated); expect(execute).not.toHaveBeenCalled(); const truncatedEntry = await truncated.session.getEntry(truncated.resultEntryIds[0]!, BACKGROUND_CONTEXT); @@ -519,8 +585,18 @@ describe("durable tool batch", () => { ? truncatedEntry.message.content[0] : undefined, ).toMatchObject({ type: "text", text: expect.stringContaining("arguments may be truncated") }); + expect(truncatedAfterTool).not.toHaveBeenCalled(); + expect(truncated.events.filter((event) => event.type === "tool_start")).toHaveLength(1); + expect(truncated.events.filter((event) => event.type === "tool_end")).toHaveLength(1); + expect(truncated.observations.indexOf("outcome_commit")).toBeLessThan( + truncated.observations.indexOf("tool_start"), + ); + expect(truncated.observations.indexOf("tool_start")).toBeLessThan(truncated.observations.indexOf("tool_end")); + expect(truncated.observations.indexOf("tool_end")).toBeLessThan(truncated.observations.indexOf("entry_added")); const missing = await createFixture({ calls: [{ name: "missing" }], tools: [] }); + const missingAfterTool = vi.fn(() => undefined); + missing.hooks.on("after_tool", missingAfterTool); await driveTools(missing); const missingEntry = await missing.session.getEntry(missing.resultEntryIds[0]!, BACKGROUND_CONTEXT); expect(missingEntry?.type === "message" ? missingEntry.message : undefined).toMatchObject({ @@ -533,6 +609,17 @@ describe("durable tool batch", () => { ? Object.hasOwn(missingEntry.message, "details") : true, ).toBe(false); + expect(missingAfterTool).not.toHaveBeenCalled(); + expect(missing.events.filter((event) => event.type === "tool_start")).toHaveLength(1); + const missingEnds = missing.events.filter((event) => event.type === "tool_end"); + expect(missingEnds).toHaveLength(1); + expect(missing.events.find((event) => event.type === "tool_start")).toMatchObject({ + args: { value: "missing" }, + }); + expect(missingEnds[0]).toMatchObject({ + result: { content: [{ type: "text", text: 'Tool "missing" is unavailable' }], details: undefined }, + isError: true, + }); }); it("awaits update delivery and checkpoint persistence before after_tool", async () => { diff --git a/packages/agent/test/harness/runtime/reducer.test.ts b/packages/agent/test/harness/runtime/reducer.test.ts index 4ca39cbcf..8f1475829 100644 --- a/packages/agent/test/harness/runtime/reducer.test.ts +++ b/packages/agent/test/harness/runtime/reducer.test.ts @@ -175,6 +175,96 @@ describe("lane snapshot reducer", () => { expect(running).toMatchObject({ operation: { id: "run", kind: "run" } }); }); + it("retains settled parallel tools until each source-ordered result is placed", async () => { + const { lane } = await createFixture(); + const snapshot = (await lane.watch(BACKGROUND_CONTEXT)).snapshot; + snapshot.operation = { + id: "run", + kind: "run", + startedAt: 1, + fromTipId: null, + status: "open", + runningTools: [], + }; + const result = (text: string) => ({ content: [{ type: "text" as const, text }], details: { text } }); + const start = (index: number): HarnessEvent => ({ + type: "tool_start", + lane: "main", + runId: "run", + turnId: "turn", + toolCallId: `call-${index}`, + toolName: `tool-${index}`, + args: { index }, + }); + const end = (index: number): HarnessEvent => ({ + type: "tool_end", + lane: "main", + runId: "run", + turnId: "turn", + toolCallId: `call-${index}`, + toolName: `tool-${index}`, + result: result(`done-${index}`), + isError: false, + terminate: false, + }); + const entry = (index: number): HarnessEvent => ({ + type: "entry_added", + lane: "main", + entry: { + id: `result-${index}`, + parentId: index === 0 ? null : `result-${index - 1}`, + seq: index + 1, + timestamp: index + 1, + type: "message", + message: { + role: "toolResult", + toolCallId: `call-${index}`, + toolName: `tool-${index}`, + content: [{ type: "text", text: `done-${index}` }], + isError: false, + timestamp: index + 1, + }, + }, + }); + + for (let index = 0; index < 3; index += 1) reduceLaneSnapshot(snapshot, start(index)); + expect(snapshot.operation.runningTools).toHaveLength(3); + reduceLaneSnapshot(snapshot, start(0)); + expect(snapshot.operation.runningTools).toHaveLength(3); + + reduceLaneSnapshot(snapshot, { + type: "tool_update", + lane: "main", + runId: "stale", + turnId: "turn", + toolCallId: "call-1", + toolName: "tool-1", + partialResult: result("stale"), + }); + expect(snapshot.operation.runningTools[1]).not.toHaveProperty("result"); + + reduceLaneSnapshot(snapshot, end(2)); + expect(snapshot.operation.runningTools.find(({ toolCallId }) => toolCallId === "call-2")).toMatchObject({ + status: "settled", + args: { index: 2 }, + result: result("done-2"), + }); + + reduceLaneSnapshot(snapshot, end(0)); + expect(snapshot.operation.runningTools.map(({ status }) => status)).toEqual(["settled", "running", "settled"]); + reduceLaneSnapshot(snapshot, entry(0)); + expect(snapshot.operation.runningTools.map(({ toolCallId }) => toolCallId)).toEqual(["call-1", "call-2"]); + expect(snapshot.transcript.map(({ id }) => id)).toEqual(["result-0"]); + + reduceLaneSnapshot(snapshot, end(1)); + expect(snapshot.operation.runningTools.every(({ status }) => status === "settled")).toBe(true); + reduceLaneSnapshot(snapshot, entry(1)); + expect(snapshot.operation.runningTools.map(({ toolCallId }) => toolCallId)).toEqual(["call-2"]); + reduceLaneSnapshot(snapshot, entry(2)); + expect(snapshot.operation.runningTools).toEqual([]); + expect(snapshot.transcript.map(({ id }) => id)).toEqual(["result-0", "result-1", "result-2"]); + }); + it("marks navigation completion for rebase", async () => { const { lane } = await createFixture(); const snapshot = (await lane.watch(BACKGROUND_CONTEXT)).snapshot; diff --git a/packages/agent/test/harness/runtime/watch.test.ts b/packages/agent/test/harness/runtime/watch.test.ts index 76258b960..cfdb8e645 100644 --- a/packages/agent/test/harness/runtime/watch.test.ts +++ b/packages/agent/test/harness/runtime/watch.test.ts @@ -257,7 +257,7 @@ describe("runtime lane watch", () => { watch.unsubscribe(); }); - it("reduces assistant frames and renders only effect-pending tools with full-content indexes", async () => { + it("reduces assistant frames and projects running and settled tools with full-content indexes", async () => { const frameSession = await createSession(); const partial = fauxAssistantMessage([], { stopReason: "pending" }); const frames: AssistantMessageFrame[] = [ @@ -306,10 +306,27 @@ describe("runtime lane watch", () => { frameWatch.unsubscribe(); const toolSession = await createSession(); + const toolUsage = { + input: 1, + output: 2, + cacheRead: 3, + cacheWrite: 4, + totalTokens: 10, + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, + }; const assistant = fauxAssistantMessage([ { type: "text", text: "before" }, - { type: "toolCall", id: "call", name: "read", arguments: { live: false } }, - { type: "toolCall", id: "call-without-checkpoint", name: "write", arguments: { live: false } }, + { type: "toolCall", id: "call-completed", name: "completed", arguments: { source: "completed" } }, + { type: "toolCall", id: "call-running", name: "read", arguments: { source: "running" } }, + { + type: "toolCall", + id: "call-without-checkpoint", + name: "write", + arguments: { source: "without-checkpoint" }, + }, + { type: "toolCall", id: "call-ready", name: "real", arguments: { source: "real" } }, + { type: "toolCall", id: "call-synthetic", name: "missing", arguments: { source: "synthetic" } }, + { type: "toolCall", id: "call-planned", name: "planned", arguments: { source: "planned" } }, ]); const toolOperationId = toolSession.idGenerator.next(); await commit(toolSession, [ @@ -329,25 +346,66 @@ describe("runtime lane watch", () => { configuration, turnId: "turn", calls: [ - { status: "planned", sourceIndex: 1, resultEntryId: "planned" }, - { status: "effect_pending", sourceIndex: 1, resultEntryId: "result", replay: "safe" }, + { status: "completed", sourceIndex: 1, resultEntryId: "completed", terminate: false }, + { status: "effect_pending", sourceIndex: 2, resultEntryId: "result", replay: "safe" }, { status: "effect_pending", - sourceIndex: 2, + sourceIndex: 3, resultEntryId: "without-checkpoint", replay: "never", }, - { status: "outcome_ready", sourceIndex: 1, resultEntryId: "ready", terminate: false }, + { status: "outcome_ready", sourceIndex: 4, resultEntryId: "ready", terminate: true }, + { status: "outcome_ready", sourceIndex: 5, resultEntryId: "synthetic", terminate: false }, + { status: "planned", sourceIndex: 6, resultEntryId: "planned" }, ], }, }), - storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 1), { path: "file" }), - storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 2), { path: "output" }), + sessionWrites.insertEntry({ + id: "completed", + parentId: "assistant", + type: "message", + message: { + role: "toolResult", + toolCallId: "call-completed", + toolName: "completed", + content: [{ type: "text", text: "completed" }], + isError: false, + timestamp: 2, + }, + }), + storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 2), { path: "file" }), + storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 3), { path: "output" }), + storedValues.setValue(storedValues.operationToolArgs(toolOperationId, "turn", 4), { path: "settled" }), storedValues.setValue(storedValues.pendingToolOutput(toolOperationId, "result"), { content: [{ type: "text", text: "partial" }], details: { bytes: 1 }, }), - storedValues.setValue(storedValues.branchTip("main"), "assistant"), + storedValues.setValue(storedValues.pendingEntry("ready"), { + type: "message", + payload: { + role: "toolResult", + toolCallId: "call-ready", + toolName: "real", + content: [{ type: "text", text: "settled" }], + details: { kind: "real" }, + usage: toolUsage, + addedToolNames: ["later"], + isError: false, + timestamp: 3, + }, + }), + storedValues.setValue(storedValues.pendingEntry("synthetic"), { + type: "message", + payload: { + role: "toolResult", + toolCallId: "call-synthetic", + toolName: "missing", + content: [{ type: "text", text: "unavailable" }], + isError: true, + timestamp: 4, + }, + }), + storedValues.setValue(storedValues.branchTip("main"), "completed"), storedValues.setValue(storedValues.laneState("main"), { currentOperationId: toolOperationId, lastOperationId: null, @@ -356,23 +414,110 @@ describe("runtime lane watch", () => { ]); const toolHarness = await attach(toolSession); const toolWatch = await toolHarness.watch(BACKGROUND_CONTEXT); + expect(toolWatch.snapshot.transcript.map(({ id }) => id)).toEqual(["assistant", "completed"]); expect(toolWatch.snapshot.operation?.runningTools).toEqual([ { - toolCallId: "call", + status: "running", + toolCallId: "call-running", toolName: "read", args: { path: "file" }, - partialResult: { content: [{ type: "text", text: "partial" }], details: { bytes: 1 } }, + result: { content: [{ type: "text", text: "partial" }], details: { bytes: 1 } }, }, { + status: "running", toolCallId: "call-without-checkpoint", toolName: "write", args: { path: "output" }, }, + { + status: "settled", + toolCallId: "call-ready", + toolName: "real", + args: { path: "settled" }, + result: { + content: [{ type: "text", text: "settled" }], + details: { kind: "real" }, + usage: toolUsage, + addedToolNames: ["later"], + terminate: true, + }, + isError: false, + }, + { + status: "settled", + toolCallId: "call-synthetic", + toolName: "missing", + args: { source: "synthetic" }, + result: { + content: [{ type: "text", text: "unavailable" }], + details: undefined, + }, + isError: true, + }, ]); - expect(toolWatch.snapshot.operation?.runningTools[1]).not.toHaveProperty("partialResult"); + expect(toolWatch.snapshot.operation?.runningTools[1]).not.toHaveProperty("result"); toolWatch.unsubscribe(); }); + it("faults missing or mismatched staged outcome-ready results", async () => { + for (const corruption of ["missing", "mismatched"] as const) { + const session = await createSession(); + const operationId = session.idGenerator.next(); + const writes: Write[] = [ + sessionWrites.insertEntry({ + id: "assistant", + parentId: null, + type: "message", + message: fauxAssistantMessage([ + { type: "toolCall", id: "call", name: "read", arguments: { path: "file" } }, + ]), + }), + storedValues.setValue(storedValues.operationMeta(operationId), { + operationId, + lane: "main", + sourceTipId: null, + startedAt: 1, + intent: { kind: "run", promptEntryIds: [] }, + }), + storedValues.setValue(storedValues.operationState(operationId), { + ...runScope(), + at: "tools", + batch: { + assistantEntryId: "assistant", + configuration, + turnId: "turn", + calls: [{ status: "outcome_ready", sourceIndex: 0, resultEntryId: "result", terminate: false }], + }, + }), + storedValues.setValue(storedValues.branchTip("main"), "assistant"), + storedValues.setValue(storedValues.laneState("main"), { + currentOperationId: operationId, + lastOperationId: null, + inbox: [], + }), + ]; + if (corruption === "mismatched") { + writes.push( + storedValues.setValue(storedValues.pendingEntry("result"), { + type: "message", + payload: { + role: "toolResult", + toolCallId: "other-call", + toolName: "read", + content: [], + isError: false, + timestamp: 2, + }, + }), + ); + } + await commit(session, writes); + const harness = await attach(session); + + await expect(harness.watch(BACKGROUND_CONTEXT)).rejects.toBeInstanceOf(HarnessFault); + } + }); + it("faults required payload corruption and unsubscribes the incomplete watcher", async () => { const session = await createSession(); await commit(session, [ diff --git a/packages/agent/test/harness/types.test.ts b/packages/agent/test/harness/types.test.ts index fa7a5a53f..39b8e8782 100644 --- a/packages/agent/test/harness/types.test.ts +++ b/packages/agent/test/harness/types.test.ts @@ -33,6 +33,7 @@ import type { LaneConfiguration, LaneExecutionInfo, LaneSnapshot, + LaneSnapshotTool, NewEntry, OperationAdmissionResult, OperationAt, @@ -443,6 +444,7 @@ it("covers Part 5 results, events, hooks, snapshots, tools, and stream options", systemPrompt: string; }>(); expectTypeOf().not.toEqualTypeOf(); + expectTypeOf().toEqualTypeOf<"running" | "settled">(); expectTypeOf().returns.toEqualTypeOf>(); expectTypeOf().toEqualTypeOf<{ operationId: string; diff --git a/packages/coding-agent/src/experimental/client-tui-chat.ts b/packages/coding-agent/src/experimental/client-tui-chat.ts index f452314c9..7584ae658 100644 --- a/packages/coding-agent/src/experimental/client-tui-chat.ts +++ b/packages/coding-agent/src/experimental/client-tui-chat.ts @@ -42,12 +42,11 @@ export class ExperimentalChatView { this.#syncStreaming(snapshot.operation?.streamingMessage); for (const tool of snapshot.operation?.runningTools ?? []) { const component = this.#tool(tool.toolName, tool.toolCallId, tool.args); - component.markExecutionStarted(); - if (tool.partialResult) { - component.updateResult( - { content: tool.partialResult.content, details: tool.partialResult.details, isError: false }, - true, - ); + if (tool.status === "running") { + component.markExecutionStarted(); + if (tool.result !== undefined) component.updateResult({ ...tool.result, isError: false }, true); + } else { + component.updateResult({ ...tool.result, isError: tool.isError }, false); } } this.#syncQueues(snapshot.queues); diff --git a/packages/coding-agent/src/experimental/mini/tui/view.ts b/packages/coding-agent/src/experimental/mini/tui/view.ts index 52150471f..295d178c5 100644 --- a/packages/coding-agent/src/experimental/mini/tui/view.ts +++ b/packages/coding-agent/src/experimental/mini/tui/view.ts @@ -251,12 +251,11 @@ class MiniTui { this.#syncStreaming(snapshot.operation?.streamingMessage); for (const tool of snapshot.operation?.runningTools ?? []) { const component = this.#tool(tool.toolName, tool.toolCallId, tool.args); - component.markExecutionStarted(); - if (tool.partialResult) { - component.updateResult( - { content: tool.partialResult.content, details: tool.partialResult.details, isError: false }, - true, - ); + if (tool.status === "running") { + component.markExecutionStarted(); + if (tool.result !== undefined) component.updateResult({ ...tool.result, isError: false }, true); + } else { + component.updateResult({ ...tool.result, isError: tool.isError }, false); } } this.#syncQueues(snapshot.queues);