From 176b3b6040ab91fa0d1b6d8e58c9fdc847d6ad8b Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Wed, 16 Sep 2026 01:11:09 -0700 Subject: [PATCH 1/2] fix(supervise): heal a released slot the sweep's process died before recording A release sweep appends a retained child's environment-teardown receipt and then its terminal record. A process that died between the two left an open cursor slot beside a destroyed environment; the next resume treated the node as interrupted, tried to recover an executor whose environment was gone or left it open forever, and every reader charged the ceiling the pool had already refunded. The reconciled record now carries the settlement the driver received (reason, infra, trace, outRef, providerModel, harnessTranscript), the withheld overspend, the cancellation source, and settledSeq, the cursor seq next() stamped on the delivery, through the same field spread the terminal record uses. On the root resume, healReleasedSlots walks the journal forest and, for a spawned node with no terminal record whose receipts after that record all read destroyed: true and name the environment its last admission named, writes the released record with the sweep's own builder at settledSeq and the reconciled at, then emits the sweep's agent.child event. The node is never interrupted and no recovery is attempted. A destroyed: false receipt, an empty receipt set, a receipt for an unadmitted environment, or a reconciled record without settledSeq leaves the slot open as before. The resumed cursor starts past every open node's settledSeq; a journal that already closes that seq fails the resume and writes nothing. terminalDownEvent, settledNodeEvidence and the release payload move to supervise/terminal-record.ts so scope.ts and recover-executors.ts share one builder without a runtime import cycle. The refund at the reconcile is unchanged. Version 0.234.0: SpawnEvent's shape moved. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit c5b7b69c7f0dacbb805ce767e2a05598d117527e) --- CHANGELOG.md | 28 + api-surface.json | 2 +- docs/api/primitive-catalog.md | 2 +- docs/api/runtime.md | 94 ++- docs/canonical-api.md | 2 +- package.json | 2 +- src/runtime/supervise/recover-executors.ts | 171 +++++- src/runtime/supervise/scope.ts | 154 ++--- src/runtime/supervise/terminal-record.ts | 159 +++++ src/runtime/supervise/types.ts | 58 +- .../fixtures/agent-improvement-proposal.json | 10 +- .../agent-profile-improvement-proposal.json | 6 +- tests/kernel/recover-released-slot.test.ts | 480 +++++++++++++++ .../retained-environment-release.test.ts | 551 ++++++++++++++++++ 14 files changed, 1580 insertions(+), 139 deletions(-) create mode 100644 src/runtime/supervise/terminal-record.ts create mode 100644 tests/kernel/recover-released-slot.test.ts 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..c7a519a6 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( @@ -386,6 +535,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 +575,18 @@ 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: Math.max( + maxSeqOf(prior, closesCursorSlot), + prior.reduce( + (max, event) => + event.kind === 'reconciled' && event.settledSeq !== undefined + ? Math.max(max, event.settledSeq) + : max, + -1, + ), + ), 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..352f86c1 --- /dev/null +++ b/tests/kernel/recover-released-slot.test.ts @@ -0,0 +1,480 @@ +/** + * 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 { + 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('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, + }) + }) +}) From 161756a0f1ff4c25205b86f50f3c9158739fad72 Mon Sep 17 00:00:00 2001 From: Drew Stone Date: Wed, 16 Sep 2026 04:43:01 -0700 Subject: [PATCH 2/2] fix(supervise): every writer in a resumed process mints its cursor seq past the reserved floor Independent check of the heal found the recorded-result branch of prepareInterruptedExecutors seeding its seq one past the last CLOSED record only, while the resumed scope's cursor already folded open nodes' reserved settledSeq. A recorded result could therefore be journaled on the seq an open retained node had reserved, exactly the collision the heal refuses. One reservedCursorFloor(events) now seeds both; a test pins the shape (open node at settledSeq 1, recorded result settles at 2) and fails on the closed-only floor. Co-Authored-By: Claude Fable 5.1 (cherry picked from commit 00de2dd31a811330b2c9e700f2562b51ec81d8c4) --- src/runtime/supervise/recover-executors.ts | 40 ++++++++++----- tests/kernel/recover-released-slot.test.ts | 60 ++++++++++++++++++++++ 2 files changed, 87 insertions(+), 13 deletions(-) diff --git a/src/runtime/supervise/recover-executors.ts b/src/runtime/supervise/recover-executors.ts index c7a519a6..b27406ed 100644 --- a/src/runtime/supervise/recover-executors.ts +++ b/src/runtime/supervise/recover-executors.ts @@ -312,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) @@ -450,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 @@ -577,16 +600,7 @@ export async function prepareScopeResume( maxSpawnOrdinal: maxSeqOf(prior, (event) => event.kind === 'spawned'), // 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: Math.max( - maxSeqOf(prior, closesCursorSlot), - prior.reduce( - (max, event) => - event.kind === 'reconciled' && event.settledSeq !== undefined - ? Math.max(max, event.settledSeq) - : max, - -1, - ), - ), + maxCursorSeq: reservedCursorFloor(prior), maxWaitOrdinal: maxSeqOf(prior, (event) => event.kind === 'waiting'), waits: pendingWaits(prior), keys: keyedAssignments(prior, settled), diff --git a/tests/kernel/recover-released-slot.test.ts b/tests/kernel/recover-released-slot.test.ts index 352f86c1..37ad827d 100644 --- a/tests/kernel/recover-released-slot.test.ts +++ b/tests/kernel/recover-released-slot.test.ts @@ -20,6 +20,10 @@ 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, @@ -442,6 +446,62 @@ describe('healReleasedSlots on resume', () => { 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,