diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index 0b4e94af0e..ee84887804 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -99,225 +99,11 @@ fn resolve_created_avatar_url( .or_else(|| managed_agent_avatar_url(agent_command)) } -#[cfg(feature = "mesh-llm")] -async fn ensure_relay_mesh_for_record( - app: &AppHandle, - model_id: Option<&str>, - allow_fresh_create_start: bool, -) -> Result<(), String> { - crate::commands::ensure_relay_mesh_for_record(app, model_id, allow_fresh_create_start).await -} - -#[cfg(not(feature = "mesh-llm"))] -async fn ensure_relay_mesh_for_record( - _app: &AppHandle, - _model_id: Option<&str>, - _allow_fresh_create_start: bool, -) -> Result<(), String> { - Ok(()) -} - -pub(super) async fn start_local_agent_pairs_with_preflight( - app: &AppHandle, - state: &AppState, - pubkey: &str, - relay_urls: &[String], -) -> Result { - let record_snapshot = { - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - load_managed_agents(app)? - .into_iter() - .find(|record| record.pubkey == pubkey) - .ok_or_else(|| format!("agent {pubkey} not found"))? - }; - if record_snapshot.backend != BackendKind::Local { - return Err(format!("agent {pubkey} is not a local agent")); - } - let personas_for_preflight = load_personas(app).unwrap_or_default(); - let global_for_preflight = - crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); - let mesh_model_id = - crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( - &record_snapshot, - &personas_for_preflight, - &global_for_preflight, - ); - ensure_relay_mesh_for_record(app, mesh_model_id.as_deref(), false).await?; - - { - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let mut records = load_managed_agents(app)?; - let record = find_managed_agent_mut(&mut records, pubkey)?; - let personas = load_personas(app).unwrap_or_default(); - if let Some(persona_id) = record.persona_id.clone() { - if let Some(persona) = personas.iter().find(|persona| persona.id == persona_id) { - crate::managed_agents::persona_events::apply_persona_snapshot(record, persona); - record.updated_at = crate::util::now_iso(); - } - } - save_managed_agents(app, &records)?; - if let Some(saved_record) = records.iter().find(|record| record.pubkey == pubkey) { - retain_managed_agent_pending(app, state, saved_record); - } - } - - let mut errors = Vec::new(); - for relay_url in relay_urls { - if let Err(error) = crate::managed_agents::start_managed_agent_runtime_pair_lazy( - pubkey.to_string(), - relay_url.clone(), - app.clone(), - ) { - errors.push(format!("{relay_url}: {error}")); - } - } - if !errors.is_empty() { - return Err(format!( - "failed to restart one or more managed-agent runtime pairs: {}", - errors.join("; ") - )); - } - - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let records = load_managed_agents(app)?; - let runtimes = state - .managed_agent_processes - .lock() - .map_err(|e| e.to_string())?; - let record = records - .iter() - .find(|record| record.pubkey == pubkey) - .ok_or_else(|| format!("agent {pubkey} not found"))?; - summarize_from_disk(app, record, &runtimes) -} - -pub(super) async fn start_local_agent_with_preflight( - app: &AppHandle, - state: &AppState, - pubkey: &str, - allow_fresh_create_start: bool, - expected_relay_url: Option<&str>, - expected_signer_pubkey: Option<&str>, -) -> Result { - let record_snapshot = { - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let records = load_managed_agents(app)?; - records - .iter() - .find(|record| record.pubkey == pubkey) - .cloned() - .ok_or_else(|| format!("agent {pubkey} not found"))? - }; - - if record_snapshot.backend != BackendKind::Local { - return Err(format!("agent {pubkey} is not a local agent")); - } - - // 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: this reads the CURRENT definition - // directly, so a definition edit that flips `provider` to/from relay-mesh - // between saves is reflected here without needing a prospective re-snapshot; - // for a global-inherited blank definition, it also folds in the global - // default, which record-byte sniffing could never see. - let personas = load_personas(app).unwrap_or_default(); - let global = crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); - let mesh_model_id = - crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( - &record_snapshot, - &personas, - &global, - ); - ensure_relay_mesh_for_record(app, mesh_model_id.as_deref(), allow_fresh_create_start).await?; - - // The mesh preflight above is the suspension window Projects callbacks - // capture their scope against: a community switch during that await - // would otherwise spawn this pair keyed to the *new* workspace relay. - // Read the workspace relay ONCE, assert the caller's captured scope - // against that exact read, and hand the same bound value to the spawn - // below — the check is tied to its use, so a switch landing after this - // point can no longer retarget the spawn (it only changes state this - // call no longer consults). - let workspace_relay_url = crate::relay::bind_expected_relay_scope( - expected_relay_url, - crate::relay::relay_ws_url_with_override(state), - )?; - // Bind the active owner after the same final await as the relay. A - // same-relay identity replacement during mesh preflight must not release - // the stale preflight owner to spawn. - let workspace_owner = - crate::relay::bind_expected_signer(expected_signer_pubkey, workspace_owner_hex(state)?)?; - - let _store_guard = state - .managed_agents_store_lock - .lock() - .map_err(|e| e.to_string())?; - let mut records = load_managed_agents(app)?; - let mut runtimes = state - .managed_agent_processes - .lock() - .map_err(|e| e.to_string())?; - let record = find_managed_agent_mut(&mut records, pubkey)?; - if record.backend != BackendKind::Local { - return Err(format!("agent {pubkey} is no longer a local agent")); - } - // Re-snapshot the persona onto the record at every spawn so the agent always - // starts with the current persona config (system_prompt, model, provider, - // runtime). This clears the "out of date" drift badge without requiring a - // delete+recreate. See `apply_persona_snapshot` for the precedence and - // env-override self-heal rules. - // Load personas once: used for snapshot application below and summary build - // at the end — avoids a second disk read for the same file in the same call. - let personas = load_personas(app).unwrap_or_default(); - if let Some(persona_id) = record.persona_id.clone() { - match personas.iter().find(|p| p.id == persona_id) { - Some(persona) => { - crate::managed_agents::persona_events::apply_persona_snapshot(record, persona); - record.updated_at = crate::util::now_iso(); - } - None => { - return Err( - crate::managed_agents::effective_config::ORPHANED_INSTANCE_ERROR.to_string(), - ); - } - } - } - start_managed_agent_process( - app, - record, - &mut runtimes, - Some(workspace_owner.as_str()), - &workspace_relay_url, - )?; - save_managed_agents(app, &records)?; - if let Some(saved_record) = records.iter().find(|r| r.pubkey == pubkey) { - retain_managed_agent_pending(app, state, saved_record); - } - let record = records - .iter() - .find(|record| record.pubkey == pubkey) - .ok_or_else(|| format!("agent {pubkey} not found"))?; - build_managed_agent_summary( - app, - record, - &runtimes, - &personas, - &load_teams(app).unwrap_or_default(), - &crate::managed_agents::load_global_agent_config(app).unwrap_or_default(), - ) -} +#[path = "agents_local_start.rs"] +mod local_start; +pub(super) use local_start::{ + start_local_agent_pairs_with_preflight, start_local_agent_with_preflight, +}; pub(crate) use provider_deploy::deploy_to_provider; @@ -953,37 +739,13 @@ pub async fn start_managed_agent( let result = match target { StartTarget::Local => { - let app_for_start = app.clone(); - let agent_for_start = pubkey.clone(); - let relay_for_start = reconcile_relay.as_str().to_owned(); - let resumed = tokio::task::spawn_blocking(move || { - crate::managed_agents::start_after_exact_stop( - &app_for_start, - &agent_for_start, - &relay_for_start, - ) - }) - .await - .map_err(|_| "explicit Start task failed")??; - if resumed { - let records = load_managed_agents(&app)?; - let runtimes = state - .managed_agent_processes - .lock() - .map_err(|e| e.to_string())?; - let record = records - .iter() - .find(|r| r.pubkey == pubkey) - .ok_or("agent not found")?; - return summarize_from_disk(&app, record, &runtimes); - } start_local_agent_with_preflight( &app, &state, &pubkey, false, - expected_relay_url.as_deref(), - expected_signer_pubkey.as_deref(), + Some(reconcile_relay.as_str()), + Some(&owner_hex), ) .await } @@ -1004,8 +766,8 @@ pub async fn start_managed_agent( &config, agent_json, cached_binary_path.as_deref(), - expected_relay_url.as_deref(), - expected_signer_pubkey.as_deref(), + Some(reconcile_relay.as_str()), + Some(&owner_hex), ) .await?; diff --git a/desktop/src-tauri/src/commands/agents_local_start.rs b/desktop/src-tauri/src/commands/agents_local_start.rs new file mode 100644 index 0000000000..935ddad921 --- /dev/null +++ b/desktop/src-tauri/src/commands/agents_local_start.rs @@ -0,0 +1,262 @@ +//! Local launch preflight shared by explicit controls and config-driven starts. +use super::*; + +#[cfg(feature = "mesh-llm")] +async fn ensure_relay_mesh_for_record( + app: &AppHandle, + model_id: Option<&str>, + allow_fresh_create_start: bool, +) -> Result<(), String> { + crate::commands::ensure_relay_mesh_for_record(app, model_id, allow_fresh_create_start).await +} + +#[cfg(not(feature = "mesh-llm"))] +async fn ensure_relay_mesh_for_record( + _app: &AppHandle, + _model_id: Option<&str>, + _allow_fresh_create_start: bool, +) -> Result<(), String> { + Ok(()) +} + +pub(in crate::commands) async fn start_local_agent_pairs_with_preflight( + app: &AppHandle, + state: &AppState, + pubkey: &str, + relay_urls: &[String], +) -> Result { + let record_snapshot = { + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + load_managed_agents(app)? + .into_iter() + .find(|record| record.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))? + }; + if record_snapshot.backend != BackendKind::Local { + return Err(format!("agent {pubkey} is not a local agent")); + } + let personas_for_preflight = load_personas(app).unwrap_or_default(); + let global_for_preflight = + crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); + let mesh_model_id = + crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( + &record_snapshot, + &personas_for_preflight, + &global_for_preflight, + ); + ensure_relay_mesh_for_record(app, mesh_model_id.as_deref(), false).await?; + + { + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let mut records = load_managed_agents(app)?; + let record = find_managed_agent_mut(&mut records, pubkey)?; + let personas = load_personas(app).unwrap_or_default(); + if let Some(persona_id) = record.persona_id.clone() { + if let Some(persona) = personas.iter().find(|persona| persona.id == persona_id) { + crate::managed_agents::persona_events::apply_persona_snapshot(record, persona); + record.updated_at = crate::util::now_iso(); + } + } + save_managed_agents(app, &records)?; + if let Some(saved_record) = records.iter().find(|record| record.pubkey == pubkey) { + retain_managed_agent_pending(app, state, saved_record); + } + } + + let mut errors = Vec::new(); + for relay_url in relay_urls { + if let Err(error) = crate::managed_agents::start_managed_agent_runtime_pair_lazy( + pubkey.to_string(), + relay_url.clone(), + app.clone(), + ) { + errors.push(format!("{relay_url}: {error}")); + } + } + if !errors.is_empty() { + return Err(format!( + "failed to restart one or more managed-agent runtime pairs: {}", + errors.join("; ") + )); + } + + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(app)?; + let runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + let record = records + .iter() + .find(|record| record.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + summarize_from_disk(app, record, &runtimes) +} + +pub(in crate::commands) async fn start_local_agent_with_preflight( + app: &AppHandle, + state: &AppState, + pubkey: &str, + allow_fresh_create_start: bool, + expected_relay_url: Option<&str>, + expected_signer_pubkey: Option<&str>, +) -> Result { + let record_snapshot = { + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let records = load_managed_agents(app)?; + records + .iter() + .find(|record| record.pubkey == pubkey) + .cloned() + .ok_or_else(|| format!("agent {pubkey} not found"))? + }; + + if record_snapshot.backend != BackendKind::Local { + return Err(format!("agent {pubkey} is not a local agent")); + } + + // 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: this reads the CURRENT definition + // directly, so a definition edit that flips `provider` to/from relay-mesh + // between saves is reflected here without needing a prospective re-snapshot; + // for a global-inherited blank definition, it also folds in the global + // default, which record-byte sniffing could never see. + let personas = load_personas(app).unwrap_or_default(); + let global = crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); + let mesh_model_id = + crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( + &record_snapshot, + &personas, + &global, + ); + ensure_relay_mesh_for_record(app, mesh_model_id.as_deref(), allow_fresh_create_start).await?; + + // The mesh preflight above is the suspension window Projects callbacks + // capture their scope against: a community switch during that await + // would otherwise spawn this pair keyed to the *new* workspace relay. + // Read the workspace relay ONCE, assert the caller's captured scope + // against that exact read, and hand the same bound value to the spawn + // below — the check is tied to its use, so a switch landing after this + // point can no longer retarget the spawn (it only changes state this + // call no longer consults). + let workspace_relay_url = crate::relay::bind_expected_relay_scope( + expected_relay_url, + crate::relay::relay_ws_url_with_override(state), + )?; + // Bind the active owner after the same final await as the relay. A + // same-relay identity replacement during mesh preflight must not release + // the stale preflight owner to spawn. + let workspace_owner = + crate::relay::bind_expected_signer(expected_signer_pubkey, workspace_owner_hex(state)?)?; + + if !allow_fresh_create_start { + let app_for_start = app.clone(); + let agent_for_start = pubkey.to_owned(); + let relay_for_start = workspace_relay_url.as_str().to_owned(); + let owner_for_start = workspace_owner.as_str().to_owned(); + let resumed = tokio::task::spawn_blocking(move || { + crate::managed_agents::start_after_exact_stop( + &app_for_start, + &agent_for_start, + &relay_for_start, + &owner_for_start, + ) + }) + .await + .map_err(|_| "explicit Start task failed")??; + if resumed { + let records = load_managed_agents(app)?; + let runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + let record = records + .iter() + .find(|r| r.pubkey == pubkey) + .ok_or("agent not found")?; + return summarize_from_disk(app, record, &runtimes); + } + } + + // Recovery adds another suspension even when there is no journal yet. + // Revalidate the captured pair/owner before the legacy first-start path; + // never let that await retarget or authorize a stale workspace action. + crate::relay::assert_expected_relay_scope( + Some(workspace_relay_url.as_str()), + &crate::relay::relay_ws_url_with_override(state), + )?; + crate::relay::assert_expected_signer( + Some(workspace_owner.as_str()), + &workspace_owner_hex(state)?, + )?; + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| e.to_string())?; + let mut records = load_managed_agents(app)?; + let mut runtimes = state + .managed_agent_processes + .lock() + .map_err(|e| e.to_string())?; + let record = find_managed_agent_mut(&mut records, pubkey)?; + if record.backend != BackendKind::Local { + return Err(format!("agent {pubkey} is no longer a local agent")); + } + // Re-snapshot the persona onto the record at every spawn so the agent always + // starts with the current persona config (system_prompt, model, provider, + // runtime). This clears the "out of date" drift badge without requiring a + // delete+recreate. See `apply_persona_snapshot` for the precedence and + // env-override self-heal rules. + // Load personas once: used for snapshot application below and summary build + // at the end — avoids a second disk read for the same file in the same call. + let personas = load_personas(app).unwrap_or_default(); + if let Some(persona_id) = record.persona_id.clone() { + match personas.iter().find(|p| p.id == persona_id) { + Some(persona) => { + crate::managed_agents::persona_events::apply_persona_snapshot(record, persona); + record.updated_at = crate::util::now_iso(); + } + None => { + return Err( + crate::managed_agents::effective_config::ORPHANED_INSTANCE_ERROR.to_string(), + ); + } + } + } + start_managed_agent_process( + app, + record, + &mut runtimes, + Some(workspace_owner.as_str()), + &workspace_relay_url, + )?; + save_managed_agents(app, &records)?; + if let Some(saved_record) = records.iter().find(|r| r.pubkey == pubkey) { + retain_managed_agent_pending(app, state, saved_record); + } + let record = records + .iter() + .find(|record| record.pubkey == pubkey) + .ok_or_else(|| format!("agent {pubkey} not found"))?; + build_managed_agent_summary( + app, + record, + &runtimes, + &personas, + &load_teams(app).unwrap_or_default(), + &crate::managed_agents::load_global_agent_config(app).unwrap_or_default(), + ) +} diff --git a/desktop/src-tauri/src/managed_agents/execution.rs b/desktop/src-tauri/src/managed_agents/execution.rs index 04d24b4cd8..4d7b065a10 100644 --- a/desktop/src-tauri/src/managed_agents/execution.rs +++ b/desktop/src-tauri/src/managed_agents/execution.rs @@ -42,14 +42,31 @@ pub(super) fn config_revision( pub(crate) fn local_execution_config( app: &AppHandle, record: &ManagedAgentRecord, +) -> Result { + execution_config(app, record, true) +} + +fn execution_config( + app: &AppHandle, + record: &ManagedAgentRecord, + remote: bool, ) -> Result { local_execution_prerequisites(record)?; let personas = load_personas(app)?; let teams = load_teams(app)?; let global = load_global_agent_config(app)?; let descriptor = resolve_effective_harness_descriptor(record, &personas, &global)?; - let runtime = known_acp_runtime(&descriptor.command) - .ok_or("destination runtime is not in the Rust catalog")?; + let runtime = known_acp_runtime(&descriptor.command); + let revision = config_revision(record, &personas, &global, &teams, &descriptor)?; + if !remote { + // Ordinary local launch keeps its existing readiness/custom-runtime and + // agents-everywhere semantics. Spawn still rechecks this exact revision. + return Ok(LocalExecutionConfig { + runtime: runtime.map_or("custom", |runtime| runtime.id).into(), + revision, + }); + } + let runtime = runtime.ok_or("destination runtime is not in the Rust catalog")?; let effective = effective_config::resolve_effective_config(record, &personas, &global) .require_resolved()?; // Destination-local mesh preflight needs its own readiness gate. Do not @@ -70,7 +87,7 @@ pub(crate) fn local_execution_config( } Ok(LocalExecutionConfig { runtime: runtime.id.into(), - revision: config_revision(record, &personas, &global, &teams, &descriptor)?, + revision, }) } @@ -142,6 +159,72 @@ pub(crate) fn execute_host_operation( command_id: &str, request: &Command, compatible_runtime: bool, +) -> Result { + execute_operation( + app, + owner, + command_id, + request, + compatible_runtime, + Admission::Remote, + ) +} + +enum Admission<'a> { + Remote, + // Exact predecessor captured by the explicit local action; rechecked while + // holding both the transition lock and OS journal lock, never auto-reconcile. + Local { predecessor: &'a str }, +} + +fn local_start_predecessor(entry: &Entry) -> Result<(), String> { + match (&entry.request.action, &entry.outcome) { + (Action::Stop { .. }, Outcome::Stopped) | (Action::Start { .. }, Outcome::Rejected) => { + Ok(()) + } + _ => Err( + "Previous execution is not proven stopped or rejected; replacement remains blocked" + .into(), + ), + } +} + +impl Admission<'_> { + fn validate_predecessor(&self, ledger: &Ledger, request: &Command) -> Result<(), String> { + if let Admission::Local { predecessor } = self { + let current = ledger + .current() + .ok_or("local Start predecessor disappeared")?; + if current.command_id != *predecessor || !matches!(request.action, Action::Start { .. }) + { + return Err("local Start predecessor changed; refresh runtime status".into()); + } + local_start_predecessor(current)?; + } + Ok(()) + } + + fn admits_record(&self, record: &ManagedAgentRecord, owner: &str, relay: &str) -> bool { + match self { + Self::Local { .. } => local_execution_prerequisites(record).is_ok(), + Self::Remote => { + buzz_core_pkg::relay::normalize_relay_url(&record.relay_url) + .ok() + .as_deref() + == Some(relay) + && execution_agent_owner(record, owner).is_ok() + } + } + } +} + +fn execute_operation( + app: &AppHandle, + owner: &str, + command_id: &str, + request: &Command, + compatible_runtime: bool, + admission: Admission<'_>, ) -> Result { let state = app.state::(); let _transition = state @@ -178,16 +261,12 @@ pub(crate) fn execute_host_operation( if request.expires_at <= nostr::Timestamp::now().as_secs() { return Err("execution command expired without a recorded outcome".into()); } + admission.validate_predecessor(&ledger, request)?; let mut records = load_managed_agents(app)?; if matches!(request.action, Action::Start { .. }) - && !records.iter().any(|r| { - r.pubkey == request.agent - && buzz_core_pkg::relay::normalize_relay_url(&r.relay_url) - .ok() - .as_deref() - == Some(request.relay.as_str()) - && execution_agent_owner(r, owner).is_ok() - }) + && !records + .iter() + .any(|r| r.pubkey == request.agent && admission.admits_record(r, owner, &request.relay)) { ledger.begin(command_id, request)?; return ledger.finish(&request.operation, Outcome::Rejected); @@ -217,7 +296,7 @@ pub(crate) fn execute_host_operation( let receipt_path = managed_agents_base_dir(app)? .join("agent-pids") .join(format!("{}.json", key.runtime_id())); - let preflight = local_execution_config(app, record); + let preflight = execution_config(app, record, matches!(admission, Admission::Remote)); let compatible = preflight .is_ok_and(|config| config.runtime == *runtime && config.revision == *revision); if !compatible_runtime @@ -238,6 +317,10 @@ pub(crate) fn execute_host_operation( Some((&request.operation, revision)), ) { Ok(process) => process, + // spawn_agent_child_for_run returns Err only before a child is + // created (including OS spawn failure); after spawn it always + // returns the retained Child. Never classify post-spawn receipt, + // startup timeout or root-exit failures as definite rejection. Err(_) => return ledger.finish(&request.operation, Outcome::Rejected), }; let now = crate::util::now_iso(); @@ -357,32 +440,27 @@ pub(super) fn stop_local_selected_run( Ok(()) } -/// Explicit ordinary Start after an exact Stop still uses the execution ledger. -/// Automatic reconciliation never calls this and remains fenced. +/// Explicit ordinary Start after an exact Stop or a definite rejected Start +/// uses the same journal, with ordinary local admission rather than remote +/// provisioning grants. Automatic reconciliation never calls this. pub(crate) fn start_after_exact_stop( app: &AppHandle, pubkey: &str, relay: &str, + owner: &str, ) -> Result { - let state = app.state::(); - let owner = state.signing_keys()?.public_key().to_hex(); let key = ManagedAgentRuntimeKey::new(pubkey, relay)?; - let prior = ledger(app, &key, &owner)?.current().cloned(); + let prior = ledger(app, &key, owner)?.current().cloned(); let Some(prior) = prior else { return Ok(false); }; - if !matches!(prior.request.action, Action::Stop { .. }) { - return Ok(false); - } - if prior.outcome != Outcome::Stopped { - return Err("Selected Stop unconfirmed; replacement remains blocked".into()); - } + local_start_predecessor(&prior)?; let records = load_managed_agents(app)?; let record = records .iter() .find(|r| r.pubkey == pubkey) .ok_or("agent not found")?; - let config = local_execution_config(app, record)?; + let config = execution_config(app, record, false)?; let request = Command { v: 1, operation: uuid::Uuid::new_v4().simple().to_string(), @@ -395,12 +473,15 @@ pub(crate) fn start_after_exact_stop( }, }; let bytes = serde_json::to_vec(&request).map_err(|_| "invalid local Start")?; - let result = execute_host_operation( + let result = execute_operation( app, - &owner, + owner, &hex::encode(Sha256::digest(bytes)), &request, true, + Admission::Local { + predecessor: &prior.command_id, + }, )?; if result.outcome != Outcome::Spawned { return Err("Explicit Start not confirmed; inspect destination setup".into()); @@ -463,6 +544,14 @@ mod tests { // Availability only: this preflight must not inspect or export key bytes. record.private_key_nsec = "synthetic-nonempty-placeholder".into(); assert!(local_execution_prerequisites(&record).is_ok()); + // A legacy pin and absent attestation do not revoke ordinary local + // community-pair authority. They still deny remote destination Start. + let local = Admission::Local { + predecessor: "unused", + }; + assert!(local.admits_record(&record, &"aa".repeat(32), "wss://other.example")); + assert!(!Admission::Remote.admits_record(&record, &"aa".repeat(32), "wss://other.example")); + assert!(!Admission::Remote.admits_record(&record, &"aa".repeat(32), "wss://relay.example")); } #[test] @@ -476,3 +565,7 @@ mod tests { assert!(!exact_generation_matches("legacy", "legacy")); } } + +#[cfg(test)] +#[path = "execution_local_recovery_tests.rs"] +mod local_recovery_tests; diff --git a/desktop/src-tauri/src/managed_agents/execution_local_recovery_tests.rs b/desktop/src-tauri/src/managed_agents/execution_local_recovery_tests.rs new file mode 100644 index 0000000000..9571baf343 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/execution_local_recovery_tests.rs @@ -0,0 +1,172 @@ +//! Ordinary recovery uses the real persisted ledger, never clearing its fence. +use super::*; + +fn start(operation: &str) -> Command { + Command { + v: 1, + operation: operation.repeat(16), + relay: "wss://community-b.example".into(), + agent: nostr::Keys::generate().public_key().to_hex(), + expires_at: nostr::Timestamp::now().as_secs() + 120, + action: Action::Start { + runtime: "buzz-agent".into(), + revision: "ff".repeat(32), + }, + } +} + +#[test] +fn ordinary_rejected_start_recovers_after_reopen_without_losing_fence() { + let dir = tempfile::tempdir().unwrap(); + let initial = start("aa"); + let mut stop = initial.clone(); + stop.operation = "bb".repeat(16); + stop.action = Action::Stop { + run: initial.operation.clone(), + }; + let mut failed = initial.clone(); + failed.operation = "cc".repeat(16); + let stopped_id = "22".repeat(32); + let rejected_id = "33".repeat(32); + { + let mut ledger = Ledger::open(dir.path(), "placement").unwrap(); + ledger.begin(&"11".repeat(32), &initial).unwrap(); + ledger.finish(&initial.operation, Outcome::Spawned).unwrap(); + ledger.begin(&stopped_id, &stop).unwrap(); + ledger.finish(&stop.operation, Outcome::Stopped).unwrap(); + let admission = Admission::Local { + predecessor: &stopped_id, + }; + admission.validate_predecessor(&ledger, &failed).unwrap(); + ledger.begin(&rejected_id, &failed).unwrap(); + ledger.finish(&failed.operation, Outcome::Rejected).unwrap(); + } + let mut ledger = Ledger::open(dir.path(), "placement").unwrap(); + assert!( + ledger.is_fenced(), + "automatic/legacy launch remains blocked" + ); + let mut retry = failed.clone(); + retry.operation = "dd".repeat(16); + retry.action = Action::Start { + runtime: "buzz-agent".into(), + revision: "ee".repeat(32), + }; + assert!(Admission::Local { + predecessor: &stopped_id + } + .validate_predecessor(&ledger, &retry) + .is_err()); + let admission = Admission::Local { + predecessor: &rejected_id, + }; + admission.validate_predecessor(&ledger, &retry).unwrap(); + assert!(matches!( + ledger.begin(&"44".repeat(32), &retry).unwrap(), + Begin::Execute + )); + // A second callback captured before the first intent cannot create another + // attempt even when both started from the same definite rejection. + assert!(admission.validate_predecessor(&ledger, &retry).is_err()); + assert!(ledger.is_fenced()); + ledger.finish(&retry.operation, Outcome::Spawned).unwrap(); + assert_eq!( + ledger + .replay(&rejected_id, &failed) + .unwrap() + .unwrap() + .outcome, + Outcome::Rejected + ); + assert!(local_start_predecessor(ledger.current().unwrap()).is_err()); +} + +#[test] +fn ordinary_recovery_never_relabels_uncertainty_as_rejection() { + let request = start("aa"); + for outcome in [ + Outcome::Accepted, + Outcome::Unknown, + Outcome::RootExited, + Outcome::Spawned, + Outcome::Listening, + Outcome::Ready, + ] { + let mut entry = Entry { + command_id: "11".repeat(32), + request: request.clone(), + outcome: outcome.clone(), + observed_at: 1, + }; + assert!(local_start_predecessor(&entry).is_err(), "{outcome:?}"); + entry.request.action = Action::Stop { + run: "bb".repeat(16), + }; + assert!(local_start_predecessor(&entry).is_err(), "Stop {outcome:?}"); + } +} + +#[test] +fn recovery_serializes_across_controllers_and_rechecks_after_reopen() { + let dir = tempfile::tempdir().unwrap(); + let first = start("aa"); + let id = "11".repeat(32); + let mut ledger = Ledger::open(dir.path(), "placement").unwrap(); + ledger.begin(&id, &first).unwrap(); + ledger.finish(&first.operation, Outcome::Rejected).unwrap(); + let mut next = first.clone(); + next.operation = "bb".repeat(16); + let admission = Admission::Local { predecessor: &id }; + admission.validate_predecessor(&ledger, &next).unwrap(); + let path = dir.path().to_owned(); + assert!( + std::thread::spawn(move || Ledger::open(&path, "placement").is_err()) + .join() + .unwrap() + ); + ledger.begin(&"22".repeat(32), &next).unwrap(); + // Simulate a crash in the intent/spawn window, not a definite rejection. + drop(ledger); + let ledger = Ledger::open(dir.path(), "placement").unwrap(); + assert!(admission.validate_predecessor(&ledger, &next).is_err()); + assert!(local_start_predecessor(ledger.current().unwrap()).is_err()); + assert!(ledger.is_fenced()); + assert_eq!( + ledger + .replay(&"22".repeat(32), &next) + .unwrap() + .unwrap() + .outcome, + Outcome::Unknown + ); +} + +#[test] +fn post_spawn_or_root_exit_cannot_be_reclassified_for_local_retry() { + for outcome in [Outcome::Spawned, Outcome::Unknown] { + let dir = tempfile::tempdir().unwrap(); + let request = start("aa"); + let mut ledger = Ledger::open(dir.path(), "placement").unwrap(); + ledger.begin(&"11".repeat(32), &request).unwrap(); + ledger.finish(&request.operation, outcome.clone()).unwrap(); + assert!(ledger + .finish(&request.operation, Outcome::Rejected) + .is_err()); + drop(ledger); + let ledger = Ledger::open(dir.path(), "placement").unwrap(); + assert_eq!(ledger.current().unwrap().outcome, outcome); + assert!(local_start_predecessor(ledger.current().unwrap()).is_err()); + } + let dir = tempfile::tempdir().unwrap(); + let mut stop = start("aa"); + stop.action = Action::Stop { + run: "bb".repeat(16), + }; + let mut ledger = Ledger::open(dir.path(), "placement").unwrap(); + ledger.begin(&"11".repeat(32), &stop).unwrap(); + ledger.finish(&stop.operation, Outcome::RootExited).unwrap(); + assert!(ledger.finish(&stop.operation, Outcome::Rejected).is_err()); + drop(ledger); + let ledger = Ledger::open(dir.path(), "placement").unwrap(); + assert!(local_start_predecessor(ledger.current().unwrap()).is_err()); +} diff --git a/desktop/src-tauri/src/managed_agents/runtime_commands.rs b/desktop/src-tauri/src/managed_agents/runtime_commands.rs index 2ff2a627e9..b48ed68517 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_commands.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_commands.rs @@ -238,7 +238,12 @@ pub fn start_managed_agent_runtime( relay_url: String, app: AppHandle, ) -> Result { - if super::execution::start_after_exact_stop(&app, &pubkey, &relay_url)? { + let owner = app + .state::() + .signing_keys()? + .public_key() + .to_hex(); + if super::execution::start_after_exact_stop(&app, &pubkey, &relay_url, &owner)? { let state = app.state::(); let records = load_managed_agents(&app)?; let record = records diff --git a/desktop/src/features/agents/lib/managedAgentControlActions.test.mjs b/desktop/src/features/agents/lib/managedAgentControlActions.test.mjs index 70b355af6b..3daa9ebca1 100644 --- a/desktop/src/features/agents/lib/managedAgentControlActions.test.mjs +++ b/desktop/src/features/agents/lib/managedAgentControlActions.test.mjs @@ -12,6 +12,7 @@ function agent(overrides = {}) { name: "Mesh Agent", personaId: null, relayUrl: "ws://localhost:3000", + selectedRelayUrl: "ws://localhost:3000", acpCommand: "buzz-acp", agentCommand: "goose", agentArgs: [], @@ -56,7 +57,10 @@ test("relay-mesh agents delegate start to the backend preflight", async () => { calledWith = pubkey; }, }); - assert.equal(calledWith, meshAgent.pubkey); + assert.deepEqual(calledWith, { + pubkey: meshAgent.pubkey, + expectedRelayUrl: meshAgent.selectedRelayUrl, + }); // Backend preflight failures (e.g. no live serve target) propagate as-is. await assert.rejects( @@ -78,7 +82,10 @@ test("ordinary local agents still start normally", async () => { calledWith = pubkey; }, }); - assert.equal(calledWith, "deadbeef".repeat(8)); + assert.deepEqual(calledWith, { + pubkey: "deadbeef".repeat(8), + expectedRelayUrl: "ws://localhost:3000", + }); }); // --- respawnManagedAgentWithRules: stop→clear→start boundary tests ----------- @@ -193,3 +200,40 @@ test("ordinary Stop preserves clicked generation even when a successor appears", expectedRelayUrl: "wss://clicked.example", }); }); + +test("Restart retains clicked community across the Stop await", async () => { + const clicked = agent({ status: "running", selectedRunId: "aa".repeat(16) }); + const original = clicked.selectedRelayUrl; + let stop, start; + await respawnManagedAgentWithRules({ + agent: clicked, + stopManagedAgent: async (input) => { + stop = input; + await Promise.resolve(); + clicked.selectedRelayUrl = "wss://switched.example"; + }, + startManagedAgent: async (input) => { + start = input; + }, + }); + assert.equal(stop.expectedRelayUrl, original); + assert.equal(start.expectedRelayUrl, original); + assert.equal(stop.selectedRunId, "aa".repeat(16)); +}); + +test("missing community fails before either Restart leg, never falls back to record pin", async () => { + let calls = 0; + await assert.rejects( + respawnManagedAgentWithRules({ + agent: agent({ status: "running", selectedRelayUrl: null }), + stopManagedAgent: async () => { + calls++; + }, + startManagedAgent: async () => { + calls++; + }, + }), + /selected community/, + ); + assert.equal(calls, 0); +}); diff --git a/desktop/src/features/agents/lib/managedAgentControlActions.ts b/desktop/src/features/agents/lib/managedAgentControlActions.ts index ad1615405d..60b6f80e6c 100644 --- a/desktop/src/features/agents/lib/managedAgentControlActions.ts +++ b/desktop/src/features/agents/lib/managedAgentControlActions.ts @@ -12,7 +12,10 @@ type DeleteManagedAgentInput = { forceRemoteDelete?: boolean; }; -type StartManagedAgent = (pubkey: string) => Promise; +export type StartManagedAgent = (input: { + pubkey: string; + expectedRelayUrl: string; +}) => Promise; type StopManagedAgent = (input: { pubkey: string; selectedRunId?: string | null; @@ -26,6 +29,14 @@ function selectedStop(agent: ManagedAgent) { expectedRelayUrl: agent.selectedRelayUrl, }; } +function selectedStart(agent: ManagedAgent) { + if (!agent.selectedRelayUrl) { + throw new Error( + "Cannot start without a selected community; refresh agent status", + ); + } + return { pubkey: agent.pubkey, expectedRelayUrl: agent.selectedRelayUrl }; +} type DeleteManagedAgent = (input: DeleteManagedAgentInput) => Promise; type ManagedAgentChannelContext = { @@ -97,7 +108,7 @@ export async function startManagedAgentWithRules({ // Relay-mesh agents are no longer blocked here: the backend start preflight // (ensure_relay_mesh_for_record) re-resolves a live serve target and dials // it, failing with an actionable error when no peer serves the model. - await startManagedAgent(agent.pubkey); + await startManagedAgent(selectedStart(agent)); } export async function respawnManagedAgentWithRules({ @@ -113,12 +124,13 @@ export async function respawnManagedAgentWithRules({ * clear stale working badges at the right boundary. */ onStopped?: () => void; }) { + const start = selectedStart(agent); if (agent.backend.type === "local" && isManagedAgentActive(agent)) { await stopManagedAgent(selectedStop(agent)); onStopped?.(); } - await startManagedAgent(agent.pubkey); + await startManagedAgent(start); } export async function stopManagedAgentWithRules({ diff --git a/desktop/src/features/profile/ui/useAgentLifecycleActions.ts b/desktop/src/features/profile/ui/useAgentLifecycleActions.ts index f52b1a6171..cea54a4e2a 100644 --- a/desktop/src/features/profile/ui/useAgentLifecycleActions.ts +++ b/desktop/src/features/profile/ui/useAgentLifecycleActions.ts @@ -2,6 +2,7 @@ import * as React from "react"; import { toast } from "sonner"; import { + type StartManagedAgent, isManagedAgentActive, respawnManagedAgentWithRules, startManagedAgentWithRules, @@ -20,7 +21,7 @@ export function useAgentLifecycleActions({ channels: readonly Channel[] | undefined; managedAgent: ManagedAgent | undefined; relayAgents: readonly RelayAgent[] | undefined; - startManagedAgent: (pubkey: string) => Promise; + startManagedAgent: StartManagedAgent; stopManagedAgent: (input: { pubkey: string; selectedRunId?: string | null; diff --git a/docs/host-execution.md b/docs/host-execution.md index b83ccb9eb7..e2b8ca0f33 100644 --- a/docs/host-execution.md +++ b/docs/host-execution.md @@ -122,9 +122,26 @@ RootExited and Unknown return an error and keep replacement fenced. A repeated ordinary Stop reuses its original immutable local request; a live successor cannot be reported as stopped by replaying that earlier success. -An explicit ordinary Start after confirmed exact Stop, and the existing pair -Restart action, enter the same ledger for a fresh generation. Automatic reconcile -remains fenced. Spawn is not Ready: an immediate Stop during ACP pool initialization +Summary/profile Start captures the selected community, and Restart retains that +same community across its Stop await. The native Start command binds its owner +and community before preflight, then revalidates after suspension; a stale +continuation cannot start in the newly selected workspace. + +An explicit ordinary Start after confirmed exact Stop or definite rejected Start, +and the existing pair Restart action, enter the same ledger for a fresh generation. +Local recovery uses ordinary local pair/config/readiness semantics (including +legacy stored relay pins, setup-listener mode and custom runtimes), not remote +provisioning grants. Remote Start still requires its pinned relay, verified owner +attestation and advertised compatible runtime. Both paths recheck the exact +configuration revision at spawn. The captured local predecessor is rechecked +under the transition and OS journal locks; a concurrent/newer intent invalidates +it. Rejected means no child was created by that attempt: spawn failures before +child creation qualify, but post-spawn persistence failure, timeout and root exit +do not. Accepted/Unknown/RootExited/Spawned remain fenced, including after reopen; +old commands replay their immutable results. Automatic reconcile remains fenced. +No agent-configuration model rules changed. + +Spawn is not Ready: an immediate Stop during ACP pool initialization can return RootExited, even when partial children were drained. It cannot certify replacement; the actual tracer preserves this negative case rather than forcing Stopped. Provider `!shutdown` behavior is unchanged. Config-edit, delete