diff --git a/migrations/20260810000002_pending_questions.sql b/migrations/20260810000002_pending_questions.sql new file mode 100644 index 000000000..126a1dbf5 --- /dev/null +++ b/migrations/20260810000002_pending_questions.sql @@ -0,0 +1,25 @@ +-- Pending questions store for the ask tool. +-- +-- When an agent calls the ask tool, the question + options are persisted here +-- so inbound interaction clicks can be correlated back to the original question. + +CREATE TABLE IF NOT EXISTS pending_questions ( + question_id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + channel_id TEXT NOT NULL, + question TEXT NOT NULL, + options TEXT NOT NULL, -- JSON array of AskOption + multi_select INTEGER NOT NULL DEFAULT 0, + message_ref TEXT, -- platform message id, for disabling buttons after answer + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + resolved_at TIMESTAMP, + answer TEXT -- JSON array of picked labels +); + +-- Fast lookup by channel for pruning +CREATE INDEX IF NOT EXISTS idx_pending_questions_channel + ON pending_questions(channel_id, created_at DESC); + +-- Fast lookup for resolution via inbound interaction click +CREATE INDEX IF NOT EXISTS idx_pending_questions_resolved + ON pending_questions(resolved_at); diff --git a/prompts/en/tools/ask_description.md.j2 b/prompts/en/tools/ask_description.md.j2 new file mode 100644 index 000000000..70cc511d2 --- /dev/null +++ b/prompts/en/tools/ask_description.md.j2 @@ -0,0 +1 @@ +Ask the user a question with selectable answer options. Use this when you need the user to pick between choices (which environment, which approach, proceed or not). Renders as buttons or a select menu on platforms that support them, and as a numbered list on text-only channels. The answer arrives as a future message — do not speculate about the answer; end your turn after asking. The user can also type a free-form response instead of clicking an option. diff --git a/src/agent/channel.rs b/src/agent/channel.rs index 2dd64eee0..b1fa09a92 100644 --- a/src/agent/channel.rs +++ b/src/agent/channel.rs @@ -133,6 +133,83 @@ fn is_control_command(message: &InboundMessage) -> bool { } } +/// Enrich an ask-tool interaction with the original question context. +/// +/// When an inbound `Interaction` has an `action_id` matching the `ask:` prefix, +/// this looks up the pending question, resolves it, and returns a human-readable +/// enrichment like `Alice answered "Which environment?": staging`. +/// +/// Non-ask interactions pass through as-is. Expired or already-resolved questions +/// get an `(expired)` marker. +async fn enrich_ask_interaction( + pool: &sqlx::SqlitePool, + sender_name: &str, + action_id: &str, + values: &[String], +) -> String { + let (question_id, option_idx) = match crate::tools::ask::parse_ask_custom_id(action_id) { + Some(parsed) => parsed, + None => { + // Not an ask interaction — use standard display + if !values.is_empty() { + return format!("[interaction: {action_id} → {}]", values.join(", ")); + } + return format!("[interaction: {action_id}]"); + } + }; + + let store = crate::questions::QuestionStore::new(pool.clone()); + + match store.get(question_id).await { + Ok(Some(q)) if q.resolved_at.is_none() => { + let answer_labels: Vec = match option_idx { + Some(idx) => { + // Button click: use the option at this index + q.options + .get(idx) + .map(|opt| vec![opt.label.clone()]) + .unwrap_or_default() + } + None => { + // Select menu: parse values to get indices + values + .iter() + .filter_map(|value| { + let (_, idx) = crate::tools::ask::parse_ask_custom_id(value)?; + idx.and_then(|i| q.options.get(i).map(|opt| opt.label.clone())) + }) + .collect() + } + }; + + if answer_labels.is_empty() { + return format!("[interaction: {action_id}] (expired — no matching options)"); + } + + // Resolve the question so duplicate clicks get the expired path. + // Failure is non-fatal — the answer still rendered correctly. + if let Err(e) = store.resolve(question_id, &answer_labels).await { + tracing::warn!( + question_id, + error = %e, + "failed to resolve pending question; duplicate clicks may re-fire" + ); + } + + let labels_str = answer_labels.join(", "); + format!("{sender_name} answered \"{}\": {labels_str}", q.question) + } + Ok(Some(_)) => { + // Already resolved + format!("[interaction: {action_id}] (expired)") + } + _ => { + // Question not found or store error + format!("[interaction: {action_id}] (expired)") + } + } +} + fn should_flush_coalesce_buffer_for_event(event: &ProcessEvent) -> bool { matches!( event, @@ -1957,8 +2034,19 @@ impl Channel { } // Render interactions and commands as their Display form // so the LLM sees plain text. - crate::MessageContent::Interaction { .. } - | crate::MessageContent::Command { .. } => { + crate::MessageContent::Interaction { + action_id, values, .. + } => { + let text = enrich_ask_interaction( + &self.deps.sqlite_pool, + &sender_name, + action_id, + values, + ) + .await; + (text, Vec::new()) + } + crate::MessageContent::Command { .. } => { (message.content.to_string(), Vec::new()) } }; @@ -2353,9 +2441,16 @@ impl Channel { // Render interactions and commands as their Display form so the // LLM sees plain text; a Command renders as "/name args" and is // dispatched by the same parse below. - crate::MessageContent::Interaction { .. } | crate::MessageContent::Command { .. } => { - (message.content.to_string(), Vec::new()) + crate::MessageContent::Interaction { + action_id, values, .. + } => { + let sender = participant_display_name(&message); + let raw_text = + enrich_ask_interaction(&self.deps.sqlite_pool, &sender, action_id, values) + .await; + (raw_text, Vec::new()) } + crate::MessageContent::Command { .. } => (message.content.to_string(), Vec::new()), }; // Save attachments to disk when enabled, capturing bytes for LLM reuse diff --git a/src/lib.rs b/src/lib.rs index bfd083eae..5c22de647 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -26,6 +26,7 @@ pub mod openai_auth; pub mod opencode; pub mod projects; pub mod prompts; +pub mod questions; pub mod sandbox; pub mod schedule; pub mod secrets; diff --git a/src/prompts/text.rs b/src/prompts/text.rs index 37c52d3c9..54da17748 100644 --- a/src/prompts/text.rs +++ b/src/prompts/text.rs @@ -321,6 +321,7 @@ fn lookup(lang: &str, key: &str) -> &'static str { ("en", "tools/project_manage") => { include_str!("../../prompts/en/tools/project_manage_description.md.j2") } + ("en", "tools/ask") => include_str!("../../prompts/en/tools/ask_description.md.j2"), ("en", "tools/attachment_recall") => { include_str!("../../prompts/en/tools/attachment_recall_description.md.j2") } diff --git a/src/questions.rs b/src/questions.rs new file mode 100644 index 000000000..a1c761b6f --- /dev/null +++ b/src/questions.rs @@ -0,0 +1,225 @@ +//! Pending question store for the ask tool. +//! +//! Persists questions that the agent has asked the user so inbound interaction +//! clicks can be correlated back to the original question context. Restart-safe +//! by construction — questions survive process restarts. + +use crate::error::Result; +use anyhow::Context as _; +use serde::{Deserialize, Serialize}; +use sqlx::{Row as _, SqlitePool}; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +/// Default TTL for pending questions (7 days). +pub const DEFAULT_QUESTION_TTL_DAYS: i64 = 7; + +/// A single option for the ask tool. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AskOption { + pub label: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub description: Option, +} + +/// A persisted pending question row. +#[derive(Debug, Clone)] +pub struct PendingQuestion { + pub question_id: String, + pub agent_id: String, + pub channel_id: String, + pub question: String, + pub options: Vec, + pub multi_select: bool, + pub message_ref: Option, + pub created_at: String, + pub resolved_at: Option, + pub answer: Option>, +} + +/// Input for creating a pending question. +#[derive(Debug, Clone)] +pub struct NewQuestion { + pub question_id: String, + pub agent_id: String, + pub channel_id: String, + pub question: String, + pub options: Vec, + pub multi_select: bool, + pub message_ref: Option, +} + +// --------------------------------------------------------------------------- +// Store +// --------------------------------------------------------------------------- + +#[derive(Debug, Clone)] +pub struct QuestionStore { + pool: SqlitePool, +} + +/// Timestamp matching SQLite's CURRENT_TIMESTAMP format so comparisons +/// against `datetime('now', …)` are consistent. +fn now_sqlite() -> String { + chrono::Utc::now().format("%Y-%m-%d %H:%M:%S").to_string() +} + +impl QuestionStore { + pub fn new(pool: SqlitePool) -> Self { + Self { pool } + } + + /// Insert a new pending question. + pub async fn insert(&self, question: &NewQuestion) -> Result<()> { + let options_json = serde_json::to_string(&question.options) + .context("failed to serialize question options")?; + + sqlx::query( + r#" + INSERT INTO pending_questions + (question_id, agent_id, channel_id, question, options, multi_select, message_ref) + VALUES (?, ?, ?, ?, ?, ?, ?) + "#, + ) + .bind(&question.question_id) + .bind(&question.agent_id) + .bind(&question.channel_id) + .bind(&question.question) + .bind(&options_json) + .bind(question.multi_select as i64) + .bind(&question.message_ref) + .execute(&self.pool) + .await + .context("failed to insert pending question")?; + + // Prune expired questions in the background so the table does not + // grow unbounded. Failure is non-fatal — the next write retries. + let prune_pool = self.pool.clone(); + tokio::spawn(async move { + if let Err(error) = QuestionStore::new(prune_pool) + .prune_expired(DEFAULT_QUESTION_TTL_DAYS) + .await + { + tracing::warn!(%error, "background prune of pending_questions failed"); + } + }); + + Ok(()) + } + + /// Look up a pending question by ID. + pub async fn get(&self, question_id: &str) -> Result> { + let row = sqlx::query( + r#" + SELECT question_id, agent_id, channel_id, question, options, multi_select, + message_ref, created_at, resolved_at, answer + FROM pending_questions + WHERE question_id = ? + "#, + ) + .bind(question_id) + .fetch_optional(&self.pool) + .await + .context("failed to fetch pending question")?; + + match row { + Some(row) => Ok(Some(question_from_row(row)?)), + None => Ok(None), + } + } + + /// Resolve a pending question with the given answer labels. + /// Returns Ok(true) if the question was found and unresolved, Ok(false) if + /// already resolved or not found. + pub async fn resolve(&self, question_id: &str, answer: &[String]) -> Result { + let answer_json = serde_json::to_string(answer).context("failed to serialize answer")?; + let now = now_sqlite(); + + let affected = sqlx::query( + r#" + UPDATE pending_questions + SET resolved_at = ?, answer = ? + WHERE question_id = ? AND resolved_at IS NULL + "#, + ) + .bind(&now) + .bind(&answer_json) + .bind(question_id) + .execute(&self.pool) + .await + .context("failed to resolve pending question")? + .rows_affected(); + + Ok(affected > 0) + } + + /// Prune resolved questions older than the TTL, and unanswered questions + /// older than the TTL (expired). Returns the count of removed rows. + pub async fn prune_expired(&self, ttl_days: i64) -> Result { + let cutoff = format!("-{} days", ttl_days); + + let affected = sqlx::query( + r#" + DELETE FROM pending_questions + WHERE (resolved_at IS NOT NULL AND resolved_at < datetime('now', ?)) + OR (resolved_at IS NULL AND created_at < datetime('now', ?)) + "#, + ) + .bind(&cutoff) + .bind(&cutoff) + .execute(&self.pool) + .await + .context("failed to prune expired questions")? + .rows_affected(); + + Ok(affected) + } +} + +// --------------------------------------------------------------------------- +// Row mapping +// --------------------------------------------------------------------------- + +fn question_from_row(row: sqlx::sqlite::SqliteRow) -> Result { + let options_json: String = row + .try_get("options") + .context("failed to read question options")?; + let options: Vec = + serde_json::from_str(&options_json).context("failed to parse question options")?; + + let answer_json: Option = row + .try_get::, _>("answer") + .context("failed to read answer")?; + let answer = match answer_json { + Some(json) => Some(serde_json::from_str(&json).context("failed to parse question answer")?), + None => None, + }; + + Ok(PendingQuestion { + question_id: row + .try_get("question_id") + .context("failed to read question_id")?, + agent_id: row.try_get("agent_id").context("failed to read agent_id")?, + channel_id: row + .try_get("channel_id") + .context("failed to read channel_id")?, + question: row.try_get("question").context("failed to read question")?, + options, + multi_select: row + .try_get::("multi_select") + .context("failed to read multi_select")? + != 0, + message_ref: row + .try_get::, _>("message_ref") + .context("failed to read message_ref")?, + created_at: row + .try_get("created_at") + .context("failed to read created_at")?, + resolved_at: row + .try_get::, _>("resolved_at") + .context("failed to read resolved_at")?, + answer, + }) +} diff --git a/src/tools.rs b/src/tools.rs index 8c93cfbfa..81a4fedb8 100644 --- a/src/tools.rs +++ b/src/tools.rs @@ -31,6 +31,7 @@ //! - branch + worker tool superset plus `spacebot_docs`, `config_inspect`, `spawn_worker`, //! and `restart` +pub mod ask; pub mod attachment_recall; pub mod autonomy_complete; pub mod branch_tool; @@ -90,6 +91,7 @@ pub mod factory_search_context; pub mod factory_update_config; pub mod factory_update_identity; +pub use ask::{AskArgs, AskError, AskOptionArg, AskOutput, AskTool}; pub use attachment_recall::{ AttachmentRecallArgs, AttachmentRecallError, AttachmentRecallOutput, AttachmentRecallTool, }; @@ -497,12 +499,24 @@ pub async fn add_channel_tools( .unwrap_or_else(|| state.deps.agent_id.to_string()); handle .add_tool(ReplyTool::new( - reply_target, + reply_target.clone(), conversation_id.clone(), state.conversation_logger.clone(), state.channel_id.clone(), replied_flag.clone(), + agent_display_name.clone(), + state.deps.api_state.clone(), + )) + .await?; + handle + .add_tool(AskTool::new( + crate::questions::QuestionStore::new(state.deps.sqlite_pool.clone()), + response_tx.clone(), + state.conversation_logger.clone(), + state.channel_id.clone(), + state.deps.agent_id.to_string(), agent_display_name, + replied_flag.clone(), state.deps.api_state.clone(), )) .await?; @@ -852,6 +866,7 @@ pub async fn remove_channel_tools( ) -> Result<(), rig::tool::server::ToolServerError> { if allow_direct_reply { handle.remove_tool(ReplyTool::NAME).await?; + handle.remove_tool(AskTool::NAME).await?; } handle.remove_tool(BranchTool::NAME).await?; handle.remove_tool(SpawnWorkerTool::NAME).await?; diff --git a/src/tools/ask.rs b/src/tools/ask.rs new file mode 100644 index 000000000..6aabd8ba1 --- /dev/null +++ b/src/tools/ask.rs @@ -0,0 +1,352 @@ +//! Ask tool: presents the user with a question and selectable answer options. +//! +//! Renders buttons or a select menu on platforms that support them, and as a +//! numbered list on text-only channels. Answers arrive as enriched interaction +//! messages that include the original question text for context. + +use crate::api::ApiState; +use crate::conversation::ConversationLogger; +use crate::questions::{NewQuestion, QuestionStore}; +use crate::{ChannelId, OutboundResponse, RoutedSender}; +use rig::completion::ToolDefinition; +use rig::tool::Tool; +use schemars::JsonSchema; +use serde::{Deserialize, Serialize}; +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use crate::tools::reply::RepliedFlag; + +/// Generate a short random question ID for the custom_id prefix. +fn new_question_id() -> String { + uuid::Uuid::new_v4().to_string()[..8].to_string() +} + +/// Tool that asks the user a question with selectable options. +#[derive(Clone)] +pub struct AskTool { + question_store: QuestionStore, + sender: RoutedSender, + conversation_logger: ConversationLogger, + channel_id: ChannelId, + agent_id: String, + agent_display_name: String, + replied_flag: RepliedFlag, + api_state: Option>, +} + +impl std::fmt::Debug for AskTool { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AskTool") + .field("channel_id", &self.channel_id) + .field("agent_id", &self.agent_id) + .field("agent_display_name", &self.agent_display_name) + .finish() + } +} + +impl AskTool { + /// Create a new ask tool bound to a conversation. + #[allow(clippy::too_many_arguments)] + pub fn new( + question_store: QuestionStore, + sender: RoutedSender, + conversation_logger: ConversationLogger, + channel_id: ChannelId, + agent_id: String, + agent_display_name: String, + replied_flag: RepliedFlag, + api_state: Option>, + ) -> Self { + Self { + question_store, + sender, + conversation_logger, + channel_id, + agent_id, + agent_display_name, + replied_flag, + api_state, + } + } +} + +/// Error type for ask tool. +#[derive(Debug, thiserror::Error)] +#[error("Ask failed: {0}")] +pub struct AskError(String); + +/// Arguments for ask tool. +#[derive(Debug, Deserialize, JsonSchema)] +pub struct AskArgs { + /// The question to ask. + pub question: String, + /// Selectable answers. 2 to 10 options. + pub options: Vec, + /// Allow picking more than one option. Renders as a select menu + /// with multi-select where supported. + #[serde(default)] + pub multi_select: bool, +} + +#[derive(Debug, Clone, Deserialize, JsonSchema)] +pub struct AskOptionArg { + /// Short label shown on the button (keep under ~40 chars). + pub label: String, + /// Optional longer description, shown where the platform supports it + /// (select menu descriptions, portal UI) and in the text fallback. + #[serde(default)] + pub description: Option, +} + +/// Output from ask tool. +#[derive(Debug, Serialize)] +pub struct AskOutput { + pub question_id: String, + pub question: String, + pub options_count: usize, + pub message: String, +} + +fn build_text_fallback(question: &str, options: &[AskOptionArg]) -> String { + let mut text = format!("{}\n", question.trim_end()); + for (i, opt) in options.iter().enumerate() { + match &opt.description { + Some(desc) if !desc.trim().is_empty() => { + text.push_str(&format!( + "{}. {} — {}\n", + i + 1, + opt.label.trim(), + desc.trim() + )); + } + _ => { + text.push_str(&format!("{}. {}\n", i + 1, opt.label.trim())); + } + } + } + text.trim_end().to_string() +} + +fn build_interactive_elements( + question_id: &str, + options: &[AskOptionArg], + multi_select: bool, +) -> Vec { + if multi_select || options.len() > 5 { + // Select menu + let select_options: Vec = options + .iter() + .enumerate() + .map(|(idx, opt)| crate::SelectOption { + label: opt.label.clone(), + value: format!("ask:{}:{}", question_id, idx), + description: opt.description.clone(), + emoji: None, + }) + .collect(); + + let placeholder = if multi_select { + "Select one or more options…".to_string() + } else { + "Select an option…".to_string() + }; + + vec![crate::InteractiveElements::Select { + select: crate::SelectMenu { + custom_id: format!("ask:{}:menu", question_id), + options: select_options, + placeholder: Some(placeholder), + }, + }] + } else { + // Buttons + let buttons: Vec = options + .iter() + .enumerate() + .map(|(idx, opt)| crate::Button { + label: opt.label.clone(), + custom_id: Some(format!("ask:{question_id}:{idx}")), + style: crate::ButtonStyle::Primary, + url: None, + }) + .collect(); + vec![crate::InteractiveElements::Buttons { buttons }] + } +} + +pub(crate) const ASK_CUSTOM_ID_PREFIX: &str = "ask:"; + +/// Parse an ask custom_id into (question_id, option_index). +/// custom_id format: `ask:{question_id}:{idx}` or `ask:{question_id}:menu` for selects. +pub fn parse_ask_custom_id(custom_id: &str) -> Option<(&str, Option)> { + let stripped = custom_id.strip_prefix(ASK_CUSTOM_ID_PREFIX)?; + let (question_id, idx_part) = stripped.rsplit_once(':')?; + if idx_part == "menu" { + Some((question_id, None)) + } else { + let idx: usize = idx_part.parse().ok()?; + Some((question_id, Some(idx))) + } +} + +impl Tool for AskTool { + const NAME: &'static str = "ask"; + + type Error = AskError; + type Args = AskArgs; + type Output = AskOutput; + + async fn definition(&self, _prompt: String) -> ToolDefinition { + let parameters = serde_json::json!({ + "type": "object", + "properties": { + "question": { + "type": "string", + "description": "The question to ask the user." + }, + "options": { + "type": "array", + "description": "Selectable answer options. Minimum 2, maximum 10.", + "items": { + "type": "object", + "properties": { + "label": { + "type": "string", + "description": "Short label shown on the button or select option." + }, + "description": { + "type": "string", + "description": "Optional longer description for the option." + } + }, + "required": ["label"] + } + }, + "multi_select": { + "type": "boolean", + "description": "Allow the user to pick more than one option. Renders as a multi-select menu where supported. Defaults to false." + } + }, + "required": ["question", "options"] + }); + + ToolDefinition { + name: Self::NAME.to_string(), + description: crate::prompts::text::get("tools/ask").to_string(), + parameters, + } + } + + async fn call(&self, args: Self::Args) -> Result { + let question = args.question.trim().to_string(); + if question.is_empty() { + return Err(AskError("question must not be empty".into())); + } + + // Trim and validate options + let options: Vec = args + .options + .into_iter() + .map(|mut opt| { + opt.label = opt.label.trim().to_string(); + opt.description = opt.description.map(|d| d.trim().to_string()); + opt + }) + .filter(|opt| !opt.label.is_empty()) + .collect(); + + if options.len() < 2 { + return Err(AskError("at least 2 non-empty options are required".into())); + } + if options.len() > 10 { + return Err(AskError("at most 10 options are allowed".into())); + } + + let question_id = new_question_id(); + + // Build text fallback (numbered list) — this IS the message on + // text-only channels, and provides context on button channels. + let text = build_text_fallback(&question, &options); + + // Build interactive elements + let interactive_elements = + build_interactive_elements(&question_id, &options, args.multi_select); + + tracing::info!( + question_id = %question_id, + channel_id = %self.channel_id, + question_len = question.len(), + options_count = options.len(), + multi_select = args.multi_select, + "ask tool sent question" + ); + + // Persist the pending question + let store_options: Vec = options + .iter() + .map(|opt| crate::questions::AskOption { + label: opt.label.clone(), + description: opt.description.clone(), + }) + .collect(); + + let new_q = NewQuestion { + question_id: question_id.clone(), + agent_id: self.agent_id.clone(), + channel_id: self.channel_id.to_string(), + question: question.clone(), + options: store_options, + multi_select: args.multi_select, + message_ref: None, + }; + + if let Err(e) = self.question_store.insert(&new_q).await { + tracing::error!(error = %e, "failed to persist pending question"); + return Err(AskError(format!("failed to store question: {e}"))); + } + + // Send via RichMessage + let response = OutboundResponse::RichMessage { + text, + blocks: vec![], + cards: vec![], + interactive_elements, + poll: None, + }; + + self.sender + .send(response) + .await + .map_err(|e| AskError(format!("failed to send question: {e}")))?; + + // Drain accumulated channel tool calls and pack into message metadata + let tool_calls_json = if let Some(ref api_state) = self.api_state { + let calls = api_state.take_channel_tool_calls(&self.channel_id).await; + if calls.is_empty() { + None + } else { + serde_json::to_string(&calls).ok() + } + } else { + None + }; + + self.conversation_logger.log_bot_message_with_metadata( + &self.channel_id, + &format!("[ask] {question}"), + Some(&self.agent_display_name), + tool_calls_json, + ); + + // Mark turn as handled so the channel doesn't send fallback text + self.replied_flag.store(true, Ordering::Relaxed); + + Ok(AskOutput { + question_id, + question, + options_count: options.len(), + message: "Question sent. The answer will arrive as a future message. End your turn now — do not speculate about the answer.".to_string(), + }) + } +}