diff --git a/lib/crates/fabro-acp/src/error.rs b/lib/crates/fabro-acp/src/error.rs index 79304d654..631620634 100644 --- a/lib/crates/fabro-acp/src/error.rs +++ b/lib/crates/fabro-acp/src/error.rs @@ -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), diff --git a/lib/crates/fabro-acp/src/session.rs b/lib/crates/fabro-acp/src/session.rs index 415e1c80b..e3af3d3c2 100644 --- a/lib/crates/fabro-acp/src/session.rs +++ b/lib/crates/fabro-acp/src/session.rs @@ -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 { - 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 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>, 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")); } } } diff --git a/lib/crates/fabro-acp/src/test_support.rs b/lib/crates/fabro-acp/src/test_support.rs index 6aab49027..167d639d6 100644 --- a/lib/crates/fabro-acp/src/test_support.rs +++ b/lib/crates/fabro-acp/src/test_support.rs @@ -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({ diff --git a/lib/crates/fabro-acp/src/transport.rs b/lib/crates/fabro-acp/src/transport.rs index 81b4a8ad6..d1c8db029 100644 --- a/lib/crates/fabro-acp/src/transport.rs +++ b/lib/crates/fabro-acp/src/transport.rs @@ -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>>, - stderr: Arc>>, + handle: Arc>>, + stderr: Arc>>, } 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 for SandboxAcpTransport { - async fn connect_to( - self, - client: impl ConnectTo, - ) -> agent_client_protocol::Result<()> { + async fn connect_to(self, client: impl ConnectTo) -> AcpProtocolResult<()> { let mut env = self.command.env().clone(); env.extend(self.env); @@ -90,15 +96,15 @@ impl ConnectTo 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> + Send>>; - let outgoing_sink = Box::pin(futures::sink::unfold( + let incoming_lines = Box::pin(BufReader::new(process.stdout.compat()).lines()) + as Pin> + 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 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}" ))) } diff --git a/lib/crates/fabro-acp/tests/session.rs b/lib/crates/fabro-acp/tests/session.rs index 0ff425ac4..3812a625a 100644 --- a/lib/crates/fabro-acp/tests/session.rs +++ b/lib/crates/fabro-acp/tests/session.rs @@ -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 = 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, timeout_ms: Option, cancel_token: CancellationToken, -) -> Result { +) -> Result { 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 = 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({ diff --git a/lib/crates/fabro-cli/tests/it/workflow/acp.rs b/lib/crates/fabro-cli/tests/it/workflow/acp.rs index 09ca27c63..780e9f89c 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/acp.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/acp.rs @@ -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] diff --git a/lib/crates/fabro-sandbox/src/details.rs b/lib/crates/fabro-sandbox/src/details.rs index 2abb93cad..317b8ee6d 100644 --- a/lib/crates/fabro-sandbox/src/details.rs +++ b/lib/crates/fabro-sandbox/src/details.rs @@ -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::parse_from_rfc3339(value) .ok() diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index af39b82e5..a541fc416 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/lib/crates/fabro-sandbox/src/docker.rs @@ -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>, + termination: TokioMutex>, } #[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>, cancel_token: Option, ) -> crate::Result { - 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> = 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 { diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index a373442a2..f6ed2b9a5 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -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, - termination: tokio::sync::Mutex>, + child: TokioMutex, + termination: TokioMutex>, } #[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 { diff --git a/lib/crates/fabro-sandbox/src/sandbox.rs b/lib/crates/fabro-sandbox/src/sandbox.rs index 14b424b88..90655d48e 100644 --- a/lib/crates/fabro-sandbox/src/sandbox.rs +++ b/lib/crates/fabro-sandbox/src/sandbox.rs @@ -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>>, + inner: Arc>>, 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(&self, mut reader: R) -> tokio::task::JoinHandle<()> + pub fn spawn_reader(&self, mut reader: R) -> JoinHandle<()> where R: AsyncRead + Unpin + Send + 'static, { diff --git a/lib/crates/fabro-sandbox/src/test_support.rs b/lib/crates/fabro-sandbox/src/test_support.rs index 09df77fbe..bd2bcf013 100644 --- a/lib/crates/fabro-sandbox/src/test_support.rs +++ b/lib/crates/fabro-sandbox/src/test_support.rs @@ -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), diff --git a/lib/crates/fabro-workflow/src/handler/llm/acp.rs b/lib/crates/fabro-workflow/src/handler/llm/acp.rs index 72935b864..aecbea6c7 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/acp.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/acp.rs @@ -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") diff --git a/lib/crates/fabro-workflow/src/handler/llm/cli.rs b/lib/crates/fabro-workflow/src/handler/llm/cli.rs index 2f91c5c72..4faa0c43f 100644 --- a/lib/crates/fabro-workflow/src/handler/llm/cli.rs +++ b/lib/crates/fabro-workflow/src/handler/llm/cli.rs @@ -781,9 +781,9 @@ pub(crate) fn process_env_var(name: &str) -> Option { /// Routes codergen invocations to API, CLI, or ACP backends based on node /// attributes and model type. pub struct BackendRouter { - api_backend: Box, - cli_backend: AgentCliBackend, - acp_backend: AgentAcpBackend, + api: Box, + 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 { + fn select_backend(node: &Node) -> Result { 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 { + fn select_one_shot_backend(node: &Node) -> Result { 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>, cancel_token: CancellationToken, ) -> Result { - 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, cancel_token: CancellationToken, ) -> Result { - 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) { - 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" diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs index 73e8a8962..fcf933c6b 100644 --- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs @@ -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 {{