diff --git a/codi-rs/src/orchestrate/child_agent.rs b/codi-rs/src/orchestrate/child_agent.rs index a06bdca..e53b8fc 100644 --- a/codi-rs/src/orchestrate/child_agent.rs +++ b/codi-rs/src/orchestrate/child_agent.rs @@ -8,7 +8,7 @@ //! //! # Lifecycle //! -//! 1. Connect to commander's IPC socket +//! 1. Connect to commander's IPC endpoint //! 2. Perform handshake with worker config //! 3. Execute task with agent loop //! 4. Request permissions via IPC when needed diff --git a/codi-rs/src/orchestrate/commander.rs b/codi-rs/src/orchestrate/commander.rs index 0e30e95..0fb19ef 100644 --- a/codi-rs/src/orchestrate/commander.rs +++ b/codi-rs/src/orchestrate/commander.rs @@ -12,8 +12,8 @@ //! ┌─────────────────────────────────────────────────────────────┐ //! │ Commander │ //! │ ┌──────────────────────────────────────────────────────┐ │ -//! │ │ IPC Server (Unix domain socket) │ │ -//! │ │ ~/.codi/orchestrator.sock │ │ +//! │ │ IPC Server (socket/pipe) │ │ +//! │ │ ~/.codi/orchestrator.sock or \\\\.\\pipe\\... │ │ //! │ └──────────────────┬───────────────────────────────────┘ │ //! │ │ │ //! │ ┌──────────────────┴───────────────────────────────────┐ │ diff --git a/codi-rs/src/orchestrate/ipc/client.rs b/codi-rs/src/orchestrate/ipc/client.rs index a3074ba..7fb8518 100644 --- a/codi-rs/src/orchestrate/ipc/client.rs +++ b/codi-rs/src/orchestrate/ipc/client.rs @@ -3,7 +3,7 @@ //! IPC client for worker agents. //! -//! The client connects to the commander's Unix domain socket and handles +//! The client connects to the commander's IPC endpoint and handles //! bidirectional communication for permission requests and status updates. use std::collections::HashMap; @@ -12,7 +12,6 @@ use std::sync::Arc; use std::time::Duration; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::UnixStream; use tokio::sync::{mpsc, oneshot, Mutex}; use tracing::{debug, error, info, warn}; @@ -22,6 +21,7 @@ use crate::types::TokenUsage; use super::protocol::{ decode, encode, CommanderMessage, PermissionResult, WorkerMessage, }; +use super::transport::{self, IpcStream}; use super::super::types::{WorkerConfig, WorkerResult, WorkerStatus, WorkspaceInfo}; /// Error type for IPC client operations. @@ -75,12 +75,12 @@ struct PendingPermission { /// IPC client for worker-commander communication. pub struct IpcClient { - /// Path to the Unix socket. + /// Path to the IPC endpoint. socket_path: PathBuf, /// Worker ID. worker_id: String, - /// Writer half of the socket. - writer: Option>, + /// Writer half of the stream. + writer: Option>, /// Pending permission requests by request ID. pending_permissions: Arc>>, /// Channel for cancel signals. @@ -105,9 +105,9 @@ impl IpcClient { } } - /// Connect to the commander's socket. + /// Connect to the commander's endpoint. pub async fn connect(&mut self) -> Result<(), IpcClientError> { - let stream = UnixStream::connect(&self.socket_path).await?; + let stream = transport::connect(&self.socket_path).await?; let (read_half, write_half) = tokio::io::split(stream); self.writer = Some(write_half); diff --git a/codi-rs/src/orchestrate/ipc/mod.rs b/codi-rs/src/orchestrate/ipc/mod.rs index 3fc32bb..b9905ff 100644 --- a/codi-rs/src/orchestrate/ipc/mod.rs +++ b/codi-rs/src/orchestrate/ipc/mod.rs @@ -3,8 +3,8 @@ //! IPC (Inter-Process Communication) module for commander-worker communication. //! -//! This module provides Unix domain socket-based communication between the -//! commander (orchestrator) and worker (child agent) processes. +//! This module provides cross-platform IPC between the commander (orchestrator) +//! and worker (child agent) processes. //! //! # Architecture //! @@ -12,8 +12,8 @@ //! ┌─────────────────┐ ┌─────────────────┐ //! │ Commander │ │ Worker │ //! │ │ │ │ -//! │ ┌───────────┐ │ Unix │ ┌───────────┐ │ -//! │ │ Server │◄─┼──Socket──────┼──│ Client │ │ +//! │ ┌───────────┐ │ Socket/ │ ┌───────────┐ │ +//! │ │ Server │◄─┼──Pipe────────┼──│ Client │ │ //! │ └───────────┘ │ │ └───────────┘ │ //! └─────────────────┘ └─────────────────┘ //! ``` @@ -23,6 +23,10 @@ //! Messages are newline-delimited JSON (NDJSON). Each message is a complete //! JSON object followed by a newline character. //! +//! Transport: +//! - Unix: domain sockets +//! - Windows: named pipes +//! //! ## Worker → Commander Messages //! //! - `handshake` - Initial connection from worker @@ -44,6 +48,7 @@ pub mod protocol; pub mod server; pub mod client; +pub mod transport; pub use protocol::{ WorkerMessage, CommanderMessage, PermissionResult, @@ -52,3 +57,70 @@ pub use protocol::{ }; pub use server::IpcServer; pub use client::IpcClient; + +#[cfg(test)] +mod tests { + use super::{CommanderMessage, IpcClient, IpcServer, WorkerMessage}; + use crate::orchestrate::types::{WorkerConfig, WorkspaceInfo}; + use std::path::PathBuf; + use std::sync::Arc; + + #[cfg(windows)] + #[tokio::test] + async fn test_named_pipe_handshake_roundtrip() { + let pipe_name = format!(r"\\.\pipe\codi-ipc-handshake-{}", uuid::Uuid::new_v4()); + let socket_path = PathBuf::from(pipe_name); + + let mut server = IpcServer::new(&socket_path); + server.start().await.expect("server start failed"); + + let mut rx = server.take_receiver().expect("receiver already taken"); + let server = Arc::new(server); + + let accept_server = Arc::clone(&server); + let accept_task = tokio::spawn(async move { + accept_server.accept().await.expect("accept failed") + }); + + let ack_server = Arc::clone(&server); + let ack_task = tokio::spawn(async move { + let (worker_id, msg) = rx.recv().await.expect("handshake missing"); + assert_eq!(worker_id, "worker-1"); + assert!(matches!(msg, WorkerMessage::Handshake { .. })); + + let ack = CommanderMessage::handshake_ack( + true, + vec!["read_file".to_string()], + vec!["rm -rf".to_string()], + 1_234, + ); + + ack_server + .send(&worker_id, &ack) + .await + .expect("ack send failed"); + }); + + let mut client = IpcClient::new(&socket_path, "worker-1"); + client.connect().await.expect("client connect failed"); + + let workspace = WorkspaceInfo::GitWorktree { + path: PathBuf::from("."), + branch: "feat/test".to_string(), + base_branch: "main".to_string(), + }; + let config = WorkerConfig::new("worker-1", "feat/test", "task"); + + let ack = client + .handshake(&config, &workspace) + .await + .expect("handshake failed"); + + assert_eq!(ack.auto_approve, vec!["read_file".to_string()]); + assert_eq!(ack.dangerous_patterns, vec!["rm -rf".to_string()]); + assert_eq!(ack.timeout_ms, 1_234); + + accept_task.await.expect("accept task failed"); + ack_task.await.expect("ack task failed"); + } +} diff --git a/codi-rs/src/orchestrate/ipc/protocol.rs b/codi-rs/src/orchestrate/ipc/protocol.rs index b64ae2f..910cfcf 100644 --- a/codi-rs/src/orchestrate/ipc/protocol.rs +++ b/codi-rs/src/orchestrate/ipc/protocol.rs @@ -3,7 +3,7 @@ //! IPC protocol for commander-worker communication. //! -//! Uses newline-delimited JSON over Unix domain sockets. +//! Uses newline-delimited JSON over a platform-specific IPC transport. use serde::{Deserialize, Serialize}; use uuid::Uuid; diff --git a/codi-rs/src/orchestrate/ipc/server.rs b/codi-rs/src/orchestrate/ipc/server.rs index 32defd8..490f846 100644 --- a/codi-rs/src/orchestrate/ipc/server.rs +++ b/codi-rs/src/orchestrate/ipc/server.rs @@ -3,21 +3,21 @@ //! IPC server for the commander. //! -//! The server listens on a Unix domain socket and handles connections -//! from worker processes. +//! The server listens on a platform-specific IPC transport and handles +//! connections from worker processes. use std::collections::HashMap; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; -use tokio::net::{UnixListener, UnixStream}; use tokio::sync::{mpsc, Mutex, RwLock}; use tracing::{debug, error, info, warn}; use super::protocol::{ decode, encode, CommanderMessage, WorkerMessage, }; +use super::transport::{self, IpcListener, IpcStream}; /// Error type for IPC operations. #[derive(Debug, thiserror::Error)] @@ -43,18 +43,18 @@ pub enum IpcError { /// A connected worker client. struct ConnectedWorker { - /// Write half of the socket. - writer: tokio::io::WriteHalf, + /// Write half of the stream. + writer: tokio::io::WriteHalf, /// Worker ID (stored for logging/diagnostics). _worker_id: String, } /// IPC server for commander-worker communication. pub struct IpcServer { - /// Path to the Unix socket. + /// Path to the IPC endpoint. socket_path: PathBuf, /// Listener (set after start). - listener: Option, + listener: Option, /// Connected workers by ID. workers: Arc>>>>, /// Channel for incoming messages (worker_id, message). @@ -76,25 +76,14 @@ impl IpcServer { } } - /// Get the socket path. + /// Get the IPC endpoint path. pub fn socket_path(&self) -> &Path { &self.socket_path } /// Start the server. pub async fn start(&mut self) -> Result<(), IpcError> { - // Remove existing socket file if present - if self.socket_path.exists() { - std::fs::remove_file(&self.socket_path)?; - } - - // Create parent directory if needed - if let Some(parent) = self.socket_path.parent() { - std::fs::create_dir_all(parent)?; - } - - // Bind the listener - let listener = UnixListener::bind(&self.socket_path)?; + let listener = transport::bind(&self.socket_path).await?; info!("IPC server listening on {:?}", self.socket_path); self.listener = Some(listener); @@ -107,10 +96,7 @@ impl IpcServer { let mut workers = self.workers.write().await; workers.clear(); - // Remove socket file - if self.socket_path.exists() { - std::fs::remove_file(&self.socket_path)?; - } + transport::cleanup(&self.socket_path)?; self.listener = None; info!("IPC server stopped"); @@ -131,7 +117,7 @@ impl IpcServer { pub async fn accept(&self) -> Result { let listener = self.listener.as_ref().ok_or(IpcError::NotStarted)?; - let (stream, _addr) = listener.accept().await?; + let stream = listener.accept().await?; debug!("New connection accepted"); let (read_half, write_half) = tokio::io::split(stream); @@ -177,7 +163,7 @@ impl IpcServer { /// Background task to read messages from a worker. async fn read_worker_messages( - mut reader: BufReader>, + mut reader: BufReader>, worker_id: String, workers: Arc>>>>, tx: mpsc::Sender<(String, WorkerMessage)>, @@ -269,10 +255,7 @@ impl IpcServer { impl Drop for IpcServer { fn drop(&mut self) { - // Clean up socket file - if self.socket_path.exists() { - let _ = std::fs::remove_file(&self.socket_path); - } + let _ = transport::cleanup(&self.socket_path); } } @@ -287,12 +270,15 @@ mod tests { let socket_path = dir.path().join("test.sock"); let mut server = IpcServer::new(&socket_path); + #[cfg(not(windows))] assert!(!socket_path.exists()); server.start().await.unwrap(); + #[cfg(not(windows))] assert!(socket_path.exists()); server.stop().await.unwrap(); + #[cfg(not(windows))] assert!(!socket_path.exists()); } diff --git a/codi-rs/src/orchestrate/ipc/transport.rs b/codi-rs/src/orchestrate/ipc/transport.rs new file mode 100644 index 0000000..2eacd70 --- /dev/null +++ b/codi-rs/src/orchestrate/ipc/transport.rs @@ -0,0 +1,150 @@ +// Copyright 2026 Layne Penney +// SPDX-License-Identifier: AGPL-3.0-or-later + +//! Cross-platform transport helpers for IPC. + +use std::io; +use std::path::Path; + +use tokio::io::{AsyncRead, AsyncWrite}; + +pub type IpcStream = Box; + +#[cfg(unix)] +use tokio::net::UnixListener; +#[cfg(unix)] +use tokio::net::UnixStream; + +#[cfg(windows)] +use tokio::net::windows::named_pipe::{ClientOptions, ServerOptions}; + +pub struct IpcListener { + #[cfg(unix)] + inner: UnixListener, + #[cfg(windows)] + name: String, +} + +pub async fn bind(path: &Path) -> io::Result { + #[cfg(unix)] + { + if path.exists() { + let _ = std::fs::remove_file(path); + } + + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + + let inner = UnixListener::bind(path)?; + Ok(IpcListener { inner }) + } + + #[cfg(windows)] + { + Ok(IpcListener { + name: pipe_name_from_path(path), + }) + } +} + +pub async fn connect(path: &Path) -> io::Result { + #[cfg(unix)] + { + let stream = UnixStream::connect(path).await?; + Ok(Box::new(stream)) + } + + #[cfg(windows)] + { + let name = pipe_name_from_path(path); + let mut attempts = 0; + loop { + match ClientOptions::new().open(&name) { + Ok(client) => return Ok(Box::new(client)), + Err(err) if attempts < 50 => { + attempts += 1; + tokio::time::sleep(tokio::time::Duration::from_millis(50)).await; + continue; + } + Err(err) => return Err(err), + } + } + } +} + +impl IpcListener { + pub async fn accept(&self) -> io::Result { + #[cfg(unix)] + { + let (stream, _addr) = self.inner.accept().await?; + Ok(Box::new(stream)) + } + + #[cfg(windows)] + { + let server = ServerOptions::new().create(&self.name)?; + server.connect().await?; + Ok(Box::new(server)) + } + } +} + +pub fn cleanup(path: &Path) -> io::Result<()> { + #[cfg(unix)] + { + if path.exists() { + let _ = std::fs::remove_file(path); + } + } + + #[cfg(windows)] + { + let _ = path; + } + + Ok(()) +} + +#[cfg(windows)] +fn pipe_name_from_path(path: &Path) -> String { + let name = path.to_string_lossy().to_string(); + if name.starts_with(r"\\.\pipe\") { + name + } else { + format!(r"\\.\pipe\{}", name) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + #[cfg(windows)] + #[tokio::test] + async fn test_named_pipe_roundtrip() { + let pipe_path = Path::new(r"\\.\pipe\codi-ipc-test"); + + let listener = bind(pipe_path).await.expect("bind failed"); + + let server_task = tokio::spawn(async move { + let mut stream = listener.accept().await.expect("accept failed"); + let mut buf = [0u8; 5]; + stream.read_exact(&mut buf).await.expect("read failed"); + assert_eq!(&buf, b"hello"); + stream.write_all(b"world").await.expect("write failed"); + stream.flush().await.expect("flush failed"); + }); + + let mut client = connect(pipe_path).await.expect("connect failed"); + client.write_all(b"hello").await.expect("client write failed"); + client.flush().await.expect("client flush failed"); + + let mut buf = [0u8; 5]; + client.read_exact(&mut buf).await.expect("client read failed"); + assert_eq!(&buf, b"world"); + + server_task.await.expect("server task failed"); + } +} diff --git a/codi-rs/src/orchestrate/mod.rs b/codi-rs/src/orchestrate/mod.rs index 7b081b2..62f953e 100644 --- a/codi-rs/src/orchestrate/mod.rs +++ b/codi-rs/src/orchestrate/mod.rs @@ -19,7 +19,7 @@ //! - **WorkspaceIsolator**: Abstraction for creating isolated workspaces, supporting //! both git worktrees (single repo) and griptrees (multi-repo gitgrip). //! -//! - **IPC**: Unix domain socket-based communication between Commander and workers. +//! - **IPC**: Cross-platform IPC (Unix domain sockets on Unix, named pipes on Windows). //! //! # Workspace Isolation //! @@ -78,7 +78,7 @@ //! //! # IPC Protocol //! -//! Communication uses newline-delimited JSON over Unix domain sockets. +//! Communication uses newline-delimited JSON over a platform-specific IPC transport. //! //! ## Worker → Commander Messages //! diff --git a/codi-rs/src/orchestrate/types.rs b/codi-rs/src/orchestrate/types.rs index b86233b..3937e66 100644 --- a/codi-rs/src/orchestrate/types.rs +++ b/codi-rs/src/orchestrate/types.rs @@ -8,6 +8,8 @@ use std::collections::HashSet; use std::path::{Path, PathBuf}; +#[cfg(windows)] +use std::hash::{Hash, Hasher}; use std::time::Instant; use chrono::{DateTime, Utc}; @@ -352,7 +354,7 @@ impl WorkerState { /// Configuration for the commander (orchestrator). #[derive(Debug, Clone)] pub struct CommanderConfig { - /// Path to the IPC socket. + /// Path to the IPC endpoint. pub socket_path: PathBuf, /// Maximum number of concurrent workers. pub max_workers: usize, @@ -369,7 +371,8 @@ pub struct CommanderConfig { impl CommanderConfig { /// Create configuration for a specific project. /// - /// The socket will be created at `/.codi/orchestrator.sock`. + /// The endpoint will be created at `/.codi/orchestrator.sock` on Unix, + /// and a named pipe on Windows. pub fn for_project(project_root: &Path) -> Self { Self { socket_path: socket_path_for_project(project_root), @@ -384,9 +387,8 @@ impl CommanderConfig { impl Default for CommanderConfig { fn default() -> Self { - // Default uses a temp directory for tests Self { - socket_path: PathBuf::from("/tmp/codi-orchestrator.sock"), + socket_path: default_socket_path(), max_workers: 4, base_branch: "main".to_string(), cleanup_on_exit: true, @@ -398,9 +400,33 @@ impl Default for CommanderConfig { /// Get the socket path for a project. /// -/// Returns `/.codi/orchestrator.sock`. +/// Returns `/.codi/orchestrator.sock` on Unix, and a named pipe +/// on Windows. pub fn socket_path_for_project(project_root: &Path) -> PathBuf { - project_root.join(".codi").join("orchestrator.sock") + #[cfg(not(windows))] + { + project_root.join(".codi").join("orchestrator.sock") + } + + #[cfg(windows)] + { + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + project_root.to_string_lossy().hash(&mut hasher); + let hash = hasher.finish(); + PathBuf::from(format!(r"\\.\pipe\codi-orchestrator-{hash:x}")) + } +} + +fn default_socket_path() -> PathBuf { + #[cfg(not(windows))] + { + PathBuf::from("/tmp/codi-orchestrator.sock") + } + + #[cfg(windows)] + { + PathBuf::from(r"\\.\pipe\codi-orchestrator-default") + } } // ============================================================================ @@ -591,17 +617,23 @@ mod tests { #[test] fn test_commander_config_for_project() { let config = CommanderConfig::for_project(Path::new("/workspace/my-project")); + #[cfg(not(windows))] assert_eq!( config.socket_path, PathBuf::from("/workspace/my-project/.codi/orchestrator.sock") ); + #[cfg(windows)] + assert!(config.socket_path.to_string_lossy().starts_with(r"\\.\pipe\codi-orchestrator-")); assert_eq!(config.max_workers, 4); } #[test] fn test_socket_path_for_project() { let path = socket_path_for_project(Path::new("/home/user/project")); + #[cfg(not(windows))] assert_eq!(path, PathBuf::from("/home/user/project/.codi/orchestrator.sock")); + #[cfg(windows)] + assert!(path.to_string_lossy().starts_with(r"\\.\pipe\codi-orchestrator-")); } #[test]