diff --git a/codi-rs/docs/ROADMAP.md b/codi-rs/docs/ROADMAP.md index eef6ac9d..79a56e61 100644 --- a/codi-rs/docs/ROADMAP.md +++ b/codi-rs/docs/ROADMAP.md @@ -9,10 +9,10 @@ This roadmap focuses on the Rust CLI (`codi-rs`) and its TUI/orchestration stack ## P0: Stability and Cross-Platform Foundations -1) Cross-platform IPC for multi-agent -- Replace Unix domain sockets with an IPC abstraction. -- Implement Windows named pipes (or a transport-agnostic layer). -- Ensure commander/worker handshake is deterministic with explicit timeouts. +1) Cross-platform IPC for multi-agent (complete) +- Transport abstraction with Windows named pipes. +- Deterministic commander/worker handshake with explicit timeouts. +- Windows IPC tests for roundtrip + handshake/permission flows. 2) Cancellation and lifecycle correctness - Wire the TUI cancel flow to actual worker cancellation. diff --git a/codi-rs/src/agent/mod.rs b/codi-rs/src/agent/mod.rs index deb8aa2f..1699eea7 100644 --- a/codi-rs/src/agent/mod.rs +++ b/codi-rs/src/agent/mod.rs @@ -44,6 +44,8 @@ pub use types::{ use std::sync::Arc; use std::time::{Duration, Instant}; +use tokio::sync::watch; + use crate::error::{AgentError, Result}; use crate::types::{ BoxedProvider, ContentBlock, Message, Role, StreamEvent, @@ -440,6 +442,27 @@ impl Agent { /// Takes a user message, sends it to the model, handles any tool calls, /// and returns the final text response. pub async fn chat(&mut self, user_message: &str) -> Result { + self.chat_with_cancel_internal(user_message, None).await + } + + /// The main agentic loop with a cancellation signal. + /// + /// If `cancel_rx` is triggered, the request short-circuits with + /// `AgentError::UserCancelled`. + pub async fn chat_with_cancel( + &mut self, + user_message: &str, + cancel_rx: watch::Receiver, + ) -> Result { + self.chat_with_cancel_internal(user_message, Some(cancel_rx)).await + } + + async fn chat_with_cancel_internal( + &mut self, + user_message: &str, + cancel_rx: Option>, + ) -> Result { + let mut cancel_rx = cancel_rx; let start_time = Instant::now(); let max_duration = Duration::from_millis(self.config.max_turn_duration_ms); @@ -459,6 +482,12 @@ impl Agent { // Main loop loop { + if let Some(rx) = cancel_rx.as_ref() { + if *rx.borrow() { + return Err(AgentError::UserCancelled.into()); + } + } + self.state.current_iteration += 1; // Check iteration limit @@ -487,23 +516,56 @@ impl Agent { let on_stream_event = self.callbacks.on_stream_event.clone(); // Call the provider with streaming - let response = self.provider.stream_chat( - &self.state.messages, - tools.as_deref(), - Some(&system_context), - Box::new(move |event| { - // Forward raw stream events - if let Some(ref cb) = on_stream_event { - cb(&event); - } - // Fire on_text for text deltas - if let StreamEvent::TextDelta(ref text) = event { - if let Some(ref cb) = on_text { - cb(text); + let response = if let Some(rx) = cancel_rx.as_mut() { + if *rx.borrow() { + return Err(AgentError::UserCancelled.into()); + } + tokio::select! { + res = self.provider.stream_chat( + &self.state.messages, + tools.as_deref(), + Some(&system_context), + Box::new(move |event| { + // Forward raw stream events + if let Some(ref cb) = on_stream_event { + cb(&event); + } + // Fire on_text for text deltas + if let StreamEvent::TextDelta(ref text) = event { + if let Some(ref cb) = on_text { + cb(text); + } + } + }), + ) => res?, + _ = rx.changed() => { + if *rx.borrow() { + return Err(AgentError::UserCancelled.into()); } + continue; } - }), - ).await?; + } + } else { + self.provider + .stream_chat( + &self.state.messages, + tools.as_deref(), + Some(&system_context), + Box::new(move |event| { + // Forward raw stream events + if let Some(ref cb) = on_stream_event { + cb(&event); + } + // Fire on_text for text deltas + if let StreamEvent::TextDelta(ref text) = event { + if let Some(ref cb) = on_text { + cb(text); + } + } + }), + ) + .await? + }; // Update token stats if let Some(ref usage) = response.usage { @@ -543,7 +605,24 @@ impl Agent { } // Process tool calls - match self.process_tool_calls(&response.tool_calls, &mut turn_stats).await { + let tool_result = if let Some(rx) = cancel_rx.as_mut() { + if *rx.borrow() { + return Err(AgentError::UserCancelled.into()); + } + tokio::select! { + res = self.process_tool_calls(&response.tool_calls, &mut turn_stats) => res, + _ = rx.changed() => { + if *rx.borrow() { + return Err(AgentError::UserCancelled.into()); + } + continue; + } + } + } else { + self.process_tool_calls(&response.tool_calls, &mut turn_stats).await + }; + + match tool_result { Ok((results, has_error)) => { // Add tool results to history self.add_tool_results(results); @@ -599,6 +678,12 @@ impl Agent { #[cfg(test)] mod tests { use super::*; + use async_trait::async_trait; + use tokio::sync::watch; + use tokio::time::Duration; + + use crate::error::ProviderError; + use crate::types::{Provider, ProviderResponse}; #[test] fn test_agent_config_default() { @@ -735,4 +820,73 @@ mod tests { // Invalid pattern should not cause a panic assert_eq!(config.matches_dangerous_pattern("hello"), None); } + + struct SlowProvider { + delay: Duration, + } + + #[async_trait] + impl Provider for SlowProvider { + async fn chat( + &self, + _messages: &[Message], + _tools: Option<&[ToolDefinition]>, + _system_prompt: Option<&str>, + ) -> std::result::Result { + self.stream_chat(_messages, _tools, _system_prompt, Box::new(|_| {})) + .await + } + + async fn stream_chat( + &self, + _messages: &[Message], + _tools: Option<&[ToolDefinition]>, + _system_prompt: Option<&str>, + on_event: Box, + ) -> std::result::Result { + on_event(StreamEvent::TextDelta("partial".to_string())); + tokio::time::sleep(self.delay).await; + Ok(ProviderResponse::text("done")) + } + + fn supports_tool_use(&self) -> bool { + false + } + + fn name(&self) -> &str { + "slow" + } + + fn model(&self) -> &str { + "slow-model" + } + } + + #[tokio::test] + async fn test_chat_with_cancel_returns_user_cancelled() { + let provider: BoxedProvider = Box::new(SlowProvider { + delay: Duration::from_millis(200), + }); + let registry = Arc::new(ToolRegistry::with_defaults()); + let mut agent = Agent::new(AgentOptions { + provider, + tool_registry: registry, + system_prompt: None, + config: AgentConfig::default(), + callbacks: AgentCallbacks::default(), + }); + + let (tx, rx) = watch::channel(false); + tokio::spawn(async move { + tokio::time::sleep(Duration::from_millis(10)).await; + let _ = tx.send(true); + }); + + let result = agent.chat_with_cancel("hello", rx).await; + let err = result.unwrap_err(); + let cancelled = err + .downcast_ref::() + .is_some_and(|e| matches!(e, AgentError::UserCancelled)); + assert!(cancelled); + } } diff --git a/codi-rs/src/config/loader.rs b/codi-rs/src/config/loader.rs index 36d0f1fd..a3b27ada 100644 --- a/codi-rs/src/config/loader.rs +++ b/codi-rs/src/config/loader.rs @@ -170,6 +170,14 @@ mod tests { assert!(dir.ends_with(".codi")); } + #[cfg(windows)] + #[test] + fn test_global_config_dir_windows() { + let home = dirs::home_dir().expect("home dir"); + let expected = home.join(GLOBAL_CONFIG_DIR); + assert_eq!(get_global_config_dir().unwrap(), expected); + } + #[test] fn test_load_workspace_config_not_found() { let temp = TempDir::new().unwrap(); diff --git a/codi-rs/src/orchestrate/ipc/transport.rs b/codi-rs/src/orchestrate/ipc/transport.rs index cf2451ae..acb457bd 100644 --- a/codi-rs/src/orchestrate/ipc/transport.rs +++ b/codi-rs/src/orchestrate/ipc/transport.rs @@ -8,10 +8,11 @@ use std::path::Path; use tokio::io::{AsyncRead, AsyncWrite}; -pub trait IpcStreamTrait: AsyncRead + AsyncWrite {} -impl IpcStreamTrait for T {} +pub trait IpcIo: AsyncRead + AsyncWrite + Unpin + Send {} -pub type IpcStream = Box; +impl IpcIo for T where T: AsyncRead + AsyncWrite + Unpin + Send {} + +pub type IpcStream = Box; #[cfg(unix)] use tokio::net::UnixListener; diff --git a/codi-rs/src/tools/handlers/bash.rs b/codi-rs/src/tools/handlers/bash.rs index 41d380f3..2309afdb 100644 --- a/codi-rs/src/tools/handlers/bash.rs +++ b/codi-rs/src/tools/handlers/bash.rs @@ -276,12 +276,44 @@ mod tests { use super::*; use tempfile::tempdir; + fn echo_command() -> &'static str { + if cfg!(windows) { + "echo hello world" + } else { + "echo 'hello world'" + } + } + + fn list_command() -> &'static str { + if cfg!(windows) { + "dir /B" + } else { + "ls" + } + } + + fn stderr_command() -> &'static str { + if cfg!(windows) { + "echo error 1>&2" + } else { + "echo 'error' >&2" + } + } + + fn timeout_command() -> &'static str { + if cfg!(windows) { + "ping -n 5 127.0.0.1 > NUL" + } else { + "sleep 10" + } + } + #[tokio::test] async fn test_bash_echo() { let handler = BashHandler; let result = handler .execute(serde_json::json!({ - "command": "echo 'hello world'" + "command": echo_command() })) .await .unwrap(); @@ -298,7 +330,7 @@ mod tests { let handler = BashHandler; let result = handler .execute(serde_json::json!({ - "command": "ls", + "command": list_command(), "cwd": temp.path().to_str().unwrap() })) .await @@ -327,7 +359,7 @@ mod tests { let handler = BashHandler; let result = handler .execute(serde_json::json!({ - "command": "echo 'error' >&2" + "command": stderr_command() })) .await .unwrap(); @@ -366,7 +398,7 @@ mod tests { let handler = BashHandler; let result = handler .execute(serde_json::json!({ - "command": "sleep 10", + "command": timeout_command(), "timeout": 100 // 100ms timeout })) .await diff --git a/codi-rs/src/tui/app.rs b/codi-rs/src/tui/app.rs index b91893ba..c49c89d9 100644 --- a/codi-rs/src/tui/app.rs +++ b/codi-rs/src/tui/app.rs @@ -10,14 +10,14 @@ use std::sync::Arc; use crossterm::event::{KeyCode, KeyModifiers}; use ratatui::prelude::*; use ratatui::text::Line; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, watch}; use crate::agent::{ Agent, AgentCallbacks, AgentConfig, AgentOptions, ConfirmationResult, ToolConfirmation, TurnStats, }; use crate::config::ResolvedConfig; -use crate::error::{Result as CodiResult, ToolError}; +use crate::error::{AgentError, Result as CodiResult, ToolError}; use crate::completion::{complete_line, get_completion_matches}; use crate::orchestrate::{Commander, CommanderConfig, WorkerConfig, WorkerStatus, WorkspaceInfo, PermissionResult}; use crate::session::{Session, SessionInfo, SessionService}; @@ -220,6 +220,10 @@ pub struct App { // Background agent task /// Receiver for agent returning from a background chat task. pending_agent: Option)>>, + /// Cancellation signal for the in-flight agent task. + pending_agent_cancel: Option>, + /// Whether a cancel request is in flight. + cancel_requested: bool, // Tool execution visualization /// Manager for tool execution cells. pub exec_cells: crate::tui::components::ExecCellManager, @@ -272,6 +276,8 @@ impl App { config: None, auto_approve_all: false, pending_agent: None, + pending_agent_cancel: None, + cancel_requested: false, exec_cells: crate::tui::components::ExecCellManager::new(), commander: None, pending_worker_permissions: Vec::new(), @@ -433,23 +439,44 @@ impl App { Ok((agent, result)) => { self.agent = Some(agent); self.pending_agent = None; + self.pending_agent_cancel = None; + self.cancel_requested = false; match result { Ok(_) => { // Response was streamed via callbacks; TurnComplete will finalize } Err(e) => { - self.status = Some(format!("Error: {}", e)); - self.mode = AppMode::Normal; - self.finalize_streaming(); + let cancelled = e + .downcast_ref::() + .is_some_and(|err| matches!(err, AgentError::UserCancelled)); + if cancelled { + self.status = Some("Cancelled".to_string()); + self.mode = AppMode::Normal; + self.turn_start_time = None; + self.finalize_streaming(); + } else { + self.status = Some(format!("Error: {}", e)); + self.mode = AppMode::Normal; + self.finalize_streaming(); + } } } } Err(tokio::sync::oneshot::error::TryRecvError::Closed) => { // Task panicked or was dropped self.pending_agent = None; - self.status = Some("Agent task failed unexpectedly".to_string()); - self.mode = AppMode::Normal; - self.finalize_streaming(); + self.pending_agent_cancel = None; + if self.cancel_requested { + self.status = Some("Cancelled".to_string()); + self.mode = AppMode::Normal; + self.turn_start_time = None; + self.finalize_streaming(); + } else { + self.status = Some("Agent task failed unexpectedly".to_string()); + self.mode = AppMode::Normal; + self.finalize_streaming(); + } + self.cancel_requested = false; } Err(tokio::sync::oneshot::error::TryRecvError::Empty) => { // Still running, keep waiting @@ -524,6 +551,10 @@ impl App { /// Handle text delta from streaming. fn handle_text_delta(&mut self, text: &str) { + if self.cancel_requested { + return; + } + // Initialize stream controller if needed if self.stream_controller.is_none() { let width = self.terminal_width.map(|w| (w.saturating_sub(4)) as usize); @@ -832,9 +863,19 @@ impl App { /// Handle key while waiting for response. fn handle_waiting_key(&mut self, key: KeyCode) { if key == KeyCode::Esc { - // Cancel request (TODO: actually cancel) - self.mode = AppMode::Normal; - self.status = Some("Cancelled".to_string()); + self.request_cancel(); + } + } + + fn request_cancel(&mut self) { + if self.cancel_requested { + return; + } + + if let Some(tx) = self.pending_agent_cancel.as_ref() { + let _ = tx.send(true); + self.cancel_requested = true; + self.status = Some("Cancelling...".to_string()); self.finalize_streaming(); } } @@ -902,9 +943,12 @@ impl App { let (tx, rx) = tokio::sync::oneshot::channel(); self.pending_agent = Some(rx); + let (cancel_tx, cancel_rx) = watch::channel(false); + self.pending_agent_cancel = Some(cancel_tx); + self.cancel_requested = false; tokio::spawn(async move { - let result = agent.chat(&prompt).await; + let result = agent.chat_with_cancel(&prompt, cancel_rx).await; let _ = tx.send((agent, result)); }); } @@ -928,10 +972,13 @@ impl App { // Create a oneshot channel to get the agent back when done let (tx, rx) = tokio::sync::oneshot::channel(); self.pending_agent = Some(rx); + let (cancel_tx, cancel_rx) = watch::channel(false); + self.pending_agent_cancel = Some(cancel_tx); + self.cancel_requested = false; // Spawn the agent chat on a background task tokio::spawn(async move { - let result = agent.chat(&input).await; + let result = agent.chat_with_cancel(&input, cancel_rx).await; // Send the agent and result back (ignore error if receiver dropped) let _ = tx.send((agent, result)); }); diff --git a/codi-rs/src/tui/terminal_ui.rs b/codi-rs/src/tui/terminal_ui.rs index 08856097..12361901 100644 --- a/codi-rs/src/tui/terminal_ui.rs +++ b/codi-rs/src/tui/terminal_ui.rs @@ -15,7 +15,7 @@ use std::sync::Arc; use std::time::Instant; use crossterm::{ - style::{Color, Print, ResetColor, SetForegroundColor}, + style::{Color, Print, ResetColor, SetForegroundColor, Stylize}, ExecutableCommand, };