feat(mcp): elicitation during tool execution (phase 2) (#2545)

* feat(mcp): elicitation during tool execution — phase 2 (#2522)

- Fix deadlock: drain elicitation_rx concurrently with tool tier futures
  via tokio::select! loop in execute_native_tool_calls
- TUI: ElicitationDialogState widget with field navigation and render
- TUI: TuiChannel::elicit() via AgentEvent::ElicitationRequest oneshot
- Telegram: TelegramChannel::elicit() with sequential prompts, 120s timeout
- Security: sanitize all MCP-provided strings (ANSI, control chars)
- ELICITATION_TIMEOUT constant in zeph-channels

* test(telegram): add elicit() unit tests; sanitize JSON field keys

- Add 4 unit tests for TelegramChannel::elicit():
  - happy path string field → Accepted with correct key/value
  - /cancel command → Cancelled
  - timeout logic (rx-level isolation, matching confirm pattern)
  - sanitize_field_key strips dashes, spaces, special chars

- Fix sanitize_field_key test: function only strips non-alphanumeric/
  underscore chars, not ANSI sequences — corrected test assertion

- Update coverage-status.md: add row for MCP elicitation phase 2
  (status: Untested) in the zeph-mcp section

Author: bug-ops

CHANGELOG.md

@@ -12,6 +12,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- feat(mcp): elicitation during tool execution — phase 2 (#2522)
+  - **Core deadlock fix**: agent loop now drains MCP elicitation events concurrently with tool tier futures via `tokio::select!` in `execute_native_tool_calls`; without this, MCP servers that send an elicitation mid-tool-call would deadlock (tool awaits elicitation response, agent loop awaits tool result)
+  - **TUI interactive dialog**: modal overlay (`ElicitationDialogState` + `widgets::elicitation`) renders elicitation fields (string/integer/number/boolean/enum) with keyboard navigation: `Tab`/`Shift+Tab` to move between fields, `Space` to toggle booleans, `Up`/`Down` for enum selection, `Enter` to submit, `Esc` to cancel; overrides vi-mode while active
+  - **Telegram sequential prompts**: `TelegramChannel::elicit()` sends one prompt per field with 120 s per-field timeout; boolean uses yes/no reply; enum accepts 1-based index or exact match; `/cancel` command dismisses at any point; `ELICITATION_TIMEOUT` constant added to `zeph-channels`
+  - **Security**: all MCP-supplied strings (server name, message, field names, enum values) are sanitized to strip ANSI escape sequences and control characters before rendering in TUI or Telegram; Telegram field prompts use 1-based numeric indexes to avoid the 64-byte callback_data limit
+
+
+
 - feat(tools): structured shell output envelope (#2488) — `execute_bash` now captures stdout and stderr as separate streams at the process level using a tagged `(bool, String)` channel; `ShellOutputEnvelope { stdout, stderr, exit_code, truncated }` is built post-execution and serialized into `ToolOutput.raw_response` for ACP/audit consumers; LLM context continues using the interleaved combined output in `summary`; `AuditEntry` gains optional `exit_code: Option<i32>` and `truncated: bool` fields (`skip_serializing_if` for backward compat)
 - feat(tools): per-path read allow/deny sandbox for file tool (#2489) — new `[tools.file]` config section with `deny_read` and `allow_read` glob pattern lists; evaluation order: deny-then-allow; all patterns matched against canonicalized absolute paths (prevents symlink bypass); `FileExecutor::with_read_sandbox()` builder method applies the sandbox; `handle_read()` checks sandbox before `read_to_string`; `grep_recursive` skips denied files before reading content; `FileConfig` exported from `zeph-tools`
 

crates/zeph-channels/src/lib.rs

@@ -21,3 +21,5 @@ pub use cli::CliChannel;
 /// Used by Telegram, Discord, and Slack `confirm()` implementations to ensure
 /// consistent deny-on-timeout behavior.
 pub const CONFIRM_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
+/// Per-field timeout for interactive elicitation dialogs on remote channels (Telegram, etc.).
+pub const ELICITATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(120);

crates/zeph-channels/src/telegram.rs

@@ -7,7 +7,10 @@ use crate::markdown::markdown_to_telegram;
 use teloxide::prelude::*;
 use teloxide::types::{BotCommand, ChatAction, MessageId, ParseMode};
 use tokio::sync::mpsc;
-use zeph_core::channel::{Attachment, AttachmentKind, Channel, ChannelError, ChannelMessage};
+use zeph_core::channel::{
+    Attachment, AttachmentKind, Channel, ChannelError, ChannelMessage, ElicitationField,
+    ElicitationFieldType, ElicitationRequest, ElicitationResponse,
+};
 
 const MAX_MESSAGE_LEN: usize = 4096;
 const MAX_IMAGE_BYTES: u32 = 20 * 1024 * 1024;
@@ -501,6 +504,144 @@ impl Channel for TelegramChannel {
             }
         }
     }
+
+    async fn elicit(
+        &mut self,
+        request: ElicitationRequest,
+    ) -> Result<ElicitationResponse, ChannelError> {
+        let timeout = crate::ELICITATION_TIMEOUT;
+
+        self.send(&format!(
+            "*[MCP server '{}' is requesting input]*\n{}\n\n_Reply /cancel to cancel. \
+             Timeout: {}s._",
+            sanitize_markdown(&request.server_name),
+            sanitize_markdown(&request.message),
+            timeout.as_secs(),
+        ))
+        .await?;
+
+        let mut values = serde_json::Map::new();
+        for field in &request.fields {
+            let prompt = build_telegram_field_prompt(field);
+            self.send(&prompt).await?;
+
+            let incoming = match tokio::time::timeout(timeout, self.rx.recv()).await {
+                Ok(Some(msg)) => msg,
+                Ok(None) => {
+                    tracing::warn!(server = request.server_name, "elicitation channel closed");
+                    return Ok(ElicitationResponse::Declined);
+                }
+                Err(_) => {
+                    tracing::warn!(server = request.server_name, "elicitation timed out");
+                    let _ = self
+                        .send("Elicitation timed out — request cancelled.")
+                        .await;
+                    return Ok(ElicitationResponse::Cancelled);
+                }
+            };
+
+            let text = incoming.text.trim().to_owned();
+
+            if text.eq_ignore_ascii_case("/cancel") {
+                let _ = self.send("Elicitation cancelled.").await;
+                return Ok(ElicitationResponse::Cancelled);
+            }
+
+            let Some(value) = coerce_telegram_field(&text, &field.field_type) else {
+                let _ = self
+                    .send(&format!("Invalid value for '{}'. Declining.", field.name))
+                    .await;
+                return Ok(ElicitationResponse::Declined);
+            };
+            values.insert(sanitize_field_key(&field.name), value);
+        }
+
+        Ok(ElicitationResponse::Accepted(serde_json::Value::Object(
+            values,
+        )))
+    }
+}
+
+/// Strip Markdown special characters to prevent injection in Telegram messages.
+fn sanitize_markdown(s: &str) -> String {
+    s.chars()
+        .filter(|c| !matches!(c, '*' | '_' | '[' | ']' | '`' | '\x1b'))
+        .collect()
+}
+
+/// Sanitize a field name for use as a JSON key.
+///
+/// Keeps only alphanumeric characters and underscores to prevent injection via
+/// malicious MCP server field names (e.g. keys with special chars that could
+/// confuse downstream consumers).
+fn sanitize_field_key(s: &str) -> String {
+    s.chars()
+        .filter(|c| c.is_alphanumeric() || *c == '_')
+        .collect()
+}
+
+fn build_telegram_field_prompt(field: &ElicitationField) -> String {
+    let req = if field.required { " (required)" } else { "" };
+    match &field.field_type {
+        ElicitationFieldType::Boolean => {
+            format!("*{}*{}: Reply *yes* or *no*", field.name, req)
+        }
+        ElicitationFieldType::Enum(opts) => {
+            // Use short numeric indexes to avoid Telegram 64-byte callback_data limit
+            let list: String = opts
+                .iter()
+                .enumerate()
+                .map(|(i, o)| format!("{}: {}", i + 1, sanitize_markdown(o)))
+                .collect::<Vec<_>>()
+                .join("\n");
+            format!("*{}*{}: Reply with the number:\n{}", field.name, req, list)
+        }
+        ElicitationFieldType::Integer => {
+            format!("*{}*{}: Reply with an integer", field.name, req)
+        }
+        ElicitationFieldType::Number => {
+            format!("*{}*{}: Reply with a number", field.name, req)
+        }
+        ElicitationFieldType::String => {
+            format!("*{}*{}: Reply with text", field.name, req)
+        }
+    }
+}
+
+fn coerce_telegram_field(text: &str, kind: &ElicitationFieldType) -> Option<serde_json::Value> {
+    match kind {
+        ElicitationFieldType::String => Some(serde_json::Value::String(text.to_owned())),
+        ElicitationFieldType::Boolean => {
+            if text.eq_ignore_ascii_case("yes") || text == "1" {
+                Some(serde_json::Value::Bool(true))
+            } else if text.eq_ignore_ascii_case("no") || text == "0" {
+                Some(serde_json::Value::Bool(false))
+            } else {
+                None
+            }
+        }
+        ElicitationFieldType::Integer => text
+            .parse::<i64>()
+            .ok()
+            .map(|n| serde_json::Value::Number(n.into())),
+        ElicitationFieldType::Number => text
+            .parse::<f64>()
+            .ok()
+            .and_then(|n| serde_json::Number::from_f64(n).map(serde_json::Value::Number)),
+        ElicitationFieldType::Enum(opts) => {
+            // Accept numeric index (1-based) or exact match
+            if let Ok(idx) = text.parse::<usize>()
+                && idx >= 1
+                && idx <= opts.len()
+            {
+                return Some(serde_json::Value::String(opts[idx - 1].clone()));
+            }
+            // Exact match (case-insensitive)
+            opts.iter()
+                .find(|o| o.eq_ignore_ascii_case(text))
+                .map(|o| serde_json::Value::String(o.clone()))
+        }
+    }
 }
 
 #[cfg(test)]
@@ -846,4 +987,93 @@ mod tests {
             requests.len()
         );
     }
+
+    // ---------------------------------------------------------------------------
+    // elicit() — happy path, timeout, /cancel, field-key sanitization
+    // All tests that exercise elicit() need the mock server because elicit()
+    // calls self.send() (which calls the Telegram Bot API) before reading rx.
+    // ---------------------------------------------------------------------------
+
+    #[tokio::test]
+    async fn elicit_happy_path_string_field_returns_accepted() {
+        let server = MockServer::start().await;
+        let (mut channel, tx) = make_mocked_channel(&server, vec![]).await;
+
+        let request = ElicitationRequest {
+            server_name: "test-server".to_owned(),
+            message: "Please provide your name".to_owned(),
+            fields: vec![ElicitationField {
+                name: "username".to_owned(),
+                description: None,
+                field_type: ElicitationFieldType::String,
+                required: true,
+            }],
+        };
+
+        // Send the answer before calling elicit() so it is buffered in the channel.
+        tx.send(plain_message("alice")).await.unwrap();
+
+        let response = channel.elicit(request).await.unwrap();
+
+        match response {
+            ElicitationResponse::Accepted(val) => {
+                assert_eq!(val["username"], "alice");
+            }
+            other => panic!("expected Accepted, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn elicit_cancel_command_returns_cancelled() {
+        let server = MockServer::start().await;
+        let (mut channel, tx) = make_mocked_channel(&server, vec![]).await;
+
+        let request = ElicitationRequest {
+            server_name: "test-server".to_owned(),
+            message: "Provide a value".to_owned(),
+            fields: vec![ElicitationField {
+                name: "token".to_owned(),
+                description: None,
+                field_type: ElicitationFieldType::String,
+                required: true,
+            }],
+        };
+
+        tx.send(plain_message("/cancel")).await.unwrap();
+
+        let response = channel.elicit(request).await.unwrap();
+        assert!(
+            matches!(response, ElicitationResponse::Cancelled),
+            "expected Cancelled, got {response:?}"
+        );
+    }
+
+    /// Verify the timeout branch of elicit() at the rx level, matching the
+    /// same pattern used in confirm_timeout_logic_denies_on_timeout.
+    #[tokio::test]
+    async fn elicit_timeout_logic_cancels_on_timeout() {
+        tokio::time::pause();
+        let (_tx, mut rx) = mpsc::channel::<IncomingMessage>(1);
+        let timeout_fut = tokio::time::timeout(crate::ELICITATION_TIMEOUT, rx.recv());
+        tokio::time::advance(crate::ELICITATION_TIMEOUT + Duration::from_millis(1)).await;
+        let result = timeout_fut.await;
+        assert!(
+            result.is_err(),
+            "expected Err(Elapsed) for elicitation timeout, got recv result"
+        );
+    }
+
+    // ---------------------------------------------------------------------------
+    // sanitize_field_key — pure unit test (no network)
+    // ---------------------------------------------------------------------------
+
+    #[test]
+    fn sanitize_field_key_strips_special_chars() {
+        assert_eq!(sanitize_field_key("hello world"), "helloworld");
+        assert_eq!(sanitize_field_key("field-name"), "fieldname");
+        assert_eq!(sanitize_field_key("__ok__"), "__ok__");
+        assert_eq!(sanitize_field_key("a.b.c"), "abc");
+        // Alphanumeric chars and underscores are kept; everything else stripped.
+        assert_eq!(sanitize_field_key("key!@#val"), "keyval");
+    }
 }

crates/zeph-core/src/agent/mcp.rs

@@ -511,7 +511,7 @@ impl<C: Channel> Agent<C> {
     }
 
     /// Handle a single elicitation event by routing it to the active channel.
-    async fn handle_elicitation_event(&mut self, event: zeph_mcp::ElicitationEvent) {
+    pub(super) async fn handle_elicitation_event(&mut self, event: zeph_mcp::ElicitationEvent) {
         use crate::channel::{ElicitationRequest, ElicitationResponse};
 
         let decline = CreateElicitationResult {

crates/zeph-core/src/agent/tool_execution/native.rs

@@ -1246,18 +1246,40 @@ impl<C: Channel> Agent<C> {
             // runtime. For CPU-bound tool work, the semaphore limits oversubscription.
             let (indices, futs): (Vec<usize>, Vec<ToolExecFut>) = tier_futs.into_iter().unzip();
 
-            let tier_results = tokio::select! {
-                results = futures::future::join_all(futs) => results,
-                () = cancel.cancelled() => {
-                    self.tool_executor.set_skill_env(None);
-                    tracing::info!("tool execution cancelled by user");
-                    self.update_metrics(|m| m.cancellations += 1);
-                    self.channel.send("[Cancelled]").await?;
-                    // Persist tombstone ToolResult for all tool_calls so the assistant ToolUse
-                    // persisted above is always paired in the DB (prevents cross-session orphan).
-                    self.persist_cancelled_tool_results(tool_calls).await;
-                    return Ok(());
-                }
+            // Poll tier futures, cancellation, and MCP elicitation requests concurrently.
+            // Elicitation events arrive from MCP server handlers that are blocked waiting on a
+            // oneshot response. Without draining them here the tier join never completes (deadlock).
+            let tier_results = {
+                let mut join_fut = std::pin::pin!(futures::future::join_all(futs));
+                // Take elicitation_rx out of self so we can hold &mut self for handling.
+                let mut elicitation_rx = self.mcp.elicitation_rx.take();
+                let result = loop {
+                    tokio::select! {
+                        results = &mut join_fut => break results,
+                        () = cancel.cancelled() => {
+                            self.mcp.elicitation_rx = elicitation_rx;
+                            self.tool_executor.set_skill_env(None);
+                            tracing::info!("tool execution cancelled by user");
+                            self.update_metrics(|m| m.cancellations += 1);
+                            self.channel.send("[Cancelled]").await?;
+                            // Persist tombstone ToolResult for all tool_calls so the assistant ToolUse
+                            // persisted above is always paired in the DB (prevents cross-session orphan).
+                            self.persist_cancelled_tool_results(tool_calls).await;
+                            return Ok(());
+                        }
+                        event = recv_elicitation(&mut elicitation_rx) => {
+                            if let Some(ev) = event {
+                                self.handle_elicitation_event(ev).await;
+                            } else {
+                                // Channel closed — stop polling it
+                                tracing::debug!("elicitation channel closed during tier exec");
+                                elicitation_rx = None;
+                            }
+                        }
+                    }
+                };
+                self.mcp.elicitation_rx = elicitation_rx;
+                result
             };
 
             // Store results and collect failed tool_use_ids for dependency propagation.
@@ -2252,6 +2274,19 @@ impl<C: Channel> Agent<C> {
     }
 }
 
+/// Receive the next elicitation event from an optional channel without blocking.
+///
+/// Returns `None` when the receiver is absent (no MCP elicitation configured) or the channel
+/// is closed, causing the `select!` branch to be disabled rather than polling indefinitely.
+async fn recv_elicitation(
+    rx: &mut Option<tokio::sync::mpsc::Receiver<zeph_mcp::ElicitationEvent>>,
+) -> Option<zeph_mcp::ElicitationEvent> {
+    match rx {
+        Some(r) => r.recv().await,
+        None => std::future::pending().await,
+    }
+}
+
 // T-CRIT-02: handle_focus_tool tests — happy path, error paths, checkpoint pinning (S5 fix).
 #[cfg(all(test, feature = "context-compression"))]
 mod tests {