From 4b813c34c1bb6e41e5e7c75e59403e5899b3a6ff Mon Sep 17 00:00:00 2001 From: Tee8z Date: Mon, 19 Oct 2026 14:57:09 -0400 Subject: [PATCH] feat(acp): bind the harness to one owner thread Add a strict owner respond-to mode that admits only the human owner's signed events, in DMs and channels, without sibling or workflow attribution. Add an optional bound session that restricts ingress and control commands to one channel and canonical thread root, and refuse bound configurations that allow non-owner, non-mention, non-thread, or autonomous prompts. Add the agent-full-access permission mode for externally isolated containers. The adapter must advertise and confirm it; unlike other modes it never falls back to per-tool approval. Signed-off-by: Tee8z --- crates/buzz-acp/README.md | 34 +- crates/buzz-acp/src/acp.rs | 5 + crates/buzz-acp/src/config.rs | 198 +++++++- crates/buzz-acp/src/lib.rs | 459 +++++++++++++++++- crates/buzz-acp/src/pool.rs | 85 +--- crates/buzz-acp/src/pool/permission_mode.rs | 200 ++++++++ crates/buzz-acp/src/setup_mode.rs | 44 +- crates/buzz-acp/src/setup_mode_bound_tests.rs | 149 ++++++ 8 files changed, 1070 insertions(+), 104 deletions(-) create mode 100644 crates/buzz-acp/src/pool/permission_mode.rs create mode 100644 crates/buzz-acp/src/setup_mode_bound_tests.rs 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(); + } +}