diff --git a/.mcp.json b/.mcp.json index 91a12906..f85935d7 100644 --- a/.mcp.json +++ b/.mcp.json @@ -2,7 +2,10 @@ "mcpServers": { "huskies": { "type": "http", - "url": "http://localhost:3001/mcp" + "url": "http://localhost:3001/mcp", + "headers": { + "X-Huskies-Session": "1228" + } } } } diff --git a/server/src/agent_mode/context.rs b/server/src/agent_mode/context.rs index 06140d9e..c485b86b 100644 --- a/server/src/agent_mode/context.rs +++ b/server/src/agent_mode/context.rs @@ -66,6 +66,12 @@ pub(super) fn build_agent_app_context( perm_rx, Arc::clone(&permission_registry), ); + let (question_tx, question_rx) = tokio::sync::mpsc::unbounded_channel(); + let question_registry = crate::service::question_router::QuestionResponderRegistry::new(); + crate::service::question_router::spawn_question_router( + question_rx, + Arc::clone(&question_registry), + ); let timer_store = Arc::new(crate::service::timer::TimerStore::load( project_root.join(".huskies").join("timers.json"), )); @@ -83,6 +89,9 @@ pub(super) fn build_agent_app_context( pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry, + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new(), + question_timeout_secs: 120, status: agents.status_broadcaster(), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }); @@ -96,6 +105,7 @@ pub(super) fn build_agent_app_context( watcher_tx, reconciliation_tx, perm_tx, + question_tx, qa_app_process: Arc::new(std::sync::Mutex::new(None)), bot_shutdown: None, matrix_shutdown_tx: None, diff --git a/server/src/chat/transport/matrix/bot/context.rs b/server/src/chat/transport/matrix/bot/context.rs index 15160cc3..a0a96685 100644 --- a/server/src/chat/transport/matrix/bot/context.rs +++ b/server/src/chat/transport/matrix/bot/context.rs @@ -318,6 +318,10 @@ mod tests { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_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/handle_message.rs b/server/src/chat/transport/matrix/bot/messages/handle_message.rs index acd5c586..11802a37 100644 --- a/server/src/chat/transport/matrix/bot/messages/handle_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/handle_message.rs @@ -95,7 +95,8 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( String::new() }; let prompt = format!( - "{event_log_ctx}{seed_prefix}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n{active_project_ctx}\n{}", + "{event_log_ctx}{seed_prefix}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n[{}]\n{active_project_ctx}\n{}", + crate::chat::util::QUESTION_FORMAT_INSTRUCTION, format_user_prompt(&sender, &user_message) ); 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 bea7b867..f728fead 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 @@ -336,6 +336,60 @@ async fn try_handle_stop_command( true } +/// Parse a chat reply to a pending [`QuestionSpec`](crate::http::context::QuestionSpec) +/// into the answer it represents (story 1228, AC2/AC4/AC5). +/// +/// Tries, in order: a comma-separated list of 1-based option numbers (e.g. +/// `"2"` or `"1,3"`, validated against `num_options` and `multi_select`); a +/// case-insensitive exact match against one of `labels`; and finally free +/// text (the always-available "Other" path). Returns `None` only when the +/// reply looks like an attempted numeric selection but is out of range or +/// violates `multi_select` — the caller must re-prompt in that case rather +/// than guessing or picking a default (AC4). +fn parse_question_reply( + body: &str, + num_options: usize, + multi_select: bool, + labels: &[String], +) -> Option { + let body_trimmed = body.trim(); + // Strip a leading "@mention " prefix (e.g. "@timmy 1") entirely, not just + // the '@' character, so a mention-prefixed reply still parses as a + // number/label rather than falling through to free text. + let trimmed = if body_trimmed.starts_with('@') { + body_trimmed + .split_once(char::is_whitespace) + .map(|(_, rest)| rest.trim_start()) + .unwrap_or(body_trimmed) + } else { + body_trimmed + }; + let tokens: Vec<&str> = trimmed.split(',').map(str::trim).collect(); + let parsed_indices: Option> = + tokens.iter().map(|t| t.parse::().ok()).collect(); + match parsed_indices { + Some(indices) if !trimmed.is_empty() => { + if indices.iter().any(|&i| i == 0 || i > num_options) { + None + } else { + let mut zero_based: Vec = indices.into_iter().map(|i| i - 1).collect(); + zero_based.dedup(); + if zero_based.len() > 1 && !multi_select { + None + } else { + Some(crate::http::context::QuestionAnswer::Selected(zero_based)) + } + } + } + _ => match labels.iter().position(|l| l.eq_ignore_ascii_case(trimmed)) { + Some(idx) => Some(crate::http::context::QuestionAnswer::Selected(vec![idx])), + None => Some(crate::http::context::QuestionAnswer::FreeText( + trimmed.to_string(), + )), + }, + } +} + pub(in crate::chat::transport::matrix::bot) async fn on_room_message( ev: OriginalSyncRoomMessageEvent, room: Room, @@ -471,6 +525,59 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( } } + // If there is a pending question for this room, interpret the message as + // an answer instead of starting a new chat (story 1228). Checked before + // the pending-permission block below so the two never conflate (AC4): + // each is a fully independent store, and this only fires when a question + // is actually pending for this room. + if let Some((num_options, multi_select, labels)) = ctx + .services + .pending_question_replies + .peek_oldest_meta(incoming_room_id.as_str()) + .await + { + let outcome = parse_question_reply(&body, num_options, multi_select, &labels); + + match outcome { + None => { + // Invalid attempted selection — re-prompt without consuming + // the pending question (AC4: never strand the agent or + // silently pick a default). + let msg = "I didn't understand that reply. Please reply with a number \ + (or numbers separated by commas), an option's label, or your \ + own answer."; + let html = markdown_to_html(msg); + if let Ok(msg_id) = ctx.transport.send_message(&room_id_str, msg, &html).await + && let Ok(event_id) = msg_id.parse() + { + ctx.bot_sent_event_ids.lock().await.insert(event_id); + } + return; + } + Some(answer) => { + if let Some(tx) = ctx + .services + .pending_question_replies + .resolve_oldest(incoming_room_id.as_str()) + .await + { + let _ = tx.send(Ok(answer)); + } + let confirmation = "Got it — thanks for answering."; + let html = markdown_to_html(confirmation); + if let Ok(msg_id) = ctx + .transport + .send_message(&room_id_str, confirmation, &html) + .await + && let Ok(event_id) = msg_id.parse() + { + ctx.bot_sent_event_ids.lock().await.insert(event_id); + } + return; + } + } + } + // If there is a pending permission prompt for this room, interpret the // message as a yes/no response instead of starting a new chat. { @@ -1496,14 +1603,74 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message( mod tests { use super::{ eval_gateway_overview_command, eval_gateway_status_command, eval_switch_command, - try_handle_compact_command, try_handle_stop_command, + parse_question_reply, try_handle_compact_command, try_handle_stop_command, }; use crate::chat::{ChatTransport, MessageId}; + use crate::http::context::QuestionAnswer; use crate::service::gateway::config::ProjectEntry; use async_trait::async_trait; use std::collections::BTreeMap; use tokio::sync::RwLock; + // -- parse_question_reply (story 1228) ----------------------------------- + + fn sample_labels() -> Vec { + vec!["Fast".to_string(), "Safe".to_string()] + } + + #[test] + fn parse_question_reply_single_number_selects_option() { + let outcome = parse_question_reply("2", 2, false, &sample_labels()); + assert_eq!(outcome, Some(QuestionAnswer::Selected(vec![1]))); + } + + #[test] + fn parse_question_reply_multi_select_comma_list() { + let outcome = parse_question_reply("1,2", 2, true, &sample_labels()); + assert_eq!(outcome, Some(QuestionAnswer::Selected(vec![0, 1]))); + } + + #[test] + fn parse_question_reply_multi_numbers_rejected_when_not_multi_select() { + let outcome = parse_question_reply("1,2", 2, false, &sample_labels()); + assert_eq!(outcome, None, "must re-prompt, not silently pick one"); + } + + #[test] + fn parse_question_reply_out_of_range_number_is_invalid() { + let outcome = parse_question_reply("5", 2, false, &sample_labels()); + assert_eq!(outcome, None); + } + + #[test] + fn parse_question_reply_zero_is_invalid() { + let outcome = parse_question_reply("0", 2, false, &sample_labels()); + assert_eq!(outcome, None); + } + + #[test] + fn parse_question_reply_matches_option_label_case_insensitively() { + let outcome = parse_question_reply("fast", 2, false, &sample_labels()); + assert_eq!(outcome, Some(QuestionAnswer::Selected(vec![0]))); + } + + #[test] + fn parse_question_reply_unmatched_text_is_free_text() { + let outcome = parse_question_reply("Let's do something else", 2, false, &sample_labels()); + assert_eq!( + outcome, + Some(QuestionAnswer::FreeText( + "Let's do something else".to_string() + )) + ); + } + + #[test] + fn parse_question_reply_strips_leading_mention() { + let outcome = parse_question_reply("@timmy 1", 2, false, &sample_labels()); + assert_eq!(outcome, Some(QuestionAnswer::Selected(vec![0]))); + } + /// Regression test: `switch` reads from the live store, not a snapshot Vec. /// /// Seeds an empty store, inserts a project at runtime, then asserts the diff --git a/server/src/chat/transport/matrix/bot/mod.rs b/server/src/chat/transport/matrix/bot/mod.rs index 3ce15a67..3b5cfa85 100644 --- a/server/src/chat/transport/matrix/bot/mod.rs +++ b/server/src/chat/transport/matrix/bot/mod.rs @@ -12,6 +12,10 @@ pub mod messages; /// 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; +/// Question listener — registers as a question responder for the bot's +/// lifetime and forwards `ask_question` requests to the configured Matrix +/// room, rendered as numbered text (story 1228). +pub mod question_listener; /// Bot run loop — the main async task that drives the Matrix sync loop. pub mod run; /// Device verification — handles Matrix cross-signing and emoji verification flows. diff --git a/server/src/chat/transport/matrix/bot/permission_listener.rs b/server/src/chat/transport/matrix/bot/permission_listener.rs index aa06a465..0441bcb2 100644 --- a/server/src/chat/transport/matrix/bot/permission_listener.rs +++ b/server/src/chat/transport/matrix/bot/permission_listener.rs @@ -174,6 +174,10 @@ mod tests { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_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/question_listener.rs b/server/src/chat/transport/matrix/bot/question_listener.rs new file mode 100644 index 00000000..a99fa665 --- /dev/null +++ b/server/src/chat/transport/matrix/bot/question_listener.rs @@ -0,0 +1,335 @@ +//! Background task that registers as a question responder for the bot's +//! lifetime and forwards `ask_question` requests to the configured Matrix +//! room, rendering them as numbered text (story 1228). +//! +//! Mirrors `permission_listener.rs`'s always-on registration pattern so an +//! agent's question reaches chat even when the bot isn't mid-turn. + +use crate::chat::ChatTransport; +use crate::http::context::QuestionSpec; +use crate::services::Services; +use crate::slog; +use matrix_sdk::ruma::{OwnedEventId, OwnedRoomId}; +use std::collections::HashSet; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::Mutex as TokioMutex; + +use super::format::markdown_to_html; + +/// Render a [`QuestionSpec`] as readable, numbered chat text — the question, +/// then each option numbered with its label and description — never raw +/// JSON (story 1228, AC1). +pub fn format_question_as_text(question: &QuestionSpec) -> String { + let mut out = format!("**{}**\n\n{}\n\n", question.header, question.question); + for (i, opt) in question.options.iter().enumerate() { + out.push_str(&format!("{}. {} — {}\n", i + 1, opt.label, opt.description)); + } + out.push('\n'); + if question.multi_select { + out.push_str( + "Reply with a number, or multiple numbers separated by commas (e.g. \"1,3\"), \ + or type your own answer.", + ); + } else { + out.push_str("Reply with a number, or type your own answer."); + } + out +} + +/// Spawn a background task that registers with `services.question_registry` +/// for the bot's lifetime and forwards each incoming question request to +/// `target_room` as a chat message. Replies are resolved by `on_room_message` +/// via `pending_question_replies`. +pub fn spawn_question_listener( + services: Arc, + transport: Arc, + target_room: OwnedRoomId, + bot_sent_event_ids: Arc>>, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let (_responder_guard, mut question_rx) = services.question_registry.register(); + let target_room_str = target_room.as_str().to_string(); + slog!("[matrix-bot] question listener started; forwarding requests to {target_room_str}"); + + while let Some(q_fwd) = question_rx.recv().await { + let prompt_msg = format_question_as_text(&q_fwd.question); + let html = markdown_to_html(&prompt_msg); + if let Ok(msg_id) = transport + .send_message(&target_room_str, &prompt_msg, &html) + .await + && let Ok(event_id) = msg_id.parse::() + { + bot_sent_event_ids.lock().await.insert(event_id); + } + + let num_options = q_fwd.question.options.len(); + let multi_select = q_fwd.question.multi_select; + let labels: Vec = q_fwd + .question + .options + .iter() + .map(|o| o.label.clone()) + .collect(); + services + .pending_question_replies + .insert( + target_room.to_string(), + q_fwd.request_id.clone(), + num_options, + multi_select, + labels, + q_fwd.response_tx, + ) + .await; + + // Spawn a per-request timeout: give up (fail-closed) if the user + // does not respond within `question_timeout_secs`. + let pending = Arc::clone(&services.pending_question_replies); + let timeout_room_key = target_room.to_string(); + let timeout_request_id = q_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.question_timeout_secs; + tokio::spawn(async move { + tokio::time::sleep(Duration::from_secs(timeout_secs)).await; + if let Some(tx) = pending + .remove_by_request_id(&timeout_room_key, &timeout_request_id) + .await + { + let _ = tx.send(Err("Question timed out waiting for a reply.".to_string())); + let msg = "Question timed out — no answer received."; + let html = markdown_to_html(msg); + if let Ok(msg_id) = timeout_transport + .send_message(&timeout_room_str, msg, &html) + .await + && let Ok(event_id) = msg_id.parse::() + { + timeout_sent_ids.lock().await.insert(event_id); + } + } + }); + } + + slog!("[matrix-bot] question listener exiting (channel closed)"); + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::http::context::{QuestionAnswer, QuestionForward, QuestionOption}; + use async_trait::async_trait; + + struct RecordingTransport { + sent: Arc>>, + } + + #[async_trait] + impl crate::chat::ChatTransport for RecordingTransport { + async fn send_message( + &self, + room_id: &str, + plain: &str, + _html: &str, + ) -> Result { + self.sent + .lock() + .unwrap() + .push((room_id.to_string(), plain.to_string())); + Ok("$test_event_id:example.com".to_string()) + } + + async fn edit_message(&self, _: &str, _: &str, _: &str, _: &str) -> Result<(), String> { + Ok(()) + } + + async fn send_typing(&self, _: &str, _: bool) -> Result<(), String> { + Ok(()) + } + } + + 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())), + permission_registry: crate::service::permission_router::ResponderRegistry::new(), + pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), + permission_timeout_secs: 120, + remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_timeout_secs: 120, + status: Arc::new(crate::service::status::StatusBroadcaster::new()), + chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), + }) + } + + fn sample_question(multi_select: bool) -> QuestionSpec { + QuestionSpec { + header: "Approach".to_string(), + question: "Which approach should we use?".to_string(), + options: vec![ + QuestionOption { + label: "Fast".to_string(), + description: "Ship quickly".to_string(), + }, + QuestionOption { + label: "Safe".to_string(), + description: "Take more time".to_string(), + }, + ], + multi_select, + } + } + + #[test] + fn format_question_as_text_renders_numbered_options_never_json() { + let text = format_question_as_text(&sample_question(false)); + assert!(text.contains("**Approach**")); + assert!(text.contains("Which approach should we use?")); + assert!(text.contains("1. Fast — Ship quickly")); + assert!(text.contains("2. Safe — Take more time")); + assert!(!text.contains('{'), "must never render raw JSON: {text}"); + } + + #[test] + fn format_question_as_text_mentions_multi_select_syntax_when_enabled() { + let text = format_question_as_text(&sample_question(true)); + assert!(text.contains("1,3")); + } + + #[tokio::test] + async fn listener_forwards_question_to_target_room_and_registers_pending_reply() { + 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 = "!question:example.com".parse().unwrap(); + let bot_sent_event_ids = Arc::new(TokioMutex::new(HashSet::new())); + + spawn_question_listener( + Arc::clone(&services), + Arc::clone(&transport), + target_room.clone(), + Arc::clone(&bot_sent_event_ids), + ); + + assert!( + wait_until( + || !services.question_registry.is_empty(), + std::time::Duration::from_secs(2) + ) + .await, + "listener never registered as a responder" + ); + + let (response_tx, _response_rx) = tokio::sync::oneshot::channel(); + services.question_registry.dispatch(QuestionForward { + request_id: "req-1".to_string(), + question: sample_question(false), + response_tx, + }); + + assert!( + wait_until( + || !sent.lock().unwrap().is_empty(), + std::time::Duration::from_secs(2) + ) + .await, + "listener never sent the question prompt" + ); + + let recorded = sent.lock().unwrap().clone(); + assert_eq!(recorded[0].0, target_room.as_str()); + assert!(recorded[0].1.contains("1. Fast — Ship quickly")); + + let mut resolved = None; + for _ in 0..50 { + resolved = services + .pending_question_replies + .peek_oldest_meta(target_room.as_str()) + .await; + if resolved.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + assert_eq!( + resolved, + Some((2, false, vec!["Fast".to_string(), "Safe".to_string()])), + "pending_question_replies missing entry with correct option count" + ); + } + + #[tokio::test] + async fn resolved_reply_delivers_answer_through_response_channel() { + 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 = "!question2:example.com".parse().unwrap(); + let bot_sent_event_ids = Arc::new(TokioMutex::new(HashSet::new())); + + spawn_question_listener( + Arc::clone(&services), + Arc::clone(&transport), + target_room.clone(), + Arc::clone(&bot_sent_event_ids), + ); + assert!( + wait_until( + || !services.question_registry.is_empty(), + std::time::Duration::from_secs(2) + ) + .await + ); + + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + services.question_registry.dispatch(QuestionForward { + request_id: "req-2".to_string(), + question: sample_question(false), + response_tx, + }); + + let mut tx = None; + for _ in 0..50 { + tx = services + .pending_question_replies + .resolve_oldest(target_room.as_str()) + .await; + if tx.is_some() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(5)).await; + } + let tx = tx.expect("pending reply must be present"); + let _ = tx.send(Ok(QuestionAnswer::Selected(vec![1]))); + assert_eq!( + response_rx.await.unwrap().unwrap(), + QuestionAnswer::Selected(vec![1]) + ); + } +} diff --git a/server/src/chat/transport/matrix/bot/run.rs b/server/src/chat/transport/matrix/bot/run.rs index f598ffb4..46df9ddf 100644 --- a/server/src/chat/transport/matrix/bot/run.rs +++ b/server/src/chat/transport/matrix/bot/run.rs @@ -306,6 +306,12 @@ pub async fn run_bot( target_room.clone(), Arc::clone(&bot_sent_event_ids), ); + super::question_listener::spawn_question_listener( + Arc::clone(&services), + Arc::clone(&transport), + target_room.clone(), + Arc::clone(&bot_sent_event_ids), + ); } // The forwarder only needs live (future) events — resubscribe is fine. diff --git a/server/src/chat/transport/matrix/health.rs b/server/src/chat/transport/matrix/health.rs index 2ccfce93..5713a6ec 100644 --- a/server/src/chat/transport/matrix/health.rs +++ b/server/src/chat/transport/matrix/health.rs @@ -625,6 +625,10 @@ mod tests { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_timeout_secs: 120, status: Arc::new(crate::service::status::StatusBroadcaster::new()), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }); @@ -657,6 +661,10 @@ mod tests { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_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/whatsapp/commands/mod.rs b/server/src/chat/transport/whatsapp/commands/mod.rs index 8df30dde..6f85a786 100644 --- a/server/src/chat/transport/whatsapp/commands/mod.rs +++ b/server/src/chat/transport/whatsapp/commands/mod.rs @@ -315,6 +315,10 @@ mod tests { pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_timeout_secs: 120, chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }); Arc::new(WhatsAppWebhookContext { diff --git a/server/src/chat/util.rs b/server/src/chat/util.rs index 7c313d28..5b509cbb 100644 --- a/server/src/chat/util.rs +++ b/server/src/chat/util.rs @@ -3,6 +3,23 @@ //! These functions are transport-agnostic helpers for processing chat messages: //! prefix stripping, bot-mention handling, and paragraph buffering. +/// Instruction prepended to chat-bot prompts telling the model how to ask a +/// clarifying multiple-choice question over a chat protocol that has no +/// interactive question UI (story 1228). +/// +/// Claude Code's built-in `AskUserQuestion` tool requires an interactive +/// terminal/IDE to render and answer; in the headless `claude -p` sessions +/// this server spawns for chat transports it either isn't offered to the +/// model at all, or (if forced into scope) degrades into the model emitting +/// its attempted call as raw tool-call syntax in plain text. Disallowing it +/// (`--disallowedTools AskUserQuestion` on the CLI invocation) closes that +/// leak; this instruction points the model at the huskies-owned MCP +/// `ask_question` tool as the replacement, which renders the numbered +/// options in chat, blocks until a reply resolves it, and returns the +/// answer directly to this tool call — the model does not need to format +/// the question itself or wait for a future turn to see the reply. +pub const QUESTION_FORMAT_INSTRUCTION: &str = "This chat has no interactive question UI, and the built-in AskUserQuestion tool is unavailable here. When you need to ask the user a clarifying multiple-choice question, call the `ask_question` MCP tool instead — never emit JSON or tool-call syntax as plain text. Pass a `question`, an optional `header`, and at least two `options` (each with a `label` and `description`); set `multi_select: true` if the user may choose more than one. The tool renders the options as a numbered list in chat and blocks until the user replies, then returns the selected option(s) or free-text answer for you to continue with."; + /// Truncate `s` to at most `max_bytes` bytes without splitting a UTF-8 codepoint. /// /// If `s.len() <= max_bytes` the original slice is returned unchanged. @@ -318,6 +335,29 @@ mod tests { assert_eq!(truncate_at_char_boundary("hi", 100), "hi"); } + // -- QUESTION_FORMAT_INSTRUCTION (story 1228) --------------------------- + + #[test] + fn question_format_instruction_directs_model_to_ask_question_tool() { + assert!(QUESTION_FORMAT_INSTRUCTION.contains("ask_question")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("AskUserQuestion")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("never emit JSON or tool-call syntax")); + } + + #[test] + fn question_format_instruction_describes_required_arguments() { + assert!(QUESTION_FORMAT_INSTRUCTION.contains("question")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("options")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("label")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("multi_select")); + } + + #[test] + fn question_format_instruction_explains_blocking_reply_behavior() { + assert!(QUESTION_FORMAT_INSTRUCTION.contains("blocks until the user replies")); + assert!(QUESTION_FORMAT_INSTRUCTION.contains("free-text")); + } + // -- is_permission_approval --------------------------------------------- #[test] diff --git a/server/src/http/context.rs b/server/src/http/context.rs index 9b27748a..913164a8 100644 --- a/server/src/http/context.rs +++ b/server/src/http/context.rs @@ -37,6 +37,46 @@ pub struct PermissionForward { pub response_tx: oneshot::Sender, } +/// A single selectable choice within a [`QuestionSpec`]. +#[derive(Debug, Clone)] +pub struct QuestionOption { + pub label: String, + pub description: String, +} + +/// A multiple-choice question forwarded from the MCP `ask_question` tool to a +/// chat transport for rendering as numbered text (story 1228). +#[derive(Debug, Clone)] +pub struct QuestionSpec { + pub header: String, + pub question: String, + pub options: Vec, + pub multi_select: bool, +} + +/// The user's reply to a forwarded [`QuestionSpec`]. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum QuestionAnswer { + /// 0-based indices into `QuestionSpec::options` the user selected. + Selected(Vec), + /// Freeform text the user typed instead of selecting a listed option + /// (the always-available "Other" path, AC5). + FreeText(String), +} + +/// A question request forwarded from the MCP `ask_question` tool to the +/// active chat transport. The MCP handler blocks on `response_tx` until a +/// chat reply resolves it (or it times out). +/// +/// Kept structurally separate from `PermissionForward` / permission-router +/// plumbing (see `service::question_router`) so a reply answering one is +/// never misinterpreted as answering the other (story 1228, AC4). +pub struct QuestionForward { + pub request_id: String, + pub question: QuestionSpec, + pub response_tx: oneshot::Sender>, +} + #[derive(Clone)] /// Shared application state threaded through all HTTP handlers via Poem's `Data` extractor. pub struct AppContext { @@ -56,6 +96,10 @@ pub struct AppContext { /// `prompt_permission` tool. The MCP handler sends a [`PermissionForward`] /// and awaits the oneshot response. pub perm_tx: mpsc::UnboundedSender, + /// Sender for questions originating from the MCP `ask_question` tool. + /// The MCP handler sends a [`QuestionForward`] and awaits the oneshot + /// response (story 1228). + pub question_tx: mpsc::UnboundedSender, /// Child process of the QA app launched for manual testing. /// Only one instance runs at a time. pub qa_app_process: Arc>>, @@ -101,6 +145,8 @@ impl AppContext { let (reconciliation_tx, _) = broadcast::channel(64); let (perm_tx, perm_rx) = mpsc::unbounded_channel(); let permission_registry = crate::service::permission_router::ResponderRegistry::new(); + let (question_tx, question_rx) = mpsc::unbounded_channel(); + let question_registry = crate::service::question_router::QuestionResponderRegistry::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. @@ -109,6 +155,10 @@ impl AppContext { perm_rx, Arc::clone(&permission_registry), ); + crate::service::question_router::spawn_question_router( + question_rx, + Arc::clone(&question_registry), + ); } let timer_store = Arc::new(TimerStore::load( project_root.join(".huskies").join("timers.json"), @@ -130,6 +180,10 @@ impl AppContext { pending_perm_replies: crate::service::permission_router::PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry, + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_timeout_secs: 120, status: agents.status_broadcaster(), chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }); @@ -141,6 +195,7 @@ impl AppContext { watcher_tx, reconciliation_tx, perm_tx, + question_tx, qa_app_process: Arc::new(std::sync::Mutex::new(None)), bot_shutdown: None, matrix_shutdown_tx: None, diff --git a/server/src/http/mcp/diagnostics/mod.rs b/server/src/http/mcp/diagnostics/mod.rs index d3d387f6..4d456f18 100644 --- a/server/src/http/mcp/diagnostics/mod.rs +++ b/server/src/http/mcp/diagnostics/mod.rs @@ -7,10 +7,12 @@ use serde_json::{Value, json}; mod chat_telemetry; mod permission; +mod question; mod usage; pub(crate) use chat_telemetry::tool_chat_telemetry; pub(crate) use permission::tool_prompt_permission; +pub(crate) use question::tool_ask_question; pub(crate) use usage::tool_get_token_usage; pub(crate) fn tool_get_server_logs(args: &Value) -> Result { diff --git a/server/src/http/mcp/diagnostics/question.rs b/server/src/http/mcp/diagnostics/question.rs new file mode 100644 index 00000000..c7014710 --- /dev/null +++ b/server/src/http/mcp/diagnostics/question.rs @@ -0,0 +1,229 @@ +//! MCP `ask_question` tool — presents a multiple-choice question to the user +//! via chat transports that lack a native question UI (story 1228). + +use serde_json::{Value, json}; + +use crate::http::context::{ + AppContext, QuestionAnswer, QuestionForward, QuestionOption, QuestionSpec, +}; + +/// MCP tool called by an agent to ask the user a multiple-choice question. +/// +/// Forwards the question through the shared channel to the active chat +/// transport (currently Matrix), which renders it as numbered text. Blocks +/// until a reply resolves it or the question times out. +pub(crate) async fn tool_ask_question(args: &Value, ctx: &AppContext) -> Result { + let header = args + .get("header") + .and_then(|v| v.as_str()) + .unwrap_or("Question") + .to_string(); + let question_text = args + .get("question") + .and_then(|v| v.as_str()) + .ok_or_else(|| "Missing required argument: question".to_string())? + .to_string(); + let multi_select = args + .get("multi_select") + .and_then(|v| v.as_bool()) + .unwrap_or(false); + let options: Vec = args + .get("options") + .and_then(|v| v.as_array()) + .ok_or_else(|| "Missing required argument: options".to_string())? + .iter() + .map(|o| QuestionOption { + label: o + .get("label") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + description: o + .get("description") + .and_then(|v| v.as_str()) + .unwrap_or_default() + .to_string(), + }) + .collect(); + + if options.len() < 2 { + return Err("options must contain at least 2 entries".to_string()); + } + + let question = QuestionSpec { + header, + question: question_text, + options, + multi_select, + }; + + if ctx.services.question_registry.is_empty() { + crate::slog!("[question] No interactive session active — cannot ask question"); + return serde_json::to_string_pretty(&json!({ + "answered": false, + "message": "No interactive session active. Nobody is available to answer this question." + })) + .map_err(|e| format!("Serialization error: {e}")); + } + + let request_id = uuid::Uuid::new_v4().to_string(); + let (response_tx, response_rx) = tokio::sync::oneshot::channel(); + let option_labels: Vec = question.options.iter().map(|o| o.label.clone()).collect(); + + if ctx + .question_tx + .send(QuestionForward { + request_id: request_id.clone(), + question, + response_tx, + }) + .is_err() + { + return serde_json::to_string_pretty(&json!({ + "answered": false, + "message": "Failed to forward question — no active session." + })) + .map_err(|e| format!("Serialization error: {e}")); + } + + let timeout_secs = ctx.services.question_timeout_secs; + let outcome = tokio::time::timeout(std::time::Duration::from_secs(timeout_secs), response_rx) + .await + .map_err(|_| format!("Question timed out after {timeout_secs} seconds with no reply"))? + .map_err(|_| "Question response channel closed unexpectedly".to_string())?; + + match outcome { + Ok(QuestionAnswer::Selected(indices)) => { + let labels: Vec<&String> = indices + .iter() + .filter_map(|&i| option_labels.get(i)) + .collect(); + serde_json::to_string_pretty(&json!({ + "answered": true, + "selected_indices": indices, + "selected_labels": labels, + })) + .map_err(|e| format!("Serialization error: {e}")) + } + Ok(QuestionAnswer::FreeText(text)) => serde_json::to_string_pretty(&json!({ + "answered": true, + "free_text": text, + })) + .map_err(|e| format!("Serialization error: {e}")), + Err(message) => serde_json::to_string_pretty(&json!({ + "answered": false, + "message": message, + })) + .map_err(|e| format!("Serialization error: {e}")), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::http::test_helpers::test_ctx; + + #[tokio::test] + async fn tool_ask_question_no_interactive_session_returns_not_answered() { + let tmp = tempfile::tempdir().unwrap(); + let ctx = test_ctx(tmp.path()); + + let result = tool_ask_question( + &json!({ + "question": "Which approach?", + "options": [ + {"label": "A", "description": "First"}, + {"label": "B", "description": "Second"} + ] + }), + &ctx, + ) + .await + .expect("must return Ok even when no session is active"); + + let parsed: Value = serde_json::from_str(&result).unwrap(); + assert_eq!(parsed["answered"], false); + } + + #[tokio::test] + async fn tool_ask_question_requires_at_least_two_options() { + let tmp = tempfile::tempdir().unwrap(); + let ctx = test_ctx(tmp.path()); + + let result = tool_ask_question( + &json!({ + "question": "Pick one?", + "options": [{"label": "Only", "description": "one"}] + }), + &ctx, + ) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn tool_ask_question_selected_answer_returns_labels() { + let tmp = tempfile::tempdir().unwrap(); + let ctx = test_ctx(tmp.path()); + + let (guard, mut rx) = ctx.services.question_registry.register(); + tokio::spawn(async move { + if let Some(forward) = rx.recv().await { + let _ = forward + .response_tx + .send(Ok(QuestionAnswer::Selected(vec![1]))); + } + drop(guard); + }); + + let result = tool_ask_question( + &json!({ + "question": "Which approach?", + "options": [ + {"label": "A", "description": "First"}, + {"label": "B", "description": "Second"} + ] + }), + &ctx, + ) + .await + .expect("should succeed"); + + let parsed: Value = serde_json::from_str(&result).unwrap(); + assert_eq!(parsed["answered"], true); + assert_eq!(parsed["selected_labels"][0], "B"); + } + + #[tokio::test] + async fn tool_ask_question_free_text_answer() { + let tmp = tempfile::tempdir().unwrap(); + let ctx = test_ctx(tmp.path()); + + let (guard, mut rx) = ctx.services.question_registry.register(); + tokio::spawn(async move { + if let Some(forward) = rx.recv().await { + let _ = forward + .response_tx + .send(Ok(QuestionAnswer::FreeText("Something else".to_string()))); + } + drop(guard); + }); + + let result = tool_ask_question( + &json!({ + "question": "Which approach?", + "options": [ + {"label": "A", "description": "First"}, + {"label": "B", "description": "Second"} + ] + }), + &ctx, + ) + .await + .expect("should succeed"); + + let parsed: Value = serde_json::from_str(&result).unwrap(); + assert_eq!(parsed["answered"], true); + assert_eq!(parsed["free_text"], "Something else"); + } +} diff --git a/server/src/http/mcp/dispatch.rs b/server/src/http/mcp/dispatch.rs index b4d3b288..201e2449 100644 --- a/server/src/http/mcp/dispatch.rs +++ b/server/src/http/mcp/dispatch.rs @@ -86,6 +86,7 @@ pub async fn dispatch_tool_call( "get_version" => diagnostics::tool_get_version(ctx), // Permission bridge (Claude Code → frontend dialog) "prompt_permission" => diagnostics::tool_prompt_permission(&args, ctx).await, + "ask_question" => diagnostics::tool_ask_question(&args, ctx).await, // Token usage "get_token_usage" => diagnostics::tool_get_token_usage(&args, ctx), // Chat turn telemetry (story 1209) diff --git a/server/src/http/mcp/tools_list/mod.rs b/server/src/http/mcp/tools_list/mod.rs index 70d7bef3..9aff4232 100644 --- a/server/src/http/mcp/tools_list/mod.rs +++ b/server/src/http/mcp/tools_list/mod.rs @@ -120,7 +120,8 @@ mod tests { assert!(names.contains(&"write")); assert!(names.contains(&"gc")); assert!(names.contains(&"chat_telemetry")); - assert_eq!(tools.len(), 88); + assert!(names.contains(&"ask_question")); + assert_eq!(tools.len(), 89); } #[test] diff --git a/server/src/http/mcp/tools_list/system_tools.rs b/server/src/http/mcp/tools_list/system_tools.rs index b2e8089b..f62c8661 100644 --- a/server/src/http/mcp/tools_list/system_tools.rs +++ b/server/src/http/mcp/tools_list/system_tools.rs @@ -60,6 +60,40 @@ pub(super) fn system_tools() -> Vec { "required": ["tool_name", "input"] } }), + json!({ + "name": "ask_question", + "description": "Ask the user a multiple-choice question via the active chat transport (e.g. Matrix). Renders as numbered options in chat protocols without a native question UI; the reply is parsed back (a number, a comma-separated list for multi_select, an option label, or free text) and returned here. Blocks until answered or timed out.", + "inputSchema": { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question to ask, e.g. 'Which approach should we use?'" + }, + "header": { + "type": "string", + "description": "Short label for the question (e.g. 'Approach'). Defaults to 'Question'." + }, + "options": { + "type": "array", + "items": { + "type": "object", + "properties": { + "label": {"type": "string"}, + "description": {"type": "string"} + }, + "required": ["label", "description"] + }, + "description": "At least 2 selectable options, each with a label and description." + }, + "multi_select": { + "type": "boolean", + "description": "If true, the user may select multiple options (e.g. reply '1,3'). Default false." + } + }, + "required": ["question", "options"] + } + }), json!({ "name": "get_token_usage", "description": "Return per-agent token usage records from the persistent log. Shows input tokens, output tokens, cache tokens, and cost in USD for each agent session. Optionally filter by story_id.", diff --git a/server/src/llm/providers/claude_code/mod.rs b/server/src/llm/providers/claude_code/mod.rs index 2a05b5cb..6cb3642a 100644 --- a/server/src/llm/providers/claude_code/mod.rs +++ b/server/src/llm/providers/claude_code/mod.rs @@ -292,6 +292,14 @@ fn run_pty_session( // a tool requires user approval, instead of using PTY stdin/stdout. cmd.arg("--permission-prompt-tool"); cmd.arg("mcp__huskies__prompt_permission"); + // AskUserQuestion requires an interactive terminal/IDE to render and + // answer; every session this provider spawns is headless (stdin is + // dropped below), so the tool is either unavailable or, if the model + // attempts it anyway, degrades into raw tool-call syntax leaking into + // the output text. Disallowing it forces the model to ask clarifying + // questions as plain text instead (story 1228). + cmd.arg("--disallowedTools"); + cmd.arg("AskUserQuestion"); // Note: --system is not a valid Claude Code CLI flag. System-level // instructions (like bot name) are prepended to the user prompt instead. cmd.cwd(cwd); diff --git a/server/src/main.rs b/server/src/main.rs index ec8dee4c..4a2edb73 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -219,6 +219,7 @@ async fn main() -> Result<(), std::io::Error> { // Reconciliation progress channel and permission channel. let (reconciliation_tx, _) = broadcast::channel::(64); let (perm_tx, perm_rx) = tokio::sync::mpsc::unbounded_channel(); + let (question_tx, question_rx) = tokio::sync::mpsc::unbounded_channel(); let watcher_tx_for_bot = watcher_tx.clone(); let watcher_rx_for_whatsapp = watcher_tx.subscribe(); @@ -227,6 +228,8 @@ async fn main() -> Result<(), std::io::Error> { let permission_registry = service::permission_router::ResponderRegistry::new(); service::permission_router::spawn_permission_router(perm_rx, Arc::clone(&permission_registry)); + let question_registry = service::question_router::QuestionResponderRegistry::new(); + service::question_router::spawn_question_router(question_rx, Arc::clone(&question_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(); @@ -257,6 +260,12 @@ async fn main() -> Result<(), std::io::Error> { .map(|c| c.permission_timeout_secs) .unwrap_or(120), remembered_permissions: service::permission_router::RememberedPermissions::new(), + question_registry: Arc::clone(&question_registry), + pending_question_replies: service::question_router::PendingQuestionReplies::new(), + question_timeout_secs: bot_cfg + .as_ref() + .map(|c| c.permission_timeout_secs) + .unwrap_or(120), status: agents.status_broadcaster(), chat_dispatcher: std::sync::Arc::new(chat::dispatcher::ChatDispatcher::new( bot_cfg @@ -359,6 +368,7 @@ async fn main() -> Result<(), std::io::Error> { watcher_tx, reconciliation_tx, perm_tx, + question_tx, qa_app_process: Arc::new(std::sync::Mutex::new(None)), bot_shutdown: bot_ctxs.shutdown_notifier.clone(), matrix_shutdown_tx: Some(Arc::clone(&bot_ctxs.matrix_shutdown_tx)), diff --git a/server/src/service/bot_command/io.rs b/server/src/service/bot_command/io.rs index a15df154..a1a4244d 100644 --- a/server/src/service/bot_command/io.rs +++ b/server/src/service/bot_command/io.rs @@ -142,6 +142,9 @@ pub(super) fn call_sync( pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new(), + question_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 68a93c03..9356a942 100644 --- a/server/src/service/gateway/io.rs +++ b/server/src/service/gateway/io.rs @@ -742,6 +742,9 @@ pub fn spawn_gateway_bot( .map(|c| c.permission_timeout_secs) .unwrap_or(120), remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new(), + question_timeout_secs: 120, chat_dispatcher: std::sync::Arc::new(crate::chat::dispatcher::ChatDispatcher::new( bot_cfg .as_ref() diff --git a/server/src/service/mod.rs b/server/src/service/mod.rs index f5aa681c..b679a369 100644 --- a/server/src/service/mod.rs +++ b/server/src/service/mod.rs @@ -48,6 +48,10 @@ pub mod pipeline; pub mod project; /// QA — request, approve, and reject code reviews. pub mod qa; +/// Question router — responder registry and pending-reply tracking for the +/// MCP `ask_question` tool, kept separate from `permission_router` so a +/// question answer is never conflated with a permission decision (story 1228). +pub mod question_router; /// Project settings read/write and validation. pub mod settings; /// Shell command safety, sandboxing, and output helpers. diff --git a/server/src/service/question_router.rs b/server/src/service/question_router.rs new file mode 100644 index 00000000..54d64859 --- /dev/null +++ b/server/src/service/question_router.rs @@ -0,0 +1,401 @@ +//! Question router — mirrors `permission_router.rs`'s `ResponderRegistry` / +//! pending-reply pattern for the MCP `ask_question` tool (story 1228). +//! +//! Kept as a fully separate registry and pending-reply store from +//! `permission_router.rs` rather than reusing those types with a flag: AC4 of +//! story 1228 requires that a chat reply answering a permission prompt is +//! never treated as answering a pending question (and vice versa). Two +//! independent stores make that conflation structurally impossible instead of +//! relying on careful conditionals over a shared one. + +use crate::http::context::{QuestionAnswer, QuestionForward}; +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. +pub const QUESTION_RESPONDER_CHANNEL_CAPACITY: usize = 16; + +struct ResponderSlot { + id: u64, + tx: mpsc::Sender, +} + +/// Tracks which tasks are currently registered to receive forwarded question +/// requests. Mirrors `permission_router::ResponderRegistry` exactly, but for +/// `ask_question` forwards instead of `prompt_permission` ones. +pub struct QuestionResponderRegistry { + next_id: AtomicU64, + slots: StdMutex>, +} + +impl QuestionResponderRegistry { + /// 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_ask_question` uses this to fail closed immediately (returning a + /// "no interactive session" result to the agent) instead of forwarding a + /// question 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. + pub fn register(self: &Arc) -> (QuestionResponderGuard, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(QUESTION_RESPONDER_CHANNEL_CAPACITY); + let id = self.next_id.fetch_add(1, Ordering::Relaxed); + self.slots.lock().unwrap().push(ResponderSlot { id, tx }); + ( + QuestionResponderGuard { + 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 an error result rather than dropped + /// silently. + pub fn dispatch(&self, forward: QuestionForward) { + 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(Err( + "No interactive session is available to answer this question.".to_string(), + )); + } + + fn unregister(&self, id: u64) { + self.slots.lock().unwrap().retain(|s| s.id != id); + } +} + +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 [`QuestionResponderRegistry::register`]. +pub struct QuestionResponderGuard { + id: u64, + registry: Arc, +} + +impl Drop for QuestionResponderGuard { + 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 [`QuestionResponderRegistry::dispatch`], +/// which is itself non-blocking. Exits when `question_rx` closes (server +/// shutdown). +pub fn spawn_question_router( + mut question_rx: mpsc::UnboundedReceiver, + registry: Arc, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + while let Some(forward) = question_rx.recv().await { + registry.dispatch(forward); + } + }) +} + +// ── Pending replies ────────────────────────────────────────────────────── + +/// A pending question awaiting a chat reply, together with the metadata a +/// transport needs to parse that reply (option count, single/multi-select). +struct PendingQuestion { + tx: oneshot::Sender>, + num_options: usize, + multi_select: bool, + labels: Vec, +} + +/// Tracks questions awaiting a chat reply, keyed by `request_id` with a +/// per-location FIFO index — mirrors `permission_router::PendingPermReplies`. +pub struct PendingQuestionReplies { + inner: TokioMutex, +} + +#[derive(Default)] +struct PendingInner { + by_request_id: HashMap, + by_location: HashMap>, +} + +impl PendingQuestionReplies { + /// 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, + num_options: usize, + multi_select: bool, + labels: Vec, + 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, + PendingQuestion { + tx, + num_options, + multi_select, + labels, + }, + ); + } + + /// Peek the metadata (option count, multi-select, option labels) of the + /// oldest pending question queued for `location`, without consuming it. + /// Used so an invalid reply can trigger a re-prompt without losing the + /// pending request (AC4), and so a plain-text reply can be matched + /// against option labels (AC2). + pub async fn peek_oldest_meta(&self, location: &str) -> Option<(usize, bool, Vec)> { + let inner = self.inner.lock().await; + let request_id = inner.by_location.get(location)?.front()?; + inner + .by_request_id + .get(request_id) + .map(|p| (p.num_options, p.multi_select, p.labels.clone())) + } + + /// Resolve the oldest pending question queued for `location`, removing it + /// from both the location queue and the request_id map. + 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(pending) = inner.by_request_id.remove(&request_id) { + return Some(pending.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. 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).map(|p| p.tx) + } +} + +// ── Tests ───────────────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + use crate::http::context::QuestionSpec; + + fn make_forward( + request_id: &str, + ) -> ( + QuestionForward, + oneshot::Receiver>, + ) { + let (tx, rx) = oneshot::channel(); + ( + QuestionForward { + request_id: request_id.to_string(), + question: QuestionSpec { + header: "Test".to_string(), + question: "Pick one?".to_string(), + options: vec![], + multi_select: false, + }, + response_tx: tx, + }, + rx, + ) + } + + #[tokio::test] + async fn registry_starts_empty() { + let registry = QuestionResponderRegistry::new(); + assert!(registry.is_empty()); + } + + #[tokio::test] + async fn register_makes_registry_non_empty_until_guard_dropped() { + let registry = QuestionResponderRegistry::new(); + let (guard, _rx) = registry.register(); + assert!(!registry.is_empty()); + drop(guard); + assert!(registry.is_empty()); + } + + #[tokio::test] + async fn dispatch_delivers_to_registered_responder() { + let registry = QuestionResponderRegistry::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_fails_closed_when_no_responder_registered() { + let registry = QuestionResponderRegistry::new(); + let (fwd, response_rx) = make_forward("req-2"); + registry.dispatch(fwd); + let result = response_rx.await.expect("must receive a result"); + assert!( + result.is_err(), + "no registered responder must fail-closed with an error" + ); + } + + #[tokio::test] + async fn pending_replies_two_concurrent_requests_same_location_both_resolve() { + let pending = PendingQuestionReplies::new(); + let (tx1, rx1) = oneshot::channel(); + let (tx2, rx2) = oneshot::channel(); + pending + .insert( + "room-1", + "req-a", + 3, + false, + vec!["A".to_string(), "B".to_string(), "C".to_string()], + tx1, + ) + .await; + pending + .insert("room-1", "req-b", 2, true, vec![], tx2) + .await; + + let first = pending + .resolve_oldest("room-1") + .await + .expect("first pending reply must still be present"); + let _ = first.send(Ok(QuestionAnswer::Selected(vec![0]))); + assert_eq!( + rx1.await.unwrap().unwrap(), + QuestionAnswer::Selected(vec![0]) + ); + + let second = pending + .resolve_oldest("room-1") + .await + .expect("second pending reply must still be present"); + let _ = second.send(Ok(QuestionAnswer::FreeText("other".to_string()))); + assert_eq!( + rx2.await.unwrap().unwrap(), + QuestionAnswer::FreeText("other".to_string()) + ); + } + + #[tokio::test] + async fn peek_oldest_meta_does_not_consume() { + let pending = PendingQuestionReplies::new(); + let (tx, _rx) = oneshot::channel(); + pending + .insert( + "room-1", + "req-a", + 4, + true, + vec!["X".to_string(), "Y".to_string()], + tx, + ) + .await; + + let meta = pending.peek_oldest_meta("room-1").await; + assert_eq!( + meta, + Some((4, true, vec!["X".to_string(), "Y".to_string()])) + ); + + // Peeking again must return the same entry — it was not consumed. + let meta_again = pending.peek_oldest_meta("room-1").await; + assert_eq!( + meta_again, + Some((4, true, vec!["X".to_string(), "Y".to_string()])) + ); + } + + #[tokio::test] + async fn remove_by_request_id_prevents_later_resolution() { + let pending = PendingQuestionReplies::new(); + let (tx, _rx) = oneshot::channel(); + pending + .insert("room-1", "req-timeout", 2, false, vec![], tx) + .await; + + let removed = pending.remove_by_request_id("room-1", "req-timeout").await; + assert!(removed.is_some()); + + let resolved = pending.resolve_oldest("room-1").await; + assert!(resolved.is_none()); + assert!(pending.peek_oldest_meta("room-1").await.is_none()); + } + + #[tokio::test] + async fn resolve_oldest_returns_none_for_unknown_location() { + let pending = PendingQuestionReplies::new(); + assert!(pending.resolve_oldest("no-such-room").await.is_none()); + } +} diff --git a/server/src/services.rs b/server/src/services.rs index d082ae09..06be7670 100644 --- a/server/src/services.rs +++ b/server/src/services.rs @@ -10,6 +10,7 @@ use crate::chat::dispatcher::ChatDispatcher; use crate::service::permission_router::{ PendingPermReplies, RememberedPermissions, ResponderRegistry, }; +use crate::service::question_router::{PendingQuestionReplies, QuestionResponderRegistry}; use crate::service::status::StatusBroadcaster; use std::collections::HashSet; use std::path::PathBuf; @@ -48,6 +49,18 @@ pub struct Services { /// request to chat; never persisted to disk and never affects a /// different session's agent. pub remembered_permissions: Arc, + /// Registry of tasks currently registered to receive forwarded MCP + /// `ask_question` requests (story 1228). Kept fully separate from + /// `permission_registry` so a question answer is never conflated with a + /// permission decision. + pub question_registry: Arc, + /// Pending question replies awaiting a chat reply, keyed by `request_id` + /// with a per-location FIFO index, mirroring `pending_perm_replies` but + /// for `ask_question` (story 1228). + pub pending_question_replies: Arc, + /// Seconds to wait for a user to answer a question before giving up + /// (fail-closed): the MCP tool returns an error to the asking agent. + pub question_timeout_secs: u64, /// Project-scoped status broadcaster. /// /// Consumers (chat transports, Web UI, agent context) call @@ -81,6 +94,9 @@ impl Services { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: RememberedPermissions::new(), + question_registry: QuestionResponderRegistry::new(), + pending_question_replies: PendingQuestionReplies::new(), + question_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 8746a392..c32d7526 100644 --- a/server/src/sled_uplink.rs +++ b/server/src/sled_uplink.rs @@ -484,6 +484,10 @@ mod tests { pending_perm_replies: PendingPermReplies::new(), permission_timeout_secs: 120, remembered_permissions: crate::service::permission_router::RememberedPermissions::new(), + question_registry: crate::service::question_router::QuestionResponderRegistry::new(), + pending_question_replies: crate::service::question_router::PendingQuestionReplies::new( + ), + question_timeout_secs: 120, chat_dispatcher: Arc::new(crate::chat::dispatcher::ChatDispatcher::new(1_500)), }) }