diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 34f3a0713b0..0ea3368f11a 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -36,8 +36,8 @@ pub struct AppState { /// init and takes priority over env vars and compile-time defaults. pub relay_url_override: Mutex>, /// Set during backend setup when managed agents are eligible for launch restore. - /// `apply_workspace` consumes it after installing the workspace relay and - /// identity, so agents never start against the fallback relay. + /// The frontend consumes it after workspace apply and authoritative relay + /// backfill, so agents never start against fallback or stale configuration. pub managed_agent_restore_pending: AtomicBool, /// Whether desktop may repair managed-agent kind:0 profiles from local records. /// Disabled by the experiment so agent profile updates survive start/restore. diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs index 15516f4ca0a..dbdc8c9ee53 100644 --- a/desktop/src-tauri/src/commands/workspace.rs +++ b/desktop/src-tauri/src/commands/workspace.rs @@ -164,6 +164,14 @@ pub async fn apply_workspace( }; // ── Apply all state changes (nothing below can fail) ────────────────── + // Serialize workspace mutation with launch-time restore. If restore + // already crossed into spawning, this switch waits for its children to + // be tracked. If the switch wins, restore's post-lock scope check stops + // the stale launch. + let restore_transition = state + .managed_agent_runtime_transition + .lock() + .map_err(|e| e.to_string())?; // Serialize the scope transition with inbound private-config handling. // Inbound holds this lock from scope resolution through overlay insert, // so a patch decrypted for the old workspace cannot land after this clear. @@ -196,6 +204,7 @@ pub async fn apply_workspace( state .managed_agent_profile_reconcile_enabled .store(!agent_managed_profiles.unwrap_or(false), Ordering::Release); + drop(restore_transition); // ── Filesystem side-effect (non-fatal) ──────────────────────────────── // Persist the *effective* repos_dir (None when the candidate failed @@ -230,37 +239,39 @@ pub async fn apply_workspace( // Backfill this exact relay+owner scope only after the workspace has been // applied. Running at process boot would target the fallback relay and // collapse every community into one pending-event store. - match crate::managed_agents::retention::active_retention_scope(&restore_app, &state) { - Ok(scope) => { - // Adopt whatever the pre-scoping release left queued in the global - // retention database BEFORE the scoped reconcile and flush run, so - // stranded tombstones and archive requests publish on this boot - // instead of being abandoned by the storage cutover. - migrate_legacy_retention_into(&restore_app, &scope); - crate::event_sync::spawn_event_sync( - restore_app.clone(), - scope.owner_keys, - scope.db_path, - ) - } - Err(error) => { - eprintln!("buzz-desktop: scoped event-sync unavailable after workspace apply: {error}"); - } - } - - let restore_pending = state - .managed_agent_restore_pending - .swap(false, Ordering::AcqRel); + let event_sync_task = + match crate::managed_agents::retention::active_retention_scope(&restore_app, &state) { + Ok(scope) => { + // Adopt whatever the pre-scoping release left queued in the global + // retention database BEFORE the scoped reconcile and flush run, so + // stranded tombstones and archive requests publish on this boot + // instead of being abandoned by the storage cutover. + migrate_legacy_retention_into(&restore_app, &scope); + let owner_pubkey = scope.owner_keys.public_key().to_hex(); + let db_path = scope.db_path.clone(); + let task = crate::event_sync::spawn_event_sync( + restore_app.clone(), + scope.owner_keys, + scope.db_path, + ); + Some((owner_pubkey, db_path, task)) + } + Err(error) => { + eprintln!( + "buzz-desktop: scoped event-sync unavailable after workspace apply: {error}" + ); + None + } + }; + crate::event_sync::replace_event_sync_task(event_sync_task)?; - // The coordinator starts before React applies the selected workspace, so - // its startup publication may have used the fallback relay and placeholder - // identity. Correct it off the command path so an unavailable relay cannot - // hold the frontend on its loading gate. On initial launch, restore MeshLLM - // first so a slow stopped-status request cannot overwrite a newly restored - // serving status, then restore managed agents after the admission identity - // has been published (or the bounded publication attempt has timed out). + // Managed-agent restore waits for the frontend's authoritative relay + // backfill. Starting it here would race both that backfill and the local + // event-sync task above. Mesh restore remains independent and can proceed + // while the agent configuration boundary is settling. #[cfg(feature = "mesh-llm")] { + let restore_pending = state.managed_agent_restore_pending.load(Ordering::Acquire); let app = restore_app.clone(); tauri::async_runtime::spawn(async move { let state = app.state::(); @@ -272,28 +283,84 @@ pub async fn apply_workspace( } } crate::mesh_llm::publish_current_status_once(&app, "workspace apply").await; - if restore_pending { - if let Err(error) = - restore_managed_agents_on_launch(&app, &state.shutdown_started).await - { - eprintln!("buzz-desktop: failed to restore managed agents: {error}"); - } - } }); } - #[cfg(not(feature = "mesh-llm"))] - if restore_pending { - let app = restore_app.clone(); - tauri::async_runtime::spawn(async move { - let state = app.state::(); - if let Err(error) = - restore_managed_agents_on_launch(&app, &state.shutdown_started).await - { - eprintln!("buzz-desktop: failed to restore managed agents: {error}"); - } - }); + Ok(()) +} + +/// Finish launch-time restore only after the selected community's private +/// agent backfill has been fully reconciled. +/// +/// The relay and owner are supplied by the subscription that completed. A +/// stale completion from a previous workspace is ignored before it can consume +/// the one-shot restore request. +#[tauri::command] +pub async fn complete_managed_agent_bootstrap( + owner_pubkey: String, + arrival_relay_url: String, + app: AppHandle, +) -> Result<(), String> { + let state = app.state::(); + let Some(scope) = crate::managed_agents::retention::arrival_retention_scope( + &app, + &state, + &arrival_relay_url, + )? + else { + return Ok(()); + }; + if !scope + .owner_keys + .public_key() + .to_hex() + .eq_ignore_ascii_case(owner_pubkey.trim()) + { + return Ok(()); } - Ok(()) + let event_sync_task = crate::event_sync::take_event_sync_task( + &scope.owner_keys.public_key().to_hex(), + &scope.db_path, + )?; + if let Some(task) = event_sync_task { + task.await + .map_err(|error| format!("managed-agent event sync failed: {error}"))?; + } + + let sync_app = app.clone(); + let sync_owner_keys = scope.owner_keys.clone(); + let sync_db_path = scope.db_path.clone(); + tauri::async_runtime::spawn_blocking(move || { + crate::event_sync::run_managed_agent_event_sync(&sync_app, &sync_owner_keys, &sync_db_path) + }) + .await + .map_err(|error| format!("managed-agent event sync failed: {error}"))??; + + // A community switch can complete while the blocking reconcile above is + // running. Revalidate before consuming the process-wide one-shot restore + // flag; an old community must never launch agents in the newly selected one. + let Some(current_scope) = crate::managed_agents::retention::arrival_retention_scope( + &app, + &state, + &arrival_relay_url, + )? + else { + return Ok(()); + }; + if current_scope.db_path != scope.db_path + || current_scope.owner_keys.public_key() != scope.owner_keys.public_key() + { + return Ok(()); + } + + if !state.managed_agent_restore_pending.load(Ordering::Acquire) { + return Ok(()); + } + let restore_scope = crate::managed_agents::ManagedAgentRestoreScope { + owner_pubkey: scope.owner_keys.public_key().to_hex(), + relay_url: scope.relay_url, + db_path: scope.db_path, + }; + restore_managed_agents_on_launch(&app, &state.shutdown_started, &restore_scope).await } diff --git a/desktop/src-tauri/src/event_sync.rs b/desktop/src-tauri/src/event_sync.rs index 6829779406c..f19bbf90421 100644 --- a/desktop/src-tauri/src/event_sync.rs +++ b/desktop/src-tauri/src/event_sync.rs @@ -1,14 +1,69 @@ //! Boot-time disk→relay event reconcile ("event sync"). //! -//! Reconciles the on-disk JSON stores (`personas.json`, `teams.json`, -//! `managed-agents.json`) into signed retention events queued for relay -//! publish. Runs after identity resolution (event signing needs the owner -//! keys), unlike the pre-identity migrations in [`crate::migration`]. +//! Reconciles on-disk persona/team stores into signed retention events, then +//! reconciles managed agents only after authoritative relay backfill. Runs +//! after identity resolution (event signing needs the owner keys), unlike the +//! pre-identity migrations in [`crate::migration`]. -use std::path::Path; +use std::{path::Path, sync::OnceLock}; +use tauri::Manager; -/// Reconcile personas, teams, and managed agents into signed retention -/// events. All readers consume the already-synced +type EventSyncTask = tauri::async_runtime::JoinHandle<()>; + +struct ScopedEventSyncTask { + owner_pubkey: String, + db_path: std::path::PathBuf, + task: EventSyncTask, +} + +fn event_sync_task_slot() -> &'static std::sync::Mutex> { + static SLOT: OnceLock>> = OnceLock::new(); + SLOT.get_or_init(|| std::sync::Mutex::new(None)) +} + +/// Replace the workspace-scoped boot reconcile task, aborting work retained +/// from a previous workspace. Passing `None` clears the slot when the selected +/// workspace has no valid retention scope. +pub fn replace_event_sync_task( + task: Option<(String, std::path::PathBuf, EventSyncTask)>, +) -> Result<(), String> { + let task = task.map(|(owner_pubkey, db_path, task)| ScopedEventSyncTask { + owner_pubkey, + db_path, + task, + }); + let mut slot = event_sync_task_slot() + .lock() + .map_err(|error| error.to_string())?; + let previous = std::mem::replace(&mut *slot, task); + drop(slot); + if let Some(previous) = previous { + previous.task.abort(); + } + Ok(()) +} + +/// Take the current workspace's boot reconcile task so the post-backfill +/// managed-agent phase can wait for its local retention writes. +pub fn take_event_sync_task( + owner_pubkey: &str, + db_path: &Path, +) -> Result, String> { + let mut slot = event_sync_task_slot() + .lock() + .map_err(|error| error.to_string())?; + let matches_scope = slot.as_ref().is_some_and(|scoped| { + scoped.owner_pubkey.eq_ignore_ascii_case(owner_pubkey) && scoped.db_path == db_path + }); + if matches_scope { + Ok(slot.take().map(|scoped| scoped.task)) + } else { + Ok(None) + } +} + +/// Reconcile personas and teams into signed retention events. Both readers +/// consume the already-synced /// `personas.json`/`teams.json`/`managed-agents.json` that /// `sync_team_personas` wrote in [`crate::migration::run_boot_migrations`] /// (see its `# Ordering` guard). Event signing needs the resolved owner keys, @@ -16,8 +71,43 @@ use std::path::Path; pub fn run_event_sync(app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: &Path) { migrate_personas_to_events(app, owner_keys, db_path); migrate_teams_to_events(app, owner_keys, db_path); - crate::managed_agents::reconcile::reconcile_agents_to_events(app, owner_keys, db_path); - hydrate_private_config_overlay(app, owner_keys, db_path); +} + +/// Reconcile managed-agent projections only after the frontend has completed +/// the selected relay's authoritative backfill. +/// +/// On a fresh device the scoped retention store has no private-config head yet. +/// Running this before relay backfill would treat a stale disk record as the +/// first kind:30179 generation and publish it over the configuration another +/// device already owns. The post-backfill order makes an existing relay head +/// visible before boot reconcile decides whether a first private projection is +/// actually missing. +pub fn run_managed_agent_event_sync( + app: &tauri::AppHandle, + owner_keys: &nostr::Keys, + db_path: &Path, +) -> Result<(), String> { + let state = app.state::(); + // Keep workspace apply outside the entire disk→retention→overlay + // transaction. Otherwise a stale A completion could write B's disk record + // into A's retention store before hydration notices that the workspace + // changed. + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|error| error.to_string())?; + let active = crate::managed_agents::retention::active_retention_scope(app, &state)?; + if active.db_path != db_path || active.owner_keys.public_key() != owner_keys.public_key() { + return Err("workspace changed before managed-agent event sync completed".into()); + } + crate::managed_agents::reconcile::reconcile_agents_to_events(app, owner_keys, db_path)?; + let hydrated = hydrate_private_config_overlay(app, owner_keys, db_path)?; + if hydrated > 0 { + eprintln!( + "buzz-desktop: private-config-overlay: hydrated {hydrated} agents from retention" + ); + } + Ok(()) } /// Rebuild the relay-config overlay from the retained kind:30179 rows. @@ -31,31 +121,25 @@ fn hydrate_private_config_overlay( app: &tauri::AppHandle, owner_keys: &nostr::Keys, db_path: &Path, -) { - use tauri::Manager; - - let result = (|| -> Result { - let conn = crate::managed_agents::retention::open_retention_db(db_path)?; - let hydrated = crate::managed_agents::private_config_overlay::hydrate_from_retention( - &conn, owner_keys, - )?; - let count = hydrated.len(); - let state = app.state::(); - *state - .private_managed_agent_overlay - .lock() - .map_err(|error| error.to_string())? = hydrated; - Ok(count) - })(); - match result { - Ok(0) => {} - Ok(count) => { - eprintln!( - "buzz-desktop: private-config-overlay: hydrated {count} agents from retention" - ) - } - Err(error) => eprintln!("buzz-desktop: private-config-overlay: {error}"), +) -> Result { + let state = app.state::(); + // The caller holds `managed_agents_store_lock` across validation, + // reconciliation, and this assignment. Workspace apply therefore either + // runs after A installs and clears it, or runs first and makes A fail its + // scope check before writing anything. + let active = crate::managed_agents::retention::active_retention_scope(app, &state)?; + if active.db_path != db_path || active.owner_keys.public_key() != owner_keys.public_key() { + return Err("workspace changed before private-config hydration completed".into()); } + let conn = crate::managed_agents::retention::open_retention_db(db_path)?; + let hydrated = + crate::managed_agents::private_config_overlay::hydrate_from_retention(&conn, owner_keys)?; + let count = hydrated.len(); + *state + .private_managed_agent_overlay + .lock() + .map_err(|error| error.to_string())? = hydrated; + Ok(count) } /// Spawn the best-effort event reconcile off the synchronous Tauri setup path. @@ -68,7 +152,7 @@ pub fn spawn_event_sync( app: tauri::AppHandle, owner_keys: nostr::Keys, db_path: std::path::PathBuf, -) { +) -> tauri::async_runtime::JoinHandle<()> { tauri::async_runtime::spawn(async move { if let Err(e) = tauri::async_runtime::spawn_blocking(move || { run_event_sync(&app, &owner_keys, &db_path); @@ -77,7 +161,7 @@ pub fn spawn_event_sync( { eprintln!("buzz-desktop: event-sync: spawn_blocking failed: {e}"); } - }); + }) } /// Reconcile `personas.json` into the persona-event retention store. diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 4f935631b60..bde34eb3357 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -884,6 +884,7 @@ pub fn run() { confirm_pairing_sas, cancel_pairing, apply_workspace, + complete_managed_agent_bootstrap, validate_repos_dir, get_active_workspace, fetch_workspace_icon, diff --git a/desktop/src-tauri/src/managed_agents/reconcile.rs b/desktop/src-tauri/src/managed_agents/reconcile.rs index 2ddd3111908..c26675a6113 100644 --- a/desktop/src-tauri/src/managed_agents/reconcile.rs +++ b/desktop/src-tauri/src/managed_agents/reconcile.rs @@ -40,10 +40,8 @@ pub(crate) fn reconcile_agents_to_events( app: &tauri::AppHandle, keys: &nostr::Keys, db_path: &Path, -) { - let Ok(base_dir) = super::managed_agents_base_dir(app) else { - return; - }; +) -> Result<(), String> { + let base_dir = super::managed_agents_base_dir(app)?; match reconcile_agents_in_dir_at(&base_dir, keys, db_path) { Ok(0) => {} @@ -52,10 +50,9 @@ pub(crate) fn reconcile_agents_to_events( "buzz-desktop: agent-event-reconcile: {reconciled} agents reconciled to retention" ); } - Err(e) => { - eprintln!("buzz-desktop: agent-event-reconcile: {e}"); - } + Err(error) => return Err(format!("agent-event-reconcile: {error}")), } + Ok(()) } /// Core reconcile logic, decoupled from the Tauri `AppHandle` for testing. diff --git a/desktop/src-tauri/src/managed_agents/restore.rs b/desktop/src-tauri/src/managed_agents/restore.rs index 25dadbeec60..5a5130b371a 100644 --- a/desktop/src-tauri/src/managed_agents/restore.rs +++ b/desktop/src-tauri/src/managed_agents/restore.rs @@ -5,9 +5,70 @@ use super::{ }; use crate::app_state::AppState; use crate::util; +use std::collections::HashSet; +use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use tauri::Manager; +/// Workspace boundary captured after authoritative launch state is hydrated. +/// Restore may do preparatory work without holding the runtime-transition lock, +/// but it must still own this exact boundary when it crosses into spawning. +pub struct ManagedAgentRestoreScope { + pub owner_pubkey: String, + pub relay_url: String, + pub db_path: PathBuf, +} + +/// A restore remains pending until the production spawn/persistence seam says +/// it completed. Dropping an attempt after any error deliberately leaves the +/// flag set so a later authoritative bootstrap can retry. +struct PendingRestoreAttempt<'a> { + pending: &'a AtomicBool, + failed_agents: HashSet, +} + +impl<'a> PendingRestoreAttempt<'a> { + fn begin(pending: &'a AtomicBool) -> Option { + pending.load(Ordering::Acquire).then_some(Self { + pending, + failed_agents: HashSet::new(), + }) + } + + fn record_failure(&mut self, pubkey: &str) { + self.failed_agents.insert(pubkey.to_string()); + } + + fn finish(self) -> Result<(), String> { + if self.failed_agents.is_empty() { + self.pending.store(false, Ordering::Release); + Ok(()) + } else { + let agent_label = if self.failed_agents.len() == 1 { + "agent" + } else { + "agents" + }; + Err(format!( + "managed-agent restore incomplete for {} {agent_label}; reconnect to retry", + self.failed_agents.len(), + )) + } + } +} + +fn restore_scope_matches( + expected: &ManagedAgentRestoreScope, + active: &super::retention::RetentionScope, +) -> bool { + active.db_path == expected.db_path + && active + .owner_keys + .public_key() + .to_hex() + .eq_ignore_ascii_case(&expected.owner_pubkey) +} + /// Outcome of a Phase B spawn attempt for one restore candidate. /// /// `Skipped` covers the case where a concurrently-running startup reconcile @@ -17,15 +78,73 @@ use tauri::Manager; /// live-child guard in `start_pair` (`runtime_commands.rs`). Without this, /// restore would kill reconcile's lazy child by its receipt and replace it with /// an eager one, flipping the pair's laziness on a startup race. +struct PendingSpawnedProcess(Option>); + +impl PendingSpawnedProcess { + fn new(process: ManagedAgentProcess) -> Self { + Self(Some(Box::new(process))) + } + + fn process_mut(&mut self) -> Option<&mut ManagedAgentProcess> { + self.0.as_deref_mut() + } + + fn adopt(mut self) -> Option> { + self.0.take() + } +} + +impl Drop for PendingSpawnedProcess { + fn drop(&mut self) { + let Some(mut process) = self.0.take() else { + return; + }; + let _ = super::terminate_process(process.child.id()); + let _ = process.child.wait(); + } +} + enum SpawnOutcome { /// Boxed: the spawned process carries its full spawn-config snapshot, so an /// inline variant would make every `Skipped`/`Failed` outcome pay for it. - Spawned(super::ManagedAgentRuntimeKey, Box), + Spawned(super::ManagedAgentRuntimeKey, PendingSpawnedProcess), Skipped, Failed(String), } type AgentSpawnResult = (String, SpawnOutcome); +/// Resolve the exact records Phase B may hand to `spawn_agent_child`. +/// +/// Candidate selection remains device-local (`start_on_app_launch` and process +/// lifecycle), while every portable/private launch field comes from the +/// authoritative overlay. The current device-local definition is applied last +/// because its prompt/model/provider/runtime remain template-owned. +fn resolve_restore_spawn_records( + records: &[super::ManagedAgentRecord], + candidate_pubkeys: &HashSet<&str>, + overlay: &super::private_config_overlay::PrivateConfigOverlay, + personas: &[super::AgentDefinition], + updated_at: &str, +) -> Vec { + records + .iter() + .filter(|record| candidate_pubkeys.contains(record.pubkey.as_str())) + .filter_map(|record| { + let mut resolved = overlay.resolve_local_record(record); + if resolved.backend != BackendKind::Local { + return None; + } + if let Some(persona_id) = resolved.persona_id.clone() { + if let Some(persona) = personas.iter().find(|persona| persona.id == persona_id) { + super::persona_events::apply_persona_snapshot(&mut resolved, persona); + resolved.updated_at = updated_at.to_string(); + } + } + Some(resolved) + }) + .collect() +} + /// Backfill the pinned persona snapshot for pre-existing agents created before /// the record became the spawn source of truth. Runs once at launch, before /// `restore_managed_agents_on_launch` spawns anything, so no agent boots from an @@ -94,6 +213,7 @@ pub fn backfill_persona_snapshots(app: &tauri::AppHandle) -> Result<(), String> pub async fn restore_managed_agents_on_launch( app: &tauri::AppHandle, shutdown_started: &AtomicBool, + expected_scope: &ManagedAgentRestoreScope, ) -> Result<(), String> { if shutdown_started.load(Ordering::SeqCst) { return Ok(()); @@ -167,7 +287,7 @@ pub async fn restore_managed_agents_on_launch( let candidates: Vec = records .iter() - .filter(|record| record.start_on_app_launch && record.backend == BackendKind::Local) + .filter(|record| record.start_on_app_launch) .map(|record| record.pubkey.clone()) .collect(); @@ -214,34 +334,38 @@ pub async fn restore_managed_agents_on_launch( record.updated_at = util::now_iso(); changed = true; } - // Re-collect to_start from the updated records so Phase B spawns the refreshed config. - agents_to_start = records + // Build the actual spawn records from the authoritative private-config + // overlay, then re-apply the device-local definition. The disk records + // above remain lifecycle/migration state and must never become the + // launch source merely because restore runs at boot. + let candidate_pubkeys: HashSet<_> = agents_to_start .iter() - .filter(|r| agents_to_start.iter().any(|s| s.pubkey == r.pubkey)) - .cloned() + .map(|record| record.pubkey.as_str()) .collect(); + let overlay = state + .private_managed_agent_overlay + .lock() + .map_err(|error| error.to_string())?; + agents_to_start = resolve_restore_spawn_records( + &records, + &candidate_pubkeys, + &overlay, + &personas_for_snapshot, + &util::now_iso(), + ); + drop(overlay); if changed { save_managed_agents(app, &records)?; } } - if agents_to_start.is_empty() { - return Ok(()); - } - - // Snapshot the workspace owner pubkey once for the legacy auth_tag fallback. - // Read outside the per-agent spawn loop so all parallel spawns see the same - // value and we don't lock `state.keys` repeatedly. - let owner_hex: Option = state - .keys - .lock() - .map_err(|e| e.to_string()) - .ok() - .map(|k| k.public_key().to_hex()); + // The owner and relay used by every child come from the same boundary that + // completed backfill, never from mutable workspace state during restore. + let owner_hex = Some(expected_scope.owner_pubkey.clone()); #[cfg(feature = "mesh-llm")] - let agents_to_start = { + let (agents_to_start, mesh_preflight_failures) = { // Preflight against the same resolution spawn uses — `resolve_effective_config` // (definition → global fallback). A linked instance's own `provider`/`model`/ // `relay_mesh` bytes never contribute. See `start_local_agent_with_preflight` @@ -267,19 +391,20 @@ pub async fn restore_managed_agents_on_launch( mesh_preflight_failures.insert(record.pubkey.clone()); } } - agents_to_start - .into_iter() - .filter(|record| !mesh_preflight_failures.contains(&record.pubkey)) - .collect::>() + ( + agents_to_start + .into_iter() + .filter(|record| !mesh_preflight_failures.contains(&record.pubkey)) + .collect::>(), + mesh_preflight_failures, + ) }; - if agents_to_start.is_empty() { - return Ok(()); - } - + #[cfg(not(feature = "mesh-llm"))] + let mesh_preflight_failures = HashSet::::new(); // Serialize spawning and runtime registration with shutdown cleanup. The - // shutdown flag is rechecked after taking the lock so shutdown either - // prevents this transition or waits until every child is tracked and can - // be terminated. + // same lock also serializes workspace mutation. Revalidate the captured + // scope after taking it: whichever transition wins determines whether this + // restore runs or becomes a harmless stale completion. let restore_transition = state .managed_agent_runtime_transition .lock() @@ -287,6 +412,28 @@ pub async fn restore_managed_agents_on_launch( if shutdown_started.load(Ordering::SeqCst) { return Ok(()); } + let Some(active_scope) = + super::retention::arrival_retention_scope(app, &state, &expected_scope.relay_url)? + else { + return Ok(()); + }; + if !restore_scope_matches(expected_scope, &active_scope) { + return Ok(()); + } + // Another completion may have waited on this transition while the first + // completed. Recheck inside the serialized boundary so it cannot launch a + // duplicate restore. + let Some(mut restore_attempt) = + PendingRestoreAttempt::begin(&state.managed_agent_restore_pending) + else { + return Ok(()); + }; + for pubkey in mesh_preflight_failures { + restore_attempt.record_failure(&pubkey); + } + if agents_to_start.is_empty() { + return restore_attempt.finish(); + } // ── Phase B (transition lock held): resolve commands and spawn in parallel ── let spawn_results: Vec = std::thread::scope(|scope| { @@ -296,11 +443,9 @@ pub async fn restore_managed_agents_on_launch( .filter(|_| !shutdown_started.load(Ordering::SeqCst)) .map(|record| { let handle = scope.spawn(move || { - let workspace_relay = - crate::relay::relay_ws_url_with_override(&app.state::()); let relay_url = crate::relay::effective_agent_relay_url( &record.relay_url, - &workspace_relay, + &expected_scope.relay_url, ); let outcome = match super::ManagedAgentRuntimeKey::new(record.pubkey.clone(), &relay_url) @@ -340,9 +485,10 @@ pub async fn restore_managed_agents_on_launch( owner_hex_ref, ) }) { - Ok(process) => { - SpawnOutcome::Spawned(key, Box::new(process)) - } + Ok(process) => SpawnOutcome::Spawned( + key, + PendingSpawnedProcess::new(process), + ), Err(error) => SpawnOutcome::Failed(error), } } @@ -359,7 +505,7 @@ pub async fn restore_managed_agents_on_launch( }); if spawn_results.is_empty() { - return Ok(()); + return restore_attempt.finish(); } // ── Phase C (re-acquire lock): write back PIDs and status to records ── @@ -380,10 +526,13 @@ pub async fn restore_managed_agents_on_launch( // Skipped means a concurrent reconcile already owns a live child for // this pair; leave its runtime and record state untouched. SpawnOutcome::Skipped => continue, - SpawnOutcome::Spawned(key, mut process) => { + SpawnOutcome::Spawned(key, mut pending_process) => { let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else { continue; }; + let Some(process) = pending_process.process_mut() else { + continue; + }; let now = util::now_iso(); let receipt = super::ManagedAgentRuntimeReceipt { key: key.clone(), @@ -392,8 +541,7 @@ pub async fn restore_managed_agents_on_launch( started_at: now.clone(), }; if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) { - let _ = super::terminate_process(process.child.id()); - let _ = process.child.wait(); + restore_attempt.record_failure(&pubkey); record.updated_at = now; record.last_error = Some(error); continue; @@ -404,10 +552,14 @@ pub async fn restore_managed_agents_on_launch( record.last_stopped_at = None; record.last_exit_code = None; record.last_error = None; + let Some(process) = pending_process.adopt() else { + continue; + }; runtimes.insert(key, super::ManagedAgentPairRuntime::starting(*process)); successfully_spawned.push(pubkey); } SpawnOutcome::Failed(error) => { + restore_attempt.record_failure(&pubkey); let Ok(record) = find_managed_agent_mut(&mut records, &pubkey) else { continue; }; @@ -449,6 +601,10 @@ pub async fn restore_managed_agents_on_launch( .collect(); save_managed_agents(app, &records)?; + // A restore request is consumed only when every requested agent crossed the + // spawn, receipt, adoption, and persistence boundary. Successful agents are + // already tracked, so a retry skips them and retries only the failed agents. + let restore_result = restore_attempt.finish(); drop(runtimes); drop(_store_guard); drop(restore_transition); @@ -469,7 +625,7 @@ pub async fn restore_managed_agents_on_launch( }); } - Ok(()) + restore_result } #[cfg(feature = "mesh-llm")] @@ -489,3 +645,214 @@ fn persist_restore_error( record.last_error = Some(error); save_managed_agents(app, &records) } + +#[cfg(test)] +mod tests { + use super::*; + use buzz_core_pkg::private_managed_agent::{ + Payload, PrivateConfig, PrivateIdentity, FORMAT, VERSION, + }; + use serde_json::{json, Map}; + use std::collections::BTreeMap; + + #[test] + fn failed_restore_attempt_remains_pending_for_retry() { + let pending = AtomicBool::new(true); + + { + let _first_attempt = PendingRestoreAttempt::begin(&pending).unwrap(); + // Leaving this scope models the production restore returning an error. + } + assert!(pending.load(Ordering::Acquire)); + + PendingRestoreAttempt::begin(&pending) + .unwrap() + .finish() + .unwrap(); + assert!(!pending.load(Ordering::Acquire)); + assert!(PendingRestoreAttempt::begin(&pending).is_none()); + } + + #[test] + fn partial_restore_failure_stays_pending_until_retry_succeeds() { + let pending = AtomicBool::new(true); + let mut first_attempt = PendingRestoreAttempt::begin(&pending).unwrap(); + + // One agent crossed the production boundary; another did not. Only + // failures are recorded, so successful agents remain live and Phase A + // will skip them when the pending attempt is retried. + first_attempt.record_failure(&"bb".repeat(32)); + let error = first_attempt.finish().unwrap_err(); + assert!(error.contains("1 agent")); + assert!(pending.load(Ordering::Acquire)); + + PendingRestoreAttempt::begin(&pending) + .unwrap() + .finish() + .unwrap(); + assert!(!pending.load(Ordering::Acquire)); + } + + #[test] + fn restore_scope_rejects_a_workspace_that_changed_before_spawn() { + let owner_a = nostr::Keys::generate(); + let owner_b = nostr::Keys::generate(); + let expected = ManagedAgentRestoreScope { + owner_pubkey: owner_a.public_key().to_hex(), + relay_url: "wss://community-a.example".into(), + db_path: PathBuf::from("scope-a.db"), + }; + let matching = super::super::retention::RetentionScope { + db_path: PathBuf::from("scope-a.db"), + relay_url: "wss://community-a.example".into(), + owner_keys: owner_a, + }; + let switched = super::super::retention::RetentionScope { + db_path: PathBuf::from("scope-b.db"), + relay_url: "wss://community-b.example".into(), + owner_keys: owner_b, + }; + + assert!(restore_scope_matches(&expected, &matching)); + assert!(!restore_scope_matches(&expected, &switched)); + } + + #[cfg(unix)] + #[test] + fn spawn_guard_terminates_child_when_phase_c_cannot_adopt() { + use std::os::unix::process::CommandExt; + use std::process::{Command, Stdio}; + + let mut command = Command::new("/bin/sh"); + command + .args(["-c", "sleep 30"]) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + command.process_group(0); + let child = command.spawn().expect("spawn external restore child"); + let pid = child.id(); + let process = ManagedAgentProcess { + child, + log_path: PathBuf::new(), + spawn_config: super::super::spawn_snapshot::prospective_spawn_config_snapshot( + &disk_record(&"aa".repeat(32)), + &[], + &[], + "wss://relay.example", + &Default::default(), + ), + setup_mode: false, + adapter_availability: None, + start_nonce: "restore-test".into(), + }; + + drop(PendingSpawnedProcess::new(process)); + + assert!(!super::super::process_is_running(pid)); + } + + fn disk_record(pubkey: &str) -> super::super::ManagedAgentRecord { + serde_json::from_value(json!({ + "pubkey": pubkey, + "name": "Disk agent", + "private_key_nsec": "nsec-disk", + "auth_tag": "disk-auth", + "relay_url": "wss://relay.example", + "acp_command": "buzz-acp", + "agent_command": "goose", + "agent_args": [], + "mcp_command": "", + "turn_timeout_seconds": 320, + "parallelism": 1, + "system_prompt": "STALE disk prompt", + "model": "disk-model", + "env_vars": {"API_TOKEN": "disk-token"}, + "start_on_app_launch": true, + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z" + })) + .unwrap() + } + + #[test] + fn restore_spawn_record_uses_relay_config_after_delayed_backfill() { + let pubkey = "aa".repeat(32); + let disk = disk_record(&pubkey); + let mut overlay = super::super::private_config_overlay::PrivateConfigOverlay::default(); + overlay + .insert(Payload { + format: FORMAT.into(), + version: VERSION, + agent_pubkey: pubkey.clone(), + owner_pubkey: "bb".repeat(32), + generation: 2, + previous_event_id: Some("cc".repeat(32)), + updated_at: "2026-08-07T00:00:00Z".into(), + identity: PrivateIdentity { + private_key_nsec: "nsec-relay".into(), + auth_tag: Some("relay-auth".into()), + }, + config: PrivateConfig { + relay_url: "wss://relay.example".into(), + name: "Relay agent".into(), + persona_id: None, + runtime: Some("goose".into()), + model: Some("relay-model".into()), + provider: None, + system_prompt: Some("FRESH relay prompt".into()), + parallelism: Some(7), + respond_to: None, + respond_to_allowlist: vec![], + agent_command_override: None, + agent_args: vec![], + idle_timeout_seconds: None, + max_turn_duration_seconds: None, + env_vars: BTreeMap::from([("API_TOKEN".into(), "relay-token".into())]), + backend: json!({"type":"local"}), + backend_agent_id: None, + team_id: None, + persona_name_in_team: None, + relay_mesh: None, + extra: Map::new(), + }, + extensions: BTreeMap::new(), + extra: Map::new(), + }) + .unwrap(); + let candidates = HashSet::from([pubkey.as_str()]); + + let resolved = resolve_restore_spawn_records( + &[disk], + &candidates, + &overlay, + &[], + "2026-08-07T00:00:01Z", + ); + + assert_eq!(resolved.len(), 1); + assert_eq!(resolved[0].name, "Relay agent"); + assert_eq!(resolved[0].private_key_nsec, "nsec-relay"); + assert_eq!(resolved[0].auth_tag.as_deref(), Some("relay-auth")); + assert_eq!( + resolved[0].system_prompt.as_deref(), + Some("FRESH relay prompt") + ); + assert_eq!(resolved[0].model.as_deref(), Some("relay-model")); + assert_eq!(resolved[0].parallelism, 7); + assert_eq!(resolved[0].env_vars["API_TOKEN"], "relay-token"); + + let spawn_snapshot = super::super::spawn_snapshot::prospective_spawn_config_snapshot( + &resolved[0], + &[], + &[], + "wss://relay.example", + &Default::default(), + ) + .canonical(); + assert_eq!(spawn_snapshot["system_prompt"], "FRESH relay prompt"); + assert_eq!(spawn_snapshot["model"], "relay-model"); + assert_eq!(spawn_snapshot["env"]["API_TOKEN"], "relay-token"); + assert_eq!(spawn_snapshot["parallelism"], 7); + } +} diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index b929dbb6131..3dea9dd82fe 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -168,6 +168,17 @@ with a TypeScript lookup table or an id comparison in a component. `getAgentAccessOwnerOnly()` is true, every managed agent's access control is locked to owner-only, including provider-backed agents. A provider backend does not prove remote execution and must never create a policy carve-out. +12. **Launch-time restore waits for authoritative agent state.** A saved agent + must not auto-start from disk while the selected community's private-agent + backfill is incomplete. The frontend buffers live definition events during + backfill, drains them before completing bootstrap, and leaves restore pending + after any reconciliation or restore failure so reconnect can retry. The + backend captures the workspace scope, revalidates it under the runtime + transition lock, resolves each candidate through the relay-primary overlay, + and keeps `start_on_app_launch` device-local. Incomplete or stale authority + fails closed: the agent stays stopped. Every spawned child remains owned by + a cleanup guard until its receipt is durable and the runtime registry adopts + it; errors or deletions before adoption must terminate and reap the child. ## The tests that enforce this @@ -195,6 +206,10 @@ with a TypeScript lookup table or an id comparison in a component. - Rust: `runtime_metadata_env_vars` tests pin spawn-time key application. - Rust: persona sharing/retention tests pin relay+owner scoping, durable enqueue errors, relay rejection/unavailability, and accepted publication. +- `lib/usePersonaSync.test.mjs` and the restore tests in + `desktop/src-tauri/src/managed_agents/restore.rs` pin the authoritative + backfill barrier, live-event drain, retry, workspace-scope, overlay, and + pre-adoption child-cleanup contracts. ## Keep this file true diff --git a/desktop/src/features/agents/lib/usePersonaSync.test.mjs b/desktop/src/features/agents/lib/usePersonaSync.test.mjs index f3d9b93acfc..eb735a98163 100644 --- a/desktop/src/features/agents/lib/usePersonaSync.test.mjs +++ b/desktop/src/features/agents/lib/usePersonaSync.test.mjs @@ -9,7 +9,10 @@ import { KIND_PRIVATE_MANAGED_AGENT, KIND_TEAM, } from "@/shared/constants/kinds"; -import { startPersonaSync } from "./usePersonaSync.ts"; +import { + fetchPersonaSyncBackfill, + startPersonaSync, +} from "./usePersonaSync.ts"; const EXPECTED_KINDS = [ KIND_PERSONA, @@ -19,15 +22,68 @@ const EXPECTED_KINDS = [ KIND_DELETION, ]; +function mockConnectedRelay() { + return mock.method(relayClient, "subscribeToConnectionState", (listener) => { + listener("connected"); + return () => {}; + }); +} + +test("fetchPersonaSyncBackfill drains timestamp boundaries before paging", async () => { + const filters = []; + const firstPage = Array.from({ length: 500 }, (_, index) => ({ + id: `head-${index}`, + created_at: index === 499 ? 10 : 20, + })); + const pages = [ + firstPage, + [ + { id: "head-499", created_at: 10 }, + { id: "boundary-peer", created_at: 10 }, + ], + [{ id: "older", created_at: 9 }], + ]; + + const events = await fetchPersonaSyncBackfill("owner-pubkey", (filter) => { + filters.push(filter); + return Promise.resolve(pages.shift() ?? []); + }); + + assert.equal(events.length, 502); + assert.deepEqual(filters[1], { + kinds: EXPECTED_KINDS, + authors: ["owner-pubkey"], + limit: 500, + since: 10, + until: 10, + }); + assert.equal(filters[2].until, 9); +}); + +test("fetchPersonaSyncBackfill rejects an unexhausted timestamp bucket", async () => { + const fullPage = Array.from({ length: 500 }, (_, index) => ({ + id: `event-${index}`, + created_at: 10, + })); + + await assert.rejects( + fetchPersonaSyncBackfill("owner-pubkey", () => Promise.resolve(fullPage)), + /too many events share one timestamp/, + ); +}); + // Regression guard for the fresh-start backfill gap (F3): a device that comes // online AFTER another published gets zero history from a live-only `limit: 0` // subscription, because reconnect-replay's since-cursor is undefined until the // first live event. `startPersonaSync` MUST do a one-shot history fetch up // front, and both the backfill and the live sub MUST carry the deletion kind // so tombstones catch up too. -test("startPersonaSync backfills history including the deletion kind", () => { +test("startPersonaSync backfills history including the deletion kind", async () => { const fetchCalls = []; const liveCalls = []; + globalThis.window = { + __TAURI_INTERNALS__: { invoke: () => Promise.resolve() }, + }; mock.method(relayClient, "fetchEvents", (filter) => { fetchCalls.push(filter); return Promise.resolve([]); @@ -36,8 +92,10 @@ test("startPersonaSync backfills history including the deletion kind", () => { liveCalls.push(filter); return Promise.resolve(() => Promise.resolve()); }); + mockConnectedRelay(); startPersonaSync("owner-pubkey", "wss://relay.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); assert.equal(fetchCalls.length, 1, "must do exactly one backfill fetch"); assert.deepEqual( @@ -59,6 +117,7 @@ test("startPersonaSync backfills history including the deletion kind", () => { ); mock.reset(); + delete globalThis.window; }); // Regression guard for the arrival-scope fix (F6): the reconcile must carry the @@ -87,6 +146,7 @@ test("startPersonaSync forwards its own relay as the event arrival relay", async mock.method(relayClient, "subscribeLive", () => Promise.resolve(() => Promise.resolve()), ); + mockConnectedRelay(); startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); // Let the backfill promise chain and the reconcile invoke settle. @@ -110,3 +170,304 @@ test("startPersonaSync forwards its own relay as the event arrival relay", async mock.reset(); delete globalThis.window; }); + +test("managed-agent restore waits for every backfill event to reconcile", async () => { + const invokes = []; + const pendingReconciles = []; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + if (cmd !== "reconcile_inbound_persona_event") { + return Promise.resolve(); + } + return new Promise((resolve) => pendingReconciles.push(resolve)); + }, + }, + }; + + mock.method(relayClient, "subscribeLive", () => + Promise.resolve(() => Promise.resolve()), + ); + mock.method(relayClient, "fetchEvents", () => + Promise.resolve([ + { id: "first", pubkey: "owner-pubkey", kind: KIND_PRIVATE_MANAGED_AGENT }, + { id: "second", pubkey: "owner-pubkey", kind: KIND_MANAGED_AGENT }, + ]), + ); + mockConnectedRelay(); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(pendingReconciles.length, 1, "reconciliation is serialized"); + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + "restore must not start while the first retained event is pending", + ); + + pendingReconciles.shift()(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + pendingReconciles.length, + 1, + "second event starts after the first", + ); + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + "restore must not start while any retained event is pending", + ); + + pendingReconciles.shift()(); + await new Promise((resolve) => setImmediate(resolve)); + const completions = invokes.filter( + (call) => call.cmd === "complete_managed_agent_bootstrap", + ); + assert.equal(completions.length, 1); + assert.deepEqual(completions[0].args, { + ownerPubkey: "owner-pubkey", + arrivalRelayUrl: "wss://community-a.example", + }); + + mock.reset(); + delete globalThis.window; +}); + +test("managed-agent restore drains a live event received during backfill", async () => { + const invokes = []; + const pendingReconciles = []; + let liveListener = () => {}; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + if (cmd !== "reconcile_inbound_persona_event") { + return Promise.resolve(); + } + return new Promise((resolve) => pendingReconciles.push(resolve)); + }, + }, + }; + mock.method(relayClient, "subscribeLive", (_filter, listener) => { + liveListener = listener; + return Promise.resolve(() => Promise.resolve()); + }); + mock.method(relayClient, "fetchEvents", () => + Promise.resolve([ + { + id: "backfill", + pubkey: "owner-pubkey", + kind: KIND_PRIVATE_MANAGED_AGENT, + }, + ]), + ); + mockConnectedRelay(); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(pendingReconciles.length, 1); + + liveListener({ + id: "live-during-backfill", + pubkey: "owner-pubkey", + kind: KIND_PRIVATE_MANAGED_AGENT, + }); + pendingReconciles.shift()(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + pendingReconciles.length, + 1, + "the buffered live event must reconcile before restore", + ); + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + ); + + pendingReconciles.shift()(); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + invokes.filter((call) => call.cmd === "complete_managed_agent_bootstrap") + .length, + 1, + ); + + mock.reset(); + delete globalThis.window; +}); + +test("managed-agent restore fails closed when the live bootstrap buffer overflows", async () => { + const invokes = []; + let liveListener = () => {}; + let finishBackfill = () => {}; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + return Promise.resolve(); + }, + }, + }; + mock.method(relayClient, "subscribeLive", (_filter, listener) => { + liveListener = listener; + return Promise.resolve(() => Promise.resolve()); + }); + mock.method( + relayClient, + "fetchEvents", + () => + new Promise((resolve) => { + finishBackfill = () => resolve([]); + }), + ); + mockConnectedRelay(); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + for (let index = 0; index <= 20_000; index += 1) { + liveListener({ + id: `live-${index}`, + pubkey: "owner-pubkey", + kind: KIND_PRIVATE_MANAGED_AGENT, + }); + } + finishBackfill(); + await new Promise((resolve) => setImmediate(resolve)); + await new Promise((resolve) => setImmediate(resolve)); + + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + ); + + mock.reset(); + delete globalThis.window; +}); + +test("failed backfill reconciliation keeps managed-agent restore paused", async () => { + const invokes = []; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + return cmd === "reconcile_inbound_persona_event" + ? Promise.reject(new Error("retention unavailable")) + : Promise.resolve(); + }, + }, + }; + mock.method(relayClient, "subscribeLive", () => + Promise.resolve(() => Promise.resolve()), + ); + mock.method(relayClient, "fetchEvents", () => + Promise.resolve([ + { + id: "broken", + pubkey: "owner-pubkey", + kind: KIND_PRIVATE_MANAGED_AGENT, + }, + ]), + ); + mockConnectedRelay(); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + ); + + mock.reset(); + delete globalThis.window; +}); + +test("managed-agent bootstrap retries after the relay reconnects", async () => { + const invokes = []; + let connectionListener = () => {}; + let fetchAttempt = 0; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + return Promise.resolve(); + }, + }, + }; + mock.method(relayClient, "subscribeLive", () => + Promise.resolve(() => Promise.resolve()), + ); + mock.method(relayClient, "subscribeToConnectionState", (listener) => { + connectionListener = listener; + listener("connected"); + return () => {}; + }); + mock.method(relayClient, "fetchEvents", () => { + fetchAttempt += 1; + return fetchAttempt === 1 + ? Promise.reject(new Error("relay unavailable")) + : Promise.resolve([]); + }); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + invokes.some((call) => call.cmd === "complete_managed_agent_bootstrap"), + false, + ); + + connectionListener("reconnecting"); + connectionListener("connected"); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal( + invokes.filter((call) => call.cmd === "complete_managed_agent_bootstrap") + .length, + 1, + ); + + mock.reset(); + delete globalThis.window; +}); + +test("live agent-settings sync retries after the relay reconnects", async () => { + const invokes = []; + let connectionListener = () => {}; + let subscribeAttempt = 0; + globalThis.window = { + __TAURI_INTERNALS__: { + invoke: (cmd, args) => { + invokes.push({ cmd, args }); + return Promise.resolve(); + }, + }, + }; + mock.method(relayClient, "subscribeToConnectionState", (listener) => { + connectionListener = listener; + listener("connected"); + return () => {}; + }); + mock.method(relayClient, "subscribeLive", () => { + subscribeAttempt += 1; + return subscribeAttempt === 1 + ? Promise.reject(new Error("subscription unavailable")) + : Promise.resolve(() => Promise.resolve()); + }); + mock.method(relayClient, "fetchEvents", () => Promise.resolve([])); + + startPersonaSync("owner-pubkey", "wss://community-a.example", () => false); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(subscribeAttempt, 1); + + connectionListener("reconnecting"); + connectionListener("connected"); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(subscribeAttempt, 2); + assert.equal( + invokes.filter((call) => call.cmd === "complete_managed_agent_bootstrap") + .length, + 1, + ); + + mock.reset(); + delete globalThis.window; +}); diff --git a/desktop/src/features/agents/lib/usePersonaSync.ts b/desktop/src/features/agents/lib/usePersonaSync.ts index cc34adb45bb..b05e7cfc3b2 100644 --- a/desktop/src/features/agents/lib/usePersonaSync.ts +++ b/desktop/src/features/agents/lib/usePersonaSync.ts @@ -1,7 +1,11 @@ import * as React from "react"; +import { toast } from "sonner"; import { relayClient } from "@/shared/api/relayClient"; -import { reconcileInboundPersonaEvent } from "@/shared/api/tauriPersonas"; +import { + completeManagedAgentBootstrap, + reconcileInboundPersonaEvent, +} from "@/shared/api/tauriPersonas"; import type { RelayEvent } from "@/shared/api/types"; import { KIND_DELETION, @@ -21,9 +25,74 @@ const PERSONA_SYNC_KINDS = [ KIND_PRIVATE_MANAGED_AGENT, KIND_DELETION, ]; +const PERSONA_SYNC_BACKFILL_LIMIT = 500; +const PERSONA_SYNC_MAX_BACKFILL_PAGES = 40; +const PERSONA_SYNC_MAX_BUFFERED_LIVE_EVENTS = + PERSONA_SYNC_BACKFILL_LIMIT * PERSONA_SYNC_MAX_BACKFILL_PAGES; + +type FetchPersonaSyncPage = (filter: { + kinds: number[]; + authors: string[]; + limit: number; + since?: number; + until?: number; +}) => Promise; + +/** + * Read the complete owner-authored projection history without silently + * accepting the relay's per-request limit as completeness. + * + * Nostr's `until` cursor is inclusive, so a full page cannot advance beyond + * its oldest second until that entire timestamp bucket is known to fit in one + * response. Failing closed here leaves saved agents stopped; it is safer than + * launching one from a partial private-config or deletion history. + */ +export async function fetchPersonaSyncBackfill( + pubkey: string, + fetchPage: FetchPersonaSyncPage = (filter) => relayClient.fetchEvents(filter), +): Promise { + const byId = new Map(); + let until: number | undefined; + + for ( + let pageIndex = 0; + pageIndex < PERSONA_SYNC_MAX_BACKFILL_PAGES; + pageIndex += 1 + ) { + const page = await fetchPage({ + kinds: PERSONA_SYNC_KINDS, + authors: [pubkey], + limit: PERSONA_SYNC_BACKFILL_LIMIT, + ...(until === undefined ? {} : { until }), + }); + for (const event of page) byId.set(event.id, event); + if (page.length < PERSONA_SYNC_BACKFILL_LIMIT) { + return [...byId.values()]; + } + + const oldest = Math.min(...page.map((event) => event.created_at)); + const boundary = await fetchPage({ + kinds: PERSONA_SYNC_KINDS, + authors: [pubkey], + limit: PERSONA_SYNC_BACKFILL_LIMIT, + since: oldest, + until: oldest, + }); + for (const event of boundary) byId.set(event.id, event); + if (boundary.length >= PERSONA_SYNC_BACKFILL_LIMIT) { + throw new Error( + "Agent settings cannot be completely synchronized because too many events share one timestamp.", + ); + } + if (oldest <= 0) return [...byId.values()]; + until = oldest - 1; + } + + throw new Error("Agent settings synchronization exceeded its safety limit."); +} // Start the persona/team/agent/deletion sync for `pubkey` on `relayUrl`: -// one-shot backfill of existing heads + tombstones, then a live subscription. +// establish the live edge, then backfill existing heads + tombstones. // Returns a disposer that closes the live subscription. Extracted from the hook // so the wiring is unit-testable without a React renderer (see // `usePersonaSync.test.mjs`). @@ -37,42 +106,143 @@ export function startPersonaSync( relayUrl: string, onCancelled: () => boolean, ): () => Promise { - const reconcile = (event: RelayEvent) => { + let reconcileTail = Promise.resolve(); + let reconcileFailed = false; + let bootstrapComplete = false; + let bootstrapAttempt: Promise | null = null; + let liveSetup: Promise | null = null; + let bufferLiveEvents = true; + let liveBufferOverflowed = false; + const bufferedLiveEvents = new Map(); + let notifiedPaused = false; + const notifyPaused = () => { + if (notifiedPaused || onCancelled()) return; + notifiedPaused = true; + toast.error("Automatic agent startup is paused", { + description: "Reconnect to try again.", + }); + }; + const queueReconcile = (event: RelayEvent) => { if (event.pubkey !== pubkey) return; - void reconcileInboundPersonaEvent(JSON.stringify(event), relayUrl).catch( - (error) => { - console.warn("[usePersonaSync] reconcile failed:", error); - }, + const operation = reconcileTail.then(() => + reconcileInboundPersonaEvent(JSON.stringify(event), relayUrl), ); + reconcileTail = operation.catch((error) => { + reconcileFailed = true; + console.warn("[usePersonaSync] reconcile failed:", error); + }); + }; + const reconcileLiveEvent = (event: RelayEvent) => { + if (event.pubkey !== pubkey) return; + if (bufferLiveEvents) { + if ( + !bufferedLiveEvents.has(event.id) && + bufferedLiveEvents.size >= PERSONA_SYNC_MAX_BUFFERED_LIVE_EVENTS + ) { + liveBufferOverflowed = true; + return; + } + bufferedLiveEvents.set(event.id, event); + return; + } + queueReconcile(event); }; - // One-shot backfill of existing heads + tombstones (closes the fresh-start - // gap that live-only subscription + reconnect-replay cannot recover). - void relayClient - .fetchEvents({ kinds: PERSONA_SYNC_KINDS, authors: [pubkey], limit: 500 }) - .then((events) => { + let unsub: (() => Promise) | null = null; + let unsubscribeConnectionState: (() => void) | null = null; + const attemptBootstrap = () => { + if (bootstrapComplete || bootstrapAttempt || onCancelled()) return; + reconcileFailed = false; + bufferLiveEvents = true; + liveBufferOverflowed = false; + bufferedLiveEvents.clear(); + bootstrapAttempt = (async () => { + const events = await fetchPersonaSyncBackfill(pubkey); if (onCancelled()) return; - for (const event of events) reconcile(event); - }) - .catch((error) => { - console.warn("[usePersonaSync] backfill failed:", error); - }); + for (const event of events) queueReconcile(event); + await reconcileTail; + if (onCancelled()) return; + if (reconcileFailed) { + notifyPaused(); + return; + } - let unsub: (() => Promise) | null = null; - void relayClient - .subscribeLive( - { kinds: PERSONA_SYNC_KINDS, authors: [pubkey], limit: 0 }, - reconcile, - ) - .then((dispose) => { - if (onCancelled()) { - void dispose(); - } else { + // Live delivery starts before history so no event can fall between the + // two. Drain everything received during backfill to quiescence, then + // make one synchronous transition to normal live reconciliation. A + // callback cannot interleave between the empty check and that transition. + while (bufferedLiveEvents.size > 0) { + const batch = [...bufferedLiveEvents.values()]; + bufferedLiveEvents.clear(); + for (const event of batch) queueReconcile(event); + await reconcileTail; + if (onCancelled()) return; + if (reconcileFailed || liveBufferOverflowed) { + notifyPaused(); + return; + } + } + if (liveBufferOverflowed) { + notifyPaused(); + return; + } + bufferLiveEvents = false; + + await completeManagedAgentBootstrap(pubkey, relayUrl); + bootstrapComplete = true; + })() + .catch((error) => { + console.warn( + "[usePersonaSync] authoritative backfill failed; managed-agent restore remains paused:", + error, + ); + notifyPaused(); + }) + .finally(() => { + bootstrapAttempt = null; + }); + }; + + const ensureLiveSync = () => { + if (unsub || liveSetup || onCancelled()) return; + liveSetup = relayClient + .subscribeLive( + { kinds: PERSONA_SYNC_KINDS, authors: [pubkey], limit: 0 }, + reconcileLiveEvent, + ) + .then(async (dispose) => { + if (onCancelled()) { + await dispose(); + return; + } unsub = dispose; + attemptBootstrap(); + }) + .catch((error) => { + console.warn( + "[usePersonaSync] live agent-settings sync failed; managed-agent restore remains paused:", + error, + ); + notifyPaused(); + }) + .finally(() => { + liveSetup = null; + }); + }; + + unsubscribeConnectionState = relayClient.subscribeToConnectionState( + (state) => { + if (state !== "connected") return; + if (unsub) { + attemptBootstrap(); + } else { + ensureLiveSync(); } - }); + }, + ); return async () => { + unsubscribeConnectionState?.(); if (unsub) await unsub(); }; } diff --git a/desktop/src/shared/api/tauriPersonas.ts b/desktop/src/shared/api/tauriPersonas.ts index 3cd9734ae26..4291d8ffab0 100644 --- a/desktop/src/shared/api/tauriPersonas.ts +++ b/desktop/src/shared/api/tauriPersonas.ts @@ -457,3 +457,17 @@ export async function reconcileInboundPersonaEvent( arrivalRelayUrl, }); } + +/** + * Release launch-time agent restore after this exact relay and owner have + * completed their authoritative private-config backfill. + */ +export async function completeManagedAgentBootstrap( + ownerPubkey: string, + arrivalRelayUrl: string, +): Promise { + await invokeTauri("complete_managed_agent_bootstrap", { + ownerPubkey, + arrivalRelayUrl, + }); +}