docs: the session framework slices that landed, and the two that are blocked

take_events kept returning Option<EventBatchView> and defaulting through the
question-mark operator, so a batch missing a field read as end of stream and the
ConsumerEvents enum added in the previous round described nothing. It returns
Batch or Unreadable now, and a test publishes a batch with no droppedBefore,
asserts it is reported as unreadable naming the field, and asserts the next real
batch still reads.

The checkpoint-envelope-v1 amendment cites the rule that lets a required
manifest field land with envelopeVersion still 1 while no production file
exists.

docs/design/session-framework/checkpoint-envelope-v1.md

@@ -100,11 +100,21 @@ names; a manifest missing any of them is not a complete checkpoint.
 | `compositionDigest` | Coordinator scheduler and configuration identity |
 | `portMap` | The exact port-to-agent map, `[{portId, agentId}]` |
 | `compatibility` | Backend, content, patch, controller, parser and state-format identities |
-| `agents` | Per agent: profile, dataset and model identities, resolved seed, tick count, remainder and the payload name holding its state |
+| `agents` | Per agent: profile, dataset, **index** and model identities, resolved seed, tick count, remainder and the payload name holding its state |
 | `coordinator` | Task ledger, prior world inspection, per-agent executor state, admission state and event watermarks, each as a payload name or an inline value |
 | `helperState` | External-helper state required for exact resume, as payload names |
 | `payloads` | `[{name, byteLength, digest}]`, mirroring the payload table |
 
+**Amendment, 2026-09-22 (PUBLISH-01).** The `agents` row gains `indexDigest`, the index the
+agent attested to at `Agent.Initialize`, and it joins that agent's compatibility identity.
+Without it a replacement fly that built another graph -- the same dataset, the same neuron
+count, another index -- passed the group check and was then published under its predecessor's
+`indexDigest`, which is a graph identity crossing a recovery and exactly what section 5's rules
+exist to prevent. It is recorded from the worker's attestation rather than recomputed from the
+dataset, because the point is that the two can disagree. `envelopeVersion` stays `1`, which the
+required-manifest-field rule below allows only while no production `FLYSESS1` file exists; once
+one does, adding a required manifest field must bump it.
+
 **Amendment, 2026-09-22 (STATE-01).** The table above names a holder for every payload except
 the environment's own, although section 6's fixture has one (`world`) and a group install has
 to map it by name like any other participant's. The manifest therefore also records:

docs/design/session-framework/implementation.md

@@ -160,6 +160,10 @@ delayed rendering retains its handle. Distinguish AssetRef from transient Artifa
 
 **Depends on:** SESSION-02, MEDIA-01 and the profile/identity foundation in the broader backlog.
 
+**2026-09-22:** blocked. The profile/identity foundation is FOUNDATION-02
+(`feat/brain-profile-contract`) in the [MaleCNS backlog](../malecns-modular-implementation.md),
+which has not been built. Not started.
+
 **Implement:** adapter over existing LIF, plasticity, retina and fixed readout primitives;
 reference-first composition/goldens; independently seeded agent state and shared immutable data.
 Avoid using the old whole-frame `tick` wrapper if it changes the specified phase ordering.
@@ -172,6 +176,10 @@ dispatch order and varying worker count preserves results. Keep 64-role limits e
 
 **Depends on:** AGENT-01 and environment/task extraction in the broader backlog.
 
+**2026-09-22:** blocked. AGENT-01 is blocked, and environment/task extraction is
+RUNTIME-01 (`refactor/environment-task-boundary`) in the same backlog, which has not been
+built. Not started.
+
 **Implement:** binjgb environment, task-local memory inspector and identity/existing action
 adapter. Keep `legacy-gameboy-v1` separately routed with exact old ordering/hash semantics.
 

docs/design/session-framework/publishing-v1.md

@@ -42,6 +42,20 @@ Example addresses (chosen by composition, not recognized by router code):
 | `app.pokemon.cues` | Pub/sub: application narrative/presentation events under declared delivery policy |
 | `app.pokemon` | RPC: application queries/admission, e.g. restore UI state or request a supported effect |
 
+**Amendment, 2026-09-22 (PUBLISH-01).** The repair path above needs exact methods, and
+"exact methods require session API schemas" left the row unbuildable. The session registers
+one **read-only** service, `session.<id>.query`, with exactly two methods, both ordinary
+[session RPCs](ipc-v1.md) answering from what the session already published:
+`Session.GetDescriptor` takes an optional `{revision: U64}` and returns that `SessionDescriptor`
+or, with no revision, the newest; `Session.GetSnapshot` takes no parameters and returns the
+latest `CommittedSnapshot`. A revision the session never published is `IDENTITY_MISMATCH`, not
+an empty answer. Nothing on this service mutates, selects a participant or reaches a worker, so
+it is not the controller API section 7 rules out; adding a third method that did would be.
+These two names are **internal and provisional**: they are what the internal boundary needs in
+order to be buildable now, and the later public v2 step is free to rename them, supersede them
+or expose a different repair surface entirely. Nothing about them is browser-facing, and the
+public step does not inherit them by default merely because they landed first.
+
 Descriptor revisions and scope link observations to schemas. Cross-topic ordering is not
 guaranteed; a subscriber receiving an unknown descriptor revision must fetch it through the
 application/session query contract or buffer a bounded number of snapshots, not infer shape.
@@ -77,6 +91,16 @@ interface CommittedSnapshot {
 }
 ```
 
+**Amendment, 2026-09-22 (PUBLISH-01).** "Null at initial boundary 0" is the rule for a
+boundary this epoch *produced*. A group restore ([state/media](state-media-v1.md) section 5)
+re-establishes a committed boundary `k > 0` that this epoch did not run a transition into, and
+the abandoned epoch's decisions are not this session's to republish under a new epoch. So the
+rule is: `selectedDecision` and `appliedControls` are null at boundary 0 and at a boundary
+*installed* by a restore, present otherwise, and always **together** and for **every agent or
+none**. A snapshot where one fly carries an action and another does not would be two different
+boundaries in one value, and is refused. Without this, the section 6 requirement to publish the
+recovery could not be met at all: the restored boundary's snapshot would be unrepresentable.
+
 Publish only after all agent commits establish Ready(k). Decisions/controls describe the
 transition ending at that boundary, null at initial boundary 0. Health updates are separate
 and never claim an uncommitted future boundary. Every transient media reference is a declared

docs/design/session-framework/state-media-v1.md

@@ -117,6 +117,21 @@ If it violates configured resource policy, disconnect/restart that observer inst
 live data or silently skipping simulation input. Global store exhaustion is an explicit fault
 or pause condition; the router cannot guess that a particular live object is disposable.
 
+**Amendment, 2026-09-22 (PUBLISH-01).** "Disconnect/restart that observer" names an action
+no participant can take under [Flybus v1](bus-v1.md). Section 5 there makes publish admission
+all or nothing -- "for a bounded subscriber overflow, reject the **whole** publish; no partial
+fan-out or retained-latest update" -- and the router exposes no per-subscriber eviction, so a
+session meeting a full bounded queue cannot drop that one subscriber and deliver to the rest.
+The realisable reading, which the session now implements, is three-part: observation topics
+are published `latest`, and a latest subscriber can never refuse a publication (it loses its
+own queued value and is told how many by `replaced`); a bounded subscriber's refusal, which
+`bus-v1` section 6 explicitly permits, is a named and counted publication outcome that takes
+no world step, stalls nothing and fences no epoch, and the exact value stays recoverable
+through the [publishing-v1](publishing-v1.md) section 2 query path; and disconnecting the
+offender is an operator action against the topic the ledger names, not something the session
+performs. A per-subscriber drop would need a router operation Flybus v1 does not have, and
+inventing one here would be a transport change written into the wrong document.
+
 No coordinator tracks per-reader socket acknowledgments or calls a producer's reclaim method.
 The SDK and bus perform that bookkeeping. File-backed immutable mappings are safe after
 unlink; physical pages disappear when all OS mappings close. Pooled reuse is deferred until

docs/design/session-framework/workers-v1.md

@@ -96,9 +96,29 @@ interface AgentInitializeResult {
   warmupTicks: U64; committedStep: U64; // committedStep == "0"
   decisionContextDigest: Digest;
   telemetry: AgentTelemetry;
+  graph: AgentGraph;
+}
+interface AgentGraph {
+  datasetDigest: Digest; indexDigest: Digest; neuronCount: U64;
+  rateRoles: Id[];          // <=64, unique; AgentTelemetry.rates is in this order
+  supportedStimuli: Id[];   // <=64, unique; an undeclared kind is UNSUPPORTED
 }
 ```
 
+**Amendment, 2026-09-22 (PUBLISH-01).** `AgentInitializeResult` gains `graph`, because
+[publishing-v1](publishing-v1.md) section 3 requires `datasetDigest`, `indexDigest`,
+`neuronCount`, `rateRoles` and `supportedStimuli` in every published `AgentDescriptor` and no
+worker method carried any of them. Without this the only available source is the composition
+that asked for the agent, so a descriptor could only ever agree with itself and the section 3
+rule that "geometry/spike mapping requires indexDigest, not merely the same number of neurons"
+would have nothing to compare. Initialize is where the agent has just loaded its dataset and
+built its index, so the attestation belongs there. `rateRoles` is the "profile-defined order"
+section 1 already requires `AgentTelemetry.rates` to be in, and the result is refused when the
+two disagree; `supportedStimuli` is the profile capability section 1 already requires a
+stimulus kind to resolve through, and a kind outside it is refused with `UNSUPPORTED` before
+the model is touched. It changes `contractDigest`, which [session RPC](ipc-v1.md) section 4
+already provides for.
+
 **Amendment, 2026-09-22 (SESSION-02).** `HelloResult.limits` gains `workerThreads`, an
 integer >=1 reporting the allocation the launcher started that worker within, because
 "within launcher allocation" above had no wire-level proof: the launcher passes the number to

docs/stream-mvp-plan.md

@@ -748,3 +748,29 @@ rewritten separately.
   inside one commit), so the run restarted from rung 8, VIRIDIAN CITY, with
   `fly-reset-to-milestone 8`; the v5 checkpoint migrated to v6 as designed. The previous state is
   archived beside the store.
+
+## 2026-09-22 - session framework: what landed and what stopped
+
+The session framework slices from docs/design/session-framework/implementation.md were built
+in ordered waves, each on its own branch with an independent review before merge.
+
+Landed on main: CONTRACT-01, BUS-01 through BUS-03, SESSION-01, SESSION-02, MEDIA-01,
+STATE-01 and PUBLISH-01. Together they give the repo an executable session contract with a
+TypeScript oracle, a conforming bus with a written conformance table, a lockstep session that
+runs in process, on threads or as one process per fly, native frame and audio observations
+with spectator isolation, a coherent all-participant checkpoint with group restore and a
+liftable fence, and an internal publication boundary with committed snapshots over the same
+bus. Every contract silence met on the way was closed by a dated amendment in the affected
+document rather than by convention; none touched doctrine.
+
+Also landed: the bus test suite asserts guarantees rather than the machine's timing, and a
+coordinator defect found through one of those flakes is fixed, where a lifecycle
+acknowledgement that legitimately releases nothing was treated as a fault.
+
+Stopped: AGENT-01 and ENV-01 are blocked on FOUNDATION-02 and RUNTIME-01 from the MaleCNS
+backlog, which do not exist yet. The profile contract and the environment boundary are
+decisions for the operator, so the swarm stopped here. DOLPHIN-01 was never in scope.
+
+Measured on the development box, not capacity claims: bus RPC near one millisecond at the
+median; a two-fly transition near 10 to 12 ms at the median in every execution mode; about
+5.7 MiB per participant process when split.

packages/session-types/src/publishing.ts

@@ -16,6 +16,7 @@ import {
   type EnvironmentDescriptor,
   MAX_AGENTS,
   MAX_RATE_ROLES,
+  MAX_SUPPORTED_STIMULI,
   type PortControl,
   findPort,
   readAgentTelemetry,
@@ -26,8 +27,9 @@ import {
   validateTelemetryRoles,
 } from './workers';
 
+export { MAX_SUPPORTED_STIMULI } from './workers';
+
 /** Not stated by a document; this crate's choices, published in the schema set. */
-export const MAX_SUPPORTED_STIMULI = 64;
 export const MAX_ASSETS = 64;
 export const MAX_SNAPSHOT_EVENTS = 64;
 
@@ -166,16 +168,24 @@ export function readCommittedSnapshot(value: unknown): CommittedSnapshot {
   const atBoundaryZero = u64(scope.step) === 0n;
   for (const agent of agents) {
     // "Decisions/controls describe the transition ending at that boundary, null at initial
-    // boundary 0." (publishing-v1 section 3)
+    // boundary 0." (publishing-v1 section 3, and its 2026-09-22 amendment for a boundary that
+    // was installed rather than produced.)
     if (atBoundaryZero && (agent.selectedDecision !== null || agent.appliedControls !== null)) {
       fail('CommittedSnapshot: at boundary 0 selectedDecision and appliedControls are null');
     }
-    if (!atBoundaryZero && (agent.selectedDecision === null || agent.appliedControls === null)) {
+    if ((agent.selectedDecision === null) !== (agent.appliedControls === null)) {
       fail(
-        'CommittedSnapshot: past boundary 0 every agent has a decision and applied controls',
+        'CommittedSnapshot: selectedDecision and appliedControls are null together or present together',
       );
     }
   }
+  // A boundary is produced by a transition or installed by one, and the whole snapshot says
+  // which: every agent carries the transition that ended here, or none does.
+  if (agents.some((a) => (a.selectedDecision === null) !== (agents[0].selectedDecision === null))) {
+    fail(
+      'CommittedSnapshot: either every agent carries the transition that ended here, or none does',
+    );
+  }
   return {
     descriptorRevision,
     publisherIncarnation,

packages/session-types/src/workers.ts

@@ -35,6 +35,8 @@ import { readSchemaRef, readTypedValue, readNullableTypedValue } from './common'
 export const MAX_AGENTS = 4;
 export const MAX_PORTS = 4;
 export const MAX_RATE_ROLES = 64;
+/** Declared stimulus kinds per agent. Not a stated bound; recorded in the schema set. */
+export const MAX_SUPPORTED_STIMULI = 64;
 export const MAX_STIMULI = 64;
 export const MAX_REWARDS = 64;
 export const MAX_BUTTONS = 32;
@@ -257,6 +259,14 @@ export interface AgentInitializeParams {
   workerThreads: number;
 }
 
+export interface AgentGraph {
+  datasetDigest: Digest;
+  indexDigest: Digest;
+  neuronCount: U64;
+  rateRoles: Id[];
+  supportedStimuli: Id[];
+}
+
 export interface AgentInitializeResult {
   agentId: Id;
   profileDigest: Digest;
@@ -265,6 +275,7 @@ export interface AgentInitializeResult {
   committedStep: U64;
   decisionContextDigest: Digest;
   telemetry: AgentTelemetry;
+  graph: AgentGraph;
 }
 
 export interface PrepareParams {
@@ -313,6 +324,21 @@ export function readAgentInitializeParams(value: unknown): AgentInitializeParams
   return params;
 }
 
+export function readAgentGraph(value: unknown): AgentGraph {
+  const reader = new Reader(value, 'AgentGraph');
+  const graph: AgentGraph = {
+    datasetDigest: reader.digest('datasetDigest'),
+    indexDigest: reader.digest('indexDigest'),
+    neuronCount: reader.u64('neuronCount'),
+    rateRoles: reader.idList('rateRoles', 0, MAX_RATE_ROLES),
+    supportedStimuli: reader.idList('supportedStimuli', 0, MAX_SUPPORTED_STIMULI),
+  };
+  reader.finish();
+  requireUnique(graph.rateRoles, 'AgentGraph.rateRoles');
+  requireUnique(graph.supportedStimuli, 'AgentGraph.supportedStimuli');
+  return graph;
+}
+
 export function readAgentInitializeResult(value: unknown): AgentInitializeResult {
   const reader = new Reader(value, 'AgentInitializeResult');
   const result: AgentInitializeResult = {
@@ -323,12 +349,15 @@ export function readAgentInitializeResult(value: unknown): AgentInitializeResult
     committedStep: reader.u64('committedStep'),
     decisionContextDigest: reader.digest('decisionContextDigest'),
     telemetry: readAgentTelemetry(reader.value('telemetry')),
+    graph: readAgentGraph(reader.value('graph')),
   };
   reader.finish();
   requirePositiveRational(result.tickDuration, 'AgentInitializeResult.tickDuration');
   if (u64(result.committedStep) !== 0n) {
     fail('AgentInitializeResult: committedStep must be "0"');
   }
+  // The rates a worker reports and the role order it declares are one statement.
+  validateTelemetryRoles(result.telemetry, result.graph.rateRoles);
   return result;
 }
 

services/flysim/crates/fly-session-types/fixtures/contract-digest.json

@@ -1,9 +1,9 @@
 {
   "description": "contractDigest is the SHA-256 of the canonical schema set in schema-set.json.",
-  "contractDigest": "d8f29a49b5df05ad8f75f7f5790a3f8cde9c5ad23a685137474c649c3c9da36d",
+  "contractDigest": "7f4b11d6737e5097c6889657479527174ddf496e50b60322b281e6c7490bcb4a",
   "schemaSetVersion": 1,
-  "schemaSetBytes": 26814,
-  "types": 53,
+  "schemaSetBytes": 27470,
+  "types": 54,
   "enums": 11,
   "limits": 26
 }

services/flysim/crates/fly-session-types/fixtures/invalid.json

@@ -2746,6 +2746,19 @@
             "changed": "2",
             "signal": 0.5
           }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
         }
       },
       "reason": "initialization establishes Ready(0)"
@@ -2782,10 +2795,256 @@
             "changed": "2",
             "signal": 0.5
           }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
         }
       },
       "reason": "durations are positive"
     },
