From bca433e828ae4b3516d7a33953814f329f4977da Mon Sep 17 00:00:00 2001 From: Tee8z Date: Tue, 3 Nov 2026 19:54:21 -0500 Subject: [PATCH] feat(agents): admit shared turns through the running harness Signed-off-by: Tee8z --- Cargo.lock | 2 + crates/buzz-acp/Cargo.toml | 2 + crates/buzz-acp/src/acp.rs | 3 + crates/buzz-acp/src/config.rs | 43 ++- crates/buzz-acp/src/lib.rs | 107 +++++- crates/buzz-acp/src/pool.rs | 36 ++ crates/buzz-acp/src/runtime.rs | 9 +- crates/buzz-acp/src/sharing.rs | 276 ++++++++++++++ crates/buzz-acp/src/sharing/notices.rs | 136 +++++++ crates/buzz-acp/src/sharing/tests.rs | 357 ++++++++++++++++++ crates/buzz-cli/src/commands/agents.rs | 43 +++ .../src/commands/agents_sharing_tests.rs | 81 ++++ crates/buzz-cli/src/lib.rs | 17 +- crates/buzz-sandbox-reaper/src/manager.rs | 65 +++- .../src/manager_tests/shared.rs | 39 ++ 15 files changed, 1182 insertions(+), 34 deletions(-) create mode 100644 crates/buzz-acp/src/sharing.rs create mode 100644 crates/buzz-acp/src/sharing/notices.rs create mode 100644 crates/buzz-acp/src/sharing/tests.rs diff --git a/Cargo.lock b/Cargo.lock index 21c90681f2a..e00dadeebb7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1449,6 +1449,8 @@ dependencies = [ "buzz-core", "buzz-persona", "buzz-sdk", + "buzz-session-control", + "buzz-ws-client", "chrono", "clap", "evalexpr", diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index 244bae144ed..dbb5835968d 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -19,6 +19,8 @@ path = "src/main.rs" # Internal buzz-core = { workspace = true } buzz-sdk = { workspace = true } +buzz-session-control = { path = "../buzz-session-control" } +buzz-ws-client = { workspace = true } buzz-persona = { path = "../buzz-persona" } # Git multicall helpers and ephemeral key ownership diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 1d55af6ef36..1ffdd4e5766 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -113,6 +113,9 @@ pub enum AcpError { #[error("Protocol error: {0}")] Protocol(String), + #[error("Shared task permission: {0}")] + TaskPermission(String), + #[error("Agent reported error (code {code}): {message}")] AgentError { code: i64, message: String }, diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 1a1c92b6deb..54e16fb373b 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -12,7 +12,7 @@ use buzz_core::kind::{ }; use clap::Parser; use clap::ValueEnum; -use nostr::Keys; +use nostr::{Keys, PublicKey}; use thiserror::Error; use url::Url; use uuid::Uuid; @@ -594,6 +594,10 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_BOUND_SESSION", value_parser = parse_bound_session)] pub bound_session: Option, + /// Session manager that authorizes shared channel turns. Requires a bound thread. + #[arg(long, env = "BUZZ_ACP_SHARING_MANAGER")] + pub sharing_manager: Option, + /// Save and verify a private workspace checkpoint after draining the agents. #[arg(long, env = "BUZZ_ACP_CHECKPOINT_ON_STOP", default_value_t = false)] pub checkpoint_on_stop: bool, @@ -716,6 +720,8 @@ pub struct Config { /// Whether to emit bounded JSONL activity for platform collection. pub activity_log: bool, pub bound_session: Option, + /// Manager for fresh shared-task admission; absent preserves owner-only input. + pub sharing_manager: Option, /// Whether to run `buzz-agent-checkpoint save` after draining on stop and /// fail shutdown when it does not produce a verified receipt. pub checkpoint_on_stop: bool, @@ -1042,6 +1048,16 @@ impl Config { /// `propagate_legacy_env_vars()` before the tokio runtime starts (in the /// sync `fn main()` wrapper) — see Rust 2024 edition safety. pub fn from_args(mut args: CliArgs) -> Result { + if args.sharing_manager.is_some() + && !matches!( + args.bound_session.as_ref(), + Some(BoundSession::Thread { .. }) + ) + { + return Err(ConfigError::ConfigFile( + "shared tasks require a bound channel thread".into(), + )); + } if let Some(binding) = &args.bound_session { let direct_message = matches!(binding, BoundSession::DirectMessage { .. }); let subscription = if direct_message { @@ -1333,7 +1349,9 @@ impl Config { max_turns_per_session: args.max_turns_per_session, presence_enabled: !args.no_presence, typing_enabled: !args.no_typing, - memory_enabled: args.memory && !args.no_memory, + // A later sharing grant must never expose personal core memory + // already injected into an owner-started channel session. + memory_enabled: args.memory && !args.no_memory && args.sharing_manager.is_none(), model, effort_level: args.effort_level, session_title: args @@ -1349,6 +1367,7 @@ impl Config { relay_observer: args.relay_observer, activity_log: args.activity_log, bound_session: args.bound_session, + sharing_manager: args.sharing_manager, checkpoint_on_stop: args.checkpoint_on_stop, checkpoint_after_turn_secs: args.checkpoint_after_turn_secs, exit_after_inactivity_secs: args.exit_after_inactivity, @@ -1801,6 +1820,13 @@ mod tests { .unwrap() }; assert!(Config::from_args(args()).is_ok()); + let mut private = args(); + private.memory = true; + assert!(Config::from_args(private).unwrap().memory_enabled); + let mut shared = args(); + shared.sharing_manager = Some(Keys::generate().public_key()); + let error = Config::from_args(shared).err().unwrap().to_string(); + assert!(error.contains("shared tasks require a bound channel thread")); let mut mentions = args(); mentions.subscribe = SubscribeMode::Mentions; assert!(Config::from_args(mentions).is_err()); @@ -1899,6 +1925,18 @@ mod tests { "0", ]; assert!(Config::from_args(CliArgs::try_parse_from(base).unwrap()).is_ok()); + let mut shared = CliArgs::try_parse_from(base).unwrap(); + shared.sharing_manager = Some(Keys::generate().public_key()); + let configured = Config::from_args(shared).unwrap(); + assert!( + !configured.memory_enabled, + "shared channels must not preload personal core memory" + ); + assert_eq!(configured.respond_to, RespondTo::Owner); + let mut unbound = CliArgs::try_parse_from(base).unwrap(); + unbound.sharing_manager = Some(Keys::generate().public_key()); + unbound.bound_session = None; + assert!(Config::from_args(unbound).is_err()); for (variation, flag) in [ (0, "--respond-to owner"), (1, "--no-mention-filter"), @@ -1969,6 +2007,7 @@ mod tests { relay_observer: false, activity_log: false, bound_session: None, + sharing_manager: None, checkpoint_on_stop: false, checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index fc19a869e83..fabcbe0803c 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -22,6 +22,8 @@ mod recovery_wake; mod relay; mod run_task; mod runtime; +/// Shared channel task admission through the existing encrypted control plane. +pub mod sharing; use runtime::{AgentRuntime, PoolStartup, SessionMode}; mod scope; mod setup_mode; @@ -622,6 +624,7 @@ mod inbound_author_gate { relay_self: Option, // None means no authoritative NIP-11 result yet, including at startup. refreshed_generation: Option, + sharing: Option>, } pub(crate) fn refresh_needed(refreshed_generation: Option, event_generation: u64) -> bool { @@ -640,6 +643,7 @@ mod inbound_author_gate { agent_pubkey_hex: agent_pubkey_hex.to_string(), relay_self, refreshed_generation: completed.then_some(0), + sharing: None, } } @@ -727,6 +731,73 @@ mod inbound_author_gate { } } + pub(crate) fn enable_sharing( + &mut self, + sharing: Option>, + ) { + self.sharing = sharing; + } + + pub(crate) async fn authorize_shared_listener_event( + &mut self, + buzz_event: relay::BuzzEvent, + respond_to: &RespondTo, + allowlist: &HashSet, + owner_cache: &OwnerCache, + channel_info: &pool::ChannelInfoResolver, + rest_client: &relay::RestClient, + ) -> Result, crate::sharing::Error> { + if self.sharing.is_none() { + return Ok(self + .authorize_listener_event( + buzz_event, + respond_to, + allowlist, + owner_cache, + channel_info, + rest_client, + ) + .await); + } + let decision = self + .evaluate_listener_event( + &buzz_event, + respond_to, + allowlist, + owner_cache, + channel_info, + rest_client, + ) + .await; + if decision.allowed { + return Ok(Some(AuthorizedListenerEvent { + buzz_event, + effective_author: decision.effective_author, + })); + } + if matches!(respond_to, RespondTo::Owner) && !decision.is_dm { + if let Some(sharing) = &self.sharing { + if sharing.accepts(&buzz_event.event) { + let refusal = match sharing.authorize(&buzz_event.event).await { + Ok(None) => { + return Ok(Some(AuthorizedListenerEvent { + buzz_event, + effective_author: decision.effective_author, + })) + } + Ok(Some(reason)) => reason, + Err(error) => error.to_string(), + }; + // Persist a visible refusal before handling a transient + // outage; never abandon the task or kill an in-flight turn. + sharing.refuse(&buzz_event.event, &refusal).await?; + return Ok(None); + } + } + } + Ok(None) + } + pub(crate) async fn authorize_listener_event( &mut self, buzz_event: relay::BuzzEvent, @@ -970,9 +1041,9 @@ async fn authorize_normal_listener_event( owner_cache: &OwnerCache, channel_info: &pool::ChannelInfoResolver, rest_client: &relay::RestClient, -) -> Option { +) -> std::result::Result, sharing::Error> { author_gate - .authorize_listener_event( + .authorize_shared_listener_event( buzz_event, respond_to, allowlist, @@ -3139,6 +3210,11 @@ async fn run_harness( channel_info_map, SessionMode::Conversation, )?); + author_gate_ctx.enable_sharing(ctx.sharing.clone()); + let _shared_notice_worker = ctx + .sharing + .as_ref() + .map(|client| client.clone().start_notices()); if !config.memory_enabled { tracing::info!( @@ -3838,7 +3914,7 @@ async fn run_harness( &ctx.channel_info, &ctx.rest_client, ) - .await + .await? else { continue; }; @@ -7937,18 +8013,17 @@ mod author_gate_tests { event: relay_signed_workflow_dispatch(relay_keys, workflow_owner, &agent), }; let authorized = match listener { - ListenerBoundary::Normal => { - authorize_normal_listener_event( - &mut gate, - event, - &respond_to, - &allowlist, - &owner_cache, - &channel_info, - &rest_client, - ) - .await - } + ListenerBoundary::Normal => authorize_normal_listener_event( + &mut gate, + event, + &respond_to, + &allowlist, + &owner_cache, + &channel_info, + &rest_client, + ) + .await + .unwrap(), ListenerBoundary::Setup => { setup_mode::authorize_setup_listener_event( &mut gate, @@ -10609,6 +10684,7 @@ mod build_mcp_servers_tests { relay_observer: false, activity_log: false, bound_session: None, + sharing_manager: None, checkpoint_on_stop: false, checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, @@ -11710,6 +11786,7 @@ mod error_outcome_emission_tests { relay_observer: false, activity_log: false, bound_session: None, + sharing_manager: None, checkpoint_on_stop: false, checkpoint_after_turn_secs: 0, exit_after_inactivity_secs: 0, diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index ece50ae541a..ec13fe652d7 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -934,6 +934,8 @@ pub struct PromptContext { /// `` section. On by default; disabled via /// `--no-memory` / `BUZZ_ACP_NO_MEMORY`. pub memory_enabled: bool, + /// Fresh authorization for bound shared-channel tasks. + pub sharing: Option>, /// Harness identity string for NIP-AM `harness` field. Derived from the /// configured `agent_command` at startup (e.g. `"goose"`, `"buzz-agent"`). pub harness_name: String, @@ -2510,6 +2512,39 @@ pub async fn run_prompt_task( .unwrap_or_default(); let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone()); + // Queueing may outlive a grant. Recheck before session creation, memory + // fetches, or any ACP prompt; the listener separately guards native steering. + if let (Some(sharing), Some(pending)) = (&ctx.sharing, &batch) { + let mut denied = None; + for item in &pending.events { + if Some(item.event.pubkey) == ctx.agent_owner_pubkey { + continue; + } + match sharing.authorize(&item.event).await { + Ok(None) => {} + Ok(Some(reason)) => { + denied = Some(reason); + break; + } + Err(error) => { + denied = Some(error.to_string()); + break; + } + } + } + if let Some(reason) = denied { + send_prompt_result( + &result_tx, + &turn_id, + agent, + source, + PromptOutcome::Error(AcpError::TaskPermission(reason)), + requeue_batch_if_queue(&ctx, batch), + ); + return; + } + } + // Resolve project authority exactly once, before any ACP session creation or // initial-message delivery. An indeterminate result is a local relay-state // outcome: fail closed and preserve the batch without poisoning the healthy @@ -10490,6 +10525,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'" agent_keys: agent_keys.clone(), agent_owner_pubkey: owner_pubkey, memory_enabled: false, + sharing: None, harness_name: "goose".to_string(), relay_url: "ws://127.0.0.1:3000".to_string(), } diff --git a/crates/buzz-acp/src/runtime.rs b/crates/buzz-acp/src/runtime.rs index e6745b63313..444a0b4e837 100644 --- a/crates/buzz-acp/src/runtime.rs +++ b/crates/buzz-acp/src/runtime.rs @@ -82,6 +82,8 @@ fn make_prompt_context( ) -> Result { let base_prompt_content = config.base_prompt_content.as_ref(); let cwd = current_working_directory()?; + let sharing = + crate::sharing::Client::from_config(config, &rest_client)?.map(std::sync::Arc::new); Ok(PromptContext { mcp_servers: build_mcp_servers(config), initial_message: config.initial_message.clone(), @@ -91,7 +93,11 @@ fn make_prompt_context( dedup_mode: config.dedup_mode, system_prompt: config.system_prompt.clone(), session_title: config.session_title.clone(), - team_instructions: config.team_instructions.clone(), + team_instructions: if sharing.is_some() { + Some(format!("{}\n\nTo delegate to another shared agent, use `buzz agents handoff --from --message ''`. Use the current incoming task's event ID. Delegate specific work or reviews; avoid acknowledgement loops. Sharing preserves the original requester and does not confer owner lifecycle rights.",config.team_instructions.as_deref().unwrap_or(""))) + } else { + config.team_instructions.clone() + }, base_prompt: if config.no_base_prompt { None } else { @@ -119,6 +125,7 @@ fn make_prompt_context( .as_deref() .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()), memory_enabled: config.memory_enabled, + sharing, harness_name: crate::config::normalize_agent_command_identity(&config.agent_command), relay_url: config.relay_url.clone(), }) diff --git a/crates/buzz-acp/src/sharing.rs b/crates/buzz-acp/src/sharing.rs new file mode 100644 index 00000000000..594336c775c --- /dev/null +++ b/crates/buzz-acp/src/sharing.rs @@ -0,0 +1,276 @@ +//! Fresh shared-task admission. The manager owns authorization and its durable +//! budget. A retry uses a new transport request ID, so its response cache cannot +//! preserve an approval after revocation; the task receipt remains idempotent. + +use buzz_session_control::{build_request, open_reply, Op, Request, Status, KIND}; +use buzz_ws_client::{NostrWsConnection, RelayMessage}; +use nostr::{Event, EventBuilder, Keys, Kind, PublicKey, Tag, Timestamp}; +use serde_json::json; +use std::time::Duration; + +mod notices; + +#[cfg(test)] +mod tests; + +/// Admission transport failure. It propagates to the task owner for recovery. +#[derive(Debug, thiserror::Error)] +#[error("shared task admission unavailable; retry or recover this task")] +pub struct Error; + +/// Agent credentials used only for encrypted requests and signed channel replies. +/// Deliberately has no Debug implementation. +#[derive(Clone)] +pub struct Client { + keys: Keys, + auth: String, + relay: String, + manager: PublicKey, + scope: buzz_sdk::sharing::task::Scope, + outbox: std::sync::Arc>, +} + +impl Client { + pub(crate) fn from_config( + config: &crate::config::Config, + rest: &crate::relay::RestClient, + ) -> Result, Error> { + let Some(manager) = config.sharing_manager else { + return Ok(None); + }; + let auth = rest.auth_tag_json.clone().ok_or(Error)?; + let crate::config::BoundSession::Thread { + channel_id, + thread_root, + } = config.bound_session.as_ref().ok_or(Error)? + else { + return Err(Error); + }; + let scope = buzz_sdk::sharing::task::Scope { + channel: *channel_id, + root: nostr::EventId::from_hex(thread_root).map_err(|_| Error)?, + }; + let directory = std::env::current_dir() + .map_err(|_| Error)? + .join(".buzz/shared-failures") + .join(config.keys.public_key().to_hex()) + .join(scope.root.to_hex()); + Ok(Some(Self { + keys: config.keys.clone(), + auth, + relay: config.relay_url.clone(), + manager, + scope, + outbox: std::sync::Arc::new(tokio::sync::Mutex::new(notices::Outbox::new(directory))), + })) + } + + pub(crate) fn accepts(&self, event: &Event) -> bool { + buzz_sdk::sharing::task::scope(event, self.keys.public_key(), Timestamp::now().as_secs()) + .ok() + == Some(self.scope) + } + + /// Return a permanent refusal separately from a retryable transport failure. + pub(crate) async fn authorize(&self, event: &Event) -> Result, Error> { + if !self.accepts(event) { + return Ok(Some("task is outside this bound conversation".into())); + } + tokio::time::timeout(Duration::from_secs(50), self.exchange(event)) + .await + .map_err(|_| Error)? + } + + async fn exchange(&self, trigger: &Event) -> Result, Error> { + let auth = buzz_sdk::nip_oa::parse_auth_tag(&self.auth).map_err(|_| Error)?; + let request = Request::new( + Op::Admit, + uuid::Uuid::new_v4().to_string(), + json!({"relay_url":self.relay,"trigger":trigger}), + ); + let event = build_request( + &self.keys, + &self.manager, + Some(&self.auth), + &request, + Timestamp::now(), + ) + .map_err(|_| Error)?; + let id = event.id; + let mut connection = + NostrWsConnection::connect_authenticated(&self.relay, &self.keys, Some(&auth)) + .await + .map_err(|_| Error)?; + connection.send_raw(&json!(["REQ","shared-reply",{"kinds":[KIND],"authors":[self.manager],"#p":[self.keys.public_key()],"#e":[id],"limit":0}])).await.map_err(|_|Error)?; + let mut ready = false; + for _ in 0..8 { + match connection + .next_event(Duration::from_secs(10)) + .await + .map_err(|_| Error)? + { + RelayMessage::Eose { subscription_id } if subscription_id == "shared-reply" => { + ready = true; + break; + } + RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(Error), + _ => {} + } + } + if !ready { + return Err(Error); + } + if !connection + .send_event(event) + .await + .map_err(|_| Error)? + .accepted + { + return Err(Error); + } + for _ in 0..32 { + match connection + .next_event(Duration::from_secs(45)) + .await + .map_err(|_| Error)? + { + RelayMessage::Event { + subscription_id, + event, + } if subscription_id == "shared-reply" => { + let reply = + open_reply(&event, &self.keys, &self.manager, &id).map_err(|_| Error)?; + if reply.request_id != request.request_id { + return Err(Error); + } + if reply.status != Status::Done { + continue; + } + let response = reply.response.ok_or(Error)?; + if response["ok"] == true { + return Ok(None); + } + if response["code"] == "shared_task_retry" { + return Err(Error); + } + let reason = response["error"] + .as_str() + .unwrap_or("task permission was refused"); + return Ok(Some(reason.chars().take(500).collect())); + } + RelayMessage::Closed { .. } | RelayMessage::Auth { .. } => return Err(Error), + _ => {} + } + } + Err(Error) + } + + pub(crate) async fn refuse(&self, trigger: &Event, reason: &str) -> Result<(), Error> { + if !self.accepts(trigger) { + return Err(Error); + } + { + let scope = buzz_sdk::sharing::task::scope( + trigger, + self.keys.public_key(), + Timestamp::now().as_secs(), + ) + .map_err(|_| Error)?; + let tags = [ + vec!["h".into(), scope.channel.to_string()], + vec!["e".into(), scope.root.to_hex(), "".into(), "root".into()], + vec!["e".into(), trigger.id.to_hex(), "".into(), "reply".into()], + ] + .into_iter() + .map(Tag::parse) + .collect::, _>>() + .map_err(|_| Error)?; + // No p tag: a refusal must not wake the handing-off agent in a loop. + let notice = EventBuilder::new( + Kind::Custom(9), + format!("Could not accept this shared task: {reason}"), + ) + .tags(tags) + .sign_with_keys(&self.keys) + .map_err(|_| Error)?; + self.outbox.lock().await.reserve(trigger.id, notice)?; + } + self.flush_notice().await + } + + pub(crate) async fn flush_notice(&self) -> Result<(), Error> { + let outbox = self.outbox.lock().await; + let now = Timestamp::now().as_secs(); + let Some((path, mut notice)) = outbox.next(now)? else { + return Ok(()); + }; + if notice.event.pubkey != self.keys.public_key() + || notice.event.verify().is_err() + || !notice + .event + .tags + .iter() + .any(|t| t.as_slice() == ["h", &self.scope.channel.to_string()]) + || !notice + .event + .tags + .iter() + .any(|t| t.as_slice() == ["e", &self.scope.root.to_hex(), "", "root"]) + || notice + .event + .tags + .iter() + .any(|t| t.as_slice().first().map(String::as_str) == Some("p")) + { + return Err(Error); + } + outbox.attempt(&path, &mut notice, now)?; + let delivered = tokio::time::timeout(Duration::from_secs(10), async { + let auth = buzz_sdk::nip_oa::parse_auth_tag(&self.auth).map_err(|_| Error)?; + let mut connection = + NostrWsConnection::connect_authenticated(&self.relay, &self.keys, Some(&auth)) + .await + .map_err(|_| Error)?; + if connection + .send_event(notice.event) + .await + .map_err(|_| Error)? + .accepted + { + Ok(()) + } else { + Err(Error) + } + }) + .await; + if matches!(delivered, Ok(Ok(()))) { + outbox.complete(&path)?; + } else { + tracing::warn!("shared task refusal retained for bounded delivery retry"); + } + Ok(()) + } + + pub(crate) fn start_notices(self: std::sync::Arc) -> NoticeWorker { + NoticeWorker(tokio::spawn(async move { + let mut tick = tokio::time::interval(Duration::from_secs(10)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tick.tick().await; + if self.flush_notice().await.is_err() { + tracing::error!( + "shared refusal journal needs repair; persisted notices retained" + ); + return; + } + } + })) + } +} + +pub(crate) struct NoticeWorker(tokio::task::JoinHandle<()>); +impl Drop for NoticeWorker { + fn drop(&mut self) { + self.0.abort(); + } +} diff --git a/crates/buzz-acp/src/sharing/notices.rs b/crates/buzz-acp/src/sharing/notices.rs new file mode 100644 index 00000000000..3366fba7ec0 --- /dev/null +++ b/crates/buzz-acp/src/sharing/notices.rs @@ -0,0 +1,136 @@ +//! Durable, bounded refusal notices. Store before sending, preserve the exact +//! signed event after a lost acknowledgement, and retain exhausted records. + +use super::Error; +use nostr::{Event, EventId}; +use serde::{Deserialize, Serialize}; +use std::{ + fs, + io::{Read, Write}, + path::{Path, PathBuf}, +}; + +const MAX_FILES: usize = 32; +const MAX_BYTES: u64 = 4096; + +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Notice { + pub trigger: EventId, + pub event: Event, + pub attempts: u8, + pub next_try: u64, +} + +pub(super) struct Outbox { + directory: PathBuf, +} + +impl Outbox { + pub fn new(directory: PathBuf) -> Self { + Self { directory } + } + + fn files(&self) -> Result, Error> { + if !self.directory.exists() { + return Ok(Vec::new()); + } + let mut files = Vec::new(); + for entry in fs::read_dir(&self.directory) + .map_err(|_| Error)? + .take(MAX_FILES + 1) + { + let entry = entry.map_err(|_| Error)?; + if !entry.file_type().map_err(|_| Error)?.is_file() { + return Err(Error); + } + files.push(entry.path()); + } + if files.len() > MAX_FILES { + return Err(Error); + } + files.sort(); + Ok(files) + } + + pub fn reserve(&self, trigger: EventId, event: Event) -> Result<(), Error> { + fs::create_dir_all(&self.directory).map_err(|_| Error)?; + let path = self.directory.join(format!("{trigger}.json")); + if let Ok(metadata) = fs::symlink_metadata(&path) { + return if metadata.file_type().is_file() { + Ok(()) + } else { + Err(Error) + }; + } + if self.files()?.len() >= MAX_FILES { + return Err(Error); + } + let notice = Notice { + trigger, + event, + attempts: 0, + next_try: 0, + }; + self.save(&path, ¬ice, true) + } + + pub fn next(&self, now: u64) -> Result, Error> { + for path in self.files()? { + let mut bytes = Vec::new(); + fs::File::open(&path) + .map_err(|_| Error)? + .take(MAX_BYTES + 1) + .read_to_end(&mut bytes) + .map_err(|_| Error)?; + if bytes.len() as u64 > MAX_BYTES { + return Err(Error); + } + let notice: Notice = serde_json::from_slice(&bytes).map_err(|_| Error)?; + if path.file_name().and_then(|s| s.to_str()) + != Some(&format!("{}.json", notice.trigger)) + { + return Err(Error); + } + if notice.attempts < 8 && notice.next_try <= now { + return Ok(Some((path, notice))); + } + } + Ok(None) + } + + pub fn attempt(&self, path: &Path, notice: &mut Notice, now: u64) -> Result<(), Error> { + notice.attempts = notice.attempts.saturating_add(1); + notice.next_try = now.saturating_add((5u64 << notice.attempts.min(4)).min(60)); + self.save(path, notice, false) + } + + pub fn complete(&self, path: &Path) -> Result<(), Error> { + fs::remove_file(path).map_err(|_| Error) + } + + fn save(&self, path: &Path, notice: &Notice, create: bool) -> Result<(), Error> { + let data = serde_json::to_vec(notice).map_err(|_| Error)?; + if data.len() as u64 > MAX_BYTES { + return Err(Error); + } + let mut file = tempfile::NamedTempFile::new_in(&self.directory).map_err(|_| Error)?; + file.write_all(&data).map_err(|_| Error)?; + file.as_file().sync_all().map_err(|_| Error)?; + if create { + if let Err(error) = file.persist_noclobber(path) { + if error.error.kind() != std::io::ErrorKind::AlreadyExists { + return Err(Error); + } + } + } else { + file.persist(path).map_err(|_| Error)?; + } + #[cfg(unix)] + fs::File::open(&self.directory) + .map_err(|_| Error)? + .sync_all() + .map_err(|_| Error)?; + Ok(()) + } +} diff --git a/crates/buzz-acp/src/sharing/tests.rs b/crates/buzz-acp/src/sharing/tests.rs new file mode 100644 index 00000000000..2d756d5244a --- /dev/null +++ b/crates/buzz-acp/src/sharing/tests.rs @@ -0,0 +1,357 @@ +use super::*; +use futures_util::{SinkExt, StreamExt}; +use nostr::{EventBuilder, Kind, Tag}; +use std::sync::{Arc, Mutex}; +use tokio_tungstenite::tungstenite::Message; + +fn fixture(relay: String, manager: PublicKey, directory: std::path::PathBuf) -> (Client, Event) { + let owner = Keys::generate(); + let agent = Keys::generate(); + let channel = uuid::Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(9), "team task") + .tags([ + Tag::parse(["h", &channel.to_string()]).unwrap(), + Tag::public_key(agent.public_key()), + ]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + let scope = + buzz_sdk::sharing::task::scope(&event, agent.public_key(), Timestamp::now().as_secs()) + .unwrap(); + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap(); + ( + Client { + keys: agent, + auth, + relay, + manager, + scope, + outbox: Arc::new(tokio::sync::Mutex::new(notices::Outbox::new(directory))), + }, + event, + ) +} + +#[tokio::test] +async fn encrypted_admission_uses_fresh_ids_and_request_scoped_reply_filters() { + let directory = tempfile::tempdir().unwrap(); + let manager = Keys::generate(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("ws://{}", listener.local_addr().unwrap()); + let requests = Arc::new(Mutex::new(Vec::::new())); + let captured = requests.clone(); + let manager_pub = manager.public_key(); + let server = tokio::spawn(async move { + for _ in 0..5 { + let (stream, _) = listener.accept().await.unwrap(); + let mut socket = tokio_tungstenite::accept_async(stream).await.unwrap(); + socket + .send(Message::Text(json!(["AUTH", "fixture"]).to_string().into())) + .await + .unwrap(); + let mut subscribed = None; + while let Some(Ok(message)) = socket.next().await { + let Message::Text(text) = message else { + continue; + }; + let message: serde_json::Value = serde_json::from_str(&text).unwrap(); + match message[0].as_str().unwrap() { + "AUTH" => socket + .send(Message::Text( + json!(["OK", message[1]["id"], true, ""]).to_string().into(), + )) + .await + .unwrap(), + "REQ" => { + subscribed = Some(message[2]["#e"][0].as_str().unwrap().to_owned()); + socket + .send(Message::Text( + json!(["EOSE", "shared-reply"]).to_string().into(), + )) + .await + .unwrap(); + } + "EVENT" => { + let event: Event = serde_json::from_value(message[1].clone()).unwrap(); + if event.kind.as_u16() == 9 { + assert!(!event.tags.iter().any(|t| t + .as_slice() + .first() + .map(String::as_str) + == Some("p"))); + socket + .send(Message::Text( + json!(["OK", event.id, true, ""]).to_string().into(), + )) + .await + .unwrap(); + break; + } + assert_eq!(subscribed.as_deref(), Some(event.id.to_hex().as_str())); + assert!( + serde_json::from_str::(&event.content).is_err(), + "control content must be encrypted" + ); + let opened = buzz_session_control::open_request(&event, &manager).unwrap(); + assert_eq!(opened.request.op, Op::Admit); + let attempt = captured.lock().unwrap().len(); + captured + .lock() + .unwrap() + .push(opened.request.request_id.clone()); + socket + .send(Message::Text( + json!(["OK", event.id, true, ""]).to_string().into(), + )) + .await + .unwrap(); + let response = if attempt == 0 || attempt == 3 { + json!({"ok":true}) + } else { + json!({"ok":false,"code":"shared_task_denied","error":"sharing was revoked"}) + }; + let reply = + buzz_session_control::Reply::done(opened.request.request_id, response); + let event = buzz_session_control::build_reply( + &manager, + &opened.signer, + &event.id, + &reply, + Timestamp::now(), + ) + .unwrap(); + socket + .send(Message::Text( + json!(["EVENT", "shared-reply", event]).to_string().into(), + )) + .await + .unwrap(); + break; + } + other => panic!("unexpected frame {other}"), + } + } + } + }); + let (client, event) = fixture(url, manager_pub, directory.path().into()); + assert!(client.authorize(&event).await.unwrap().is_none()); + assert_eq!( + client.authorize(&event).await.unwrap().as_deref(), + Some("sharing was revoked") + ); + let (rest, rest_server) = + crate::author_gate_tests::nip11_server(json!({"self":manager_pub})).await; + let channels = crate::pool::ChannelInfoResolver::new( + std::collections::HashMap::from([( + client.scope.channel, + crate::relay::ChannelInfo { + name: "shared".into(), + channel_type: "stream".into(), + description: None, + }, + )]), + rest.clone(), + ); + let owner = buzz_sdk::nip_oa::verify_auth_tag(&client.auth, &client.keys.public_key()).unwrap(); + let owner = crate::OwnerCache::new(Some(owner.to_hex())); + let mut gate = + crate::InboundAuthorGate::connect(&rest, &client.keys.public_key().to_hex(), "shared-test") + .await; + gate.enable_sharing(Some(Arc::new(client.clone()))); + let inbound = || crate::relay::BuzzEvent { + channel_id: client.scope.channel, + connection_generation: 0, + event: event.clone(), + }; + assert!(crate::authorize_normal_listener_event( + &mut gate, + inbound(), + &crate::config::RespondTo::Owner, + &Default::default(), + &owner, + &channels, + &rest + ) + .await + .unwrap() + .is_none()); + let allowed = crate::authorize_normal_listener_event( + &mut gate, + inbound(), + &crate::config::RespondTo::Owner, + &Default::default(), + &owner, + &channels, + &rest, + ) + .await + .unwrap() + .unwrap(); + assert_eq!( + allowed.into_parts().1, + event.pubkey.to_hex(), + "shared requests retain the actual signer" + ); + rest_server.abort(); + server.await.unwrap(); + let ids = requests.lock().unwrap(); + assert_eq!(ids.len(), 4); + assert_ne!( + ids[0], ids[1], + "a cached transport approval must not survive revocation" + ); +} + +#[tokio::test] +async fn failed_refusal_delivery_is_durable_and_does_not_generate_agent_mentions() { + let directory = tempfile::tempdir().unwrap(); + let (client, event) = fixture( + "ws://127.0.0.1:1".into(), + Keys::generate().public_key(), + directory.path().into(), + ); + client + .refuse(&event, "authorization unavailable; mention me again") + .await + .unwrap(); + let outbox = notices::Outbox::new(directory.path().into()); + let (_, notice) = outbox + .next(Timestamp::now().as_secs() + 60) + .unwrap() + .unwrap(); + assert_eq!(notice.trigger, event.id); + assert_eq!(notice.attempts, 1); + assert!(notice.event.verify().is_ok()); + assert!(!notice + .event + .tags + .iter() + .any(|t| t.as_slice().first().map(String::as_str) == Some("p"))); + assert!(notice.event.content.contains("authorization unavailable")); + let outside = EventBuilder::new(Kind::Custom(9), "different thread") + .tags([ + Tag::parse(["h", &client.scope.channel.to_string()]).unwrap(), + Tag::public_key(client.keys.public_key()), + ]) + .sign_with_keys(&Keys::generate()) + .unwrap(); + assert!(!client.accepts(&outside)); + assert!(client.refuse(&outside, "no").await.is_err()); +} + +#[test] +fn outbox_replays_exact_signatures_and_bounds_storage() { + let directory = tempfile::tempdir().unwrap(); + let owner = Keys::generate(); + let event = EventBuilder::new(Kind::Custom(9), "refusal") + .sign_with_keys(&owner) + .unwrap(); + let outbox = notices::Outbox::new(directory.path().into()); + outbox.reserve(event.id, event.clone()).unwrap(); + let (path, mut notice) = outbox.next(0).unwrap().unwrap(); + outbox.attempt(&path, &mut notice, 0).unwrap(); + let recovered = notices::Outbox::new(directory.path().into()); + assert!(recovered.next(9).unwrap().is_none()); + let (_, retry) = recovered.next(10).unwrap().unwrap(); + assert_eq!(retry.event, event); + recovered.complete(&path).unwrap(); + assert!(recovered.next(10).unwrap().is_none()); + let huge = EventBuilder::new(Kind::Custom(9), "x".repeat(5000)) + .sign_with_keys(&owner) + .unwrap(); + assert!(outbox.reserve(huge.id, huge).is_err()); + for i in 0..32 { + outbox + .reserve(nostr::EventId::from_byte_array([i; 32]), event.clone()) + .unwrap(); + } + assert!(outbox + .reserve(nostr::EventId::from_byte_array([33; 32]), event) + .is_err()); +} + +#[tokio::test] +async fn queued_shared_task_is_rechecked_before_context_or_acp_session_creation() { + use crate::pool::{OwnedAgent, PromptOutcome, SessionState}; + let directory = tempfile::tempdir().unwrap(); + let (client, event) = fixture( + "ws://127.0.0.1:1".into(), + Keys::generate().public_key(), + directory.path().join("notices"), + ); + let capture = directory.path().join("acp-wire.ndjson"); + let quoted = capture.to_string_lossy().replace('\'', "'\\''"); + let script = + format!("while IFS= read -r line; do printf '%s\\n' \"$line\" >> '{quoted}'; done"); + let acp = crate::acp::AcpClient::spawn("bash", &["-c".into(), script], &[], false) + .await + .unwrap(); + let agent = OwnedAgent { + index: 0, + acp, + state: SessionState::default(), + model_capabilities: None, + desired_model: None, + model_overridden: false, + desired_model_request_id: None, + desired_model_pending_ack: false, + startup_effort: None, + agent_name: "sharing-boundary".into(), + goose_system_prompt_supported: None, + protocol_version: 1, + }; + let mut ctx = crate::pool::tests::make_prompt_context_no_owner(); + ctx.agent_owner_pubkey = + Some(buzz_sdk::nip_oa::verify_auth_tag(&client.auth, &client.keys.public_key()).unwrap()); + ctx.agent_keys = client.keys.clone(); + ctx.sharing = Some(Arc::new(client.clone())); + ctx.memory_enabled = true; + ctx.initial_message = Some("private setup context must not be sent".into()); + ctx.dedup_mode = crate::config::DedupMode::Queue; + let batch = crate::queue::FlushBatch { + channel_id: client.scope.channel, + scope: crate::scope::SessionScope::Thread { + channel_id: client.scope.channel, + root_event_id: client.scope.root.to_hex(), + }, + events: vec![crate::queue::BatchEvent { + event: event.clone(), + prompt_tag: "shared".into(), + received_at: std::time::Instant::now(), + edit: None, + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let (send, mut receive) = tokio::sync::mpsc::unbounded_channel(); + tokio::time::timeout( + Duration::from_secs(3), + crate::pool::run_prompt_task( + agent, + Some(batch), + None, + Arc::new(ctx), + send, + None, + "queued-shared-task".into(), + Default::default(), + ), + ) + .await + .unwrap(); + let mut result = receive.recv().await.unwrap(); + assert!(matches!( + result.outcome, + PromptOutcome::Error(crate::acp::AcpError::TaskPermission(_)) + )); + assert_eq!( + result.batch.take().unwrap().events[0].event.id, + event.id, + "retain the original task for bounded retry" + ); + result.agent.acp.shutdown().await; + assert!( + !capture.exists(), + "an indeterminate shared task must not reach memory or ACP session setup" + ); +} diff --git a/crates/buzz-cli/src/commands/agents.rs b/crates/buzz-cli/src/commands/agents.rs index eedc2d40a4e..3cf9d16b287 100644 --- a/crates/buzz-cli/src/commands/agents.rs +++ b/crates/buzz-cli/src/commands/agents.rs @@ -15,6 +15,49 @@ mod sharing_tests; pub async fn dispatch(command: AgentsCmd, client: &BuzzClient) -> Result<(), CliError> { match command { + AgentsCmd::Handoff { + agent, + from, + message, + } => { + if client.auth_tag_owner_hex().is_none() { + return Err(CliError::Auth("handoff requires an agent identity".into())); + } + validate_hex64(&agent)?; + let target = PublicKey::from_hex(&agent).map_err(|e| CliError::Usage(e.to_string()))?; + let from = crate::validate::parse_event_id(&from)?; + let raw = client + .query(&json!({"kinds":[9,40002],"ids":[from],"limit":1})) + .await?; + let mut events: Vec = serde_json::from_str(&raw) + .map_err(|_| CliError::Other("invalid incoming task response".into()))?; + if events.len() != 1 { + return Err(CliError::Usage( + "incoming task is not visible to this agent".into(), + )); + } + let source = events + .pop() + .ok_or_else(|| CliError::Other("incoming task was not returned".into()))?; + if source.id != from { + return Err(CliError::Other( + "relay returned a different incoming task".into(), + )); + } + let builder = buzz_sdk::sharing::task::handoff( + &source, + client.keys().public_key(), + target, + &read_or_stdin(&message)?, + nostr::Timestamp::now().as_secs(), + ) + .map_err(|e| CliError::Usage(e.to_string()))?; + println!( + "{}", + client.submit_event(client.sign_event(builder)?).await? + ); + Ok(()) + } AgentsCmd::Share { agent, channel } => sharing(client, &agent, channel, true).await, AgentsCmd::Unshare { agent, channel } => sharing(client, &agent, channel, false).await, AgentsCmd::Sharing { diff --git a/crates/buzz-cli/src/commands/agents_sharing_tests.rs b/crates/buzz-cli/src/commands/agents_sharing_tests.rs index e4cdead739b..72221070fff 100644 --- a/crates/buzz-cli/src/commands/agents_sharing_tests.rs +++ b/crates/buzz-cli/src/commands/agents_sharing_tests.rs @@ -3,6 +3,87 @@ use axum::{extract::State, routing::post, Json, Router}; use nostr::{Event, EventBuilder, Keys, Kind}; use std::sync::{Arc, Mutex}; +#[tokio::test] +async fn handoff_preserves_signed_request_scope_and_rejects_unaddressed_sources() { + let owner = Keys::generate(); + let agent = Keys::generate(); + let colleague = Keys::generate(); + let human = Keys::generate(); + let channel = uuid::Uuid::new_v4(); + let source = EventBuilder::new(Kind::Custom(9), "review this change") + .tag(nostr::Tag::parse(["h".to_owned(), channel.to_string()]).unwrap()) + .tag(nostr::Tag::public_key(agent.public_key())) + .sign_with_keys(&human) + .unwrap(); + let query_source = source.clone(); + let writes = Arc::new(Mutex::new(Vec::::new())); + let app = Router::new() + .route("/query", post(move |Json(filters): Json>| { + let source = query_source.clone(); + async move { + assert_eq!(filters[0]["ids"], json!([source.id])); + assert_eq!(filters[0]["kinds"], json!([9, 40002])); + Json(vec![source]) + } + })) + .route("/events", post(|State(writes): State>>>, Json(event): Json| async move { + event.verify().unwrap(); + let id = event.id; + writes.lock().unwrap().push(event); + Json(json!({"event_id":id,"accepted":true,"message":""})) + })).with_state(writes.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + let command = || AgentsCmd::Handoff { + agent: colleague.public_key().to_hex(), + from: source.id.to_hex(), + message: "check the permission boundary".into(), + }; + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &agent.public_key(), "").unwrap(); + let client = BuzzClient::new( + url.clone(), + agent.clone(), + Some(buzz_sdk::nip_oa::parse_auth_tag(&auth).unwrap()), + None, + ) + .unwrap(); + dispatch(command(), &client).await.unwrap(); + let human_client = BuzzClient::new(url.clone(), human, None, None).unwrap(); + assert!(dispatch(command(), &human_client).await.is_err()); + let other = Keys::generate(); + let auth = buzz_sdk::nip_oa::compute_auth_tag(&owner, &other.public_key(), "").unwrap(); + let unaddressed = BuzzClient::new( + url, + other, + Some(buzz_sdk::nip_oa::parse_auth_tag(&auth).unwrap()), + None, + ) + .unwrap(); + assert!(dispatch(command(), &unaddressed).await.is_err()); + let events = writes.lock().unwrap(); + assert_eq!(events.len(), 1); + let event = &events[0]; + assert_eq!(event.pubkey, agent.public_key()); + assert_eq!( + buzz_sdk::sharing::task::scope( + event, + colleague.public_key(), + nostr::Timestamp::now().as_secs() + ) + .unwrap(), + buzz_sdk::sharing::task::Scope { + channel, + root: source.id + } + ); + let origin = buzz_sdk::sharing::task::origin(event).unwrap().unwrap(); + assert_eq!(origin.request, source.id); + assert_eq!(origin.agent, agent.public_key()); + assert_eq!(origin.parent, source.id); + server.abort(); +} + #[tokio::test] async fn share_and_unshare_publish_signed_scoped_heads_after_owner_verification() { let owner = Keys::generate(); diff --git a/crates/buzz-cli/src/lib.rs b/crates/buzz-cli/src/lib.rs index c92bf1eecbb..3a4224a2763 100644 --- a/crates/buzz-cli/src/lib.rs +++ b/crates/buzz-cli/src/lib.rs @@ -301,7 +301,19 @@ impl RespondToArg { #[derive(Subcommand)] pub enum AgentsCmd { - /// Let current members invoke your agent in one channel using explicit mentions + /// Delegate work to another shared agent, retaining the incoming task's provenance + Handoff { + /// Target agent pubkey (hex) + agent: String, + /// Incoming task event ID; must explicitly address this signing agent + #[arg(long)] + from: String, + /// Task text, or '-' to read stdin + #[arg(long)] + message: String, + }, + /// Let current channel members invoke your agent and its configured tools by explicit mention. + /// Ownership and lifecycle control stay with you; DMs remain private. Share { /// Agent pubkey (hex); you must sign as its human owner agent: String, @@ -2463,6 +2475,7 @@ mod tests { "archived", "draft-create", "draft-update", + "handoff", "share", "sharing", "unarchive", @@ -2609,7 +2622,7 @@ mod tests { #[test] fn subcommand_counts_are_stable() { let expected: Vec<(&str, usize)> = vec![ - ("agents", 8), + ("agents", 9), ("canvas", 4), ("channels", 16), ("dms", 4), diff --git a/crates/buzz-sandbox-reaper/src/manager.rs b/crates/buzz-sandbox-reaper/src/manager.rs index c9cd70f4614..5115cb55495 100644 --- a/crates/buzz-sandbox-reaper/src/manager.rs +++ b/crates/buzz-sandbox-reaper/src/manager.rs @@ -48,6 +48,8 @@ pub struct ManagerConfig { developers: HashMap, /// The bucket every developer's checkpoints go to, under `/`. checkpoint_bucket: String, + /// Enable only after deploying a matching shared-task agent image. + shared_tasks: bool, } /// One configured developer. @@ -72,6 +74,8 @@ struct ConfigFile { developers: BTreeMap, /// The checkpoint bucket, shared by every developer under `/`. checkpoint_bucket: String, + #[serde(default)] + shared_tasks: bool, } #[derive(Deserialize)] @@ -196,6 +200,7 @@ impl ManagerConfig { Ok(Self { developers, checkpoint_bucket: file.checkpoint_bucket, + shared_tasks: file.shared_tasks, }) } @@ -478,7 +483,7 @@ impl Manager { agent, owner, defaults: self.defaults.clone(), - env: self.config.launch_env(developer), + env: self.shared_env(developer), }, } .run(self) @@ -571,6 +576,11 @@ impl Manager { developer: developer.clone(), } } else if op == Op::Admit { + if !self.config.shared_tasks { + return answer( + serde_json::json!({"ok":false,"code":"shared_task_denied","error":"shared tasks are not enabled on this manager"}), + ); + } let request = match serde_json::from_value(opened.request.payload.clone()) { Ok(request) => request, Err(_) => { @@ -600,7 +610,7 @@ impl Manager { agent: start.agent(), owner, defaults: self.defaults.clone(), - env: self.config.launch_env(developer), + env: self.shared_env(developer), }; Work::Start { start, target } } else { @@ -618,18 +628,31 @@ impl Manager { }; match self.payload(&opened) { // 6. The namespace comes from the verified owner, never the request. - Ok(request) => Work::Provider { - request, - registration, - target: Target::Managed { - client: self.client.clone(), - namespace: developer.namespace.clone(), - agent: opened.signer, - owner, - defaults: self.defaults.clone(), - env: self.config.launch_env(developer), - }, - }, + Ok(request) => { + let thread = match &request { + Request::Deploy(d) | Request::Stop(d) => { + d.provider_config["sandbox"]["thread_root"].is_string() + } + Request::Info => false, + }; + let env = if thread { + self.shared_env(developer) + } else { + self.config.launch_env(developer) + }; + Work::Provider { + request, + registration, + target: Target::Managed { + client: self.client.clone(), + namespace: developer.namespace.clone(), + agent: opened.signer, + owner, + defaults: self.defaults.clone(), + env, + }, + } + } Err(response) => return answer(response), } }; @@ -778,6 +801,17 @@ impl Manager { }) } + fn shared_env(&self, developer: &Developer) -> BTreeMap { + let mut env = self.config.launch_env(developer); + if self.config.shared_tasks { + env.insert( + "BUZZ_ACP_SHARING_MANAGER".into(), + self.keys.public_key().to_hex(), + ); + } + env + } + pub(crate) async fn admit_shared( &self, owner: PublicKey, @@ -786,6 +820,9 @@ impl Manager { request: &crate::shared_tasks::Request, ) -> serde_json::Value { use buzz_sdk::sharing::policy::Error as PolicyError; + if !self.config.shared_tasks { + return serde_json::json!({"ok":false,"code":"shared_task_denied","error":"shared tasks are not enabled on this manager"}); + } let result=tokio::time::timeout(std::time::Duration::from_secs(40),async { let task=match crate::shared_tasks::verify(self.client.clone(),namespace,owner,agent,request).await { Ok(task)=>task, diff --git a/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs b/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs index e931d3c68aa..0c20affd350 100644 --- a/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs +++ b/crates/buzz-sandbox-reaper/src/manager_tests/shared.rs @@ -1,5 +1,11 @@ use super::*; +fn manager() -> Manager { + let mut manager = super::manager(); + manager.config.shared_tasks = true; + manager +} + fn task_request() -> Sent { let trigger = EventBuilder::new(Kind::Custom(9), "teammate task") .tags([ @@ -15,6 +21,39 @@ fn task_request() -> Sent { } } +#[tokio::test] +async fn sharing_rollout_is_disabled_by_default_and_supplies_a_confined_manager_key() { + let disabled = super::manager(); + assert_eq!(admit(&disabled, &task_request()), "shared_task_denied"); + let enabled = manager(); + let sent = Sent::deploy(); + let Admission::Start { job, .. } = enabled.admit(&sent.event(), now()) else { + panic!("thread deploy refused") + }; + let Work::Provider { + target: Target::Managed { env, .. }, + .. + } = job.work + else { + panic!("not a managed thread") + }; + assert_eq!( + env.get("BUZZ_ACP_SHARING_MANAGER"), + Some(&manager_keys().public_key().to_hex()) + ); + let Admission::Start { job, .. } = disabled.admit(&sent.event(), now()) else { + panic!("legacy deploy refused") + }; + let Work::Provider { + target: Target::Managed { env, .. }, + .. + } = job.work + else { + panic!("not a managed thread") + }; + assert!(!env.contains_key("BUZZ_ACP_SHARING_MANAGER")); +} + #[tokio::test] async fn shared_admission_is_agent_attested_and_confined_to_a_registered_identity() { let manager = manager();