diff --git a/crates/buzz-provider-deploy/src/lib.rs b/crates/buzz-provider-deploy/src/lib.rs index 0bc669e05db..56b25b7cdf0 100644 --- a/crates/buzz-provider-deploy/src/lib.rs +++ b/crates/buzz-provider-deploy/src/lib.rs @@ -16,8 +16,37 @@ //! desktop resolves `~/.buzz`; a headless daemon may have none) and, for a //! caller acting on a signed launch bundle, pinning the expected binary //! digest via [`provider_deploy_pinned`] — see that function's doc for why. +//! +//! # The child's environment +//! +//! Every entry point takes an `env: Option<&HashMap>`. `None` +//! (every caller before `buzz-waker`'s dynamic multi-tenant enrolment) +//! leaves the child's environment exactly as `std::process::Command` +//! defaults it — the parent process's own environment, untouched — so this +//! parameter changed no existing caller's behavior when it was added. +//! +//! **`Some(map)` is isolation, not an overlay.** The child's environment is +//! cleared ([`std::process::Command::env_clear`]), then +//! [`tenant_child_environment_baseline`] is applied (today: `HOME` — required +//! by `buzz-backend-sprites::credentials::resolve` before it even checks +//! `SPRITE_TOKEN`, for its keychain fallback and `~/.sprites` metadata dir — +//! and `PATH`, which that same binary's own provisioning code needs to +//! resolve `bash` by relative name), then `map`'s keys are layered on top, +//! overriding any baseline variable with the same name. This stops the +//! child from also seeing whatever else the daemon's own environment +//! carries (`WAKER_IDENTITY_NSEC`, another tenant's already-resolved +//! provider credential, proxy/TLS controls) — the gap the original overlay- +//! only version of this parameter left open, since in `buzz-waker`'s +//! multi-tenant model the tenant's own bundle authorizes which provider +//! binary/digest runs, so an unisolated child could otherwise read and +//! exfiltrate secrets across the shared-daemon boundary. +//! +//! `TENANT_BASELINE_VARS` is deliberately small and reviewed per addition, +//! not "whatever the parent happens to have" — growing it back to the full +//! parent environment would silently undo the isolation this exists for. use sha2::{Digest, Sha256}; +use std::collections::HashMap; use std::io::{BufReader, Read, Write}; use std::path::{Path, PathBuf}; use std::sync::mpsc; @@ -29,6 +58,27 @@ const STDERR_CAP: usize = 65536; const STDOUT_CAP: usize = 1_048_576; // 1 MB const PROVIDER_PROTOCOL_VERSION: u64 = 1; +/// Variables kept from this process's own environment when a tenant-scoped +/// `env` overlay is supplied — see the module doc's The child's environment +/// section for why each one is here. Provider-agnostic on purpose: this +/// crate has no notion of which provider binary it is invoking, so it keeps +/// the same small baseline regardless of provider, rather than branching on +/// provider identity. +const TENANT_BASELINE_VARS: &[&str] = &["HOME", "PATH"]; + +/// Build the fixed baseline applied under a tenant-scoped `env` overlay, +/// read fresh from this process's own environment (never from `env` itself). +fn tenant_child_environment_baseline() -> HashMap { + TENANT_BASELINE_VARS + .iter() + .filter_map(|&key| { + std::env::var(key) + .ok() + .map(|value| (key.to_string(), value)) + }) + .collect() +} + /// On Windows, a console-subsystem child gets a fresh, briefly-visible /// console window per invocation unless `CREATE_NO_WINDOW` is set. A pure /// no-op on non-Windows platforms, so callers can call this unconditionally. @@ -102,7 +152,8 @@ fn validate_provider_info(info: &serde_json::Value) -> Result<(), String> { /// `workdir` is the child's working directory (`None` inherits the caller's /// own CWD) — callers with a notion of a stable agent home (the desktop's /// `~/.buzz`) should resolve and pass it; a caller without one may pass -/// `None`. +/// `None`. `Some` isolates the child's environment — see the module doc's +/// The child's environment section. /// /// Reader threads stream lines/chunks over channels so the caller can receive /// data as it arrives and time-box the wait. No `read_to_end` — if a provider @@ -119,6 +170,7 @@ pub fn invoke_provider( request: &serde_json::Value, timeout: Duration, workdir: Option<&Path>, + env: Option<&HashMap>, ) -> Result { let request_bytes = format!( "{}\n", @@ -129,6 +181,11 @@ pub fn invoke_provider( if let Some(workdir) = workdir { cmd.current_dir(workdir); } + if let Some(env) = env { + cmd.env_clear(); + cmd.envs(tenant_child_environment_baseline()); + cmd.envs(env); + } configure_no_window(&mut cmd); let mut child = cmd .stdin(std::process::Stdio::piped()) @@ -563,7 +620,8 @@ fn stage_provider( /// Deploy through one immutable staged copy: negotiate protocol v1 before the /// secret-bearing request, then invoke deploy on those exact same bytes. /// -/// `workdir` is passed straight through to [`invoke_provider`] — see its doc. +/// `workdir` and `env` are passed straight through to [`invoke_provider`] — +/// see its doc. /// /// # Errors /// See [`invoke_provider`] and [`stage_provider`] — every path returns a @@ -573,8 +631,9 @@ pub fn provider_deploy( agent: &serde_json::Value, provider_config: &serde_json::Value, workdir: Option<&Path>, + env: Option<&HashMap>, ) -> Result { - deploy(binary, agent, provider_config, workdir, None) + deploy(binary, agent, provider_config, workdir, env, None) } /// Like [`provider_deploy`], but refuses to run unless the staged binary's @@ -595,11 +654,13 @@ pub fn provider_deploy( /// # Errors /// A digest mismatch is reported before any process is spawned. Otherwise /// see [`provider_deploy`]. +#[allow(clippy::too_many_arguments)] pub fn provider_deploy_pinned( binary: &Path, agent: &serde_json::Value, provider_config: &serde_json::Value, workdir: Option<&Path>, + env: Option<&HashMap>, expected_sha256_hex: &str, ) -> Result { deploy( @@ -607,15 +668,18 @@ pub fn provider_deploy_pinned( agent, provider_config, workdir, + env, Some(expected_sha256_hex), ) } +#[allow(clippy::too_many_arguments)] fn deploy( binary: &Path, agent: &serde_json::Value, provider_config: &serde_json::Value, workdir: Option<&Path>, + env: Option<&HashMap>, expected_sha256_hex: Option<&str>, ) -> Result { let (_directory, staged, digest, _execution_guard) = stage_provider(binary)?; @@ -633,7 +697,13 @@ fn deploy( "op": "info", "request_id": uuid::Uuid::new_v4().to_string(), }); - let info = invoke_provider(&staged, &info_request, Duration::from_secs(10), workdir)?; + let info = invoke_provider( + &staged, + &info_request, + Duration::from_secs(10), + workdir, + env, + )?; validate_provider_info(&info)?; let request = serde_json::json!({ @@ -642,7 +712,7 @@ fn deploy( "agent": agent, "provider_config": provider_config, }); - let resp = invoke_provider(&staged, &request, Duration::from_secs(600), workdir)?; + let resp = invoke_provider(&staged, &request, Duration::from_secs(600), workdir, env)?; let agent_id = resp["agent_id"] .as_str() .map(String::from) diff --git a/crates/buzz-provider-deploy/src/tests.rs b/crates/buzz-provider-deploy/src/tests.rs index 79dad73eb6f..d3357fc49a8 100644 --- a/crates/buzz-provider-deploy/src/tests.rs +++ b/crates/buzz-provider-deploy/src/tests.rs @@ -162,6 +162,7 @@ esac"#, &serde_json::json!({}), &serde_json::json!({}), None, + None, ) .expect("staged deploy"); assert_eq!(outcome.agent_id, "remote-1"); @@ -208,6 +209,7 @@ esac"# &serde_json::json!({}), &serde_json::json!({}), None, + None, ) .expect("staged deploy"); assert_eq!(outcome.fresh_generation, parsed, "wire value {wire_value}"); @@ -257,6 +259,7 @@ esac"#, &serde_json::json!({}), &serde_json::json!({}), None, + None, ) .expect("deploy from immutable staged copy") .agent_id; @@ -304,6 +307,7 @@ esac"#, &serde_json::json!({}), &serde_json::json!({}), None, + None, ) .expect("deploy from immutable staged copy") .agent_id; @@ -342,6 +346,7 @@ esac"#, &serde_json::json!({"private_key_nsec": "nsec1must-not-cross"}), &serde_json::json!({}), None, + None, ) .unwrap_err(); assert!(error.contains("protocol version 2"), "{error}"); @@ -365,6 +370,7 @@ printf '%s\n' '{"ok":true,"version":"1.0.0"}'"#, &serde_json::json!({}), &serde_json::json!({}), None, + None, ) .unwrap_err(); assert!( @@ -391,6 +397,7 @@ fn provider_deploy_pinned_refuses_a_digest_mismatch_before_any_negotiation() { &serde_json::json!({"private_key_nsec": "nsec1must-not-cross"}), &serde_json::json!({}), None, + None, &"f".repeat(64), ) .unwrap_err(); @@ -424,12 +431,144 @@ esac"#, &serde_json::json!({}), &serde_json::json!({}), None, + None, &expected, ) .expect("digest matched, deploy proceeds"); assert_eq!(outcome.agent_id, "pinned-1"); } +/// The headline case `env` exists for: a caller's overlay must actually +/// reach the child process, not just be accepted and silently dropped. +#[cfg(unix)] +#[test] +fn provider_deploy_env_overlay_reaches_the_child_process() { + let directory = tempfile::tempdir().unwrap(); + let provider = directory.path().join("provider"); + write_test_provider( + &provider, + r#"read request +case "$request" in + *\"op\":\"info\"*) printf '%s\n' '{"ok":true,"name":"test","version":"1.0.0","protocol_version":1,"description":"test provider","config_schema":{}}' ;; + *\"op\":\"deploy\"*) printf '{"ok":true,"agent_id":"%s"}\n' "$TEST_TENANT_TOKEN" ;; +esac"#, + ); + + let mut env = std::collections::HashMap::new(); + env.insert( + "TEST_TENANT_TOKEN".to_string(), + "sprt-tenant-abc".to_string(), + ); + + let outcome = provider_deploy( + &provider, + &serde_json::json!({}), + &serde_json::json!({}), + None, + Some(&env), + ) + .expect("deploy with an env overlay"); + assert_eq!( + outcome.agent_id, "sprt-tenant-abc", + "the child must see the overlaid variable" + ); +} + +/// `Some(map)` overrides an inherited variable of the same name — the +/// module doc's own contract, and the property `buzz-waker` relies on if a +/// baseline var and a tenant var ever collide. +#[cfg(unix)] +#[test] +fn provider_deploy_env_overlay_overrides_an_inherited_variable() { + let directory = tempfile::tempdir().unwrap(); + let provider = directory.path().join("provider"); + write_test_provider( + &provider, + r#"read request +case "$request" in + *\"op\":\"info\"*) printf '%s\n' '{"ok":true,"name":"test","version":"1.0.0","protocol_version":1,"description":"test provider","config_schema":{}}' ;; + *\"op\":\"deploy\"*) printf '{"ok":true,"agent_id":"%s"}\n' "$TEST_TENANT_TOKEN" ;; +esac"#, + ); + + // SAFETY: test-only, single-threaded at this point in the process + // (no other test in this crate reads or writes this exact variable). + unsafe { + std::env::set_var("TEST_TENANT_TOKEN", "inherited-from-parent"); + } + let mut env = std::collections::HashMap::new(); + env.insert( + "TEST_TENANT_TOKEN".to_string(), + "overlaid-value".to_string(), + ); + + let outcome = provider_deploy( + &provider, + &serde_json::json!({}), + &serde_json::json!({}), + None, + Some(&env), + ) + .expect("deploy with an env overlay"); + assert_eq!(outcome.agent_id, "overlaid-value"); + + unsafe { + std::env::remove_var("TEST_TENANT_TOKEN"); + } +} + +/// A tenant-scoped `env` overlay must isolate the child from this process's +/// own environment, not merely overlay on top of it — an unrelated inherited +/// variable (standing in for a daemon secret like `WAKER_IDENTITY_NSEC` or +/// another tenant's already-resolved provider credential) must not reach the +/// child, even though it is never mentioned in `env` or in +/// `TENANT_BASELINE_VARS`. +#[cfg(unix)] +#[test] +fn provider_deploy_env_overlay_clears_unrelated_inherited_variables() { + let directory = tempfile::tempdir().unwrap(); + let provider = directory.path().join("provider"); + write_test_provider( + &provider, + r#"read request +case "$request" in + *\"op\":\"info\"*) printf '%s\n' '{"ok":true,"name":"test","version":"1.0.0","protocol_version":1,"description":"test provider","config_schema":{}}' ;; + *\"op\":\"deploy\"*) + if [ -z "${TEST_SHOULD_NOT_LEAK:-}" ]; then + printf '{"ok":true,"agent_id":"cleared"}\n' + else + printf '{"ok":true,"agent_id":"leaked"}\n' + fi + ;; +esac"#, + ); + + // SAFETY: test-only, single-threaded at this point in the process + // (no other test in this crate reads or writes this exact variable). + unsafe { + std::env::set_var("TEST_SHOULD_NOT_LEAK", "daemon-secret"); + } + let env = std::collections::HashMap::new(); + + let outcome = provider_deploy( + &provider, + &serde_json::json!({}), + &serde_json::json!({}), + None, + Some(&env), + ) + .expect("deploy with an empty tenant env overlay"); + assert_eq!( + outcome.agent_id, "cleared", + "a variable inherited from this process but absent from both the tenant overlay \ + and TENANT_BASELINE_VARS must not reach the child" + ); + + unsafe { + std::env::remove_var("TEST_SHOULD_NOT_LEAK"); + } +} + #[test] fn provider_info_requires_the_complete_flat_wire_shape() { let complete = serde_json::json!({ diff --git a/crates/buzz-waker/src/effects.rs b/crates/buzz-waker/src/effects.rs index fb973b88f0d..4c873f9b3a4 100644 --- a/crates/buzz-waker/src/effects.rs +++ b/crates/buzz-waker/src/effects.rs @@ -45,20 +45,27 @@ //! `select_wake_candidates` filtered against can be minutes stale. The //! desktop's version re-checks the full managed-agent roster (local ∪ //! relay-registered). This daemon has no such roster — it only knows the -//! agents it was configured to watch — so its baseline is that watch list. -//! Documented in `PLANS/BUZZ_WAKER_DESIGN.md` as an accepted gap: an author -//! that is a *managed agent this daemon does not watch* is not caught here. -//! It is still caught by the synchronous baseline at admission time whenever -//! that baseline is populated the same way; the re-check only narrows the -//! window, and narrowing it to "this daemon's own agents" is strictly better -//! than not re-checking at all. - +//! agents it was configured to watch — so its baseline is that watch list, +//! held in a [`crate::watch_list::WatchList`] and read live rather than +//! snapshotted, since the dynamic supervisor (`PLANS/BUZZ_WAKER_DESIGN.md` +//! §12 build order step 3) can add or remove a watched agent at any time — +//! see that module's own doc for why a frozen snapshot would let a +//! roster-added agent's mention wake another agent undetected. Still +//! documented as an accepted gap: an author that is a *managed agent this +//! daemon does not watch* is not caught here. It is still caught by the +//! synchronous baseline at admission time whenever that baseline is +//! populated the same way; the re-check only narrows the window, and +//! narrowing it to "this daemon's own agents" is strictly better than not +//! re-checking at all. + +use std::collections::HashMap; use std::sync::Arc; use crate::attempt::{HeartbeatObservation, WakeEffects}; use crate::bundle::LaunchBundleBody; use crate::decide::normalize_pubkey; use crate::presence_feed::{PresenceError, PresenceState}; +use crate::watch_list::WatchList; use buzz_core::PresenceStatus; use tokio_util::sync::CancellationToken; @@ -138,10 +145,12 @@ pub struct RealWakeEffects { /// The presence tap shared with every attempt for this agent — one tap /// per watched agent, not per attempt. presence_state: Arc, - /// This daemon's full watch list, normalized. Used only by + /// This daemon's live watch list. Used only by /// `confirm_author_not_known_agent`; see the module note on why this is - /// the accepted baseline rather than a full managed-agent roster. - watch_list: Arc<[String]>, + /// the accepted baseline rather than a full managed-agent roster, and + /// `crate::watch_list`'s own doc on why it must be read live rather than + /// snapshotted once an agent can be added or removed at runtime. + watch_list: WatchList, /// The pubkey of the agent this attempt is scoped to — this daemon's own /// watched identity, never derived from the bundle. Compared against /// `bundle.agent_pubkey` before any deploy, so a bundle transport bug @@ -158,6 +167,17 @@ pub struct RealWakeEffects { /// This attempt's launch bundle, if one is available. `None` until /// bundle transport is wired into this daemon — see the module note. bundle: Option>, + /// This agent's own provider credential, if it has one — a dynamically + /// (roster-)enrolled agent's [`crate::enrolment::ProviderCredential`], + /// already converted to its wire-shape environment variables + /// ([`crate::enrolment::ProviderCredential::to_env`]) at credential + /// delivery time. `None` for a statically configured agent (no + /// per-tenant credential exists) and for a roster-enrolled agent whose + /// delivered credential didn't carry one. Passed as the deploy + /// subprocess's `env` — see `buzz_provider_deploy`'s own module doc for + /// what that does (isolates the child's environment, not an overlay on + /// this daemon's own). + provider_env: Option>>, /// Fires on daemon shutdown. Deliberately **not** tied to the mention /// feed's own connection lifecycle: a wake attempt does not touch that /// socket, so a feed reconnect must not cancel an attempt that is still @@ -178,11 +198,12 @@ impl RealWakeEffects { #[allow(clippy::too_many_arguments)] pub fn new( presence_state: Arc, - watch_list: Arc<[String]>, + watch_list: WatchList, watched_agent_pubkey: &str, trigger_author: &str, trigger_created_at: u64, bundle: Option>, + provider_env: Option>>, cancel: CancellationToken, on_deployed: impl Fn() + Send + Sync + 'static, ) -> Self { @@ -193,6 +214,7 @@ impl RealWakeEffects { trigger_author: normalize_pubkey(trigger_author), trigger_created_at, bundle, + provider_env, cancel, on_deployed: Box::new(on_deployed), } @@ -244,13 +266,13 @@ impl WakeEffects for RealWakeEffects { async fn confirm_author_not_known_agent(&self) -> Result { // `Ok(true)` means "confirmed not a known agent" — see the trait doc. - // Every entry in the watch list is, by definition, a known agent this - // daemon manages, so the author is clear exactly when it matches - // none of them. - Ok(!self - .watch_list - .iter() - .any(|watched| watched == &self.trigger_author)) + // Every member of the watch list is, by definition, a known agent + // this daemon manages, so the author is clear exactly when it is not + // currently a member. Read live (`WatchList::contains`), not from a + // snapshot taken when this attempt was constructed — the whole point + // of a fresh re-check is to catch a watch-list change since the + // synchronous baseline ran. + Ok(!self.watch_list.contains(&self.trigger_author)) } async fn start_managed_agent(&self) -> Result, Self::Error> { @@ -328,6 +350,7 @@ impl WakeEffects for RealWakeEffects { serde_json::Value::String(self.trigger_created_at.to_string()); let provider_config = bundle.provider.provider_config.clone(); + let provider_env = self.provider_env.clone(); let outcome = tokio::task::spawn_blocking(move || { buzz_provider_deploy::provider_deploy_pinned( @@ -335,6 +358,7 @@ impl WakeEffects for RealWakeEffects { &agent_json, &provider_config, None, + provider_env.as_deref(), &expected_digest, ) }) @@ -364,7 +388,7 @@ mod tests { #[allow(clippy::too_many_arguments)] fn effects_with( presence_state: Arc, - watch_list: Arc<[String]>, + watch_list: WatchList, watched_agent_pubkey: &str, trigger_author: &str, bundle: Option>, @@ -378,6 +402,7 @@ mod tests { trigger_author, 1_000, bundle, + None, cancel, on_deployed, ) @@ -406,7 +431,7 @@ mod tests { async fn an_unresolved_presence_tap_reports_unavailable() { let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, @@ -427,7 +452,7 @@ mod tests { presence_state.observe("ev1", PresenceStatus::Online, 1_000); let effects = effects_with( presence_state, - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, @@ -443,7 +468,7 @@ mod tests { let watched = "bb".repeat(32); let effects = effects_with( state(), - Arc::from(vec![watched.clone()]), + WatchList::from(vec![watched.clone()]), "aa".repeat(32).as_str(), &watched, None, @@ -458,7 +483,7 @@ mod tests { async fn an_author_off_the_watch_list_is_confirmed_clear() { let effects = effects_with( state(), - Arc::from(vec!["bb".repeat(32)]), + WatchList::from(vec!["bb".repeat(32)]), "aa".repeat(32).as_str(), "cc".repeat(32).as_str(), None, @@ -474,7 +499,7 @@ mod tests { let watched = "BB".repeat(32); let effects = effects_with( state(), - Arc::from(vec![normalize_pubkey(&watched)]), + WatchList::from(vec![normalize_pubkey(&watched)]), "aa".repeat(32).as_str(), &watched, None, @@ -489,7 +514,7 @@ mod tests { async fn start_managed_agent_without_a_bundle_reports_no_bundle() { let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, @@ -511,7 +536,7 @@ mod tests { let bundle = bundle_for(&watched, u64::MAX); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), &watched, "aa".repeat(32).as_str(), Some(bundle), @@ -534,7 +559,7 @@ mod tests { let bundle = bundle_for(&other_agent, u64::MAX); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), &watched, "aa".repeat(32).as_str(), Some(bundle), @@ -562,7 +587,7 @@ mod tests { let bundle = bundle_for(&watched, 1); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), &watched, "aa".repeat(32).as_str(), Some(bundle), @@ -590,7 +615,7 @@ mod tests { let called_clone = called.clone(); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, @@ -607,7 +632,7 @@ mod tests { let cancel = CancellationToken::new(); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, @@ -624,7 +649,7 @@ mod tests { let cancel = CancellationToken::new(); let effects = effects_with( state(), - Arc::from(vec![]), + WatchList::from(vec![]), "aa".repeat(32).as_str(), "aa".repeat(32).as_str(), None, diff --git a/crates/buzz-waker/src/enrolment.rs b/crates/buzz-waker/src/enrolment.rs index dc4aaee9f2c..34cf02294dc 100644 --- a/crates/buzz-waker/src/enrolment.rs +++ b/crates/buzz-waker/src/enrolment.rs @@ -384,6 +384,26 @@ impl std::fmt::Debug for ProviderCredential { } } +impl ProviderCredential { + /// This credential as the environment variable(s) a provider subprocess + /// reads it from — `buzz-provider-deploy`'s `env` overlay param, laid + /// over the deployed process's environment for exactly one tenant's own + /// deploy call. One key per variant today + /// (`credentials::resolve()` in `crates/buzz-backend-sprites` for + /// `Sprites`'s own `SPRITE_TOKEN`), but returns a map rather than a + /// single pair since a future provider variant may need more than one. + #[must_use] + pub fn to_env(&self) -> std::collections::HashMap { + let mut env = std::collections::HashMap::new(); + match self { + Self::Sprites { sprite_token } => { + env.insert("SPRITE_TOKEN".to_string(), sprite_token.clone()); + } + } + env + } +} + /// The signed content of one agent's delivered credential. /// /// `Debug` is implemented by hand and redacts [`Self::nsec`] and @@ -1017,6 +1037,19 @@ mod tests { assert!(rendered.contains("")); } + #[test] + fn sprites_credential_maps_to_sprite_token_env() { + let credential = ProviderCredential::Sprites { + sprite_token: "sprt-real-value".to_string(), + }; + let env = credential.to_env(); + assert_eq!( + env.get("SPRITE_TOKEN"), + Some(&"sprt-real-value".to_string()) + ); + assert_eq!(env.len(), 1); + } + #[test] fn a_signed_credential_debug_impl_redacts_body_json() { let owner = keypair(); diff --git a/crates/buzz-waker/src/lib.rs b/crates/buzz-waker/src/lib.rs index 05c340ad650..916dbae9e93 100644 --- a/crates/buzz-waker/src/lib.rs +++ b/crates/buzz-waker/src/lib.rs @@ -49,11 +49,11 @@ //! [`bundle_feed`]'s connect/backoff/idle-timeout shape, also authenticated //! as the waker's own identity, decrypting and admitting one agent's //! delivered `nsec` against a per-agent [`floors::FloorStore`]. -//! `docs/waker-agent-enrolment.md` (design) and `PLANS/BUZZ_WAKER_DESIGN.md` -//! §12 (build order) — the dynamic per-agent supervisor `main.rs` needs to -//! diff [`roster_feed::RosterState`] against the daemon's watch list and -//! spawn/cancel [`credential_feed::run_credential_tap`] instances is the -//! next phase, not yet implemented. +//! - [`watch_list`] — [`watch_list::WatchList`], this daemon's live known-agent +//! set. `main.rs`'s dynamic supervisor (`PLANS/BUZZ_WAKER_DESIGN.md` §12 +//! build order step 3) diffs [`roster_feed::RosterState`] against it, +//! spawning/cancelling [`credential_feed::run_credential_tap`] plus each +//! agent's presence/bundle/wake-loop tasks as the roster changes. //! //! Each exists because of a specific review finding and carries the gate id //! (`G1`–`G4`) it discharges, so the reason is not lost. @@ -73,6 +73,7 @@ pub mod presence_feed; pub mod relay_feed; pub mod roster_feed; pub mod wake_loop; +pub mod watch_list; pub use attempt::{ is_managed_agent_live, is_presumed_delivered_by_floor, is_wake_attempt_debounced, diff --git a/crates/buzz-waker/src/main.rs b/crates/buzz-waker/src/main.rs index 82773b946a1..5b51f965d0b 100644 --- a/crates/buzz-waker/src/main.rs +++ b/crates/buzz-waker/src/main.rs @@ -4,31 +4,65 @@ //! `crates/buzz-relay/src/main.rs` and `crates/buzz-pair-relay/src/main.rs` //! for the pattern this follows. JSON-structured logs, graceful shutdown on //! SIGTERM/Ctrl+C via a shared [`CancellationToken`], and three tasks spawned -//! per configured agent: the mention-feed loop +//! per watched agent: the mention-feed loop //! ([`buzz_waker::wake_loop::run_wake_loop`]), the presence tap //! ([`buzz_waker::presence_feed::run_presence_tap`]), and the bundle-delivery //! tap ([`buzz_waker::bundle_feed::run_bundle_tap`]). //! +//! # Two ways an agent gets watched +//! +//! **Statically**, from `WAKER_AGENTS_CONFIG_PATH` — read once at startup, +//! same as before this module's dynamic supervisor (below) existed. +//! +//! **Dynamically**, via the roster tap +//! ([`buzz_waker::roster_feed::run_roster_tap`]), when `WAKER_OWNER_PUBKEYS` +//! is non-empty (`PLANS/BUZZ_WAKER_DESIGN.md` §12 build order step 3). This +//! daemon's own reconciliation loop diffs every authorized owner's current +//! roster against its own `supervised` map, spawns a per-agent credential tap +//! ([`buzz_waker::credential_feed::run_credential_tap`]) to fetch a +//! newly-listed agent's `nsec` before it can be watched at all, then calls +//! the same [`spawn_agent_watch`] a static agent uses once that credential +//! arrives — and cancels a previously roster-added agent's tasks the moment +//! it drops off every authorized owner's roster. A statically configured +//! agent always wins a pubkey collision against a roster-discovered one: see +//! [`compute_desired_roster_agents`]'s own doc. +//! //! # Configuration //! //! | Env var | Required | Meaning | //! |---|---|---| -//! | `WAKER_RELAY_URL` | yes | The relay every watched agent's mention feed, presence tap, and bundle tap connects to. | -//! | `WAKER_STATE_DIR` | yes | Base directory for durable per-agent state (`//{cursor,floor}.json`). Created if missing. | -//! | `WAKER_AGENTS_CONFIG_PATH` | yes | Path to a JSON file listing the agents to watch — see [`AgentConfig`]. | +//! | `WAKER_RELAY_URL` | yes | The relay every watched agent's mention feed, presence tap, and bundle tap connects to — also the relay the roster/credential taps below connect to. | +//! | `WAKER_STATE_DIR` | yes | Base directory for durable per-agent state (`//{cursor,floor,credential_floor}.json`) and the roster's own per-owner floors (`/roster-floors/.json`). Created if missing. | +//! | `WAKER_AGENTS_CONFIG_PATH` | yes | Path to a JSON file listing the agents to statically watch — see [`AgentConfig`]. | +//! | `WAKER_OWNER_PUBKEYS` | no | Comma-separated list of owner pubkeys this daemon discovers agents for dynamically. Empty or unset disables dynamic enrolment entirely — see [`buzz_waker::enrolment::parse_authorized_owners`]'s own fail-closed doc. | +//! | `WAKER_IDENTITY_NSEC` | only if `WAKER_OWNER_PUBKEYS` is set | This daemon's own Nostr identity — the roster and credential taps decrypt as this key, never as any watched agent's. | +//! | `WAKER_MAX_AGENTS` | only if `WAKER_OWNER_PUBKEYS` is set | Total ceiling on supervised agents — config, pending, and running together — dynamic enrolment can ever push this daemon to. Refuse-not-evict: an authorized owner's roster past this ceiling is refused new admissions, never made to cancel an existing agent to make room. See [`reconcile_roster`]'s own doc. | //! | `RUST_LOG` | no | `tracing-subscriber` env filter. Defaults to `buzz_waker=info`. | //! //! # What is still deliberately not here //! -//! Agent identities, the watch list, and each agent's owner pubkey (pinned +//! Every *statically* configured agent's identity and owner pubkey (pinned //! into its [`buzz_waker::floors::FloorStore`] on first run, **G2**) are read //! from local config, matching the ecosystem's existing agent-identity //! provisioning story — nothing about *that* pin can come from a delivered -//! bundle without defeating the pin's own purpose. +//! bundle without defeating the pin's own purpose. A dynamically discovered +//! agent's owner is instead the roster entry's own owner, already proven +//! against `WAKER_OWNER_PUBKEYS` before this daemon ever trusts it (see +//! [`buzz_waker::roster_feed`]'s module doc). +//! +//! A dynamically watched agent's `provider_credential`, when its delivered +//! credential carries one, becomes that agent's own deploy subprocess +//! environment (`buzz_provider_deploy`'s `env` parameter): the child's +//! environment is cleared and rebuilt from a small fixed baseline plus this +//! credential, not this daemon's own inherited environment. See +//! `buzz_provider_deploy`'s own module doc, The child's environment +//! section, for exactly what that baseline is and why. +use std::collections::HashMap; use std::collections::HashSet; -use std::path::PathBuf; +use std::path::{Path, PathBuf}; use std::sync::Arc; +use std::time::Duration; use nostr::{Keys, Tag}; use serde::Deserialize; @@ -37,10 +71,14 @@ use tokio_util::sync::CancellationToken; use tracing_subscriber::{fmt, prelude::*, EnvFilter}; use buzz_waker::bundle_feed::{run_bundle_tap, BundleState}; +use buzz_waker::credential_feed::{run_credential_tap, CredentialState}; use buzz_waker::decide::normalize_pubkey; -use buzz_waker::floors::FloorStore; +use buzz_waker::enrolment::{parse_authorized_owners, RosterEntry}; +use buzz_waker::floors::{FloorError, FloorStore}; use buzz_waker::presence_feed::{run_presence_tap, PresenceState}; +use buzz_waker::roster_feed::{run_roster_tap, RosterState}; use buzz_waker::wake_loop::{run_wake_loop, WakeLoopConfig}; +use buzz_waker::watch_list::WatchList; /// One watched agent, as read from `WAKER_AGENTS_CONFIG_PATH`. #[derive(Debug, Deserialize)] @@ -109,8 +147,38 @@ fn ensure_owner_pin_matches( if pinned_owner != configured_owner { anyhow::bail!( "floor store for {pubkey} is pinned to owner {pinned_owner}, but \ - WAKER_AGENTS_CONFIG_PATH now configures owner {configured_owner}; refusing \ - to run with disagreeing owners" + the configured owner is now {configured_owner}; refusing to run with \ + disagreeing owners" + ); + } + Ok(()) +} + +/// `WAKER_MAX_AGENTS` is documented as a total ceiling over config, pending, +/// and running agents together, not just a dynamic-admission threshold — +/// [`reconcile_roster`]'s own refuse-not-evict check only ever guards +/// roster *additions* against it, so an oversized static baseline would +/// otherwise start unchecked and stay that way for the daemon's whole life. +/// Called once at startup, before any agent (static or otherwise) is +/// spawned or gets per-agent state on disk — the one place this can still +/// fail closed. A no-op when dynamic enrolment is disabled +/// (`authorized_owners` empty): `max_agents` is `0` in that case by +/// definition (see [`main`]'s own parsing), not a real ceiling to enforce. +/// +/// # Errors +/// Dynamic enrolment is enabled and `static_agent_count` alone already +/// exceeds `max_agents`. +fn ensure_static_agent_count_fits_cap( + static_agent_count: usize, + max_agents: usize, + dynamic_enrolment_enabled: bool, +) -> anyhow::Result<()> { + if dynamic_enrolment_enabled && static_agent_count > max_agents { + anyhow::bail!( + "WAKER_MAX_AGENTS={max_agents} but WAKER_AGENTS_CONFIG_PATH already lists \ + {static_agent_count} statically configured agents; WAKER_MAX_AGENTS is a total \ + ceiling over config, pending, and running agents together, so the static \ + baseline alone must fit within it" ); } Ok(()) @@ -169,6 +237,680 @@ async fn shutdown_signal() { } } +/// How a watched pubkey came to be watched — governs both dedup ordering +/// (config always wins a collision) and how loudly this daemon reacts to +/// one of its tasks exiting unsolicited: see [`classify_exit`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum AgentSource { + /// Listed in `WAKER_AGENTS_CONFIG_PATH`. + Config, + /// Discovered via an authorized owner's roster. + Roster, +} + +/// This daemon's bookkeeping for one currently-watched (or being-bootstrapped) +/// pubkey. +struct SupervisedAgent { + /// Shared by every task this pubkey owns (credential tap, presence tap, + /// bundle tap, wake loop) — a [`CancellationToken::child_token`] of the + /// daemon's own global token. Cancelling it stops exactly this pubkey's + /// tasks, nothing else; the global token cancelling stops it too, along + /// with everything else, on shutdown. + cancel: CancellationToken, + source: AgentSource, + /// The owner that published this pubkey's roster entry. Unused for + /// [`AgentSource::Config`] (each config entry already carries its own + /// `owner_pubkey` separately, read once at spawn time). + owner_pubkey: String, + /// `Some` for [`AgentSource::Roster`] only — the credential tap's live + /// state, watched for this agent's **entire** supervised lifetime, not + /// just while bootstrapping. A rotation or revocation delivered after + /// this agent is already running has to be seen too, or the old + /// identity (and, for a revocation, a credential the owner explicitly + /// withdrew) would keep running indefinitely — the exact gap Alex's + /// review round caught. `None` for [`AgentSource::Config`]. + credential_state: Option>, + /// `Some` for [`AgentSource::Roster`] only — the `credential_version` + /// the *most recent* roster reconciliation observed for this pubkey. + /// Compared against `credential_state.current()`'s own + /// `credential_version` every tick: promotion requires an exact match + /// (a stale or not-yet-caught-up delivery must not start this agent), + /// and the roster diff tears this agent down the moment the roster's + /// own claimed version moves, so it re-bootstraps from the new one + /// rather than continuing to run under the old identity. `None` for + /// [`AgentSource::Config`], which has no roster-published version at + /// all. + expected_credential_version: Option, + /// Whether this agent's presence/bundle/wake tasks have been spawned. + /// `true` immediately for [`AgentSource::Config`]. `false` for a + /// [`AgentSource::Roster`] agent until its credential (at the expected + /// version) arrives and [`spawn_agent_watch`] succeeds. + running: bool, +} + +/// Why one of this daemon's tasks finished, for the join-handling loop in +/// [`main`] to classify via [`classify_exit`]. +enum TaskExit { + /// One of a watched agent's three tasks (`"presence_tap"`, + /// `"bundle_tap"`, or `"wake_loop"`). + /// + /// `generation` is a clone of the exact [`SupervisedAgent::cancel`] + /// token this task was spawned under, captured at spawn time — not + /// re-derived from `supervised` at exit time. A roster version change + /// tears down and re-adopts the same pubkey within one + /// [`reconcile_roster`] pass, so by the time a *predecessor* + /// generation's task actually finishes, `supervised` already holds the + /// *replacement* generation's fresh, uncancelled token under that same + /// pubkey — looking it up by pubkey at exit time would misclassify the + /// predecessor's expected exit as the replacement's unsolicited one. + /// Checking this captured token's own `is_cancelled()` instead is + /// correct regardless of what currently occupies that pubkey, because + /// [`tear_down_roster_agent`] always cancels a generation's token + /// before anything replaces its `supervised` entry. + Agent { + pubkey: String, + task: &'static str, + generation: CancellationToken, + }, + /// A [`AgentSource::Roster`] agent's credential tap, tracked separately + /// from `Agent` only so log lines name it correctly — it shares that + /// agent's own [`SupervisedAgent::cancel`] and is classified exactly the + /// same way, `generation` included for the same reason. + CredentialTap { + pubkey: String, + generation: CancellationToken, + }, + /// A daemon-wide task with no per-agent scope: the roster tap. Expected + /// to run until the global token cancels; any other exit is fatal. + Component(&'static str), +} + +/// What an unsolicited task exit means for the daemon. +#[derive(Debug, PartialEq, Eq)] +enum ExitDisposition { + /// Already accounted for — the owning entry was already removed from + /// `supervised` (an earlier sibling's exit already tore this pubkey + /// down), or its own token was already cancelled deliberately. No + /// further action. + Expected, + /// This daemon cannot recover from this exit; stop everything. + FatalDaemon, + /// Tear down just this one roster-discovered agent's remaining tasks + /// and keep running everything else — a single tenant's agent + /// misbehaving must not take down a daemon serving several. + TearDownAgent, +} + +/// Classify one task's exit, given whether it was already-known-cancelled at +/// the moment it finished and which agent (if any) it belonged to. +/// +/// `was_cancelled = true` covers both a deliberate per-agent cancellation +/// (roster removed this pubkey) and a global shutdown (every child token +/// cancels transitively) — either way, the exit is expected, not a failure +/// this function needs to react to. +/// +/// A statically configured agent's unsolicited exit is fatal for the whole +/// daemon — the historical behavior, preserved exactly: before this +/// module's dynamic supervisor, *every* watched agent was config-sourced, +/// so this is also what makes today's default (no `WAKER_OWNER_PUBKEYS`) +/// behave identically to before this file changed. +/// +/// `source = None` only reaches this function as `(None, true)` from the +/// real call site in [`main`] — it derives `was_cancelled` via +/// `supervised.get(pubkey).map(|a| a.cancel.is_cancelled()).unwrap_or(true)`, +/// so a pubkey no longer in `supervised` (already torn down by an earlier +/// sibling task's exit) always reports `was_cancelled = true` and short +/// circuits above. `(None, false)` is therefore not reachable from `main` +/// today; it is still handled here, conservatively, as fatal rather than +/// `unreachable!` — a caller that ever changes that default should fail +/// loud, not silently swallow an exit this function has no real source for. +fn classify_exit(source: Option, was_cancelled: bool) -> ExitDisposition { + if was_cancelled { + return ExitDisposition::Expected; + } + match source { + None | Some(AgentSource::Config) => ExitDisposition::FatalDaemon, + Some(AgentSource::Roster) => ExitDisposition::TearDownAgent, + } +} + +/// One agent this daemon should be watching because an authorized owner's +/// roster lists it. +struct DesiredRosterAgent { + pubkey: String, + owner_pubkey: String, + /// The [`RosterEntry::credential_version`] the roster currently expects + /// — the exact version [`SupervisedAgent::expected_credential_version`] + /// must match before this agent is promoted, or must still match for + /// an already-running agent to keep running unchanged. + credential_version: u64, +} + +/// Diff every authorized owner's current roster into the set of pubkeys this +/// daemon should be watching dynamically. +/// +/// `rosters` is `(owner_pubkey, entries)` for every owner this daemon +/// currently has a tracked roster for (from +/// [`buzz_waker::roster_feed::RosterState`] — read at the call site, not +/// here, so this stays a plain function over data rather than needing a live +/// `RosterState` to unit test). `existing_sources` is a snapshot of +/// `main`'s own `supervised` map, pubkey to [`AgentSource`] — enough to +/// enforce the one dedup rule that matters: **a statically configured +/// pubkey is never touched by this diff**, in either direction. It is never +/// added again (it is already supervised) and never removed (removal below +/// only ever targets [`AgentSource::Roster`] entries) — see +/// `PLANS/BUZZ_WAKER_DESIGN.md` §12's own note that a broken enrolment path +/// must never be load-bearing for an agent an operator explicitly +/// configured. +/// +/// Returns the desired set plus every pubkey more than one authorized owner +/// claims in this pass — informational for the caller to log loudly (an +/// operator misconfiguration, not something this function can resolve on +/// its own); the first owner encountered wins that pubkey, deterministic +/// only in `rosters`' own iteration order. +fn compute_desired_roster_agents( + rosters: &[(String, Vec)], + existing_sources: &HashMap, +) -> (Vec, Vec) { + let mut desired = Vec::new(); + let mut claimed_by: HashMap = HashMap::new(); + let mut conflicts = Vec::new(); + + for (owner_pubkey, entries) in rosters { + for entry in entries { + let pubkey = normalize_pubkey(&entry.agent_pubkey); + if existing_sources.get(&pubkey) == Some(&AgentSource::Config) { + continue; + } + match claimed_by.get(&pubkey) { + Some(first_owner) if first_owner != owner_pubkey => { + conflicts.push(pubkey); + } + Some(_) => {} + None => { + claimed_by.insert(pubkey.clone(), owner_pubkey.clone()); + desired.push(DesiredRosterAgent { + pubkey, + owner_pubkey: owner_pubkey.clone(), + credential_version: entry.credential_version, + }); + } + } + } + } + + (desired, conflicts) +} + +/// Open (or, the first time this daemon has ever seen `pubkey` under this +/// exact floor path, enroll) a [`FloorStore`] pinned to `owner_pubkey`, and +/// refuse it if a previous pin disagrees. +/// +/// Shared by the bundle floor and the credential floor, and by both a +/// statically configured agent (whose owner never changes across restarts) +/// and a roster-discovered one (whose claimed owner is re-validated against +/// the pin on every reconciliation tick that touches it). +/// +/// # Errors +/// The store cannot be created/opened, or its pinned owner disagrees with +/// `owner_pubkey`. +fn open_pinned_floor_store(path: &Path, owner_pubkey: &str) -> anyhow::Result { + let store = match FloorStore::open(path) { + Ok(store) => store, + Err(FloorError::NotEnrolled { .. }) => { + FloorStore::enroll(path, owner_pubkey).map_err(|e| { + anyhow::anyhow!("could not enroll floor store at {}: {e}", path.display()) + })? + } + Err(e) => anyhow::bail!("could not open floor store at {}: {e}", path.display()), + }; + let pinned_owner = normalize_pubkey(&store.snapshot().owner_pubkey); + ensure_owner_pin_matches(&path.display().to_string(), &pinned_owner, owner_pubkey)?; + Ok(store) +} + +/// Spawn one agent's presence tap, bundle tap, and wake loop under `cancel`, +/// and register it in `watch_list`. +/// +/// The extracted "per-agent spawn block" +/// `PLANS/BUZZ_WAKER_DESIGN.md` §12 build order step 3 calls for — shared by +/// both a statically configured agent (called once per entry at startup, +/// always with `provider_env: None` — no per-tenant credential exists for +/// one) and a roster-discovered one (called once its credential tap +/// delivers a first `nsec` at the expected version, `provider_env` derived +/// from that same delivery's `provider_credential` if it carried one), so +/// there is exactly one place this wiring can drift. +/// +/// # Errors +/// The agent's state directory or bundle [`FloorStore`] cannot be +/// created/opened, or the store's pinned owner disagrees with +/// `owner_pubkey` — see [`open_pinned_floor_store`]. +#[allow(clippy::too_many_arguments)] +fn spawn_agent_watch( + relay_url: &str, + state_dir: &Path, + keys: &Keys, + auth_tag: Option<&Tag>, + owner_pubkey: &str, + provider_env: Option>>, + watch_list: &WatchList, + cancel: CancellationToken, + tasks: &mut JoinSet, +) -> anyhow::Result<()> { + let pubkey = normalize_pubkey(&keys.public_key().to_hex()); + let agent_dir = state_dir.join(&pubkey); + std::fs::create_dir_all(&agent_dir) + .map_err(|e| anyhow::anyhow!("could not create state dir {}: {e}", agent_dir.display()))?; + let cursor_path = agent_dir.join("cursor.json"); + let floor_path = agent_dir.join("floor.json"); + + let mut floor_store = open_pinned_floor_store(&floor_path, owner_pubkey)?; + + let presence_state = Arc::new(PresenceState::new()); + let bundle_state = Arc::new(BundleState::new()); + + tracing::info!(agent = %pubkey, owner = %owner_pubkey, "buzz-waker: watching agent"); + + { + let relay_url = relay_url.to_string(); + let keys = keys.clone(); + let auth_tag = auth_tag.cloned(); + let presence_state = Arc::clone(&presence_state); + let cancel = cancel.clone(); + let pubkey = pubkey.clone(); + tasks.spawn(async move { + run_presence_tap( + &relay_url, + &keys, + auth_tag.as_ref(), + &presence_state, + &cancel, + ) + .await; + TaskExit::Agent { + pubkey, + task: "presence_tap", + generation: cancel, + } + }); + } + + { + let relay_url = relay_url.to_string(); + let keys = keys.clone(); + let auth_tag = auth_tag.cloned(); + let owner_pubkey = owner_pubkey.to_string(); + let bundle_state = Arc::clone(&bundle_state); + let cancel = cancel.clone(); + let pubkey = pubkey.clone(); + tasks.spawn(async move { + run_bundle_tap( + &relay_url, + &keys, + auth_tag.as_ref(), + &owner_pubkey, + &mut floor_store, + &bundle_state, + &cancel, + ) + .await; + TaskExit::Agent { + pubkey, + task: "bundle_tap", + generation: cancel, + } + }); + } + + { + let config = WakeLoopConfig { + relay_url: relay_url.to_string(), + keys: keys.clone(), + auth_tag: auth_tag.cloned(), + cursor_path, + presence_state, + watch_list: watch_list.clone(), + bundle_state, + provider_env: provider_env.clone(), + }; + let cancel = cancel.clone(); + let generation = cancel.clone(); + let pubkey = pubkey.clone(); + tasks.spawn(async move { + run_wake_loop(config, cancel).await; + TaskExit::Agent { + pubkey, + task: "wake_loop", + generation, + } + }); + } + + watch_list.insert(&pubkey); + Ok(()) +} + +/// How often the reconciliation loop re-diffs every authorized owner's +/// current [`RosterState`] against `supervised`. +/// +/// This only reads state this daemon already holds in memory (the roster +/// tap keeps `RosterState` current via its own live subscription, not +/// polling) and checks each pending agent's [`CredentialState`] — both +/// cheap — so a short interval costs nothing but stays far from a busy +/// loop. +const RECONCILE_INTERVAL: Duration = Duration::from_secs(5); + +/// Cancel and drop one roster-sourced agent's entry — shared by every +/// tear-down path in [`reconcile_roster`] (no longer listed, version +/// changed, revoked while running, or a spawn/parse failure) so each stays +/// a one-line call rather than a repeated three-statement block. +fn tear_down_roster_agent( + supervised: &mut HashMap, + watch_list: &WatchList, + pubkey: &str, +) { + if let Some(agent) = supervised.remove(pubkey) { + agent.cancel.cancel(); + } + watch_list.remove(pubkey); +} + +/// One reconciliation pass, in order: +/// +/// 1. Tear down [`AgentSource::Roster`] agents no longer listed by any +/// authorized owner's roster. +/// 2. Tear down [`AgentSource::Roster`] agents still listed but whose +/// roster-claimed `credential_version` has moved since the last pass — +/// pending or already running, so a rotation cancels the old identity +/// immediately rather than only affecting a not-yet-started one. +/// 3. Adopt newly-listed agents (spawn a credential tap for each), bounded +/// by `max_agents` counting every currently supervised pubkey +/// (config, pending, and running together) — see [`main`]'s own +/// `WAKER_MAX_AGENTS` doc. +/// 4. For every [`AgentSource::Roster`] agent (pending or running), +/// re-check its credential tap's live state against the version it is +/// expected to be at: promote a pending agent whose delivery now +/// matches exactly, tear down a running agent whose credential was +/// revoked (the tap's state going from `Some` to `None`). +#[allow(clippy::too_many_arguments)] +fn reconcile_roster( + relay_url: &str, + state_dir: &Path, + waker_keys: &Keys, + authorized_owners: &[String], + max_agents: usize, + roster_state: &RosterState, + supervised: &mut HashMap, + watch_list: &WatchList, + cancel: &CancellationToken, + tasks: &mut JoinSet, +) { + let rosters: Vec<(String, Vec)> = authorized_owners + .iter() + .filter_map(|owner| { + roster_state + .current(owner) + .map(|body| (owner.clone(), body.entries.clone())) + }) + .collect(); + + let existing_sources: HashMap = supervised + .iter() + .map(|(pubkey, agent)| (pubkey.clone(), agent.source)) + .collect(); + let (desired, conflicts) = compute_desired_roster_agents(&rosters, &existing_sources); + for pubkey in conflicts { + tracing::warn!( + agent = %pubkey, + "buzz-waker: more than one authorized owner's roster claims this agent; \ + the first one seen this pass wins, this is almost certainly a \ + misconfiguration" + ); + } + let desired_versions: HashMap<&str, u64> = desired + .iter() + .map(|d| (d.pubkey.as_str(), d.credential_version)) + .collect(); + + // Pass 1: no longer listed anywhere. + let no_longer_listed: Vec = supervised + .iter() + .filter(|(pubkey, agent)| { + agent.source == AgentSource::Roster && !desired_versions.contains_key(pubkey.as_str()) + }) + .map(|(pubkey, _)| pubkey.clone()) + .collect(); + for pubkey in no_longer_listed { + tear_down_roster_agent(supervised, watch_list, &pubkey); + tracing::info!( + agent = %pubkey, + "buzz-waker: no authorized owner's roster lists this agent anymore; \ + cancelling its watch tasks" + ); + } + + // Pass 2: still listed, but the roster's own claimed version moved — + // pending or already running, tear down either way so re-adoption + // (pass 3, same tick) starts fresh under the new version rather than + // leaving the old identity running or a stale bootstrap in place. + let version_changed: Vec = supervised + .iter() + .filter(|(pubkey, agent)| { + agent.source == AgentSource::Roster + && desired_versions + .get(pubkey.as_str()) + .is_some_and(|&v| Some(v) != agent.expected_credential_version) + }) + .map(|(pubkey, _)| pubkey.clone()) + .collect(); + for pubkey in version_changed { + tear_down_roster_agent(supervised, watch_list, &pubkey); + tracing::info!( + agent = %pubkey, + "buzz-waker: roster's credential_version for this agent changed; \ + cancelling and re-bootstrapping from the new version" + ); + } + + // Pass 3: adopt anything not currently supervised (newly listed, or + // just torn down above for a version change), bounded by max_agents. + for desired_agent in &desired { + if supervised.contains_key(&desired_agent.pubkey) { + continue; + } + if supervised.len() >= max_agents { + tracing::warn!( + agent = %desired_agent.pubkey, + max_agents, + "buzz-waker: refusing to adopt this roster-discovered agent; \ + WAKER_MAX_AGENTS reached (refuse, not evict — an existing agent is never \ + cancelled to make room)" + ); + continue; + } + let agent_cancel = cancel.child_token(); + let agent_dir = state_dir.join(&desired_agent.pubkey); + if let Err(error) = std::fs::create_dir_all(&agent_dir) { + tracing::error!( + agent = %desired_agent.pubkey, + %error, + "buzz-waker: could not create state dir for a roster-discovered agent; skipping this pass" + ); + continue; + } + let credential_floor_path = agent_dir.join("credential_floor.json"); + let mut credential_floor_store = match open_pinned_floor_store( + &credential_floor_path, + &desired_agent.owner_pubkey, + ) { + Ok(store) => store, + Err(error) => { + tracing::error!( + agent = %desired_agent.pubkey, + %error, + "buzz-waker: could not open this agent's credential floor; skipping this pass" + ); + continue; + } + }; + + // Known the moment this daemon adopts the pubkey, ahead of the + // credential that proves this daemon can actually run it — see + // `crate::watch_list`'s own doc for why that ordering is the safe + // one for `confirm_author_not_known_agent`. + watch_list.insert(&desired_agent.pubkey); + + let credential_state = Arc::new(CredentialState::new()); + { + let relay_url = relay_url.to_string(); + let waker_keys = waker_keys.clone(); + let owner_pubkey = desired_agent.owner_pubkey.clone(); + let agent_pubkey = desired_agent.pubkey.clone(); + let credential_state = Arc::clone(&credential_state); + let tap_cancel = agent_cancel.clone(); + let pubkey_for_exit = desired_agent.pubkey.clone(); + tasks.spawn(async move { + run_credential_tap( + &relay_url, + &waker_keys, + None, + &owner_pubkey, + &agent_pubkey, + &mut credential_floor_store, + &credential_state, + &tap_cancel, + ) + .await; + TaskExit::CredentialTap { + pubkey: pubkey_for_exit, + generation: tap_cancel, + } + }); + } + + tracing::info!( + agent = %desired_agent.pubkey, + owner = %desired_agent.owner_pubkey, + credential_version = desired_agent.credential_version, + "buzz-waker: roster lists a new agent; waiting for its credential" + ); + supervised.insert( + desired_agent.pubkey.clone(), + SupervisedAgent { + cancel: agent_cancel, + source: AgentSource::Roster, + owner_pubkey: desired_agent.owner_pubkey.clone(), + credential_state: Some(credential_state), + expected_credential_version: Some(desired_agent.credential_version), + running: false, + }, + ); + } + + // Pass 4: re-check every roster-sourced agent's credential tap against + // the version it is expected to be at — pending or already running. + let pubkeys_to_check: Vec = supervised + .iter() + .filter(|(_, agent)| agent.credential_state.is_some()) + .map(|(pubkey, _)| pubkey.clone()) + .collect(); + for pubkey in pubkeys_to_check { + let Some(agent) = supervised.get(&pubkey) else { + continue; + }; + let Some(credential_state) = &agent.credential_state else { + continue; + }; + let expected_version = agent.expected_credential_version; + let running = agent.running; + let owner_pubkey = agent.owner_pubkey.clone(); + let agent_cancel = agent.cancel.clone(); + + match credential_state.current() { + None => { + if running { + tear_down_roster_agent(supervised, watch_list, &pubkey); + tracing::error!( + agent = %pubkey, + "buzz-waker: this agent's credential was revoked while running; \ + cancelling its watch tasks" + ); + } + // Not yet running: still waiting for the first delivery, + // nothing to do this tick. + } + Some(body) if Some(body.credential_version) == expected_version => { + if running { + continue; // already running this exact version + } + let keys = match Keys::parse(&body.nsec) { + Ok(keys) => keys, + Err(error) => { + tracing::error!( + agent = %pubkey, + %error, + "buzz-waker: delivered credential's nsec does not parse; tearing down this agent" + ); + tear_down_roster_agent(supervised, watch_list, &pubkey); + continue; + } + }; + let auth_tag = match body.auth_tag.clone().map(Tag::parse).transpose() { + Ok(auth_tag) => auth_tag, + Err(error) => { + tracing::error!( + agent = %pubkey, + %error, + "buzz-waker: delivered credential's auth_tag does not parse; tearing down this agent" + ); + tear_down_roster_agent(supervised, watch_list, &pubkey); + continue; + } + }; + let provider_env = body + .provider_credential + .as_ref() + .map(|credential| Arc::new(credential.to_env())); + + match spawn_agent_watch( + relay_url, + state_dir, + &keys, + auth_tag.as_ref(), + &owner_pubkey, + provider_env, + watch_list, + agent_cancel, + tasks, + ) { + Ok(()) => { + if let Some(agent) = supervised.get_mut(&pubkey) { + agent.running = true; + } + tracing::info!(agent = %pubkey, "buzz-waker: roster-discovered agent's credential arrived; now watching it"); + } + Err(error) => { + tracing::error!( + agent = %pubkey, + %error, + "buzz-waker: could not start watching a roster-discovered agent; tearing it down" + ); + tear_down_roster_agent(supervised, watch_list, &pubkey); + } + } + } + Some(_stale_or_mismatched) => { + // A delivery that doesn't match the roster's current + // expectation — a lagging reconnect replay, or the + // credential simply hasn't caught up to a just-bumped + // roster yet. Left in place, not acted on; either the tap + // eventually delivers the right version (self-heals) or + // the roster catches up (pass 2 next tick). + } + } + } +} + #[tokio::main] async fn main() -> anyhow::Result<()> { // Install the ring CryptoProvider before any wss:// feed opens. Both ring @@ -187,6 +929,44 @@ async fn main() -> anyhow::Result<()> { let relay_url = env_var("WAKER_RELAY_URL")?; let state_dir = PathBuf::from(env_var("WAKER_STATE_DIR")?); let agents_config_path = env_var("WAKER_AGENTS_CONFIG_PATH")?; + let authorized_owners = + parse_authorized_owners(&std::env::var("WAKER_OWNER_PUBKEYS").unwrap_or_default())?; + let waker_keys = if authorized_owners.is_empty() { + None + } else { + let nsec = env_var("WAKER_IDENTITY_NSEC").map_err(|_| { + anyhow::anyhow!( + "WAKER_OWNER_PUBKEYS is set but WAKER_IDENTITY_NSEC is not; the roster and \ + credential taps have no identity to connect as" + ) + })?; + Some(Keys::parse(&nsec).map_err(|e| anyhow::anyhow!("invalid WAKER_IDENTITY_NSEC: {e}"))?) + }; + // Required alongside WAKER_IDENTITY_NSEC, same reasoning: dynamic + // enrolment needs an explicit ceiling on how many agents an authorized + // owner's roster can cause this daemon to run — refuse-not-evict, per + // the approved multi-tenant design (`PLANS/BUZZ_WAKER_DESIGN.md` §12's + // multi-tenant extension). Counts every currently supervised pubkey — + // config, pending, and running together — not roster-sourced agents + // alone, so a daemon cannot be pushed past the operator's own stated + // ceiling regardless of source. + let max_agents: usize = if authorized_owners.is_empty() { + 0 + } else { + let raw = env_var("WAKER_MAX_AGENTS").map_err(|_| { + anyhow::anyhow!( + "WAKER_OWNER_PUBKEYS is set but WAKER_MAX_AGENTS is not; dynamic enrolment \ + needs an explicit total-agent ceiling" + ) + })?; + let parsed: usize = raw + .parse() + .map_err(|e| anyhow::anyhow!("invalid WAKER_MAX_AGENTS {raw:?}: {e}"))?; + if parsed == 0 { + anyhow::bail!("WAKER_MAX_AGENTS must be at least 1, got 0"); + } + parsed + }; let agent_configs = load_agents(&agents_config_path)?; @@ -209,15 +989,11 @@ async fn main() -> anyhow::Result<()> { keys_by_agent.push((keys, auth_tag, owner_pubkey)); } - // This daemon's whole known-agent baseline — see `effects`'s module doc - // on why this is the accepted simplification for - // `confirm_author_not_known_agent` rather than the full managed-agent - // roster. - let watch_list: Arc<[String]> = keys_by_agent - .iter() - .map(|(keys, _, _)| normalize_pubkey(&keys.public_key().to_hex())) - .collect::>() - .into(); + ensure_static_agent_count_fits_cap( + keys_by_agent.len(), + max_agents, + !authorized_owners.is_empty(), + )?; std::fs::create_dir_all(&state_dir).map_err(|e| { anyhow::anyhow!( @@ -227,114 +1003,161 @@ async fn main() -> anyhow::Result<()> { })?; let cancel = CancellationToken::new(); - let mut tasks: JoinSet<(String, &'static str)> = JoinSet::new(); + let mut tasks: JoinSet = JoinSet::new(); + let watch_list = WatchList::new(); + let mut supervised: HashMap = HashMap::new(); for (keys, auth_tag, owner_pubkey) in keys_by_agent { let pubkey = normalize_pubkey(&keys.public_key().to_hex()); - let agent_dir = state_dir.join(&pubkey); - std::fs::create_dir_all(&agent_dir).map_err(|e| { - anyhow::anyhow!("could not create state dir {}: {e}", agent_dir.display()) - })?; - let cursor_path = agent_dir.join("cursor.json"); - let floor_path = agent_dir.join("floor.json"); - - let presence_state = Arc::new(PresenceState::new()); - let bundle_state = Arc::new(BundleState::new()); - - // Open-or-enroll, matching `CursorStore::open_or_start`'s idempotent - // shape: a fresh state dir enrolls fresh, an existing one re-opens - // its durable floors (G2) rather than resetting them. - let mut floor_store = match FloorStore::open(&floor_path) { - Ok(store) => store, - Err(buzz_waker::floors::FloorError::NotEnrolled { .. }) => { - FloorStore::enroll(&floor_path, &owner_pubkey).map_err(|e| { - anyhow::anyhow!("could not enroll floor store for {pubkey}: {e}") - })? - } - Err(e) => { - anyhow::bail!("could not open floor store for {pubkey}: {e}") - } - }; - - let pinned_owner = normalize_pubkey(&floor_store.snapshot().owner_pubkey); - ensure_owner_pin_matches(&pubkey, &pinned_owner, &owner_pubkey)?; - - tracing::info!(agent = %pubkey, owner = %owner_pubkey, "buzz-waker: watching agent"); + let agent_cancel = cancel.child_token(); + spawn_agent_watch( + &relay_url, + &state_dir, + &keys, + auth_tag.as_ref(), + &owner_pubkey, + None, + &watch_list, + agent_cancel.clone(), + &mut tasks, + )?; + supervised.insert( + pubkey, + SupervisedAgent { + cancel: agent_cancel, + source: AgentSource::Config, + owner_pubkey, + credential_state: None, + expected_credential_version: None, + running: true, + }, + ); + } + let roster_state = if let Some(waker_keys) = &waker_keys { + let roster_state = Arc::new(RosterState::new()); + let roster_floor_dir = state_dir.join("roster-floors"); { let relay_url = relay_url.clone(); - let keys = keys.clone(); - let auth_tag = auth_tag.clone(); - let presence_state = Arc::clone(&presence_state); + let waker_keys = waker_keys.clone(); + let authorized_owners = authorized_owners.clone(); + let roster_state = Arc::clone(&roster_state); let cancel = cancel.clone(); - let pubkey = pubkey.clone(); tasks.spawn(async move { - run_presence_tap( + run_roster_tap( &relay_url, - &keys, - auth_tag.as_ref(), - &presence_state, + &waker_keys, + None, + &authorized_owners, + &roster_floor_dir, + &roster_state, &cancel, ) .await; - (pubkey, "presence_tap") + TaskExit::Component("roster_tap") }); } + Some(roster_state) + } else { + None + }; - { - let relay_url = relay_url.clone(); - let keys = keys.clone(); - let auth_tag = auth_tag.clone(); - let owner_pubkey = owner_pubkey.clone(); - let bundle_state = Arc::clone(&bundle_state); - let cancel = cancel.clone(); - let pubkey = pubkey.clone(); - tasks.spawn(async move { - run_bundle_tap( + // A watch task can finish on its own, outside the shutdown path: a + // corrupt cursor makes `run_wake_loop` return immediately, and any task + // can panic. This loop keeps running for the daemon's whole life (not + // just a one-shot race at startup) because reconciliation and per-agent + // teardown are now ongoing, ordinary events, not only something that + // happens once at shutdown. + let mut ticker = tokio::time::interval(RECONCILE_INTERVAL); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let fatal: Option = loop { + tokio::select! { + () = shutdown_signal() => { + tracing::info!("buzz-waker: shutdown signal received; stopping"); + break None; + } + _ = ticker.tick(), if roster_state.is_some() => { + let (Some(waker_keys), Some(roster_state)) = (&waker_keys, &roster_state) else { + unreachable!("ticker only fires when roster_state is Some, which only happens alongside waker_keys"); + }; + reconcile_roster( &relay_url, - &keys, - auth_tag.as_ref(), - &owner_pubkey, - &mut floor_store, - &bundle_state, + &state_dir, + waker_keys, + &authorized_owners, + max_agents, + roster_state, + &mut supervised, + &watch_list, &cancel, - ) - .await; - (pubkey, "bundle_tap") - }); - } - - { - let config = WakeLoopConfig { - relay_url: relay_url.clone(), - keys, - auth_tag, - cursor_path, - presence_state, - watch_list: Arc::clone(&watch_list), - bundle_state, - }; - let cancel = cancel.clone(); - let pubkey = pubkey.clone(); - tasks.spawn(async move { - run_wake_loop(config, cancel).await; - (pubkey, "wake_loop") - }); - } - } - - // A watch task can also finish on its own, outside the shutdown path: a - // corrupt cursor makes `run_wake_loop` return immediately, and either - // task can panic. If that happens before `cancel` fires, the daemon must - // not keep running with that agent silently unwatched and reporting - // healthy — race the first such completion against the shutdown signal - // and treat an early one as fatal for the whole process. - let early_exit = tokio::select! { - () = shutdown_signal() => { - tracing::info!("buzz-waker: shutdown signal received; stopping"); - None + &mut tasks, + ); + } + result = tasks.join_next(), if !tasks.is_empty() => { + let Some(result) = result else { continue }; + match result { + Err(join_error) => { + break Some(anyhow::anyhow!( + "buzz-waker: a watch task panicked: {join_error}" + )); + } + Ok(TaskExit::Component(name)) => { + if !cancel.is_cancelled() { + break Some(anyhow::anyhow!( + "buzz-waker: component {name} exited before shutdown was requested" + )); + } + } + Ok(exit) => { + let pubkey = match &exit { + TaskExit::Agent { pubkey, .. } | TaskExit::CredentialTap { pubkey, .. } => pubkey.clone(), + TaskExit::Component(_) => unreachable!("handled above"), + }; + let task_name: &'static str = match &exit { + TaskExit::Agent { task, .. } => task, + TaskExit::CredentialTap { .. } => "credential_tap", + TaskExit::Component(_) => unreachable!("handled above"), + }; + // Classify against the exact token this task was + // spawned under, not whatever `supervised` currently + // holds for this pubkey — a version change tears + // down and re-adopts the same pubkey within one + // `reconcile_roster` pass, so a predecessor + // generation's exit can land after `supervised` + // already holds the replacement's fresh, uncancelled + // token. See `TaskExit::Agent`'s own doc. + let was_cancelled = match &exit { + TaskExit::Agent { generation, .. } + | TaskExit::CredentialTap { generation, .. } => generation.is_cancelled(), + TaskExit::Component(_) => unreachable!("handled above"), + }; + let source = supervised.get(&pubkey).map(|agent| agent.source); + match classify_exit(source, was_cancelled) { + ExitDisposition::Expected => {} + ExitDisposition::FatalDaemon => { + break Some(anyhow::anyhow!( + "buzz-waker: {task_name} for agent {pubkey} exited before \ + shutdown was requested; that agent stopped being watched — \ + treating as fatal rather than running silently degraded" + )); + } + ExitDisposition::TearDownAgent => { + tracing::error!( + agent = %pubkey, + task = %task_name, + "buzz-waker: a roster-discovered agent's task exited unexpectedly; \ + tearing down this agent only, daemon continues" + ); + if let Some(agent) = supervised.remove(&pubkey) { + agent.cancel.cancel(); + } + watch_list.remove(&pubkey); + } + } + } + } + } } - result = tasks.join_next() => Some(result), }; cancel.cancel(); @@ -347,20 +1170,9 @@ async fn main() -> anyhow::Result<()> { tracing::info!("buzz-waker: shutdown complete"); - match early_exit { - Some(Some(Ok((pubkey, task)))) => { - anyhow::bail!( - "buzz-waker: {task} for agent {pubkey} exited before shutdown was requested; \ - that agent stopped being watched — treating as fatal rather than running \ - silently degraded" - ) - } - Some(Some(Err(error))) => { - anyhow::bail!( - "buzz-waker: a watch task panicked before shutdown was requested: {error}" - ) - } - Some(None) | None => Ok(()), + match fatal { + Some(error) => Err(error), + None => Ok(()), } } @@ -431,6 +1243,25 @@ mod tests { assert!(error.to_string().contains("disagreeing owners"), "{error}"); } + #[test] + fn a_static_baseline_within_the_cap_is_accepted() { + assert!(ensure_static_agent_count_fits_cap(10, 10, true).is_ok()); + assert!(ensure_static_agent_count_fits_cap(5, 10, true).is_ok()); + } + + #[test] + fn a_static_baseline_exceeding_the_cap_is_refused_when_dynamic_enrolment_is_enabled() { + let error = ensure_static_agent_count_fits_cap(20, 10, true).unwrap_err(); + let message = error.to_string(); + assert!(message.contains("WAKER_MAX_AGENTS=10"), "{message}"); + assert!(message.contains("20"), "{message}"); + } + + #[test] + fn a_static_baseline_exceeding_the_cap_is_ignored_when_dynamic_enrolment_is_disabled() { + assert!(ensure_static_agent_count_fits_cap(20, 0, false).is_ok()); + } + #[test] fn a_valid_agent_list_round_trips_through_load_agents() { let dir = tempfile::tempdir().expect("tempdir"); @@ -447,4 +1278,147 @@ mod tests { let agents = load_agents(path.to_str().expect("utf8 path")).expect("loads"); assert_eq!(agents.len(), 1); } + + fn entry(pubkey: &str) -> RosterEntry { + entry_at_version(pubkey, 1) + } + + fn entry_at_version(pubkey: &str, credential_version: u64) -> RosterEntry { + RosterEntry { + agent_pubkey: pubkey.to_string(), + credential_version, + } + } + + #[test] + fn a_config_sourced_pubkey_is_never_desired_via_roster() { + let pubkey = "a".repeat(64); + let owner = "b".repeat(64); + let mut existing = HashMap::new(); + existing.insert(normalize_pubkey(&pubkey), AgentSource::Config); + + let (desired, conflicts) = + compute_desired_roster_agents(&[(owner, vec![entry(&pubkey)])], &existing); + + assert!(desired.is_empty()); + assert!(conflicts.is_empty()); + } + + #[test] + fn a_roster_only_pubkey_is_desired_under_its_owner() { + let pubkey = "a".repeat(64); + let owner = "b".repeat(64); + + let (desired, conflicts) = compute_desired_roster_agents( + &[(owner.clone(), vec![entry(&pubkey)])], + &HashMap::new(), + ); + + assert_eq!(desired.len(), 1); + assert_eq!(desired[0].pubkey, normalize_pubkey(&pubkey)); + assert_eq!(desired[0].owner_pubkey, owner); + assert_eq!(desired[0].credential_version, 1); + assert!(conflicts.is_empty()); + } + + #[test] + fn the_roster_entrys_credential_version_is_carried_through() { + let pubkey = "a".repeat(64); + let owner = "b".repeat(64); + + let (desired, _) = compute_desired_roster_agents( + &[(owner, vec![entry_at_version(&pubkey, 7)])], + &HashMap::new(), + ); + + assert_eq!( + desired[0].credential_version, 7, + "the exact version reconcile_roster gates promotion on must survive the fold" + ); + } + + #[test] + fn an_already_roster_supervised_pubkey_is_still_desired_so_it_is_not_torn_down() { + let pubkey = "a".repeat(64); + let owner = "b".repeat(64); + let mut existing = HashMap::new(); + existing.insert(normalize_pubkey(&pubkey), AgentSource::Roster); + + let (desired, _) = + compute_desired_roster_agents(&[(owner, vec![entry(&pubkey)])], &existing); + + assert_eq!( + desired.len(), + 1, + "an existing roster-sourced agent still listed must remain desired" + ); + } + + #[test] + fn two_owners_claiming_the_same_pubkey_is_a_conflict_and_the_first_wins() { + let pubkey = "a".repeat(64); + let owner_a = "b".repeat(64); + let owner_b = "c".repeat(64); + + let (desired, conflicts) = compute_desired_roster_agents( + &[ + (owner_a.clone(), vec![entry(&pubkey)]), + (owner_b, vec![entry(&pubkey)]), + ], + &HashMap::new(), + ); + + assert_eq!(desired.len(), 1); + assert_eq!( + desired[0].owner_pubkey, owner_a, + "the first owner seen wins" + ); + assert_eq!(conflicts, vec![normalize_pubkey(&pubkey)]); + } + + #[test] + fn a_pubkey_missing_from_every_roster_is_not_desired() { + let (desired, conflicts) = compute_desired_roster_agents(&[], &HashMap::new()); + assert!(desired.is_empty()); + assert!(conflicts.is_empty()); + } + + #[test] + fn a_cancelled_exit_is_always_expected_regardless_of_source() { + assert_eq!( + classify_exit(Some(AgentSource::Config), true), + ExitDisposition::Expected + ); + assert_eq!( + classify_exit(Some(AgentSource::Roster), true), + ExitDisposition::Expected + ); + assert_eq!(classify_exit(None, true), ExitDisposition::Expected); + } + + #[test] + fn an_unsolicited_config_exit_is_fatal() { + assert_eq!( + classify_exit(Some(AgentSource::Config), false), + ExitDisposition::FatalDaemon + ); + } + + #[test] + fn an_unsolicited_exit_of_an_untracked_pubkey_is_fatal() { + // `None` means the exiting task's pubkey has no entry in + // `supervised` at all — never true for a real roster-sourced agent + // (removal always cancels first), so this can only be an + // accounting bug or a statically configured agent whose entry was + // somehow lost. Treated the same as `Config`: fatal. + assert_eq!(classify_exit(None, false), ExitDisposition::FatalDaemon); + } + + #[test] + fn an_unsolicited_roster_exit_tears_down_only_that_agent() { + assert_eq!( + classify_exit(Some(AgentSource::Roster), false), + ExitDisposition::TearDownAgent + ); + } } diff --git a/crates/buzz-waker/src/wake_loop.rs b/crates/buzz-waker/src/wake_loop.rs index 3744db3f8a5..83d74713c24 100644 --- a/crates/buzz-waker/src/wake_loop.rs +++ b/crates/buzz-waker/src/wake_loop.rs @@ -54,6 +54,7 @@ use crate::feed::{ }; use crate::presence_feed::PresenceState; use crate::relay_feed::RelayFeed; +use crate::watch_list::WatchList; /// Configuration for one agent's wake loop. #[derive(Clone)] @@ -71,16 +72,21 @@ pub struct WakeLoopConfig { pub cursor_path: PathBuf, /// The presence tap shared with every wake attempt for this agent. pub presence_state: Arc, - /// This daemon's full watch list, normalized — see `effects`'s module - /// doc for why this is the accepted `confirm_author_not_known_agent` - /// baseline. - pub watch_list: Arc<[String]>, + /// This daemon's live watch list — see `effects`'s module doc for why + /// this is the accepted `confirm_author_not_known_agent` baseline, and + /// `crate::watch_list`'s own doc for why it is read live rather than + /// snapshotted. + pub watch_list: WatchList, /// The live cache [`crate::bundle_feed::run_bundle_tap`] writes this /// agent's admitted bundle into. Read fresh at the moment each attempt is /// spawned (never captured once at loop-construction time) so a reissue /// admitted mid-run takes effect on the very next wake, with no daemon /// restart required. pub bundle_state: Arc, + /// This agent's own provider credential's environment overlay, if it has + /// one — see `effects::RealWakeEffects`'s own doc for what this is and + /// why it's `None` for a statically configured agent. + pub provider_env: Option>>, } fn now_secs() -> u64 { @@ -339,8 +345,9 @@ pub async fn run_wake_loop(config: WakeLoopConfig, cancel: CancellationToken) { agent_pubkey.clone(), Arc::clone(&attempt_state), Arc::clone(&config.presence_state), - Arc::clone(&config.watch_list), + config.watch_list.clone(), config.bundle_state.current(), + config.provider_env.clone(), cancel.clone(), ); } @@ -592,8 +599,9 @@ fn spawn_attempt( agent_pubkey: String, attempt_state: Arc, presence_state: Arc, - watch_list: Arc<[String]>, + watch_list: WatchList, bundle: Option>, + provider_env: Option>>, cancel: CancellationToken, ) { attempts.spawn(async move { @@ -607,6 +615,7 @@ fn spawn_attempt( &event.author, event.created_at, bundle, + provider_env, cancel, move || { tracing::info!( diff --git a/crates/buzz-waker/src/watch_list.rs b/crates/buzz-waker/src/watch_list.rs new file mode 100644 index 00000000000..4d07d81e310 --- /dev/null +++ b/crates/buzz-waker/src/watch_list.rs @@ -0,0 +1,126 @@ +//! This daemon's own live known-agent set — `crate::effects`'s baseline for +//! `confirm_author_not_known_agent`, `PLANS/BUZZ_WAKER_DESIGN.md` §12 build +//! order step 3. +//! +//! Before the dynamic supervisor, this was a frozen `Arc<[String]>` snapshot +//! taken once at startup from `WAKER_AGENTS_CONFIG_PATH` — safe only because +//! the set never changed for the life of the process. Once agents can be +//! added or removed at runtime (a roster reissue), a frozen snapshot goes +//! stale: an agent added after startup would not be "known" to +//! `confirm_author_not_known_agent`, and a mention it authored could then +//! wake another agent — exactly the agent-to-agent wake loop that guard +//! exists to prevent (`crate::decide::select_wake_candidates`'s own doc). +//! [`WatchList`] fixes that by being read live rather than snapshotted: +//! every clone shares the same underlying set, so an `insert`/`remove` from +//! the supervisor is visible to every in-flight wake attempt's +//! `confirm_author_not_known_agent` check immediately. + +use std::collections::HashSet; +use std::sync::{Arc, Mutex, PoisonError}; + +use crate::decide::normalize_pubkey; + +/// A cheaply-cloneable, live-shared set of normalized agent pubkeys. +#[derive(Debug, Clone, Default)] +pub struct WatchList { + inner: Arc>>, +} + +impl WatchList { + /// An empty watch list. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Recover from a poisoned lock rather than propagating it — a panic in + /// one reader must not permanently blind every future watch-list check. + fn lock(&self) -> std::sync::MutexGuard<'_, HashSet> { + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Add `pubkey` to the set. + pub fn insert(&self, pubkey: &str) { + self.lock().insert(normalize_pubkey(pubkey)); + } + + /// Remove `pubkey` from the set. A no-op if it was never present. + pub fn remove(&self, pubkey: &str) { + self.lock().remove(&normalize_pubkey(pubkey)); + } + + /// Whether `pubkey` is currently in the set, comparison + /// case/whitespace-insensitive (both sides normalized). + #[must_use] + pub fn contains(&self, pubkey: &str) -> bool { + self.lock().contains(&normalize_pubkey(pubkey)) + } +} + +impl From> for WatchList { + /// Build a watch list already populated with `pubkeys` — the shape + /// existing tests already construct a frozen `Arc<[String]>` with + /// (`Arc::from(vec![...])`), so callers only need to swap the type, not + /// the construction pattern. + fn from(pubkeys: Vec) -> Self { + let set = pubkeys.iter().map(|p| normalize_pubkey(p)).collect(); + Self { + inner: Arc::new(Mutex::new(set)), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_fresh_list_is_empty() { + let list = WatchList::new(); + assert!(!list.contains(&"a".repeat(64))); + } + + #[test] + fn an_inserted_pubkey_is_found() { + let list = WatchList::new(); + let pubkey = "a".repeat(64); + list.insert(&pubkey); + assert!(list.contains(&pubkey)); + } + + #[test] + fn a_removed_pubkey_is_no_longer_found() { + let list = WatchList::new(); + let pubkey = "a".repeat(64); + list.insert(&pubkey); + list.remove(&pubkey); + assert!(!list.contains(&pubkey)); + } + + #[test] + fn membership_is_case_and_whitespace_insensitive() { + let list = WatchList::new(); + list.insert(" AA "); + assert!(list.contains("aa")); + } + + #[test] + fn clones_share_the_same_underlying_set() { + let list = WatchList::new(); + let clone = list.clone(); + let pubkey = "a".repeat(64); + + clone.insert(&pubkey); + + assert!( + list.contains(&pubkey), + "a clone must be a shared handle, not an independent copy" + ); + } + + #[test] + fn from_vec_normalizes_every_entry() { + let list = WatchList::from(vec![" AA ".to_string()]); + assert!(list.contains("aa")); + } +} diff --git a/desktop/src-tauri/src/managed_agents/backend.rs b/desktop/src-tauri/src/managed_agents/backend.rs index c4a533c4a4c..45d5ff7bafd 100644 --- a/desktop/src-tauri/src/managed_agents/backend.rs +++ b/desktop/src-tauri/src/managed_agents/backend.rs @@ -23,6 +23,7 @@ pub fn invoke_provider( request, timeout, super::default_agent_workdir().as_deref(), + None, ) } @@ -41,6 +42,7 @@ pub fn provider_deploy( agent, provider_config, super::default_agent_workdir().as_deref(), + None, ) }