+    {
+      "name": "agent initialize result without a graph identity",
+      "type": "AgentInitializeResult",
+      "value": {
+        "agentId": "fly-a",
+        "profileDigest": "1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0",
+        "tickDuration": {
+          "numerator": "1000000",
+          "denominator": "1"
+        },
+        "warmupTicks": "2500",
+        "committedStep": "0",
+        "decisionContextDigest": "ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4",
+        "telemetry": {
+          "brainTicks": "2500",
+          "populationRateHz": 12.5,
+          "rates": [
+            {
+              "roleId": "kenyon",
+              "hz": 3.25
+            },
+            {
+              "roleId": "mbon",
+              "hz": 0.0
+            }
+          ],
+          "learning": {
+            "enabled": true,
+            "updates": "4",
+            "changed": "2",
+            "signal": 0.5
+          }
+        }
+      },
+      "reason": "publishing-v1 3 needs datasetDigest, indexDigest, neuronCount, rateRoles and supportedStimuli, and only the worker knows them"
+    },
+    {
+      "name": "agent initialize result whose index digest is not a digest",
+      "type": "AgentInitializeResult",
+      "value": {
+        "agentId": "fly-a",
+        "profileDigest": "1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0",
+        "tickDuration": {
+          "numerator": "1000000",
+          "denominator": "1"
+        },
+        "warmupTicks": "2500",
+        "committedStep": "0",
+        "decisionContextDigest": "ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4",
+        "telemetry": {
+          "brainTicks": "2500",
+          "populationRateHz": 12.5,
+          "rates": [
+            {
+              "roleId": "kenyon",
+              "hz": 3.25
+            },
+            {
+              "roleId": "mbon",
+              "hz": 0.0
+            }
+          ],
+          "learning": {
+            "enabled": true,
+            "updates": "4",
+            "changed": "2",
+            "signal": 0.5
+          }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "not-a-digest",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
+        }
+      },
+      "reason": "digests are 64 lowercase hex digits"
+    },
+    {
+      "name": "agent initialize result whose rates are not in the declared role order",
+      "type": "AgentInitializeResult",
+      "value": {
+        "agentId": "fly-a",
+        "profileDigest": "1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0",
+        "tickDuration": {
+          "numerator": "1000000",
+          "denominator": "1"
+        },
+        "warmupTicks": "2500",
+        "committedStep": "0",
+        "decisionContextDigest": "ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4",
+        "telemetry": {
+          "brainTicks": "2500",
+          "populationRateHz": 12.5,
+          "rates": [
+            {
+              "roleId": "kenyon",
+              "hz": 3.25
+            },
+            {
+              "roleId": "mbon",
+              "hz": 0.0
+            }
+          ],
+          "learning": {
+            "enabled": true,
+            "updates": "4",
+            "changed": "2",
+            "signal": 0.5
+          }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "mbon",
+            "kenyon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
+        }
+      },
+      "reason": "AgentTelemetry.rates is in graph.rateRoles order"
+    },
+    {
+      "name": "agent initialize result declaring a role it reports no rate for",
+      "type": "AgentInitializeResult",
+      "value": {
+        "agentId": "fly-a",
+        "profileDigest": "1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0",
+        "tickDuration": {
+          "numerator": "1000000",
+          "denominator": "1"
+        },
+        "warmupTicks": "2500",
+        "committedStep": "0",
+        "decisionContextDigest": "ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4",
+        "telemetry": {
+          "brainTicks": "2500",
+          "populationRateHz": 12.5,
+          "rates": [
+            {
+              "roleId": "kenyon",
+              "hz": 3.25
+            },
+            {
+              "roleId": "mbon",
+              "hz": 0.0
+            }
+          ],
+          "learning": {
+            "enabled": true,
+            "updates": "4",
+            "changed": "2",
+            "signal": 0.5
+          }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon",
+            "pn"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
+        }
+      },
+      "reason": "AgentTelemetry.rates is exactly graph.rateRoles"
+    },
+    {
+      "name": "agent initialize result repeating a supported stimulus",
+      "type": "AgentInitializeResult",
+      "value": {
+        "agentId": "fly-a",
+        "profileDigest": "1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0",
+        "tickDuration": {
+          "numerator": "1000000",
+          "denominator": "1"
+        },
+        "warmupTicks": "2500",
+        "committedStep": "0",
+        "decisionContextDigest": "ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4",
+        "telemetry": {
+          "brainTicks": "2500",
+          "populationRateHz": 12.5,
+          "rates": [
+            {
+              "roleId": "kenyon",
+              "hz": 3.25
+            },
+            {
+              "roleId": "mbon",
+              "hz": 0.0
+            }
+          ],
+          "learning": {
+            "enabled": true,
+            "updates": "4",
+            "changed": "2",
+            "signal": 0.5
+          }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "sugar"
+          ]
+        }
+      },
+      "reason": "supportedStimuli is unique"
+    },
     {
       "name": "hello result for an agent without agent-step-v1",
       "type": "HelloResult",
@@ -3909,7 +4168,179 @@
         },
         "eventIds": []
       },
-      "reason": "a committed transition has applied controls"
+      "reason": "selectedDecision and appliedControls are null together or present together"
+    },
+    {
+      "name": "snapshot where one agent carries the transition and another does not",
+      "type": "CommittedSnapshot",
+      "value": {
+        "descriptorRevision": "7",
+        "publisherIncarnation": "pub-1",
+        "scope": {
+          "sessionId": "demo",
+          "epoch": "epoch-1",
+          "step": "42"
+        },
+        "episodeId": "episode-1",
+        "sequence": "42",
+        "worldTime": {
+          "numerator": "700000000",
+          "denominator": "1"
+        },
+        "agents": [
+          {
+            "agentId": "fly-a",
+            "telemetry": {
+              "brainTicks": "2534",
+              "populationRateHz": 12.5,
+              "rates": [
+                {
+                  "roleId": "kenyon",
+                  "hz": 3.25
+                },
+                {
+                  "roleId": "mbon",
+                  "hz": 0.0
+                }
+              ],
+              "learning": {
+                "enabled": true,
+                "updates": "4",
+                "changed": "2",
+                "signal": 0.5
+              }
+            },
+            "selectedDecision": {
+              "schema": {
+                "id": "gameboy.intent.v1",
+                "version": 1,
+                "digest": "e70451b26fb5462ec63729d8355912dad01c3963a95bec0e6a8452375717ad9d"
+              },
+              "value": {
+                "press": "a"
+              }
+            },
+            "appliedControls": {
+              "portId": "port-1",
+              "buttons": [
+                {
+                  "id": "a",
+                  "down": true
+                },
+                {
+                  "id": "b",
+                  "down": false
+                },
+                {
+                  "id": "start",
+                  "down": false
+                },
+                {
+                  "id": "select",
+                  "down": false
+                },
+                {
+                  "id": "up",
+                  "down": false
+                },
+                {
+                  "id": "down",
+                  "down": false
+                },
+                {
+                  "id": "left",
+                  "down": false
+                },
+                {
+                  "id": "right",
+                  "down": false
+                }
+              ],
+              "axes": [
+                {
+                  "id": "stick-x",
+                  "value": 0.0
+                },
+                {
+                  "id": "trigger",
+                  "value": 0.0
+                }
+              ]
+            }
+          },
+          {
+            "agentId": "fly-b",
+            "telemetry": {
+              "brainTicks": "2534",
+              "populationRateHz": 12.5,
+              "rates": [
+                {
+                  "roleId": "kenyon",
+                  "hz": 3.25
+                },
+                {
+                  "roleId": "mbon",
+                  "hz": 0.0
+                }
+              ],
+              "learning": {
+                "enabled": true,
+                "updates": "4",
+                "changed": "2",
+                "signal": 0.5
+              }
+            },
+            "selectedDecision": null,
+            "appliedControls": null
+          }
+        ],
+        "progress": {
+          "schema": {
+            "id": "pokemon.progress.v1",
+            "version": 1,
+            "digest": "80fa973157f334bb3208dfff6f7a2f298e79f73f07c72c33f5725975c6cff8e1"
+          },
+          "value": {
+            "rank": 10
+          }
+        },
+        "media": {
+          "views": [
+            {
+              "viewId": "screen",
+              "producedStep": "42",
+              "pixels": {
+                "storeId": "store-1",
+                "artifactId": "frame-1",
+                "generation": "1",
+                "byteLength": "92160",
+                "contentType": "image/x-rgba8",
+                "digest": null
+              }
+            }
+          ],
+          "audio": [
+            {
+              "streamId": "mix",
+              "firstSample": "33600",
+              "sampleFrames": 800,
+              "samples": {
+                "storeId": "store-1",
+                "artifactId": "audio-1",
+                "generation": "1",
+                "byteLength": "6400",
+                "contentType": "audio/x-f32le",
+                "digest": null
+              },
+              "discontinuity": false
+            }
+          ]
+        },
+        "eventIds": [
+          "evt-1"
+        ]
+      },
+      "reason": "a boundary is produced or installed for the whole composition, never per agent"
     },
     {
       "name": "trace whose commit acknowledgment is the old boundary",

services/flysim/crates/fly-session-types/fixtures/valid.json

@@ -470,11 +470,24 @@
             "changed": "2",
             "signal": 0.5
           }
+        },
+        "graph": {
+          "datasetDigest": "6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52",
+          "indexDigest": "52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42",
+          "neuronCount": "139255",
+          "rateRoles": [
+            "kenyon",
+            "mbon"
+          ],
+          "supportedStimuli": [
+            "sugar",
+            "shock"
+          ]
         }
       },
       "note": "",
-      "canonical": "{\"agentId\":\"fly-a\",\"committedStep\":\"0\",\"decisionContextDigest\":\"ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4\",\"profileDigest\":\"1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0\",\"telemetry\":{\"brainTicks\":\"2500\",\"learning\":{\"changed\":\"2\",\"enabled\":true,\"signal\":0.5,\"updates\":\"4\"},\"populationRateHz\":12.5,\"rates\":[{\"hz\":3.25,\"roleId\":\"kenyon\"},{\"hz\":0,\"roleId\":\"mbon\"}]},\"tickDuration\":{\"denominator\":\"1\",\"numerator\":\"1000000\"},\"warmupTicks\":\"2500\"}",
-      "digest": "a220160c75d758720fe43145089939201ee35c86bb100fae0c4efdd3c4eafaa6"
+      "canonical": "{\"agentId\":\"fly-a\",\"committedStep\":\"0\",\"decisionContextDigest\":\"ea7792a26f405e2ae9c6f49ca93bbe6076ceac0a1fc53d83426c7d7f2d9377e4\",\"graph\":{\"datasetDigest\":\"6c0af1f0784ef63a393ee77d614e8246c625051360f3f1a48838374c5d355b52\",\"indexDigest\":\"52a7a9b441f686ebea398e3d65dbaa87b4fde4f8b2231fff951700548f6a4e42\",\"neuronCount\":\"139255\",\"rateRoles\":[\"kenyon\",\"mbon\"],\"supportedStimuli\":[\"sugar\",\"shock\"]},\"profileDigest\":\"1900eab6c028483d7126599ee6f50de0d27907b5c65fa90524580b4b0f9852b0\",\"telemetry\":{\"brainTicks\":\"2500\",\"learning\":{\"changed\":\"2\",\"enabled\":true,\"signal\":0.5,\"updates\":\"4\"},\"populationRateHz\":12.5,\"rates\":[{\"hz\":3.25,\"roleId\":\"kenyon\"},{\"hz\":0,\"roleId\":\"mbon\"}]},\"tickDuration\":{\"denominator\":\"1\",\"numerator\":\"1000000\"},\"warmupTicks\":\"2500\"}",
+      "digest": "707803be4867dc2f0f3d4bccedfaa7b97b1e52b9af78d21ae6957caba1a7e3f8"
     },
     {
       "name": "prepare params",
@@ -2598,6 +2611,100 @@
       "canonical": "{\"agents\":[{\"agentId\":\"fly-a\",\"appliedControls\":{\"axes\":[{\"id\":\"stick-x\",\"value\":0},{\"id\":\"trigger\",\"value\":0}],\"buttons\":[{\"down\":true,\"id\":\"a\"},{\"down\":false,\"id\":\"b\"},{\"down\":false,\"id\":\"start\"},{\"down\":false,\"id\":\"select\"},{\"down\":false,\"id\":\"up\"},{\"down\":false,\"id\":\"down\"},{\"down\":false,\"id\":\"left\"},{\"down\":false,\"id\":\"right\"}],\"portId\":\"port-1\"},\"selectedDecision\":{\"schema\":{\"digest\":\"e70451b26fb5462ec63729d8355912dad01c3963a95bec0e6a8452375717ad9d\",\"id\":\"gameboy.intent.v1\",\"version\":1},\"value\":{\"press\":\"a\"}},\"telemetry\":{\"brainTicks\":\"2534\",\"learning\":{\"changed\":\"2\",\"enabled\":true,\"signal\":0.5,\"updates\":\"4\"},\"populationRateHz\":12.5,\"rates\":[{\"hz\":3.25,\"roleId\":\"kenyon\"},{\"hz\":0,\"roleId\":\"mbon\"}]}}],\"descriptorRevision\":\"7\",\"episodeId\":\"episode-1\",\"eventIds\":[\"evt-1\"],\"media\":{\"audio\":[{\"discontinuity\":false,\"firstSample\":\"33600\",\"sampleFrames\":800,\"samples\":{\"artifactId\":\"audio-1\",\"byteLength\":\"6400\",\"contentType\":\"audio/x-f32le\",\"digest\":null,\"generation\":\"1\",\"storeId\":\"store-1\"},\"streamId\":\"mix\"}],\"views\":[{\"pixels\":{\"artifactId\":\"frame-1\",\"byteLength\":\"92160\",\"contentType\":\"image/x-rgba8\",\"digest\":null,\"generation\":\"1\",\"storeId\":\"store-1\"},\"producedStep\":\"42\",\"viewId\":\"screen\"}]},\"progress\":{\"schema\":{\"digest\":\"80fa973157f334bb3208dfff6f7a2f298e79f73f07c72c33f5725975c6cff8e1\",\"id\":\"pokemon.progress.v1\",\"version\":1},\"value\":{\"rank\":10}},\"publisherIncarnation\":\"pub-1\",\"scope\":{\"epoch\":\"epoch-1\",\"sessionId\":\"demo\",\"step\":\"42\"},\"sequence\":\"42\",\"worldTime\":{\"denominator\":\"1\",\"numerator\":\"700000000\"}}",
       "digest": "d5ed83ccb866318b3cf17b53c8a5b1dea2f404b10649f6b640bf5f2b88a31435"
     },
