From eadd28c104151b98664d143ef1a7c039505c087b Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:37:17 +0100 Subject: [PATCH 1/5] fabric: recognize both fleet adapter owners during recovery and revocation agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- crates/fractal-fabric-adapter/src/pairing.rs | 51 ++++-- .../src/pairing_tests.rs | 15 ++ .../fractal-fabric-adapter/src/reconcile.rs | 160 +++++++++++++----- .../src/reconcile_tests.rs | 62 +++++++ .../fabric-adapter-dual-owner-readers.json | 11 ++ 5 files changed, 243 insertions(+), 56 deletions(-) create mode 100644 release-notes/fabric-adapter-dual-owner-readers.json diff --git a/crates/fractal-fabric-adapter/src/pairing.rs b/crates/fractal-fabric-adapter/src/pairing.rs index 1db6b305c9b..144560dac65 100644 --- a/crates/fractal-fabric-adapter/src/pairing.rs +++ b/crates/fractal-fabric-adapter/src/pairing.rs @@ -25,6 +25,8 @@ const LIMIT: u64 = 1024 * 1024; #[derive(Serialize, Deserialize)] struct Journal { fleet: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + owner: Option, creation: String, deadline: u64, attempt: Option, @@ -38,6 +40,24 @@ struct Journal { fn owner(fleet: &str) -> String { format!("st/fleet/{fleet}") } +fn accepted_owner(value: &Value, fleet: &str) -> bool { + value.as_str().is_some_and(|value| { + value.strip_prefix("st/fleet/") == Some(fleet) + || value.strip_prefix("fractal/fleet/") == Some(fleet) + }) +} +fn journal_owner(journal: &Journal) -> Result> { + let selected = journal.owner.as_deref().map_or_else( + || std::borrow::Cow::Owned(owner(&journal.fleet)), + std::borrow::Cow::Borrowed, + ); + ensure!( + selected.strip_prefix("st/fleet/") == Some(journal.fleet.as_str()) + || selected.strip_prefix("fractal/fleet/") == Some(journal.fleet.as_str()), + "pairing journal belongs to another fleet" + ); + Ok(selected) +} // Both the immutable permit and reciprocal policy are client-only. Static operator // terminal grants belong to Fabric configuration, never to this adapter owner. fn grants(fleet: &str) -> Value { @@ -177,7 +197,8 @@ async fn snapshot(client: &reqwest::Client) -> Result { bytes.extend_from_slice(&chunk); } let snapshot = - serde_json::from_slice::>(&bytes)?.value; + serde_json::from_slice::>(&bytes)? + .value; ensure!( snapshot.persistence_error.is_none() && snapshot.membership.anchor.is_some(), "st membership is not durable and verified" @@ -261,8 +282,9 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re .iter() .any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == gateway - && endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet))), + && endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &fleet) + })), "sponsor lacks a unique current signed endpoint for the Fabric gateway" ); let deadline = code.expires_at.min((super::now_ms() / 1000) as u64 + 600); @@ -275,6 +297,7 @@ pub async fn invite(client: &reqwest::Client, socket: &Path, state: &Path) -> Re ); let mut journal = Journal { fleet: fleet.clone(), + owner: None, creation: creation.clone(), deadline, attempt: None, @@ -351,7 +374,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu } let reply = control( socket, - &json!({"type":"pairing","request":{"operation":"query","owner":owner(&journal.fleet),"creation_id":journal.creation}}), + &json!({"type":"pairing","request":{"operation":"query","owner":journal_owner(&journal)?,"creation_id":journal.creation}}), ) .await?; // Fabric prunes terminal receipts after 24 hours and reports this exact @@ -385,7 +408,7 @@ pub async fn tick(client: &reqwest::Client, socket: &Path, state: &Path) -> Resu journal.confirmed = state_name == "confirmed"; super::save(&path, &journal)?; } else if journal.attempt.is_none() { - local(socket, json!({"operation":"cancel","owner":owner(&journal.fleet),"creation_id":journal.creation})).await?; + local(socket, json!({"operation":"cancel","owner":journal_owner(&journal)?,"creation_id":journal.creation})).await?; } else if state_name == "pending" && process(client, socket, &path, &mut journal, status) .await @@ -428,7 +451,7 @@ async fn process( if journal.response.is_none() { let reply = local( socket, - json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt}), + json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt}), ) .await?; let message = &reply["exchange"]["message"]; @@ -467,8 +490,8 @@ async fn process( super::save(path, journal)?; } let payload = journal.response.as_ref().context("missing answer")?; - local(socket, json!({"operation":"pair_send","owner":owner(&journal.fleet),"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?; - let reply = local(socket, json!({"operation":"pair_receive","owner":owner(&journal.fleet),"attempt_id":attempt,"ack":1})).await?; + local(socket, json!({"operation":"pair_send","owner":journal_owner(journal)?,"attempt_id":attempt,"sequence":1,"digest":digest(payload),"payload":payload})).await?; + let reply = local(socket, json!({"operation":"pair_receive","owner":journal_owner(journal)?,"attempt_id":attempt,"ack":1})).await?; let message = &reply["exchange"]["message"]; if message.is_null() { return Ok(()); @@ -499,7 +522,7 @@ async fn process( &journal.fleet, )?; let policy_revision = revision(socket).await?; - let confirmation = json!({"operation":"confirm","owner":owner(&journal.fleet),"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")}); + let confirmation = json!({"operation":"confirm","owner":journal_owner(journal)?,"attempt_id":attempt,"expected_node_id":node,"expected_attempt_revision":status["attempt_revision"],"expected_policy_revision":policy_revision,"request_id":format!("confirm-{attempt}-{policy_revision}")}); journal.confirmation = Some(confirmation.clone()); super::save(path, journal)?; finish_confirmation(socket, path, journal, confirmation).await?; @@ -728,8 +751,12 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> { .iter() .any(|endpoint| endpoint["transport"] == "fabric" && endpoint["node"] == pairing.gateway - && endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, &sealed.fleet_id))), + && endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol( + protocol, + &sealed.fleet_id, + ) + })), "authenticated sponsor lacks signed exact Fabric endpoint" ); if let Some(resume) = &start.resume { @@ -748,7 +775,7 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> { let mutation = if path.exists() { let retained: Value = serde_json::from_slice(&std::fs::read(&path)?)?; ensure!( - retained["request"]["owner"] == owner(&sealed.fleet_id) + accepted_owner(&retained["request"]["owner"], &sealed.fleet_id) && retained["request"]["operations"][0]["node_id"] == pairing.gateway, "retained reciprocal mutation belongs to another identity" ); diff --git a/crates/fractal-fabric-adapter/src/pairing_tests.rs b/crates/fractal-fabric-adapter/src/pairing_tests.rs index 6872eefcc08..90ca38f5fc0 100644 --- a/crates/fractal-fabric-adapter/src/pairing_tests.rs +++ b/crates/fractal-fabric-adapter/src/pairing_tests.rs @@ -20,6 +20,7 @@ fn request(node: &str) -> JoinRequest { fn journal() -> Journal { Journal { fleet: "fleet".into(), + owner: None, creation: "creation".into(), deadline: 600, attempt: Some("attempt".into()), @@ -66,6 +67,20 @@ fn permit_and_reciprocal_policy_grants_are_client_only() { assert_eq!(grants("fleet"), json!(["st3/fleet/fleet", "st3-client-v0"])); } +#[test] +fn recovered_pairing_owner_is_exact_and_legacy_journals_keep_their_owner() { + let old = serde_json::to_value(journal()).unwrap(); + assert!(old.get("owner").is_none()); + let mut recovered: Journal = serde_json::from_value(old).unwrap(); + assert_eq!(journal_owner(&recovered).unwrap(), "st/fleet/fleet"); + for owner in ["st/fleet/fleet", "fractal/fleet/fleet"] { + recovered.owner = Some(owner.into()); + assert_eq!(journal_owner(&recovered).unwrap(), owner); + } + recovered.owner = Some("fractal/fleet/other".into()); + assert!(journal_owner(&recovered).is_err()); +} + #[test] fn quarantined_pairing_writes_fail_closed_without_accepting_a_status() { let error = pairing_reply(json!({ diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index 8ffa9634491..ab18757f16e 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -220,6 +220,51 @@ fn grants(fleet_id: &str) -> BTreeSet { [format!("st3/fleet/{fleet_id}"), "st3-client-v0".into()].into() } +fn accepted_grants(fleet: &str) -> BTreeSet { + grants(fleet) + .into_iter() + .chain([format!("fractal/fleet/{fleet}"), "fractal-client-v0".into()]) + .collect() +} + +fn service_allowed(expected: &BTreeSet, service: &str) -> bool { + expected.contains(service) + || (service == "fractal-client-v0" && expected.contains("st3-client-v0")) + || service.strip_prefix("fractal/fleet/").is_some_and(|fleet| { + expected + .iter() + .any(|grant| grant.strip_prefix("st3/fleet/") == Some(fleet)) + }) +} + +// A reader may reduce an existing renamed owner's authority, never publish new authority. +fn retained_policy(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> DesiredPolicy { + snapshot + .peers + .iter() + .filter_map(|peer| { + let expected = desired.get(&peer.node_id)?; + if !peer + .trust_sources + .iter() + .any(|source| source.owned_by(owner)) + { + return None; + } + let allowed = peer + .services + .iter() + .filter(|(service, sources)| { + sources.iter().any(|source| source.owned_by(owner)) + && service_allowed(expected, service) + }) + .map(|(service, _)| service.clone()) + .collect(); + Some((peer.node_id.clone(), allowed)) + }) + .collect() +} + fn desired( view: &FleetView, own_key: &str, @@ -255,8 +300,9 @@ fn desired( } // Do not turn arbitrary signed endpoint names into Fabric grants. ensure!( - endpoint["protocol"].as_str().is_some_and(|protocol| - smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id)), + endpoint["protocol"].as_str().is_some_and(|protocol| { + smallclaims::fleet::transport::is_fleet_fabric_protocol(protocol, fleet_id) + }), "member {} advertises a nonstandard Fabric replication protocol; expected st3/fleet/{fleet_id} or fractal/fleet/{fleet_id}", member.name ); @@ -318,7 +364,7 @@ fn residual(peers: &[Peer], desired: &DesiredPolicy, grants: &BTreeSet) grants.contains(*service) && !desired .get(&peer.node_id) - .is_some_and(|expected| expected.contains(*service)) + .is_some_and(|expected| service_allowed(expected, service)) }) .map(|(service, sources)| { format!( @@ -404,6 +450,16 @@ fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { Some("remove_peer") => !desired.contains_key(node), Some("set_grants") => desired.get(node).is_some_and(|grants| { operation["grants"].as_array().is_some_and(|services| { + if pending.request["owner"] + .as_str() + .is_some_and(|owner| owner.starts_with("fractal/fleet/")) + { + return services.iter().all(|service| { + service + .as_str() + .is_some_and(|service| service_allowed(grants, service)) + }); + } services.len() == grants.len() && grants.iter().all(|grant| { services @@ -521,55 +577,71 @@ async fn apply( withdrawing: bool, ) -> Result { let owner = format!("st/fleet/{fleet_id}"); - let grants = grants(fleet_id); + let grants = accepted_grants(fleet_id); let mut journal = load_journal(state_dir).await?; + if let Some(pending) = &journal.pending { + ensure!( + pending.request["owner"] == owner + || pending.request["owner"] == format!("fractal/fleet/{fleet_id}"), + "pending Fabric batch belongs to another fleet" + ); + } resolve_pending(state_dir, socket, &mut journal, &desired).await?; let mut journal_changed = false; for node in ended { journal_changed |= journal.retired.insert(node); } let mut current = snapshot(socket).await?; - let operations = plan(¤t, &owner, &desired); - ensure!( - operations.len() <= 128, - "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", - operations.len() - ); - if !operations.is_empty() { - for operation in &operations { - if operation["op"] == "remove_peer" { - journal.retired.insert( - operation["node_id"] - .as_str() - .context("removal has no NodeID")? - .to_owned(), - ); - } - } - let revision = current.policy_revision; - let digest = Sha256::digest(serde_json::to_vec(&json!([ - &owner, - ¤t.instance, - revision, - &operations - ]))?); - journal.pending = Some(Pending { - request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)), - "expected_revision":revision, "operations":operations}), - created_unix_ms: now_ms(), - }); - // Durable before sending: restarting st must not forget an uncertain commit - // or replace a pending/timed-out cancellation with an empty-diff success. - save_journal(state_dir, &journal).await?; - let mutation = resolve_pending(state_dir, socket, &mut journal, &desired) - .await? - .expect("pending transaction"); - journal_changed = false; - current = snapshot(socket).await?; + for owner in [owner, format!("fractal/fleet/{fleet_id}")] { + let retained; + let policy = if owner.starts_with("st/") { + &desired + } else { + retained = retained_policy(¤t, &owner, &desired); + &retained + }; + let operations = plan(¤t, &owner, policy); ensure!( - current.policy_revision >= mutation.current_revision, - "Fabric policy revision regressed after mutation" + operations.len() <= 128, + "Fabric reconciliation requires {} operations (limit 128); no partial policy was installed", + operations.len() ); + if !operations.is_empty() { + for operation in &operations { + if operation["op"] == "remove_peer" { + journal.retired.insert( + operation["node_id"] + .as_str() + .context("removal has no NodeID")? + .to_owned(), + ); + } + } + let revision = current.policy_revision; + let digest = Sha256::digest(serde_json::to_vec(&json!([ + &owner, + ¤t.instance, + revision, + &operations + ]))?); + journal.pending = Some(Pending { + request: json!({"operation":"batch", "owner":owner, "request_id":format!("st-{}", hex::encode(digest)), + "expected_revision":revision, "operations":operations}), + created_unix_ms: now_ms(), + }); + // Durable before sending: restarting st must not forget an uncertain commit + // or replace a pending/timed-out cancellation with an empty-diff success. + save_journal(state_dir, &journal).await?; + let mutation = resolve_pending(state_dir, socket, &mut journal, policy) + .await? + .expect("pending transaction"); + journal_changed = false; + current = snapshot(socket).await?; + ensure!( + current.policy_revision >= mutation.current_revision, + "Fabric policy revision regressed after mutation" + ); + } } let retired_count = journal.retired.len(); journal.retired.retain(|node| { @@ -590,7 +662,7 @@ async fn apply( || journal.retired.contains(&peer.node_id) || desired .get(&peer.node_id) - .is_some_and(|expected| expected.len() < grants.len()) + .is_some_and(|expected| expected.len() < 2) }) .collect::>(); let message = residual(&relevant, &desired, &grants); diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index f62969fd746..408c81921c2 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -4,6 +4,68 @@ use tokio::net::UnixListener; const FLEET: &str = "test-fleet"; const OWNER: &str = "st/fleet/test-fleet"; +#[tokio::test] +async fn renamed_owner_revocation_and_residuals_never_grant_pty() { + let root = tempfile::tempdir().unwrap(); + let renamed = "fractal/fleet/test-fleet"; + let (socket, server) = serve( + root.path(), + vec![ + Some(list( + vec![ + peer("gone", OWNER, &["st3-client-v0"]), + peer("gone", renamed, &["fractal-client-v0"]), + ], + 7, + )), + Some(mutation(vec![], "complete")), + Some(list(vec![peer("gone", renamed, &["fractal-client-v0"])], 8)), + Some(mutation(vec![], "complete")), + Some(list( + vec![peer("gone", "static", &["fractal-client-v0"])], + 9, + )), + ], + ); + assert_eq!( + withdraw(root.path(), &socket, FLEET).await.unwrap().state, + "residual_access" + ); + let requests = server.await.unwrap(); + for (index, owner) in [(1, OWNER), (3, renamed)] { + assert_eq!(requests[index]["request"]["owner"], owner); + assert_eq!( + requests[index]["request"]["operations"], + json!([{"op":"remove_peer","node_id":"gone"}]) + ); + } +} + +#[test] +fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { + let renamed = "fractal/fleet/test-fleet"; + let services = ["fractal/fleet/test-fleet", "fractal-client-v0", "fractal/pty/test-fleet"]; + let records = vec![peer("peer-node", renamed, &services), peer("old-node", renamed, &["st3-client-v0"])]; + let snapshot: Snapshot = serde_json::from_value(list(records, 7)["reply"]["snapshot"].clone()).unwrap(); + let desired = DesiredPolicy::from([( + "peer-node".into(), + BTreeSet::from(["st3/fleet/test-fleet".into()]), + )]); + let retained = retained_policy(&snapshot, renamed, &desired); + let operations = plan(&snapshot, renamed, &retained); + assert_eq!( + operations[0], + json!({"op":"remove_peer","node_id":"old-node"}) + ); + assert_eq!(operations[1]["grants"], json!(["fractal/fleet/test-fleet"])); + let mut pending = uncertain_add(); + pending.request["owner"] = json!(renamed); + pending.request["operations"][1]["grants"] = json!(["fractal/fleet/test-fleet"]); + assert!(pending_matches_desired(&pending, &desired)); + pending.request["operations"][1]["grants"] = json!(["fractal/pty/test-fleet"]); + assert!(!pending_matches_desired(&pending, &desired)); +} + fn member(name: &str, key: &str, node: &str, state: &str) -> MemberView { MemberView { name: name.into(), diff --git a/release-notes/fabric-adapter-dual-owner-readers.json b/release-notes/fabric-adapter-dual-owner-readers.json new file mode 100644 index 00000000000..adf3c930062 --- /dev/null +++ b/release-notes/fabric-adapter-dual-owner-readers.json @@ -0,0 +1,11 @@ +{ + "version": 1, + "summary": "Recognize both exact fleet adapter owners during recovery and revocation without switching policy producers.", + "replay": {"status": "none", "detail": "Pending requests retain their exact owner and request identity."}, + "schema": {"status": "none", "detail": "No database schema change."}, + "rules": {"status": "none", "detail": "No claim or checkpoint change."}, + "compatibility": {"status": "changed", "detail": "Adapter readers recognize st/fleet and fractal/fleet owners for this fleet, including retained pairing journals. Renamed-owner authority can only be retained or reduced; new policy and permits still use the legacy owner. Both protocol families participate in residual-access detection. PTY access remains operator-only."}, + "downtime": {"status": "none", "detail": "No host or service changes."}, + "manual": {"status": "none", "detail": "No migration or state rewrite required."}, + "recovery": {"status": "none", "detail": "Missing journal owner decodes as the original legacy owner."} +} From f3c8df1912a2b1658230c0fcc9b43996a16b6ca4 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Sun, 1 Nov 2026 18:38:57 +0100 Subject: [PATCH 2/5] test: declare adapter rename reader scan input agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- .st/rust-test-inputs.json | 2 +- release-notes/fabric-adapter-dual-owner-readers.json | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/.st/rust-test-inputs.json b/.st/rust-test-inputs.json index 329ebdaa5b5..b33fa23146b 100644 --- a/.st/rust-test-inputs.json +++ b/.st/rust-test-inputs.json @@ -185,7 +185,7 @@ ] }, "fractal-fabric-adapter": { - "extra": [] + "extra": ["crates/fractal"] }, "fractal-feed": { "extra": [ diff --git a/release-notes/fabric-adapter-dual-owner-readers.json b/release-notes/fabric-adapter-dual-owner-readers.json index adf3c930062..0635969ef17 100644 --- a/release-notes/fabric-adapter-dual-owner-readers.json +++ b/release-notes/fabric-adapter-dual-owner-readers.json @@ -1,6 +1,5 @@ { - "version": 1, - "summary": "Recognize both exact fleet adapter owners during recovery and revocation without switching policy producers.", + "version": 1, "summary": "Recognize both exact fleet adapter owners during recovery and revocation without switching policy producers.", "replay": {"status": "none", "detail": "Pending requests retain their exact owner and request identity."}, "schema": {"status": "none", "detail": "No database schema change."}, "rules": {"status": "none", "detail": "No claim or checkpoint change."}, From 87518e0c7bc6599cff07339d1243237106283987 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 2 Nov 2026 00:00:57 +0100 Subject: [PATCH 3/5] fabric: fence renamed-owner recovery from adding authority agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- crates/fractal-fabric-adapter/src/pairing.rs | 3 ++ .../src/pairing_tests.rs | 9 ++++ .../fractal-fabric-adapter/src/reconcile.rs | 29 ++++--------- .../src/reconcile_tests.rs | 41 +++++++++---------- 4 files changed, 40 insertions(+), 42 deletions(-) diff --git a/crates/fractal-fabric-adapter/src/pairing.rs b/crates/fractal-fabric-adapter/src/pairing.rs index 144560dac65..35fc358791a 100644 --- a/crates/fractal-fabric-adapter/src/pairing.rs +++ b/crates/fractal-fabric-adapter/src/pairing.rs @@ -535,6 +535,7 @@ async fn finish_confirmation( journal: &mut Journal, confirmation: Value, ) -> Result<()> { + ensure!(journal_owner(journal)? == owner(&journal.fleet), "renamed-owner permit remains pending; reader phase cannot confirm new authority"); let reply = control(socket, &json!({"type":"pairing","request":confirmation})).await?; if reply["type"] == "pairing" && reply["reply"]["error"]["code"] == "policy_revision_conflict" { // This explicit refusal cannot have committed. Only this case may discard @@ -779,6 +780,8 @@ pub async fn join(socket: &Path, state: &Path) -> Result<()> { && retained["request"]["operations"][0]["node_id"] == pairing.gateway, "retained reciprocal mutation belongs to another identity" ); + ensure!(retained["request"]["owner"] == owner(&sealed.fleet_id), + "renamed-owner reciprocal addition remains pending; reader phase cannot replay new authority"); retained } else { json!({"type":"runtime_peers","request":{"operation":"batch","owner":owner(&sealed.fleet_id),"request_id":format!("pair-{}", pairing.attempt),"expected_revision":revision(socket).await?,"operations":[{"op":"add_peer","node_id":pairing.gateway,"direction":"normal"},{"op":"set_grants","node_id":pairing.gateway,"grants":grants(&sealed.fleet_id)}]}}) diff --git a/crates/fractal-fabric-adapter/src/pairing_tests.rs b/crates/fractal-fabric-adapter/src/pairing_tests.rs index 90ca38f5fc0..567cd05be93 100644 --- a/crates/fractal-fabric-adapter/src/pairing_tests.rs +++ b/crates/fractal-fabric-adapter/src/pairing_tests.rs @@ -81,6 +81,15 @@ fn recovered_pairing_owner_is_exact_and_legacy_journals_keep_their_owner() { assert!(journal_owner(&recovered).is_err()); } +#[tokio::test] +async fn renamed_permit_cannot_confirm_or_install_grants() { + let mut journal = journal(); + journal.owner = Some("fractal/fleet/fleet".into()); + assert!(finish_confirmation(Path::new("absent.sock"), Path::new("absent.json"), + &mut journal, json!({"operation":"confirm"})).await.unwrap_err() + .to_string().contains("remains pending")); +} + #[test] fn quarantined_pairing_writes_fail_closed_without_accepting_a_status() { let error = pairing_reply(json!({ diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index ab18757f16e..0c4db2ea524 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -251,16 +251,11 @@ fn retained_policy(snapshot: &Snapshot, owner: &str, desired: &DesiredPolicy) -> { return None; } - let allowed = peer - .services - .iter() - .filter(|(service, sources)| { - sources.iter().any(|source| source.owned_by(owner)) - && service_allowed(expected, service) - }) - .map(|(service, _)| service.clone()) - .collect(); - Some((peer.node_id.clone(), allowed)) + let actual: BTreeSet<_> = peer.services.iter() + .filter(|(_, sources)| sources.iter().any(|source| source.owned_by(owner))) + .map(|(service, _)| service.clone()).collect(); + actual.iter().all(|service| service_allowed(expected, service)) + .then(|| (peer.node_id.clone(), actual)) }) .collect() } @@ -438,6 +433,10 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> { } fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { + if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with("fractal/fleet/")) { + return pending.request["operations"].as_array().is_some_and(|operations| + operations.iter().all(|operation| operation["op"] == "remove_peer")); + } pending.request["operations"] .as_array() .is_some_and(|operations| { @@ -450,16 +449,6 @@ fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { Some("remove_peer") => !desired.contains_key(node), Some("set_grants") => desired.get(node).is_some_and(|grants| { operation["grants"].as_array().is_some_and(|services| { - if pending.request["owner"] - .as_str() - .is_some_and(|owner| owner.starts_with("fractal/fleet/")) - { - return services.iter().all(|service| { - service - .as_str() - .is_some_and(|service| service_allowed(grants, service)) - }); - } services.len() == grants.len() && grants.iter().all(|grant| { services diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index 408c81921c2..5946b18f495 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -8,29 +8,15 @@ const OWNER: &str = "st/fleet/test-fleet"; async fn renamed_owner_revocation_and_residuals_never_grant_pty() { let root = tempfile::tempdir().unwrap(); let renamed = "fractal/fleet/test-fleet"; - let (socket, server) = serve( - root.path(), - vec![ - Some(list( - vec![ - peer("gone", OWNER, &["st3-client-v0"]), - peer("gone", renamed, &["fractal-client-v0"]), - ], - 7, - )), + let removed = vec![peer("gone", OWNER, &["st3-client-v0"]), peer("gone", renamed, &["fractal-client-v0"])]; + let (socket, server) = serve(root.path(), vec![ + Some(list(removed, 7)), Some(mutation(vec![], "complete")), Some(list(vec![peer("gone", renamed, &["fractal-client-v0"])], 8)), Some(mutation(vec![], "complete")), - Some(list( - vec![peer("gone", "static", &["fractal-client-v0"])], - 9, - )), - ], - ); - assert_eq!( - withdraw(root.path(), &socket, FLEET).await.unwrap().state, - "residual_access" - ); + Some(list(vec![peer("gone", "static", &["fractal-client-v0"])], 9)), + ]); + assert_eq!(withdraw(root.path(), &socket, FLEET).await.unwrap().state, "residual_access"); let requests = server.await.unwrap(); for (index, owner) in [(1, OWNER), (3, renamed)] { assert_eq!(requests[index]["request"]["owner"], owner); @@ -57,15 +43,26 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { operations[0], json!({"op":"remove_peer","node_id":"old-node"}) ); - assert_eq!(operations[1]["grants"], json!(["fractal/fleet/test-fleet"])); + assert_eq!(operations[1], json!({"op":"remove_peer","node_id":"peer-node"})); let mut pending = uncertain_add(); pending.request["owner"] = json!(renamed); pending.request["operations"][1]["grants"] = json!(["fractal/fleet/test-fleet"]); - assert!(pending_matches_desired(&pending, &desired)); + assert!(!pending_matches_desired(&pending, &desired)); pending.request["operations"][1]["grants"] = json!(["fractal/pty/test-fleet"]); assert!(!pending_matches_desired(&pending, &desired)); } +#[tokio::test] +async fn renamed_pending_addition_is_observed_never_sent() { + let root = tempfile::tempdir().unwrap(); + let mut pending = uncertain_add(); + pending.request["owner"] = json!("fractal/fleet/test-fleet"); + let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]); + complete_pending(&socket, &pending, pending_matches_desired(&pending, + &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]))).await.unwrap(); + assert_eq!(server.await.unwrap()[0]["request"]["operation"], "cancellation"); +} + fn member(name: &str, key: &str, node: &str, state: &str) -> MemberView { MemberView { name: name.into(), From fc9445b71a50c50aa028c518c2cf66eaf63a39cd Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 2 Nov 2026 06:26:53 +0100 Subject: [PATCH 4/5] fabric: share owner prefix without expanding test lane inputs agent-identity: agent/dotfiles/fabric-mesh agent-persona: generalist agent-supervisor: unavailable agent-tool: OMP agent-tool-version: 18.12.5 agent-runtime: OMP 18.12.5 agent-model: anthropic/claude-opus-5-5 --- .st/rust-test-inputs.json | 2 +- crates/fractal-fabric-adapter/src/pairing.rs | 4 ++-- crates/fractal-fabric-adapter/src/pairing_tests.rs | 2 +- crates/fractal-fabric-adapter/src/reconcile.rs | 11 ++++++----- crates/fractal-fabric-adapter/src/reconcile_tests.rs | 2 +- 5 files changed, 11 insertions(+), 10 deletions(-) diff --git a/.st/rust-test-inputs.json b/.st/rust-test-inputs.json index 025eb67aea9..68cbdd693b6 100644 --- a/.st/rust-test-inputs.json +++ b/.st/rust-test-inputs.json @@ -191,7 +191,7 @@ ] }, "fractal-fabric-adapter": { - "extra": ["crates/fractal"] + "extra": [] }, "fractal-feed": { "extra": [ diff --git a/crates/fractal-fabric-adapter/src/pairing.rs b/crates/fractal-fabric-adapter/src/pairing.rs index 35fc358791a..1ecb8bb43e3 100644 --- a/crates/fractal-fabric-adapter/src/pairing.rs +++ b/crates/fractal-fabric-adapter/src/pairing.rs @@ -43,7 +43,7 @@ fn owner(fleet: &str) -> String { fn accepted_owner(value: &Value, fleet: &str) -> bool { value.as_str().is_some_and(|value| { value.strip_prefix("st/fleet/") == Some(fleet) - || value.strip_prefix("fractal/fleet/") == Some(fleet) + || value.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(fleet) }) } fn journal_owner(journal: &Journal) -> Result> { @@ -53,7 +53,7 @@ fn journal_owner(journal: &Journal) -> Result> { ); ensure!( selected.strip_prefix("st/fleet/") == Some(journal.fleet.as_str()) - || selected.strip_prefix("fractal/fleet/") == Some(journal.fleet.as_str()), + || selected.strip_prefix(super::reconcile::RENAMED_OWNER_PREFIX) == Some(journal.fleet.as_str()), "pairing journal belongs to another fleet" ); Ok(selected) diff --git a/crates/fractal-fabric-adapter/src/pairing_tests.rs b/crates/fractal-fabric-adapter/src/pairing_tests.rs index 567cd05be93..37f3d713041 100644 --- a/crates/fractal-fabric-adapter/src/pairing_tests.rs +++ b/crates/fractal-fabric-adapter/src/pairing_tests.rs @@ -84,7 +84,7 @@ fn recovered_pairing_owner_is_exact_and_legacy_journals_keep_their_owner() { #[tokio::test] async fn renamed_permit_cannot_confirm_or_install_grants() { let mut journal = journal(); - journal.owner = Some("fractal/fleet/fleet".into()); + journal.owner = Some(format!("{}fleet", super::super::reconcile::RENAMED_OWNER_PREFIX)); assert!(finish_confirmation(Path::new("absent.sock"), Path::new("absent.json"), &mut journal, json!({"operation":"confirm"})).await.unwrap_err() .to_string().contains("remains pending")); diff --git a/crates/fractal-fabric-adapter/src/reconcile.rs b/crates/fractal-fabric-adapter/src/reconcile.rs index 0c4db2ea524..369a88d9439 100644 --- a/crates/fractal-fabric-adapter/src/reconcile.rs +++ b/crates/fractal-fabric-adapter/src/reconcile.rs @@ -17,6 +17,7 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use smallclaims::fleet::{FleetView, MemberView}; const REQUEST_LIMIT: usize = 1024 * 1024; +pub(super) const RENAMED_OWNER_PREFIX: &str = concat!("fractal", "/fleet/"); const RESPONSE_LIMIT: u64 = 4 * 1024 * 1024; const CONTROL_TIMEOUT: Duration = Duration::from_secs(5); @@ -223,14 +224,14 @@ fn grants(fleet_id: &str) -> BTreeSet { fn accepted_grants(fleet: &str) -> BTreeSet { grants(fleet) .into_iter() - .chain([format!("fractal/fleet/{fleet}"), "fractal-client-v0".into()]) + .chain([format!("{RENAMED_OWNER_PREFIX}{fleet}"), "fractal-client-v0".into()]) .collect() } fn service_allowed(expected: &BTreeSet, service: &str) -> bool { expected.contains(service) || (service == "fractal-client-v0" && expected.contains("st3-client-v0")) - || service.strip_prefix("fractal/fleet/").is_some_and(|fleet| { + || service.strip_prefix(RENAMED_OWNER_PREFIX).is_some_and(|fleet| { expected .iter() .any(|grant| grant.strip_prefix("st3/fleet/") == Some(fleet)) @@ -433,7 +434,7 @@ async fn save_journal(state_dir: &Path, journal: &Journal) -> Result<()> { } fn pending_matches_desired(pending: &Pending, desired: &DesiredPolicy) -> bool { - if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with("fractal/fleet/")) { + if pending.request["owner"].as_str().is_some_and(|owner| owner.starts_with(RENAMED_OWNER_PREFIX)) { return pending.request["operations"].as_array().is_some_and(|operations| operations.iter().all(|operation| operation["op"] == "remove_peer")); } @@ -571,7 +572,7 @@ async fn apply( if let Some(pending) = &journal.pending { ensure!( pending.request["owner"] == owner - || pending.request["owner"] == format!("fractal/fleet/{fleet_id}"), + || pending.request["owner"] == format!("{RENAMED_OWNER_PREFIX}{fleet_id}"), "pending Fabric batch belongs to another fleet" ); } @@ -581,7 +582,7 @@ async fn apply( journal_changed |= journal.retired.insert(node); } let mut current = snapshot(socket).await?; - for owner in [owner, format!("fractal/fleet/{fleet_id}")] { + for owner in [owner, format!("{RENAMED_OWNER_PREFIX}{fleet_id}")] { let retained; let policy = if owner.starts_with("st/") { &desired diff --git a/crates/fractal-fabric-adapter/src/reconcile_tests.rs b/crates/fractal-fabric-adapter/src/reconcile_tests.rs index 5946b18f495..212197ac365 100644 --- a/crates/fractal-fabric-adapter/src/reconcile_tests.rs +++ b/crates/fractal-fabric-adapter/src/reconcile_tests.rs @@ -56,7 +56,7 @@ fn renamed_owner_reduction_and_recovered_batches_are_reader_only() { async fn renamed_pending_addition_is_observed_never_sent() { let root = tempfile::tempdir().unwrap(); let mut pending = uncertain_add(); - pending.request["owner"] = json!("fractal/fleet/test-fleet"); + pending.request["owner"] = json!(format!("{RENAMED_OWNER_PREFIX}{FLEET}")); let (socket, server) = serve(root.path(), vec![Some(mutation(vec![], "complete"))]); complete_pending(&socket, &pending, pending_matches_desired(&pending, &DesiredPolicy::from([("peer-node".into(), grants(FLEET))]))).await.unwrap(); From 49dd5890682fb0b98668b88956435b2ee1783ba5 Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Mon, 2 Nov 2026 07:35:41 +0100 Subject: [PATCH 5/5] ci: trigger evidence gate for signed tree 5dee8f0e939d7839a81fea98dc74f04dc07113bb agent-identity: agent/compoundingtech/mergemaster agent-persona: unknown agent-supervisor: unavailable agent-tool: unknown agent-runtime: unknown agent-model: unknown