From 6c46f2350ac1aed4a0ed69cb2ca4a50dcc656d89 Mon Sep 17 00:00:00 2001 From: Storme Drone <49c46e84758b2ebff4abf5abbbd44ee4ce788fc3b55db9fa703eec124eead621@buzz.block.builderlab.xyz> Date: Mon, 31 Aug 2026 17:43:42 -0600 Subject: [PATCH] Make team membership bindings durable Signed-off-by: Storme Drone <49c46e84758b2ebff4abf5abbbd44ee4ce788fc3b55db9fa703eec124eead621@buzz.block.builderlab.xyz> --- .../src/commands/personas/inbound.rs | 67 ++++++++----- .../personas/inbound/inbound_tests.rs | 71 ++++++++++++-- desktop/src-tauri/src/commands/teams/mod.rs | 94 ++++++++++--------- desktop/src-tauri/src/commands/teams/tests.rs | 22 +++-- .../teams/tests/membership_recovery.rs | 55 +++++++++-- 5 files changed, 214 insertions(+), 95 deletions(-) diff --git a/desktop/src-tauri/src/commands/personas/inbound.rs b/desktop/src-tauri/src/commands/personas/inbound.rs index 2080630742..8a5aa82894 100644 --- a/desktop/src-tauri/src/commands/personas/inbound.rs +++ b/desktop/src-tauri/src/commands/personas/inbound.rs @@ -219,6 +219,10 @@ fn reconcile_inbound_persona_event_blocking( .managed_agents_store_lock .lock() .map_err(|error| error.to_string())?; + // A prior create, update, or inbound team write can have reached teams.json + // before its agent-store write failed. Replay it before accepting any next + // inbound event, so an identical retained event cannot consume that intent. + crate::commands::replay_pending_team_membership(&app)?; // Resolve inbound vs. any pending local edit before touching the store, in // the scope the event ARRIVED on. A workspace switch since arrival leaves @@ -305,9 +309,11 @@ fn reconcile_inbound_persona_event_blocking( &mut teams, d_tag, team_content_from_event(&event)?, + |pending| crate::commands::teams::save_pending_team_membership(&app, pending), |teams| save_teams(&app, teams), || load_managed_agents(&app), |records| save_managed_agents(&app, records), + || crate::commands::teams::clear_pending_team_membership(&app), ) })?; if outcome == InboundOutcome::Skipped { @@ -751,29 +757,23 @@ fn apply_inbound_managed_agent( false } -/// In-memory core of the inbound `KIND_TEAM` reconcile: capture the matched -/// team's roster *before* applying the inbound projection, apply it, persist -/// teams authoritatively, then propagate the prior→current membership delta to -/// live instances best-effort — the same binding semantics the local -/// create/update commands use. Without this, a 30176 team edit from another -/// device lands on `teams.json` but never touches `ManagedAgentRecord.team_id`: -/// an added persona's running instances stay unbound (member in roster, not in -/// behavior) and a removed persona's instances keep drawing the old team's -/// instructions at spawn until restart. -/// -/// A no-match insert has no prior roster, so its whole roster is the added -/// delta — symmetric with `commit_team_create`. Injected persistence keeps it -/// `AppHandle`-free so the prior-roster capture and delta direction are -/// unit-testable; a `persist_teams` error propagates, agent IO is best-effort -/// (mirrors the local command path: the authoritative team write already -/// landed, and boot repair is the designed retry for a stale binding). +/// In-memory core of the inbound `KIND_TEAM` reconcile. It stages a changed +/// roster before the team write, then commits the team and agent stores. It +/// clears the stage only after both writes succeed. This makes an agent-store +/// failure retryable when the team contains a persona shared by another team. +/// The caller advances relay retention only after this function succeeds. +#[allow(clippy::too_many_arguments)] fn commit_inbound_team( teams: &mut Vec, d_tag: String, inbound: TeamEventContent, + save_pending: impl FnOnce( + &crate::commands::teams::PendingTeamMembershipUpdate, + ) -> Result<(), String>, persist_teams: impl FnOnce(&[TeamRecord]) -> Result<(), String>, load_agents: impl FnOnce() -> Result, String>, save_agents: impl FnOnce(&[ManagedAgentRecord]) -> Result<(), String>, + clear_pending: impl FnOnce() -> Result<(), String>, ) -> Result<(), String> { let team_id = d_tag.clone(); let previous_persona_ids = teams @@ -787,14 +787,33 @@ fn commit_inbound_team( .find(|record| record.id == team_id) .map(|record| record.persona_ids.clone()) .unwrap_or_default(); - persist_teams(teams)?; - crate::commands::teams::propagate_membership_best_effort( - &team_id, - &previous_persona_ids, - ¤t_persona_ids, - load_agents, - save_agents, - ); + let membership_changed = previous_persona_ids != current_persona_ids; + if membership_changed { + save_pending(&crate::commands::teams::PendingTeamMembershipUpdate { + team_id: team_id.clone(), + previous_persona_ids: previous_persona_ids.clone(), + current_persona_ids: current_persona_ids.clone(), + })?; + persist_teams(teams)?; + crate::commands::teams::propagate_membership( + &team_id, + &previous_persona_ids, + ¤t_persona_ids, + load_agents, + save_agents, + ) + .map_err(|error| format!("could not update inbound team agents: {error}"))?; + clear_pending()?; + } else { + persist_teams(teams)?; + crate::commands::teams::propagate_membership_best_effort( + &team_id, + &previous_persona_ids, + ¤t_persona_ids, + load_agents, + save_agents, + ); + } Ok(()) } diff --git a/desktop/src-tauri/src/commands/personas/inbound/inbound_tests.rs b/desktop/src-tauri/src/commands/personas/inbound/inbound_tests.rs index ab93243755..c7f3ac73c2 100644 --- a/desktop/src-tauri/src/commands/personas/inbound/inbound_tests.rs +++ b/desktop/src-tauri/src/commands/personas/inbound/inbound_tests.rs @@ -594,11 +594,13 @@ fn inbound_team_add_binds_unbound_instance_through_wiring() { persona_ids: Some(vec!["p-existing".to_string(), "p-added".to_string()]), }, |_| Ok(()), + |_| Ok(()), || Ok(existing.clone()), |records| { *saved.borrow_mut() = Some(records.to_vec()); Ok(()) }, + || Ok(()), ) .expect("inbound add succeeds"); @@ -638,11 +640,13 @@ fn inbound_team_removal_detaches_instance_through_wiring() { persona_ids: Some(vec![]), }, |_| Ok(()), + |_| Ok(()), || Ok(existing.clone()), |records| { *saved.borrow_mut() = Some(records.to_vec()); Ok(()) }, + || Ok(()), ) .expect("inbound removal succeeds"); @@ -671,11 +675,13 @@ fn inbound_team_omitted_roster_leaves_bindings_untouched() { TEAM_ID.to_string(), team_content_omitting_optional_fields("Renamed"), |_| Ok(()), + |_| Ok(()), || Ok(existing.clone()), |records| { *saved.borrow_mut() = Some(records.to_vec()); Ok(()) }, + || Ok(()), ) .expect("inbound metadata-only edit succeeds"); @@ -685,15 +691,24 @@ fn inbound_team_omitted_roster_leaves_bindings_untouched() { ); } -/// A failing agent-store write after the authoritative `save_teams` is -/// swallowed: the inbound reconcile still succeeds (boot repair is the retry), -/// so a secondary-store hiccup never aborts an inbound event whose team write -/// already landed. +/// An inbound add persists a recovery stage before the team write. A failed +/// agent write leaves that stage available to bind a shared persona on restart. +/// The error stops `commit_inbound_with_store`, so the retention head does not +/// advance until the binding becomes durable. #[test] -fn inbound_team_swallows_agent_store_failure() { - let mut teams = vec![local_team()]; +fn inbound_team_failure_keeps_a_durable_shared_persona_replay_delta() { + let pending_file = tempfile::NamedTempFile::new().expect("temporary stage"); + std::fs::write(pending_file.path(), "null").expect("initialize stage"); + let mut teams = vec![local_team(), { + let mut other = local_team(); + other.id = "other-team".to_string(); + other.persona_ids = vec!["p-added".to_string()]; + other + }]; teams[0].persona_ids = vec![]; - commit_inbound_team( + let agents = RefCell::new(vec![team_instance('a', "p-added", None)]); + + let error = commit_inbound_team( &mut teams, TEAM_ID.to_string(), TeamEventContent { @@ -702,11 +717,45 @@ fn inbound_team_swallows_agent_store_failure() { instructions: None, persona_ids: Some(vec!["p-added".to_string()]), }, + |pending| { + crate::commands::teams::save_pending_team_membership_at( + pending_file.path(), + Some(pending), + ) + }, |_| Ok(()), - || Err("agent store unreadable".to_string()), - |_| Ok(()), + || Ok(agents.borrow().clone()), + |_| Err("agent store unwritable".to_string()), + || crate::commands::teams::save_pending_team_membership_at(pending_file.path(), None), ) - .expect("inbound reconcile swallows secondary-store failure"); + .expect_err("inbound reconcile must report a lost membership binding"); + assert!( + error.contains("could not update inbound team agents"), + "{error}" + ); + assert_eq!( + teams[0].persona_ids, + vec!["p-added"], + "the team write landed" + ); + + let pending = crate::commands::teams::load_pending_team_membership_at(pending_file.path()) + .expect("read staged delta") + .expect("the failed inbound add keeps its stage"); + crate::commands::teams::propagate_membership( + &pending.team_id, + &pending.previous_persona_ids, + &pending.current_persona_ids, + || Ok(agents.borrow().clone()), + |records| { + *agents.borrow_mut() = records.to_vec(); + Ok(()) + }, + ) + .expect("a restart replay binds the explicit inbound add"); + crate::commands::teams::save_pending_team_membership_at(pending_file.path(), None) + .expect("clear replayed stage"); + assert_eq!(agents.borrow()[0].team_id.as_deref(), Some(TEAM_ID)); } /// A `persist_teams` error propagates — the authoritative team write failing is @@ -718,9 +767,11 @@ fn inbound_team_propagates_persist_teams_error() { &mut teams, TEAM_ID.to_string(), team_content("Team"), + |_| Ok(()), |_| Err("disk full".to_string()), || Ok(vec![]), |_| Ok(()), + || Ok(()), ) .expect_err("a failed team persist must propagate"); assert_eq!(err, "disk full"); diff --git a/desktop/src-tauri/src/commands/teams/mod.rs b/desktop/src-tauri/src/commands/teams/mod.rs index 5d50a2fa32..36192bcece 100644 --- a/desktop/src-tauri/src/commands/teams/mod.rs +++ b/desktop/src-tauri/src/commands/teams/mod.rs @@ -29,20 +29,20 @@ fn trim_optional(value: Option) -> Option { }) } -/// A staged team update. The record persists before the two stores change so a -/// later save or launch can replay the original membership delta. +/// A staged team membership change. The record persists before the team and +/// agent stores change so a later save or launch can replay the original delta. #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -struct PendingTeamMembershipUpdate { - team_id: String, - previous_persona_ids: Vec, - current_persona_ids: Vec, +pub(in crate::commands) struct PendingTeamMembershipUpdate { + pub(in crate::commands) team_id: String, + pub(in crate::commands) previous_persona_ids: Vec, + pub(in crate::commands) current_persona_ids: Vec, } -fn pending_team_membership_path(app: &AppHandle) -> Result { +fn pending_team_membership_path(app: &AppHandle) -> Result { Ok(crate::managed_agents::managed_agents_base_dir(app)?.join("pending-team-membership.json")) } -fn save_pending_team_membership_at( +pub(in crate::commands) fn save_pending_team_membership_at( path: &std::path::Path, pending: Option<&PendingTeamMembershipUpdate>, ) -> Result<(), String> { @@ -51,7 +51,7 @@ fn save_pending_team_membership_at( crate::managed_agents::storage::atomic_write_json(path, &payload) } -fn load_pending_team_membership_at( +pub(in crate::commands) fn load_pending_team_membership_at( path: &std::path::Path, ) -> Result, String> { if !path.exists() { @@ -63,20 +63,22 @@ fn load_pending_team_membership_at( .map_err(|error| format!("failed to parse pending team update: {error}")) } -fn save_pending_team_membership( - app: &AppHandle, +pub(in crate::commands) fn save_pending_team_membership( + app: &AppHandle, pending: &PendingTeamMembershipUpdate, ) -> Result<(), String> { save_pending_team_membership_at(&pending_team_membership_path(app)?, Some(pending)) } -fn load_pending_team_membership( - app: &AppHandle, +fn load_pending_team_membership( + app: &AppHandle, ) -> Result, String> { load_pending_team_membership_at(&pending_team_membership_path(app)?) } -fn clear_pending_team_membership(app: &AppHandle) -> Result<(), String> { +pub(in crate::commands) fn clear_pending_team_membership( + app: &AppHandle, +) -> Result<(), String> { // Write `null` through the link instead of unlinking it. The pending file // is shared by dev worktrees, and `atomic_write_json` preserves the link. save_pending_team_membership_at(&pending_team_membership_path(app)?, None) @@ -112,7 +114,9 @@ fn pending_replay_delta( } /// Replay a staged membership delta. Callers hold `managed_agents_store_lock`. -pub(crate) fn replay_pending_team_membership(app: &AppHandle) -> Result<(), String> { +pub(crate) fn replay_pending_team_membership( + app: &AppHandle, +) -> Result<(), String> { let Some(pending) = load_pending_team_membership(app)? else { return Ok(()); }; @@ -234,25 +238,9 @@ fn propagate_membership_with_roster( Ok(()) } -/// Propagate a team's membership *change* to its members' already-running -/// instances, best-effort: any load/save error is logged and swallowed. -/// -/// Used by [`commit_team_create`] and by the inbound reconcile path. For a -/// create, the team write already landed, and failing the command would make a -/// UI retry mint a duplicate team; a missing backfill only costs a member the -/// team instructions until boot repair runs, and it blocks nothing. -/// -/// [`commit_team_update`] does **not** use this policy. A removal that does not -/// reach the agent store leaves an agent bound to the team, and the delete guard -/// then refuses the team — so an update reports the failure instead. -/// -/// `load_agents`/`save_agents` are injected so the command wiring (prior-roster -/// capture, delta direction, and this best-effort policy) is unit-testable -/// without an `AppHandle`; the commands pass the real store IO. -/// -/// Shared with the inbound reconcile path (`commands::personas::inbound`): a -/// 30176 team edit arriving from another device must bind/detach instances the -/// same way a local edit does, so both call this one wrapper. +/// Apply a membership change without failing a metadata-only producer. The +/// caller must use [`propagate_membership`] when a roster change needs durable +/// recovery. pub(in crate::commands) fn propagate_membership_best_effort( team_id: &str, previous_persona_ids: &[String], @@ -260,32 +248,45 @@ pub(in crate::commands) fn propagate_membership_best_effort( load_agents: impl FnOnce() -> Result, String>, save_agents: impl FnOnce(&[crate::managed_agents::ManagedAgentRecord]) -> Result<(), String>, ) { - if let Err(e) = propagate_membership( + if let Err(error) = propagate_membership( team_id, previous_persona_ids, current_persona_ids, load_agents, save_agents, ) { - eprintln!("buzz-desktop: team-membership-propagate: {e}"); + eprintln!("buzz-desktop: team-membership-propagate: {error}"); } } -/// In-memory core of [`create_team`]: push the built team, persist teams -/// authoritatively, then propagate its whole roster (no prior members ⇒ the -/// whole roster is the added delta) to live instances best-effort. Decoupled -/// from the `AppHandle` shell via injected persistence so the create wiring is -/// unit-testable. A `persist_teams` error propagates; agent IO is best-effort. +/// In-memory core of [`create_team`]: stage, persist, and bind the new team's +/// full roster to already-running instances. The function clears the stage +/// only after both stores succeed. A failure returns an error because a success +/// response must mean the membership binding is durable. fn commit_team_create( teams: &mut Vec, team: TeamRecord, + save_pending: impl FnOnce(&PendingTeamMembershipUpdate) -> Result<(), String>, persist_teams: impl FnOnce(&[TeamRecord]) -> Result<(), String>, load_agents: impl FnOnce() -> Result, String>, save_agents: impl FnOnce(&[crate::managed_agents::ManagedAgentRecord]) -> Result<(), String>, + clear_pending: impl FnOnce() -> Result<(), String>, ) -> Result { + let membership_changed = !team.persona_ids.is_empty(); + if membership_changed { + save_pending(&PendingTeamMembershipUpdate { + team_id: team.id.clone(), + previous_persona_ids: Vec::new(), + current_persona_ids: team.persona_ids.clone(), + })?; + } teams.push(team.clone()); persist_teams(teams)?; - propagate_membership_best_effort(&team.id, &[], &team.persona_ids, load_agents, save_agents); + if membership_changed { + propagate_membership(&team.id, &[], &team.persona_ids, load_agents, save_agents) + .map_err(|error| format!("could not update the new team's agents: {error}"))?; + clear_pending()?; + } Ok(team) } @@ -302,8 +303,8 @@ fn commit_team_create( /// agent write leaves the stage file in place. The next update or launch replays /// that original delta before it accepts another team edit. /// -/// A create has no stable id, so it keeps a best-effort policy to avoid a -/// duplicate team on retry. +/// A create also stages its full roster before it writes the team. A retry +/// replays that stage, so the command never reports a lost member binding. #[allow(clippy::too_many_arguments)] fn commit_team_update( teams: &mut [TeamRecord], @@ -677,11 +678,14 @@ pub async fn create_team(input: CreateTeamRequest, app: AppHandle) -> Result = Vec::new(); - let created = commit_team_create( + let error = commit_team_create( &mut teams, team("team-a", &["duncan"]), |_| Ok(()), + |_| Ok(()), || Ok(vec![instance('a', "duncan", None)]), |_| Err("disk full".to_string()), + || Ok(()), ) - .expect("create swallows secondary-store failure"); - assert_eq!(created.id, "team-a"); + .expect_err("create reports a lost membership binding"); + + assert!(error.contains("could not update the new team's agents")); + assert_eq!( + teams.len(), + 1, + "the team write occurs before the agent write" + ); } } diff --git a/desktop/src-tauri/src/commands/teams/tests/membership_recovery.rs b/desktop/src-tauri/src/commands/teams/tests/membership_recovery.rs index 4860969f05..5477a4515f 100644 --- a/desktop/src-tauri/src/commands/teams/tests/membership_recovery.rs +++ b/desktop/src-tauri/src/commands/teams/tests/membership_recovery.rs @@ -313,11 +313,13 @@ fn commit_team_create_treats_full_roster_as_added() { &mut teams, team("team-a", &["duncan"]), |_| Ok(()), + |_| Ok(()), || Ok(existing.clone()), |records| { spy.borrow_mut().saved = Some(records.to_vec()); Ok(()) }, + || Ok(()), ) .expect("create succeeds"); @@ -330,21 +332,52 @@ fn commit_team_create_treats_full_roster_as_added() { ); } -/// A failing secondary agent write after successful `save_teams` is -/// swallowed by `create`: it still returns the persisted team. Otherwise a UI -/// retry of a create whose team already landed would mint a duplicate. +/// A failed create keeps its staged delta. The retry path can bind a persona +/// even when another team also contains that persona. #[test] -fn commit_create_returns_ok_when_agent_save_fails() { - let mut teams: Vec = Vec::new(); - let created = commit_team_create( +fn failed_create_keeps_a_durable_shared_persona_replay_delta() { + let pending_file = tempfile::NamedTempFile::new().expect("temporary stage"); + std::fs::write(pending_file.path(), "null").expect("initialize stage"); + let mut teams = vec![team("team-a", &["duncan"])]; + let agents = RefCell::new(vec![instance('a', "duncan", None)]); + + let error = commit_team_create( &mut teams, - team("team-a", &["duncan"]), + team("team-b", &["duncan"]), + |pending| save_pending_team_membership_at(pending_file.path(), Some(pending)), |_| Ok(()), - || Ok(vec![instance('a', "duncan", None)]), + || Ok(agents.borrow().clone()), |_| Err("disk full".to_string()), + || save_pending_team_membership_at(pending_file.path(), None), ) - .expect("create swallows secondary-store failure"); - assert_eq!(created.id, "team-a"); + .expect_err("a create must report an undurable member binding"); + assert!( + error.contains("could not update the new team's agents"), + "{error}" + ); + assert_eq!( + teams.len(), + 2, + "the team write landed before the failed binding" + ); + + let pending = load_pending_team_membership_at(pending_file.path()) + .expect("read staged delta") + .expect("the failed create keeps its stage"); + assert_eq!(pending.team_id, "team-b"); + propagate_membership( + &pending.team_id, + &pending.previous_persona_ids, + &pending.current_persona_ids, + || Ok(agents.borrow().clone()), + |records| { + *agents.borrow_mut() = records.to_vec(); + Ok(()) + }, + ) + .expect("a launch replay binds the shared persona"); + save_pending_team_membership_at(pending_file.path(), None).expect("clear replayed stage"); + assert_eq!(agents.borrow()[0].team_id.as_deref(), Some("team-b")); } /// `update` must NOT swallow an agent-store failure while emptying a roster. @@ -689,11 +722,13 @@ fn replay_before_create_preserves_the_new_team_binding() { &mut teams, team("team-b", &["duncan"]), |_| Ok(()), + |_| Ok(()), || Ok(agents.borrow().clone()), |records| { *agents.borrow_mut() = records.to_vec(); Ok(()) }, + || Ok(()), ) .expect("the new team binds the now-unbound persona");