+    {
+      "name": "committed snapshot at a boundary installed by a restore",
+      "type": "CommittedSnapshot",
+      "value": {
+        "descriptorRevision": "7",
+        "publisherIncarnation": "pub-1",
+        "scope": {
+          "sessionId": "demo",
+          "epoch": "epoch-1",
+          "step": "42"
+        },
+        "episodeId": "episode-1",
+        "sequence": "42",
+        "worldTime": {
+          "numerator": "700000000",
+          "denominator": "1"
+        },
+        "agents": [
+          {
+            "agentId": "fly-a",
+            "telemetry": {
+              "brainTicks": "2534",
+              "populationRateHz": 12.5,
+              "rates": [
+                {
+                  "roleId": "kenyon",
+                  "hz": 3.25
+                },
+                {
+                  "roleId": "mbon",
+                  "hz": 0.0
+                }
+              ],
+              "learning": {
+                "enabled": true,
+                "updates": "4",
+                "changed": "2",
+                "signal": 0.5
+              }
+            },
+            "selectedDecision": null,
+            "appliedControls": null
+          }
+        ],
+        "progress": {
+          "schema": {
+            "id": "pokemon.progress.v1",
+            "version": 1,
+            "digest": "80fa973157f334bb3208dfff6f7a2f298e79f73f07c72c33f5725975c6cff8e1"
+          },
+          "value": {
+            "rank": 10
+          }
+        },
+        "media": {
+          "views": [
+            {
+              "viewId": "screen",
+              "producedStep": "42",
+              "pixels": {
+                "storeId": "store-1",
+                "artifactId": "frame-1",
+                "generation": "1",
+                "byteLength": "92160",
+                "contentType": "image/x-rgba8",
+                "digest": null
+              }
+            }
+          ],
+          "audio": [
+            {
+              "streamId": "mix",
+              "firstSample": "33600",
+              "sampleFrames": 800,
+              "samples": {
+                "storeId": "store-1",
+                "artifactId": "audio-1",
+                "generation": "1",
+                "byteLength": "6400",
+                "contentType": "audio/x-f32le",
+                "digest": null
+              },
+              "discontinuity": false
+            }
+          ]
+        },
+        "eventIds": [
+          "evt-1"
+        ]
+      },
+      "note": "a restore re-establishes a committed boundary this epoch did not run a transition into",
+      "canonical": "{\"agents\":[{\"agentId\":\"fly-a\",\"appliedControls\":null,\"selectedDecision\":null,\"telemetry\":{\"brainTicks\":\"2534\",\"learning\":{\"changed\":\"2\",\"enabled\":true,\"signal\":0.5,\"updates\":\"4\"},\"populationRateHz\":12.5,\"rates\":[{\"hz\":3.25,\"roleId\":\"kenyon\"},{\"hz\":0,\"roleId\":\"mbon\"}]}}],\"descriptorRevision\":\"7\",\"episodeId\":\"episode-1\",\"eventIds\":[\"evt-1\"],\"media\":{\"audio\":[{\"discontinuity\":false,\"firstSample\":\"33600\",\"sampleFrames\":800,\"samples\":{\"artifactId\":\"audio-1\",\"byteLength\":\"6400\",\"contentType\":\"audio/x-f32le\",\"digest\":null,\"generation\":\"1\",\"storeId\":\"store-1\"},\"streamId\":\"mix\"}],\"views\":[{\"pixels\":{\"artifactId\":\"frame-1\",\"byteLength\":\"92160\",\"contentType\":\"image/x-rgba8\",\"digest\":null,\"generation\":\"1\",\"storeId\":\"store-1\"},\"producedStep\":\"42\",\"viewId\":\"screen\"}]},\"progress\":{\"schema\":{\"digest\":\"80fa973157f334bb3208dfff6f7a2f298e79f73f07c72c33f5725975c6cff8e1\",\"id\":\"pokemon.progress.v1\",\"version\":1},\"value\":{\"rank\":10}},\"publisherIncarnation\":\"pub-1\",\"scope\":{\"epoch\":\"epoch-1\",\"sessionId\":\"demo\",\"step\":\"42\"},\"sequence\":\"42\",\"worldTime\":{\"denominator\":\"1\",\"numerator\":\"700000000\"}}",
+      "digest": "79a0ef10ed65c0720c34c7d99de1fe87ec2f0c26028994562cc4655b0f6acd1f"
+    },
     {
       "name": "transition trace",
       "type": "TransitionTrace",

services/flysim/crates/fly-session-types/src/publishing.rs

@@ -15,8 +15,9 @@ use crate::workers::{
     AgentTelemetry, AssetRef, EnvironmentDescriptor, MAX_AGENTS, MAX_RATE_ROLES, PortControl,
 };
 
-/// Declared stimulus kinds per agent. Not a stated bound; recorded in the schema set.
-pub const MAX_SUPPORTED_STIMULI: usize = 64;
+/// Declared stimulus kinds per agent, re-exported from its defining module.
+pub use crate::workers::MAX_SUPPORTED_STIMULI;
+
 /// Installed assets in one descriptor. Not a stated bound; recorded in the schema set.
 pub const MAX_ASSETS: usize = 64;
 /// Scoped event ids in one snapshot. Not a stated bound; recorded in the schema set.
@@ -413,7 +414,8 @@ impl DomainType for CommittedSnapshot {
                 controls.validate()?;
             }
             // "Decisions/controls describe the transition ending at that boundary, null at
-            // initial boundary 0." (publishing-v1 section 3)
+            // initial boundary 0." (publishing-v1 section 3, and its 2026-09-22 amendment for
+            // a boundary that was installed rather than produced.)
             if self.scope.step == 0
                 && (agent.selected_decision.is_some() || agent.applied_controls.is_some())
             {
@@ -421,14 +423,24 @@ impl DomainType for CommittedSnapshot {
                     "CommittedSnapshot: at boundary 0 selectedDecision and appliedControls are null",
                 );
             }
-            if self.scope.step > 0
-                && (agent.selected_decision.is_none() || agent.applied_controls.is_none())
-            {
+            if agent.selected_decision.is_some() != agent.applied_controls.is_some() {
                 return err(
-                    "CommittedSnapshot: past boundary 0 every agent has a decision and applied controls",
+                    "CommittedSnapshot: selectedDecision and appliedControls are null together or present together",
                 );
             }
         }
+        // A boundary is produced by a transition or installed by one, and the whole snapshot
+        // says which: every agent carries the transition that ended here, or none does. A
+        // mixture would be one fly's action beside another fly's silence at the same boundary.
+        if self
+            .agents
+            .iter()
+            .any(|a| a.selected_decision.is_some() != self.agents[0].selected_decision.is_some())
+        {
+            return err(
+                "CommittedSnapshot: either every agent carries the transition that ended here, or none does",
+            );
+        }
         self.progress.validate()?;
         if self.views.len() > MAX_VIEWS {
             return err("CommittedSnapshot: at most 8 views");

services/flysim/crates/fly-session-types/src/schema.rs

@@ -438,6 +438,22 @@ pub const SCHEMAS: &[TypeSchema] = &[
             req("committedStep", "U64", "\"0\""),
             req("decisionContextDigest", "Digest", ""),
             req("telemetry", "AgentTelemetry", ""),
+            req("graph", "AgentGraph", "rates are in graph.rateRoles order"),
+        ],
+    },
+    TypeSchema {
+        name: "AgentGraph",
+        source: "workers-v1 2",
+        fields: &[
+            req("datasetDigest", "Digest", ""),
+            req(
+                "indexDigest",
+                "Digest",
+                "geometry mapping needs this, not neuronCount",
+            ),
+            req("neuronCount", "U64", ""),
+            req("rateRoles", "array<Id>", "<= 64, unique"),
+            req("supportedStimuli", "array<Id>", "<= 64, unique"),
         ],
     },
     TypeSchema {
@@ -924,12 +940,12 @@ pub const SCHEMAS: &[TypeSchema] = &[
             opt(
                 "selectedDecision",
                 "TypedValue|null",
-                "null exactly at boundary 0",
+                "null at boundary 0 and at an installed boundary; null or present for every agent together",
             ),
             opt(
                 "appliedControls",
                 "PortControl|null",
-                "null exactly at boundary 0; the agent's assigned port",
+                "null with selectedDecision; the agent's assigned port",
             ),
         ],
     },

services/flysim/crates/fly-session-types/src/workers.rs

@@ -19,6 +19,8 @@ pub const MAX_AGENTS: usize = 4;
 pub const MAX_PORTS: usize = 4;
 /// 64 rate roles per agent.
 pub const MAX_RATE_ROLES: usize = 64;
+/// Declared stimulus kinds per agent. Not a stated bound; recorded in the schema set.
+pub const MAX_SUPPORTED_STIMULI: usize = 64;
 /// Arrays of stimuli or rewards are bounded to 64 per operation (workers-v1 section 1).
 pub const MAX_STIMULI: usize = 64;
 /// Arrays of stimuli or rewards are bounded to 64 per operation.
@@ -706,6 +708,92 @@ pub struct AgentInitializeResult {
     pub committed_step: u64,
     pub decision_context_digest: String,
     pub telemetry: AgentTelemetry,
+    /// The graph identity this agent actually loaded, which is what a descriptor publishes.
+    ///
+    /// `publishing-v1` section 3 requires `datasetDigest`, `indexDigest`, `neuronCount`,
+    /// `rateRoles` and `supportedStimuli` in every `AgentDescriptor`, and before the
+    /// 2026-09-22 amendment to `workers-v1` section 2 no worker method carried them: a
+    /// coordinator could only have restated its own configuration. The worker attests
+    /// instead, so a fly that built another index is a visible mismatch rather than a
+    /// descriptor that agrees with itself.
+    pub graph: AgentGraph,
+}
+
+/// What one agent's loaded graph is, as the agent reports it.
+///
+/// `neuronCount` does not identify a mapping: "geometry/spike mapping requires indexDigest,
+/// not merely the same number of neurons" (publishing-v1 section 3), so both travel and a
+/// consumer compares the digest.
+#[derive(Clone, Debug, PartialEq)]
+pub struct AgentGraph {
+    pub dataset_digest: String,
+    pub index_digest: String,
+    pub neuron_count: u64,
+    /// The profile-defined rate-role order. `AgentTelemetry.rates` is in exactly this order.
+    pub rate_roles: Vec<String>,
+    pub supported_stimuli: Vec<String>,
+}
+
+impl AgentGraph {
+    pub fn from_json(value: &Value) -> Result<AgentGraph> {
+        let mut f = Fields::new(value, "AgentGraph")?;
+        let dataset_digest = f.string("datasetDigest")?.to_owned();
+        let index_digest = f.string("indexDigest")?.to_owned();
+        let neuron_count = f.u64_string("neuronCount")?;
+        let rate_roles = id_list(&mut f, "rateRoles", 0, MAX_RATE_ROLES)?;
+        let supported_stimuli = id_list(&mut f, "supportedStimuli", 0, MAX_SUPPORTED_STIMULI)?;
+        f.finish()?;
+        let g = AgentGraph {
+            dataset_digest,
+            index_digest,
+            neuron_count,
+            rate_roles,
+            supported_stimuli,
+        };
+        g.validate()?;
+        Ok(g)
+    }
+
+    pub fn to_json(&self) -> Value {
+        obj(vec![
+            ("datasetDigest", self.dataset_digest.clone().into()),
+            ("indexDigest", self.index_digest.clone().into()),
+            ("neuronCount", u64_json(self.neuron_count)),
+            (
+                "rateRoles",
+                Value::Array(self.rate_roles.iter().map(|r| r.clone().into()).collect()),
+            ),
+            (
+                "supportedStimuli",
+                Value::Array(
+                    self.supported_stimuli
+                        .iter()
+                        .map(|s| s.clone().into())
+                        .collect(),
+                ),
+            ),
+        ])
+    }
+
+    pub fn validate(&self) -> Result<()> {
+        if !is_digest(&self.dataset_digest) || !is_digest(&self.index_digest) {
+            return err("AgentGraph: datasetDigest and indexDigest must be 64 lowercase hex digits");
+        }
+        if self.rate_roles.len() > MAX_RATE_ROLES {
+            return err("AgentGraph: at most 64 rate roles");
+        }
+        if self.supported_stimuli.len() > MAX_SUPPORTED_STIMULI {
+            return err("AgentGraph: at most 64 supported stimuli");
+        }
+        require_unique(
+            self.rate_roles.iter().map(String::as_str),
+            "AgentGraph.rateRoles",
+        )?;
+        require_unique(
+            self.supported_stimuli.iter().map(String::as_str),
+            "AgentGraph.supportedStimuli",
+        )
+    }
 }
 
 impl DomainType for AgentInitializeResult {
@@ -720,6 +808,7 @@ impl DomainType for AgentInitializeResult {
         let committed_step = f.u64_string("committedStep")?;
         let decision_context_digest = f.string("decisionContextDigest")?.to_owned();
         let telemetry = AgentTelemetry::from_json(f.value("telemetry")?)?;
+        let graph = AgentGraph::from_json(f.value("graph")?)?;
         f.finish()?;
         let r = AgentInitializeResult {
             agent_id,
@@ -729,6 +818,7 @@ impl DomainType for AgentInitializeResult {
             committed_step,
             decision_context_digest,
             telemetry,
+            graph,
         };
         r.validate()?;
         Ok(r)
@@ -746,6 +836,7 @@ impl DomainType for AgentInitializeResult {
                 self.decision_context_digest.clone().into(),
             ),
             ("telemetry", self.telemetry.to_json()),
+            ("graph", self.graph.to_json()),
         ])
     }
 
@@ -762,7 +853,10 @@ impl DomainType for AgentInitializeResult {
         if self.committed_step != 0 {
             return err("AgentInitializeResult: committedStep must be \"0\"");
         }
-        self.telemetry.validate()
+        self.graph.validate()?;
+        // The rates a worker reports and the role order it declares are one statement, so a
+        // descriptor built from the second can never mislabel the first.
+        self.telemetry.validate_against_roles(&self.graph.rate_roles)
     }
 }
 

services/flysim/crates/fly-session/README.md

@@ -39,6 +39,7 @@ Ready(k) ─ Prepare all agents concurrently ───────────
 | `environment` | The counter arena: one complete batch per advance, one native frame |
 | `task` | The task and executor traits, the deterministic counter task, the identity executor |
 | `rpc` | Domain calls: `req-<U64>` serials, incarnation pinning, the retry rule |
+| `publish` | The publication boundary: declared delivery policies, named publication outcomes, the bounded event batch, the read-only repair service, an application channel and a fake multi-agent consumer |
 | `coordinator` | The transaction, the trace, the failure rules and the publication boundary |
 | `launcher` | The supervisor: thread budget, identities, start, health check, reap |
 | `metrics` | Latency percentiles and the machine's core and memory counters |
@@ -217,6 +218,22 @@ The durable store is `state`, over the `FLYSESS1` layout the contract crate owns
   derived from the epoch, so a resumed run's behaviour is compared through
   `EpochRebase`, which rewrites exactly those and fails on anything it does not recognise.
 
+## Before you add a check to a reply
+
+Ask which kind of reply it is. Is the far side **reporting what it did**, in which case a
+subset or an empty answer is permitted and must be accepted? Or is it **being held to a
+requirement**, in which case exactness is the rule and must be enforced? `Worker.Acknowledge`
+is the only reply of the first kind in this crate, because `ipc-v1` section 5 explicitly makes
+it idempotent -- "Already released/unknown IDs are ignored" -- so a second one legitimately
+releases nothing, and the section 6 resolution turns any slow Acknowledge into exactly that
+second one. Demanding the whole list back there fenced healthy sessions until
+`an_acknowledge_that_releases_nothing_is_not_a_failure` was written.
+
+The commit and batch checks are the second kind and must stay exact: `commit_all` requires
+every agent (`step-v1` section 3 phase D, section 7) and `check_batch` requires every declared
+port (`workers-v1` section 3). Loosening those in the name of tolerance is the same mistake
+pointing the other way -- they are what make a partial commit and an incomplete batch fail.
+
 ## Where this crate narrows or adds to the contract crate
 
 - **Required views.** `WorldObservation::validate_against` checks the views a result carries
@@ -231,6 +248,29 @@ The durable store is `state`, over the `FLYSESS1` layout the contract crate owns
   it takes -- the transition finishes, then the session pauses at the boundary it just
   committed -- is now written into the section 2 machine as a dated amendment.
 
+## The publication boundary
+
+`publishing-v1` on the same bus, with nothing added to the router:
+
+| Address | Delivery | Contents |
+| --- | --- | --- |
+| `session.<id>.descriptor` | retained latest | `SessionDescriptor`, built from what each participant attested to |
+| `session.<id>.snapshots` | retained latest | `CommittedSnapshot` plus the boundary's media handles |
+| `session.<id>.events` | bounded, depth 64 | the transition's task events, with a `droppedBefore` count |
+| `session.<id>.query` | RPC, read-only | `Session.GetDescriptor`, `Session.GetSnapshot` |
+| `<app>.state`, `<app>.cues` | the application's own | whatever the experience needs, under the application's schema |
+
+Every publication returns a named outcome: `Accepted`, `RefusedByObserver` or `Faulted`. Only
+`BACKPRESSURE` is an observer's refusal, and a refusal takes no world step, stalls nothing and
+fences no epoch -- it is counted per topic in the ledger and the exact value stays readable
+through the query service. Anything else is the session's own fault and fails the epoch. A
+snapshot is checked before it is published and again when it is read: every frame comes from
+the boundary its declared delay implies, every handle is the artifact its reference names,
+audio never goes backwards, and the snapshot agrees with the descriptor revision it names.
+
+What is **not** here: the approved public v2 wire schemas and the stage adapters that speak
+them. `implementation.md` sequences those after this slice and together with each other.
+
 ## Limitations
 
 - **Fake workers.** There is no neural model and no emulator. What is modelled exactly is the
