fix(desktop): preserve runtime community authority and refuse lossy receipts

Signed-off-by: Logan Johnson <loganj@squareup.com>
This commit is contained in:
Logan Johnson
2026-09-05 00:20:33 -04:00
parent ef58314dd5
commit 546f5ec16a
21 changed files with 1324 additions and 211 deletions

View File

@@ -1,4 +1,4 @@
//! Canonical relay identities shared by runtime components.
//! Legacy canonical relay identities shared by compatibility consumers.
use thiserror::Error;
use url::{Host, Url};
@@ -23,17 +23,16 @@ pub enum NormalizeRelayUrlError {
MissingHost,
}
/// Canonicalize a WebSocket relay URL for use as a runtime identity key.
/// Canonicalize a WebSocket relay URL for legacy equivalence consumers.
///
/// This is the sole normalizer for `(agent, relay)` process identity. It keeps
/// the WebSocket scheme, lowercases DNS hosts, folds all loopback spellings to
/// `127.0.0.1`, removes default ports and a root slash, and preserves non-root
/// paths and queries. It deliberately is **not** the NIP-42 AUTH comparison
/// helper in `buzz-auth`: AUTH validation is a security boundary with narrower
/// equivalence rules and must not be widened by runtime-key canonicalization.
/// Bestie scope and pollen/profile migration retain this historical behavior:
/// keep the WebSocket scheme, lowercase DNS hosts, fold all loopback spellings
/// to `127.0.0.1`, remove default ports and trailing slashes, and preserve
/// queries. Managed-agent process identity intentionally uses a scoped
/// host-preserving normalizer because relay hosts are tenant authorities.
///
/// Connection code may retain the configured URL; this canonical form is for
/// identity, receipts, status and deduplication.
/// This deliberately is **not** the NIP-42 AUTH comparison helper in
/// `buzz-auth`; changing either equivalence contract requires a separate review.
pub fn normalize_relay_url(raw: &str) -> Result<String, NormalizeRelayUrlError> {
let mut url = Url::parse(raw.trim())
.map_err(|error| NormalizeRelayUrlError::InvalidUrl(error.to_string()))?;

View File

@@ -176,8 +176,8 @@ async fn start_local_agent_with_preflight(
replay_floor_unix: Option<u64>,
) -> Result<ManagedAgentSummary, String> {
let launch_owner = workspace_owner_hex(state)?;
// Runtime keys fold loopback aliases for process bookkeeping, not tenant
// identity. Preserve the workspace authority across the preflight await.
// Runtime keys preserve the workspace host authority. Bind that same
// authority across the preflight await so the eventual spawn cannot move.
let launch_relay = crate::relay::bind_expected_relay_scope(
expected_relay_url,
relay_ws_url_with_override(state),

View File

@@ -267,7 +267,10 @@ mod tests {
let community = "ws://localhost:3037";
let agent = Keys::generate().public_key().to_hex();
let key = ManagedAgentRuntimeKey::new(&agent, community).unwrap();
assert_ne!(key.relay_url, community);
assert_eq!(key.relay_url, community);
let numeric_community = "ws://127.0.0.1:3037";
let numeric_key = ManagedAgentRuntimeKey::new(&agent, numeric_community).unwrap();
assert_ne!(key, numeric_key);
let scope = super::super::retention::RetentionScope {
db_path: scoped_retention_db_path(&root.join("agents"), community, &owner),
relay_url: community.into(),
@@ -293,9 +296,16 @@ mod tests {
)
.unwrap();
assert!(check_launch(app.handle(), &key, community, Some(&owner), None).is_err());
// The numeric community is a different authority, even though the
// process bookkeeping key historically folds the two spellings.
assert!(check_launch(app.handle(), &key, &key.relay_url, Some(&owner), None).is_ok());
// The numeric community is a different authority with its own runtime
// identity and fence database.
assert!(check_launch(
app.handle(),
&numeric_key,
numeric_community,
Some(&owner),
None
)
.is_ok());
let resume = capture_resume(app.handle(), &key, community, &owner).unwrap();
assert!(check_launch(app.handle(), &key, community, Some(&owner), Some(&resume)).is_ok());
let newer = request(&keys, &target, 101);

View File

@@ -397,12 +397,12 @@ pub async fn restore_managed_agents_on_launch(
continue;
};
let now = util::now_iso();
let receipt = super::ManagedAgentRuntimeReceipt {
key: key.clone(),
pid: process.child.id(),
desktop_instance_id: super::current_instance_id(app),
started_at: now.clone(),
};
let receipt = super::ManagedAgentRuntimeReceipt::new(
key.clone(),
process.child.id(),
super::current_instance_id(app),
now.clone(),
);
if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) {
let _ = super::terminate_process(process.child.id());
let _ = process.child.wait();

View File

@@ -41,11 +41,13 @@ 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,
select_pair_runtime_receipt_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,
with_pair_runtime_receipt_authority,
};
mod orphan_sweep;
@@ -108,8 +110,8 @@ fn persona_drift_state(
/// 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(
app: &AppHandle,
pub(crate) fn workspace_pair_key<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &ManagedAgentRecord,
) -> Option<ManagedAgentRuntimeKey> {
let state = app.state::<crate::app_state::AppState>();
@@ -444,8 +446,8 @@ pub(crate) fn spawn_with_effort_proof(
/// publishes the triggering message before this spawn and passes its send
/// timestamp here so the harness's first REQ replays past that message no
/// matter how long the spawn takes. buzz-acp clamps stale floors to ~15 min.
pub fn spawn_agent_child(
app: &AppHandle,
pub fn spawn_agent_child<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &ManagedAgentRecord,
relay_url: &str,
lazy: bool,
@@ -466,8 +468,8 @@ pub fn spawn_agent_child(
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn spawn_agent_child_with_broker(
app: &AppHandle,
pub(crate) fn spawn_agent_child_with_broker<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &ManagedAgentRecord,
relay_url: &str,
lazy: bool,
@@ -917,8 +919,8 @@ pub(crate) fn spawn_agent_child_with_broker(
/// exact workspace-relay read the caller's scope assertion passed on; it never
/// re-reads the mutable override (see `relay::scope`). The key comes from
/// [`bound_runtime_key`] — the seam the spawn-key regressions exercise.
pub fn start_managed_agent_process(
app: &AppHandle,
pub fn start_managed_agent_process<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &mut ManagedAgentRecord,
runtimes: &mut HashMap<ManagedAgentRuntimeKey, ManagedAgentPairRuntime>,
owner_hex: Option<&str>,
@@ -944,6 +946,10 @@ pub fn start_managed_agent_process(
// Scalar PIDs are migration-only and never establish pair liveness.
record.runtime_pid = None;
// A prior-session receipt is the only untracked process this pair may
// replace. Selection enforces host-preserving authority provenance and
// uses the ordinary process-tree termination contract.
terminate_untracked_pair_runtime(app, &key)?;
let mut process = spawn_agent_child(
app,
record,
@@ -954,12 +960,12 @@ pub fn start_managed_agent_process(
resume,
)?;
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(),
};
let receipt = super::ManagedAgentRuntimeReceipt::new(
key.clone(),
process.child.id(),
current_instance_id(app),
now.clone(),
);
if let Err(error) = super::write_agent_runtime_receipt(app, &receipt) {
let _ = terminate_process(process.child.id());
let _ = process.child.wait();
@@ -979,7 +985,7 @@ pub fn start_managed_agent_process(
}
#[cfg(test)]
mod test_fixtures;
pub(super) mod test_fixtures;
#[cfg(test)]
mod tests;

View File

@@ -0,0 +1,364 @@
//! Runtime authority migration and production Start regressions.
use super::super as runtime;
use super::receipt_fixture;
#[test]
fn legacy_receipt_validation_uses_legacy_rendering_for_global_ownership() {
let mut receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new(
"aa".repeat(32),
"wss://relay.example?mode=one",
)
.unwrap(),
);
receipt.authority_version = 0;
// url::Url serialization retained the root slash before a query in V0,
// while the scoped runtime renderer intentionally removes that root slash.
receipt.key.relay_url = "wss://relay.example/?mode=one".into();
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
assert!(runtime::valid_agent_runtime_receipt_with(
&path,
&receipt,
"test-instance",
|_| true,
|_, _| true,
));
}
#[test]
fn legacy_normalizer_loss_boundaries_are_pinned_to_the_real_url_renderer() {
let normalize = buzz_core_pkg::relay::normalize_relay_url;
assert_eq!(
normalize("wss://relay.example/room/").unwrap(),
"wss://relay.example/room"
);
assert_eq!(
normalize("wss://relay.example/room?tail=/").unwrap(),
"wss://relay.example/room?tail="
);
assert_eq!(
normalize("wss://relay.example/?mode=one").unwrap(),
"wss://relay.example/?mode=one"
);
assert_eq!(
normalize("wss://relay.example/?").unwrap(),
"wss://relay.example/?"
);
}
#[test]
fn replacement_removes_receipt_only_after_confirmed_exit() {
use std::cell::{Cell, RefCell};
let receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap(),
);
let path = std::path::Path::new("pair.json");
let terminated = Cell::new(None);
let polls = Cell::new(0);
let removed = RefCell::new(None);
runtime::terminate_runtime_receipt_with(
path,
&receipt,
|pid| {
terminated.set(Some(pid));
Ok(())
},
|_| {
let poll = polls.get() + 1;
polls.set(poll);
poll < 2
},
|path| *removed.borrow_mut() = Some(path.to_path_buf()),
)
.unwrap();
assert_eq!(terminated.get(), Some(receipt.pid));
assert_eq!(polls.get(), 2);
assert_eq!(removed.into_inner().as_deref(), Some(path));
}
#[test]
fn replacement_failure_keeps_receipt() {
use std::cell::Cell;
let receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap(),
);
let removed = Cell::new(false);
let error = runtime::terminate_runtime_receipt_with(
std::path::Path::new("pair.json"),
&receipt,
|_| Err("signal failed".into()),
|_| false,
|_| removed.set(true),
)
.unwrap_err();
assert_eq!(error, "signal failed");
assert!(!removed.get());
}
#[cfg(unix)]
#[test]
fn production_start_refuses_live_unversioned_receipt_before_spawn() {
let _path_guard = crate::managed_agents::lock_path_mutex();
let temp = tempfile::tempdir().unwrap();
let old_home = std::env::var_os("HOME");
let old_xdg = std::env::var_os("XDG_DATA_HOME");
struct RestoreEnv(Option<std::ffi::OsString>, Option<std::ffi::OsString>);
impl Drop for RestoreEnv {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var("HOME", value),
None => std::env::remove_var("HOME"),
}
match self.1.take() {
Some(value) => std::env::set_var("XDG_DATA_HOME", value),
None => std::env::remove_var("XDG_DATA_HOME"),
}
}
}
let _restore_env = RestoreEnv(old_home, old_xdg);
std::env::set_var("HOME", temp.path());
std::env::set_var("XDG_DATA_HOME", temp.path());
let relay = "wss://relay.example";
let pubkey = "aa".repeat(32);
let app = tauri::test::mock_builder()
.manage(crate::app_state::build_app_state())
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.unwrap();
let instance_id = runtime::current_instance_id(app.handle());
let mut child = runtime::test_fixtures::MarkedTestChild::spawn(&instance_id).unwrap();
assert!(runtime::process_has_buzz_marker(child.id(), &instance_id));
let key = crate::managed_agents::ManagedAgentRuntimeKey::new(&pubkey, relay).unwrap();
let receipt = crate::managed_agents::ManagedAgentRuntimeReceipt {
authority_version: 0,
key: key.clone(),
pid: child.id(),
desktop_instance_id: instance_id,
started_at: "now".into(),
};
crate::managed_agents::write_agent_runtime_receipt(app.handle(), &receipt).unwrap();
let mut record = runtime::test_fixtures::fixture(
crate::managed_agents::RespondTo::OwnerOnly,
Vec::new(),
None,
);
record.pubkey = pubkey;
record.acp_command = "a-command-that-must-not-be-resolved".into();
let bound = crate::relay::bind_expected_relay_scope(None, relay.into()).unwrap();
let mut runtimes = std::collections::HashMap::new();
let error = runtime::start_managed_agent_process(
app.handle(),
&mut record,
&mut runtimes,
None,
&bound,
None,
None,
)
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
assert!(!error.contains("crash"));
assert!(runtimes.is_empty());
assert!(child.child_mut().try_wait().unwrap().is_none());
assert!(
crate::managed_agents::read_all_agent_runtime_receipts(app.handle())
.iter()
.any(|(_, candidate)| candidate == &receipt)
);
crate::managed_agents::remove_agent_runtime_receipt(app.handle(), &key);
}
#[test]
fn receipt_selection_refuses_ambiguous_unversioned_loopback_authority() {
let mut receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://127.0.0.1:3000")
.unwrap(),
);
receipt.authority_version = 0;
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
for requested_relay in [
"ws://127.0.0.1:3000",
"ws://localhost:3000",
"ws://127.0.0.2:3000",
"ws://[::1]:3000",
] {
let requested =
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), requested_relay)
.unwrap();
let error = runtime::select_pair_runtime_receipt_with(
vec![(path.clone(), receipt.clone())],
&requested,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap_err();
assert!(
error.contains("cannot prove the requested community authority"),
"legacy receipt must not prove {requested_relay}"
);
}
}
#[test]
fn receipt_selection_keeps_versioned_loopback_authorities_disjoint() {
let receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://127.0.0.1:3000")
.unwrap(),
);
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
let requested =
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "ws://localhost:3000")
.unwrap();
let selected = runtime::select_pair_runtime_receipt_with(
vec![(path, receipt)],
&requested,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap();
assert!(selected.is_none());
}
#[test]
fn receipt_selection_refuses_unversioned_non_loopback_lossy_urls() {
let pubkey = "aa".repeat(32);
for (stored_relay, requested_relay) in [
("wss://relay.example/room", "wss://relay.example/room"),
("wss://relay.example/room", "wss://relay.example/room/"),
(
"wss://relay.example/?mode=one",
"wss://relay.example?mode=one",
),
(
"wss://relay.example/room?tail=",
"wss://relay.example/room?tail=/",
),
] {
let mut receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new(
pubkey.clone(),
"wss://relay.example",
)
.unwrap(),
);
receipt.authority_version = 0;
receipt.key.relay_url = stored_relay.into();
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
let requested =
crate::managed_agents::ManagedAgentRuntimeKey::new(pubkey.clone(), requested_relay)
.unwrap();
let error = runtime::select_pair_runtime_receipt_with(
vec![(path, receipt)],
&requested,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap_err();
assert!(
error.contains("cannot prove the requested community authority"),
"legacy {stored_relay} must not prove {requested_relay}"
);
}
}
#[test]
fn legacy_renderer_collapses_repeated_root_slashes_but_modern_keys_do_not() {
let parsed = url::Url::parse("wss://relay.example//").unwrap();
assert_eq!(parsed.path(), "//");
assert_eq!(
buzz_core_pkg::relay::normalize_relay_url("wss://relay.example//").unwrap(),
"wss://relay.example"
);
let root =
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap();
let repeated = crate::managed_agents::ManagedAgentRuntimeKey::new(
"aa".repeat(32),
"wss://relay.example//",
)
.unwrap();
assert_ne!(root, repeated);
}
#[test]
fn receipt_selection_refuses_unversioned_root_authority() {
let key =
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap();
let mut receipt = receipt_fixture(key.clone());
receipt.authority_version = 0;
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
let error = runtime::select_pair_runtime_receipt_with(
vec![(path, receipt)],
&key,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
}
#[test]
fn repeated_root_request_is_refused_for_colliding_unversioned_receipt() {
let pubkey = "aa".repeat(32);
let stored =
crate::managed_agents::ManagedAgentRuntimeKey::new(&pubkey, "wss://relay.example").unwrap();
let requested =
crate::managed_agents::ManagedAgentRuntimeKey::new(&pubkey, "wss://relay.example//")
.unwrap();
let mut receipt = receipt_fixture(stored);
receipt.authority_version = 0;
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
let error = runtime::select_pair_runtime_receipt_with(
vec![(path, receipt)],
&requested,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
}
#[test]
fn receipt_selection_refuses_unknown_future_authority_version() {
let key =
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap();
let mut receipt = receipt_fixture(key.clone());
receipt.authority_version = crate::managed_agents::RUNTIME_AUTHORITY_RECEIPT_VERSION + 1;
let path = std::path::PathBuf::from(format!("{}.json", receipt.key.runtime_id()));
let error = runtime::select_pair_runtime_receipt_with(
vec![(path, receipt)],
&key,
"test-instance",
|_| true,
|_, _| true,
)
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
}
// ── workspace pair-key resolution (summary/stop scoping) ────────────────

View File

@@ -395,7 +395,9 @@ pub(crate) fn valid_agent_runtime_receipt(
/// Injectable version of `valid_agent_runtime_receipt` for testing.
/// `is_running(pid)` and `has_marker(pid, instance_id)` can be substituted by
/// test doubles without spawning real processes.
/// test doubles without spawning real processes. Validity here proves only
/// instance ownership for global cleanup; pair actions must additionally use
/// `select_pair_runtime_receipt_with` to establish authority provenance.
pub(crate) fn valid_agent_runtime_receipt_with(
path: &std::path::Path,
receipt: &super::super::ManagedAgentRuntimeReceipt,
@@ -403,12 +405,21 @@ pub(crate) fn valid_agent_runtime_receipt_with(
is_running: impl Fn(u32) -> bool,
has_marker: impl Fn(u32, &str) -> bool,
) -> bool {
let Ok(canonical) =
let key_rendering_is_valid = if receipt.authority_version == 0 {
receipt.key.pubkey.len() == 64
&& receipt
.key
.pubkey
.bytes()
.all(|byte| byte.is_ascii_hexdigit())
&& receipt.key.pubkey == receipt.key.pubkey.to_ascii_lowercase()
&& buzz_core_pkg::relay::normalize_relay_url(&receipt.key.relay_url)
.is_ok_and(|legacy| legacy == receipt.key.relay_url)
} else {
ManagedAgentRuntimeKey::new(receipt.key.pubkey.clone(), &receipt.key.relay_url)
else {
return false;
.is_ok_and(|canonical| canonical == receipt.key)
};
canonical == receipt.key
key_rendering_is_valid
&& path.file_name().and_then(|name| name.to_str())
== Some(&format!("{}.json", receipt.key.runtime_id()))
&& receipt.desktop_instance_id == instance_id
@@ -441,20 +452,93 @@ pub(super) fn terminate_runtime_receipt_with(
))
}
fn receipt_has_proven_pair_authority(receipt: &super::super::ManagedAgentRuntimeReceipt) -> bool {
receipt.authority_version == super::super::RUNTIME_AUTHORITY_RECEIPT_VERSION
}
fn unversioned_receipt_may_ambiguously_match(
receipt: &super::super::ManagedAgentRuntimeReceipt,
key: &ManagedAgentRuntimeKey,
) -> bool {
if receipt.authority_version != 0 || !receipt.key.pubkey.eq_ignore_ascii_case(&key.pubkey) {
return false;
}
buzz_core_pkg::relay::normalize_relay_url(&key.relay_url)
.is_ok_and(|legacy_relay| legacy_relay == receipt.key.relay_url)
}
/// Select a receipt only when it proves the requested pair authority.
///
/// Unversioned receipts used a lossy normalizer that folded loopback hosts and
/// stripped every terminal slash. They can establish instance ownership for
/// global cleanup but cannot prove which new runtime key a pair-scoped action
/// owns, including an apparently exact root URL: `wss://h//` and `wss://h`
/// share the V0 rendering but are distinct modern keys.
pub(crate) fn select_pair_runtime_receipt_with(
entries: Vec<(std::path::PathBuf, super::super::ManagedAgentRuntimeReceipt)>,
key: &ManagedAgentRuntimeKey,
instance_id: &str,
is_running: impl Fn(u32) -> bool,
has_marker: impl Fn(u32, &str) -> bool,
) -> Result<Option<(std::path::PathBuf, super::super::ManagedAgentRuntimeReceipt)>, String> {
let mut selected = None;
for (path, receipt) in entries {
if !valid_agent_runtime_receipt_with(&path, &receipt, instance_id, &is_running, &has_marker)
|| !receipt.key.pubkey.eq_ignore_ascii_case(&key.pubkey)
{
continue;
}
let exact = receipt.key == *key;
if (exact && !receipt_has_proven_pair_authority(&receipt))
|| unversioned_receipt_may_ambiguously_match(&receipt, key)
{
return Err(
"Runtime receipt cannot prove the requested community authority; quit Buzz Desktop normally, then reopen it before retrying this Start or Stop"
.into(),
);
}
if exact {
selected = Some((path, receipt));
}
}
Ok(selected)
}
/// Run a pair action only after every live receipt that could name the pair
/// has proven authority. The check itself performs no process termination.
pub(crate) fn with_pair_runtime_receipt_authority<R: tauri::Runtime, T>(
app: &AppHandle<R>,
key: &ManagedAgentRuntimeKey,
effect: impl FnOnce() -> Result<T, String>,
) -> Result<T, String> {
let instance_id = current_instance_id(app);
select_pair_runtime_receipt_with(
super::super::read_all_agent_runtime_receipts(app),
key,
&instance_id,
process_is_running,
process_has_buzz_marker,
)?;
effect()
}
/// 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
/// another start.
pub(crate) fn terminate_untracked_pair_runtime(
app: &AppHandle,
pub(crate) fn terminate_untracked_pair_runtime<R: tauri::Runtime>(
app: &AppHandle<R>,
key: &ManagedAgentRuntimeKey,
) -> 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)
})
let Some((path, receipt)) = select_pair_runtime_receipt_with(
super::super::read_all_agent_runtime_receipts(app),
key,
&instance_id,
process_is_running,
process_has_buzz_marker,
)?
else {
return Ok(());
};

View File

@@ -81,4 +81,21 @@ mod tests {
let key = bound_runtime_key(&record, &bound).expect("keyable record and relay");
assert_eq!(key.relay_url, "wss://tenant-a.example");
}
#[test]
fn production_spawn_key_preserves_loopback_community_authority() {
let record = record(&"cc".repeat(32), "");
let localhost =
crate::relay::bind_expected_relay_scope(None, "ws://localhost:3000".to_string())
.unwrap();
let numeric =
crate::relay::bind_expected_relay_scope(None, "ws://127.0.0.1:3000".to_string())
.unwrap();
let localhost_key = bound_runtime_key(&record, &localhost).unwrap();
let numeric_key = bound_runtime_key(&record, &numeric).unwrap();
assert_eq!(localhost_key.relay_url, "ws://localhost:3000");
assert_eq!(numeric_key.relay_url, "ws://127.0.0.1:3000");
assert_ne!(localhost_key, numeric_key);
}
}

View File

@@ -120,31 +120,33 @@ fn stop_legacy_scalar_pid<R: tauri::Runtime>(
/// 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.
pub fn stop_managed_agent_workspace_pair(
app: &AppHandle,
pub fn stop_managed_agent_workspace_pair<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &mut ManagedAgentRecord,
runtimes: &mut HashMap<ManagedAgentRuntimeKey, ManagedAgentPairRuntime>,
) -> Result<(), String> {
use tauri::Manager;
let state = app.state::<crate::app_state::AppState>();
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)?;
Some(pair_key) => super::with_pair_runtime_receipt_authority(app, &pair_key, || {
if runtimes.contains_key(&pair_key) {
stop_managed_agent_pair(app, record, runtimes, &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;
} else {
// No tracked pair here — a pubkey-wide cache clear would
// disturb live pairs in other communities, so stay scoped.
super::terminate_untracked_pair_runtime(app, &pair_key)?;
stop_legacy_scalar_pid(app, record)?;
}
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) => {
// 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)?;
state.clear_agent_session_cache(&pair_key);
}
Ok(())
})?,
None => {
stop_legacy_scalar_pid(app, record)?;
state.clear_agent_session_caches(&record.pubkey);
@@ -250,4 +252,151 @@ mod tests {
selected.sort_by(|left, right| left.relay_url.cmp(&right.relay_url));
assert_eq!(selected, vec![first, second]);
}
#[cfg(unix)]
fn local_stop_refuses_ambiguous_receipt_before_side_effects(tracked: bool) {
use tauri::Manager as _;
let _path_guard = crate::managed_agents::lock_path_mutex();
let temp = tempfile::tempdir().unwrap();
let old_home = std::env::var_os("HOME");
let old_xdg = std::env::var_os("XDG_DATA_HOME");
struct RestoreEnv(Option<std::ffi::OsString>, Option<std::ffi::OsString>);
impl Drop for RestoreEnv {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var("HOME", value),
None => std::env::remove_var("HOME"),
}
match self.1.take() {
Some(value) => std::env::set_var("XDG_DATA_HOME", value),
None => std::env::remove_var("XDG_DATA_HOME"),
}
}
}
let _restore_env = RestoreEnv(old_home, old_xdg);
std::env::set_var("HOME", temp.path());
std::env::set_var("XDG_DATA_HOME", temp.path());
let requested_relay = "wss://relay.example/room/";
let stored_relay = "wss://relay.example/room";
let pubkey = "aa".repeat(32);
let state = crate::app_state::build_app_state();
*state.relay_url_override.lock().unwrap() = Some(requested_relay.into());
let app = tauri::test::mock_builder()
.manage(state)
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.unwrap();
let instance_id = super::super::current_instance_id(app.handle());
let mut child =
Some(super::super::test_fixtures::MarkedTestChild::spawn(&instance_id).unwrap());
let pid = child.as_ref().unwrap().id();
let _process_guard = super::super::test_fixtures::MarkedProcessGuard::new(pid);
assert!(super::super::process_has_buzz_marker(pid, &instance_id));
let stored_key = ManagedAgentRuntimeKey::new(&pubkey, stored_relay).unwrap();
let requested_key = ManagedAgentRuntimeKey::new(&pubkey, requested_relay).unwrap();
let receipt = super::super::super::ManagedAgentRuntimeReceipt {
authority_version: 0,
key: stored_key.clone(),
pid,
desktop_instance_id: instance_id,
started_at: "now".into(),
};
super::super::super::write_agent_runtime_receipt(app.handle(), &receipt).unwrap();
let mut record: ManagedAgentRecord = serde_json::from_value(serde_json::json!({
"pubkey": pubkey,
"name": "test",
"private_key_nsec": "",
"relay_url": "",
"acp_command": "buzz-acp",
"agent_command": "buzz-agent",
"agent_args": [],
"mcp_command": "",
"turn_timeout_seconds": 320,
"env_vars": {},
"created_at": "2026-01-01T00:00:00Z",
"updated_at": "before"
}))
.unwrap();
let mut runtimes = HashMap::new();
if tracked {
let process = crate::managed_agents::ManagedAgentProcess {
child: child.take().unwrap().into_child(),
log_path: Default::default(),
spawn_config:
crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot(
&record,
&[],
&[],
requested_relay,
&Default::default(),
false,
crate::managed_agents::AcpSessionPolicy::Channel,
),
setup_mode: false,
adapter_availability: None,
start_nonce: "test-nonce".into(),
};
runtimes.insert(
requested_key.clone(),
ManagedAgentPairRuntime::starting(process),
);
}
let cache: crate::managed_agents::config_bridge::SessionConfigCache =
serde_json::from_value(serde_json::json!({
"configOptions": [],
"availableModes": [],
"availableModels": [],
"currentModel": null,
"modelOverridden": false,
"gooseNativeConfig": null,
"capturedAt": "now"
}))
.unwrap();
app.state::<crate::app_state::AppState>()
.put_session_cache(requested_key.clone(), cache);
let error = stop_managed_agent_workspace_pair(app.handle(), &mut record, &mut runtimes)
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
assert_eq!(record.updated_at, "before");
assert!(record.last_stopped_at.is_none());
assert!(app
.state::<crate::app_state::AppState>()
.get_session_cache(&requested_key)
.is_some());
assert!(
super::super::super::read_all_agent_runtime_receipts(app.handle())
.iter()
.any(|(_, candidate)| candidate == &receipt)
);
if tracked {
let runtime = runtimes.get_mut(&requested_key).unwrap();
assert!(runtime.child.try_wait().unwrap().is_none());
let mut runtime = runtimes.remove(&requested_key).unwrap();
let _ = runtime.child.kill();
let _ = runtime.child.wait();
} else {
let child = child.as_mut().unwrap();
assert!(child.child_mut().try_wait().unwrap().is_none());
}
super::super::super::remove_agent_runtime_receipt(app.handle(), &stored_key);
}
#[cfg(unix)]
#[test]
fn tracked_local_stop_has_no_side_effect_before_ambiguous_receipt_refusal() {
local_stop_refuses_ambiguous_receipt_before_side_effects(true);
}
#[cfg(unix)]
#[test]
fn untracked_local_stop_has_no_side_effect_before_ambiguous_receipt_refusal() {
local_stop_refuses_ambiguous_receipt_before_side_effects(false);
}
}

View File

@@ -1,5 +1,128 @@
use crate::managed_agents::types::{ManagedAgentRecord, RespondTo};
#[cfg(unix)]
const MARKED_CHILD_FIXTURE_ENV: &str = "BUZZ_TEST_MARKED_CHILD_FIXTURE";
#[cfg(unix)]
const MARKED_CHILD_READY_ENV: &str = "BUZZ_TEST_MARKED_CHILD_READY";
/// Test-executable child whose environment is stable and directly observable
/// through the production process-marker reader.
#[cfg(unix)]
pub(in crate::managed_agents) struct MarkedTestChild {
child: Option<std::process::Child>,
_ready_dir: tempfile::TempDir,
}
#[cfg(unix)]
impl MarkedTestChild {
pub(in crate::managed_agents) fn spawn(instance_id: &str) -> Result<Self, String> {
use std::os::unix::process::CommandExt as _;
use std::process::{Command, Stdio};
let ready_dir = tempfile::tempdir().map_err(|error| error.to_string())?;
let ready_path = ready_dir.path().join("ready");
let executable = std::env::current_exe().map_err(|error| error.to_string())?;
let mut child = Command::new(executable)
.args([
"--exact",
"managed_agents::runtime::test_fixtures::marked_child_process_fixture",
"--nocapture",
])
.env_clear()
.env(MARKED_CHILD_FIXTURE_ENV, "1")
.env(MARKED_CHILD_READY_ENV, &ready_path)
.env("BUZZ_MANAGED_AGENT", instance_id)
.process_group(0)
.stdin(Stdio::null())
.stdout(Stdio::null())
.stderr(Stdio::null())
.spawn()
.map_err(|error| error.to_string())?;
for _ in 0..100 {
if ready_path.is_file() {
return Ok(Self {
child: Some(child),
_ready_dir: ready_dir,
});
}
match child.try_wait() {
Ok(Some(status)) => {
return Err(format!(
"marked child fixture exited before readiness: {status}"
));
}
Ok(None) => {}
Err(error) => {
let _ = super::terminate_process(child.id());
let _ = child.wait();
return Err(format!("failed to inspect marked child fixture: {error}"));
}
}
std::thread::sleep(std::time::Duration::from_millis(20));
}
let _ = super::terminate_process(child.id());
let _ = child.wait();
Err("marked child fixture did not become ready".into())
}
pub(in crate::managed_agents) fn id(&self) -> u32 {
self.child.as_ref().expect("child is present").id()
}
pub(in crate::managed_agents) fn child_mut(&mut self) -> &mut std::process::Child {
self.child.as_mut().expect("child is present")
}
pub(in crate::managed_agents) fn into_child(mut self) -> std::process::Child {
self.child.take().expect("child is present")
}
}
#[cfg(unix)]
impl Drop for MarkedTestChild {
fn drop(&mut self) {
if let Some(child) = self.child.as_mut() {
let _ = super::terminate_process(child.id());
let _ = child.wait();
}
}
}
/// Backstop for children whose owned `Child` handle is moved into production
/// runtime state. A failed assertion still terminates the complete process
/// group; successful tests explicitly wait through the owned handle.
#[cfg(unix)]
pub(in crate::managed_agents) struct MarkedProcessGuard(u32);
#[cfg(unix)]
impl MarkedProcessGuard {
pub(in crate::managed_agents) fn new(pid: u32) -> Self {
Self(pid)
}
}
#[cfg(unix)]
impl Drop for MarkedProcessGuard {
fn drop(&mut self) {
let _ = super::terminate_process(self.0);
}
}
#[cfg(unix)]
#[test]
fn marked_child_process_fixture() {
if std::env::var_os(MARKED_CHILD_FIXTURE_ENV).is_none() {
return;
}
let ready_path = std::env::var_os(MARKED_CHILD_READY_ENV)
.expect("marked child fixture requires a readiness path");
std::fs::write(ready_path, b"ready").expect("write marked child readiness handshake");
loop {
std::thread::park_timeout(std::time::Duration::from_secs(60));
}
}
pub(super) const EXPECTED_ACCESS_ENV: &str = "BUZZ_TEST_EXPECTED_AGENT_ACCESS_OWNER_ONLY";
pub(super) fn expected_owner_only() -> bool {
@@ -30,7 +153,7 @@ pub(super) fn expected_mode(oss_mode: &'static str) -> &'static str {
}
/// Construct a minimal record fixture for runtime tests.
pub(super) fn fixture(
pub(in crate::managed_agents) fn fixture(
respond_to: RespondTo,
allowlist: Vec<String>,
auth_tag: Option<String>,

View File

@@ -1,5 +1,8 @@
use crate::managed_agents::known_acp_runtime;
#[path = "authority_tests.rs"]
mod authority_tests;
#[path = "cli_tests.rs"]
mod cli_tests;
@@ -860,6 +863,7 @@ fn receipt_fixture(
key: crate::managed_agents::ManagedAgentRuntimeKey,
) -> crate::managed_agents::ManagedAgentRuntimeReceipt {
crate::managed_agents::ManagedAgentRuntimeReceipt {
authority_version: crate::managed_agents::RUNTIME_AUTHORITY_RECEIPT_VERSION,
key,
pid: std::process::id(),
desktop_instance_id: "test-instance".into(),
@@ -895,64 +899,6 @@ fn receipt_validation_rejects_wrong_pair_filename() {
));
}
#[test]
fn replacement_removes_receipt_only_after_confirmed_exit() {
use std::cell::{Cell, RefCell};
let receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap(),
);
let path = std::path::Path::new("pair.json");
let terminated = Cell::new(None);
let polls = Cell::new(0);
let removed = RefCell::new(None);
super::terminate_runtime_receipt_with(
path,
&receipt,
|pid| {
terminated.set(Some(pid));
Ok(())
},
|_| {
let poll = polls.get() + 1;
polls.set(poll);
poll < 2
},
|path| *removed.borrow_mut() = Some(path.to_path_buf()),
)
.unwrap();
assert_eq!(terminated.get(), Some(receipt.pid));
assert_eq!(polls.get(), 2);
assert_eq!(removed.into_inner().as_deref(), Some(path));
}
#[test]
fn replacement_failure_keeps_receipt() {
use std::cell::Cell;
let receipt = receipt_fixture(
crate::managed_agents::ManagedAgentRuntimeKey::new("aa".repeat(32), "wss://relay.example")
.unwrap(),
);
let removed = Cell::new(false);
let error = super::terminate_runtime_receipt_with(
std::path::Path::new("pair.json"),
&receipt,
|_| Err("signal failed".into()),
|_| false,
|_| removed.set(true),
)
.unwrap_err();
assert_eq!(error, "signal failed");
assert!(!removed.get());
}
// ── workspace pair-key resolution (summary/stop scoping) ────────────────
#[test]
fn unpinned_record_resolves_pair_key_per_workspace() {
// Community-scoped truth: an unpinned agent running only on relay A must
@@ -967,6 +913,17 @@ fn unpinned_record_resolves_pair_key_per_workspace() {
assert!(!runtimes.contains_key(&key_b));
}
#[test]
fn workspace_pair_resolution_distinguishes_loopback_communities() {
let pubkey = "aa".repeat(32);
let localhost = super::resolve_workspace_pair_key(&pubkey, "", "ws://localhost:3000").unwrap();
let numeric = super::resolve_workspace_pair_key(&pubkey, "", "ws://127.0.0.1:3000").unwrap();
let runtimes = std::collections::HashMap::from([(localhost.clone(), ())]);
assert!(runtimes.contains_key(&localhost));
assert!(!runtimes.contains_key(&numeric));
}
#[test]
fn stored_relay_pin_is_ignored_in_pair_key_resolution() {
// Legacy pins are ignored (#2122): a record carrying a creation-era

View File

@@ -14,8 +14,8 @@ use crate::app_state::AppState;
const STATUS_EVENT: &str = "managed-agent-runtime-status";
fn status_for(
app: &AppHandle,
fn status_for<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &super::ManagedAgentRecord,
key: &ManagedAgentRuntimeKey,
runtime: Option<&ManagedAgentPairRuntime>,
@@ -43,8 +43,8 @@ struct StatusInputs<'a> {
global: &'a super::GlobalAgentConfig,
}
fn status_for_with(
app: &AppHandle,
fn status_for_with<R: tauri::Runtime>(
app: &AppHandle<R>,
record: &super::ManagedAgentRecord,
key: &ManagedAgentRuntimeKey,
runtime: Option<&ManagedAgentPairRuntime>,
@@ -72,7 +72,7 @@ fn status_for_with(
}
}
fn emit_status(app: &AppHandle, status: &ManagedAgentRuntimeStatus) {
fn emit_status<R: tauri::Runtime>(app: &AppHandle<R>, status: &ManagedAgentRuntimeStatus) {
let _ = app.emit(STATUS_EVENT, status);
}
@@ -339,12 +339,12 @@ pub(crate) fn start_pair_locked(
broker,
)?;
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(),
};
let receipt = ManagedAgentRuntimeReceipt::new(
key.clone(),
process.child.id(),
current_instance_id(&app),
now.clone(),
);
if let Err(error) = write_agent_runtime_receipt(&app, &receipt) {
let _ = terminate_process(process.child.id());
let _ = process.child.wait();
@@ -379,10 +379,10 @@ pub fn stop_managed_agent_runtime(
}
// Caller owns managed_agent_runtime_transition for the whole admission/effect.
pub(crate) fn stop_pair_locked(
pub(crate) fn stop_pair_locked<R: tauri::Runtime>(
pubkey: String,
relay_url: String,
app: AppHandle,
app: AppHandle<R>,
) -> Result<ManagedAgentRuntimeStatus, String> {
let state = app.state::<AppState>();
let _store = state
@@ -396,30 +396,36 @@ pub(crate) fn stop_pair_locked(
.managed_agent_processes
.lock()
.map_err(|e| e.to_string())?;
if runtimes.contains_key(&key) {
// Use ordinary Desktop Stop, including its platform-specific child/job
// ownership. Remote control must not grow a second teardown contract.
super::stop_managed_agent_pair(&app, record, &mut runtimes, &key)?;
} else {
terminate_untracked_pair_runtime(&app, &key)?;
}
// Old scalar records have no community-bound receipt. Do not erase a live
// child or claim success for it when this request cannot establish scope.
reject_unscoped_live_child(
record.runtime_pid.filter(|pid| process_is_running(*pid)),
runtimes.values().map(|runtime| runtime.child.id()),
)?;
super::remove_agent_runtime_receipt(&app, &key);
state.clear_agent_session_cache(&key);
if record
.runtime_pid
.is_some_and(|pid| !process_is_running(pid))
{
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);
// V0 receipt normalization was lossy. Wrap every tracked/untracked Stop
// side effect so an ambiguous receipt cannot be killed, deleted, cleared
// from cache, persisted as stopped, or reported stopped.
let status = super::with_pair_runtime_receipt_authority(&app, &key, || {
if runtimes.contains_key(&key) {
// Use ordinary Desktop Stop, including its platform-specific
// child/job ownership. Remote control must not grow a second
// teardown contract.
super::stop_managed_agent_pair(&app, record, &mut runtimes, &key)?;
} else {
terminate_untracked_pair_runtime(&app, &key)?;
}
// Old scalar records have no community-bound receipt. Do not erase a
// live child or claim success when this request cannot establish scope.
reject_unscoped_live_child(
record.runtime_pid.filter(|pid| process_is_running(*pid)),
runtimes.values().map(|runtime| runtime.child.id()),
)?;
super::remove_agent_runtime_receipt(&app, &key);
state.clear_agent_session_cache(&key);
if record
.runtime_pid
.is_some_and(|pid| !process_is_running(pid))
{
record.runtime_pid = None;
}
record.updated_at = crate::util::now_iso();
record.last_stopped_at = Some(record.updated_at.clone());
Ok(status_for(&app, record, &key, None, None))
})?;
drop(runtimes);
save_managed_agents(&app, &records)?;
emit_status(&app, &status);
@@ -753,6 +759,24 @@ mod tests {
assert_ne!(key, observer_lifecycle_key(&other.pubkey, &other).unwrap());
}
#[test]
fn observer_lifecycle_key_does_not_cross_loopback_communities() {
let localhost = payload(
"ws://localhost:3000",
ManagedAgentRuntimeLifecycle::Ready,
None,
);
let numeric = payload(
"ws://127.0.0.1:3000",
ManagedAgentRuntimeLifecycle::Ready,
None,
);
assert_ne!(
observer_lifecycle_key(&localhost.pubkey, &localhost).unwrap(),
observer_lifecycle_key(&numeric.pubkey, &numeric).unwrap()
);
}
#[test]
fn observer_lifecycle_rejects_cross_agent_and_desktop_states() {
let ready = payload(
@@ -792,6 +816,147 @@ mod tests {
mod stop_scope_tests {
use super::reject_unscoped_live_child;
#[cfg(unix)]
fn assert_remote_stop_effect_is_blocked(tracked: bool) {
use tauri::Manager as _;
let _path_guard = crate::managed_agents::lock_path_mutex();
let temp = tempfile::tempdir().unwrap();
let old_home = std::env::var_os("HOME");
let old_xdg = std::env::var_os("XDG_DATA_HOME");
struct RestoreEnv(Option<std::ffi::OsString>, Option<std::ffi::OsString>);
impl Drop for RestoreEnv {
fn drop(&mut self) {
match self.0.take() {
Some(value) => std::env::set_var("HOME", value),
None => std::env::remove_var("HOME"),
}
match self.1.take() {
Some(value) => std::env::set_var("XDG_DATA_HOME", value),
None => std::env::remove_var("XDG_DATA_HOME"),
}
}
}
let _restore_env = RestoreEnv(old_home, old_xdg);
std::env::set_var("HOME", temp.path());
std::env::set_var("XDG_DATA_HOME", temp.path());
let requested_relay = "wss://relay.example/room/";
let stored_relay = "wss://relay.example/room";
let pubkey = "aa".repeat(32);
let app = tauri::test::mock_builder()
.manage(crate::app_state::build_app_state())
.build(tauri::test::mock_context(tauri::test::noop_assets()))
.unwrap();
let instance_id = super::super::current_instance_id(app.handle());
let mut child = Some(
super::super::runtime::test_fixtures::MarkedTestChild::spawn(&instance_id).unwrap(),
);
let pid = child.as_ref().unwrap().id();
let _process_guard = super::super::runtime::test_fixtures::MarkedProcessGuard::new(pid);
assert!(super::super::process_has_buzz_marker(pid, &instance_id));
let mut record = super::super::runtime::test_fixtures::fixture(
super::super::RespondTo::OwnerOnly,
Vec::new(),
None,
);
record.pubkey = pubkey.clone();
record.updated_at = "before".into();
super::super::save_managed_agents(app.handle(), &[record.clone()]).unwrap();
let stored_key = super::ManagedAgentRuntimeKey::new(&pubkey, stored_relay).unwrap();
let requested_key = super::ManagedAgentRuntimeKey::new(&pubkey, requested_relay).unwrap();
let receipt = super::ManagedAgentRuntimeReceipt {
authority_version: 0,
key: stored_key.clone(),
pid,
desktop_instance_id: instance_id,
started_at: "now".into(),
};
super::super::write_agent_runtime_receipt(app.handle(), &receipt).unwrap();
if tracked {
let process = crate::managed_agents::ManagedAgentProcess {
child: child.take().unwrap().into_child(),
log_path: Default::default(),
spawn_config:
crate::managed_agents::spawn_snapshot::prospective_spawn_config_snapshot(
&record,
&[],
&[],
requested_relay,
&Default::default(),
false,
crate::managed_agents::AcpSessionPolicy::Channel,
),
setup_mode: false,
adapter_availability: None,
start_nonce: "test-nonce".into(),
};
app.state::<crate::app_state::AppState>()
.managed_agent_processes
.lock()
.unwrap()
.insert(
requested_key.clone(),
super::ManagedAgentPairRuntime::starting(process),
);
}
let cache: crate::managed_agents::config_bridge::SessionConfigCache =
serde_json::from_value(serde_json::json!({
"configOptions": [],
"availableModes": [],
"availableModels": [],
"currentModel": null,
"modelOverridden": false,
"gooseNativeConfig": null,
"capturedAt": "now"
}))
.unwrap();
app.state::<crate::app_state::AppState>()
.put_session_cache(requested_key.clone(), cache);
let error =
super::stop_pair_locked(pubkey.clone(), requested_relay.into(), app.handle().clone())
.unwrap_err();
assert!(error.contains("cannot prove the requested community authority"));
assert_eq!(
super::super::load_managed_agents(app.handle()).unwrap()[0].updated_at,
"before"
);
assert!(app
.state::<crate::app_state::AppState>()
.get_session_cache(&requested_key)
.is_some());
assert!(super::super::read_all_agent_runtime_receipts(app.handle())
.iter()
.any(|(_, candidate)| candidate == &receipt));
if tracked {
let mut runtime = app
.state::<crate::app_state::AppState>()
.managed_agent_processes
.lock()
.unwrap()
.remove(&requested_key)
.unwrap();
assert!(runtime.child.try_wait().unwrap().is_none());
let _ = runtime.child.kill();
let _ = runtime.child.wait();
} else {
assert!(child
.as_mut()
.unwrap()
.child_mut()
.try_wait()
.unwrap()
.is_none());
}
super::super::remove_agent_runtime_receipt(app.handle(), &stored_key);
}
#[test]
fn live_legacy_child_cannot_be_erased_or_reported_stopped() {
assert!(reject_unscoped_live_child(Some(12), [].into_iter()).is_err());
@@ -799,4 +964,16 @@ mod stop_scope_tests {
assert!(reject_unscoped_live_child(Some(12), [12].into_iter()).is_ok());
assert!(reject_unscoped_live_child(None, [13].into_iter()).is_ok());
}
#[cfg(unix)]
#[test]
fn tracked_remote_stop_has_no_effect_before_ambiguous_receipt_refusal() {
assert_remote_stop_effect_is_blocked(true);
}
#[cfg(unix)]
#[test]
fn untracked_remote_stop_has_no_effect_before_ambiguous_receipt_refusal() {
assert_remote_stop_effect_is_blocked(false);
}
}

View File

@@ -1,8 +1,67 @@
use serde::{Deserialize, Serialize};
use sha2::{Digest as _, Sha256};
use url::{Host, Url};
use super::ManagedAgentProcess;
pub(crate) const RUNTIME_AUTHORITY_RECEIPT_VERSION: u8 = 1;
/// Canonicalize only URL syntax that cannot distinguish relay authorities.
///
/// In particular, loopback host spellings stay literal: relay tenancy and
/// lifecycle fences distinguish `localhost`, `127.*`, and `::1`. The shared
/// buzz-core normalizer predates that boundary and deliberately remains in use
/// by Bestie and migration consumers whose compatibility rules differ.
fn normalize_runtime_relay_url(raw: &str) -> Result<String, String> {
let mut url = Url::parse(raw.trim()).map_err(|error| format!("invalid relay URL: {error}"))?;
if !matches!(url.scheme(), "ws" | "wss") {
return Err("relay URL scheme must be ws or wss".into());
}
if !url.username().is_empty() || url.password().is_some() {
return Err("relay URL must not contain credentials".into());
}
if url.fragment().is_some() {
return Err("relay URL must not contain a fragment".into());
}
let host = url
.host()
.ok_or_else(|| "relay URL must contain a host".to_string())?;
if let Host::Domain(domain) = host {
let lowercase = domain.to_ascii_lowercase();
url.set_host(Some(&lowercase))
.map_err(|_| "relay URL must contain a host".to_string())?;
}
let default_port = match url.scheme() {
"ws" => Some(80),
"wss" => Some(443),
_ => None,
};
if url.port() == default_port {
url.set_port(None)
.map_err(|_| "relay URL scheme must be ws or wss".to_string())?;
}
let host = match url
.host()
.ok_or_else(|| "relay URL must contain a host".to_string())?
{
Host::Domain(domain) => domain.to_string(),
Host::Ipv4(address) => address.to_string(),
Host::Ipv6(address) => format!("[{address}]"),
};
let port = url
.port()
.map(|port| format!(":{port}"))
.unwrap_or_default();
let path = if url.path() == "/" { "" } else { url.path() };
let query = url
.query()
.map(|query| format!("?{query}"))
.unwrap_or_default();
Ok(format!("{}://{host}{port}{path}{query}", url.scheme()))
}
/// Canonical identity of one managed-agent harness on one relay.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Hash)]
#[serde(rename_all = "camelCase")]
@@ -19,8 +78,7 @@ impl ManagedAgentRuntimeKey {
}
Ok(Self {
pubkey: pubkey.to_ascii_lowercase(),
relay_url: buzz_core_pkg::relay::normalize_relay_url(relay_url)
.map_err(|error| error.to_string())?,
relay_url: normalize_runtime_relay_url(relay_url)?,
})
}
@@ -113,8 +171,106 @@ pub struct ManagedAgentCommunityTarget {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct ManagedAgentRuntimeReceipt {
/// Version 0 is an unversioned legacy receipt. Its lossy host/path rendering
/// cannot prove pair authority; it is usable only for instance-wide cleanup.
#[serde(default)]
pub authority_version: u8,
pub key: ManagedAgentRuntimeKey,
pub pid: u32,
pub desktop_instance_id: String,
pub started_at: String,
}
impl ManagedAgentRuntimeReceipt {
pub(crate) fn new(
key: ManagedAgentRuntimeKey,
pid: u32,
desktop_instance_id: String,
started_at: String,
) -> Self {
Self {
authority_version: RUNTIME_AUTHORITY_RECEIPT_VERSION,
key,
pid,
desktop_instance_id,
started_at,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn key(relay_url: &str) -> ManagedAgentRuntimeKey {
ManagedAgentRuntimeKey::new("aa".repeat(32), relay_url).unwrap()
}
#[test]
fn runtime_identity_preserves_distinct_loopback_authorities() {
let localhost = key("ws://localhost:3000");
let ipv4 = key("ws://127.0.0.1:3000");
let other_ipv4 = key("ws://127.0.0.2:3000");
let ipv6 = key("ws://[::1]:3000");
assert_eq!(localhost.relay_url, "ws://localhost:3000");
assert_eq!(ipv4.relay_url, "ws://127.0.0.1:3000");
assert_eq!(other_ipv4.relay_url, "ws://127.0.0.2:3000");
assert_eq!(ipv6.relay_url, "ws://[::1]:3000");
assert_ne!(localhost, ipv4);
assert_ne!(ipv4, other_ipv4);
assert_ne!(ipv4, ipv6);
}
#[test]
fn runtime_identity_preserves_paths_queries_and_meaningful_trailing_slashes() {
assert_eq!(
key(" WSS://Relay.Example:443/community/?mode=one ").relay_url,
"wss://relay.example/community/?mode=one"
);
assert_ne!(
key("wss://relay.example/community").relay_url,
key("wss://relay.example/community/").relay_url
);
assert_eq!(
key("wss://relay.example/?").relay_url,
"wss://relay.example?"
);
assert_ne!(
key("wss://relay.example").relay_url,
key("wss://relay.example/?").relay_url
);
}
#[test]
fn runtime_identity_rejects_non_websocket_credentials_and_fragments() {
for relay_url in [
"https://relay.example",
"wss://user@relay.example",
"wss://relay.example/#",
"wss://relay.example/#fragment",
] {
assert!(ManagedAgentRuntimeKey::new("aa".repeat(32), relay_url).is_err());
}
}
#[test]
fn receipt_authority_version_distinguishes_new_and_unversioned_receipts() {
let receipt = ManagedAgentRuntimeReceipt::new(
key("ws://localhost:3000"),
42,
"instance".into(),
"now".into(),
);
assert_eq!(receipt.authority_version, RUNTIME_AUTHORITY_RECEIPT_VERSION);
let legacy: ManagedAgentRuntimeReceipt = serde_json::from_value(serde_json::json!({
"key": receipt.key,
"pid": 42,
"desktopInstanceId": "instance",
"startedAt": "now"
}))
.unwrap();
assert_eq!(legacy.authority_version, 0);
}
}

View File

@@ -79,8 +79,8 @@ pub(crate) fn apply_acp_session_policy_env(
/// Resolve the effective policy, apply it to `command`, and return it so the
/// caller can stamp the same value onto the spawn snapshot (env and badge can
/// never disagree about what the child launched with).
pub(crate) fn apply_app_acp_session_policy_env(
app: &AppHandle,
pub(crate) fn apply_app_acp_session_policy_env<R: tauri::Runtime>(
app: &AppHandle<R>,
command: &mut std::process::Command,
) -> AcpSessionPolicy {
let policy = acp_session_policy(app.state::<AppState>().inner());

View File

@@ -48,7 +48,7 @@ pub(crate) fn managed_agents_store_path<R: tauri::Runtime>(
Ok(managed_agents_base_dir(app)?.join("managed-agents.json"))
}
fn managed_agents_logs_dir(app: &AppHandle) -> Result<PathBuf, String> {
fn managed_agents_logs_dir<R: tauri::Runtime>(app: &AppHandle<R>) -> Result<PathBuf, String> {
let dir = managed_agents_base_dir(app)?.join("logs");
fs::create_dir_all(&dir).map_err(|error| format!("failed to create logs dir: {error}"))?;
Ok(dir)
@@ -88,8 +88,8 @@ pub fn managed_agent_log_path(app: &AppHandle, pubkey: &str) -> Result<PathBuf,
/// Pair-scoped log path for a managed runtime. The relay URL never appears in
/// the filename; the suffix is a hash of the canonical URL.
pub fn managed_agent_runtime_log_path(
app: &AppHandle,
pub fn managed_agent_runtime_log_path<R: tauri::Runtime>(
app: &AppHandle<R>,
key: &ManagedAgentRuntimeKey,
) -> Result<PathBuf, String> {
Ok(managed_agents_logs_dir(app)?.join(format!("{}.log", key.runtime_id())))
@@ -813,8 +813,8 @@ fn agent_pids_dir<R: tauri::Runtime>(app: &AppHandle<R>) -> Result<PathBuf, Stri
/// Persist a pair-scoped runtime receipt atomically. Callers must register the
/// process in memory in the same runtime transition; on write failure they must
/// terminate the child before releasing that transition.
pub fn write_agent_runtime_receipt(
app: &AppHandle,
pub fn write_agent_runtime_receipt<R: tauri::Runtime>(
app: &AppHandle<R>,
receipt: &ManagedAgentRuntimeReceipt,
) -> Result<(), String> {
let path = agent_pids_dir(app)?.join(format!("{}.json", receipt.key.runtime_id()));
@@ -836,8 +836,8 @@ pub fn remove_agent_runtime_receipt_path(path: &Path) {
let _ = fs::remove_file(path);
}
pub fn read_all_agent_runtime_receipts(
app: &AppHandle,
pub fn read_all_agent_runtime_receipts<R: tauri::Runtime>(
app: &AppHandle<R>,
) -> Vec<(PathBuf, ManagedAgentRuntimeReceipt)> {
let Ok(dir) = agent_pids_dir(app) else {
return Vec::new();

View File

@@ -132,7 +132,7 @@ mod tests {
let runtime =
crate::managed_agents::ManagedAgentRuntimeKey::new("a".repeat(64), captured.as_str())
.unwrap();
assert_eq!(runtime.relay_url, "ws://127.0.0.1:3037");
assert_eq!(runtime.relay_url, relay);
assert_eq!(captured.revalidate(relay.into()).unwrap().as_str(), relay);
}

View File

@@ -21,7 +21,7 @@ test("canonicalCommunityRelays dedupes by canonical form, keeps stored spelling"
const relays = canonicalCommunityRelays(
[
{ relayUrl: "ws://localhost:3000" },
// Same relay, different spelling — folds onto the first entry.
// A distinct loopback authority is a distinct community.
{ relayUrl: "ws://127.0.0.1:3000" },
{ relayUrl: "wss://relay.example" },
// Unparsable entries are dropped rather than reconciled.
@@ -32,7 +32,8 @@ test("canonicalCommunityRelays dedupes by canonical form, keeps stored spelling"
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"],
],
);
@@ -40,7 +41,8 @@ test("canonicalCommunityRelays dedupes by canonical form, keeps stored spelling"
test("pendingReconcileRelays skips reconciled and in-flight relays", () => {
const canonicalToRequested = new Map([
["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://a.example", "wss://a.example"],
["wss://b.example", "wss://b.example"],
]);
@@ -49,7 +51,7 @@ test("pendingReconcileRelays skips reconciled and in-flight relays", () => {
new Set(["wss://a.example"]),
new Set(["ws://127.0.0.1:3000"]),
);
assert.deepEqual(pending, ["wss://b.example"]);
assert.deepEqual(pending, ["ws://localhost:3000", "wss://b.example"]);
});
test("classifyReconcileResult marks the whole batch failed when the call throws", () => {
@@ -64,7 +66,7 @@ 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"];
const attempted = ["ws://localhost:3000", "wss://b.example"];
const rows = [
// Started cleanly on the loopback relay — reconciled.
{
@@ -92,7 +94,7 @@ test("classifyReconcileResult splits by Failed rows, matching on requested URL",
assert.deepEqual(
classifyReconcileResult(attempted, rows, canonicalRelayUrl),
{
succeeded: ["ws://127.0.0.1:3000"],
succeeded: ["ws://localhost:3000"],
failed: ["wss://b.example"],
},
);

View File

@@ -47,3 +47,16 @@ test("startup reconcile preserves unrelated runtime rows", () => {
[discovered, existing],
);
});
test("startup reconcile keeps same-agent loopback authority rows distinct", () => {
const localhost = runtime({ relayUrl: "ws://localhost:3000" });
const numeric = runtime({
relayUrl: "ws://127.0.0.1:3000",
lifecycle: "ready",
});
assert.deepEqual(
mergeManagedAgentRuntimeStatuses([localhost], [localhost], [numeric]),
[numeric, localhost],
);
});

View File

@@ -4,6 +4,7 @@ import test from "node:test";
import {
agentCommunityAvailability,
agentCommunityStatusDetail,
canonicalBestieRelayUrl,
canonicalRelayUrl,
findManagedAgentRuntime,
managedAgentRuntimeKey,
@@ -78,9 +79,8 @@ test("selects one relay without collapsing same-pubkey pairs", () => {
});
test("canonicalRelayUrl mirrors the backend pair-key normalization", () => {
// Loopback folding + default-port and trailing-slash stripping — the
// standard dev setup that previously broke pair matching.
assert.equal(canonicalRelayUrl("ws://localhost:3000"), "ws://127.0.0.1:3000");
assert.equal(canonicalRelayUrl("ws://localhost:3000"), "ws://localhost:3000");
assert.equal(canonicalRelayUrl("ws://127.0.0.1:3000"), "ws://127.0.0.1:3000");
assert.equal(
canonicalRelayUrl("WSS://Relay.Example:443/"),
"wss://relay.example",
@@ -91,23 +91,57 @@ test("canonicalRelayUrl mirrors the backend pair-key normalization", () => {
);
assert.equal(
canonicalRelayUrl("wss://relay.example/path/"),
"wss://relay.example/path",
"wss://relay.example/path/",
);
assert.equal(canonicalRelayUrl("ws://[::1]:3000"), "ws://[::1]:3000");
assert.equal(
canonicalRelayUrl("wss://relay.example/community/?mode=one"),
"wss://relay.example/community/?mode=one",
);
assert.equal(
canonicalRelayUrl("wss://relay.example/?"),
"wss://relay.example?",
);
assert.notEqual(
canonicalRelayUrl("wss://relay.example/?"),
canonicalRelayUrl("wss://relay.example"),
);
assert.equal(canonicalRelayUrl("ws://[::1]:3000"), "ws://127.0.0.1:3000");
assert.equal(canonicalRelayUrl("https://relay.example"), null);
assert.equal(canonicalRelayUrl("wss://user@relay.example"), null);
assert.equal(canonicalRelayUrl("wss://relay.example/#"), null);
assert.equal(canonicalRelayUrl("wss://relay.example/#fragment"), null);
assert.equal(canonicalRelayUrl("not a url"), null);
});
test("matches a stored community URL against canonical backend rows", () => {
test("Bestie retains its Rust legacy equivalence without widening runtime identity", () => {
assert.equal(
canonicalBestieRelayUrl("ws://localhost:3000"),
"ws://127.0.0.1:3000",
);
assert.equal(
canonicalBestieRelayUrl("wss://relay.example/path/"),
"wss://relay.example/path",
);
assert.equal(
canonicalBestieRelayUrl("wss://relay.example/?mode=one"),
"wss://relay.example/?mode=one",
);
assert.equal(
canonicalBestieRelayUrl("wss://relay.example/?"),
"wss://relay.example/?",
);
});
test("runtime lookup never crosses loopback community authorities", () => {
const runtimes = [
runtime({ relayUrl: "ws://127.0.0.1:3000", lifecycle: "ready" }),
];
assert.equal(
findManagedAgentRuntime(runtimes, "aa", "ws://localhost:3000")?.lifecycle,
"ready",
);
assert.equal(
findManagedAgentRuntime(runtimes, "aa", "ws://localhost:3001"),
findManagedAgentRuntime(runtimes, "aa", "ws://localhost:3000"),
undefined,
);
assert.equal(
findManagedAgentRuntime(runtimes, "aa", "ws://127.0.0.1:3000"),
runtimes[0],
);
});

View File

@@ -63,30 +63,54 @@ export const MANAGED_AGENT_PAIR_ACTION_LABELS: Record<
};
/**
* 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.
* Canonicalize a relay URL the way the backend keys runtime pairs. Host
* spellings remain distinct because the relay authority is the community:
* `localhost`, `127.*`, and `::1` must never select one another's process.
* DNS case/default ports and a root slash are syntax-only; non-root paths,
* queries, and meaningful trailing slashes are preserved.
*/
export function canonicalRelayUrl(raw: string): string | null {
const input = raw.trim();
let url: URL;
try {
url = new URL(raw.trim());
url = new URL(input);
} catch {
return null;
}
if (url.protocol !== "ws:" && url.protocol !== "wss:") return null;
if (url.username !== "" || url.password !== "" || input.includes("#"))
return null;
const host = url.hostname.toLowerCase();
const defaultPort = url.protocol === "ws:" ? "80" : "443";
const port = url.port && url.port !== defaultPort ? `:${url.port}` : "";
const path = url.pathname === "/" ? "" : url.pathname;
const query = url.search || (url.href.endsWith("?") ? "?" : "");
return `${url.protocol}//${host}${port}${path}${query}`;
}
/**
* Bestie's Rust scope check intentionally retains buzz-core's legacy
* loopback-folding equivalence. Keep its React Query cache key aligned without
* reusing that broader equivalence for managed-runtime identity.
*/
export function canonicalBestieRelayUrl(raw: string): string | null {
const input = raw.trim();
let url: URL;
try {
url = new URL(input);
} catch {
return null;
}
if (url.protocol !== "ws:" && url.protocol !== "wss:") return null;
if (url.username !== "" || url.password !== "" || input.includes("#"))
return null;
let host = url.hostname.toLowerCase();
if (host === "localhost" || host === "[::1]" || host.startsWith("127.")) {
host = "127.0.0.1";
}
const defaultPort = url.protocol === "ws:" ? "80" : "443";
const port = url.port && url.port !== defaultPort ? `:${url.port}` : "";
const path = url.pathname === "/" ? "" : url.pathname;
// The backend trims trailing slashes from the final rendered URL.
return `${url.protocol}//${host}${port}${path}${url.search}`.replace(
const port = url.port ? `:${url.port}` : "";
const query = url.search || (url.href.endsWith("?") ? "?" : "");
return `${url.protocol}//${host}${port}${url.pathname}${query}`.replace(
/\/+$/,
"",
);
@@ -98,10 +122,8 @@ export function findManagedAgentRuntime(
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.
// Backend rows carry the canonical pair URL; compare syntax-equivalent
// spellings while preserving distinct host authorities.
const canonical = canonicalRelayUrl(relayUrl);
return runtimes.find(
(runtime) =>

View File

@@ -8,7 +8,7 @@ import {
useManagedAgentRuntimesQuery,
} from "@/features/agents/managedAgentRuntimeHooks";
import {
canonicalRelayUrl,
canonicalBestieRelayUrl,
findManagedAgentRuntime,
managedAgentPairAction,
} from "@/features/agents/managedAgentRuntimeStatus";
@@ -36,7 +36,7 @@ export function bestieAssignmentQueryKey(
) {
return [
"bestie-assignment",
canonicalRelayUrl(relayUrl) ?? relayUrl,
canonicalBestieRelayUrl(relayUrl) ?? relayUrl,
ownerPubkey.toLowerCase(),
] as const;
}