fix(desktop): replay team membership stages safely

Signed-off-by: Storme Drone <49c46e84758b2ebff4abf5abbbd44ee4ce788fc3b55db9fa703eec124eead621@buzz.block.builderlab.xyz>
2026-10-29 13:31:44 +00:00 · 2026-08-27 17:27:59 -06:00
parent 38cb69aeb7
commit 36e2f452c3
2 changed files with 213 additions and 83 deletions
--- a/desktop/src-tauri/src/commands/teams.rs
+++ b/desktop/src-tauri/src/commands/teams.rs
@@ -31,7 +31,7 @@ fn trim_optional(value: Option<String>) -> Option<String> {

 /// 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)]
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 struct PendingTeamMembershipUpdate {
    team_id: String,
    previous_persona_ids: Vec<String>,
@@ -42,60 +42,73 @@ fn pending_team_membership_path(app: &AppHandle) -> Result<PathBuf, String> {
    Ok(crate::managed_agents::managed_agents_base_dir(app)?.join("pending-team-membership.json"))
 }

+fn save_pending_team_membership_at(
+    path: &std::path::Path,
+    pending: Option<&PendingTeamMembershipUpdate>,
+) -> Result<(), String> {
+    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_at(
+    path: &std::path::Path,
+) -> Result<Option<PendingTeamMembershipUpdate>, String> {
+    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_err(|error| format!("failed to parse pending team update: {error}"))
+}
+
 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)
+    save_pending_team_membership_at(&pending_team_membership_path(app)?, Some(pending))
 }

 fn load_pending_team_membership(
    app: &AppHandle,
 ) -> Result<Option<PendingTeamMembershipUpdate>, 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}"))
+    load_pending_team_membership_at(&pending_team_membership_path(app)?)
 }

 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}")),
-    }
+    // 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)
 }

-enum PendingTeamMembershipState {
-    Pending,
-    Superseded,
-    MissingTeam,
-    UnexpectedRoster,
-}
-
-fn pending_team_membership_state(
+/// The part of a staged delta that still agrees with the current team roster.
+///
+/// An inbound event can extend or reorder the roster while a local agent-store
+/// write is pending. It does not erase the local add or removal evidence that
+/// still holds. An inbound reversal does erase that evidence, so the replay
+/// leaves that membership direction alone.
+fn pending_replay_delta(
    pending: &PendingTeamMembershipUpdate,
-    teams: &[TeamRecord],
-) -> Result<PendingTeamMembershipState, String> {
-    let Some(team) = teams.iter().find(|team| team.id == pending.team_id) else {
-        return Ok(PendingTeamMembershipState::MissingTeam);
-    };
-    if team.persona_ids == pending.current_persona_ids {
-        Ok(PendingTeamMembershipState::Pending)
-    } else if team.persona_ids == pending.previous_persona_ids {
-        Ok(PendingTeamMembershipState::Superseded)
-    } else {
-        Ok(PendingTeamMembershipState::UnexpectedRoster)
-    }
+    current_persona_ids: &[String],
+) -> (Vec<String>, Vec<String>) {
+    let removed = pending
+        .previous_persona_ids
+        .iter()
+        .filter(|id| {
+            !pending.current_persona_ids.contains(*id) && !current_persona_ids.contains(*id)
+        })
+        .cloned()
+        .collect();
+    let added = pending
+        .current_persona_ids
+        .iter()
+        .filter(|id| {
+            !pending.previous_persona_ids.contains(*id) && current_persona_ids.contains(*id)
+        })
+        .cloned()
+        .collect();
+    (removed, added)
 }

 /// Replay a staged membership delta. Callers hold `managed_agents_store_lock`.
@@ -103,27 +116,25 @@ pub(crate) fn replay_pending_team_membership(app: &AppHandle) -> Result<(), Stri
    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 => {}
-        PendingTeamMembershipState::MissingTeam => eprintln!(
+    let teams = load_teams(app)?;
+    let Some(team) = teams.iter().find(|team| team.id == pending.team_id) else {
+        eprintln!(
            "buzz-desktop: pending-team-membership: discarding staged update for missing team {:?}",
            pending.team_id
-        ),
-        PendingTeamMembershipState::UnexpectedRoster => eprintln!(
-            "buzz-desktop: pending-team-membership: discarding staged update for team {:?} with a different roster",
-            pending.team_id
-        ),
-    }
+        );
+        return clear_pending_team_membership(app);
+    };
+    let (previous_persona_ids, current_persona_ids) =
+        pending_replay_delta(&pending, &team.persona_ids);
+    propagate_membership_with_roster(
+        &pending.team_id,
+        &previous_persona_ids,
+        &current_persona_ids,
+        &team.persona_ids,
+        || load_managed_agents(app),
+        |records| save_managed_agents(app, records),
+    )
+    .map_err(|error| format!("could not replay pending team update: {error}"))?;
    clear_pending_team_membership(app)
 }