@@ -239,6 +279,14 @@ The durable store is `state`, over the `FLYSESS1` layout the contract crate owns
   restore refuses one taken under another backend, content, patch, controller or parser
   identity. It does not migrate between compositions, and it does not try.
 - **No audience input.** The admitted pre-step stimulation list exists and is always empty.
+- **One descriptor revision.** A revision changes when the composition does, and the only
+  in-session path to that is a group restore into a fresh epoch, which is STATE-01's. The
+  session publishes revision 1; the repair path, the revision cache and the index-change rule
+  are exercised against a second revision published by a `Publisher` of a second composition.
+- **No per-subscriber eviction.** A bounded subscriber may refuse a publication, and Flybus v1
+  has no operation to drop that one subscriber, so the refusal costs every subscriber that
+  boundary's delivery on a stream whose contract is "latest". See the 2026-09-22 amendment to
+  `state-media-v1` section 3.
 - **Pacing is coarse.** The pacing deadline rounds one step to whole nanoseconds for sleeping
   only; simulation time stays rational and that rounding never re-enters the accumulator.
 
@@ -298,9 +346,13 @@ The three integration suites do not all run over both transports, and cannot:
 - `tests/processes.rs` runs over the Unix socket only, in all three execution modes. A
   participant in a process of its own has no in-memory transport to reach the router by, so
   the mode is the axis that suite varies and the transport is fixed.
-- `tests/media.rs` and `tests/state.rs` run over both transports *and* in all three execution
-  modes: each acceptance body is written once and registered twice, by `both_transports!` in
-  the in-process composition and by `all_modes!` over the socket.
+- `tests/media.rs`, `tests/state.rs` and `tests/publishing.rs` run over both transports *and*
+  in the execution modes: each acceptance body is written once and registered twice, by
+  `both_transports!` in the in-process composition and by `all_modes!` over the socket.
+  `tests/publishing.rs` registers a subset that way rather than all of it, because the
+  publication boundary lives in the coordinator: unlike the render counter and the sensor log
+  it crosses no process boundary and stays fully observable in all three modes, which
+  `the_publication_boundary_holds_in_every_execution_mode` asserts rather than assumes.
 
 - `tests/session.rs`: one world advance per complete batch; every agent Prepared before the
   advance; one task evaluation per transition; every agent committed before the next Prepare or
@@ -317,6 +369,14 @@ The three integration suites do not all run over both transports, and cannot:
   allocation -- plus the sequential/reversed/parallel trace comparison across all three modes
   and the two process-mode section 4 rows: a router restart during a world advance, and an old
   worker's reply after a restart.
+- `tests/publishing.rs`: the PUBLISH-01 acceptance bullets over both transports -- a consumer
+  that disconnects and one that stops consuming, a bounded observer's named refusal, every
+  boundary's media belonging to that boundary, a frame and a handle from another boundary
+  refused, an unheld revision repaired rather than inferred, a revision that was never
+  published, an index that moved under a mapped consumer, boundary 0's null decision, the
+  committed action being the transition that just ended, one snapshot carrying every agent,
+  application-owned state and cues, a held event batch, and the read-only query service --
+  plus the first two generated once per execution mode by `all_modes!`.
 - `tests/state.rs`: the STATE-01 acceptance bullets -- an uninterrupted run and a resumed run
   committing the same behaviour once the epoch metadata is rebased, a corrupt payload failing
   the install as a group for every participant and for the coordinator's own ledger, a lost

services/flysim/crates/fly-session/src/agent.rs

@@ -199,6 +199,9 @@ pub struct AgentFaults {
     /// Refuse `State.ActivateRestore` after this worker has already staged, so a group meets
     /// a failure halfway through activation.
     pub fail_activate_restore: bool,
+    /// Add this id to every `Worker.Acknowledge` reply, so the caller meets a worker
+    /// reporting about an id it was never asked about.
+    pub acknowledge_extra_id: Option<Id>,
 }
 
 /// One fake agent worker's configuration.
@@ -212,6 +215,10 @@ pub struct AgentConfig {
     /// The thread allocation the launcher started this worker within. `workers-v1` requires
     /// `Agent.Initialize`'s `workerThreads` to lie inside it.
     pub worker_threads: usize,
+    /// Which graph this fly built. Two variants have the same `neuronCount` and different
+    /// `indexDigest`, which is the case `publishing-v1` section 3 says a consumer must not
+    /// mistake for the same mapping.
+    pub graph_variant: u64,
     /// Records every view this agent read, so a test can see which artifact reached it.
     ///
     /// It is this process's log: an agent with a process of its own writes to its own copy,
@@ -456,6 +463,9 @@ impl FakeAgentWorker {
             committed_step: 0,
             decision_context_digest: self.context_digest.clone().expect("just set"),
             telemetry: self.model.telemetry(),
+            // The worker attests to the graph it loaded. A descriptor built from this can
+            // disagree with the composition; one built from the composition never could.
+            graph: synthetic_graph(&self.config.agent_id, self.config.graph_variant),
         };
         Ok(HandlerReply::from(&result))
     }
@@ -507,6 +517,7 @@ impl FakeAgentWorker {
         }
         for stimulus in &params.pre_step_stimulations {
             stimulus.validate().map_err(DomainError::invalid)?;
+            check_supported(stimulus)?;
         }
         let available =
             FakeAgentWorker::available_actions(self.context.as_ref().expect("initialized"))?;
@@ -594,6 +605,7 @@ impl FakeAgentWorker {
         }
         for stimulus in &params.task_stimulations {
             stimulus.validate().map_err(DomainError::invalid)?;
+            check_supported(stimulus)?;
         }
         params.next_decision_context.validate().map_err(DomainError::invalid)?;
         FakeAgentWorker::available_actions(&params.next_decision_context)?;
@@ -689,6 +701,10 @@ impl WorkerEndpoint for FakeAgentWorker {
         self.config.worker_threads as u64
     }
 
