scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix: complete ACP backend verification

Browse source
This commit is contained in:
Bryan Helmkamp 2026-05-11 12:04:27 -04:00
parent 317e440b22
commit ffc7c2a148
No known key found for this signature in database
14 changed files with 187 additions and 140 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,7 +1,9 @@
use crate::command::AcpCommandError;
#[derive(Debug, thiserror::Error)]
pub enum AcpError {
#[error(transparent)]
Command(#[from] crate::command::AcpCommandError),
Command(#[from] AcpCommandError),
#[error(transparent)]
Sandbox(#[from] fabro_sandbox::Error),

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

@ -7,10 +7,11 @@ use agent_client_protocol::schema::{
ProtocolVersion, RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
SelectedPermissionOutcome, SessionNotification, SessionUpdate, StopReason,
};
use agent_client_protocol::util::MatchDispatch;
use agent_client_protocol::{ActiveSession, Agent, Client, SessionMessage};
use agent_client_protocol::util::{MatchDispatch, internal_error};
use agent_client_protocol::{ActiveSession, Agent, Client, Error as ProtocolError, SessionMessage};
use fabro_sandbox::Sandbox;
use fabro_util::time::elapsed_ms;
use tokio::time::{sleep, timeout};
use tokio_util::sync::CancellationToken;
use crate::command::AcpCommand;
@ -90,14 +91,13 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result<AcpRunResult, AcpErr
let outcome = match request.timeout_ms {
Some(timeout_ms) => {
match tokio::time::timeout(Duration::from_millis(timeout_ms), run).await {
Ok(result) => result,
Err(_) => {
state.terminate().await?;
return Err(AcpError::TimedOut {
stderr: state.stderr_tail().await,
});
}
if let Ok(result) = timeout(Duration::from_millis(timeout_ms), run).await {
result
} else {
state.terminate().await?;
return Err(AcpError::TimedOut {
stderr: state.stderr_tail().await,
});
}
}
None => run.await,
@ -113,7 +113,7 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result<AcpRunResult, AcpErr
})
}
fn map_protocol_error(error: agent_client_protocol::Error) -> AcpError {
fn map_protocol_error(error: ProtocolError) -> AcpError {
let message = error.to_string();
if message.contains("ACP turn was cancelled") {
AcpError::Cancelled
@ -160,7 +160,7 @@ async fn read_turn(
cancel_token: &CancellationToken,
on_activity: Option<&Arc<dyn Fn() + Send + Sync>>,
state: &TransportState,
) -> Result<(String, StopReason), agent_client_protocol::Error> {
) -> Result<(String, StopReason), ProtocolError> {
let mut text = String::new();
let mut cancel_sent = false;
@ -188,16 +188,16 @@ async fn read_turn(
SessionMessage::StopReason(stop_reason) => {
return match stop_reason {
StopReason::EndTurn | StopReason::Refusal => Ok((text, stop_reason)),
StopReason::Cancelled => Err(agent_client_protocol::util::internal_error(
"ACP turn was cancelled",
)),
StopReason::Cancelled => {
Err(internal_error("ACP turn was cancelled"))
}
StopReason::MaxTokens | StopReason::MaxTurnRequests => {
Err(agent_client_protocol::util::internal_error(format!(
Err(internal_error(format!(
"ACP prompt stopped with {}: {text}",
stop_reason_name(stop_reason)
)))
}
_ => Err(agent_client_protocol::util::internal_error(format!(
_ => Err(internal_error(format!(
"ACP prompt stopped with {}: {text}",
stop_reason_name(stop_reason)
))),
@ -213,9 +213,9 @@ async fn read_turn(
CancelNotification::new(session.session_id().clone()),
)?;
}
() = tokio::time::sleep(Duration::from_millis(500)), if cancel_sent => {
state.terminate().await.map_err(agent_client_protocol::Error::into_internal_error)?;
return Err(agent_client_protocol::util::internal_error("ACP turn was cancelled"));
() = 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"));
}
}
}

9
lib/crates/fabro-acp/src/test_support.rs
View file

@ -36,6 +36,11 @@ def send(message):
def respond(message, result):
send({"jsonrpc": "2.0", "id": message["id"], "result": result})
def record_methods():
if os.environ.get("ACP_RECORD"):
with open(os.environ["ACP_RECORD"], "w", encoding="utf-8") as record:
record.write("\n".join(methods) + "\n")
for line in sys.stdin:
message = json.loads(line)
method = message.get("method")
@ -100,10 +105,8 @@ for line in sys.stdin:
}
}
})
record_methods()
respond(message, {"stopReason": os.environ.get("ACP_STOP_REASON", "end_turn")})
if os.environ.get("ACP_RECORD"):
with open(os.environ["ACP_RECORD"], "w", encoding="utf-8") as record:
record.write("\n".join(methods) + "\n")
break
else:
send({

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

@ -1,10 +1,19 @@
use std::collections::HashMap;
use std::io::Result as IoResult;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;
use agent_client_protocol::{Client, ConnectTo, Lines};
use fabro_sandbox::{Sandbox, StderrCollector, StdioProcessHandle};
use futures::{AsyncBufReadExt, AsyncWriteExt};
use agent_client_protocol::util::internal_error;
use agent_client_protocol::{
Agent, Client, ConnectTo, Error as ProtocolError, Lines, Result as AcpProtocolResult,
};
use fabro_sandbox::{Result as SandboxResult, Sandbox, StderrCollector, StdioProcessHandle};
use futures::io::BufReader;
use futures::sink::unfold;
use futures::{AsyncBufReadExt, AsyncWriteExt, Stream};
use tokio::sync::Mutex as TokioMutex;
use tokio::time::timeout;
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};
use tokio_util::sync::CancellationToken;
@ -12,15 +21,15 @@ use crate::command::AcpCommand;
#[derive(Clone)]
pub(crate) struct TransportState {
handle: Arc<tokio::sync::Mutex<Option<StdioProcessHandle>>>,
stderr: Arc<tokio::sync::Mutex<Option<StderrCollector>>>,
handle: Arc<TokioMutex<Option<StdioProcessHandle>>>,
stderr: Arc<TokioMutex<Option<StderrCollector>>>,
}
impl TransportState {
pub(crate) fn new() -> Self {
Self {
handle: Arc::new(tokio::sync::Mutex::new(None)),
stderr: Arc::new(tokio::sync::Mutex::new(None)),
handle: Arc::new(TokioMutex::new(None)),
stderr: Arc::new(TokioMutex::new(None)),
}
}
@ -29,7 +38,7 @@ impl TransportState {
*self.stderr.lock().await = Some(stderr);
}
pub(crate) async fn terminate(&self) -> fabro_sandbox::Result<()> {
pub(crate) async fn terminate(&self) -> SandboxResult<()> {
if let Some(handle) = self.handle.lock().await.as_ref().cloned() {
handle.terminate().await?;
}
@ -74,10 +83,7 @@ impl SandboxAcpTransport {
}
impl ConnectTo<Client> for SandboxAcpTransport {
async fn connect_to(
self,
client: impl ConnectTo<agent_client_protocol::Agent>,
) -> agent_client_protocol::Result<()> {
async fn connect_to(self, client: impl ConnectTo<Agent>) -> AcpProtocolResult<()> {
let mut env = self.command.env().clone();
env.extend(self.env);
@ -90,15 +96,15 @@ impl ConnectTo<Client> for SandboxAcpTransport {
Some(self.cancel_token),
)
.await
.map_err(agent_client_protocol::Error::into_internal_error)?;
.map_err(ProtocolError::into_internal_error)?;
let handle = process.handle.clone();
let stderr = process.stderr.clone();
self.state.set_process(handle.clone(), stderr.clone()).await;
let incoming_lines = Box::pin(futures::io::BufReader::new(process.stdout.compat()).lines())
as Pin<Box<dyn futures::Stream<Item = std::io::Result<String>> + Send>>;
let outgoing_sink = Box::pin(futures::sink::unfold(
let incoming_lines = Box::pin(BufReader::new(process.stdout.compat()).lines())
as Pin<Box<dyn Stream<Item = IoResult<String>> + Send>>;
let outgoing_sink = Box::pin(unfold(
process.stdin.compat_write(),
async move |mut writer, line: String| {
let mut bytes = line.into_bytes();
@ -114,13 +120,13 @@ impl ConnectTo<Client> for SandboxAcpTransport {
);
tokio::select! {
result = protocol => {
let _ = tokio::time::timeout(std::time::Duration::from_millis(500), handle.wait()).await;
let _ = timeout(Duration::from_millis(500), handle.wait()).await;
result
}
termination = handle.wait() => {
let termination = termination.map_err(agent_client_protocol::Error::into_internal_error)?;
let termination = termination.map_err(ProtocolError::into_internal_error)?;
let stderr = stderr.tail_string().await;
Err(agent_client_protocol::util::internal_error(format!(
Err(internal_error(format!(
"ACP process exited before protocol completed: termination={termination:?}, stderr={stderr}"
)))
}

94
lib/crates/fabro-acp/tests/session.rs
View file

@ -1,23 +1,28 @@
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use agent_client_protocol::schema::StopReason;
use fabro_acp::{AcpError, AcpRunRequest, resolve_acp_command, run_acp_turn};
use fabro_acp::{AcpError, AcpRunRequest, AcpRunResult, resolve_acp_command, run_acp_turn};
use fabro_model::Provider;
use fabro_sandbox::{LocalSandbox, Sandbox, shell_quote};
use tokio::fs::{read_to_string, write};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
#[tokio::test]
async fn session_lifecycle_initializes_sends_prompt_and_aggregates_text() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let script_path = tempdir.path().join("fake_acp_agent.py");
let record_path = tempdir.path().join("methods.txt");
tokio::fs::write(&script_path, fake_agent_script())
write(&script_path, fake_agent_script())
.await
.unwrap();
.expect("write fake ACP agent");
let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
let command = resolve_acp_command(Provider::OpenAi, Some(&raw_command)).unwrap();
let command =
resolve_acp_command(Provider::OpenAi, Some(&raw_command)).expect("resolve ACP command");
let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.path().to_path_buf()));
let result = run_acp_turn(AcpRunRequest {
@ -34,19 +39,21 @@ async fn session_lifecycle_initializes_sends_prompt_and_aggregates_text() {
on_activity: None,
})
.await
.unwrap();
.expect("run ACP turn");
assert_eq!(result.text, "hello from acp");
assert_eq!(result.stop_reason, StopReason::EndTurn);
assert_eq!(
tokio::fs::read_to_string(record_path).await.unwrap(),
read_to_string(record_path)
.await
.expect("read method record"),
"initialize\nsession/new\nsession/prompt\n"
);
}
#[tokio::test]
async fn permission_request_selects_allow_always() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let permission_path = tempdir.path().join("permission.json");
let result = run_fake_agent(
@ -62,17 +69,19 @@ async fn permission_request_selects_allow_always() {
CancellationToken::new(),
)
.await
.unwrap();
.expect("run ACP turn");
assert_eq!(result.text, "hello from acp");
let permission = tokio::fs::read_to_string(permission_path).await.unwrap();
let permission = read_to_string(permission_path)
.await
.expect("read permission record");
assert!(permission.contains(r#""outcome":"selected""#));
assert!(permission.contains(r#""optionId":"always""#));
}
#[tokio::test]
async fn runs_inside_sandbox_and_uses_requested_cwd() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let cwd_path = tempdir.path().join("session_new.json");
let result = run_fake_agent(
@ -88,26 +97,26 @@ async fn runs_inside_sandbox_and_uses_requested_cwd() {
CancellationToken::new(),
)
.await
.unwrap();
.expect("run ACP turn");
assert_eq!(result.text, "hello from acp");
assert_eq!(
tokio::fs::read_to_string(tempdir.path().join("hello.txt"))
read_to_string(tempdir.path().join("hello.txt"))
.await
.unwrap(),
.expect("read sandbox output file"),
"hello from sandbox\n"
);
assert!(
tokio::fs::read_to_string(cwd_path)
read_to_string(cwd_path)
.await
.unwrap()
.expect("read session/new params")
.contains(&tempdir.path().to_string_lossy().into_owned())
);
}
#[tokio::test]
async fn cancellation_sends_session_cancel_and_returns_cancelled() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let cancel_path = tempdir.path().join("cancel.txt");
let cancel_token = CancellationToken::new();
let cancel_for_task = cancel_token.clone();
@ -128,16 +137,19 @@ async fn cancellation_sends_session_cancel_and_returns_cancelled() {
.await
});
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
sleep(Duration::from_millis(100)).await;
cancel_token.cancel();
let err = task.await.unwrap().unwrap_err();
let err = task
.await
.expect("join cancellation task")
.expect_err("cancelled turn should error");
assert!(matches!(err, AcpError::Cancelled));
}
#[tokio::test]
async fn refusal_stop_reason_returns_text() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let result = run_fake_agent(
tempdir.path(),
@ -146,7 +158,7 @@ async fn refusal_stop_reason_returns_text() {
CancellationToken::new(),
)
.await
.unwrap();
.expect("run ACP turn");
assert_eq!(result.text, "hello from acp");
assert_eq!(result.stop_reason, StopReason::Refusal);
@ -154,7 +166,7 @@ async fn refusal_stop_reason_returns_text() {
#[tokio::test]
async fn max_tokens_stop_reason_returns_partial_text_error() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let err = run_fake_agent(
tempdir.path(),
@ -163,7 +175,7 @@ async fn max_tokens_stop_reason_returns_partial_text_error() {
CancellationToken::new(),
)
.await
.unwrap_err();
.expect_err("max_tokens should return stop reason error");
let AcpError::StopReason { stop_reason, text } = err else {
panic!("expected stop reason error");
@ -174,7 +186,7 @@ async fn max_tokens_stop_reason_returns_partial_text_error() {
#[tokio::test]
async fn max_turn_requests_stop_reason_returns_partial_text_error() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let err = run_fake_agent(
tempdir.path(),
@ -186,7 +198,7 @@ async fn max_turn_requests_stop_reason_returns_partial_text_error() {
CancellationToken::new(),
)
.await
.unwrap_err();
.expect_err("max_turn_requests should return stop reason error");
let AcpError::StopReason { stop_reason, text } = err else {
panic!("expected stop reason error");
@ -197,7 +209,7 @@ async fn max_turn_requests_stop_reason_returns_partial_text_error() {
#[tokio::test]
async fn timeout_terminates_process_and_returns_timeout() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let err = run_fake_agent(
tempdir.path(),
@ -206,14 +218,14 @@ async fn timeout_terminates_process_and_returns_timeout() {
CancellationToken::new(),
)
.await
.unwrap_err();
.expect_err("timeout should error");
assert!(matches!(err, AcpError::TimedOut { .. }));
}
#[tokio::test]
async fn malformed_json_returns_protocol_error() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let err = run_fake_agent(
tempdir.path(),
@ -222,14 +234,14 @@ async fn malformed_json_returns_protocol_error() {
CancellationToken::new(),
)
.await
.unwrap_err();
.expect_err("malformed JSON should error");
assert!(matches!(err, AcpError::Protocol(_)));
}
#[tokio::test]
async fn early_exit_returns_protocol_error_with_stderr() {
let tempdir = tempfile::tempdir().unwrap();
let tempdir = tempfile::tempdir().expect("create tempdir");
let err = run_fake_agent(
tempdir.path(),
@ -238,23 +250,24 @@ async fn early_exit_returns_protocol_error_with_stderr() {
CancellationToken::new(),
)
.await
.unwrap_err();
.expect_err("early exit should error");
assert!(matches!(err, AcpError::Protocol(_)));
}
async fn run_fake_agent(
tempdir: &std::path::Path,
tempdir: &Path,
env: HashMap<String, String>,
timeout_ms: Option<u64>,
cancel_token: CancellationToken,
) -> Result<fabro_acp::AcpRunResult, AcpError> {
) -> Result<AcpRunResult, AcpError> {
let script_path = tempdir.join("fake_acp_agent.py");
tokio::fs::write(&script_path, fake_agent_script())
write(&script_path, fake_agent_script())
.await
.unwrap();
.expect("write fake ACP agent");
let raw_command = format!("python3 {}", shell_quote(&script_path.to_string_lossy()));
let command = resolve_acp_command(Provider::OpenAi, Some(&raw_command)).unwrap();
let command =
resolve_acp_command(Provider::OpenAi, Some(&raw_command)).expect("resolve ACP command");
let sandbox: Arc<dyn Sandbox> = Arc::new(LocalSandbox::new(tempdir.to_path_buf()));
run_acp_turn(AcpRunRequest {
@ -286,6 +299,11 @@ def send(message):
def respond(message, result):
send({"jsonrpc": "2.0", "id": message["id"], "result": result})
def record_methods():
if os.environ.get("ACP_RECORD"):
with open(os.environ["ACP_RECORD"], "w", encoding="utf-8") as record:
record.write("\n".join(methods) + "\n")
for line in sys.stdin:
message = json.loads(line)
method = message.get("method")
@ -360,10 +378,8 @@ for line in sys.stdin:
}
}
})
record_methods()
respond(message, {"stopReason": os.environ.get("ACP_STOP_REASON", "end_turn")})
if os.environ.get("ACP_RECORD"):
with open(os.environ["ACP_RECORD"], "w", encoding="utf-8") as record:
record.write("\n".join(methods) + "\n")
break
else:
send({

9
lib/crates/fabro-cli/tests/it/workflow/acp.rs
View file

@ -14,13 +14,18 @@ use super::{find_run_dir, fixture, has_event, read_conclusion, run_events, run_s
#[test]
fn acp_backend_workflow() {
let context = test_context!();
let mut context = test_context!();
context.write_home(
".fabro/settings.toml",
"[server.auth]\nmethods = [\"dev-token\"]\n",
);
context.isolated_server();
seed_openai_vault(&context.storage_dir);
let fake_agent = fixture("fake_acp_agent.py");
let workflow = context.temp_dir.join("acp_backend.fabro");
context.write_temp(
"acp_backend.fabro",
&format!(
format!(
r#"digraph ACP {{
graph [goal="Exercise ACP backend"]
start [shape=Mdiamond]

2
lib/crates/fabro-sandbox/src/details.rs
View file

@ -1,6 +1,7 @@
use std::collections::BTreeMap;
use anyhow::Result;
#[cfg(any(feature = "docker", feature = "daytona"))]
use chrono::{DateTime, Utc};
use fabro_types::{
RunId, RunSandbox, SandboxDetails, SandboxProvider, SandboxResources, SandboxState,
@ -55,6 +56,7 @@ fn local_details(record: &RunSandbox) -> SandboxDetails {
}
}
#[cfg(any(feature = "docker", feature = "daytona"))]
fn parse_rfc3339_utc(value: &str) -> Option<DateTime<Utc>> {
DateTime::parse_from_rfc3339(value)
.ok()

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

@ -19,14 +19,14 @@ use fabro_github::GitHubCredentials;
use fabro_types::{CommandOutputStream, CommandTermination, RunId};
use fabro_util::time::elapsed_ms;
use futures::StreamExt;
use tokio::io::AsyncWriteExt;
use tokio::sync::OnceCell;
use tokio::io::{AsyncWriteExt, duplex};
use tokio::sync::{Mutex as TokioMutex, OnceCell};
use tokio::{fs, time};
use tokio_util::sync::CancellationToken;
use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
use crate::redact::redact_auth_url;
use crate::sandbox::{optional_timeout, resolve_path};
use crate::sandbox::{StdioProcessControl, optional_timeout, resolve_path};
use crate::{
CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector,
@ -882,11 +882,11 @@ struct DockerStdioProcessControl {
container_id: String,
exec_id: String,
stop_file: String,
termination: tokio::sync::Mutex<Option<CommandTermination>>,
termination: TokioMutex<Option<CommandTermination>>,
}
#[async_trait]
impl crate::sandbox::StdioProcessControl for DockerStdioProcessControl {
impl StdioProcessControl for DockerStdioProcessControl {
async fn terminate(&self) -> crate::Result<()> {
if self.termination.lock().await.is_some() {
return Ok(());
@ -1452,9 +1452,10 @@ impl Sandbox for DockerSandbox {
env_vars: Option<&HashMap<String, String>>,
cancel_token: Option<CancellationToken>,
) -> crate::Result<StdioProcess> {
let effective_dir = working_dir
.map(Self::resolve_container_path)
.unwrap_or_else(|| WORKING_DIRECTORY.to_string());
let effective_dir = working_dir.map_or_else(
|| WORKING_DIRECTORY.to_string(),
Self::resolve_container_path,
);
let env: Option<Vec<String>> =
env_vars.map(|vars| vars.iter().map(|(k, v)| format!("{k}={v}")).collect());
let (stop_file, pid_file) = docker_exec_control_paths();
@ -1483,7 +1484,7 @@ 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) = tokio::io::duplex(64 * 1024);
let (mut stdout_writer, stdout_reader) = duplex(64 * 1024);
tokio::spawn(async move {
while let Some(chunk) = output.next().await {
match chunk {
@ -1511,7 +1512,7 @@ impl Sandbox for DockerSandbox {
container_id,
exec_id,
stop_file,
termination: tokio::sync::Mutex::new(None),
termination: TokioMutex::new(None),
});
if let Some(token) = cancel_token {

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

@ -7,11 +7,12 @@ 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::task::spawn_blocking;
use tokio::{fs, time};
use tokio_util::sync::CancellationToken;
use crate::sandbox::optional_timeout;
use crate::sandbox::{StdioProcessControl, optional_timeout};
use crate::{
CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, StderrCollector,
@ -143,12 +144,12 @@ where
}
struct LocalStdioProcessControl {
child: tokio::sync::Mutex<Child>,
termination: tokio::sync::Mutex<Option<CommandTermination>>,
child: TokioMutex<Child>,
termination: TokioMutex<Option<CommandTermination>>,
}
#[async_trait]
impl crate::sandbox::StdioProcessControl for LocalStdioProcessControl {
impl StdioProcessControl for LocalStdioProcessControl {
async fn terminate(&self) -> crate::Result<()> {
if self.termination.lock().await.is_some() {
return Ok(());
@ -515,8 +516,8 @@ impl Sandbox for LocalSandbox {
stderr_collector.spawn_reader(stderr);
let handle = StdioProcessHandle::new(LocalStdioProcessControl {
child: tokio::sync::Mutex::new(child),
termination: tokio::sync::Mutex::new(None),
child: TokioMutex::new(child),
termination: TokioMutex::new(None),
});
if let Some(token) = cancel_token {

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

@ -10,6 +10,8 @@ use async_trait::async_trait;
use fabro_types::{CommandOutputStream, CommandTermination};
use serde::{Deserialize, Serialize};
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite};
use tokio::sync::Mutex as TokioMutex;
use tokio::task::JoinHandle;
use tokio::time;
use tokio_util::sync::CancellationToken;
@ -688,7 +690,7 @@ pub struct StdioProcess {
#[derive(Debug, Clone)]
pub struct StderrCollector {
inner: Arc<tokio::sync::Mutex<Vec<u8>>>,
inner: Arc<TokioMutex<Vec<u8>>>,
max_bytes: usize,
}
@ -696,7 +698,7 @@ impl StderrCollector {
#[must_use]
pub fn new(max_bytes: usize) -> Self {
Self {
inner: Arc::new(tokio::sync::Mutex::new(Vec::new())),
inner: Arc::new(TokioMutex::new(Vec::new())),
max_bytes,
}
}
@ -715,7 +717,7 @@ impl StderrCollector {
String::from_utf8_lossy(&tail).into_owned()
}
pub fn spawn_reader<R>(&self, mut reader: R) -> tokio::task::JoinHandle<()>
pub fn spawn_reader<R>(&self, mut reader: R) -> JoinHandle<()>
where
R: AsyncRead + Unpin + Send + 'static,
{

8
lib/crates/fabro-sandbox/src/test_support.rs
View file

@ -4,8 +4,10 @@ use std::sync::Mutex;
use async_trait::async_trait;
use fabro_types::CommandTermination;
use tokio::fs;
use tokio::io::duplex;
use tokio_util::sync::CancellationToken;
use crate::sandbox::StdioProcessControl;
use crate::{
DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent,
SandboxEventCallback, StderrCollector, StdioProcess, StdioProcessHandle,
@ -110,7 +112,7 @@ impl Default for MockSandbox {
struct MockStdioProcessControl;
#[async_trait]
impl crate::sandbox::StdioProcessControl for MockStdioProcessControl {
impl StdioProcessControl for MockStdioProcessControl {
async fn terminate(&self) -> crate::Result<()> {
Ok(())
}
@ -224,8 +226,8 @@ impl Sandbox for MockSandbox {
.lock()
.expect("captured_env_vars lock poisoned") = env_vars.cloned();
let (stdin, _stdin_read) = tokio::io::duplex(1024);
let (_stdout_write, stdout) = tokio::io::duplex(1024);
let (stdin, _stdin_read) = duplex(1024);
let (_stdout_write, stdout) = duplex(1024);
Ok(StdioProcess {
stdin: Box::pin(stdin),
stdout: Box::pin(stdout),

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

@ -548,6 +548,10 @@ for line in sys.stdin:
"#
}
#[expect(
clippy::disallowed_methods,
reason = "unit test initializes an isolated git repository with the system git binary"
)]
fn init_git(path: &std::path::Path) {
let output = std::process::Command::new("git")
.arg("init")

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

@ -781,9 +781,9 @@ pub(crate) fn process_env_var(name: &str) -> Option<String> {
/// Routes codergen invocations to API, CLI, or ACP backends based on node
/// attributes and model type.
pub struct BackendRouter {
api_backend: Box<dyn CodergenBackend>,
cli_backend: AgentCliBackend,
acp_backend: AgentAcpBackend,
api: Box<dyn CodergenBackend>,
cli: AgentCliBackend,
acp: AgentAcpBackend,
}
impl BackendRouter {
@ -794,13 +794,13 @@ impl BackendRouter {
acp_backend: AgentAcpBackend,
) -> Self {
Self {
api_backend,
cli_backend,
acp_backend,
api: api_backend,
cli: cli_backend,
acp: acp_backend,
}
}
fn select_backend(&self, node: &Node) -> Result<SelectedBackend, Error> {
fn select_backend(node: &Node) -> Result<SelectedBackend, Error> {
match node.backend() {
None => {
if node.model().is_some_and(is_cli_only_model) {
@ -818,7 +818,7 @@ impl BackendRouter {
}
}
fn select_one_shot_backend(&self, node: &Node) -> Result<SelectedBackend, Error> {
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),
@ -829,8 +829,8 @@ impl BackendRouter {
}
#[cfg(test)]
fn should_use_cli(&self, node: &Node) -> bool {
matches!(self.select_backend(node), Ok(SelectedBackend::Cli))
fn should_use_cli(node: &Node) -> bool {
matches!(Self::select_backend(node), Ok(SelectedBackend::Cli))
}
}
@ -854,9 +854,9 @@ impl CodergenBackend for BackendRouter {
tool_hooks: Option<Arc<dyn fabro_agent::ToolHookCallback>>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
match self.select_backend(node)? {
match Self::select_backend(node)? {
SelectedBackend::Api => {
self.api_backend
self.api
.run(
node,
prompt,
@ -870,7 +870,7 @@ impl CodergenBackend for BackendRouter {
.await
}
SelectedBackend::Cli => {
self.cli_backend
self.cli
.run(
node,
prompt,
@ -884,7 +884,7 @@ impl CodergenBackend for BackendRouter {
.await
}
SelectedBackend::Acp => {
self.acp_backend
self.acp
.run(
node,
prompt,
@ -910,9 +910,9 @@ impl CodergenBackend for BackendRouter {
sandbox: &Arc<dyn Sandbox>,
cancel_token: CancellationToken,
) -> Result<CodergenResult, Error> {
match self.select_one_shot_backend(node)? {
match Self::select_one_shot_backend(node)? {
SelectedBackend::Acp => {
self.acp_backend
self.acp
.one_shot(
node,
prompt,
@ -925,7 +925,7 @@ impl CodergenBackend for BackendRouter {
.await
}
SelectedBackend::Api | SelectedBackend::Cli => {
self.api_backend
self.api
.one_shot(
node,
prompt,
@ -941,7 +941,7 @@ impl CodergenBackend for BackendRouter {
}
async fn shutdown(&self, emitter: &Arc<Emitter>) {
self.api_backend.shutdown(emitter).await;
self.api.shutdown(emitter).await;
}
}
@ -1387,16 +1387,14 @@ mod tests {
node.attrs
.insert("backend".to_string(), AttrValue::String("cli".to_string()));
let router = test_router();
assert!(router.should_use_cli(&node));
assert!(BackendRouter::should_use_cli(&node));
}
#[test]
fn router_uses_api_by_default() {
let node = Node::new("test");
let router = test_router();
assert!(!router.should_use_cli(&node));
assert!(!BackendRouter::should_use_cli(&node));
}
#[test]
@ -1407,8 +1405,7 @@ mod tests {
AttrValue::String("claude-opus-4-6".to_string()),
);
let router = test_router();
assert!(!router.should_use_cli(&node));
assert!(!BackendRouter::should_use_cli(&node));
}
#[test]
@ -1417,8 +1414,10 @@ mod tests {
node.attrs
.insert("backend".to_string(), AttrValue::String("api".to_string()));
let router = test_router();
assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Api);
assert_eq!(
BackendRouter::select_backend(&node).unwrap(),
SelectedBackend::Api
);
}
#[test]
@ -1427,8 +1426,10 @@ mod tests {
node.attrs
.insert("backend".to_string(), AttrValue::String("cli".to_string()));
let router = test_router();
assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Cli);
assert_eq!(
BackendRouter::select_backend(&node).unwrap(),
SelectedBackend::Cli
);
}
#[test]
@ -1437,8 +1438,10 @@ mod tests {
node.attrs
.insert("backend".to_string(), AttrValue::String("acp".to_string()));
let router = test_router();
assert_eq!(router.select_backend(&node).unwrap(), SelectedBackend::Acp);
assert_eq!(
BackendRouter::select_backend(&node).unwrap(),
SelectedBackend::Acp
);
}
#[test]
@ -1449,8 +1452,7 @@ mod tests {
AttrValue::String("codex".to_string()),
);
let router = test_router();
let err = router.select_backend(&node).unwrap_err();
let err = BackendRouter::select_backend(&node).unwrap_err();
assert_eq!(
err.to_string(),
"Validation error: unsupported LLM backend \"codex\"; expected one of: api, cli, acp"

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

@ -687,6 +687,7 @@ mod tests {
use fabro_types::{EventBody, RunEvent, RunId, WorkflowSettings, fixtures};
use fabro_vault::{SecretType, Vault};
use object_store::memory::InMemory;
use tokio::fs::{create_dir_all, write};
use tokio::sync::RwLock as AsyncRwLock;
use super::*;
@ -1015,9 +1016,9 @@ mod tests {
async fn initialize_executes_acp_backend_node_from_registry() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
create_dir_all(&run_dir).await.unwrap();
let script_path = temp.path().join("fake_acp_agent.py");
std::fs::write(&script_path, fake_acp_agent_script()).unwrap();
write(&script_path, fake_acp_agent_script()).await.unwrap();
let source = format!(
r#"digraph test {{