diff --git a/CHANGELOG.md b/CHANGELOG.md index d301b1ab..bb57ec29 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,33 @@ # Changelog +## 0.234.0 + +**A resume heals the 0.233.0 crash window.** The `reconciled` record now carries the settlement +the driver received (`reason`, `infra`, `trace`, `outRef`, `providerModel`, `harnessTranscript`), +the withheld overspend, the cancellation source when there was one, and `settledSeq` — the +cursor seq `next()` stamped on the delivery. It is written through the same field spread as the +terminal record, so the two cannot disagree. + +- A spawned child with no terminal record whose `environment-teardown` receipts after its latest + `reconciled` record all read `destroyed: true`, and which name the environment its last + admission named, gets its released terminal record on the next resume (`healReleasedSlots`, + before interrupted executors are prepared) from the release sweep's own builder, at + `settledSeq` and the reconciled `at`: the bytes a completed sweep would have written. + `fleetYield` counts it `down` (or `cancelled`, with its `source`) and `releasedUnrecovered`, + `spendGaps` names it `unreported` rather than `never-settled`, replay yields it where the driver + saw it, and the `agent.child` `:released` event is emitted on the resumed stream. It is never + treated as interrupted, and no recovery is attempted against the destroyed environment. +- A `destroyed: false` receipt, an empty receipt set, a receipt journaled before the latest + `reconciled` record, a receipt for an environment the journal never admitted, or a `reconciled` + record written before `settledSeq` existed (0.230.0–0.233.1) leaves the slot open exactly as + before. Nothing is invented for those journals. +- A resumed cursor now starts past every open node's `settledSeq`, so a resumed scope can never + mint an open node's seq for another settlement; a journal that already closes that seq fails the + resume with `RuntimeRunStateError` and writes nothing. +- The refund at the reconcile is unchanged. `terminalDownEvent`, `settledNodeEvidence` and the + release payload move to `supervise/terminal-record.ts`, the one module both the live sweep and + the resume heal import. + ## 0.233.1 Support Sandbox 0.41 through the published peer range. diff --git a/api-surface.json b/api-surface.json index f0ac24c8..c87ac1b2 100644 --- a/api-surface.json +++ b/api-surface.json @@ -1217,7 +1217,7 @@ "ShapeRegistry": "type b567a5be55ec", "Shell": "type c676fe970f79", "ShotSpec": "type bf645fd74234", - "SpawnEvent": "type f9c5663b41cd", + "SpawnEvent": "type 1497e8b7c300", "SpawnForest": "type 7308f34da226", "SpawnForestEvent": "type 230e369c6548", "SpawnForestInDoubtNode": "type 4e4bddd6a7d9", diff --git a/docs/api/primitive-catalog.md b/docs/api/primitive-catalog.md index 48a1e3b3..b3d6c0b2 100644 --- a/docs/api/primitive-catalog.md +++ b/docs/api/primitive-catalog.md @@ -7,7 +7,7 @@ # Primitive catalog — the never-stale anti-reinvention inventory -> **GENERATED** from `@tangle-network/agent-runtime@0.233.1` and `@tangle-network/agent-eval@0.182.0` by `scripts/gen-primitive-catalog.mjs`. Do NOT hand-edit — run `pnpm run docs:api`. This is the mechanical companion to the JUDGMENT in `canonical-api.md` (§2 decision table + §1.5 AgentProfile law): that doc says WHICH primitive to reach for and what NOT to build; this catalog proves WHAT exists. Per-symbol signatures + `file:line` live in the per-module pages under `docs/api/`. +> **GENERATED** from `@tangle-network/agent-runtime@0.234.0` and `@tangle-network/agent-eval@0.182.0` by `scripts/gen-primitive-catalog.mjs`. Do NOT hand-edit — run `pnpm run docs:api`. This is the mechanical companion to the JUDGMENT in `canonical-api.md` (§2 decision table + §1.5 AgentProfile law): that doc says WHICH primitive to reach for and what NOT to build; this catalog proves WHAT exists. Per-symbol signatures + `file:line` live in the per-module pages under `docs/api/`. ## 1. agent-runtime — own public surface diff --git a/docs/api/runtime.md b/docs/api/runtime.md index 4291db76..7acd26e8 100644 --- a/docs/api/runtime.md +++ b/docs/api/runtime.md @@ -22820,7 +22820,9 @@ the paid execution inside it. slot with a terminal record marked `retainedExecution: 'released'`, so `spendGaps` names it `unreported` (a floor) rather than `never-settled` (a ceiling) and `fleetYield.releasedUnrecovered` counts it; a refused release leaves the slot open and - the node in `teardownUnconfirmed`. + the node in `teardownUnconfirmed`. A process that dies between the last `destroyed: true` + receipt and that record leaves the slot open only until the next resume, whose + `healReleasedSlots` writes the identical record from the `reconciled` record. - `'keep'`: a later process may resume this run, so the environments stay for its recovery. Default: `'keep'` when `resume` is true (a durable run a later process may continue), else @@ -28747,7 +28749,7 @@ Epoch ms parsed from the durable settlement/cancellation record when available. ### SpawnEvent -> **SpawnEvent** = \{ `kind`: `"spawned"`; `id`: [`NodeId`](#nodeid-6); `parent?`: [`NodeId`](#nodeid-6); `label`: `string`; `key?`: `string`; `assignmentId?`: `string`; `budget`: [`Budget`](#budget-18); `runtime`: [`Runtime`](#runtime-7); `ownedTreeRoot?`: [`NodeId`](#nodeid-6); `identity?`: [`NodeExecutionIdentity`](#nodeexecutionidentity); `profileRef?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-input"`; `id`: [`NodeId`](#nodeid-6); `taskRef`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-admitted"`; `id`: [`NodeId`](#nodeid-6); `admission`: [`RetainedRunAdmission`](#retainedrunadmission); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-result"`; `outcome?`: `Pick`\<`AgentTurnResult`, `"success"` \| `"error"`\>; `id`: [`NodeId`](#nodeid-6); `outRef`: `string`; `spent`: [`Spend`](#spend); `verdict?`: `DefaultVerdict`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-bound"`; `id`: [`NodeId`](#nodeid-6); `binding`: [`ExecutionBindingReceipt`](#executionbindingreceipt); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"materialized"`; `id`: [`NodeId`](#nodeid-6); `receipt`: [`ProfileMaterializationReceipt`](#profilematerializationreceipt); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"settled"`; `id`: [`NodeId`](#nodeid-6); `status`: `"done"` \| `"down"`; `outRef?`: `string`; `verdict?`: `DefaultVerdict`; `spent`: [`Spend`](#spend); `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `infra?`: `boolean`; `reason?`: `string`; `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `retainedExecution?`: `Extract`\<[`RetainedExecutionState`](#retainedexecutionstate), `"released"`\>; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"cancelled"`; `id`: [`NodeId`](#nodeid-6); `reason`: `string`; `source?`: `string`; `infra?`: `boolean`; `spent?`: [`Spend`](#spend); `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `outRef?`: `string`; `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `retainedExecution?`: `Extract`\<[`RetainedExecutionState`](#retainedexecutionstate), `"released"`\>; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"node-inputs-resolved"`; `id`: [`NodeId`](#nodeid-6); `node`: `string`; `instance`: `string`; `inputRef`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"edge-verdict"`; `id`: [`NodeId`](#nodeid-6); `edge`: `string`; `fired`: `boolean`; `sourceStatus`: `"done"` \| `"down"` \| `"invalid"`; `capped?`: `boolean`; `inputRef?`: `string`; `toInstance?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"join-state"`; `id`: [`NodeId`](#nodeid-6); `node`: `string`; `rule`: `"all"` \| `"any"` \| `"any_failed"` \| `"all_done"`; `satisfiedBy`: `ReadonlyArray`\<`string`\>; `consumedPending`: `ReadonlyArray`\<`string`\>; `instance`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"waiting"`; `id`: [`NodeId`](#nodeid-6); `parent?`: [`NodeId`](#nodeid-6); `label`: `string`; `spec`: [`WaitSpec`](#waitspec); `armedAt`: `number`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"woken"`; `id`: [`NodeId`](#nodeid-6); `by`: `"fired"` \| `"timeout"` \| `"cancelled"` \| `"expired"`; `outRef?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"metered"`; `id`: [`NodeId`](#nodeid-6); `spend`: [`Spend`](#spend); `accountingOnly?`: `true`; `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"progress"`; `id`: [`NodeId`](#nodeid-6); `spend`: [`Spend`](#spend); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"reconciled"`; `id`: [`NodeId`](#nodeid-6); `spent`: [`Spend`](#spend); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"teardown-unconfirmed"`; `id`: [`NodeId`](#nodeid-6); `label`: `string`; `runtime`: [`Runtime`](#runtime-7); `status`: [`NodeStatus`](#nodestatus); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"environment-teardown"`; `id`: [`NodeId`](#nodeid-6); `provider`: `string`; `environmentId`: `string`; `destroyed`: `boolean`; `detail?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"edge"`; `id`: [`NodeId`](#nodeid-6); `edge`: \{ `kind`: `"delegates"` \| `"analyzes"` \| `"data"`; `from`: `string`; `to`: `string`; `directive?`: `string`; `port?`: `string`; \}; `traversal`: `number`; `outcome`: `"delivered"` \| `"stripped"` \| `"empty"` \| `"unpropagated"`; `continuity?`: `"fresh"` \| `"resume"` \| `"steer"`; `bytes`: `number`; `reason?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"trace-unpropagated"`; `id`: [`NodeId`](#nodeid-6); `expectedTraceId`: `string`; `backend`: `string`; `reason`: `"no-env-channel"` \| `"no-worker-process"` \| `"caller-omitted"`; `seq`: `number`; `at`: `string`; \} +> **SpawnEvent** = \{ `kind`: `"spawned"`; `id`: [`NodeId`](#nodeid-6); `parent?`: [`NodeId`](#nodeid-6); `label`: `string`; `key?`: `string`; `assignmentId?`: `string`; `budget`: [`Budget`](#budget-18); `runtime`: [`Runtime`](#runtime-7); `ownedTreeRoot?`: [`NodeId`](#nodeid-6); `identity?`: [`NodeExecutionIdentity`](#nodeexecutionidentity); `profileRef?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-input"`; `id`: [`NodeId`](#nodeid-6); `taskRef`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-admitted"`; `id`: [`NodeId`](#nodeid-6); `admission`: [`RetainedRunAdmission`](#retainedrunadmission); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-result"`; `outcome?`: `Pick`\<`AgentTurnResult`, `"success"` \| `"error"`\>; `id`: [`NodeId`](#nodeid-6); `outRef`: `string`; `spent`: [`Spend`](#spend); `verdict?`: `DefaultVerdict`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"execution-bound"`; `id`: [`NodeId`](#nodeid-6); `binding`: [`ExecutionBindingReceipt`](#executionbindingreceipt); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"materialized"`; `id`: [`NodeId`](#nodeid-6); `receipt`: [`ProfileMaterializationReceipt`](#profilematerializationreceipt); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"settled"`; `id`: [`NodeId`](#nodeid-6); `status`: `"done"` \| `"down"`; `outRef?`: `string`; `verdict?`: `DefaultVerdict`; `spent`: [`Spend`](#spend); `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `infra?`: `boolean`; `reason?`: `string`; `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `retainedExecution?`: `Extract`\<[`RetainedExecutionState`](#retainedexecutionstate), `"released"`\>; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"cancelled"`; `id`: [`NodeId`](#nodeid-6); `reason`: `string`; `source?`: `string`; `infra?`: `boolean`; `spent?`: [`Spend`](#spend); `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `outRef?`: `string`; `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `retainedExecution?`: `Extract`\<[`RetainedExecutionState`](#retainedexecutionstate), `"released"`\>; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"node-inputs-resolved"`; `id`: [`NodeId`](#nodeid-6); `node`: `string`; `instance`: `string`; `inputRef`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"edge-verdict"`; `id`: [`NodeId`](#nodeid-6); `edge`: `string`; `fired`: `boolean`; `sourceStatus`: `"done"` \| `"down"` \| `"invalid"`; `capped?`: `boolean`; `inputRef?`: `string`; `toInstance?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"join-state"`; `id`: [`NodeId`](#nodeid-6); `node`: `string`; `rule`: `"all"` \| `"any"` \| `"any_failed"` \| `"all_done"`; `satisfiedBy`: `ReadonlyArray`\<`string`\>; `consumedPending`: `ReadonlyArray`\<`string`\>; `instance`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"waiting"`; `id`: [`NodeId`](#nodeid-6); `parent?`: [`NodeId`](#nodeid-6); `label`: `string`; `spec`: [`WaitSpec`](#waitspec); `armedAt`: `number`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"woken"`; `id`: [`NodeId`](#nodeid-6); `by`: `"fired"` \| `"timeout"` \| `"cancelled"` \| `"expired"`; `outRef?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"metered"`; `id`: [`NodeId`](#nodeid-6); `spend`: [`Spend`](#spend); `accountingOnly?`: `true`; `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"progress"`; `id`: [`NodeId`](#nodeid-6); `spend`: [`Spend`](#spend); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"reconciled"`; `id`: [`NodeId`](#nodeid-6); `spent`: [`Spend`](#spend); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `settledSeq?`: `number`; `reason?`: `string`; `infra?`: `boolean`; `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `outRef?`: `string`; `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `cancellation?`: \{ `source`: `string`; \}; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"teardown-unconfirmed"`; `id`: [`NodeId`](#nodeid-6); `label`: `string`; `runtime`: [`Runtime`](#runtime-7); `status`: [`NodeStatus`](#nodestatus); `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"environment-teardown"`; `id`: [`NodeId`](#nodeid-6); `provider`: `string`; `environmentId`: `string`; `destroyed`: `boolean`; `detail?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"edge"`; `id`: [`NodeId`](#nodeid-6); `edge`: \{ `kind`: `"delegates"` \| `"analyzes"` \| `"data"`; `from`: `string`; `to`: `string`; `directive?`: `string`; `port?`: `string`; \}; `traversal`: `number`; `outcome`: `"delivered"` \| `"stripped"` \| `"empty"` \| `"unpropagated"`; `continuity?`: `"fresh"` \| `"resume"` \| `"steer"`; `bytes`: `number`; `reason?`: `string`; `seq`: `number`; `at`: `string`; \} \| \{ `kind`: `"trace-unpropagated"`; `id`: [`NodeId`](#nodeid-6); `expectedTraceId`: `string`; `backend`: `string`; `reason`: `"no-env-channel"` \| `"no-worker-process"` \| `"caller-omitted"`; `seq`: `number`; `at`: `string`; \} Journaled spawn-tree events (B1/B2). `seq` is the cursor order; `at` is an ISO timestamp for human inspection only (NOT a replay input). @@ -29052,8 +29054,13 @@ Present when the reconciled spend exceeded the reservation, on either status. > `optional` **retainedExecution?**: `Extract`\<[`RetainedExecutionState`](#retainedexecutionstate), `"released"`\> -Written only by the release sweep, on the same tree, after every `environment-teardown` - receipt for the node reads `destroyed: true` and the executor's own teardown confirmed. +Written by the release sweep in the settling process, on the same tree, after every + `environment-teardown` receipt for the node reads `destroyed: true` and the executor's + own teardown confirmed — or by `healReleasedSlots` on the next resume, when that process + died between the last `destroyed: true` receipt and this record: the same builder, `seq` + and `at`, read from the `reconciled` record, so the bytes are identical either way. A + `destroyed: false` receipt, an empty receipt set, or a `reconciled` record without + `settledSeq` leaves the slot open. `spent` is this node's child-work component of the reconcile the pool committed — for a leaf the streamed floor itself, for a recursive executor its `accounting().reported` split with the remainder on its `metered` records — never the reservation ceiling. @@ -29444,7 +29451,7 @@ without charging the same spend twice. ##### Type Literal -\{ `kind`: `"reconciled"`; `id`: [`NodeId`](#nodeid-6); `spent`: [`Spend`](#spend); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `seq`: `number`; `at`: `string`; \} +\{ `kind`: `"reconciled"`; `id`: [`NodeId`](#nodeid-6); `spent`: [`Spend`](#spend); `harnessTranscript?`: [`HarnessTranscriptEvidence`](#harnesstranscriptevidence); `settledSeq?`: `number`; `reason?`: `string`; `infra?`: `boolean`; `trace?`: [`WorkerTraceEvidence`](#workertraceevidence); `outRef?`: `string`; `providerModel?`: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence); `budgetViolation?`: [`BudgetViolation`](#budgetviolation-3); `cancellation?`: \{ `source`: `string`; \}; `seq`: `number`; `at`: `string`; \} ###### kind @@ -29454,11 +29461,15 @@ A retained child's reservation was reconciled at the child-work floor its execut observed, while its cursor slot stays OPEN so a resume can recover the execution. It stands in for the `settled` record an open node cannot carry: cost readers and a restored pool charge this floor for the node instead of its declared ceiling, and a later `settled` or - `cancelled` record for the same node supersedes it. That record has two writers: a - resumed process's recovered settlement, or the release sweep's terminal record marked - `retainedExecution: 'released'`. A driver's own inference travels on its - `metered` record as on every other path, so `reconciled + metered` is what the pool - committed. Its `seq` lives outside the cursor-uniqueness namespace. + `cancelled` record for the same node supersedes it. That record has three writers: a + resumed process's recovered settlement, the release sweep's terminal record marked + `retainedExecution: 'released'`, or `healReleasedSlots` on the next resume when the + process died between the last `environment-teardown` receipt and that record. The last + two write the same bytes, because both read the settlement from THIS record: it is + literally the settlement it stands in for, minus the cursor position it cannot hold. A + driver's own inference travels on its `metered` record as on every other path, so + `reconciled + metered` is what the pool committed. Its `seq` lives outside the + cursor-uniqueness namespace; `settledSeq` is the cursor seq. ###### id @@ -29477,6 +29488,58 @@ The transcript receipt of a retained-pending child. This record is the ONLY dura #1244 children exactly — dropped mid-run with a live box the capture read. A later terminal record for the node carries the same receipt forward. +###### settledSeq? + +> `optional` **settledSeq?**: `number` + +The cursor seq `next()` stamped on the delivery this floor stands in for — what + `finalizeSettlement` holds as `child.settledSeq` and the release sweep writes the + terminal record under. A resume that heals a crashed sweep reads the seq back from here, + so a record without it cannot be healed. Optional only so journals written before this + field existed remain replayable. + +###### reason? + +> `optional` **reason?**: `string` + +The settlement the driver received, verbatim, as `settled`/`cancelled` carry it. Optional + only so journals written before these fields existed remain replayable. + +###### infra? + +> `optional` **infra?**: `boolean` + +###### trace? + +> `optional` **trace?**: [`WorkerTraceEvidence`](#workertraceevidence) + +###### outRef? + +> `optional` **outRef?**: `string` + +###### providerModel? + +> `optional` **providerModel?**: [`ProviderModelExecutionEvidence`](#providermodelexecutionevidence) + +###### budgetViolation? + +> `optional` **budgetViolation?**: [`BudgetViolation`](#budgetviolation-3) + +The overspend the RETAINED reconcile returned, which the open-slot surfaces withhold + (`materializeTreeView` folds only `spent` from this record); journaled so the released + record carries it without recomputation. + +###### cancellation? + +> `optional` **cancellation?**: `object` + +Present iff the child had a `RunCancellationReason` when it settled; decides `settled` + vs `cancelled` and its `source` on the released record. + +###### cancellation.source + +> `readonly` **source**: `string` + ###### seq > **seq**: `number` @@ -29545,8 +29608,10 @@ One provider environment a settled retained-pending child held, released at root in `detail`, and the node is then also journaled as `teardown-unconfirmed`. When every receipt for a node is `destroyed: true` and the executor confirms teardown, the node's terminal `settled`/`cancelled` record with `retainedExecution: 'released'` follows on - the same tree and closes the cursor slot; a `destroyed: false` receipt or an unconfirmed - teardown leaves the slot open because the environment may still exist. + the same tree and closes the cursor slot; when the settling process dies between this + receipt and that record, `healReleasedSlots` writes the same record on the next resume + from the `reconciled` record. A `destroyed: false` receipt, an empty receipt set, or an + unconfirmed teardown leaves the slot open because the environment may still exist. Informational: replay, `materializeTreeView`, and cost readers skip it, and its `seq` is per node, outside the cursor-uniqueness namespace. @@ -29757,7 +29822,10 @@ path). Not a failure classification — the child is `down` either way. that nothing could be released. - `'released'`: root settlement under `retainedAtSettlement: 'release'` destroyed the environment (executor-confirmed) before any process recovered the execution; this is the - node's terminal record. The pool's own admission fault, if the reconcile raised one, is not + node's terminal record, written by the release sweep in the settling process or by + `healReleasedSlots` on the next resume when that process died between the last + `destroyed: true` receipt and the record (same builder, `seq` and `at`, from the + `reconciled` record). The pool's own admission fault, if the reconcile raised one, is not on this record. Absent = an ordinary child. The live `Settled` a driver branched on carried `'pending'` where diff --git a/docs/canonical-api.md b/docs/canonical-api.md index 09203fd3..e9999b8d 100644 --- a/docs/canonical-api.md +++ b/docs/canonical-api.md @@ -4,7 +4,7 @@ Generated signatures and the complete export list live in docs/api/. Run pnpm docs:freshness after editing this file. --> -> **Version 0.233.1.** +> **Version 0.234.0.** > [`docs/api/primitive-catalog.md`](./api/primitive-catalog.md) lists every export and import path. > `agent-eval` must satisfy `>=0.182.0 <0.183.0`. > `sandbox` must satisfy `>=0.36.4 <0.42.0`. diff --git a/package.json b/package.json index 2d72a50f..e5dc8f2e 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/agent-runtime", - "version": "0.233.1", + "version": "0.234.0", "description": "Shared task-lifecycle skeleton for agents: a recursive loop kernel for chat turns, one-shot tasks, and multi-attempt loops, with trace capture and eval-gated self-improvement. Domain behavior lives in adapters; scoring and ship-gates in @tangle-network/agent-eval.", "homepage": "https://github.com/tangle-network/agent-runtime#readme", "repository": { diff --git a/src/runtime/supervise/recover-executors.ts b/src/runtime/supervise/recover-executors.ts index c244d61a..b27406ed 100644 --- a/src/runtime/supervise/recover-executors.ts +++ b/src/runtime/supervise/recover-executors.ts @@ -6,12 +6,14 @@ import { import { closesCursorSlot, contentAddress, + loadSpawnForest, materializeTreeView, ownedTreeRootSpawn, pendingWaits, replaySpawnTree, } from '../../durable/spawn-journal' import { RuntimeRunStateError } from '../../errors' +import { notifyRuntimeHookEvent } from '../../runtime-hooks' import { addSpend, zeroSpend } from '../util' import { assertValidSpend, @@ -24,6 +26,7 @@ import { addResourceSpend, withBudgetResources } from './resources' import { prepareRetainedExecutor, type RetainedChildRecovery } from './retained-executor' import type { ScopeArgs } from './scope' import { detachedSnapshot } from './snapshot' +import { releasedChildPayload, terminalDownEvent } from './terminal-record' import { nestedDriverTreeRoot } from './tree-key' import type { Budget, @@ -36,9 +39,155 @@ import type { SupervisorOpts, } from './types' -type ResumeStores = Pick +type ResumeStores = Pick< + SupervisorOpts, + 'runId' | 'journal' | 'blobs' | 'recoverExecutor' | 'hooks' +> type Spawned = Extract type RecordedResult = Extract +type Reconciled = Extract +type TeardownReceipt = Extract + +/** The environment id the node's latest durable admission names — the executor's own rule + * (`admittedEnvironmentId` in environment-provider.ts), read off the journal. */ +function journaledEnvironmentId(owned: ReadonlyArray): string | undefined { + for (const event of [...owned].reverse()) { + if (event.kind !== 'execution-admitted') continue + if (event.admission.phase === 'dispatched') return event.admission.controlRef.environmentId + if (event.admission.phase === 'environment') return event.admission.environmentId + } + return undefined +} + +/** + * Close the cursor slot of every node the release sweep destroyed but never recorded: the + * settling process died between the last `environment-teardown` receipt and the terminal record + * (the 0.233.0 crash window). The record is the sweep's own — the same builder, the cursor seq + * `next()` stamped on the delivery and the settlement instant, all read back from the node's + * latest `reconciled` record — so a healed journal replays as one the sweep completed. + * + * The gate is fail-closed, every clause required, because the heal cannot re-ask the executor + * to confirm teardown the way the sweep does: + * - the latest `reconciled` record carries `settledSeq`, `reason`, `infra` and `trace` (a record + * written before those fields existed names no cursor position and is not healed); + * - at least one receipt sits after that record and every such receipt reads `destroyed: true` + * (an empty set must not close the slot vacuously, and a receipt from an earlier process + * belongs to an earlier settlement); + * - the environment the latest admission names is among those receipts, so a receipt for an + * environment the journal never admitted, or a re-admitted node whose receipt names the + * earlier environment, is left open where the live sweep would have closed it. + * A node with an `execution-result` is left to the recorded-result branch of + * {@link prepareInterruptedExecutors}, which settles it from the durable blob and takes + * precedence. The walk is over the whole forest because a nested manager settled on the ordinary + * path is never restored, so a per-tree heal inside the resume of its tree would never reach its + * grandchild. Returns the number of records written. + */ +export async function healReleasedSlots( + opts: ResumeStores, + signal: AbortSignal, + now: () => number, +): Promise { + const forest = await loadSpawnForest(opts.journal, opts.runId) + let healed = 0 + for (const tree of forest.trees) { + signal.throwIfAborted() + const events = tree.events + const closed = new Set(events.filter(closesCursorSlot).map((event) => event.id)) + const recorded = new Set( + events.flatMap((event) => (event.kind === 'execution-result' ? [event.id] : [])), + ) + for (const spawned of events) { + if (spawned.kind !== 'spawned' || spawned.parent === undefined) continue + if (closed.has(spawned.id) || recorded.has(spawned.id)) continue + const owned = events.filter((event) => event.id === spawned.id) + const floor = owned.reduce( + (latest, event) => + event.kind === 'reconciled' && (latest === undefined || event.seq > latest.seq) + ? event + : latest, + undefined, + ) + if ( + floor?.settledSeq === undefined || + floor.reason === undefined || + floor.infra === undefined || + floor.trace === undefined + ) + continue + const receipts = owned + .slice(owned.indexOf(floor) + 1) + .filter((event): event is TeardownReceipt => event.kind === 'environment-teardown') + if (receipts.length === 0 || !receipts.every((receipt) => receipt.destroyed)) continue + const held = journaledEnvironmentId(owned) + if (held === undefined || !receipts.some((receipt) => receipt.environmentId === held)) + continue + const settledSeq = floor.settledSeq + // The journal's own duplicate-cursor guard is the backstop; this names the node first, and + // writes nothing. A fixed process cannot produce it: the resumed cursor starts past every + // open node's `settledSeq` (see `maxCursorSeq` below). + if (events.some((event) => closesCursorSlot(event) && event.seq === settledSeq)) { + throw new RuntimeRunStateError( + `retained child '${spawned.id}' cannot be released at cursor seq ${settledSeq}: tree '${tree.root}' already closes that seq`, + ) + } + const subject = { + id: spawned.id, + spent: floor.spent, + ...(floor.budgetViolation ? { budgetViolation: floor.budgetViolation } : {}), + ...(floor.cancellation ? { cancellationReason: floor.cancellation } : {}), + } + const settlement = { + kind: 'down' as const, + reason: floor.reason, + infra: floor.infra, + trace: floor.trace, + ...(floor.outRef ? { outRef: floor.outRef } : {}), + ...(floor.providerModel ? { providerModel: floor.providerModel } : {}), + ...(floor.harnessTranscript ? { harnessTranscript: floor.harnessTranscript } : {}), + } + await opts.journal.appendEvent( + tree.root, + terminalDownEvent(subject, settlement, settledSeq, floor.at, 'released'), + ) + healed += 1 + // The sweep's `agent.child` on the resumed stream, so a projection whose node state comes + // only from that target flips the node to released as it does live. `startedAt` is the + // runtime's own rule for a recovered child (the spawn instant), `releasedAt` the last + // receipt's instant; the journal record above is the byte-identical surface. + const materialized = owned.find((event) => event.kind === 'materialized') + const lastReceipt = receipts[receipts.length - 1]! + notifyRuntimeHookEvent( + opts.hooks, + { + id: `${spawned.id}:released`, + runId: tree.root, + target: 'agent.child', + phase: 'after', + timestamp: now(), + stepIndex: settledSeq, + parentId: tree.ownerNodeId ?? opts.runId, + payload: releasedChildPayload( + { + ...subject, + runtime: materialized?.receipt.runtime ?? spawned.runtime, + startedAt: Date.parse(spawned.at), + ...(materialized === undefined ? {} : { materialization: materialized.receipt }), + executionBindings: owned.flatMap((event) => + event.kind === 'execution-bound' ? [event.binding] : [], + ), + ...(floor.providerModel ? { providerModel: floor.providerModel } : {}), + }, + settlement, + Date.parse(floor.at), + Date.parse(lastReceipt.at), + ), + }, + { signal }, + ) + } + } + return healed +} /** Validate durable inputs without waiting for work that may need its resumed manager. */ export async function prepareInterruptedExecutors( @@ -163,9 +312,9 @@ export async function prepareInterruptedExecutors( factory: prepared?.factory ?? opts.recoverExecutor, }) } - let seq = - events.reduce((max, event) => (closesCursorSlot(event) ? Math.max(max, event.seq) : max), -1) + - 1 + // Past every closed AND reserved seq: an open node's `settledSeq` is where its terminal record + // lands, and a recovered result written there would be the collision the heal refuses. + let seq = reservedCursorFloor(events) + 1 for (const { result, fault, budgetViolation } of accepted) { signal.throwIfAborted() const failureReason = executorFailureReason(result) @@ -301,6 +450,29 @@ export function maxSeqOf(events: SpawnEvent[], pred: (ev: SpawnEvent) => boolean return max } +/** + * The highest cursor seq any record in the journal already owns, closed OR reserved. + * + * A closed slot owns its seq outright. An open retained node owns the `settledSeq` its + * `reconciled` record carries: that is the seq the driver already branched on, and the seq the + * heal (or a later release sweep) writes the terminal record under. Every writer that mints a + * new cursor seq in a resumed process — the recorded-result settlements below and the resumed + * scope's own cursor — must start past BOTH, or a recovered result lands on a seq an open node + * has reserved and the heal can only ever collide. One function so no writer can drift. + */ +export function reservedCursorFloor(events: SpawnEvent[]): number { + return Math.max( + maxSeqOf(events, closesCursorSlot), + events.reduce( + (max, event) => + event.kind === 'reconciled' && event.settledSeq !== undefined + ? Math.max(max, event.settledSeq) + : max, + -1, + ), + ) +} + /** Per-channel sum over a journaled event list: `settled` = spawned-child work (reconciled), plus * the reconciled floor of every node still open, plus the declared ceiling of every open node * nothing reconciled; `metered` = driver inference (re-homed up the tree, so a single root-tree @@ -386,6 +558,13 @@ export async function prepareScopeResume( now: () => number, parentId = opts.runId, ): Promise<{ resumeFrom: ScopeResumeState; poolRestore: BudgetPoolRestore }> { + // The root resume walks the whole forest once; the nested restore, whose `parentId` is the + // manager node, does not walk again. A healed node is then terminal to everything below: + // never interrupted, never a recovery, charged its floor as `settled.spent` instead of the + // reconciled floor (the same object), and replayed at the driver's own seq. + if (parentId === opts.runId && (await healReleasedSlots(opts, signal, now)) > 0) { + events = (await opts.journal.loadTree(opts.runId)) ?? events + } const prepared = await prepareInterruptedExecutors(opts, events, signal, now, parentId) const prior = prepared.events const recovering = new Set(prepared.recoveries.map((item) => item.spawned.id)) @@ -419,7 +598,9 @@ export async function prepareScopeResume( settled, view: materializeTreeView(prior), maxSpawnOrdinal: maxSeqOf(prior, (event) => event.kind === 'spawned'), - maxCursorSeq: maxSeqOf(prior, closesCursorSlot), + // An open node's journaled cursor seq is reserved across processes: the resumed scope must + // never mint it for another node, or the heal above could only ever collide. + maxCursorSeq: reservedCursorFloor(prior), maxWaitOrdinal: maxSeqOf(prior, (event) => event.kind === 'waiting'), waits: pendingWaits(prior), keys: keyedAssignments(prior, settled), diff --git a/src/runtime/supervise/scope.ts b/src/runtime/supervise/scope.ts index 8e74a1e7..0d8d8614 100644 --- a/src/runtime/supervise/scope.ts +++ b/src/runtime/supervise/scope.ts @@ -100,6 +100,14 @@ import { releaseScopeRetainedOwnerEnvironment, } from './retained-scope-owner' import { detachedSnapshot } from './snapshot' +import { + type DownSettlement, + releasedChildPayload, + settledNodeEvidence, + settlementFields, + type TerminalDownSubject, + terminalDownEvent, +} from './terminal-record' import { captureWorkerTraceEvidence } from './trace-evidence' import type { TraceSource } from './trace-source' import { runtimeOwnedNestedDriverTreeRoot } from './tree-key' @@ -375,7 +383,9 @@ interface LiveChild { settledAt?: number /** The cursor seq `next()` stamped on this child's settlement. A retained-pending child's * terminal record is written later by the release sweep under THIS seq, so replay yields it - * at the position the driver saw it; the sweep never mints a new `cursorSeq` for it. */ + * at the position the driver saw it; the sweep never mints a new `cursorSeq` for it, and + * neither does a resume that heals a crashed sweep — it reads this seq back from the + * reconciled record's `settledSeq`. */ settledSeq?: number /** Mirrors the journal's statement about a retained execution so `makeTreeView` reports it. */ retainedExecution?: RetainedExecutionState @@ -436,7 +446,7 @@ interface LiveChild { /** A child's terminal settlement before the cursor stamps the monotonic `seq`. A wait-state's * `done` carries a `WaitOutcome` as its `out` and a zero `spent` — waiting is free by type, not * by measurement. */ -type PreSeqSettled = +export type PreSeqSettled = | { kind: 'done' out: unknown @@ -2057,9 +2067,11 @@ export function createScope(args: ScopeArgs): Scope { // `cleanupConfirmed` the barrier reads: a `destroyed: false` receipt, an executor throw, an // environment-less `[]` answer or a refused `confirmTeardown` leave the slot open, because // the environment may still exist and `[].every` would otherwise close it vacuously. A - // crash between the receipt and this record leaves the slot open too; nothing below can - // run without the executor's confirmation in hand. A manager settled on the ordinary path - // already has its record and never passes `recoveryPending`. + // crash between the last receipt and this record no longer strands the slot: the + // reconciled record carries this settlement and the cursor seq, and `healReleasedSlots` + // (recover-executors.ts) writes this same record from it on the next resume, under the + // same seq. A manager settled on the ordinary path already has its record and never + // passes `recoveryPending`. if ( child.cleanupConfirmed && child.recoveryPending === true && @@ -2095,17 +2107,7 @@ export function createScope(args: ScopeArgs): Scope { timestamp: releasedAt, stepIndex: child.settledSeq, parentId: args.parentId, - payload: { - childId: child.id, - status: 'down', - retainedExecution: 'released', - releasedAt, - ...(child.resolved.outRef === undefined ? {} : { outRef: child.resolved.outRef }), - reason: child.resolved.reason, - infra: child.resolved.infra, - spent: child.spent, - ...settledNodeEvidence(child, { ...child.resolved, metered: undefined }, settledAt), - }, + payload: releasedChildPayload(child, child.resolved, settledAt, releasedAt), }, { signal: args.signal }, ) @@ -2571,21 +2573,27 @@ async function appendSettlementMetering( /** An open node's reconciled floor has the same per-child sequence discipline as its metering: * outside the cursor namespace, monotonic per node, so a node reconciled once per process (a - * retained failure, a recovery, a second retained failure) keeps its records ordered. */ + * retained failure, a recovery, a second retained failure) keeps its records ordered. It carries + * the whole settlement and the cursor seq (`settledSeq`) beside the floor, through the same + * field spread the terminal record uses, so a resume that finds the release sweep died before + * its record can write that record byte-for-byte instead of inventing one. */ async function appendReconciledFloor( journal: SpawnJournal, root: NodeId, - id: NodeId, - spent: Spend, + subject: TerminalDownSubject, + settlement: DownSettlement, + settledSeq: number, at: string, - harnessTranscript?: HarnessTranscriptEvidence, ): Promise { - const seq = await nextPerNodeSeq(journal, root, 'reconciled', id) + const seq = await nextPerNodeSeq(journal, root, 'reconciled', subject.id) await appendAcknowledged(journal, root, { kind: 'reconciled', - id, - spent, - ...(harnessTranscript ? { harnessTranscript } : {}), + id: subject.id, + ...settlementFields(subject, settlement), + ...(subject.cancellationReason === undefined + ? {} + : { cancellation: { source: subject.cancellationReason.source } }), + settledSeq, seq, at, }) @@ -2684,20 +2692,35 @@ async function finalizeSettlement( // ceiling the pool refunded: measured 2026-09-11, the reported `childWork` carried 4M per // retained child against a metered 10 (#1190). The floor is journaled in the settlement's // place, outside the cursor namespace, so the slot stays open and the ledgers agree. The - // slot closes in exactly two ways: a resume recovers the execution and settles it on the - // ordinary path, or root settlement under `retainedAtSettlement: 'release'` destroys the + // slot closes in exactly three ways: a resume recovers the execution and settles it on the + // ordinary path; root settlement under `retainedAtSettlement: 'release'` destroys the // environment and the release sweep (`retainedReleasers`, above) writes this settlement as - // the terminal record with `retainedExecution: 'released'`, under the seq stamped here. + // the terminal record with `retainedExecution: 'released'`, under the seq stamped here; or + // the next resume's `healReleasedSlots` writes that same record from this reconciled record + // and the receipts, when the sweep's process died between them. The reconciled record + // therefore carries the settlement and `seq` in full: `spent` is the floor the pool committed + // (`live.spent` at the retained branch, the same object `child.spent` holds), the withheld + // overspend is `retainedViolation`, and `cancellationReason` is final because + // `recordCancellation` only assigns before `executorDone`. else { child.retainedExecution = 'pending' if (settlement.reconciled !== undefined) await appendReconciledFloor( args.journal, args.root, - child.id, - settlement.reconciled, + { + id: child.id, + spent: settlement.reconciled, + ...(child.retainedViolation === undefined + ? {} + : { budgetViolation: child.retainedViolation }), + ...(child.cancellationReason === undefined + ? {} + : { cancellationReason: child.cancellationReason }), + }, + settlement, + seq, at, - settlement.harnessTranscript, ) } // The in-memory down and the first `agent.child` are the only surfaces that exist while the @@ -2800,77 +2823,6 @@ async function finalizeSettlement( } } -/** - * The evidence a settled node carries beyond its status: the receipts that name what actually - * ran, the own-inference spend the parent tree re-homes as a `metered` event, and the wall-clock - * window. One builder feeds BOTH terminal paths, so an observer never sees a `down` node - * described in different terms from a `done` one. Every field is omitted when the fact is - * absent — an unreported receipt must not read as an empty one. Snapshots are detached because - * an observer may serialize them after the live child has moved on. - */ -/** The one builder of a down child's terminal record, for both its writers: the settle path at - * the reconcile and the release sweep closing a retained slot later. Two hand-written literals - * for one record shape is how a field lands on one path and not the other (#1244 was exactly - * that for `harnessTranscript`). `cancelled` keeps its `source` from the child's own - * cancellation reason, so a released cancelled child records the same source it settled with. */ -function terminalDownEvent( - child: LiveChild, - settlement: Extract, - seq: number, - at: string, - retainedExecution?: 'released', -): Extract { - const cancellation = child.cancellationReason - return { - ...(cancellation === undefined - ? { kind: 'settled' as const, status: 'down' as const } - : { kind: 'cancelled' as const, source: cancellation.source }), - id: child.id, - spent: child.spent, - infra: settlement.infra, - reason: settlement.reason, - ...(settlement.outRef ? { outRef: settlement.outRef } : {}), - ...(settlement.providerModel ? { providerModel: settlement.providerModel } : {}), - ...(child.budgetViolation ? { budgetViolation: child.budgetViolation } : {}), - trace: settlement.trace, - ...(settlement.harnessTranscript ? { harnessTranscript: settlement.harnessTranscript } : {}), - ...(retainedExecution === undefined ? {} : { retainedExecution }), - seq, - at, - } -} - -function settledNodeEvidence( - child: LiveChild, - settlement: PreSeqSettled, - settledAt: number, -): Record { - return { - runtime: child.runtime, - startedAt: child.startedAt, - settledAt, - ...(settlement.metered ? { metered: detachedSnapshot(settlement.metered, 'metered') } : {}), - ...(child.providerModel - ? { providerModel: detachedSnapshot(child.providerModel, 'provider model evidence') } - : {}), - ...(child.materialization - ? { materialization: detachedSnapshot(child.materialization, 'materialization receipt') } - : {}), - ...(child.executionBindings.length > 0 - ? { - executionBindings: detachedSnapshot( - [...child.executionBindings], - 'execution binding receipts', - ), - } - : {}), - ...(child.budgetViolation - ? { budgetViolation: detachedSnapshot(child.budgetViolation, 'budget violation') } - : {}), - trace: detachedSnapshot(settlement.trace, 'worker trace evidence'), - } -} - /** Journal a wait-state's settlement as a `woken` event and project it onto the same `Settled` * the driver branches on. `by` names WHY it woke — `fired` / `timeout` / `cancelled` — which is * the fact a resumed reader needs and cannot recover from a payload it may never fetch. */ diff --git a/src/runtime/supervise/terminal-record.ts b/src/runtime/supervise/terminal-record.ts new file mode 100644 index 00000000..3f96cb35 --- /dev/null +++ b/src/runtime/supervise/terminal-record.ts @@ -0,0 +1,159 @@ +/** + * The one builder of a down child's terminal record and of the evidence an observer sees beside + * it, shared by every writer: the settle path at the reconcile, the release sweep closing a + * retained slot at root settlement, and `healReleasedSlots` closing that slot on the next resume + * when the settling process died between the last `environment-teardown` receipt and the record. + * + * This is its own module rather than a corner of scope.ts because scope.ts value-imports + * recover-executors.ts, so a value import in the other direction would be a runtime ESM cycle + * that biome does not flag. Every import of './scope' here is type-only and erased. + * + * Two hand-written literals for one record shape is how a field lands on one path and not the + * other (#1244 was exactly that for `harnessTranscript`), so the subject is a structural type a + * live `LiveChild` satisfies unchanged and a journaled `reconciled` record can be projected onto. + */ + +import type { PreSeqSettled } from './scope' +import { detachedSnapshot } from './snapshot' +import type { + BudgetViolation, + ExecutionBindingReceipt, + NodeId, + NodeSnapshot, + ProfileMaterializationReceipt, + ProviderModelExecutionEvidence, + SpawnEvent, + Spend, +} from './types' + +export type DownSettlement = Extract + +/** What the terminal record needs from the node beyond its settlement. `RunCancellationReason` + * has `readonly source: string`, so a live child's `cancellationReason` fits as it is. */ +export type TerminalDownSubject = { + readonly id: NodeId + readonly spent: Spend + readonly budgetViolation?: BudgetViolation + readonly cancellationReason?: { readonly source: string } +} + +/** The settlement fields every writer copies, in the terminal record's key order. The + * `reconciled` record is built from this same spread so it and the terminal record can never + * disagree on a field. */ +export function settlementFields( + subject: TerminalDownSubject, + settlement: DownSettlement, +): { + spent: Spend + infra: boolean + reason: string + outRef?: string + providerModel?: ProviderModelExecutionEvidence + budgetViolation?: BudgetViolation + trace: DownSettlement['trace'] + harnessTranscript?: DownSettlement['harnessTranscript'] +} { + return { + spent: subject.spent, + infra: settlement.infra, + reason: settlement.reason, + ...(settlement.outRef ? { outRef: settlement.outRef } : {}), + ...(settlement.providerModel ? { providerModel: settlement.providerModel } : {}), + ...(subject.budgetViolation ? { budgetViolation: subject.budgetViolation } : {}), + trace: settlement.trace, + ...(settlement.harnessTranscript ? { harnessTranscript: settlement.harnessTranscript } : {}), + } +} + +/** `cancelled` keeps its `source` from the subject's own cancellation reason, so a released + * cancelled child records the same source it settled with. */ +export function terminalDownEvent( + subject: TerminalDownSubject, + settlement: DownSettlement, + seq: number, + at: string, + retainedExecution?: 'released', +): Extract { + const cancellation = subject.cancellationReason + return { + ...(cancellation === undefined + ? { kind: 'settled' as const, status: 'down' as const } + : { kind: 'cancelled' as const, source: cancellation.source }), + id: subject.id, + ...settlementFields(subject, settlement), + ...(retainedExecution === undefined ? {} : { retainedExecution }), + seq, + at, + } +} + +/** What the observer evidence needs from the node beyond its settlement. */ +export type SettledEvidenceSubject = { + readonly runtime: NodeSnapshot['runtime'] + readonly startedAt: number + readonly providerModel?: ProviderModelExecutionEvidence + readonly materialization?: ProfileMaterializationReceipt + readonly executionBindings: ReadonlyArray + readonly budgetViolation?: BudgetViolation +} + +/** + * The evidence a settled node carries beyond its status: the receipts that name what actually + * ran, the own-inference spend the parent tree re-homes as a `metered` event, and the wall-clock + * window. One builder feeds BOTH terminal paths, so an observer never sees a `down` node + * described in different terms from a `done` one. Every field is omitted when the fact is + * absent — an unreported receipt must not read as an empty one. Snapshots are detached because + * an observer may serialize them after the live child has moved on. + */ +export function settledNodeEvidence( + subject: SettledEvidenceSubject, + settlement: PreSeqSettled, + settledAt: number, +): Record { + return { + runtime: subject.runtime, + startedAt: subject.startedAt, + settledAt, + ...(settlement.metered ? { metered: detachedSnapshot(settlement.metered, 'metered') } : {}), + ...(subject.providerModel + ? { providerModel: detachedSnapshot(subject.providerModel, 'provider model evidence') } + : {}), + ...(subject.materialization + ? { materialization: detachedSnapshot(subject.materialization, 'materialization receipt') } + : {}), + ...(subject.executionBindings.length > 0 + ? { + executionBindings: detachedSnapshot( + [...subject.executionBindings], + 'execution binding receipts', + ), + } + : {}), + ...(subject.budgetViolation + ? { budgetViolation: detachedSnapshot(subject.budgetViolation, 'budget violation') } + : {}), + trace: detachedSnapshot(settlement.trace, 'worker trace evidence'), + } +} + +/** The `agent.child` payload of the release: the settlement restated as released, `settledAt` + * kept at the settlement instant and `metered` omitted so the driver's inference is not summed + * twice. The sweep and the resume heal emit it from the same builder. */ +export function releasedChildPayload( + subject: TerminalDownSubject & SettledEvidenceSubject, + settlement: DownSettlement, + settledAt: number, + releasedAt: number, +): Record { + return { + childId: subject.id, + status: 'down', + retainedExecution: 'released', + releasedAt, + ...(settlement.outRef === undefined ? {} : { outRef: settlement.outRef }), + reason: settlement.reason, + infra: settlement.infra, + spent: subject.spent, + ...settledNodeEvidence(subject, { ...settlement, metered: undefined }, settledAt), + } +} diff --git a/src/runtime/supervise/types.ts b/src/runtime/supervise/types.ts index 8ca72110..2e7b2fa3 100644 --- a/src/runtime/supervise/types.ts +++ b/src/runtime/supervise/types.ts @@ -1359,8 +1359,13 @@ export type SpawnEvent = harnessTranscript?: HarnessTranscriptEvidence /** Present when the reconciled spend exceeded the reservation, on either status. */ budgetViolation?: BudgetViolation - /** Written only by the release sweep, on the same tree, after every `environment-teardown` - * receipt for the node reads `destroyed: true` and the executor's own teardown confirmed. + /** Written by the release sweep in the settling process, on the same tree, after every + * `environment-teardown` receipt for the node reads `destroyed: true` and the executor's + * own teardown confirmed — or by `healReleasedSlots` on the next resume, when that process + * died between the last `destroyed: true` receipt and this record: the same builder, `seq` + * and `at`, read from the `reconciled` record, so the bytes are identical either way. A + * `destroyed: false` receipt, an empty receipt set, or a `reconciled` record without + * `settledSeq` leaves the slot open. * `spent` is this node's child-work component of the reconcile the pool committed — for * a leaf the streamed floor itself, for a recursive executor its `accounting().reported` * split with the remainder on its `metered` records — never the reservation ceiling. @@ -1501,11 +1506,15 @@ export type SpawnEvent = * observed, while its cursor slot stays OPEN so a resume can recover the execution. It stands * in for the `settled` record an open node cannot carry: cost readers and a restored pool * charge this floor for the node instead of its declared ceiling, and a later `settled` or - * `cancelled` record for the same node supersedes it. That record has two writers: a - * resumed process's recovered settlement, or the release sweep's terminal record marked - * `retainedExecution: 'released'`. A driver's own inference travels on its - * `metered` record as on every other path, so `reconciled + metered` is what the pool - * committed. Its `seq` lives outside the cursor-uniqueness namespace. */ + * `cancelled` record for the same node supersedes it. That record has three writers: a + * resumed process's recovered settlement, the release sweep's terminal record marked + * `retainedExecution: 'released'`, or `healReleasedSlots` on the next resume when the + * process died between the last `environment-teardown` receipt and that record. The last + * two write the same bytes, because both read the settlement from THIS record: it is + * literally the settlement it stands in for, minus the cursor position it cannot hold. A + * driver's own inference travels on its `metered` record as on every other path, so + * `reconciled + metered` is what the pool committed. Its `seq` lives outside the + * cursor-uniqueness namespace; `settledSeq` is the cursor seq. */ kind: 'reconciled' id: NodeId spent: Spend @@ -1514,6 +1523,26 @@ export type SpawnEvent = * #1244 children exactly — dropped mid-run with a live box the capture read. A later * terminal record for the node carries the same receipt forward. */ harnessTranscript?: HarnessTranscriptEvidence + /** The cursor seq `next()` stamped on the delivery this floor stands in for — what + * `finalizeSettlement` holds as `child.settledSeq` and the release sweep writes the + * terminal record under. A resume that heals a crashed sweep reads the seq back from here, + * so a record without it cannot be healed. Optional only so journals written before this + * field existed remain replayable. */ + settledSeq?: number + /** The settlement the driver received, verbatim, as `settled`/`cancelled` carry it. Optional + * only so journals written before these fields existed remain replayable. */ + reason?: string + infra?: boolean + trace?: WorkerTraceEvidence + outRef?: string + providerModel?: ProviderModelExecutionEvidence + /** The overspend the RETAINED reconcile returned, which the open-slot surfaces withhold + * (`materializeTreeView` folds only `spent` from this record); journaled so the released + * record carries it without recomputation. */ + budgetViolation?: BudgetViolation + /** Present iff the child had a `RunCancellationReason` when it settled; decides `settled` + * vs `cancelled` and its `source` on the released record. */ + cancellation?: { readonly source: string } seq: number at: string } @@ -1541,8 +1570,10 @@ export type SpawnEvent = * in `detail`, and the node is then also journaled as `teardown-unconfirmed`. When every * receipt for a node is `destroyed: true` and the executor confirms teardown, the node's * terminal `settled`/`cancelled` record with `retainedExecution: 'released'` follows on - * the same tree and closes the cursor slot; a `destroyed: false` receipt or an unconfirmed - * teardown leaves the slot open because the environment may still exist. + * the same tree and closes the cursor slot; when the settling process dies between this + * receipt and that record, `healReleasedSlots` writes the same record on the next resume + * from the `reconciled` record. A `destroyed: false` receipt, an empty receipt set, or an + * unconfirmed teardown leaves the slot open because the environment may still exist. * Informational: replay, `materializeTreeView`, and cost readers skip it, and its `seq` is * per node, outside the cursor-uniqueness namespace. */ kind: 'environment-teardown' @@ -1706,7 +1737,9 @@ export interface SupervisorOpts { * slot with a terminal record marked `retainedExecution: 'released'`, so `spendGaps` names * it `unreported` (a floor) rather than `never-settled` (a ceiling) and * `fleetYield.releasedUnrecovered` counts it; a refused release leaves the slot open and - * the node in `teardownUnconfirmed`. + * the node in `teardownUnconfirmed`. A process that dies between the last `destroyed: true` + * receipt and that record leaves the slot open only until the next resume, whose + * `healReleasedSlots` writes the identical record from the `reconciled` record. * - `'keep'`: a later process may resume this run, so the environments stay for its recovery. * * Default: `'keep'` when `resume` is true (a durable run a later process may continue), else @@ -1834,7 +1867,10 @@ export interface SpendGap { * that nothing could be released. * - `'released'`: root settlement under `retainedAtSettlement: 'release'` destroyed the * environment (executor-confirmed) before any process recovered the execution; this is the - * node's terminal record. The pool's own admission fault, if the reconcile raised one, is not + * node's terminal record, written by the release sweep in the settling process or by + * `healReleasedSlots` on the next resume when that process died between the last + * `destroyed: true` receipt and the record (same builder, `seq` and `at`, from the + * `reconciled` record). The pool's own admission fault, if the reconcile raised one, is not * on this record. * * Absent = an ordinary child. The live `Settled` a driver branched on carried `'pending'` where diff --git a/src/testing/fixtures/agent-improvement-proposal.json b/src/testing/fixtures/agent-improvement-proposal.json index e2ae9bb9..a73b5484 100644 --- a/src/testing/fixtures/agent-improvement-proposal.json +++ b/src/testing/fixtures/agent-improvement-proposal.json @@ -1,6 +1,6 @@ { "changedSurfaces": ["prompt"], - "digest": "sha256:fcda921d124e3a3c3ffd35d6c762f2863dcc29960b7098b9ca889a4a3842b79d", + "digest": "sha256:a58a7ee58716613986d3256e2cb47e6e64718e2f356ee86bf4932f194424eadd", "evaluation": { "decision": { "contributingChecks": [ @@ -4882,7 +4882,7 @@ ], "metadata": { "fixture": "agent-improvement-proposal", - "runtimeVersion": "0.233.1" + "runtimeVersion": "0.234.0" }, "objectives": [ { @@ -4993,8 +4993,8 @@ "baselineContentHash": "sha256:5c21ee53e513fc604cb09754e21c392b24a424da0ef37dbf8f1ee4a8a0b08f09", "candidateContentHash": "sha256:60fcbb1c728194bd51d7d19cb732d1c3f1881dce7e0a6266b41c8b98cfd65693", "kind": "agent-eval-loop", - "recordDigest": "sha256:b13fd55539afd15bdff24e2b258fb8793d103a629511f433758c32eeedc68182", - "runId": "agent-runtime-0.233.1-proposal-fixture", + "recordDigest": "sha256:b3d9230879ed4ffaa0e60253ed1868db36bda7913b479ae5af420d005bdde5c8", + "runId": "agent-runtime-0.234.0-proposal-fixture", "schema": "agent-candidate-experiment" } }, @@ -5021,5 +5021,5 @@ ], "kind": "agent-improvement-proposal", "proposedAt": "2026-07-10T01:00:00.000Z", - "runId": "agent-runtime-0.233.1-proposal-fixture" + "runId": "agent-runtime-0.234.0-proposal-fixture" } diff --git a/src/testing/fixtures/agent-profile-improvement-proposal.json b/src/testing/fixtures/agent-profile-improvement-proposal.json index 7eb2b161..271f8f14 100644 --- a/src/testing/fixtures/agent-profile-improvement-proposal.json +++ b/src/testing/fixtures/agent-profile-improvement-proposal.json @@ -1,6 +1,6 @@ { "changedSurfaces": ["prompt", "skills"], - "digest": "sha256:820acd9c859c559efd8d38971a9b54772e4198d96f4cba32a094cf9859476699", + "digest": "sha256:59878d6e5cf517ac69c6432f0c5078756c4b40be4a222af55661bd411ab68e1b", "evaluation": { "decision": { "contributingChecks": [ @@ -1715,7 +1715,7 @@ ], "metadata": { "fixture": "agent-profile-improvement-proposal", - "runtimeVersion": "0.233.1" + "runtimeVersion": "0.234.0" }, "objectives": [ { @@ -1826,7 +1826,7 @@ "baselineContentHash": "sha256:21c495a37c418c10bde64fbaa188beddeed31f1f051ea60a6a6582a9ee0db704", "candidateContentHash": "sha256:103f77bc8481601eef1ad5fe6ba84a40dffabc3a44f421f8c8559121edab84e9", "kind": "agent-eval-loop", - "recordDigest": "sha256:e8e375b262bf2dbaa895afcf863ffb4ffe09c383988a66aa08e09c017d949809", + "recordDigest": "sha256:c145255a0e299d8876bfa1285bef32c76391634c20c1e3785e25f5c01c71eebe", "runId": "profile-improvement-1", "schema": "agent-profile-improvement-experiment" } diff --git a/tests/kernel/recover-released-slot.test.ts b/tests/kernel/recover-released-slot.test.ts new file mode 100644 index 00000000..37ad827d --- /dev/null +++ b/tests/kernel/recover-released-slot.test.ts @@ -0,0 +1,540 @@ +/** + * The 0.233.0 crash window, healed on resume. The release sweep appends a node's + * `environment-teardown` receipt and then its terminal record; a process that dies between the + * two leaves an open cursor slot beside a destroyed environment, and the next process treated + * that node as interrupted: it tried to recover an executor whose environment was gone, or left + * it open forever and charged the ceiling. + * + * `healReleasedSlots` closes that slot with the sweep's own record. The `reconciled` record now + * carries the settlement and the cursor seq the driver saw, so the heal writes the identical + * bytes at the identical seq — nothing is invented. Every clause of the gate is exercised here on + * hand-built journals, because each one is a way the heal could otherwise close a slot the live + * sweep would have left open. + */ + +import { + canonicalAgentProfileDigest, + canonicalCandidateDigest, +} from '@tangle-network/agent-interface' +import { describe, expect, it } from 'vitest' +import { contentAddress, materializeTreeView } from '../../src/durable/spawn-journal' +import { RuntimeRunStateError } from '../../src/errors' +import type { RetainedRunAdmission } from '../../src/runtime/retained-run-types' +import { + knownExecutionBindingReceipt, + knownMaterializationReceipt, +} from '../../src/runtime/supervise/materialization' +import { + healReleasedSlots, + prepareScopeResume, + sumSpendFromEvents, +} from '../../src/runtime/supervise/recover-executors' +import { createInMemoryRunContext } from '../../src/runtime/supervise/run-context' +import type { Executor, SpawnEvent, Spend } from '../../src/runtime/supervise/types' +import type { RuntimeHookEvent } from '../../src/runtime-hooks' +import { testAgentProfile } from './test-agent-profile' + +const at = '2026-09-15T10:00:00.000Z' +const later = '2026-09-15T10:00:05.000Z' +const budget = { maxIterations: 4, maxTokens: 4000 } +const profile = testAgentProfile('retained') +const profileRef = contentAddress(profile) +const task = 'retained task' +const taskRef = contentAddress(task) +const identity = { + profileDigest: canonicalAgentProfileDigest(profile), + taskDigest: canonicalCandidateDigest(task), +} +const trace = { status: 'unavailable', reason: 'execution-did-not-start' } as const +const reason = 'retained provider execution requires reconciliation before replacement' + +function floor(input: number, output: number): Spend { + return { + iterations: 1, + tokens: { input, output }, + usd: 0, + ms: 5, + tokensKnown: false, + usdKnown: false, + } +} + +function spawned( + id: string, + seq: number, + overrides: Partial> = {}, +): SpawnEvent { + return { + kind: 'spawned', + id, + parent: 'r', + label: id, + budget, + runtime: 'router', + profileRef, + identity, + seq, + at, + ...overrides, + } +} + +const intent = (id: string): RetainedRunAdmission => ({ + phase: 'intent', + provider: 'provider', + idempotencyKey: `${id}-key`, + turnId: `${id}-turn`, + sessionId: `${id}-session`, + executionId: `${id}-execution`, + runId: 'r', + requestedProfileDigest: identity.profileDigest, + requestDigest: canonicalCandidateDigest({ request: id }), +}) +const environment = (id: string, environmentId: string): RetainedRunAdmission => ({ + phase: 'environment', + provider: 'provider', + environmentId, + idempotencyKey: `${id}-key`, + turnId: `${id}-turn`, + sessionId: `${id}-session`, + executionId: `${id}-execution`, +}) +const dispatched = (id: string, environmentId: string): RetainedRunAdmission => ({ + phase: 'dispatched', + idempotencyKey: `${id}-key`, + turnId: `${id}-turn`, + controlRef: { + provider: 'provider', + environmentId, + sessionId: `${id}-session`, + executionId: `${id}-execution`, + runId: 'r', + requestDigest: canonicalCandidateDigest({ request: `${id}-exact` }), + }, +}) + +/** The admission chain a leaf writes before its execution: input, intent, environment. */ +function admitted(id: string, environmentId: string): SpawnEvent[] { + return [ + { kind: 'execution-input', id, taskRef, seq: 0, at }, + { kind: 'execution-admitted', id, admission: intent(id), seq: 0, at }, + { kind: 'execution-admitted', id, admission: environment(id, environmentId), seq: 1, at }, + ] +} + +function reconciled( + id: string, + seq: number, + spent: Spend, + extra: Partial> = {}, +): SpawnEvent { + return { kind: 'reconciled', id, spent, seq, at, ...extra } +} + +/** The whole settlement beside the floor, as 0.234.0's settle path writes it. */ +const settlement = (settledSeq: number) => ({ settledSeq, reason, infra: true, trace }) + +function receipt(id: string, seq: number, environmentId: string, destroyed = true): SpawnEvent { + return { + kind: 'environment-teardown', + id, + provider: 'provider', + environmentId, + destroyed, + seq, + at: later, + } +} + +const root: SpawnEvent = { + kind: 'spawned', + id: 'r', + label: 'root', + budget: { maxIterations: 100, maxTokens: 100_000 }, + runtime: 'inline', + seq: 0, + at, +} + +async function journaled(events: ReadonlyArray) { + const context = createInMemoryRunContext() + await context.journal.beginTree('r', at) + for (const event of events) await context.journal.appendEvent('r', event) + await context.blobs.put(profileRef, profile) + await context.blobs.put(taskRef, task) + return context +} + +const neverExecutes: Executor = { + runtime: 'router', + execute: () => { + throw new Error('recovery must not execute in these cases') + }, + teardown: async () => ({ destroyed: true }), + resultArtifact: () => { + throw new Error('no result') + }, +} + +const terminalRecords = (events: ReadonlyArray, id: string) => + events.filter( + (event) => event.id === id && (event.kind === 'settled' || event.kind === 'cancelled'), + ) + +async function resume( + context: Awaited>, + options: { recover?: boolean; hooks?: RuntimeHookEvent[] } = {}, +) { + const events = (await context.journal.loadTree('r')) ?? [] + return prepareScopeResume( + { + runId: 'r', + journal: context.journal, + blobs: context.blobs, + ...(options.recover ? { recoverExecutor: () => neverExecutes } : {}), + ...(options.hooks + ? { + hooks: { + onEvent: (event: RuntimeHookEvent) => { + options.hooks?.push(event) + }, + }, + } + : {}), + }, + events, + new AbortController().signal, + () => 1_000, + ) +} + +describe('healReleasedSlots on resume', () => { + it('writes the released record from the reconciled record at the seq the driver saw', async () => { + const spent = floor(7, 3) + const harnessTranscript = { status: 'unavailable', reason: 'capture-unsupported' } as const + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, spent, { ...settlement(4), harnessTranscript }), + receipt('r:s0', 0, 'env-1'), + ]) + const before = (await context.journal.loadTree('r')) ?? [] + // The withheld contract: the open node's view carries the floor and no overspend or marker. + expect(materializeTreeView(before).nodes.find((node) => node.id === 'r:s0')).toMatchObject({ + status: 'pending', + spent, + }) + const hooks: RuntimeHookEvent[] = [] + const restored = await resume(context, { recover: true, hooks }) + const events = (await context.journal.loadTree('r')) ?? [] + expect(events).toHaveLength(before.length + 1) + const reconciledRecord = before.find((event) => event.kind === 'reconciled')! + expect(terminalRecords(events, 'r:s0')).toEqual([ + { + kind: 'settled', + status: 'down', + id: 'r:s0', + spent, + infra: true, + reason, + trace, + harnessTranscript, + retainedExecution: 'released', + seq: 4, + at: reconciledRecord.at, + }, + ]) + // Never interrupted: no recovery against the destroyed environment, and the floor is charged + // exactly once, now as the terminal record's spend. + expect(restored.resumeFrom.recoveries).toEqual([]) + expect(restored.poolRestore.uncertainReservations).toEqual([]) + expect(sumSpendFromEvents(events).childWork.tokens).toMatchObject({ input: 7, output: 3 }) + expect(restored.resumeFrom.maxCursorSeq).toBe(4) + expect(restored.resumeFrom.settled).toMatchObject([ + { kind: 'down', retainedExecution: 'released', seq: 4, harnessTranscript }, + ]) + expect(restored.resumeFrom.view.nodes.find((node) => node.id === 'r:s0')).toMatchObject({ + status: 'failed', + retainedExecution: 'released', + spent, + }) + // The sweep's own `agent.child` on the resumed stream, so a projection flips the node. + expect(hooks).toMatchObject([ + { + id: 'r:s0:released', + runId: 'r', + target: 'agent.child', + stepIndex: 4, + parentId: 'r', + timestamp: 1_000, + payload: { + childId: 'r:s0', + status: 'down', + retainedExecution: 'released', + reason, + infra: true, + spent, + runtime: 'router', + startedAt: Date.parse(at), + settledAt: Date.parse(reconciledRecord.at), + releasedAt: Date.parse(later), + }, + }, + ]) + expect(hooks[0]?.payload).not.toHaveProperty('metered') + }) + + it('keeps the cancelled kind and its source, and carries the withheld overspend', async () => { + const spent = floor(900, 0) + const budgetViolation = { + overspent: [{ channel: 'tokens', reserved: 800, spent: 900 }], + } + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, spent, { + ...settlement(2), + budgetViolation, + cancellation: { source: 'signal' }, + }), + receipt('r:s0', 0, 'env-1'), + ]) + const before = (await context.journal.loadTree('r')) ?? [] + // The open-slot view withholds the overspend the pool committed. + expect(materializeTreeView(before).nodes.find((node) => node.id === 'r:s0')).not.toHaveProperty( + 'budgetViolation', + ) + await resume(context) + const events = (await context.journal.loadTree('r')) ?? [] + expect(terminalRecords(events, 'r:s0')).toMatchObject([ + { + kind: 'cancelled', + source: 'signal', + retainedExecution: 'released', + spent, + budgetViolation, + seq: 2, + }, + ]) + expect(materializeTreeView(events).nodes.find((node) => node.id === 'r:s0')).toMatchObject({ + status: 'cancelled', + retainedExecution: 'released', + budgetViolation, + }) + }) + + it('leaves a reconciled record written before settledSeq existed open, still interrupted', async () => { + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, floor(7, 3)), + receipt('r:s0', 0, 'env-1'), + ]) + const before = (await context.journal.loadTree('r')) ?? [] + const restored = await resume(context, { recover: true }) + expect(await context.journal.loadTree('r')).toEqual(before) + expect(restored.resumeFrom.recoveries.map((recovery) => recovery.spawned.id)).toEqual(['r:s0']) + expect(restored.resumeFrom.maxCursorSeq).toBe(-1) + expect(sumSpendFromEvents(before).childWork.tokens).toMatchObject({ input: 7, output: 3 }) + }) + + it('refuses to reuse a cursor seq another record already closes', async () => { + const context = await journaled([ + root, + spawned('r:s0', 0), + spawned('r:s1', 1), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, floor(7, 3), settlement(4)), + receipt('r:s0', 0, 'env-1'), + { + kind: 'settled', + id: 'r:s1', + status: 'done', + outRef: contentAddress('other'), + spent: { iterations: 1, tokens: { input: 1, output: 1 }, usd: 0, ms: 1 }, + trace, + seq: 4, + at, + }, + ]) + const before = (await context.journal.loadTree('r')) ?? [] + await expect(resume(context)).rejects.toThrow(RuntimeRunStateError) + await expect(resume(context)).rejects.toThrow(/'r:s0' cannot be released at cursor seq 4/) + expect(await context.journal.loadTree('r')).toEqual(before) + }) + + it.each([ + { + name: 'the receipt sits before the latest reconciled record', + tail: [receipt('r:s0', 0, 'env-1'), reconciled('r:s0', 0, floor(7, 3), settlement(4))], + }, + { + name: 'one of two receipts is destroyed: false', + tail: [ + reconciled('r:s0', 0, floor(7, 3), settlement(4)), + receipt('r:s0', 0, 'env-1'), + receipt('r:s0', 1, 'env-1b', false), + ], + }, + { + name: 'the receipt names an environment the last admission does not', + tail: [reconciled('r:s0', 0, floor(7, 3), settlement(4)), receipt('r:s0', 0, 'env-2')], + }, + { + name: 'there is no receipt at all', + tail: [reconciled('r:s0', 0, floor(7, 3), settlement(7))], + }, + ])('leaves the slot open when $name', async ({ tail }) => { + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + ...tail, + ]) + const before = (await context.journal.loadTree('r')) ?? [] + const restored = await resume(context, { recover: true }) + expect(await context.journal.loadTree('r')).toEqual(before) + expect(terminalRecords(before, 'r:s0')).toEqual([]) + expect(restored.resumeFrom.recoveries.map((recovery) => recovery.spawned.id)).toEqual(['r:s0']) + }) + + it('reserves an open node’s settledSeq on the resumed cursor even when it is not healed', async () => { + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, floor(7, 3), settlement(7)), + ]) + const restored = await resume(context) + expect(restored.resumeFrom.maxCursorSeq).toBe(7) + }) + + it('leaves a node with a recorded result to the recorded-result branch', async () => { + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + { + kind: 'execution-admitted', + id: 'r:s0', + admission: dispatched('r:s0', 'env-1'), + seq: 2, + at, + }, + reconciled('r:s0', 0, floor(7, 3), settlement(4)), + receipt('r:s0', 0, 'env-1'), + { + kind: 'execution-result', + id: 'r:s0', + outRef: contentAddress('result'), + spent: { iterations: 1, tokens: { input: 3, output: 2 }, usd: 0, ms: 1 }, + seq: 0, + at: later, + }, + ]) + const before = (await context.journal.loadTree('r')) ?? [] + expect( + await healReleasedSlots( + { runId: 'r', journal: context.journal, blobs: context.blobs }, + new AbortController().signal, + () => 1_000, + ), + ).toBe(0) + expect(await context.journal.loadTree('r')).toEqual(before) + }) + + it('settles a recorded result past an open node’s reserved settledSeq, never on it', async () => { + // r:s0 is retained-pending with settledSeq 1 and no receipt, so it stays open. r:s1 has a + // recorded result. Before the shared floor, the recorded-result branch minted seq 1 (one past + // the last CLOSED record) and wrote r:s1's settlement on the seq r:s0 had reserved. + const materialized = knownMaterializationReceipt({ + authoredProfileDigest: identity.profileDigest, + runtime: 'router', + declaration: { + effectiveProfile: profile, + backend: 'router', + model: { status: 'known', id: 'test/model' }, + execution: { kind: 'request', id: 'r:s1-execution' }, + materializer: 'test-router', + plan: { kind: 'completion', model: 'test/model' }, + }, + }) + const bound = knownExecutionBindingReceipt(materialized, { + attemptId: 'r:s1:attempt:1', + binding: { endpoint: 'https://router.example.test', executionId: 'r:s1-execution' }, + descriptor: { kind: 'router-request', transport: 'http' }, + }) + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, floor(7, 3), settlement(1)), + spawned('r:s1', 1), + ...admitted('r:s1', 'env-2'), + { kind: 'materialized', id: 'r:s1', receipt: materialized, seq: 0, at }, + { kind: 'execution-bound', id: 'r:s1', binding: bound, seq: 0, at }, + { + kind: 'execution-admitted', + id: 'r:s1', + admission: dispatched('r:s1', 'env-2'), + seq: 2, + at, + }, + { + kind: 'execution-result', + id: 'r:s1', + outRef: contentAddress('result'), + spent: { iterations: 1, tokens: { input: 3, output: 2 }, usd: 0, ms: 1 }, + seq: 0, + at: later, + }, + ]) + await context.blobs.put(contentAddress('result'), 'result') + const restored = await resume(context) + const events = (await context.journal.loadTree('r')) ?? [] + const s1 = terminalRecords(events, 'r:s1') + expect(s1).toHaveLength(1) + expect(s1[0]?.seq).toBe(2) + expect(terminalRecords(events, 'r:s0')).toEqual([]) + expect(restored.resumeFrom.maxCursorSeq).toBe(2) + }) + + it('writes nothing twice: a released record already present is left alone', async () => { + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, floor(7, 3), settlement(4)), + receipt('r:s0', 0, 'env-1'), + ]) + await resume(context) + const healed = (await context.journal.loadTree('r')) ?? [] + const hooks: RuntimeHookEvent[] = [] + await resume(context, { hooks }) + expect(await context.journal.loadTree('r')).toEqual(healed) + expect(hooks).toEqual([]) + }) + + it('takes the latest reconciled record: its settledSeq, spent and settlement', async () => { + const first = floor(7, 3) + const second = floor(9, 4) + const context = await journaled([ + root, + spawned('r:s0', 0), + ...admitted('r:s0', 'env-1'), + reconciled('r:s0', 0, first, settlement(1)), + receipt('r:s0', 0, 'env-1'), + reconciled('r:s0', 1, second, { ...settlement(3), reason: 'second failure' }), + receipt('r:s0', 1, 'env-1'), + ]) + await resume(context) + const events = (await context.journal.loadTree('r')) ?? [] + expect(terminalRecords(events, 'r:s0')).toMatchObject([ + { spent: second, reason: 'second failure', seq: 3, retainedExecution: 'released' }, + ]) + expect(sumSpendFromEvents(events).childWork.tokens).toMatchObject({ input: 9, output: 4 }) + }) +}) diff --git a/tests/kernel/retained-environment-release.test.ts b/tests/kernel/retained-environment-release.test.ts index defe0c27..433524cd 100644 --- a/tests/kernel/retained-environment-release.test.ts +++ b/tests/kernel/retained-environment-release.test.ts @@ -33,6 +33,8 @@ import type { AgentEnvironmentProvider, } from '@tangle-network/agent-interface/environment-provider' import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { FileObserverJournal } from '../../src/durable/observer-journal' +import { projectPursuit } from '../../src/durable/observer-projection' import { closesCursorSlot, materializeTreeView, @@ -40,6 +42,7 @@ import { } from '../../src/durable/spawn-journal' import { providerAsExecutor } from '../../src/runtime/environment-provider' import { driverChild } from '../../src/runtime/supervise/driver-executor' +import { sumSpendFromEvents } from '../../src/runtime/supervise/recover-executors' import { RetainedExecutionPendingError } from '../../src/runtime/supervise/retained-executor' import { createFileRunContext, @@ -50,9 +53,11 @@ import type { Agent, Executor, ExecutorFactory, + NodeId, Scope, Settled, SpawnEvent, + SpawnJournal, SupervisorOpts, } from '../../src/runtime/supervise/types' import type { RuntimeHookEvent } from '../../src/runtime-hooks' @@ -833,3 +838,549 @@ describe('retained environments at root settlement', () => { }) }) }) + +/** + * The process dies between the last `destroyed: true` receipt and the released record: the + * journal ends with the receipt, the slot is open, the environment is gone. Everything the release + * sweep would have written is captured so the heal can be compared byte-for-byte against it. + * Every other journal call forwards to the real file journal with its own `this`, so the stamp + * check and the receipt's acknowledgement are untouched. + */ +function crashBeforeReleasedRecord(journal: SpawnJournal) { + const dropped: SpawnEvent[] = [] + const proxied = new Proxy(journal, { + get(target, property, receiver) { + if (property !== 'appendEvent') return Reflect.get(target, property, receiver) + return (root: NodeId, event: SpawnEvent) => { + if ( + (event.kind === 'settled' || event.kind === 'cancelled') && + event.retainedExecution === 'released' + ) { + dropped.push(event) + throw new Error('process died before the released record') + } + return target.appendEvent(root, event) + } + }, + }) + return { journal: proxied, dropped } +} + +/** An ordinary leaf that completes at once, for a settlement after the healed one. */ +function completingWorker(): Agent { + const spent = { iterations: 1, tokens: { input: 1, output: 1 }, usd: 0, ms: 0 } + const executor: Executor = { + runtime: 'router', + execute(): AsyncIterable<{ kind: 'tokens'; input: number; output: number }> { + return (async function* () { + yield { kind: 'tokens' as const, input: 1, output: 1 } + })() + }, + teardown: async () => ({ destroyed: true }), + resultArtifact: () => ({ out: 'ok', outRef: canonicalCandidateDigest('ok'), spent }), + } + return Object.assign( + { name: 'plain', act: async () => 'unused' }, + { executorSpec: { profile: testAgentProfile('plain'), harness: null, executor } }, + ) +} + +const reconciledRecords = (events: ReadonlyArray, id: string) => + events.flatMap((event) => (event.kind === 'reconciled' && event.id === id ? [event] : [])) + +/** A deterministic clock so two runs of the same shape journal the same instants. */ +const fixedClock = () => { + let tick = Date.parse('2026-09-15T12:00:00.000Z') + return () => (tick += 1000) +} + +/** Two runs of one shape differ only in the attempt ids the root's own binding mints, and a + * replayed handle carries a method; compare the recorded data with the ids pinned. */ +const recorded = (value: unknown) => + JSON.parse( + JSON.stringify(value).replace(/"attemptId":"[^"]+"/g, '"attemptId":""'), + ) as unknown + +describe('a crash between the receipt and the released record heals on the next resume', () => { + let directory: string + beforeEach(async () => { + directory = await mkdtemp(join(tmpdir(), 'retained-heal-')) + }) + afterEach(async () => { + await rm(directory, { recursive: true, force: true }) + }) + + const common = (runId: string) => + ({ + runId, + budget: { maxIterations: 4, maxTokens: 40 }, + rootIdentity: { + profileDigest: canonicalCandidateDigest({ name: 'root' }), + taskDigest: canonicalCandidateDigest('task'), + }, + retainedAtSettlement: 'release', + }) satisfies Partial + + it('writes the sweep’s own record at the driver’s seq, treats nothing as interrupted, and reports the floor', async () => { + const fleet = retainedProvider(directory) + const runDirectory = join(directory, 'run') + const observerPath = join(directory, 'observer.jsonl') + const crashed = crashBeforeReleasedRecord(createFileRunContext(runDirectory).journal) + await createSupervisor().run( + { + name: 'root', + async act(_task, scope) { + await spawnAndAwait(scope, retainedWorker(providerAsExecutor(fleet.provider()))) + return 'finished' + }, + }, + 'task', + { + ...createFileRunContext(runDirectory), + journal: crashed.journal, + ...common('heal'), + hooks: new FileObserverJournal(observerPath, 'pursuit:heal').hooks(), + }, + ) + // The window exactly: the environment is destroyed, the receipt is the last record, the slot + // is open, and the reconciled record carries the settlement and the cursor seq beside the floor. + expect(fleet.state.destroys).toBe(1) + expect(fleet.environments()).toEqual([]) + expect(crashed.dropped).toHaveLength(1) + const before = (await createFileRunContext(runDirectory).journal.loadTree('heal')) ?? [] + const receipt = releaseReceipts(before)[0]! + expect(receipt).toMatchObject({ id: 'heal:s0', destroyed: true }) + expect(before.at(-1)).toBe(receipt) + expect(terminalRecords(before, 'heal:s0')).toEqual([]) + const reconciled = reconciledRecords(before, 'heal:s0')[0]! + expect(reconciled).toMatchObject({ + settledSeq: 0, + infra: true, + reason: expect.stringContaining('reconciliation'), + trace: expect.anything(), + harnessTranscript: expect.anything(), + }) + expect(reconciled).not.toHaveProperty('cancellation') + + fleet.state.resultLost = false + const restarted = createFileRunContext(runDirectory) + const hookEvents: RuntimeHookEvent[] = [] + const observer = new FileObserverJournal(observerPath, 'pursuit:heal').hooks() + let secondChild: Settled | undefined + let actError: unknown + const resumed = await createSupervisor().run( + { + name: 'root', + async act(_task, scope) { + try { + expect(scope.resume).toBeDefined() + // Healed before the root acts: nothing in flight, the node already down and released. + expect(scope.view.inFlight).toBe(0) + expect(scope.resume?.settled).toMatchObject([ + { kind: 'down', retainedExecution: 'released', seq: 0 }, + ]) + // The resumed cursor starts past the healed seq, so a new settlement never collides. + const plain = scope.spawn(completingWorker(), 'plain task', { + budget: { maxIterations: 1, maxTokens: 10 }, + }) + expect(plain, JSON.stringify(plain)).toMatchObject({ ok: true }) + secondChild = (await scope.next()) ?? undefined + } catch (error) { + actError = error + throw error + } + return 'resumed' + }, + }, + 'task', + { + ...restarted, + ...common('heal'), + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + hooks: { + onEvent: (event, context) => { + hookEvents.push(event) + return observer.onEvent?.(event, context) + }, + }, + }, + ) + expect(actError).toBeUndefined() + expect(resumed.kind, JSON.stringify(resumed)).toBe('winner') + if (resumed.kind !== 'winner') return + expect(resumed.out).toBe('resumed') + expect(secondChild).toMatchObject({ kind: 'done', seq: 1 }) + // No recovery was attempted against the destroyed environment. + expect(fleet.state.destroys).toBe(1) + expect(fleet.environments()).toEqual([]) + + const events = (await restarted.journal.loadTree('heal')) ?? [] + const terminal = terminalRecords(events, 'heal:s0') + expect(terminal).toEqual([crashed.dropped[0]]) + expect(events.indexOf(terminal[0]!)).toBeGreaterThan(events.indexOf(receipt)) + expect(terminal[0]).toMatchObject({ + kind: 'settled', + status: 'down', + retainedExecution: 'released', + seq: 0, + at: reconciled.at, + spent: reconciled.spent, + harnessTranscript: reconciled.harnessTranscript, + }) + expect(await replaySpawnTree(restarted.journal, restarted.blobs, 'heal')).toMatchObject([ + { + kind: 'down', + retainedExecution: 'released', + seq: 0, + harnessTranscript: reconciled.harnessTranscript, + }, + { kind: 'done', seq: 1 }, + ]) + // Every reader agrees: down and released, an unreported floor rather than a never-settled + // ceiling, and the same child work the pool committed at the reconcile. + expect(resumed.fleetYield).toEqual({ + spawned: 2, + done: 1, + down: 1, + cancelled: 0, + neverSettled: 0, + releasedUnrecovered: 1, + }) + expect(resumed.spendGaps).toEqual([ + expect.objectContaining({ id: 'heal:s0', kind: 'unreported' }), + ]) + expect(resumed.spentTotal.tokensKnown).toBe(false) + const retainedOnly = (list: ReadonlyArray) => + list.filter( + (event) => + event.id === 'heal:s0' || (event.kind === 'spawned' && event.parent === undefined), + ) + expect(sumSpendFromEvents(retainedOnly(events)).childWork).toEqual( + sumSpendFromEvents(retainedOnly(before)).childWork, + ) + // The sweep's `agent.child` on the resumed stream, and the projection that reads it. + const released = childPayloads(hookEvents, 'heal:s0') + expect(released).toHaveLength(1) + expect(released[0]).toMatchObject({ + stepIndex: 0, + payload: { + status: 'down', + retainedExecution: 'released', + settledAt: Date.parse(reconciled.at), + releasedAt: Date.parse(receipt.at), + spent: reconciled.spent, + }, + }) + expect(released[0]?.payload).not.toHaveProperty('metered') + expect(hookEvents.find((event) => event.id === 'heal:s0:released')?.runId).toBe('heal') + const projection = projectPursuit( + await new FileObserverJournal(observerPath, 'pursuit:heal').read(), + ) + expect(projection.nodes.find((node) => node.id === 'heal:s0')).toMatchObject({ + status: 'down', + retainedExecution: 'released', + releasedAt: Date.parse(receipt.at), + spent: expect.objectContaining({ tokens: reconciled.spent.tokens }), + }) + expect(projection.runs[0]?.spendGaps).toEqual([ + expect.objectContaining({ id: 'heal:s0', kind: 'unreported' }), + ]) + }) + + it('replays as the run the sweep completed, and a second resume writes nothing more', async () => { + const fleet = retainedProvider(directory) + const root: Agent = { + name: 'root', + async act(_task, scope) { + if (scope.resume !== undefined) return 'resumed' + await spawnAndAwait(scope, retainedWorker(providerAsExecutor(fleet.provider()))) + return 'finished' + }, + } + // The control: the same run, uncrashed, on the same clock. + const controlDirectory = join(directory, 'control') + const control = createFileRunContext(controlDirectory) + const controlRun = await createSupervisor().run(root, 'task', { + ...control, + ...common('same'), + now: fixedClock(), + }) + expect(controlRun.kind, JSON.stringify(controlRun)).toBe('winner') + const controlEvents = (await control.journal.loadTree('same')) ?? [] + expect(terminalRecords(controlEvents, 'same:s0')).toMatchObject([ + { retainedExecution: 'released' }, + ]) + + const runDirectory = join(directory, 'run') + const crashed = crashBeforeReleasedRecord(createFileRunContext(runDirectory).journal) + await createSupervisor().run(root, 'task', { + ...createFileRunContext(runDirectory), + journal: crashed.journal, + ...common('same'), + now: fixedClock(), + }) + const crashedEvents = (await createFileRunContext(runDirectory).journal.loadTree('same')) ?? [] + expect(terminalRecords(crashedEvents, 'same:s0')).toEqual([]) + expect(recorded(crashedEvents)).toEqual(recorded(controlEvents.slice(0, -1))) + + const restarted = createFileRunContext(runDirectory) + const resumed = await createSupervisor().run(root, 'task', { + ...restarted, + ...common('same'), + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + }) + expect(resumed.kind, JSON.stringify(resumed)).toBe('winner') + const healedEvents = (await restarted.journal.loadTree('same')) ?? [] + // The resumed process adds only its own root binding; every child record is the control's. + const childRecords = (list: ReadonlyArray) => + list.filter((event) => !(event.kind === 'execution-bound' && event.id === 'same')) + expect(recorded(childRecords(healedEvents))).toEqual(recorded(childRecords(controlEvents))) + expect(recorded(await replaySpawnTree(restarted.journal, restarted.blobs, 'same'))).toEqual( + recorded(await replaySpawnTree(control.journal, control.blobs, 'same')), + ) + const childNodes = (list: ReadonlyArray) => + materializeTreeView(list).nodes.filter((node) => node.parent !== undefined) + expect(recorded(childNodes(healedEvents))).toEqual(recorded(childNodes(controlEvents))) + expect(resumed.fleetYield).toEqual(controlRun.fleetYield) + + const again = createFileRunContext(runDirectory) + const hookEvents: RuntimeHookEvent[] = [] + const second = await createSupervisor().run(root, 'task', { + ...again, + ...common('same'), + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + hooks: { + onEvent: (event) => { + hookEvents.push(event) + }, + }, + }) + expect(second.kind, JSON.stringify(second)).toBe('winner') + expect(childRecords((await again.journal.loadTree('same')) ?? [])).toEqual( + childRecords(healedEvents), + ) + expect(childPayloads(hookEvents, 'same:s0')).toEqual([]) + expect(fleet.state.destroys).toBe(2) + }) + + it('keeps the cancelled kind and its source on a healed cancelled child', async () => { + const fleet = retainedProvider(directory) + const runDirectory = join(directory, 'run') + const root: Agent = { + name: 'root', + async act(_task, scope) { + if (scope.resume === undefined) { + await spawnAndAwait(scope, retainedWorker(providerAsExecutor(fleet.provider()))) + return 'finished' + } + expect(scope.view.inFlight).toBe(0) + return 'resumed' + }, + } + const abort = new AbortController() + fleet.state.observe = async (signal) => { + abort.abort(new Error('operator stopped the run')) + await new Promise((_resolve, reject) => { + const fail = () => reject(signal?.reason ?? new Error('observation aborted')) + if (signal === undefined || signal.aborted) fail() + else signal.addEventListener('abort', fail, { once: true }) + }) + } + const crashed = crashBeforeReleasedRecord(createFileRunContext(runDirectory).journal) + await createSupervisor().run(root, 'task', { + ...createFileRunContext(runDirectory), + journal: crashed.journal, + ...common('abort-heal'), + signal: abort.signal, + }) + expect(fleet.state.destroys).toBe(1) + const before = (await createFileRunContext(runDirectory).journal.loadTree('abort-heal')) ?? [] + expect(terminalRecords(before, 'abort-heal:s0')).toEqual([]) + expect(reconciledRecords(before, 'abort-heal:s0')[0]).toMatchObject({ + settledSeq: 0, + cancellation: { source: 'signal' }, + }) + expect(crashed.dropped).toMatchObject([{ kind: 'cancelled', source: 'signal' }]) + + fleet.state.observe = undefined + fleet.state.resultLost = false + const restarted = createFileRunContext(runDirectory) + const resumed = await createSupervisor().run(root, 'task', { + ...restarted, + ...common('abort-heal'), + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + }) + expect(resumed.kind, JSON.stringify(resumed)).toBe('winner') + expect(fleet.state.destroys).toBe(1) + const events = (await restarted.journal.loadTree('abort-heal')) ?? [] + expect(terminalRecords(events, 'abort-heal:s0')).toEqual(crashed.dropped) + expect(terminalRecords(events, 'abort-heal:s0')).toMatchObject([ + { kind: 'cancelled', source: 'signal', retainedExecution: 'released', seq: 0 }, + ]) + expect(resumed.fleetYield).toEqual({ + spawned: 1, + done: 0, + down: 0, + cancelled: 1, + neverSettled: 0, + releasedUnrecovered: 1, + }) + }) + + it('leaves a refused release open: the node is still interrupted and the resume recovers it', async () => { + const fleet = retainedProvider(directory) + fleet.state.destroyFailure = new Error('409 Conflict: environment is still stopping') + const runDirectory = join(directory, 'run') + const root: Agent = { + name: 'root', + async act(_task, scope) { + if (scope.resume === undefined) { + await spawnAndAwait(scope, retainedWorker(providerAsExecutor(fleet.provider()))) + return 'finished' + } + // Still interrupted: the resumed process adopts the retained child before the root acts. + expect(scope.view.inFlight).toBe(1) + const settled = await scope.next() + return settled?.kind === 'done' ? 'recovered' : 'lost' + }, + } + const first = createFileRunContext(runDirectory) + const refused = await createSupervisor().run(root, 'task', { + ...first, + ...common('refused-heal'), + }) + expect(refused.kind, JSON.stringify(refused)).toBe('winner') + expect(fleet.environments()).toHaveLength(1) + const before = (await first.journal.loadTree('refused-heal')) ?? [] + expect(releaseReceipts(before)).toMatchObject([{ id: 'refused-heal:s0', destroyed: false }]) + expect(reconciledRecords(before, 'refused-heal:s0')[0]).toMatchObject({ settledSeq: 0 }) + expect(terminalRecords(before, 'refused-heal:s0')).toEqual([]) + + fleet.state.destroyFailure = undefined + fleet.state.resultLost = false + const restarted = createFileRunContext(runDirectory) + const resumed = await createSupervisor().run(root, 'task', { + ...restarted, + ...common('refused-heal'), + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + }) + expect(resumed.kind, JSON.stringify(resumed)).toBe('winner') + if (resumed.kind !== 'winner') return + expect(resumed.out).toBe('recovered') + const events = (await restarted.journal.loadTree('refused-heal')) ?? [] + const terminal = terminalRecords(events, 'refused-heal:s0') + expect(terminal).toMatchObject([{ kind: 'settled', status: 'done' }]) + expect(terminal[0]).not.toHaveProperty('retainedExecution') + expect(resumed.fleetYield).toEqual({ + spawned: 1, + done: 1, + down: 0, + cancelled: 0, + neverSettled: 0, + releasedUnrecovered: 0, + }) + }) + + it("heals a nested manager's grandchild in its own tree from the root resume", async () => { + const fleet = retainedProvider(directory) + const runDirectory = join(directory, 'run') + const crashed = crashBeforeReleasedRecord( + createFileRunContext(runDirectory, { withDriver: true }).journal, + ) + const managerFor = (journal: SpawnJournal) => + driverChild( + testAgentProfile('manager'), + { + name: 'manager', + async act(_task, scope) { + const settled = await spawnAndAwait( + scope, + retainedWorker(providerAsExecutor(fleet.provider())), + ) + expect(settled?.kind).toBe('down') + return 'finalized manager' + }, + }, + journal, + ) + const rootFor = (manager: Agent): Agent => ({ + name: 'root', + async act(_task, scope) { + if (scope.resume !== undefined) { + expect(scope.view.inFlight).toBe(0) + return 'resumed' + } + expect( + scope.spawn(manager, 'manage', { budget: { maxIterations: 2, maxTokens: 20 } }).ok, + ).toBe(true) + expect((await scope.next())?.kind).toBe('down') + return 'finished' + }, + }) + await createSupervisor().run(rootFor(managerFor(crashed.journal)), 'task', { + ...createFileRunContext(runDirectory, { withDriver: true }), + journal: crashed.journal, + ...common('root'), + budget: { maxIterations: 4, maxTokens: 100 }, + }) + expect(fleet.environments()).toEqual([]) + expect(crashed.dropped).toHaveLength(1) + const firstContext = createFileRunContext(runDirectory, { withDriver: true }) + const nestedBefore = (await firstContext.journal.loadTree('root/root:s0')) ?? [] + expect(releaseReceipts(nestedBefore)).toMatchObject([{ id: 'root:s0:s0', destroyed: true }]) + expect(terminalRecords(nestedBefore, 'root:s0:s0')).toEqual([]) + // The manager settled on the ordinary path in the root tree: nothing there to restore. + const rootBefore = (await firstContext.journal.loadTree('root')) ?? [] + expect(terminalRecords(rootBefore, 'root:s0')).toHaveLength(1) + + fleet.state.resultLost = false + const restarted = createFileRunContext(runDirectory, { withDriver: true }) + const hookEvents: RuntimeHookEvent[] = [] + const resumed = await createSupervisor().run( + rootFor(managerFor(restarted.journal)), + 'task', + { + ...restarted, + ...common('root'), + budget: { maxIterations: 4, maxTokens: 100 }, + resume: true, + recoverExecutor: providerAsExecutor(fleet.provider()), + hooks: { + onEvent: (event) => { + hookEvents.push(event) + }, + }, + }, + ) + expect(resumed.kind, JSON.stringify(resumed)).toBe('winner') + expect(fleet.state.destroys).toBe(1) + const nested = (await restarted.journal.loadTree('root/root:s0')) ?? [] + const nestedTerminal = terminalRecords(nested, 'root:s0:s0') + expect(nestedTerminal).toEqual(crashed.dropped) + expect(nested.indexOf(nestedTerminal[0]!)).toBeGreaterThan( + nested.indexOf(releaseReceipts(nested)[0]!), + ) + // The root tree gains only the resumed root's own binding: no receipt, no second record. + const rootAfter = (await restarted.journal.loadTree('root')) ?? [] + expect(rootAfter.filter(closesCursorSlot)).toEqual(rootBefore.filter(closesCursorSlot)) + expect(releaseReceipts(rootAfter)).toEqual([]) + expect(hookEvents.find((event) => event.id === 'root:s0:s0:released')).toMatchObject({ + runId: 'root/root:s0', + parentId: 'root:s0', + target: 'agent.child', + }) + expect(resumed.fleetYield).toEqual({ + spawned: 2, + done: 0, + down: 2, + cancelled: 0, + neverSettled: 0, + releasedUnrecovered: 1, + }) + }) +})