alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

session: a graph identity does not cross a recovery

Browse source 
The index an agent attested to at Agent.Initialize joins its compatibility
identity and its checkpoint manifest row, so a replacement fly that built
another graph -- the same dataset, the same neuron count, another index --
cannot install a checkpoint taken under the first one. It was accepted before,
because agent_compatibility digested a dataset digest recomputed from a free
function rather than what the worker attested to, and the restored composition
was then published under its predecessor's indexDigest.

The refusal names what the worker is, not just that two digests differ. Dated
amendment to checkpoint-envelope-v1's agents row; no wire type and no schema
text change, so the contract digest is unchanged.
This commit is contained in:
dev 2026-09-22 20:36:51 +00:00
parent 6bf5687aca
commit 305a50d8df
5 changed files with 136 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
docs/design/session-framework/checkpoint-envelope-v1.md
View file

@ -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:

22
services/flysim/crates/fly-session/src/agent.rs
View file

@ -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

49
services/flysim/crates/fly-session/src/coordinator.rs
View file

@ -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<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.
@ -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,
),
));
}

6
services/flysim/crates/fly-session/src/state.rs
View file

@ -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,

68
services/flysim/crates/fly-session/tests/publishing.rs
View file

@ -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;