+    fn acknowledge_extra_id(&self) -> Option<Id> {
+        self.config.faults.acknowledge_extra_id.clone()
+    }
+
     fn methods(&self) -> Vec<&'static str> {
         vec![
             "Agent.Initialize",
@@ -733,13 +749,61 @@ pub fn agent_op_class(method: &str) -> Option<OpClass> {
 }
 
 /// A synthetic profile asset for one agent. The digest covers its effective identities.
+/// Refuses a stimulus kind this profile does not resolve, before the model is touched.
+///
+/// `supportedStimuli` in a published descriptor is exactly this list, so the declaration is
+/// what the worker enforces rather than a label printed beside it.
+fn check_supported(stimulus: &Stimulus) -> DomainResult<()> {
+    if SUPPORTED_STIMULI.contains(&stimulus.kind_id.as_str()) {
+        return Ok(());
+    }
+    Err(DomainError::before(
+        ErrorCode::Unsupported,
+        format!(
+            "stimulus kind {} is not one this profile resolves",
+            stimulus.kind_id
+        ),
+    ))
+}
+
+/// The rate roles this fake model reports, in the order it reports them.
+pub const RATE_ROLES: [&str; 2] = ["kc", "mbon"];
+
+/// The stimulus kinds this synthetic profile resolves. An undeclared kind is refused before
+/// the model is touched, so `supportedStimuli` in a descriptor is what the worker enforces
+/// rather than a label beside it.
+pub const SUPPORTED_STIMULI: [&str; 1] = ["arena.milestone"];
+
+/// This fly's graph identity. Every variant has the same neuron count and its own index, so
+/// "the same number of neurons" can never be mistaken for the same mapping.
+pub const NEURON_COUNT: u64 = 1024;
+
+pub fn synthetic_graph(agent_id: &Id, variant: u64) -> AgentGraph {
+    AgentGraph {
+        dataset_digest: digest_of_bytes(
+            format!("arena-dataset-v1\nvariant={variant}\n").as_bytes(),
+        ),
+        index_digest: digest_of_bytes(
+            format!(
+                "arena-index-v1\nagent={agent_id}\nvariant={variant}\nneurons={NEURON_COUNT}\n"
+            )
+            .as_bytes(),
+        ),
+        neuron_count: NEURON_COUNT,
+        rate_roles: RATE_ROLES.iter().map(|r| id(r)).collect(),
+        supported_stimuli: SUPPORTED_STIMULI.iter().map(|s| id(s)).collect(),
+    }
+}
+
 pub fn synthetic_profile(agent_id: &Id, tick_duration: &RationalNs, warmup_ticks: u64) -> AssetRef {
     let text = format!(
         "arena-direct-v1\nagent={agent_id}\ntick={}/{}\nwarmup={warmup_ticks}\n",
         tick_duration.numerator, tick_duration.denominator
     );
     AssetRef {
-        id: id("arena-direct-v1"),
+        // One installed asset per fly: a descriptor's `assets` are unique by id, and two
+        // profiles that differ in content are two assets, not one id with two digests.
+        id: parse_id(&format!("arena-direct-v1-{agent_id}")).expect("a prefix plus an agent id"),
         digest: digest_of_bytes(text.as_bytes()),
         byte_length: text.len() as u64,
         format: id("fly-profile-v1"),
@@ -787,6 +851,7 @@ pub fn agent_compatibility_digest(
     model_version: &str,
     plasticity_version: &str,
     seed: i32,
+    index_digest: &Digest,
 ) -> Digest {
     let value = serde_json::json!({
         "agentId": agent_id.as_str(),
@@ -795,6 +860,11 @@ pub fn agent_compatibility_digest(
         "modelVersion": model_version,
         "plasticityVersion": plasticity_version,
         "seed": seed,
+        // The index the worker actually built, not a value recomputed from the dataset: the
+        // whole point is that the two can disagree. Without it a replacement fly that built
+        // another graph restores cleanly and is then published under its predecessor's
+        // `indexDigest`, which is the predecessor's graph identity crossing a recovery.
+        "indexDigest": index_digest.as_str(),
     });
     digest_of(&value).expect("an agent compatibility block canonicalizes")
 }
@@ -883,13 +953,15 @@ struct StagedAgent {
 impl FakeAgentWorker {
     /// This worker's own compatibility identity, from its configuration and a resolved seed.
     fn compatibility_digest(&self, profile: &AssetRef, seed: i32) -> Digest {
+        let graph = synthetic_graph(&self.config.agent_id, self.config.graph_variant);
         agent_compatibility_digest(
             &self.config.agent_id,
             &profile.digest,
-            &dataset_digest(),
+            &graph.dataset_digest,
             MODEL_VERSION,
             PLASTICITY_VERSION,
             seed,
+            &graph.index_digest,
         )
     }
 
@@ -1090,10 +1162,16 @@ worker; this worker is {other:?}"
         // of another agent's brain, fails here and never reaches activation.
         let computed = self.compatibility_digest(&profile, model.seed());
         if computed != params.compatibility_digest {
+            let graph = synthetic_graph(&self.config.agent_id, self.config.graph_variant);
             return Err(incompatible(format!(
-                "the staged state's compatibility {computed} is not the {} the restore \
-requires",
-                params.compatibility_digest
+                "the staged state's compatibility {} is not the {computed} this worker is: \
+profile {}, dataset {}, index {}, model {MODEL_VERSION}, plasticity {PLASTICITY_VERSION}, \
+seed {}",
+                params.compatibility_digest,
+                profile.digest,
+                graph.dataset_digest,
+                graph.index_digest,
+                model.seed()
             )));
         }
         let accumulator_value = value

services/flysim/crates/fly-session/src/cli.rs

@@ -204,6 +204,7 @@ fn serve(role: &str, options: &Options) -> Result<(), String> {
             tick_duration: options.rational(flags::TICK_NUMERATOR, flags::TICK_DENOMINATOR)?,
             warmup_ticks: options.u64(flags::WARMUP_TICKS, 0)?,
             worker_threads: threads,
+            graph_variant: options.u64(flags::GRAPH_VARIANT, 0)?,
             // This process's own log. The supervisor reads what crosses the bus, not this.
             sensors: crate::media::SensorLog::new(),
             faults: AgentFaults {
@@ -212,6 +213,9 @@ fn serve(role: &str, options: &Options) -> Result<(), String> {
                 commit_delay_ms: options.u64(flags::COMMIT_DELAY_MS, 0)?,
                 fail_stage_restore: options.flag(flags::FAIL_STAGE_RESTORE)?,
                 fail_activate_restore: options.flag(flags::FAIL_ACTIVATE_RESTORE)?,
+                // A worker process is never asked to misbehave this way: the subset refusal
+                // is a caller-side check and its test runs the worker in-process.
+                acknowledge_extra_id: None,
             },
             client_id: client_id.clone(),
             service: service.clone(),

services/flysim/crates/fly-session/src/coordinator.rs

@@ -16,6 +16,7 @@ use serde_json::{Map, Value, json};
 
 use crate::clock::Pacing;
 use crate::media::{self, AudioTimelines};
+use crate::publish::PublicationOutcome;
 use crate::metrics::Metrics;
 use crate::phase::{Phase, PhaseMachine};
 use crate::rpc::{self, DomainReply, Serials, WorkerRef};
@@ -54,6 +55,18 @@ pub struct Injections {
     pub altered_advance_controls: bool,
     /// Read and release the Advance result's frame, then replay the same operation.
     pub consume_advance_artifact_then_retry: bool,
+    /// Publish this boundary's snapshot naming the previous boundary's frame: new agent
+    /// state beside an older observation.
+    pub stale_published_view: bool,
+    /// Publish this boundary's snapshot with a handle that is not the artifact the snapshot
+    /// references: the same name, the same shape, another object.
+    pub substituted_published_handle: bool,
+    /// Ask an agent to apply a stimulus kind its published descriptor does not declare.
+    pub undeclared_stimulus: bool,
+    /// Acknowledge the lifecycle replies twice, which is what the `ipc-v1` section 6
+    /// resolution does to any Acknowledge whose first reply outran the probe. The second one
+    /// legitimately releases nothing, and bootstrap must accept it.
+    pub duplicate_lifecycle_acknowledge: bool,
 }
 
 /// What an injection produced, for a test to assert on.
@@ -183,6 +196,14 @@ impl Default for Deadlines {
     }
 }
 
+/// The descriptor revision this slice publishes.
+///
+/// A revision changes when the composition does -- a replaced fly with another index, a
+/// different port assignment -- and the only in-session path to that is a group restore into
+/// a fresh epoch, which is STATE-01's. So a session establishes revision 1 and the repair
+/// path, not a revision counter with nothing to count.
+pub const DESCRIPTOR_REVISION: u64 = 1;
+
 /// The bus addresses this session publishes on. Chosen by the composition, not the router.
 #[derive(Clone, Debug)]
 pub struct Topics {
@@ -217,6 +238,11 @@ pub struct AgentSlot {
     pub tick_duration: RationalNs,
     pub warmup_ticks: u64,
     pub committed_step: u64,
+    /// The graph this fly attested to at `Agent.Initialize`, which is what the published
+    /// descriptor says about it. `None` before initialization.
+    pub graph: Option<AgentGraph>,
+    /// The telemetry of the last committed boundary, which is what the snapshot publishes.
+    pub telemetry: Option<AgentTelemetry>,
     /// The tick count and remainder this agent last reported, which are what the checkpoint
     /// manifest records for it. They are metadata about the payload, never a substitute for
     /// it: the agent's own capture is the state that is restored.
@@ -246,6 +272,8 @@ impl AgentSlot {
             tick_duration: RationalNs::ZERO,
             warmup_ticks: 0,
             committed_step: 0,
+            graph: None,
+            telemetry: None,
             brain_ticks: 0,
             remainder: RationalNs::ZERO,
             context: TypedValue::new(crate::task::context_schema(), Value::Object(Map::new()))
@@ -294,6 +322,16 @@ pub struct Coordinator {
     media_names: Vec<String>,
     serials: Serials,
     topics: Topics,
+    /// The publication boundary. Everything this session publishes goes through it, and
+    /// every outcome it returns is a named one.
+    publisher: crate::publish::Publisher,
+    /// The composition as published. Built from what the live participants attested to,
+    /// never restated from the configuration that asked for them.
+    session_descriptor: Option<SessionDescriptor>,
+    /// The revision the next descriptor publication carries.
+    descriptor_revision: u64,
+    /// The read-only repair service. Held so it stops with the session.
+    query: Option<crate::publish::QueryService>,
     pacing: Option<Pacing>,
     /// Set by whoever asks for a normal pause, possibly while a transition is in flight.
     pause: std::sync::Arc<std::sync::atomic::AtomicBool>,
@@ -312,6 +350,8 @@ pub struct Coordinator {
     pub resolutions: u64,
     /// How the last resolution ended, so a test or a supervisor can tell which bound fired.
     pub last_resolution: Option<ResolutionEnd>,
+    /// How many attempts the last resolution spent. Counted, not inferred from the clock.
+    pub last_resolution_attempts: u32,
     /// The caller-side failure-detection budgets of `ipc-v1` section 6.
     pub deadlines: Deadlines,
     /// Per-method and critical-path latency samples. Local synthetic timings, never a
@@ -331,6 +371,17 @@ pub struct Coordinator {
     started: std::time::Instant,
     last_advance_request: Option<DomainRequestId>,
     last_commit_requests: Vec<TraceRequest>,
+    /// The previous committed boundary's broadcast references. Data only: no handle, no owner,
+    /// no retention, and nothing reads it but the publication fault injections.
+    previous_broadcast_views: Vec<ViewRef>,
+}
+
+/// The broadcast references of an observation, or none when there is no observation yet.
+fn observation_views_of(observation: &Option<WorldObservation>) -> Vec<ViewRef> {
+    observation
+        .as_ref()
+        .map(|o| o.broadcast_views.clone())
+        .unwrap_or_default()
 }
 
 impl Coordinator {
@@ -349,6 +400,7 @@ impl Coordinator {
         // Sorted agent-id order is the executor and control order, so it is fixed here once.
         agents.sort_by(|a, b| a.agent_id.cmp(&b.agent_id));
         let topics = Topics::for_session(&session_id);
+        let publisher = crate::publish::Publisher::new(bus.clone(), &session_id, &epoch, &topics);
         Coordinator {
             bus,
             session_id,
@@ -371,6 +423,10 @@ impl Coordinator {
                 media::audio_attachment(crate::environment::AUDIO_STREAM_ID),
             ],
             serials: Serials::default(),
+            publisher,
+            session_descriptor: None,
+            descriptor_revision: DESCRIPTOR_REVISION,
+            query: None,
             topics,
             pacing: None,
             pause: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
@@ -385,6 +441,7 @@ impl Coordinator {
             in_progress_replies: 0,
             resolutions: 0,
             last_resolution: None,
+            last_resolution_attempts: 0,
             deadlines: Deadlines::default(),
             metrics: Metrics::default(),
             blame: None,
@@ -395,6 +452,7 @@ impl Coordinator {
             started: std::time::Instant::now(),
             last_advance_request: None,
             last_commit_requests: Vec::new(),
+            previous_broadcast_views: Vec::new(),
         }
     }
 
@@ -410,6 +468,37 @@ impl Coordinator {
         &self.topics
     }
 
+    /// The composition this session published, once it has.
+    pub fn session_descriptor(&self) -> Option<&SessionDescriptor> {
+        self.session_descriptor.as_ref()
+    }
+
+    /// The revision the last published descriptor carried.
+    pub fn descriptor_revision(&self) -> u64 {
+        self.descriptor_revision
+    }
+
+    /// The sequence the next published snapshot will carry.
+    pub fn published_sequence(&self) -> u64 {
+        self.publisher.sequence()
+    }
+
+    /// What this session published and what became of it: accepted, refused by an observer,
+    /// or faulted, per topic.
+    pub fn ledger(&self) -> &crate::publish::Ledger {
+        self.publisher.ledger()
+    }
+
+    /// The events the bounded batch is still holding because an observer refused them.
+    pub fn pending_events(&self) -> usize {
+        self.publisher.outbox().len()
+    }
+
+    /// The read-only state the repair service answers from.
+    pub fn published_state(&self) -> crate::publish::SharedState {
+        self.publisher.state()
+    }
+
     pub fn epoch(&self) -> &Id {
         &self.epoch
     }
@@ -665,26 +754,14 @@ impl Coordinator {
         Ok(())
     }
 
+    /// Declares every framework topic under the delivery policy the publisher holds.
     async fn declare_topics(&mut self) -> Outcome<()> {
-        for (name, retained) in [
-            (self.topics.descriptor.clone(), flybus::Retained::Latest),
-            (self.topics.snapshots.clone(), flybus::Retained::Latest),
-            (self.topics.events.clone(), flybus::Retained::None),
-            // Checkpoint events are a stream of distinct facts, not a latest value: a
-            // "committed" that replaced a "queued" would erase the distinction the durable
-            // commit rules are built on.
-            (self.topics.checkpoints.clone(), flybus::Retained::None),
-        ] {
-            self.bus.declare_topic(&name, retained).await.map_err(|e| {
-                let error = DomainError::new(
-                    ErrorCode::BackendFailure,
-                    format!("declaring {name}: {}", e.message),
-                    MutationCertainty::None,
-                );
-                self.fail_now(error, "declare-topic")
-            })?;
+        // Every framework topic, including the checkpoint stream, is declared in one place
+        // under the delivery policy the publisher holds.
+        match self.publisher.declare().await {
+            Ok(()) => Ok(()),
+            Err(e) => Err(self.fail_now(e, "declare-topic")),
         }
-        Ok(())
     }
 
     async fn initialize_environment(&mut self) -> Outcome<()> {
@@ -864,6 +941,10 @@ impl Coordinator {
             .telemetry
             .validate()
             .map_err(|e| self.fail_now(DomainError::invalid(e), "agent-initialize"))?;
+        // What the fly says it built. The descriptor publishes this, so a composition that
+        // loaded another index is visible in the descriptor rather than only in a log line.
+        self.agents[index].graph = Some(result.graph.clone());
+        self.agents[index].telemetry = Some(result.telemetry.clone());
         self.agents[index].tick_duration = result.tick_duration;
         self.agents[index].warmup_ticks = result.warmup_ticks;
         self.agents[index].committed_step = 0;
@@ -892,23 +973,71 @@ impl Coordinator {
                 .push(request_id);
         }
         for (worker, ids) in by_worker.into_values() {
-            let params = AcknowledgeParams { request_ids: ids.clone() };
-            let reply = self
-                .call(&worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
-                .await?;
-            let result: AcknowledgeResult =
-                reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
-            if result.acknowledged.len() != ids.len() {
-                return Err(self.fail_now(
-                    DomainError::invalid("a worker did not acknowledge every lifecycle reply"),
-                    "acknowledge",
-                ));
+            if self.injections.duplicate_lifecycle_acknowledge {
+                // Release them first, out of sight, so the call this method then makes and
+                // checks is already the *second* one -- which is the shape the section 6
+                // resolution produces when an Acknowledge's first reply outruns the probe,
+                // and the shape the original defect fenced a healthy session on. Adding a
+                // second call after the checked one would not reproduce it: the first reply
+                // is always complete, so a length check on it would pass.
+                let first = self.acknowledge_replies(&worker, &ids).await?;
+                if first.len() != ids.len() {
+                    return Err(self.fail_now(
+                        DomainError::invalid("the first Acknowledge did not release everything"),
+                        "acknowledge",
+                    ));
+                }
             }
+            self.acknowledge_replies(&worker, &ids).await?;
         }
         self.audit.push("acknowledge.lifecycle".to_owned());
         Ok(())
     }
 
+    /// Releases a worker's retained lifecycle replies, and accepts a short answer.
+    ///
+    /// **`ipc-v1` section 5: "Already released/unknown IDs are ignored."** The reply lists what
+    /// *this* call released, which is not always everything it asked about, and the contract
+    /// type already holds that list to a subset of the request. So a second Acknowledge of the
+    /// same ids answers with an empty list by design, and an empty list is success.
+    ///
+    /// This matters beyond tidiness. The `ipc-v1` section 6 resolution turns any Acknowledge
+    /// whose reply is slower than the probe into a second Acknowledge of the same ids, so the
+    /// short answer is not an edge case -- it is what the contract produces on an ordinarily
+    /// slow worker. Requiring the whole list back made the contract's own idempotence a failed
+    /// epoch, which is what
+    /// `an_acknowledge_that_releases_nothing_is_not_a_failure` guards against.
+    ///
+    /// A short list is accepted; a list about something else is not. The worker reports what
+    /// *it* released, so fewer ids than asked for is success -- but it is still only entitled
+    /// to report about the ids it was asked about, and an id outside the request is a worker
+    /// talking about another caller's cache. That half is exact-demanded, and
+    /// `AcknowledgeResult::validate_against` is what says so.
+    ///
+    /// Returns the ids the worker actually released.
+    pub async fn acknowledge_replies(
+        &mut self,
+        worker: &WorkerRef,
+        request_ids: &[DomainRequestId],
+    ) -> Outcome<Vec<DomainRequestId>> {
+        let params = AcknowledgeParams { request_ids: request_ids.to_vec() };
+        let reply = self
+            .call(worker, "Worker.Acknowledge", None, object(params.to_json()), &[], &[])
+            .await?;
+        let result: AcknowledgeResult =
+            reply.parse().map_err(|e| self.fail_now(e, "acknowledge"))?;
+        if let Err(e) = result.validate_against(&params) {
+            return Err(self.fail_now(
+                DomainError::before(
+                    ErrorCode::IdentityMismatch,
+                    format!("a worker acknowledged an id this session never asked about: {e}"),
+                ),
+                "acknowledge",
+            ));
+        }
+        Ok(result.acknowledged)
+    }
+
     /// Queries one worker's status without waiting for its current mutation.
     pub async fn status(&mut self, worker: &WorkerRef) -> Outcome<StatusResult> {
         let reply = self
@@ -1200,16 +1329,22 @@ impl Coordinator {
         let attempts = self.deadlines.resolve_attempts;
         let started = Instant::now();
         self.resolutions += 1;
+        // Both, together: a resolution that ends before its first attempt would otherwise
+        // report the previous one's count.
         self.last_resolution = None;
+        self.last_resolution_attempts = 0;
         self.audit.push(format!("resolve:{}:{method}", worker.worker_id));
         // The budget is the working limit and the attempt count is a guard; whichever runs
         // out is recorded, so "it gave up" is never an unexplained number.
         let mut end = ResolutionEnd::AttemptsExhausted;
+        let mut spent = 0u32;
         for _ in 0..attempts {
             if started.elapsed() >= budget {
                 end = ResolutionEnd::BudgetExpired;
                 break;
             }
+            spent += 1;
+            self.last_resolution_attempts = spent;
             let outcome = call_owned(
                 self.bus.clone(),
                 worker.clone(),
@@ -1587,8 +1722,30 @@ impl Coordinator {
             self.agents[index].context_digest = context.digest();
             self.agents[index].context = context;
             self.agents[index].committed_step = k + 1;
+            // The telemetry of the transition that just ended, which is what this boundary's
+            // snapshot publishes. Without this the slot would keep whatever `Agent.Initialize`
+            // reported and every snapshot would label warm-up telemetry as boundary k.
+            let telemetry = commits
+                .iter()
+                .find(|(id, _)| *id == agent_id)
+                .map(|(_, result)| result.telemetry.clone());
+            match telemetry {
+                Some(telemetry) => self.agents[index].telemetry = Some(telemetry),
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::before(
+                            ErrorCode::IdentityMismatch,
+                            format!("agent {agent_id} committed without telemetry"),
+                        ),
+                        "commit",
+                    ));
+                }
+            }
         }
         // The previous boundary's handles are no longer needed; the new ones take over.
+        // The references -- which are data, not ownership -- are kept for one boundary, so a
+        // publication fault injection can name an older frame without retaining it.
+        self.previous_broadcast_views = observation_views_of(&self.observation);
         self.views = new_views;
         self.audio = new_audio;
         self.pending_views.clear();
@@ -2250,13 +2407,23 @@ impl Coordinator {
                 .expect("every agent prepared");
             let outcome = outcomes.get(&agent_id).cloned().unwrap_or_default();
             let next_context = next_contexts.get(&agent_id).cloned().expect("checked");
+            let mut task_stimulations = outcome.stimulations.clone();
+            if self.injections.undeclared_stimulus && self.injections.at_step == k {
+                // A kind outside the agent's published `supportedStimuli`. The declaration is
+                // only worth publishing if the worker enforces it.
+                task_stimulations.push(Stimulus {
+                    id: parse_id(&format!("stim-undeclared-{k}")).expect("a serial makes an Id"),
+                    kind_id: id("arena.undeclared"),
+                    duration_ms: 1.0,
+                });
+            }
             let params = CommitParams {
                 agent_id: agent_id.clone(),
                 prepared_request_id: prepared_request.clone(),
                 next_input: self.sensory_input(observation, k + 1),
                 next_decision_context: next_context,
                 rewards: outcome.rewards.clone(),
-                task_stimulations: outcome.stimulations.clone(),
+                task_stimulations,
             };
             let params = match params.to_json() {
                 Value::Object(m) => m,
@@ -2513,59 +2680,115 @@ impl Coordinator {
     // -----------------------------------------------------------------------------------
     // Publication
 
-    async fn publish(
-        &mut self,
-        topic: &str,
-        payload: Map<String, Value>,
-        attachments: Vec<(String, flybus::Artifact)>,
-    ) -> Outcome<()> {
-        let refs: Vec<(&str, &flybus::Artifact)> =
-            attachments.iter().map(|(n, a)| (n.as_str(), a)).collect();
-        match self.bus.publish(topic, payload, &refs).await {
-            Ok(_) => Ok(()),
-            Err(e) => {
-                // A disconnected or backpressured observer never stalls the world; only a
-                // real resource fault reaches here, and it fails the epoch honestly.
-                let error = DomainError::new(
-                    ErrorCode::BackendFailure,
-                    format!("publishing {topic}: {}", e.message),
-                    MutationCertainty::None,
-                );
-                Err(self.fail_now(error, "publish"))
+    /// Sends one publication and turns its outcome into the session's response to it.
+    ///
+    /// An observer's refusal is counted and the world carries on: "ordinary snapshot
+    /// publication is latest/bounded and never waits for a spectator to consume it"
+    /// (publishing-v1 section 3), and `bus-v1` section 6 allows a bounded subscriber to reject
+    /// a publication. A session resource fault is not an observer and fails the epoch.
+    fn settle(&mut self, outcome: PublicationOutcome, detail: &str) -> Outcome<PublicationOutcome> {
+        match outcome.fault() {
+            Some(error) => Err(self.fail_now(error, detail)),
+            None => {
+                if outcome.is_refused() {
+                    self.audit.push(format!("refused:{}", outcome.topic()));
+                }
+                Ok(outcome)
             }
         }
     }
 
+    /// The composition as the live participants attested to it.
+    ///
+    /// Every agent row comes from that agent's own `Agent.Initialize` reply, so a descriptor
+    /// can disagree with the configuration that asked for the composition. One restated from
+    /// the configuration never could, and `publishing-v1` section 3 needs the disagreement to
+    /// be visible: "geometry/spike mapping requires indexDigest, not merely the same number
+    /// of neurons".
+    fn build_descriptor(&self, revision: u64) -> DomainResult<SessionDescriptor> {
+        let environment = self.descriptor.clone().ok_or_else(|| {
+            DomainError::before(ErrorCode::InvalidPhase, "no environment descriptor")
+        })?;
+        let mut agents = Vec::new();
+        let mut assets = Vec::new();
+        for slot in &self.agents {
+            let graph = slot.graph.clone().ok_or_else(|| {
+                DomainError::before(
+                    ErrorCode::InvalidPhase,
+                    format!("agent {} has not attested to a graph", slot.agent_id),
+                )
+            })?;
+            agents.push(AgentDescriptor {
+                agent_id: slot.agent_id.clone(),
+                port_id: slot.port_id.clone(),
+                profile_digest: slot.profile.digest.clone(),
+                dataset_digest: graph.dataset_digest,
+                index_digest: graph.index_digest,
+                neuron_count: graph.neuron_count,
+                rate_roles: graph.rate_roles,
+                supported_stimuli: graph.supported_stimuli,
+            });
+            assets.push(slot.profile.clone());
+        }
+        let descriptor = SessionDescriptor {
+            session_id: self.session_id.clone(),
+            revision,
+            composition_digest: self.composition_digest(),
+            environment,
+            task_schema: self.task.schema(),
+            agents,
+            assets,
+        };
+        descriptor.validate().map_err(DomainError::invalid)?;
+        Ok(descriptor)
+    }
+
+    /// Publishes the composition and starts the read-only repair service beside it.
+    /// Publishes the composition, advancing the revision when the composition changed.
+    ///
+    /// A revision identifies a composition, so republishing an unchanged one keeps its number
+    /// and a changed one takes the next: a group restore establishes a fresh epoch, which is a
+    /// new `compositionDigest`, and a consumer that held the old revision has to be told rather
+    /// than handed the same number with different contents. The publisher refuses the second
+    /// case outright, so this is where the number moves.
     async fn publish_descriptor(&mut self) -> Outcome<()> {
-        let descriptor = self.descriptor.clone().expect("bootstrapped");
-        let agents: Vec<Value> = self
-            .agents
-            .iter()
-            .map(|slot| {
-                json!({
-                    "agentId": slot.agent_id.as_str(),
-                    "portId": slot.port_id.as_str(),
-                    "profileDigest": slot.profile.digest.as_str(),
-                    "tickDuration": slot.tick_duration.to_json(),
-                    "warmupTicks": slot.warmup_ticks.to_string(),
-                })
-            })
-            .collect();
-        let payload = json!({
-            "sessionId": self.session_id.as_str(),
-            "revision": "1",
-            "compositionDigest": self.composition_digest().as_str(),
-            "schedulerId": "lockstep-v1",
-            "environment": descriptor.to_json(),
-            "taskSchema": self.task.schema().to_json(),
-            "agents": agents,
-        });
-        let topic = self.topics.descriptor.clone();
-        self.publish(&topic, match payload {
-            Value::Object(m) => m,
-            _ => Map::new(),
-        }, Vec::new())
-        .await
+        let mut descriptor = match self.build_descriptor(self.descriptor_revision) {
+            Ok(descriptor) => descriptor,
+            Err(e) => return Err(self.fail_now(e, "descriptor")),
+        };
+        if let Some(published) = &self.session_descriptor {
+            let mut same = descriptor.clone();
+            same.revision = published.revision;
+            if same != *published {
+                self.descriptor_revision += 1;
+                descriptor.revision = self.descriptor_revision;
+                self.audit
+                    .push(format!("descriptor-revision:{}", self.descriptor_revision));
+            }
+        }
+        let outcome = match self.publisher.publish_descriptor(&descriptor).await {
+            Ok(outcome) => outcome,
+            Err(e) => return Err(self.fail_now(e, "descriptor")),
+        };
+        self.settle(outcome, "descriptor")?;
+        self.session_descriptor = Some(descriptor);
+        if self.query.is_none() {
+            let state = self.publisher.state();
+            let service =
+                crate::publish::QueryService::start(self.bus.clone(), &self.session_id, state)
+                    .await
+                    .map_err(|e| {
+                        let error = DomainError::new(
+                            ErrorCode::BackendFailure,
+                            format!("registering the session query service: {}", e.message),
+                            MutationCertainty::None,
+                        );
+                        self.fail_now(error, "query-service")
+                    })?;
+            self.query = Some(service);
+        }
+        self.audit.push("publish:descriptor".to_owned());
+        Ok(())
     }
 
     /// The composition identity: session, epoch, agents, ports and the contract revision.
@@ -2585,22 +2808,15 @@ impl Coordinator {
         digest_of_bytes(text.as_bytes())
     }
 
+    /// Offers this boundary's events to the bounded batch and publishes what it holds.
     async fn publish_events(&mut self, source_step: u64, events: &[TaskEvent]) -> Outcome<()> {
-        if events.is_empty() {
-            return Ok(());
+        match self.publisher.publish_events(source_step, events).await {
+            None => Ok(()),
+            Some(outcome) => {
+                self.settle(outcome, "events")?;
+                Ok(())
+            }
         }
-        let payload = json!({
-            "sessionId": self.session_id.as_str(),
-            "epoch": self.epoch.as_str(),
-            "sourceStep": source_step.to_string(),
-            "events": Value::Array(events.iter().map(DomainType::to_json).collect()),
-        });
-        let topic = self.topics.events.clone();
-        self.publish(&topic, match payload {
-            Value::Object(m) => m,
-            _ => Map::new(),
-        }, Vec::new())
-        .await
     }
 
     /// Publishes the committed boundary. Never an in-progress mix of new agent state and an
@@ -2621,55 +2837,180 @@ impl Coordinator {
                 "publish",
             ));
         }
+        let descriptor = match self.session_descriptor.clone() {
+            Some(descriptor) => descriptor,
+            None => {
+                return Err(self.fail_now(
+                    DomainError::before(ErrorCode::InvalidPhase, "no descriptor was published"),
+                    "publish",
+                ));
+            }
+        };
         let observation = self.observation.clone().expect("bootstrapped");
-        let agents: Vec<Value> = self
-            .agents
-            .iter()
-            .map(|slot| {
-                let control = controls
-                    .iter()
-                    .find(|c| c.port_id == slot.port_id)
-                    .map(|c| c.to_json());
-                json!({
-                    "agentId": slot.agent_id.as_str(),
-                    "selectedDecision": decisions
-                        .get(&slot.agent_id)
-                        .map(|d| d.to_json()),
-                    "appliedControls": control,
-                    "committedStep": slot.committed_step.to_string(),
-                })
-            })
-            .collect();
-        let payload = json!({
-            "descriptorRevision": "1",
-            "publisherIncarnation": self.bus.info().connection_id.clone(),
-            "scope": self.scope(boundary).to_json(),
-            "episodeId": self.episode_id.as_str(),
-            "sequence": self.stats.publications.to_string(),
-            "worldTime": observation.world_time.to_json(),
-            "agents": agents,
-            "progress": self.task.progress().to_json(),
-            "media": json!({
-                "views": Value::Array(observation.broadcast_views.iter().map(DomainType::to_json).collect()),
-                "audio": Value::Array(observation.audio.iter().map(DomainType::to_json).collect()),
-            }),
-            "eventIds": event_ids.iter().map(Id::as_str).collect::<Vec<_>>(),
-        });
-        // The same owned handles the agents were given, published once for presentation.
-        let mut attachments = self.view_attachments();
-        attachments.extend(self.audio.iter().map(|(n, a)| (n.clone(), a.clone())));
-        let topic = self.topics.snapshots.clone();
-        self.publish(
-            &topic,
-            match payload {
-                Value::Object(m) => m,
-                _ => Map::new(),
-            },
-            attachments,
-        )
-        .await?;
-        self.stats.publications += 1;
-        self.audit.push(format!("publish:{boundary}"));
+        let mut agents = Vec::new();
+        for slot in &self.agents {
+            if slot.committed_step != boundary {
+                // A snapshot names one boundary. An agent that is not at it would be future
+                // state beside this world, which is the thing this check exists to refuse.
+                return Err(self.fail_now(
+                    DomainError::before(
+                        ErrorCode::IdentityMismatch,
+                        format!(
+                            "agent {} is committed at {} and the snapshot is boundary {boundary}",
+                            slot.agent_id, slot.committed_step
+                        ),
+                    ),
+                    "publish",
+                ));
+            }
+            let telemetry = match slot.telemetry.clone() {
+                Some(telemetry) => telemetry,
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::before(
+                            ErrorCode::InvalidPhase,
+                            format!("agent {} reported no telemetry", slot.agent_id),
+                        ),
+                        "publish",
+                    ));
+                }
+            };
+            agents.push(SnapshotAgent {
+                agent_id: slot.agent_id.clone(),
+                telemetry,
+                // "Decisions/controls describe the transition ending at that boundary, null at
+                // initial boundary 0." These are the decisions of the transition that ended
+                // here, never the ones prepared for the transition about to start.
+                selected_decision: decisions.get(&slot.agent_id).cloned(),
+                applied_controls: controls.iter().find(|c| c.port_id == slot.port_id).cloned(),
+            });
+        }
+        let mut views = observation.broadcast_views.clone();
+        if self.injections.stale_published_view && self.injections.at_step + 1 == boundary {
+            // The injection of "new agent state with old media": the agents are at this
+            // boundary and the frame is the previous one's.
+            let stale = self.previous_broadcast_views.clone();
+            if stale.is_empty() {
+                return Err(self.fail_now(
+                    DomainError::invalid("no previous boundary to take a stale view from"),
+                    "publish",
+                ));
+            }
+            views = stale;
+            self.injection_log.push(InjectionOutcome {
+                what: "stale-published-view".to_owned(),
+                code: None,
+                identical: false,
+            });
+        }
+        let snapshot = CommittedSnapshot {
+            descriptor_revision: descriptor.revision,
+            publisher_incarnation: self.publisher.incarnation(),
+            scope: self.scope(boundary),
+            episode_id: self.episode_id.clone(),
+            sequence: self.publisher.sequence(),
+            world_time: observation.world_time,
+            agents,
+            progress: self.task.progress(),
+            views,
+            audio: observation.audio.clone(),
+            event_ids: event_ids.to_vec(),
+        };
+        // The same owned handles the agents were given, published once for presentation. A
+        // referenced frame with no handle, or a handle that is another boundary's object, is
+        // refused by `check_publication` before anything reaches a subscriber.
+        let mut attachments = Vec::new();
+        for view in &snapshot.views {
+            let name = media::view_attachment(&view.view_id);
+            match self.views.get(&name) {
+                Some(artifact) => attachments.push((name, artifact.clone())),
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::new(
+                            ErrorCode::BufferInvalid,
+                            format!("this boundary holds no handle for view {}", view.view_id),
+                            MutationCertainty::None,
+                        ),
+                        "publish",
+                    ));
+                }
+            }
+        }
+        for chunk in &snapshot.audio {
+            let name = media::audio_attachment(&chunk.stream_id);
+            match self.audio.get(&name) {
+                Some(artifact) => attachments.push((name, artifact.clone())),
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::new(
+                            ErrorCode::BufferInvalid,
+                            format!(
+                                "this boundary holds no handle for stream {}",
+                                chunk.stream_id
+                            ),
+                            MutationCertainty::None,
+                        ),
+                        "publish",
+                    ));
+                }
+            }
+        }
+        if self.injections.substituted_published_handle && self.injections.at_step + 1 == boundary {
+            // The same attachment name and the same bytes, a different object. Only the
+            // artifact identity sees it, which is why the check compares that and not names.
+            let (name, artifact) = match attachments.first() {
+                Some(first) => first.clone(),
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::invalid("no attachment to substitute"),
+                        "publish",
+                    ));
+                }
+            };
+            let bytes = match artifact.read_all().await {
+                Ok(bytes) => bytes,
+                Err(e) => {
+                    return Err(self.fail_now(
+                        DomainError::new(
+                            ErrorCode::BufferInvalid,
+                            e.message,
+                            MutationCertainty::None,
+                        ),
+                        "publish",
+                    ));
+                }
+            };
+            let copy = match media::seal_copy(
+                &self.bus,
+                artifact.reference().content_type.clone(),
+                &bytes,
+            )
+            .await
+            {
+                Ok(copy) => copy,
+                Err(e) => return Err(self.fail_now(e, "publish")),
+            };
+            attachments[0] = (name, copy);
+            self.injection_log.push(InjectionOutcome {
+                what: "substituted-published-handle".to_owned(),
+                code: None,
+                identical: false,
+            });
+        }
+        let positions = self.timelines.positions();
+        let outcome = match self
+            .publisher
+            .publish_snapshot(&descriptor, &snapshot, &attachments, &positions)
+            .await
+        {
+            Ok(outcome) => outcome,
+            Err(e) => return Err(self.fail_now(e, "publish")),
+        };
+        let outcome = self.settle(outcome, "publish")?;
+        if outcome.is_accepted() {
+            self.stats.publications += 1;
+            self.audit.push(format!("publish:{boundary}"));
+        }
         Ok(())
     }
 }
@@ -2756,15 +3097,33 @@ impl Coordinator {
         }
     }
 
-    fn agent_compatibility(&self, slot: &AgentSlot) -> Digest {
-        crate::agent::agent_compatibility_digest(
+    /// One agent's compatibility identity, from what that agent attested to at
+    /// `Agent.Initialize` rather than from anything this coordinator recomputed.
+    ///
+    /// The graph belongs here because `state-media-v1`'s recovery rules say old parser or
+    /// media state must not cross a recovery, and a graph identity is exactly that: without it
+    /// a replacement fly that built another index passes the group check and is then published
+    /// under its predecessor's `indexDigest`. An agent that has not attested yet has no
+    /// compatibility, which is a refusal rather than a guessed digest.
+    fn agent_compatibility(&self, slot: &AgentSlot) -> DomainResult<Digest> {
+        let graph = slot.graph.as_ref().ok_or_else(|| {
+            DomainError::before(
+                ErrorCode::InvalidPhase,
+                format!(
+                    "agent {} has not attested to a graph, so it has no compatibility identity",
+                    slot.agent_id
+                ),
+            )
+        })?;
+        Ok(crate::agent::agent_compatibility_digest(
             &slot.agent_id,
             &slot.profile.digest,
-            &crate::agent::dataset_digest(),
+            &graph.dataset_digest,
             crate::agent::MODEL_VERSION,
             crate::agent::PLASTICITY_VERSION,
             slot.seed,
-        )
+            &graph.index_digest,
+        ))
     }
 
     /// The coordinator's own session record: what it must hold again to resume this boundary.
@@ -2953,7 +3312,10 @@ impl Coordinator {
         for index in 0..self.agents.len() {
             let slot_worker = self.agents[index].worker.clone();
             let agent_id = self.agents[index].agent_id.clone();
-            let expected = self.agent_compatibility(&self.agents[index]);
+            let expected = match self.agent_compatibility(&self.agents[index]) {
+                Ok(expected) => expected,
+                Err(e) => return Err(self.fail_now(e, "capture")),
+            };
             let reply = self
                 .call(
                     &slot_worker,
@@ -2971,10 +3333,25 @@ impl Coordinator {
             payloads.push(payload);
             acknowledge.push((slot_worker, reply.request_id.clone()));
             let slot = &self.agents[index];
+            // The graph identities come from what this agent attested to, not from a value
+            // the coordinator recomputed; that is the whole point of recording them.
+            let graph = match slot.graph.clone() {
+                Some(graph) => graph,
+                None => {
+                    return Err(self.fail_now(
+                        DomainError::before(
+                            ErrorCode::InvalidPhase,
+                            format!("agent {agent_id} has not attested to a graph"),
+                        ),
+                        "capture",
+                    ));
+                }
+            };
             agent_rows.push(crate::state::AgentEntry {
                 agent_id: agent_id.clone(),
                 profile_digest: slot.profile.digest.clone(),
-                dataset_digest: crate::agent::dataset_digest(),
+                dataset_digest: graph.dataset_digest,
+                index_digest: graph.index_digest,
                 model_version: crate::agent::MODEL_VERSION.to_owned(),
                 plasticity_version: crate::agent::PLASTICITY_VERSION.to_owned(),
                 seed: slot.seed,
@@ -3262,8 +3639,9 @@ impl Coordinator {
             "boundary": boundary.to_string(),
             "detail": detail.map_or(Value::Null, |d| Value::String(d.to_owned())),
         });
-        let topic = self.topics.checkpoints.clone();
-        self.publish(&topic, object(payload), Vec::new()).await
+        let outcome = self.publisher.publish_checkpoint(object(payload)).await;
+        self.settle(outcome, "checkpoint-event")?;
+        Ok(())
     }
 
     // ---------------------------------------------------------------------------------------
@@ -3552,6 +3930,7 @@ impl Coordinator {
                     &row.model_version,
                     &row.plasticity_version,
                     row.seed,
+                    &row.index_digest,
                 ),
             ));
         }

services/flysim/crates/fly-session/src/harness.rs

@@ -47,6 +47,10 @@ pub struct AgentSpec {
     /// The threads this agent asks the launcher for. `Agent.Initialize` carries exactly what
     /// the launcher allocated, which `workers-v1` requires it to lie within.
     pub worker_threads: usize,
+    /// Which graph this fly builds. A replacement worker started on another variant has the
+    /// same neuron count and another `indexDigest`, which is the composition change a
+    /// descriptor revision exists to make visible.
+    pub graph_variant: u64,
 }
 
 impl AgentSpec {
@@ -58,6 +62,7 @@ impl AgentSpec {
             seed,
             faults: AgentFaults::default(),
             worker_threads: 1,
+            graph_variant: 0,
         }
     }
 }
@@ -164,6 +169,14 @@ fn agent_client(agent_id: &Id) -> String {
     format!("worker-{agent_id}")
 }
 
+/// What a presentation consumer may do: subscribe, and ask the read-only repair service.
+fn consumer_grants() -> Grants {
+    grants(|g| {
+        g.subscribe = vec![Pattern::prefix("session."), Pattern::prefix("app.")];
+        g.call = vec![Pattern::prefix("session.")];
+    })
+}
+
 fn grants(f: impl FnOnce(&mut Grants)) -> Grants {
     let mut g = Grants::default();
     f(&mut g);
@@ -193,6 +206,8 @@ pub struct SessionHarness {
     /// The supervisor. It owns every participant's lifetime and thread allocation.
     pub launcher: Launcher,
     observers: Mutex<Vec<Client>>,
+    /// Which configured observer identity the next consumer takes.
+    next_observer: std::sync::atomic::AtomicUsize,
     /// Which generation of each participant is running: 1 is the one the composition started.
     generations: BTreeMap<Id, u32>,
     /// Where the durable checkpoint store lives, for a test that reads the files themselves.
@@ -221,6 +236,10 @@ impl SessionHarness {
                     g.call = vec![Pattern::prefix("agent."), Pattern::prefix("env.")];
                     g.publish = vec![Pattern::prefix("session.")];
                     g.manage_topics = vec![Pattern::prefix("session.")];
+                    // The read-only repair service of publishing-v1 section 2. It is the
+                    // session's own address and answers two queries; naming it is not
+                    // authority over anything, and no method on it mutates.
+                    g.register = vec![Pattern::prefix("session.")];
                 }),
             )
             .client(
@@ -232,7 +251,36 @@ impl SessionHarness {
             // The writer publishes the checkpoint events and never calls a participant.
             .client(WRITER_CLIENT, grants(|g| g.publish = vec![Pattern::prefix("session.")]))
             .client(ENV_CLIENT, grants(|g| g.register = vec![Pattern::exact(ENV_SERVICE)]))
-            .client("observer", grants(|g| g.subscribe = vec![Pattern::prefix("session.")]));
+            // A presentation consumer subscribes and may call the repair service. It can
+            // publish nothing, register nothing and reach no worker: "viewers/browser clients
+            // never obtain worker control" (publishing-v1 section 7). A bus client id is one
+            // connection, so a composition with several consumers configures several of them;
+            // they are the same grants, because a second viewer is not a more privileged one.
+            .client("observer", consumer_grants())
+            .client("observer-2", consumer_grants())
+            .client("observer-3", consumer_grants())
+            .client("observer-4", consumer_grants())
+            // The application's own publisher. Its addresses are its own, and it has no
+            // reach into the session's.
+            .client(
+                "application",
+                grants(|g| {
+                    g.publish = vec![Pattern::prefix("app.")];
+                    g.manage_topics = vec![Pattern::prefix("app.")];
+                    g.subscribe = vec![Pattern::prefix("session.")];
+                }),
+            )
+            // The publication boundary, when a composition places it on a client of its own
+            // rather than on the coordinator's.
+            .client(
+                "publisher",
+                grants(|g| {
+                    g.publish = vec![Pattern::prefix("session.")];
+                    g.manage_topics = vec![Pattern::prefix("session.")];
+                }),
+            );
+        // A replacement environment connects under its own client id, one per generation;
+        // this subsumes the single `-r2` identity the publication slice had configured.
         for generation in 2..=MAX_GENERATIONS {
             policy = policy.client(
                 &format!("{ENV_CLIENT}-r{generation}"),
@@ -307,6 +355,7 @@ impl SessionHarness {
                     tick_duration,
                     warmup_ticks: config.warmup_ticks,
                     worker_threads: spec.worker_threads,
+                    graph_variant: spec.graph_variant,
                     sensors: sensors[&spec.agent_id].clone(),
                     faults: spec.faults.clone(),
                     client_id: agent_client(&spec.agent_id),
@@ -374,6 +423,7 @@ impl SessionHarness {
             sensors,
             launcher,
             observers: Mutex::new(Vec::new()),
+            next_observer: std::sync::atomic::AtomicUsize::new(0),
             generations: BTreeMap::new(),
             checkpoint_root,
         })
@@ -410,17 +460,65 @@ impl SessionHarness {
     }
 
     /// An extra subscriber, for a test that watches the published boundaries.
+    ///
+    /// Each call takes the next configured observer identity: one bus client id is one
+    /// connection, so two consumers are two configured participants and not one identity
+    /// used twice.
     pub async fn observer(&self) -> Result<Client, flybus::BusError> {
-        let client = self.launcher.connect("observer").await?;
+        let index = self
+            .next_observer
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        let id = match index {
+            0 => "observer".to_owned(),
+            n => format!("observer-{}", n + 1),
+        };
+        let client = self.launcher.connect(&id).await?;
         self.observers.lock().expect("not poisoned").push(client.clone());
         Ok(client)
     }
 
+    /// A client for the application that owns its own state and cues.
+    pub async fn application(&self) -> Result<Client, flybus::BusError> {
+        let client = self.launcher.connect("application").await?;
+        self.observers.lock().expect("not poisoned").push(client.clone());
+        Ok(client)
+    }
+
+    /// A client for a publication boundary of its own.
+    pub async fn publisher(&self) -> Result<Client, flybus::BusError> {
+        let client = self.launcher.connect("publisher").await?;
+        self.observers.lock().expect("not poisoned").push(client.clone());
+        Ok(client)
+    }
+
+    /// A fake multi-agent presentation consumer attached to this session's topics.
+    pub async fn consumer(&self) -> Result<crate::publish::PresentationConsumer, flybus::BusError> {
+        let client = self.observer().await?;
+        crate::publish::PresentationConsumer::attach(
+            client,
+            &self.config.session_id,
+            self.coordinator.topics(),
+        )
+        .await
+    }
+
     /// Replaces one agent's worker with a fresh incarnation, as a restore would.
     ///
     /// The coordinator still pins the old registration, so its next call to that agent fails
     /// rather than silently reaching another brain.
     pub async fn restart_agent(&mut self, agent_id: &Id) -> Result<Restarted, flybus::BusError> {
+        self.restart_agent_on_graph(agent_id, None).await
+    }
+
+    /// Replaces one agent's worker, optionally with a fly that built another graph.
+    ///
+    /// `Some(variant)` is the composition change a descriptor revision exists for: the same
+    /// neuron count, another `indexDigest`.
+    pub async fn restart_agent_on_graph(
+        &mut self,
+        agent_id: &Id,
+        graph_variant: Option<u64>,
+    ) -> Result<Restarted, flybus::BusError> {
         let spec = self
             .config
             .agents
@@ -442,6 +540,7 @@ impl SessionHarness {
                 tick_duration,
                 warmup_ticks: self.config.warmup_ticks,
                 worker_threads: spec.worker_threads,
+                graph_variant: graph_variant.unwrap_or(spec.graph_variant),
                 // The same log: a replacement worker in this process keeps writing where its
                 // predecessor wrote, so a restore's sensory input is visible beside it.
                 sensors: self.sensors.get(agent_id).cloned().unwrap_or_default(),
@@ -550,6 +649,22 @@ impl SessionHarness {
         }
     }
 
+    /// Changes which graph one agent builds, so the replacement the next restart launches is
+    /// a fly with the same neuron count and another index.
+    ///
+    /// The same relaunch rule as a fault: the worker running now keeps what it was started
+    /// with, and the change reaches the composition through the next replacement.
+    pub fn set_agent_graph(&mut self, agent_id: &Id, graph_variant: u64) {
+        if let Some(spec) = self
+            .config
+            .agents
+            .iter_mut()
+            .find(|spec| spec.agent_id == *agent_id)
+        {
+            spec.graph_variant = graph_variant;
+        }
+    }
+
     /// Changes the environment's injected faults, with the same relaunch rule.
     pub fn set_environment_faults(&mut self, faults: EnvironmentFaults) {
         self.config.environment_faults = faults;

services/flysim/crates/fly-session/src/launcher.rs

@@ -231,6 +231,9 @@ pub struct AgentLaunch {
     /// gets a fresh log in that process, which the supervisor cannot read.
     pub sensors: crate::media::SensorLog,
     pub faults: AgentFaults,
+    /// Which graph this fly builds. Crosses a process boundary as argv, like every other
+    /// thing a worker is started with.
+    pub graph_variant: u64,
     /// The configured client id. A replacement worker connects under its own.
     pub client_id: String,
     pub service: String,
@@ -1231,6 +1234,7 @@ pub(crate) mod flags {
     pub const PREPARE_DELAY_MS: &str = "prepare-delay-ms";
     pub const COMMIT_DELAY_MS: &str = "commit-delay-ms";
     pub const FAIL_COMMIT_AT_STEP: &str = "fail-commit-at-step";
+    pub const GRAPH_VARIANT: &str = "graph-variant";
     pub const FAIL_STAGE_RESTORE: &str = "fail-stage-restore";
     pub const FAIL_ACTIVATE_RESTORE: &str = "fail-activate-restore";
 
@@ -1273,6 +1277,7 @@ pub(crate) mod flags {
         PREPARE_DELAY_MS,
         COMMIT_DELAY_MS,
         FAIL_COMMIT_AT_STEP,
+        GRAPH_VARIANT,
         FAIL_STAGE_RESTORE,
         FAIL_ACTIVATE_RESTORE,
     ];
@@ -1333,6 +1338,7 @@ impl Started {
                     arg(flags::WARMUP_TICKS, spec.warmup_ticks),
                     arg(flags::PREPARE_DELAY_MS, spec.faults.prepare_delay_ms),
                     arg(flags::COMMIT_DELAY_MS, spec.faults.commit_delay_ms),
+                    arg(flags::GRAPH_VARIANT, spec.graph_variant),
                     arg(flags::FAIL_STAGE_RESTORE, u64::from(spec.faults.fail_stage_restore)),
                     arg(
                         flags::FAIL_ACTIVATE_RESTORE,
@@ -1393,6 +1399,7 @@ pub(crate) fn agent_config(spec: &AgentLaunch, worker_threads: usize) -> AgentCo
         tick_duration: spec.tick_duration,
         warmup_ticks: spec.warmup_ticks,
         worker_threads,
+        graph_variant: spec.graph_variant,
         sensors: spec.sensors.clone(),
         faults: spec.faults.clone(),
     }
@@ -1504,6 +1511,7 @@ mod flag_tests {
             tick_duration: RationalNs::new(1, 1_000).expect("a tick"),
             warmup_ticks: 10,
             worker_threads: 1,
+            graph_variant: 3,
             sensors: crate::media::SensorLog::new(),
             faults: AgentFaults {
                 fail_commit_at_step: Some(2),
@@ -1511,6 +1519,9 @@ mod flag_tests {
                 commit_delay_ms: 2,
                 fail_stage_restore: true,
                 fail_activate_restore: true,
+                // Argv carries the injected faults a worker process can have; this one is
+                // in-process only, because the check it drives is the caller's.
+                acknowledge_extra_id: None,
             },
             client_id: "worker-fly-a".to_owned(),
             service: "agent.fly-a".to_owned(),

services/flysim/crates/fly-session/src/lib.rs

@@ -33,6 +33,7 @@ pub mod measure;
 pub mod media;
 pub mod metrics;
 pub mod phase;
+pub mod publish;
 pub mod rpc;
 pub mod state;
 pub mod task;

services/flysim/crates/fly-session/src/media.rs

@@ -442,6 +442,18 @@ impl AudioSource {
     }
 }
 
+/// Allocates, writes and seals one immutable artifact of an arbitrary content type.
+///
+/// Used where a test needs a second object with the same bytes, so that "the handle is not
+/// the artifact the payload names" can be produced without corrupting the bytes.
+pub async fn seal_copy(
+    client: &flybus::Client,
+    content_type: String,
+    bytes: &[u8],
+) -> DomainResult<flybus::Artifact> {
+    seal(client, &content_type, bytes).await
+}
+
 /// Allocates, writes and seals one immutable artifact.
 async fn seal(
     client: &flybus::Client,

services/flysim/crates/fly-session/src/state.rs

@@ -599,6 +599,10 @@ pub struct AgentEntry {
     pub agent_id: Id,
     pub profile_digest: Digest,
     pub dataset_digest: Digest,
+    /// The index the agent attested to at `Agent.Initialize`. It is part of the agent's
+    /// compatibility identity, so a replacement that built another graph cannot install this
+    /// payload -- the graph identity does not cross the recovery.
+    pub index_digest: Digest,
     pub model_version: String,
     pub plasticity_version: String,
     pub seed: i32,
@@ -613,6 +617,7 @@ impl AgentEntry {
             "agentId": self.agent_id.as_str(),
             "profileDigest": self.profile_digest.as_str(),
             "datasetDigest": self.dataset_digest.as_str(),
+            "indexDigest": self.index_digest.as_str(),
             "modelVersion": self.model_version.as_str(),
             "plasticityVersion": self.plasticity_version.as_str(),
             "seed": self.seed,
@@ -646,6 +651,7 @@ impl AgentEntry {
             agent_id: parse_id(&text("agentId")?)?,
             profile_digest: text("profileDigest")?,
             dataset_digest: text("datasetDigest")?,
+            index_digest: text("indexDigest")?,
             model_version: text("modelVersion")?,
             plasticity_version: text("plasticityVersion")?,
             seed,

services/flysim/crates/fly-session/src/types.rs

@@ -32,9 +32,12 @@ pub use fly_session_types::schema::contract_digest;
 pub use fly_session_types::trace::{
     TraceAgent, TraceBehaviour, TraceObservation, TraceOperational, TraceRequest, TransitionTrace,
 };
+pub use fly_session_types::publishing::{
+    AgentDescriptor, CommittedSnapshot, SessionDescriptor, SnapshotAgent,
+};
 pub use fly_session_types::workers::{
     AcknowledgeParams, AcknowledgeResult, AdvanceParams, AgentCommitResult, AgentInitializeParams,
-    AgentInitializeResult, AgentTelemetry, AssetRef, AxisRange, AxisSchema, AxisValue, ButtonState,
+    AgentGraph, AgentInitializeResult, AgentTelemetry, AssetRef, AxisRange, AxisSchema, AxisValue, ButtonState,
     CommitParams, ControllerSchema, Determinism, EnvironmentDescriptor,
     EnvironmentInitializeParams, EnvironmentInitializeResult, EpisodeRequest, HelloParams,
     HelloResult, LearningTelemetry, MAX_ACKNOWLEDGE, MAX_AGENTS, MAX_PORTS, MAX_RATE_ROLES,

services/flysim/crates/fly-session/src/worker.rs

@@ -194,6 +194,13 @@ pub trait WorkerEndpoint: Send + 'static {
     /// allocation" can read the allocation instead of being told it out of band.
     fn worker_threads(&self) -> u64;
 
+    /// An id this worker will add to every `Worker.Acknowledge` reply, for a test that needs a
+    /// worker reporting about something it was never asked about. `None` for a worker that
+    /// behaves.
+    fn acknowledge_extra_id(&self) -> Option<Id> {
+        None
+    }
+
     /// The domain methods this endpoint implements, beyond the common `Worker.*` set.
     /// Anything else returns UNSUPPORTED without entering the endpoint.
     fn methods(&self) -> Vec<&'static str>;
@@ -281,7 +288,8 @@ async fn run<E: WorkerEndpoint>(
 ) {
     // Identity and capabilities are fixed for the endpoint's lifetime, so the shell reads them
     // once and never takes the endpoint mutex to answer Hello or Status.
-    let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads) = {
+    #[allow(clippy::type_complexity)]
+    let (worker_id, incarnation_id, session_id, role, capabilities, status, methods, threads, extra_ack) = {
         let e = endpoint.lock().await;
         (
             e.worker_id(),
@@ -292,6 +300,7 @@ async fn run<E: WorkerEndpoint>(
             e.status_cell(),
             e.methods(),
             e.worker_threads(),
+            e.acknowledge_extra_id(),
         )
     };
     let mut running: Vec<tokio::task::JoinHandle<()>> = Vec::new();
@@ -345,7 +354,7 @@ async fn run<E: WorkerEndpoint>(
                 continue;
             }
             "Worker.Acknowledge" => {
-                let outcome = match acknowledge(&request, &cache).await {
+                let outcome = match acknowledge(&request, &cache, extra_ack.as_ref()).await {
                     Ok(result) => success(&request, &worker_id, &incarnation_id, result),
                     Err(e) => {
                         failure(&request.request_id, &worker_id, &incarnation_id, request.scope.clone(), e)
@@ -717,16 +726,24 @@ fn hello(
 async fn acknowledge(
     request: &SessionRpcRequest,
     cache: &Arc<tokio::sync::Mutex<ResultCache>>,
+    extra: Option<&Id>,
 ) -> DomainResult<Map<String, Value>> {
     let params: AcknowledgeParams = AcknowledgeParams::from_json(&request.params)
         .map_err(|e| DomainError::invalid(format!("Worker.Acknowledge: {e}")))?;
     if params.request_ids.is_empty() || params.request_ids.len() > MAX_ACKNOWLEDGE {
         return Err(DomainError::invalid("Worker.Acknowledge takes 1..=16 request ids"));
     }
-    let acknowledged = {
+    let mut acknowledged = {
         let mut c = cache.lock().await;
         c.acknowledge(&params.request_ids)
     };
+    // A deliberately misbehaving worker, for the caller-side subset check to refuse.
+    if let Some(extra) = extra
+        && let Ok(id) = DomainRequestId::parse(extra)
+        && !params.request_ids.contains(&id)
+    {
+        acknowledged.push(id);
+    }
     let result = AcknowledgeResult { acknowledged };
     Ok(object(result.to_json()))
 }

services/flysim/crates/fly-session/tests/processes.rs

@@ -16,13 +16,15 @@ use common::{at, count, fly_a, fly_b, mode_fixture, within};
 use fly_session::agent::AgentFaults;
 use fly_session::coordinator::{DispatchOrder, Injections};
 use fly_session::environment::EnvironmentFaults;
-use fly_session::harness::{ExecutionMode, HarnessConfig, Via};
+use fly_session::harness::{AgentSpec, ExecutionMode, HarnessConfig, Via};
 use fly_session::launcher::{ReapOutcome, ThreadBudget};
 use fly_session::ResolutionEnd;
 use fly_session::phase::Phase;
 use fly_session::types::*;
 
 all_modes!(
+    an_acknowledge_that_releases_nothing_is_not_a_failure,
+    bootstrap_survives_the_second_acknowledge_its_resolution_makes,
     a_slow_participant_is_resolved_rather_than_failed,
     a_resolution_says_which_of_its_two_bounds_ended_it,
     a_delayed_one_agent_result_holds_the_world,
@@ -84,6 +86,111 @@ async fn sequential_reversed_and_parallel_completion_agree() {
     }
 }
 
+// -------------------------------------------------------------------------------------------
+// ipc-v1 section 5: an Acknowledge that releases nothing is success
+
+/// `ipc-v1` section 5: "Already released/unknown IDs are ignored."
+///
+/// A second `Worker.Acknowledge` of ids the worker has already released answers with an empty
+/// list. That is the contract working, not a worker misbehaving, and the coordinator must
+/// accept it and carry on. The session's own bootstrap releases every lifecycle reply, so
+/// asking again for the same ids is exactly that case -- driven directly here rather than by
+/// making something slow, because it is a rule about the reply and not about timing.
+///
+/// The rule has teeth because of section 6: any Acknowledge whose reply outruns the probe is
+/// resolved, and the resolution *is* a second Acknowledge of the same ids. A coordinator that
+/// demands the whole list back therefore fences a healthy session the first time a worker is
+/// slow to answer. It did, on this branch's parent; this test fails if that check returns.
+async fn an_acknowledge_that_releases_nothing_is_not_a_failure(mode: ExecutionMode) {
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    // Bootstrap acknowledges every lifecycle reply, so afterwards the worker holds none.
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
+    let worker = f.harness.coordinator.agent_ref(&fly_a()).cloned().unwrap();
+
+    // The ids bootstrap already released. The worker ignores them and releases nothing.
+    let already: Vec<DomainRequestId> =
+        (1..=3).map(DomainRequestId::from_serial).collect();
+    let released = within(
+        "acknowledge",
+        f.harness.coordinator.acknowledge_replies(&worker, &already),
+    )
+    .await
+    .expect("a second Acknowledge of released ids is success, not a failed epoch");
+    assert!(
+        released.is_empty(),
+        "already released ids are ignored, so this call released nothing: {released:?}"
+    );
+
+    // The session is untouched by it: not fenced, still at its boundary, and still plays.
+    assert!(!f.harness.coordinator.is_fenced(), "an empty acknowledgment is not a fault");
+    assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0));
+    let report = within("step", f.harness.coordinator.step())
+        .await
+        .expect("the session continues after an Acknowledge that released nothing");
+    assert_eq!(report.boundary, 1);
+    assert_eq!(f.harness.coordinator.stats().advances, 1);
+    f.shutdown().await;
+}
+
+/// The same rule, on the path `bootstrap` actually uses.
+///
+/// The test above calls `acknowledge_replies` directly, which guards the check where it lives
+/// now but not where it lived before: a length check reintroduced into `acknowledge_lifecycle`
+/// after that call would leave it green. This one drives bootstrap itself, with the
+/// `duplicate_lifecycle_acknowledge` injection doing exactly what the section 6 resolution
+/// does -- the same ids again, to a worker that has already released them -- so the second,
+/// empty answer has to be accepted by every check on bootstrap's path.
+async fn bootstrap_survives_the_second_acknowledge_its_resolution_makes(mode: ExecutionMode) {
+    let mut f = mode_fixture(mode, two_agents(mode)).await;
+    f.harness.coordinator.injections = Injections {
+        duplicate_lifecycle_acknowledge: true,
+        ..Injections::default()
+    };
+    within("bootstrap", f.harness.coordinator.bootstrap())
+        .await
+        .expect("bootstrap accepts the second, empty acknowledgment of its own lifecycle ids");
+    assert!(!f.harness.coordinator.is_fenced());
+    assert_eq!(f.harness.coordinator.phase(), Phase::Ready(0));
+    let report = within("step", f.harness.coordinator.step()).await.expect("and still plays");
+    assert_eq!(report.boundary, 1);
+    f.shutdown().await;
+}
+
+/// The other half of the rule: a short list is accepted, an id outside the request is not.
+///
+/// A worker reports what *it* released, so fewer ids than asked for is success -- but it is
+/// only entitled to report about the ids it was asked about. An id from outside the request is
+/// a worker talking about another caller's cache, and
+/// `AcknowledgeResult::validate_against` is what refuses it. Without this, dropping the length
+/// check left nothing checking the reply against the request at all.
+///
+/// Not generated per mode, deliberately. The check is the *caller's*, so the mode of the
+/// worker that misbehaves is irrelevant to it, and the alternative -- carrying the
+/// misbehaviour to a separate process over argv -- would put a flag in the shipped binary
+/// whose only purpose is to make a worker lie about its acknowledgments.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn an_acknowledged_id_outside_the_request_is_refused() {
+    let mode = ExecutionMode::InProcess;
+    let mut config = two_agents(mode);
+    // This worker adds an id nobody asked about to every acknowledgment.
+    config.agents[0].faults = AgentFaults {
+        acknowledge_extra_id: Some(id("req-9999")),
+        ..AgentFaults::default()
+    };
+    let mut f = mode_fixture(mode, config).await;
+    let failure = within("bootstrap", f.harness.coordinator.bootstrap())
+        .await
+        .expect_err("a worker may not acknowledge an id this session never asked about");
+    assert_eq!(failure.error.code, ErrorCode::IdentityMismatch);
+    assert!(
+        failure.error.message.contains("never asked about"),
+        "the refusal says what was wrong: {failure}"
+    );
+    assert_eq!(failure.detail, "acknowledge");
+    assert_eq!(failure.error.mutation, MutationCertainty::None, "refused before any mutation");
+    f.shutdown().await;
+}
+
 // -------------------------------------------------------------------------------------------
 // ipc-v1 section 6: an uncertain call is resolved, not failed
 
@@ -114,17 +221,21 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode)
     config.environment_faults =
         EnvironmentFaults { advance_delay_ms: 500, ..EnvironmentFaults::default() };
     let mut f = mode_fixture(mode, config).await;
+    // Bootstrap first, at ordinary deadlines: its lifecycle calls are not what this test is
+    // about, and squeezing them through the probe below only tests the machine's luck.
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     // A probe well inside both delays, and a resolution budget well outside them: the point is
-    // a call that expires and an operation that is nevertheless fine.
+    // a call that expires and an operation that is nevertheless fine. The guard is out of
+    // reach so the budget is the only bound in play, and the budget is far above what the
+    // delays need, so neither ends this resolution -- the answer does.
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(120),
-        resolve: Duration::from_secs(20),
-        resolve_attempts: 4096,
+        resolve: Duration::from_secs(15),
+        resolve_attempts: u32::MAX,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
         durable: Duration::from_secs(60),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let reports = within("run", f.harness.coordinator.run(2))
         .await
         .expect("a slow participant is resolved, not failed");
@@ -176,28 +287,46 @@ async fn a_slow_participant_is_resolved_rather_than_failed(mode: ExecutionMode)
     f.shutdown().await;
 }
 
+/// One agent, one port, and a participant that will not answer this side of the test's own
+/// timeout. The composition for the bound tests: one participant means one possible name in
+/// the failure, so which agent is blamed is not a race.
+fn one_silent_agent(mode: ExecutionMode) -> HarnessConfig {
+    HarnessConfig {
+        agents: vec![AgentSpec {
+            // Ten minutes. The suite's own `within` gives up at twenty seconds, so if the step
+            // returns at all, a bound ended it and not the participant. That is a claim about
+            // the code rather than about how fast this machine happens to be.
+            faults: AgentFaults { prepare_delay_ms: 600_000, ..AgentFaults::default() },
+            ..AgentSpec::new("fly-a", "p1", 7)
+        }],
+        mode,
+        ..HarnessConfig::default()
+    }
+}
+
 /// The resolution has two bounds, and which one ended it is never left to be guessed.
 ///
-/// `resolve` is the working limit at the default values -- the attempt guard is over sixteen
-/// seconds of pauses against an eight-second budget -- so an unresponsive participant runs the
-/// budget out. Setting the guard low instead ends the same resolution the other way, and the
-/// failure says so both in `last_resolution` and in its own message.
+/// Both halves are arranged so the bound under test is the only one that *can* fire: the
+/// other is set orders of magnitude out of reach, so no amount of scheduling delay flips them.
+/// The claim is the contract's -- a resolution ends by budget or by guard, records which, and
+/// names it in the failure -- and nothing here is timed.
+///
+/// The deadlines are installed after `bootstrap`, deliberately. Bootstrap makes lifecycle
+/// calls of its own, and squeezing them through a fifty-millisecond probe tests the harness's
+/// luck rather than the resolution.
 async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode) {
-    // The budget is what ends it at ordinary settings: a generous attempt guard, a short
-    // budget, and a participant far slower than either.
-    let mut config = two_agents(mode);
-    config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
-    let mut f = mode_fixture(mode, config).await;
+    // Half one: the budget fires, because the guard cannot. `u32::MAX` attempts at the two
+    // millisecond pause is over ninety days; the budget is a fifth of a second.
+    let mut f = mode_fixture(mode, one_silent_agent(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(50),
-        resolve: Duration::from_millis(300),
-        resolve_attempts: 8192,
+        resolve: Duration::from_millis(200),
+        resolve_attempts: u32::MAX,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
         durable: Duration::from_secs(60),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
-    let started = Instant::now();
     let failure = within("step", f.harness.coordinator.step())
         .await
         .expect_err("a participant that never answers exhausts the resolution");
@@ -206,38 +335,42 @@ async fn a_resolution_says_which_of_its_two_bounds_ended_it(mode: ExecutionMode)
         failure.error.message.contains("resolution budget"),
         "the message names the bound that fired: {failure}"
     );
-    assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
+    // The budget ended it with attempts still in hand, which is what makes it the budget. A
+    // 200 ms budget at a 50 ms probe cannot spend more than a handful, and `u32::MAX` was
+    // never in reach; asserting against the guard's own size would be vacuous.
+    let spent = f.harness.coordinator.last_resolution_attempts;
+    assert!(spent >= 1, "the resolution made at least one attempt");
+    assert!(spent < 100, "and nowhere near its guard: {spent}");
+    assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
     assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
-    assert!(
-        started.elapsed() < Duration::from_secs(20),
-        "the budget, not the 30-second participant, is what ended it"
-    );
     assert!(f.harness.coordinator.is_fenced());
     f.shutdown().await;
 
-    // The guard is what ends it when it is set below the budget: three attempts against a
-    // budget the participant could never reach anyway.
-    let mut config = two_agents(mode);
-    config.agents[1].faults = AgentFaults { prepare_delay_ms: 30_000, ..AgentFaults::default() };
-    let mut f = mode_fixture(mode, config).await;
+    // Half two: the guard fires, because the budget cannot. Three attempts against an hour.
+    let mut f = mode_fixture(mode, one_silent_agent(mode)).await;
+    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     f.harness.coordinator.deadlines = fly_session::Deadlines {
         probe: Duration::from_millis(50),
-        resolve: Duration::from_secs(600),
+        resolve: Duration::from_secs(3_600),
         resolve_attempts: 3,
         boot: Duration::from_secs(30),
         capture: Duration::from_secs(30),
         durable: Duration::from_secs(60),
     };
-    within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let failure = within("step", f.harness.coordinator.step())
         .await
         .expect_err("three attempts are not enough to resolve a silent participant");
     assert_eq!(f.harness.coordinator.last_resolution, Some(ResolutionEnd::AttemptsExhausted));
     assert!(
-        failure.error.message.contains("attempt guard") && failure.error.message.contains("3 attempts"),
+        failure.error.message.contains("attempt guard")
+            && failure.error.message.contains("3 attempts"),
         "the message names the bound that fired and its size: {failure}"
     );
-    assert_eq!(failure.participant.as_deref(), Some(fly_b().as_str()));
+    // Counted, not timed: the guard was spent exactly, and the hour never came near.
+    assert_eq!(f.harness.coordinator.last_resolution_attempts, 3);
+    assert_eq!(failure.participant.as_deref(), Some(fly_a().as_str()));
+    assert_eq!(failure.error.mutation, MutationCertainty::Unknown);
+    assert!(f.harness.coordinator.is_fenced(), "an exhausted guard fences the epoch too");
     f.shutdown().await;
 }
 
@@ -294,6 +427,35 @@ async fn a_delayed_one_agent_result_holds_the_world(mode: ExecutionMode) {
 // -------------------------------------------------------------------------------------------
 // Acceptance: worker or helper death has a bounded diagnosed outcome
 
+/// Waits until `worker` is provably inside the operation, then kills it.
+///
+/// Sleeping a fixed time before the kill asserts a race: under load the kill can land before
+/// the call is even dispatched, and then `MutationCertainty::None` is the *correct* answer
+/// because the participant never received anything. The certainty the death rows are about --
+/// `unknown`, because the participant died with work in its hands -- only holds if the work
+/// reached it, so the test waits for the worker's own status to say so rather than guessing
+/// from the clock.
+async fn kill_once_it_is_working(
+    launcher: &mut fly_session::Launcher,
+    worker: &Id,
+    inside: impl Fn(&StatusResult) -> bool,
+) -> ReapOutcome {
+    let deadline = Instant::now() + Duration::from_secs(15);
+    loop {
+        if let Ok(status) = launcher.health_check(worker).await
+            && inside(&status)
+        {
+            break;
+        }
+        assert!(
+            Instant::now() < deadline,
+            "{worker} never reported itself inside the operation"
+        );
+        tokio::time::sleep(Duration::from_millis(5)).await;
+    }
+    launcher.kill(worker).await
+}
+
 /// One agent dies in the middle of its Prepare. The epoch fails with a typed cause naming
 /// that agent, within the caller's own budget, and nothing continues on the remainder.
 async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
@@ -301,22 +463,21 @@ async fn a_worker_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     config.agents[1].faults = AgentFaults { prepare_delay_ms: 5_000, ..AgentFaults::default() };
     let mut f = mode_fixture(mode, config).await;
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
-    let started = Instant::now();
 
+    let victim = fly_b();
     let (coordinator, launcher) = f.harness.parts();
     let (stepped, reaped) = tokio::join!(
         async { within("step", coordinator.step()).await },
-        async {
-            tokio::time::sleep(Duration::from_millis(80)).await;
-            launcher.kill(&fly_b()).await
-        }
+        // Killed once it has the Prepare in its hands, not after a fixed sleep: the row is
+        // about a participant that dies *with work*, so the work has to have reached it.
+        kill_once_it_is_working(launcher, &victim, |status| {
+            status.state == WorkerState::Preparing && status.active_request_id.is_some()
+        })
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
+    // Boundedness is the suite's own `within` above: the participant is five seconds slow and
+    // `within` gives up at twenty, so returning at all is the claim.
     let failure = stepped.expect_err("a dead participant is a failed epoch, not a slow one");
-    assert!(
-        started.elapsed() < Duration::from_secs(20),
-        "the outcome must be bounded, not a hang"
-    );
     assert_eq!(
         failure.participant.as_deref(),
         Some(fly_b().as_str()),
@@ -350,19 +511,20 @@ async fn a_helper_death_has_a_bounded_diagnosed_outcome(mode: ExecutionMode) {
     let mut f = mode_fixture(mode, config).await;
     within("bootstrap", f.harness.coordinator.bootstrap()).await.unwrap();
     let environment = f.harness.environment_id();
-    let started = Instant::now();
 
     let (coordinator, launcher) = f.harness.parts();
     let (stepped, reaped) = tokio::join!(
         async { within("step", coordinator.step()).await },
-        async {
-            tokio::time::sleep(Duration::from_millis(200)).await;
-            launcher.kill(&environment).await
-        }
+        // Killed once the world has recorded the batch, which the arena does before its
+        // injected delay. So the Advance provably reached it and the certainty is `unknown`
+        // rather than `none`; a fixed sleep could land before dispatch under load, and then
+        // `none` would be right and this row would be asserting a race.
+        kill_once_it_is_working(launcher, &environment, |status| {
+            status.last_batch_id.is_some()
+        })
     );
     assert_eq!(reaped, ReapOutcome::Terminated);
     let failure = stepped.expect_err("a dead world is a failed epoch");
-    assert!(started.elapsed() < Duration::from_secs(20), "bounded, not a hang");
     assert_eq!(
         failure.participant.as_deref(),
         Some(environment.as_str()),