diff --git a/docs/plans/2026-05-11-add-acp-backend-test-plan.md b/docs/plans/2026-05-11-add-acp-backend-test-plan.md index c957ca96f..3a54ee109 100644 --- a/docs/plans/2026-05-11-add-acp-backend-test-plan.md +++ b/docs/plans/2026-05-11-add-acp-backend-test-plan.md @@ -287,15 +287,15 @@ The accepted testing strategy still holds, with scoped additions from the implem - Disposition: new - Harness: repository test suites named by the implementation plan. - Preconditions: all implementation tasks complete. - - Actions: run: - `ulimit -n 4096 && cargo nextest run -p fabro-acp`; - `ulimit -n 4096 && cargo nextest run -p fabro-sandbox -E 'test(stdio)'`; - `ulimit -n 4096 && cargo nextest run -p fabro-workflow -E 'test(router_) | test(acp_backend) | test(agent_acp) | test(initialize.*acp)'`; - `ulimit -n 4096 && cargo nextest run -p fabro-validate -E 'test(backend_valid)'`; - `ulimit -n 4096 && cargo nextest run -p fabro-store -E 'test(agent_acp)'`; - `ulimit -n 4096 && cargo nextest run -p fabro-server -E 'test(acp.*steer|steer.*acp)'`; - `ulimit -n 4096 && cargo nextest run -p fabro-cli -E 'test(acp_backend_workflow)'`. - - Expected outcome: every targeted suite passes without live provider credentials. Source of truth: accepted strategy final verification and implementation plan Task 10. + - Actions: run: + `ulimit -n 4096 && cargo nextest run -p fabro-acp --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-sandbox --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-workflow --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-validate --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-store --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-server --run-ignored all --no-fail-fast`; + `ulimit -n 4096 && cargo nextest run -p fabro-cli --run-ignored all --no-fail-fast`. + - Expected outcome: every suite passes without skipped tests or live provider credentials. Source of truth: accepted strategy final verification and implementation plan Task 10. - Interactions: all changed crates and user-visible workflow/server surfaces. 30. **Workspace-wide build, formatting, and lint gates pass** @@ -303,8 +303,8 @@ The accepted testing strategy still holds, with scoped additions from the implem - Disposition: existing - Harness: repository-wide Cargo/rustfmt/clippy commands. - Preconditions: targeted tests pass. - - Actions: run `cargo build --workspace`, `cargo +nightly-2026-04-14 fmt --check --all`, and `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings`. - - Expected outcome: build, formatting, and clippy all pass. Source of truth: repository `AGENTS.md` build/test commands. + - Actions: run `cargo build --workspace`, `ulimit -n 4096 && cargo nextest run --workspace --run-ignored all --no-fail-fast`, `cargo +nightly-2026-04-14 fmt --check --all`, and `cargo +nightly-2026-04-14 clippy --workspace --all-targets -- -D warnings`. + - Expected outcome: build, workspace tests, formatting, and clippy all pass with zero skipped tests. Source of truth: repository `AGENTS.md` build/test commands and the no-skipped-tests final-run requirement. - Interactions: full workspace dependency graph, feature flags, generated code boundaries. ## Coverage Summary diff --git a/lib/crates/fabro-acp/src/session.rs b/lib/crates/fabro-acp/src/session.rs index e3af3d3c2..8349379e9 100644 --- a/lib/crates/fabro-acp/src/session.rs +++ b/lib/crates/fabro-acp/src/session.rs @@ -41,6 +41,7 @@ pub async fn run_acp_turn(request: AcpRunRequest) -> Result Result Result run.await, }; - let (text, stop_reason) = outcome.map_err(map_protocol_error)?; + let (text, stop_reason) = match outcome { + Ok(result) => result, + Err(_) if run_cancel_token.is_cancelled() => { + state.terminate().await?; + return Err(AcpError::Cancelled); + } + Err(error) => { + state.terminate().await?; + return Err(map_protocol_error(error)); + } + }; + state.terminate().await?; let stderr = state.stderr_tail().await; Ok(AcpRunResult { text, diff --git a/lib/crates/fabro-acp/src/transport.rs b/lib/crates/fabro-acp/src/transport.rs index d1c8db029..a5ad283fa 100644 --- a/lib/crates/fabro-acp/src/transport.rs +++ b/lib/crates/fabro-acp/src/transport.rs @@ -120,6 +120,9 @@ impl ConnectTo for SandboxAcpTransport { ); tokio::select! { result = protocol => { + if let Err(err) = handle.terminate().await { + tracing::warn!(error = %err, "Failed to terminate ACP process after protocol completion"); + } let _ = timeout(Duration::from_millis(500), handle.wait()).await; result } diff --git a/lib/crates/fabro-acp/tests/session.rs b/lib/crates/fabro-acp/tests/session.rs index 3812a625a..fd7e09ff2 100644 --- a/lib/crates/fabro-acp/tests/session.rs +++ b/lib/crates/fabro-acp/tests/session.rs @@ -8,6 +8,7 @@ use fabro_acp::{AcpError, AcpRunRequest, AcpRunResult, resolve_acp_command, run_ use fabro_model::Provider; use fabro_sandbox::{LocalSandbox, Sandbox, shell_quote}; use tokio::fs::{read_to_string, write}; +use tokio::process::Command; use tokio::time::sleep; use tokio_util::sync::CancellationToken; @@ -147,6 +148,62 @@ async fn cancellation_sends_session_cancel_and_returns_cancelled() { assert!(matches!(err, AcpError::Cancelled)); } +#[tokio::test] +async fn pre_session_cancellation_returns_cancelled() { + let tempdir = tempfile::tempdir().expect("create tempdir"); + let cancel_token = CancellationToken::new(); + cancel_token.cancel(); + + let err = run_fake_agent( + tempdir.path(), + HashMap::from([("ACP_MODE".to_string(), "slow_initialize".to_string())]), + Some(1_000), + cancel_token, + ) + .await + .expect_err("pre-session cancellation should error"); + + assert!(matches!(err, AcpError::Cancelled)); +} + +#[tokio::test] +async fn successful_turn_terminates_lingering_agent_process() { + let tempdir = tempfile::tempdir().expect("create tempdir"); + let pid_path = tempdir.path().join("agent.pid"); + + let result = run_fake_agent( + tempdir.path(), + HashMap::from([ + ("ACP_MODE".to_string(), "linger_after_response".to_string()), + ( + "ACP_PID_RECORD".to_string(), + pid_path.to_string_lossy().into_owned(), + ), + ]), + Some(5_000), + CancellationToken::new(), + ) + .await + .expect("run ACP turn"); + + sleep(Duration::from_millis(100)).await; + let pid = read_to_string(&pid_path).await.expect("read agent pid"); + let still_running = process_is_running(pid.trim()).await; + if still_running { + let _ = Command::new("kill") + .arg("-TERM") + .arg(pid.trim()) + .status() + .await; + } + + assert_eq!(result.text, "hello from acp"); + assert!( + !still_running, + "successful ACP turn should not leave lingering agent process" + ); +} + #[tokio::test] async fn refusal_stop_reason_returns_text() { let tempdir = tempfile::tempdir().expect("create tempdir"); @@ -283,16 +340,53 @@ async fn run_fake_agent( .await } +async fn process_is_running(pid: &str) -> bool { + let Ok(status) = Command::new("kill").arg("-0").arg(pid).status().await else { + return false; + }; + if !status.success() { + return false; + } + + let Ok(output) = Command::new("ps") + .args(["-ww", "-o", "stat=", "-p", pid]) + .output() + .await + else { + return true; + }; + if !output.status.success() { + return false; + } + String::from_utf8_lossy(&output.stdout) + .chars() + .find(|ch| !ch.is_whitespace()) + .is_none_or(|state| !matches!(state, 'Z' | 'z')) +} + fn fake_agent_script() -> &'static str { r#" import json import os +import signal import sys import time methods = [] session_id = "sess-1" +if os.environ.get("ACP_PID_RECORD"): + with open(os.environ["ACP_PID_RECORD"], "w", encoding="utf-8") as record: + record.write(str(os.getpid())) + +def handle_sigterm(signum, frame): + if os.environ.get("ACP_LINGER_TERMINATED"): + with open(os.environ["ACP_LINGER_TERMINATED"], "w", encoding="utf-8") as record: + record.write("terminated\n") + sys.exit(0) + +signal.signal(signal.SIGTERM, handle_sigterm) + def send(message): print(json.dumps(message), flush=True) @@ -310,6 +404,8 @@ for line in sys.stdin: methods.append(method) if method == "initialize": + if os.environ.get("ACP_MODE") == "slow_initialize": + time.sleep(60) respond(message, {"protocolVersion": 1, "agentCapabilities": {}}) elif method == "session/new": if os.environ.get("ACP_SESSION_NEW_PARAMS"): @@ -380,6 +476,9 @@ for line in sys.stdin: }) record_methods() respond(message, {"stopReason": os.environ.get("ACP_STOP_REASON", "end_turn")}) + if mode == "linger_after_response": + while True: + time.sleep(1) break else: send({ diff --git a/lib/crates/fabro-cli/tests/it/cmd/doctor.rs b/lib/crates/fabro-cli/tests/it/cmd/doctor.rs index 075a74f98..9c9092b47 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/doctor.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/doctor.rs @@ -5,7 +5,11 @@ use std::process::Output; +use fabro_auth::{AuthCredential, AuthDetails}; +use fabro_config::Storage; +use fabro_model::Provider; use fabro_test::{fabro_snapshot, test_context, twin_openai}; +use fabro_vault::{SecretType, Vault}; async fn run_success_output(mut cmd: assert_cmd::Command) -> Output { tokio::task::spawn_blocking(move || cmd.assert().success().get_output().clone()) @@ -13,6 +17,35 @@ async fn run_success_output(mut cmd: assert_cmd::Command) -> Output { .expect("blocking command task should complete") } +fn toml_path(path: &std::path::Path) -> String { + path.display() + .to_string() + .replace('\\', "\\\\") + .replace('"', "\\\"") +} + +fn seed_openai_vault(storage_dir: &std::path::Path, base_url: &str, api_key: &str) { + let mut vault = + Vault::load(Storage::new(storage_dir).secrets_path()).expect("test vault should load"); + vault + .set( + "openai", + &serde_json::to_string(&AuthCredential { + provider: Provider::OpenAi, + details: AuthDetails::ApiKey { + key: api_key.to_string(), + }, + }) + .expect("OpenAI test credential should serialize"), + SecretType::Credential, + None, + ) + .expect("OpenAI credential should store in test vault"); + vault + .set("OPENAI_BASE_URL", base_url, SecretType::Environment, None) + .expect("OpenAI base URL should store in test vault"); +} + #[test] fn help() { let context = test_context!(); @@ -64,9 +97,28 @@ fn live_doctor() { #[fabro_macros::e2e_test(twin)] async fn twin_doctor() { - let context = test_context!(); + let mut context = test_context!(); let twin = twin_openai().await; let namespace = format!("{}::{}", module_path!(), line!()); + let storage_dir = context.temp_dir.join("doctor-server-storage"); + context.write_home( + ".fabro/settings.toml", + format!( + r#"[server.storage] +root = "{}" + +[server.auth] +methods = ["dev-token"] + +[server.integrations.github] +strategy = "app" +"#, + toml_path(&storage_dir) + ), + ); + seed_openai_vault(&storage_dir, &twin.base_url, &namespace); + context.isolated_server(); + let mut cmd = context.doctor(); cmd.arg("--verbose"); cmd.env_clear(); diff --git a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs index 0eadec381..ee26e2fbf 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs @@ -26,14 +26,17 @@ fn local_run_lifecycle() { }; // 1. Run a workflow - cmd(&[ - "run", - "--auto-approve", - "--sandbox", - "local", - fixture("command_pipeline.fabro").to_str().unwrap(), - ]) - .success(); + context + .run_cmd() + .args([ + "--auto-approve", + "--sandbox", + "local", + fixture("command_pipeline.fabro").to_str().unwrap(), + ]) + .timeout(timeout_for("local")) + .assert() + .success(); // 2. ps -a --json — should list exactly one run let label = context.test_case_label(); diff --git a/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs b/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs index 1718e8428..447ef05b4 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/command_agent_mixed.rs @@ -49,8 +49,8 @@ fn scenario_command_agent_mixed(sandbox: &str) { let export_dir = dump_export(&context, &run_id_for(&run_dir)); let stdout = - std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("stdout.log")) - .expect("verify stdout.log should exist"); + std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("output.log")) + .expect("verify output.log should exist"); assert!( stdout.contains("SCENARIO_FLAG_42"), "verify stdout should contain SCENARIO_FLAG_42, got: {stdout}" diff --git a/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs b/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs index ff6e583a9..2f0d4c94e 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs @@ -49,8 +49,8 @@ fn scenario_command_pipeline(sandbox: &str) { let export_dir = dump_export(&context, &run_id_for(&run_dir)); let stdout1 = - std::fs::read_to_string(stage_dump_dir(&export_dir, "step1@1").join("stdout.log")) - .expect("step1 stdout.log should exist"); + std::fs::read_to_string(stage_dump_dir(&export_dir, "step1@1").join("output.log")) + .expect("step1 output.log should exist"); assert!( stdout1.contains("hello-from-step1"), "step1 stdout should contain hello-from-step1, got: {stdout1}" diff --git a/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs b/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs index fef6e0578..3ba6c9caa 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs @@ -74,8 +74,8 @@ fn scenario_full_stack(sandbox: &str) { // Verify node stdout should contain PASS let export_dir = dump_export(&context, &run_id_for(&run_dir)); let stdout = - std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("stdout.log")) - .expect("verify stdout.log should exist"); + std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("output.log")) + .expect("verify output.log should exist"); assert!( stdout.contains("PASS"), "verify stdout should contain PASS, got: {stdout}" diff --git a/lib/crates/fabro-cli/tests/it/workflow/hooks.rs b/lib/crates/fabro-cli/tests/it/workflow/hooks.rs index c2eecb762..85638b0af 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/hooks.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/hooks.rs @@ -11,9 +11,17 @@ use std::process::Output; -use fabro_test::{TestMode, TwinScenario, TwinScenarios, TwinToolCall, test_context, twin_openai}; +use fabro_auth::{AuthCredential, AuthDetails}; +use fabro_config::Storage; +use fabro_model::Provider; +use fabro_test::{ + TestMode, TwinOpenAi, TwinScenario, TwinScenarios, TwinToolCall, expect_reqwest_status, + test_context, twin_openai, +}; +use fabro_vault::{SecretType, Vault}; -use super::{find_run_dir, read_conclusion}; +use super::run_id_for; +use crate::cmd::support::server_endpoint; async fn run_success_output(mut cmd: assert_cmd::Command) -> Output { tokio::task::spawn_blocking(move || cmd.assert().success().get_output().clone()) @@ -51,6 +59,73 @@ fn stage_provider() -> &'static str { } } +fn toml_path(path: &std::path::Path) -> String { + path.display() + .to_string() + .replace('\\', "\\\\") + .replace('"', "\\\"") +} + +fn twin_server_storage_dir(context: &fabro_test::TestContext) -> std::path::PathBuf { + context.temp_dir.join("hook-server-storage") +} + +fn settings_with_hook(context: &fabro_test::TestContext, hook: &str) -> String { + if TestMode::from_env().is_twin() { + format!( + r#"[server.storage] +root = "{}" + +[server.auth] +methods = ["dev-token"] + +{hook}"#, + toml_path(&twin_server_storage_dir(context)), + ) + } else { + hook.to_string() + } +} + +fn write_hook_settings(context: &fabro_test::TestContext, hook: &str) { + let settings = settings_with_hook(context, hook); + if settings.trim().is_empty() { + return; + } + context.write_home(".fabro/settings.toml", settings); +} + +fn seed_openai_vault(storage_dir: &std::path::Path, base_url: &str, api_key: &str) { + let mut vault = + Vault::load(Storage::new(storage_dir).secrets_path()).expect("test vault should load"); + vault + .set( + "openai", + &serde_json::to_string(&AuthCredential { + provider: Provider::OpenAi, + details: AuthDetails::ApiKey { + key: api_key.to_string(), + }, + }) + .expect("OpenAI test credential should serialize"), + SecretType::Credential, + None, + ) + .expect("OpenAI credential should store in test vault"); + vault + .set("OPENAI_BASE_URL", base_url, SecretType::Environment, None) + .expect("OpenAI base URL should store in test vault"); +} + +fn configure_twin_server( + context: &mut fabro_test::TestContext, + twin: &TwinOpenAi, + namespace: &str, +) { + seed_openai_vault(&twin_server_storage_dir(context), &twin.base_url, namespace); + context.isolated_server(); +} + fn write_workflow(context: &fabro_test::TestContext, name: &str, dot: &str) -> std::path::PathBuf { context.write_temp(name, dot); context.temp_dir.join(name) @@ -69,9 +144,24 @@ fn configure_hook_env(cmd: &mut assert_cmd::Command, hook_model: &str) { cmd.arg("--model").arg(hook_model); } -fn conclusion_status(context: &fabro_test::TestContext) -> String { - let run_dir = find_run_dir(&context); - read_conclusion(&run_dir)["status"] +async fn conclusion_status(context: &fabro_test::TestContext) -> String { + let run_dir = context.single_run_dir(); + let run_id = run_id_for(&run_dir); + let (client, base_url) = + server_endpoint(&context.storage_dir).expect("server endpoint should exist"); + let response = client + .get(format!("{base_url}/api/v1/runs/{run_id}/state")) + .send() + .await + .expect("run state request should succeed"); + let response = expect_reqwest_status( + response, + fabro_http::StatusCode::OK, + format!("GET /api/v1/runs/{run_id}/state"), + ) + .await; + let state: serde_json::Value = response.json().await.expect("run state should parse"); + state["conclusion"]["status"] .as_str() .expect("conclusion should include a string status") .to_string() @@ -79,15 +169,14 @@ fn conclusion_status(context: &fabro_test::TestContext) -> String { #[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))] async fn hook_prompt_proceed_allows_run() { - let context = test_context!(); - context.write_home( - ".fabro/settings.toml", + let mut context = test_context!(); + write_hook_settings( + &context, &format!( r#" -[[hooks]] +[[run.hooks]] name = "prompt-proceed" event = "run_start" -type = "prompt" prompt = "A workflow is starting. Always approve. Respond with {{\"ok\": true}}." model = "{model}" "#, @@ -111,6 +200,7 @@ model = "{model}" .scenario(TwinScenario::responses("gpt-5.4-mini").text(r#"{"ok":true}"#)) .load(twin) .await; + configure_twin_server(&mut context, twin, &namespace); let mut cmd = context.run_cmd(); configure_hook_env(&mut cmd, stage_model()); twin.configure_command(&mut cmd, &namespace); @@ -123,20 +213,19 @@ model = "{model}" run_success_output(cmd).await; } - assert_eq!(conclusion_status(&context), "succeeded"); + assert_eq!(conclusion_status(&context).await, "succeeded"); } #[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))] async fn hook_prompt_block_prevents_run() { - let context = test_context!(); - context.write_home( - ".fabro/settings.toml", + let mut context = test_context!(); + write_hook_settings( + &context, &format!( r#" -[[hooks]] +[[run.hooks]] name = "prompt-block" event = "run_start" -type = "prompt" prompt = "Check: is 2+2 equal to 5? If the statement is true, respond {{\"ok\": true}}. If false, respond {{\"ok\": false, \"reason\": \"math check failed\"}}." model = "{model}" "#, @@ -163,6 +252,7 @@ model = "{model}" ) .load(twin) .await; + configure_twin_server(&mut context, twin, &namespace); let mut cmd = context.run_cmd(); configure_hook_env(&mut cmd, stage_model()); twin.configure_command(&mut cmd, &namespace); @@ -184,18 +274,18 @@ model = "{model}" #[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))] async fn hook_agent_proceed_allows_run() { - let context = test_context!(); - context.write_home( - ".fabro/settings.toml", + let mut context = test_context!(); + write_hook_settings( + &context, &format!( r#" -[[hooks]] +[[run.hooks]] name = "agent-proceed" event = "run_start" -type = "agent" prompt = "A workflow is starting. Always approve. Respond with {{\"ok\": true}}. Do not use any tools." model = "{model}" max_tool_rounds = 1 +agent = "enabled" "#, model = hook_model() ), @@ -217,6 +307,7 @@ max_tool_rounds = 1 .scenario(TwinScenario::responses("gpt-5.4-mini").text(r#"{"ok":true}"#)) .load(twin) .await; + configure_twin_server(&mut context, twin, &namespace); let mut cmd = context.run_cmd(); configure_hook_env(&mut cmd, stage_model()); twin.configure_command(&mut cmd, &namespace); @@ -229,25 +320,25 @@ max_tool_rounds = 1 run_success_output(cmd).await; } - assert_eq!(conclusion_status(&context), "succeeded"); + assert_eq!(conclusion_status(&context).await, "succeeded"); } #[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))] async fn hook_agent_with_tool_use() { - let context = test_context!(); + let mut context = test_context!(); let marker = context.temp_dir.join("hook_check.txt"); std::fs::write(&marker, "READY").unwrap(); - context.write_home( - ".fabro/settings.toml", + write_hook_settings( + &context, &format!( r#" -[[hooks]] +[[run.hooks]] name = "agent-tools" event = "run_start" -type = "agent" prompt = "Read the file at {path} using the read_file tool. If it contains 'READY', respond with {{\"ok\": true}}. Otherwise respond with {{\"ok\": false, \"reason\": \"not ready\"}}." model = "{model}" max_tool_rounds = 5 +agent = "enabled" "#, path = marker.display(), model = hook_model() @@ -274,6 +365,7 @@ max_tool_rounds = 5 .scenario(TwinScenario::responses("gpt-5.4-mini").text(r#"{"ok":true}"#)) .load(twin) .await; + configure_twin_server(&mut context, twin, &namespace); let mut cmd = context.run_cmd(); configure_hook_env(&mut cmd, stage_model()); twin.configure_command(&mut cmd, &namespace); @@ -286,12 +378,13 @@ max_tool_rounds = 5 run_success_output(cmd).await; } - assert_eq!(conclusion_status(&context), "succeeded"); + assert_eq!(conclusion_status(&context).await, "succeeded"); } #[fabro_macros::e2e_test(twin, live("ANTHROPIC_API_KEY"))] async fn arc_e2e_with_real_llm() { - let context = test_context!(); + let mut context = test_context!(); + write_hook_settings(&context, ""); let hello = context.temp_dir.join("hello.txt"); let workflow = write_workflow( &context, @@ -328,6 +421,7 @@ async fn arc_e2e_with_real_llm() { ) .load(twin) .await; + configure_twin_server(&mut context, twin, &namespace); let mut cmd = context.run_cmd(); configure_hook_env(&mut cmd, stage_model()); twin.configure_command(&mut cmd, &namespace); @@ -345,5 +439,5 @@ async fn arc_e2e_with_real_llm() { "Hello from LLM", "workflow should create the expected file" ); - assert_eq!(conclusion_status(&context), "succeeded"); + assert_eq!(conclusion_status(&context).await, "succeeded"); } diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index f6ed2b9a5..c29156bcf 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/lib/crates/fabro-sandbox/src/local.rs @@ -473,9 +473,7 @@ impl Sandbox for LocalSandbox { 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())); - } + filtered_env.push((k.clone(), v.clone())); } } @@ -484,7 +482,7 @@ impl Sandbox for LocalSandbox { let mut cmd = Command::new("/bin/bash"); cmd.arg("-lc") - .arg(command) + .arg(format!("exec {command}")) .current_dir(&effective_dir) .env_clear() .envs(filtered_env) @@ -1018,6 +1016,30 @@ mod tests { std::fs::remove_dir_all(&dir).unwrap(); } + #[tokio::test] + async fn stdio_process_forwards_explicit_provider_credentials() { + let dir = temp_dir(); + let sandbox = LocalSandbox::new(dir.clone()); + let env = HashMap::from([("OPENAI_API_KEY".to_string(), "test-key".to_string())]); + let process = sandbox + .spawn_stdio_process( + "python3 -u -c 'import os; print(os.environ.get(\"OPENAI_API_KEY\", \"missing\"), flush=True)'", + None, + Some(&env), + None, + ) + .await + .unwrap(); + + let mut stdout = BufReader::new(process.stdout); + let mut line = String::new(); + stdout.read_line(&mut line).await.unwrap(); + assert_eq!(line.trim_end(), "test-key"); + + process.handle.wait().await.unwrap(); + std::fs::remove_dir_all(&dir).unwrap(); + } + #[tokio::test] async fn exec_command_exit_code() { let dir = temp_dir(); diff --git a/lib/crates/fabro-workflow/Cargo.toml b/lib/crates/fabro-workflow/Cargo.toml index 406c12391..a2f106c3c 100644 --- a/lib/crates/fabro-workflow/Cargo.toml +++ b/lib/crates/fabro-workflow/Cargo.toml @@ -71,7 +71,7 @@ toml.workspace = true fabro-vault = { path = "../fabro-vault" } [dev-dependencies] base64.workspace = true -fabro-sandbox = { path = "../fabro-sandbox", features = ["daytona", "test-support"] } +fabro-sandbox = { path = "../fabro-sandbox", features = ["daytona", "docker", "test-support"] } fabro-mcp = { path = "../fabro-mcp" } tokio = { workspace = true, features = ["test-util", "macros"] } object_store.workspace = true diff --git a/lib/crates/fabro-workflow/tests/it/cp_integration.rs b/lib/crates/fabro-workflow/tests/it/cp_integration.rs index 3aba03426..897b775c3 100644 --- a/lib/crates/fabro-workflow/tests/it/cp_integration.rs +++ b/lib/crates/fabro-workflow/tests/it/cp_integration.rs @@ -17,6 +17,9 @@ use fabro_sandbox::reconnect::reconnect; use fabro_types::{RunSandbox, RunSandboxRuntime, SandboxProvider}; +const DOCKER_MANAGED_LABEL: &str = "sh.fabro.managed"; +const DOCKER_CP_IMAGE: &str = "buildpack-deps:noble"; + // --------------------------------------------------------------------------- // Local sandbox // --------------------------------------------------------------------------- @@ -143,14 +146,84 @@ fn docker_record(container_id: &str) -> RunSandbox { } } +struct DockerCpContainer { + id: String, + cleanup: bool, +} + +impl Drop for DockerCpContainer { + fn drop(&mut self) { + if self.cleanup { + let _ = std::process::Command::new("docker") + .args(["rm", "-f", &self.id]) + .output(); + } + } +} + +fn docker_cp_container() -> DockerCpContainer { + if let Ok(id) = std::env::var("FABRO_DOCKER_CP_CONTAINER") { + return DockerCpContainer { id, cleanup: false }; + } + + ensure_docker_image(DOCKER_CP_IMAGE); + let output = std::process::Command::new("docker") + .args([ + "run", + "-d", + "--label", + &format!("{DOCKER_MANAGED_LABEL}=true"), + "--workdir", + "/workspace", + DOCKER_CP_IMAGE, + "sh", + "-c", + "mkdir -p /workspace && sleep 300", + ]) + .output() + .expect("docker run should execute"); + assert!( + output.status.success(), + "docker run failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let id = String::from_utf8(output.stdout) + .expect("docker run stdout should be UTF-8") + .trim() + .to_string(); + assert!(!id.is_empty(), "docker run should return a container id"); + DockerCpContainer { id, cleanup: true } +} + +fn ensure_docker_image(image: &str) { + let inspect = std::process::Command::new("docker") + .args(["image", "inspect", image]) + .output() + .expect("docker image inspect should execute"); + if inspect.status.success() { + return; + } + + let pull = std::process::Command::new("docker") + .args(["pull", image]) + .output() + .expect("docker pull should execute"); + assert!( + pull.status.success(), + "docker pull {image} failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&pull.stdout), + String::from_utf8_lossy(&pull.stderr) + ); +} + #[tokio::test] #[ignore] // requires Docker daemon async fn docker_cp_upload_download_round_trip() { - let container_id = std::env::var("FABRO_DOCKER_CP_CONTAINER") - .expect("set FABRO_DOCKER_CP_CONTAINER to an initialized Fabro-managed container ID"); + let container = docker_cp_container(); let scratch = tempfile::tempdir().unwrap(); - let record = docker_record(&container_id); + let record = docker_record(&container.id); let sandbox = reconnect(&record, None).await.expect("reconnect docker"); // Upload a text file @@ -176,11 +249,10 @@ async fn docker_cp_upload_download_round_trip() { #[tokio::test] #[ignore] // requires Docker daemon async fn docker_cp_binary_round_trip() { - let container_id = std::env::var("FABRO_DOCKER_CP_CONTAINER") - .expect("set FABRO_DOCKER_CP_CONTAINER to an initialized Fabro-managed container ID"); + let container = docker_cp_container(); let scratch = tempfile::tempdir().unwrap(); - let record = docker_record(&container_id); + let record = docker_record(&container.id); let sandbox = reconnect(&record, None).await.expect("reconnect docker"); let binary: Vec = (0..=255).collect(); @@ -204,11 +276,10 @@ async fn docker_cp_binary_round_trip() { #[tokio::test] #[ignore] // requires Docker daemon async fn docker_cp_creates_parent_dirs() { - let container_id = std::env::var("FABRO_DOCKER_CP_CONTAINER") - .expect("set FABRO_DOCKER_CP_CONTAINER to an initialized Fabro-managed container ID"); + let container = docker_cp_container(); let scratch = tempfile::tempdir().unwrap(); - let record = docker_record(&container_id); + let record = docker_record(&container.id); let sandbox = reconnect(&record, None).await.expect("reconnect docker"); let content = b"nested docker file\n";