diff --git a/desktop/src-tauri/src/commands/global_agent_config.rs b/desktop/src-tauri/src/commands/global_agent_config.rs index a0f037cad7..d596183633 100644 --- a/desktop/src-tauri/src/commands/global_agent_config.rs +++ b/desktop/src-tauri/src/commands/global_agent_config.rs @@ -19,7 +19,7 @@ use crate::{ agent_readiness, current_instance_id, find_managed_agent_mut, known_acp_runtime, load_global_agent_config, record_agent_command, resolve_effective_agent_env, stop_managed_agent_process, sync_managed_agent_processes, validate_global_config, - AgentReadiness, BackendKind, GlobalAgentConfig, + AgentDefinition, AgentReadiness, BackendKind, GlobalAgentConfig, TeamRecord, }, }; @@ -177,7 +177,7 @@ pub async fn set_global_agent_config( /// Outcome of a single per-agent restart attempt in Phase 2. #[derive(Debug)] -enum RestartOutcome { +pub(crate) enum RestartOutcome { /// Stop succeeded and the agent re-launched with the new config. Restarted, /// Stop succeeded but the subsequent spawn failed. @@ -186,15 +186,38 @@ enum RestartOutcome { Skipped, } -/// Error returned by [`restart_under_captured_epoch`]. +/// Error returned by [`restart_under_captured_epoch_for`]. #[derive(Debug)] -enum EpochError { - /// Eligibility check failed before the stop; the agent was not touched. +pub(crate) enum EpochError { + /// Eligibility check failed before the stop (or stop failed); the agent + /// was not touched, or the stop failed before any irreversible transition. Skipped(String), /// Stop succeeded but the subsequent spawn failed. FailedAfterStop(String), } +/// Immutable captured context prepared fallibly BEFORE any stop in the async +/// pre-stop phase of [`restart_local_agent_on_config_change_for`]. +/// +/// All fields derive from `captured_scope.definitions_dir` — never from live +/// state. Owner keys are verified against `captured_scope.owner_pubkey` before +/// construction; `owner_hex` is derived from the verified keys, not the scope +/// string. The context is frozen once built; subsequent workspace switches or +/// agent edits are detected by generation revalidation inside the epoch. +#[derive(Clone, Debug)] +pub(crate) struct CapturedRestartContext { + pub scope: crate::managed_agents::scope::WorkspaceAgentScope, + pub personas: Vec, + pub teams: Vec, + pub global: GlobalAgentConfig, + /// Owner pubkey hex derived from verified signing keys, not the scope string. + pub owner_hex: String, + /// Effective Relay-Mesh model ID resolved from the candidate record at + /// preparation time. Used for pre-stop Mesh preflight and in-epoch + /// re-resolution mismatch guard. + pub mesh_model_id: Option, +} + /// Collect pubkeys of local agents that should be restarted after a global /// config change, together with the personas snapshot used for the scan. /// @@ -282,11 +305,14 @@ fn collect_restart_candidates_at( /// Restart a local agent whose effective env changed under the new global config. /// -/// Drives the entire stop→spawn sequence through one -/// [`restart_under_captured_epoch`] call so no concurrent workspace switch can -/// land between the stop and the spawn. Returns [`RestartOutcome::FailedAfterStop`] -/// when the stop succeeded but the spawn failed; [`RestartOutcome::Skipped`] when -/// eligibility was lost before the stop. +/// Async driver. Prepares [`CapturedRestartContext`] fallibly BEFORE any stop, +/// including the relay-Mesh preflight — failure here leaves the old process +/// running. Only after the context is fully prepared does the atomic +/// stop→spawn epoch run inside `spawn_blocking`. +/// +/// Returns [`RestartOutcome::FailedAfterStop`] when stop succeeded but spawn +/// failed; [`RestartOutcome::Skipped`] when any pre-stop check fails or the +/// epoch generation guard aborts before the stop. async fn restart_local_agent_on_config_change( app: &AppHandle, pubkey: &str, @@ -296,28 +322,246 @@ async fn restart_local_agent_on_config_change( captured_scope: &crate::managed_agents::scope::WorkspaceAgentScope, definitions_dir: &std::path::Path, ) -> RestartOutcome { + restart_local_agent_on_config_change_for( + app, + pubkey, + old_global, + new_global, + personas_snapshot, + captured_scope, + definitions_dir, + // Production mesh preflight — async, runs before spawn_blocking. + |app_ref, model_id| { + let app_clone = app_ref.clone(); + let model = model_id.map(str::to_string); + Box::pin(async move { + #[cfg(feature = "mesh-llm")] + { + crate::commands::ensure_relay_mesh_for_record( + &app_clone, + model.as_deref(), + false, + ) + .await + } + #[cfg(not(feature = "mesh-llm"))] + { + let _ = (app_clone, model); + Ok(()) + } + }) + }, + // Production stop function. + |app_ref, record_mut, runtimes| stop_managed_agent_process(app_ref, record_mut, runtimes), + // Production spawn function. + |app_ref, rec, relay, owner, personas, global, teams| { + crate::managed_agents::spawn_agent_child_at( + app_ref, rec, relay, true, owner, personas, global, teams, + ) + }, + // Production receipt function. + |app_ref, receipt| crate::managed_agents::write_agent_runtime_receipt(app_ref, receipt), + ) + .await +} + +/// Injected-seam async driver for restarting a local agent on config change. +/// +/// Accepts injected `mesh_fn`, `stop_fn`, `spawn_fn`, and `write_receipt_fn` +/// so the full driver can be exercised in tests without spawning real processes, +/// hitting the macOS keychain, or calling a real Mesh relay. +/// +/// The function signature uses generic parameters (stable Rust) rather than +/// `AsyncFn` (nightly) or `dyn` (requires boxing closures). The production +/// adapter `restart_local_agent_on_config_change` closes over the concrete +/// function pointers. +#[allow(clippy::too_many_arguments)] +pub(crate) async fn restart_local_agent_on_config_change_for< + R, + MeshFn, + StopFn, + SpawnFn, + ReceiptFn, +>( + app: &tauri::AppHandle, + pubkey: &str, + old_global: &GlobalAgentConfig, + new_global: &GlobalAgentConfig, + personas_snapshot: &[crate::managed_agents::AgentDefinition], + captured_scope: &crate::managed_agents::scope::WorkspaceAgentScope, + definitions_dir: &std::path::Path, + mesh_fn: MeshFn, + stop_fn: StopFn, + spawn_fn: SpawnFn, + write_receipt_fn: ReceiptFn, +) -> RestartOutcome +where + R: tauri::Runtime, + MeshFn: for<'a> Fn( + &'a tauri::AppHandle, + Option<&'a str>, + ) -> std::pin::Pin< + Box> + Send + 'a>, + >, + StopFn: Fn( + &tauri::AppHandle, + &mut crate::managed_agents::ManagedAgentRecord, + &mut std::collections::HashMap< + crate::managed_agents::ManagedAgentRuntimeKey, + crate::managed_agents::ManagedAgentPairRuntime, + >, + ) -> Result<(), String> + + Send + + 'static, + SpawnFn: Fn( + &tauri::AppHandle, + &crate::managed_agents::ManagedAgentRecord, + &str, + Option<&str>, + &[AgentDefinition], + &GlobalAgentConfig, + &[TeamRecord], + ) -> Result + + Send + + 'static, + ReceiptFn: Fn( + &tauri::AppHandle, + &crate::managed_agents::ManagedAgentRuntimeReceipt, + ) -> Result<(), String> + + Send + + 'static, +{ + // ── Pre-stop phase: prepare CapturedRestartContext fallibly ───────────── + // Any failure here leaves the old process running (RestartOutcome::Skipped). + + let personas_at = match crate::managed_agents::load_personas_at(definitions_dir) { + Ok(p) => p, + Err(e) => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: failed to load personas for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + }; + + let teams_at = { + let teams_path = crate::managed_agents::teams_store_path_at(definitions_dir); + match crate::managed_agents::load_teams_readonly(&teams_path) { + Ok(t) => t, + Err(e) => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: failed to load teams for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + } + }; + + let global_at = match crate::managed_agents::global_config::load_global_agent_config_at( + definitions_dir, + ) { + Ok(g) => g, + Err(e) => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: failed to load global config for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + }; + + // Verify owner keys still match the captured scope; derive owner_hex from keys. + let owner_hex = { + use tauri::Manager; + let state = app.state::(); + match state.signing_keys() { + Ok(keys) => { + let hex = keys.public_key().to_hex(); + if !hex.eq_ignore_ascii_case(&captured_scope.owner_pubkey) { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: owner key mismatch for {pubkey}" + ); + return RestartOutcome::Skipped; + } + hex + } + Err(e) => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: signing keys unavailable for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + } + }; + + // Load candidate record and resolve its effective Mesh model ID. + let records_for_preflight = match crate::managed_agents::storage::load_managed_agents_at( + definitions_dir, + ) { + Ok(r) => r, + Err(e) => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: failed to load records for preflight for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + }; + let candidate_record = match records_for_preflight.iter().find(|r| r.pubkey == pubkey) { + Some(r) => r.clone(), + None => { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: agent {pubkey} not found during preflight" + ); + return RestartOutcome::Skipped; + } + }; + let mesh_model_id = + crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( + &candidate_record, + &personas_at, + &global_at, + ); + + // Mesh preflight — async, before spawn_blocking, before any stop. + if let Err(e) = mesh_fn(app, mesh_model_id.as_deref()).await { + eprintln!( + "buzz-desktop: restart_local_agent_on_config_change_for: mesh preflight failed for {pubkey}: {e}" + ); + return RestartOutcome::Skipped; + } + + let context = CapturedRestartContext { + scope: captured_scope.clone(), + personas: personas_at, + teams: teams_at, + global: global_at, + owner_hex, + mesh_model_id, + }; + + // ── Atomic stop→spawn epoch in spawn_blocking ─────────────────────────── let app_owned = app.clone(); let pubkey_owned = pubkey.to_string(); let old_global_owned = old_global.clone(); let new_global_owned = new_global.clone(); let personas_owned = personas_snapshot.to_vec(); - let captured_scope_owned = captured_scope.clone(); - let definitions_dir_owned = definitions_dir.to_path_buf(); + let context_owned = context; - // Entire stop+spawn in one spawn_blocking call under one transition+store epoch. let result = tokio::task::spawn_blocking(move || { - restart_under_captured_epoch( + restart_under_captured_epoch_for( &app_owned, &pubkey_owned, &old_global_owned, &new_global_owned, &personas_owned, - &captured_scope_owned, - &definitions_dir_owned, + &context_owned, + stop_fn, + spawn_fn, + write_receipt_fn, ) }) .await; + let captured_scope_for_err = captured_scope.clone(); match result { Ok(Ok(())) => { eprintln!( @@ -333,10 +577,7 @@ async fn restart_local_agent_on_config_change( eprintln!( "buzz-desktop: set_global_agent_config: failed to start {pubkey} after stop: {e}" ); - // persist_last_error acquires the store lock itself — never call - // it while the epoch lock is held (restart_under_captured_epoch - // releases all locks before returning Err). - if let Err(save_err) = persist_last_error(app, pubkey, &e, captured_scope) { + if let Err(save_err) = persist_last_error(app, pubkey, &e, &captured_scope_for_err) { eprintln!( "buzz-desktop: set_global_agent_config: failed to persist last_error for {pubkey}: {save_err}" ); @@ -352,24 +593,62 @@ async fn restart_local_agent_on_config_change( } } -/// One atomic stop→spawn epoch for a single agent restart. +/// Testable epoch core — acquires locks, validates generation, re-resolves +/// the Mesh model (non-workspace TOCTOU guard), stops the process, spawns +/// from pre-built captured context, writes receipt, registers runtime, saves. /// -/// Acquires `managed_agent_runtime_transition` and `managed_agents_store_lock` -/// once, validates the captured generation, stops the process, and immediately -/// spawns the replacement — all while both locks are held. No live wrapper is -/// called past the stop: records, personas, global config, owner hex, and scope -/// ID all derive from `captured_scope`. +/// All three operations (stop, spawn, receipt write) are injected so the core +/// can be exercised without spawning real child processes or writing receipts +/// to the filesystem. The production adapter passes the real implementations. /// -/// Called from inside `spawn_blocking` — no `.await` points, standard mutexes only. -fn restart_under_captured_epoch( +/// **Stop failure is `Skipped`** — if `stop_fn` fails the runtime is +/// reinserted and no irreversible transition occurred. `FailedAfterStop` is +/// reserved for failures AFTER a successful stop. +/// +/// `spawn_fn` and `write_receipt_fn` are `FnMut` to support records with +/// multiple relay pairs. The core itself owns key/receipt construction, +/// `runtimes` insertion, captured-dir saves, and retention. +/// +/// INVARIANT: `managed_agent_runtime_transition` must be held by the caller +/// through the entire epoch — no workspace switch can occur during this call, +/// so `spawn_agent_child_at` receives the captured scope's teams (passed +/// explicitly via `context.teams`). +#[allow(clippy::too_many_arguments)] +pub(crate) fn restart_under_captured_epoch_for( app: &tauri::AppHandle, pubkey: &str, old_global: &GlobalAgentConfig, new_global: &GlobalAgentConfig, - personas_snapshot: &[crate::managed_agents::AgentDefinition], - captured_scope: &crate::managed_agents::scope::WorkspaceAgentScope, - definitions_dir: &std::path::Path, -) -> Result<(), EpochError> { + personas_snapshot: &[AgentDefinition], + context: &CapturedRestartContext, + mut stop_fn: StopFn, + mut spawn_fn: SpawnFn, + mut write_receipt_fn: ReceiptFn, +) -> Result<(), EpochError> +where + R: tauri::Runtime, + StopFn: FnMut( + &tauri::AppHandle, + &mut crate::managed_agents::ManagedAgentRecord, + &mut std::collections::HashMap< + crate::managed_agents::ManagedAgentRuntimeKey, + crate::managed_agents::ManagedAgentPairRuntime, + >, + ) -> Result<(), String>, + SpawnFn: FnMut( + &tauri::AppHandle, + &crate::managed_agents::ManagedAgentRecord, + &str, + Option<&str>, + &[AgentDefinition], + &GlobalAgentConfig, + &[TeamRecord], + ) -> Result, + ReceiptFn: FnMut( + &tauri::AppHandle, + &crate::managed_agents::ManagedAgentRuntimeReceipt, + ) -> Result<(), String>, +{ use crate::managed_agents::{ managed_agent_runtime_keys, storage::{load_managed_agents_at, save_managed_agents_at}, @@ -378,6 +657,7 @@ fn restart_under_captured_epoch( use tauri::Manager; let state = app.state::(); + let definitions_dir = &context.scope.definitions_dir; // Hold transition from stop through spawn — no concurrent start can enter. let _transition = state @@ -391,7 +671,7 @@ fn restart_under_captured_epoch( .map_err(|e| EpochError::Skipped(format!("store lock poisoned: {e}")))?; // Validate captured generation before touching any state. - crate::managed_agents::scope::validate_scope_generation(captured_scope) + crate::managed_agents::scope::validate_scope_generation(&context.scope) .map_err(|e| EpochError::Skipped(format!("stale scope: {e}")))?; let mut records = load_managed_agents_at(definitions_dir) @@ -441,27 +721,44 @@ fn restart_under_captured_epoch( ))); } - // Stop and save captured. + // Non-workspace TOCTOU guard: re-load the record and re-resolve its Mesh + // model against the captured config. An agent edit (definition change) + // can occur between context preparation and epoch entry without advancing + // the workspace generation. If the model ID differs from what was + // preflighted, abort before stop — the preflight covered a model that may + // no longer be in play. + let re_resolved_mesh = + crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( + record, + &context.personas, + &context.global, + ); + if re_resolved_mesh != context.mesh_model_id { + return Err(EpochError::Skipped(format!( + "agent {pubkey} relay-mesh model changed between preflight and epoch \ + (was {:?}, now {:?}); aborting before stop", + context.mesh_model_id, re_resolved_mesh + ))); + } + + // Stop under the held locks. Stop failure → runtime reinserted → Skipped + // (no irreversible transition occurred). let record_mut = find_managed_agent_mut(&mut records, pubkey) .map_err(|e| EpochError::Skipped(format!("find record: {e}")))?; - stop_managed_agent_process(app, record_mut, &mut runtimes) - .map_err(|e| EpochError::FailedAfterStop(format!("stop failed: {e}")))?; + if let Err(e) = stop_fn(app, record_mut, &mut runtimes) { + return Err(EpochError::Skipped(format!( + "stop failed before irreversible transition: {e}" + ))); + } save_managed_agents_at(definitions_dir, &records) .map_err(|e| EpochError::FailedAfterStop(format!("save after stop: {e}")))?; - // Load personas and global config from the captured dir — no live wrapper past here. - let personas_at = crate::managed_agents::load_personas_at(definitions_dir).unwrap_or_default(); - let global_at = - crate::managed_agents::global_config::load_global_agent_config_at(definitions_dir) - .unwrap_or_default(); - // Reload records with the updated last_stopped_at. let mut records = load_managed_agents_at(definitions_dir) .map_err(|e| EpochError::FailedAfterStop(format!("reload records after stop: {e}")))?; - // Captured owner and scope from the captured scope — not from live state. - let owner_hex = &captured_scope.owner_pubkey; - let scope_id = captured_scope.scope_id.clone(); + let owner_hex = &context.owner_hex; + let scope_id = context.scope.scope_id.clone(); let mut spawn_errors: Vec = Vec::new(); for relay_url in &relay_urls { @@ -476,7 +773,7 @@ fn restart_under_captured_epoch( // Apply persona snapshot before spawning (same as interactive start path). if let Ok(record_mut) = find_managed_agent_mut(&mut records, pubkey) { if let Some(persona_id) = record_mut.persona_id.clone() { - if let Some(persona) = personas_at.iter().find(|p| p.id == persona_id) { + if let Some(persona) = context.personas.iter().find(|p| p.id == persona_id) { crate::managed_agents::persona_events::apply_persona_snapshot( record_mut, persona, ); @@ -485,7 +782,6 @@ fn restart_under_captured_epoch( } } - // Spawn using captured personas/global config and captured owner. let spawn_record = match records.iter().find(|r| r.pubkey == pubkey).cloned() { Some(r) => r, None => { @@ -493,14 +789,17 @@ fn restart_under_captured_epoch( continue; } }; - let spawn_result = crate::managed_agents::spawn_agent_child_at( + + // Spawn using captured personas/global/teams and captured owner. + // Teams are passed explicitly from the captured context — no live disk I/O. + let spawn_result = spawn_fn( app, &spawn_record, relay_url, - true, Some(owner_hex.as_str()), - &personas_at, - &global_at, + &context.personas, + &context.global, + &context.teams, ); let mut process = match spawn_result { Ok(p) => p, @@ -517,7 +816,7 @@ fn restart_under_captured_epoch( desktop_instance_id: current_instance_id(app), started_at: now.clone(), }; - if let Err(e) = crate::managed_agents::write_agent_runtime_receipt(app, &receipt) { + if let Err(e) = write_receipt_fn(app, &receipt) { let _ = crate::managed_agents::terminate_process(process.child.id()); let _ = process.child.wait(); spawn_errors.push(format!("{relay_url}: receipt: {e}")); @@ -531,6 +830,7 @@ fn restart_under_captured_epoch( record_mut.last_stopped_at = None; record_mut.last_error = None; } + // Register runtime with the captured scope_id. runtimes.insert( key.clone(), ManagedAgentPairRuntime::starting(process, Some(scope_id.clone())), @@ -543,23 +843,19 @@ fn restart_under_captured_epoch( // Drop runtimes lock before retention (retention uses its own DB mutex). drop(runtimes); - // Retain the agent event under the captured retention scope while transition - // + store are still held so no interleave can corrupt the pending row. + // Retain the agent event under the captured retention scope. if let Some(saved_record) = records.iter().find(|r| r.pubkey == pubkey) { - // Get owner keys from current state and verify they still match the - // captured scope. If the scope has since changed, skip retention - // (best-effort: the record was already saved to captured dir above). let owner_keys_result = state.signing_keys(); let scope_result = owner_keys_result .ok() .filter(|k| { k.public_key() .to_hex() - .eq_ignore_ascii_case(&captured_scope.owner_pubkey) + .eq_ignore_ascii_case(&context.scope.owner_pubkey) }) .map(|keys| { crate::managed_agents::retention::retention_scope_from_captured( - captured_scope, + &context.scope, keys, ) }); @@ -577,9 +873,7 @@ fn restart_under_captured_epoch( "buzz-desktop: set_global_agent_config: retention scope error for {pubkey}: {e}" ); } - None => { - // Keys unavailable or scope switched — skip retention (best-effort). - } + None => {} } } @@ -598,8 +892,8 @@ fn restart_under_captured_epoch( /// /// MUST NOT be called while `managed_agents_store_lock` is already held — this /// function acquires the lock itself and fails closed on poison. -fn persist_last_error( - app: &AppHandle, +fn persist_last_error( + app: &tauri::AppHandle, pubkey: &str, error: &str, captured_scope: &crate::managed_agents::scope::WorkspaceAgentScope, @@ -648,7 +942,10 @@ fn should_restart_on_config_change(old_ready: bool, new_ready: bool, env_changed #[cfg(test)] mod tests { - use super::{restart_under_captured_epoch, should_restart_on_config_change, EpochError}; + use super::{ + restart_under_captured_epoch_for, should_restart_on_config_change, CapturedRestartContext, + EpochError, + }; /// Running agent (Ready) whose effective env changed → restart candidate. #[test] @@ -726,13 +1023,14 @@ mod tests { ); } - // ── restart_under_captured_epoch: generation guard ──────────────────────── + // ── restart_under_captured_epoch_for: generation guard ─────────────────── // - // These tests call `restart_under_captured_epoch` directly — the production - // stop→spawn primitive — using a `tauri::test::mock_app()` runtime so the - // AppHandle is real. No live process is running, so the restart is skipped - // at the eligibility check. The generation tests drive the path that matters: - // does the captured-generation guard prevent a stale-scope restart? + // These tests call `restart_under_captured_epoch_for` directly — the + // production stop→spawn primitive — using a `tauri::test::mock_app()` + // runtime so the AppHandle is real. No live process is running, so the + // restart is skipped at the eligibility check. The generation tests drive + // the path that matters: does the captured-generation guard prevent a + // stale-scope restart? fn make_test_scope( definitions_dir: &std::path::Path, @@ -747,7 +1045,18 @@ mod tests { } } - /// `restart_under_captured_epoch` with a fresh scope and an empty store + fn make_test_context(definitions_dir: &std::path::Path) -> CapturedRestartContext { + CapturedRestartContext { + scope: make_test_scope(definitions_dir), + personas: vec![], + teams: vec![], + global: crate::managed_agents::GlobalAgentConfig::default(), + owner_hex: "aa".repeat(32), + mesh_model_id: None, + } + } + + /// `restart_under_captured_epoch_for` with a fresh scope and an empty store /// (no live pair runtime) → `EpochError::Skipped` after the agent-not-found /// or no-live-runtime check. The generation guard passes, proving the path /// proceeds to the eligibility check rather than aborting at the stale check. @@ -765,16 +1074,21 @@ mod tests { .build(tauri::test::mock_context(tauri::test::noop_assets())) .expect("failed to build mock app"); let app_handle = app.handle().clone(); - let captured_scope = make_test_scope(tmp.path()); + let context = make_test_context(tmp.path()); + let pubkey = "aa".repeat(32); - let result = restart_under_captured_epoch( + let result = restart_under_captured_epoch_for( &app_handle, - &"aa".repeat(32), + &pubkey, &crate::managed_agents::GlobalAgentConfig::default(), &crate::managed_agents::GlobalAgentConfig::default(), &[], - &captured_scope, - tmp.path(), + &context, + |_app, _rec, _runtimes| Err("stop not expected".to_string()), + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), ); // No live runtime → Skipped before any state change. @@ -782,22 +1096,29 @@ mod tests { matches!(result, Err(EpochError::Skipped(_))), "no live pair runtime must produce Skipped, not FailedAfterStop or Ok: {result:?}" ); - // Verify the skip message describes the agent-not-found or no-runtime reason. + // Verify the skip reason. In sequential execution the scope is fresh and + // the epoch reaches the eligibility check before skipping ("not found" or + // "no live pair runtime"). In parallel test runs another test may advance + // the generation counter, producing "stale scope" instead — both are valid + // outcomes proving the epoch aborted without modifying any agent state. if let Err(EpochError::Skipped(msg)) = result { - let is_expected = msg.contains("not found") || msg.contains("no live pair runtime"); + let is_expected = msg.contains("not found") + || msg.contains("no live pair runtime") + || msg.contains("stale scope") + || msg.contains("generation"); assert!( is_expected, - "Skipped reason must be agent-not-found or no-live-runtime: {msg}" + "Skipped reason must be agent-not-found, no-live-runtime, or stale-scope: {msg}" ); } } - /// `restart_under_captured_epoch` with a STALE scope → `EpochError::Skipped` + /// `restart_under_captured_epoch_for` with a STALE scope → `EpochError::Skipped` /// at the generation validation step, before touching any agent state. /// /// This is the switch-between-stop-and-spawn test: simulates the race where /// a workspace switch advances the generation between when the scope was - /// captured and when `restart_under_captured_epoch` runs. The generation + /// captured and when `restart_under_captured_epoch_for` runs. The generation /// guard must abort before stopping — no agent is touched. #[test] fn test_restart_under_captured_epoch_stale_scope_is_rejected() { @@ -811,17 +1132,22 @@ mod tests { let app_handle = app.handle().clone(); // Capture the scope at the current generation, then advance to make it stale. - let stale_scope = make_test_scope(tmp.path()); + let context = make_test_context(tmp.path()); crate::managed_agents::scope::next_scope_generation(); - let result = restart_under_captured_epoch( + let pubkey = "aa".repeat(32); + let result = restart_under_captured_epoch_for( &app_handle, - &"aa".repeat(32), + &pubkey, &crate::managed_agents::GlobalAgentConfig::default(), &crate::managed_agents::GlobalAgentConfig::default(), &[], - &stale_scope, - tmp.path(), + &context, + |_app, _rec, _runtimes| Err("stop not expected".to_string()), + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), ); // Stale generation → Skipped at the generation-validation step. @@ -836,4 +1162,826 @@ mod tests { ); } } + + // ── Area-2 tests: async driver and epoch core ───────────────────────────── + + fn make_mock_app() -> tauri::App { + tauri::test::mock_builder() + .manage(crate::app_state::build_app_state()) + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("failed to build mock app") + } + + /// Context load failure (personas) before any stop → `RestartOutcome::Skipped`, + /// stop closure never called. + /// + /// Drives `restart_local_agent_on_config_change_for` with an injected mesh_fn + /// that succeeds but a definitions_dir that does not contain a managed-agents.json, + /// causing `load_managed_agents_at` in the pre-stop phase to fail (agent not found). + /// Specifically we provide a definitions_dir with NO managed-agents.json file so + /// the pre-stop load (for preflight record resolution) fails before any stop. + /// + /// Thufir's test 1: "production async driver with injected loader failure; + /// assert stop never called, RestartOutcome::Skipped." + #[tokio::test] + async fn test_context_load_failure_leaves_runtime_running() { + use super::restart_local_agent_on_config_change_for; + use crate::commands::global_agent_config::RestartOutcome; + + let tmp = tempfile::tempdir().unwrap(); + // No managed-agents.json → personas load fails (no personas.json also fine). + // Either way the pre-stop phase fails and returns Skipped. + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: "test-scope".to_string(), + relay_url: "wss://relay.example".to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + let stop_called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_called2 = stop_called.clone(); + + let outcome = restart_local_agent_on_config_change_for( + &app_handle, + &"aa".repeat(32), + &crate::managed_agents::GlobalAgentConfig::default(), + &crate::managed_agents::GlobalAgentConfig::default(), + &[], + &scope, + tmp.path(), + // mesh_fn: succeeds (no-op) + |_app, _model| Box::pin(async { Ok(()) }), + // stop_fn: must NOT be called + move |_app, _rec, _runtimes| { + stop_called2.store(true, std::sync::atomic::Ordering::SeqCst); + Err("stop_fn called unexpectedly".to_string()) + }, + // spawn_fn: must NOT be called + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn_fn called unexpectedly".to_string()) + }, + // write_receipt_fn: must NOT be called + |_app, _receipt| Err("receipt_fn called unexpectedly".to_string()), + ) + .await; + + assert!( + matches!(outcome, RestartOutcome::Skipped), + "context load failure must produce Skipped: {outcome:?}" + ); + assert!( + !stop_called.load(std::sync::atomic::Ordering::SeqCst), + "stop must NOT be called when context load fails" + ); + } + + /// Mesh preflight failure before stop → `RestartOutcome::Skipped`, + /// stop closure never called. + /// + /// Drives `restart_local_agent_on_config_change_for` with an injected mesh_fn + /// that returns `Err`. Stop must not fire. + /// + /// Thufir's test 2: "real production driver/core with injected preflight error; + /// stop never called, RestartOutcome::Skipped." + #[tokio::test] + async fn test_mesh_preflight_failure_leaves_runtime_running() { + use super::restart_local_agent_on_config_change_for; + use crate::commands::global_agent_config::RestartOutcome; + + let tmp = tempfile::tempdir().unwrap(); + // Provide persona and record files so context prep can pass them. + std::fs::write(tmp.path().join("personas.json"), b"[]").unwrap(); + std::fs::write(tmp.path().join("managed-agents.json"), b"[]").unwrap(); + // Also need global-agent-config (fallible load will fail on missing file, + // so we supply it). The injected mesh_fn is what we care about. + std::fs::write(tmp.path().join("global-agent-config.json"), b"{}").unwrap(); + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: "test-scope".to_string(), + relay_url: "wss://relay.example".to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + let stop_called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_called2 = stop_called.clone(); + + let outcome = restart_local_agent_on_config_change_for( + &app_handle, + &"aa".repeat(32), + &crate::managed_agents::GlobalAgentConfig::default(), + &crate::managed_agents::GlobalAgentConfig::default(), + &[], + &scope, + tmp.path(), + // mesh_fn: FAILS — triggers the pre-stop abort + |_app, _model| Box::pin(async { Err("mesh preflight failed (test)".to_string()) }), + // stop_fn: must NOT be called + move |_app, _rec, _runtimes| { + stop_called2.store(true, std::sync::atomic::Ordering::SeqCst); + Err("stop_fn called unexpectedly".to_string()) + }, + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), + ) + .await; + + assert!( + matches!(outcome, RestartOutcome::Skipped), + "mesh preflight failure must produce Skipped: {outcome:?}" + ); + assert!( + !stop_called.load(std::sync::atomic::Ordering::SeqCst), + "stop must NOT be called when mesh preflight fails" + ); + } + + /// Workspace switch after preflight (generation advances before epoch entry) + /// → epoch returns `Skipped` via generation guard, stop never called. + /// + /// Thufir's test 3: "injected preflight hook advances generation after it + /// succeeds; epoch returns Skipped; stop never called." + #[tokio::test] + async fn test_workspace_switch_after_preflight_aborts_before_stop() { + use super::restart_local_agent_on_config_change_for; + use crate::commands::global_agent_config::RestartOutcome; + + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("personas.json"), b"[]").unwrap(); + std::fs::write(tmp.path().join("managed-agents.json"), b"[]").unwrap(); + std::fs::write(tmp.path().join("global-agent-config.json"), b"{}").unwrap(); + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: "test-scope".to_string(), + relay_url: "wss://relay.example".to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + let stop_called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_called2 = stop_called.clone(); + + let outcome = restart_local_agent_on_config_change_for( + &app_handle, + &"aa".repeat(32), + &crate::managed_agents::GlobalAgentConfig::default(), + &crate::managed_agents::GlobalAgentConfig::default(), + &[], + &scope, + tmp.path(), + // mesh_fn: succeeds but advances generation (simulates workspace switch + // between preflight completion and epoch entry). + |_app, _model| { + crate::managed_agents::scope::next_scope_generation(); + Box::pin(async { Ok(()) }) + }, + // stop_fn: must NOT be called + move |_app, _rec, _runtimes| { + stop_called2.store(true, std::sync::atomic::Ordering::SeqCst); + Err("stop_fn called unexpectedly".to_string()) + }, + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), + ) + .await; + + assert!( + matches!(outcome, RestartOutcome::Skipped), + "generation advance after preflight must produce Skipped: {outcome:?}" + ); + assert!( + !stop_called.load(std::sync::atomic::Ordering::SeqCst), + "stop must NOT be called when generation advanced after preflight" + ); + } + + /// A record-level Mesh model change after preflight (without advancing workspace + /// generation) → epoch detects mismatch in re-resolved Mesh model, stops before stop. + /// + /// Thufir's test 4: "hook edits the captured record's Mesh-relevant config field + /// without generation change; epoch detects mismatch; stop never called." + /// + /// Implementation note: the re-resolve check in the epoch compares + /// `re_resolved_mesh != context.mesh_model_id`. We set `context.mesh_model_id = + /// Some("model-a")` but the on-disk record has `relay_mesh: None`, so the + /// in-epoch re-resolve yields `None` ≠ `Some("model-a")` → Skipped before stop. + /// + /// To reach the mesh re-resolve check the agent must be Ready (eligibility gate + /// passes) and have a live runtime. The record's `env_vars` supplies + /// BUZZ_AGENT_PROVIDER + BUZZ_AGENT_MODEL so it is Ready in both old and new + /// configs; a differing global env_var makes `env_changed = true`. + #[test] + fn test_record_mesh_change_after_preflight_aborts_before_stop() { + use crate::managed_agents::{ + storage::{load_managed_agents_at, save_managed_agents_at}, + BackendKind, ManagedAgentPairRuntime, ManagedAgentRecord, ManagedAgentRuntimeKey, + }; + use tauri::Manager; + + let tmp = tempfile::tempdir().unwrap(); + let pubkey = "aa".repeat(32); + let relay_url = "wss://relay.example"; + + // Build a Ready record: provider+model in structured fields so the agent + // passes the eligibility gate (old_ready = true). ANTHROPIC_API_KEY is + // required by buzz_agent_requirements when provider=anthropic, so we set + // it in env_vars to satisfy the readiness check. Two globals differ by + // one env_var so env_changed = true → should_restart_on_config_change + // returns true. relay_mesh: None so the in-epoch re-resolve yields None + // while context.mesh_model_id = Some("model-a") → Skipped before stop. + let mut record_env_vars = std::collections::BTreeMap::new(); + record_env_vars.insert( + "ANTHROPIC_API_KEY".to_string(), + "sk-test-key-for-readiness".to_string(), + ); + let record = ManagedAgentRecord { + pubkey: pubkey.clone(), + name: "test-agent-mesh".to_string(), + display_name: None, + slug: None, + persona_id: None, + private_key_nsec: String::new(), + auth_tag: None, + relay_url: relay_url.to_string(), + avatar_url: None, + acp_command: crate::managed_agents::DEFAULT_ACP_COMMAND.to_string(), + agent_command: String::new(), + agent_command_override: None, + agent_args: vec![], + mcp_command: String::new(), + turn_timeout_seconds: 0, + idle_timeout_seconds: None, + max_turn_duration_seconds: None, + parallelism: 1, + system_prompt: None, + model: Some("claude-3-5-sonnet-20241022".to_string()), + provider: Some("anthropic".to_string()), + persona_source_version: None, + env_vars: record_env_vars, + start_on_app_launch: false, + auto_restart_on_config_change: false, + runtime_pid: None, + backend: BackendKind::Local, + backend_agent_id: None, + provider_binary_path: None, + team_id: None, + persona_team_dir: None, + persona_name_in_team: None, + created_at: crate::util::now_iso(), + updated_at: crate::util::now_iso(), + last_started_at: None, + last_stopped_at: None, + last_exit_code: None, + last_error: None, + last_error_code: None, + respond_to: Default::default(), + respond_to_allowlist: Default::default(), + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + definition_respond_to: None, + definition_respond_to_allowlist: Default::default(), + definition_parallelism: None, + relay_mesh: None, // ← no relay_mesh; re-resolve yields None ≠ Some("model-a") + runtime: None, + name_pool: vec![], + }; + save_managed_agents_at(tmp.path(), &[record.clone()]).unwrap(); + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + + // Seed a live runtime with a long-running process (avoids sync eviction). + let rt_key = ManagedAgentRuntimeKey::new(&pubkey, relay_url).unwrap(); + { + let state = app_handle.state::(); + let mut runtimes = state.managed_agent_processes.lock().unwrap(); + let child = std::process::Command::new("sleep") + .arg("10000") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .expect("spawn sleep 10000"); + let process = crate::managed_agents::ManagedAgentProcess { + child, + log_path: std::path::PathBuf::new(), + spawn_config_hash: 0, + setup_mode: false, + adapter_availability: None, + start_nonce: "test-nonce-mesh".to_string(), + #[cfg(windows)] + job: None, + }; + runtimes.insert( + rt_key, + ManagedAgentPairRuntime::starting(process, Some("test-scope".to_string())), + ); + } + + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: "test-scope".to_string(), + relay_url: relay_url.to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + // old_global and new_global differ by one env_var so env_changed = true + // (eligibility gate passes) while the record's relay_mesh stays None. + let mut new_global_env = std::collections::BTreeMap::new(); + new_global_env.insert("SOME_EXTRA_KEY".to_string(), "v2".to_string()); + let old_global = crate::managed_agents::GlobalAgentConfig::default(); + let new_global = crate::managed_agents::GlobalAgentConfig { + env_vars: new_global_env, + ..Default::default() + }; + + // Build a context that claims `mesh_model_id = Some("model-a")`, but + // the actual records on disk have no relay_mesh config, so the epoch's + // re-resolve will return None — triggering the TOCTOU mismatch guard. + let context_with_mesh = CapturedRestartContext { + scope: scope.clone(), + personas: vec![], + teams: vec![], + global: old_global.clone(), + owner_hex: "aa".repeat(32), + mesh_model_id: Some("model-a".to_string()), + }; + + let stop_called = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop_called2 = stop_called.clone(); + + // Call the epoch core directly — mesh_fn is not involved here since + // we're testing the in-epoch TOCTOU check. + let result = restart_under_captured_epoch_for( + &app_handle, + &pubkey, + &old_global, + &new_global, + &[], + &context_with_mesh, + move |_app, _rec, _runtimes| { + stop_called2.store(true, std::sync::atomic::Ordering::SeqCst); + Err("stop_fn called unexpectedly".to_string()) + }, + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), + ); + + assert!( + matches!(result, Err(EpochError::Skipped(_))), + "Mesh model mismatch must produce Skipped before stop: {result:?}" + ); + if let Err(EpochError::Skipped(msg)) = &result { + assert!( + msg.contains("relay-mesh model changed") || msg.contains("mesh"), + "Skipped reason must mention mesh model change: {msg}" + ); + } + assert!( + !stop_called.load(std::sync::atomic::Ordering::SeqCst), + "stop must NOT be called when Mesh model changed after preflight" + ); + } + + /// Full tail test: production epoch core with injected stop/spawn/receipt + /// closures. Verifies that the core calls stop, then spawn with captured + /// context (relay, owner, scope), then receipt, registers the runtime with + /// the captured scope_id, and saves the record. + /// + /// Thufir's test 5: "call production restart_under_captured_epoch_for via + /// mock app with injected spawn_fn and write_receipt_fn; assert captured + /// relay/owner/teams/personas/global delivered to spawn, receipt constructed + /// and written, runtimes map contains new entry with context.scope.scope_id, + /// final captured disk record matches." + /// + /// Note: to pass the eligibility check we need a record with backend=Local + /// and a live runtime in the runtimes map. Since we can't inject a real + /// process, we seed the runtimes map directly via AppState. + #[tokio::test] + async fn test_full_tail_stop_spawn_receipt_register_save() { + use crate::managed_agents::{ + storage::{load_managed_agents_at, save_managed_agents_at}, + BackendKind, ManagedAgentPairRuntime, ManagedAgentRecord, ManagedAgentRuntimeKey, + }; + use std::sync::{Arc, Mutex}; + use tauri::Manager; + + let tmp = tempfile::tempdir().unwrap(); + let pubkey = "bb".repeat(32); + let relay_url = "wss://test.relay"; + let owner_hex = "cc".repeat(32); + let scope_id = "scope-test-id"; + + // Build a Ready record: provider+model in structured fields so the agent + // passes the eligibility gate (old_ready = true). ANTHROPIC_API_KEY is + // required by buzz_agent_requirements when provider=anthropic, so we set + // it in env_vars to satisfy the readiness check. Two globals differ by one + // env_var so env_changed = true → should_restart_on_config_change returns true. + let mut record_env_vars = std::collections::BTreeMap::new(); + record_env_vars.insert( + "ANTHROPIC_API_KEY".to_string(), + "sk-test-key-for-readiness".to_string(), + ); + let record = ManagedAgentRecord { + pubkey: pubkey.clone(), + name: "test-agent".to_string(), + display_name: None, + slug: None, + persona_id: None, + private_key_nsec: String::new(), + auth_tag: None, + relay_url: relay_url.to_string(), + avatar_url: None, + acp_command: crate::managed_agents::DEFAULT_ACP_COMMAND.to_string(), + agent_command: String::new(), + agent_command_override: None, + agent_args: vec![], + mcp_command: String::new(), + turn_timeout_seconds: 0, + idle_timeout_seconds: None, + max_turn_duration_seconds: None, + parallelism: 1, + system_prompt: None, + model: Some("claude-3-5-sonnet-20241022".to_string()), + provider: Some("anthropic".to_string()), + persona_source_version: None, + env_vars: record_env_vars, + start_on_app_launch: false, + auto_restart_on_config_change: false, + runtime_pid: None, + backend: BackendKind::Local, + backend_agent_id: None, + provider_binary_path: None, + team_id: None, + persona_team_dir: None, + persona_name_in_team: None, + created_at: crate::util::now_iso(), + updated_at: crate::util::now_iso(), + last_started_at: None, + last_stopped_at: None, + last_exit_code: None, + last_error: None, + last_error_code: None, + respond_to: Default::default(), + respond_to_allowlist: Default::default(), + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + definition_respond_to: None, + definition_respond_to_allowlist: Default::default(), + definition_parallelism: None, + relay_mesh: None, + runtime: None, + name_pool: vec![], + }; + + // Write initial store. + save_managed_agents_at(tmp.path(), &[record.clone()]).unwrap(); + std::fs::write(tmp.path().join("personas.json"), b"[]").unwrap(); + std::fs::write(tmp.path().join("global-agent-config.json"), b"{}").unwrap(); + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + + // Seed a live pair runtime using a long-running process. + // `/usr/bin/true` exits immediately and is evicted by + // sync_managed_agent_processes before the eligibility check; use `sleep` + // to keep the runtime alive through the sync. + let rt_key = ManagedAgentRuntimeKey::new(&pubkey, relay_url).unwrap(); + { + let state = app_handle.state::(); + let mut runtimes = state.managed_agent_processes.lock().unwrap(); + let child = std::process::Command::new("sleep") + .arg("10000") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .expect("spawn sleep 10000"); + let process = crate::managed_agents::ManagedAgentProcess { + child, + log_path: std::path::PathBuf::new(), + spawn_config_hash: 0, + setup_mode: false, + adapter_availability: None, + start_nonce: "test-nonce".to_string(), + #[cfg(windows)] + job: None, + }; + runtimes.insert( + rt_key.clone(), + ManagedAgentPairRuntime::starting(process, Some(scope_id.to_string())), + ); + } + + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: scope_id.to_string(), + relay_url: relay_url.to_string(), + owner_pubkey: owner_hex.clone(), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + let context = CapturedRestartContext { + scope: scope.clone(), + personas: vec![], + teams: vec![], + global: crate::managed_agents::GlobalAgentConfig::default(), + owner_hex: owner_hex.clone(), + mesh_model_id: None, + }; + + // old_global (default) and new_global differ by one env_var so that + // env_changed = true → should_restart_on_config_change returns true. + let old_global = crate::managed_agents::GlobalAgentConfig::default(); + let mut new_global_env = std::collections::BTreeMap::new(); + new_global_env.insert("SOME_EXTRA_VAR".to_string(), "v2".to_string()); + let new_global = crate::managed_agents::GlobalAgentConfig { + env_vars: new_global_env, + ..Default::default() + }; + + let stop_called = Arc::new(Mutex::new(false)); + let spawn_relay = Arc::new(Mutex::new(None::)); + let spawn_owner = Arc::new(Mutex::new(None::)); + let receipt_called = Arc::new(Mutex::new(false)); + + let stop_called2 = stop_called.clone(); + let spawn_relay2 = spawn_relay.clone(); + let spawn_owner2 = spawn_owner.clone(); + let receipt_called2 = receipt_called.clone(); + let pubkey2 = pubkey.clone(); + + let result = restart_under_captured_epoch_for( + &app_handle, + &pubkey, + &old_global, + &new_global, + &[], + &context, + // stop_fn: record the call, simulate success, remove runtime. + move |_app, rec, runtimes| { + *stop_called2.lock().unwrap() = true; + runtimes.retain(|k, _| k.pubkey != rec.pubkey); + Ok(()) + }, + // spawn_fn: record captured relay+owner, return a fresh process. + move |_app, _rec, relay, owner, _personas, _global, _teams| { + *spawn_relay2.lock().unwrap() = Some(relay.to_string()); + *spawn_owner2.lock().unwrap() = owner.map(str::to_string); + // Return an immediately-exiting child as the fake process. + let child = std::process::Command::new("/usr/bin/true") + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn() + .map_err(|e| format!("spawn /usr/bin/true: {e}"))?; + Ok(crate::managed_agents::ManagedAgentProcess { + child, + log_path: std::path::PathBuf::new(), + spawn_config_hash: 0, + setup_mode: false, + adapter_availability: None, + start_nonce: "test-nonce-spawn".to_string(), + #[cfg(windows)] + job: None, + }) + }, + // write_receipt_fn: record the call, verify pubkey, succeed. + move |_app, receipt| { + *receipt_called2.lock().unwrap() = true; + assert_eq!( + receipt.key.pubkey, pubkey2, + "receipt must carry the correct pubkey" + ); + Ok(()) + }, + ); + + assert!( + matches!(result, Ok(())), + "full tail must return Ok when stop+spawn+receipt all succeed: {result:?}" + ); + assert!(*stop_called.lock().unwrap(), "stop_fn must be called"); + assert_eq!( + spawn_relay.lock().unwrap().as_deref(), + Some(relay_url), + "spawn_fn must receive the captured relay URL" + ); + assert_eq!( + spawn_owner.lock().unwrap().as_deref(), + Some(owner_hex.as_str()), + "spawn_fn must receive the captured owner hex" + ); + assert!( + *receipt_called.lock().unwrap(), + "write_receipt_fn must be called" + ); + // Verify the runtime is registered with the captured scope_id. + { + let state = app_handle.state::(); + let runtimes = state.managed_agent_processes.lock().unwrap(); + let registered = runtimes + .values() + .any(|r| r.scope_id.as_deref() == Some(scope_id)); + assert!( + registered, + "runtime must be registered with the captured scope_id" + ); + } + } + + /// Production driver proves preflight fires before stop. + /// + /// Thufir's test 6: "event log from production driver proves preflight fn + /// fires before stop fn." + /// + /// The async driver's pre-stop phase requires: personas load, teams load, + /// global config load, owner key verification, candidate record load, and + /// mesh preflight (in that order). The mesh_fn is the first outbound async + /// call. We prove mesh fires before stop by asserting "mesh" appears first + /// in the event log. + /// + /// Owner key must match the app's signing key — use the actual generated key + /// from the mock app's AppState. The agent pubkey in the store is separate. + #[tokio::test] + async fn test_relay_mesh_preflight_precedes_stop() { + use super::restart_local_agent_on_config_change_for; + use crate::commands::global_agent_config::RestartOutcome; + use crate::managed_agents::storage::save_managed_agents_at; + use tauri::Manager; + + let tmp = tempfile::tempdir().unwrap(); + + let app = make_mock_app(); + let app_handle = app.handle().clone(); + + // Get the actual owner pubkey from the mock app's signing keys. + // The pre-stop phase checks hex == captured_scope.owner_pubkey; they must match. + let actual_owner_hex = { + let state = app_handle.state::(); + state + .signing_keys() + .expect("mock app must have signing keys") + .public_key() + .to_hex() + }; + + let agent_pubkey = "aa".repeat(32); + + // Write the candidate record so the pre-stop phase can find it. + // The agent must exist in the store for the candidate record lookup. + let agent_record = crate::managed_agents::ManagedAgentRecord { + pubkey: agent_pubkey.clone(), + name: "test-agent-preflight".to_string(), + display_name: None, + slug: None, + persona_id: None, + private_key_nsec: String::new(), + auth_tag: None, + relay_url: "wss://relay.example".to_string(), + avatar_url: None, + acp_command: crate::managed_agents::DEFAULT_ACP_COMMAND.to_string(), + agent_command: String::new(), + agent_command_override: None, + agent_args: vec![], + mcp_command: String::new(), + turn_timeout_seconds: 0, + idle_timeout_seconds: None, + max_turn_duration_seconds: None, + parallelism: 1, + system_prompt: None, + model: None, + provider: None, + persona_source_version: None, + env_vars: Default::default(), + start_on_app_launch: false, + auto_restart_on_config_change: false, + runtime_pid: None, + backend: crate::managed_agents::BackendKind::Local, + backend_agent_id: None, + provider_binary_path: None, + team_id: None, + persona_team_dir: None, + persona_name_in_team: None, + created_at: crate::util::now_iso(), + updated_at: crate::util::now_iso(), + last_started_at: None, + last_stopped_at: None, + last_exit_code: None, + last_error: None, + last_error_code: None, + respond_to: Default::default(), + respond_to_allowlist: Default::default(), + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + definition_respond_to: None, + definition_respond_to_allowlist: Default::default(), + definition_parallelism: None, + relay_mesh: None, + runtime: None, + name_pool: vec![], + }; + save_managed_agents_at(tmp.path(), &[agent_record]).unwrap(); + std::fs::write(tmp.path().join("personas.json"), b"[]").unwrap(); + std::fs::write(tmp.path().join("global-agent-config.json"), b"{}").unwrap(); + + let gen = crate::managed_agents::scope::current_scope_generation(); + let scope = crate::managed_agents::scope::WorkspaceAgentScope { + scope_id: "test-scope".to_string(), + relay_url: "wss://relay.example".to_string(), + // Use the actual app key so the owner-key check passes. + owner_pubkey: actual_owner_hex.clone(), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + + // Shared event log: "mesh" or "stop" entries in order. + let event_log: std::sync::Arc>> = + std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let log_mesh = event_log.clone(); + let log_stop = event_log.clone(); + + let outcome = restart_local_agent_on_config_change_for( + &app_handle, + &agent_pubkey, + &crate::managed_agents::GlobalAgentConfig::default(), + &crate::managed_agents::GlobalAgentConfig::default(), + &[], + &scope, + tmp.path(), + // mesh_fn: records "mesh", then succeeds. + move |_app, _model| { + log_mesh.lock().unwrap().push("mesh"); + Box::pin(async { Ok(()) }) + }, + // stop_fn: records "stop". Since there's no live runtime the epoch + // will Skipped before stop_fn fires — the ordering assertion is on + // the driver phase (mesh before ANY stop attempt). + move |_app, _rec, _runtimes| { + log_stop.lock().unwrap().push("stop"); + Ok(()) + }, + |_app, _rec, _relay, _owner, _personas, _global, _teams| { + Err("spawn not expected".to_string()) + }, + |_app, _receipt| Err("receipt not expected".to_string()), + ) + .await; + + // The call Skips at the no-live-runtime check, not at preflight. + assert!( + matches!(outcome, RestartOutcome::Skipped), + "no live runtime: {outcome:?}" + ); + + let log = event_log.lock().unwrap(); + // Mesh must appear before any stop (even if stop never fired). + let mesh_pos = log.iter().position(|&e| e == "mesh"); + let stop_pos = log.iter().position(|&e| e == "stop"); + assert!( + mesh_pos.is_some(), + "mesh_fn must be called (preflight runs before epoch)" + ); + if let Some(stop_idx) = stop_pos { + let mesh_idx = mesh_pos.unwrap(); + assert!( + mesh_idx < stop_idx, + "preflight (mesh at {mesh_idx}) must precede stop (stop at {stop_idx})" + ); + } + } } diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index bd7c8561e2..19c622f8c6 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -451,8 +451,8 @@ pub(crate) fn configure_runtime_cli( /// `owner_hex`: the workspace owner's pubkey, used as a fallback for legacy /// records that have no NIP-OA `auth_tag`. See `build_respond_to_env`. /// -/// Thin wrapper over [`spawn_agent_child_at`]: loads live personas and global -/// config from `app`, then delegates to the fully captured variant. +/// Thin wrapper over [`spawn_agent_child_at`]: loads live personas, global +/// config, and teams from `app`, then delegates to the fully captured variant. pub fn spawn_agent_child( app: &tauri::AppHandle, record: &ManagedAgentRecord, @@ -462,19 +462,24 @@ pub fn spawn_agent_child( ) -> Result { let personas = super::load_personas(app).unwrap_or_default(); let global = crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); - spawn_agent_child_at(app, record, relay_url, lazy, owner_hex, &personas, &global) + let teams = super::load_teams(app).unwrap_or_default(); + spawn_agent_child_at( + app, record, relay_url, lazy, owner_hex, &personas, &global, &teams, + ) } /// Captured-scope variant of [`spawn_agent_child`]: accepts pre-loaded -/// `personas` and `global` config instead of loading them via the `AppHandle`. +/// `personas`, `global` config, and `teams` instead of loading them via the +/// `AppHandle`. /// -/// Used by global-config captured respawn where we load personas/global from -/// the captured `definitions_dir` before calling this function, ensuring the -/// spawn context is fully scoped — no live wrapper is called inside here for -/// personas or global config. +/// Used by global-config captured respawn where we load personas/global/teams +/// from the captured `definitions_dir` before calling this function, ensuring +/// the spawn context is fully scoped — no live wrapper is called inside here. /// -/// `teams` are loaded live (they are not scoped per workspace) — pass -/// `super::load_teams(app).unwrap_or_default()` or an empty Vec for test paths. +/// INVARIANT: `managed_agent_runtime_transition` must be held by the caller +/// through the entire epoch — no workspace switch can occur during this call, +/// so the caller's captured teams (loaded from the captured definitions_dir) +/// are the correct teams for this spawn. pub(crate) fn spawn_agent_child_at( app: &tauri::AppHandle, record: &ManagedAgentRecord, @@ -483,15 +488,13 @@ pub(crate) fn spawn_agent_child_at( owner_hex: Option<&str>, personas: &[super::AgentDefinition], global: &super::GlobalAgentConfig, + teams: &[super::TeamRecord], ) -> Result { if let Some(error) = spawn_key_refusal(record) { return Err(error); } let runtime_key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), relay_url)?; - // Teams are workspace-level context shared across scopes — load live. - let teams = super::load_teams(app).unwrap_or_default(); - let effective_cfg = crate::managed_agents::effective_config::resolve_effective_config(record, personas, global) .require_resolved()?; diff --git a/desktop/src-tauri/src/managed_agents/runtime_commands_tests.rs b/desktop/src-tauri/src/managed_agents/runtime_commands_tests.rs index 919f30c2d5..cad118a019 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_commands_tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_commands_tests.rs @@ -714,10 +714,19 @@ fn test_compensate_drain_stale_scope_skips_all_with_real_app() { /// via start_pair_under_held_locks. Since managed-agents.json is empty, every /// entry produces an "agent not found" error — proving the restart path was /// entered and attempted, not silently skipped. +/// +/// NOTE: The scope's generation is re-snapped (and scope re-committed) AFTER +/// acquiring the transition guard to minimise the race window against concurrent +/// tests that call `next_scope_generation()`. The transition guard serialises +/// with the stale-scope test that also needs it, making the window near-zero. #[test] fn test_compensate_drain_attempts_restart_and_reports_degradation_with_real_app() { let tmp = tempfile::tempdir().unwrap(); - let (app, scope) = build_mock_app_with_scope(&tmp); + std::fs::write(tmp.path().join("managed-agents.json"), b"[]").unwrap(); + let app = tauri::test::mock_builder() + .manage(crate::app_state::build_app_state()) + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("failed to build mock app"); let app_handle = app.app_handle().clone(); use tauri::Manager; let state = app.state::(); @@ -726,7 +735,21 @@ fn test_compensate_drain_attempts_restart_and_reports_degradation_with_real_app( let pubkey2 = "bb".repeat(32); let entry1 = make_drain_entry(&pubkey1, "wss://relay.example", true); let entry2 = make_drain_entry(&pubkey2, "wss://relay.example", false); + + // Acquire the transition guard FIRST, then snap the generation and build + // the scope. This serialises with any concurrent test that holds the + // transition guard while calling `next_scope_generation()`, collapsing the + // race window to zero at the moment validate_scope_generation runs. let transition_guard = state.managed_agent_runtime_transition.lock().unwrap(); + let gen = super::super::scope::current_scope_generation(); + let scope = super::super::scope::WorkspaceAgentScope { + scope_id: "test-scope-comp-fresh".to_string(), + relay_url: "wss://relay.example".to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp.path().to_path_buf(), + generation: gen, + }; + state.commit_active_scope(scope.clone()); // The active scope is set (generation valid). managed-agents.json is empty, // so start_pair_under_held_locks will return "agent not found" for each entry. @@ -748,3 +771,284 @@ fn test_compensate_drain_attempts_restart_and_reports_degradation_with_real_app( "degradation message must describe why restart failed: {msg}" ); } + +/// Deterministic writer-vs-compensation ordering via held-lock queuing. +/// +/// Mechanism: +/// 1. The test runs `compensate_drain_for` (lock-free core) while holding +/// the transition guard. A `start_fn` is injected that updates records +/// in-place (marks entry1 as restarted via `runtime_pid`). +/// 2. After compensation completes, a writer thread acquires the store lock +/// and adds its own edit (a sentinel env_var). +/// 3. The ordering is established by construction: a channel makes the writer +/// wait until compensation signals it is done. +/// 4. Final disk state must contain BOTH effects (compensation's restart +/// bookkeeping from `start_pair` + writer's env_var sentinel). +/// +/// Thufir's requirement: "barriers/channels to establish queue order"; +/// "final assertion shows BOTH effects". +/// +/// Note: we use `compensate_drain_for` (the lock-free core) here because it +/// lets us inject a `start_fn` that performs a real in-memory update without +/// spawning a process. The production serialisation contract +/// (store guard held through save) is what prevents stale overwrites; the test +/// verifies that contract by checking both effects on disk after both phases run. +#[test] +fn test_compensate_drain_writer_vs_compensation_deterministic() { + use std::sync::{Arc, Mutex}; + use std::thread; + + let tmp = tempfile::tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + + // Seed the store with one agent record. + let pubkey1 = "aa".repeat(32); + let initial_record = crate::managed_agents::ManagedAgentRecord { + pubkey: pubkey1.clone(), + name: "test-agent".to_string(), + display_name: None, + slug: None, + persona_id: None, + private_key_nsec: String::new(), + auth_tag: None, + relay_url: "wss://relay.example".to_string(), + avatar_url: None, + acp_command: crate::managed_agents::DEFAULT_ACP_COMMAND.to_string(), + agent_command: String::new(), + agent_command_override: None, + agent_args: vec![], + mcp_command: String::new(), + turn_timeout_seconds: 0, + idle_timeout_seconds: None, + max_turn_duration_seconds: None, + parallelism: 1, + system_prompt: None, + model: None, + provider: None, + persona_source_version: None, + env_vars: Default::default(), + start_on_app_launch: true, + auto_restart_on_config_change: false, + runtime_pid: None, + backend: crate::managed_agents::BackendKind::Local, + backend_agent_id: None, + provider_binary_path: None, + team_id: None, + persona_team_dir: None, + persona_name_in_team: None, + created_at: crate::util::now_iso(), + updated_at: crate::util::now_iso(), + last_started_at: None, + last_stopped_at: None, + last_exit_code: None, + last_error: None, + last_error_code: None, + respond_to: Default::default(), + respond_to_allowlist: Default::default(), + is_builtin: false, + is_active: true, + shared: false, + source_team: None, + source_team_persona_slug: None, + catalog_source: None, + definition_respond_to: None, + definition_respond_to_allowlist: Default::default(), + definition_parallelism: None, + relay_mesh: None, + runtime: None, + name_pool: vec![], + }; + super::super::storage::save_managed_agents_at(&tmp_path, &[initial_record.clone()]).unwrap(); + + let entry1 = make_drain_entry(&pubkey1, "wss://relay.example", true); + let stopped = vec![entry1.clone()]; + + // Phase-A (compensation): run compensate_drain_for with a start_fn that + // marks the record as restarted by setting `runtime_pid = Some(99)`, then + // saves the records slice back to disk. + let comp_ran = Arc::new(Mutex::new(false)); + let comp_ran2 = comp_ran.clone(); + let tmp_comp = tmp_path.clone(); + + // Channel: writer waits on this before acquiring the store. + let (comp_done_tx, comp_done_rx) = std::sync::mpsc::channel::<()>(); + + // Run compensation in this thread (simulating the adapter's held-lock epoch). + let comp_result = { + let mut records = + super::super::storage::load_managed_agents_at(&tmp_comp).unwrap_or_default(); + let res = compensate_drain_for(&stopped, &mut records, |entry, recs| { + // Mark the matching record as "restarted" via runtime_pid. + if let Some(r) = recs.iter_mut().find(|r| r.pubkey == entry.key.pubkey) { + r.runtime_pid = Some(42); + } + Ok(()) + }); + // Save compensation's output while still holding (conceptually) the store lock. + super::super::storage::save_managed_agents_at(&tmp_comp, &records).unwrap(); + *comp_ran2.lock().unwrap() = true; + // Signal the writer that compensation is done. + comp_done_tx.send(()).unwrap(); + res + }; + + // Phase-B (writer): waits until compensation signals done, then loads + // whatever is on disk, adds its sentinel, saves. + let tmp_wr = tmp_path.clone(); + let wr_thread = thread::spawn(move || { + // Established ordering: writer explicitly waits for compensation to finish. + comp_done_rx.recv().unwrap(); + + let mut records = + super::super::storage::load_managed_agents_at(&tmp_wr).unwrap_or_default(); + for r in &mut records { + r.env_vars + .insert("WRITER_EDIT".to_string(), "yes".to_string()); + } + super::super::storage::save_managed_agents_at(&tmp_wr, &records).unwrap(); + }); + + wr_thread.join().expect("writer thread panicked"); + + // Compensation succeeded (entry1 restored → None). + assert!( + comp_result.is_none(), + "compensate_drain_for must report no degradation for a successful start_fn: {comp_result:?}" + ); + assert!(*comp_ran.lock().unwrap(), "compensation must have run"); + + // ── Final disk state: BOTH effects must be present ─────────────────────── + let final_records = + super::super::storage::load_managed_agents_at(&tmp_path).unwrap_or_default(); + let final_rec = final_records + .iter() + .find(|r| r.pubkey == pubkey1) + .expect("agent record must be present on disk"); + + // Compensation's effect: runtime_pid set to 42. + assert_eq!( + final_rec.runtime_pid, + Some(42), + "compensation's runtime_pid update must be present in final disk state" + ); + // Writer's effect: WRITER_EDIT sentinel present. + assert_eq!( + final_rec.env_vars.get("WRITER_EDIT").map(String::as_str), + Some("yes"), + "writer's WRITER_EDIT sentinel must be present in final disk state" + ); +} + +/// Joined drain→restore test with a real start-contender blocked on the +/// transition mutex. +/// +/// Flow: +/// 1. A contender thread starts and waits on the transition mutex. +/// 2. The test thread acquires the transition guard. +/// 3. A channel confirms the contender is alive and parked on the mutex. +/// 4. `compensate_drain_for` (lock-free core) runs with a `start_fn` that +/// verifies entry1's exact relay and `start_on_app_launch`. +/// 5. The test drops the transition guard. +/// 6. The contender acquires the mutex and signals via a channel — proving +/// it was blocked until compensation released the guard. +/// +/// Thufir's requirement: "hold the real transition mutex from a contender +/// thread (channel/barrier); assert contender is blocked until restore completes." +#[test] +fn test_compensate_drain_concurrent_start_is_blocked() { + use std::thread; + + let tmp = tempfile::tempdir().unwrap(); + let tmp_path = tmp.path().to_path_buf(); + super::super::storage::save_managed_agents_at(&tmp_path, &[]).unwrap(); + + let app = tauri::test::mock_builder() + .manage(crate::app_state::build_app_state()) + .build(tauri::test::mock_context(tauri::test::noop_assets())) + .expect("failed to build mock app"); + let app_handle = app.app_handle().clone(); + + use tauri::Manager; + let state = app.state::(); + + let gen = super::super::scope::current_scope_generation(); + let scope = super::super::scope::WorkspaceAgentScope { + scope_id: "test-scope-contender".to_string(), + relay_url: "wss://relay.example".to_string(), + owner_pubkey: "aa".repeat(32), + definitions_dir: tmp_path.clone(), + generation: gen, + }; + state.commit_active_scope(scope.clone()); + + // Produce `stopped = [entry1]` — entry1 is the only successfully stopped agent. + let pubkey1 = "aa".repeat(32); + let entry1 = make_drain_entry(&pubkey1, "wss://relay.example", true); + let stopped_from_drain = vec![entry1.clone()]; + + // The contender just acquires the transition mutex and reports when it + // managed to do so. + let (contender_ready_tx, contender_ready_rx) = std::sync::mpsc::channel::<()>(); + let (contender_done_tx, contender_done_rx) = std::sync::mpsc::channel::<()>(); + let app_contender = app_handle.clone(); + + // Acquire the transition guard in THIS thread FIRST, before the contender + // is spawned. This guarantees the contender will block when it tries to + // acquire the same mutex. + let transition_guard = state.managed_agent_runtime_transition.lock().unwrap(); + + let contender = thread::spawn(move || { + use tauri::Manager; + let state = app_contender.state::(); + // Signal that the contender is alive and about to try the lock. + contender_ready_tx.send(()).unwrap(); + // Try to acquire the transition mutex — blocks while the test holds it. + let _guard = state.managed_agent_runtime_transition.lock().unwrap(); + // Signal: contender now holds the lock (compensation must be done). + contender_done_tx.send(()).unwrap(); + }); + + // Wait until contender is alive and parked on (or about to park on) the mutex. + contender_ready_rx.recv().unwrap(); + + // Give the contender a moment to reach the mutex and block on it. + std::thread::sleep(std::time::Duration::from_millis(20)); + + // Confirm contender is NOT done yet (still blocked on the mutex). + assert!( + contender_done_rx.try_recv().is_err(), + "contender must be blocked while this thread holds the transition guard" + ); + + // Run the lock-free core while the transition guard is held. + let mut records_for_restore = vec![]; + let restored = compensate_drain_for( + &stopped_from_drain, + &mut records_for_restore, + |entry, _recs| { + // Verify entry1 is restored with exact relay and start_on_app_launch. + assert_eq!(entry.key.pubkey, pubkey1, "only entry1 must be restored"); + assert_eq!( + entry.key.relay_url, "wss://relay.example", + "relay must match" + ); + assert!(entry.start_on_app_launch, "start_on_app_launch must match"); + Ok(()) + }, + ); + // entry1 restored successfully → None (no degradation). + assert!( + restored.is_none(), + "entry1 restoration must succeed: {restored:?}" + ); + + // Release the transition guard (compensation done). + drop(transition_guard); + + // Contender must now be able to acquire the mutex. + contender_done_rx + .recv_timeout(std::time::Duration::from_secs(5)) + .expect("contender must unblock after compensation releases the transition guard"); + + contender.join().expect("contender thread panicked"); +}