feat(agents): admit shared turns through the running harness

crates/buzz-acp/Cargo.toml

@@ -19,6 +19,8 @@ path = "src/main.rs"
 # Internal
 buzz-core = { workspace = true }
 buzz-sdk = { workspace = true }
+buzz-session-control = { path = "../buzz-session-control" }
+buzz-ws-client = { workspace = true }
 buzz-persona = { path = "../buzz-persona" }
 
 # Git multicall helpers and ephemeral key ownership

crates/buzz-acp/src/acp.rs

@@ -113,6 +113,9 @@ pub enum AcpError {
     #[error("Protocol error: {0}")]
     Protocol(String),
 
+    #[error("Shared task permission: {0}")]
+    TaskPermission(String),
+
     #[error("Agent reported error (code {code}): {message}")]
     AgentError { code: i64, message: String },
 

crates/buzz-acp/src/config.rs

@@ -12,7 +12,7 @@ use buzz_core::kind::{
 };
 use clap::Parser;
 use clap::ValueEnum;
-use nostr::Keys;
+use nostr::{Keys, PublicKey};
 use thiserror::Error;
 use url::Url;
 use uuid::Uuid;
@@ -594,6 +594,10 @@ pub struct CliArgs {
     #[arg(long, env = "BUZZ_ACP_BOUND_SESSION", value_parser = parse_bound_session)]
     pub bound_session: Option<BoundSession>,
 
+    /// Session manager that authorizes shared channel turns. Requires a bound thread.
+    #[arg(long, env = "BUZZ_ACP_SHARING_MANAGER")]
+    pub sharing_manager: Option<PublicKey>,
+
     /// Save and verify a private workspace checkpoint after draining the agents.
     #[arg(long, env = "BUZZ_ACP_CHECKPOINT_ON_STOP", default_value_t = false)]
     pub checkpoint_on_stop: bool,
@@ -716,6 +720,8 @@ pub struct Config {
     /// Whether to emit bounded JSONL activity for platform collection.
     pub activity_log: bool,
     pub bound_session: Option<BoundSession>,
+    /// Manager for fresh shared-task admission; absent preserves owner-only input.
+    pub sharing_manager: Option<PublicKey>,
     /// Whether to run `buzz-agent-checkpoint save` after draining on stop and
     /// fail shutdown when it does not produce a verified receipt.
     pub checkpoint_on_stop: bool,
@@ -1042,6 +1048,16 @@ impl Config {
     /// `propagate_legacy_env_vars()` before the tokio runtime starts (in the
     /// sync `fn main()` wrapper) — see Rust 2024 edition safety.
     pub fn from_args(mut args: CliArgs) -> Result<Self, ConfigError> {
+        if args.sharing_manager.is_some()
+            && !matches!(
+                args.bound_session.as_ref(),
+                Some(BoundSession::Thread { .. })
+            )
+        {
+            return Err(ConfigError::ConfigFile(
+                "shared tasks require a bound channel thread".into(),
+            ));
+        }
         if let Some(binding) = &args.bound_session {
             let direct_message = matches!(binding, BoundSession::DirectMessage { .. });
             let subscription = if direct_message {
@@ -1333,7 +1349,9 @@ impl Config {
             max_turns_per_session: args.max_turns_per_session,
             presence_enabled: !args.no_presence,
             typing_enabled: !args.no_typing,
-            memory_enabled: args.memory && !args.no_memory,
+            // A later sharing grant must never expose personal core memory
+            // already injected into an owner-started channel session.
+            memory_enabled: args.memory && !args.no_memory && args.sharing_manager.is_none(),
             model,
             effort_level: args.effort_level,
             session_title: args
@@ -1349,6 +1367,7 @@ impl Config {
             relay_observer: args.relay_observer,
             activity_log: args.activity_log,
             bound_session: args.bound_session,
+            sharing_manager: args.sharing_manager,
             checkpoint_on_stop: args.checkpoint_on_stop,
             checkpoint_after_turn_secs: args.checkpoint_after_turn_secs,
             exit_after_inactivity_secs: args.exit_after_inactivity,
@@ -1801,6 +1820,13 @@ mod tests {
             .unwrap()
         };
         assert!(Config::from_args(args()).is_ok());
+        let mut private = args();
+        private.memory = true;
+        assert!(Config::from_args(private).unwrap().memory_enabled);
+        let mut shared = args();
+        shared.sharing_manager = Some(Keys::generate().public_key());
+        let error = Config::from_args(shared).err().unwrap().to_string();
+        assert!(error.contains("shared tasks require a bound channel thread"));
         let mut mentions = args();
         mentions.subscribe = SubscribeMode::Mentions;
         assert!(Config::from_args(mentions).is_err());
@@ -1899,6 +1925,18 @@ mod tests {
             "0",
         ];
         assert!(Config::from_args(CliArgs::try_parse_from(base).unwrap()).is_ok());
+        let mut shared = CliArgs::try_parse_from(base).unwrap();
+        shared.sharing_manager = Some(Keys::generate().public_key());
+        let configured = Config::from_args(shared).unwrap();
+        assert!(
+            !configured.memory_enabled,
+            "shared channels must not preload personal core memory"
+        );
+        assert_eq!(configured.respond_to, RespondTo::Owner);
+        let mut unbound = CliArgs::try_parse_from(base).unwrap();
+        unbound.sharing_manager = Some(Keys::generate().public_key());
+        unbound.bound_session = None;
+        assert!(Config::from_args(unbound).is_err());
         for (variation, flag) in [
             (0, "--respond-to owner"),
             (1, "--no-mention-filter"),
@@ -1969,6 +2007,7 @@ mod tests {
             relay_observer: false,
             activity_log: false,
             bound_session: None,
+            sharing_manager: None,
             checkpoint_on_stop: false,
             checkpoint_after_turn_secs: 0,
             exit_after_inactivity_secs: 0,

crates/buzz-acp/src/lib.rs

@@ -22,6 +22,8 @@ mod recovery_wake;
 mod relay;
 mod run_task;
 mod runtime;
+/// Shared channel task admission through the existing encrypted control plane.
+pub mod sharing;
 use runtime::{AgentRuntime, PoolStartup, SessionMode};
 mod scope;
 mod setup_mode;
@@ -622,6 +624,7 @@ mod inbound_author_gate {
         relay_self: Option<String>,
         // None means no authoritative NIP-11 result yet, including at startup.
         refreshed_generation: Option<u64>,
+        sharing: Option<std::sync::Arc<crate::sharing::Client>>,
     }
 
     pub(crate) fn refresh_needed(refreshed_generation: Option<u64>, event_generation: u64) -> bool {
@@ -640,6 +643,7 @@ mod inbound_author_gate {
                 agent_pubkey_hex: agent_pubkey_hex.to_string(),
                 relay_self,
                 refreshed_generation: completed.then_some(0),
+                sharing: None,
             }
         }
 
@@ -727,6 +731,73 @@ mod inbound_author_gate {
             }
         }
 
+        pub(crate) fn enable_sharing(
+            &mut self,
+            sharing: Option<std::sync::Arc<crate::sharing::Client>>,
+        ) {
+            self.sharing = sharing;
+        }
+
+        pub(crate) async fn authorize_shared_listener_event(
+            &mut self,
+            buzz_event: relay::BuzzEvent,
+            respond_to: &RespondTo,
+            allowlist: &HashSet<String>,
+            owner_cache: &OwnerCache,
+            channel_info: &pool::ChannelInfoResolver,
+            rest_client: &relay::RestClient,
+        ) -> Result<Option<AuthorizedListenerEvent>, crate::sharing::Error> {
+            if self.sharing.is_none() {
+                return Ok(self
+                    .authorize_listener_event(
+                        buzz_event,
+                        respond_to,
+                        allowlist,
+                        owner_cache,
+                        channel_info,
+                        rest_client,
+                    )
+                    .await);
+            }
+            let decision = self
+                .evaluate_listener_event(
+                    &buzz_event,
+                    respond_to,
+                    allowlist,
+                    owner_cache,
+                    channel_info,
+                    rest_client,
+                )
+                .await;
+            if decision.allowed {
+                return Ok(Some(AuthorizedListenerEvent {
+                    buzz_event,
+                    effective_author: decision.effective_author,
+                }));
+            }
+            if matches!(respond_to, RespondTo::Owner) && !decision.is_dm {
+                if let Some(sharing) = &self.sharing {
+                    if sharing.accepts(&buzz_event.event) {
+                        let refusal = match sharing.authorize(&buzz_event.event).await {
+                            Ok(None) => {
+                                return Ok(Some(AuthorizedListenerEvent {
+                                    buzz_event,
+                                    effective_author: decision.effective_author,
+                                }))
+                            }
+                            Ok(Some(reason)) => reason,
+                            Err(error) => error.to_string(),
+                        };
+                        // Persist a visible refusal before handling a transient
+                        // outage; never abandon the task or kill an in-flight turn.
+                        sharing.refuse(&buzz_event.event, &refusal).await?;
+                        return Ok(None);
+                    }
+                }
+            }
+            Ok(None)
+        }
+
         pub(crate) async fn authorize_listener_event(
             &mut self,
             buzz_event: relay::BuzzEvent,
@@ -970,9 +1041,9 @@ async fn authorize_normal_listener_event(
     owner_cache: &OwnerCache,
     channel_info: &pool::ChannelInfoResolver,
     rest_client: &relay::RestClient,
-) -> Option<AuthorizedListenerEvent> {
+) -> std::result::Result<Option<AuthorizedListenerEvent>, sharing::Error> {
     author_gate
-        .authorize_listener_event(
+        .authorize_shared_listener_event(
             buzz_event,
             respond_to,
             allowlist,
@@ -3139,6 +3210,11 @@ async fn run_harness(
         channel_info_map,
         SessionMode::Conversation,
     )?);
+    author_gate_ctx.enable_sharing(ctx.sharing.clone());
+    let _shared_notice_worker = ctx
+        .sharing
+        .as_ref()
+        .map(|client| client.clone().start_notices());
 
     if !config.memory_enabled {
         tracing::info!(
@@ -3838,7 +3914,7 @@ async fn run_harness(
                                 &ctx.channel_info,
                                 &ctx.rest_client,
                             )
-                            .await
+                            .await?
                             else {
                                 continue;
                             };
@@ -7937,18 +8013,17 @@ mod author_gate_tests {
             event: relay_signed_workflow_dispatch(relay_keys, workflow_owner, &agent),
         };
         let authorized = match listener {
-            ListenerBoundary::Normal => {
-                authorize_normal_listener_event(
-                    &mut gate,
-                    event,
-                    &respond_to,
-                    &allowlist,
-                    &owner_cache,
-                    &channel_info,
-                    &rest_client,
-                )
-                .await
-            }
+            ListenerBoundary::Normal => authorize_normal_listener_event(
+                &mut gate,
+                event,
+                &respond_to,
+                &allowlist,
+                &owner_cache,
+                &channel_info,
+                &rest_client,
+            )
+            .await
+            .unwrap(),
             ListenerBoundary::Setup => {
                 setup_mode::authorize_setup_listener_event(
                     &mut gate,
@@ -10609,6 +10684,7 @@ mod build_mcp_servers_tests {
             relay_observer: false,
             activity_log: false,
             bound_session: None,
+            sharing_manager: None,
             checkpoint_on_stop: false,
             checkpoint_after_turn_secs: 0,
             exit_after_inactivity_secs: 0,
@@ -11710,6 +11786,7 @@ mod error_outcome_emission_tests {
             relay_observer: false,
             activity_log: false,
             bound_session: None,
+            sharing_manager: None,
             checkpoint_on_stop: false,
             checkpoint_after_turn_secs: 0,
             exit_after_inactivity_secs: 0,

crates/buzz-acp/src/pool.rs

@@ -934,6 +934,8 @@ pub struct PromptContext {
     /// `<core-memory>` section. On by default; disabled via
     /// `--no-memory` / `BUZZ_ACP_NO_MEMORY`.
     pub memory_enabled: bool,
+    /// Fresh authorization for bound shared-channel tasks.
+    pub sharing: Option<Arc<crate::sharing::Client>>,
     /// Harness identity string for NIP-AM `harness` field. Derived from the
     /// configured `agent_command` at startup (e.g. `"goose"`, `"buzz-agent"`).
     pub harness_name: String,
@@ -2510,6 +2512,39 @@ pub async fn run_prompt_task(
         .unwrap_or_default();
     let _reaction_guard = ReactionGuard::new(ctx.rest_client.clone(), reaction_ids.clone());
 
+    // Queueing may outlive a grant. Recheck before session creation, memory
+    // fetches, or any ACP prompt; the listener separately guards native steering.
+    if let (Some(sharing), Some(pending)) = (&ctx.sharing, &batch) {
+        let mut denied = None;
+        for item in &pending.events {
+            if Some(item.event.pubkey) == ctx.agent_owner_pubkey {
+                continue;
+            }
+            match sharing.authorize(&item.event).await {
+                Ok(None) => {}
+                Ok(Some(reason)) => {
+                    denied = Some(reason);
+                    break;
+                }
+                Err(error) => {
+                    denied = Some(error.to_string());
+                    break;
+                }
+            }
+        }
+        if let Some(reason) = denied {
+            send_prompt_result(
+                &result_tx,
+                &turn_id,
+                agent,
+                source,
+                PromptOutcome::Error(AcpError::TaskPermission(reason)),
+                requeue_batch_if_queue(&ctx, batch),
+            );
+            return;
+        }
+    }
+
     // Resolve project authority exactly once, before any ACP session creation or
     // initial-message delivery. An indeterminate result is a local relay-state
     // outcome: fail closed and preserve the batch without poisoning the healthy
@@ -10490,6 +10525,7 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             agent_keys: agent_keys.clone(),
             agent_owner_pubkey: owner_pubkey,
             memory_enabled: false,
+            sharing: None,
             harness_name: "goose".to_string(),
             relay_url: "ws://127.0.0.1:3000".to_string(),
         }

crates/buzz-acp/src/runtime.rs

@@ -82,6 +82,8 @@ fn make_prompt_context(
 ) -> Result<PromptContext> {
     let base_prompt_content = config.base_prompt_content.as_ref();
     let cwd = current_working_directory()?;
+    let sharing =
+        crate::sharing::Client::from_config(config, &rest_client)?.map(std::sync::Arc::new);
     Ok(PromptContext {
         mcp_servers: build_mcp_servers(config),
         initial_message: config.initial_message.clone(),
@@ -91,7 +93,11 @@ fn make_prompt_context(
         dedup_mode: config.dedup_mode,
         system_prompt: config.system_prompt.clone(),
         session_title: config.session_title.clone(),
-        team_instructions: config.team_instructions.clone(),
+        team_instructions: if sharing.is_some() {
+            Some(format!("{}\n\nTo delegate to another shared agent, use `buzz agents handoff <agent-pubkey> --from <incoming-event-id> --message '<task>'`. Use the current incoming task's event ID. Delegate specific work or reviews; avoid acknowledgement loops. Sharing preserves the original requester and does not confer owner lifecycle rights.",config.team_instructions.as_deref().unwrap_or("")))
+        } else {
+            config.team_instructions.clone()
+        },
         base_prompt: if config.no_base_prompt {
             None
         } else {
@@ -119,6 +125,7 @@ fn make_prompt_context(
             .as_deref()
             .and_then(|hex| nostr::PublicKey::from_hex(hex).ok()),
         memory_enabled: config.memory_enabled,
+        sharing,
         harness_name: crate::config::normalize_agent_command_identity(&config.agent_command),
         relay_url: config.relay_url.clone(),
     })