feat(desktop): surface tool-permission asks as thread-reply cards (#4938)

crates/buzz-acp/src/observer.rs

@@ -30,6 +30,33 @@ pub struct ObserverContext {
     pub started_at: Option<String>,
 }
 
+/// Authorization envelope attached to permission-related observer events.
+///
+/// Present on the single `acp_read` emitted after a permission request passes
+/// the admission preflight, and on the corresponding `acp_write` after the
+/// response is confirmed written.
+#[derive(Clone, Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct AuthorizationEnvelope {
+    /// Single-use nonce bound to this request — delivered to the desktop and
+    /// consumed exactly once when the owner makes a decision.
+    pub request_nonce: String,
+    /// `true` when the owner can take action (policy=ask, preflight passed,
+    /// owner/observer available). `false` for auto-deny / fail-closed paths.
+    pub actionable: bool,
+    /// Human-readable reason when `actionable` is `false`.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub reason: Option<String>,
+    /// Wire card-expiry (unix seconds) for an actionable card — the same value
+    /// stored in the kind-9 sentinel. The desktop bounds its
+    /// retransmit-until-acked loop by this deadline, so a decision published
+    /// while the harness socket is down keeps being resent until the card
+    /// expires (never past it). `None` on non-actionable / already-resolved
+    /// frames, where no owner decision is awaited.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub expires_at: Option<u64>,
+}
+
 /// Handle used by the harness to publish local observer events.
 #[derive(Clone)]
 pub struct ObserverHandle {
@@ -54,7 +81,7 @@ fn new_observer_handle() -> ObserverHandle {
 }
 
 /// Event delivered through the in-process observer bus.
-#[derive(Clone, Serialize)]
+#[derive(Clone, Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ObserverEvent {
     /// Monotonic process-local sequence number.
@@ -74,6 +101,12 @@ pub struct ObserverEvent {
     /// RFC3339 timestamp at which the current turn began, when known.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub started_at: Option<String>,
+    /// Authorization envelope — present only on permission `acp_read` /
+    /// `acp_write` frames, and on the observer-only `permission_terminal` frame
+    /// (which carries `reason = "uncertain"` and is never sent on the ACP wire).
+    /// `None` on all other event kinds.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub authorization: Option<AuthorizationEnvelope>,
     /// Raw or semantic event payload.
     pub payload: serde_json::Value,
 }
@@ -107,6 +140,31 @@ impl ObserverHandle {
         agent_index: Option<usize>,
         context: &ObserverContext,
         payload: serde_json::Value,
+    ) {
+        self.emit_inner(kind, agent_index, context, None, payload);
+    }
+
+    /// Emit a local observer event with an authorization envelope.
+    ///
+    /// Used for permission `acp_read` and `acp_write` frames.
+    pub fn emit_authorized(
+        &self,
+        kind: impl Into<String>,
+        agent_index: Option<usize>,
+        context: &ObserverContext,
+        authorization: AuthorizationEnvelope,
+        payload: serde_json::Value,
+    ) {
+        self.emit_inner(kind, agent_index, context, Some(authorization), payload);
+    }
+
+    fn emit_inner(
+        &self,
+        kind: impl Into<String>,
+        agent_index: Option<usize>,
+        context: &ObserverContext,
+        authorization: Option<AuthorizationEnvelope>,
+        payload: serde_json::Value,
     ) {
         let event = ObserverEvent {
             seq: self.inner.seq.fetch_add(1, Ordering::Relaxed),
@@ -117,6 +175,7 @@ impl ObserverHandle {
             session_id: context.session_id.clone(),
             turn_id: context.turn_id.clone(),
             started_at: context.started_at.clone(),
+            authorization,
             payload,
         };
 

crates/buzz-acp/src/pool.rs

@@ -34,7 +34,9 @@ use crate::acp::{
     model_in_catalog, resolve_model_switch_method, AcpClient, AcpError, EnvVar, McpServer,
     ModelSwitchMethod, StopReason, SystemPromptTransport,
 };
-use crate::config::{compose_scoped_session_title, DedupMode, PermissionMode};
+use crate::config::{
+    compose_scoped_session_title, DedupMode, PermissionMode, ResolvedPermissionConfig,
+};
 use crate::observer;
 use crate::prompt_project::{pick_authoritative_project_home, PromptProjectInfo};
 use crate::queue::{
@@ -79,6 +81,13 @@ pub struct TaskMeta {
     /// tasks only — all prompt tasks install a steer channel regardless
     /// of the agent's name.
     pub steer_tx: Option<tokio::sync::mpsc::Sender<SteerRequest>>,
+    /// Permission decision channel — delivers `permission_decision` control
+    /// frames from the observer dispatch loop into the read loop's decision
+    /// arm. `None` until the first `ask`-policy permission request arrives
+    /// (installed per-session by the pool dispatch path). Cloned from the
+    /// sender end of the channel installed on `AcpClient` via
+    /// `install_permission_decision_rx`.
+    pub permission_decision_tx: Option<tokio::sync::mpsc::Sender<crate::acp::PermissionDecision>>,
     /// Successful non-cancelling steers acknowledged while this task owned the
     /// live session. The session ID prevents a late ack from contaminating a
     /// replacement session after task return.
@@ -330,6 +339,15 @@ pub struct AgentPool {
     result_rx: mpsc::UnboundedReceiver<PromptResult>,
     pub join_set: JoinSet<()>,
     task_map: HashMap<tokio::task::Id, TaskMeta>,
+    /// Nonces of permission decisions already forwarded to a read loop, with the
+    /// instant each was recorded. The desktop retransmits a decision until it
+    /// sees a `control_result`; a copy that arrives after the read loop applied
+    /// the decision and its task ended would otherwise get `no_active_turn` /
+    /// `channel_closed` and flip the resolved card to failed. Recording the
+    /// nonce on first delivery lets [`Self::was_recently_decided`] recognize
+    /// such a late duplicate and ack it success-shaped instead. Pruned to
+    /// [`DECIDED_NONCE_RETENTION`] (≥ the card's max expiry) on every write.
+    recently_decided: HashMap<String, tokio::time::Instant>,
     /// Authoritative directory of which worker most recently owned each session
     /// scope's provider session. Survives while a worker is checked out (its
     /// `SessionState` is invisible to the pool then), so a busy owner does not
@@ -339,6 +357,11 @@ pub struct AgentPool {
     session_owners: HashMap<SessionScope, usize>,
 }
 
+/// Retention for [`AgentPool::recently_decided`]. Matches the maximum card
+/// expiry (`PERMISSION_ASK_TIMEOUT_SECS`) so a nonce stays recognized for as
+/// long as the desktop could still be retransmitting it, then is reclaimed.
+const DECIDED_NONCE_RETENTION: Duration = Duration::from_secs(300);
+
 /// Result returned by a completed prompt task.
 pub struct PromptResult {
     pub agent: OwnedAgent,
@@ -785,8 +808,8 @@ pub struct PromptContext {
     pub context_message_limit: u32,
     /// Max turns per session before proactive rotation. 0 = disabled.
     pub max_turns_per_session: u32,
-    /// Permission mode to apply after session creation. `Default` = skip.
-    pub permission_mode: PermissionMode,
+    /// Resolved permission configuration — policy, effective ACP mode, and how to transmit.
+    pub permission_config: ResolvedPermissionConfig,
     /// Agent identity — used to derive the NIP-AE conversation key at
     /// session creation for core injection.
     pub agent_keys: nostr::Keys,
@@ -806,6 +829,11 @@ pub struct PromptContext {
     /// the desktop keys per (agent, relay) pair, e.g. `session_config_captured`,
     /// mirroring the `managed_agent_runtime_lifecycle` frames.
     pub relay_url: String,
+    /// Publisher for kind-9 sentinel cards and kind-40003 edits.
+    /// When set, `run_prompt_task` wires it into `AcpClient` so permission
+    /// cards appear in the channel thread. `None` disables sentinel publishing
+    /// (observer feed path remains).
+    pub relay_event_publisher: Option<crate::relay::RelayEventPublisher>,
 }
 
 impl AgentPool {
@@ -823,10 +851,31 @@ impl AgentPool {
             result_rx,
             join_set: JoinSet::new(),
             task_map: HashMap::new(),
+            recently_decided: HashMap::new(),
             session_owners: HashMap::new(),
         }
     }
 
+    /// Record a permission-decision nonce as delivered to a read loop and prune
+    /// entries older than [`DECIDED_NONCE_RETENTION`]. Called when a decision is
+    /// first forwarded so a later retransmit of the same nonce is recognized.
+    pub fn record_permission_decision(&mut self, nonce: &str) {
+        let now = tokio::time::Instant::now();
+        self.recently_decided
+            .retain(|_, at| now.duration_since(*at) < DECIDED_NONCE_RETENTION);
+        self.recently_decided.insert(nonce.to_string(), now);
+    }
+
+    /// Whether `nonce` was recently forwarded to a read loop and is still within
+    /// the retention window. A late retransmit that matches is a duplicate the
+    /// harness has already forwarded — the caller acks it success-shaped rather
+    /// than failing the resolved card.
+    pub fn was_recently_decided(&self, nonce: &str) -> bool {
+        self.recently_decided.get(nonce).is_some_and(|at| {
+            tokio::time::Instant::now().duration_since(*at) < DECIDED_NONCE_RETENTION
+        })
+    }
+
     /// Record which worker is handling `scope` so a later dispatch can detect a
     /// busy owner and avoid opening a duplicate session on another worker.
     pub fn record_scope_owner(&mut self, scope: SessionScope, agent_index: usize) {
@@ -1515,14 +1564,20 @@ async fn create_session_and_apply_model(
         }),
     );
 
-    // Apply permission mode if not the agent's built-in default AND the agent
-    // 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.is_default()
-        && agent_supports_mode(&resp.raw, ctx.permission_mode.as_wire_str())
+    // Apply permission mode whenever the agent advertises it (including `default`).
+    // The `transmit_mode` flag handles any future cases where transmission should be skipped.
+    if ctx.permission_config.transmit_mode
+        && agent_supports_mode(
+            &resp.raw,
+            ctx.permission_config.effective_mode.as_wire_str(),
+        )
     {
-        apply_permission_mode(&mut agent.acp, &resp.session_id, &ctx.permission_mode).await?;
+        apply_permission_mode(
+            &mut agent.acp,
+            &resp.session_id,
+            &ctx.permission_config.effective_mode,
+        )
+        .await?;
     }
 
     Ok(resp.session_id)
@@ -1769,11 +1824,7 @@ 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
+/// Check whether 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 {
@@ -1789,7 +1840,11 @@ fn agent_supports_mode(session_new_result: &serde_json::Value, mode_wire: &str)
         .unwrap_or(false)
 }
 
-/// per-tool auto-approval in `handle_permission_request`.
+/// 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 mode, and any interactive permission request is rejected by
+/// `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.
@@ -1829,7 +1884,7 @@ async fn apply_permission_mode(
         Ok(Err(e)) => {
             tracing::warn!(
                 target: "pool::permission",
-                "failed to set permission mode {wire:?}: {e} — falling back to per-tool auto-approval"
+                "failed to set permission mode {wire:?}: {e} — falling back to per-tool rejection"
             );
         }
         Err(_) => {
@@ -2052,6 +2107,44 @@ pub async fn run_prompt_task(
         turn_id.clone(),
         turn_started_at.clone(),
     ));
+
+    // Wire permission configuration and owner-knowledge into the ACP client so
+    // `handle_permission_request` can evaluate the ask availability gate. These
+    // values come from `PromptContext` (resolved once at startup from CLI args and
+    // desktop-injected env vars) and are idempotent to re-apply across turns.
+    agent
+        .acp
+        .set_permission_config(ctx.permission_config.clone());
+    agent
+        .acp
+        .set_owner_pubkey_known(ctx.agent_owner_pubkey.is_some());
+
+    // Wire sentinel card publisher, agent signing keys, owner pubkey, and
+    // per-turn context for D7-final admission and kind-9/40003 publishing.
+    if let Some(publisher) = ctx.relay_event_publisher.clone() {
+        agent
+            .acp
+            .set_relay_publisher(publisher, ctx.agent_keys.clone());
+    }
+    agent
+        .acp
+        .set_agent_owner_pubkey_hex(ctx.agent_owner_pubkey.as_ref().map(|pk| pk.to_hex()));
+    // D7-final: record the turn initiator from the first event in the batch.
+    let turn_initiator = batch
+        .as_ref()
+        .and_then(|b| b.events.first())
+        .map(|be| be.event.pubkey);
+    agent.acp.set_turn_initiator_pubkey(turn_initiator);
+    // Sentinel routing: channel UUID and reply anchor from batch.
+    let batch_channel_id = batch.as_ref().map(|b| b.channel_id);
+    let thread_reply_event_id = batch
+        .as_ref()
+        .and_then(|b| b.events.first())
+        .map(|be| be.event.id.to_hex());
+    agent
+        .acp
+        .set_turn_channel_context(batch_channel_id, thread_reply_event_id);
+
     let triggering_event_ids: Vec<String> = batch
         .as_ref()
         .map(|b| b.events.iter().map(|be| be.event.id.to_hex()).collect())
@@ -5147,6 +5240,60 @@ mod tests {
         }
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn recently_decided_recognizes_a_nonce_within_retention() {
+        let mut pool = AgentPool::from_slots(vec![None]);
+        assert!(
+            !pool.was_recently_decided("n1"),
+            "unknown nonce is not recently decided"
+        );
+        pool.record_permission_decision("n1");
+        assert!(
+            pool.was_recently_decided("n1"),
+            "just-recorded nonce is recognized"
+        );
+        assert!(
+            !pool.was_recently_decided("n2"),
+            "a different nonce is not recognized"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn recently_decided_expires_after_retention_window() {
+        let mut pool = AgentPool::from_slots(vec![None]);
+        pool.record_permission_decision("n1");
+        // Just inside the window: still recognized.
+        tokio::time::advance(DECIDED_NONCE_RETENTION - Duration::from_secs(1)).await;
+        assert!(
+            pool.was_recently_decided("n1"),
+            "nonce inside retention is still recognized"
+        );
+        // Past the window: no longer recognized (bounds the set's growth and
+        // stops acking retransmits for cards that have long since expired).
+        tokio::time::advance(Duration::from_secs(2)).await;
+        assert!(
+            !pool.was_recently_decided("n1"),
+            "nonce past retention is forgotten"
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn recording_prunes_entries_past_retention() {
+        let mut pool = AgentPool::from_slots(vec![None]);
+        pool.record_permission_decision("old");
+        tokio::time::advance(DECIDED_NONCE_RETENTION + Duration::from_secs(1)).await;
+        // Recording a new nonce prunes the stale one so the map cannot grow
+        // without bound over a long-lived process.
+        pool.record_permission_decision("new");
+        assert!(!pool.was_recently_decided("old"), "stale entry was pruned");
+        assert!(pool.was_recently_decided("new"), "fresh entry retained");
+        assert_eq!(
+            pool.recently_decided.len(),
+            1,
+            "only the fresh entry remains"
+        );
+    }
+
     #[test]
     fn delivery_receipt_line_sorts_event_ids() {
         let channel_id = Uuid::nil();
@@ -8681,12 +8828,17 @@ printf '%s\n' '{{"jsonrpc":"2.0","id":0,"result":{{"stopReason":"end_turn"}}}}'"
             ),
             context_message_limit: 0,
             max_turns_per_session: 0,
-            permission_mode: PermissionMode::Default,
+            permission_config: ResolvedPermissionConfig::resolve(
+                crate::config::PermissionPolicy::Reject,
+                None,
+            )
+            .expect("test config"),
             agent_keys: agent_keys.clone(),
             agent_owner_pubkey: owner_pubkey,
             memory_enabled: false,
             harness_name: "goose".to_string(),
             relay_url: "ws://127.0.0.1:3000".to_string(),
+            relay_event_publisher: None,
         }
     }