@@ -187,6 +198,27 @@ pub(in crate::commands) fn propagate_membership(
    current_persona_ids: &[String],
    load_agents: impl FnOnce() -> Result<Vec<crate::managed_agents::ManagedAgentRecord>, String>,
    save_agents: impl FnOnce(&[crate::managed_agents::ManagedAgentRecord]) -> Result<(), String>,
+) -> Result<(), MembershipPropagationError> {
+    propagate_membership_with_roster(
+        team_id,
+        previous_persona_ids,
+        current_persona_ids,
+        current_persona_ids,
+        load_agents,
+        save_agents,
+    )
+}
+
+/// Apply a membership delta, then reconcile bindings with the authoritative
+/// roster. Replay uses a smaller delta when an inbound edit has changed the
+/// roster after staging, while the reconciliation always uses that latest roster.
+fn propagate_membership_with_roster(
+    team_id: &str,
+    previous_persona_ids: &[String],
+    current_persona_ids: &[String],
+    authoritative_persona_ids: &[String],
+    load_agents: impl FnOnce() -> Result<Vec<crate::managed_agents::ManagedAgentRecord>, String>,
+    save_agents: impl FnOnce(&[crate::managed_agents::ManagedAgentRecord]) -> Result<(), String>,
 ) -> Result<(), MembershipPropagationError> {
    let mut records = load_agents().map_err(MembershipPropagationError::Load)?;
    let delta_changed = apply_team_membership_delta(
@@ -195,7 +227,7 @@ pub(in crate::commands) fn propagate_membership(
        previous_persona_ids,
        current_persona_ids,
    );
-    let detached = detach_agents_outside_roster(&mut records, team_id, current_persona_ids);
+    let detached = detach_agents_outside_roster(&mut records, team_id, authoritative_persona_ids);
    if delta_changed || detached {
        save_agents(&records).map_err(MembershipPropagationError::Save)?;
    }
@@ -512,6 +544,7 @@ pub async fn create_team(input: CreateTeamRequest, app: AppHandle) -> Result<Tea
            .managed_agents_store_lock
            .lock()
            .map_err(|error| error.to_string())?;
+        replay_pending_team_membership(&app)?;
        let personas = load_personas(&app)?;
        ensure_persona_ids_are_active(&personas, &input.persona_ids)?;
        let mut teams = load_teams(&app)?;
--- a/desktop/src-tauri/src/commands/teams_tests.rs
+++ b/desktop/src-tauri/src/commands/teams_tests.rs
@@ -4,8 +4,8 @@

 use super::{
    apply_team_membership_delta, commit_team_create, commit_team_update,
-    detach_agents_outside_roster, pending_team_membership_state, propagate_membership,
-    PendingTeamMembershipState, PendingTeamMembershipUpdate,
+    detach_agents_outside_roster, load_pending_team_membership_at, pending_replay_delta,
+    propagate_membership, save_pending_team_membership_at, PendingTeamMembershipUpdate,
 };
 use crate::managed_agents::{ManagedAgentRecord, TeamRecord};
 use std::cell::RefCell;
@@ -553,34 +553,59 @@ fn detach_outside_roster_is_scoped_to_this_team_and_absent_personas() {
    );
 }

-/// A staged update replays only when the team keeps the staged roster. The
-/// prior roster means the team write did not land. A missing team or a different
-/// roster means an inbound event superseded the stage. Each stale stage is safe
-/// to clear without a membership change.
+/// A replay keeps each staged direction whose evidence still matches the
+/// current roster. An inbound extension or reorder does not discard a local
+/// add. An inbound reversal does discard the obsolete direction.
 #[test]
-fn pending_membership_state_distinguishes_replay_and_stale_stages() {
+fn pending_replay_delta_merges_with_an_inbound_roster_change() {
+    let pending = PendingTeamMembershipUpdate {
+        team_id: "team-a".to_string(),
+        previous_persona_ids: ids(&["duncan"]),
+        current_persona_ids: ids(&["duncan", "ada"]),
+    };
+
+    assert_eq!(
+        pending_replay_delta(&pending, &ids(&["ada", "duncan", "paul"])),
+        (ids(&[]), ids(&["ada"])),
+        "a reorder and an inbound extension preserve the staged add"
+    );
+    assert_eq!(
+        pending_replay_delta(&pending, &ids(&["duncan"])),
+        (ids(&[]), ids(&[])),
+        "an inbound removal makes the staged add obsolete"
+    );
+}
+
+/// Writing an empty pending state must preserve the app-facing symlink. On a
+/// later launch the worktree sync can retain that link without reviving the old
+/// canonical stage.
+#[cfg(unix)]
+#[test]
+fn clearing_a_pending_stage_preserves_the_shared_symlink_on_relaunch() {
+    let directory = tempfile::tempdir().expect("temporary directory");
+    let canonical = directory.path().join("canonical.json");
+    let worktree = directory.path().join("worktree.json");
+    std::fs::write(&canonical, "null").expect("create canonical pending file");
+    std::os::unix::fs::symlink(&canonical, &worktree).expect("create shared link");
    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)
-    ));
-    assert!(matches!(
-        pending_team_membership_state(&pending, &[]),
-        Ok(PendingTeamMembershipState::MissingTeam)
-    ));
-    assert!(matches!(
-        pending_team_membership_state(&pending, &[team("team-a", &["paul"])]),
-        Ok(PendingTeamMembershipState::UnexpectedRoster)
-    ));
+    save_pending_team_membership_at(&worktree, Some(&pending)).expect("stage through link");
+    assert_eq!(
+        load_pending_team_membership_at(&canonical).expect("read staged canonical file"),
+        Some(pending.clone())
+    );
+    save_pending_team_membership_at(&worktree, None).expect("clear through link");
+
+    assert!(worktree.is_symlink(), "the worktree path stays a symlink");
+    assert_eq!(
+        load_pending_team_membership_at(&worktree).expect("read after relaunch"),
+        None,
+        "the canonical file keeps the cleared state"
+    );
 }

 /// The durable replay uses the original replace delta after the first agent
