scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-07 08:27:12 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: route workflow stages to ACP backend

Browse source
This commit is contained in:
Bryan Helmkamp 2026-05-11 11:13:27 -04:00
parent 505547b962
commit 1c3edfe5f7
No known key found for this signature in database
24 changed files with 1178 additions and 227 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -2480,6 +2480,7 @@ dependencies = [
"bytes",
"chrono",
"dirs",
"fabro-acp",
"fabro-agent",
"fabro-auth",
"fabro-checkpoint",

49
lib/crates/fabro-acp/src/command.rs
View file

@ -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<String>,
env: HashMap<String, String>,
args: Vec<String>,
env: HashMap<String, String>,
}
impl AcpCommand {
@ -72,26 +70,33 @@ impl From<agent_client_protocol::Error> 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<AcpCommand, AcpCommandError> {
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<const N: usize>(
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(),
}
}

5
lib/crates/fabro-acp/src/error.rs
View file

@ -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<agent_client_protocol::Error> for AcpError {

56
lib/crates/fabro-acp/src/session.rs
View file

@ -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<u64>,
pub env: HashMap<String, String>,
pub sandbox: Arc<dyn Sandbox>,
pub command: AcpCommand,
pub prompt: String,
pub cwd: String,
pub timeout_ms: Option<u64>,
pub env: HashMap<String, String>,
pub sandbox: Arc<dyn Sandbox>,
pub cancel_token: CancellationToken,
pub on_activity: Option<Arc<dyn Fn() + Send + Sync>>,
pub on_activity: Option<Arc<dyn Fn() + Send + Sync>>,
}
#[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<AcpRunResult, AcpErr
.block_task()
.run_until(async |mut session| {
session.send_prompt(prompt)?;
read_turn(&mut session, &cancel_token, on_activity.as_ref(), &state_for_run).await
read_turn(
&mut session,
&cancel_token,
on_activity.as_ref(),
&state_for_run,
)
.await
})
.await
});
let outcome = match request.timeout_ms {
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,
});
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)

20
lib/crates/fabro-acp/src/transport.rs
View file

@ -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<String, String>,
sandbox: Arc<dyn Sandbox>,
command: AcpCommand,
cwd: String,
env: HashMap<String, String>,
sandbox: Arc<dyn Sandbox>,
cancel_token: CancellationToken,
state: TransportState,
state: TransportState,
}
impl SandboxAcpTransport {
@ -106,8 +108,10 @@ impl ConnectTo<Client> for SandboxAcpTransport {
},
));
let protocol =
agent_client_protocol::ConnectTo::<Client>::connect_to(Lines::new(outgoing_sink, incoming_lines), client);
let protocol = agent_client_protocol::ConnectTo::<Client>::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;

35
lib/crates/fabro-sandbox/src/docker.rs
View file

@ -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<String>, 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<Option<CommandTermination>>,
exec_id: String,
stop_file: String,
termination: tokio::sync::Mutex<Option<CommandTermination>>,
}
#[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,

5
lib/crates/fabro-sandbox/src/lib.rs
View file

@ -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};

6
lib/crates/fabro-sandbox/src/local.rs
View file

@ -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,

7
lib/crates/fabro-sandbox/src/read_guard.rs
View file

@ -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()
)]);
}
}

5
lib/crates/fabro-types/src/graph.rs
View file

@ -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")

84
lib/crates/fabro-validate/src/rules/backend_valid.rs Normal file
View file

@ -0,0 +1,84 @@
use fabro_graphviz::graph::{AttrValue, Graph};
use crate::{Diagnostic, LintRule, Severity};
pub(super) fn rule() -> Box<dyn LintRule> {
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<Diagnostic> {
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")
);
}
}

2
lib/crates/fabro-validate/src/rules/mod.rs
View file

