From 1c3edfe5f7feb53e59a516c260726dec910e5c78 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Mon, 11 May 2026 11:13:27 -0400 Subject: [PATCH] feat: route workflow stages to ACP backend --- Cargo.lock | 1 + lib/crates/fabro-acp/src/command.rs | 49 +- lib/crates/fabro-acp/src/error.rs | 5 +- lib/crates/fabro-acp/src/session.rs | 56 ++- lib/crates/fabro-acp/src/transport.rs | 20 +- lib/crates/fabro-sandbox/src/docker.rs | 35 +- lib/crates/fabro-sandbox/src/lib.rs | 5 +- lib/crates/fabro-sandbox/src/local.rs | 6 +- lib/crates/fabro-sandbox/src/read_guard.rs | 7 +- lib/crates/fabro-types/src/graph.rs | 5 + .../fabro-validate/src/rules/backend_valid.rs | 84 ++++ lib/crates/fabro-validate/src/rules/mod.rs | 2 + lib/crates/fabro-workflow/Cargo.toml | 1 + .../fabro-workflow/src/handler/agent.rs | 2 + .../fabro-workflow/src/handler/llm/acp.rs | 473 ++++++++++++++++++ .../fabro-workflow/src/handler/llm/api.rs | 2 + .../src/handler/llm/changed_files.rs | 75 +++ .../fabro-workflow/src/handler/llm/cli.rs | 469 ++++++++++++----- .../fabro-workflow/src/handler/llm/mod.rs | 4 + .../src/handler/llm/node_runtime.rs | 39 ++ .../fabro-workflow/src/handler/prompt.rs | 8 + .../fabro-workflow/src/pipeline/initialize.rs | 19 +- .../fabro-workflow/src/transforms/import.rs | 17 +- .../fabro-workflow/tests/it/integration.rs | 21 +- 24 files changed, 1178 insertions(+), 227 deletions(-) create mode 100644 lib/crates/fabro-validate/src/rules/backend_valid.rs create mode 100644 lib/crates/fabro-workflow/src/handler/llm/acp.rs create mode 100644 lib/crates/fabro-workflow/src/handler/llm/changed_files.rs create mode 100644 lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs diff --git a/Cargo.lock b/Cargo.lock index 92854b102..f7fc18003 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2480,6 +2480,7 @@ dependencies = [ "bytes", "chrono", "dirs", + "fabro-acp", "fabro-agent", "fabro-auth", "fabro-checkpoint", diff --git a/lib/crates/fabro-acp/src/command.rs b/lib/crates/fabro-acp/src/command.rs index 9fb85cfb9..834c0daab 100644 --- a/lib/crates/fabro-acp/src/command.rs +++ b/lib/crates/fabro-acp/src/command.rs @@ -1,8 +1,6 @@ -use std::{ - collections::HashMap, - path::{Path, PathBuf}, - str::FromStr, -}; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use std::str::FromStr; use agent_client_protocol::schema::McpServer; use agent_client_protocol_tokio::AcpAgent; @@ -12,8 +10,8 @@ use fabro_model::Provider; pub struct AcpCommand { display: String, program: PathBuf, - args: Vec, - env: HashMap, + args: Vec, + env: HashMap, } impl AcpCommand { @@ -72,26 +70,33 @@ impl From for AcpCommandError { #[must_use] pub fn default_acp_command(provider: Provider) -> AcpCommand { match provider { - Provider::Anthropic => command_from_parts( - "npx -y @zed-industries/claude-code-acp@latest", - "npx", - ["-y", "@zed-industries/claude-code-acp@latest"], - ), + Provider::Anthropic => { + command_from_parts("npx -y @zed-industries/claude-code-acp@latest", "npx", [ + "-y", + "@zed-industries/claude-code-acp@latest", + ]) + } Provider::Gemini => command_from_parts( "npx -y -- @google/gemini-cli@latest --experimental-acp", "npx", - ["-y", "--", "@google/gemini-cli@latest", "--experimental-acp"], + [ + "-y", + "--", + "@google/gemini-cli@latest", + "--experimental-acp", + ], ), Provider::OpenAi | Provider::Kimi | Provider::Zai | Provider::Minimax | Provider::Inception - | Provider::OpenAiCompatible => command_from_parts( - "npx -y @zed-industries/codex-acp@latest", - "npx", - ["-y", "@zed-industries/codex-acp@latest"], - ), + | Provider::OpenAiCompatible => { + command_from_parts("npx -y @zed-industries/codex-acp@latest", "npx", [ + "-y", + "@zed-industries/codex-acp@latest", + ]) + } } } @@ -121,8 +126,8 @@ fn parse_acp_command(raw: &str) -> Result { Ok(AcpCommand { display: raw.to_string(), program: stdio.command, - args: stdio.args, - env: stdio + args: stdio.args, + env: stdio .env .into_iter() .map(|env| (env.name, env.value)) @@ -154,8 +159,8 @@ fn command_from_parts( AcpCommand { display: display.into(), program: program.into(), - args: args.into_iter().map(str::to_string).collect(), - env: HashMap::new(), + args: args.into_iter().map(str::to_string).collect(), + env: HashMap::new(), } } diff --git a/lib/crates/fabro-acp/src/error.rs b/lib/crates/fabro-acp/src/error.rs index d7865d42e..79304d654 100644 --- a/lib/crates/fabro-acp/src/error.rs +++ b/lib/crates/fabro-acp/src/error.rs @@ -19,7 +19,10 @@ pub enum AcpError { ProcessExited { stderr: String }, #[error("ACP prompt stopped with {stop_reason}: {text}")] - StopReason { stop_reason: String, text: String }, + StopReason { + stop_reason: String, + text: String, + }, } impl From for AcpError { diff --git a/lib/crates/fabro-acp/src/session.rs b/lib/crates/fabro-acp/src/session.rs index d6f9fd70c..415e1c80b 100644 --- a/lib/crates/fabro-acp/src/session.rs +++ b/lib/crates/fabro-acp/src/session.rs @@ -1,10 +1,11 @@ -use std::{collections::HashMap, sync::Arc, time::Duration}; +use std::collections::HashMap; +use std::sync::Arc; +use std::time::Duration; use agent_client_protocol::schema::{ CancelNotification, ContentBlock, ContentChunk, InitializeRequest, PermissionOptionKind, - ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, - RequestPermissionResponse, SelectedPermissionOutcome, SessionNotification, SessionUpdate, - StopReason, + ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse, + SelectedPermissionOutcome, SessionNotification, SessionUpdate, StopReason, }; use agent_client_protocol::util::MatchDispatch; use agent_client_protocol::{ActiveSession, Agent, Client, SessionMessage}; @@ -17,21 +18,21 @@ use crate::error::AcpError; use crate::transport::{SandboxAcpTransport, TransportState}; pub struct AcpRunRequest { - pub command: AcpCommand, - pub prompt: String, - pub cwd: String, - pub timeout_ms: Option, - pub env: HashMap, - pub sandbox: Arc, + pub command: AcpCommand, + pub prompt: String, + pub cwd: String, + pub timeout_ms: Option, + pub env: HashMap, + pub sandbox: Arc, pub cancel_token: CancellationToken, - pub on_activity: Option>, + pub on_activity: Option>, } #[derive(Debug)] pub struct AcpRunResult { - pub text: String, + pub text: String, pub stop_reason: StopReason, - pub stderr: String, + pub stderr: String, pub duration_ms: u64, } @@ -76,22 +77,29 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result match tokio::time::timeout(Duration::from_millis(timeout_ms), run).await - { - Ok(result) => result, - Err(_) => { - state.terminate().await?; - return Err(AcpError::TimedOut { - stderr: state.stderr_tail().await, - }); + Some(timeout_ms) => { + match tokio::time::timeout(Duration::from_millis(timeout_ms), run).await { + Ok(result) => result, + Err(_) => { + state.terminate().await?; + return Err(AcpError::TimedOut { + stderr: state.stderr_tail().await, + }); + } } - }, + } None => run.await, }; let (text, stop_reason) = outcome.map_err(map_protocol_error)?; @@ -115,7 +123,7 @@ fn map_protocol_error(error: agent_client_protocol::Error) -> AcpError { .map_or((rest, ""), |(stop_reason, text)| (stop_reason, text)); AcpError::StopReason { stop_reason: stop_reason.to_string(), - text: text.trim_end_matches('"').to_string(), + text: text.trim_end_matches('"').to_string(), } } else { AcpError::Protocol(error) diff --git a/lib/crates/fabro-acp/src/transport.rs b/lib/crates/fabro-acp/src/transport.rs index a934134e4..81b4a8ad6 100644 --- a/lib/crates/fabro-acp/src/transport.rs +++ b/lib/crates/fabro-acp/src/transport.rs @@ -1,4 +1,6 @@ -use std::{collections::HashMap, pin::Pin, sync::Arc}; +use std::collections::HashMap; +use std::pin::Pin; +use std::sync::Arc; use agent_client_protocol::{Client, ConnectTo, Lines}; use fabro_sandbox::{Sandbox, StderrCollector, StdioProcessHandle}; @@ -43,12 +45,12 @@ impl TransportState { } pub(crate) struct SandboxAcpTransport { - command: AcpCommand, - cwd: String, - env: HashMap, - sandbox: Arc, + command: AcpCommand, + cwd: String, + env: HashMap, + sandbox: Arc, cancel_token: CancellationToken, - state: TransportState, + state: TransportState, } impl SandboxAcpTransport { @@ -106,8 +108,10 @@ impl ConnectTo for SandboxAcpTransport { }, )); - let protocol = - agent_client_protocol::ConnectTo::::connect_to(Lines::new(outgoing_sink, incoming_lines), client); + let protocol = agent_client_protocol::ConnectTo::::connect_to( + Lines::new(outgoing_sink, incoming_lines), + client, + ); tokio::select! { result = protocol => { let _ = tokio::time::timeout(std::time::Duration::from_millis(500), handle.wait()).await; diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index df5898d92..af39b82e5 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -29,8 +29,8 @@ use crate::redact::redact_auth_url; use crate::sandbox::{optional_timeout, resolve_path}; use crate::{ CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, - ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, - StderrCollector, StdioProcess, StdioProcessHandle, format_lines_numbered, shell_quote, + ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector, + StdioProcess, StdioProcessHandle, format_lines_numbered, shell_quote, }; const WORKING_DIRECTORY: &str = "/workspace"; @@ -800,12 +800,12 @@ fn docker_stdio_exec_options( ) -> (CreateExecOptions, StartExecOptions) { ( CreateExecOptions { - attach_stdin: Some(true), + attach_stdin: Some(true), attach_stdout: Some(true), attach_stderr: Some(true), - tty: Some(false), - cmd: Some(vec!["/bin/bash".to_string(), "-lc".to_string(), command]), - working_dir: Some(working_dir), + tty: Some(false), + cmd: Some(vec!["/bin/bash".to_string(), "-lc".to_string(), command]), + working_dir: Some(working_dir), env, ..Default::default() }, @@ -824,11 +824,7 @@ async fn request_docker_exec_stop_with( ) -> crate::Result<()> { let command = format!("touch {}", shell_quote(stop_file)); let exec_opts = CreateExecOptions { - cmd: Some(vec![ - "/bin/bash".to_string(), - "-lc".to_string(), - command, - ]), + cmd: Some(vec!["/bin/bash".to_string(), "-lc".to_string(), command]), attach_stdout: Some(true), attach_stderr: Some(true), working_dir: Some("/".to_string()), @@ -855,7 +851,12 @@ async fn request_docker_exec_stop_with( stderr.push_str(&String::from_utf8_lossy(&message)); } Ok(_) => {} - Err(e) => return Err(crate::Error::context("Error reading stop request output", e)), + Err(e) => { + return Err(crate::Error::context( + "Error reading stop request output", + e, + )); + } } } } @@ -877,11 +878,11 @@ async fn request_docker_exec_stop_with( } struct DockerStdioProcessControl { - docker: Docker, + docker: Docker, container_id: String, - exec_id: String, - stop_file: String, - termination: tokio::sync::Mutex>, + exec_id: String, + stop_file: String, + termination: tokio::sync::Mutex>, } #[async_trait] @@ -1524,7 +1525,7 @@ impl Sandbox for DockerSandbox { } Ok(StdioProcess { - stdin: input, + stdin: input, stdout: Box::pin(stdout_reader), stderr: stderr_collector, handle, diff --git a/lib/crates/fabro-sandbox/src/lib.rs b/lib/crates/fabro-sandbox/src/lib.rs index 9d3a419e1..fb2e98753 100644 --- a/lib/crates/fabro-sandbox/src/lib.rs +++ b/lib/crates/fabro-sandbox/src/lib.rs @@ -41,9 +41,8 @@ pub use reconnect::{reconnect, reconnect_for_run, reconnect_for_run_with_callbac pub use sandbox::{ CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, ExecStreamingResult, GitRunInfo, GitSetupIntent, GrepOptions, Sandbox, SandboxEvent, - SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, - format_lines_numbered, git_push_via_exec, redacted_output_tail, setup_git_via_exec, - shell_quote, + SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle, format_lines_numbered, + git_push_via_exec, redacted_output_tail, setup_git_via_exec, shell_quote, }; pub use sandbox_spec::SandboxSpec; pub use terminal::{TerminalSession, TerminalSize, open_terminal_for_run}; diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index bba0c041d..a373442a2 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -14,8 +14,8 @@ use tokio_util::sync::CancellationToken; use crate::sandbox::optional_timeout; use crate::{ CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, - ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, - StderrCollector, StdioProcess, StdioProcessHandle, format_lines_numbered, + ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector, + StdioProcess, StdioProcessHandle, format_lines_numbered, }; pub struct LocalSandbox { @@ -530,7 +530,7 @@ impl Sandbox for LocalSandbox { } Ok(StdioProcess { - stdin: Box::pin(stdin), + stdin: Box::pin(stdin), stdout: Box::pin(stdout), stderr: stderr_collector, handle, diff --git a/lib/crates/fabro-sandbox/src/read_guard.rs b/lib/crates/fabro-sandbox/src/read_guard.rs index 34669801c..588bd960d 100644 --- a/lib/crates/fabro-sandbox/src/read_guard.rs +++ b/lib/crates/fabro-sandbox/src/read_guard.rs @@ -291,9 +291,8 @@ mod tests { *mock.captured_command.lock().unwrap(), Some("python fake_agent.py".to_string()) ); - assert_eq!( - *mock.captured_working_dirs.lock().unwrap(), - vec![Some("/work/sub".to_string())] - ); + assert_eq!(*mock.captured_working_dirs.lock().unwrap(), vec![Some( + "/work/sub".to_string() + )]); } } diff --git a/lib/crates/fabro-types/src/graph.rs b/lib/crates/fabro-types/src/graph.rs index 90e9307f4..ba43af5e3 100644 --- a/lib/crates/fabro-types/src/graph.rs +++ b/lib/crates/fabro-types/src/graph.rs @@ -266,6 +266,11 @@ impl Node { self.str_attr("backend") } + #[must_use] + pub fn acp_command(&self) -> Option<&str> { + self.str_attr("acp_command") + } + #[must_use] pub fn selection(&self) -> &str { self.str_attr("selection").unwrap_or("deterministic") diff --git a/lib/crates/fabro-validate/src/rules/backend_valid.rs b/lib/crates/fabro-validate/src/rules/backend_valid.rs new file mode 100644 index 000000000..1ca5d9cfa --- /dev/null +++ b/lib/crates/fabro-validate/src/rules/backend_valid.rs @@ -0,0 +1,84 @@ +use fabro_graphviz::graph::{AttrValue, Graph}; + +use crate::{Diagnostic, LintRule, Severity}; + +pub(super) fn rule() -> Box { + Box::new(Rule) +} + +struct Rule; + +const VALID_BACKENDS: &[&str] = &["api", "cli", "acp"]; + +impl LintRule for Rule { + fn name(&self) -> &'static str { + "backend_valid" + } + + fn apply(&self, graph: &Graph) -> Vec { + let mut diagnostics = Vec::new(); + for node in graph.nodes.values() { + if let Some(backend) = node.attrs.get("backend").and_then(AttrValue::as_str) { + if !VALID_BACKENDS.contains(&backend) { + diagnostics.push(Diagnostic { + rule: self.name().to_string(), + severity: Severity::Error, + message: format!( + "unsupported LLM backend \"{backend}\"; expected one of: api, cli, acp" + ), + node_id: Some(node.id.clone()), + edge: None, + fix: Some("Use one of: api, cli, acp".to_string()), + }); + } + } + } + diagnostics + } +} + +#[cfg(test)] +mod tests { + use fabro_graphviz::graph::{AttrValue, Node}; + + use super::Rule; + use crate::rules::test_support::minimal_graph; + use crate::{LintRule, Severity}; + + #[test] + fn backend_valid_accepts_absent_api_cli_and_acp() { + for backend in [None, Some("api"), Some("cli"), Some("acp")] { + let mut graph = minimal_graph(); + let mut node = Node::new("work"); + if let Some(backend) = backend { + node.attrs.insert( + "backend".to_string(), + AttrValue::String(backend.to_string()), + ); + } + graph.nodes.insert("work".to_string(), node); + + assert!(Rule.apply(&graph).is_empty(), "backend: {backend:?}"); + } + } + + #[test] + fn backend_valid_rejects_unknown_backend() { + let mut graph = minimal_graph(); + let mut node = Node::new("work"); + node.attrs.insert( + "backend".to_string(), + AttrValue::String("codex".to_string()), + ); + graph.nodes.insert("work".to_string(), node); + + let diagnostics = Rule.apply(&graph); + assert_eq!(diagnostics.len(), 1); + assert_eq!(diagnostics[0].severity, Severity::Error); + assert!( + diagnostics[0] + .message + .contains("unsupported LLM backend \"codex\"; expected one of: api, cli, acp") + ); + } +} diff --git a/lib/crates/fabro-validate/src/rules/mod.rs b/lib/crates/fabro-validate/src/rules/mod.rs index 1d01da29f..208fe0bcf 100644 --- a/lib/crates/fabro-validate/src/rules/mod.rs +++ b/lib/crates/fabro-validate/src/rules/mod.rs @@ -1,4 +1,5 @@ mod all_conditional_edges; +mod backend_valid; mod condition_syntax; mod direction_valid; mod edge_target_exists; @@ -43,6 +44,7 @@ pub fn built_in_rules() -> Vec> { condition_syntax::rule(), stylesheet_syntax::rule(), type_known::rule(), + backend_valid::rule(), fidelity_valid::rule(), retry_target_exists::rule(), goal_gate_has_retry::rule(), diff --git a/lib/crates/fabro-workflow/Cargo.toml b/lib/crates/fabro-workflow/Cargo.toml index 91dd3dac0..406c12391 100644 --- a/lib/crates/fabro-workflow/Cargo.toml +++ b/lib/crates/fabro-workflow/Cargo.toml @@ -19,6 +19,7 @@ workspace = true [dependencies] anyhow.workspace = true fabro-auth = { path = "../fabro-auth" } +fabro-acp = { path = "../fabro-acp" } fabro-agent = { path = "../fabro-agent" } fabro-config = { path = "../fabro-config" } fabro-graphviz = { path = "../fabro-graphviz" } diff --git a/lib/crates/fabro-workflow/src/handler/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index 1be5cb25e..936d89d79 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -56,6 +56,8 @@ pub trait CodergenBackend: Send + Sync { _system_prompt: Option<&str>, _emitter: &Arc, _stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, ) -> Result { Err(Error::Validation( "one_shot mode not supported by this backend".into(), diff --git a/lib/crates/fabro-workflow/src/handler/llm/acp.rs b/lib/crates/fabro-workflow/src/handler/llm/acp.rs new file mode 100644 index 000000000..48e2b09bc --- /dev/null +++ b/lib/crates/fabro-workflow/src/handler/llm/acp.rs @@ -0,0 +1,473 @@ +//! Workflow adapter for ACP-backed LLM stages. + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; +use fabro_acp::{AcpError, AcpRunRequest, default_acp_command, resolve_acp_command}; +use fabro_agent::{Sandbox, StaticEnvProvider, ToolEnvProvider}; +use fabro_auth::{CliAgentKind, CredentialResolver, CredentialUsage, ResolvedCredential}; +use fabro_graphviz::graph::Node; +use fabro_model::Provider; +use tokio_util::sync::CancellationToken; + +use super::super::agent::{CodergenBackend, CodergenResult}; +use super::cli::{AgentCli, process_env_var}; +use super::{changed_files, node_runtime}; +use crate::context::Context; +use crate::error::Error; +use crate::event::{Emitter, RunNoticeCode, RunNoticeLevel, StageScope}; + +pub struct AgentAcpBackend { + model: String, + provider: Provider, + tool_env: Option>, + github_token_refresh_managed: bool, + resolver: Option, +} + +impl AgentAcpBackend { + #[must_use] + pub fn new(model: String, provider: Provider, resolver: CredentialResolver) -> Self { + Self { + model, + provider, + tool_env: None, + github_token_refresh_managed: false, + resolver: Some(resolver), + } + } + + #[must_use] + pub fn new_from_env(model: String, provider: Provider) -> Self { + Self { + model, + provider, + tool_env: None, + github_token_refresh_managed: false, + resolver: None, + } + } + + #[must_use] + pub fn with_env(mut self, env: HashMap) -> Self { + self.tool_env = Some(Arc::new(StaticEnvProvider(env))); + self + } + + #[must_use] + pub fn with_tool_env_provider( + mut self, + provider: Arc, + github_token_refresh_managed: bool, + ) -> Self { + self.tool_env = Some(provider); + self.github_token_refresh_managed = github_token_refresh_managed; + self + } + + async fn run_turn( + &self, + node: &Node, + prompt: String, + emitter: &Arc, + sandbox: &Arc, + cancel_token: CancellationToken, + ) -> Result { + let files_before = changed_files::detect_changed_files(sandbox).await; + let _model = node.model().unwrap_or(&self.model); + let provider = node + .provider() + .and_then(|value| value.parse::().ok()) + .unwrap_or(self.provider); + let explicit_command = node.acp_command(); + let command = resolve_acp_command(provider, explicit_command) + .map_err(|err| Error::handler_with_source("Failed to resolve ACP command", &err))?; + + if explicit_command.is_none() + && command.program() == default_acp_command(provider).program() + { + node_runtime::ensure_node_runtime(sandbox, &cancel_token).await?; + } + + let launch_env = self + .launch_env(provider, emitter, sandbox, &cancel_token) + .await?; + let on_activity = { + let emitter = Arc::clone(emitter); + Arc::new(move || emitter.touch()) as Arc + }; + + let result = fabro_acp::run_acp_turn(AcpRunRequest { + command, + prompt, + cwd: sandbox.working_directory().to_string(), + timeout_ms: node.timeout().map(crate::millis_u64), + env: launch_env, + sandbox: Arc::clone(sandbox), + cancel_token: cancel_token.child_token(), + on_activity: Some(on_activity), + }) + .await + .map_err(acp_error_to_workflow)?; + + let (files_touched, last_file_touched) = + changed_files::files_touched_since(sandbox, &files_before).await; + + Ok(CodergenResult::Text { + text: result.text, + usage: None, + files_touched, + last_file_touched, + }) + } + + async fn launch_env( + &self, + provider: Provider, + emitter: &Arc, + sandbox: &Arc, + cancel_token: &CancellationToken, + ) -> Result, Error> { + let cli_agent = match AgentCli::for_provider(provider) { + AgentCli::Claude => CliAgentKind::Claude, + AgentCli::Codex => CliAgentKind::Codex, + AgentCli::Gemini => CliAgentKind::Gemini, + }; + let mut launch_env = if let Some(resolver) = &self.resolver { + let resolved = resolver + .resolve(provider, CredentialUsage::CliAgent(cli_agent)) + .await + .map_err(|err| { + Error::handler_with_source("Failed to resolve ACP credential", &err) + })?; + let ResolvedCredential::Cli(cli_credential) = resolved else { + return Err(Error::handler("Expected CLI credential".to_string())); + }; + if let Some(login_cmd) = &cli_credential.login_command { + let login_result = sandbox + .exec_command( + login_cmd, + 30_000, + None, + None, + Some(cancel_token.child_token()), + ) + .await + .map_err(|err| { + Error::handler_with_source("ACP credential login failed", &err) + })?; + if !login_result.is_success() { + tracing::warn!( + exit_code = login_result.display_exit_code(), + "ACP credential login failed: {}", + login_result.stderr + ); + } + } + cli_credential.env_vars + } else { + let mut env = HashMap::new(); + for name in provider.api_key_env_vars() { + if let Some(value) = process_env_var(name) { + env.insert((*name).to_string(), value); + } + } + env + }; + + if let Some(provider) = &self.tool_env { + if self.github_token_refresh_managed { + emitter.notice( + RunNoticeLevel::Info, + RunNoticeCode::GithubTokenRefreshLimited, + "ACP agent stages receive GitHub tokens at process launch; stages running \ + beyond token expiry may need to be retried.", + ); + } + let tool_env = provider.resolve().await.map_err(|err| { + Error::handler_with_anyhow("Failed to resolve ACP agent env", &err) + })?; + launch_env.extend(tool_env); + } + + Ok(launch_env) + } +} + +#[async_trait] +impl CodergenBackend for AgentAcpBackend { + async fn run( + &self, + node: &Node, + prompt: &str, + _context: &Context, + _thread_id: Option<&str>, + emitter: &Arc, + sandbox: &Arc, + _tool_hooks: Option>, + cancel_token: CancellationToken, + ) -> Result { + self.run_turn(node, prompt.to_string(), emitter, sandbox, cancel_token) + .await + } + + async fn one_shot( + &self, + node: &Node, + prompt: &str, + system_prompt: Option<&str>, + emitter: &Arc, + _stage_scope: &StageScope, + sandbox: &Arc, + cancel_token: CancellationToken, + ) -> Result { + let prompt = match system_prompt.filter(|prompt| !prompt.is_empty()) { + Some(system_prompt) => format!("System:\n{system_prompt}\n\nUser:\n{prompt}"), + None => prompt.to_string(), + }; + self.run_turn(node, prompt, emitter, sandbox, cancel_token) + .await + } +} + +fn acp_error_to_workflow(error: AcpError) -> Error { + match error { + AcpError::Cancelled => Error::Cancelled, + AcpError::TimedOut { stderr } => { + if stderr.is_empty() { + Error::handler("ACP turn timed out") + } else { + Error::handler(format!("ACP turn timed out: {stderr}")) + } + } + AcpError::StopReason { stop_reason, text } => { + Error::handler(format!("ACP prompt stopped with {stop_reason}: {text}")) + } + other => Error::handler_with_source("ACP turn failed", &other), + } +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::sync::Arc; + + use fabro_agent::{LocalSandbox, Sandbox, shell_quote}; + use fabro_graphviz::graph::{AttrValue, Node}; + use fabro_model::Provider; + use tokio_util::sync::CancellationToken; + + use super::AgentAcpBackend; + use crate::context::Context; + use crate::event::{Emitter, StageScope}; + use crate::handler::agent::{CodergenBackend, CodergenResult}; + + #[tokio::test] + async fn acp_backend_run_sends_prompt_and_returns_text() { + let tempdir = tempfile::tempdir().unwrap(); + init_git(tempdir.path()); + let script_path = tempdir.path().join("fake_acp_agent.py"); + tokio::fs::write(&script_path, fake_agent_script()) + .await + .unwrap(); + + let mut node = Node::new("work"); + node.attrs.insert( + "provider".to_string(), + AttrValue::String("openai".to_string()), + ); + node.attrs.insert( + "model".to_string(), + AttrValue::String("fake-acp".to_string()), + ); + node.attrs + .insert("backend".to_string(), AttrValue::String("acp".to_string())); + node.attrs.insert( + "acp_command".to_string(), + AttrValue::String(format!( + "python3 {}", + shell_quote(&script_path.to_string_lossy()) + )), + ); + + let backend = AgentAcpBackend::new_from_env("fake-acp".to_string(), Provider::OpenAi); + let sandbox: Arc = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf())); + let result = backend + .run( + &node, + "write hello", + &Context::new(), + None, + &Arc::new(Emitter::default()), + &sandbox, + None, + CancellationToken::new(), + ) + .await + .unwrap(); + + let CodergenResult::Text { + text, + files_touched, + .. + } = result + else { + panic!("expected text result"); + }; + assert_eq!(text, "hello from acp"); + assert_eq!(files_touched, vec!["hello.txt"]); + } + + #[tokio::test] + async fn acp_backend_one_shot_combines_system_prompt_and_uses_passed_sandbox() { + let tempdir = tempfile::tempdir().unwrap(); + let script_path = tempdir.path().join("fake_acp_agent.py"); + let prompt_record_path = tempdir.path().join("prompt.json"); + tokio::fs::write(&script_path, fake_agent_script()) + .await + .unwrap(); + + let mut node = Node::new("prompt"); + node.attrs.insert( + "provider".to_string(), + AttrValue::String("openai".to_string()), + ); + node.attrs + .insert("backend".to_string(), AttrValue::String("acp".to_string())); + node.attrs.insert( + "acp_command".to_string(), + AttrValue::String(format!( + "python3 {}", + shell_quote(&script_path.to_string_lossy()) + )), + ); + + let backend = AgentAcpBackend::new_from_env("fake-acp".to_string(), Provider::OpenAi) + .with_env(HashMap::from([( + "ACP_PROMPT_RECORD".to_string(), + prompt_record_path.to_string_lossy().into_owned(), + )])); + let sandbox: Arc = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf())); + let result = backend + .one_shot( + &node, + "User prompt", + Some("System prompt"), + &Arc::new(Emitter::default()), + &StageScope::for_handler(&Context::new(), "prompt"), + &sandbox, + CancellationToken::new(), + ) + .await + .unwrap(); + + assert!(matches!(result, CodergenResult::Text { .. })); + let recorded = tokio::fs::read_to_string(prompt_record_path).await.unwrap(); + assert!(recorded.contains("System:\\nSystem prompt\\n\\nUser:\\nUser prompt")); + assert_eq!( + tokio::fs::read_to_string(tempdir.path().join("hello.txt")) + .await + .unwrap(), + "hello from sandbox\n" + ); + } + + #[tokio::test] + async fn acp_backend_cancelled_stop_reason_maps_to_cancelled_error() { + let tempdir = tempfile::tempdir().unwrap(); + let script_path = tempdir.path().join("fake_acp_agent.py"); + tokio::fs::write(&script_path, fake_agent_script()) + .await + .unwrap(); + + let mut node = Node::new("work"); + node.attrs.insert( + "provider".to_string(), + AttrValue::String("openai".to_string()), + ); + node.attrs.insert( + "acp_command".to_string(), + AttrValue::String(format!( + "python3 {}", + shell_quote(&script_path.to_string_lossy()) + )), + ); + + let backend = + AgentAcpBackend::new_from_env("fake-acp".to_string(), Provider::OpenAi).with_env( + HashMap::from([("ACP_STOP_REASON".to_string(), "cancelled".to_string())]), + ); + let sandbox: Arc = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf())); + let result = backend + .run( + &node, + "cancel", + &Context::new(), + None, + &Arc::new(Emitter::default()), + &sandbox, + None, + CancellationToken::new(), + ) + .await; + let Err(err) = result else { + panic!("expected cancellation error"); + }; + + assert!(matches!(err, crate::error::Error::Cancelled)); + } + + fn fake_agent_script() -> &'static str { + r#" +import json +import os +import sys + +session_id = "sess-1" + +def send(message): + print(json.dumps(message), flush=True) + +def respond(message, result): + send({"jsonrpc": "2.0", "id": message["id"], "result": result}) + +for line in sys.stdin: + message = json.loads(line) + method = message.get("method") + if method == "initialize": + respond(message, {"protocolVersion": 1, "agentCapabilities": {}}) + elif method == "session/new": + respond(message, {"sessionId": session_id}) + elif method == "session/prompt": + if os.environ.get("ACP_PROMPT_RECORD"): + with open(os.environ["ACP_PROMPT_RECORD"], "w", encoding="utf-8") as record: + record.write(json.dumps(message.get("params", {}))) + with open("hello.txt", "w", encoding="utf-8") as file: + file.write("hello from sandbox\n") + for text in ["hello ", "from acp"]: + send({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": session_id, + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": text} + } + } + }) + respond(message, {"stopReason": os.environ.get("ACP_STOP_REASON", "end_turn")}) + break +"# + } + + fn init_git(path: &std::path::Path) { + let output = std::process::Command::new("git") + .arg("init") + .current_dir(path) + .output() + .unwrap(); + assert!(output.status.success()); + } +} diff --git a/lib/crates/fabro-workflow/src/handler/llm/api.rs b/lib/crates/fabro-workflow/src/handler/llm/api.rs index a65093d85..4c65fe462 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/api.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/api.rs @@ -483,6 +483,8 @@ impl CodergenBackend for AgentApiBackend { system_prompt: Option<&str>, emitter: &Arc, stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, ) -> Result { let client = Client::from_source(self.source.as_ref()) .await diff --git a/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs b/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs new file mode 100644 index 000000000..d5e302515 --- /dev/null +++ b/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs @@ -0,0 +1,75 @@ +use std::sync::Arc; + +use fabro_agent::{Sandbox, shell_quote}; + +pub async fn detect_changed_files(sandbox: &Arc) -> Vec { + let diff_result = sandbox + .exec_command("git diff --name-only", 30_000, None, None, None) + .await; + let untracked_result = sandbox + .exec_command( + "git ls-files --others --exclude-standard", + 30_000, + None, + None, + None, + ) + .await; + + let mut files: Vec = Vec::new(); + if let Ok(result) = diff_result { + if result.is_success() { + files.extend( + result + .stdout + .lines() + .filter(|line| !line.trim().is_empty()) + .map(String::from), + ); + } + } + if let Ok(result) = untracked_result { + if result.is_success() { + files.extend( + result + .stdout + .lines() + .filter(|line| !line.trim().is_empty()) + .map(String::from), + ); + } + } + + files.sort(); + files.dedup(); + files +} + +pub async fn files_touched_since( + sandbox: &Arc, + files_before: &[String], +) -> (Vec, Option) { + let files_after = detect_changed_files(sandbox).await; + let files_touched: Vec = files_after + .into_iter() + .filter(|file| !files_before.contains(file)) + .collect(); + + let last_file_touched = if files_touched.is_empty() { + None + } else { + let quoted_files: Vec = + files_touched.iter().map(|file| shell_quote(file)).collect(); + let cmd = format!("ls -t {} | head -1", quoted_files.join(" ")); + sandbox + .exec_command(&cmd, 5_000, None, None, None) + .await + .ok() + .and_then(|result| { + let trimmed = result.stdout.trim().to_string(); + (result.is_success() && !trimmed.is_empty()).then_some(trimmed) + }) + }; + + (files_touched, last_file_touched) +} diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index c69f8d99c..2f91c5c72 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -39,6 +39,8 @@ fn cli_failure_detail(stdout: &str, stderr: &str, command: &str) -> String { } use super::super::agent::{CodergenBackend, CodergenResult}; +use super::acp::AgentAcpBackend; +use super::{changed_files, node_runtime}; use crate::context::Context; use crate::error::Error; use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; @@ -132,10 +134,8 @@ async fn ensure_cli( // Install Node.js (if needed) and the CLI in a single shell so PATH persists let install_cmd = format!( - "export PATH=\"$HOME/.local/bin:$PATH\" && \ - (node --version >/dev/null 2>&1 || \ - (mkdir -p ~/.local && curl -fsSL https://nodejs.org/dist/v22.14.0/node-v22.14.0-linux-x64.tar.gz | tar -xz --strip-components=1 -C ~/.local)) && \ - npm install -g {}", + "{} && npm install -g {}", + node_runtime::ensure_node_runtime_shell(), cli.npm_package() ); let install_result = sandbox @@ -455,56 +455,6 @@ impl AgentCliBackend { self.poll_interval = interval; self } - - /// Detect changed files by comparing git state before and after the CLI - /// run. - async fn detect_changed_files(&self, sandbox: &Arc) -> Vec { - // Get unstaged changes - let diff_result = sandbox - .exec_command("git diff --name-only", 30_000, None, None, None) - .await; - - // Get untracked files - let untracked_result = sandbox - .exec_command( - "git ls-files --others --exclude-standard", - 30_000, - None, - None, - None, - ) - .await; - - let mut files: Vec = Vec::new(); - - if let Ok(result) = diff_result { - if result.is_success() { - files.extend( - result - .stdout - .lines() - .filter(|l| !l.trim().is_empty()) - .map(String::from), - ); - } - } - - if let Ok(result) = untracked_result { - if result.is_success() { - files.extend( - result - .stdout - .lines() - .filter(|l| !l.trim().is_empty()) - .map(String::from), - ); - } - } - - files.sort(); - files.dedup(); - files - } } #[async_trait] @@ -521,7 +471,7 @@ impl CodergenBackend for AgentCliBackend { cancel_token: CancellationToken, ) -> Result { // 1. Snapshot git state before the CLI run - let files_before = self.detect_changed_files(sandbox).await; + let files_before = changed_files::detect_changed_files(sandbox).await; // 2. Generate unique paths for this run let run_id = uuid::Uuid::new_v4().to_string(); @@ -801,29 +751,8 @@ impl CodergenBackend for AgentCliBackend { .ok_or_else(|| Error::handler("Failed to parse CLI output".to_string()))?; // 5. Detect changed files - let files_after = self.detect_changed_files(sandbox).await; - let files_touched: Vec = files_after - .into_iter() - .filter(|f| !files_before.contains(f)) - .collect(); - - // Find the most recently modified file by mtime - let last_file_touched = if files_touched.is_empty() { - None - } else { - let quoted_files: Vec = files_touched.iter().map(|f| shell_quote(f)).collect(); - let cmd = format!("ls -t {} | head -1", quoted_files.join(" ")); - if let Ok(result) = sandbox.exec_command(&cmd, 5_000, None, None, None).await { - let trimmed = result.stdout.trim().to_string(); - if result.is_success() && !trimmed.is_empty() { - Some(trimmed) - } else { - None - } - } else { - None - } - }; + let (files_touched, last_file_touched) = + changed_files::files_touched_since(sandbox, &files_before).await; let stage_usage = billed_model_usage_from_llm(model, provider, node.speed(), &TokenCounts { @@ -845,45 +774,71 @@ impl CodergenBackend for AgentCliBackend { clippy::disallowed_methods, reason = "CLI agent fallback credentials intentionally read provider API-key env vars." )] -fn process_env_var(name: &str) -> Option { +pub(crate) fn process_env_var(name: &str) -> Option { std::env::var(name).ok() } -/// Routes codergen invocations to either the API backend or CLI backend -/// based on node attributes and model type. +/// Routes codergen invocations to API, CLI, or ACP backends based on node +/// attributes and model type. pub struct BackendRouter { api_backend: Box, cli_backend: AgentCliBackend, + acp_backend: AgentAcpBackend, } impl BackendRouter { #[must_use] - pub fn new(api_backend: Box, cli_backend: AgentCliBackend) -> Self { + pub fn new( + api_backend: Box, + cli_backend: AgentCliBackend, + acp_backend: AgentAcpBackend, + ) -> Self { Self { api_backend, cli_backend, + acp_backend, } } - #[allow( - clippy::unused_self, - reason = "CLI backend selection lives on the router even though it only inspects the node." - )] - fn should_use_cli(&self, node: &Node) -> bool { - // Explicit backend="cli" attribute on the node - if node.backend() == Some("cli") { - return true; - } - - // CLI-only model on the node - if let Some(model) = node.model() { - if is_cli_only_model(model) { - return true; + fn select_backend(&self, node: &Node) -> Result { + match node.backend() { + None => { + if node.model().is_some_and(is_cli_only_model) { + Ok(SelectedBackend::Cli) + } else { + Ok(SelectedBackend::Api) + } } + Some("api") => Ok(SelectedBackend::Api), + Some("cli") => Ok(SelectedBackend::Cli), + Some("acp") => Ok(SelectedBackend::Acp), + Some(other) => Err(Error::Validation(format!( + "unsupported LLM backend \"{other}\"; expected one of: api, cli, acp" + ))), } - - false } + + fn select_one_shot_backend(&self, node: &Node) -> Result { + match node.backend() { + Some("acp") => Ok(SelectedBackend::Acp), + Some("api" | "cli") | None => Ok(SelectedBackend::Api), + Some(other) => Err(Error::Validation(format!( + "unsupported LLM backend \"{other}\"; expected one of: api, cli, acp" + ))), + } + } + + #[cfg(test)] + fn should_use_cli(&self, node: &Node) -> bool { + matches!(self.select_backend(node), Ok(SelectedBackend::Cli)) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SelectedBackend { + Api, + Cli, + Acp, } #[async_trait] @@ -899,32 +854,49 @@ impl CodergenBackend for BackendRouter { tool_hooks: Option>, cancel_token: CancellationToken, ) -> Result { - if self.should_use_cli(node) { - self.cli_backend - .run( - node, - prompt, - context, - thread_id, - emitter, - sandbox, - tool_hooks, - cancel_token, - ) - .await - } else { - self.api_backend - .run( - node, - prompt, - context, - thread_id, - emitter, - sandbox, - tool_hooks, - cancel_token, - ) - .await + match self.select_backend(node)? { + SelectedBackend::Api => { + self.api_backend + .run( + node, + prompt, + context, + thread_id, + emitter, + sandbox, + tool_hooks, + cancel_token, + ) + .await + } + SelectedBackend::Cli => { + self.cli_backend + .run( + node, + prompt, + context, + thread_id, + emitter, + sandbox, + tool_hooks, + cancel_token, + ) + .await + } + SelectedBackend::Acp => { + self.acp_backend + .run( + node, + prompt, + context, + thread_id, + emitter, + sandbox, + tool_hooks, + cancel_token, + ) + .await + } } } @@ -935,11 +907,37 @@ impl CodergenBackend for BackendRouter { system_prompt: Option<&str>, emitter: &Arc, stage_scope: &StageScope, + sandbox: &Arc, + cancel_token: CancellationToken, ) -> Result { - // CLI backend doesn't support one_shot, always route to API - self.api_backend - .one_shot(node, prompt, system_prompt, emitter, stage_scope) - .await + match self.select_one_shot_backend(node)? { + SelectedBackend::Acp => { + self.acp_backend + .one_shot( + node, + prompt, + system_prompt, + emitter, + stage_scope, + sandbox, + cancel_token, + ) + .await + } + SelectedBackend::Api | SelectedBackend::Cli => { + self.api_backend + .one_shot( + node, + prompt, + system_prompt, + emitter, + stage_scope, + sandbox, + cancel_token, + ) + .await + } + } } async fn shutdown(&self, emitter: &Arc) { @@ -951,6 +949,7 @@ impl CodergenBackend for BackendRouter { mod tests { use std::path::Path; + use fabro_agent::LocalSandbox; use fabro_agent::sandbox::ExecResult; use fabro_graphviz::graph::AttrValue; @@ -1388,8 +1387,7 @@ mod tests { node.attrs .insert("backend".to_string(), AttrValue::String("cli".to_string())); - let cli_backend = AgentCliBackend::new_from_env("model".into(), Provider::Anthropic); - let router = BackendRouter::new(Box::new(StubBackend), cli_backend); + let router = test_router(); assert!(router.should_use_cli(&node)); } @@ -1397,8 +1395,7 @@ mod tests { fn router_uses_api_by_default() { let node = Node::new("test"); - let cli_backend = AgentCliBackend::new_from_env("model".into(), Provider::Anthropic); - let router = BackendRouter::new(Box::new(StubBackend), cli_backend); + let router = test_router(); assert!(!router.should_use_cli(&node)); } @@ -1410,11 +1407,197 @@ mod tests { AttrValue::String("claude-opus-4-6".to_string()), ); - let cli_backend = AgentCliBackend::new_from_env("model".into(), Provider::Anthropic); - let router = BackendRouter::new(Box::new(StubBackend), cli_backend); + let router = test_router(); assert!(!router.should_use_cli(&node)); } + #[test] + fn router_uses_api_for_backend_api() { + let mut node = Node::new("test"); + node.attrs + .insert("backend".to_string(), AttrValue::String("api".to_string())); + + let router = test_router(); + assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Api); + } + + #[test] + fn router_uses_cli_for_backend_cli() { + let mut node = Node::new("test"); + node.attrs + .insert("backend".to_string(), AttrValue::String("cli".to_string())); + + let router = test_router(); + assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Cli); + } + + #[test] + fn router_uses_acp_for_backend_acp() { + let mut node = Node::new("test"); + node.attrs + .insert("backend".to_string(), AttrValue::String("acp".to_string())); + + let router = test_router(); + assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Acp); + } + + #[test] + fn router_rejects_unknown_backend() { + let mut node = Node::new("test"); + node.attrs.insert( + "backend".to_string(), + AttrValue::String("codex".to_string()), + ); + + let router = test_router(); + let err = router.select_backend(&node).unwrap_err(); + assert_eq!( + err.to_string(), + "Validation error: unsupported LLM backend \"codex\"; expected one of: api, cli, acp" + ); + } + + #[tokio::test] + async fn router_routes_one_shot_to_acp_for_backend_acp() { + let tempdir = tempfile::tempdir().unwrap(); + let script_path = tempdir.path().join("fake_acp_agent.py"); + tokio::fs::write(&script_path, fake_acp_agent_script()) + .await + .unwrap(); + let sandbox: Arc = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf())); + let mut node = Node::new("test"); + node.attrs + .insert("backend".to_string(), AttrValue::String("acp".to_string())); + node.attrs.insert( + "acp_command".to_string(), + AttrValue::String(format!( + "python3 {}", + shell_quote(&script_path.to_string_lossy()) + )), + ); + + let context = Context::new(); + let router = test_router(); + let result = router + .one_shot( + &node, + "prompt", + None, + &Arc::new(Emitter::default()), + &StageScope::for_handler(&context, "test"), + &sandbox, + CancellationToken::new(), + ) + .await + .unwrap(); + + let CodergenResult::Text { text, .. } = result else { + panic!("expected text result"); + }; + assert_eq!(text, "hello from acp"); + } + + #[tokio::test] + async fn router_routes_one_shot_to_api_by_default() { + let node = Node::new("test"); + let sandbox: Arc = Arc::new(LocalSandbox::new( + tempfile::tempdir().unwrap().path().to_path_buf(), + )); + let context = Context::new(); + let router = test_router(); + + let result = router + .one_shot( + &node, + "prompt", + None, + &Arc::new(Emitter::default()), + &StageScope::for_handler(&context, "test"), + &sandbox, + CancellationToken::new(), + ) + .await + .unwrap(); + + let CodergenResult::Text { text, .. } = result else { + panic!("expected text result"); + }; + assert_eq!(text, "api one-shot"); + } + + #[tokio::test] + async fn router_routes_one_shot_to_api_for_legacy_cli_backend() { + let mut node = Node::new("test"); + node.attrs + .insert("backend".to_string(), AttrValue::String("cli".to_string())); + let sandbox: Arc = Arc::new(LocalSandbox::new( + tempfile::tempdir().unwrap().path().to_path_buf(), + )); + let context = Context::new(); + let router = test_router(); + + let result = router + .one_shot( + &node, + "prompt", + None, + &Arc::new(Emitter::default()), + &StageScope::for_handler(&context, "test"), + &sandbox, + CancellationToken::new(), + ) + .await + .unwrap(); + + let CodergenResult::Text { text, .. } = result else { + panic!("expected text result"); + }; + assert_eq!(text, "api one-shot"); + } + + fn test_router() -> BackendRouter { + let cli_backend = AgentCliBackend::new_from_env("model".into(), Provider::Anthropic); + let acp_backend = AgentAcpBackend::new_from_env("model".into(), Provider::Anthropic); + BackendRouter::new(Box::new(StubBackend), cli_backend, acp_backend) + } + + fn fake_acp_agent_script() -> &'static str { + r#" +import json +import sys + +session_id = "sess-1" + +def send(message): + print(json.dumps(message), flush=True) + +def respond(message, result): + send({"jsonrpc": "2.0", "id": message["id"], "result": result}) + +for line in sys.stdin: + message = json.loads(line) + method = message.get("method") + if method == "initialize": + respond(message, {"protocolVersion": 1, "agentCapabilities": {}}) + elif method == "session/new": + respond(message, {"sessionId": session_id}) + elif method == "session/prompt": + send({ + "jsonrpc": "2.0", + "method": "session/update", + "params": { + "sessionId": session_id, + "update": { + "sessionUpdate": "agent_message_chunk", + "content": {"type": "text", "text": "hello from acp"} + } + } + }) + respond(message, {"stopReason": "end_turn"}) + break +"# + } + /// Minimal stub backend for testing routing logic. struct StubBackend; @@ -1438,6 +1621,24 @@ mod tests { last_file_touched: None, }) } + + async fn one_shot( + &self, + _node: &Node, + _prompt: &str, + _system_prompt: Option<&str>, + _emitter: &Arc, + _stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, + ) -> Result { + Ok(CodergenResult::Text { + text: "api one-shot".to_string(), + usage: None, + files_touched: Vec::new(), + last_file_touched: None, + }) + } } /// Sandbox stub whose `exec_command_streaming` returns a configurable diff --git a/lib/crates/fabro-workflow/src/handler/llm/mod.rs b/lib/crates/fabro-workflow/src/handler/llm/mod.rs index f210015ff..2310334da 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/mod.rs @@ -1,7 +1,11 @@ +pub mod acp; pub mod activation_lease; pub mod api; +pub mod changed_files; pub mod cli; +pub mod node_runtime; pub mod preamble; +pub use acp::AgentAcpBackend; pub use api::AgentApiBackend; pub use cli::{AgentCliBackend, BackendRouter, parse_cli_response}; diff --git a/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs b/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs new file mode 100644 index 000000000..18787773c --- /dev/null +++ b/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs @@ -0,0 +1,39 @@ +use std::sync::Arc; + +use fabro_agent::Sandbox; +use tokio_util::sync::CancellationToken; + +use crate::error::Error; + +pub fn ensure_node_runtime_shell() -> String { + "export PATH=\"$HOME/.local/bin:$PATH\" && \ + (node --version >/dev/null 2>&1 && npm --version >/dev/null 2>&1 && npx --version >/dev/null 2>&1 || \ + (mkdir -p ~/.local && curl -fsSL https://nodejs.org/dist/v22.14.0/node-v22.14.0-linux-x64.tar.gz | tar -xz --strip-components=1 -C ~/.local))" + .to_string() +} + +pub async fn ensure_node_runtime( + sandbox: &Arc, + cancel_token: &CancellationToken, +) -> Result<(), Error> { + let command = ensure_node_runtime_shell(); + let result = sandbox + .exec_command( + &command, + 180_000, + None, + None, + Some(cancel_token.child_token()), + ) + .await + .map_err(|err| Error::handler_with_source("Failed to ensure Node runtime", &err))?; + + if result.is_success() { + Ok(()) + } else { + Err(Error::handler(format!( + "Node runtime install exited with code {}", + result.display_exit_code() + ))) + } +} diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index d62654241..5c59041fc 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -126,6 +126,8 @@ impl Handler for PromptHandler { system_prompt.as_deref(), &services.run.emitter, &stage_scope, + &services.run.sandbox, + services.run.cancel_token(), ) .await; match result { @@ -335,6 +337,8 @@ mod tests { _system_prompt: Option<&str>, _emitter: &Arc, _stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, ) -> Result { Ok(CodergenResult::Text { text: "one-shot response".to_string(), @@ -398,6 +402,8 @@ mod tests { _system_prompt: Option<&str>, _emitter: &Arc, _stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, ) -> Result { Ok(CodergenResult::Text { text: "one-shot response".to_string(), @@ -458,6 +464,8 @@ mod tests { system_prompt: Option<&str>, _emitter: &Arc, _stage_scope: &StageScope, + _sandbox: &Arc, + _cancel_token: CancellationToken, ) -> Result { *self.captured_prompt.lock().unwrap() = Some(prompt.to_string()); *self.captured_system_prompt.lock().unwrap() = Some(system_prompt.map(String::from)); diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 79ee3c3f7..fe8ed6fb4 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -28,7 +28,7 @@ use crate::devcontainer_bridge::{devcontainer_to_snapshot_config, run_devcontain use crate::error::Error; use crate::event::{Event, RunNoticeCode, RunNoticeLevel}; use crate::github_token_source::{AppIatMinter, GitHubTokenSource}; -use crate::handler::llm::{AgentApiBackend, AgentCliBackend, BackendRouter}; +use crate::handler::llm::{AgentAcpBackend, AgentApiBackend, AgentCliBackend, BackendRouter}; use crate::handler::{HandlerRegistry, default_registry}; use crate::run_metadata::{RunMetadataRuntime, build_metadata_writer, metadata_branch_name}; use crate::run_options::{GitCheckpointOptions, RunOptions}; @@ -181,8 +181,21 @@ async fn build_registry( || AgentCliBackend::new_from_env(model.clone(), provider), |resolver| AgentCliBackend::new(model.clone(), provider, resolver), ) - .with_tool_env_provider(tool_env_provider, github_token_refresh_managed); - Some(Box::new(BackendRouter::new(Box::new(api), cli))) + .with_tool_env_provider( + tool_env_provider.clone(), + github_token_refresh_managed, + ); + let acp = cli_resolver + .clone() + .map_or_else( + || AgentAcpBackend::new_from_env(model.clone(), provider), + |resolver| AgentAcpBackend::new(model.clone(), provider, resolver), + ) + .with_tool_env_provider( + tool_env_provider.clone(), + github_token_refresh_managed, + ); + Some(Box::new(BackendRouter::new(Box::new(api), cli, acp))) })); Ok((registry, false)) } diff --git a/lib/crates/fabro-workflow/src/transforms/import.rs b/lib/crates/fabro-workflow/src/transforms/import.rs index 464692846..cce49e9c6 100644 --- a/lib/crates/fabro-workflow/src/transforms/import.rs +++ b/lib/crates/fabro-workflow/src/transforms/import.rs @@ -534,6 +534,7 @@ impl ImportTransform { | "reasoning_effort" | "speed" | "backend" + | "acp_command" | "fidelity" | "max_retries" | "thread_id" @@ -732,7 +733,7 @@ mod tests { let graph = apply_import( r#"digraph Deploy { start [shape=Mdiamond] - validate [import="./validate.fabro", model="haiku", class="fast, shared"] + validate [import="./validate.fabro", model="haiku", backend="acp", acp_command="python fake_agent.py", class="fast, shared"] exit [shape=Msquare] start -> validate -> exit }"#, @@ -772,6 +773,20 @@ mod tests { .iter() .any(|class_name| class_name == "validate") ); + assert_eq!( + graph.nodes["validate.test"] + .attrs + .get("backend") + .and_then(AttrValue::as_str), + Some("acp") + ); + assert_eq!( + graph.nodes["validate.test"] + .attrs + .get("acp_command") + .and_then(AttrValue::as_str), + Some("python fake_agent.py") + ); } #[test] diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index ad659b05a..200de9e24 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -42,6 +42,7 @@ use fabro_workflow::handler::command::CommandHandler; use fabro_workflow::handler::conditional::ConditionalHandler; use fabro_workflow::handler::exit::ExitHandler; use fabro_workflow::handler::human::HumanHandler; +use fabro_workflow::handler::llm::AgentAcpBackend; use fabro_workflow::handler::llm::cli::{AgentCliBackend, BackendRouter, parse_cli_response}; use fabro_workflow::handler::manager_loop::SubWorkflowHandler; use fabro_workflow::handler::start::StartHandler; @@ -6315,6 +6316,8 @@ mod real_llm { _system_prompt: Option<&str>, _emitter: &Arc, _stage_scope: &fabro_workflow::event::StageScope, + _sandbox: &Arc, + _cancel_token: tokio_util::sync::CancellationToken, ) -> Result { self.complete(prompt).await } @@ -10264,6 +10267,10 @@ async fn cli_backend_run_returns_text_and_usage() { // -- BackendRouter e2e: delegates to correct backend -- +fn test_acp_backend() -> AgentAcpBackend { + AgentAcpBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) +} + #[tokio::test] async fn backend_router_delegates_to_cli_for_cli_node() { let claude_output = r#"{"type":"result","result":"CLI response","usage":{"input_tokens":10,"output_tokens":5}}"#; @@ -10272,7 +10279,7 @@ async fn backend_router_delegates_to_cli_for_cli_node() { let api_backend = Box::new(MockCodergenBackend); // would return "Response for ..." let cli = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - let router = BackendRouter::new(api_backend, cli); + let router = BackendRouter::new(api_backend, cli, test_acp_backend()); let mut node = Node::new("cli_step"); node.attrs @@ -10317,7 +10324,7 @@ async fn backend_router_delegates_to_api_for_normal_node() { let api_backend = Box::new(MockCodergenBackend); let cli = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - let router = BackendRouter::new(api_backend, cli); + let router = BackendRouter::new(api_backend, cli, test_acp_backend()); let mut node = Node::new("api_step"); node.attrs.insert( @@ -10361,7 +10368,7 @@ async fn backend_router_delegates_to_cli_for_backend_attr() { let api_backend = Box::new(MockCodergenBackend); let cli = AgentCliBackend::new_from_env("gpt-5.3-codex".into(), Provider::OpenAi) .with_poll_interval(Duration::from_millis(10)); - let router = BackendRouter::new(api_backend, cli); + let router = BackendRouter::new(api_backend, cli, test_acp_backend()); let mut node = Node::new("codex_step"); node.attrs @@ -10455,7 +10462,7 @@ async fn full_pipeline_with_cli_backend_node() { let api = MockCodergenBackend; let cli = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - let router = BackendRouter::new(Box::new(api), cli); + let router = BackendRouter::new(Box::new(api), cli, test_acp_backend()); let codergen_handler = AgentHandler::new(Some(Box::new(router))); let mut registry = HandlerRegistry::new(Box::new(codergen_handler)); @@ -10468,7 +10475,7 @@ async fn full_pipeline_with_cli_backend_node() { let api2 = MockCodergenBackend; let cli2 = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - BackendRouter::new(Box::new(api2), cli2) + BackendRouter::new(Box::new(api2), cli2, test_acp_backend()) })))), ); @@ -10577,7 +10584,7 @@ async fn stylesheet_backend_property_routes_to_cli() { let api = MockCodergenBackend; let cli = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - let router = BackendRouter::new(Box::new(api), cli); + let router = BackendRouter::new(Box::new(api), cli, test_acp_backend()); let mut registry = HandlerRegistry::new(Box::new(AgentHandler::new(Some(Box::new(router))))); registry.register("start", Box::new(StartHandler)); @@ -10585,7 +10592,7 @@ async fn stylesheet_backend_property_routes_to_cli() { let api2 = MockCodergenBackend; let cli2 = AgentCliBackend::new_from_env("claude-opus-4-6".into(), Provider::Anthropic) .with_poll_interval(Duration::from_millis(10)); - let router2 = BackendRouter::new(Box::new(api2), cli2); + let router2 = BackendRouter::new(Box::new(api2), cli2, test_acp_backend()); registry.register( "agent", Box::new(AgentHandler::new(Some(Box::new(router2)))),