mirror of
https://github.com/badlogic/pi-mono.git
synced 2026-11-05 23:37:25 +00:00
fix(agent): retain settled tools until placement
This commit is contained in:
Mario Zechner
parent
6e8b9c8ea6
commit
e26afb63a4
15 changed files with 643 additions and 233 deletions
|
|
@ -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<unknown> }>;
|
||||
runningTools: Array<
|
||||
| { status: "running"; toolCallId: string; toolName: string; args: unknown;
|
||||
result?: AgentToolResult<unknown> }
|
||||
| { status: "settled"; toolCallId: string; toolName: string; args: unknown;
|
||||
result: AgentToolResult<unknown>; isError: boolean }
|
||||
>;
|
||||
};
|
||||
queues: LaneQueuedItem[];
|
||||
faulted: boolean;
|
||||
|
|
@ -1129,7 +1135,7 @@ interface WatchHandle<T> {
|
|||
|
||||
`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.
|
||||
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -191,6 +191,23 @@ export interface LaneInfo {
|
|||
operation: CurrentOperationInfo | null;
|
||||
}
|
||||
|
||||
export type LaneSnapshotTool =
|
||||
| {
|
||||
status: "running";
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
args: unknown;
|
||||
result?: AgentToolResult<unknown>;
|
||||
}
|
||||
| {
|
||||
status: "settled";
|
||||
toolCallId: string;
|
||||
toolName: string;
|
||||
args: unknown;
|
||||
result: AgentToolResult<unknown>;
|
||||
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<unknown>;
|
||||
}[];
|
||||
runningTools: LaneSnapshotTool[];
|
||||
};
|
||||
queues: LaneQueuedItem[];
|
||||
faulted: boolean;
|
||||
|
|
|
|||
|
|
@ -180,6 +180,20 @@ export function finalizeToolCall<TContext extends object | undefined>(
|
|||
};
|
||||
}
|
||||
|
||||
/** Reconstruct the canonical tool result represented by a staged transcript message. */
|
||||
export function toolResultFromMessage(
|
||||
message: ToolResultMessage<unknown>,
|
||||
terminate: boolean,
|
||||
): AgentToolResult<unknown> {
|
||||
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 {
|
||||
|
|
|
|||
|
|
@ -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<void> };
|
||||
type ToolOutcome = { message: ToolResultMessage<unknown>; terminate: boolean };
|
||||
type ToolOutcome = { toolCall: AgentToolCall; message: ToolResultMessage<unknown>; terminate: boolean };
|
||||
type PreparedToolInvocation<TContext extends object | undefined> =
|
||||
| { kind: "ready"; cleared: ClearedToolCall<TContext> }
|
||||
| { kind: "outcome"; outcome: ToolOutcome };
|
||||
|
|
@ -146,32 +147,41 @@ function syntheticMessage(
|
|||
};
|
||||
}
|
||||
|
||||
function abortedMessage(toolCall: AgentToolCall): ToolResultMessage<unknown> {
|
||||
return syntheticMessage(toolCall, [{ type: "text", text: "Tool execution was cancelled before completion." }]);
|
||||
}
|
||||
|
||||
function interruptedMessage(
|
||||
toolCall: AgentToolCall,
|
||||
checkpoint: AgentToolResult<unknown> | undefined,
|
||||
): ToolResultMessage<unknown> {
|
||||
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<unknown> {
|
||||
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<unknown> | 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<TContext extends object | undefined>(
|
||||
|
|
@ -179,8 +189,10 @@ async function publishToolIntent<TContext extends object | undefined>(
|
|||
drive: Drive,
|
||||
run: ToolsOperation,
|
||||
planned: Extract<ToolCall, { status: "planned" }>,
|
||||
toolCall: AgentToolCall,
|
||||
args: Record<string, JsonValue>,
|
||||
replay: "never" | "safe",
|
||||
recovery: boolean,
|
||||
): Promise<ContinueOperationResult<Extract<ToolCall, { status: "effect_pending" }>>> {
|
||||
return lane.continueOperation(
|
||||
run,
|
||||
|
|
@ -196,6 +208,18 @@ async function publishToolIntent<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
|||
lane: Lane<TContext>,
|
||||
drive: Drive,
|
||||
capability: ToolsOperation,
|
||||
call: ToolCall,
|
||||
message: ToolResultMessage<unknown>,
|
||||
terminate: boolean,
|
||||
call: Extract<ToolCall, { status: "planned" | "effect_pending" }>,
|
||||
finalized: ToolOutcome,
|
||||
recovery: boolean,
|
||||
): Promise<void> {
|
||||
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<ToolCall, { status: "outcome_ready" }> = {
|
||||
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<TContext extends object | undefined>(
|
|||
drive: Drive,
|
||||
batch: ToolBatch,
|
||||
call: Extract<ToolCall, { status: "effect_pending" }>,
|
||||
toolCall: AgentToolCall,
|
||||
): Promise<Record<string, JsonValue>> {
|
||||
return lane.command(async (state, reader) => {
|
||||
const stored = await reader.getValue(
|
||||
|
|
@ -257,6 +312,18 @@ async function clearReplayCheckpoint<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
|||
const capability = invocationCapability(lane, drive, batch, call);
|
||||
const progress = openToolProgress(lane, drive, batch.turnId, call.sourceIndex, call.resultEntryId);
|
||||
let latestUpdateDelivery: Promise<void> = Promise.resolve();
|
||||
let startDelivered: Promise<void> = Promise.resolve();
|
||||
let startQueued = false;
|
||||
const bufferedUpdates: AgentToolResult<unknown>[] = [];
|
||||
const publishUpdate = (partial: AgentToolResult<unknown>): void => {
|
||||
latestUpdateDelivery = lane.emitBatch(
|
||||
[
|
||||
|
|
@ -320,48 +384,26 @@ async function performToolInvocation<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
||||
|
|
@ -418,7 +442,7 @@ async function prepareToolInvocation<TContext extends object | undefined>(
|
|||
): Promise<PreparedToolInvocation<TContext>> {
|
||||
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<TContext extends object | undefined>(
|
|||
} 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<TContext extends object | undefined>(
|
|||
): Promise<ToolCallTask> {
|
||||
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<TContext extends object | undefined>(
|
|||
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<TContext extends object | undefined>(
|
|||
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<TContext> = { 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<TContext extends object | undefined>(
|
|||
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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<TContext extends object | undefined> 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<TContext extends object | undefined> 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;
|
||||
|
|
|
|||
|
|
@ -3,6 +3,14 @@ import type { OperationResultRecord } from "../session/types.ts";
|
|||
|
||||
export type LaneSnapshotReduction = "rebase" | undefined;
|
||||
|
||||
type LaneOperationSnapshot = NonNullable<LaneSnapshot["operation"]>;
|
||||
|
||||
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;
|
||||
|
|
|
|||
|
|
@ -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<undefined>;
|
||||
observations: string[];
|
||||
}
|
||||
|
||||
interface FixtureOptions {
|
||||
|
|
@ -69,6 +69,32 @@ function unusedWatch<T>(): WatchHandle<T> {
|
|||
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<T = void>(): { promise: Promise<T>; resolve(value: T): void } {
|
||||
let resolvePromise: ((value: T) => void) | undefined;
|
||||
const promise = new Promise<T>((resolve) => {
|
||||
|
|
@ -113,7 +139,7 @@ function tool(
|
|||
}
|
||||
|
||||
async function createFixture(options: FixtureOptions): Promise<Fixture> {
|
||||
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<Fixture> {
|
|||
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<Fixture> {
|
|||
(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<Fixture> {
|
|||
);
|
||||
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 () => {
|
||||
|
|
|
|||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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, [
|
||||
|
|
|
|||
|
|
@ -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<LaneSnapshot["operation"]>().not.toEqualTypeOf<SessionSnapshot>();
|
||||
expectTypeOf<LaneSnapshotTool["status"]>().toEqualTypeOf<"running" | "settled">();
|
||||
expectTypeOf<AgentLane["getResult"]>().returns.toEqualTypeOf<Promise<OperationResultRecord | undefined>>();
|
||||
expectTypeOf<SuspendedRun>().toEqualTypeOf<{
|
||||
operationId: string;
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
|
|
@ -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);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue