use std::sync::{mpsc, Arc, Mutex};
use async_trait::async_trait;
use crate::mcp::{ElicitationRequest, ElicitationResponse, McpElicitationHandler};
use crate::permissions::{ApprovalOutcome, ApprovalRequest, PermissionsApprovalHandler};
use crate::subagents::QueuedApproval;
use super::bridge::{
PendingApprovalRequest, PendingChildApproval, PendingElicitation, PendingOAuthDisplay,
};
pub struct TuiApprovalHandler {
tx: mpsc::Sender<PendingApprovalRequest>,
}
impl TuiApprovalHandler {
pub fn new(tx: mpsc::Sender<PendingApprovalRequest>) -> Self {
TuiApprovalHandler { tx }
}
}
impl PermissionsApprovalHandler for TuiApprovalHandler {
fn ask(&self, req: &ApprovalRequest) -> ApprovalOutcome {
let (reply_tx, reply_rx) = mpsc::channel();
let pending = PendingApprovalRequest {
tool: req.tool.to_string(),
subject: req.subject.map(String::from),
raw_args: req.raw_args.clone(),
reply_tx,
};
if self.tx.send(pending).is_err() {
return ApprovalOutcome::Deny;
}
reply_rx.recv().unwrap_or(ApprovalOutcome::Deny)
}
}
pub struct TuiChildApprovalHandler {
child_agent_id: String,
queue: Arc<Mutex<Vec<QueuedApproval>>>,
tx: mpsc::Sender<PendingChildApproval>,
}
impl TuiChildApprovalHandler {
pub fn new(
child_agent_id: String,
queue: Arc<Mutex<Vec<QueuedApproval>>>,
tx: mpsc::Sender<PendingChildApproval>,
) -> Self {
TuiChildApprovalHandler {
child_agent_id,
queue,
tx,
}
}
}
impl PermissionsApprovalHandler for TuiChildApprovalHandler {
fn ask(&self, req: &ApprovalRequest) -> ApprovalOutcome {
if let Ok(mut q) = self.queue.lock() {
q.push(QueuedApproval {
child_agent_id: self.child_agent_id.clone(),
tool: req.tool.to_string(),
subject: req.subject.map(String::from),
queued_at_ms: now_ms(),
});
}
let (reply_tx, reply_rx) = mpsc::channel();
let pending = PendingChildApproval {
child_agent_id: self.child_agent_id.clone(),
tool: req.tool.to_string(),
subject: req.subject.map(String::from),
raw_args: req.raw_args.clone(),
reply_tx,
};
if self.tx.send(pending).is_err() {
return ApprovalOutcome::Deny;
}
reply_rx.recv().unwrap_or(ApprovalOutcome::Deny)
}
}
fn now_ms() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0)
}
pub struct TuiElicitationHandler {
tx: mpsc::Sender<PendingElicitation>,
}
impl TuiElicitationHandler {
pub fn new(tx: mpsc::Sender<PendingElicitation>) -> Self {
TuiElicitationHandler { tx }
}
}
#[async_trait]
impl McpElicitationHandler for TuiElicitationHandler {
async fn handle(&self, request: &ElicitationRequest) -> ElicitationResponse {
let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
let pending = PendingElicitation {
message: request.message.clone(),
requested_schema: request.requested_schema.clone(),
reply_tx,
};
if self.tx.send(pending).is_err() {
return ElicitationResponse {
action: crate::mcp::ElicitationAction::Cancel,
content: None,
};
}
reply_rx.await.unwrap_or(ElicitationResponse {
action: crate::mcp::ElicitationAction::Cancel,
content: None,
})
}
}
pub struct TuiBridge {
approval_tx: mpsc::Sender<PendingApprovalRequest>,
pub approval_rx: mpsc::Receiver<PendingApprovalRequest>,
child_approval_tx: mpsc::Sender<PendingChildApproval>,
pub child_approval_rx: mpsc::Receiver<PendingChildApproval>,
elicitation_tx: mpsc::Sender<PendingElicitation>,
pub elicitation_rx: mpsc::Receiver<PendingElicitation>,
oauth_tx: mpsc::Sender<PendingOAuthDisplay>,
pub oauth_rx: mpsc::Receiver<PendingOAuthDisplay>,
}
impl Default for TuiBridge {
fn default() -> Self {
Self::new()
}
}
impl TuiBridge {
pub fn new() -> Self {
let (approval_tx, approval_rx) = mpsc::channel();
let (child_approval_tx, child_approval_rx) = mpsc::channel();
let (elicitation_tx, elicitation_rx) = mpsc::channel();
let (oauth_tx, oauth_rx) = mpsc::channel();
TuiBridge {
approval_tx,
approval_rx,
child_approval_tx,
child_approval_rx,
elicitation_tx,
elicitation_rx,
oauth_tx,
oauth_rx,
}
}
pub fn install_on(&self, agent: &mut crate::agent::Agent) {
agent.set_permissions_approval_handler(TuiApprovalHandler::new(self.approval_tx.clone()));
agent.set_child_approval_handler_factory({
let tx = self.child_approval_tx.clone();
move |child_id, queue| {
Arc::new(TuiChildApprovalHandler::new(child_id, queue, tx.clone()))
as Arc<dyn PermissionsApprovalHandler>
}
});
}
pub fn elicitation_handler(&self) -> Arc<dyn McpElicitationHandler> {
Arc::new(TuiElicitationHandler::new(self.elicitation_tx.clone()))
}
pub fn oauth_sender(&self) -> mpsc::Sender<PendingOAuthDisplay> {
self.oauth_tx.clone()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::permissions::approval::resolve_ask;
use crate::permissions::ApprovalCache;
#[test]
fn approval_ask_blocks_until_render_loop_replies_allow_for_session_and_cache_then_skips_handler(
) {
let (tx, rx) = mpsc::channel::<PendingApprovalRequest>();
let handler = Arc::new(TuiApprovalHandler::new(tx));
let cache = Arc::new(ApprovalCache::new());
let h = handler.clone();
let c = cache.clone();
let worker = std::thread::spawn(move || {
let args = serde_json::json!({});
let req = ApprovalRequest {
tool: "bash",
subject: Some("ls -la"),
raw_args: &args,
};
resolve_ask(&c, Some(h.as_ref()), &req)
});
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
assert_eq!(pending.tool, "bash");
assert_eq!(pending.subject.as_deref(), Some("ls -la"));
pending
.reply_tx
.send(ApprovalOutcome::AllowForSession)
.unwrap();
let approved = worker.join().unwrap();
assert!(approved, "first Ask must be approved via the modal");
let args = serde_json::json!({});
let req2 = ApprovalRequest {
tool: "bash",
subject: Some("ls -la"),
raw_args: &args,
};
let approved2 = resolve_ask(&cache, Some(handler.as_ref()), &req2);
assert!(approved2);
assert!(
rx.try_recv().is_err(),
"a cached AllowForSession must skip the handler entirely"
);
}
#[test]
fn approval_ask_deny_is_not_cached() {
let (tx, rx) = mpsc::channel::<PendingApprovalRequest>();
let handler = Arc::new(TuiApprovalHandler::new(tx));
let cache = Arc::new(ApprovalCache::new());
let h = handler.clone();
let c = cache.clone();
let worker = std::thread::spawn(move || {
let args = serde_json::json!({});
let req = ApprovalRequest {
tool: "bash",
subject: Some("curl evil.example"),
raw_args: &args,
};
resolve_ask(&c, Some(h.as_ref()), &req)
});
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
pending.reply_tx.send(ApprovalOutcome::Deny).unwrap();
let approved = worker.join().unwrap();
assert!(!approved);
assert!(!cache.is_approved(&ApprovalCache::key("bash", Some("curl evil.example"))));
}
#[test]
fn approval_ask_fails_closed_when_render_loop_is_gone() {
let (tx, rx) = mpsc::channel::<PendingApprovalRequest>();
let handler = TuiApprovalHandler::new(tx);
drop(rx); let args = serde_json::json!({});
let req = ApprovalRequest {
tool: "bash",
subject: None,
raw_args: &args,
};
assert_eq!(handler.ask(&req), ApprovalOutcome::Deny);
}
#[test]
fn approval_ask_fails_closed_when_reply_sender_is_dropped_without_replying() {
let (tx, rx) = mpsc::channel::<PendingApprovalRequest>();
let handler = Arc::new(TuiApprovalHandler::new(tx));
let h = handler.clone();
let worker = std::thread::spawn(move || {
let args = serde_json::json!({});
let req = ApprovalRequest {
tool: "bash",
subject: None,
raw_args: &args,
};
h.ask(&req)
});
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
drop(pending.reply_tx); assert_eq!(worker.join().unwrap(), ApprovalOutcome::Deny);
}
#[test]
fn approval_handler_never_consulted_when_rule_engine_already_denies() {
use crate::permissions::approval::decision_to_approved;
use crate::permissions::rules::Decision;
let (tx, rx) = mpsc::channel::<PendingApprovalRequest>();
let handler = TuiApprovalHandler::new(tx);
let approved = decision_to_approved(Decision::Deny, || {
handler.ask(&ApprovalRequest {
tool: "bash",
subject: None,
raw_args: &serde_json::json!({}),
}) == ApprovalOutcome::Allow
});
assert!(!approved);
assert!(rx.try_recv().is_err(), "handler must never have been asked");
}
#[test]
fn child_approval_handler_blocks_for_an_answer_instead_of_immediate_deny() {
let (tx, rx) = mpsc::channel::<PendingChildApproval>();
let queue = Arc::new(Mutex::new(Vec::new()));
let handler = Arc::new(TuiChildApprovalHandler::new(
"agent-bg-7".to_string(),
queue.clone(),
tx,
));
let h = handler.clone();
let worker = std::thread::spawn(move || {
let args = serde_json::json!({});
let subject = "/workspace/out.txt".to_string();
let req = ApprovalRequest {
tool: "write_file",
subject: Some(subject.as_str()),
raw_args: &args,
};
h.ask(&req)
});
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
assert_eq!(pending.child_agent_id, "agent-bg-7");
assert_eq!(pending.tool, "write_file");
pending.reply_tx.send(ApprovalOutcome::Allow).unwrap();
assert_eq!(worker.join().unwrap(), ApprovalOutcome::Allow);
let recorded = queue.lock().unwrap();
assert_eq!(recorded.len(), 1);
assert_eq!(recorded[0].child_agent_id, "agent-bg-7");
}
#[test]
fn child_approval_handler_fails_closed_when_nobody_answers() {
let (tx, rx) = mpsc::channel::<PendingChildApproval>();
let queue = Arc::new(Mutex::new(Vec::new()));
let handler = Arc::new(TuiChildApprovalHandler::new(
"agent-bg-8".to_string(),
queue,
tx,
));
let h = handler.clone();
let worker = std::thread::spawn(move || {
let args = serde_json::json!({});
let req = ApprovalRequest {
tool: "bash",
subject: None,
raw_args: &args,
};
h.ask(&req)
});
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
drop(pending); assert_eq!(worker.join().unwrap(), ApprovalOutcome::Deny);
}
#[tokio::test]
async fn elicitation_handler_returns_the_modals_accept_answer() {
let (tx, rx) = mpsc::channel::<PendingElicitation>();
let handler = TuiElicitationHandler::new(tx);
let request = ElicitationRequest {
message: "What's the deploy tag?".to_string(),
requested_schema: serde_json::json!({"properties": {"tag": {"type": "string"}}}),
};
let handle_fut = handler.handle(&request);
let reply_task = tokio::task::spawn_blocking(move || {
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
assert_eq!(pending.message, "What's the deploy tag?");
pending
.reply_tx
.send(ElicitationResponse {
action: crate::mcp::ElicitationAction::Accept,
content: Some(serde_json::json!({"tag": "v1.2.3"})),
})
.unwrap();
});
let (resp, _) = tokio::join!(handle_fut, reply_task);
assert_eq!(resp.action, crate::mcp::ElicitationAction::Accept);
assert_eq!(resp.content, Some(serde_json::json!({"tag": "v1.2.3"})));
}
#[tokio::test]
async fn elicitation_handler_cancels_when_render_loop_is_gone() {
let (tx, rx) = mpsc::channel::<PendingElicitation>();
let handler = TuiElicitationHandler::new(tx);
drop(rx);
let request = ElicitationRequest {
message: "…".to_string(),
requested_schema: serde_json::json!({}),
};
let resp = handler.handle(&request).await;
assert_eq!(resp.action, crate::mcp::ElicitationAction::Cancel);
assert_eq!(resp.content, None);
}
#[tokio::test]
async fn elicitation_handler_cancels_when_reply_sender_dropped_without_replying() {
let (tx, rx) = mpsc::channel::<PendingElicitation>();
let handler = TuiElicitationHandler::new(tx);
let request = ElicitationRequest {
message: "…".to_string(),
requested_schema: serde_json::json!({}),
};
let handle_fut = handler.handle(&request);
let drop_task = tokio::task::spawn_blocking(move || {
let pending = rx.recv_timeout(std::time::Duration::from_secs(5)).unwrap();
drop(pending); });
let (resp, _) = tokio::join!(handle_fut, drop_task);
assert_eq!(resp.action, crate::mcp::ElicitationAction::Cancel);
}
#[test]
fn bridge_install_on_wires_a_working_approval_handler() {
let bridge = TuiBridge::new();
let mut agent = crate::agent::Agent::new(
crate::Config::builder()
.model("test/model")
.api_key("test-key")
.build(),
)
.expect("agent construction");
bridge.install_on(&mut agent);
assert!(bridge.approval_rx.try_recv().is_err());
assert!(bridge.child_approval_rx.try_recv().is_err());
}
#[test]
fn bridge_elicitation_handler_feeds_the_bridges_receiver() {
let bridge = TuiBridge::new();
let handler = bridge.elicitation_handler();
let h = handler.clone();
let worker = std::thread::spawn(move || {
let rt = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap();
rt.block_on(async {
let req = ElicitationRequest {
message: "hi".to_string(),
requested_schema: serde_json::json!({}),
};
h.handle(&req).await
})
});
let pending = bridge
.elicitation_rx
.recv_timeout(std::time::Duration::from_secs(5))
.unwrap();
pending
.reply_tx
.send(ElicitationResponse {
action: crate::mcp::ElicitationAction::Decline,
content: None,
})
.unwrap();
let resp = worker.join().unwrap();
assert_eq!(resp.action, crate::mcp::ElicitationAction::Decline);
}
}