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

fix: simplify ACP backend cleanup

Browse source 
Address review findings around ACP process control, backend routing reuse, and launch environment duplication.
This commit is contained in:
Bryan Helmkamp 2026-05-11 15:18:55 -04:00
parent 2f8958ccaa
commit 537ab7a511
No known key found for this signature in database
19 changed files with 615 additions and 451 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -2453,6 +2453,7 @@ version = "0.230.0-nightly.0"
dependencies = [
"fabro-graphviz",
"fabro-model",
"fabro-types",
"serde",
"thiserror 2.0.18",
]

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

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

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

@ -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<AcpRunResult, AcpError> {
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<AcpRunResult, AcpErr
session.send_prompt(prompt)?;
read_turn(
&mut session,
&cancel_token,
&read_cancel_token,
on_activity.as_ref(),
&state_for_run,
)
@ -90,21 +97,34 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result<AcpRunResult, AcpErr
.await
});
let outcome = match request.timeout_ms {
Some(timeout_ms) => {
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<AcpRunResult, AcpErr
}
};
match stop_reason {
StopReason::EndTurn | StopReason::Refusal => {}
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<AcpRunResult, AcpErr
}
fn map_protocol_error(error: ProtocolError) -> 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:?}"))
}

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

@ -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<DockerStdioProcessState>,
}
#[derive(Default)]
struct DockerStdioProcessState {
stop_requested: TokioMutex<bool>,
termination: TokioMutex<Option<StdioProcessTermination>>,
stop_requested: AtomicBool,
termination: TokioMutex<Option<StdioProcessTermination>>,
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<DockerStdioProcessState>,
) {
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]

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

@ -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<String, String>>,
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<R>(mut pipe: Option<R>, stream: CommandOutputStream) -> String
where
R: AsyncRead + Unpin,
@ -143,37 +171,74 @@ where
buf
}
type LocalStdioOutcome = Result<StdioProcessTermination, String>;
struct LocalStdioProcessControl {
child: TokioMutex<Child>,
termination: TokioMutex<Option<StdioProcessTermination>>,
terminate_tx: watch::Sender<bool>,
termination_rx: watch::Receiver<Option<LocalStdioOutcome>>,
}
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<StdioProcessTermination> {
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<StdioProcessTermination> {
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<ExecResult> {
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<ExecStreamingResult> {
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<String, String>>,
cancel_token: Option<CancellationToken>,
) -> crate::Result<StdioProcess> {
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();

70
lib/crates/fabro-store/src/run_state.rs
View file

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

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

@ -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<Result<LlmBackend, strum::ParseError>> {
self.backend().map(str::parse)
}
#[must_use]
pub fn acp_command(&self) -> Option<&str> {
self.str_attr("acp_command")

2
lib/crates/fabro-types/src/lib.rs
View file

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

28
lib/crates/fabro-types/src/llm_backend.rs Normal file
View file

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

3
lib/crates/fabro-validate/Cargo.toml
View file

@ -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 }
thiserror = { workspace = true }

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

@ -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<dyn LintRule> {
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::<LlmBackend>().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)),
});
}
}

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

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

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

@ -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<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;
const DIFF_MARKER: &str = "__FABRO_CHANGED_FILES_DIFF__";
const UNTRACKED_MARKER: &str = "__FABRO_CHANGED_FILES_UNTRACKED__";
pub async fn detect_changed_files(sandbox: &Arc<dyn Sandbox>) -> Vec<String> {
let mut files: Vec<String> = 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<String>, Option<String>) {
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<String> = 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<Item = String> + '_ {
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::<Vec<_>>();
assert_eq!(files, vec!["src/main.rs", "README.md"]);
}
}

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

@ -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<SelectedBackend, Error> {
match node.backend() {
fn select_backend(node: &Node) -> Result<LlmBackend, Error> {
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<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"
))),
fn select_one_shot_backend(node: &Node) -> Result<LlmBackend, Error> {
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<CodergenResult, Error> {
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<CodergenResult, Error> {
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
);
}

107
lib/crates/fabro-workflow/src/handler/llm/launch_env.rs Normal file
View file

@ -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<dyn ToolEnvProvider>>,
pub github_token_refresh_managed: bool,
pub stage_label: &'static str,
pub emitter: &'a Arc<Emitter>,
pub sandbox: &'a Arc<dyn Sandbox>,
pub cancel_token: &'a CancellationToken,
}
pub(crate) async fn resolve_agent_launch_env(
request: AgentLaunchEnvRequest<'_>,
) -> Result<HashMap<String, String>, 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)
}

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

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

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

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

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

@ -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<dyn CredentialSource>,
cli_resolver: Option<CredentialResolver>,
) -> Result<(Arc<HandlerRegistry>, 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<AsyncRwLock<Vault>>>) -> Arc<dyn CredentialSource> {
match vault {
Some(vault) => Arc::new(VaultCredentialSource::new(vault)),

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

@ -9659,14 +9659,25 @@ impl fabro_agent::Sandbox for CliTestEnv {
) -> fabro_sandbox::Result<fabro_agent::ExecResult> {
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()