diff --git a/Cargo.lock b/Cargo.lock index f7fc18003..a013f06ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2453,6 +2453,7 @@ version = "0.230.0-nightly.0" dependencies = [ "fabro-graphviz", "fabro-model", + "fabro-types", "serde", "thiserror 2.0.18", ] diff --git a/lib/crates/fabro-acp/src/error.rs b/lib/crates/fabro-acp/src/error.rs index 631620634..37ae511e5 100644 --- a/lib/crates/fabro-acp/src/error.rs +++ b/lib/crates/fabro-acp/src/error.rs @@ -17,9 +17,6 @@ pub enum AcpError { #[error("ACP turn timed out")] TimedOut { stderr: String }, - #[error("ACP process exited before the protocol completed")] - ProcessExited { stderr: String }, - #[error("ACP prompt stopped with {stop_reason}: {text}")] StopReason { stop_reason: String, diff --git a/lib/crates/fabro-acp/src/session.rs b/lib/crates/fabro-acp/src/session.rs index 8349379e9..10053b32e 100644 --- a/lib/crates/fabro-acp/src/session.rs +++ b/lib/crates/fabro-acp/src/session.rs @@ -7,7 +7,7 @@ use agent_client_protocol::schema::{ ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse, SelectedPermissionOutcome, SessionNotification, SessionUpdate, StopReason, }; -use agent_client_protocol::util::{MatchDispatch, internal_error}; +use agent_client_protocol::util::MatchDispatch; use agent_client_protocol::{ActiveSession, Agent, Client, Error as ProtocolError, SessionMessage}; use fabro_sandbox::Sandbox; use fabro_util::time::elapsed_ms; @@ -38,21 +38,28 @@ pub struct AcpRunResult { } pub async fn run_acp_turn(request: AcpRunRequest) -> Result { + let AcpRunRequest { + command, + prompt, + cwd, + timeout_ms, + env, + sandbox, + cancel_token, + on_activity, + } = request; let start = std::time::Instant::now(); let state = TransportState::new(); - let cancel_token = request.cancel_token.clone(); - let run_cancel_token = request.cancel_token.clone(); - let permission_cancel_token = request.cancel_token.clone(); - let prompt = request.prompt.clone(); - let cwd = request.cwd.clone(); - let on_activity = request.on_activity.clone(); + let read_cancel_token = cancel_token.clone(); + let run_cancel_token = cancel_token.clone(); + let permission_cancel_token = cancel_token.clone(); let state_for_run = state.clone(); let transport = SandboxAcpTransport::new( - request.command, - request.cwd, - request.env, - request.sandbox, - request.cancel_token.clone(), + command, + cwd.clone(), + env, + sandbox, + CancellationToken::new(), state.clone(), ); @@ -81,7 +88,7 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result Result { - if let Ok(result) = timeout(Duration::from_millis(timeout_ms), run).await { - result - } else { - state.terminate().await?; - if run_cancel_token.is_cancelled() { - return Err(AcpError::Cancelled); + let cancel_deadline_token = cancel_token.clone(); + let run_outcome = async { + match timeout_ms { + Some(timeout_ms) => { + if let Ok(result) = timeout(Duration::from_millis(timeout_ms), run).await { + Ok(result) + } else { + state.terminate().await?; + if run_cancel_token.is_cancelled() { + return Err(AcpError::Cancelled); + } + Err(AcpError::TimedOut { + stderr: state.stderr_tail().await, + }) } - return Err(AcpError::TimedOut { - stderr: state.stderr_tail().await, - }); } + None => Ok(run.await), + } + }; + let outcome = tokio::select! { + result = run_outcome => result?, + () = async { + cancel_deadline_token.cancelled().await; + sleep(Duration::from_millis(500)).await; + } => { + state.terminate().await?; + return Err(AcpError::Cancelled); } - None => run.await, }; let (text, stop_reason) = match outcome { Ok(result) => result, @@ -118,6 +138,21 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result {} + StopReason::Cancelled => { + state.terminate().await?; + return Err(AcpError::Cancelled); + } + _ => { + state.terminate().await?; + return Err(AcpError::StopReason { + stop_reason: stop_reason_to_string(stop_reason), + text, + }); + } + } + state.terminate().await?; let stderr = state.stderr_tail().await; Ok(AcpRunResult { @@ -129,20 +164,7 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result AcpError { - let message = error.to_string(); - if message.contains("ACP turn was cancelled") { - AcpError::Cancelled - } else if let Some(rest) = message.split("ACP prompt stopped with ").nth(1) { - let (stop_reason, text) = rest - .split_once(": ") - .map_or((rest, ""), |(stop_reason, text)| (stop_reason, text)); - AcpError::StopReason { - stop_reason: stop_reason.to_string(), - text: text.trim_end_matches('"').to_string(), - } - } else { - AcpError::Protocol(error) - } + AcpError::Protocol(error) } fn select_permission_outcome(request: &RequestPermissionRequest) -> RequestPermissionOutcome { @@ -201,22 +223,7 @@ async fn read_turn( .otherwise_ignore()?; } SessionMessage::StopReason(stop_reason) => { - return match stop_reason { - StopReason::EndTurn | StopReason::Refusal => Ok((text, stop_reason)), - StopReason::Cancelled => { - Err(internal_error("ACP turn was cancelled")) - } - StopReason::MaxTokens | StopReason::MaxTurnRequests => { - Err(internal_error(format!( - "ACP prompt stopped with {}: {text}", - stop_reason_name(stop_reason) - ))) - } - _ => Err(internal_error(format!( - "ACP prompt stopped with {}: {text}", - stop_reason_name(stop_reason) - ))), - }; + return Ok((text, stop_reason)); } _ => {} } @@ -230,19 +237,15 @@ async fn read_turn( } () = sleep(Duration::from_millis(500)), if cancel_sent => { state.terminate().await.map_err(ProtocolError::into_internal_error)?; - return Err(internal_error("ACP turn was cancelled")); + return Ok((text, StopReason::Cancelled)); } } } } -fn stop_reason_name(stop_reason: StopReason) -> &'static str { - match stop_reason { - StopReason::EndTurn => "end_turn", - StopReason::MaxTokens => "max_tokens", - StopReason::MaxTurnRequests => "max_turn_requests", - StopReason::Refusal => "refusal", - StopReason::Cancelled => "cancelled", - _ => "unknown", - } +fn stop_reason_to_string(stop_reason: StopReason) -> String { + serde_json::to_value(stop_reason) + .ok() + .and_then(|value| value.as_str().map(str::to_string)) + .unwrap_or_else(|| format!("{stop_reason:?}")) } diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index dae54b658..6259547d7 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -1,7 +1,8 @@ use std::collections::HashMap; use std::fmt::Write as _; use std::io::Cursor; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::{Instant, SystemTime, UNIX_EPOCH}; use async_trait::async_trait; @@ -20,7 +21,7 @@ use fabro_types::{CommandOutputStream, CommandTermination, RunId}; use fabro_util::time::elapsed_ms; use futures::StreamExt; use tokio::io::{AsyncWriteExt, duplex}; -use tokio::sync::{Mutex as TokioMutex, OnceCell}; +use tokio::sync::{Mutex as TokioMutex, Notify, OnceCell}; use tokio::{fs, time}; use tokio_util::sync::CancellationToken; @@ -884,13 +885,14 @@ struct DockerStdioProcessControl { container_id: String, exec_id: String, stop_file: String, - state: DockerStdioProcessState, + state: Arc, } #[derive(Default)] struct DockerStdioProcessState { - stop_requested: TokioMutex, - termination: TokioMutex>, + stop_requested: AtomicBool, + termination: TokioMutex>, + termination_notify: Notify, } impl DockerStdioProcessState { @@ -898,27 +900,36 @@ impl DockerStdioProcessState { *self.termination.lock().await } - async fn should_request_stop(&self) -> bool { - self.cached_termination().await.is_none() && !*self.stop_requested.lock().await - } - - async fn mark_stop_requested(&self) { - *self.stop_requested.lock().await = true; + async fn request_stop_once(&self) -> bool { + self.cached_termination().await.is_none() + && !self.stop_requested.swap(true, Ordering::AcqRel) } async fn cache_termination(&self, termination: StdioProcessTermination) { - *self.termination.lock().await = Some(termination); + let mut cached = self.termination.lock().await; + if cached.is_none() { + *cached = Some(termination); + self.termination_notify.notify_waiters(); + } + } + + async fn wait_for_cached_termination(&self) -> StdioProcessTermination { + loop { + if let Some(termination) = self.cached_termination().await { + return termination; + } + self.termination_notify.notified().await; + } } } #[async_trait] impl StdioProcessControl for DockerStdioProcessControl { async fn terminate(&self) -> crate::Result<()> { - if !self.state.should_request_stop().await { + if !self.state.request_stop_once().await { return Ok(()); } request_docker_exec_stop_with(&self.docker, &self.container_id, &self.stop_file).await?; - self.state.mark_stop_requested().await; Ok(()) } @@ -927,7 +938,11 @@ impl StdioProcessControl for DockerStdioProcessControl { return Ok(termination); } + let mut poll_interval = time::interval(std::time::Duration::from_secs(1)); loop { + if let Some(termination) = self.state.cached_termination().await { + return Ok(termination); + } let inspect = self .docker .inspect_exec(&self.exec_id) @@ -939,7 +954,29 @@ impl StdioProcessControl for DockerStdioProcessControl { self.state.cache_termination(termination).await; return Ok(termination); } - time::sleep(std::time::Duration::from_millis(50)).await; + tokio::select! { + termination = self.state.wait_for_cached_termination() => return Ok(termination), + _ = poll_interval.tick() => {} + } + } + } +} + +async fn cache_docker_stdio_completion( + docker: Docker, + exec_id: String, + state: Arc, +) { + match docker.inspect_exec(&exec_id).await { + Ok(inspect) if inspect.running != Some(true) => { + let exit_code = inspect.exit_code.and_then(|code| i32::try_from(code).ok()); + state + .cache_termination(StdioProcessTermination::exited(exit_code)) + .await; + } + Ok(_) => {} + Err(err) => { + tracing::warn!(error = %err, "Failed to inspect completed Docker stdio exec"); } } } @@ -1513,13 +1550,17 @@ impl Sandbox for DockerSandbox { let stderr_collector = StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES); let stderr_for_output = stderr_collector.clone(); let (mut stdout_writer, stdout_reader) = duplex(64 * 1024); + let state = Arc::new(DockerStdioProcessState::default()); + let state_for_output = Arc::clone(&state); + let docker_for_output = self.docker.clone(); + let exec_id_for_output = exec_id.clone(); tokio::spawn(async move { while let Some(chunk) = output.next().await { match chunk { Ok(LogOutput::StdOut { message }) => { if let Err(err) = stdout_writer.write_all(&message).await { tracing::warn!(error = %err, "Failed to forward Docker stdio stdout"); - return; + break; } } Ok(LogOutput::StdErr { message }) => { @@ -1529,10 +1570,12 @@ impl Sandbox for DockerSandbox { Err(err) => { let message = format!("Docker stdio output stream error: {err}"); stderr_for_output.push(message.as_bytes()).await; - return; + break; } } } + cache_docker_stdio_completion(docker_for_output, exec_id_for_output, state_for_output) + .await; }); let handle = StdioProcessHandle::new(DockerStdioProcessControl { @@ -1540,7 +1583,7 @@ impl Sandbox for DockerSandbox { container_id, exec_id, stop_file, - state: DockerStdioProcessState::default(), + state, }); if let Some(token) = cancel_token { @@ -2088,15 +2131,14 @@ mod tests { async fn docker_stdio_process_state_does_not_cache_cancelled_on_stop_request() { let state = DockerStdioProcessState::default(); - assert!(state.should_request_stop().await); - state.mark_stop_requested().await; + assert!(state.request_stop_once().await); assert_eq!(state.cached_termination().await, None); - assert!(!state.should_request_stop().await); + assert!(!state.request_stop_once().await); let termination = StdioProcessTermination::exited(Some(143)); state.cache_termination(termination).await; assert_eq!(state.cached_termination().await, Some(termination)); - assert!(!state.should_request_stop().await); + assert!(!state.request_stop_once().await); } #[tokio::test] diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index 5418fb567..dc94c35f3 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -7,7 +7,7 @@ use fabro_types::{CommandOutputStream, CommandTermination}; use fabro_util::time::elapsed_ms; use tokio::io::{AsyncRead, AsyncReadExt}; use tokio::process::{Child, Command}; -use tokio::sync::Mutex as TokioMutex; +use tokio::sync::watch; use tokio::task::spawn_blocking; use tokio::{fs, time}; use tokio_util::sync::CancellationToken; @@ -130,6 +130,34 @@ fn process_env_vars() -> Vec<(String, String)> { std::env::vars().collect() } +#[derive(Debug, Clone, Copy)] +enum ExplicitEnvPolicy { + FilterSensitive, + TrustCaller, +} + +fn filtered_env_vars( + env_vars: Option<&std::collections::HashMap>, + explicit_policy: ExplicitEnvPolicy, +) -> Vec<(String, String)> { + let mut filtered_env: Vec<(String, String)> = process_env_vars() + .into_iter() + .filter(|(key, _)| !LocalSandbox::should_filter_env_var(key)) + .collect(); + + if let Some(extra) = env_vars { + for (key, value) in extra { + if matches!(explicit_policy, ExplicitEnvPolicy::TrustCaller) + || !LocalSandbox::should_filter_env_var(key) + { + filtered_env.push((key.clone(), value.clone())); + } + } + } + + filtered_env +} + async fn drain_pipe(mut pipe: Option, stream: CommandOutputStream) -> String where R: AsyncRead + Unpin, @@ -143,37 +171,74 @@ where buf } +type LocalStdioOutcome = Result; + struct LocalStdioProcessControl { - child: TokioMutex, - termination: TokioMutex>, + terminate_tx: watch::Sender, + termination_rx: watch::Receiver>, +} + +impl LocalStdioProcessControl { + fn new(mut child: Child) -> Self { + let (terminate_tx, mut terminate_rx) = watch::channel(false); + let (termination_tx, termination_rx) = watch::channel(None); + + tokio::spawn(async move { + let outcome = tokio::select! { + status = child.wait() => { + status + .map(|status| StdioProcessTermination::exited(status.code())) + .map_err(|err| format!("Failed to wait for stdio process: {err}")) + } + changed = terminate_rx.changed() => { + if changed.is_err() || !*terminate_rx.borrow() { + child.wait() + .await + .map(|status| StdioProcessTermination::exited(status.code())) + .map_err(|err| format!("Failed to wait for stdio process: {err}")) + } else { + sigterm_then_kill(&mut child).await; + Ok(StdioProcessTermination::cancelled()) + } + } + }; + let _ = termination_tx.send(Some(outcome)); + }); + + Self { + terminate_tx, + termination_rx, + } + } + + async fn wait_for_termination(&self) -> crate::Result { + let mut termination_rx = self.termination_rx.clone(); + loop { + if let Some(outcome) = termination_rx.borrow().clone() { + return outcome.map_err(crate::Error::message); + } + termination_rx.changed().await.map_err(|_| { + crate::Error::message( + "stdio process supervisor stopped before reporting termination", + ) + })?; + } + } } #[async_trait] impl StdioProcessControl for LocalStdioProcessControl { async fn terminate(&self) -> crate::Result<()> { - if self.termination.lock().await.is_some() { + if self.termination_rx.borrow().is_some() { return Ok(()); } - let mut child = self.child.lock().await; - sigterm_then_kill(&mut child).await; - *self.termination.lock().await = Some(StdioProcessTermination::cancelled()); - Ok(()) + self.terminate_tx.send_replace(true); + self.wait_for_termination().await.map(|_| ()) } async fn wait(&self) -> crate::Result { - if let Some(termination) = *self.termination.lock().await { - return Ok(termination); - } - - let mut child = self.child.lock().await; - let status = child - .wait() - .await - .map_err(|e| crate::Error::context("Failed to wait for stdio process", e))?; - let termination = StdioProcessTermination::exited(status.code()); - *self.termination.lock().await = Some(termination); - Ok(termination) + self.wait_for_termination().await } } @@ -288,18 +353,7 @@ impl Sandbox for LocalSandbox { ) -> crate::Result { let start = Instant::now(); - let mut filtered_env: Vec<(String, String)> = process_env_vars() - .into_iter() - .filter(|(key, _)| !Self::should_filter_env_var(key)) - .collect(); - - if let Some(extra) = env_vars { - for (k, v) in extra { - if !Self::should_filter_env_var(k) { - filtered_env.push((k.clone(), v.clone())); - } - } - } + let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::FilterSensitive); let effective_dir = working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from); @@ -376,18 +430,7 @@ impl Sandbox for LocalSandbox { ) -> crate::Result { let start = Instant::now(); - let mut filtered_env: Vec<(String, String)> = process_env_vars() - .into_iter() - .filter(|(key, _)| !Self::should_filter_env_var(key)) - .collect(); - - if let Some(extra) = env_vars { - for (k, v) in extra { - if !Self::should_filter_env_var(k) { - filtered_env.push((k.clone(), v.clone())); - } - } - } + let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::FilterSensitive); let effective_dir = working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from); @@ -467,16 +510,7 @@ impl Sandbox for LocalSandbox { env_vars: Option<&std::collections::HashMap>, cancel_token: Option, ) -> crate::Result { - let mut filtered_env: Vec<(String, String)> = process_env_vars() - .into_iter() - .filter(|(key, _)| !Self::should_filter_env_var(key)) - .collect(); - - if let Some(extra) = env_vars { - for (k, v) in extra { - filtered_env.push((k.clone(), v.clone())); - } - } + let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::TrustCaller); let effective_dir = working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from); @@ -514,10 +548,7 @@ impl Sandbox for LocalSandbox { let stderr_collector = StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES); stderr_collector.spawn_reader(stderr); - let handle = StdioProcessHandle::new(LocalStdioProcessControl { - child: TokioMutex::new(child), - termination: TokioMutex::new(None), - }); + let handle = StdioProcessHandle::new(LocalStdioProcessControl::new(child)); if let Some(token) = cancel_token { let handle_for_cancel = handle.clone(); diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 5f79dd4c2..176b32c5a 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -405,7 +405,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_cli_terminal( + apply_agent_terminal( + "agent.cli", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -416,7 +417,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_cli_terminal( + apply_agent_terminal( + "agent.cli", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -427,7 +429,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_cli_terminal( + apply_agent_terminal( + "agent.cli", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -438,7 +441,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_acp_terminal( + apply_agent_terminal( + "agent.acp", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -449,7 +453,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_acp_terminal( + apply_agent_terminal( + "agent.acp", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -460,7 +465,8 @@ impl RunProjectionReducer for RunProjection { let Some(stage) = stage_at_current_visit(self, stored, event.seq) else { return Ok(()); }; - apply_agent_acp_terminal( + apply_agent_terminal( + "agent.acp", stage, props, merge_agent_cli_output(&props.stdout, &props.stderr), @@ -878,51 +884,37 @@ fn provider_used_from_agent_session_activated(props: &AgentSessionActivatedProps } fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value { - let mut provider_used = serde_json::Map::new(); - provider_used.insert("mode".to_string(), Value::String("cli".to_string())); - provider_used.insert( - "provider".to_string(), - Value::String(props.provider.clone()), - ); - provider_used.insert("model".to_string(), Value::String(props.model.clone())); - provider_used.insert("command".to_string(), Value::String(props.command.clone())); - Value::Object(provider_used) + provider_used_from_agent_process_started("cli", &props.provider, &props.model, &props.command) } fn provider_used_from_agent_acp_started(props: &AgentAcpStartedProps) -> Value { + provider_used_from_agent_process_started("acp", &props.provider, &props.model, &props.command) +} + +fn provider_used_from_agent_process_started( + mode: &str, + provider: &str, + model: &str, + command: &str, +) -> Value { let mut provider_used = serde_json::Map::new(); - provider_used.insert("mode".to_string(), Value::String("acp".to_string())); - provider_used.insert( - "provider".to_string(), - Value::String(props.provider.clone()), - ); - provider_used.insert("model".to_string(), Value::String(props.model.clone())); - provider_used.insert("command".to_string(), Value::String(props.command.clone())); + provider_used.insert("mode".to_string(), Value::String(mode.to_string())); + provider_used.insert("provider".to_string(), Value::String(provider.to_string())); + provider_used.insert("model".to_string(), Value::String(model.to_string())); + provider_used.insert("command".to_string(), Value::String(command.to_string())); Value::Object(provider_used) } -fn apply_agent_cli_terminal( +fn apply_agent_terminal( + event_prefix: &str, stage: &mut StageProjection, props: &impl serde::Serialize, output: String, termination: CommandTermination, ) -> Result<()> { - let script_timing = serde_json::to_value(props) - .map_err(|err| Error::InvalidEvent(format!("invalid agent.cli terminal payload: {err}")))?; - stage.output = Some(output); - stage.termination = Some(termination); - stage.script_timing = Some(script_timing); - Ok(()) -} - -fn apply_agent_acp_terminal( - stage: &mut StageProjection, - props: &impl serde::Serialize, - output: String, - termination: CommandTermination, -) -> Result<()> { - let script_timing = serde_json::to_value(props) - .map_err(|err| Error::InvalidEvent(format!("invalid agent.acp terminal payload: {err}")))?; + let script_timing = serde_json::to_value(props).map_err(|err| { + Error::InvalidEvent(format!("invalid {event_prefix} terminal payload: {err}")) + })?; stage.output = Some(output); stage.termination = Some(termination); stage.script_timing = Some(script_timing); diff --git a/lib/crates/fabro-types/src/graph.rs b/lib/crates/fabro-types/src/graph.rs index ba43af5e3..786fe269e 100644 --- a/lib/crates/fabro-types/src/graph.rs +++ b/lib/crates/fabro-types/src/graph.rs @@ -3,6 +3,8 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; +use crate::LlmBackend; + /// Typed attribute values for nodes, edges, and graph-level attributes. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub enum AttrValue { @@ -266,6 +268,11 @@ impl Node { self.str_attr("backend") } + #[must_use] + pub fn llm_backend(&self) -> Option> { + self.backend().map(str::parse) + } + #[must_use] pub fn acp_command(&self) -> Option<&str> { self.str_attr("acp_command") diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index dcac895f2..8931662dd 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/lib/crates/fabro-types/src/lib.rs @@ -13,6 +13,7 @@ pub mod event_envelope; pub mod failure_signature; pub mod graph; pub mod interview; +pub mod llm_backend; pub mod outcome; pub mod principal; pub mod pull_request; @@ -57,6 +58,7 @@ pub use graph::{ shape_to_handler_type, }; pub use interview::{InterviewQuestionRecord, QuestionType}; +pub use llm_backend::LlmBackend; pub use outcome::{ FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageOutcome, StageState, }; diff --git a/lib/crates/fabro-types/src/llm_backend.rs b/lib/crates/fabro-types/src/llm_backend.rs new file mode 100644 index 000000000..0dc510acc --- /dev/null +++ b/lib/crates/fabro-types/src/llm_backend.rs @@ -0,0 +1,28 @@ +use serde::{Deserialize, Serialize}; +use strum::{Display, EnumString, IntoStaticStr, VariantArray}; + +#[derive( + Debug, + Clone, + Copy, + PartialEq, + Eq, + Hash, + Serialize, + Deserialize, + Display, + EnumString, + IntoStaticStr, + VariantArray, +)] +#[serde(rename_all = "snake_case")] +#[strum(serialize_all = "snake_case")] +pub enum LlmBackend { + Api, + Cli, + Acp, +} + +impl LlmBackend { + pub const EXPECTED: &'static str = "api, cli, acp"; +} diff --git a/lib/crates/fabro-validate/Cargo.toml b/lib/crates/fabro-validate/Cargo.toml index 0dc049b04..2ea44ae2d 100644 --- a/lib/crates/fabro-validate/Cargo.toml +++ b/lib/crates/fabro-validate/Cargo.toml @@ -15,5 +15,6 @@ workspace = true [dependencies] fabro-graphviz = { path = "../fabro-graphviz" } fabro-model = { path = "../fabro-model" } +fabro-types = { path = "../fabro-types" } serde = { workspace = true } -thiserror = { workspace = true } \ No newline at end of file +thiserror = { workspace = true } diff --git a/lib/crates/fabro-validate/src/rules/backend_valid.rs b/lib/crates/fabro-validate/src/rules/backend_valid.rs index 1ca5d9cfa..41489c762 100644 --- a/lib/crates/fabro-validate/src/rules/backend_valid.rs +++ b/lib/crates/fabro-validate/src/rules/backend_valid.rs @@ -1,4 +1,5 @@ use fabro_graphviz::graph::{AttrValue, Graph}; +use fabro_types::LlmBackend; use crate::{Diagnostic, LintRule, Severity}; @@ -8,8 +9,6 @@ pub(super) fn rule() -> Box { struct Rule; -const VALID_BACKENDS: &[&str] = &["api", "cli", "acp"]; - impl LintRule for Rule { fn name(&self) -> &'static str { "backend_valid" @@ -19,16 +18,17 @@ impl LintRule for Rule { 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) { + if backend.parse::().is_err() { diagnostics.push(Diagnostic { rule: self.name().to_string(), severity: Severity::Error, message: format!( - "unsupported LLM backend \"{backend}\"; expected one of: api, cli, acp" + "unsupported LLM backend \"{backend}\"; expected one of: {}", + LlmBackend::EXPECTED ), node_id: Some(node.id.clone()), edge: None, - fix: Some("Use one of: api, cli, acp".to_string()), + fix: Some(format!("Use one of: {}", LlmBackend::EXPECTED)), }); } } diff --git a/lib/crates/fabro-workflow/src/handler/llm/acp.rs b/lib/crates/fabro-workflow/src/handler/llm/acp.rs index 6f138aa6c..d074fd8f1 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/acp.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/acp.rs @@ -6,18 +6,19 @@ 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_auth::CredentialResolver; use fabro_graphviz::graph::Node; use fabro_model::Provider; use fabro_util::time::elapsed_ms; use tokio_util::sync::CancellationToken; use super::super::agent::{CodergenBackend, CodergenResult}; -use super::cli::{AgentCli, process_env_var}; +use super::cli::AgentCli; +use super::launch_env::{AgentLaunchEnvRequest, resolve_agent_launch_env}; use super::{changed_files, node_runtime}; use crate::context::Context; use crate::error::Error; -use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; +use crate::event::{Emitter, Event, StageScope}; pub struct AgentAcpBackend { model: String, @@ -94,9 +95,18 @@ impl AgentAcpBackend { None }; - let mut launch_env = self - .launch_env(provider, emitter, sandbox, &cancel_token) - .await?; + let mut launch_env = resolve_agent_launch_env(AgentLaunchEnvRequest { + provider, + cli: AgentCli::for_provider(provider), + resolver: self.resolver.as_ref(), + tool_env: self.tool_env.as_ref(), + github_token_refresh_managed: self.github_token_refresh_managed, + stage_label: "ACP", + emitter, + sandbox, + cancel_token: &cancel_token, + }) + .await?; if let Some(runtime_env) = node_runtime_env { node_runtime::apply_node_runtime_env(&mut launch_env, runtime_env); } @@ -197,78 +207,6 @@ impl AgentAcpBackend { 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] diff --git a/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs b/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs index d5e302515..6b07e2f5d 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/changed_files.rs @@ -1,42 +1,25 @@ +use std::collections::HashSet; 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; +const DIFF_MARKER: &str = "__FABRO_CHANGED_FILES_DIFF__"; +const UNTRACKED_MARKER: &str = "__FABRO_CHANGED_FILES_UNTRACKED__"; +pub async fn detect_changed_files(sandbox: &Arc) -> Vec { let mut files: Vec = Vec::new(); - if let Ok(result) = diff_result { + let command = format!( + "printf '%s\\n' {diff}; git diff --name-only || true; \ + printf '%s\\n' {untracked}; git ls-files --others --exclude-standard || true", + diff = shell_quote(DIFF_MARKER), + untracked = shell_quote(UNTRACKED_MARKER), + ); + if let Ok(result) = sandbox + .exec_command(&command, 30_000, None, None, None) + .await + { 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.extend(parse_changed_files(&result.stdout)); } } @@ -50,9 +33,10 @@ pub async fn files_touched_since( files_before: &[String], ) -> (Vec, Option) { let files_after = detect_changed_files(sandbox).await; + let files_before: HashSet<&str> = files_before.iter().map(String::as_str).collect(); let files_touched: Vec = files_after .into_iter() - .filter(|file| !files_before.contains(file)) + .filter(|file| !files_before.contains(file.as_str())) .collect(); let last_file_touched = if files_touched.is_empty() { @@ -73,3 +57,27 @@ pub async fn files_touched_since( (files_touched, last_file_touched) } + +fn parse_changed_files(stdout: &str) -> impl Iterator + '_ { + stdout.lines().filter_map(|line| { + let trimmed = line.trim(); + (!trimmed.is_empty() && trimmed != DIFF_MARKER && trimmed != UNTRACKED_MARKER) + .then(|| trimmed.to_string()) + }) +} + +#[cfg(test)] +mod tests { + use super::parse_changed_files; + + #[test] + fn parse_changed_files_ignores_section_markers() { + let files = parse_changed_files( + "__FABRO_CHANGED_FILES_DIFF__\nsrc/main.rs\n\ + __FABRO_CHANGED_FILES_UNTRACKED__\nREADME.md\n", + ) + .collect::>(); + + assert_eq!(files, vec!["src/main.rs", "README.md"]); + } +} diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 4faa0c43f..681df81cb 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -8,11 +8,11 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use fabro_agent::{Sandbox, StaticEnvProvider, ToolEnvProvider, shell_quote}; -use fabro_auth::{CliAgentKind, CredentialResolver, CredentialUsage, ResolvedCredential}; +use fabro_auth::CredentialResolver; use fabro_graphviz::graph::Node; use fabro_llm::types::TokenCounts; use fabro_model::Provider; -use fabro_types::{CommandOutputStream, CommandTermination}; +use fabro_types::{CommandOutputStream, CommandTermination, LlmBackend}; use fabro_util::time::elapsed_ms; use tokio_util::sync::CancellationToken; @@ -40,10 +40,11 @@ fn cli_failure_detail(stdout: &str, stderr: &str, command: &str) -> String { use super::super::agent::{CodergenBackend, CodergenResult}; use super::acp::AgentAcpBackend; +use super::launch_env::{AgentLaunchEnvRequest, resolve_agent_launch_env}; use super::{changed_files, node_runtime}; use crate::context::Context; use crate::error::Error; -use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; +use crate::event::{Emitter, Event, StageScope}; use crate::outcome::billed_model_usage_from_llm; /// Maps a provider to its corresponding CLI tool metadata. @@ -509,67 +510,18 @@ impl CodergenBackend for AgentCliBackend { &stage_scope, ); - // Forward provider API key and custom env vars so the CLI tool can - // authenticate. Resolve credentials and run any pre-login command - // before the main CLI invocation. - let cli_agent = match cli { - 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(|e| Error::handler_with_source("Failed to resolve CLI credential", &e))?; - 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(|e| Error::handler_with_source("codex login failed", &e))?; - if !login_result.is_success() { - tracing::warn!( - exit_code = login_result.display_exit_code(), - "codex login --with-api-key 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(val) = process_env_var(name) { - env.insert((*name).to_string(), val); - } - } - env - }; - if let Some(provider) = &self.tool_env { - if self.github_token_refresh_managed { - emitter.notice( - RunNoticeLevel::Info, - RunNoticeCode::GithubTokenRefreshLimited, - "CLI 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 CLI agent env", &err) - })?; - for (name, val) in tool_env { - launch_env.insert(name, val); - } - } + let launch_env = resolve_agent_launch_env(AgentLaunchEnvRequest { + provider, + cli, + resolver: self.resolver.as_ref(), + tool_env: self.tool_env.as_ref(), + github_token_refresh_managed: self.github_token_refresh_managed, + stage_label: "CLI", + emitter, + sandbox, + cancel_token: &cancel_token, + }) + .await?; // Write env file so the inner shell that runs the CLI command picks up // PATH and provider env vars; we still pass `launch_env` to @@ -800,45 +752,43 @@ impl BackendRouter { } } - fn select_backend(node: &Node) -> Result { - match node.backend() { + fn select_backend(node: &Node) -> Result { + match node.llm_backend() { None => { if node.model().is_some_and(is_cli_only_model) { - Ok(SelectedBackend::Cli) + Ok(LlmBackend::Cli) } else { - Ok(SelectedBackend::Api) + Ok(LlmBackend::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" - ))), + Some(Ok(backend)) => Ok(backend), + Some(Err(_)) => Err(unsupported_backend_error( + node.backend().unwrap_or_default(), + )), } } - fn select_one_shot_backend(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" - ))), + fn select_one_shot_backend(node: &Node) -> Result { + match node.llm_backend() { + Some(Ok(LlmBackend::Acp)) => Ok(LlmBackend::Acp), + Some(Ok(LlmBackend::Api | LlmBackend::Cli)) | None => Ok(LlmBackend::Api), + Some(Err(_)) => Err(unsupported_backend_error( + node.backend().unwrap_or_default(), + )), } } #[cfg(test)] fn should_use_cli(node: &Node) -> bool { - matches!(Self::select_backend(node), Ok(SelectedBackend::Cli)) + matches!(Self::select_backend(node), Ok(LlmBackend::Cli)) } } -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum SelectedBackend { - Api, - Cli, - Acp, +fn unsupported_backend_error(raw: &str) -> Error { + Error::Validation(format!( + "unsupported LLM backend \"{raw}\"; expected one of: {}", + LlmBackend::EXPECTED + )) } #[async_trait] @@ -855,7 +805,7 @@ impl CodergenBackend for BackendRouter { cancel_token: CancellationToken, ) -> Result { match Self::select_backend(node)? { - SelectedBackend::Api => { + LlmBackend::Api => { self.api .run( node, @@ -869,7 +819,7 @@ impl CodergenBackend for BackendRouter { ) .await } - SelectedBackend::Cli => { + LlmBackend::Cli => { self.cli .run( node, @@ -883,7 +833,7 @@ impl CodergenBackend for BackendRouter { ) .await } - SelectedBackend::Acp => { + LlmBackend::Acp => { self.acp .run( node, @@ -911,7 +861,7 @@ impl CodergenBackend for BackendRouter { cancel_token: CancellationToken, ) -> Result { match Self::select_one_shot_backend(node)? { - SelectedBackend::Acp => { + LlmBackend::Acp => { self.acp .one_shot( node, @@ -924,7 +874,7 @@ impl CodergenBackend for BackendRouter { ) .await } - SelectedBackend::Api | SelectedBackend::Cli => { + LlmBackend::Api | LlmBackend::Cli => { self.api .one_shot( node, @@ -1416,7 +1366,7 @@ mod tests { assert_eq!( BackendRouter::select_backend(&node).unwrap(), - SelectedBackend::Api + LlmBackend::Api ); } @@ -1428,7 +1378,7 @@ mod tests { assert_eq!( BackendRouter::select_backend(&node).unwrap(), - SelectedBackend::Cli + LlmBackend::Cli ); } @@ -1440,7 +1390,7 @@ mod tests { assert_eq!( BackendRouter::select_backend(&node).unwrap(), - SelectedBackend::Acp + LlmBackend::Acp ); } diff --git a/lib/crates/fabro-workflow/src/handler/llm/launch_env.rs b/lib/crates/fabro-workflow/src/handler/llm/launch_env.rs new file mode 100644 index 000000000..134dea313 --- /dev/null +++ b/lib/crates/fabro-workflow/src/handler/llm/launch_env.rs @@ -0,0 +1,107 @@ +use std::collections::HashMap; +use std::sync::Arc; + +use fabro_agent::{Sandbox, ToolEnvProvider}; +use fabro_auth::{CliAgentKind, CredentialResolver, CredentialUsage, ResolvedCredential}; +use fabro_model::Provider; +use tokio_util::sync::CancellationToken; + +use super::cli::{AgentCli, process_env_var}; +use crate::error::Error; +use crate::event::{Emitter, RunNoticeCode, RunNoticeLevel}; + +pub(crate) struct AgentLaunchEnvRequest<'a> { + pub provider: Provider, + pub cli: AgentCli, + pub resolver: Option<&'a CredentialResolver>, + pub tool_env: Option<&'a Arc>, + pub github_token_refresh_managed: bool, + pub stage_label: &'static str, + pub emitter: &'a Arc, + pub sandbox: &'a Arc, + pub cancel_token: &'a CancellationToken, +} + +pub(crate) async fn resolve_agent_launch_env( + request: AgentLaunchEnvRequest<'_>, +) -> Result, Error> { + let cli_agent = match request.cli { + AgentCli::Claude => CliAgentKind::Claude, + AgentCli::Codex => CliAgentKind::Codex, + AgentCli::Gemini => CliAgentKind::Gemini, + }; + + let mut launch_env = if let Some(resolver) = request.resolver { + let resolved = resolver + .resolve(request.provider, CredentialUsage::CliAgent(cli_agent)) + .await + .map_err(|err| { + Error::handler_with_source( + format!("Failed to resolve {} credential", request.stage_label), + &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 = request + .sandbox + .exec_command( + login_cmd, + 30_000, + None, + None, + Some(request.cancel_token.child_token()), + ) + .await + .map_err(|err| { + Error::handler_with_source( + format!("{} credential login failed", request.stage_label), + &err, + ) + })?; + if !login_result.is_success() { + tracing::warn!( + exit_code = login_result.display_exit_code(), + stage = request.stage_label, + "{} credential login failed: {}", + request.stage_label, + login_result.stderr + ); + } + } + cli_credential.env_vars + } else { + let mut env = HashMap::new(); + for name in request.provider.api_key_env_vars() { + if let Some(value) = process_env_var(name) { + env.insert((*name).to_string(), value); + } + } + env + }; + + if let Some(provider) = request.tool_env { + if request.github_token_refresh_managed { + request.emitter.notice( + RunNoticeLevel::Info, + RunNoticeCode::GithubTokenRefreshLimited, + format!( + "{} agent stages receive GitHub tokens at process launch; stages running \ + beyond token expiry may need to be retried.", + request.stage_label + ), + ); + } + let tool_env = provider.resolve().await.map_err(|err| { + Error::handler_with_anyhow( + format!("Failed to resolve {} agent env", request.stage_label), + &err, + ) + })?; + launch_env.extend(tool_env); + } + + Ok(launch_env) +} diff --git a/lib/crates/fabro-workflow/src/handler/llm/mod.rs b/lib/crates/fabro-workflow/src/handler/llm/mod.rs index 2310334da..0669e91bd 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/mod.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/mod.rs @@ -3,6 +3,7 @@ pub mod activation_lease; pub mod api; pub mod changed_files; pub mod cli; +pub mod launch_env; pub mod node_runtime; pub mod preamble; diff --git a/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs b/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs index 838917584..8b1398051 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/node_runtime.rs @@ -14,9 +14,24 @@ pub struct NodeRuntimeEnv { } 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))" + r#"export PATH="$HOME/.local/bin:$PATH" && \ +if node --version >/dev/null 2>&1 && npm --version >/dev/null 2>&1 && npx --version >/dev/null 2>&1; then \ + true; \ +else \ + os="$(uname -s)"; \ + if [ "$os" != "Linux" ]; then \ + echo "Node.js, npm, and npx are required for default ACP/CLI commands on $os" >&2; \ + exit 127; \ + fi; \ + arch="$(uname -m)"; \ + case "$arch" in \ + x86_64|amd64) node_arch="x64" ;; \ + aarch64|arm64) node_arch="arm64" ;; \ + *) echo "Unsupported Linux architecture for Node.js install: $arch" >&2; exit 127 ;; \ + esac; \ + mkdir -p "$HOME/.local" && \ + curl -fsSL "https://nodejs.org/dist/v22.14.0/node-v22.14.0-linux-${node_arch}.tar.gz" | tar -xz --strip-components=1 -C "$HOME/.local"; \ +fi"# .to_string() } diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index fcf933c6b..f77245c43 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -15,6 +15,7 @@ use fabro_sandbox::{ reconnect_for_run_with_callback, }; use fabro_static::EnvVars; +use fabro_types::LlmBackend; use fabro_vault::Vault; use futures::future::try_join_all; use shlex::try_quote; @@ -28,6 +29,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::cli::is_cli_only_model; 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}; @@ -124,7 +126,13 @@ async fn build_registry( llm_source: Arc, cli_resolver: Option, ) -> Result<(Arc, bool), Error> { - let build_no_backend = || Arc::new(default_registry(Arc::clone(&interviewer), || None)); + let no_backend_interviewer = Arc::clone(&interviewer); + let build_no_backend = move || { + Arc::new(default_registry( + Arc::clone(&no_backend_interviewer), + || None, + )) + }; if spec.dry_run { return Ok((build_no_backend(), true)); @@ -135,6 +143,51 @@ async fn build_registry( .values() .any(|n| graph::is_llm_handler_type(n.handler_type())); + if !graph_needs_llm { + return Ok((build_no_backend(), false)); + } + + let build_llm_registry = || { + let model = spec.model.clone(); + let provider = spec.provider; + let fallback_chain = spec.fallback_chain.clone(); + let mcp_servers = spec.mcp_servers.clone(); + let llm_source_for_api = Arc::clone(&llm_source); + let steering_hub_for_api = Arc::clone(&steering_hub); + let tool_env_provider_for_backend = Arc::clone(&tool_env_provider); + Arc::new(default_registry(interviewer, move || { + let tool_env_provider = Arc::clone(&tool_env_provider_for_backend); + let api = AgentApiBackend::new( + model.clone(), + provider, + fallback_chain.clone(), + Arc::clone(&llm_source_for_api), + Arc::clone(&steering_hub_for_api), + ) + .with_tool_env_provider(tool_env_provider.clone()) + .with_mcp_servers(mcp_servers.clone()); + let cli = cli_resolver + .clone() + .map_or_else( + || AgentCliBackend::new_from_env(model.clone(), provider), + |resolver| AgentCliBackend::new(model.clone(), provider, resolver), + ) + .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))) + })) + }; + + if !graph_needs_api_backend(graph) { + return Ok((build_llm_registry(), false)); + } + match llm_source.resolve().await { Ok(result) if result.credentials.is_empty() => { if graph_needs_llm { @@ -156,49 +209,7 @@ async fn build_registry( } Ok((build_no_backend(), false)) } - Ok(_result) => { - let model = spec.model.clone(); - let provider = spec.provider; - let fallback_chain = spec.fallback_chain.clone(); - let mcp_servers = spec.mcp_servers.clone(); - let llm_source_for_api = Arc::clone(&llm_source); - let steering_hub_for_api = Arc::clone(&steering_hub); - let tool_env_provider_for_backend = Arc::clone(&tool_env_provider); - let registry = Arc::new(default_registry(interviewer, move || { - let tool_env_provider = Arc::clone(&tool_env_provider_for_backend); - let api = AgentApiBackend::new( - model.clone(), - provider, - fallback_chain.clone(), - Arc::clone(&llm_source_for_api), - Arc::clone(&steering_hub_for_api), - ) - .with_tool_env_provider(tool_env_provider.clone()) - .with_mcp_servers(mcp_servers.clone()); - let cli = cli_resolver - .clone() - .map_or_else( - || AgentCliBackend::new_from_env(model.clone(), provider), - |resolver| AgentCliBackend::new(model.clone(), provider, resolver), - ) - .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)) - } + Ok(_result) => Ok((build_llm_registry(), false)), Err(e) => { if graph_needs_llm { return Err(Error::Precondition(format!( @@ -210,6 +221,25 @@ async fn build_registry( } } +fn graph_needs_api_backend(graph: &graph::Graph) -> bool { + graph.nodes.values().any(node_needs_api_backend) +} + +fn node_needs_api_backend(node: &graph::Node) -> bool { + if !graph::is_llm_handler_type(node.handler_type()) { + return false; + } + + match node.handler_type() { + Some("prompt" | "one_shot") => !matches!(node.llm_backend(), Some(Ok(LlmBackend::Acp))), + _ => match node.llm_backend() { + Some(Ok(LlmBackend::Api)) => true, + Some(Ok(LlmBackend::Cli | LlmBackend::Acp) | Err(_)) => false, + None => !node.model().is_some_and(is_cli_only_model), + }, + } +} + fn build_llm_source(vault: Option>>) -> Arc { match vault { Some(vault) => Arc::new(VaultCredentialSource::new(vault)), diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index edd547af4..e13d7404a 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -9659,14 +9659,25 @@ impl fabro_agent::Sandbox for CliTestEnv { ) -> fabro_sandbox::Result { self.commands.lock().unwrap().push(command.to_string()); - // git diff calls: first pair returns empty (before), second pair returns - // configured files - if command.starts_with("git diff") || command.starts_with("git ls-files") { + // Changed-file snapshot calls: first returns empty (before), second + // returns configured files (after). + if command.contains("__FABRO_CHANGED_FILES_DIFF__") + || command.starts_with("git diff") + || command.starts_with("git ls-files") + { let call_num = self .git_diff_call_count .fetch_add(1, std::sync::atomic::Ordering::SeqCst); - // Calls 0,1 = before snapshot (empty), calls 2,3 = after snapshot - let stdout = if call_num >= 2 && command.starts_with("git diff") { + let stdout = if command.contains("__FABRO_CHANGED_FILES_DIFF__") { + if call_num >= 1 { + format!( + "__FABRO_CHANGED_FILES_DIFF__\n{}__FABRO_CHANGED_FILES_UNTRACKED__\n", + self.git_diff_after + ) + } else { + "__FABRO_CHANGED_FILES_DIFF__\n__FABRO_CHANGED_FILES_UNTRACKED__\n".to_string() + } + } else if call_num >= 2 && command.starts_with("git diff") { self.git_diff_after.clone() } else { String::new()