From eca28126024b6c2f898be656621dfbb5b7518989 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 19 Sep 2026 10:15:15 -0400 Subject: [PATCH] Fix what the gates found after the removal sweep The CLI artifact scenario seeded its run through the deleted upload route; it now runs a real Petri workflow whose hooks collect the artifacts, and the fabro artifact list and cp assertions read those. A real command retry is not producible from a command node (a plain failure or a timeout routes onward), so the retry dimension of the old fixture goes; the stage, node, and retry filters, the tree copies, the cross-stage ambiguity, and the filename collision stay covered. The archive guard test drops its upload row (the blob write row covers an octet-stream mutation). A dangling doc comment and two absolute paths clippy flagged are fixed. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-cli/tests/it/cmd/support.rs | 78 ----- .../fabro-cli/tests/it/scenario/artifacts.rs | 308 ++++++++++-------- lib/apps/fabro-server/src/server.rs | 11 +- .../fabro-server/src/server/handler/usage.rs | 3 +- .../fabro-server/tests/it/scenario/archive.rs | 6 - lib/components/fabro-sandbox/src/sandbox.rs | 5 - 6 files changed, 185 insertions(+), 226 deletions(-) diff --git a/lib/apps/fabro-cli/tests/it/cmd/support.rs b/lib/apps/fabro-cli/tests/it/cmd/support.rs index 6a7787ecb..7cd2e4bbf 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/support.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/support.rs @@ -484,10 +484,6 @@ pub(crate) fn setup_detached_dry_run(context: &TestContext) -> RunSetup { run } -pub(crate) fn setup_seeded_artifact_run(context: &TestContext) -> RunSetup { - seed_artifact_run(context) -} - pub(crate) fn setup_project_fixture(context: &TestContext) -> ProjectFixture { let project_dir = context.temp_dir.join("project"); let fabro_root = project_dir.join(".fabro"); @@ -993,41 +989,6 @@ async fn seed_dry_run(context: &TestContext) -> RunSetup { .await } -/// A completed dry run of the artifact workflow, with artifacts uploaded -/// for its stages through the API. -fn seed_artifact_run(context: &TestContext) -> RunSetup { - let workflow = context.temp_dir.join("artifact_run.fabro"); - write_text_file(&workflow, artifact_workflow_source()); - let run = run_completed_dry_run(context, &workflow); - - let (client, base_url) = server_endpoint(&context.storage_dir) - .expect("test server endpoint should be available for seeded artifacts"); - block_on(async { - for (stage_id, retry, path, contents) in [ - ("create_assets@1", 1, "assets/node_a/summary.txt", "alpha"), - ("create_assets@1", 1, "assets/shared/report.txt", "one"), - ("create_assets@2", 1, "assets/shared/report.txt", "two"), - ("create_colliding@1", 1, "assets/other/summary.txt", "beta"), - ("create_colliding@1", 1, "assets/retry/report.txt", "second"), - ("retry_assets@1", 1, "assets/retry/report.txt", "first"), - ("retry_assets@1", 2, "assets/retry/report.txt", "second"), - ] { - upload_seeded_artifact( - &client, - &base_url, - &run.run_id, - stage_id, - retry, - path, - contents, - ) - .await; - } - }); - - run -} - async fn create_seeded_run( context: &TestContext, target_path: &str, @@ -1086,32 +1047,6 @@ async fn create_seeded_run( } } -async fn upload_seeded_artifact( - client: &fabro_http::HttpClient, - base_url: &str, - run_id: &str, - stage_id: &str, - retry: u32, - path: &str, - contents: &str, -) { - let response = client - .post(format!( - "{base_url}/api/v1/runs/{run_id}/stages/{stage_id}/artifacts?filename={path}&retry={retry}" - )) - .header(fabro_http::header::CONTENT_TYPE, "application/octet-stream") - .body(contents.to_string()) - .send() - .await - .unwrap_or_else(|err| panic!("seeded artifact upload should execute: {err}")); - expect_reqwest_status( - response, - fabro_http::StatusCode::NO_CONTENT, - format!("POST /api/v1/runs/{run_id}/stages/{stage_id}/artifacts ({path}, retry {retry})"), - ) - .await; -} - fn test_label_map(context: &TestContext) -> std::collections::HashMap { test_labels(context) .into_iter() @@ -1144,19 +1079,6 @@ fn fast_simple_workflow_source() -> &'static str { "# } -fn artifact_workflow_source() -> &'static str { - r#"digraph ArtifactRun { - graph [goal="Exercise artifact commands", default_max_retries=0] - start [shape=Mdiamond] - exit [shape=Msquare] - create_assets [shape=parallelogram, script="true", max_retries=0] - retry_assets [shape=parallelogram, script="true", retry_policy="linear", timeout="500ms"] - create_colliding [shape=parallelogram, script="true", max_retries=0] - start -> create_assets -> retry_assets -> create_colliding -> exit -} -"# -} - pub(crate) fn text_tree(root: &Path) -> Vec { fn visit(root: &Path, dir: &Path, entries: &mut Vec) { let mut children: Vec<_> = std::fs::read_dir(dir) diff --git a/lib/apps/fabro-cli/tests/it/scenario/artifacts.rs b/lib/apps/fabro-cli/tests/it/scenario/artifacts.rs index d35e3ec53..234c4fe61 100644 --- a/lib/apps/fabro-cli/tests/it/scenario/artifacts.rs +++ b/lib/apps/fabro-cli/tests/it/scenario/artifacts.rs @@ -1,26 +1,63 @@ +//! `fabro artifact list` and `fabro artifact cp` over a run whose artifacts +//! the engine's hooks collected: every file under `[run.artifacts] include` +//! in a stage's workspace, once per content, into the blob table. + +use std::path::PathBuf; use std::time::Duration; use fabro_test::{fabro_snapshot, test_context}; -use crate::cmd::support::{read_text, setup_seeded_artifact_run, text_tree}; +use super::petri::{RunningServer, host_plugin, run_detached, wait_for_success}; +use crate::cmd::support::{read_text, text_tree}; -fn artifact_filters(context: &fabro_test::TestContext) -> Vec<(String, String)> { - let mut filters = context.filters(); - filters.push(( - r"\[STORAGE_DIR\]/scratch/\d{8}-\[ULID\]".to_string(), - "[RUN_DIR]".to_string(), - )); - filters +/// Three command stages that leave files under `assets/`. The second and +/// third write different contents to the same path, so the path names an +/// artifact of each; the third also writes a `summary.txt` that collides +/// by filename with the first stage's. +#[expect( + clippy::disallowed_methods, + reason = "the fixture files are written before the run starts" +)] +fn artifact_workspace(context: &fabro_test::TestContext) -> PathBuf { + let workspace = context.temp_dir.join("artifact-workspace"); + std::fs::create_dir_all(&workspace).expect("the workspace creates"); + std::fs::write( + workspace.join("workflow.fabro"), + "digraph ArtifactRun {\n graph [goal=\"Exercise artifact commands\", \ + default_max_retries=0]\n start [shape=Mdiamond]\n exit [shape=Msquare]\n \ + create_assets [shape=parallelogram, script=\"mkdir -p assets/node_a assets/shared && \ + printf alpha > assets/node_a/summary.txt && printf one > \ + assets/shared/report.txt\"]\n update_assets [shape=parallelogram, script=\"mkdir -p \ + assets/retry && printf second > assets/retry/report.txt\"]\n create_colliding \ + [shape=parallelogram, script=\"mkdir -p assets/other && printf beta > \ + assets/other/summary.txt && printf third > assets/retry/report.txt\"]\n start -> \ + create_assets -> update_assets -> create_colliding -> exit\n}\n", + ) + .expect("the workflow writes"); + std::fs::write( + workspace.join("workflow.toml"), + "_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\n\n[run]\ngoal = \"Exercise \ + artifact commands\"\n\n[run.artifacts]\ninclude = [\"assets/**\"]\n", + ) + .expect("the settings write"); + workspace } -#[test] -fn artifact_commands_share_populated_run_fixture() { +#[tokio::test(flavor = "multi_thread")] +async fn artifact_commands_read_the_artifacts_the_hooks_collected() { + if host_plugin().is_none() { + return; + } let context = test_context!(); - let run = setup_seeded_artifact_run(&context); - let filters = artifact_filters(&context); + let server = RunningServer::start().await; + let workspace = artifact_workspace(&context); + let run_id = run_detached(&context, &server, &workspace); + wait_for_success(&server, &run_id).await; + let target = server.target(); + let filters = context.filters(); let mut list_json = context.command(); - list_json.args(["artifact", "list", &run.run_id, "--json"]); + list_json.args(["artifact", "list", &run_id, "--json", "--server", &target]); fabro_snapshot!(filters.clone(), list_json, @r#" success: true exit_code: 0 @@ -41,12 +78,75 @@ fn artifact_commands_share_populated_run_fixture() { "size": 3 }, { - "stage_id": "create_assets@2", - "node_slug": "create_assets", + "stage_id": "create_colliding@1", + "node_slug": "create_colliding", "retry": 1, - "relative_path": "assets/shared/report.txt", - "size": 3 + "relative_path": "assets/other/summary.txt", + "size": 4 }, + { + "stage_id": "create_colliding@1", + "node_slug": "create_colliding", + "retry": 1, + "relative_path": "assets/retry/report.txt", + "size": 5 + }, + { + "stage_id": "update_assets@1", + "node_slug": "update_assets", + "retry": 1, + "relative_path": "assets/retry/report.txt", + "size": 6 + } + ] + ----- stderr ----- + "#); + + let mut list_filtered = context.command(); + list_filtered.args([ + "artifact", + "list", + &run_id, + "--node", + "update_assets", + "--retry", + "1", + "--json", + "--server", + &target, + ]); + fabro_snapshot!(filters.clone(), list_filtered, @r#" + success: true + exit_code: 0 + ----- stdout ----- + [ + { + "stage_id": "update_assets@1", + "node_slug": "update_assets", + "retry": 1, + "relative_path": "assets/retry/report.txt", + "size": 6 + } + ] + ----- stderr ----- + "#); + + let mut list_stage_filtered = context.command(); + list_stage_filtered.args([ + "artifact", + "list", + &run_id, + "--stage", + "create_colliding@1", + "--json", + "--server", + &target, + ]); + fabro_snapshot!(filters.clone(), list_stage_filtered, @r#" + success: true + exit_code: 0 + ----- stdout ----- + [ { "stage_id": "create_colliding@1", "node_slug": "create_colliding", @@ -59,73 +159,7 @@ fn artifact_commands_share_populated_run_fixture() { "node_slug": "create_colliding", "retry": 1, "relative_path": "assets/retry/report.txt", - "size": 6 - }, - { - "stage_id": "retry_assets@1", - "node_slug": "retry_assets", - "retry": 1, - "relative_path": "assets/retry/report.txt", "size": 5 - }, - { - "stage_id": "retry_assets@1", - "node_slug": "retry_assets", - "retry": 2, - "relative_path": "assets/retry/report.txt", - "size": 6 - } - ] - ----- stderr ----- - "#); - - let mut list_filtered = context.command(); - list_filtered.args([ - "artifact", - "list", - &run.run_id, - "--node", - "retry_assets", - "--retry", - "2", - "--json", - ]); - fabro_snapshot!(filters.clone(), list_filtered, @r#" - success: true - exit_code: 0 - ----- stdout ----- - [ - { - "stage_id": "retry_assets@1", - "node_slug": "retry_assets", - "retry": 2, - "relative_path": "assets/retry/report.txt", - "size": 6 - } - ] - ----- stderr ----- - "#); - - let mut list_stage_filtered = context.command(); - list_stage_filtered.args([ - "artifact", - "list", - &run.run_id, - "--stage", - "create_assets@2", - "--json", - ]); - fabro_snapshot!(filters.clone(), list_stage_filtered, @r#" - success: true - exit_code: 0 - ----- stdout ----- - [ - { - "stage_id": "create_assets@2", - "node_slug": "create_assets", - "retry": 1, - "relative_path": "assets/shared/report.txt", - "size": 3 } ] ----- stderr ----- @@ -136,90 +170,98 @@ fn artifact_commands_share_populated_run_fixture() { cp_single.args([ "artifact", "cp", - &format!("{}:assets/shared/report.txt", run.run_id), + &format!("{run_id}:assets/retry/report.txt"), single_dest.to_str().unwrap(), "--stage", - "create_assets@2", + "create_colliding@1", + "--server", + &target, ]); - fabro_snapshot!(context.filters(), cp_single, @" + fabro_snapshot!(filters.clone(), cp_single, @" success: true exit_code: 0 ----- stdout ----- - Copied assets/shared/report.txt to [TEMP_DIR]/artifact-one/report.txt + Copied assets/retry/report.txt to [TEMP_DIR]/artifact-one/report.txt ----- stderr ----- "); - assert_eq!(read_text(&single_dest.join("report.txt")), "two"); + assert_eq!(read_text(&single_dest.join("report.txt")), "third"); + + let node_dest = context.temp_dir.join("artifact-node"); + let mut cp_node = context.command(); + cp_node.args([ + "artifact", + "cp", + &format!("{run_id}:assets/retry/report.txt"), + node_dest.to_str().unwrap(), + "--node", + "update_assets", + "--server", + &target, + ]); + fabro_snapshot!(filters.clone(), cp_node, @" + success: true + exit_code: 0 + ----- stdout ----- + Copied assets/retry/report.txt to [TEMP_DIR]/artifact-node/report.txt + ----- stderr ----- + "); + assert_eq!(read_text(&node_dest.join("report.txt")), "second"); let stage_tree_dest = context.temp_dir.join("artifact-stage-tree"); let mut cp_stage_tree = context.command(); cp_stage_tree.args([ "artifact", "cp", - &run.run_id, + &run_id, stage_tree_dest.to_str().unwrap(), "--stage", - "create_assets@2", + "create_colliding@1", "--tree", + "--server", + &target, ]); - fabro_snapshot!(context.filters(), cp_stage_tree, @" + fabro_snapshot!(filters.clone(), cp_stage_tree, @" success: true exit_code: 0 ----- stdout ----- - Copied 1 artifact(s) to [TEMP_DIR]/artifact-stage-tree + Copied 2 artifact(s) to [TEMP_DIR]/artifact-stage-tree ----- stderr ----- "); insta::assert_snapshot!( text_tree(&stage_tree_dest).join("\n"), - @"create_assets/visit_2/retry_1/assets/shared/report.txt = two" + @r" + create_colliding/retry_1/assets/other/summary.txt = beta + create_colliding/retry_1/assets/retry/report.txt = third + " ); - let repeated_visit_dest = context.temp_dir.join("artifact-repeated-visit"); - let mut cp_repeated_visit = context.command(); - cp_repeated_visit.args([ - "artifact", - "cp", - &format!("{}:assets/shared/report.txt", run.run_id), - repeated_visit_dest.to_str().unwrap(), - "--node", - "create_assets", - "--retry", - "1", - ]); - fabro_snapshot!(context.filters(), cp_repeated_visit, @" - success: false - exit_code: 1 - ----- stdout ----- - ----- stderr ----- - × Path 'assets/shared/report.txt' matches multiple artifacts: create_assets@1:retry_1, create_assets@2:retry_1. Use --stage and/or --retry to disambiguate. - "); - let tree_dest = context.temp_dir.join("artifact-tree"); let mut cp_tree = context.command(); cp_tree.args([ "artifact", "cp", - &run.run_id, + &run_id, tree_dest.to_str().unwrap(), "--tree", + "--server", + &target, ]); cp_tree.timeout(Duration::from_secs(30)); - fabro_snapshot!(context.filters(), cp_tree, @" + fabro_snapshot!(filters.clone(), cp_tree, @" success: true exit_code: 0 ----- stdout ----- - Copied 7 artifact(s) to [TEMP_DIR]/artifact-tree + Copied 5 artifact(s) to [TEMP_DIR]/artifact-tree ----- stderr ----- "); insta::assert_snapshot!( text_tree(&tree_dest).join("\n"), @r" - create_assets/visit_1/retry_1/assets/node_a/summary.txt = alpha - create_assets/visit_1/retry_1/assets/shared/report.txt = one - create_assets/visit_2/retry_1/assets/shared/report.txt = two + create_assets/retry_1/assets/node_a/summary.txt = alpha + create_assets/retry_1/assets/shared/report.txt = one create_colliding/retry_1/assets/other/summary.txt = beta - create_colliding/retry_1/assets/retry/report.txt = second - retry_assets/retry_1/assets/retry/report.txt = first - retry_assets/retry_2/assets/retry/report.txt = second + create_colliding/retry_1/assets/retry/report.txt = third + update_assets/retry_1/assets/retry/report.txt = second " ); @@ -228,25 +270,35 @@ fn artifact_commands_share_populated_run_fixture() { cp_ambiguous.args([ "artifact", "cp", - &format!("{}:assets/retry/report.txt", run.run_id), + &format!("{run_id}:assets/retry/report.txt"), ambiguous_dest.to_str().unwrap(), + "--server", + &target, ]); - fabro_snapshot!(context.filters(), cp_ambiguous, @" + fabro_snapshot!(filters.clone(), cp_ambiguous, @" success: false exit_code: 1 ----- stdout ----- ----- stderr ----- - × Path 'assets/retry/report.txt' matches multiple artifacts: create_colliding@1:retry_1, retry_assets@1:retry_1, retry_assets@1:retry_2. Use --stage and/or --retry to disambiguate. + × Path 'assets/retry/report.txt' matches multiple artifacts: create_colliding@1:retry_1, update_assets@1:retry_1. Use --stage and/or --retry to disambiguate. "); let flat_dest = context.temp_dir.join("artifact-flat"); let mut cp_flat = context.command(); - cp_flat.args(["artifact", "cp", &run.run_id, flat_dest.to_str().unwrap()]); - fabro_snapshot!(context.filters(), cp_flat, @" + cp_flat.args([ + "artifact", + "cp", + &run_id, + flat_dest.to_str().unwrap(), + "--server", + &target, + ]); + fabro_snapshot!(filters, cp_flat, @" success: false exit_code: 1 ----- stdout ----- ----- stderr ----- - × Filename collision: 'report.txt' exists in both create_assets@1:retry_1 and create_assets@2:retry_1. Use --tree to preserve directory structure, or --stage and/or --retry to filter. + × Filename collision: 'summary.txt' exists in both create_assets@1:retry_1 and create_colliding@1:retry_1. Use --tree to preserve directory structure, or --stage and/or --retry to filter. "); + server.shutdown(); } diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index c0a40627a..0686efab0 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -93,6 +93,7 @@ use fabro_types::settings::run::NotificationRouteSettings; use fabro_types::settings::server::{ GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination, }; +use fabro_types::usage_rollup::{ProjectionUsageRollup, usage_rollup_from_projection}; use fabro_types::{ AskFabro, AskFabroUnavailableReason, BlobHash, FailureReason, InterviewQuestionRecord, ModelRef, ModelTestMode, PendingReason, Principal, PullRequestLink, QuestionType, @@ -1352,16 +1353,10 @@ pub(crate) fn accumulate_concluded_run_usage( .aggregate_usage .lock() .expect("aggregate_usage lock poisoned"); - accumulate_usage_rollup( - &mut agg, - &fabro_types::usage_rollup::usage_rollup_from_projection(final_state), - ); + accumulate_usage_rollup(&mut agg, &usage_rollup_from_projection(final_state)); } -fn accumulate_usage_rollup( - accumulator: &mut UsageAccumulator, - rollup: &fabro_types::usage_rollup::ProjectionUsageRollup, -) { +fn accumulate_usage_rollup(accumulator: &mut UsageAccumulator, rollup: &ProjectionUsageRollup) { accumulator.total_runs += 1; accumulator.total_timing = accumulator.total_timing.saturating_add(&rollup.timing); for model in &rollup.by_model { diff --git a/lib/apps/fabro-server/src/server/handler/usage.rs b/lib/apps/fabro-server/src/server/handler/usage.rs index f20a8a99d..1cffc13f2 100644 --- a/lib/apps/fabro-server/src/server/handler/usage.rs +++ b/lib/apps/fabro-server/src/server/handler/usage.rs @@ -2,6 +2,7 @@ use std::collections::HashMap; use std::sync::Arc; use chrono::{DateTime, Utc}; +use fabro_types::usage_rollup::usage_rollup_from_projection; use fabro_types::{ Graph, RunProjection, StageHandler, StageId, StageProjection, StageState, StageTiming, usage_is_empty, @@ -93,7 +94,7 @@ async fn get_run_usage( Err(err) => return err.into_response(), }; - let rollup = fabro_types::usage_rollup::usage_rollup_from_projection(&projection); + let rollup = usage_rollup_from_projection(&projection); let by_model = rollup .by_model .iter() diff --git a/lib/apps/fabro-server/tests/it/scenario/archive.rs b/lib/apps/fabro-server/tests/it/scenario/archive.rs index 9844a1c29..5074cd94d 100644 --- a/lib/apps/fabro-server/tests/it/scenario/archive.rs +++ b/lib/apps/fabro-server/tests/it/scenario/archive.rs @@ -67,12 +67,6 @@ async fn archived_runs_reject_mutations_with_actionable_body() { r#"{"kind":"text","text":"x"}"#, "application/json", ), - ( - "POST", - format!("/runs/{run_id}/stages/fake@1/artifacts?filename=smoke.txt&retry=1"), - "payload", - "application/octet-stream", - ), ( "PUT", format!("/runs/{run_id}/sandbox/file?path=smoke.txt"), diff --git a/lib/components/fabro-sandbox/src/sandbox.rs b/lib/components/fabro-sandbox/src/sandbox.rs index d998db0a6..c992ffe69 100644 --- a/lib/components/fabro-sandbox/src/sandbox.rs +++ b/lib/components/fabro-sandbox/src/sandbox.rs @@ -83,11 +83,6 @@ pub(crate) fn join_sandbox_path(base: &str, relative_path: &str) -> String { format!("{}/{relative_path}", base.trim_end_matches('/')) } -/// Creates the run branch in the sandbox's checkout through the driver's -/// git facet: a new run branches from `HEAD`, a fork from the source run's -/// checkpoint. The branch is created at that base, or moved to it when an -/// earlier attempt already created it. - #[cfg(test)] mod tests { #[test]