diff --git a/desktop/src-tauri/src/commands/agent_discovery.rs b/desktop/src-tauri/src/commands/agent_discovery.rs index 95534854a0b..e4030339ddb 100644 --- a/desktop/src-tauri/src/commands/agent_discovery.rs +++ b/desktop/src-tauri/src/commands/agent_discovery.rs @@ -555,9 +555,10 @@ async fn restart_single_agent_after_install( if record.backend != BackendKind::Local { return Err(format!("agent {pubkey_owned} is no longer a local agent")); } - let runtime_keys = - crate::managed_agents::managed_agent_runtime_keys(&runtimes, &pubkey_owned); - if runtime_keys.is_empty() { + // Dial targets (configured spellings) must be collected before the stop. + let restart_targets = + crate::managed_agents::managed_agent_restart_targets(&runtimes, &pubkey_owned); + if restart_targets.is_empty() { return Err(format!( "agent {pubkey_owned} no longer has a live pair runtime after sync" )); @@ -599,12 +600,12 @@ async fn restart_single_agent_after_install( stop_managed_agent_process(&app_for_stop, record_mut, &mut runtimes)?; save_managed_agents(&app_for_stop, &records)?; - Ok(runtime_keys) + Ok(restart_targets) }) .await; - let runtime_keys = match stop_result { - Ok(Ok(runtime_keys)) => runtime_keys, + let relay_urls = match stop_result { + Ok(Ok(restart_targets)) => restart_targets, Ok(Err(e)) => { eprintln!("buzz-desktop: install_acp_runtime: skipping restart of {pubkey}: {e}"); return InstallRestartOutcome::Skipped; @@ -617,7 +618,6 @@ async fn restart_single_agent_after_install( } }; - let relay_urls: Vec<_> = runtime_keys.into_iter().map(|key| key.relay_url).collect(); let state = app.state::(); match super::agents::start_local_agent_pairs_with_preflight(app, &state, pubkey, &relay_urls) .await diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index ed7a33d397d..58b9fc0740e 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -336,7 +336,7 @@ pub(super) async fn start_local_agent_pairs_with_preflight( relay_url.clone(), app.clone(), ) { - errors.push(format!("{relay_url}: {error}")); + errors.push(error); } } if !errors.is_empty() { diff --git a/desktop/src-tauri/src/commands/global_agent_config.rs b/desktop/src-tauri/src/commands/global_agent_config.rs index 91219bafb9c..73dce035439 100644 --- a/desktop/src-tauri/src/commands/global_agent_config.rs +++ b/desktop/src-tauri/src/commands/global_agent_config.rs @@ -292,9 +292,12 @@ async fn restart_local_agent_on_config_change( if record.backend != BackendKind::Local { return Err(format!("agent {pubkey_owned} is no longer a local agent")); } - let runtime_keys = - crate::managed_agents::managed_agent_runtime_keys(&runtimes, &pubkey_owned); - if runtime_keys.is_empty() { + // Collect the restart dial targets (each pair's configured connection + // URL, not the canonical key spelling) BEFORE the stop below drops the + // pairs — and their URLs — from the runtimes map. + let restart_targets = + crate::managed_agents::managed_agent_restart_targets(&runtimes, &pubkey_owned); + if restart_targets.is_empty() { return Err(format!( "agent {pubkey_owned} no longer has a live pair runtime after sync" )); @@ -327,12 +330,12 @@ async fn restart_local_agent_on_config_change( stop_managed_agent_process(&app_for_stop, record_mut, &mut runtimes)?; save_managed_agents(&app_for_stop, &records)?; - Ok(runtime_keys) + Ok(restart_targets) }) .await; - let runtime_keys = match stop_result { - Ok(Ok(runtime_keys)) => runtime_keys, + let relay_urls = match stop_result { + Ok(Ok(restart_targets)) => restart_targets, Ok(Err(e)) => { eprintln!("buzz-desktop: set_global_agent_config: skipping restart of {pubkey}: {e}"); return RestartOutcome::Skipped; @@ -345,7 +348,6 @@ async fn restart_local_agent_on_config_change( } }; - let relay_urls: Vec<_> = runtime_keys.into_iter().map(|key| key.relay_url).collect(); use tauri::Manager; let state = app.state::(); match super::agents::start_local_agent_pairs_with_preflight(app, &state, pubkey, &relay_urls) diff --git a/desktop/src-tauri/src/managed_agents/agent_env.rs b/desktop/src-tauri/src/managed_agents/agent_env.rs index 59b300d9d17..39376e88b82 100644 --- a/desktop/src-tauri/src/managed_agents/agent_env.rs +++ b/desktop/src-tauri/src/managed_agents/agent_env.rs @@ -137,6 +137,14 @@ pub(crate) fn parse_agent_env_lines(raw: &str) -> Vec<(&str, &str)> { .collect() } +pub(super) fn child_rust_log_filter() -> String { + match std::env::var("RUST_LOG") { + Ok(existing) if existing.contains("buzz_acp") => existing, + Ok(existing) if !existing.trim().is_empty() => format!("{existing},buzz_acp=info"), + _ => "buzz_acp=info".to_string(), + } +} + #[cfg(test)] mod tests { use super::{ diff --git a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs index 479d6ec913e..c14160b59c3 100644 --- a/desktop/src-tauri/src/managed_agents/process_lifecycle.rs +++ b/desktop/src-tauri/src/managed_agents/process_lifecycle.rs @@ -133,6 +133,7 @@ pub fn taskkill_tree(pid: u32) -> Result<(), String> { pub fn finish_spawn( child: std::process::Child, log_path: std::path::PathBuf, + connect_relay_url: String, spawn_config: super::spawn_snapshot::SpawnConfigSnapshot, setup_mode: bool, adapter_availability: Option, @@ -149,6 +150,7 @@ pub fn finish_spawn( super::ManagedAgentProcess { child, log_path, + connect_relay_url, spawn_config, setup_mode, adapter_availability, diff --git a/desktop/src-tauri/src/managed_agents/restore.rs b/desktop/src-tauri/src/managed_agents/restore.rs index 895aad712a8..b90c437cac2 100644 --- a/desktop/src-tauri/src/managed_agents/restore.rs +++ b/desktop/src-tauri/src/managed_agents/restore.rs @@ -26,6 +26,25 @@ enum SpawnOutcome { } type AgentSpawnResult = (String, SpawnOutcome); +/// Phase-A lookup for the exact requested workspace pair. A canonical-key hit +/// is not enough because loopback spellings can name distinct tenants. +fn phase_a_has_live_requested_pair( + runtimes: &mut std::collections::HashMap, + pubkey: &str, + requested_relay_url: &str, + target_matches: impl Fn(&T, &str) -> bool, + mut is_live: impl FnMut(&mut T) -> bool, +) -> bool { + let Ok(key) = super::ManagedAgentRuntimeKey::new(pubkey.to_string(), requested_relay_url) + else { + return false; + }; + let Some(runtime) = runtimes.get_mut(&key) else { + return false; + }; + target_matches(runtime, requested_relay_url) && is_live(runtime) +} + /// Backfill the pinned persona snapshot for pre-existing agents created before /// the record became the spawn source of truth. Runs once at launch, before /// `restore_managed_agents_on_launch` spawns anything, so no agent boots from an @@ -165,31 +184,26 @@ pub async fn restore_managed_agents_on_launch( // replacing the three separate kernel enumerations. super::sweep_untracked_bundle_harnesses(&tracked_pids); - let candidates: Vec = records + let workspace_relay = crate::relay::relay_ws_url_with_override(&state); + let mut to_start = Vec::new(); + for record in records .iter() .filter(|record| record.start_on_app_launch && record.backend == BackendKind::Local) - .map(|record| record.pubkey.clone()) - .collect(); - - let mut to_start = Vec::new(); - for pubkey in &candidates { - if let Some(runtime) = runtimes - .iter_mut() - .find(|(key, _)| key.pubkey == *pubkey) - .map(|(_, runtime)| runtime) - { - if runtime.child.try_wait().ok().flatten().is_none() { - continue; - } - } - if let Some(record) = records.iter().find(|r| r.pubkey == *pubkey) { - if let Some(pid) = record.runtime_pid { - if super::process_is_running(pid) { - continue; - } - } - to_start.push(record.clone()); + { + let requested_relay = + crate::relay::effective_agent_relay_url(&record.relay_url, &workspace_relay); + if phase_a_has_live_requested_pair( + &mut runtimes, + &record.pubkey, + &requested_relay, + |runtime, requested| { + super::connection_targets_match(&runtime.connect_relay_url, requested) + }, + |runtime| runtime.child.try_wait().ok().flatten().is_none(), + ) { + continue; } + to_start.push(record.clone()); } agents_to_start = to_start; @@ -308,38 +322,48 @@ pub async fn restore_managed_agents_on_launch( Ok(key) => { // F2: if a concurrent startup reconcile already // tracked a live child for this exact pair during - // the Phase A window, leave it alone. Mirrors the + // the Phase A window, leave it alone - but only + // when it dials the requested URL. Mirrors the // live-child guard in `start_pair`. - let already_live = app + let tracked_outcome = app .state::() .managed_agent_processes .lock() .ok() .and_then(|mut runtimes| { - runtimes.get_mut(&key).map(|runtime| { - runtime.child.try_wait().ok().flatten().is_none() - }) - }) - .unwrap_or(false); - if already_live { - SpawnOutcome::Skipped + let runtime = runtimes.get_mut(&key)?; + if runtime.child.try_wait().ok().flatten().is_none() { + return Some(live_pair_outcome(runtime, &relay_url)); + } + // A dead tracked entry can be replaced only + // when it belonged to this same target; a + // mismatched entry may carry tenant-scoped + // session cache under the colliding key. + super::ensure_pair_connection_matches(runtime, &relay_url) + .err() + .map(SpawnOutcome::Failed) + }); + if let Some(outcome) = tracked_outcome { + outcome } else { - match super::terminate_untracked_pair_runtime(app, &key) - .and_then(|()| { - // F1: restore spawns lazy, matching - // reconcile and manual start. Eager on - // restore buys nothing — a crashed - // mid-turn session is not resumed by an - // eager child — and silently reintroduces - // N idle brains on every launch. - spawn_agent_child( - app, - record, - &key.relay_url, - true, - owner_hex_ref, - ) - }) { + match super::terminate_untracked_pair_runtime( + app, &key, &relay_url, + ) + .and_then(|()| { + // F1: restore spawns lazy, matching + // reconcile and manual start. Eager on + // restore buys nothing — a crashed + // mid-turn session is not resumed by an + // eager child — and silently reintroduces + // N idle brains on every launch. + spawn_agent_child( + app, + record, + &relay_url, + true, + owner_hex_ref, + ) + }) { Ok(process) => { SpawnOutcome::Spawned(key, Box::new(process)) } @@ -390,6 +414,11 @@ pub async fn restore_managed_agents_on_launch( pid: process.child.id(), desktop_instance_id: super::current_instance_id(app), started_at: now.clone(), + // Phase B stamped the dial URL onto the process at spawn; + // reading it back here (not recomputing from the record) + // keeps the receipt truthful even if the workspace relay + // changed between phases. + connect_relay_url: Some(process.connect_relay_url.clone()), }; if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) { let _ = super::terminate_process(process.child.id()); @@ -557,3 +586,83 @@ fn persist_restore_error( record.last_error = Some(error); save_managed_agents(app, &records) } + +/// Phase-B decision for an already-tracked live pair: reuse (skip spawning) +/// only when the live child dials the requested URL; a cross-spelling child +/// is a connection-target conflict recorded as a failed outcome so Phase C +/// persists the sanitized error instead of silently keeping the wrong tenant. +fn live_pair_outcome( + runtime: &super::ManagedAgentPairRuntime, + requested_relay_url: &str, +) -> SpawnOutcome { + match super::ensure_pair_connection_matches(runtime, requested_relay_url) { + Ok(()) => SpawnOutcome::Skipped, + Err(error) => SpawnOutcome::Failed(error), + } +} + +#[cfg(test)] +mod tests { + use super::SpawnOutcome; + use crate::managed_agents::make_pair_runtime_with_connect_url; + + #[test] + fn phase_a_skips_only_the_live_requested_pair() { + let pubkey = "aa".repeat(32); + let key = crate::managed_agents::ManagedAgentRuntimeKey::new( + pubkey.clone(), + "ws://localhost:3100", + ) + .unwrap(); + let mut runtimes = + std::collections::HashMap::from([(key, ("ws://127.0.0.1:3100".to_string(), true))]); + let target_matches = |runtime: &(String, bool), requested: &str| { + super::super::connection_targets_match(&runtime.0, requested) + }; + let is_live = |runtime: &mut (String, bool)| runtime.1; + + // Same canonical key, different loopback tenant: Phase A must not + // suppress Phase B's explicit connection-target conflict. + assert!(!super::phase_a_has_live_requested_pair( + &mut runtimes, + &pubkey, + "ws://localhost:3100", + target_matches, + is_live, + )); + assert!(super::phase_a_has_live_requested_pair( + &mut runtimes, + &pubkey, + "ws://127.0.0.1:3100", + target_matches, + is_live, + )); + assert!(!super::phase_a_has_live_requested_pair( + &mut runtimes, + &pubkey, + "wss://other.example", + target_matches, + is_live, + )); + } + + #[test] + fn restore_reuses_live_pair_only_for_matching_spelling() { + let matching = make_pair_runtime_with_connect_url("ws://localhost:3100"); + assert!(matches!( + super::live_pair_outcome(&matching, " ws://localhost:3100 "), + SpawnOutcome::Skipped + )); + + // localhost and 127.0.0.1 share a canonical key but are distinct + // tenants: restore must record the conflict, not keep the wrong one. + let foreign = make_pair_runtime_with_connect_url("ws://127.0.0.1:3100"); + match super::live_pair_outcome(&foreign, "ws://localhost:3100") { + SpawnOutcome::Failed(error) => { + assert!(error.contains("connection-target conflict")); + assert!(!error.contains("3100"), "error must not echo URLs"); + } + _ => panic!("cross-spelling live pair must fail, not be reused"), + } + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 923530d34b9..459140b6db8 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -2,7 +2,9 @@ use std::collections::HashMap; use tauri::AppHandle; -use super::agent_env::{build_buzz_agent_provider_defaults, idle_pool_sleep_env}; +use super::agent_env::{ + build_buzz_agent_provider_defaults, child_rust_log_filter, idle_pool_sleep_env, +}; use crate::{ managed_agents::{ @@ -22,13 +24,16 @@ pub(crate) use path::{compose_path_entries, should_skip_claude_executable, shoul pub(crate) use super::access_policy::{build_respond_to_env_with_policy, RespondToEnv}; mod metadata; +use metadata::persona_drift_state; pub(crate) use metadata::{ apply_agent_display_env, resolve_session_title, runtime_metadata_env_vars, DISPLAY_NAME_ENV_VAR, SESSION_TITLE_ENV_VAR, }; mod stop; -pub(crate) use stop::managed_agent_runtime_keys; +pub(crate) use stop::{ + clear_inactive_legacy_scalar_pid, managed_agent_restart_targets, managed_agent_runtime_keys, +}; pub use stop::{stop_managed_agent_process, stop_managed_agent_workspace_pair}; mod sweep; @@ -38,11 +43,14 @@ mod process; #[cfg(test)] use process::{ buzz_marker_entry, name_matches_interpreter, name_matches_known_binary, - terminate_runtime_receipt_with, valid_agent_runtime_receipt_with, + terminate_runtime_receipt_for_target_with, terminate_runtime_receipt_with, + valid_agent_runtime_receipt_with, }; pub(crate) use process::{ - current_instance_id, process_belongs_to_us, process_has_buzz_marker, process_is_running, - terminate_process, terminate_untracked_pair_runtime, valid_agent_runtime_receipt, + connection_relay_url, connection_targets_match, current_instance_id, + ensure_pair_connection_matches, ensure_pair_receipt_connection_matches, process_belongs_to_us, + process_has_buzz_marker, process_is_running, terminate_process, + terminate_untracked_pair_runtime, tracked_pair_runtime_for_target, valid_agent_runtime_receipt, }; mod orphan_sweep; @@ -69,64 +77,33 @@ mod lifecycle; use lifecycle::kill_stale_tracked_processes_with; pub use lifecycle::{kill_stale_tracked_processes, sync_managed_agent_processes}; -/// Classify an agent's persona against the live catalog for the Agents-menu -/// drift indicator. Returns `(out_of_date, orphaned)`. -/// -/// Drift basis is the RECORD's `persona_source_version`, never the engram: -/// - persona_id set + persona present: out_of_date when the snapshot hash -/// differs from the persona's current content hash. -/// - persona_id set + persona gone: orphaned (no current hash to respawn into, -/// so never out_of_date — we must not tell the user to respawn into nothing). -/// - no persona_id: neither — a hand-built agent has no persona to drift from. -fn persona_drift_state( - record: &ManagedAgentRecord, - personas: &[crate::managed_agents::types::AgentDefinition], -) -> (bool, bool) { - let Some(persona_id) = record.persona_id.as_deref() else { - return (false, false); - }; - let Some(persona) = personas.iter().find(|p| p.id == persona_id) else { - return (false, true); - }; - let current = crate::managed_agents::persona_events::persona_content_hash( - &crate::managed_agents::persona_events::persona_event_content(persona), - ); - let out_of_date = record - .persona_source_version - .as_deref() - .is_some_and(|pinned| pinned != current); - (out_of_date, false) -} - -/// Resolve the runtime-pair key this record maps to for the active -/// workspace: always the active workspace relay (the legacy per-record relay -/// pin is ignored — see `effective_agent_relay_url`). Returns `None` for -/// records that cannot form a valid pair key yet (e.g. key-less agents that -/// mint keys on first start). -pub(crate) fn workspace_pair_key( +/// Resolve both the canonical runtime key and exact requested relay for this +/// record in the active workspace. The latter remains authoritative when +/// canonical loopback spellings collide. +pub(crate) fn workspace_pair_target( app: &AppHandle, record: &ManagedAgentRecord, -) -> Option { +) -> Option<(ManagedAgentRuntimeKey, String)> { use tauri::Manager; let state = app.state::(); - resolve_workspace_pair_key( + resolve_workspace_pair_target( &record.pubkey, &record.relay_url, &crate::relay::relay_ws_url_with_override(&state), ) } -/// Pure core of [`workspace_pair_key`]: workspace-relay resolution (legacy -/// record pins ignored) plus canonical key construction, kept `AppHandle`-free -/// so summary/stop scoping semantics are unit-testable. -pub(crate) fn resolve_workspace_pair_key( +/// AppHandle-free core of [`workspace_pair_target`]. Legacy record pins remain +/// ignored by `effective_agent_relay_url`. +pub(crate) fn resolve_workspace_pair_target( pubkey: &str, record_relay_url: &str, workspace_relay_url: &str, -) -> Option { - let effective_relay = +) -> Option<(ManagedAgentRuntimeKey, String)> { + let requested_relay = crate::relay::effective_agent_relay_url(record_relay_url, workspace_relay_url); - ManagedAgentRuntimeKey::new(pubkey.to_string(), &effective_relay).ok() + let key = ManagedAgentRuntimeKey::new(pubkey.to_string(), &requested_relay).ok()?; + Some((key, requested_relay)) } pub fn build_managed_agent_summary( @@ -143,8 +120,14 @@ pub fn build_managed_agent_summary( // workspace relay. An agent running only in another community must read // as stopped here — matching by pubkey alone would show every community a // green light as long as any pair anywhere is alive. - let pair_key = workspace_pair_key(app, record); - let pair_runtime = pair_key.as_ref().and_then(|key| runtimes.get(key)); + let pair_target = workspace_pair_target(app, record); + // A canonical-key collision can belong to another loopback tenant. That + // runtime is not an error for this workspace; it simply is not running here. + let pair_runtime = pair_target.as_ref().and_then(|(key, requested_relay)| { + runtimes + .get(key) + .filter(|runtime| connection_targets_match(&runtime.connect_relay_url, requested_relay)) + }); let (status, pid, log_path) = if record.backend != BackendKind::Local { // Two-axis status model for remote agents: @@ -169,21 +152,12 @@ pub fn build_managed_agent_summary( }; (status, None, String::new()) } else { - let persisted_pid = record.runtime_pid.filter(|pid| process_is_running(*pid)); if let Some(runtime) = pair_runtime { ( "running".to_string(), Some(runtime.child.id()), runtime.log_path.display().to_string(), ) - } else if let Some(pid) = persisted_pid { - ( - "running".to_string(), - Some(pid), - managed_agent_log_path(app, &record.pubkey)? - .display() - .to_string(), - ) } else { ( "stopped".to_string(), @@ -227,11 +201,10 @@ pub fn build_managed_agent_summary( // Restart badge: the running process stamped the effective spawn config // it was launched with; recompute a prospective one from current disk // state and report every differing field. Only the tracked live pair for - // THIS workspace can drift — stopped agents spawn fresh, adopted - // (runtime_pid-only) processes have no stamp to compare, and pairs running - // for other communities are judged in their own community (comparing them - // against this workspace's relay would flag a spurious restart on every - // community switch). + // THIS workspace can drift — stopped agents spawn fresh, legacy + // `runtime_pid` bookkeeping is not pair-scoped, and pairs running for other + // communities are judged in their own community (comparing them against + // this workspace's relay would flag a spurious restart on every switch). // // Adapter-availability drift (codex only) contributes its own synthetic // entry, so an out-of-band adapter change (manual npm install/downgrade) @@ -244,17 +217,20 @@ pub fn build_managed_agent_summary( // The prospective side is computed only for a tracked pair: an unstamped // agent has nothing to compare against. - let tracked_spawn = pair_key.as_ref().zip(pair_runtime).map(|(key, runtime)| { - let current = crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot( - record, - personas, - teams, - &key.relay_url, - global_config, - super::owner_only_access_build(), - ); - (runtime, current) - }); + let tracked_spawn = pair_target + .as_ref() + .zip(pair_runtime) + .map(|((key, _), runtime)| { + let current = crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot( + record, + personas, + teams, + &key.relay_url, + global_config, + super::owner_only_access_build(), + ); + (runtime, current) + }); let restart_diff = crate::managed_agents::spawn_snapshot::eligible_restart_diff( persona_orphaned, tracked_spawn.as_ref().map(|(runtime, current)| { @@ -496,9 +472,10 @@ pub fn spawn_agent_child( .map(|p| p.display().to_string()) .unwrap_or_else(|| effective_command.clone()); - // The caller supplies the explicit canonical pair relay. This is the only - // relay this child may connect to, regardless of the record/workspace default. - let effective_relay_url = runtime_key.relay_url.clone(); + // The canonical relay is only the pair identity. Preserve the configured + // authority for the network connection because relay communities are + // host-derived (`localhost` and `127.0.0.1` can be distinct tenants). + let effective_relay_url = connection_relay_url(relay_url); // Augment PATH for DMG launches so child processes can find: // - bundled CLI via ~/.local/bin symlink @@ -853,7 +830,11 @@ pub fn spawn_agent_child( super::spawn_snapshot::SpawnConfigInputs { record, descriptor: &descriptor, - relay_url: &effective_relay_url, + // Snapshot the canonical key relay, not the connection URL: the + // restart-drift check recomputes the prospective snapshot with + // `key.relay_url`, and the two must agree even when the configured + // spelling differs from the canonical one. + relay_url: &runtime_key.relay_url, team_instructions: team_instructions.as_deref(), system_prompt: effective_prompt.as_deref(), model: effective_model.as_deref(), @@ -909,6 +890,7 @@ pub fn spawn_agent_child( return Ok(super::process_lifecycle::finish_spawn( child, log_path, + effective_relay_url, spawn_config, spawned_setup_mode, spawned_adapter_availability, @@ -919,6 +901,7 @@ pub fn spawn_agent_child( Ok(crate::managed_agents::ManagedAgentProcess { child, log_path, + connect_relay_url: effective_relay_url, spawn_config, setup_mode: spawned_setup_mode, adapter_availability: spawned_adapter_availability, @@ -926,14 +909,6 @@ pub fn spawn_agent_child( }) } -fn child_rust_log_filter() -> String { - match std::env::var("RUST_LOG") { - Ok(existing) if existing.contains("buzz_acp") => existing, - Ok(existing) if !existing.trim().is_empty() => format!("{existing},buzz_acp=info"), - _ => "buzz_acp=info".to_string(), - } -} - pub fn start_managed_agent_process( app: &AppHandle, record: &mut ManagedAgentRecord, @@ -949,7 +924,10 @@ pub fn start_managed_agent_process( ) }; let key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), &relay_url)?; - if let Some(runtime) = runtimes.get_mut(&key) { + if tracked_pair_runtime_for_target(runtimes, &key, &relay_url)?.is_some() { + let runtime = runtimes + .get_mut(&key) + .ok_or_else(|| "managed-agent runtime changed during start".to_string())?; if runtime .child .try_wait() @@ -959,20 +937,25 @@ pub fn start_managed_agent_process( return Ok(()); } + ensure_pair_receipt_connection_matches(app, &key, &relay_url)?; runtimes.remove(&key); - super::remove_agent_runtime_receipt(app, &key); } + terminate_untracked_pair_runtime(app, &key, &relay_url)?; // Scalar PIDs are migration-only and never establish pair liveness. record.runtime_pid = None; - let mut process = spawn_agent_child(app, record, &key.relay_url, false, owner_hex)?; + let mut process = spawn_agent_child(app, record, &relay_url, false, owner_hex)?; let now = now_iso(); let receipt = super::ManagedAgentRuntimeReceipt { key: key.clone(), pid: process.child.id(), desktop_instance_id: current_instance_id(app), started_at: now.clone(), + // The URL the child actually dialed, not the local `relay_url` + // binding — same string by construction here, but reading it off the + // process keeps every receipt site identical to the spawn. + connect_relay_url: Some(process.connect_relay_url.clone()), }; if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) { let _ = terminate_process(process.child.id()); @@ -993,6 +976,10 @@ pub fn start_managed_agent_process( #[cfg(test)] mod test_fixtures; +#[cfg(test)] +pub(crate) use test_fixtures::make_pair_runtime_with_connect_url; +#[cfg(test)] +mod connect_url_tests; #[cfg(test)] mod tests; diff --git a/desktop/src-tauri/src/managed_agents/runtime/connect_url_tests.rs b/desktop/src-tauri/src/managed_agents/runtime/connect_url_tests.rs new file mode 100644 index 00000000000..215cd332d2f --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/runtime/connect_url_tests.rs @@ -0,0 +1,338 @@ +//! Connection-URL persistence tests: receipt back-compat and validation, +//! restart-target selection, and the pair-reuse conflict guard. + +use super::test_fixtures::{make_pair_runtime_with_connect_url, receipt_fixture}; + +#[test] +fn receipt_without_connect_url_deserializes_and_validates() { + // Receipts persisted before `connectRelayUrl` existed must keep loading + // (field absent -> None) and keep validating. + let key = + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(); + let json = format!( + r#"{{"key":{{"pubkey":"{}","relayUrl":"{}"}},"pid":{},"desktopInstanceId":"test-instance","startedAt":"now"}}"#, + key.pubkey, + key.relay_url, + std::process::id(), + ); + let receipt: crate::managed_agents::ManagedAgentRuntimeReceipt = + serde_json::from_str(&json).expect("pre-connect-url receipt must deserialize"); + assert_eq!(receipt.connect_relay_url, None); + + let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id())); + assert!(super::valid_agent_runtime_receipt_with( + &path, + &receipt, + "test-instance", + |_| true, + |_, _| true, + )); +} + +#[test] +fn receipt_connect_url_matching_pair_validates() { + // The configured spelling (`localhost`) canonicalizes to the receipt's own + // key (`127.0.0.1`), so the receipt is valid — this is the normal shape + // written by every spawn on a loopback workspace. + let mut receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + receipt.connect_relay_url = Some("ws://localhost:3100".into()); + let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id())); + assert!(super::valid_agent_runtime_receipt_with( + &path, + &receipt, + "test-instance", + |_| true, + |_, _| true, + )); +} + +#[test] +fn receipt_connect_url_foreign_pair_rejected() { + // A connection URL that canonicalizes to a DIFFERENT pair key is a corrupt + // or cross-wired receipt — it must fail validation, not fall back. + let mut receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + receipt.connect_relay_url = Some("ws://localhost:4000".into()); + let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id())); + assert!(!super::valid_agent_runtime_receipt_with( + &path, + &receipt, + "test-instance", + |_| true, + |_, _| true, + )); +} + +#[test] +fn receipt_connect_url_roundtrips_through_persistence() { + // Present field serializes (camelCase) and deserializes unchanged, so a + // restart in a later session re-dials the exact configured spelling. + let mut receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + receipt.connect_relay_url = Some("ws://localhost:3100".into()); + let json = serde_json::to_string(&receipt).expect("serialize receipt"); + assert!(json.contains("\"connectRelayUrl\":\"ws://localhost:3100\"")); + let restored: crate::managed_agents::ManagedAgentRuntimeReceipt = + serde_json::from_str(&json).expect("deserialize receipt"); + assert_eq!(restored, receipt); +} + +#[test] +fn receipt_target_mismatch_is_rejected_before_termination_selection() { + use std::cell::Cell; + + let mut receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + receipt.connect_relay_url = Some("ws://localhost:3100".into()); + let validated = Cell::new(false); + let terminated = Cell::new(false); + let error = super::terminate_runtime_receipt_for_target_with( + std::path::Path::new("pair.json"), + &receipt, + "ws://127.0.0.1:3100", + |_, _| { + validated.set(true); + true + }, + |_| { + terminated.set(true); + Ok(()) + }, + |_| false, + |_| {}, + ) + .unwrap_err(); + + assert!(error.contains("connection-target conflict")); + assert!( + !error.contains("3100"), + "error must not disclose either URL" + ); + assert!( + validated.get(), + "receipt ownership must be validated before target selection" + ); + assert!( + !terminated.get(), + "mismatched receipt must never be terminated" + ); +} + +#[test] +fn invalid_receipt_is_ignored_without_target_selection() { + use std::cell::Cell; + + let mut receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + receipt.connect_relay_url = Some("ws://localhost:3100".into()); + let terminated = Cell::new(false); + + super::terminate_runtime_receipt_for_target_with( + std::path::Path::new("pair.json"), + &receipt, + "ws://127.0.0.1:3100", + |_, _| false, + |_| { + terminated.set(true); + Ok(()) + }, + |_| false, + |_| {}, + ) + .expect("an invalid receipt cannot select a process"); + + assert!(!terminated.get()); +} + +#[test] +fn legacy_receipt_uses_its_historical_canonical_dial_target() { + use std::cell::Cell; + + let receipt = receipt_fixture( + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(), + ); + let terminated = Cell::new(false); + let error = super::terminate_runtime_receipt_for_target_with( + std::path::Path::new("pair.json"), + &receipt, + "ws://localhost:3100", + |_, _| true, + |_| { + terminated.set(true); + Ok(()) + }, + |_| false, + |_| {}, + ) + .unwrap_err(); + + // Legacy children dialed key.relay_url (`127.0.0.1` after canonical + // normalization), so a localhost request must fail closed. + assert!(error.contains("connection-target conflict")); + assert!(!terminated.get()); +} + +#[test] +fn tracked_runtime_selection_requires_the_requested_connection_target() { + let key = + crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3100") + .unwrap(); + let runtimes = std::collections::HashMap::from([( + key.clone(), + make_pair_runtime_with_connect_url("ws://localhost:3100"), + )]); + + assert!( + super::tracked_pair_runtime_for_target(&runtimes, &key, "ws://LocalHost:3100/",) + .unwrap() + .is_some() + ); + let error = + super::tracked_pair_runtime_for_target(&runtimes, &key, "ws://127.0.0.1:3100").unwrap_err(); + assert!(error.contains("connection-target conflict")); + assert!( + runtimes.contains_key(&key), + "failed selection must not mutate the map" + ); +} + +#[test] +fn restart_targets_preserve_configured_spelling_per_pair() { + // Restart targets come from each live pair's stamped connection URL — + // the configured loopback spelling survives (never the canonical fold), + // and only the requested agent's pairs are selected. + let agent_a = "aa".repeat(32); + let agent_b = "bb".repeat(32); + let mut runtimes = std::collections::HashMap::new(); + runtimes.insert( + crate::managed_agents::ManagedAgentRuntimeKey::new(agent_a.clone(), "ws://localhost:3100") + .unwrap(), + make_pair_runtime_with_connect_url("ws://localhost:3100"), + ); + runtimes.insert( + crate::managed_agents::ManagedAgentRuntimeKey::new(agent_a.clone(), "wss://other.example") + .unwrap(), + make_pair_runtime_with_connect_url("wss://other.example"), + ); + runtimes.insert( + crate::managed_agents::ManagedAgentRuntimeKey::new(agent_b, "ws://localhost:9999").unwrap(), + make_pair_runtime_with_connect_url("ws://localhost:9999"), + ); + + let mut targets = super::managed_agent_restart_targets(&runtimes, &agent_a); + targets.sort(); + assert_eq!( + targets, + vec![ + "ws://localhost:3100".to_string(), + "wss://other.example".to_string(), + ], + ); +} + +#[test] +fn pair_reuse_with_matching_spelling_is_allowed() { + let runtime = make_pair_runtime_with_connect_url("ws://localhost:3100"); + assert!(super::ensure_pair_connection_matches(&runtime, " ws://localhost:3100 ").is_ok()); +} + +#[test] +fn pair_reuse_across_spellings_is_a_connection_target_conflict() { + // localhost and 127.0.0.1 share a canonical key but are distinct tenants: + // reuse must fail loudly instead of reporting the requested tenant started. + let runtime = make_pair_runtime_with_connect_url("ws://127.0.0.1:3100"); + let err = super::ensure_pair_connection_matches(&runtime, "ws://localhost:3100").unwrap_err(); + assert!(err.contains("connection-target conflict")); + // No URL disclosure: the message must not echo either spelling. + assert!(!err.contains("3100")); +} + +#[test] +fn pair_reuse_folds_connection_equivalent_spellings() { + // Host case, explicit default port, root slash, and the FQDN root dot + // are connection-equivalent per the tenancy authority - none of these + // may read as a conflict after harmless config formatting drift. + for (live, requested) in [ + ("ws://localhost:3000", "ws://LocalHost:3000"), + ("wss://relay.example", "wss://relay.example:443"), + ("ws://localhost:3000", "ws://localhost:3000/"), + ("wss://relay.example", "wss://relay.example."), + ("ws://relay.example:80/ws", "ws://Relay.Example:80/ws"), + ("ws://relay.example?token=x", "ws://relay.example/?token=x"), + ("ws://[::1]:3000", "ws://[0:0:0:0:0:0:0:1]:3000"), + ] { + let runtime = make_pair_runtime_with_connect_url(live); + assert!( + super::ensure_pair_connection_matches(&runtime, requested).is_ok(), + "equivalent spellings must not conflict: {live} vs {requested}", + ); + } +} + +#[test] +fn pair_reuse_keeps_tenancy_significant_differences_conflicting() { + // Scheme, non-default port, and path differences are real target + // differences - and the loopback split stays a conflict (see + // pair_reuse_across_spellings_is_a_connection_target_conflict). + for (live, requested) in [ + ("ws://relay.example:3000", "wss://relay.example:3000"), + ("ws://relay.example:3000", "ws://relay.example:3001"), + ("ws://relay.example:3000/a", "ws://relay.example:3000/b"), + ("ws://relay.example:443", "ws://relay.example"), + ("wss://relay.example:80", "wss://relay.example"), + ("ws://relay.example?token=x", "ws://relay.example?token=y"), + ("ws://[::1]:3000", "ws://localhost:3000"), + ("ws://[::1]:3000", "ws://127.0.0.1:3000"), + ] { + let runtime = make_pair_runtime_with_connect_url(live); + assert!( + super::ensure_pair_connection_matches(&runtime, requested).is_err(), + "distinct targets must conflict: {live} vs {requested}", + ); + } +} + +#[test] +fn invalid_connection_targets_do_not_alias_after_parse() { + assert!(!super::connection_targets_match( + "wss://alice@relay.example", + "wss://bob@relay.example", + )); + assert!(!super::connection_targets_match( + "wss://alice@relay.example", + "wss://relay.example", + )); + assert!(!super::connection_targets_match( + "wss://relay.example#east", + "wss://relay.example#west", + )); + assert!(!super::connection_targets_match( + "wss://relay.example#east", + "wss://relay.example", + )); + assert!(!super::connection_targets_match( + "https://relay.example", + "https://RELAY.example/", + )); + assert!(super::connection_targets_match( + "wss://alice@relay.example", + "wss://alice@relay.example", + )); + assert!(super::connection_targets_match( + " wss://relay.example#east ", + "wss://relay.example#east", + )); +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/metadata.rs b/desktop/src-tauri/src/managed_agents/runtime/metadata.rs index 5aef424ea61..41da7d9a940 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/metadata.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/metadata.rs @@ -74,6 +74,35 @@ pub(crate) fn resolve_session_title(display_name: Option<&str>, name: &str) -> O .find(|value| !value.is_empty()) } +/// Classify an agent's persona against the live catalog for the Agents-menu +/// drift indicator. Returns `(out_of_date, orphaned)`. +/// +/// Drift basis is the RECORD's `persona_source_version`, never the engram: +/// - persona_id set + persona present: out_of_date when the snapshot hash +/// differs from the persona's current content hash. +/// - persona_id set + persona gone: orphaned (no current hash to respawn into, +/// so never out_of_date — we must not tell the user to respawn into nothing). +/// - no persona_id: neither — a hand-built agent has no persona to drift from. +pub(super) fn persona_drift_state( + record: &crate::managed_agents::ManagedAgentRecord, + personas: &[crate::managed_agents::types::AgentDefinition], +) -> (bool, bool) { + let Some(persona_id) = record.persona_id.as_deref() else { + return (false, false); + }; + let Some(persona) = personas.iter().find(|p| p.id == persona_id) else { + return (false, true); + }; + let current = crate::managed_agents::persona_events::persona_content_hash( + &crate::managed_agents::persona_events::persona_event_content(persona), + ); + let out_of_date = record + .persona_source_version + .as_deref() + .is_some_and(|pinned| pinned != current); + (out_of_date, false) +} + #[cfg(test)] mod tests { use super::resolve_session_title; diff --git a/desktop/src-tauri/src/managed_agents/runtime/process.rs b/desktop/src-tauri/src/managed_agents/runtime/process.rs index 37eb5659a4a..763456cc41c 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/process.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/process.rs @@ -408,7 +408,16 @@ pub(crate) fn valid_agent_runtime_receipt_with( else { return false; }; + // A stored connection URL must belong to this pair: canonicalizing it has + // to reproduce the receipt's own key. Absent is fine (pre-field receipts); + // present-but-foreign or unparseable is a corrupt receipt, not a fallback. + let connect_url_matches_key = match receipt.connect_relay_url.as_deref() { + None => true, + Some(connect_url) => ManagedAgentRuntimeKey::new(receipt.key.pubkey.clone(), connect_url) + .is_ok_and(|from_connect| from_connect == receipt.key), + }; canonical == receipt.key + && connect_url_matches_key && path.file_name().and_then(|name| name.to_str()) == Some(&format!("{}.json", receipt.key.runtime_id())) && receipt.desktop_instance_id == instance_id @@ -436,11 +445,72 @@ pub(super) fn terminate_runtime_receipt_with( std::thread::sleep(std::time::Duration::from_millis(100)); } Err(format!( - "prior runtime {} for pair {} on {} did not exit", - receipt.pid, receipt.key.pubkey, receipt.key.relay_url + "prior runtime {} for agent {} did not exit", + receipt.pid, receipt.key.pubkey )) } +fn receipt_connection_relay_url(receipt: &super::super::ManagedAgentRuntimeReceipt) -> &str { + receipt + .connect_relay_url + .as_deref() + // Before `connectRelayUrl` was persisted, children were spawned with + // the canonical key URL. That concrete historical target is the only + // safe fallback for a legacy receipt. + .unwrap_or(&receipt.key.relay_url) +} + +fn valid_runtime_receipt_for_key( + app: &AppHandle, + key: &ManagedAgentRuntimeKey, + instance_id: &str, +) -> Option<(std::path::PathBuf, super::super::ManagedAgentRuntimeReceipt)> { + super::super::read_all_agent_runtime_receipts(app) + .into_iter() + .find(|(path, receipt)| { + receipt.key == *key && valid_agent_runtime_receipt(path, receipt, instance_id) + }) +} + +fn ensure_receipt_connection_matches( + receipt: &super::super::ManagedAgentRuntimeReceipt, + requested_relay_url: &str, +) -> Result<(), String> { + ensure_connection_targets_match(receipt_connection_relay_url(receipt), requested_relay_url) +} + +/// Validate any receipt occupying `key` before a tracked runtime or its +/// receipt/cache is removed through a caller-supplied connection URL. +pub(crate) fn ensure_pair_receipt_connection_matches( + app: &AppHandle, + key: &ManagedAgentRuntimeKey, + requested_relay_url: &str, +) -> Result<(), String> { + let instance_id = current_instance_id(app); + if let Some((_, receipt)) = valid_runtime_receipt_for_key(app, key, &instance_id) { + ensure_receipt_connection_matches(&receipt, requested_relay_url)?; + } + Ok(()) +} + +pub(super) fn terminate_runtime_receipt_for_target_with( + path: &std::path::Path, + receipt: &super::super::ManagedAgentRuntimeReceipt, + requested_relay_url: &str, + is_valid: impl FnOnce(&std::path::Path, &super::super::ManagedAgentRuntimeReceipt) -> bool, + terminate: impl FnOnce(u32) -> Result<(), String>, + is_running: impl FnMut(u32) -> bool, + remove: impl FnOnce(&std::path::Path), +) -> Result<(), String> { + if !is_valid(path, receipt) { + return Ok(()); + } + // Only a valid, owned receipt may select a process. Once selected, require + // its stamped target to match before termination. + ensure_receipt_connection_matches(receipt, requested_relay_url)?; + terminate_runtime_receipt_with(path, receipt, terminate, is_running, remove) +} + /// Replace a valid prior-session process before registering a new child for /// the same pair. The caller must hold the runtime transition lock so receipt /// inspection, termination, spawn, and registration cannot race shutdown or @@ -448,22 +518,159 @@ pub(super) fn terminate_runtime_receipt_with( pub(crate) fn terminate_untracked_pair_runtime( app: &AppHandle, key: &ManagedAgentRuntimeKey, + requested_relay_url: &str, ) -> Result<(), String> { let instance_id = current_instance_id(app); - let Some((path, receipt)) = super::super::read_all_agent_runtime_receipts(app) - .into_iter() - .find(|(path, receipt)| { - receipt.key == *key && valid_agent_runtime_receipt(path, receipt, &instance_id) - }) - else { + let Some((path, receipt)) = valid_runtime_receipt_for_key(app, key, &instance_id) else { return Ok(()); }; - terminate_runtime_receipt_with( + terminate_runtime_receipt_for_target_with( &path, &receipt, + requested_relay_url, + |_, _| true, terminate_process, process_is_running, super::super::remove_agent_runtime_receipt_path, ) } + +/// The URL a child is dialed with: the configured spelling, trimmed. The +/// canonical form is identity-only; connection code preserves the authority. +pub(crate) fn connection_relay_url(configured_relay_url: &str) -> String { + configured_relay_url.trim().to_string() +} + +/// Comparable connection target, parsed with `url::Url`: lowercased scheme, +/// case-folded host with a single FQDN root dot stripped (mirroring the +/// tenancy authority `tenant::normalize_host`), a port only when it is not +/// the scheme's own default, a root-slash-folded path, and the query kept +/// verbatim. Folds spellings that reach the same tenant while preserving +/// tenancy-significant differences: `localhost`, `127.0.0.1`, and `[::1]` +/// stay three distinct hosts, and `ws` vs `wss`, non-default ports (including +/// the OTHER scheme's default), paths, and query strings stay distinct. +/// `None` for unparsable URLs, non-`ws(s)` schemes, userinfo, or fragments — +/// the caller falls back to exact comparison. +fn connection_target(raw: &str) -> Option { + let url = url::Url::parse(raw.trim()).ok()?; + let scheme = url.scheme().to_ascii_lowercase(); + if scheme != "ws" && scheme != "wss" { + return None; + } + if !url.username().is_empty() || url.password().is_some() || url.fragment().is_some() { + return None; + } + let host = { + let host = url.host_str()?.to_ascii_lowercase(); + host.strip_suffix('.').map(str::to_string).unwrap_or(host) + }; + let default_port = match scheme.as_str() { + "ws" => Some(80), + "wss" => Some(443), + _ => None, + }; + let port = url + .port_or_known_default() + .filter(|port| Some(*port) != default_port); + let path = match url.path() { + "/" => String::new(), + path => path.to_string(), + }; + let query = url.query().map(str::to_string); + Some(ConnectionTarget { + scheme, + host, + port, + path, + query, + }) +} + +/// See [`connection_target`]. +#[derive(PartialEq)] +struct ConnectionTarget { + scheme: String, + host: String, + port: Option, + path: String, + query: Option, +} + +/// Compare the actual relay connection authorities without applying canonical +/// runtime-key loopback folding. Unparsable inputs fail closed to exact, +/// trimmed equality. +pub(crate) fn connection_targets_match(live: &str, requested: &str) -> bool { + match (connection_target(live), connection_target(requested)) { + (Some(live), Some(requested)) => live == requested, + _ => connection_relay_url(live) == connection_relay_url(requested), + } +} + +fn ensure_connection_targets_match(live: &str, requested: &str) -> Result<(), String> { + if connection_targets_match(live, requested) { + Ok(()) + } else { + // Never include either URL: query strings may contain relay tokens and + // this error is persisted into `last_error` and rendered by the UI. + Err(concat!( + "connection-target conflict: a managed-agent runtime under this canonical identity ", + "belongs to a different connection target; manage it through its configured community" + ) + .into()) + } +} + +/// A live pair may only be reused for a start request that dials the same +/// connection target it already holds. Canonical keys fold host spellings, +/// so two distinct tenants can share one key; silently reusing across +/// spellings would report the requested tenant as started while the child +/// stays connected to the old one, and reconciliation would stop retrying. +/// Equivalence is by [`connection_target`], so harmless formatting drift +/// (host case, default port, root slash, FQDN dot) never reads as a +/// conflict. The error deliberately omits both URLs: they may carry query +/// tokens, and this string lands in `last_error` and the UI. +pub(crate) fn ensure_pair_connection_matches( + runtime: &ManagedAgentPairRuntime, + requested_relay_url: &str, +) -> Result<(), String> { + ensure_connection_targets_match(&runtime.connect_relay_url, requested_relay_url) +} + +/// Select a tracked canonical-key entry only when its stamped dial target is +/// connection-equivalent to the caller's requested community. +pub(crate) fn tracked_pair_runtime_for_target<'a>( + runtimes: &'a std::collections::HashMap, + key: &ManagedAgentRuntimeKey, + requested_relay_url: &str, +) -> Result, String> { + let Some(runtime) = runtimes.get(key) else { + return Ok(None); + }; + ensure_pair_connection_matches(runtime, requested_relay_url)?; + Ok(Some(runtime)) +} + +/// Hand-written so `connect_relay_url` never renders verbatim: +/// `normalize_relay_url` rejects userinfo but deliberately preserves query +/// strings, so `wss://relay.example/ws?token=...` is a valid value. Same +/// masking policy as `SpawnConfigSnapshot`'s `relay_url` (its single +/// redaction authority), pinned by the owning-process Debug sentinel test. +impl std::fmt::Debug for crate::managed_agents::ManagedAgentProcess { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + let mut s = f.debug_struct("ManagedAgentProcess"); + s.field("child", &self.child) + .field("log_path", &self.log_path) + .field( + "connect_relay_url", + &crate::managed_agents::spawn_snapshot::diff::MASK, + ) + .field("spawn_config", &self.spawn_config) + .field("setup_mode", &self.setup_mode) + .field("adapter_availability", &self.adapter_availability) + .field("start_nonce", &self.start_nonce); + #[cfg(windows)] + s.field("job", &self.job); + s.finish() + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/stop.rs b/desktop/src-tauri/src/managed_agents/runtime/stop.rs index 08bca15febb..ec717c2d554 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/stop.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/stop.rs @@ -4,8 +4,9 @@ use tauri::AppHandle; use super::{ append_log_marker, current_instance_id, now_iso, process_belongs_to_us, - process_has_buzz_marker, process_is_running, terminate_process, ManagedAgentPairRuntime, - ManagedAgentRecord, ManagedAgentRuntimeKey, + process_has_buzz_marker, process_is_running, terminate_process, + tracked_pair_runtime_for_target, ManagedAgentPairRuntime, ManagedAgentRecord, + ManagedAgentRuntimeKey, }; pub(crate) fn managed_agent_runtime_keys( @@ -19,6 +20,23 @@ pub(crate) fn managed_agent_runtime_keys( .collect() } +/// The relay URLs to dial when restarting `pubkey`'s live pairs: each pair's +/// stamped connection URL. Restart paths must use this instead of +/// `key.relay_url` — the canonical spelling folds loopback hosts, and on a +/// host-scoped multi-tenant relay that lands the child in the wrong (empty) +/// community. Collect before stopping: stopping removes the pair, and with it +/// the only in-memory copy of the configured spelling. +pub(crate) fn managed_agent_restart_targets( + runtimes: &HashMap, + pubkey: &str, +) -> Vec { + runtimes + .iter() + .filter(|(key, _)| key.pubkey.eq_ignore_ascii_case(pubkey)) + .map(|(_, runtime)| runtime.connect_relay_url.clone()) + .collect() +} + #[cfg(test)] pub(crate) fn managed_agent_runtime_relay_urls( runtimes: &HashMap, @@ -42,7 +60,12 @@ fn stop_managed_agent_pair( record: &mut ManagedAgentRecord, runtimes: &mut HashMap, key: &ManagedAgentRuntimeKey, + requested_relay_url: &str, ) -> Result<(), String> { + if tracked_pair_runtime_for_target(runtimes, key, requested_relay_url)?.is_none() { + return Ok(()); + } + super::ensure_pair_receipt_connection_matches(app, key, requested_relay_url)?; let Some(mut runtime) = runtimes.remove(key) else { return Ok(()); }; @@ -95,19 +118,57 @@ fn stop_managed_agent_pair( /// Terminate a legacy scalar-PID child (pre-pair records) and remove the /// agent-scoped pid file. Pair receipts are restored separately. fn stop_legacy_scalar_pid(app: &AppHandle, record: &mut ManagedAgentRecord) -> Result<(), String> { - if let Some(pid) = record.runtime_pid.take() { + if let Some(pid) = record.runtime_pid { if process_is_running(pid) && process_belongs_to_us(pid) && process_has_buzz_marker(pid, ¤t_instance_id(app)) { terminate_process(pid)?; } + record.runtime_pid = None; record.updated_at = now_iso(); } super::super::remove_agent_pid_file(app, &record.pubkey); Ok(()) } +fn clear_inactive_legacy_scalar_pid_with( + record: &mut ManagedAgentRecord, + is_live_owned: impl FnOnce(u32) -> bool, +) -> Result { + let Some(pid) = record.runtime_pid else { + return Ok(false); + }; + if is_live_owned(pid) { + return Err(concat!( + "connection-target conflict: a live legacy managed-agent runtime has no ", + "connection-target receipt; stop all runtimes for this agent before managing ", + "one community" + ) + .into()); + } + record.runtime_pid = None; + record.updated_at = now_iso(); + Ok(true) +} + +/// Pair-scoped stop may clear stale scalar bookkeeping, but a live legacy PID +/// has no tenant stamp and must be left to the explicit agent-wide stop path. +pub(crate) fn clear_inactive_legacy_scalar_pid( + app: &AppHandle, + record: &mut ManagedAgentRecord, +) -> Result<(), String> { + let cleared = clear_inactive_legacy_scalar_pid_with(record, |pid| { + process_is_running(pid) + && process_belongs_to_us(pid) + && process_has_buzz_marker(pid, ¤t_instance_id(app)) + })?; + if cleared { + super::super::remove_agent_pid_file(app, &record.pubkey); + } + Ok(()) +} + /// Stop the runtime pair this record resolves to for the active workspace /// (explicit relay pin, else the active workspace relay) — the pair-scoped /// counterpart of [`stop_managed_agent_process`], which drains every pair. @@ -115,8 +176,8 @@ fn stop_legacy_scalar_pid(app: &AppHandle, record: &mut ManagedAgentRecord) -> R /// Community-scoped surfaces (profile panel, Agents tab, auto-restart) stop /// through here so stopping an agent in one community never tears down its /// pairs in other communities. Clears the matching agent session cache -/// (pair-scoped when a pair key resolves). When no pair is tracked for this -/// workspace, only legacy scalar-PID cleanup runs. +/// (pair-scoped when a pair target resolves). A prior-session receipt must +/// match that same target before any pair-scoped cleanup proceeds. pub fn stop_managed_agent_workspace_pair( app: &AppHandle, record: &mut ManagedAgentRecord, @@ -124,22 +185,22 @@ pub fn stop_managed_agent_workspace_pair( ) -> Result<(), String> { use tauri::Manager; let state = app.state::(); - match super::workspace_pair_key(app, record) { - Some(pair_key) if runtimes.contains_key(&pair_key) => { - stop_managed_agent_pair(app, record, runtimes, &pair_key)?; + match super::workspace_pair_target(app, record) { + Some((pair_key, requested_relay)) if runtimes.contains_key(&pair_key) => { + stop_managed_agent_pair(app, record, runtimes, &pair_key, &requested_relay)?; state.clear_agent_session_cache(&pair_key); - super::super::remove_agent_pid_file(app, &record.pubkey); let now = now_iso(); - record.runtime_pid = None; record.updated_at = now.clone(); record.last_stopped_at = Some(now); record.last_error = None; record.last_error_code = None; } - Some(pair_key) => { + Some((pair_key, requested_relay)) => { // No tracked pair here — a pubkey-wide cache clear would disturb - // live pairs in other communities, so stay pair-scoped. - stop_legacy_scalar_pid(app, record)?; + // live pairs in other communities, so validate any receipt and + // stay pair-scoped. + super::terminate_untracked_pair_runtime(app, &pair_key, &requested_relay)?; + clear_inactive_legacy_scalar_pid(app, record)?; state.clear_agent_session_cache(&pair_key); } None => { @@ -162,27 +223,33 @@ pub fn stop_managed_agent_process( let mut errors = Vec::new(); for key in keys { - if let Err(error) = stop_managed_agent_pair(app, record, runtimes, &key) { - errors.push(format!("{}: {error}", key.relay_url)); + let Some(requested_relay) = runtimes + .get(&key) + .map(|runtime| runtime.connect_relay_url.clone()) + else { + continue; + }; + if let Err(error) = stop_managed_agent_pair(app, record, runtimes, &key, &requested_relay) { + errors.push(error); } } + if !errors.is_empty() { + return Err(format!( + "failed to stop one or more managed-agent runtimes: {}", + errors.join("; ") + )); + } + + // This is the explicit agent-wide path, so it is also the safe migration + // escape hatch for a live scalar PID whose community cannot be proven. + stop_legacy_scalar_pid(app, record)?; let now = now_iso(); - record.runtime_pid = None; record.updated_at = now.clone(); record.last_stopped_at = Some(now); record.last_error = None; record.last_error_code = None; - super::super::remove_agent_pid_file(app, &record.pubkey); - - if errors.is_empty() { - Ok(()) - } else { - Err(format!( - "failed to stop one or more managed-agent runtimes: {}", - errors.join("; ") - )) - } + Ok(()) } #[cfg(test)] @@ -247,4 +314,25 @@ mod tests { selected.sort_by(|left, right| left.relay_url.cmp(&right.relay_url)); assert_eq!(selected, vec![first, second]); } + + #[test] + fn pair_scoped_stop_refuses_a_live_unscoped_legacy_pid() { + let mut record = super::super::test_fixtures::minimal_record(&"aa".repeat(32)); + record.runtime_pid = Some(42); + + let error = clear_inactive_legacy_scalar_pid_with(&mut record, |_| true).unwrap_err(); + + assert!(error.contains("connection-target conflict")); + assert!(!error.contains("ws://")); + assert_eq!(record.runtime_pid, Some(42)); + } + + #[test] + fn pair_scoped_stop_clears_only_inactive_legacy_bookkeeping() { + let mut record = super::super::test_fixtures::minimal_record(&"aa".repeat(32)); + record.runtime_pid = Some(42); + + assert!(clear_inactive_legacy_scalar_pid_with(&mut record, |_| false).unwrap()); + assert_eq!(record.runtime_pid, None); + } } diff --git a/desktop/src-tauri/src/managed_agents/runtime/test_fixtures.rs b/desktop/src-tauri/src/managed_agents/runtime/test_fixtures.rs index 9076766b2e6..857b93f0c40 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/test_fixtures.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/test_fixtures.rs @@ -93,3 +93,84 @@ pub(super) fn fixture( effort_level: None, } } + +pub(crate) fn make_pair_runtime_with_connect_url( + connect_relay_url: &str, +) -> crate::managed_agents::ManagedAgentPairRuntime { + use std::process::{Command, Stdio}; + // Spawn a real child so ManagedAgentProcess's Child field is satisfied. + // `true` exits immediately with 0 — just a handle we need for type purposes. + // + // Absolute `/usr/bin/true` on unix (present on both macOS and Linux): + // parallel tests holding `lock_path_mutex` swap PATH to a tempdir, and a + // bare `true` lookup during that window fails with NotFound (observed + // flake). Windows keeps the PATH lookup — no test there swaps PATH. + #[cfg(unix)] + let program = "/usr/bin/true"; + #[cfg(windows)] + let program = "true"; + let child = Command::new(program) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::null()) + .spawn() + .expect("spawn true for placeholder"); + let process = crate::managed_agents::ManagedAgentProcess { + child, + log_path: std::path::PathBuf::new(), + connect_relay_url: connect_relay_url.to_string(), + spawn_config: crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot( + &minimal_record(&"cc".repeat(32)), + &[], + &[], + "wss://relay.example", + &Default::default(), + false, + ), + setup_mode: false, + adapter_availability: None, + start_nonce: "test-nonce".to_string(), + #[cfg(windows)] + job: None, + }; + crate::managed_agents::ManagedAgentPairRuntime::starting(process) +} + +pub(super) fn minimal_record(pubkey: &str) -> crate::managed_agents::ManagedAgentRecord { + serde_json::from_str(&format!( + r#"{{ + "pubkey": "{pubkey}", + "name": "test", + "private_key_nsec": "nsec1fake", + "relay_url": "", + "acp_command": "buzz-acp", + "agent_command": "buzz-agent", + "agent_args": [], + "mcp_command": "", + "turn_timeout_seconds": 320, + "system_prompt": null, + "model": null, + "provider": null, + "env_vars": {{}}, + "created_at": "2026-01-01T00:00:00Z", + "updated_at": "2026-01-01T00:00:00Z", + "last_started_at": null, + "last_stopped_at": null, + "last_exit_code": null, + "last_error": null + }}"# + )) + .expect("minimal_record fixture") +} + +pub(super) fn receipt_fixture( + key: crate::managed_agents::ManagedAgentRuntimeKey, +) -> crate::managed_agents::ManagedAgentRuntimeReceipt { + crate::managed_agents::ManagedAgentRuntimeReceipt { + key, + pid: std::process::id(), + desktop_instance_id: "test-instance".into(), + started_at: "now".into(), + connect_relay_url: None, + } +} diff --git a/desktop/src-tauri/src/managed_agents/runtime/tests.rs b/desktop/src-tauri/src/managed_agents/runtime/tests.rs index edb4fad422e..44e0945043b 100644 --- a/desktop/src-tauri/src/managed_agents/runtime/tests.rs +++ b/desktop/src-tauri/src/managed_agents/runtime/tests.rs @@ -1,5 +1,13 @@ use crate::managed_agents::known_acp_runtime; +#[test] +fn agent_connection_preserves_loopback_authority() { + assert_eq!( + super::connection_relay_url(" ws://localhost:3200/ "), + "ws://localhost:3200/" + ); +} + // ── desktop binary name tests ─────────────────────────────────────────── #[test] @@ -117,7 +125,9 @@ fn unknown_command_returns_none() { // ── build_respond_to_env tests ─────────────────────────────────────── -use super::test_fixtures::{expected_mode, expected_owner_only, fixture}; +use super::test_fixtures::{ + expected_mode, expected_owner_only, fixture, minimal_record, receipt_fixture, +}; use super::{build_respond_to_env, build_respond_to_env_with_policy}; use crate::managed_agents::types::{ManagedAgentRecord, RespondTo}; @@ -884,17 +894,6 @@ fn own_group_grandchild_detected_by_ancestor_walk() { // ── pair receipt validation tests ─────────────────────────────────────── -fn receipt_fixture( - key: crate::managed_agents::ManagedAgentRuntimeKey, -) -> crate::managed_agents::ManagedAgentRuntimeReceipt { - crate::managed_agents::ManagedAgentRuntimeReceipt { - key, - pid: std::process::id(), - desktop_instance_id: "test-instance".into(), - started_at: "now".into(), - } -} - #[test] fn receipt_validation_rejects_noncanonical_identity() { let mut receipt = receipt_fixture( @@ -987,12 +986,16 @@ fn unpinned_record_resolves_pair_key_per_workspace() { // read as running in workspace A and stopped in workspace B — the pair // key the summary looks up differs per workspace. let pubkey = "aa".repeat(32); - let key_a = super::resolve_workspace_pair_key(&pubkey, "", "wss://one.example").unwrap(); - let key_b = super::resolve_workspace_pair_key(&pubkey, "", "wss://two.example").unwrap(); + let (key_a, target_a) = + super::resolve_workspace_pair_target(&pubkey, "", "wss://one.example").unwrap(); + let (key_b, target_b) = + super::resolve_workspace_pair_target(&pubkey, "", "wss://two.example").unwrap(); let runtimes = std::collections::HashMap::from([(key_a.clone(), ())]); assert!(runtimes.contains_key(&key_a)); assert!(!runtimes.contains_key(&key_b)); + assert_eq!(target_a, "wss://one.example"); + assert_eq!(target_b, "wss://two.example"); } #[test] @@ -1001,32 +1004,48 @@ fn stored_relay_pin_is_ignored_in_pair_key_resolution() { // `relay_url` resolves the same per-workspace pair key an unpinned record // does, so summaries/stop act on the community being viewed. let pubkey = "aa".repeat(32); - let from_a = - super::resolve_workspace_pair_key(&pubkey, "wss://pinned.example", "wss://one.example") + let (from_a, target_a) = + super::resolve_workspace_pair_target(&pubkey, "wss://pinned.example", "wss://one.example") .unwrap(); - let from_b = - super::resolve_workspace_pair_key(&pubkey, "wss://pinned.example", "wss://two.example") + let (from_b, target_b) = + super::resolve_workspace_pair_target(&pubkey, "wss://pinned.example", "wss://two.example") .unwrap(); assert_ne!(from_a, from_b); assert_eq!(from_a.relay_url, "wss://one.example"); assert_eq!(from_b.relay_url, "wss://two.example"); + assert_eq!(target_a, "wss://one.example"); + assert_eq!(target_b, "wss://two.example"); } #[test] -fn workspace_pair_key_is_canonical() { - // Spawn stamps the canonical key; lookup must hit the same entry even - // when the workspace relay is written in a non-canonical form. +fn workspace_pair_target_keeps_canonical_key_and_exact_request() { let pubkey = "aa".repeat(32); - let stamped = super::resolve_workspace_pair_key(&pubkey, "", "wss://one.example").unwrap(); - let viewed = super::resolve_workspace_pair_key(&pubkey, "", "WSS://One.Example:443/").unwrap(); - assert_eq!(stamped, viewed); + let (stamped_key, stamped_target) = + super::resolve_workspace_pair_target(&pubkey, "", "wss://one.example").unwrap(); + let (viewed_key, viewed_target) = + super::resolve_workspace_pair_target(&pubkey, "", "WSS://One.Example:443/").unwrap(); + assert_eq!(stamped_key, viewed_key); + assert!(super::connection_targets_match( + &stamped_target, + &viewed_target + )); + + let (localhost_key, localhost_target) = + super::resolve_workspace_pair_target(&pubkey, "", "ws://localhost:3100").unwrap(); + let (ipv4_key, ipv4_target) = + super::resolve_workspace_pair_target(&pubkey, "", "ws://127.0.0.1:3100").unwrap(); + assert_eq!(localhost_key, ipv4_key); + assert!(!super::connection_targets_match( + &localhost_target, + &ipv4_target + )); } #[test] fn invalid_pubkey_resolves_no_pair_key() { // Key-less records (keys minted on first start) cannot form a pair key; // the summary must fall back to the stopped/legacy-pid path, not panic. - assert!(super::resolve_workspace_pair_key("not-a-key", "", "wss://one.example").is_none()); + assert!(super::resolve_workspace_pair_target("not-a-key", "", "wss://one.example").is_none()); } // ── Custom-harness orphan sweep coverage ───────────────────────────────────── @@ -1208,67 +1227,12 @@ fn receipt_invalid_when_process_not_running() { // ── Test helpers ──────────────────────────────────────────────────────────── -fn minimal_record(pubkey: &str) -> crate::managed_agents::ManagedAgentRecord { - serde_json::from_str(&format!( - r#"{{ - "pubkey": "{pubkey}", - "name": "test", - "private_key_nsec": "nsec1fake", - "relay_url": "", - "acp_command": "buzz-acp", - "agent_command": "buzz-agent", - "agent_args": [], - "mcp_command": "", - "turn_timeout_seconds": 320, - "system_prompt": null, - "model": null, - "provider": null, - "env_vars": {{}}, - "created_at": "2026-01-01T00:00:00Z", - "updated_at": "2026-01-01T00:00:00Z", - "last_started_at": null, - "last_stopped_at": null, - "last_exit_code": null, - "last_error": null - }}"# - )) - .expect("minimal_record fixture") +fn make_pair_runtime_placeholder() -> crate::managed_agents::ManagedAgentPairRuntime { + make_pair_runtime_with_connect_url("wss://relay.example") } -fn make_pair_runtime_placeholder() -> crate::managed_agents::ManagedAgentPairRuntime { - use std::process::{Command, Stdio}; - // Spawn a real child so ManagedAgentProcess's Child field is satisfied. - // `true` exits immediately with 0 — just a handle we need for type purposes. - // Absolute `/usr/bin/true` on unix (present on both macOS and Linux): - // parallel tests holding `lock_path_mutex` swap PATH to a tempdir, and a - // bare `true` lookup during that window fails with NotFound (observed - // flake). Windows keeps the PATH lookup — no test there swaps PATH. - #[cfg(unix)] - let program = "/usr/bin/true"; - #[cfg(windows)] - let program = "true"; - let child = Command::new(program) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("spawn true for placeholder"); - let process = crate::managed_agents::ManagedAgentProcess { - child, - log_path: Default::default(), - spawn_config: crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot( - &minimal_record(&"cc".repeat(32)), - &[], - &[], - "wss://relay.example", - &Default::default(), - false, - ), - setup_mode: false, - adapter_availability: None, - start_nonce: "test-nonce".to_string(), - #[cfg(windows)] - job: None, - }; - crate::managed_agents::ManagedAgentPairRuntime::starting(process) +fn make_pair_runtime_with_connect_url( + connect_relay_url: &str, +) -> crate::managed_agents::ManagedAgentPairRuntime { + super::test_fixtures::make_pair_runtime_with_connect_url(connect_relay_url) } diff --git a/desktop/src-tauri/src/managed_agents/runtime_commands.rs b/desktop/src-tauri/src/managed_agents/runtime_commands.rs index c0e55184b19..0f1a3effdee 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_commands.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_commands.rs @@ -3,13 +3,14 @@ use std::sync::atomic::Ordering; use tauri::{AppHandle, Emitter, Manager}; use super::{ - agent_readiness, append_log_marker, current_instance_id, find_managed_agent_mut, + agent_readiness, append_log_marker, clear_inactive_legacy_scalar_pid, connection_relay_url, + current_instance_id, ensure_pair_receipt_connection_matches, find_managed_agent_mut, load_global_agent_config, load_managed_agents, load_personas, managed_agent_runtime_log_path, process_is_running, record_agent_command, resolve_effective_agent_env, save_managed_agents, spawn_agent_child, terminate_process, terminate_untracked_pair_runtime, - write_agent_runtime_receipt, AgentReadiness, BackendKind, ManagedAgentPairRuntime, - ManagedAgentRuntimeKey, ManagedAgentRuntimeLifecycle, ManagedAgentRuntimeReceipt, - ManagedAgentRuntimeStatus, + tracked_pair_runtime_for_target, write_agent_runtime_receipt, AgentReadiness, BackendKind, + ManagedAgentPairRuntime, ManagedAgentRuntimeKey, ManagedAgentRuntimeLifecycle, + ManagedAgentRuntimeReceipt, ManagedAgentRuntimeStatus, }; use crate::app_state::AppState; @@ -60,7 +61,12 @@ fn status_for_with( ManagedAgentRuntimeStatus { pubkey: key.pubkey.clone(), relay_url: key.relay_url.clone(), - requested_relay_url, + // Callers that know the exact descriptor URL (reconcile) pass it in; + // otherwise fall back to the live pair's stamped connection URL so + // frontend start/restart actions re-dial the configured spelling + // instead of the canonical form. + requested_relay_url: requested_relay_url + .or_else(|| runtime.map(|runtime| runtime.connect_relay_url.clone())), local_setup, lifecycle: runtime .map(|runtime| runtime.lifecycle.clone()) @@ -160,16 +166,17 @@ pub fn list_managed_agent_runtimes( .managed_agent_processes .lock() .map_err(|e| e.to_string())?; - let exited_keys: Vec<_> = runtimes + let exited_pairs: Vec<_> = runtimes .iter_mut() .filter_map(|(key, runtime)| match runtime.child.try_wait() { - Ok(Some(_)) | Err(_) => Some(key.clone()), + Ok(Some(_)) | Err(_) => Some((key.clone(), runtime.connect_relay_url.clone())), Ok(None) => None, }) .collect(); - let records_changed = !exited_keys.is_empty(); + let records_changed = !exited_pairs.is_empty(); let mut statuses = Vec::new(); - for key in exited_keys { + for (key, requested_relay) in exited_pairs { + ensure_pair_receipt_connection_matches(&app, &key, &requested_relay)?; runtimes.remove(&key); super::remove_agent_runtime_receipt(&app, &key); state.clear_agent_session_cache(&key); @@ -184,7 +191,7 @@ pub fn list_managed_agent_runtimes( record, &key, None, - None, + Some(requested_relay), StatusInputs { personas: &personas, global: &global, @@ -268,28 +275,32 @@ fn start_pair( .managed_agent_processes .lock() .map_err(|e| e.to_string())?; - if runtimes - .get_mut(&key) - .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) - { - let status = status_for(&app, record, &key, runtimes.get(&key), None); - return Ok(status); + if tracked_pair_runtime_for_target(&runtimes, &key, &relay_url)?.is_some() { + if runtimes + .get_mut(&key) + .is_some_and(|runtime| runtime.child.try_wait().ok().flatten().is_none()) + { + let status = status_for(&app, record, &key, runtimes.get(&key), None); + return Ok(status); + } + ensure_pair_receipt_connection_matches(&app, &key, &relay_url)?; + runtimes.remove(&key); } - runtimes.remove(&key); - terminate_untracked_pair_runtime(&app, &key)?; + terminate_untracked_pair_runtime(&app, &key, &relay_url)?; let owner = state .keys .lock() .ok() .map(|keys| keys.public_key().to_hex()); - let mut process = spawn_agent_child(&app, record, &key.relay_url, lazy, owner.as_deref())?; + let mut process = spawn_agent_child(&app, record, &relay_url, lazy, owner.as_deref())?; let now = crate::util::now_iso(); let receipt = ManagedAgentRuntimeReceipt { key: key.clone(), pid: process.child.id(), desktop_instance_id: current_instance_id(&app), started_at: now.clone(), + connect_relay_url: Some(process.connect_relay_url.clone()), }; if let Err(error) = write_agent_runtime_receipt(&app, &receipt) { let _ = terminate_process(process.child.id()); @@ -331,7 +342,12 @@ pub fn stop_managed_agent_runtime( .managed_agent_processes .lock() .map_err(|e| e.to_string())?; - if let Some(mut runtime) = runtimes.remove(&key) { + let tracked = tracked_pair_runtime_for_target(&runtimes, &key, &relay_url)?.is_some(); + if tracked { + ensure_pair_receipt_connection_matches(&app, &key, &relay_url)?; + let Some(mut runtime) = runtimes.remove(&key) else { + return Err("managed-agent runtime changed during stop".into()); + }; let stop_result = if process_is_running(runtime.child.id()) { terminate_process(runtime.child.id()) } else { @@ -362,14 +378,20 @@ pub fn stop_managed_agent_runtime( // failure the receipt stays on disk (terminate_untracked_pair_runtime // only removes it after the child exits), mirroring the tracked // path's keep-until-success invariant. - terminate_untracked_pair_runtime(&app, &key)?; + terminate_untracked_pair_runtime(&app, &key, &relay_url)?; + clear_inactive_legacy_scalar_pid(&app, record)?; } super::remove_agent_runtime_receipt(&app, &key); state.clear_agent_session_cache(&key); - record.runtime_pid = None; record.updated_at = crate::util::now_iso(); record.last_stopped_at = Some(record.updated_at.clone()); - let status = status_for(&app, record, &key, None, None); + let status = status_for( + &app, + record, + &key, + None, + Some(connection_relay_url(&relay_url)), + ); drop(runtimes); save_managed_agents(&app, &records)?; emit_status(&app, &status); @@ -403,7 +425,10 @@ async fn probe_agent_relay_access( let key = ManagedAgentRuntimeKey::new(record.pubkey.clone(), &requested_relay_url)?; let keys = nostr::Keys::parse(record.private_key_nsec.trim()) .map_err(|error| format!("invalid managed-agent key: {error}"))?; - let api_base = crate::relay::relay_http_base_url(&key.relay_url); + // Probe the community the user actually configured: relay tenancy is + // host-derived, so the canonical key spelling can resolve to a different + // (or unmapped) community than the requested URL. + let api_base = crate::relay::relay_http_base_url(&requested_relay_url); tokio::time::timeout( std::time::Duration::from_secs(10), crate::relay::query_relay_at_with_keys( @@ -504,7 +529,10 @@ pub async fn reconcile_managed_agent_runtimes( Ok((record, key, requested)) => { match start_pair( record.pubkey.clone(), - key.relay_url.clone(), + // Start with the requested spelling so the spawned + // child connects to the community that was probed. + // start_pair re-derives the same canonical key. + requested.clone(), true, Some(&record.updated_at), app.clone(), diff --git a/desktop/src-tauri/src/managed_agents/runtime_types.rs b/desktop/src-tauri/src/managed_agents/runtime_types.rs index 4862cedbae3..d65a45dcf45 100644 --- a/desktop/src-tauri/src/managed_agents/runtime_types.rs +++ b/desktop/src-tauri/src/managed_agents/runtime_types.rs @@ -83,8 +83,10 @@ impl ManagedAgentPairRuntime { pub struct ManagedAgentRuntimeStatus { pub pubkey: String, pub relay_url: String, - /// Exact descriptor URL echoed only by reconcile result rows so callers can - /// correlate a canonical response without normalizing on the frontend. + /// The requested (non-canonical) URL when known: the exact submitted + /// descriptor on reconcile result rows, otherwise the live pair's actual + /// connection spelling. Lets callers correlate rows across spellings + /// without normalizing on the frontend. #[serde(skip_serializing_if = "Option::is_none")] pub requested_relay_url: Option, pub local_setup: bool, @@ -117,4 +119,12 @@ pub struct ManagedAgentRuntimeReceipt { pub pid: u32, pub desktop_instance_id: String, pub started_at: String, + /// The exact relay URL the child was dialed with, persisted alongside the + /// pair identity. Optional so receipts written before this field existed + /// still deserialize; absent marks a valid legacy receipt whose connection + /// spelling is unknown. When present it must canonicalize back to + /// `key.relay_url` (see `valid_agent_runtime_receipt_with`), so a receipt + /// can never smuggle a connection URL belonging to a different pair. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub connect_relay_url: Option, } diff --git a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff.rs b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff.rs index 0ae3009bae3..a8c8fda13bc 100644 --- a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff.rs +++ b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff.rs @@ -21,7 +21,7 @@ use crate::managed_agents::AcpAvailabilityStatus; /// the process was spawned with. const ADAPTER_AVAILABILITY_FIELD: &str = "adapter_availability"; -const MASK: &str = "••••"; +pub(crate) const MASK: &str = "••••"; /// One changed field. `field` is a dotted path built from serde field names, /// with dynamic map keys appended verbatim (`env.OPENAI_API_KEY`). The UI diff --git a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs index e21dc4735c7..280c58f76d2 100644 --- a/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs +++ b/desktop/src-tauri/src/managed_agents/spawn_snapshot/diff/tests.rs @@ -440,6 +440,9 @@ fn no_sentinel_reaches_the_owning_process_debug_output() { let process = crate::managed_agents::ManagedAgentProcess { child, log_path: std::path::PathBuf::new(), + // Token-bearing URL: the process Debug impl must mask this field, + // exactly like the snapshot masks its own relay_url. + connect_relay_url: RELAY_WITH_TOKEN.to_string(), spawn_config: seeded_with_sentinels(), setup_mode: false, adapter_availability: None, diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index 9049482de3a..f847396ef90 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -1,5 +1,5 @@ use serde::{Deserialize, Serialize}; -use std::{collections::BTreeMap, path::PathBuf, process::Child}; +use std::{collections::BTreeMap, path::PathBuf}; #[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)] #[serde(tag = "type", rename_all = "snake_case")] @@ -446,38 +446,6 @@ pub struct ManagedAgentRecord { pub effort_level: Option, } -#[derive(Debug)] -pub struct ManagedAgentProcess { - pub child: Child, - pub log_path: PathBuf, - /// The effective spawn config this process was launched with (see - /// `spawn_snapshot::SpawnConfigSnapshot`). Runtime-only — never persisted. - /// The summary builder recomputes a prospective snapshot and reports - /// differing fields via `ManagedAgentSummary::restart_diff`. Agents - /// adopted via `runtime_pid` have none; their config is unknown. - pub spawn_config: super::spawn_snapshot::SpawnConfigSnapshot, - /// Whether this process was spawned in setup-listener mode (i.e. - /// `BUZZ_ACP_SETUP_PAYLOAD` was set at launch because the agent was - /// `NotReady`). Runtime-only — never persisted. Used by - /// `install_acp_runtime` to target only stuck agents for auto-restart, - /// excluding healthy in-pool agents. - pub setup_mode: bool, - /// Adapter availability status stamped at spawn time for runtimes with a - /// version gate (currently codex only; `None` for all others). Runtime-only - /// — never persisted. The summary builder compares this against the current - /// cached availability and sets `needs_restart` on drift, catching out-of- - /// band adapter changes that Phase-1 auto-restart doesn't cover. - pub adapter_availability: Option, - /// Unpredictable identity shared only with this harness generation. - pub start_nonce: String, - /// Win32 Job Object owning the harness + its entire process tree. Closing - /// the handle (via `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`) kills the whole - /// tree — the Windows mirror of the Unix process-group teardown. `None` - /// if job creation/assignment failed (we fall back to `Child::kill()`). - #[cfg(windows)] - pub job: Option, -} - #[derive(Debug, Clone, Serialize)] pub struct ManagedAgentSummary { pub pubkey: String, @@ -978,6 +946,8 @@ pub fn resolve_mint_behavioral_defaults( mod catalog_source; pub use catalog_source::CatalogSource; +mod process; +pub use process::ManagedAgentProcess; mod relay_mesh; pub use relay_mesh::RelayMeshConfig; mod requests; diff --git a/desktop/src-tauri/src/managed_agents/types/process.rs b/desktop/src-tauri/src/managed_agents/types/process.rs new file mode 100644 index 00000000000..aeace8824d5 --- /dev/null +++ b/desktop/src-tauri/src/managed_agents/types/process.rs @@ -0,0 +1,39 @@ +//! Runtime-only process handle, split from `types.rs` (file-size cap). + +use std::{path::PathBuf, process::Child}; + +use super::AcpAvailabilityStatus; + +pub struct ManagedAgentProcess { + pub child: Child, + pub log_path: PathBuf, + /// The exact URL this child dials (`BUZZ_RELAY_URL`); may differ from the + /// canonical key spelling. Restarts must reuse it; receipts persist it. + pub connect_relay_url: String, + /// The effective spawn config this process was launched with (see + /// `spawn_snapshot::SpawnConfigSnapshot`). Runtime-only — never persisted. + /// The summary builder recomputes a prospective snapshot and reports + /// differing fields via `ManagedAgentSummary::restart_diff`. Agents + /// adopted via `runtime_pid` have none; their config is unknown. + pub spawn_config: crate::managed_agents::spawn_snapshot::SpawnConfigSnapshot, + /// Whether this process was spawned in setup-listener mode (i.e. + /// `BUZZ_ACP_SETUP_PAYLOAD` was set at launch because the agent was + /// `NotReady`). Runtime-only — never persisted. Used by + /// `install_acp_runtime` to target only stuck agents for auto-restart, + /// excluding healthy in-pool agents. + pub setup_mode: bool, + /// Adapter availability status stamped at spawn time for runtimes with a + /// version gate (currently codex only; `None` for all others). Runtime-only + /// — never persisted. The summary builder compares this against the current + /// cached availability and sets `needs_restart` on drift, catching out-of- + /// band adapter changes that Phase-1 auto-restart doesn't cover. + pub adapter_availability: Option, + /// Unpredictable identity shared only with this harness generation. + pub start_nonce: String, + /// Win32 Job Object owning the harness + its entire process tree. Closing + /// the handle (via `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE`) kills the whole + /// tree — the Windows mirror of the Unix process-group teardown. `None` + /// if job creation/assignment failed (we fall back to `Child::kill()`). + #[cfg(windows)] + pub job: Option, +} diff --git a/desktop/src/features/agents/managedAgentReconciliationPlan.test.mjs b/desktop/src/features/agents/managedAgentReconciliationPlan.test.mjs index 21327c30d92..b4e1f66cc47 100644 --- a/desktop/src/features/agents/managedAgentReconciliationPlan.test.mjs +++ b/desktop/src/features/agents/managedAgentReconciliationPlan.test.mjs @@ -2,12 +2,12 @@ import assert from "node:assert/strict"; import test from "node:test"; import { - canonicalCommunityRelays, classifyReconcileResult, + connectionTargetCommunityRelays, pendingReconcileRelays, reconcileRetryDelayMs, } from "./managedAgentReconciliationPlan.ts"; -import { canonicalRelayUrl } from "./managedAgentRuntimeStatus.ts"; +import { connectionTargetUrl } from "./managedAgentRuntimeStatus.ts"; test("reconcileRetryDelayMs walks a capped backoff then gives up", () => { assert.equal(reconcileRetryDelayMs(1), 5_000); @@ -17,37 +17,40 @@ test("reconcileRetryDelayMs walks a capped backoff then gives up", () => { assert.equal(reconcileRetryDelayMs(0), null); }); -test("canonicalCommunityRelays dedupes by canonical form, keeps stored spelling", () => { - const relays = canonicalCommunityRelays( +test("connectionTargetCommunityRelays attempts localhost and 127 separately", () => { + const relays = connectionTargetCommunityRelays( [ { relayUrl: "ws://localhost:3000" }, - // Same relay, different spelling — folds onto the first entry. + // Connection-equivalent formatting dedupes onto the first entry. + { relayUrl: "ws://LOCALHOST:3000/" }, + // Canonical runtime-key aliases are still distinct tenant attempts. { relayUrl: "ws://127.0.0.1:3000" }, { relayUrl: "wss://relay.example" }, // Unparsable entries are dropped rather than reconciled. { relayUrl: "not a url" }, ], - canonicalRelayUrl, + connectionTargetUrl, ); assert.deepEqual( [...relays.entries()], [ - ["ws://127.0.0.1:3000", "ws://localhost:3000"], + ["ws://localhost:3000", "ws://localhost:3000"], + ["ws://127.0.0.1:3000", "ws://127.0.0.1:3000"], ["wss://relay.example", "wss://relay.example"], ], ); }); test("pendingReconcileRelays skips reconciled and in-flight relays", () => { - const canonicalToRequested = new Map([ - ["ws://127.0.0.1:3000", "ws://localhost:3000"], + const targetToRequested = new Map([ + ["ws://localhost:3000", "ws://localhost:3000"], ["wss://a.example", "wss://a.example"], ["wss://b.example", "wss://b.example"], ]); const pending = pendingReconcileRelays( - canonicalToRequested, + targetToRequested, new Set(["wss://a.example"]), - new Set(["ws://127.0.0.1:3000"]), + new Set(["ws://localhost:3000"]), ); assert.deepEqual(pending, ["wss://b.example"]); }); @@ -55,7 +58,7 @@ test("pendingReconcileRelays skips reconciled and in-flight relays", () => { test("classifyReconcileResult marks the whole batch failed when the call throws", () => { const attempted = ["wss://a.example", "wss://b.example"]; assert.deepEqual( - classifyReconcileResult(attempted, null, canonicalRelayUrl), + classifyReconcileResult(attempted, null, connectionTargetUrl), { succeeded: [], failed: attempted, @@ -63,8 +66,8 @@ test("classifyReconcileResult marks the whole batch failed when the call throws" ); }); -test("classifyReconcileResult splits by Failed rows, matching on requested URL", () => { - const attempted = ["ws://127.0.0.1:3000", "wss://b.example"]; +test("classifyReconcileResult keeps colliding loopback targets separate", () => { + const attempted = ["ws://localhost:3000", "ws://127.0.0.1:3000"]; const rows = [ // Started cleanly on the loopback relay — reconciled. { @@ -77,23 +80,23 @@ test("classifyReconcileResult splits by Failed rows, matching on requested URL", error: null, logPath: null, }, - // Failed on b.example — stays failing so it is retried. + // The canonical-key collision for 127 is explicit and retried separately. { pubkey: "aa", - relayUrl: "wss://b.example", - requestedRelayUrl: "wss://b.example", + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://127.0.0.1:3000", localSetup: true, lifecycle: "failed", pid: null, - error: "relay access probe timed out", + error: "connection-target conflict", logPath: null, }, ]; assert.deepEqual( - classifyReconcileResult(attempted, rows, canonicalRelayUrl), + classifyReconcileResult(attempted, rows, connectionTargetUrl), { - succeeded: ["ws://127.0.0.1:3000"], - failed: ["wss://b.example"], + succeeded: ["ws://localhost:3000"], + failed: ["ws://127.0.0.1:3000"], }, ); }); @@ -102,7 +105,7 @@ test("classifyReconcileResult treats a relay with no rows as reconciled", () => // A community with no eligible auto-start agents produces no rows; it must // still count as reconciled so the hook stops retrying it. assert.deepEqual( - classifyReconcileResult(["wss://a.example"], [], canonicalRelayUrl), + classifyReconcileResult(["wss://a.example"], [], connectionTargetUrl), { succeeded: ["wss://a.example"], failed: [] }, ); }); diff --git a/desktop/src/features/agents/managedAgentReconciliationPlan.ts b/desktop/src/features/agents/managedAgentReconciliationPlan.ts index 58d9b58b73a..a8731d1075e 100644 --- a/desktop/src/features/agents/managedAgentReconciliationPlan.ts +++ b/desktop/src/features/agents/managedAgentReconciliationPlan.ts @@ -30,37 +30,37 @@ export function reconcileRetryDelayMs(failureCount: number): number | null { } /** - * Canonicalize the configured community relays, dropping duplicates and - * unparsable entries. Maps canonical URL -> the raw `relayUrl` to submit to the - * backend (first occurrence wins), so the reconcile call still speaks the - * community's stored spelling while all bookkeeping is keyed canonically. + * Group configured communities by connection-equivalent relay target, dropping + * duplicates and unparsable entries. Distinct loopback host spellings remain + * separate targets so the backend can return an explicit canonical-key conflict. + * Maps target -> raw stored `relayUrl` (first equivalent occurrence wins). */ -export function canonicalCommunityRelays( +export function connectionTargetCommunityRelays( communities: readonly { relayUrl: string }[], - canonicalize: (url: string) => string | null, + connectionTarget: (url: string) => string | null, ): Map { - const byCanonical = new Map(); + const byTarget = new Map(); for (const community of communities) { - const canonical = canonicalize(community.relayUrl); - if (canonical === null || byCanonical.has(canonical)) continue; - byCanonical.set(canonical, community.relayUrl); + const target = connectionTarget(community.relayUrl); + if (target === null || byTarget.has(target)) continue; + byTarget.set(target, community.relayUrl); } - return byCanonical; + return byTarget; } /** * Configured relays that still need a reconcile attempt: not yet reconciled - * cleanly and not currently in flight. Returns canonical URLs. + * cleanly and not currently in flight. Returns connection-target URLs. */ export function pendingReconcileRelays( - canonicalToRequested: ReadonlyMap, + targetToRequested: ReadonlyMap, reconciled: ReadonlySet, inFlight: ReadonlySet, ): string[] { const pending: string[] = []; - for (const canonical of canonicalToRequested.keys()) { - if (reconciled.has(canonical) || inFlight.has(canonical)) continue; - pending.push(canonical); + for (const target of targetToRequested.keys()) { + if (reconciled.has(target) || inFlight.has(target)) continue; + pending.push(target); } return pending; } @@ -74,21 +74,21 @@ export function pendingReconcileRelays( export function classifyReconcileResult( attempted: readonly string[], rows: readonly ManagedAgentRuntimeStatus[] | null, - canonicalize: (url: string) => string | null, + connectionTarget: (url: string) => string | null, ): { succeeded: string[]; failed: string[] } { if (rows === null) { return { succeeded: [], failed: [...attempted] }; } - const failedRelays = new Set(); + const failedTargets = new Set(); for (const row of rows) { if (row.lifecycle !== "failed") continue; - const canonical = canonicalize(row.requestedRelayUrl ?? row.relayUrl); - if (canonical !== null) failedRelays.add(canonical); + const target = connectionTarget(row.requestedRelayUrl ?? row.relayUrl); + if (target !== null) failedTargets.add(target); } const succeeded: string[] = []; const failed: string[] = []; for (const relay of attempted) { - if (failedRelays.has(relay)) failed.push(relay); + if (failedTargets.has(relay)) failed.push(relay); else succeeded.push(relay); } return { succeeded, failed }; diff --git a/desktop/src/features/agents/managedAgentRuntimeHooks.test.mjs b/desktop/src/features/agents/managedAgentRuntimeHooks.test.mjs index 3e961b58544..4d9752d99cb 100644 --- a/desktop/src/features/agents/managedAgentRuntimeHooks.test.mjs +++ b/desktop/src/features/agents/managedAgentRuntimeHooks.test.mjs @@ -1,7 +1,12 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { restartManagedAgentPair } from "./managedAgentRuntimeHooks.ts"; +import { + mergeManagedAgentRuntimeStatuses, + replaceManagedAgentRuntimeStatus, + restartManagedAgentPair, + stoppedPairMatchesActiveCommunity, +} from "./managedAgentRuntimeHooks.ts"; // --------------------------------------------------------------------------- // restartManagedAgentPair: discriminating regression tests for the pair @@ -17,15 +22,89 @@ const PUBKEY = "deadbeef".repeat(8); const RELAY = "wss://relay.example"; /** Returns a resolved-status stub sufficient for the return-type assertion. */ -function makeStatus() { +function makeStatus(overrides = {}) { return { pubkey: PUBKEY, relayUrl: RELAY, localSetup: true, lifecycle: "running", + ...overrides, }; } +test("cache merge retains colliding localhost success and 127 failure rows", () => { + const liveLocalhost = makeStatus({ + pubkey: PUBKEY.toUpperCase(), + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://localhost:3000", + lifecycle: "ready", + }); + const failedIpv4 = makeStatus({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://127.0.0.1:3000", + lifecycle: "failed", + error: "connection-target conflict", + }); + const baseline = [liveLocalhost]; + + const merged = mergeManagedAgentRuntimeStatuses(baseline, baseline, [ + failedIpv4, + ]); + + assert.equal(merged.length, 2); + assert.equal( + merged.find((row) => row.requestedRelayUrl.includes("localhost")) + ?.lifecycle, + "ready", + ); + assert.equal( + merged.find((row) => row.requestedRelayUrl.includes("127.0.0.1")) + ?.lifecycle, + "failed", + ); +}); + +test("action replacement updates only the authoritative connection target", () => { + const liveLocalhost = makeStatus({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://localhost:3000", + lifecycle: "ready", + }); + const failedIpv4 = makeStatus({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://127.0.0.1:3000", + lifecycle: "failed", + }); + const stoppedLocalhost = makeStatus({ + pubkey: PUBKEY.toUpperCase(), + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://LOCALHOST:3000/", + lifecycle: "stopped", + }); + + const replaced = replaceManagedAgentRuntimeStatus( + [liveLocalhost, failedIpv4], + stoppedLocalhost, + ); + assert.equal(replaced.length, 2); + assert.equal(replaced[0], stoppedLocalhost); + assert.equal(replaced[1], failedIpv4); +}); + +test("stop badge clearing never crosses loopback tenants", () => { + assert.equal( + stoppedPairMatchesActiveCommunity( + "ws://localhost:3000", + "ws://127.0.0.1:3000", + ), + false, + ); + assert.equal( + stoppedPairMatchesActiveCommunity("ws://LOCALHOST:80/", "ws://localhost"), + true, + ); +}); + test("test_pair_restart_stop_success_start_failure_clear_still_ran", async () => { // Stop succeeds, start throws. The clear must have fired — badge is gone // regardless of the start failure. On the old combined-command approach, diff --git a/desktop/src/features/agents/managedAgentRuntimeHooks.ts b/desktop/src/features/agents/managedAgentRuntimeHooks.ts index 96a3abc78d4..6923258afcb 100644 --- a/desktop/src/features/agents/managedAgentRuntimeHooks.ts +++ b/desktop/src/features/agents/managedAgentRuntimeHooks.ts @@ -17,7 +17,10 @@ import { stopManagedAgentRuntime, } from "@/shared/api/tauriManagedAgents"; import type { ManagedAgentRuntimeStatus } from "@/shared/api/types"; -import { canonicalRelayUrl } from "./managedAgentRuntimeStatus"; +import { + connectionTargetsMatch, + managedAgentRuntimeKey, +} from "./managedAgentRuntimeStatus"; export const managedAgentRuntimesQueryKey = ["managed-agent-runtimes"] as const; @@ -27,14 +30,20 @@ export function mergeManagedAgentRuntimeStatuses( reconciled: readonly ManagedAgentRuntimeStatus[], ): ManagedAgentRuntimeStatus[] { const baselineByPair = new Map( - (baseline ?? []).map((runtime) => [runtimePairKey(runtime), runtime]), + (baseline ?? []).map((runtime) => [ + managedAgentRuntimeKey(runtime), + runtime, + ]), ); const currentByPair = new Map( - (current ?? []).map((runtime) => [runtimePairKey(runtime), runtime]), + (current ?? []).map((runtime) => [ + managedAgentRuntimeKey(runtime), + runtime, + ]), ); const reconciledPairs = new Set(); const merged = reconciled.map((runtime) => { - const key = runtimePairKey(runtime); + const key = managedAgentRuntimeKey(runtime); reconciledPairs.add(key); const currentRuntime = currentByPair.get(key); const baselineRuntime = baselineByPair.get(key); @@ -47,13 +56,25 @@ export function mergeManagedAgentRuntimeStatuses( }); for (const runtime of current ?? []) { - if (!reconciledPairs.has(runtimePairKey(runtime))) merged.push(runtime); + if (!reconciledPairs.has(managedAgentRuntimeKey(runtime))) { + merged.push(runtime); + } } return merged; } -function runtimePairKey(runtime: ManagedAgentRuntimeStatus): string { - return JSON.stringify([runtime.pubkey, runtime.relayUrl]); +export function replaceManagedAgentRuntimeStatus( + current: readonly ManagedAgentRuntimeStatus[], + runtime: ManagedAgentRuntimeStatus, +): ManagedAgentRuntimeStatus[] { + const runtimeKey = managedAgentRuntimeKey(runtime); + const index = current.findIndex( + (candidate) => managedAgentRuntimeKey(candidate) === runtimeKey, + ); + if (index === -1) return [...current, runtime]; + return current.map((candidate, candidateIndex) => + candidateIndex === index ? runtime : candidate, + ); } export function cacheReconciledManagedAgentRuntimes( @@ -131,12 +152,8 @@ export function clearActiveTurnsForAgentOnStop( // Pair-scoped: only clear when the stopped pair's relay matches the active // community. A mismatch means the stop targets a different community's // store — leave it alone. - const activeCanonical = canonicalRelayUrl(activeCommunity.relayUrl); - const stoppedCanonical = canonicalRelayUrl(relayUrl); if ( - activeCanonical === null || - stoppedCanonical === null || - activeCanonical !== stoppedCanonical + !stoppedPairMatchesActiveCommunity(activeCommunity.relayUrl, relayUrl) ) { return; } @@ -147,6 +164,13 @@ export function clearActiveTurnsForAgentOnStop( clearActiveTurnsForAgent(pubkey); } +export function stoppedPairMatchesActiveCommunity( + activeRelayUrl: string, + stoppedRelayUrl: string, +): boolean { + return connectionTargetsMatch(activeRelayUrl, stoppedRelayUrl); +} + /** * Execute a pair restart as stop → relay-scoped badge clear → start. * @@ -206,22 +230,15 @@ export function useManagedAgentRuntimeAction() { // For stop-only: clear stale working badges immediately. The restart // path already clears at the stop-success boundary inside mutationFn. if (action === "stop") { - clearActiveTurnsForAgentOnStop(runtime.pubkey, runtime.relayUrl); + clearActiveTurnsForAgentOnStop( + runtime.pubkey, + runtime.requestedRelayUrl ?? runtime.relayUrl, + ); } queryClient.setQueryData( managedAgentRuntimesQueryKey, - (current = []) => { - const index = current.findIndex( - (candidate) => - candidate.pubkey === runtime.pubkey && - candidate.relayUrl === runtime.relayUrl, - ); - if (index === -1) return [...current, runtime]; - return current.map((candidate, candidateIndex) => - candidateIndex === index ? runtime : candidate, - ); - }, + (current = []) => replaceManagedAgentRuntimeStatus(current, runtime), ); }, }); diff --git a/desktop/src/features/agents/managedAgentRuntimeStatus.test.mjs b/desktop/src/features/agents/managedAgentRuntimeStatus.test.mjs index edd368eccd6..2abc49cd188 100644 --- a/desktop/src/features/agents/managedAgentRuntimeStatus.test.mjs +++ b/desktop/src/features/agents/managedAgentRuntimeStatus.test.mjs @@ -5,6 +5,8 @@ import { agentCommunityAvailability, agentCommunityStatusDetail, canonicalRelayUrl, + connectionTargetsMatch, + connectionTargetUrl, findManagedAgentRuntime, managedAgentRuntimeKey, } from "./managedAgentRuntimeStatus.ts"; @@ -60,6 +62,31 @@ test("pair key cannot collide at component boundaries", () => { managedAgentRuntimeKey(runtime({ pubkey: "ab", relayUrl: "c" })), managedAgentRuntimeKey(runtime({ pubkey: "a", relayUrl: "bc" })), ); + assert.equal( + managedAgentRuntimeKey( + runtime({ + pubkey: "AA", + requestedRelayUrl: "ws://LOCALHOST:80/", + }), + ), + managedAgentRuntimeKey( + runtime({ pubkey: "aa", requestedRelayUrl: "ws://localhost" }), + ), + ); + assert.notEqual( + managedAgentRuntimeKey( + runtime({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://localhost:3000", + }), + ), + managedAgentRuntimeKey( + runtime({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://127.0.0.1:3000", + }), + ), + ); }); test("selects one relay without collapsing same-pubkey pairs", () => { @@ -96,6 +123,17 @@ test("canonicalRelayUrl mirrors the backend pair-key normalization", () => { assert.equal(canonicalRelayUrl("ws://[::1]:3000"), "ws://127.0.0.1:3000"); assert.equal(canonicalRelayUrl("https://relay.example"), null); assert.equal(canonicalRelayUrl("not a url"), null); + assert.equal(canonicalRelayUrl("wss://alice@relay.example"), null); + assert.equal(canonicalRelayUrl("wss://relay.example#east"), null); + assert.equal(canonicalRelayUrl("wss://relay.example#"), null); + assert.equal( + canonicalRelayUrl("wss://@relay.example"), + "wss://relay.example", + ); + assert.equal( + canonicalRelayUrl("wss://:@relay.example"), + "wss://relay.example", + ); }); test("matches a stored community URL against canonical backend rows", () => { @@ -111,3 +149,175 @@ test("matches a stored community URL against canonical backend rows", () => { undefined, ); }); + +test("requestedRelayUrl is authoritative: distinct loopback tenants never alias", () => { + // One agent, one canonical key, child actually dialed to localhost while + // both loopback communities are configured simultaneously. + const runtimes = [ + runtime({ + relayUrl: "ws://127.0.0.1:3000", + requestedRelayUrl: "ws://localhost:3000", + lifecycle: "ready", + }), + ]; + // The community that was actually dialed resolves the runtime... + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "ws://localhost:3000")?.lifecycle, + "ready", + ); + // ...the OTHER loopback community must not: its card would otherwise show + // this agent as running there, and its Stop/Restart action could target + // the localhost child through the shared canonical key. + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "ws://127.0.0.1:3000"), + undefined, + ); +}); + +test("requested-URL matching folds connection-equivalent spellings only", () => { + const runtimes = [ + runtime({ + relayUrl: "ws://127.0.0.1:80", + requestedRelayUrl: "ws://localhost:80", + }), + ]; + for (const spelling of [ + "ws://LocalHost:80", + "ws://localhost", + "ws://localhost/", + ]) { + assert.ok( + findManagedAgentRuntime(runtimes, "aa", spelling), + `equivalent spelling must match: ${spelling}`, + ); + } + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "wss://localhost"), + undefined, + ); +}); + +test("connectionTargetUrl folds formatting but preserves tenant hosts", () => { + assert.equal(connectionTargetUrl("ws://LocalHost:80/"), "ws://localhost"); + assert.equal( + connectionTargetUrl("wss://relay.example."), + "wss://relay.example", + ); + assert.notEqual( + connectionTargetUrl("ws://localhost:3000"), + connectionTargetUrl("ws://127.0.0.1:3000"), + ); + assert.equal(connectionTargetUrl("https://relay.example"), null); + // Query rides along; the root slash folds with or without one. + assert.equal( + connectionTargetUrl("ws://relay.example?token=x"), + connectionTargetUrl("ws://relay.example/?token=x"), + ); + assert.notEqual( + connectionTargetUrl("ws://relay.example?token=x"), + connectionTargetUrl("ws://relay.example?token=y"), + ); + // The other scheme's default port is a real port, not foldable. + assert.notEqual( + connectionTargetUrl("ws://relay.example:443"), + connectionTargetUrl("ws://relay.example"), + ); + // Three loopback spellings, three distinct tenants. + assert.notEqual( + connectionTargetUrl("ws://[::1]:3000"), + connectionTargetUrl("ws://localhost:3000"), + ); + assert.notEqual( + connectionTargetUrl("ws://[::1]:3000"), + connectionTargetUrl("ws://127.0.0.1:3000"), + ); + assert.equal( + connectionTargetsMatch(" ws://LOCALHOST:80/ ", "ws://localhost"), + true, + ); + assert.equal( + connectionTargetsMatch("ws://localhost:3000", "ws://127.0.0.1:3000"), + false, + ); + assert.equal(connectionTargetsMatch(" invalid ", "invalid"), true); +}); + +test("invalid connection targets keep exact-string fallback instead of aliasing", () => { + assert.equal(connectionTargetUrl("wss://alice@relay.example"), null); + assert.equal(connectionTargetUrl("wss://relay.example#east"), null); + assert.equal( + connectionTargetsMatch( + "wss://alice@relay.example", + "wss://bob@relay.example", + ), + false, + ); + assert.equal( + connectionTargetsMatch("wss://alice@relay.example", "wss://relay.example"), + false, + ); + assert.equal( + connectionTargetsMatch( + "wss://relay.example#east", + "wss://relay.example#west", + ), + false, + ); + assert.equal( + connectionTargetsMatch("wss://relay.example#east", "wss://relay.example"), + false, + ); + assert.equal( + connectionTargetsMatch( + "wss://alice@relay.example", + "wss://alice@relay.example", + ), + true, + ); + assert.equal( + connectionTargetsMatch( + " wss://relay.example#east ", + "wss://relay.example#east", + ), + true, + ); + assert.equal( + connectionTargetsMatch("ws://LOCALHOST:80/", "ws://localhost"), + true, + ); + assert.equal( + connectionTargetsMatch("ws://localhost:3000", "ws://127.0.0.1:3000"), + false, + ); + assert.equal( + connectionTargetUrl("wss://@relay.example"), + "wss://relay.example", + ); + assert.equal(connectionTargetUrl("wss://relay.example#"), null); + assert.equal( + connectionTargetsMatch("wss://@relay.example", "wss://relay.example"), + true, + ); + assert.equal( + connectionTargetsMatch("wss://relay.example#", "wss://relay.example"), + false, + ); +}); + +test("legacy rows without requestedRelayUrl reject fragment and credential aliases", () => { + const runtimes = [runtime({ relayUrl: "wss://relay.example" })]; + assert.ok(findManagedAgentRuntime(runtimes, "aa", "wss://relay.example")); + assert.ok(findManagedAgentRuntime(runtimes, "aa", "wss://@relay.example")); + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "wss://alice@relay.example"), + undefined, + ); + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "wss://relay.example#east"), + undefined, + ); + assert.equal( + findManagedAgentRuntime(runtimes, "aa", "wss://relay.example#"), + undefined, + ); +}); diff --git a/desktop/src/features/agents/managedAgentRuntimeStatus.ts b/desktop/src/features/agents/managedAgentRuntimeStatus.ts index c3a952f7d5d..b9fa11be7dc 100644 --- a/desktop/src/features/agents/managedAgentRuntimeStatus.ts +++ b/desktop/src/features/agents/managedAgentRuntimeStatus.ts @@ -36,9 +36,16 @@ export function agentCommunityStatusDetail( } export function managedAgentRuntimeKey( - runtime: Pick, + runtime: Pick< + ManagedAgentRuntimeStatus, + "pubkey" | "relayUrl" | "requestedRelayUrl" + >, ): string { - return JSON.stringify([runtime.pubkey, runtime.relayUrl]); + const requestedRelay = runtime.requestedRelayUrl ?? runtime.relayUrl; + return JSON.stringify([ + runtime.pubkey.toLowerCase(), + connectionTargetUrl(requestedRelay) ?? requestedRelay, + ]); } export type ManagedAgentPairAction = "start" | "stop" | "restart"; @@ -62,22 +69,30 @@ export const MANAGED_AGENT_PAIR_ACTION_LABELS: Record< restart: "Restart Agent", }; +function hasExplicitFragment(raw: string): boolean { + return raw.includes("#"); +} + /** * Canonicalize a relay URL the way the backend keys runtime pairs, so a * stored community URL (e.g. `ws://localhost:3000`) matches backend rows * (`ws://127.0.0.1:3000`). Mirrors buzz-core's `normalize_relay_url` * (`crates/buzz-core/src/relay.rs`): lowercase host, loopback hosts folded * to 127.0.0.1, default ports and root-path trailing slash stripped. - * Returns null when the URL cannot be parsed as ws/wss. + * Returns null when the URL cannot be parsed as ws/wss, or when it carries + * credentials or an explicit fragment. Empty userinfo (`wss://@host`) is + * treated as no credentials, matching the backend normalizer. */ export function canonicalRelayUrl(raw: string): string | null { + const trimmed = raw.trim(); let url: URL; try { - url = new URL(raw.trim()); + url = new URL(trimmed); } catch { return null; } if (url.protocol !== "ws:" && url.protocol !== "wss:") return null; + if (url.username || url.password || hasExplicitFragment(trimmed)) return null; let host = url.hostname.toLowerCase(); if (host === "localhost" || host === "[::1]" || host.startsWith("127.")) { host = "127.0.0.1"; @@ -92,22 +107,62 @@ export function canonicalRelayUrl(raw: string): string | null { ); } +/** + * Comparable connection target mirroring the backend's tenancy authority + * (buzz-core's `tenant::normalize_host`): lowercase host, strip an explicit + * default port and the FQDN root dot, fold the root-path slash - WITHOUT + * folding loopback spellings, which are distinct tenants on a host-scoped + * relay. Returns null when the URL cannot be parsed as ws/wss, or when it + * carries credentials or an explicit fragment; callers then fall back to + * exact comparison. Empty userinfo is treated as no credentials. + */ +export function connectionTargetUrl(raw: string): string | null { + const trimmed = raw.trim(); + let url: URL; + try { + url = new URL(trimmed); + } catch { + return null; + } + if (url.protocol !== "ws:" && url.protocol !== "wss:") return null; + if (url.username || url.password || hasExplicitFragment(trimmed)) return null; + const host = url.hostname.toLowerCase().replace(/\.$/, ""); + const defaultPort = url.protocol === "ws:" ? "80" : "443"; + const port = url.port && url.port !== defaultPort ? `:${url.port}` : ""; + const path = url.pathname === "/" ? "" : url.pathname; + return `${url.protocol}//${host}${port}${path}${url.search}`; +} + +/** Match relay connection authorities without canonical loopback folding. */ +export function connectionTargetsMatch(left: string, right: string): boolean { + const leftTarget = connectionTargetUrl(left); + const rightTarget = connectionTargetUrl(right); + if (leftTarget !== null && rightTarget !== null) { + return leftTarget === rightTarget; + } + return left.trim() === right.trim(); +} + export function findManagedAgentRuntime( runtimes: readonly ManagedAgentRuntimeStatus[], pubkey: string, relayUrl: string, ): ManagedAgentRuntimeStatus | undefined { const normalizedPubkey = pubkey.toLowerCase(); - // Backend rows carry the canonical pair URL; the caller passes the - // community's stored URL, which may differ in spelling (localhost vs - // 127.0.0.1, default port, trailing slash). Compare canonically, keeping - // the exact-string checks as a fallback for unparsable stored URLs. const canonical = canonicalRelayUrl(relayUrl); - return runtimes.find( - (runtime) => - runtime.pubkey.toLowerCase() === normalizedPubkey && - (runtime.relayUrl === relayUrl || - runtime.requestedRelayUrl === relayUrl || - (canonical !== null && runtime.relayUrl === canonical)), - ); + return runtimes.find((runtime) => { + if (runtime.pubkey.toLowerCase() !== normalizedPubkey) return false; + // A row carrying the actual dial spelling is authoritative: canonical + // matching would alias distinct loopback tenants that share one runtime + // key, letting the wrong community's card claim - and stop - this child. + if (runtime.requestedRelayUrl != null) { + return connectionTargetsMatch(runtime.requestedRelayUrl, relayUrl); + } + // Legacy rows without the dial spelling keep the canonical fallback + // (exact-string check first, for unparsable stored URLs). + return ( + runtime.relayUrl === relayUrl || + (canonical !== null && runtime.relayUrl === canonical) + ); + }); } diff --git a/desktop/src/features/agents/useManagedAgentRuntimeReconciliation.ts b/desktop/src/features/agents/useManagedAgentRuntimeReconciliation.ts index f2fb2416b92..638cd5facfa 100644 --- a/desktop/src/features/agents/useManagedAgentRuntimeReconciliation.ts +++ b/desktop/src/features/agents/useManagedAgentRuntimeReconciliation.ts @@ -2,8 +2,8 @@ import { useQueryClient } from "@tanstack/react-query"; import * as React from "react"; import { - canonicalCommunityRelays, classifyReconcileResult, + connectionTargetCommunityRelays, pendingReconcileRelays, reconcileRetryDelayMs, } from "@/features/agents/managedAgentReconciliationPlan"; @@ -11,7 +11,7 @@ import { cacheReconciledManagedAgentRuntimes, managedAgentRuntimesQueryKey, } from "@/features/agents/managedAgentRuntimeHooks"; -import { canonicalRelayUrl } from "@/features/agents/managedAgentRuntimeStatus"; +import { connectionTargetUrl } from "@/features/agents/managedAgentRuntimeStatus"; import type { ManagedAgentRuntimeStatus } from "@/shared/api/types"; import { reconcileManagedAgentRuntimes } from "@/shared/api/tauriManagedAgents"; @@ -19,7 +19,7 @@ import { reconcileManagedAgentRuntimes } from "@/shared/api/tauriManagedAgents"; * Bootstrap a lazy harness pair for every auto-start local agent in every * configured community, incrementally and with retry. * - * Reconciliation is keyed by canonical relay URL: each configured relay is + * Reconciliation bookkeeping is keyed by connection target: each configured relay is * reconciled once it appears (so adding a community mid-session spawns pairs * there without needing the add flow to also switch communities), and a relay * whose reconcile fails is retried with a capped backoff (5s / 30s / 2m) rather @@ -31,11 +31,11 @@ export function useManagedAgentRuntimeReconciliation( communities: readonly { relayUrl: string }[], ): void { const queryClient = useQueryClient(); - // Canonical relay URLs that have reconciled cleanly — never re-hit. + // Connection targets that have reconciled cleanly — never re-hit. const reconciledRef = React.useRef>(new Set()); - // Canonical relay URLs with a reconcile call in flight — not re-dispatched. + // Connection targets with a reconcile call in flight — not re-dispatched. const inFlightRef = React.useRef>(new Set()); - // Consecutive failures per canonical relay URL, driving the retry backoff. + // Consecutive failures per connection target, driving the retry backoff. const failuresRef = React.useRef>(new Map()); const retryTimerRef = React.useRef | null>( null, @@ -73,23 +73,23 @@ export function useManagedAgentRuntimeReconciliation( }; const runReconcile = () => { - const canonicalToRequested = canonicalCommunityRelays( + const targetToRequested = connectionTargetCommunityRelays( communities, - canonicalRelayUrl, + connectionTargetUrl, ); // Forget bookkeeping for relays that are no longer configured so the sets // stay bounded and re-adding a removed community reconciles it afresh. for (const done of [...reconciledRef.current]) { - if (!canonicalToRequested.has(done)) reconciledRef.current.delete(done); + if (!targetToRequested.has(done)) reconciledRef.current.delete(done); } for (const failing of [...failuresRef.current.keys()]) { - if (!canonicalToRequested.has(failing)) { + if (!targetToRequested.has(failing)) { failuresRef.current.delete(failing); } } const pending = pendingReconcileRelays( - canonicalToRequested, + targetToRequested, reconciledRef.current, inFlightRef.current, ); @@ -100,7 +100,7 @@ export function useManagedAgentRuntimeReconciliation( for (const relay of pending) inFlightRef.current.add(relay); const targets = pending.map((relay) => ({ - relayUrl: canonicalToRequested.get(relay) as string, + relayUrl: targetToRequested.get(relay) as string, })); const baseline = queryClient.getQueryData( managedAgentRuntimesQueryKey, @@ -109,11 +109,15 @@ export function useManagedAgentRuntimeReconciliation( void reconcileManagedAgentRuntimes(targets) .then((runtimes) => { cacheReconciledManagedAgentRuntimes(queryClient, baseline, runtimes); - return classifyReconcileResult(pending, runtimes, canonicalRelayUrl); + return classifyReconcileResult( + pending, + runtimes, + connectionTargetUrl, + ); }) .catch((error) => { console.warn("[managed-agent-runtimes] reconcile failed:", error); - return classifyReconcileResult(pending, null, canonicalRelayUrl); + return classifyReconcileResult(pending, null, connectionTargetUrl); }) .then(({ succeeded, failed }) => { for (const relay of pending) inFlightRef.current.delete(relay); diff --git a/desktop/src/shared/api/types.ts b/desktop/src/shared/api/types.ts index 41c63f7be97..9ab248b4f5d 100644 --- a/desktop/src/shared/api/types.ts +++ b/desktop/src/shared/api/types.ts @@ -289,7 +289,7 @@ export type ManagedAgentRuntimeLifecycle = export type ManagedAgentRuntimeStatus = { pubkey: string; - /** Exact submitted descriptor, present only on startup reconcile results. */ + /** Requested (non-canonical) URL: reconcile descriptor or live pair connection spelling. */ requestedRelayUrl?: string; /** Canonical, backend-owned pair identity component. Do not normalize in TS. */ relayUrl: string;