diff --git a/desktop/src-tauri/src/commands/mod.rs b/desktop/src-tauri/src/commands/mod.rs index 7cb2d8e3b8..37d5dc5a19 100644 --- a/desktop/src-tauri/src/commands/mod.rs +++ b/desktop/src-tauri/src/commands/mod.rs @@ -120,6 +120,7 @@ pub use relay_members::*; pub use relay_reconnect::*; pub use social::*; pub use team_snapshot::*; +pub(crate) use teams::replay_pending_team_membership; pub use teams::*; pub use updater::*; pub use window_chrome::*; diff --git a/desktop/src-tauri/src/commands/teams.rs b/desktop/src-tauri/src/commands/teams.rs index dd52e934a4..fdf61fd53f 100644 --- a/desktop/src-tauri/src/commands/teams.rs +++ b/desktop/src-tauri/src/commands/teams.rs @@ -1,3 +1,6 @@ +use std::path::PathBuf; + +use serde::{Deserialize, Serialize}; use tauri::AppHandle; use uuid::Uuid; @@ -26,6 +29,98 @@ 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. +#[derive(Debug, Clone, Serialize, Deserialize)] +struct PendingTeamMembershipUpdate { + team_id: String, + previous_persona_ids: Vec, + current_persona_ids: Vec, +} + +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( + app: &AppHandle, + pending: &PendingTeamMembershipUpdate, +) -> Result<(), String> { + let path = pending_team_membership_path(app)?; + let payload = serde_json::to_vec_pretty(pending) + .map_err(|error| format!("failed to serialize pending team update: {error}"))?; + crate::managed_agents::storage::atomic_write_json(&path, &payload) +} + +fn load_pending_team_membership( + app: &AppHandle, +) -> Result, String> { + let path = pending_team_membership_path(app)?; + if !path.exists() { + return Ok(None); + } + let payload = std::fs::read_to_string(&path) + .map_err(|error| format!("failed to read pending team update: {error}"))?; + serde_json::from_str(&payload) + .map(Some) + .map_err(|error| format!("failed to parse pending team update: {error}")) +} + +fn clear_pending_team_membership(app: &AppHandle) -> Result<(), String> { + let path = pending_team_membership_path(app)?; + match std::fs::remove_file(&path) { + Ok(()) => Ok(()), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()), + Err(error) => Err(format!("failed to clear pending team update: {error}")), + } +} + +enum PendingTeamMembershipState { + Pending, + Superseded, +} + +fn pending_team_membership_state( + pending: &PendingTeamMembershipUpdate, + teams: &[TeamRecord], +) -> Result { + let team = teams + .iter() + .find(|team| team.id == pending.team_id) + .ok_or_else(|| format!("pending team {} no longer exists", pending.team_id))?; + if team.persona_ids == pending.current_persona_ids { + Ok(PendingTeamMembershipState::Pending) + } else if team.persona_ids == pending.previous_persona_ids { + Ok(PendingTeamMembershipState::Superseded) + } else { + Err(format!( + "pending team {} has an unexpected roster; save the team again", + pending.team_id + )) + } +} + +/// Replay a staged membership delta. Callers hold `managed_agents_store_lock`. +pub(crate) fn replay_pending_team_membership(app: &AppHandle) -> Result<(), String> { + let Some(pending) = load_pending_team_membership(app)? else { + return Ok(()); + }; + match pending_team_membership_state(&pending, &load_teams(app)?)? { + PendingTeamMembershipState::Pending => { + propagate_membership( + &pending.team_id, + &pending.previous_persona_ids, + &pending.current_persona_ids, + || load_managed_agents(app), + |records| save_managed_agents(app, records), + ) + .map_err(|error| format!("could not replay pending team update: {error}"))?; + } + PendingTeamMembershipState::Superseded => {} + } + clear_pending_team_membership(app) +} + /// Clear `team_id` on every instance that is bound to `team_id` but whose /// persona is absent from `current_persona_ids` or unset. Reports whether /// anything changed. @@ -60,24 +155,17 @@ fn detach_agents_outside_roster( changed } -/// Reports a membership propagation failure. The update command reports a -/// failure for a roster removal or a failed stale-binding detach. Other changes -/// keep the best-effort policy. +/// Reports a membership propagation failure. +#[derive(Debug)] pub(in crate::commands) enum MembershipPropagationError { Load(String), - Save { error: String, detached: bool }, -} - -impl MembershipPropagationError { - fn must_report(&self) -> bool { - matches!(self, Self::Save { detached: true, .. }) - } + Save(String), } impl std::fmt::Display for MembershipPropagationError { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::Load(error) | Self::Save { error, .. } => formatter.write_str(error), + Self::Load(error) | Self::Save(error) => formatter.write_str(error), } } } @@ -103,8 +191,7 @@ pub(in crate::commands) fn propagate_membership( ); let detached = detach_agents_outside_roster(&mut records, team_id, current_persona_ids); if delta_changed || detached { - save_agents(&records) - .map_err(|error| MembershipPropagationError::Save { error, detached })?; + save_agents(&records).map_err(MembershipPropagationError::Save)?; } Ok(()) } @@ -172,21 +259,13 @@ fn commit_team_create( /// keeps it `AppHandle`-free; a `persist_teams` error propagates. Returns the /// updated team. /// -/// Unlike [`commit_team_create`], an update reports an agent-store failure for -/// a roster removal or a failed stale-binding detach. A removal clears `team_id` -/// on the removed member. If that write does not land, the agent keeps the -/// binding and `delete_team_with_cascade` refuses the team, so the user gets an -/// empty team that they cannot delete — the exact defect this command must not -/// create. The command must not report success while the delete guard can still -/// refuse. +/// Keeps the team roster and instance bindings recoverable across two stores. +/// It stages the prior→current delta before it writes either store. A failed +/// agent write leaves the stage file in place. The next update or launch replays +/// that original delta before it accepts another team edit. /// -/// When this reports an agent-store error after the team write, `update_team` -/// does not retain the team for relay sync until a retry or the next launch -/// migration runs. -/// -/// Reporting is safe here because an update is idempotent: it targets an -/// existing team by id, so a retry cannot mint a duplicate team. A create has no -/// id yet, which is why it keeps the best-effort policy. +/// A create has no stable id, so it keeps a best-effort policy to avoid a +/// duplicate team on retry. #[allow(clippy::too_many_arguments)] fn commit_team_update( teams: &mut [TeamRecord], @@ -215,10 +294,8 @@ fn commit_team_update( team.updated_at = now; let updated = team.clone(); + let membership_changed = previous_persona_ids != updated.persona_ids; persist_teams(teams)?; - let has_removal = previous_persona_ids - .iter() - .any(|persona_id| !updated.persona_ids.contains(persona_id)); if let Err(error) = propagate_membership( &updated.id, &previous_persona_ids, @@ -226,10 +303,9 @@ fn commit_team_update( load_agents, save_agents, ) { - if has_removal || error.must_report() { + if membership_changed || matches!(error, MembershipPropagationError::Save(_)) { return Err(format!( - "Saved the team, but could not update its agents: {error}. The team can refuse deletion \ - until this succeeds. Save the team again." + "Saved the team, but could not update its agents: {error}. Save the team again." )); } eprintln!("buzz-desktop: team-membership-propagate: {error}"); @@ -475,9 +551,27 @@ pub async fn update_team(input: UpdateTeamRequest, app: AppHandle) -> Result Result Result<(), String> { .managed_agents_store_lock .lock() .map_err(|error| error.to_string())?; + replay_pending_team_membership(&app)?; let cascaded_persona_d_tags = delete_team_with_cascade(&app, &id)?; // delete_team_with_cascade rejects built-in teams via validate_team_deletion, // so reaching here means this team was owner-published — tombstone it. The diff --git a/desktop/src-tauri/src/commands/teams_tests.rs b/desktop/src-tauri/src/commands/teams_tests.rs index f6a50a9988..793a25a0ae 100644 --- a/desktop/src-tauri/src/commands/teams_tests.rs +++ b/desktop/src-tauri/src/commands/teams_tests.rs @@ -4,7 +4,8 @@ use super::{ apply_team_membership_delta, commit_team_create, commit_team_update, - detach_agents_outside_roster, + detach_agents_outside_roster, pending_team_membership_state, propagate_membership, + PendingTeamMembershipState, PendingTeamMembershipUpdate, }; use crate::managed_agents::{ManagedAgentRecord, TeamRecord}; use std::cell::RefCell; @@ -200,13 +201,13 @@ fn commit_update_ignores_agent_load_failure_for_metadata_only_edit() { assert_eq!(teams[0].name, "Renamed team"); } -/// An add-only update also keeps its disk-authoritative result when the -/// agent store cannot load. A missed bind does not block team deletion. +/// An add-only update reports an agent-store load failure. The staged delta +/// remains available for replay on the next save or launch. #[test] -fn commit_update_ignores_agent_load_failure_for_add_only_edit() { +fn commit_update_reports_agent_load_failure_for_add_only_edit() { let mut teams = vec![team("team-a", &["duncan"])]; - let updated = commit_team_update( + let error = commit_team_update( &mut teams, "team-a", "team-a".to_string(), @@ -218,9 +219,9 @@ fn commit_update_ignores_agent_load_failure_for_add_only_edit() { || Err("corrupt managed-agents.json".to_string()), |_| Ok(()), ) - .expect("add-only update keeps the best-effort policy"); + .expect_err("an add-only update must report the lost binding"); - assert_eq!(updated.persona_ids, ids(&["duncan", "ada"])); + assert!(error.contains("could not update its agents"), "{error}"); assert_eq!(teams[0].persona_ids, ids(&["duncan", "ada"])); } @@ -552,6 +553,65 @@ fn detach_outside_roster_is_scoped_to_this_team_and_absent_personas() { ); } +/// A staged update only replays when the team still has the staged roster. +/// A prior roster means the team write did not land, so the stale stage is safe +/// to clear without a membership change. +#[test] +fn pending_membership_state_distinguishes_pending_and_superseded_writes() { + let pending = PendingTeamMembershipUpdate { + team_id: "team-a".to_string(), + previous_persona_ids: ids(&["duncan"]), + current_persona_ids: ids(&["ada"]), + }; + + assert!(matches!( + pending_team_membership_state(&pending, &[team("team-a", &["ada"])]), + Ok(PendingTeamMembershipState::Pending) + )); + assert!(matches!( + pending_team_membership_state(&pending, &[team("team-a", &["duncan"])]), + Ok(PendingTeamMembershipState::Superseded) + )); +} + +/// The durable replay uses the original replace delta after the first agent +/// save fails. It binds Ada on retry even though the persisted roster already +/// contains Ada and therefore supplies no new delta. +#[test] +fn replayed_replace_delta_binds_the_added_instance() { + let previous = ids(&["duncan"]); + let current = ids(&["ada"]); + let store = RefCell::new(vec![ + instance('a', "duncan", Some("team-a")), + instance('b', "ada", None), + ]); + + let error = propagate_membership( + "team-a", + &previous, + ¤t, + || Ok(store.borrow().clone()), + |_| Err("disk full".to_string()), + ) + .expect_err("the first save fails"); + assert!(error.to_string().contains("disk full")); + + propagate_membership( + "team-a", + &previous, + ¤t, + || Ok(store.borrow().clone()), + |records| { + *store.borrow_mut() = records.to_vec(); + Ok(()) + }, + ) + .expect("the durable replay succeeds"); + + assert_eq!(store.borrow()[0].team_id, None); + assert_eq!(store.borrow()[1].team_id.as_deref(), Some("team-a")); +} + /// A roster with every binding already correct writes nothing. #[test] fn detach_outside_roster_is_inert_when_nothing_is_stale() { diff --git a/desktop/src-tauri/src/migration.rs b/desktop/src-tauri/src/migration.rs index 1e22d7aaec..e7e384972a 100644 --- a/desktop/src-tauri/src/migration.rs +++ b/desktop/src-tauri/src/migration.rs @@ -31,6 +31,7 @@ const LEGACY_RELEASE_IDENTIFIER: &str = "xyz.block.sprout.app"; /// receive their identity via the `BUZZ_PRIVATE_KEY` env var. const SHARED_AGENT_FILES: &[&str] = &[ "agents/managed-agents.json", + "agents/pending-team-membership.json", "agents/personas.json", "agents/teams.json", ]; @@ -186,6 +187,7 @@ fn run_boot_migrations_inner(app: &tauri::AppHandle, reset_completed: bool) { // Repair dropped team↔member links, then detach directory-backed teams, // gated on a clean repair so a failure preserves `source_dir` for a retry. team_membership::repair_then_detach_teams(app); + replay_pending_team_membership_update(app); reconcile_provider_mcp_commands(app); reconcile_databricks_v1_to_v2(app); materialize_agent_runtimes(app); @@ -1367,6 +1369,18 @@ pub fn migrate_persona_provider_to_runtime(app: &tauri::AppHandle) { } mod materialize; pub use materialize::materialize_agent_runtimes; + +fn replay_pending_team_membership_update(app: &tauri::AppHandle) { + let state = app.state::(); + let Ok(_store_guard) = state.managed_agents_store_lock.lock() else { + eprintln!("buzz-desktop: pending-team-membership: cannot lock agent store"); + return; + }; + if let Err(error) = crate::commands::replay_pending_team_membership(app) { + eprintln!("buzz-desktop: pending-team-membership: {error}"); + } +} + mod fold; pub use fold::fold_personas_into_agent_store; use fold::load_persona_runtimes; diff --git a/desktop/src-tauri/src/migration_tests.rs b/desktop/src-tauri/src/migration_tests.rs index 0d49bd02aa..9b98d7a40f 100644 --- a/desktop/src-tauri/src/migration_tests.rs +++ b/desktop/src-tauri/src/migration_tests.rs @@ -84,6 +84,11 @@ fn setup_sync_layout() -> (tempfile::TempDir, PathBuf, PathBuf) { ) .unwrap(); std::fs::write(canonical.join("agents/teams.json"), r#"[{"id":"team-1"}]"#).unwrap(); + std::fs::write( + canonical.join("agents/pending-team-membership.json"), + r#"{"team_id":"team-1","previous_persona_ids":[],"current_persona_ids":[]}"#, + ) + .unwrap(); // Teams installed from `.main` — canonical has no teams dir. let team_dir = main_instance.join("agents/teams/com.example.test-pack"); @@ -216,7 +221,7 @@ fn sync_files(canonical: &Path, worktree: &Path) -> u32 { fn sync_creates_symlinks_to_fresh_worktree() { let (_parent, canonical, worktree) = setup_sync_layout(); let synced = sync_files(&canonical, &worktree); - assert_eq!(synced, 4); + assert_eq!(synced, 5); for rel in SHARED_AGENT_FILES { let dst = worktree.join(rel); assert!(dst.is_symlink(), "{rel} should be a symlink"); @@ -244,7 +249,7 @@ fn sync_replaces_existing_files_with_symlinks() { let synced = sync_files(&canonical, &worktree); - assert_eq!(synced, 4); + assert_eq!(synced, 5); for rel in SHARED_AGENT_FILES { let dst = worktree.join(rel); assert!( @@ -263,7 +268,7 @@ fn sync_replaces_existing_files_with_symlinks() { #[test] fn sync_preserves_correct_symlinks() { let (_parent, canonical, worktree) = setup_sync_layout(); - assert_eq!(sync_files(&canonical, &worktree), 4); + assert_eq!(sync_files(&canonical, &worktree), 5); assert_eq!(sync_files(&canonical, &worktree), 0); for rel in SHARED_AGENT_FILES { let dst = worktree.join(rel); @@ -282,7 +287,7 @@ fn sync_replaces_wrong_symlinks() { std::os::unix::fs::symlink(&wrong_target, worktree.join(rel)).unwrap(); } let synced = sync_files(&canonical, &worktree); - assert_eq!(synced, 4); + assert_eq!(synced, 5); for rel in SHARED_AGENT_FILES { assert_eq!( std::fs::read_link(worktree.join(rel)).unwrap(), @@ -301,7 +306,7 @@ fn sync_handles_broken_symlinks() { std::os::unix::fs::symlink(&broken_target, worktree.join(rel)).unwrap(); } let synced = sync_files(&canonical, &worktree); - assert_eq!(synced, 4); + assert_eq!(synced, 5); for rel in SHARED_AGENT_FILES { let dst = worktree.join(rel); assert!(dst.is_symlink());