diff --git a/docs/design/session-framework/checkpoint-envelope-v1.md b/docs/design/session-framework/checkpoint-envelope-v1.md index d58ed98..69bdf4d 100644 --- a/docs/design/session-framework/checkpoint-envelope-v1.md +++ b/docs/design/session-framework/checkpoint-envelope-v1.md @@ -100,11 +100,19 @@ 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. + **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: diff --git a/services/flysim/crates/fly-session/src/agent.rs b/services/flysim/crates/fly-session/src/agent.rs index 21ca1a5..ab2e1a1 100644 --- a/services/flysim/crates/fly-session/src/agent.rs +++ b/services/flysim/crates/fly-session/src/agent.rs @@ -844,6 +844,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(), @@ -852,6 +853,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") } @@ -940,13 +946,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, ) } @@ -1147,10 +1155,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 diff --git a/services/flysim/crates/fly-session/src/coordinator.rs b/services/flysim/crates/fly-session/src/coordinator.rs index 8b7a444..7d68aa1 100644 --- a/services/flysim/crates/fly-session/src/coordinator.rs +++ b/services/flysim/crates/fly-session/src/coordinator.rs @@ -3036,15 +3036,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 { + 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. @@ -3233,7 +3251,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, @@ -3251,10 +3272,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, @@ -3833,6 +3869,7 @@ impl Coordinator { &row.model_version, &row.plasticity_version, row.seed, + &row.index_digest, ), )); } diff --git a/services/flysim/crates/fly-session/src/state.rs b/services/flysim/crates/fly-session/src/state.rs index aa42d3c..220310f 100644 --- a/services/flysim/crates/fly-session/src/state.rs +++ b/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, diff --git a/services/flysim/crates/fly-session/tests/publishing.rs b/services/flysim/crates/fly-session/tests/publishing.rs index bb94a1c..12f7af1 100644 --- a/services/flysim/crates/fly-session/tests/publishing.rs +++ b/services/flysim/crates/fly-session/tests/publishing.rs @@ -37,6 +37,7 @@ both_transports!( a_refused_event_batch_is_held_and_counted_not_lost, the_query_service_answers_reads_and_nothing_else, a_refused_snapshot_is_still_what_the_repair_path_answers, + a_replacement_that_built_another_index_cannot_install_the_checkpoint, the_published_descriptor_is_what_the_workers_attested_to, a_stimulus_kind_the_descriptor_does_not_declare_is_refused, a_restored_boundary_publishes_a_new_revision_and_no_transition, @@ -469,17 +470,68 @@ async fn a_revision_that_was_never_published_is_a_named_answer(via: Via) { f.shutdown().await; } +/// A graph identity does not cross a recovery. +/// +/// The index a fly attested to at `Agent.Initialize` is part of its compatibility identity, so +/// a replacement that built another graph -- the same dataset, the same neuron count, another +/// index -- cannot install a checkpoint taken under the first one, and the refusal names what +/// this worker is rather than only that two digests differ. The identical replacement still +/// installs, so the check refuses the case it is about and nothing else. +async fn a_replacement_that_built_another_index_cannot_install_the_checkpoint(via: Via) { + for (variant, refused) in [(0u64, false), (1, true)] { + let mut f = started(via).await; + let checkpoint = id("ck-graph"); + f.harness.coordinator.run(1).await.expect("one transition"); + within("checkpoint", f.harness.coordinator.checkpoint(&checkpoint)) + .await + .expect("a committed checkpoint"); + + f.harness.set_agent_graph(&fly_a(), variant); + f.harness.kill(&fly_a()).await; + f.harness + .coordinator + .step() + .await + .expect_err("a dead participant fails the epoch"); + within("replace", f.harness.replace_all_participants()) + .await + .expect("replacements"); + let outcome = within( + "restore", + f.harness.coordinator.restore(Some(&checkpoint), &id("e2")), + ) + .await; + match (refused, outcome) { + (false, Ok(_)) => {} + (false, Err(e)) => panic!("the identical replacement must still install: {e}"), + (true, Ok(_)) => panic!("a fly that built another index installed the checkpoint"), + (true, Err(failure)) => { + assert_eq!(failure.error.code, ErrorCode::IncompatibleState, "{failure:?}"); + let built = fly_session::agent::synthetic_graph(&fly_a(), 1); + assert!( + failure.error.message.contains(&built.index_digest), + "the refusal names the index this worker built: {}", + failure.error.message + ); + assert!( + f.harness.coordinator.is_fenced(), + "and the group stays fenced" + ); + } + } + f.shutdown().await; + } +} + /// The same neuron count, another index. A consumer that has mapped geometry is told, and the /// new composition is not quietly cached over the one it mapped. /// -/// The two descriptors here come from two real sessions rather than from a restore. Driving it -/// through a restore was tried and does not work yet, for a reason outside this slice: the -/// coordinator's `AgentSlot.graph` is written only by `Agent.Initialize`, and a group restore -/// installs state through `State.ActivateRestore`, so a replacement fly that built another -/// index is published under its predecessor's `indexDigest` -- and it is not refused on the way -/// in either, because `agent_compatibility` digests `agent::dataset_digest()` rather than the -/// index the worker attested to. Both halves belong to the restore contract, so this test uses -/// the compositions it can build honestly and the gap is reported rather than papered over. +/// The two descriptors here come from two real sessions, and deliberately not from a restore: +/// since the index is part of an agent's compatibility identity, a restore whose replacement +/// built another graph is now refused before it can install, which +/// `a_replacement_that_built_another_index_cannot_install_the_checkpoint` proves. A published +/// index therefore changes between compositions rather than across a recovery, and this is that +/// case. async fn a_changed_index_digest_is_named_rather_than_remapped(via: Via) { // Two real compositions that differ only in the graph one fly built. let first = started(via).await;