@@ -632,3 +657,75 @@ fn detach_outside_roster_is_inert_when_nothing_is_stale() {
    ));
    assert_eq!(records[0].team_id.as_deref(), Some("team-a"));
 }
+
+/// A failed removal must replay before a new team accepts the same persona.
+/// The replay clears the old binding, so the new team's explicit add becomes
+/// the final binding instead of an older stage clearing it after the create.
+#[test]
+fn replay_before_create_preserves_the_new_team_binding() {
+    let agents = RefCell::new(vec![instance('a', "duncan", Some("team-a"))]);
+
+    propagate_membership(
+        "team-a",
+        &ids(&["duncan"]),
+        &ids(&[]),
+        || Ok(agents.borrow().clone()),
+        |_| Err("disk full".to_string()),
+    )
+    .expect_err("the team-a removal remains staged after a failed save");
+
+    propagate_membership(
+        "team-a",
+        &ids(&["duncan"]),
+        &ids(&[]),
+        || Ok(agents.borrow().clone()),
+        |records| {
+            *agents.borrow_mut() = records.to_vec();
+            Ok(())
+        },
+    )
+    .expect("create replays the earlier removal before its validation");
+
+    let mut teams = Vec::new();
+    commit_team_create(
+        &mut teams,
+        team("team-b", &["duncan"]),
+        |_| Ok(()),
+        || Ok(agents.borrow().clone()),
+        |records| {
+            *agents.borrow_mut() = records.to_vec();
+            Ok(())
+        },
+    )
+    .expect("the new team binds the now-unbound persona");
+
+    assert_eq!(agents.borrow()[0].team_id.as_deref(), Some("team-b"));
+}
+
+/// A shared persona keeps the explicit local-add evidence after an inbound
+/// extension and reorder. The latest roster still drives stale-binding cleanup.
+#[test]
+fn replay_after_inbound_extension_binds_the_explicit_shared_add() {
+    let pending = PendingTeamMembershipUpdate {
+        team_id: "team-a".to_string(),
+        previous_persona_ids: ids(&[]),
+        current_persona_ids: ids(&["ada"]),
+    };
+    let current_roster = ids(&["paul", "ada"]);
+    let (previous, current) = pending_replay_delta(&pending, &current_roster);
+    let agents = RefCell::new(vec![instance('a', "ada", None)]);
+
+    propagate_membership(
+        &pending.team_id,
+        &previous,
+        &current,
+        || Ok(agents.borrow().clone()),
+        |records| {
+            *agents.borrow_mut() = records.to_vec();
+            Ok(())
+        },
+    )
+    .expect("the staged add binds the instance despite a shared persona");
+
+    assert_eq!(agents.borrow()[0].team_id.as_deref(), Some("team-a"));
+}