diff --git a/CHANGELOG.md b/CHANGELOG.md index 438a9286..3cf1aafa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## 0.175.0 + +### A release killed between its two journal events no longer runs its node twice + +`release()` journals the envelope pin (`node-inputs-resolved`) and then the wave consumption (`join-state`). A process killed between them left an instance pinned while its gating edges still read satisfied, so the restart both re-entered the pinned instance AND released a second one — the node executed twice in one resumed segment. A consumer's external effect fired twice through that window (agent-dev-container's workflow scheduler, which now refuses to resume non-idempotent actions through the same window). + +The fold records wave consumption on the instance, and a restart that finds a released instance without it journals the missing `join-state` before spawning — re-deriving the SAME decision the crashed process made, because its gating edges are folded exactly as they were when it released. `tests/graph/replay.test.ts` kills at the pin and asserts the node runs exactly once. + ## 0.174.1 Two graph-engine resume fixes, exposed by running agent-dev-container's real `pr-review-with-approval` workflow template on the engine (#1011). diff --git a/api-surface.json b/api-surface.json index f07476e6..3c1d75ac 100644 --- a/api-surface.json +++ b/api-surface.json @@ -775,7 +775,7 @@ "FinalizerChoice": "type 9f5b133c4aff", "FoldEdge": "type 21a7109f4e19", "FoldEdgeState": "type d818243e12e3", - "FoldInstance": "type 74b2ae32eef4", + "FoldInstance": "type df9b415f4bd5", "FoldInstanceStatus": "type 123693084eb3", "FoldNode": "type ea75dac954b8", "FoldSuspension": "type ae8d8798d242", diff --git a/docs/api/primitive-catalog.md b/docs/api/primitive-catalog.md index 26224756..0202cdc9 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.174.1` and `@tangle-network/agent-eval@0.170.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.175.0` and `@tangle-network/agent-eval@0.170.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/graph.md b/docs/api/runtime/graph.md index f44ef56d..6048319a 100644 --- a/docs/api/runtime/graph.md +++ b/docs/api/runtime/graph.md @@ -480,6 +480,16 @@ The admitted source payload this state reflects (the source settle's outRef). > `optional` **settle?**: [`GraphNodeSettle`](#graphnodesettle) +##### waveConsumed? + +> `optional` **waveConsumed?**: `boolean` + +Whether this instance's release consumed its wave — the `join-state` the +scheduler journals AFTER the envelope pin. A restart that finds a +released instance without it completes the half-journaled release rather +than leaving the gating edges satisfied, which would release the node a +second time and execute it twice. + *** ### FoldSuspension diff --git a/docs/canonical-api.md b/docs/canonical-api.md index ee1b7fb6..3d92caf5 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.174.1.** +> **Version 0.175.0.** > [`docs/api/primitive-catalog.md`](./api/primitive-catalog.md) lists every export and import path. > `agent-eval` must satisfy `>=0.163.2 <0.171.0`. > `sandbox` must satisfy `>=0.31.0 <0.32.0`. diff --git a/package.json b/package.json index f47ad8cd..7c3163d0 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@tangle-network/agent-runtime", - "version": "0.174.1", + "version": "0.175.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/graph/fold.ts b/src/runtime/graph/fold.ts index 581ef93a..98b60048 100644 --- a/src/runtime/graph/fold.ts +++ b/src/runtime/graph/fold.ts @@ -42,6 +42,14 @@ export interface FoldInstance { inputRef?: string status: FoldInstanceStatus settle?: GraphNodeSettle + /** + * Whether this instance's release consumed its wave — the `join-state` the + * scheduler journals AFTER the envelope pin. A restart that finds a + * released instance without it completes the half-journaled release rather + * than leaving the gating edges satisfied, which would release the node a + * second time and execute it twice. + */ + waveConsumed?: boolean } export interface FoldSuspension { @@ -195,6 +203,8 @@ export function applyGraphFoldEvent( return } case 'join-state': { + const released = instanceOf(state, ev.instance) + if (released) released.waveConsumed = true // A release consumes its wave: delivered consumptions count a traversal and re-arm; every // gating edge still pending is consumed-once. for (const edgeId of ev.satisfiedBy) { diff --git a/src/runtime/graph/scheduler.ts b/src/runtime/graph/scheduler.ts index 7f4281b5..960d87bc 100644 --- a/src/runtime/graph/scheduler.ts +++ b/src/runtime/graph/scheduler.ts @@ -25,7 +25,7 @@ import { type CompiledGraph, compileGraph, isEngineFired } from './compile' import { evaluateCondition } from './condition' import type { EngineGraphSpec } from './definition' import type { GraphEngine } from './engine' -import { applyGraphFoldEvent, type FoldSuspension } from './fold' +import { applyGraphFoldEvent, type FoldInstance, type FoldSuspension } from './fold' import { decideJoin, type GatingEdge } from './join' import { narrowEffects } from './kind' import { createEdgeLedger } from './ledger' @@ -406,6 +406,37 @@ async function runGraphLoop( await spawnInstance(label) } + /** + * Journal the `join-state` a crashed release never wrote. The consuming set + * is re-derived from the SAME decision the crashed process made: its gating + * edges are still folded exactly as they were when it released (their + * verdicts were journaled first), so `decideJoin` answers identically. + */ + const consumeWaveAfterCrash = async (instance: FoldInstance): Promise => { + const node = compiled.nodes.get(instance.node) + if (!node) return + const gating: GatingEdge[] = node.inbound.map((edge) => ({ + edge, + folded: state.edges.get(edge.id), + })) + const decision = decideJoin(node.join, gating) + if (!decision.release) return + const consumedPending = gating + .filter((entry) => entry.folded?.state === 'pending') + .map((entry) => entry.edge.id) + await emit({ + kind: 'join-state', + id: instance.instance, + node: instance.node, + rule: node.join, + satisfiedBy: decision.consuming.map(({ edge }) => edge.id), + consumedPending, + instance: instance.instance, + seq: engineSeq++, + at: stamp(), + }) + } + const tryRelease = async (nodeId: string): Promise => { const node = compiled.nodes.get(nodeId) const folded = state.nodes.get(nodeId) @@ -677,6 +708,12 @@ async function runGraphLoop( async function reenterAfterCrash(): Promise { for (const instance of [...state.instances.values()]) { if (instance.status === 'released') { + // The envelope was pinned but the wave consumption may not have been + // journaled — the crash window between the two events. Finish that + // release first: its gating edges stay satisfied otherwise, and the + // catch-up `tryRelease` below would release the node a SECOND time + // and execute it twice. + if (instance.waveConsumed !== true) await consumeWaveAfterCrash(instance) await spawnInstance(instance.instance) continue } diff --git a/src/testing/fixtures/agent-improvement-proposal.json b/src/testing/fixtures/agent-improvement-proposal.json index 9dbb232d..9e1c4ddf 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:97354ab43c80f5cbb36da0638ec609f8eea1afa20d1ed8b741fc96602df1cc91", + "digest": "sha256:9cbe973bcffdd0ca3dbb73f2488fa44605d465a007e519bce7dea48d1f0c14fe", "evaluation": { "decision": { "contributingChecks": [ @@ -4882,7 +4882,7 @@ ], "metadata": { "fixture": "agent-improvement-proposal", - "runtimeVersion": "0.174.1" + "runtimeVersion": "0.175.0" }, "objectives": [ { @@ -4993,8 +4993,8 @@ "baselineContentHash": "sha256:5c21ee53e513fc604cb09754e21c392b24a424da0ef37dbf8f1ee4a8a0b08f09", "candidateContentHash": "sha256:60fcbb1c728194bd51d7d19cb732d1c3f1881dce7e0a6266b41c8b98cfd65693", "kind": "agent-eval-loop", - "recordDigest": "sha256:60ab29dc9618cc1a8ae268cae19e86daabc1dec5811411d7a675c642632cb7f1", - "runId": "agent-runtime-0.174.1-proposal-fixture", + "recordDigest": "sha256:b4ed2d8bf4990e8056cc1b2a6bfbe43b49687f30aa24301b8b747d175087e110", + "runId": "agent-runtime-0.175.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.174.1-proposal-fixture" + "runId": "agent-runtime-0.175.0-proposal-fixture" } diff --git a/src/testing/fixtures/agent-profile-improvement-proposal.json b/src/testing/fixtures/agent-profile-improvement-proposal.json index 8fee46b7..a5dede38 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:e2a74c7900f58cbe4a9ae2deff007a13f006a185ea7ab9e969e0f4240e57b556", + "digest": "sha256:0840635c3fe5e602c201bd3bdfeca4d6fc73119f267153413a5171dd22bbca38", "evaluation": { "decision": { "contributingChecks": [ @@ -1715,7 +1715,7 @@ ], "metadata": { "fixture": "agent-profile-improvement-proposal", - "runtimeVersion": "0.174.1" + "runtimeVersion": "0.175.0" }, "objectives": [ { @@ -1826,7 +1826,7 @@ "baselineContentHash": "sha256:21c495a37c418c10bde64fbaa188beddeed31f1f051ea60a6a6582a9ee0db704", "candidateContentHash": "sha256:103f77bc8481601eef1ad5fe6ba84a40dffabc3a44f421f8c8559121edab84e9", "kind": "agent-eval-loop", - "recordDigest": "sha256:519eb9e02c92d6fd5362ad8e0a562dd6b8f6b358e3721de900b1e35b134bc640", + "recordDigest": "sha256:8e30ee0ea01da6b81776049aa21383e0aab6206a8ae8b9196d8414b98ab4d19a", "runId": "profile-improvement-1", "schema": "agent-profile-improvement-experiment" } diff --git a/tests/graph/replay.test.ts b/tests/graph/replay.test.ts index 047f1f6d..c16b41b2 100644 --- a/tests/graph/replay.test.ts +++ b/tests/graph/replay.test.ts @@ -201,6 +201,55 @@ describe('kill-anywhere replay — fold, never checkpoint', () => { }, 120_000) }) +describe('a release killed between its two events runs its node once (#1013)', () => { + it('completes the half-journaled release on restart instead of releasing the node twice', async () => { + const path = journalPath() + const blobs = new InMemoryResultBlobStore() + const runs = new Map() + const spec = chain(runs) + + // Kill at the append that follows the second node's envelope pin — the + // window in which the wave consumption (`join-state`) is not yet durable. + let killed = false + const killAfterPin: SpawnJournal = { + loadTree: (root) => inner.loadTree(root), + beginTree: (root, at) => inner.beginTree(root, at), + appendEvent: async (root, ev) => { + if (killed) throw new KillError('killed after the envelope pin') + if (ev.kind === 'node-inputs-resolved' && ev.instance === 'double#1') { + killed = true + } + return inner.appendEvent(root, ev) + }, + } + const inner = new FileSpawnJournal(path) + await expect( + runEngineGraph(engine(), spec, 'go', { + budget, + perNode, + journal: killAfterPin, + blobs, + runId: 'half-release', + }), + ).rejects.toBeInstanceOf(KillError) + expect(runs.get('double') ?? 0, 'the killed node never ran').toBe(0) + + const resumed = await runEngineGraph(engine(), spec, 'go', { + budget, + perNode, + journal: new FileSpawnJournal(path), + blobs, + runId: 'half-release', + resume: true, + }) + expect(resumed.kind).toBe('winner') + // THE invariant: the re-entered node executed exactly once. Before the + // fix the catch-up release fired a second instance and it ran twice. + expect(runs.get('double')).toBe(1) + expect(runs.get('sink')).toBe(1) + }) +}) + describe('suspensions survive restart (#976)', () => { it('a parked node returns suspended with a recomputable token; resume after restart settles it and the payload flows on', async () => { const path = journalPath()