@ -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<Box<dyn LintRule>> {
condition_syntax::rule(),
stylesheet_syntax::rule(),
type_known::rule(),
backend_valid::rule(),
fidelity_valid::rule(),
retry_target_exists::rule(),
goal_gate_has_retry::rule(),

1
lib/crates/fabro-workflow/Cargo.toml
View file

@ -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" }

2
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -56,6 +56,8 @@ pub trait CodergenBackend: Send + Sync {
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
_sandbox: &Arc<dyn Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
Err(Error::Validation(
"one_shot mode not supported by this backend".into(),

473
lib/crates/fabro-workflow/src/handler/llm/acp.rs Normal file
View file

@ -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<Arc<dyn ToolEnvProvider>>,
github_token_refresh_managed: bool,
resolver: Option<CredentialResolver>,
}
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<String, String>) -> Self {
self.tool_env = Some(Arc::new(StaticEnvProvider(env)));
self
}
#[must_use]
pub fn with_tool_env_provider(
mut self,
provider: Arc<dyn ToolEnvProvider>,
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<Emitter>,
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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::<Provider>().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<dyn Fn() + Send + Sync>
};
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<Emitter>,
sandbox: &Arc<dyn Sandbox>,
cancel_token: &CancellationToken,
) -> Result<HashMap<String, String>, 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<Emitter>,
sandbox: &Arc<dyn Sandbox>,
_tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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<Emitter>,
_stage_scope: &StageScope,
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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());
}
}

2
lib/crates/fabro-workflow/src/handler/llm/api.rs
View file

@ -483,6 +483,8 @@ impl CodergenBackend for AgentApiBackend {
system_prompt: Option<&str>,
emitter: &Arc<Emitter>,
stage_scope: &StageScope,
_sandbox: &Arc<dyn Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
let client = Client::from_source(self.source.as_ref())
.await

75
lib/crates/fabro-workflow/src/handler/llm/changed_files.rs Normal file
View file

@ -0,0 +1,75 @@
use std::sync::Arc;
use fabro_agent::{Sandbox, shell_quote};
pub async fn detect_changed_files(sandbox: &Arc<dyn Sandbox>) -> Vec<String> {
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<String> = 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<dyn Sandbox>,
files_before: &[String],
) -> (Vec<String>, Option<String>) {
let files_after = detect_changed_files(sandbox).await;
let files_touched: Vec<String> = 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<String> =
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)
}

469
lib/crates/fabro-workflow/src/handler/llm/cli.rs
View file

@ -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<dyn Sandbox>) -> Vec<String> {
// 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<String> = 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<CodergenResult, Error> {
// 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<String> = 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<String> = 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<String> {
pub(crate) fn process_env_var(name: &str) -> Option<String> {
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<dyn CodergenBackend>,
cli_backend: AgentCliBackend,
acp_backend: AgentAcpBackend,
}
impl BackendRouter {
#[must_use]
pub fn new(api_backend: Box<dyn CodergenBackend>, cli_backend: AgentCliBackend) -> Self {
pub fn new(
api_backend: Box<dyn CodergenBackend>,
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<SelectedBackend, Error> {
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<SelectedBackend, Error> {
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<Arc<dyn fabro_agent::ToolHookCallback>>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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<Emitter>,
stage_scope: &StageScope,
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
// 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<Emitter>) {
@ -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<dyn Sandbox> = 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<dyn Sandbox> = 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<dyn Sandbox> = 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<Emitter>,
_stage_scope: &StageScope,
_sandbox: &Arc<dyn Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
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

4
lib/crates/fabro-workflow/src/handler/llm/mod.rs
View file

@ -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};

39
lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs Normal file
View file

@ -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<dyn Sandbox>,
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()
)))
}
}

8
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -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<Emitter>,
_stage_scope: &StageScope,
_sandbox: &Arc<dyn Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
Ok(CodergenResult::Text {
text: "one-shot response".to_string(),
@ -398,6 +402,8 @@ mod tests {
_system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
_sandbox: &Arc<dyn Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
Ok(CodergenResult::Text {
text: "one-shot response".to_string(),
@ -458,6 +464,8 @@ mod tests {
system_prompt: Option<&str>,
_emitter: &Arc<Emitter>,
_stage_scope: &StageScope,
_sandbox: &Arc<dyn fabro_agent::Sandbox>,
_cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
*self.captured_prompt.lock().unwrap() = Some(prompt.to_string());
*self.captured_system_prompt.lock().unwrap() = Some(system_prompt.map(String::from));

19
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -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))
}

17
lib/crates/fabro-workflow/src/transforms/import.rs
View file

@ -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]

21
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -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<Emitter>,
_stage_scope: &fabro_workflow::event::StageScope,
_sandbox: &Arc<dyn fabro_agent::Sandbox>,
_cancel_token: tokio_util::sync::CancellationToken,
) -> Result<CodergenResult, Error> {
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)))),