mirror of
https://github.com/block/buzz.git
synced 2026-10-29 21:41:33 +00:00
fix(desktop): replay failed team membership updates
Signed-off-by: Storme Drone <49c46e84758b2ebff4abf5abbbd44ee4ce788fc3b55db9fa703eec124eead621@buzz.block.builderlab.xyz>
This commit is contained in:
@@ -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::*;
|
||||
|
||||
@@ -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<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)]
|
||||
struct PendingTeamMembershipUpdate {
|
||||
team_id: String,
|
||||
previous_persona_ids: Vec<String>,
|
||||
current_persona_ids: Vec<String>,
|
||||
}
|
||||
|
||||
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(
|
||||
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<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}"))
|
||||
}
|
||||
|
||||
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<PendingTeamMembershipState, String> {
|
||||
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<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)?;
|
||||
let previous_persona_ids = teams
|
||||
.iter()
|
||||
.find(|team| team.id == input.id)
|
||||
.ok_or_else(|| format!("team {} not found", input.id))?
|
||||
.persona_ids
|
||||
.clone();
|
||||
let membership_changed = previous_persona_ids != input.persona_ids;
|
||||
if membership_changed {
|
||||
save_pending_team_membership(
|
||||
&app,
|
||||
&PendingTeamMembershipUpdate {
|
||||
team_id: input.id.clone(),
|
||||
previous_persona_ids,
|
||||
current_persona_ids: input.persona_ids.clone(),
|
||||
},
|
||||
)?;
|
||||
}
|
||||
let updated = commit_team_update(
|
||||
&mut teams,
|
||||
&input.id,
|
||||
@@ -490,6 +584,9 @@ pub async fn update_team(input: UpdateTeamRequest, app: AppHandle) -> Result<Tea
|
||||
|| load_managed_agents(&app),
|
||||
|records| save_managed_agents(&app, records),
|
||||
)?;
|
||||
if membership_changed {
|
||||
clear_pending_team_membership(&app)?;
|
||||
}
|
||||
// Built-in teams are not owner-authored — never publish them.
|
||||
if !updated.is_builtin {
|
||||
retain_team_pending(&app, &state, &updated);
|
||||
@@ -513,6 +610,7 @@ pub async fn delete_team(id: String, app: AppHandle) -> 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
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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::<crate::app_state::AppState>();
|
||||
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;
|
||||
|
||||
@@ -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());
|
||||
|
||||
Reference in New Issue
Block a user