diff --git a/crates/buzz-acp/README.md b/crates/buzz-acp/README.md index 1161dc1034e..853117ea895 100644 --- a/crates/buzz-acp/README.md +++ b/crates/buzz-acp/README.md @@ -140,14 +140,15 @@ Controls which authors' events the harness forwards to the agent. Events from di | Flag | Env Var | Default | Description | |------|---------|---------|-------------| -| `--respond-to` | `BUZZ_ACP_RESPOND_TO` | `owner-only` | Author gate mode: `owner-only`, `allowlist`, `anyone`, `nobody`. | +| `--respond-to` | `BUZZ_ACP_RESPOND_TO` | `owner-only` | Author gate mode: `owner-only`, `owner`, `allowlist`, `anyone`, `nobody`. | | `--respond-to-allowlist` | `BUZZ_ACP_RESPOND_TO_ALLOWLIST` | — | Comma-separated 64-char hex pubkeys (required when mode is `allowlist`). Owner is always implicitly included. | **Modes:** | Mode | Behavior | |------|----------| -| `owner-only` | Forward only events from the agent's registered owner. If no owner is set, all events are dropped until the owner is resolved. | +| `owner-only` | Forward events from the agent's registered owner and from sibling agents (agents that have the same owner). Relay-signed workflow messages that target this agent count as their recorded owner. If no owner is set, all events are dropped until the owner is resolved. | +| `owner` | Stricter than `owner-only`: forward only events that the owner's key signed, in channels and DMs. Sibling agents and relay-signed workflow messages are dropped. If no owner is set, all events are dropped. Required by `--bound-session`. | | `allowlist` | Forward events from the listed pubkeys plus the owner. | | `anyone` | Forward all events (no author filtering). | | `nobody` | Drop all inbound events. Agent only acts on heartbeat prompts. | @@ -197,6 +198,35 @@ buzz-acp --respond-to anyone buzz-acp --respond-to nobody --heartbeat-interval 300 ``` +### Bound session + +| Flag | Env Var | Default | Description | +|------|---------|---------|-------------| +| `--bound-session` | `BUZZ_ACP_BOUND_SESSION` | — | JSON object `{"channel_id": "", "thread_root": "<64 lowercase hex event ID>"}`. Restricts the harness to one channel and one thread. | + +A bound harness drops every event outside the bound thread, including owner +`!shutdown`, `!cancel`, and `!rotate` commands. It resolves an edit to its +original message's thread before it admits the edit. Desktop `cancel_turn` and +`switch_model` observer controls for another channel are also dropped. + +A bound session requires `--respond-to owner`, `--subscribe mentions`, +`--session-policy thread`, and `--heartbeat-interval 0`. It refuses +`--no-mention-filter` and `--initial-message`. The startup error names the +flag that fails. + +### Permission mode + +| Flag | Env Var | Default | Description | +|------|---------|---------|-------------| +| `--permission-mode` | `BUZZ_ACP_PERMISSION_MODE` | `bypass-permissions` | Mode applied with `session/set_config_option` after each new session: `default`, `agent-full-access`, `auto`, `accept-edits`, `bypass-permissions`, `dont-ask`, `plan`. | + +If the adapter does not advertise or accept most modes, the harness falls back +to per-tool auto-approval. `agent-full-access` is for adapters (for example, +Codex) that run in an externally isolated container. It never falls back: the +adapter must advertise the mode in `session/new` and confirm it as the current +`mode` option. If it does not, the harness fails the request once with a +visible error, and it does not retry or restart the agent. + ### Host-controlled launch wrapper On Unix, a trusted host can set `BUZZ_ACP_LAUNCH_PREFIX` to a JSON argument array, for diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index 211c201d60a..0f58e555121 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -115,6 +115,11 @@ pub enum AcpError { #[error("Agent reported error (code {code}): {message}")] AgentError { code: i64, message: String }, + + /// The adapter cannot run the permission mode this harness requires. + /// The pipe is intact, and neither a retry nor a respawn can repair it. + #[error("Required permission mode unavailable: {0}")] + PermissionModeUnavailable(String), } /// Build an [`AcpError::AgentError`] from a JSON-RPC error object, diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index 24d59c2c933..4dbaee4b19a 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -19,6 +19,35 @@ use uuid::Uuid; use crate::filter::SubscriptionRule; +/// Immutable conversation binding for a remote workspace. +#[derive(Debug, Clone, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct BoundSession { + pub channel_id: Uuid, + pub thread_root: String, +} + +impl BoundSession { + pub fn accepts(&self, scope: &crate::scope::SessionScope) -> bool { + scope.channel_id() == self.channel_id && scope.root_event_id() == Some(&self.thread_root) + } +} + +fn parse_bound_session(value: &str) -> Result { + let binding: BoundSession = serde_json::from_str(value).map_err(|_| { + "bound session requires a channel UUID and canonical thread root".to_string() + })?; + if binding.thread_root.len() != 64 + || !binding + .thread_root + .bytes() + .all(|b| b.is_ascii_hexdigit() && !b.is_ascii_uppercase()) + { + return Err("bound session thread_root must be a lowercase event ID".into()); + } + Ok(binding) +} + /// Default idle timeout (seconds) when neither `--idle-timeout` nor the /// deprecated `--turn-timeout` is set. /// @@ -99,6 +128,8 @@ pub enum MultipleEventHandling { pub enum RespondTo { #[default] OwnerOnly, + /// Only the human owner's signed events; exclude sibling agents and workflows. + Owner, Allowlist, Anyone, Nobody, @@ -108,6 +139,7 @@ impl std::fmt::Display for RespondTo { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { Self::OwnerOnly => f.write_str("owner-only"), + Self::Owner => f.write_str("owner"), Self::Allowlist => f.write_str("allowlist"), Self::Anyone => f.write_str("anyone"), Self::Nobody => f.write_str("nobody"), @@ -128,6 +160,8 @@ pub enum PermissionMode { /// Agent default — permission requests per tool call. #[value(alias = "default")] Default, + /// Codex tools run inside an externally isolated development container. + AgentFullAccess, /// Auto mode — fully autonomous execution; model-gated (requires a model /// that supports `supportsAutoMode`). Degrades gracefully to `default` /// when the session's active model does not support it. @@ -153,6 +187,7 @@ impl PermissionMode { pub fn as_wire_str(&self) -> &'static str { match self { Self::Default => "default", + Self::AgentFullAccess => "agent-full-access", Self::Auto => "auto", Self::AcceptEdits => "acceptEdits", Self::BypassPermissions => "bypassPermissions", @@ -476,7 +511,7 @@ pub struct CliArgs { pub permission_mode: PermissionMode, /// Inbound author gate: which authors' events the harness forwards. - /// Modes: owner-only (default), allowlist, anyone, nobody. + /// Modes: owner-only (default), owner, allowlist, anyone, nobody. #[arg( long, env = "BUZZ_ACP_RESPOND_TO", @@ -492,7 +527,7 @@ pub struct CliArgs { /// Comma-separated list of allowed `--respond-to` modes. /// When set, the harness rejects startup if `--respond-to` is not in this list. - /// Modes: owner-only, allowlist, anyone, nobody. + /// Modes: owner-only, owner, allowlist, anyone, nobody. /// Default: empty (all modes allowed — no restriction). /// Example: `BUZZ_ACP_ALLOWED_RESPOND_TO=owner-only,allowlist` #[arg(long, env = "BUZZ_ACP_ALLOWED_RESPOND_TO", value_delimiter = ',')] @@ -506,6 +541,10 @@ pub struct CliArgs { #[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)] pub relay_observer: bool, + /// Restrict this remote workspace to one channel and canonical parent thread. + #[arg(long, env = "BUZZ_ACP_BOUND_SESSION", value_parser = parse_bound_session)] + pub bound_session: Option, + /// Exit after this many seconds with no dispatched events and no turn in flight. /// 0 disables inactivity self-termination. #[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)] @@ -611,6 +650,7 @@ pub struct Config { pub has_generated_codex_config: bool, /// Whether to publish encrypted observer frames through the relay. pub relay_observer: bool, + pub bound_session: Option, /// Seconds without dispatched events before an idle harness exits. 0 = disabled. pub exit_after_inactivity_secs: u64, /// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives. @@ -935,6 +975,28 @@ impl Config { /// tests can construct `CliArgs` via `CliArgs::try_parse_from` and exercise the full /// validation path without going through process args. pub fn from_args(mut args: CliArgs) -> Result { + if args.bound_session.is_some() { + let refusal = if args.respond_to != RespondTo::Owner { + Some("--respond-to owner") + } else if args.subscribe != SubscribeMode::Mentions { + Some("--subscribe mentions") + } else if args.no_mention_filter { + Some("no --no-mention-filter") + } else if args.session_policy != crate::scope::SessionPolicy::Thread { + Some("--session-policy thread") + } else if args.heartbeat_interval != 0 { + Some("--heartbeat-interval 0") + } else if args.initial_message.is_some() { + Some("no --initial-message") + } else { + None + }; + if let Some(requirement) = refusal { + return Err(ConfigError::ConfigFile(format!( + "--bound-session requires {requirement}" + ))); + } + } let keys = Keys::parse(&args.private_key)?; // Best-effort zeroize: overwrite the raw private key string to reduce // exposure via core dumps or heap inspection (#41). Without the `zeroize` @@ -1119,7 +1181,7 @@ impl Config { RespondTo::from_str(s.trim(), true).map_err(|_| { ConfigError::ConfigFile(format!( "invalid value in BUZZ_ACP_ALLOWED_RESPOND_TO: '{s}' \ - (valid values: owner-only, allowlist, anyone, nobody)" + (valid values: owner-only, owner, allowlist, anyone, nobody)" )) })?; } @@ -1202,6 +1264,7 @@ impl Config { persona_env_vars, has_generated_codex_config, relay_observer: args.relay_observer, + bound_session: args.bound_session, exit_after_inactivity_secs: args.exit_after_inactivity, lazy_pool: args.lazy_pool, idle_pool_sleep_secs: args.idle_pool_sleep, @@ -1539,6 +1602,124 @@ mod tests { use crate::filter::{ChannelScope, SubscriptionRule}; use clap::{Parser, ValueEnum}; + #[test] + fn bound_session_accepts_only_its_canonical_channel_and_thread() { + use crate::scope::{SessionPolicy, SessionScope}; + let channel_id = Uuid::new_v4(); + let keys = Keys::generate(); + let root = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@agent start") + .sign_with_keys(&keys) + .unwrap(); + let binding = parse_bound_session( + &serde_json::json!({ + "channel_id": channel_id, "thread_root": root.id.to_hex() + }) + .to_string(), + ) + .unwrap(); + assert!(binding.accepts(&SessionScope::derive( + SessionPolicy::Thread, + channel_id, + false, + &root + ))); + let reply = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@agent continue") + .tags([ + nostr::Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap(), + nostr::Tag::parse(["e", &"b".repeat(64), "", "reply"]).unwrap(), + ]) + .sign_with_keys(&keys) + .unwrap(); + assert!(binding.accepts(&SessionScope::derive( + SessionPolicy::Thread, + channel_id, + false, + &reply + ))); + assert!(!binding.accepts(&SessionScope::derive( + SessionPolicy::Thread, + Uuid::new_v4(), + false, + &reply + ))); + assert!(!binding.accepts(&SessionScope::derive( + SessionPolicy::Thread, + channel_id, + true, + &reply + ))); + let other = nostr::EventBuilder::new(nostr::Kind::Custom(9), "@agent new thread") + .sign_with_keys(&keys) + .unwrap(); + assert!(!binding.accepts(&SessionScope::derive( + SessionPolicy::Thread, + channel_id, + false, + &other + ))); + for invalid in ["short".to_string(), "A".repeat(64)] { + assert!(parse_bound_session( + &serde_json::json!({ + "channel_id": channel_id, "thread_root": invalid + }) + .to_string() + ) + .is_err()); + } + } + + #[test] + fn bound_session_requires_explicit_owner_mentions_without_autonomous_prompts() { + let key = Keys::generate().secret_key().to_secret_hex(); + let owner = Keys::generate().public_key().to_hex(); + let binding = + serde_json::json!({"channel_id":Uuid::new_v4(), "thread_root":"a".repeat(64)}) + .to_string(); + let base = [ + "buzz-acp", + "--private-key", + &key, + "--agent-owner", + &owner, + "--bound-session", + &binding, + "--respond-to", + "owner", + "--subscribe", + "mentions", + "--session-policy", + "thread", + "--heartbeat-interval", + "0", + ]; + assert!(Config::from_args(CliArgs::try_parse_from(base).unwrap()).is_ok()); + for (variation, flag) in [ + (0, "--respond-to owner"), + (1, "--no-mention-filter"), + (2, "--session-policy thread"), + (3, "--heartbeat-interval 0"), + (4, "--initial-message"), + (5, "--respond-to owner"), + (6, "--subscribe mentions"), + ] { + let mut args = CliArgs::try_parse_from(base).unwrap(); + match variation { + 0 => args.respond_to = RespondTo::Anyone, + 1 => args.no_mention_filter = true, + 2 => args.session_policy = crate::scope::SessionPolicy::Channel, + 3 => args.heartbeat_interval = 30, + 4 => args.initial_message = Some("autonomous".into()), + 5 => args.respond_to = RespondTo::OwnerOnly, + _ => args.subscribe = SubscribeMode::All, + } + let error = Config::from_args(args) + .err() + .unwrap_or_else(|| panic!("unsafe bound configuration {variation}")) + .to_string(); + assert!(error.contains(flag), "configuration {variation}: {error}"); + } + } + /// Build a minimal Config for testing without CLI parsing. fn test_config(mode: SubscribeMode) -> Config { Config { @@ -1580,6 +1761,7 @@ mod tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + bound_session: None, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -2407,6 +2589,10 @@ channels = "ALL" #[test] fn test_permission_mode_wire_strings() { + assert_eq!( + PermissionMode::AgentFullAccess.as_wire_str(), + "agent-full-access" + ); assert_eq!(PermissionMode::Default.as_wire_str(), "default"); assert_eq!(PermissionMode::Auto.as_wire_str(), "auto"); assert_eq!(PermissionMode::AcceptEdits.as_wire_str(), "acceptEdits"); @@ -2594,6 +2780,10 @@ channels = "ALL" #[test] fn test_respond_to_value_enum_parsing() { use clap::ValueEnum; + assert_eq!( + RespondTo::from_str("owner", true).unwrap(), + RespondTo::Owner + ); assert_eq!( RespondTo::from_str("owner-only", true).unwrap(), RespondTo::OwnerOnly @@ -2867,7 +3057,7 @@ channels = "ALL" let mode = RespondTo::from_str(s.trim(), true).map_err(|_| { ConfigError::ConfigFile(format!( "invalid value in BUZZ_ACP_ALLOWED_RESPOND_TO: '{s}' \ - (valid values: owner-only, allowlist, anyone, nobody)" + (valid values: owner-only, owner, allowlist, anyone, nobody)" )) })?; set.insert(mode); diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index ac387817f6a..a9be976586f 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -574,6 +574,10 @@ mod inbound_author_gate { owner_cache: &OwnerCache, rest_client: &relay::RestClient, ) -> bool { + // Strict owner mode applies identically in DMs and channels. + if matches!(respond_to, RespondTo::Owner) { + return owner_cache.get().is_some_and(|owner| owner == author); + } if is_dm { return match respond_to { RespondTo::Nobody => false, @@ -582,7 +586,7 @@ mod inbound_author_gate { } match respond_to { RespondTo::Anyone => true, - RespondTo::Nobody => false, + RespondTo::Nobody | RespondTo::Owner => false, RespondTo::OwnerOnly => is_owner_or_sibling(author, owner_cache, rest_client).await, RespondTo::Allowlist => { allowlist.contains(author) @@ -700,8 +704,11 @@ mod inbound_author_gate { owner_cache: &OwnerCache, rest_client: &relay::RestClient, ) -> InboundAuthorGateDecision { - let effective_author = - effective_prompt_author(event, self.relay_self.as_deref(), &self.agent_pubkey_hex); + let effective_author = if matches!(respond_to, RespondTo::Owner) { + event.pubkey.to_hex() + } else { + effective_prompt_author(event, self.relay_self.as_deref(), &self.agent_pubkey_hex) + }; let allowed = author_allowed( respond_to, allowlist, @@ -1840,6 +1847,7 @@ fn handle_relay_observer_control_event( observer: Option<&observer::ObserverHandle>, owner_pubkey_hex: &str, event_publisher: RelayEventPublisher, + bound_channel: Option, ) { // Defense-in-depth: verify signature even though the relay already checked. if let Err(e) = buzz_core::verify_event(&event) { @@ -1880,10 +1888,10 @@ fn handle_relay_observer_control_event( let command_type = payload.get("type").and_then(|value| value.as_str()); match command_type { Some("cancel_turn") => { - handle_cancel_turn_control(&payload, pool, observer); + handle_cancel_turn_control(&payload, pool, observer, bound_channel); } Some("switch_model") => { - handle_switch_model_control(&payload, pool, observer); + handle_switch_model_control(&payload, pool, observer, bound_channel); } Some("publish_project_owner_announcements") => { handle_publish_project_owner_announcements_control( @@ -2042,17 +2050,38 @@ fn emit_project_owner_control_result( } /// Handle a `cancel_turn` control frame: signal the in-flight task to cancel. -fn handle_cancel_turn_control( +/// Read a channel control frame's `channelId`. A bound harness refuses every +/// channel except its bound one. +fn control_frame_channel_id( payload: &serde_json::Value, - pool: &mut AgentPool, - observer: Option<&observer::ObserverHandle>, -) { + frame: &str, + bound_channel: Option, +) -> Option { let Some(channel_id) = payload .get("channelId") .and_then(|value| value.as_str()) .and_then(|value| value.parse::().ok()) else { - tracing::warn!("observer cancel_turn control frame missing valid channelId"); + tracing::warn!("observer {frame} control frame missing valid channelId"); + return None; + }; + if bound_channel.is_some_and(|bound| bound != channel_id) { + tracing::warn!( + channel_id = %channel_id, + "observer {frame} control frame outside the bound session — dropping" + ); + return None; + } + Some(channel_id) +} + +fn handle_cancel_turn_control( + payload: &serde_json::Value, + pool: &mut AgentPool, + observer: Option<&observer::ObserverHandle>, + bound_channel: Option, +) { + let Some(channel_id) = control_frame_channel_id(payload, "cancel_turn", bound_channel) else { return; }; @@ -2097,13 +2126,9 @@ fn handle_switch_model_control( payload: &serde_json::Value, pool: &mut AgentPool, observer: Option<&observer::ObserverHandle>, + bound_channel: Option, ) { - let Some(channel_id) = payload - .get("channelId") - .and_then(|value| value.as_str()) - .and_then(|value| value.parse::().ok()) - else { - tracing::warn!("observer switch_model control frame missing valid channelId"); + let Some(channel_id) = control_frame_channel_id(payload, "switch_model", bound_channel) else { return; }; let Some(model_id) = payload.get("modelId").and_then(|value| value.as_str()) else { @@ -2942,9 +2967,9 @@ async fn run_harness( // Warn if owner-dependent mode but no owner resolved yet. if startup_owner.is_none() { match &config.respond_to { - RespondTo::OwnerOnly => { + mode @ (RespondTo::OwnerOnly | RespondTo::Owner) => { tracing::warn!( - "respond-to=owner-only but no owner is set — all events will be \ + "respond-to={mode} but no owner is set — all events will be \ dropped. Set BUZZ_AUTH_TAG or --agent-owner, or use --respond-to=anyone." ); } @@ -3423,6 +3448,7 @@ async fn run_harness( observer.as_ref(), owner_hex, relay.event_publisher(), + config.bound_session.as_ref().map(|binding| binding.channel_id), ); } else { tracing::warn!("observer control frame received but no owner resolved — dropping"); @@ -3565,6 +3591,11 @@ async fn run_harness( continue; } + // Control commands also stay inside the bound workspace. + if !bound_session_admits(config.bound_session.as_ref(), &buzz_event, None) { + continue; + } + // Check: kind:9, content "!shutdown", from owner, mentions THIS agent. let is_shutdown = is_owner_control_command( &buzz_event.event, @@ -3746,6 +3777,13 @@ async fn run_harness( .await; let session_scope = ingress.session_scope(config.session_policy, channel_is_dm); + if !bound_session_admits( + config.bound_session.as_ref(), + &ingress.buzz_event, + Some(&session_scope), + ) { + continue; + } tracing::debug!( channel_id = %session_scope.channel_id(), scope = %session_scope.telemetry_label(), @@ -4415,6 +4453,35 @@ fn event_mentions_agent(event: &nostr::Event, agent_pubkey_hex: &str) -> bool { }) } +/// Whether a bound workspace admits a relay event; an unbound harness admits +/// everything. Before edit routing resolves (`routed_scope` is `None`), a +/// kind:9 message is checked against its own thread tags and other kinds only +/// by channel. Once routed, the resolved session scope must be the bound thread. +fn bound_session_admits( + binding: Option<&config::BoundSession>, + buzz_event: &relay::BuzzEvent, + routed_scope: Option<&scope::SessionScope>, +) -> bool { + let Some(binding) = binding else { + return true; + }; + if buzz_event.channel_id != binding.channel_id { + return false; + } + match routed_scope { + Some(scope) => binding.accepts(scope), + None => { + u32::from(buzz_event.event.kind.as_u16()) != KIND_STREAM_MESSAGE + || binding.accepts(&scope::SessionScope::derive( + scope::SessionPolicy::Thread, + buzz_event.channel_id, + false, + &buzz_event.event, + )) + } + } +} + fn is_owner_control_command( event: &nostr::Event, kind_u32: u32, @@ -5377,6 +5444,22 @@ fn handle_prompt_result( and then re-send." .to_string(); spawn_failure_notice(rest_client, &batch, content); + } else if matches!( + &result.outcome, + PromptOutcome::Error(acp::AcpError::PermissionModeUnavailable(_)) + ) { + // The adapter cannot run the required mode; a retry or respawn + // reaches the same refusal. Fail the batch once, visibly. + tracing::warn!( + channel_id = %batch.channel_id, + events = batch.events.len(), + "dead-lettering batch immediately — required permission mode unavailable" + ); + let content = "⚠️ I couldn't process the last request: this agent's adapter \ + does not support the agent-full-access permission mode it is configured \ + to require. Update the agent runtime or change its permission mode." + .to_string(); + spawn_failure_notice(rest_client, &batch, content); } else if let Some(dead) = queue.requeue(batch) { let reason = match &result.outcome { PromptOutcome::Timeout(TimeoutKind::Idle) => "the turn timed out".to_string(), @@ -6717,8 +6800,8 @@ mod owner_control_command_tests { "channelId": ch.to_string(), "modelId": "new-model", "requestId": "pick-1", }); - handle_cancel_turn_control(&payload, &mut pool, Some(&observer)); - handle_switch_model_control(&payload, &mut pool, Some(&observer)); + handle_cancel_turn_control(&payload, &mut pool, Some(&observer), None); + handle_switch_model_control(&payload, &mut pool, Some(&observer), None); let results = observer.snapshot(); assert_eq!(results.len(), 2); for result in results { @@ -6769,15 +6852,59 @@ mod owner_control_command_tests { }); match &signal { ControlSignal::Cancel => { - handle_cancel_turn_control(&payload, &mut pool, Some(&observer)) + handle_cancel_turn_control(&payload, &mut pool, Some(&observer), None) } - _ => handle_switch_model_control(&payload, &mut pool, Some(&observer)), + _ => handle_switch_model_control(&payload, &mut pool, Some(&observer), None), } assert_eq!(rx.await.unwrap(), signal); assert_eq!(observer.snapshot()[0].payload["status"], "sent"); } } + #[tokio::test] + async fn bound_harness_drops_observer_channel_controls_for_other_channels() { + for cancel in [true, false] { + let mut pool = AgentPool::from_slots(vec![]); + let bound = Uuid::new_v4(); + let foreign = Uuid::new_v4(); + let (foreign_tx, mut foreign_rx) = tokio::sync::oneshot::channel(); + insert_task_meta( + &mut pool, + 0, + scope::SessionScope::Conversation { + channel_id: foreign, + }, + foreign_tx, + ); + let (bound_tx, bound_rx) = tokio::sync::oneshot::channel(); + insert_task_meta( + &mut pool, + 1, + scope::SessionScope::Conversation { channel_id: bound }, + bound_tx, + ); + let observer = observer::ObserverHandle::in_process(); + for channel_id in [foreign, bound] { + let payload = serde_json::json!({ + "channelId": channel_id.to_string(), "modelId": "new-model", + }); + if cancel { + handle_cancel_turn_control(&payload, &mut pool, Some(&observer), Some(bound)); + } else { + handle_switch_model_control(&payload, &mut pool, Some(&observer), Some(bound)); + } + } + assert_eq!( + foreign_rx.try_recv(), + Err(tokio::sync::oneshot::error::TryRecvError::Empty) + ); + assert!(bound_rx.await.is_ok()); + let results = observer.snapshot(); + assert_eq!(results.len(), 1, "only the bound channel gets a result"); + assert_eq!(results[0].channel_id, Some(bound.to_string())); + } + } + // Fix #2: mid-turn steer/interrupt must target the exact thread scope, not // “the first task in the channel” — two threads in one channel must not // interrupt each other. @@ -7002,6 +7129,153 @@ mod owner_control_command_tests { } } +#[cfg(test)] +mod bound_session_admission_tests { + use super::*; + use nostr::{EventBuilder, Keys, Kind, Tag}; + + const AGENT: &str = "ab"; + + fn buzz_event( + channel_id: Uuid, + kind: u16, + content: &str, + root: Option<&str>, + ) -> relay::BuzzEvent { + let mut tags = vec![Tag::parse(["p", &AGENT.repeat(32)]).unwrap()]; + // NIP-10 treats a lone root marker as top-level, so a reply carries both. + if let Some(root) = root { + tags.push(Tag::parse(["e", root, "", "root"]).unwrap()); + tags.push(Tag::parse(["e", &"d".repeat(64), "", "reply"]).unwrap()); + } + relay::BuzzEvent { + connection_generation: 0, + channel_id, + event: EventBuilder::new(Kind::Custom(kind), content) + .tags(tags) + .sign_with_keys(&Keys::generate()) + .unwrap(), + } + } + + fn binding() -> config::BoundSession { + config::BoundSession { + channel_id: Uuid::new_v4(), + thread_root: "a".repeat(64), + } + } + + /// The pre-check runs before `!shutdown` / `!cancel`, so an owner command + /// in a foreign thread or channel never reaches the control handlers. + #[test] + fn owner_controls_outside_the_bound_thread_are_dropped() { + let binding = binding(); + let foreign_root = "b".repeat(64); + for command in ["!shutdown", "!cancel"] { + let foreign_thread = buzz_event( + binding.channel_id, + KIND_STREAM_MESSAGE as u16, + command, + Some(&foreign_root), + ); + assert!(is_owner_control_command( + &foreign_thread.event, + KIND_STREAM_MESSAGE, + command, + &AGENT.repeat(32) + )); + assert!(!bound_session_admits(Some(&binding), &foreign_thread, None)); + + let foreign_channel = buzz_event( + Uuid::new_v4(), + KIND_STREAM_MESSAGE as u16, + command, + Some(&binding.thread_root), + ); + assert!(!bound_session_admits( + Some(&binding), + &foreign_channel, + None + )); + + let in_thread = buzz_event( + binding.channel_id, + KIND_STREAM_MESSAGE as u16, + command, + Some(&binding.thread_root), + ); + assert!(bound_session_admits(Some(&binding), &in_thread, None)); + } + } + + #[test] + fn only_in_thread_prompts_are_admitted() { + let binding = binding(); + let in_scope = scope::SessionScope::Thread { + channel_id: binding.channel_id, + root_event_id: binding.thread_root.clone(), + }; + let reply = buzz_event( + binding.channel_id, + KIND_STREAM_MESSAGE as u16, + "@agent continue", + Some(&binding.thread_root), + ); + assert!(bound_session_admits(Some(&binding), &reply, None)); + assert!(bound_session_admits( + Some(&binding), + &reply, + Some(&in_scope) + )); + + let foreign_channel_id = Uuid::new_v4(); + let foreign_channel = buzz_event( + foreign_channel_id, + KIND_STREAM_MESSAGE as u16, + "@agent start", + Some(&binding.thread_root), + ); + let foreign_scope = scope::SessionScope::Thread { + channel_id: foreign_channel_id, + root_event_id: binding.thread_root.clone(), + }; + assert!(!bound_session_admits( + Some(&binding), + &foreign_channel, + None + )); + assert!(!bound_session_admits( + Some(&binding), + &foreign_channel, + Some(&foreign_scope) + )); + + // An edit carries its thread on the original, so the pre-check can + // only check its channel; the routed scope decides admission. + let edit_kind = buzz_core::kind::KIND_STREAM_MESSAGE_EDIT as u16; + let edit = buzz_event(binding.channel_id, edit_kind, "@agent edited", None); + assert!(bound_session_admits(Some(&binding), &edit, None)); + let edited_foreign_thread = scope::SessionScope::Thread { + channel_id: binding.channel_id, + root_event_id: "c".repeat(64), + }; + assert!(!bound_session_admits( + Some(&binding), + &edit, + Some(&edited_foreign_thread) + )); + let foreign_edit = buzz_event(foreign_channel_id, edit_kind, "@agent edited", None); + assert!(!bound_session_admits(Some(&binding), &foreign_edit, None)); + + assert!(bound_session_admits(None, &foreign_channel, None)); + assert!(bound_session_admits( + None, + &foreign_channel, + Some(&foreign_scope) + )); + } +} + #[cfg(test)] mod owner_cache_tests { use super::*; @@ -7625,6 +7899,30 @@ mod author_gate_tests { } } + #[tokio::test] + async fn owner_mode_rejects_relay_signed_workflows_in_both_listeners() { + for listener in [ListenerBoundary::Normal, ListenerBoundary::Setup] { + let relay_keys = nostr::Keys::generate(); + let owner = nostr::Keys::generate().public_key().to_hex(); + let decision = listener_boundary_scenario(ListenerBoundaryScenario { + listener, + relay_keys: &relay_keys, + workflow_owner: &owner, + responses: std::collections::VecDeque::from([Ok( + serde_json::json!({"self": relay_keys.public_key().to_hex()}), + )]), + event_generation: 0, + channel_type: "stream", + respond_to: RespondTo::Owner, + allowlist: HashSet::new(), + cache_owner: true, + cache_sibling: false, + }) + .await; + assert!(!decision.1, "strict owner mode must use the actual signer"); + } + } + /// Both production boundaries must retain DM classification when composing /// trusted workflow attribution with configured author policy. External /// allowlist entries and `Anyone` stay denied in a DM; owner and sibling @@ -8412,6 +8710,38 @@ mod author_gate_tests { ); } + #[tokio::test] + async fn strict_owner_rejects_siblings_strangers_and_missing_owner() { + let cache = cache_with_sibling(); + for is_dm in [false, true] { + for author in [OWNER, SIBLING, STRANGER] { + assert_eq!( + inbound_author_gate::test_author_allowed( + &RespondTo::Owner, + &HashSet::new(), + author, + is_dm, + &cache, + &dummy_rest_client(), + ) + .await, + author == OWNER, + ); + } + assert!( + !inbound_author_gate::test_author_allowed( + &RespondTo::Owner, + &HashSet::new(), + OWNER, + is_dm, + &OwnerCache::new(None), + &dummy_rest_client(), + ) + .await + ); + } + } + #[tokio::test] async fn test_owner_only_admits_owner_and_sibling_to_steer() { let cache = cache_with_sibling(); @@ -10130,6 +10460,7 @@ mod build_mcp_servers_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + bound_session: None, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -11227,6 +11558,7 @@ mod error_outcome_emission_tests { persona_env_vars: vec![], has_generated_codex_config: false, relay_observer: false, + bound_session: None, exit_after_inactivity_secs: 0, lazy_pool: false, idle_pool_sleep_secs: 0, @@ -12847,6 +13179,89 @@ mod error_outcome_emission_tests { ); } + /// An adapter without agent-full-access fails the same way on every retry + /// and respawn. The batch is dead-lettered once, and the healthy agent + /// returns to the pool without a crash record or respawn. + #[tokio::test] + async fn unavailable_permission_mode_fails_once_without_respawn_or_requeue() { + let event = nostr::EventBuilder::new(nostr::Kind::Custom(9), "test") + .sign_with_keys(&nostr::Keys::generate()) + .unwrap(); + let channel_id = uuid::Uuid::new_v4(); + let scope = scope::SessionScope::Conversation { channel_id }; + let batch = FlushBatch { + channel_id, + scope: scope.clone(), + events: vec![BatchEvent { + edit: None, + event, + prompt_tag: "test".into(), + received_at: std::time::Instant::now(), + }], + cancelled_events: vec![], + cancel_reason: None, + }; + let mut pool = AgentPool::from_slots(vec![None]); + let task_id = pool.join_set.spawn(async {}).id(); + pool.task_map_mut().insert( + task_id, + crate::pool::TaskMeta { + agent_index: 0, + channel_id: Some(channel_id), + scope: Some(scope.clone()), + turn_id: "test-turn-id".to_string(), + recoverable_batch: None, + control_tx: None, + steer_tx: None, + successful_steer_deliveries: HashSet::new(), + }, + ); + let mut queue = EventQueue::new(config::DedupMode::Queue); + let mut crash_history = vec![SlotCircuit { + crash_times: Vec::new(), + open_until: None, + respawn_in_flight: false, + }]; + let (respawn_tx, _respawn_rx) = mpsc::channel(8); + let mut respawn_tasks = tokio::task::JoinSet::new(); + let action = handle_prompt_result( + &mut pool, + &mut queue, + &test_config(), + PromptResult { + agent: dummy_agent(0).await, + source: PromptSource::Channel(scope), + turn_id: "test-turn-id".to_string(), + outcome: PromptOutcome::Error(acp::AcpError::PermissionModeUnavailable( + "the adapter does not advertise agent-full-access".into(), + )), + batch: Some(batch), + }, + &mut false, + &HashSet::new(), + &mut crash_history, + &respawn_tx, + &mut respawn_tasks, + None, + None, + ); + + assert!(matches!(action, LoopAction::Continue)); + assert_eq!(queue.pending_channels(), 0, "batch must not be requeued"); + assert_eq!(queue.queued_event_count(channel_id), 0); + assert_eq!( + pool.live_count(), + 1, + "the healthy agent returns to the pool" + ); + assert!( + respawn_tasks.is_empty(), + "no respawn for a configuration refusal" + ); + assert!(crash_history[0].crash_times.is_empty()); + assert!(!crash_history[0].respawn_in_flight); + } + /// Run a model-not-found turn failure for `event` through /// `handle_prompt_result` and capture the real signed notice it posts. async fn capture_model_not_found_notice( diff --git a/crates/buzz-acp/src/pool.rs b/crates/buzz-acp/src/pool.rs index 7a7e3229e70..9b23ad8195a 100644 --- a/crates/buzz-acp/src/pool.rs +++ b/crates/buzz-acp/src/pool.rs @@ -44,6 +44,9 @@ use crate::queue::{ use crate::relay::{ChannelInfo, RestClient}; use crate::scope::SessionScope; +mod permission_mode; +use permission_mode::{agent_supports_mode, apply_permission_mode}; + /// Window within which agent activity before a hard-cap death qualifies /// the turn as "recently active" (eligible for requeue instead of dead-letter). const RECENT_ACTIVITY_WINDOW: Duration = Duration::from_secs(60); @@ -1836,6 +1839,13 @@ async fn create_session_and_apply_model( // advertises the requested mode in session/new. Agents that don't support // the mode (e.g., goose crashes on unrecognized set_config_option values) // are safely skipped — the harness auto-approves via handle_permission_request. + if ctx.permission_mode == PermissionMode::AgentFullAccess + && !agent_supports_mode(&resp.raw, ctx.permission_mode.as_wire_str()) + { + return Err(AcpError::PermissionModeUnavailable( + "the adapter does not advertise agent-full-access".into(), + )); + } if !ctx.permission_mode.is_default() && agent_supports_mode(&resp.raw, ctx.permission_mode.as_wire_str()) { @@ -2147,81 +2157,6 @@ fn patch_config_option_current_value( } } -/// Set the session permission mode via `session/set_config_option`. -/// -/// Non-fatal for most errors: logs and proceeds. The agent falls back -/// to its default permission mode (`"default"`), which still works via -/// Check if the agent's `session/new` response advertises a given mode ID -/// in `result.modes.availableModes[].id`. Returns `false` if the modes -/// field is absent or the mode isn't listed. -fn agent_supports_mode(session_new_result: &serde_json::Value, mode_wire: &str) -> bool { - session_new_result - .get("modes") - .and_then(|m| m.get("availableModes")) - .and_then(|a| a.as_array()) - .map(|modes| { - modes - .iter() - .any(|m| m.get("id").and_then(|v| v.as_str()) == Some(mode_wire)) - }) - .unwrap_or(false) -} - -/// per-tool auto-approval in `handle_permission_request`. -/// -/// **Fatal exception:** if the agent process exits (e.g., goose crashes on -/// unrecognized methods), returns `Err(AgentExited)` so the caller can respawn. -async fn apply_permission_mode( - acp: &mut AcpClient, - session_id: &str, - mode: &PermissionMode, -) -> Result<(), AcpError> { - let wire = mode.as_wire_str(); - let result = tokio::time::timeout(PERMISSION_MODE_TIMEOUT, async { - acp.session_set_config_option(session_id, "mode", wire) - .await - }) - .await; - - match result { - Ok(Ok(_)) => { - tracing::info!( - target: "pool::permission", - "applied permission mode {wire:?} on session {session_id}" - ); - } - // Transport-class errors may have corrupted the stdio stream — propagate - // so the caller can respawn the agent. - Ok(Err(e @ AcpError::Io(_))) - | Ok(Err(e @ AcpError::WriteTimeout(_))) - | Ok(Err(e @ AcpError::Timeout(_))) - | Ok(Err(e @ AcpError::Protocol(_))) - | Ok(Err(e @ AcpError::AgentExited)) => { - tracing::error!( - target: "pool::permission", - "fatal error setting permission mode {wire:?}: {e}" - ); - return Err(e); - } - // Application-level errors — agent is fine, just uses default permission mode. - Ok(Err(e)) => { - tracing::warn!( - target: "pool::permission", - "failed to set permission mode {wire:?}: {e} — falling back to per-tool auto-approval" - ); - } - Err(_) => { - // Outer timeout fired — stream may be in unknown state. - tracing::error!( - target: "pool::permission", - "permission mode set timed out ({PERMISSION_MODE_TIMEOUT:?}) — treating as fatal" - ); - return Err(AcpError::Timeout(PERMISSION_MODE_TIMEOUT)); - } - } - Ok(()) -} - /// Prepend a legacy agent's standing context to a user-message body. /// /// Legacy agents (`protocol_version < 2`) don't receive standing context via diff --git a/crates/buzz-acp/src/pool/permission_mode.rs b/crates/buzz-acp/src/pool/permission_mode.rs new file mode 100644 index 00000000000..7102dbadc6c --- /dev/null +++ b/crates/buzz-acp/src/pool/permission_mode.rs @@ -0,0 +1,200 @@ +use super::{AcpClient, AcpError, PermissionMode, PERMISSION_MODE_TIMEOUT}; + +/// Read both current config options and the legacy mode advertisement. +pub(super) fn agent_supports_mode(session_new_result: &serde_json::Value, mode_wire: &str) -> bool { + session_new_result + .get("modes") + .and_then(|m| m.get("availableModes")) + .and_then(|a| a.as_array()) + .map(|modes| { + modes + .iter() + .any(|m| m.get("id").and_then(|v| v.as_str()) == Some(mode_wire)) + }) + .unwrap_or(false) + || session_new_result + .get("configOptions") + .and_then(|options| options.as_array()) + .is_some_and(|options| { + options.iter().any(|option| { + option + .get("configId") + .or_else(|| option.get("id")) + .and_then(|id| id.as_str()) + == Some("mode") + && option + .get("options") + .and_then(|values| values.as_array()) + .is_some_and(|values| { + values.iter().any(|value| { + value.get("value").and_then(|value| value.as_str()) + == Some(mode_wire) + }) + }) + }) + }) +} + +/// Set the requested mode. Full access requires explicit confirmation; +/// other modes may fall back to per-tool approval on adapter rejection. +/// Transport errors always propagate so the caller can respawn the adapter. +pub(super) async fn apply_permission_mode( + acp: &mut AcpClient, + session_id: &str, + mode: &PermissionMode, +) -> Result<(), AcpError> { + let wire = mode.as_wire_str(); + let result = tokio::time::timeout(PERMISSION_MODE_TIMEOUT, async { + acp.session_set_config_option(session_id, "mode", wire) + .await + }) + .await; + + match result { + Ok(Ok(result)) => { + if *mode == PermissionMode::AgentFullAccess + && !result + .get("configOptions") + .and_then(|v| v.as_array()) + .is_some_and(|options| { + options.iter().any(|option| { + option + .get("configId") + .or_else(|| option.get("id")) + .and_then(|v| v.as_str()) + == Some("mode") + && option.get("currentValue").and_then(|v| v.as_str()) == Some(wire) + }) + }) + { + return Err(AcpError::PermissionModeUnavailable( + "the adapter did not confirm agent-full-access".into(), + )); + } + tracing::info!( + target: "pool::permission", + "applied permission mode {wire:?} on session {session_id}" + ); + } + // Transport-class errors may have corrupted the stdio stream — propagate + // so the caller can respawn the agent. + Ok(Err(e @ AcpError::Io(_))) + | Ok(Err(e @ AcpError::WriteTimeout(_))) + | Ok(Err(e @ AcpError::Timeout(_))) + | Ok(Err(e @ AcpError::Protocol(_))) + | Ok(Err(e @ AcpError::AgentExited)) => { + tracing::error!( + target: "pool::permission", + "fatal error setting permission mode {wire:?}: {e}" + ); + return Err(e); + } + // Application-level errors — agent is fine, just uses default permission mode. + Ok(Err(e)) => { + if *mode == PermissionMode::AgentFullAccess { + return Err(AcpError::PermissionModeUnavailable(format!( + "the adapter rejected agent-full-access: {e}" + ))); + } + tracing::warn!( + target: "pool::permission", + "failed to set permission mode {wire:?}: {e} — falling back to per-tool auto-approval" + ); + } + Err(_) => { + // Outer timeout fired — stream may be in unknown state. + tracing::error!( + target: "pool::permission", + "permission mode set timed out ({PERMISSION_MODE_TIMEOUT:?}) — treating as fatal" + ); + return Err(AcpError::Timeout(PERMISSION_MODE_TIMEOUT)); + } + } + Ok(()) +} + +#[cfg(test)] +mod advertisement_tests { + use super::agent_supports_mode; + use serde_json::json; + + #[test] + fn full_access_accepts_both_config_identifiers_without_legacy_modes() { + for option in [ + json!({"id": "mode", "options": [{"value": "agent-full-access"}]}), + json!({"configId": "mode", "options": [{"value": "agent-full-access"}]}), + ] { + assert!(agent_supports_mode( + &json!({"configOptions": [option]}), + "agent-full-access" + )); + } + } + + #[test] + fn full_access_requires_an_advertised_choice_for_the_mode_option() { + for option in [ + json!({"configId": "mode", "options": [{"value": "default"}]}), + json!({"configId": "model", "id": "mode", "options": [{"value": "agent-full-access"}]}), + json!({"configId": "mode", "currentValue": "agent-full-access"}), + json!({"options": [{"value": "agent-full-access"}]}), + ] { + assert!(!agent_supports_mode( + &json!({"configOptions": [option]}), + "agent-full-access" + )); + } + } +} + +#[cfg(all(test, unix))] +mod tests { + use super::*; + + /// Reply to the single `session/set_config_option` request with `reply`. + async fn apply_with_reply(mode: PermissionMode, reply: &str) -> Result<(), AcpError> { + let script = format!( + "read -r line; printf '%s\\n' '{{\"jsonrpc\":\"2.0\",\"id\":0,{reply}}}'; sleep 5" + ); + let mut acp = AcpClient::spawn("bash", &["-c".to_string(), script], &[], false) + .await + .expect("spawn ACP script"); + let result = apply_permission_mode(&mut acp, "sess-1", &mode).await; + acp.shutdown().await; + result + } + + #[tokio::test] + async fn container_mode_requires_confirmation_and_never_falls_back() { + for confirmed in [ + r#""result":{"configOptions":[{"id":"mode","currentValue":"agent-full-access"}]}"#, + r#""result":{"configOptions":[{"configId":"mode","currentValue":"agent-full-access"}]}"#, + ] { + assert!(apply_with_reply(PermissionMode::AgentFullAccess, confirmed) + .await + .is_ok()); + } + for reply in [ + r#""result":{}"#, + r#""result":{"configOptions":[{"configId":"mode","currentValue":"default"}]}"#, + r#""result":{"configOptions":[{"configId":"model","id":"mode","currentValue":"agent-full-access"}]}"#, + r#""result":{"configOptions":[{"id":"mode","currentValue":"default"}]}"#, + r#""error":{"code":-32602,"message":"unsupported mode"}"#, + ] { + assert!( + matches!( + apply_with_reply(PermissionMode::AgentFullAccess, reply).await, + Err(AcpError::PermissionModeUnavailable(_)) + ), + "{reply}" + ); + } + // Other modes keep the per-tool auto-approval fallback. + assert!(apply_with_reply( + PermissionMode::BypassPermissions, + r#""error":{"code":-32602,"message":"unsupported mode"}"# + ) + .await + .is_ok()); + } +} diff --git a/crates/buzz-acp/src/setup_mode.rs b/crates/buzz-acp/src/setup_mode.rs index 5d08c8b2acb..827d297ac9c 100644 --- a/crates/buzz-acp/src/setup_mode.rs +++ b/crates/buzz-acp/src/setup_mode.rs @@ -428,6 +428,16 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> continue; } + if config + .bound_session + .as_ref() + .is_some_and(|binding| buzz_event.channel_id != binding.channel_id) + { + continue; + } + let channel_is_dm = config.bound_session.is_some() + && crate::is_dm_channel(buzz_event.channel_id, &channel_info).await; + // Apply the same author gate as normal mode so the nudge only goes // to authors the real agent would have answered. Same DM hardening: // in DMs only owner/siblings get a nudge (fail-closed on unknown type). @@ -454,6 +464,8 @@ pub(crate) async fn run_setup_listener(config: Config, payload: SetupPayload) -> &rest_client, &config.keys, &payload, + config.bound_session.as_ref(), + channel_is_dm, ) .await { @@ -492,6 +504,8 @@ async fn nudge_authorized_event( rest_client: &relay::RestClient, keys: &nostr::Keys, payload: &SetupPayload, + bound_session: Option<&crate::config::BoundSession>, + channel_is_dm: bool, ) -> bool { let (buzz_event, effective_author) = authorized_event.into_parts(); @@ -501,7 +515,7 @@ async fn nudge_authorized_event( .await .is_some(); - if !should_nudge_for_event(buzz_event.event.id, filter_matched, nudged_event_ids) { + if !filter_matched || nudged_event_ids.contains(&buzz_event.event.id) { return false; } @@ -511,6 +525,26 @@ async fn nudge_authorized_event( crate::edit_routing::resolve_edit(&buzz_event.event, buzz_event.channel_id, rest_client) .await; + // An unresolved edit would publish at top level. A bound workspace must + // keep silent until the original's canonical thread is known. + if let Some(binding) = bound_session { + let unresolved_edit = + crate::queue::edit_target_id(&buzz_event.event).is_some() && edit.is_none(); + let scope = crate::scope::SessionScope::derive_routed( + crate::scope::SessionPolicy::Thread, + buzz_event.channel_id, + channel_is_dm, + &buzz_event.event, + edit.as_ref(), + ); + if unresolved_edit || !binding.accepts(&scope) { + return false; + } + } + if !should_nudge_for_event(buzz_event.event.id, true, nudged_event_ids) { + return false; + } + // Build and publish the setup nudge. if let Err(e) = publish_setup_nudge( publisher, @@ -866,6 +900,8 @@ mod tests { &rest_client, &agent_keys, &payload, + None, + false, ) .await ); @@ -972,6 +1008,8 @@ mod tests { &query_rest, &agent_keys, &payload, + None, + false, ) .await ); @@ -1496,3 +1534,7 @@ mod tests { ); } } + +#[cfg(test)] +#[path = "setup_mode_bound_tests.rs"] +mod bound_session_tests; diff --git a/crates/buzz-acp/src/setup_mode_bound_tests.rs b/crates/buzz-acp/src/setup_mode_bound_tests.rs new file mode 100644 index 00000000000..9b152ec4f9a --- /dev/null +++ b/crates/buzz-acp/src/setup_mode_bound_tests.rs @@ -0,0 +1,149 @@ +use super::*; + +#[tokio::test] +async fn bound_setup_nudges_stay_in_the_canonical_owner_thread() { + for (case, expected) in [ + ("top-level", true), + ("reply", true), + ("foreign-thread", false), + ("foreign-channel", false), + ("dm", false), + ("threaded-edit", true), + ("top-level-edit", true), + ("foreign-thread-edit", false), + ("unresolved-edit", false), + ] { + let agent_keys = nostr::Keys::generate(); + let agent = agent_keys.public_key().to_hex(); + let owner_keys = nostr::Keys::generate(); + let owner = owner_keys.public_key().to_hex(); + let channel_id = Uuid::new_v4(); + let root = "11".repeat(32); + let foreign_root = "22".repeat(32); + let mut tags = vec![ + nostr::Tag::parse(["p", &agent]).unwrap(), + nostr::Tag::parse(["h", &channel_id.to_string()]).unwrap(), + ]; + if !matches!(case, "top-level" | "top-level-edit" | "unresolved-edit") { + let thread = if case.starts_with("foreign-thread") { + &foreign_root + } else { + &root + }; + tags.push(nostr::Tag::parse(["e", thread, "", "reply"]).unwrap()); + } + let original = nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_STREAM_MESSAGE as u16), + "owner asks for setup", + ) + .tags(tags) + .sign_with_keys(&owner_keys) + .unwrap(); + let event = if case.ends_with("edit") { + nostr::EventBuilder::new( + nostr::Kind::Custom(KIND_STREAM_MESSAGE_EDIT as u16), + "owner adds the mention", + ) + .tags([ + nostr::Tag::parse(["e", &original.id.to_hex()]).unwrap(), + nostr::Tag::parse(["p", &agent]).unwrap(), + nostr::Tag::parse(["h", &channel_id.to_string()]).unwrap(), + ]) + .sign_with_keys(&owner_keys) + .unwrap() + } else { + original.clone() + }; + let binding = crate::config::BoundSession { + channel_id: if case == "foreign-channel" { + Uuid::new_v4() + } else { + channel_id + }, + thread_root: if matches!(case, "top-level" | "top-level-edit" | "unresolved-edit") { + original.id.to_hex() + } else { + root + }, + }; + let (gate_rest, gate_server) = + crate::author_gate_tests::nip11_server(serde_json::json!({})).await; + let response = if case == "unresolved-edit" { + serde_json::json!([]) + } else { + serde_json::json!([original]) + }; + let (query_rest, query_server) = crate::author_gate_tests::nip11_server(response).await; + let channel_info = crate::pool::ChannelInfoResolver::new( + std::collections::HashMap::from([( + channel_id, + relay::ChannelInfo { + name: "bound setup".into(), + channel_type: if case == "dm" { "dm" } else { "stream" }.into(), + description: None, + }, + )]), + gate_rest.clone(), + ); + let mut gate = InboundAuthorGate::connect(&gate_rest, &agent, "bound setup test").await; + assert!(setup_listener_admits(&event, &agent)); + let event_id = event.id; + let authorized = authorize_setup_listener_event( + &mut gate, + relay::BuzzEvent { + connection_generation: 0, + channel_id, + event, + }, + &crate::config::RespondTo::Owner, + &HashSet::new(), + &OwnerCache::new(Some(owner)), + &channel_info, + &gate_rest, + ) + .await + .expect("owner is authorized"); + let (publisher, mut published) = RelayEventPublisher::test_pair(); + let payload = SetupPayload { + agent_name: "Bound agent".into(), + agent_pubkey: agent.clone(), + requirements: vec![], + }; + let mut nudged = HashSet::new(); + let admitted = nudge_authorized_event( + authorized, + &[mentions_rule(crate::config::default_mention_kinds())], + &agent, + &mut nudged, + &publisher, + &query_rest, + &agent_keys, + &payload, + Some(&binding), + crate::is_dm_channel(channel_id, &channel_info).await, + ) + .await; + assert_eq!(admitted, expected, "{case}"); + assert_eq!(nudged.contains(&event_id), expected, "{case}"); + if expected { + let nudge = tokio::time::timeout(std::time::Duration::from_secs(5), published.recv()) + .await + .expect("matching setup nudge reaches the publisher") + .expect("publisher stays open"); + let scope = crate::scope::SessionScope::derive( + crate::scope::SessionPolicy::Thread, + channel_id, + false, + &nudge, + ); + assert!(binding.accepts(&scope), "{case}"); + } else { + assert!( + published.try_recv().is_err(), + "{case}: no out-of-scope nudge" + ); + } + gate_server.abort(); + query_server.abort(); + } +}