From 18b065f77a86cb2ff10164660659abda199cebd8 Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Thu, 16 Jul 2026 13:13:41 +0000 Subject: [PATCH] huskies: merge 1163 story Replace perm_rx lock-as-presence-signal with a permission router --- server/src/agent_mode/context.rs | 9 +- server/src/chat/transport/discord/commands.rs | 25 +- .../src/chat/transport/matrix/bot/context.rs | 7 +- .../matrix/bot/messages/on_room_message.rs | 8 +- server/src/chat/transport/matrix/bot/mod.rs | 4 +- .../matrix/bot/permission_listener.rs | 230 ++++++++-- server/src/chat/transport/matrix/bot/run.rs | 2 +- server/src/chat/transport/matrix/health.rs | 43 +- .../src/chat/transport/slack/commands/llm.rs | 19 +- .../src/chat/transport/slack/commands/mod.rs | 8 +- .../chat/transport/whatsapp/commands/llm.rs | 19 +- .../chat/transport/whatsapp/commands/mod.rs | 13 +- server/src/cli.rs | 5 +- server/src/gateway/mod.rs | 2 +- server/src/http/context.rs | 16 +- server/src/http/gateway/mcp.rs | 14 +- server/src/http/mcp/diagnostics/permission.rs | 42 +- server/src/http/ws/mod.rs | 32 +- server/src/main.rs | 7 +- server/src/service/bot_command/io.rs | 8 +- server/src/service/gateway/io.rs | 18 +- server/src/service/gateway/mod.rs | 25 +- server/src/service/mod.rs | 3 + server/src/service/permission_router.rs | 397 ++++++++++++++++++ server/src/services.rs | 25 +- server/src/sled_uplink.rs | 135 +++--- 26 files changed, 823 insertions(+), 293 deletions(-) create mode 100644 server/src/service/permission_router.rs diff --git a/server/src/agent_mode/context.rs b/server/src/agent_mode/context.rs index fc327747..6a8a595e 100644 --- a/server/src/agent_mode/context.rs +++ b/server/src/agent_mode/context.rs @@ -61,6 +61,11 @@ pub(super) fn build_agent_app_context( ); let (reconciliation_tx, _) = broadcast::channel(64); let (perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); + let permission_registry = crate::service::permission_router::ResponderRegistry::new(); + crate::service::permission_router::spawn_permission_router( + perm_rx, + Arc::clone(&permission_registry), + ); let timer_store = Arc::new(crate::service::timer::TimerStore::load( project_root.join(".huskies").join("timers.json"), )); @@ -74,8 +79,8 @@ pub(super) fn build_agent_app_context( bot_name: "Agent".to_string(), bot_user_id: String::new(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())), + permission_registry, + pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, status: agents.status_broadcaster(), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), diff --git a/server/src/chat/transport/discord/commands.rs b/server/src/chat/transport/discord/commands.rs index 96b25669..9c57791f 100644 --- a/server/src/chat/transport/discord/commands.rs +++ b/server/src/chat/transport/discord/commands.rs @@ -47,8 +47,12 @@ pub(super) async fn handle_incoming_message( // If there is a pending permission prompt for this channel, interpret the // message as a yes/no response. { - let mut pending = ctx.services.pending_perm_replies.lock().await; - if let Some(tx) = pending.remove(channel) { + if let Some(tx) = ctx + .services + .pending_perm_replies + .resolve_oldest(channel) + .await + { let decision = if is_permission_approval(message) { PermissionDecision::Approve } else { @@ -357,14 +361,14 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use ); tokio::pin!(chat_fut); - // Lock the permission receiver for the duration of this chat session. - let mut perm_rx_guard = ctx.services.perm_rx.lock().await; + // Register as a permission responder for the duration of this chat turn. + let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); let result = loop { tokio::select! { r = &mut chat_fut => break r, - Some(perm_fwd) = perm_rx_guard.recv() => { + Some(perm_fwd) = perm_rx.recv() => { let prompt_msg = format!( "**Permission Request**\n\nTool: `{}`\n```json\n{}\n```\n\nReply **yes** to approve or **no** to deny.", perm_fwd.tool_name, @@ -374,20 +378,22 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use let formatted = markdown_to_discord(&prompt_msg); let _ = ctx.transport.send_message(channel, &formatted, "").await; + // Keyed by request_id (not just channel) so a second + // concurrent request doesn't drop the first's sender. ctx.services .pending_perm_replies - .lock() - .await - .insert(channel.to_string(), perm_fwd.response_tx); + .insert(channel.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) + .await; // Spawn a timeout task: auto-deny if the user does not respond. let pending = Arc::clone(&ctx.services.pending_perm_replies); let timeout_channel = channel.to_string(); + let timeout_request_id = perm_fwd.request_id.clone(); let timeout_transport = Arc::clone(&ctx.transport) as Arc; let timeout_secs = ctx.services.permission_timeout_secs; tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.lock().await.remove(&timeout_channel) { + if let Some(tx) = pending.remove_by_request_id(&timeout_channel, &timeout_request_id).await { let _ = tx.send(PermissionDecision::Deny); let msg = "Permission request timed out — denied (fail-closed)."; let _ = timeout_transport.send_message(&timeout_channel, msg, "").await; @@ -396,7 +402,6 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use } } }; - drop(perm_rx_guard); // Flush remaining text. let remaining = buffer.lock().unwrap().trim().to_string(); diff --git a/server/src/chat/transport/matrix/bot/context.rs b/server/src/chat/transport/matrix/bot/context.rs index f784b3f2..720f57aa 100644 --- a/server/src/chat/transport/matrix/bot/context.rs +++ b/server/src/chat/transport/matrix/bot/context.rs @@ -263,9 +263,9 @@ impl BotContext { #[cfg(test)] mod tests { use super::*; + use crate::service::permission_router::{PendingPermReplies, ResponderRegistry}; use std::collections::HashMap; use std::path::PathBuf; - use tokio::sync::mpsc; fn make_user_id(s: &str) -> OwnedUserId { s.parse().unwrap() @@ -273,15 +273,14 @@ mod tests { /// Build a test `Services` bundle with the given project root. fn test_services(project_root: PathBuf) -> Arc { - let (_perm_tx, perm_rx) = mpsc::unbounded_channel(); Arc::new(Services { project_root, agents: Arc::new(crate::agents::AgentPool::new_test(3000)), bot_name: "Assistant".to_string(), bot_user_id: "@bot:example.com".to_string(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(TokioMutex::new(perm_rx)), - pending_perm_replies: Arc::new(TokioMutex::new(HashMap::new())), + permission_registry: ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), diff --git a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs index 72b33929..c651b221 100644 --- a/server/src/chat/transport/matrix/bot/messages/on_room_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/on_room_message.rs @@ -231,8 +231,12 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( // If there is a pending permission prompt for this room, interpret the // message as a yes/no response instead of starting a new chat. { - let mut pending = ctx.services.pending_perm_replies.lock().await; - if let Some(tx) = pending.remove(incoming_room_id.as_str()) { + if let Some(tx) = ctx + .services + .pending_perm_replies + .resolve_oldest(incoming_room_id.as_str()) + .await + { let decision = if is_permission_approval(&body) { PermissionDecision::Approve } else { diff --git a/server/src/chat/transport/matrix/bot/mod.rs b/server/src/chat/transport/matrix/bot/mod.rs index b614b57e..3ce15a67 100644 --- a/server/src/chat/transport/matrix/bot/mod.rs +++ b/server/src/chat/transport/matrix/bot/mod.rs @@ -9,8 +9,8 @@ pub mod history; pub mod mentions; /// Message handlers — processes incoming Matrix room messages. pub mod messages; -/// Permission listener — holds perm_rx for the bot's lifetime and forwards -/// permission requests to the configured Matrix room. +/// Permission listener — registers as a permission responder for the bot's +/// lifetime and forwards permission requests to the configured Matrix room. pub mod permission_listener; /// Bot run loop — the main async task that drives the Matrix sync loop. pub mod run; diff --git a/server/src/chat/transport/matrix/bot/permission_listener.rs b/server/src/chat/transport/matrix/bot/permission_listener.rs index fca99f77..5a609a76 100644 --- a/server/src/chat/transport/matrix/bot/permission_listener.rs +++ b/server/src/chat/transport/matrix/bot/permission_listener.rs @@ -1,11 +1,11 @@ -//! Background task that holds `perm_rx` for the bot's lifetime and forwards -//! permission requests to the configured Matrix room. +//! Background task that registers as a permission responder for the bot's +//! lifetime and forwards permission requests to the configured Matrix room. //! //! Before story 884, each chat message handler acquired `perm_rx` for the //! duration of one chat_fut and dropped it afterwards. That meant whenever //! the bot wasn't actively responding, `prompt_permission` auto-denied any //! spawned coder bash call as "no interactive session" — making unattended -//! coder work impossible. This task holds the lock continuously while the +//! coder work impossible. This task stays registered continuously while the //! bot is connected, so requests can flow at any time. use crate::chat::ChatTransport; @@ -20,13 +20,13 @@ use tokio::sync::Mutex as TokioMutex; use super::format::markdown_to_html; -/// Spawn a background task that holds `services.perm_rx` for the bot's -/// lifetime and forwards each incoming permission request to `target_room` -/// as a chat message. Replies (yes/no) are resolved by the existing -/// `on_room_message` handler via `pending_perm_replies`. +/// Spawn a background task that registers with `services.permission_registry` +/// for the bot's lifetime and forwards each incoming permission request to +/// `target_room` as a chat message. Replies (yes/no) are resolved by the +/// existing `on_room_message` handler via `pending_perm_replies`. /// /// Returns the JoinHandle so the caller can keep ownership; the task exits -/// only when the `perm_rx` channel is closed (bot shutdown). +/// only when its private channel closes (bot shutdown). pub fn spawn_permission_listener( services: Arc, transport: Arc, @@ -34,7 +34,7 @@ pub fn spawn_permission_listener( bot_sent_event_ids: Arc>>, ) -> tokio::task::JoinHandle<()> { tokio::spawn(async move { - let mut perm_rx = services.perm_rx.lock().await; + let (_responder_guard, mut perm_rx) = services.permission_registry.register(); let target_room_str = target_room.as_str().to_string(); slog!("[matrix-bot] permission listener started; forwarding requests to {target_room_str}"); @@ -57,24 +57,33 @@ pub fn spawn_permission_listener( } // Store the MCP oneshot sender so on_room_message can resolve it - // when the user replies yes/no in the target room. + // when the user replies yes/no in the target room. Keyed by + // request_id (not just room) so a second concurrent request for + // the same room doesn't drop the first's sender. services .pending_perm_replies - .lock() - .await - .insert(target_room.to_string(), perm_fwd.response_tx); + .insert( + target_room.to_string(), + perm_fwd.request_id.clone(), + perm_fwd.response_tx, + ) + .await; // Spawn a per-request timeout: auto-deny if the user does not // respond within `permission_timeout_secs`. let pending = Arc::clone(&services.pending_perm_replies); let timeout_room_key = target_room.to_string(); + let timeout_request_id = perm_fwd.request_id.clone(); let timeout_transport = Arc::clone(&transport); let timeout_room_str = target_room_str.clone(); let timeout_sent_ids = Arc::clone(&bot_sent_event_ids); let timeout_secs = services.permission_timeout_secs; tokio::spawn(async move { tokio::time::sleep(Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.lock().await.remove(&timeout_room_key) { + if let Some(tx) = pending + .remove_by_request_id(&timeout_room_key, &timeout_request_id) + .await + { let _ = tx.send(PermissionDecision::Deny); let msg = "Permission request timed out — denied (fail-closed)."; let html = markdown_to_html(msg); @@ -89,7 +98,7 @@ pub fn spawn_permission_listener( }); } - slog!("[matrix-bot] permission listener exiting (perm_rx channel closed)"); + slog!("[matrix-bot] permission listener exiting (channel closed)"); }) } @@ -102,11 +111,11 @@ mod tests { use super::*; use crate::http::context::PermissionForward; + use crate::service::permission_router::PendingPermReplies; use crate::services::Services; use async_trait::async_trait; use serde_json::json; - use std::collections::HashMap; - use tokio::sync::{mpsc, oneshot}; + use tokio::sync::oneshot; struct RecordingTransport { sent: Arc>>, @@ -138,26 +147,40 @@ mod tests { } } - fn test_services_with_tx() -> (Arc, mpsc::UnboundedSender) { - let (perm_tx, perm_rx) = mpsc::unbounded_channel(); - let services = Arc::new(Services { + /// Poll `cond` until it returns `true` or `timeout` elapses, sleeping + /// briefly between checks. Used instead of a single fixed sleep so tests + /// don't flake under slow/loaded CI where a spawned task hasn't yet run. + async fn wait_until(mut cond: impl FnMut() -> bool, timeout: std::time::Duration) -> bool { + let start = tokio::time::Instant::now(); + loop { + if cond() { + return true; + } + if start.elapsed() > timeout { + return false; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + } + + fn test_services() -> Arc { + Arc::new(Services { project_root: std::path::PathBuf::from("/tmp/test"), agents: Arc::new(crate::agents::AgentPool::new_test(3000)), bot_name: "Assistant".to_string(), bot_user_id: "@bot:example.com".to_string(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(TokioMutex::new(perm_rx)), - pending_perm_replies: Arc::new(TokioMutex::new(HashMap::new())), + permission_registry: crate::service::permission_router::ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), - }); - (services, perm_tx) + }) } #[tokio::test] async fn listener_forwards_request_to_target_room_when_no_chat_in_flight() { - let (services, perm_tx) = test_services_with_tx(); + let services = test_services(); let sent: Arc>> = Arc::new(std::sync::Mutex::new(Vec::new())); let transport: Arc = Arc::new(RecordingTransport { @@ -173,21 +196,33 @@ mod tests { Arc::clone(&bot_sent_event_ids), ); - // Yield so the listener task acquires perm_rx and starts recv'ing. - tokio::time::sleep(std::time::Duration::from_millis(20)).await; + // Wait until the listener task registers and starts recv'ing. + assert!( + wait_until( + || !services.permission_registry.is_empty(), + std::time::Duration::from_secs(2) + ) + .await, + "listener never registered as a responder" + ); let (response_tx, _response_rx) = oneshot::channel(); - perm_tx - .send(PermissionForward { - request_id: "req-1".to_string(), - tool_name: "Bash".to_string(), - tool_input: json!({"command": "cargo test"}), - response_tx, - }) - .expect("send PermissionForward"); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-1".to_string(), + tool_name: "Bash".to_string(), + tool_input: json!({"command": "cargo test"}), + response_tx, + }); - // Give the listener a moment to process. - tokio::time::sleep(std::time::Duration::from_millis(50)).await; + // Wait for the listener to process the forward. + assert!( + wait_until( + || !sent.lock().unwrap().is_empty(), + std::time::Duration::from_secs(2) + ) + .await, + "listener never sent the permission prompt" + ); // The transport must have received exactly one send_message to the // target room with the prompt content. @@ -205,17 +240,126 @@ mod tests { recorded[0].1 ); - // pending_perm_replies must contain an entry keyed by the target room - // (so the user-reply handler can resolve the request when they reply). - let pending = services.pending_perm_replies.lock().await; + // pending_perm_replies must contain an entry resolvable for the target + // room (so the user-reply handler can resolve the request when they + // reply). The insert happens just after send_message returns, so poll + // briefly rather than assuming it's already visible. + let mut resolved = None; + for _ in 0..50 { + resolved = services + .pending_perm_replies + .resolve_oldest(target_room.as_str()) + .await; + if resolved.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } assert!( - pending.contains_key(target_room.as_str()), + resolved.is_some(), "pending_perm_replies missing entry for target room" ); // bot_sent_event_ids must have recorded the prompt's event ID so the // bot does not echo its own prompt back as user input. - let sent_ids = bot_sent_event_ids.lock().await; - assert_eq!(sent_ids.len(), 1, "expected one sent event ID recorded"); + assert!( + wait_until( + || bot_sent_event_ids + .try_lock() + .map(|s| s.len() == 1) + .unwrap_or(false), + std::time::Duration::from_secs(2) + ) + .await, + "expected one sent event ID recorded" + ); + } + + /// Regression: two concurrent permission requests for the same target + /// room must both be resolvable — keying `pending_perm_replies` by + /// request_id (rather than overwriting a single room-keyed entry) means + /// the first request's oneshot sender is never dropped by the second. + #[tokio::test] + async fn two_concurrent_requests_for_same_room_both_resolve() { + let services = test_services(); + let sent: Arc>> = + Arc::new(std::sync::Mutex::new(Vec::new())); + let transport: Arc = Arc::new(RecordingTransport { + sent: Arc::clone(&sent), + }); + let target_room: OwnedRoomId = "!perm:example.com".parse().unwrap(); + let bot_sent_event_ids = Arc::new(TokioMutex::new(HashSet::new())); + + spawn_permission_listener( + Arc::clone(&services), + Arc::clone(&transport), + target_room.clone(), + Arc::clone(&bot_sent_event_ids), + ); + assert!( + wait_until( + || !services.permission_registry.is_empty(), + std::time::Duration::from_secs(2) + ) + .await, + "listener never registered as a responder" + ); + + let (tx1, rx1) = oneshot::channel(); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-a".to_string(), + tool_name: "Bash".to_string(), + tool_input: json!({}), + response_tx: tx1, + }); + let (tx2, rx2) = oneshot::channel(); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-b".to_string(), + tool_name: "Write".to_string(), + tool_input: json!({}), + response_tx: tx2, + }); + + // Wait until the listener has processed both forwards. + assert!( + wait_until( + || sent.lock().unwrap().len() >= 2, + std::time::Duration::from_secs(2) + ) + .await, + "listener never sent both permission prompts" + ); + + // Poll rather than assuming the pending_perm_replies insert (which + // happens just after send_message returns) is already visible. + let mut first = None; + for _ in 0..50 { + first = services + .pending_perm_replies + .resolve_oldest(target_room.as_str()) + .await; + if first.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let first = first.expect("first request's sender must not have been dropped"); + let _ = first.send(PermissionDecision::Approve); + assert_eq!(rx1.await.unwrap(), PermissionDecision::Approve); + + let mut second = None; + for _ in 0..50 { + second = services + .pending_perm_replies + .resolve_oldest(target_room.as_str()) + .await; + if second.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let second = second.expect("second request's sender must still be present"); + let _ = second.send(PermissionDecision::Deny); + assert_eq!(rx2.await.unwrap(), PermissionDecision::Deny); } } diff --git a/server/src/chat/transport/matrix/bot/run.rs b/server/src/chat/transport/matrix/bot/run.rs index 5c507eae..82f02054 100644 --- a/server/src/chat/transport/matrix/bot/run.rs +++ b/server/src/chat/transport/matrix/bot/run.rs @@ -299,7 +299,7 @@ pub async fn run_bot( let bot_sent_event_ids: Arc>> = Arc::new(TokioMutex::new(HashSet::new())); - // Spawn the permission listener: holds `perm_rx` for the bot's lifetime + // Spawn the permission listener: registers as a responder for the bot's lifetime // and forwards permission requests to the first configured room. Story // 884 — replaces the per-message lock acquire previously done in // handle_message.rs, so spawned coders' bash calls reach chat even when diff --git a/server/src/chat/transport/matrix/health.rs b/server/src/chat/transport/matrix/health.rs index e21ebbbf..ebe35fa1 100644 --- a/server/src/chat/transport/matrix/health.rs +++ b/server/src/chat/transport/matrix/health.rs @@ -118,13 +118,14 @@ fn truncate_lines(mut lines: Vec) -> Vec { // ── Individual checks ──────────────────────────────────────────────────────── -/// Check the `perm_rx` receiver — PASS when the permission listener holds the lock, -/// FAIL when no task is holding it (listener has died or was never started). +/// Check the permission registry — PASS when at least one responder (e.g. the +/// Matrix permission listener) is registered, FAIL when none is (listener has +/// died or was never started). fn check_perm_rx(ctx: &BotContext) -> HealthLine { - if ctx.services.perm_rx.try_lock().is_err() { - HealthLine::pass("perm_rx") + if ctx.services.permission_registry.is_empty() { + HealthLine::fail("perm_rx", "no responder registered", "restart bot") } else { - HealthLine::fail("perm_rx", "listener not holding lock", "restart bot") + HealthLine::pass("perm_rx") } } @@ -513,25 +514,22 @@ mod tests { #[tokio::test] async fn perm_rx_pass_when_locked() { + use crate::service::permission_router::{PendingPermReplies, ResponderRegistry}; use crate::services::Services; use std::sync::Arc; - use tokio::sync::Mutex as TokioMutex; - let (perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); - let perm_rx_arc = Arc::new(TokioMutex::new(perm_rx)); + let registry = ResponderRegistry::new(); + // Register a responder to simulate the permission listener being active. + let _guard_and_rx = registry.register(); - // Acquire the lock to simulate the permission listener holding it. - let _guard = perm_rx_arc.try_lock().unwrap(); - - // Build a minimal services bundle referencing our locked perm_rx. let services = Arc::new(Services { project_root: std::path::PathBuf::from("/tmp"), agents: Arc::new(crate::agents::AgentPool::new_test(3000)), bot_name: "test".to_string(), bot_user_id: "@bot:test".to_string(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::clone(&perm_rx_arc), - pending_perm_replies: Arc::new(TokioMutex::new(std::collections::HashMap::new())), + permission_registry: registry, + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), @@ -544,30 +542,25 @@ mod tests { assert_eq!( line.status, Status::Pass, - "perm_rx should PASS when a task holds the lock" + "perm_rx should PASS when a responder is registered" ); - - drop(perm_tx); // suppress unused warning } #[tokio::test] async fn perm_rx_fail_when_unlocked() { + use crate::service::permission_router::{PendingPermReplies, ResponderRegistry}; use crate::services::Services; use std::sync::Arc; - use tokio::sync::Mutex as TokioMutex; - - let (_perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); - let perm_rx_arc = Arc::new(TokioMutex::new(perm_rx)); - // Lock is NOT held by anyone. + // No responder registered. let services = Arc::new(Services { project_root: std::path::PathBuf::from("/tmp"), agents: Arc::new(crate::agents::AgentPool::new_test(3000)), bot_name: "test".to_string(), bot_user_id: "@bot:test".to_string(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::clone(&perm_rx_arc), - pending_perm_replies: Arc::new(TokioMutex::new(std::collections::HashMap::new())), + permission_registry: ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), @@ -579,7 +572,7 @@ mod tests { assert_eq!( line.status, Status::Fail, - "perm_rx should FAIL when no task holds the lock" + "perm_rx should FAIL when no responder is registered" ); } diff --git a/server/src/chat/transport/slack/commands/llm.rs b/server/src/chat/transport/slack/commands/llm.rs index 9b304589..6379fb61 100644 --- a/server/src/chat/transport/slack/commands/llm.rs +++ b/server/src/chat/transport/slack/commands/llm.rs @@ -82,14 +82,14 @@ pub(super) async fn handle_llm_message( ); tokio::pin!(chat_fut); - // Lock the permission receiver for the duration of this chat session. - let mut perm_rx_guard = ctx.services.perm_rx.lock().await; + // Register as a permission responder for the duration of this chat turn. + let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); let result = loop { tokio::select! { r = &mut chat_fut => break r, - Some(perm_fwd) = perm_rx_guard.recv() => { + Some(perm_fwd) = perm_rx.recv() => { let prompt_msg = format!( "*Permission Request*\n\nTool: `{}`\n```json\n{}\n```\n\nReply *yes* to approve or *no* to deny.", perm_fwd.tool_name, @@ -100,20 +100,22 @@ pub(super) async fn handle_llm_message( let _ = ctx.transport.send_message(channel, &formatted, "").await; // Store the response sender so the incoming message handler - // can resolve it when the user replies yes/no. + // can resolve it when the user replies yes/no. Keyed by + // request_id (not just channel) so a second concurrent + // request doesn't drop the first's sender. ctx.services.pending_perm_replies - .lock() - .await - .insert(channel.to_string(), perm_fwd.response_tx); + .insert(channel.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) + .await; // Spawn a timeout task: auto-deny if the user does not respond. let pending = Arc::clone(&ctx.services.pending_perm_replies); let timeout_channel = channel.to_string(); + let timeout_request_id = perm_fwd.request_id.clone(); let timeout_transport = Arc::clone(&ctx.transport); let timeout_secs = ctx.services.permission_timeout_secs; tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.lock().await.remove(&timeout_channel) { + if let Some(tx) = pending.remove_by_request_id(&timeout_channel, &timeout_request_id).await { let _ = tx.send(PermissionDecision::Deny); let msg = "Permission request timed out — denied (fail-closed)."; let _ = timeout_transport.send_message(&timeout_channel, msg, "").await; @@ -122,7 +124,6 @@ pub(super) async fn handle_llm_message( } } }; - drop(perm_rx_guard); // Flush remaining text. let remaining = buffer.lock().unwrap().trim().to_string(); diff --git a/server/src/chat/transport/slack/commands/mod.rs b/server/src/chat/transport/slack/commands/mod.rs index d35d6a71..b44f6dba 100644 --- a/server/src/chat/transport/slack/commands/mod.rs +++ b/server/src/chat/transport/slack/commands/mod.rs @@ -92,8 +92,12 @@ pub(super) async fn handle_incoming_message( // If there is a pending permission prompt for this channel, interpret the // message as a yes/no response instead of starting a new command/LLM flow. { - let mut pending = ctx.services.pending_perm_replies.lock().await; - if let Some(tx) = pending.remove(channel) { + if let Some(tx) = ctx + .services + .pending_perm_replies + .resolve_oldest(channel) + .await + { let decision = if is_permission_approval(message) { PermissionDecision::Approve } else { diff --git a/server/src/chat/transport/whatsapp/commands/llm.rs b/server/src/chat/transport/whatsapp/commands/llm.rs index c9c80c0a..f89c2544 100644 --- a/server/src/chat/transport/whatsapp/commands/llm.rs +++ b/server/src/chat/transport/whatsapp/commands/llm.rs @@ -81,14 +81,14 @@ pub(super) async fn handle_llm_message( ); tokio::pin!(chat_fut); - // Lock the permission receiver for the duration of this chat session. - let mut perm_rx_guard = ctx.services.perm_rx.lock().await; + // Register as a permission responder for the duration of this chat turn. + let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); let result = loop { tokio::select! { r = &mut chat_fut => break r, - Some(perm_fwd) = perm_rx_guard.recv() => { + Some(perm_fwd) = perm_rx.recv() => { let prompt_msg = format!( "*Permission Request*\n\nTool: `{}`\n```json\n{}\n```\n\nReply *yes* to approve or *no* to deny.", perm_fwd.tool_name, @@ -101,20 +101,22 @@ pub(super) async fn handle_llm_message( } // Store the response sender so the incoming message handler - // can resolve it when the user replies yes/no. + // can resolve it when the user replies yes/no. Keyed by + // request_id (not just sender) so a second concurrent + // request doesn't drop the first's sender. ctx.services.pending_perm_replies - .lock() - .await - .insert(sender.to_string(), perm_fwd.response_tx); + .insert(sender.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx) + .await; // Spawn a timeout task: auto-deny if the user does not respond. let pending = Arc::clone(&ctx.services.pending_perm_replies); let timeout_sender = sender.to_string(); + let timeout_request_id = perm_fwd.request_id.clone(); let timeout_transport = Arc::clone(&ctx.transport); let timeout_secs = ctx.services.permission_timeout_secs; tokio::spawn(async move { tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await; - if let Some(tx) = pending.lock().await.remove(&timeout_sender) { + if let Some(tx) = pending.remove_by_request_id(&timeout_sender, &timeout_request_id).await { let _ = tx.send(PermissionDecision::Deny); let msg = "Permission request timed out — denied (fail-closed)."; let _ = timeout_transport.send_message(&timeout_sender, msg, "").await; @@ -123,7 +125,6 @@ pub(super) async fn handle_llm_message( } } }; - drop(perm_rx_guard); // Flush remaining text. let remaining = buffer.lock().unwrap().trim().to_string(); diff --git a/server/src/chat/transport/whatsapp/commands/mod.rs b/server/src/chat/transport/whatsapp/commands/mod.rs index 04da09ed..00039fbf 100644 --- a/server/src/chat/transport/whatsapp/commands/mod.rs +++ b/server/src/chat/transport/whatsapp/commands/mod.rs @@ -32,8 +32,12 @@ pub(super) async fn handle_incoming_message( // If there is a pending permission prompt for this sender, interpret the // message as a yes/no response instead of starting a new command/LLM flow. { - let mut pending = ctx.services.pending_perm_replies.lock().await; - if let Some(tx) = pending.remove(sender) { + if let Some(tx) = ctx + .services + .pending_perm_replies + .resolve_oldest(sender) + .await + { let decision = if is_permission_approval(message) { PermissionDecision::Approve } else { @@ -279,7 +283,6 @@ mod tests { let (tx, _rx) = tokio::sync::broadcast::channel::(16); let agents = Arc::new(AgentPool::new(3999, tx)); let tracker = Arc::new(MessagingWindowTracker::new()); - let (_perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); let services = Arc::new(crate::services::Services { project_root: tmp.path().to_path_buf(), status: agents.status_broadcaster(), @@ -287,8 +290,8 @@ mod tests { bot_name: "Bot".to_string(), bot_user_id: "whatsapp-bot".to_string(), ambient_rooms: Arc::new(std::sync::Mutex::new(Default::default())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(Default::default())), + permission_registry: crate::service::permission_router::ResponderRegistry::new(), + pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }); diff --git a/server/src/cli.rs b/server/src/cli.rs index 68b02a92..aa728c0c 100644 --- a/server/src/cli.rs +++ b/server/src/cli.rs @@ -23,8 +23,9 @@ pub(crate) struct CliArgs { pub(crate) gateway_url: Option, /// WebSocket URL of the upstream gateway to forward permission requests to (`--upstream-gateway`). /// - /// When set, the sled spawns a background uplink task that holds `perm_rx` and - /// forwards all `prompt_permission` tool calls to the gateway over a WebSocket. + /// When set, the sled spawns a background uplink task that registers as a + /// permission responder and forwards all `prompt_permission` tool calls to + /// the gateway over a WebSocket. /// Also readable from the `HUSKIES_UPSTREAM_GATEWAY` env var. pub(crate) upstream_gateway: Option, /// Path to a trampoline job file (`--trampoline `). diff --git a/server/src/gateway/mod.rs b/server/src/gateway/mod.rs index cefb3567..8cb84597 100644 --- a/server/src/gateway/mod.rs +++ b/server/src/gateway/mod.rs @@ -130,7 +130,7 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> { Arc::clone(&state_arc.projects), port, Some(state_arc.event_tx.clone()), - Arc::clone(&state_arc.perm_rx), + Arc::clone(&state_arc.permission_registry), ); *state_arc.bot_handle.lock().await = bot_abort; *state_arc.bot_shutdown_tx.lock().await = Some(bot_shutdown_tx); diff --git a/server/src/http/context.rs b/server/src/http/context.rs index 8694cddb..efad5f31 100644 --- a/server/src/http/context.rs +++ b/server/src/http/context.rs @@ -96,6 +96,16 @@ impl AppContext { let (watcher_tx, _) = broadcast::channel(64); let (reconciliation_tx, _) = broadcast::channel(64); let (perm_tx, perm_rx) = mpsc::unbounded_channel(); + let permission_registry = crate::service::permission_router::ResponderRegistry::new(); + // Plain `#[test]` fns (no tokio runtime) construct `AppContext` too; + // skip spawning when there's no reactor to spawn onto since those + // tests never exercise the permission plumbing. + if tokio::runtime::Handle::try_current().is_ok() { + crate::service::permission_router::spawn_permission_router( + perm_rx, + Arc::clone(&permission_registry), + ); + } let timer_store = Arc::new(TimerStore::load( project_root.join(".huskies").join("timers.json"), )); @@ -112,10 +122,8 @@ impl AppContext { bot_name: "Assistant".to_string(), bot_user_id: String::new(), ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new( - std::collections::HashMap::new(), - )), + permission_registry, + pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, status: agents.status_broadcaster(), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), diff --git a/server/src/http/gateway/mcp.rs b/server/src/http/gateway/mcp.rs index 36fbda5e..8f461c76 100644 --- a/server/src/http/gateway/mcp.rs +++ b/server/src/http/gateway/mcp.rs @@ -24,7 +24,7 @@ const GATEWAY_TOOLS: &[&str] = &[ "adopt_project", "aggregate_pipeline_status", "agents.list", - // Handled at the gateway so the Matrix bot's perm_rx listener is used + // Handled at the gateway so the Matrix bot's permission listener is used // rather than the container's (which has no interactive session attached). "prompt_permission", // One-shot container rebuild: build fresh image, swap container, preserve state. @@ -764,9 +764,9 @@ async fn handle_aggregate_pipeline_status_tool( /// Handle the `prompt_permission` tool at the gateway level. /// /// Mirrors `tool_prompt_permission` in `http/mcp/diagnostics/permission.rs` but -/// uses the gateway's `perm_tx`/`perm_rx` so requests reach the Matrix bot that -/// is listening on the gateway, not the proxied container (which has no -/// interactive session and would auto-deny immediately). +/// uses the gateway's `perm_tx`/`permission_registry` so requests reach the +/// Matrix bot that is listening on the gateway, not the proxied container +/// (which has no interactive session and would auto-deny immediately). async fn handle_prompt_permission_tool( params: &Value, state: &GatewayState, @@ -792,9 +792,9 @@ async fn handle_prompt_permission_tool( return JsonRpcResponse::success(id, json!({"content": [{"type": "text", "text": text}]})); } - // Auto-deny when no interactive session holds perm_rx (i.e. no Matrix bot - // listener is running — try_lock succeeds when nobody else holds the lock). - if state.perm_rx.try_lock().is_ok() { + // Auto-deny when no responder is registered (i.e. no Matrix bot listener + // is running). + if state.permission_registry.is_empty() { crate::slog!("[gateway/permission] Auto-denied '{tool_name}' (no interactive session)"); let text = json!({ "behavior": "deny", diff --git a/server/src/http/mcp/diagnostics/permission.rs b/server/src/http/mcp/diagnostics/permission.rs index 1adf6578..02aadf67 100644 --- a/server/src/http/mcp/diagnostics/permission.rs +++ b/server/src/http/mcp/diagnostics/permission.rs @@ -29,14 +29,14 @@ pub(crate) async fn tool_prompt_permission( return Ok(json!({"behavior": "allow", "updatedInput": tool_input}).to_string()); } - // Auto-deny immediately if no interactive session is currently listening on - // perm_rx. Story 884 made the Matrix bot hold this lock for its lifetime - // via the permission_listener task spawned at startup, so requests reach - // chat asynchronously regardless of whether a chat message is in flight. - // Other transports (Discord/Slack/WhatsApp) still acquire per message; if - // none is active, try_lock succeeds — auto-deny so background agent calls - // don't queue and flood chat at the next user session. - if ctx.services.perm_rx.try_lock().is_ok() { + // Auto-deny immediately if no responder is currently registered to + // receive forwarded permission requests. The Matrix bot's + // permission_listener task, sled uplinks, and per-message chat transports + // all register for the duration they're able to prompt a user; if none is + // registered, don't forward the request into the void — auto-deny so + // background agent calls don't queue and flood chat at the next user + // session. + if ctx.services.permission_registry.is_empty() { crate::slog!( "[permission] Auto-denied '{tool_name}' (no interactive session — agent mode)" ); @@ -141,24 +141,19 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let ctx = test_ctx(tmp.path()); - // Simulate an interactive session: lock perm_rx first, signal readiness, - // then respond with approval. The try_lock() inside tool_prompt_permission - // must fail (lock held) so the request is forwarded rather than auto-denied. - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<()>(); - let perm_rx = ctx.services.perm_rx.clone(); + // Simulate an interactive session: register a responder first so the + // registry is non-empty and the request is forwarded rather than + // auto-denied, then respond with approval. + let (guard, mut rx) = ctx.services.permission_registry.register(); tokio::spawn(async move { - let mut rx = perm_rx.lock().await; - let _ = ready_tx.send(()); // signal: lock is held if let Some(forward) = rx.recv().await { let _ = forward .response_tx .send(crate::http::context::PermissionDecision::Approve); } + drop(guard); }); - // Wait until the spawned task holds the perm_rx lock. - ready_rx.await.unwrap(); - let result = tool_prompt_permission( &json!({"tool_name": "Bash", "input": {"command": "echo hello"}}), &ctx, @@ -182,22 +177,17 @@ mod tests { let tmp = tempfile::tempdir().unwrap(); let ctx = test_ctx(tmp.path()); - // Simulate an interactive session: lock perm_rx first, then deny. - let (ready_tx, ready_rx) = tokio::sync::oneshot::channel::<()>(); - let perm_rx = ctx.services.perm_rx.clone(); + // Simulate an interactive session: register a responder, then deny. + let (guard, mut rx) = ctx.services.permission_registry.register(); tokio::spawn(async move { - let mut rx = perm_rx.lock().await; - let _ = ready_tx.send(()); // signal: lock is held if let Some(forward) = rx.recv().await { let _ = forward .response_tx .send(crate::http::context::PermissionDecision::Deny); } + drop(guard); }); - // Wait until the spawned task holds the perm_rx lock. - ready_rx.await.unwrap(); - let result = tool_prompt_permission(&json!({"tool_name": "Write", "input": {}}), &ctx) .await .expect("denial must return Ok, not Err"); diff --git a/server/src/http/ws/mod.rs b/server/src/http/ws/mod.rs index b4476295..c709fc5e 100644 --- a/server/src/http/ws/mod.rs +++ b/server/src/http/ws/mod.rs @@ -147,34 +147,20 @@ pub async fn ws_handler(ws: WebSocket, ctx: Data<&Arc>) -> impl poem ); tokio::pin!(chat_fut); - // Take perm_rx for local permission forwarding — but never - // block on it. Since story 884 the Matrix permission - // listener (and the sled uplink, when configured) hold this - // lock for the process lifetime, so a blocking - // `lock().await` here parks the entire WS connection loop - // forever: chat_fut is never polled, RPC frames on this - // socket are never answered, and everything queues behind - // the dispatcher's serial session lock. If another task - // already owns permission routing, proceed without the - // local forwarding arm — requests are handled there. - let mut perm_rx = ctx.services.perm_rx.try_lock().ok(); - if perm_rx.is_none() { - crate::slog!( - "[ws] perm_rx held by another listener; \ - skipping local permission forwarding for this chat" - ); - } + // Register as a permission responder for the duration of + // this chat turn so local (browser) approval works + // alongside any other registered responder (Matrix bot, + // sled uplink). Registration and its private channel never + // block on foreign I/O, so this can never park the WS + // connection loop — unlike the old `perm_rx.lock().await` + // workaround from commit ba0a38d4, which this replaces. + let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register(); let chat_result = loop { tokio::select! { result = &mut chat_fut => break result, - Some(perm_fwd) = async { - match perm_rx.as_mut() { - Some(rx) => rx.recv().await, - None => std::future::pending().await, - } - } => { + Some(perm_fwd) = perm_rx.recv() => { let _ = tx.send(ws::permission_request_response( &perm_fwd.request_id, &perm_fwd.tool_name, diff --git a/server/src/main.rs b/server/src/main.rs index 68e7a3a5..8b857121 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -224,7 +224,8 @@ async fn main() -> Result<(), std::io::Error> { let watcher_rx_for_discord = watcher_tx.subscribe(); let watcher_rx_for_events = watcher_tx.subscribe(); - let perm_rx = Arc::new(tokio::sync::Mutex::new(perm_rx)); + let permission_registry = service::permission_router::ResponderRegistry::new(); + service::permission_router::spawn_permission_router(perm_rx, Arc::clone(&permission_registry)); let startup_root: Option = app_state.project_root.lock().unwrap().clone(); let startup_agents = Arc::clone(&agents); let startup_reconciliation_tx = reconciliation_tx.clone(); @@ -248,8 +249,8 @@ async fn main() -> Result<(), std::io::Error> { .map(|c| c.ambient_rooms.iter().cloned().collect()) .unwrap_or_default(), )), - perm_rx: Arc::clone(&perm_rx), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(std::collections::HashMap::new())), + permission_registry: Arc::clone(&permission_registry), + pending_perm_replies: service::permission_router::PendingPermReplies::new(), permission_timeout_secs: bot_cfg .as_ref() .map(|c| c.permission_timeout_secs) diff --git a/server/src/service/bot_command/io.rs b/server/src/service/bot_command/io.rs index 74c54f44..baa83f15 100644 --- a/server/src/service/bot_command/io.rs +++ b/server/src/service/bot_command/io.rs @@ -124,22 +124,22 @@ pub(super) fn call_sync( agents: &Arc, ) -> Option { use crate::chat::commands::CommandDispatch; - use std::collections::{HashMap, HashSet}; + use crate::service::permission_router::{PendingPermReplies, ResponderRegistry}; + use std::collections::HashSet; use std::sync::Mutex; let bot_name = "__web_ui__"; let bot_user_id = "@__web_ui__:localhost"; let room_id = "__web_ui__"; - let (_, perm_rx) = tokio::sync::mpsc::unbounded_channel(); let services = Arc::new(crate::services::Services { project_root: project_root.to_path_buf(), agents: Arc::clone(agents), bot_name: bot_name.to_string(), bot_user_id: bot_user_id.to_string(), ambient_rooms: Arc::new(Mutex::new(HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(HashMap::new())), + permission_registry: ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), diff --git a/server/src/service/gateway/io.rs b/server/src/service/gateway/io.rs index 84c7cdc6..bbaa12dc 100644 --- a/server/src/service/gateway/io.rs +++ b/server/src/service/gateway/io.rs @@ -507,11 +507,7 @@ pub fn spawn_gateway_bot( gateway_projects_store: std::sync::Arc>>, port: u16, gateway_event_tx: Option>, - perm_rx: std::sync::Arc< - tokio::sync::Mutex< - tokio::sync::mpsc::UnboundedReceiver, - >, - >, + permission_registry: std::sync::Arc, ) -> ( Option, tokio::sync::watch::Sender>, @@ -550,10 +546,8 @@ pub fn spawn_gateway_bot( .map(|c| c.ambient_rooms.iter().cloned().collect()) .unwrap_or_default(), )), - perm_rx, - pending_perm_replies: std::sync::Arc::new(tokio::sync::Mutex::new( - std::collections::HashMap::new(), - )), + permission_registry, + pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: bot_cfg .as_ref() .map(|c| c.permission_timeout_secs) @@ -599,9 +593,7 @@ mod tests { let active = std::sync::Arc::new(tokio::sync::RwLock::new("proj".to_string())); let (event_tx, _) = tokio::sync::broadcast::channel(4); - let (_perm_tx, perm_rx) = - tokio::sync::mpsc::unbounded_channel::(); - let perm_rx = std::sync::Arc::new(tokio::sync::Mutex::new(perm_rx)); + let permission_registry = crate::service::permission_router::ResponderRegistry::new(); let projects_store = std::sync::Arc::new(tokio::sync::RwLock::new(std::collections::BTreeMap::new())); let (handle, shutdown_tx) = spawn_gateway_bot( @@ -610,7 +602,7 @@ mod tests { projects_store, 3001, Some(event_tx), - perm_rx, + permission_registry, ); // No bot.toml in tmp → no abort handle spawned. diff --git a/server/src/service/gateway/mod.rs b/server/src/service/gateway/mod.rs index a0c7a7e5..69f7486c 100644 --- a/server/src/service/gateway/mod.rs +++ b/server/src/service/gateway/mod.rs @@ -224,11 +224,14 @@ pub struct GatewayState { /// received from connected sleds into the gateway's Matrix bot permission /// pipeline. pub perm_tx: mpsc::UnboundedSender, - /// Receiver end of the gateway's permission channel (shared with the Matrix bot). + /// Registry of tasks registered to receive forwarded permission requests + /// (shared with the gateway's embedded Matrix bot). /// - /// The Matrix bot's `permission_listener` holds this locked for its lifetime; - /// the sled-uplink WS handler sends requests via `perm_tx`. - pub perm_rx: Arc>>, + /// The Matrix bot's `permission_listener` registers with this for its + /// lifetime; the sled-uplink WS handler sends requests via `perm_tx`, + /// which the router task (spawned in [`GatewayState::new`]) dispatches + /// into this registry. + pub permission_registry: Arc, /// Reversed sled-token map: token → project_name (sled_id). /// /// Built at startup from both [`GatewayConfig::sled_tokens`] AND the @@ -261,6 +264,16 @@ impl GatewayState { .unwrap_or(first_from_config); let (event_tx, _) = tokio::sync::broadcast::channel(EVENT_CHANNEL_CAPACITY); let (perm_tx, perm_rx) = mpsc::unbounded_channel::(); + let permission_registry = crate::service::permission_router::ResponderRegistry::new(); + // `GatewayState::new` is also called from plain `#[test]` fns with no + // tokio runtime; skip spawning when there's no reactor to spawn onto + // since those tests never exercise the permission plumbing. + if tokio::runtime::Handle::try_current().is_ok() { + crate::service::permission_router::spawn_permission_router( + perm_rx, + Arc::clone(&permission_registry), + ); + } // Build token→project_name map from two sources: // 1. Legacy top-level [sled_tokens] section (sled_id → token, reversed) @@ -287,7 +300,7 @@ impl GatewayState { bot_shutdown_tx: Arc::new(TokioMutex::new(None)), event_tx, perm_tx, - perm_rx: Arc::new(TokioMutex::new(perm_rx)), + permission_registry, sled_tokens, sled_connections: Arc::new(RwLock::new(HashMap::new())), }) @@ -683,7 +696,7 @@ pub async fn save_bot_config_and_restart(state: &GatewayState, content: &str) -> Arc::clone(&state.projects), state.port, Some(state.event_tx.clone()), - Arc::clone(&state.perm_rx), + Arc::clone(&state.permission_registry), ); *handle = new_handle; *state.bot_shutdown_tx.lock().await = Some(new_shutdown_tx); diff --git a/server/src/service/mod.rs b/server/src/service/mod.rs index 657ae4ba..b0263384 100644 --- a/server/src/service/mod.rs +++ b/server/src/service/mod.rs @@ -31,6 +31,9 @@ pub mod merge; pub mod notifications; /// OAuth 2.0 PKCE flow for Anthropic authentication. pub mod oauth; +/// Permission router — responder registry and pending-reply tracking that +/// replace holding `perm_rx`'s mutex as a presence signal. +pub mod permission_router; /// Pipeline status aggregation helpers. pub mod pipeline; /// Project open/close/list domain logic. diff --git a/server/src/service/permission_router.rs b/server/src/service/permission_router.rs new file mode 100644 index 00000000..04d4857b --- /dev/null +++ b/server/src/service/permission_router.rs @@ -0,0 +1,397 @@ +//! Permission router — replaces holding `perm_rx`'s mutex as a presence +//! signal for whether an interactive session is listening for MCP +//! `prompt_permission` requests. +//! +//! Before this module, `Services` held the MCP-side receiver behind a +//! `tokio::sync::Mutex`. Whichever transport task (Matrix bot, sled uplink, +//! WS chat session) locked it became "the" active listener for the process, +//! and `tool_prompt_permission` used `try_lock()` as a cheap presence check. +//! That conflated two unrelated concerns — "is anyone listening" and "who +//! drains the queue" — behind a single lock, and made it impossible for more +//! than one responder (e.g. a Matrix bot and several open browser tabs) to be +//! registered at once without one of them blocking forever on `.lock().await`. +//! +//! The replacement: [`spawn_permission_router`] is the sole, permanent owner +//! of the MCP-side receiver. Responders call [`ResponderRegistry::register`] +//! to obtain a private bounded channel and an RAII [`ResponderGuard`] that +//! unregisters them on drop. The router only ever `try_send`s into a +//! responder's channel — it never awaits foreign I/O, so a wedged responder +//! (registered but not draining) can stall neither the router nor any other +//! part of the server. + +use crate::http::context::{PermissionDecision, PermissionForward}; +use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; +use std::sync::Mutex as StdMutex; +use std::sync::atomic::{AtomicU64, Ordering}; +use tokio::sync::{Mutex as TokioMutex, mpsc, oneshot}; + +/// Capacity of a single responder's private inbox channel. +/// +/// Bounded so a wedged responder's queue fills up (and the router falls back +/// to the next responder, or fail-closes) instead of growing without limit. +pub const RESPONDER_CHANNEL_CAPACITY: usize = 16; + +struct ResponderSlot { + id: u64, + tx: mpsc::Sender, +} + +/// Tracks which tasks are currently registered to receive forwarded +/// permission requests. +/// +/// Registering does not grant exclusive ownership: multiple responders (the +/// Matrix bot, a sled uplink, several WS browser tabs) may be registered at +/// once. Each incoming request is dispatched to the first registered +/// responder whose private channel accepts it via `try_send`. +pub struct ResponderRegistry { + next_id: AtomicU64, + slots: StdMutex>, +} + +impl ResponderRegistry { + /// Create an empty registry. + pub fn new() -> Arc { + Arc::new(Self { + next_id: AtomicU64::new(0), + slots: StdMutex::new(Vec::new()), + }) + } + + /// `true` when no responder is currently registered. + /// + /// `tool_prompt_permission` uses this in place of the old + /// `perm_rx.try_lock().is_ok()` check to auto-deny immediately instead of + /// forwarding a request nobody is listening for. + pub fn is_empty(&self) -> bool { + self.slots.lock().unwrap().is_empty() + } + + /// Register a new responder. Returns a private receiver for forwarded + /// requests and an RAII guard — dropping the guard unregisters the + /// responder, so a disconnected transport stops being counted as present. + pub fn register(self: &Arc) -> (ResponderGuard, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(RESPONDER_CHANNEL_CAPACITY); + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + self.slots.lock().unwrap().push(ResponderSlot { id, tx }); + ( + ResponderGuard { + id, + registry: Arc::clone(self), + }, + rx, + ) + } + + /// Dispatch one forwarded request to the first responder that accepts it. + /// + /// Never blocks: uses `try_send` against each registered responder in + /// turn. If every responder's channel is full (or none are registered), + /// the request is fail-closed with [`PermissionDecision::Deny`] rather + /// than dropped silently. + pub fn dispatch(&self, forward: PermissionForward) { + let slots = self.slots.lock().unwrap().clone_senders(); + let mut remaining = forward; + for tx in &slots { + match tx.try_send(remaining) { + Ok(()) => return, + Err(mpsc::error::TrySendError::Full(fwd)) + | Err(mpsc::error::TrySendError::Closed(fwd)) => { + remaining = fwd; + } + } + } + let _ = remaining.response_tx.send(PermissionDecision::Deny); + } + + fn unregister(&self, id: u64) { + self.slots.lock().unwrap().retain(|s| s.id != id); + } +} + +/// Helper so `dispatch` can iterate over a snapshot of senders without +/// holding the `std::sync::Mutex` across the (non-blocking, but still +/// non-trivial) `try_send` calls. +trait CloneSenders { + fn clone_senders(&self) -> Vec>; +} + +impl CloneSenders for Vec { + fn clone_senders(&self) -> Vec> { + self.iter().map(|s| s.tx.clone()).collect() + } +} + +/// RAII guard returned by [`ResponderRegistry::register`]. +/// +/// Unregisters the responder on drop so a disconnected Matrix bot, sled +/// uplink, or WS session stops being counted as an active listener. +pub struct ResponderGuard { + id: u64, + registry: Arc, +} + +impl Drop for ResponderGuard { + fn drop(&mut self) { + self.registry.unregister(self.id); + } +} + +/// Spawn the router task: the sole, permanent owner of the MCP-side receiver. +/// +/// Never awaits responder I/O — only [`ResponderRegistry::dispatch`], which +/// is itself non-blocking. Exits when `perm_rx` closes (server shutdown). +pub fn spawn_permission_router( + mut perm_rx: mpsc::UnboundedReceiver, + registry: Arc, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + while let Some(forward) = perm_rx.recv().await { + registry.dispatch(forward); + } + }) +} + +// ── Pending replies ────────────────────────────────────────────────────── + +/// Tracks permission requests awaiting a plain-language (yes/no) reply from a +/// chat transport, keyed by `request_id` rather than by conversation location +/// (room / sender / channel). +/// +/// Keying by location alone meant a second concurrent request for the same +/// room silently dropped the first request's oneshot sender when it +/// overwrote the map entry. Keying by `request_id` fixes that; a per-location +/// FIFO queue lets [`PendingPermReplies::resolve_oldest`] map a bare "yes"/"no" +/// reply (which carries no request_id) back to the oldest still-pending +/// request for that location. +pub struct PendingPermReplies { + inner: TokioMutex, +} + +#[derive(Default)] +struct PendingInner { + by_request_id: HashMap>, + by_location: HashMap>, +} + +impl PendingPermReplies { + /// Create an empty tracker. + pub fn new() -> Arc { + Arc::new(Self { + inner: TokioMutex::new(PendingInner::default()), + }) + } + + /// Register a pending reply for `request_id`, queued under `location`. + pub async fn insert( + &self, + location: impl Into, + request_id: impl Into, + tx: oneshot::Sender, + ) { + let request_id = request_id.into(); + let mut inner = self.inner.lock().await; + inner + .by_location + .entry(location.into()) + .or_default() + .push_back(request_id.clone()); + inner.by_request_id.insert(request_id, tx); + } + + /// Resolve the oldest pending request queued for `location`, removing it + /// from both the location queue and the request_id map. Used when a chat + /// transport receives a bare yes/no reply with no request_id attached. + pub async fn resolve_oldest( + &self, + location: &str, + ) -> Option> { + let mut inner = self.inner.lock().await; + loop { + let queue = inner.by_location.get_mut(location)?; + let request_id = queue.pop_front()?; + if queue.is_empty() { + inner.by_location.remove(location); + } + if let Some(tx) = inner.by_request_id.remove(&request_id) { + return Some(tx); + } + // request_id was already removed (e.g. by a timeout) — try the + // next queued entry for this location instead of returning None. + } + } + + /// Remove a specific pending reply by `request_id`, also dequeuing it from + /// `location`'s FIFO so a later bare reply doesn't try to resolve an + /// already-decided (e.g. timed-out) request. Used by responder timeout + /// tasks. + pub async fn remove_by_request_id( + &self, + location: &str, + request_id: &str, + ) -> Option> { + let mut inner = self.inner.lock().await; + if let Some(queue) = inner.by_location.get_mut(location) { + queue.retain(|id| id != request_id); + if queue.is_empty() { + inner.by_location.remove(location); + } + } + inner.by_request_id.remove(request_id) + } +} + +// ── Tests ───────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn make_forward( + request_id: &str, + ) -> (PermissionForward, oneshot::Receiver) { + let (tx, rx) = oneshot::channel(); + ( + PermissionForward { + request_id: request_id.to_string(), + tool_name: "Bash".to_string(), + tool_input: json!({"command": "echo hi"}), + response_tx: tx, + }, + rx, + ) + } + + #[tokio::test] + async fn registry_starts_empty() { + let registry = ResponderRegistry::new(); + assert!(registry.is_empty()); + } + + #[tokio::test] + async fn register_makes_registry_non_empty_until_guard_dropped() { + let registry = ResponderRegistry::new(); + let (guard, _rx) = registry.register(); + assert!( + !registry.is_empty(), + "registry must be non-empty while registered" + ); + drop(guard); + assert!( + registry.is_empty(), + "dropping the guard must unregister the responder" + ); + } + + #[tokio::test] + async fn dispatch_delivers_to_registered_responder() { + let registry = ResponderRegistry::new(); + let (_guard, mut rx) = registry.register(); + let (fwd, _response_rx) = make_forward("req-1"); + registry.dispatch(fwd); + let received = rx.recv().await.expect("responder must receive forward"); + assert_eq!(received.request_id, "req-1"); + } + + #[tokio::test] + async fn dispatch_denies_when_no_responder_registered() { + let registry = ResponderRegistry::new(); + let (fwd, response_rx) = make_forward("req-2"); + registry.dispatch(fwd); + let decision = response_rx.await.expect("must receive a decision"); + assert_eq!( + decision, + PermissionDecision::Deny, + "no registered responder must fail-closed" + ); + } + + #[tokio::test] + async fn wedged_responder_falls_through_to_deny_without_blocking() { + let registry = ResponderRegistry::new(); + let (_guard, _rx) = registry.register(); + // Fill the responder's bounded channel without draining it, so it + // looks "wedged" from the router's point of view. + for i in 0..RESPONDER_CHANNEL_CAPACITY { + let (fwd, _rx) = make_forward(&format!("filler-{i}")); + registry.dispatch(fwd); + } + + let (fwd, response_rx) = make_forward("req-overflow"); + // Must return immediately — no await on responder I/O. + registry.dispatch(fwd); + let decision = tokio::time::timeout(std::time::Duration::from_secs(2), response_rx) + .await + .expect("dispatch must not stall when the responder is wedged") + .expect("must receive a decision"); + assert_eq!(decision, PermissionDecision::Deny); + } + + #[tokio::test] + async fn pending_replies_two_concurrent_requests_same_location_both_resolve() { + let pending = PendingPermReplies::new(); + let (tx1, rx1) = oneshot::channel(); + let (tx2, rx2) = oneshot::channel(); + pending.insert("room-1", "req-a", tx1).await; + pending.insert("room-1", "req-b", tx2).await; + + // Regression: previously keying by location alone meant the second + // insert overwrote (and dropped) the first request's sender. + let first = pending + .resolve_oldest("room-1") + .await + .expect("first pending reply must still be present"); + let _ = first.send(PermissionDecision::Approve); + assert_eq!(rx1.await.unwrap(), PermissionDecision::Approve); + + let second = pending + .resolve_oldest("room-1") + .await + .expect("second pending reply must still be present"); + let _ = second.send(PermissionDecision::Deny); + assert_eq!(rx2.await.unwrap(), PermissionDecision::Deny); + } + + #[tokio::test] + async fn remove_by_request_id_prevents_later_resolution() { + let pending = PendingPermReplies::new(); + let (tx, _rx) = oneshot::channel(); + pending.insert("room-1", "req-timeout", tx).await; + + let removed = pending.remove_by_request_id("room-1", "req-timeout").await; + assert!(removed.is_some(), "must return the removed sender"); + + let resolved = pending.resolve_oldest("room-1").await; + assert!( + resolved.is_none(), + "a request removed by timeout must not be resolvable afterwards" + ); + } + + #[tokio::test] + async fn resolve_oldest_skips_already_removed_entries() { + let pending = PendingPermReplies::new(); + let (tx1, _rx1) = oneshot::channel(); + let (tx2, rx2) = oneshot::channel(); + pending.insert("room-1", "req-a", tx1).await; + pending.insert("room-1", "req-b", tx2).await; + + // req-a times out (removed) before the user replies. + pending.remove_by_request_id("room-1", "req-a").await; + + // A bare "yes" reply must resolve req-b, not silently return None. + let tx = pending + .resolve_oldest("room-1") + .await + .expect("must skip the timed-out entry and resolve the next one"); + let _ = tx.send(PermissionDecision::Approve); + assert_eq!(rx2.await.unwrap(), PermissionDecision::Approve); + } + + #[tokio::test] + async fn resolve_oldest_returns_none_for_unknown_location() { + let pending = PendingPermReplies::new(); + assert!(pending.resolve_oldest("no-such-room").await.is_none()); + } +} diff --git a/server/src/services.rs b/server/src/services.rs index 29df2506..b2397b96 100644 --- a/server/src/services.rs +++ b/server/src/services.rs @@ -7,12 +7,11 @@ use crate::agents::AgentPool; use crate::chat::dispatcher::ChatDispatcher; -use crate::http::context::{PermissionDecision, PermissionForward}; +use crate::service::permission_router::{PendingPermReplies, ResponderRegistry}; use crate::service::status::StatusBroadcaster; -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; -use tokio::sync::{Mutex as TokioMutex, mpsc, oneshot}; /// Shared state bundle constructed once at startup and cloned (via `Arc`) into /// every context that needs access to the project root, agent pool, bot @@ -29,11 +28,16 @@ pub struct Services { pub bot_user_id: String, /// Set of room/channel IDs where ambient mode is active. pub ambient_rooms: Arc>>, - /// Receiver for permission requests from the MCP `prompt_permission` tool. - pub perm_rx: Arc>>, - /// Per-room pending permission reply senders, keyed by room/channel ID - /// as a plain string. - pub pending_perm_replies: Arc>>>, + /// Registry of tasks currently registered to receive forwarded MCP + /// `prompt_permission` requests (Matrix bot, sled uplink, WS chat + /// sessions, per-message chat transports). Replaces holding `perm_rx`'s + /// mutex as a presence signal. + pub permission_registry: Arc, + /// Pending permission replies awaiting a plain-language (yes/no) chat + /// reply, keyed by `request_id` with a per-location FIFO index so two + /// concurrent requests for the same room/sender/channel don't drop each + /// other's oneshot sender. + pub pending_perm_replies: Arc, /// Seconds to wait for a user to respond to a permission prompt before /// auto-denying (fail-closed). pub permission_timeout_secs: u64, @@ -58,7 +62,6 @@ impl Services { /// Build a minimal `Services` for testing with the given project root and /// bot display name. pub fn new_test(project_root: std::path::PathBuf, bot_name: String) -> std::sync::Arc { - let (_perm_tx, perm_rx) = mpsc::unbounded_channel(); let agents = std::sync::Arc::new(crate::agents::AgentPool::new_test(3000)); std::sync::Arc::new(Self { project_root, @@ -67,8 +70,8 @@ impl Services { bot_name, bot_user_id: String::new(), ambient_rooms: std::sync::Arc::new(std::sync::Mutex::new(HashSet::new())), - perm_rx: std::sync::Arc::new(TokioMutex::new(perm_rx)), - pending_perm_replies: std::sync::Arc::new(TokioMutex::new(HashMap::new())), + permission_registry: ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, chat_dispatcher: std::sync::Arc::new(ChatDispatcher::new(1_500)), }) diff --git a/server/src/sled_uplink.rs b/server/src/sled_uplink.rs index 4f979cfe..47fa9b62 100644 --- a/server/src/sled_uplink.rs +++ b/server/src/sled_uplink.rs @@ -4,9 +4,10 @@ //! When `HUSKIES_UPSTREAM_GATEWAY` is set (or `--upstream-gateway` is passed //! on the CLI), this module spawns a task that: //! -//! 1. Acquires `services.perm_rx` for its lifetime (matching the Matrix bot's -//! `permission_listener` pattern), preventing `tool_prompt_permission` from -//! auto-denying requests with "no interactive session". +//! 1. Registers with `services.permission_registry` for its lifetime (matching +//! the Matrix bot's `permission_listener` pattern), preventing +//! `tool_prompt_permission` from auto-denying requests with "no +//! interactive session". //! 2. Maintains a persistent WebSocket connection to the gateway's //! `/api/sled-uplink` endpoint. //! 3. Sends an `identity` frame announcing the sled's project name + auth @@ -85,9 +86,10 @@ pub struct UplinkEnvelope { /// /// Does nothing when `config.upstream_url` is empty (AC 8 — sleds without /// an upstream configured continue to work unchanged). When active, the task -/// holds `services.perm_rx` locked for its lifetime (preventing auto-deny in -/// `tool_prompt_permission`) and forwards all permission requests to the -/// gateway. Reconnects automatically with exponential back-off. +/// registers with `services.permission_registry` for its lifetime (preventing +/// auto-deny in `tool_prompt_permission`) and forwards all permission +/// requests to the gateway. Reconnects automatically with exponential +/// back-off. pub fn spawn_uplink_task(config: UplinkConfig, services: Arc) { if config.upstream_url.is_empty() { return; @@ -99,11 +101,12 @@ pub fn spawn_uplink_task(config: UplinkConfig, services: Arc) { } = config; slog!("[uplink] Spawning sled uplink task (gateway={upstream_url}, project={project_name})"); tokio::spawn(async move { - // Acquire perm_rx for this task's entire lifetime. While this lock is - // held, try_lock() inside tool_prompt_permission fails — meaning - // requests flow to perm_tx (which we drain here) rather than auto-deny. - let mut perm_rx = services.perm_rx.lock().await; - slog!("[uplink] Acquired perm_rx; maintaining gateway connection"); + // Register for this task's entire lifetime. While registered, + // `permission_registry.is_empty()` in `tool_prompt_permission` returns + // false — meaning requests flow to perm_tx (which we drain here) + // rather than auto-deny. + let (_responder_guard, mut perm_rx) = services.permission_registry.register(); + slog!("[uplink] Registered as permission responder; maintaining gateway connection"); let http = reqwest::Client::new(); @@ -143,7 +146,7 @@ async fn run_uplink_session( project_name: &str, local_mcp_url: &str, http: &reqwest::Client, - perm_rx: &mut tokio::sync::mpsc::UnboundedReceiver, + perm_rx: &mut tokio::sync::mpsc::Receiver, ) -> Result<(), String> { let (ws_stream, _) = tokio_tungstenite::connect_async(url) .await @@ -187,7 +190,7 @@ async fn pump_messages( impl futures::Stream> + Unpin ), - perm_rx: &mut tokio::sync::mpsc::UnboundedReceiver, + perm_rx: &mut tokio::sync::mpsc::Receiver, in_flight: &mut HashMap>, local_mcp_url: &str, http: &reqwest::Client, @@ -401,12 +404,31 @@ fn fail_close_all(in_flight: &mut HashMap Arc { + let agents = Arc::new(crate::agents::AgentPool::new_test(3000)); + Arc::new(Services { + project_root: std::path::PathBuf::from("/tmp"), + status: agents.status_broadcaster(), + agents, + bot_name: "Test".to_string(), + bot_user_id: String::new(), + ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), + permission_registry: ResponderRegistry::new(), + pending_perm_replies: PendingPermReplies::new(), + permission_timeout_secs: 120, + chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), + }) + } + // ── Pure unit tests ─────────────────────────────────────────────── #[test] @@ -520,20 +542,7 @@ mod tests { #[test] fn spawn_uplink_task_noop_when_url_empty() { - let (_perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); - let agents = Arc::new(crate::agents::AgentPool::new_test(3000)); - let services = Arc::new(Services { - project_root: std::path::PathBuf::from("/tmp"), - status: agents.status_broadcaster(), - agents, - bot_name: "Test".to_string(), - bot_user_id: String::new(), - ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(HashMap::new())), - permission_timeout_secs: 120, - chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), - }); + let services = test_services(); // Empty URL → noop; if it panicked or blocked the test would fail. spawn_uplink_task( UplinkConfig { @@ -592,20 +601,7 @@ mod tests { .unwrap(); }); - let (perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel::(); - let agents = Arc::new(crate::agents::AgentPool::new_test(3000)); - let services = Arc::new(Services { - project_root: std::path::PathBuf::from("/tmp"), - status: agents.status_broadcaster(), - agents, - bot_name: "Test".to_string(), - bot_user_id: String::new(), - ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(HashMap::new())), - permission_timeout_secs: 120, - chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), - }); + let services = test_services(); spawn_uplink_task( UplinkConfig { @@ -618,14 +614,12 @@ mod tests { tokio::time::sleep(std::time::Duration::from_millis(100)).await; let (response_tx, response_rx) = oneshot::channel(); - perm_tx - .send(PermissionForward { - request_id: "req-test-1".to_string(), - tool_name: "Bash".to_string(), - tool_input: serde_json::json!({"command": "echo hello"}), - response_tx, - }) - .unwrap(); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-test-1".to_string(), + tool_name: "Bash".to_string(), + tool_input: serde_json::json!({"command": "echo hello"}), + response_tx, + }); let decision = tokio::time::timeout(std::time::Duration::from_secs(5), response_rx) .await @@ -690,20 +684,7 @@ mod tests { } }); - let (perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel::(); - let agents = Arc::new(crate::agents::AgentPool::new_test(3000)); - let services = Arc::new(Services { - project_root: std::path::PathBuf::from("/tmp"), - status: agents.status_broadcaster(), - agents, - bot_name: "Test".to_string(), - bot_user_id: String::new(), - ambient_rooms: Arc::new(std::sync::Mutex::new(std::collections::HashSet::new())), - perm_rx: Arc::new(tokio::sync::Mutex::new(perm_rx)), - pending_perm_replies: Arc::new(tokio::sync::Mutex::new(HashMap::new())), - permission_timeout_secs: 120, - chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), - }); + let services = test_services(); spawn_uplink_task( UplinkConfig { @@ -717,14 +698,12 @@ mod tests { // First request is sent on the connection that drops → denied. let (tx1, rx1) = oneshot::channel(); - perm_tx - .send(PermissionForward { - request_id: "req-drop".to_string(), - tool_name: "Bash".to_string(), - tool_input: serde_json::json!({}), - response_tx: tx1, - }) - .unwrap(); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-drop".to_string(), + tool_name: "Bash".to_string(), + tool_input: serde_json::json!({}), + response_tx: tx1, + }); let d1 = tokio::time::timeout(std::time::Duration::from_secs(5), rx1) .await @@ -741,14 +720,12 @@ mod tests { // Second request arrives on the reconnected session → approved. let (tx2, rx2) = oneshot::channel(); - perm_tx - .send(PermissionForward { - request_id: "req-reconnect".to_string(), - tool_name: "Write".to_string(), - tool_input: serde_json::json!({}), - response_tx: tx2, - }) - .unwrap(); + services.permission_registry.dispatch(PermissionForward { + request_id: "req-reconnect".to_string(), + tool_name: "Write".to_string(), + tool_input: serde_json::json!({}), + response_tx: tx2, + }); let d2 = tokio::time::timeout(std::time::Duration::from_secs(5), rx2) .await