jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-29 13:31:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make team membership bindings durable

Browse Source 
Signed-off-by: Storme Drone <49c46e84758b2ebff4abf5abbbd44ee4ce788fc3b55db9fa703eec124eead621@buzz.block.builderlab.xyz>
This commit is contained in:
Storme Drone
2026-08-31 17:43:42 -06:00
parent d006915346
commit 6c46f2350a
5 changed files with 214 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

67
desktop/src-tauri/src/commands/personas/inbound.rs
View File

@@ -219,6 +219,10 @@ fn reconcile_inbound_persona_event_blocking<R: tauri::Runtime>(
.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<R: tauri::Runtime>(
&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<TeamRecord>,
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<Vec<ManagedAgentRecord>, 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,
&current_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,
&current_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,
&current_persona_ids,
load_agents,
save_agents,
);
}
Ok(())
}

71
desktop/src-tauri/src/commands/personas/inbound/inbound_tests.rs
View File

@@ -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");

94
desktop/src-tauri/src/commands/teams/mod.rs
View File

@@ -29,20 +29,20 @@ 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.
/// 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<String>,
current_persona_ids: Vec<String>,
pub(in crate::commands) struct PendingTeamMembershipUpdate {
pub(in crate::commands) team_id: String,
pub(in crate::commands) previous_persona_ids: Vec<String>,
pub(in crate::commands) current_persona_ids: Vec<String>,
}
fn pending_team_membership_path(app: &AppHandle) -> Result<PathBuf, String> {
fn pending_team_membership_path<R: tauri::Runtime>(app: &AppHandle<R>) -> Result<PathBuf, String> {
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<Option<PendingTeamMembershipUpdate>, 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<R: tauri::Runtime>(
app: &AppHandle<R>,
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<R: tauri::Runtime>(
app: &AppHandle<R>,
) -> Result<Option<PendingTeamMembershipUpdate>, 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<R: tauri::Runtime>(
app: &AppHandle<R>,
) -> 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<R: tauri::Runtime>(
app: &AppHandle<R>,
) -> 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<Vec<crate::managed_agents::ManagedAgentRecord>, 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<TeamRecord>,
team: TeamRecord,
save_pending: impl FnOnce(&PendingTeamMembershipUpdate) -> Result<(), String>,
persist_teams: impl FnOnce(&[TeamRecord]) -> Result<(), String>,
load_agents: impl FnOnce() -> Result<Vec<crate::managed_agents::ManagedAgentRecord>, String>,
save_agents: impl FnOnce(&[crate::managed_agents::ManagedAgentRecord]) -> Result<(), String>,
clear_pending: impl FnOnce() -> Result<(), String>,
) -> Result<TeamRecord, String> {
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<Tea
let team = commit_team_create(
&mut teams,
team,
|pending| save_pending_team_membership(&app, pending),
|teams| save_teams(&app, teams),
|| load_managed_agents(&app),
|records| save_managed_agents(&app, records),
|| clear_pending_team_membership(&app),
)?;
// Created teams are always non-builtin; publish to the relay.
// Created teams are always non-builtin. Retain only after the agent
// binding is durable, so a remote device cannot consume lost intent.
retain_team_pending(&app, &state, &team);
Ok(team)
})

22
desktop/src-tauri/src/commands/teams/tests.rs
View File

@@ -374,11 +374,13 @@ mod membership_wiring {
&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");
@@ -391,20 +393,28 @@ mod membership_wiring {
);
}
/// A create keeps its persisted result when the secondary agent write
/// fails. A retry must not create a duplicate team.
/// A failed create reports the undurable binding after it saves the team.
/// The caller can replay the staged membership delta before another write.
#[test]
fn commit_create_returns_ok_when_agent_save_fails() {
fn commit_create_reports_agent_save_failure() {
let mut teams: Vec<TeamRecord> = 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"
);
}
}

55
desktop/src-tauri/src/commands/teams/tests/membership_recovery.rs
View File

@@ -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<TeamRecord> = 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");