From d12b8bd7ece66005920a7069b97f3ac8cca97df4 Mon Sep 17 00:00:00 2001 From: rajulbhatnagar Date: Thu, 2 Apr 2026 06:06:21 -0700 Subject: [PATCH] feat: Add ACP (Agent Client Protocol) job mode for delegating to any compatible coding agent (#1600) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add ACP (Agent Client Protocol) job mode for delegating to any compatible coding agent Add a third container job mode (`JobMode::Acp`) that spawns any ACP-compliant agent (Goose, Codex, Gemini CLI, Cline, Copilot, etc.) as a subprocess inside a Docker container and communicates via the standard ACP protocol (JSON-RPC over stdio). **Bridge runtime** (`src/worker/acp_bridge.rs`): - Spawns agent subprocess, performs ACP handshake (initialize → session → prompt) - Translates ACP SessionNotification events to IronClaw's JobEventPayload stream - Auto-approves permissions (Docker container is the security boundary) - Supports follow-up prompts from the orchestrator - Detects agent process exit via oneshot channel to prevent infinite polling **User configuration** (mirrors MCP server pattern): - `ironclaw acp add/list/remove/toggle/test` CLI commands - DB-backed persistence with `~/.ironclaw/acp-agents.json` disk fallback - Per-agent `enabled` flag + global `ACP_ENABLED` toggle - `ironclaw acp test` spawns agent, verifies ACP handshake, reports capabilities **System integration**: - `ExtensionKind::AcpAgent` in extension manager (12 match arms) - `agent_name` parameter in CreateJobTool resolves agent from AcpAgentsFile - Mode stored as `"acp:"` for restart support - Doctor validation, status display, boot screen, app startup logging - Web UI: extension install mapping, job restart, follow-up prompt support Closes #1506 * test: add comprehensive ACP test coverage (22 new tests) Bridge: ToolCall, ToolCallUpdate, thought-image, max_turn_requests, session_id propagation, text_from_content_block, multibyte truncation. Config: AcpModeConfig defaults, settings resolution, env overrides. Job tool: schema includes "acp" mode + agent_name, mode="acp" requires agent_name parameter, JobMode::Acp as_str/display. Job manager: JobMode::Acp as_str/display, acp_memory_limit_mb default. CLI: parse_env_var valid/invalid/equals-in-value, command variants. * refactor: make IronClawAcpClient reusable for CLI test command Extract AcpEventSink trait so the same Client implementation (permission auto-approval, event translation) is shared between the container bridge (posts to orchestrator HTTP API) and the CLI test command (prints to stdout). Also extracts ironclaw_init_request() to avoid duplicating the ACP handshake parameters between bridge and test command. * fix(sandbox): use host.docker.internal on all platforms for orchestrator URL The orchestrator host was hardcoded to 172.17.0.1 on Linux, which is only correct for the default Docker bridge network. Environments with custom bridge IPs break container-to-host connectivity. Since all containers already set extra_hosts with host-gateway, using host.docker.internal works on all platforms and network configurations. * fix ACP PR review feedback * fix ACP DB error fallback * fix clippy after staging merge --------- Co-authored-by: Rajul Bhatnagar Co-authored-by: Firat Sertgoz --- .env.example | 6 + Cargo.lock | 56 ++ Cargo.toml | 4 + src/app.rs | 25 + src/boot_screen.rs | 9 + src/channels/web/handlers/extensions.rs | 1 + src/channels/web/handlers/jobs.rs | 168 ++++- src/channels/web/server.rs | 1 + src/cli/acp.rs | 354 ++++++++++ src/cli/doctor.rs | 60 ++ src/cli/mod.rs | 23 + ...li__tests__help_output_without_import.snap | 2 + ...ests__long_help_output_without_import.snap | 2 + src/cli/status.rs | 26 + src/config/acp.rs | 633 ++++++++++++++++++ src/config/mod.rs | 6 +- src/config/sandbox.rs | 96 +++ src/extensions/manager.rs | 56 +- src/extensions/mod.rs | 3 + src/main.rs | 12 + src/orchestrator/job_manager.rs | 125 +++- src/orchestrator/mod.rs | 5 + src/settings.rs | 5 + src/tools/builtin/job.rs | 120 +++- src/worker/acp_bridge.rs | 629 +++++++++++++++++ src/worker/mod.rs | 77 +++ 26 files changed, 2461 insertions(+), 43 deletions(-) create mode 100644 src/cli/acp.rs create mode 100644 src/config/acp.rs create mode 100644 src/worker/acp_bridge.rs diff --git a/.env.example b/.env.example index 2d78f4b64d..2cb69c7308 100644 --- a/.env.example +++ b/.env.example @@ -208,6 +208,12 @@ HEARTBEAT_NOTIFY_USER=default # SANDBOX_TIMEOUT_SECS=120 # SANDBOX_MEMORY_LIMIT_MB=2048 +# ACP (Agent Client Protocol) agents +# ACP_ENABLED=false # Enable ACP agent sandbox mode +# ACP_MEMORY_LIMIT_MB=4096 # Memory limit for ACP containers +# ACP_TIMEOUT_SECS=1800 # Maximum session timeout +# Configure agents via CLI: ironclaw acp add goose --command goose --arg "--stdio" + # Safety settings SAFETY_MAX_OUTPUT_LENGTH=100000 SAFETY_INJECTION_CHECK_ENABLED=true diff --git a/Cargo.lock b/Cargo.lock index 773ecd89a8..ef9062370f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,6 +61,37 @@ dependencies = [ "subtle", ] +[[package]] +name = "agent-client-protocol" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9c56a59cf6315e99f874d2c1f96c69d2da5ffe0087d211297fc4a41f849770a2" +dependencies = [ + "agent-client-protocol-schema", + "anyhow", + "async-broadcast", + "async-trait", + "derive_more", + "futures", + "log", + "serde", + "serde_json", +] + +[[package]] +name = "agent-client-protocol-schema" +version = "0.11.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e0497b9a95a404e35799904835c57c6f8c69b9d08ccfd3cb5b7d746425cd6789" +dependencies = [ + "anyhow", + "derive_more", + "schemars 1.2.1", + "serde", + "serde_json", + "strum", +] + [[package]] name = "ahash" version = "0.7.8" @@ -2068,6 +2099,7 @@ dependencies = [ "quote", "rustc_version", "syn 2.0.117", + "unicode-xid", ] [[package]] @@ -3402,6 +3434,7 @@ name = "ironclaw" version = "0.24.0" dependencies = [ "aes-gcm", + "agent-client-protocol", "aho-corasick", "anyhow", "async-trait", @@ -3481,6 +3514,7 @@ dependencies = [ "tokio-stream", "tokio-test", "tokio-tungstenite 0.26.2", + "tokio-util", "toml", "tower 0.5.3", "tower-http 0.6.8", @@ -6337,6 +6371,27 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "strum" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9628de9b8791db39ceda2b119bbe13134770b56c138ec1d3af810d045c04f9bd" +dependencies = [ + "strum_macros", +] + +[[package]] +name = "strum_macros" +version = "0.28.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ab85eea0270ee17587ed4156089e10b9e6880ee688791d45a905f5b1ca36f664" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "subtle" version = "2.6.1" @@ -6878,6 +6933,7 @@ checksum = "9ae9cec805b01e8fc3fd2fe289f89149a9b66dd16786abd8b19cfa7b48cb0098" dependencies = [ "bytes", "futures-core", + "futures-io", "futures-sink", "pin-project-lite", "tokio", diff --git a/Cargo.toml b/Cargo.toml index f744392405..ec6a0e9af6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,10 +39,14 @@ eula = false # Async runtime tokio = { version = "1", features = ["full"] } tokio-stream = { version = "0.1", features = ["sync"] } +tokio-util = { version = "0.7", features = ["compat"] } futures = "0.3" tokio-tungstenite = { version = "0.26", features = ["rustls-tls-native-roots"] } eventsource-stream = "0.2" +# Agent Client Protocol (ACP) — standard communication with coding agents +agent-client-protocol = "0.10" + # HTTP client reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls-native-roots", "stream"] } diff --git a/src/app.rs b/src/app.rs index d737795f59..03e3011c7a 100644 --- a/src/app.rs +++ b/src/app.rs @@ -751,6 +751,31 @@ impl AppBuilder { Some(manager) }; + // Validate ACP agent configs at startup (lightweight — no connections, just config check). + { + let acp_agents = if let Some(ref d) = self.db { + crate::config::acp::load_acp_agents_from_db(d.as_ref(), &self.config.owner_id).await + } else { + crate::config::acp::load_acp_agents().await + }; + match acp_agents { + Ok(file) => { + let enabled: Vec<_> = file.enabled_agents().collect(); + if !enabled.is_empty() { + let names: Vec<&str> = enabled.iter().map(|a| a.name.as_str()).collect(); + tracing::info!( + "ACP agents configured: {} ({} enabled)", + names.join(", "), + enabled.len() + ); + } + } + Err(e) => { + tracing::debug!("No ACP agents configured ({})", e); + } + } + } + // register_builder_tool() already calls register_dev_tools() internally, // so only register them here when the builder didn't already do it. let builder_registered_dev_tools = self.config.builder.enabled diff --git a/src/boot_screen.rs b/src/boot_screen.rs index c018abf633..99b8fd813c 100644 --- a/src/boot_screen.rs +++ b/src/boot_screen.rs @@ -25,6 +25,7 @@ pub struct BootInfo { pub sandbox_enabled: bool, pub docker_status: crate::sandbox::detect::DockerStatus, pub claude_code_enabled: bool, + pub acp_enabled: bool, pub routines_enabled: bool, pub skills_enabled: bool, pub channels: Vec, @@ -208,6 +209,11 @@ pub fn print_boot_screen(info: &BootInfo) { tags.push("claude-code".to_string()); } + // ACP agents + if info.acp_enabled { + tags.push("acp".to_string()); + } + if !tags.is_empty() { println!( " {}{: Some(crate::extensions::ExtensionKind::WasmTool), "wasm_channel" => Some(crate::extensions::ExtensionKind::WasmChannel), "channel_relay" => Some(crate::extensions::ExtensionKind::ChannelRelay), + "acp_agent" => Some(crate::extensions::ExtensionKind::AcpAgent), _ => None, }); diff --git a/src/channels/web/handlers/jobs.rs b/src/channels/web/handlers/jobs.rs index eb54c9d5ac..c1aead8019 100644 --- a/src/channels/web/handlers/jobs.rs +++ b/src/channels/web/handlers/jobs.rs @@ -23,6 +23,37 @@ fn db_error(context: &str, e: impl std::fmt::Display) -> (StatusCode, String) { ) } +async fn resolve_sandbox_restart_mode( + store: &dyn crate::db::Database, + stored_mode: &str, + user_id: &str, +) -> Result< + ( + crate::orchestrator::job_manager::JobMode, + Option, + ), + crate::config::acp::AcpConfigError, +> { + if stored_mode == "claude_code" { + return Ok((crate::orchestrator::job_manager::JobMode::ClaudeCode, None)); + } + + if let Some(agent_name) = stored_mode.strip_prefix("acp:") { + let agent = + crate::config::acp::get_enabled_acp_agent_for_user(Some(store), user_id, agent_name) + .await?; + return Ok((crate::orchestrator::job_manager::JobMode::Acp, Some(agent))); + } + + if stored_mode == "acp" { + return Err(crate::config::acp::AcpConfigError::InvalidConfig { + reason: "legacy ACP jobs without an agent name cannot be restarted".to_string(), + }); + } + + Ok((crate::orchestrator::job_manager::JobMode::Worker, None)) +} + pub async fn jobs_list_handler( State(state): State>, AuthenticatedUser(user): AuthenticatedUser, @@ -198,7 +229,9 @@ pub async fn jobs_detail_handler( } let mode = store.get_sandbox_job_mode(job.id).await.ok().flatten(); - let is_claude_code = mode.as_deref() == Some("claude_code"); + let supports_prompts = mode + .as_deref() + .is_some_and(|m| m == "claude_code" || m.starts_with("acp")); return Ok(Json(JobDetailResponse { id: job.id, @@ -215,7 +248,7 @@ pub async fn jobs_detail_handler( job_mode: mode.filter(|m| m != "worker"), transitions, can_restart: state.job_manager.is_some(), - can_prompt: is_claude_code && state.prompt_queue.is_some(), + can_prompt: supports_prompts && state.prompt_queue.is_some(), job_kind: Some("sandbox".to_string()), })); } @@ -413,6 +446,16 @@ pub async fn jobs_restart_handler( let new_job_id = Uuid::new_v4(); let now = chrono::Utc::now(); + let stored_mode = store + .get_sandbox_job_mode(old_job_id) + .await + .map_err(|e| db_error("jobs_restart_handler", e))? + .unwrap_or_default(); + + let (mode, acp_agent) = + resolve_sandbox_restart_mode(store.as_ref(), &stored_mode, &old_job.user_id) + .await + .map_err(|e| (StatusCode::CONFLICT, format!("Cannot restart job: {}", e)))?; let record = crate::history::SandboxJobRecord { id: new_job_id, task: task.clone(), @@ -429,14 +472,25 @@ pub async fn jobs_restart_handler( store .save_sandbox_job(&record) .await - .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + .map_err(|e| db_error("jobs_restart_handler", e))?; - let mode = match store.get_sandbox_job_mode(old_job_id).await { - Ok(Some(m)) if m == "claude_code" => { - crate::orchestrator::job_manager::JobMode::ClaudeCode - } - _ => crate::orchestrator::job_manager::JobMode::Worker, - }; + if mode != crate::orchestrator::job_manager::JobMode::Worker { + let mode_str = if mode == crate::orchestrator::job_manager::JobMode::Acp { + format!( + "acp:{}", + acp_agent + .as_ref() + .map(|agent| agent.name.as_str()) + .unwrap_or_default() + ) + } else { + mode.as_str().to_string() + }; + store + .update_sandbox_job_mode(new_job_id, &mode_str) + .await + .map_err(|e| db_error("jobs_restart_handler", e))?; + } let credential_grants: Vec = serde_json::from_str(&old_job.credential_grants_json).unwrap_or_else(|e| { @@ -450,7 +504,7 @@ pub async fn jobs_restart_handler( }); let project_dir = std::path::PathBuf::from(&old_job.project_dir); - let _token = jm + let create_result = jm .create_job( new_job_id, &task, @@ -458,21 +512,36 @@ pub async fn jobs_restart_handler( mode, crate::orchestrator::job_manager::JobCreationParams { credential_grants, + acp_agent, ..Default::default() }, ) - .await - .map_err(|e| { - ( + .await; + let _token = match create_result { + Ok(token) => token, + Err(e) => { + let error_text = e.to_string(); + let _ = store + .update_sandbox_job_status( + new_job_id, + "failed", + Some(false), + Some(error_text.as_str()), + None, + Some(chrono::Utc::now()), + ) + .await; + return Err(( StatusCode::INTERNAL_SERVER_ERROR, - format!("Failed to create container: {}", e), - ) - })?; + format!("Failed to create container: {}", error_text), + )); + } + }; store .update_sandbox_job_status(new_job_id, "running", None, None, Some(now), None) .await - .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?; + .map_err(|e| db_error("jobs_restart_handler", e))?; return Ok(Json(serde_json::json!({ "status": "restarted", @@ -482,7 +551,7 @@ pub async fn jobs_restart_handler( } Ok(None) => {} Err(e) => { - return Err(db_error("jobs_handler", e)); + return Err(db_error("jobs_restart_handler", e)); } } @@ -578,12 +647,15 @@ pub async fn jobs_prompt_handler( return Err((StatusCode::NOT_FOUND, "Job not found".to_string())); } - // It's a sandbox job. Check if Claude Code mode. + // It's a sandbox job. Check if Claude Code or ACP mode (both support follow-up prompts). let mode = s.get_sandbox_job_mode(job_id).await.ok().flatten(); - if mode.as_deref() == Some("claude_code") { + if mode + .as_deref() + .is_some_and(|m| m == "claude_code" || m.starts_with("acp")) + { let prompt_queue = state.prompt_queue.as_ref().ok_or(( StatusCode::NOT_IMPLEMENTED, - "Claude Code not configured".to_string(), + "Follow-up prompts are not configured".to_string(), ))?; let prompt = crate::orchestrator::api::PendingPrompt { content, done }; { @@ -830,6 +902,60 @@ pub async fn job_files_read_handler( mod tests { use super::*; + #[cfg(feature = "libsql")] + #[tokio::test] + async fn sandbox_restart_mode_uses_original_job_owner_scope() { + let (db, _tmp) = crate::testing::test_db().await; + + let mut agents = crate::config::acp::AcpAgentsFile::default(); + agents.upsert(crate::config::acp::AcpAgentConfig::new( + "codex", + "codex", + vec!["acp".into()], + std::collections::HashMap::new(), + )); + crate::config::acp::save_acp_agents_for_user(Some(db.as_ref()), "owner-123", &agents) + .await + .unwrap(); + + let (mode, agent) = resolve_sandbox_restart_mode(db.as_ref(), "acp:codex", "owner-123") + .await + .unwrap(); + + assert_eq!(mode, crate::orchestrator::job_manager::JobMode::Acp); + assert_eq!( + agent.as_ref().map(|agent| agent.name.as_str()), + Some("codex") + ); + } + + #[cfg(feature = "libsql")] + #[tokio::test] + async fn sandbox_restart_mode_rejects_disabled_acp_agent() { + let (db, _tmp) = crate::testing::test_db().await; + + let mut agents = crate::config::acp::AcpAgentsFile::default(); + let mut agent = crate::config::acp::AcpAgentConfig::new( + "codex", + "codex", + vec!["acp".into()], + std::collections::HashMap::new(), + ); + agent.enabled = false; + agents.upsert(agent); + crate::config::acp::save_acp_agents_for_user(Some(db.as_ref()), "owner-123", &agents) + .await + .unwrap(); + + let err = resolve_sandbox_restart_mode(db.as_ref(), "acp:codex", "owner-123") + .await + .unwrap_err(); + assert!(matches!( + err, + crate::config::acp::AcpConfigError::AgentDisabled { .. } + )); + } + #[test] fn test_db_error_does_not_leak_details() { let (status, body) = db_error("test_context", "relation \"jobs\" does not exist"); diff --git a/src/channels/web/server.rs b/src/channels/web/server.rs index a8c6e4d3b1..dc23787496 100644 --- a/src/channels/web/server.rs +++ b/src/channels/web/server.rs @@ -2290,6 +2290,7 @@ async fn extensions_install_handler( "mcp_server" => Some(crate::extensions::ExtensionKind::McpServer), "wasm_tool" => Some(crate::extensions::ExtensionKind::WasmTool), "wasm_channel" => Some(crate::extensions::ExtensionKind::WasmChannel), + "acp_agent" => Some(crate::extensions::ExtensionKind::AcpAgent), _ => None, }); diff --git a/src/cli/acp.rs b/src/cli/acp.rs new file mode 100644 index 0000000000..96b258c728 --- /dev/null +++ b/src/cli/acp.rs @@ -0,0 +1,354 @@ +//! ACP agent management CLI commands. +//! +//! Commands for adding, removing, listing, toggling, and testing ACP agents. +//! Mirrors the MCP server management CLI (`src/cli/mcp.rs`). + +use std::collections::HashMap; +use std::sync::Arc; + +use clap::{Args, Subcommand}; + +use crate::config::acp::{self, AcpAgentConfig, AcpAgentsFile}; +use crate::db::Database; + +/// Arguments for the `acp add` subcommand. +#[derive(Args, Debug, Clone)] +pub struct AcpAddArgs { + /// Agent name (e.g., "goose", "codex", "gemini") + pub name: String, + + /// Command to spawn the agent + #[arg(long)] + pub command: String, + + /// Command arguments (can be repeated) + #[arg(long = "arg", num_args = 1..)] + pub args: Vec, + + /// Environment variables (KEY=VALUE format, can be repeated) + #[arg(long = "env", value_parser = parse_env_var)] + pub env: Vec<(String, String)>, + + /// Agent description + #[arg(long)] + pub description: Option, +} + +fn parse_env_var(s: &str) -> Result<(String, String), String> { + let parts: Vec<&str> = s.splitn(2, '=').collect(); + if parts.len() != 2 { + return Err(format!("invalid env var format '{s}', expected KEY=VALUE")); + } + Ok((parts[0].to_string(), parts[1].to_string())) +} + +#[derive(Subcommand, Debug, Clone)] +pub enum AcpCommand { + /// Add an ACP agent + Add(Box), + + /// Remove an ACP agent + Remove { + /// Agent name to remove + name: String, + }, + + /// List configured ACP agents + List, + + /// Enable or disable an ACP agent + Toggle { + /// Agent name to toggle + name: String, + }, + + /// Test an ACP agent connection (spawn, handshake, report) + Test { + /// Agent name to test + name: String, + }, +} + +/// Run an ACP CLI command. +pub async fn run_acp_command(cmd: AcpCommand) -> anyhow::Result<()> { + match cmd { + AcpCommand::Add(args) => add_agent(*args).await, + AcpCommand::Remove { name } => remove_agent(&name).await, + AcpCommand::List => list_agents().await, + AcpCommand::Toggle { name } => toggle_agent(&name).await, + AcpCommand::Test { name } => test_agent(&name).await, + } +} + +async fn add_agent(args: AcpAddArgs) -> anyhow::Result<()> { + let env: HashMap = args.env.into_iter().collect(); + let mut config = AcpAgentConfig::new(&args.name, &args.command, args.args, env); + if let Some(desc) = args.description { + config = config.with_description(desc); + } + + config.validate().map_err(|e| anyhow::anyhow!("{}", e))?; + + let storage = resolve_storage().await; + let mut agents = load_agents(storage.db.as_deref(), &storage.owner_id).await?; + let is_update = agents.get(&args.name).is_some(); + agents.upsert(config); + save_agents(storage.db.as_deref(), &storage.owner_id, &agents).await?; + + if is_update { + println!("Updated ACP agent '{}'", args.name); + } else { + println!("Added ACP agent '{}'", args.name); + } + Ok(()) +} + +async fn remove_agent(name: &str) -> anyhow::Result<()> { + let storage = resolve_storage().await; + let mut agents = load_agents(storage.db.as_deref(), &storage.owner_id).await?; + + if !agents.remove(name) { + anyhow::bail!("ACP agent '{}' not found", name); + } + + save_agents(storage.db.as_deref(), &storage.owner_id, &agents).await?; + println!("Removed ACP agent '{}'", name); + Ok(()) +} + +async fn list_agents() -> anyhow::Result<()> { + let storage = resolve_storage().await; + let agents = load_agents(storage.db.as_deref(), &storage.owner_id).await?; + + if agents.agents.is_empty() { + println!("No ACP agents configured."); + println!(); + println!("Add one with:"); + println!(" ironclaw acp add goose --command goose --arg \"--stdio\""); + return Ok(()); + } + + println!("ACP Agents:"); + println!(); + for agent in &agents.agents { + let icon = if agent.enabled { "●" } else { "○" }; + let status = if agent.enabled { "enabled" } else { "disabled" }; + println!(" {} {} ({}) [{}]", icon, agent.name, agent.command, status); + if !agent.args.is_empty() { + println!(" args: {}", agent.args.join(" ")); + } + if !agent.env.is_empty() { + println!(" env: {} variable(s)", agent.env.len()); + } + if let Some(ref desc) = agent.description { + println!(" {}", desc); + } + } + Ok(()) +} + +async fn toggle_agent(name: &str) -> anyhow::Result<()> { + let storage = resolve_storage().await; + let mut agents = load_agents(storage.db.as_deref(), &storage.owner_id).await?; + + let agent = agents + .get_mut(name) + .ok_or_else(|| anyhow::anyhow!("ACP agent '{}' not found", name))?; + + agent.enabled = !agent.enabled; + let new_state = if agent.enabled { "enabled" } else { "disabled" }; + + save_agents(storage.db.as_deref(), &storage.owner_id, &agents).await?; + println!("ACP agent '{}' is now {}", name, new_state); + Ok(()) +} + +async fn test_agent(name: &str) -> anyhow::Result<()> { + use agent_client_protocol::{self as acp, Agent as _}; + use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; + + use crate::worker::acp_bridge; + use crate::worker::api::JobEventPayload; + + /// Event sink that prints agent output to stdout during `ironclaw acp test`. + struct PrintEventSink; + + impl acp_bridge::AcpEventSink for PrintEventSink { + async fn emit_event(&self, payload: &JobEventPayload) { + match payload.event_type.as_str() { + "message" => { + if let Some(content) = payload.data["content"].as_str() { + println!(" | {}", content); + } + } + "tool_use" => { + if let Some(tool) = payload.data["tool_name"].as_str() { + println!(" [tool: {}]", tool); + } + } + _ => {} + } + } + } + + let storage = resolve_storage().await; + let agent = crate::config::acp::get_enabled_acp_agent_for_user( + storage.db.as_deref(), + &storage.owner_id, + name, + ) + .await + .map_err(|e| anyhow::anyhow!("{}", e))?; + + println!(); + println!(" Testing ACP agent '{}'...", name); + println!(" Command: {} {}", agent.command, agent.args.join(" ")); + + // Spawn the agent subprocess + let mut child = tokio::process::Command::new(&agent.command) + .args(&agent.args) + .envs(&agent.env) + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::null()) + .kill_on_drop(true) + .spawn() + .map_err(|e| anyhow::anyhow!("Failed to spawn '{}': {}", agent.command, e))?; + + let child_stdin = child + .stdin + .take() + .ok_or_else(|| anyhow::anyhow!("failed to capture agent stdin"))?; + let child_stdout = child + .stdout + .take() + .ok_or_else(|| anyhow::anyhow!("failed to capture agent stdout"))?; + + // Run ACP handshake inside a LocalSet (!Send futures). + // Uses IronClawAcpClient with a PrintEventSink so the test exercises + // the same permission auto-approval and event translation as real jobs. + let local_set = tokio::task::LocalSet::new(); + let result = local_set + .run_until(async move { + let outgoing = child_stdin.compat_write(); + let incoming = child_stdout.compat(); + + let client = acp_bridge::IronClawAcpClient::new(PrintEventSink); + + let (conn, handle_io) = + acp::ClientSideConnection::new(client, outgoing, incoming, |fut| { + tokio::task::spawn_local(fut); + }); + tokio::task::spawn_local(handle_io); + + let handshake = tokio::time::timeout( + std::time::Duration::from_secs(15), + conn.initialize(acp_bridge::ironclaw_init_request()), + ) + .await; + + match handshake { + Ok(Ok(resp)) => { + println!(" \u{2713} ACP handshake successful!"); + println!(); + println!(" Agent info:"); + if let Some(ref info) = resp.agent_info { + println!(" Name: {}", info.name); + println!(" Version: {}", info.version); + } + println!(" Protocol: {}", resp.protocol_version); + Ok(()) + } + Ok(Err(e)) => { + println!(" \u{2717} ACP handshake failed: {}", e); + Err(anyhow::anyhow!("handshake failed: {}", e)) + } + Err(_) => { + println!(" \u{2717} ACP handshake timed out (15s)"); + Err(anyhow::anyhow!("handshake timed out")) + } + } + }) + .await; + + // Clean up child process + let _ = child.kill().await; + println!(); + result +} + +// ==================== DB / disk persistence helpers ==================== + +struct AcpCliStorage { + db: Option>, + owner_id: String, +} + +async fn resolve_storage() -> AcpCliStorage { + match crate::config::Config::from_env().await { + Ok(config) => AcpCliStorage { + db: crate::db::connect_from_config(&config.database) + .await + .ok() + .map(|db| db as Arc), + owner_id: config.owner_id, + }, + Err(_) => AcpCliStorage { + db: None, + owner_id: "default".to_string(), + }, + } +} + +async fn load_agents( + db: Option<&dyn Database>, + owner_id: &str, +) -> Result { + acp::load_acp_agents_for_user(db, owner_id) + .await + .map_err(|e| anyhow::anyhow!("{}", e)) +} + +async fn save_agents( + db: Option<&dyn Database>, + owner_id: &str, + agents: &AcpAgentsFile, +) -> Result<(), anyhow::Error> { + acp::save_acp_agents_for_user(db, owner_id, agents) + .await + .map_err(|e| anyhow::anyhow!("{}", e)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_env_var_valid() { + let (k, v) = parse_env_var("FOO=bar").unwrap(); + assert_eq!(k, "FOO"); + assert_eq!(v, "bar"); + } + + #[test] + fn test_parse_env_var_with_equals_in_value() { + let (k, v) = parse_env_var("KEY=val=ue").unwrap(); + assert_eq!(k, "KEY"); + assert_eq!(v, "val=ue"); + } + + #[test] + fn test_parse_env_var_invalid() { + let result = parse_env_var("no-equals-sign"); + assert!(result.is_err()); + } + + #[test] + fn test_acp_command_variants() { + // Verify all variants exist (compile-time check) + let _ = AcpCommand::List; + let _ = AcpCommand::Remove { name: "x".into() }; + let _ = AcpCommand::Toggle { name: "x".into() }; + let _ = AcpCommand::Test { name: "x".into() }; + } +} diff --git a/src/cli/doctor.rs b/src/cli/doctor.rs index dca408d73f..cb8fc3f532 100644 --- a/src/cli/doctor.rs +++ b/src/cli/doctor.rs @@ -10,6 +10,21 @@ use crate::bootstrap::ironclaw_base_dir; use crate::cli::fmt; use crate::settings::Settings; +async fn load_acp_agents_for_doctor() +-> Result { + match crate::config::Config::from_env().await { + Ok(config) => { + let db: Option> = + crate::db::connect_from_config(&config.database) + .await + .ok() + .map(|db| db as std::sync::Arc); + crate::config::acp::load_acp_agents_for_user(db.as_deref(), &config.owner_id).await + } + Err(_) => crate::config::acp::load_acp_agents().await, + } +} + /// Run all diagnostic checks and print results. pub async fn run_doctor_command() -> anyhow::Result<()> { println!(); @@ -102,6 +117,14 @@ pub async fn run_doctor_command() -> anyhow::Result<()> { &mut skipped, ); + check( + "ACP agents", + check_acp_config().await, + &mut passed, + &mut failed, + &mut skipped, + ); + check( "Skills", check_skills().await, @@ -520,6 +543,43 @@ async fn check_mcp_config() -> CheckResult { } } +async fn check_acp_config() -> CheckResult { + match load_acp_agents_for_doctor().await { + Ok(file) => { + let agents: Vec<_> = file.enabled_agents().collect(); + if agents.is_empty() { + return CheckResult::Skip("no ACP agents configured".into()); + } + + let mut invalid = Vec::new(); + for agent in &agents { + if let Err(e) = agent.validate() { + invalid.push(format!("{}: {}", agent.name, e)); + } + } + + if invalid.is_empty() { + CheckResult::Pass(format!("{} agent(s) configured, all valid", agents.len())) + } else { + CheckResult::Fail(format!( + "{} agent(s), {} invalid: {}", + agents.len(), + invalid.len(), + invalid.join("; ") + )) + } + } + Err(e) => { + let msg = e.to_string(); + if msg.contains("not found") || msg.contains("No such file") { + CheckResult::Skip("no ACP config file".into()) + } else { + CheckResult::Fail(format!("config error: {e}")) + } + } + } +} + // ── Skills ────────────────────────────────────────────────── async fn check_skills() -> CheckResult { diff --git a/src/cli/mod.rs b/src/cli/mod.rs index ebc97ac5b8..7fdaeb1951 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -14,6 +14,7 @@ //! - Viewing gateway logs (`logs`) //! - Checking system health (`status`) +pub mod acp; mod channels; mod completion; mod config; @@ -35,6 +36,7 @@ mod skills; pub mod status; mod tool; +pub use acp::{AcpCommand, run_acp_command}; pub use channels::{ChannelsCommand, run_channels_command}; pub use completion::Completion; pub use config::{ConfigCommand, run_config_command}; @@ -292,6 +294,14 @@ pub enum Command { max_iterations: u32, }, + /// Manage ACP (Agent Client Protocol) agents + #[command( + subcommand, + about = "Manage ACP agents", + long_about = "Add, list, remove, or test ACP-compliant coding agents.\nExample: ironclaw acp add goose --command goose --arg \"--stdio\"" + )] + Acp(AcpCommand), + /// Run as a Claude Code bridge inside a Docker container (internal use). /// Spawns the `claude` CLI and streams output back to the orchestrator. #[command(hide = true)] @@ -312,6 +322,19 @@ pub enum Command { #[arg(long, default_value = "sonnet")] model: String, }, + + /// Run as an ACP bridge inside a Docker container (internal use). + /// Spawns an ACP-compliant agent and streams output back to the orchestrator. + #[command(hide = true)] + AcpBridge { + /// Job ID to execute. + #[arg(long)] + job_id: uuid::Uuid, + + /// URL of the orchestrator's internal API. + #[arg(long, default_value = "http://host.docker.internal:50051")] + orchestrator_url: String, + }, } impl Cli { diff --git a/src/cli/snapshots/ironclaw__cli__tests__help_output_without_import.snap b/src/cli/snapshots/ironclaw__cli__tests__help_output_without_import.snap index 8fcec25eaa..e1d546276c 100644 --- a/src/cli/snapshots/ironclaw__cli__tests__help_output_without_import.snap +++ b/src/cli/snapshots/ironclaw__cli__tests__help_output_without_import.snap @@ -1,5 +1,6 @@ --- source: src/cli/mod.rs +assertion_line: 422 expression: help --- Secure personal AI assistant that protects your data and expands its capabilities @@ -26,6 +27,7 @@ Commands: status Show system status completion Generate completions login Authenticate with a provider + acp Manage ACP agents help Print this message or the help of the given subcommand(s) Options: diff --git a/src/cli/snapshots/ironclaw__cli__tests__long_help_output_without_import.snap b/src/cli/snapshots/ironclaw__cli__tests__long_help_output_without_import.snap index cb799ce74f..c4380f37ac 100644 --- a/src/cli/snapshots/ironclaw__cli__tests__long_help_output_without_import.snap +++ b/src/cli/snapshots/ironclaw__cli__tests__long_help_output_without_import.snap @@ -1,5 +1,6 @@ --- source: src/cli/mod.rs +assertion_line: 438 expression: help --- IronClaw is a secure AI assistant. Use 'ironclaw --help' for details. @@ -29,6 +30,7 @@ Commands: status Show system status completion Generate completions login Authenticate with a provider + acp Manage ACP agents help Print this message or the help of the given subcommand(s) Options: diff --git a/src/cli/status.rs b/src/cli/status.rs index 3ae825eefa..b8fd762832 100644 --- a/src/cli/status.rs +++ b/src/cli/status.rs @@ -35,6 +35,21 @@ fn load_settings_from(json_path: &std::path::Path, toml_path: &std::path::Path) settings } +async fn load_acp_agents_for_status() +-> Result { + match crate::config::Config::from_env().await { + Ok(config) => { + let db: Option> = + crate::db::connect_from_config(&config.database) + .await + .ok() + .map(|db| db as std::sync::Arc); + crate::config::acp::load_acp_agents_for_user(db.as_deref(), &config.owner_id).await + } + Err(_) => crate::config::acp::load_acp_agents().await, + } +} + /// Run the status command, printing system health info. pub async fn run_status_command() -> anyhow::Result<()> { let settings = load_settings(); @@ -182,6 +197,17 @@ pub async fn run_status_command() -> anyhow::Result<()> { }; println!("{}", fmt::kv_line("MCP Servers", &mcp_value, 12)); + // ACP agents + let acp_value = match load_acp_agents_for_status().await { + Ok(agents) => { + let enabled = agents.agents.iter().filter(|a| a.enabled).count(); + let total = agents.agents.len(); + format!("{} enabled / {} configured", enabled, total) + } + Err(_) => "none configured".to_string(), + }; + println!("{}", fmt::kv_line("ACP Agents", &acp_value, 12)); + // Config path println!(); println!( diff --git a/src/config/acp.rs b/src/config/acp.rs new file mode 100644 index 0000000000..9fd6d60bc8 --- /dev/null +++ b/src/config/acp.rs @@ -0,0 +1,633 @@ +//! ACP (Agent Client Protocol) agent configuration. +//! +//! Stores configuration for ACP-compliant coding agents that can be spawned +//! as subprocesses inside Docker containers. Configuration is persisted at +//! `~/.ironclaw/acp-agents.json` (disk fallback) and in the database settings +//! table under key `"acp_agents"`. +//! +//! Mirrors the MCP server config pattern (`src/tools/mcp/config.rs`). + +use std::collections::HashMap; +use std::path::{Path, PathBuf}; + +use serde::{Deserialize, Serialize}; +use tokio::fs; + +use crate::bootstrap::ironclaw_base_dir; + +/// Configuration for a single ACP-compliant agent. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AcpAgentConfig { + /// Unique name for this agent (e.g., "goose", "codex", "gemini"). + pub name: String, + + /// Command to spawn the agent subprocess. + pub command: String, + + /// Arguments to pass to the command. + #[serde(default)] + pub args: Vec, + + /// Additional environment variables for the agent process. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub env: HashMap, + + /// Whether this agent is enabled. + #[serde(default = "default_true")] + pub enabled: bool, + + /// Optional description for the agent. + #[serde(skip_serializing_if = "Option::is_none")] + pub description: Option, +} + +fn default_true() -> bool { + true +} + +impl AcpAgentConfig { + /// Create a new ACP agent configuration. + pub fn new( + name: impl Into, + command: impl Into, + args: Vec, + env: HashMap, + ) -> Self { + Self { + name: name.into(), + command: command.into(), + args, + env, + enabled: true, + description: None, + } + } + + /// Builder: attach a description. + pub fn with_description(mut self, desc: impl Into) -> Self { + self.description = Some(desc.into()); + self + } + + /// Validate the agent configuration. + pub fn validate(&self) -> Result<(), AcpConfigError> { + if self.name.is_empty() { + return Err(AcpConfigError::InvalidConfig { + reason: "agent name cannot be empty".to_string(), + }); + } + if self.command.is_empty() { + return Err(AcpConfigError::InvalidConfig { + reason: format!("agent '{}': command cannot be empty", self.name), + }); + } + Ok(()) + } +} + +/// Container for all configured ACP agents. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AcpAgentsFile { + /// List of configured ACP agents. + #[serde(default)] + pub agents: Vec, + + /// Schema version for future compatibility. + #[serde(default = "default_schema_version")] + pub schema_version: u32, +} + +fn default_schema_version() -> u32 { + 1 +} + +impl Default for AcpAgentsFile { + fn default() -> Self { + Self { + agents: Vec::new(), + schema_version: default_schema_version(), + } + } +} + +impl AcpAgentsFile { + /// Get an agent by name. + pub fn get(&self, name: &str) -> Option<&AcpAgentConfig> { + self.agents.iter().find(|a| a.name == name) + } + + /// Get a mutable agent by name. + pub fn get_mut(&mut self, name: &str) -> Option<&mut AcpAgentConfig> { + self.agents.iter_mut().find(|a| a.name == name) + } + + /// Add or update an agent configuration. + pub fn upsert(&mut self, config: AcpAgentConfig) { + if let Some(existing) = self.get_mut(&config.name) { + *existing = config; + } else { + self.agents.push(config); + } + } + + /// Remove an agent by name. + pub fn remove(&mut self, name: &str) -> bool { + let len_before = self.agents.len(); + self.agents.retain(|a| a.name != name); + self.agents.len() < len_before + } + + /// Get all enabled agents. + pub fn enabled_agents(&self) -> impl Iterator { + self.agents.iter().filter(|a| a.enabled) + } +} + +/// Error type for ACP configuration operations. +#[derive(Debug, thiserror::Error)] +pub enum AcpConfigError { + #[error("IO error: {0}")] + Io(#[from] std::io::Error), + + #[error("JSON error: {0}")] + Json(#[from] serde_json::Error), + + #[error("Database error: {0}")] + Database(String), + + #[error("Invalid configuration: {reason}")] + InvalidConfig { reason: String }, + + #[error("Agent not found: {name}")] + AgentNotFound { name: String }, + + #[error("Agent is disabled: {name}")] + AgentDisabled { name: String }, +} + +// ==================== Disk persistence ==================== + +/// Get the default ACP agents configuration path. +pub fn default_config_path() -> PathBuf { + ironclaw_base_dir().join("acp-agents.json") +} + +/// Load ACP agent configurations from the default location. +pub async fn load_acp_agents() -> Result { + load_acp_agents_from(default_config_path()).await +} + +/// Load ACP agent configurations from a specific path. +pub async fn load_acp_agents_from(path: impl AsRef) -> Result { + let path = path.as_ref(); + + let content = match fs::read_to_string(path).await { + Ok(c) => c, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Ok(AcpAgentsFile::default()); + } + Err(e) => return Err(e.into()), + }; + let config: AcpAgentsFile = serde_json::from_str(&content)?; + + for agent in &config.agents { + agent + .validate() + .map_err(|e| AcpConfigError::InvalidConfig { + reason: format!("Agent '{}': {}", agent.name, e), + })?; + } + + Ok(config) +} + +/// Save ACP agent configurations to the default location. +pub async fn save_acp_agents(config: &AcpAgentsFile) -> Result<(), AcpConfigError> { + save_acp_agents_to(config, default_config_path()).await +} + +/// Save ACP agent configurations to a specific path. +pub async fn save_acp_agents_to( + config: &AcpAgentsFile, + path: impl AsRef, +) -> Result<(), AcpConfigError> { + let path = path.as_ref(); + + if let Some(parent) = path.parent() { + fs::create_dir_all(parent).await?; + } + + let content = serde_json::to_string_pretty(config)?; + + // Atomic write via temp file to avoid corruption on crash. + let tmp_path = path.with_extension("json.tmp"); + fs::write(&tmp_path, content).await?; + fs::rename(&tmp_path, path).await?; + + Ok(()) +} + +/// Add a new ACP agent configuration (disk-backed). +pub async fn add_acp_agent(config: AcpAgentConfig) -> Result<(), AcpConfigError> { + config.validate()?; + + let mut agents = load_acp_agents().await?; + agents.upsert(config); + save_acp_agents(&agents).await?; + + Ok(()) +} + +/// Remove an ACP agent by name (disk-backed). +pub async fn remove_acp_agent(name: &str) -> Result<(), AcpConfigError> { + let mut agents = load_acp_agents().await?; + + if !agents.remove(name) { + return Err(AcpConfigError::AgentNotFound { + name: name.to_string(), + }); + } + + save_acp_agents(&agents).await?; + + Ok(()) +} + +/// Get a specific ACP agent configuration (disk-backed). +pub async fn get_acp_agent(name: &str) -> Result { + let agents = load_acp_agents().await?; + + agents + .get(name) + .cloned() + .ok_or_else(|| AcpConfigError::AgentNotFound { + name: name.to_string(), + }) +} + +// ==================== Database-backed persistence ==================== + +/// Load ACP agent configurations from the database settings table. +/// +/// Falls back to the disk file only if DB has no entry. +pub async fn load_acp_agents_from_db( + store: &dyn crate::db::Database, + user_id: &str, +) -> Result { + match store.get_setting(user_id, "acp_agents").await { + Ok(Some(value)) => { + let config: AcpAgentsFile = serde_json::from_value(value)?; + for agent in &config.agents { + agent + .validate() + .map_err(|e| AcpConfigError::InvalidConfig { + reason: format!("Agent '{}': {}", agent.name, e), + })?; + } + Ok(config) + } + Ok(None) => load_acp_agents().await, + Err(e) => Err(AcpConfigError::Database(e.to_string())), + } +} + +/// Load ACP agent configurations from the active persistence backend. +pub async fn load_acp_agents_for_user( + store: Option<&dyn crate::db::Database>, + user_id: &str, +) -> Result { + match store { + Some(store) => load_acp_agents_from_db(store, user_id).await, + None => load_acp_agents().await, + } +} + +/// Save ACP agent configurations to the database settings table. +pub async fn save_acp_agents_to_db( + store: &dyn crate::db::Database, + user_id: &str, + config: &AcpAgentsFile, +) -> Result<(), AcpConfigError> { + let value = serde_json::to_value(config)?; + store + .set_setting(user_id, "acp_agents", &value) + .await + .map_err(std::io::Error::other)?; + Ok(()) +} + +/// Save ACP agent configurations to the active persistence backend. +pub async fn save_acp_agents_for_user( + store: Option<&dyn crate::db::Database>, + user_id: &str, + config: &AcpAgentsFile, +) -> Result<(), AcpConfigError> { + match store { + Some(store) => save_acp_agents_to_db(store, user_id, config).await, + None => save_acp_agents(config).await, + } +} + +/// Add a new ACP agent configuration (DB-backed). +pub async fn add_acp_agent_db( + store: &dyn crate::db::Database, + user_id: &str, + config: AcpAgentConfig, +) -> Result<(), AcpConfigError> { + config.validate()?; + + let mut agents = load_acp_agents_from_db(store, user_id).await?; + agents.upsert(config); + save_acp_agents_to_db(store, user_id, &agents).await?; + + Ok(()) +} + +/// Remove an ACP agent by name (DB-backed). +pub async fn remove_acp_agent_db( + store: &dyn crate::db::Database, + user_id: &str, + name: &str, +) -> Result<(), AcpConfigError> { + let mut agents = load_acp_agents_from_db(store, user_id).await?; + + if !agents.remove(name) { + return Err(AcpConfigError::AgentNotFound { + name: name.to_string(), + }); + } + + save_acp_agents_to_db(store, user_id, &agents).await?; + Ok(()) +} + +/// Load a single ACP agent from the active persistence backend. +pub async fn get_acp_agent_for_user( + store: Option<&dyn crate::db::Database>, + user_id: &str, + name: &str, +) -> Result { + let agents = load_acp_agents_for_user(store, user_id).await?; + agents + .get(name) + .cloned() + .ok_or_else(|| AcpConfigError::AgentNotFound { + name: name.to_string(), + }) +} + +/// Load a single ACP agent and ensure it is enabled. +pub async fn get_enabled_acp_agent_for_user( + store: Option<&dyn crate::db::Database>, + user_id: &str, + name: &str, +) -> Result { + let agent = get_acp_agent_for_user(store, user_id, name).await?; + if !agent.enabled { + return Err(AcpConfigError::AgentDisabled { + name: name.to_string(), + }); + } + Ok(agent) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_agent_config_new() { + let agent = AcpAgentConfig::new("goose", "goose", vec!["--stdio".into()], HashMap::new()); + assert_eq!(agent.name, "goose"); + assert_eq!(agent.command, "goose"); + assert_eq!(agent.args, vec!["--stdio"]); + assert!(agent.enabled); + assert!(agent.description.is_none()); + } + + #[test] + fn test_agent_config_with_description() { + let agent = AcpAgentConfig::new("goose", "goose", vec![], HashMap::new()) + .with_description("Goose coding agent"); + assert_eq!(agent.description.as_deref(), Some("Goose coding agent")); + } + + #[test] + fn test_validate_empty_name() { + let agent = AcpAgentConfig::new("", "goose", vec![], HashMap::new()); + assert!(agent.validate().is_err()); + } + + #[test] + fn test_validate_empty_command() { + let agent = AcpAgentConfig::new("goose", "", vec![], HashMap::new()); + assert!(agent.validate().is_err()); + } + + #[test] + fn test_validate_ok() { + let agent = AcpAgentConfig::new("goose", "goose", vec!["--stdio".into()], HashMap::new()); + assert!(agent.validate().is_ok()); + } + + #[test] + fn test_agents_file_default() { + let file = AcpAgentsFile::default(); + assert!(file.agents.is_empty()); + assert_eq!(file.schema_version, 1); + } + + #[test] + fn test_agents_file_get() { + let mut file = AcpAgentsFile::default(); + file.agents.push(AcpAgentConfig::new( + "goose", + "goose", + vec![], + HashMap::new(), + )); + assert!(file.get("goose").is_some()); + assert!(file.get("nonexistent").is_none()); + } + + #[test] + fn test_agents_file_upsert_new() { + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec![], + HashMap::new(), + )); + assert_eq!(file.agents.len(), 1); + } + + #[test] + fn test_agents_file_upsert_existing() { + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec![], + HashMap::new(), + )); + file.upsert(AcpAgentConfig::new( + "goose", + "goose-v2", + vec!["--stdio".into()], + HashMap::new(), + )); + assert_eq!(file.agents.len(), 1); + assert_eq!(file.agents[0].command, "goose-v2"); + } + + #[test] + fn test_agents_file_remove() { + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec![], + HashMap::new(), + )); + assert!(file.remove("goose")); + assert!(file.agents.is_empty()); + assert!(!file.remove("goose")); // already removed + } + + #[test] + fn test_agents_file_enabled_agents() { + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec![], + HashMap::new(), + )); + let mut disabled = AcpAgentConfig::new("codex", "codex", vec![], HashMap::new()); + disabled.enabled = false; + file.upsert(disabled); + + let enabled: Vec<_> = file.enabled_agents().collect(); + assert_eq!(enabled.len(), 1); + assert_eq!(enabled[0].name, "goose"); + } + + #[test] + fn test_agents_file_serde_roundtrip() { + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec!["--stdio".into()], + HashMap::from([("GOOSE_TOKEN".to_string(), "secret".to_string())]), + )); + + let json = serde_json::to_string_pretty(&file).unwrap(); + let parsed: AcpAgentsFile = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed.agents.len(), 1); + assert_eq!(parsed.agents[0].name, "goose"); + assert_eq!(parsed.agents[0].command, "goose"); + assert_eq!(parsed.agents[0].args, vec!["--stdio"]); + assert!(parsed.agents[0].env.contains_key("GOOSE_TOKEN")); + } + + #[test] + fn test_default_config_path() { + let path = default_config_path(); + assert!(path.ends_with("acp-agents.json")); + } + + #[tokio::test] + async fn test_load_nonexistent_returns_empty() { + let path = std::env::temp_dir().join("ironclaw-test-nonexistent-acp.json"); + let _ = std::fs::remove_file(&path); // ensure clean + let file = load_acp_agents_from(&path).await.unwrap(); + assert!(file.agents.is_empty()); + } + + #[tokio::test] + async fn test_save_and_load_roundtrip() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("acp-agents.json"); + + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec!["--stdio".into()], + HashMap::new(), + )); + + save_acp_agents_to(&file, &path).await.unwrap(); + let loaded = load_acp_agents_from(&path).await.unwrap(); + assert_eq!(loaded.agents.len(), 1); + assert_eq!(loaded.agents[0].name, "goose"); + } + + #[tokio::test] + async fn test_load_rejects_invalid() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("acp-agents.json"); + tokio::fs::write( + &path, + r#"{"agents":[{"name":"","command":"goose","args":[]}]}"#, + ) + .await + .unwrap(); + + let result = load_acp_agents_from(&path).await; + assert!(result.is_err()); + } + + #[cfg(feature = "libsql")] + #[tokio::test] + async fn test_load_and_save_acp_agents_for_non_default_owner_scope() { + let (db, _tmp) = crate::testing::test_db().await; + + let mut file = AcpAgentsFile::default(); + file.upsert(AcpAgentConfig::new( + "goose", + "goose", + vec!["--stdio".into()], + HashMap::new(), + )); + + save_acp_agents_for_user(Some(db.as_ref()), "owner-123", &file) + .await + .unwrap(); + + let loaded = load_acp_agents_for_user(Some(db.as_ref()), "owner-123") + .await + .unwrap(); + assert_eq!( + loaded.get("goose").map(|agent| agent.command.as_str()), + Some("goose") + ); + + let default_scope = load_acp_agents_for_user(Some(db.as_ref()), "default") + .await + .unwrap(); + assert!(default_scope.get("goose").is_none()); + } + + #[cfg(feature = "libsql")] + #[tokio::test] + async fn test_get_enabled_agent_rejects_disabled_agent() { + let (db, _tmp) = crate::testing::test_db().await; + + let mut file = AcpAgentsFile::default(); + let mut agent = AcpAgentConfig::new("codex", "codex", vec!["acp".into()], HashMap::new()); + agent.enabled = false; + file.upsert(agent); + + save_acp_agents_for_user(Some(db.as_ref()), "owner-123", &file) + .await + .unwrap(); + + let err = get_enabled_acp_agent_for_user(Some(db.as_ref()), "owner-123", "codex") + .await + .unwrap_err(); + assert!(matches!(err, AcpConfigError::AgentDisabled { .. })); + } +} diff --git a/src/config/mod.rs b/src/config/mod.rs index 54d639d552..33417c2485 100644 --- a/src/config/mod.rs +++ b/src/config/mod.rs @@ -18,6 +18,7 @@ //! `DATABASE_URL` lives in `~/.ironclaw/.env` (loaded via dotenvy early //! in startup). +pub mod acp; mod agent; mod builder; mod channels; @@ -61,7 +62,7 @@ pub use self::relay::RelayConfig; pub use self::routines::RoutineConfig; pub use self::safety::SafetyConfig; use self::safety::resolve_safety_config; -pub use self::sandbox::{ClaudeCodeConfig, SandboxModeConfig}; +pub use self::sandbox::{AcpModeConfig, ClaudeCodeConfig, SandboxModeConfig}; pub use self::search::WorkspaceSearchConfig; pub use self::secrets::SecretsConfig; pub use self::skills::SkillsConfig; @@ -111,6 +112,7 @@ pub struct Config { pub routines: RoutineConfig, pub sandbox: SandboxModeConfig, pub claude_code: ClaudeCodeConfig, + pub acp: AcpModeConfig, pub skills: SkillsConfig, pub transcription: TranscriptionConfig, pub search: WorkspaceSearchConfig, @@ -184,6 +186,7 @@ impl Config { ..SandboxModeConfig::default() }, claude_code: ClaudeCodeConfig::default(), + acp: AcpModeConfig::default(), skills: SkillsConfig { enabled: true, local_dir: skills_dir, @@ -378,6 +381,7 @@ impl Config { routines: RoutineConfig::resolve(settings)?, sandbox: SandboxModeConfig::resolve(settings)?, claude_code: ClaudeCodeConfig::resolve(settings)?, + acp: AcpModeConfig::resolve(settings)?, skills: SkillsConfig::resolve(settings)?, transcription: TranscriptionConfig::resolve(settings)?, search: WorkspaceSearchConfig::resolve(settings)?, diff --git a/src/config/sandbox.rs b/src/config/sandbox.rs index 7f5adb7a01..3c5536e245 100644 --- a/src/config/sandbox.rs +++ b/src/config/sandbox.rs @@ -357,6 +357,62 @@ fn parse_oauth_access_token(json: &str) -> Option { Some(token.to_string()) } +/// ACP (Agent Client Protocol) mode configuration. +/// +/// Controls whether ACP agent delegation is available. Agent definitions +/// are stored separately in a DB blob (key `"acp_agents"`) or disk file +/// (`~/.ironclaw/acp-agents.json`), following the MCP server pattern. +#[derive(Debug, Clone)] +pub struct AcpModeConfig { + /// Whether ACP agent mode is available. + pub enabled: bool, + /// Memory limit in MB for ACP containers. + pub memory_limit_mb: u64, + /// Maximum timeout for an ACP session in seconds. + pub timeout_secs: u64, +} + +impl Default for AcpModeConfig { + fn default() -> Self { + Self { + enabled: false, + memory_limit_mb: 4096, + timeout_secs: 1800, + } + } +} + +impl AcpModeConfig { + /// Load from environment variables only (used inside containers). + pub fn from_env() -> Self { + match Self::resolve_env_only() { + Ok(c) => c, + Err(e) => { + tracing::warn!("Failed to resolve AcpModeConfig: {e}, using defaults"); + Self::default() + } + } + } + + pub(crate) fn resolve(settings: &crate::settings::Settings) -> Result { + let defaults = Self::default(); + Ok(Self { + enabled: parse_bool_env("ACP_ENABLED", settings.sandbox.acp_enabled)?, + memory_limit_mb: parse_optional_env("ACP_MEMORY_LIMIT_MB", defaults.memory_limit_mb)?, + timeout_secs: parse_optional_env("ACP_TIMEOUT_SECS", defaults.timeout_secs)?, + }) + } + + fn resolve_env_only() -> Result { + let defaults = Self::default(); + Ok(Self { + enabled: parse_bool_env("ACP_ENABLED", defaults.enabled)?, + memory_limit_mb: parse_optional_env("ACP_MEMORY_LIMIT_MB", defaults.memory_limit_mb)?, + timeout_secs: parse_optional_env("ACP_TIMEOUT_SECS", defaults.timeout_secs)?, + }) + } +} + #[cfg(test)] mod tests { use crate::config::sandbox::*; @@ -712,4 +768,44 @@ mod tests { let sandbox = config.to_sandbox_config(); assert_eq!(sandbox.policy, crate::sandbox::SandboxPolicy::ReadOnly); } + + // ── AcpModeConfig defaults ────────────────────────────────── + + #[test] + fn acp_mode_config_default_values() { + let cfg = AcpModeConfig::default(); + assert!(!cfg.enabled); + assert_eq!(cfg.memory_limit_mb, 4096); + assert_eq!(cfg.timeout_secs, 1800); + } + + #[test] + fn acp_mode_config_resolve_uses_settings() { + let _guard = crate::config::helpers::lock_env(); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.acp_enabled = true; + + let cfg = AcpModeConfig::resolve(&settings).expect("resolve"); + assert!(cfg.enabled); + } + + #[test] + fn acp_mode_config_env_overrides_settings() { + let _guard = crate::config::helpers::lock_env(); + let mut settings = crate::settings::Settings::default(); + settings.sandbox.acp_enabled = true; + + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("ACP_ENABLED", "false") }; + unsafe { std::env::set_var("ACP_MEMORY_LIMIT_MB", "8192") }; + unsafe { std::env::set_var("ACP_TIMEOUT_SECS", "3600") }; + let cfg = AcpModeConfig::resolve(&settings).expect("resolve"); + unsafe { std::env::remove_var("ACP_ENABLED") }; + unsafe { std::env::remove_var("ACP_MEMORY_LIMIT_MB") }; + unsafe { std::env::remove_var("ACP_TIMEOUT_SECS") }; + + assert!(!cfg.enabled); + assert_eq!(cfg.memory_limit_mb, 8192); + assert_eq!(cfg.timeout_secs, 3600); + } } diff --git a/src/extensions/manager.rs b/src/extensions/manager.rs index be97971d1c..e68ec57132 100644 --- a/src/extensions/manager.rs +++ b/src/extensions/manager.rs @@ -1324,6 +1324,10 @@ impl ExtensionManager { "Channel relay extensions cannot be installed by URL".to_string(), )) } + ExtensionKind::AcpAgent => Err(ExtensionError::InstallFailed( + "ACP agents are configured via 'ironclaw acp add', not the extension manager" + .to_string(), + )), } .map_err(|e| { let sanitized = sanitize_url_for_logging(url); @@ -1356,6 +1360,11 @@ impl ExtensionManager { ExtensionKind::WasmTool => self.auth_wasm_tool(name, user_id).await, ExtensionKind::WasmChannel => self.auth_wasm_channel_status(name, user_id).await, ExtensionKind::ChannelRelay => self.auth_channel_relay(name, user_id).await, + ExtensionKind::AcpAgent => Ok(AuthResult { + name: name.to_string(), + kind: ExtensionKind::AcpAgent, + status: crate::extensions::AuthStatus::NoAuthRequired, + }), } } @@ -1373,6 +1382,15 @@ impl ExtensionManager { ExtensionKind::WasmTool => self.activate_wasm_tool(name, user_id).await, ExtensionKind::WasmChannel => self.activate_wasm_channel(name, user_id).await, ExtensionKind::ChannelRelay => self.activate_channel_relay(name, user_id).await, + ExtensionKind::AcpAgent => Ok(ActivateResult { + name: name.to_string(), + kind: ExtensionKind::AcpAgent, + tools_loaded: Vec::new(), + message: format!( + "ACP agent '{}' is managed via 'ironclaw acp' commands", + name + ), + }), } } @@ -1813,6 +1831,13 @@ impl ExtensionManager { Ok(format!("Removed channel relay '{}'", name)) } + ExtensionKind::AcpAgent => { + // ACP agents are managed via `ironclaw acp remove` + Ok(format!( + "ACP agent '{}' should be removed via 'ironclaw acp remove {}'", + name, name + )) + } } } @@ -1904,7 +1929,7 @@ impl ExtensionManager { &self.wasm_channels_dir, crate::tools::wasm::WIT_CHANNEL_VERSION, ), - ExtensionKind::McpServer | ExtensionKind::ChannelRelay => { + ExtensionKind::McpServer | ExtensionKind::ChannelRelay | ExtensionKind::AcpAgent => { return UpgradeOutcome { name: name.to_string(), kind, @@ -1930,7 +1955,9 @@ impl ExtensionManager { .ok() .and_then(|c| c.wit_version) } - ExtensionKind::McpServer | ExtensionKind::ChannelRelay => None, + ExtensionKind::McpServer + | ExtensionKind::ChannelRelay + | ExtensionKind::AcpAgent => None, }; wit } @@ -2102,6 +2129,13 @@ impl ExtensionManager { }); Ok(info) } + ExtensionKind::AcpAgent => { + let info = serde_json::json!({ + "name": name, + "kind": "acp_agent", + }); + Ok(info) + } } } @@ -2291,6 +2325,9 @@ impl ExtensionManager { ), }) } + ExtensionKind::AcpAgent => Err(ExtensionError::InstallFailed( + "ACP agents are configured via 'ironclaw acp add', not the registry".to_string(), + )), } } @@ -2652,6 +2689,7 @@ impl ExtensionManager { ExtensionKind::WasmChannel => "WASM channel", ExtensionKind::McpServer => "MCP server", ExtensionKind::ChannelRelay => "channel relay", + ExtensionKind::AcpAgent => "ACP agent", }; tracing::info!( @@ -3133,7 +3171,7 @@ impl ExtensionManager { oauth_scopes_secret_name(&token_secret_name), ); } - ExtensionKind::ChannelRelay => {} + ExtensionKind::ChannelRelay | ExtensionKind::AcpAgent => {} } Ok(plan) @@ -5652,6 +5690,11 @@ impl ExtensionManager { }]; (std::collections::HashSet::new(), relay_fields) } + ExtensionKind::AcpAgent => { + return Err(ExtensionError::Other( + "ACP agents do not require setup through the extension manager".to_string(), + )); + } }; let allowed_fields: std::collections::HashSet = @@ -5937,7 +5980,7 @@ impl ExtensionManager { ExtensionKind::WasmChannel => self.activate_wasm_channel(name, user_id).await, ExtensionKind::McpServer => self.activate_mcp(name, user_id).await, ExtensionKind::ChannelRelay => self.activate_channel_relay(name, user_id).await, - ExtensionKind::WasmTool => { + ExtensionKind::WasmTool | ExtensionKind::AcpAgent => { return Ok(ConfigureResult { message: format!("Configuration saved for '{}'.", name), activated: false, @@ -6100,6 +6143,11 @@ impl ExtensionManager { ExtensionKind::ChannelRelay => { return Err(ExtensionError::AuthRequired); } + ExtensionKind::AcpAgent => { + return Err(ExtensionError::Other( + "ACP agents do not use token-based authentication".to_string(), + )); + } }; let mut secrets = std::collections::HashMap::new(); diff --git a/src/extensions/mod.rs b/src/extensions/mod.rs index 2602884d45..81d5a67027 100644 --- a/src/extensions/mod.rs +++ b/src/extensions/mod.rs @@ -40,6 +40,8 @@ pub enum ExtensionKind { WasmChannel, /// External channel via channel-relay service (Slack, etc.). ChannelRelay, + /// ACP-compliant coding agent (Goose, Codex, Gemini CLI, etc.). + AcpAgent, } impl std::fmt::Display for ExtensionKind { @@ -49,6 +51,7 @@ impl std::fmt::Display for ExtensionKind { ExtensionKind::WasmTool => write!(f, "wasm_tool"), ExtensionKind::WasmChannel => write!(f, "wasm_channel"), ExtensionKind::ChannelRelay => write!(f, "channel_relay"), + ExtensionKind::AcpAgent => write!(f, "acp_agent"), } } } diff --git a/src/main.rs b/src/main.rs index 6f4a37c570..224f181e2e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -165,6 +165,10 @@ async fn async_main() -> anyhow::Result<()> { let config = ironclaw::config::Config::from_env().await?; return ironclaw::cli::run_import_command(import_cmd, &config).await; } + Some(Command::Acp(acp_cmd)) => { + init_cli_tracing(); + return ironclaw::cli::run_acp_command(acp_cmd.clone()).await; + } Some(Command::Worker { job_id, orchestrator_url, @@ -188,6 +192,13 @@ async fn async_main() -> anyhow::Result<()> { ) .await; } + Some(Command::AcpBridge { + job_id, + orchestrator_url, + }) => { + init_worker_tracing(); + return ironclaw::worker::run_acp_bridge(*job_id, orchestrator_url).await; + } Some(Command::Login { openai_codex }) => { init_cli_tracing(); if *openai_codex { @@ -799,6 +810,7 @@ async fn async_main() -> anyhow::Result<()> { sandbox_enabled: config.sandbox.enabled, docker_status, claude_code_enabled: config.claude_code.enabled, + acp_enabled: config.acp.enabled, routines_enabled: config.routines.enabled, skills_enabled: config.skills.enabled, channels: channel_names, diff --git a/src/orchestrator/job_manager.rs b/src/orchestrator/job_manager.rs index 222d24650a..c022e26df1 100644 --- a/src/orchestrator/job_manager.rs +++ b/src/orchestrator/job_manager.rs @@ -28,6 +28,8 @@ pub enum JobMode { Worker, /// Claude Code bridge that spawns the `claude` CLI directly. ClaudeCode, + /// ACP (Agent Client Protocol) bridge that spawns any ACP-compliant agent. + Acp, } impl JobMode { @@ -35,6 +37,7 @@ impl JobMode { match self { Self::Worker => "worker", Self::ClaudeCode => "claude_code", + Self::Acp => "acp", } } } @@ -56,6 +59,8 @@ pub struct JobCreationParams { pub mcp_servers: Option>, /// Optional cap on worker agent loop iterations (clamped to 1..=500 server-side). pub max_iterations: Option, + /// ACP agent definition to inject into ACP-mode containers. + pub acp_agent: Option, } /// Configuration for the container job manager. @@ -84,6 +89,10 @@ pub struct ContainerJobConfig { pub claude_code_memory_limit_mb: u64, /// Allowed tool patterns for Claude Code (passed as CLAUDE_CODE_ALLOWED_TOOLS env var). pub claude_code_allowed_tools: Vec, + /// Memory limit for ACP containers. + pub acp_memory_limit_mb: u64, + /// Maximum runtime for ACP bridge sessions in seconds. + pub acp_timeout_secs: u64, /// Whether per-job MCP server filtering is enabled. /// When false, `mcp_servers` param on `create_job` is ignored. pub mcp_per_job_enabled: bool, @@ -102,6 +111,8 @@ impl Default for ContainerJobConfig { claude_code_max_turns: 50, claude_code_memory_limit_mb: 4096, claude_code_allowed_tools: crate::config::ClaudeCodeConfig::default().allowed_tools, + acp_memory_limit_mb: 4096, + acp_timeout_secs: 1800, mcp_per_job_enabled: false, } } @@ -247,6 +258,28 @@ impl ContainerJobManager { } } + fn extend_acp_env( + &self, + env_vec: &mut Vec, + acp_agent: Option<&crate::config::acp::AcpAgentConfig>, + ) { + env_vec.push(format!("ACP_TIMEOUT_SECS={}", self.config.acp_timeout_secs)); + + if let Some(agent) = acp_agent { + env_vec.push(format!("ACP_AGENT_COMMAND={}", agent.command)); + if !agent.args.is_empty() + && let Ok(json) = serde_json::to_string(&agent.args) + { + env_vec.push(format!("ACP_AGENT_ARGS={}", json)); + } + if !agent.env.is_empty() + && let Ok(json) = serde_json::to_string(&agent.env) + { + env_vec.push(format!("ACP_AGENT_ENV={}", json)); + } + } + } + /// Get or create a Docker connection. async fn docker(&self) -> Result { { @@ -282,8 +315,15 @@ impl ContainerJobManager { let token = self.token_store.create_token(job_id).await; // Store credential grants (revoked automatically when the token is revoked) + let JobCreationParams { + credential_grants, + mcp_servers, + max_iterations, + acp_agent, + } = params; + self.token_store - .store_grants(job_id, params.credential_grants) + .store_grants(job_id, credential_grants) .await; // Record the handle @@ -309,8 +349,9 @@ impl ContainerJobManager { &token, project_dir, mode, - params.mcp_servers, - params.max_iterations, + mcp_servers, + max_iterations, + acp_agent, ) .await { @@ -324,6 +365,7 @@ impl ContainerJobManager { } /// Inner implementation of container creation (separated for cleanup). + #[allow(clippy::too_many_arguments)] async fn create_job_inner( &self, job_id: Uuid, @@ -332,16 +374,15 @@ impl ContainerJobManager { mode: JobMode, mcp_servers: Option>, max_iterations: Option, + acp_agent: Option, ) -> Result<(), OrchestratorError> { // Connect to Docker (reuses cached connection) let docker = self.docker().await?; // Build container configuration - let orchestrator_host = if cfg!(target_os = "linux") { - "172.17.0.1" - } else { - "host.docker.internal" - }; + // Use host.docker.internal on all platforms — the extra_hosts mapping + // below resolves it to the actual host IP via Docker's host-gateway. + let orchestrator_host = "host.docker.internal"; let orchestrator_url = format!( "http://{}:{}", @@ -418,9 +459,15 @@ impl ContainerJobManager { } } - // Memory limit: Claude Code gets more memory + // ACP mode: inject runtime timeout plus per-job agent command/args/env. + if mode == JobMode::Acp { + self.extend_acp_env(&mut env_vec, acp_agent.as_ref()); + } + + // Memory limit per mode let memory_mb = match mode { JobMode::ClaudeCode => self.config.claude_code_memory_limit_mb, + JobMode::Acp => self.config.acp_memory_limit_mb, JobMode::Worker => self.config.memory_limit_mb, }; @@ -465,6 +512,13 @@ impl ContainerJobManager { "--model".to_string(), self.config.claude_code_model.clone(), ], + JobMode::Acp => vec![ + "acp-bridge".to_string(), + "--job-id".to_string(), + job_id.to_string(), + "--orchestrator-url".to_string(), + orchestrator_url, + ], }; // Add Docker labels for reaper identification and orphan detection @@ -489,6 +543,7 @@ impl ContainerJobManager { let container_name = match mode { JobMode::Worker => format!("ironclaw-worker-{}", job_id), JobMode::ClaudeCode => format!("ironclaw-claude-{}", job_id), + JobMode::Acp => format!("ironclaw-acp-{}", job_id), }; let options = CreateContainerOptions { name: container_name, @@ -916,6 +971,58 @@ mod tests { assert_eq!(handle.last_worker_status.as_deref(), Some("Iteration 3")); } + #[test] + fn test_job_mode_acp_as_str() { + assert_eq!(JobMode::Acp.as_str(), "acp"); + } + + #[test] + fn test_job_mode_acp_display() { + assert_eq!(format!("{}", JobMode::Acp), "acp"); + } + + #[test] + fn test_container_job_config_acp_memory_default() { + let config = ContainerJobConfig::default(); + assert_eq!(config.acp_memory_limit_mb, 4096); + } + + #[test] + fn test_container_job_config_acp_timeout_default() { + let config = ContainerJobConfig::default(); + assert_eq!(config.acp_timeout_secs, 1800); + } + + #[test] + fn test_extend_acp_env_includes_timeout_and_agent_details() { + let config = ContainerJobConfig { + acp_timeout_secs: 45, + ..Default::default() + }; + let manager = ContainerJobManager::new(config, TokenStore::new()); + + let agent = crate::config::acp::AcpAgentConfig::new( + "codex", + "codex", + vec!["acp".into()], + HashMap::from([("FOO".to_string(), "bar".to_string())]), + ); + let mut env_vec = Vec::new(); + manager.extend_acp_env(&mut env_vec, Some(&agent)); + + assert!(env_vec.contains(&"ACP_TIMEOUT_SECS=45".to_string())); + assert!(env_vec.contains(&"ACP_AGENT_COMMAND=codex".to_string())); + assert!( + env_vec + .iter() + .any(|entry| entry.starts_with("ACP_AGENT_ARGS=")) + ); + assert!( + env_vec + .iter() + .any(|entry| entry.starts_with("ACP_AGENT_ENV=")) + ); + } // ── generate_worker_mcp_config tests ──────────────────────────── #[tokio::test] diff --git a/src/orchestrator/mod.rs b/src/orchestrator/mod.rs index d6ecb61c2f..7902c54512 100644 --- a/src/orchestrator/mod.rs +++ b/src/orchestrator/mod.rs @@ -122,6 +122,8 @@ pub async fn setup_orchestrator( claude_code_max_turns: config.claude_code.max_turns, claude_code_memory_limit_mb: config.claude_code.memory_limit_mb, claude_code_allowed_tools: config.claude_code.allowed_tools.clone(), + acp_memory_limit_mb: config.acp.memory_limit_mb, + acp_timeout_secs: config.acp.timeout_secs, mcp_per_job_enabled: std::env::var("MCP_PER_JOB_ENABLED") .map(|v| v.eq_ignore_ascii_case("true") || v == "1") .unwrap_or(false), @@ -153,6 +155,9 @@ pub async fn setup_orchestrator( config.claude_code.max_turns ); } + if config.acp.enabled { + tracing::info!("ACP agent sandbox mode available"); + } (job_event_tx, Some(jm)) } else { (None, None) diff --git a/src/settings.rs b/src/settings.rs index 384cbf6aac..f6bc879f9e 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -700,6 +700,10 @@ pub struct SandboxSettings { /// Whether Claude Code sandbox mode is enabled. #[serde(default)] pub claude_code_enabled: bool, + + /// Whether ACP (Agent Client Protocol) agent mode is enabled. + #[serde(default)] + pub acp_enabled: bool, } fn default_sandbox_policy() -> String { @@ -734,6 +738,7 @@ impl Default for SandboxSettings { auto_pull_image: true, extra_allowed_domains: Vec::new(), claude_code_enabled: false, + acp_enabled: false, } } } diff --git a/src/tools/builtin/job.rs b/src/tools/builtin/job.rs index 940c7cfc69..9b59518084 100644 --- a/src/tools/builtin/job.rs +++ b/src/tools/builtin/job.rs @@ -355,6 +355,7 @@ impl CreateJobTool { } /// Execute via sandboxed Docker container. + #[allow(clippy::too_many_arguments)] async fn execute_sandbox( &self, task: &str, @@ -414,16 +415,21 @@ impl CreateJobTool { credential_grants_json, }); - // Persist the job mode to DB - if mode == JobMode::ClaudeCode + // Persist the job mode to DB (for non-default modes). + // For ACP, store "acp:" so restarts know which agent to use. + if mode != JobMode::Worker && let Some(store) = self.store.clone() { let job_id_copy = job_id; + let mode_str = if mode == JobMode::Acp + && let Some(ref agent) = params.acp_agent + { + format!("acp:{}", agent.name) + } else { + mode.as_str().to_string() + }; tokio::spawn(async move { - if let Err(e) = store - .update_sandbox_job_mode(job_id_copy, "claude_code") - .await - { + if let Err(e) = store.update_sandbox_job_mode(job_id_copy, &mode_str).await { tracing::warn!(job_id = %job_id_copy, "Failed to set job mode: {}", e); } }); @@ -834,9 +840,15 @@ impl Tool for CreateJobTool { }, "mode": { "type": "string", - "enum": ["worker", "claude_code"], + "enum": ["worker", "claude_code", "acp"], "description": "Execution mode. 'worker' (default) uses the IronClaw sub-agent. \ - 'claude_code' uses Claude Code CLI for full agentic software engineering." + 'claude_code' uses Claude Code CLI. \ + 'acp' uses an ACP-compliant agent (Goose, Codex, Gemini CLI)." + }, + "agent_name": { + "type": "string", + "description": "Name of the ACP agent to use (from 'ironclaw acp list'). \ + Required when mode is 'acp'." }, "project_dir": { "type": "string", @@ -910,9 +922,35 @@ impl Tool for CreateJobTool { let mode = match params.get("mode").and_then(|v| v.as_str()) { Some("claude_code") => JobMode::ClaudeCode, + Some("acp") => JobMode::Acp, _ => JobMode::Worker, }; + // Resolve ACP agent config when mode is ACP. + let acp_agent = if mode == JobMode::Acp { + let agent_name = require_str(¶ms, "agent_name")?; + Some( + crate::config::acp::get_enabled_acp_agent_for_user( + self.store.as_deref(), + &ctx.user_id, + agent_name, + ) + .await + .map_err(|e| match e { + crate::config::acp::AcpConfigError::AgentNotFound { .. } + | crate::config::acp::AcpConfigError::AgentDisabled { .. } => { + ToolError::InvalidParameters(e.to_string()) + } + _ => ToolError::ExecutionFailed(format!( + "failed to load ACP agent '{}': {}", + agent_name, e + )), + })?, + ) + } else { + None + }; + let explicit_dir = params .get("project_dir") .and_then(|v| v.as_str()) @@ -955,6 +993,7 @@ impl Tool for CreateJobTool { credential_grants, mcp_servers, max_iterations, + acp_agent, }, ctx, ) @@ -2310,4 +2349,69 @@ mod tests { let result = resolve_job_id("not-hex-at-all!", &cm).await; assert!(result.is_err()); // safety: test } + + // ── ACP mode tests ────────────────────────────────────────── + + #[test] + fn test_sandbox_schema_includes_acp_mode() { + let manager = Arc::new(ContextManager::new(5)); + let jm = Arc::new(ContainerJobManager::new( + crate::orchestrator::job_manager::ContainerJobConfig::default(), + crate::orchestrator::TokenStore::new(), + )); + let tool = CreateJobTool::new(manager).with_sandbox(jm, None); + let schema = tool.parameters_schema(); + let mode_enum = schema["properties"]["mode"]["enum"].as_array().unwrap(); // safety: test + let modes: Vec<&str> = mode_enum.iter().map(|v| v.as_str().unwrap()).collect(); // safety: test + assert!(modes.contains(&"acp"), "mode enum must include 'acp'"); + assert!(modes.contains(&"worker")); + assert!(modes.contains(&"claude_code")); + } + + #[test] + fn test_sandbox_schema_includes_agent_name() { + let manager = Arc::new(ContextManager::new(5)); + let jm = Arc::new(ContainerJobManager::new( + crate::orchestrator::job_manager::ContainerJobConfig::default(), + crate::orchestrator::TokenStore::new(), + )); + let tool = CreateJobTool::new(manager).with_sandbox(jm, None); + let schema = tool.parameters_schema(); + let props = schema.get("properties").unwrap().as_object().unwrap(); // safety: test + assert!( + /* safety: test */ + props.contains_key("agent_name"), + "sandbox schema must expose agent_name for ACP mode" + ); + } + + #[tokio::test] + async fn test_acp_mode_requires_agent_name() { + let manager = Arc::new(ContextManager::new(5)); + let jm = Arc::new(ContainerJobManager::new( + crate::orchestrator::job_manager::ContainerJobConfig::default(), + crate::orchestrator::TokenStore::new(), + )); + let tool = CreateJobTool::new(manager).with_sandbox(jm, None); + + let params = serde_json::json!({ + "title": "Test ACP job", + "description": "Test task", + "mode": "acp" + // no agent_name — should fail + }); + let result = tool.execute(params, &JobContext::default()).await; + assert!(result.is_err()); // safety: test + let err = result.unwrap_err().to_string(); // safety: test + assert!( + err.contains("agent_name"), + "error should mention missing agent_name, got: {err}" + ); + } + + #[test] + fn test_job_mode_acp_as_str() { + assert_eq!(JobMode::Acp.as_str(), "acp"); + assert_eq!(JobMode::Acp.to_string(), "acp"); + } } diff --git a/src/worker/acp_bridge.rs b/src/worker/acp_bridge.rs new file mode 100644 index 0000000000..c981dc558a --- /dev/null +++ b/src/worker/acp_bridge.rs @@ -0,0 +1,629 @@ +//! ACP (Agent Client Protocol) bridge for sandboxed execution. +//! +//! Spawns any ACP-compliant agent (Goose, Codex, Gemini CLI, etc.) as a +//! subprocess inside a Docker container and communicates via the standard +//! ACP protocol (JSON-RPC over stdio). Agent output is translated into +//! IronClaw's `JobEventPayload` stream and posted to the orchestrator. +//! +//! Security model: the Docker container is the primary security boundary +//! (cap-drop ALL, non-root user, memory limits, network isolation). +//! Agent permissions are auto-approved since the container is isolated. +//! +//! ```text +//! ┌──────────────────────────────────────────────┐ +//! │ Docker Container │ +//! │ │ +//! │ ironclaw acp-bridge --job-id │ +//! │ └─ spawns ACP agent subprocess │ +//! │ └─ ACP handshake (initialize + session) │ +//! │ └─ sends job description via prompt() │ +//! │ └─ translates ACP events → JobEventPayload │ +//! │ └─ POSTs events to orchestrator │ +//! │ └─ polls for follow-up prompts │ +//! └──────────────────────────────────────────────┘ +//! ``` + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; + +use agent_client_protocol::{self as acp, Agent as _}; +use serde_json::json; +use tokio::io::{AsyncBufReadExt, BufReader}; +use tokio::process::Command; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; +use uuid::Uuid; + +use crate::error::WorkerError; +use crate::worker::api::{CompletionReport, JobEventPayload, WorkerHttpClient}; + +/// Configuration for the ACP bridge runtime. +pub struct AcpBridgeConfig { + pub job_id: Uuid, + pub orchestrator_url: String, + pub timeout: Duration, + /// Command to spawn the ACP agent. + pub agent_command: String, + /// Arguments for the agent command. + pub agent_args: Vec, + /// Extra environment variables for the agent process. + pub agent_env: HashMap, +} + +/// The ACP bridge runtime. +pub struct AcpBridgeRuntime { + config: AcpBridgeConfig, + client: Arc, +} + +impl AcpBridgeRuntime { + /// Create a new bridge runtime. + /// + /// Reads `IRONCLAW_WORKER_TOKEN` from the environment for auth. + pub fn new(config: AcpBridgeConfig) -> Result { + let client = Arc::new(WorkerHttpClient::from_env( + config.orchestrator_url.clone(), + config.job_id, + )?); + + Ok(Self { config, client }) + } + + /// Run the bridge: fetch job, spawn ACP agent, stream events, handle follow-ups. + pub async fn run(&self) -> Result<(), WorkerError> { + // Fetch the job description from the orchestrator + let job = self.client.get_job().await?; + + tracing::info!( + job_id = %self.config.job_id, + "Starting ACP bridge for: {}", + truncate(&job.description, 100) + ); + + // Fetch credentials for injection into the spawned Command + let credentials = self.client.fetch_credentials().await?; + let mut extra_env = self.config.agent_env.clone(); + for cred in &credentials { + extra_env.insert(cred.env_var.clone(), cred.value.clone()); + } + if !credentials.is_empty() { + tracing::info!( + job_id = %self.config.job_id, + "Fetched {} credential(s) for child process injection", + credentials.len() + ); + } + + // Report that we're running + self.client + .report_status(&crate::worker::api::StatusUpdate { + state: "running".to_string(), + message: Some(format!("Spawning ACP agent: {}", self.config.agent_command)), + iteration: 0, + }) + .await?; + + // Run the ACP session + match self.run_acp_session(&job.description, &extra_env).await { + Ok(()) => { + self.client + .report_complete(&CompletionReport { + success: true, + message: Some("ACP agent session completed".to_string()), + iterations: 1, + }) + .await?; + } + Err(e) => { + tracing::error!(job_id = %self.config.job_id, "ACP session failed: {}", e); + self.client + .report_complete(&CompletionReport { + success: false, + message: Some(format!("ACP agent failed: {}", e)), + iterations: 1, + }) + .await?; + } + } + + Ok(()) + } + + /// Spawn the ACP agent and run the protocol lifecycle. + async fn run_acp_session( + &self, + prompt: &str, + extra_env: &HashMap, + ) -> Result<(), WorkerError> { + let mut cmd = Command::new(&self.config.agent_command); + cmd.args(&self.config.agent_args); + cmd.envs(extra_env); + cmd.current_dir("/workspace") + .stdin(std::process::Stdio::piped()) + .stdout(std::process::Stdio::piped()) + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); + + let mut child = cmd.spawn().map_err(|e| WorkerError::ExecutionFailed { + reason: format!( + "failed to spawn ACP agent '{}': {}", + self.config.agent_command, e + ), + })?; + + let child_stdin = child + .stdin + .take() + .ok_or_else(|| WorkerError::ExecutionFailed { + reason: "failed to capture ACP agent stdin".to_string(), + })?; + let child_stdout = child + .stdout + .take() + .ok_or_else(|| WorkerError::ExecutionFailed { + reason: "failed to capture ACP agent stdout".to_string(), + })?; + let child_stderr = child + .stderr + .take() + .ok_or_else(|| WorkerError::ExecutionFailed { + reason: "failed to capture ACP agent stderr".to_string(), + })?; + + // Spawn stderr reader that forwards lines as status events + let client_for_stderr = Arc::clone(&self.client); + let job_id = self.config.job_id; + let stderr_handle = tokio::spawn(async move { + let reader = BufReader::new(child_stderr); + let mut lines = reader.lines(); + while let Ok(Some(line)) = lines.next_line().await { + tracing::debug!(job_id = %job_id, "acp agent stderr: {}", line); + let payload = JobEventPayload { + event_type: "status".to_string(), + data: json!({ "message": line }), + }; + client_for_stderr.post_event(&payload).await; + } + }); + + // Run the ACP protocol inside a LocalSet (SDK futures are !Send) + let client_for_acp = Arc::clone(&self.client); + let prompt_owned = prompt.to_string(); + let job_id = self.config.job_id; + let timeout = self.config.timeout; + + // Clone client for follow-up loop + let client_for_followup = Arc::clone(&self.client); + + // Monitor the child process so the follow-up loop can exit if the agent dies. + // The oneshot is Send, so it crosses the LocalSet boundary cleanly. + let (child_exit_tx, child_exit_rx) = tokio::sync::oneshot::channel::>(); + tokio::spawn(async move { + let exit_code = match child.wait().await { + Ok(status) => status.code(), + Err(_) => None, + }; + let _ = child_exit_tx.send(exit_code); + }); + + let local_set = tokio::task::LocalSet::new(); + let acp_result = local_set + .run_until(async move { + let outgoing = child_stdin.compat_write(); + let incoming = child_stdout.compat(); + + // Create ACP connection + let ironclaw_client = IronClawAcpClient::new(Arc::clone(&client_for_acp)); + + let (conn, handle_io) = + acp::ClientSideConnection::new(ironclaw_client, outgoing, incoming, |fut| { + tokio::task::spawn_local(fut); + }); + tokio::task::spawn_local(handle_io); + + conn.initialize(ironclaw_init_request()) + .await + .map_err(|e| WorkerError::ExecutionFailed { + reason: format!("ACP initialize failed: {}", e), + })?; + + tracing::info!(job_id = %job_id, "ACP handshake complete"); + + // Create a new session + let workspace = std::env::current_dir().unwrap_or_else(|_| "/workspace".into()); + let session_response = conn + .new_session(acp::NewSessionRequest::new(workspace)) + .await + .map_err(|e| WorkerError::ExecutionFailed { + reason: format!("ACP new_session failed: {}", e), + })?; + + let session_id = session_response.session_id.clone(); + tracing::info!(job_id = %job_id, session_id = %session_id, "ACP session created"); + + // Send the job description as a prompt + let prompt_result = tokio::time::timeout(timeout, async { + conn.prompt(acp::PromptRequest::new( + session_id.clone(), + vec![prompt_owned.into()], + )) + .await + }) + .await; + + let prompt_response = match prompt_result { + Ok(Ok(resp)) => resp, + Ok(Err(e)) => { + return Err(WorkerError::ExecutionFailed { + reason: format!("ACP prompt failed: {}", e), + }); + } + Err(_) => { + return Err(WorkerError::ExecutionFailed { + reason: "ACP prompt timed out".to_string(), + }); + } + }; + + // Report prompt result + let result_payload = + stop_reason_to_result(&prompt_response.stop_reason, &session_id.to_string()); + client_for_acp.post_event(&result_payload).await; + + // Follow-up loop: poll for prompts, send additional prompt() calls. + // Exits when: orchestrator sends done, or agent process exits. + let mut child_exit_rx = child_exit_rx; + loop { + match client_for_followup.poll_prompt().await { + Ok(Some(follow_up)) => { + if follow_up.done { + tracing::info!(job_id = %job_id, "Orchestrator signaled done"); + break; + } + tracing::info!(job_id = %job_id, "Got follow-up prompt"); + + let follow_result = conn + .prompt(acp::PromptRequest::new( + session_id.clone(), + vec![follow_up.content.into()], + )) + .await; + + match follow_result { + Ok(resp) => { + let payload = stop_reason_to_result( + &resp.stop_reason, + &session_id.to_string(), + ); + client_for_followup.post_event(&payload).await; + } + Err(e) => { + tracing::error!( + job_id = %job_id, + "Follow-up prompt failed: {}", e + ); + client_for_followup + .post_event(&JobEventPayload { + event_type: "status".to_string(), + data: json!({ + "message": format!("Follow-up failed: {}", e), + }), + }) + .await; + } + } + } + Ok(None) => { + // No prompt available — wait, but also watch for agent exit. + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(2)) => {} + exit_code = &mut child_exit_rx => { + let code = exit_code.ok().flatten(); + tracing::info!( + job_id = %job_id, + exit_code = ?code, + "ACP agent process exited, ending follow-up loop" + ); + break; + } + } + } + Err(e) => { + tracing::warn!(job_id = %job_id, "Prompt polling error: {}", e); + tokio::select! { + _ = tokio::time::sleep(Duration::from_secs(5)) => {} + exit_code = &mut child_exit_rx => { + let code = exit_code.ok().flatten(); + tracing::info!( + job_id = %job_id, + exit_code = ?code, + "ACP agent process exited, ending follow-up loop" + ); + break; + } + } + } + } + } + + Ok::<(), WorkerError>(()) + }) + .await; + + // Wait for stderr reader to finish + let _ = stderr_handle.await; + + acp_result + } +} + +// ==================== ACP Client trait implementation ==================== + +/// Sink for ACP events translated from session notifications. +/// +/// The bridge posts events to the orchestrator via HTTP; the CLI test +/// command prints them to stdout. Both share the same `IronClawAcpClient`. +pub(crate) trait AcpEventSink: 'static { + fn emit_event(&self, payload: &JobEventPayload) -> impl std::future::Future; +} + +impl AcpEventSink for Arc { + async fn emit_event(&self, payload: &JobEventPayload) { + self.post_event(payload).await; + } +} + +/// IronClaw's implementation of the ACP Client trait. +/// +/// Handles callbacks from the agent: session notifications (streaming output) +/// and permission requests (auto-approved). Generic over the event sink so +/// both the container bridge and CLI test command can reuse it. +pub(crate) struct IronClawAcpClient { + sink: S, +} + +impl IronClawAcpClient { + pub(crate) fn new(sink: S) -> Self { + Self { sink } + } +} + +#[async_trait::async_trait(?Send)] +impl acp::Client for IronClawAcpClient { + async fn request_permission( + &self, + args: acp::RequestPermissionRequest, + ) -> acp::Result { + // Auto-approve by selecting the first option — Docker container is the + // security boundary, so we trust the agent to operate freely. + let Some(first_option) = args.options.first() else { + return Err(acp::Error::invalid_params()); + }; + let option_id = first_option.option_id.clone(); + Ok(acp::RequestPermissionResponse::new( + acp::RequestPermissionOutcome::Selected(acp::SelectedPermissionOutcome::new(option_id)), + )) + } + + async fn session_notification(&self, args: acp::SessionNotification) -> acp::Result<()> { + if let Some(payload) = session_update_to_payload(&args.update) { + self.sink.emit_event(&payload).await; + } + Ok(()) + } +} + +/// Build the standard IronClaw ACP initialization request. +pub(crate) fn ironclaw_init_request() -> acp::InitializeRequest { + acp::InitializeRequest::new(acp::ProtocolVersion::V1).client_info( + acp::Implementation::new("ironclaw", env!("CARGO_PKG_VERSION")).title("IronClaw"), + ) +} + +// ==================== Event translation ==================== + +/// Convert an ACP `SessionUpdate` into an IronClaw `JobEventPayload`. +fn session_update_to_payload(update: &acp::SessionUpdate) -> Option { + match update { + acp::SessionUpdate::AgentMessageChunk(chunk) => text_from_content_block(&chunk.content) + .map(|text| JobEventPayload { + event_type: "message".to_string(), + data: json!({ + "role": "assistant", + "content": text, + }), + }), + acp::SessionUpdate::AgentThoughtChunk(chunk) => text_from_content_block(&chunk.content) + .map(|text| JobEventPayload { + event_type: "status".to_string(), + data: json!({ + "message": text, + "type": "thought", + }), + }), + acp::SessionUpdate::ToolCall(tool_call) => Some(JobEventPayload { + event_type: "tool_use".to_string(), + data: json!({ + "tool_name": tool_call.title, + "tool_use_id": tool_call.tool_call_id.to_string(), + }), + }), + acp::SessionUpdate::ToolCallUpdate(update) => Some(JobEventPayload { + event_type: "tool_result".to_string(), + data: json!({ + "tool_use_id": update.tool_call_id.to_string(), + }), + }), + _ => Some(JobEventPayload { + event_type: "status".to_string(), + data: json!({ "message": "ACP session update" }), + }), + } +} + +/// Extract text from a `ContentBlock`, returning `None` for non-text blocks. +fn text_from_content_block(block: &acp::ContentBlock) -> Option<&str> { + match block { + acp::ContentBlock::Text(text_content) => Some(&text_content.text), + _ => None, + } +} + +/// Convert an ACP `StopReason` into a "result" `JobEventPayload`. +fn stop_reason_to_result(reason: &acp::StopReason, session_id: &str) -> JobEventPayload { + let (status, message) = match reason { + acp::StopReason::EndTurn => ("completed", "Agent completed successfully"), + acp::StopReason::MaxTokens => ("error", "Agent reached max tokens"), + acp::StopReason::MaxTurnRequests => ("error", "Agent reached max turn requests"), + acp::StopReason::Refusal => ("error", "Agent refused to continue"), + acp::StopReason::Cancelled => ("cancelled", "Agent was cancelled"), + _ => ("completed", "Agent finished"), + }; + JobEventPayload { + event_type: "result".to_string(), + data: json!({ + "status": status, + "session_id": session_id, + "message": message, + }), + } +} + +fn truncate(s: &str, max_len: usize) -> &str { + if s.len() <= max_len { + s + } else { + let mut end = max_len; + while end > 0 && !s.is_char_boundary(end) { + end -= 1; + } + &s[..end] + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_session_update_agent_message_text() { + let update = acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new( + acp::ContentBlock::Text(acp::TextContent::new("Hello world")), + )); + let payload = session_update_to_payload(&update).unwrap(); + assert_eq!(payload.event_type, "message"); + assert_eq!(payload.data["role"], "assistant"); + assert_eq!(payload.data["content"], "Hello world"); + } + + #[test] + fn test_session_update_agent_thought_text() { + let update = acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new( + acp::ContentBlock::Text(acp::TextContent::new("Thinking...")), + )); + let payload = session_update_to_payload(&update).unwrap(); + assert_eq!(payload.event_type, "status"); + assert_eq!(payload.data["type"], "thought"); + assert_eq!(payload.data["message"], "Thinking..."); + } + + #[test] + fn test_session_update_agent_message_image_ignored() { + let update = acp::SessionUpdate::AgentMessageChunk(acp::ContentChunk::new( + acp::ContentBlock::Image(acp::ImageContent::new("base64data", "image/png")), + )); + assert!(session_update_to_payload(&update).is_none()); + } + + #[test] + fn test_stop_reason_end_turn() { + let payload = stop_reason_to_result(&acp::StopReason::EndTurn, "sid-1"); + assert_eq!(payload.event_type, "result"); + assert_eq!(payload.data["status"], "completed"); + } + + #[test] + fn test_stop_reason_max_tokens() { + let payload = stop_reason_to_result(&acp::StopReason::MaxTokens, "sid-1"); + assert_eq!(payload.data["status"], "error"); + } + + #[test] + fn test_stop_reason_cancelled() { + let payload = stop_reason_to_result(&acp::StopReason::Cancelled, "sid-1"); + assert_eq!(payload.data["status"], "cancelled"); + } + + #[test] + fn test_stop_reason_refusal() { + let payload = stop_reason_to_result(&acp::StopReason::Refusal, "sid-1"); + assert_eq!(payload.data["status"], "error"); + assert_eq!(payload.data["message"], "Agent refused to continue"); + } + + #[test] + fn test_session_update_tool_call() { + let update = acp::SessionUpdate::ToolCall(acp::ToolCall::new("tc-1", "Running tests")); + let payload = session_update_to_payload(&update).unwrap(); + assert_eq!(payload.event_type, "tool_use"); + assert_eq!(payload.data["tool_name"], "Running tests"); + assert_eq!(payload.data["tool_use_id"], "tc-1"); + } + + #[test] + fn test_session_update_tool_call_update() { + let fields = acp::ToolCallUpdateFields::new(); + let update = acp::SessionUpdate::ToolCallUpdate(acp::ToolCallUpdate::new("tc-1", fields)); + let payload = session_update_to_payload(&update).unwrap(); + assert_eq!(payload.event_type, "tool_result"); + assert_eq!(payload.data["tool_use_id"], "tc-1"); + } + + #[test] + fn test_session_update_thought_image_ignored() { + let update = acp::SessionUpdate::AgentThoughtChunk(acp::ContentChunk::new( + acp::ContentBlock::Image(acp::ImageContent::new("data", "image/png")), + )); + assert!(session_update_to_payload(&update).is_none()); + } + + #[test] + fn test_stop_reason_max_turn_requests() { + let payload = stop_reason_to_result(&acp::StopReason::MaxTurnRequests, "sid-1"); + assert_eq!(payload.data["status"], "error"); + assert_eq!(payload.data["message"], "Agent reached max turn requests"); + } + + #[test] + fn test_stop_reason_includes_session_id() { + let payload = stop_reason_to_result(&acp::StopReason::EndTurn, "my-session-42"); + assert_eq!(payload.data["session_id"], "my-session-42"); + } + + #[test] + fn test_text_from_content_block_text() { + let block = acp::ContentBlock::Text(acp::TextContent::new("hello")); + assert_eq!(text_from_content_block(&block), Some("hello")); + } + + #[test] + fn test_text_from_content_block_image_returns_none() { + let block = acp::ContentBlock::Image(acp::ImageContent::new("data", "image/png")); + assert!(text_from_content_block(&block).is_none()); + } + + #[test] + fn test_truncate() { + assert_eq!(truncate("hello", 10), "hello"); + assert_eq!(truncate("hello world", 5), "hello"); + assert_eq!(truncate("", 5), ""); + } + + #[test] + fn test_truncate_multibyte_safe() { + // 2-byte UTF-8 char: "é" is 0xC3 0xA9 + let s = "café"; + assert_eq!(truncate(s, 3), "caf"); // doesn't split the é + assert_eq!(truncate(s, 5), "café"); // includes full char + } +} diff --git a/src/worker/mod.rs b/src/worker/mod.rs index dc6a2e8981..3a15ed13ae 100644 --- a/src/worker/mod.rs +++ b/src/worker/mod.rs @@ -24,6 +24,7 @@ //! └────────────────────────────────┘ //! ``` +pub mod acp_bridge; pub mod api; mod autonomous_recovery; pub mod claude_bridge; @@ -31,12 +32,17 @@ pub mod container; pub mod job; pub mod proxy_llm; +pub use acp_bridge::AcpBridgeRuntime; pub use api::WorkerHttpClient; pub use claude_bridge::ClaudeBridgeRuntime; pub use container::WorkerRuntime; pub use job::{Worker, WorkerDeps}; pub use proxy_llm::ProxyLlmProvider; +fn acp_bridge_timeout() -> std::time::Duration { + std::time::Duration::from_secs(crate::config::AcpModeConfig::from_env().timeout_secs) +} + /// Run the Worker subcommand (inside Docker containers). pub async fn run_worker( job_id: uuid::Uuid, @@ -64,6 +70,54 @@ pub async fn run_worker( .map_err(|e| anyhow::anyhow!("Worker failed: {}", e)) } +/// Run the ACP bridge subcommand (inside Docker containers). +pub async fn run_acp_bridge(job_id: uuid::Uuid, orchestrator_url: &str) -> anyhow::Result<()> { + let agent_command = std::env::var("ACP_AGENT_COMMAND").map_err(|_| { + anyhow::anyhow!("ACP_AGENT_COMMAND not set — cannot determine which agent to spawn") + })?; + + let agent_args: Vec = match std::env::var("ACP_AGENT_ARGS") { + Ok(s) => serde_json::from_str(&s).unwrap_or_else(|e| { + tracing::warn!("Failed to parse ACP_AGENT_ARGS as JSON: {e}, using empty args"); + Vec::new() + }), + Err(_) => Vec::new(), + }; + + let agent_env: std::collections::HashMap = match std::env::var("ACP_AGENT_ENV") + { + Ok(s) => serde_json::from_str(&s).unwrap_or_else(|e| { + tracing::warn!("Failed to parse ACP_AGENT_ENV as JSON: {e}, using empty env"); + std::collections::HashMap::new() + }), + Err(_) => std::collections::HashMap::new(), + }; + + tracing::info!( + "Starting ACP bridge for job {} (orchestrator: {}, agent: {} {})", + job_id, + orchestrator_url, + agent_command, + agent_args.join(" ") + ); + + let config = acp_bridge::AcpBridgeConfig { + job_id, + orchestrator_url: orchestrator_url.to_string(), + timeout: acp_bridge_timeout(), + agent_command, + agent_args, + agent_env, + }; + + let rt = AcpBridgeRuntime::new(config) + .map_err(|e| anyhow::anyhow!("ACP bridge init failed: {}", e))?; + + rt.run() + .await + .map_err(|e| anyhow::anyhow!("ACP bridge failed: {}", e)) +} + /// Run the Claude Code bridge subcommand (inside Docker containers). pub async fn run_claude_bridge( job_id: uuid::Uuid, @@ -94,3 +148,26 @@ pub async fn run_claude_bridge( .await .map_err(|e| anyhow::anyhow!("Claude bridge failed: {}", e)) } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn acp_bridge_timeout_defaults_to_1800_seconds() { + let _guard = crate::config::helpers::lock_env(); + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::remove_var("ACP_TIMEOUT_SECS") }; + assert_eq!(acp_bridge_timeout(), std::time::Duration::from_secs(1800)); + } + + #[test] + fn acp_bridge_timeout_respects_env_override() { + let _guard = crate::config::helpers::lock_env(); + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::set_var("ACP_TIMEOUT_SECS", "45") }; + assert_eq!(acp_bridge_timeout(), std::time::Duration::from_secs(45)); + // SAFETY: Under ENV_MUTEX, no concurrent env access. + unsafe { std::env::remove_var("ACP_TIMEOUT_SECS") }; + } +}