diff --git a/Cargo.lock b/Cargo.lock index d7a7d8ac1..8a1400250 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1536,7 +1536,7 @@ dependencies = [ [[package]] name = "fabro-agent" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-trait", @@ -1575,7 +1575,7 @@ dependencies = [ [[package]] name = "fabro-api" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "chrono", "fabro-config", @@ -1596,7 +1596,7 @@ dependencies = [ [[package]] name = "fabro-auth" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-trait", @@ -1620,7 +1620,7 @@ dependencies = [ [[package]] name = "fabro-checkpoint" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "chrono", "fabro-config", @@ -1636,7 +1636,7 @@ dependencies = [ [[package]] name = "fabro-cli" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "assert_cmd", @@ -1732,7 +1732,7 @@ dependencies = [ [[package]] name = "fabro-client" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "bytes", @@ -1761,7 +1761,7 @@ dependencies = [ [[package]] name = "fabro-config" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "chrono", @@ -1788,7 +1788,7 @@ dependencies = [ [[package]] name = "fabro-core" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "async-trait", "fabro-types", @@ -1803,7 +1803,7 @@ dependencies = [ [[package]] name = "fabro-dev" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "assert_cmd", @@ -1823,7 +1823,7 @@ dependencies = [ [[package]] name = "fabro-devcontainer" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "fabro-http", "fabro-static", @@ -1840,7 +1840,7 @@ dependencies = [ [[package]] name = "fabro-dump" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "bytes", @@ -1854,7 +1854,7 @@ dependencies = [ [[package]] name = "fabro-github" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "base64", @@ -1876,7 +1876,7 @@ dependencies = [ [[package]] name = "fabro-graphviz" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "fabro-types", @@ -1890,7 +1890,7 @@ dependencies = [ [[package]] name = "fabro-hooks" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "async-trait", "fabro-agent", @@ -1914,7 +1914,7 @@ dependencies = [ [[package]] name = "fabro-http" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "fabro-static", "http", @@ -1924,7 +1924,7 @@ dependencies = [ [[package]] name = "fabro-install" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "base64", @@ -1939,7 +1939,7 @@ dependencies = [ [[package]] name = "fabro-interview" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "async-trait", "dialoguer", @@ -1954,7 +1954,7 @@ dependencies = [ [[package]] name = "fabro-llm" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-trait", @@ -1986,7 +1986,7 @@ dependencies = [ [[package]] name = "fabro-macros" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "clap", "fabro-options-metadata", @@ -1997,7 +1997,7 @@ dependencies = [ [[package]] name = "fabro-mcp" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "fabro-config", @@ -2013,7 +2013,7 @@ dependencies = [ [[package]] name = "fabro-model" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "fabro-static", "insta", @@ -2024,7 +2024,7 @@ dependencies = [ [[package]] name = "fabro-oauth" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "axum", @@ -2046,7 +2046,7 @@ dependencies = [ [[package]] name = "fabro-options-metadata" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "serde", "serde_json", @@ -2054,7 +2054,7 @@ dependencies = [ [[package]] name = "fabro-proc" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "cc", "libc", @@ -2063,7 +2063,7 @@ dependencies = [ [[package]] name = "fabro-redact" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "aho-corasick", "ref-cast", @@ -2079,7 +2079,7 @@ dependencies = [ [[package]] name = "fabro-retro" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "chrono", @@ -2098,7 +2098,7 @@ dependencies = [ [[package]] name = "fabro-sandbox" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-trait", @@ -2134,7 +2134,7 @@ dependencies = [ [[package]] name = "fabro-server" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-trait", @@ -2215,7 +2215,7 @@ dependencies = [ [[package]] name = "fabro-slack" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "fabro-http", "fabro-interview", @@ -2236,18 +2236,18 @@ dependencies = [ [[package]] name = "fabro-spa" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "rust-embed", ] [[package]] name = "fabro-static" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" [[package]] name = "fabro-store" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "async-trait", "bytes", @@ -2274,7 +2274,7 @@ dependencies = [ [[package]] name = "fabro-telemetry" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "base64", @@ -2300,7 +2300,7 @@ dependencies = [ [[package]] name = "fabro-template" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "fabro-util", @@ -2312,7 +2312,7 @@ dependencies = [ [[package]] name = "fabro-test" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "assert_cmd", "axum", @@ -2335,7 +2335,7 @@ dependencies = [ [[package]] name = "fabro-tracker" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "async-trait", "fabro-github", @@ -2348,7 +2348,7 @@ dependencies = [ [[package]] name = "fabro-types" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "chrono", "clap", @@ -2369,7 +2369,7 @@ dependencies = [ [[package]] name = "fabro-util" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "console 0.15.11", @@ -2389,7 +2389,7 @@ dependencies = [ [[package]] name = "fabro-validate" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "fabro-graphviz", "fabro-model", @@ -2399,7 +2399,7 @@ dependencies = [ [[package]] name = "fabro-vault" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "chrono", "fabro-types", @@ -2411,7 +2411,7 @@ dependencies = [ [[package]] name = "fabro-workflow" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "assert_cmd", @@ -7161,7 +7161,7 @@ dependencies = [ [[package]] name = "twin-github" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "axum", "base64", @@ -7180,7 +7180,7 @@ dependencies = [ [[package]] name = "twin-openai" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" dependencies = [ "anyhow", "async-stream", diff --git a/Cargo.toml b/Cargo.toml index 145892365..9a3a113f4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ resolver = "2" [workspace.package] edition = "2021" -version = "0.220.0-nightly.2" +version = "0.221.0-nightly.0" license = "MIT" [workspace.dependencies] diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 0ff0edf2a..fe181828f 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -2064,6 +2064,7 @@ paths: parameters: - $ref: "#/components/parameters/RunId" - $ref: "#/components/parameters/StageId" + - $ref: "#/components/parameters/ArtifactRetry" - name: filename in: query required: false @@ -2109,6 +2110,7 @@ paths: - $ref: "#/components/parameters/RunId" - $ref: "#/components/parameters/StageId" - $ref: "#/components/parameters/ArtifactFilename" + - $ref: "#/components/parameters/ArtifactRetry" responses: "200": description: Artifact contents @@ -2118,7 +2120,7 @@ paths: type: string format: binary "400": - description: Missing filename + description: Missing filename or retry headers: x-request-id: $ref: "#/components/headers/XRequestId" @@ -2980,6 +2982,17 @@ components: type: string example: src/lib.rs + ArtifactRetry: + name: retry + in: query + required: true + description: Retry attempt number for the artifact. + schema: + type: integer + format: int32 + minimum: 0 + example: 1 + SinceSeq: name: since_seq in: query @@ -5097,18 +5110,32 @@ components: example: true ArtifactEntry: - description: A single artifact filename. + description: A single artifact file for a stage. type: object required: - filename + - retry + - size properties: filename: type: string description: Artifact filename. example: src/lib.rs + retry: + type: integer + format: int32 + minimum: 0 + description: Retry attempt number. + example: 1 + size: + type: integer + format: int64 + minimum: 0 + description: Artifact size in bytes. + example: 1234 ArtifactListResponse: - description: List of artifact filenames for a stage. + description: List of artifact files for a stage. type: object required: - data @@ -5179,6 +5206,7 @@ components: retry: type: integer format: int32 + minimum: 0 description: Retry attempt number. relative_path: type: string @@ -5186,6 +5214,7 @@ components: size: type: integer format: int64 + minimum: 0 description: Artifact size in bytes. RunArtifactListResponse: @@ -5257,12 +5286,22 @@ components: oneOf: - $ref: "#/components/schemas/StageCompletion" - type: "null" - provider_used: {} + provider_used: + type: ["object", "null"] + description: Provider and model metadata recorded for the stage attempt. diff: type: ["string", "null"] - script_invocation: {} - script_timing: {} - parallel_results: {} + script_invocation: + type: ["object", "null"] + description: Command and environment recorded when the stage script ran. + script_timing: + type: ["object", "null"] + description: Wall-clock and step timing metadata for the stage script. + parallel_results: + type: ["array", "null"] + items: + type: object + description: Per-branch result objects produced by a parallel stage. stdout: type: ["string", "null"] stderr: diff --git a/lib/crates/fabro-cli/src/commands/artifact/cp.rs b/lib/crates/fabro-cli/src/commands/artifact/cp.rs index bf989aac2..e33ec2cfc 100644 --- a/lib/crates/fabro-cli/src/commands/artifact/cp.rs +++ b/lib/crates/fabro-cli/src/commands/artifact/cp.rs @@ -142,7 +142,7 @@ async fn write_artifact_file( .with_context(|| format!("creating directory {}", parent.display()))?; } let bytes = client - .download_stage_artifact(run_id, &entry.stage_id, &entry.relative_path) + .download_stage_artifact(run_id, &entry.stage_id, entry.retry, &entry.relative_path) .await?; std::fs::write(dest_file, bytes) .with_context(|| format!("Failed to write {}", dest_file.display()))?; diff --git a/lib/crates/fabro-cli/src/commands/artifact/mod.rs b/lib/crates/fabro-cli/src/commands/artifact/mod.rs index b9a5ffcf7..d69caf24e 100644 --- a/lib/crates/fabro-cli/src/commands/artifact/mod.rs +++ b/lib/crates/fabro-cli/src/commands/artifact/mod.rs @@ -52,6 +52,7 @@ pub(super) async fn resolve_artifacts( entries.sort_by(|a, b| { a.stage_id .cmp(&b.stage_id) + .then_with(|| a.retry.cmp(&b.retry)) .then_with(|| a.relative_path.cmp(&b.relative_path)) }); diff --git a/lib/crates/fabro-cli/src/commands/dump.rs b/lib/crates/fabro-cli/src/commands/dump.rs index 170260398..81201fdc5 100644 --- a/lib/crates/fabro-cli/src/commands/dump.rs +++ b/lib/crates/fabro-cli/src/commands/dump.rs @@ -96,8 +96,9 @@ async fn write_run_dump( .stage_id .parse() .with_context(|| format!("server returned invalid stage id {:?}", artifact.stage_id))?; + let retry = artifact.retry.cast_unsigned(); let data = client - .download_stage_artifact(run_id, &stage_id, &artifact.relative_path) + .download_stage_artifact(run_id, &stage_id, retry, &artifact.relative_path) .await .with_context(|| { format!( @@ -105,7 +106,7 @@ async fn write_run_dump( artifact.relative_path, artifact.stage_id ) })?; - dump.add_artifact_bytes(&stage_id, &artifact.relative_path, data)?; + dump.add_artifact_bytes(&stage_id, retry, &artifact.relative_path, data)?; } let output_dir = output_dir.to_path_buf(); diff --git a/lib/crates/fabro-cli/src/commands/run/runner.rs b/lib/crates/fabro-cli/src/commands/run/runner.rs index f5aebdc82..288c45afd 100644 --- a/lib/crates/fabro-cli/src/commands/run/runner.rs +++ b/lib/crates/fabro-cli/src/commands/run/runner.rs @@ -272,6 +272,7 @@ impl StageArtifactUploader for HttpArtifactUploader { async fn upload_stage_artifacts( &self, stage_id: &fabro_types::StageId, + retry: u32, artifact_capture_dir: &Path, artifacts: &[ArtifactUpload], ) -> Result<()> { @@ -286,6 +287,7 @@ impl StageArtifactUploader for HttpArtifactUploader { .upload_stage_artifact_file( &self.run_id, stage_id, + retry, &artifact.path, &artifact_capture_dir.join(&artifact.path), &self.worker_token, @@ -297,6 +299,7 @@ impl StageArtifactUploader for HttpArtifactUploader { .upload_stage_artifact_batch( &self.run_id, stage_id, + retry, artifact_capture_dir, artifacts, &self.worker_token, diff --git a/lib/crates/fabro-cli/tests/it/cmd/dump.rs b/lib/crates/fabro-cli/tests/it/cmd/dump.rs index 9b60f7453..12b13ee9a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/dump.rs @@ -248,7 +248,10 @@ include = ["assets/**"] "run export should hydrate blob refs\n{run_json}" ); assert_eq!( - fs::read_to_string(output_dir.join("artifacts/big@1/assets/shared/report.txt")).unwrap(), + fs::read_to_string( + output_dir.join("artifacts/002-big@1/retry-0001/assets/shared/report.txt") + ) + .unwrap(), "exported" ); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index f27159af5..e295ff5d7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -869,15 +869,24 @@ async fn seed_artifact_run(context: &TestContext) -> RunSetup { let (client, base_url) = server_endpoint(&context.storage_dir) .expect("test server endpoint should be available for seeded artifacts"); append_seeded_artifact_run_events(&client, &base_url, &run, context).await; - for (stage_id, path, contents) in [ - ("create_assets@1", "assets/node_a/summary.txt", "alpha"), - ("create_assets@1", "assets/shared/report.txt", "one"), - ("create_colliding@1", "assets/other/summary.txt", "beta"), - ("create_colliding@1", "assets/retry/report.txt", "second"), - ("retry_assets@1", "assets/retry/report.txt", "first"), - ("retry_assets@2", "assets/retry/report.txt", "second"), + 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_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, path, contents).await; + upload_seeded_artifact( + &client, + &base_url, + &run.run_id, + stage_id, + retry, + path, + contents, + ) + .await; } run @@ -1368,12 +1377,13 @@ async fn upload_seeded_artifact( 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}" + "{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()) @@ -1383,7 +1393,7 @@ async fn upload_seeded_artifact( expect_reqwest_status( response, fabro_http::StatusCode::NO_CONTENT, - format!("POST /api/v1/runs/{run_id}/stages/{stage_id}/artifacts ({path})"), + format!("POST /api/v1/runs/{run_id}/stages/{stage_id}/artifacts ({path}, retry {retry})"), ) .await; } 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 d01318b69..9bdd6b7ec 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 @@ -7,7 +7,7 @@ use fabro_test::test_context; use super::{ completed_nodes, dump_export, find_run_dir, fixture, read_conclusion, run_id_for, - sandbox_tests, timeout_for, + sandbox_tests, stage_dump_dir, timeout_for, }; sandbox_tests!(command_agent_mixed, keys = ["ANTHROPIC_API_KEY"]); @@ -49,8 +49,9 @@ 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(export_dir.join("stages/verify@1/stdout.log")) - .expect("verify stdout.log should exist"); + let stdout = + std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("stdout.log")) + .expect("verify stdout.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 9120c123b..e691fb007 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/command_pipeline.rs @@ -7,7 +7,7 @@ use fabro_test::test_context; use super::{ completed_nodes, dump_export, find_run_dir, fixture, read_conclusion, run_id_for, - sandbox_tests, timeout_for, + sandbox_tests, stage_dump_dir, timeout_for, }; sandbox_tests!(command_pipeline); @@ -48,8 +48,9 @@ fn scenario_command_pipeline(sandbox: &str) { ); let export_dir = dump_export(&context, &run_id_for(&run_dir)); - let stdout1 = std::fs::read_to_string(export_dir.join("stages/step1@1/stdout.log")) - .expect("step1 stdout.log should exist"); + let stdout1 = + std::fs::read_to_string(stage_dump_dir(&export_dir, "step1@1").join("stdout.log")) + .expect("step1 stdout.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 555c53fcf..921d01855 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/full_stack.rs @@ -7,7 +7,7 @@ use fabro_test::test_context; use super::{ completed_nodes, dump_export, find_run_dir, fixture, has_event, read_conclusion, read_run_spec, - run_id_for, sandbox_tests, timeout_for, + run_id_for, sandbox_tests, stage_dump_dir, timeout_for, }; sandbox_tests!(full_stack, keys = ["ANTHROPIC_API_KEY"]); @@ -74,8 +74,9 @@ 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(export_dir.join("stages/verify@1/stdout.log")) - .expect("verify stdout.log should exist"); + let stdout = + std::fs::read_to_string(stage_dump_dir(&export_dir, "verify@1").join("stdout.log")) + .expect("verify stdout.log should exist"); assert!( stdout.contains("PASS"), "verify stdout should contain PASS, got: {stdout}" diff --git a/lib/crates/fabro-cli/tests/it/workflow/mod.rs b/lib/crates/fabro-cli/tests/it/workflow/mod.rs index b7911837a..15d8cd508 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/mod.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/mod.rs @@ -76,6 +76,34 @@ pub(super) fn dump_export(context: &TestContext, run_id: &str) -> PathBuf { output_dir } +#[expect( + clippy::disallowed_methods, + reason = "integration test helpers inspect exported files synchronously" +)] +pub(super) fn stage_dump_dir(export_dir: &Path, stage_id: &str) -> PathBuf { + let stages_dir = export_dir.join("stages"); + let mut matches: Vec<_> = std::fs::read_dir(&stages_dir) + .unwrap_or_else(|err| panic!("reading {} should succeed: {err}", stages_dir.display())) + .filter_map(|entry| entry.ok().map(|entry| entry.path())) + .filter(|path| { + path.file_name() + .and_then(|name| name.to_str()) + .is_some_and(|name| { + name == stage_id || name.split_once('-').is_some_and(|(_, id)| id == stage_id) + }) + }) + .collect(); + matches.sort(); + match matches.as_slice() { + [path] => path.clone(), + [] => panic!( + "stage dump dir for {stage_id} not found in {}", + stages_dir.display() + ), + _ => panic!("stage dump dir for {stage_id} was ambiguous: {matches:?}"), + } +} + /// Find the single run directory for this test context. pub(super) fn find_run_dir(context: &TestContext) -> PathBuf { context.single_run_dir() diff --git a/lib/crates/fabro-client/src/client.rs b/lib/crates/fabro-client/src/client.rs index bb91eb46f..86a63375c 100644 --- a/lib/crates/fabro-client/src/client.rs +++ b/lib/crates/fabro-client/src/client.rs @@ -1200,6 +1200,7 @@ impl Client { &self, run_id: &RunId, stage_id: &StageId, + retry: u32, filename: &str, ) -> Result> { let response = self @@ -1208,6 +1209,7 @@ impl Client { .get_stage_artifact() .id(run_id.to_string()) .stage_id(stage_id.to_string()) + .retry(retry.cast_signed()) .filename(filename) .send() .await @@ -1226,7 +1228,12 @@ impl Client { clippy::disallowed_types, reason = "Client builds raw server API request URLs for wire transit; logging redaction is handled at log boundaries." )] - fn stage_artifacts_url(&self, run_id: &RunId, stage_id: &StageId) -> Result { + fn stage_artifacts_url( + &self, + run_id: &RunId, + stage_id: &StageId, + retry: u32, + ) -> Result { let base_url = self.base_url(); let mut url = fabro_http::Url::parse(&base_url) .with_context(|| format!("invalid server base URL {base_url}"))?; @@ -1241,6 +1248,8 @@ impl Client { &stage_id.to_string(), "artifacts", ]); + url.query_pairs_mut() + .append_pair("retry", &retry.to_string()); Ok(url) } @@ -1248,11 +1257,12 @@ impl Client { &self, run_id: &RunId, stage_id: &StageId, + retry: u32, filename: &str, path: &Path, bearer_token: &str, ) -> Result<()> { - let mut url = self.stage_artifacts_url(run_id, stage_id)?; + let mut url = self.stage_artifacts_url(run_id, stage_id, retry)?; url.query_pairs_mut().append_pair("filename", filename); let file = File::open(path) @@ -1286,11 +1296,12 @@ impl Client { &self, run_id: &RunId, stage_id: &StageId, + retry: u32, artifact_capture_dir: &Path, artifacts: &[ArtifactUpload], bearer_token: &str, ) -> Result<()> { - let url = self.stage_artifacts_url(run_id, stage_id)?; + let url = self.stage_artifacts_url(run_id, stage_id, retry)?; let mut manifest_entries = Vec::with_capacity(artifacts.len()); let mut file_parts = Vec::with_capacity(artifacts.len()); diff --git a/lib/crates/fabro-dump/src/lib.rs b/lib/crates/fabro-dump/src/lib.rs index 57c51cbf7..69a670565 100644 --- a/lib/crates/fabro-dump/src/lib.rs +++ b/lib/crates/fabro-dump/src/lib.rs @@ -13,15 +13,34 @@ use std::path::{Component, Path, PathBuf}; use anyhow::{Context, Result, bail}; use bytes::Bytes; -use fabro_store::{EventEnvelope, RunProjection, SerializableProjection, StageId}; +use fabro_store::{ + EventEnvelope, RunProjection, SerializableProjection, StageId, retry_storage_segment, +}; use fabro_types::{RunBlobId, parse_blob_ref}; use futures::future::BoxFuture; pub type BlobReader = Box BoxFuture<'static, Result>> + Send>; +const STAGE_RANK_WIDTH: usize = 3; +const MAX_STAGES_IN_DUMP: usize = { + let mut value = 1usize; + let mut i = 0usize; + while i < STAGE_RANK_WIDTH { + value *= 10; + i += 1; + } + value - 1 +}; + +fn stage_dir_name(rank: u32, stage_id: &StageId) -> String { + format!("{rank:0>STAGE_RANK_WIDTH$}-{stage_id}") +} + #[derive(Debug, Clone)] pub struct RunDump { - entries: Vec, + entries: Vec, + stage_ranks: HashMap, + dump_log_index: Option, } #[derive(Debug, Clone)] @@ -48,15 +67,27 @@ impl RunDump { } let mut stages: Vec<_> = state.iter_stages().collect(); + if stages.len() > MAX_STAGES_IN_DUMP { + bail!( + "run dump supports at most {MAX_STAGES_IN_DUMP} stages with the current path prefix width (got {})", + stages.len() + ); + } stages.sort_by(|(left_id, left), (right_id, right)| { left.first_event_seq .cmp(&right.first_event_seq) .then_with(|| left_id.cmp(right_id)) }); + let mut stage_ranks = HashMap::new(); + for (index, (stage_id, _)) in stages.iter().enumerate() { + let rank = u32::try_from(index + 1).context("stage rank should fit in u32")?; + stage_ranks.insert((*stage_id).clone(), rank); + } + for (index, (stage_id, stage)) in stages.into_iter().enumerate() { - let rank = index + 1; - let base = PathBuf::from("stages").join(format!("{rank:03}-{stage_id}")); + let rank = u32::try_from(index + 1).context("stage rank should fit in u32")?; + let base = PathBuf::from("stages").join(stage_dir_name(rank, stage_id)); if let Some(prompt) = stage.prompt.as_ref() { entries.push(RunDumpEntry::text_path( @@ -127,7 +158,11 @@ impl RunDump { )); } - Ok(Self { entries }) + Ok(Self { + entries, + stage_ranks, + dump_log_index: None, + }) } pub fn from_store_state_and_events( @@ -158,14 +193,32 @@ impl RunDump { pub fn add_artifact_bytes( &mut self, stage_id: &StageId, + retry: u32, filename: &str, data: Vec, ) -> Result<()> { - let path = artifact_dump_path(stage_id, filename)?; + let path = artifact_dump_path(&self.stage_ranks, stage_id, retry, filename)?; + if !self.stage_ranks.contains_key(stage_id) { + self.add_orphan_notice(stage_id); + } self.entries.push(RunDumpEntry::bytes_path(&path, data)); Ok(()) } + fn add_orphan_notice(&mut self, stage_id: &StageId) { + let line = format!("notice: artifact stage {stage_id} was not present in run projection\n"); + if let Some(index) = self.dump_log_index { + if let Some(RunDumpContents::Text(text)) = + self.entries.get_mut(index).map(|entry| &mut entry.contents) + { + text.push_str(&line); + return; + } + } + self.dump_log_index = Some(self.entries.len()); + self.entries.push(RunDumpEntry::text("dump.log", line)); + } + pub fn add_file_bytes(&mut self, path: impl Into, contents: Vec) { self.entries.push(RunDumpEntry::bytes(path, contents)); } @@ -404,11 +457,21 @@ fn replace_blob_refs_in_value( Ok(()) } -fn artifact_dump_path(stage_id: &StageId, filename: &str) -> Result { +fn artifact_dump_path( + stage_ranks: &HashMap, + stage_id: &StageId, + retry: u32, + filename: &str, +) -> Result { validate_single_path_segment("node id", stage_id.node_id())?; let filename_path = validate_relative_path("artifact filename", filename)?; + let stage_dir = stage_ranks.get(stage_id).map_or_else( + || PathBuf::from("_orphans").join(stage_id.to_string()), + |rank| PathBuf::from(stage_dir_name(*rank, stage_id)), + ); Ok(PathBuf::from("artifacts") - .join(stage_id.to_string()) + .join(stage_dir) + .join(retry_storage_segment(retry)) .join(filename_path)) } @@ -609,9 +672,9 @@ mod tests { .prompt = Some("second".to_string()); let mut dump = RunDump::from_projection(&projection).unwrap(); - dump.add_artifact_bytes(&StageId::new("zebra", 1), "report.txt", b"z".to_vec()) + dump.add_artifact_bytes(&StageId::new("zebra", 1), 0, "report.txt", b"z".to_vec()) .unwrap(); - dump.add_artifact_bytes(&StageId::new("apple", 1), "report.txt", b"a".to_vec()) + dump.add_artifact_bytes(&StageId::new("apple", 1), 0, "report.txt", b"a".to_vec()) .unwrap(); let paths: Vec<&str> = dump @@ -622,8 +685,28 @@ mod tests { assert!(paths.contains(&"stages/001-zebra@1/prompt.md")); assert!(paths.contains(&"stages/002-apple@1/prompt.md")); - assert!(paths.contains(&"artifacts/zebra@1/report.txt")); - assert!(paths.contains(&"artifacts/apple@1/report.txt")); + assert!(paths.contains(&"artifacts/001-zebra@1/retry-0000/report.txt")); + assert!(paths.contains(&"artifacts/002-apple@1/retry-0000/report.txt")); + } + + #[test] + fn add_artifact_bytes_places_orphans_under_sentinel() { + let mut projection = RunProjection::default(); + projection + .stage_entry("known", 1, first_event_seq(1)) + .prompt = Some("present".to_string()); + + let mut dump = RunDump::from_projection(&projection).unwrap(); + dump.add_artifact_bytes(&StageId::new("missing", 1), 0, "report.txt", b"m".to_vec()) + .unwrap(); + + let paths: Vec<&str> = dump + .entries() + .iter() + .map(|entry| entry.path.as_str()) + .collect(); + assert!(paths.contains(&"artifacts/_orphans/missing@1/retry-0000/report.txt")); + assert!(paths.contains(&"dump.log")); } #[test] @@ -632,10 +715,12 @@ mod tests { let blob_id = fabro_types::RunBlobId::new(&blob); let legacy_ref = format!("file:///sandbox/.fabro/artifacts/{blob_id}.json"); let mut dump = RunDump { - entries: vec![RunDumpEntry::json( + entries: vec![RunDumpEntry::json( "run.json", serde_json::json!({ "stdout": legacy_ref }), )], + stage_ranks: HashMap::new(), + dump_log_index: None, }; executor::block_on(async { diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index e35d951b6..fddf93751 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -65,7 +65,8 @@ use fabro_slack::threads::ThreadRegistry; use fabro_slack::{blocks as slack_blocks, connection as slack_connection}; use fabro_static::EnvVars; use fabro_store::{ - ArtifactStore, Database, EventEnvelope, EventPayload, PendingInterviewRecord, StageId, + ArtifactKey, ArtifactStore, Database, EventEnvelope, EventPayload, NodeArtifact, + PendingInterviewRecord, StageArtifactEntry, StageId, }; #[cfg(test)] use fabro_types::BlockedReason; @@ -226,6 +227,8 @@ struct GlobalAttachParams { struct ArtifactFilenameParams { #[serde(default)] filename: Option, + #[serde(default)] + retry: Option, } #[derive(serde::Deserialize)] @@ -3452,13 +3455,12 @@ pub(crate) fn parse_blob_id_path(blob_id: &str) -> Result { #[allow( clippy::result_large_err, - reason = "Missing filename validation returns HTTP 400 responses directly." + reason = "Missing query parameter validation returns HTTP 400 responses directly." )] -fn required_filename(params: ArtifactFilenameParams) -> Result { - match params.filename { - Some(filename) if !filename.is_empty() => Ok(filename), - _ => Err(ApiError::bad_request("Missing filename query parameter.").into_response()), - } +fn required_query_param(value: Option<&T>, name: &str) -> Result { + value.cloned().ok_or_else(|| { + ApiError::bad_request(format!("Missing {name} query parameter.")).into_response() + }) } #[allow( @@ -6281,16 +6283,7 @@ async fn list_run_artifacts( match state.artifact_store.list_for_run(&id).await { Ok(entries) => Json(RunArtifactListResponse { - data: entries - .into_iter() - .map(|entry| RunArtifactEntry { - stage_id: entry.node.to_string(), - node_slug: entry.node.node_id().to_string(), - retry: entry.node.visit().cast_signed(), - relative_path: entry.filename, - size: entry.size.cast_signed(), - }) - .collect(), + data: entries.into_iter().map(run_artifact_entry_from).collect(), }) .into_response(), Err(err) => { @@ -6299,6 +6292,24 @@ async fn list_run_artifacts( } } +fn run_artifact_entry_from(entry: NodeArtifact) -> RunArtifactEntry { + RunArtifactEntry { + stage_id: entry.node.to_string(), + node_slug: entry.node.node_id().to_string(), + retry: entry.retry.cast_signed(), + relative_path: entry.filename, + size: entry.size.cast_signed(), + } +} + +fn artifact_entry_from(entry: StageArtifactEntry) -> ArtifactEntry { + ArtifactEntry { + filename: entry.filename, + retry: entry.retry.cast_signed(), + size: entry.size.cast_signed(), + } +} + async fn list_stage_artifacts( _auth: RequiredUser, State(state): State>, @@ -6317,11 +6328,8 @@ async fn list_stage_artifacts( } match state.artifact_store.list_for_node(&id, &stage_id).await { - Ok(filenames) => Json(ArtifactListResponse { - data: filenames - .into_iter() - .map(|filename| ArtifactEntry { filename }) - .collect(), + Ok(entries) => Json(ArtifactListResponse { + data: entries.into_iter().map(artifact_entry_from).collect(), }) .into_response(), Err(err) => { @@ -6502,6 +6510,7 @@ async fn upload_stage_artifact_octet_stream( state: &AppState, run_id: &RunId, stage_id: &StageId, + retry: u32, filename: String, body: Body, content_length: Option, @@ -6517,10 +6526,10 @@ async fn upload_stage_artifact_octet_stream( )); } - let mut writer = match state - .artifact_store - .writer(run_id, stage_id, &relative_path) - { + let mut writer = match state.artifact_store.writer( + run_id, + &ArtifactKey::new(stage_id.clone(), retry, relative_path), + ) { Ok(writer) => writer, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) @@ -6562,6 +6571,7 @@ async fn upload_stage_artifact_multipart( state: &AppState, run_id: &RunId, stage_id: &StageId, + retry: u32, boundary: String, body: Body, ) -> Response { @@ -6607,7 +6617,10 @@ async fn upload_stage_artifact_multipart( return bad_request_response(format!("unexpected multipart part: {part_name}")); }; - let mut writer = match state.artifact_store.writer(run_id, stage_id, &entry.path) { + let mut writer = match state.artifact_store.writer( + run_id, + &ArtifactKey::new(stage_id.clone(), retry, entry.path.clone()), + ) { Ok(writer) => writer, Err(err) => { return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()) @@ -6695,6 +6708,10 @@ async fn put_stage_artifact( if let Err(response) = load_run_spec(state.as_ref(), &id).await.map(|_| ()) { return response; } + let retry = match required_query_param(params.retry.as_ref(), "retry") { + Ok(retry) => retry, + Err(response) => return response, + }; let content_length = match content_length_from_headers(&parts.headers) { Ok(length) => length, @@ -6702,7 +6719,7 @@ async fn put_stage_artifact( }; match artifact_upload_content_type(&parts.headers) { Ok(ArtifactUploadContentType::OctetStream) => { - let filename = match required_filename(params) { + let filename = match required_query_param(params.filename.as_ref(), "filename") { Ok(filename) => filename, Err(response) => return response, }; @@ -6710,6 +6727,7 @@ async fn put_stage_artifact( state.as_ref(), &id, &stage_id, + retry, filename, body, content_length, @@ -6722,7 +6740,8 @@ async fn put_stage_artifact( "multipart upload exceeds the {MAX_MULTIPART_REQUEST_BYTES} byte limit" )); } - upload_stage_artifact_multipart(state.as_ref(), &id, &stage_id, boundary, body).await + upload_stage_artifact_multipart(state.as_ref(), &id, &stage_id, retry, boundary, body) + .await } Err(response) => response, } @@ -6742,10 +6761,14 @@ async fn get_stage_artifact( Ok(stage_id) => stage_id, Err(response) => return response, }; - let filename = match required_filename(params) { + let filename = match required_query_param(params.filename.as_ref(), "filename") { Ok(filename) => filename, Err(response) => return response, }; + let retry = match required_query_param(params.retry.as_ref(), "retry") { + Ok(retry) => retry, + Err(response) => return response, + }; let relative_path = match validate_relative_artifact_path("filename", &filename) { Ok(path) => path, Err(response) => return response, @@ -6756,7 +6779,10 @@ async fn get_stage_artifact( match state .artifact_store - .get(&id, &stage_id, &relative_path) + .get( + &id, + &ArtifactKey::new(stage_id.clone(), retry, relative_path), + ) .await { Ok(Some(bytes)) => octet_stream_response(bytes), @@ -12612,7 +12638,7 @@ slug = "fabro" let req = Request::builder() .method("POST") .uri(api(&format!( - "/runs/{run_id}/stages/{stage_id}/artifacts?filename=src/lib.rs" + "/runs/{run_id}/stages/{stage_id}/artifacts?filename=src/lib.rs&retry=1" ))) .header("content-type", "application/octet-stream") .body(Body::from("fn main() {}")) @@ -12628,6 +12654,8 @@ slug = "fabro" let response = app.clone().oneshot(req).await.unwrap(); let body = response_json!(response, StatusCode::OK).await; assert_eq!(body["data"][0]["filename"], "src/lib.rs"); + assert_eq!(body["data"][0]["retry"], 1); + assert_eq!(body["data"][0]["size"], 12); let req = Request::builder() .method("GET") @@ -12636,11 +12664,66 @@ slug = "fabro" ))) .body(Body::empty()) .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::BAD_REQUEST).await; + + let req = Request::builder() + .method("GET") + .uri(api(&format!( + "/runs/{run_id}/stages/{stage_id}/artifacts/download?filename=src/lib.rs&retry=1" + ))) + .body(Body::empty()) + .unwrap(); let response = app.oneshot(req).await.unwrap(); let bytes = response_bytes!(response, StatusCode::OK).await; assert_eq!(&bytes[..], b"fn main() {}"); } + #[tokio::test] + async fn stage_artifacts_keep_same_filename_per_retry() { + let state = create_app_state(); + let app = build_router(Arc::clone(&state), AuthMode::Disabled); + + let run_id = create_run(&app, MINIMAL_DOT).await; + let stage_id = "code@2"; + + for (retry, body) in [(1, "first"), (2, "second")] { + let req = Request::builder() + .method("POST") + .uri(api(&format!( + "/runs/{run_id}/stages/{stage_id}/artifacts?filename=logs/output.txt&retry={retry}" + ))) + .header("content-type", "application/octet-stream") + .body(Body::from(body)) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + assert_status!(response, StatusCode::NO_CONTENT).await; + } + + let req = Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/stages/{stage_id}/artifacts"))) + .body(Body::empty()) + .unwrap(); + let response = app.clone().oneshot(req).await.unwrap(); + let body = response_json!(response, StatusCode::OK).await; + assert_eq!(body["data"][0]["filename"], "logs/output.txt"); + assert_eq!(body["data"][0]["retry"], 1); + assert_eq!(body["data"][1]["filename"], "logs/output.txt"); + assert_eq!(body["data"][1]["retry"], 2); + + let req = Request::builder() + .method("GET") + .uri(api(&format!( + "/runs/{run_id}/stages/{stage_id}/artifacts/download?filename=logs/output.txt&retry=2" + ))) + .body(Body::empty()) + .unwrap(); + let response = app.oneshot(req).await.unwrap(); + let bytes = response_bytes!(response, StatusCode::OK).await; + assert_eq!(&bytes[..], b"second"); + } + #[tokio::test] async fn create_run_persists_run_spec() { let state = create_app_state(); @@ -12672,7 +12755,7 @@ slug = "fabro" let req = Request::builder() .method("POST") .uri(api(&format!( - "/runs/{run_id}/stages/code@2/artifacts?filename=../escape.txt" + "/runs/{run_id}/stages/code@2/artifacts?filename=../escape.txt&retry=1" ))) .header("content-type", "application/octet-stream") .body(Body::from("nope")) @@ -12813,7 +12896,7 @@ slug = "fabro" Request::builder() .method(Method::POST) .uri(api(&format!( - "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt" + "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt&retry=1" ))) .header(header::AUTHORIZATION, format!("Bearer {worker_token}")) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -12830,7 +12913,7 @@ slug = "fabro" Request::builder() .method(Method::POST) .uri(api(&format!( - "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt" + "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt&retry=1" ))) .header(header::AUTHORIZATION, format!("Bearer {user_jwt}")) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -12847,7 +12930,7 @@ slug = "fabro" Request::builder() .method(Method::POST) .uri(api(&format!( - "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt" + "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt&retry=1" ))) .header( header::AUTHORIZATION, @@ -12866,7 +12949,7 @@ slug = "fabro" Request::builder() .method(Method::POST) .uri(api(&format!( - "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt" + "/runs/{run_id}/stages/code@2/artifacts?filename=artifact.txt&retry=1" ))) .header(header::CONTENT_TYPE, "application/octet-stream") .body(Body::from("artifact")) @@ -13064,7 +13147,9 @@ slug = "fabro" let req = Request::builder() .method("POST") - .uri(api(&format!("/runs/{run_id}/stages/{stage_id}/artifacts"))) + .uri(api(&format!( + "/runs/{run_id}/stages/{stage_id}/artifacts?retry=1" + ))) .header( "content-type", format!("multipart/form-data; boundary={boundary}"), @@ -13085,12 +13170,16 @@ slug = "fabro" let response = app.clone().oneshot(req).await.unwrap(); let body = response_json!(response, StatusCode::OK).await; assert_eq!(body["data"][0]["filename"], "logs/output.txt"); + assert_eq!(body["data"][0]["retry"], 1); + assert_eq!(body["data"][0]["size"], log_bytes.len()); assert_eq!(body["data"][1]["filename"], "src/lib.rs"); + assert_eq!(body["data"][1]["retry"], 1); + assert_eq!(body["data"][1]["size"], source_bytes.len()); let req = Request::builder() .method("GET") .uri(api(&format!( - "/runs/{run_id}/stages/{stage_id}/artifacts/download?filename=logs/output.txt" + "/runs/{run_id}/stages/{stage_id}/artifacts/download?filename=logs/output.txt&retry=1" ))) .body(Body::empty()) .unwrap(); @@ -13112,7 +13201,9 @@ slug = "fabro" let req = Request::builder() .method("POST") - .uri(api(&format!("/runs/{run_id}/stages/code@2/artifacts"))) + .uri(api(&format!( + "/runs/{run_id}/stages/code@2/artifacts?retry=1" + ))) .header( "content-type", format!("multipart/form-data; boundary={boundary}"), diff --git a/lib/crates/fabro-server/tests/it/scenario/archive.rs b/lib/crates/fabro-server/tests/it/scenario/archive.rs index fd7a431d8..5a0c60bdf 100644 --- a/lib/crates/fabro-server/tests/it/scenario/archive.rs +++ b/lib/crates/fabro-server/tests/it/scenario/archive.rs @@ -93,7 +93,7 @@ async fn archived_runs_reject_mutations_with_actionable_body() { ), ( "POST", - format!("/runs/{run_id}/stages/fake@1/artifacts?filename=smoke.txt"), + format!("/runs/{run_id}/stages/fake@1/artifacts?filename=smoke.txt&retry=1"), "payload", "application/octet-stream", ), diff --git a/lib/crates/fabro-store/src/artifact_store.rs b/lib/crates/fabro-store/src/artifact_store.rs index d09bfcfdd..4cd4f85e1 100644 --- a/lib/crates/fabro-store/src/artifact_store.rs +++ b/lib/crates/fabro-store/src/artifact_store.rs @@ -16,9 +16,35 @@ const ARTIFACT_SEGMENT_ENCODE_SET: &AsciiSet = &NON_ALPHANUMERIC.remove(b'.').remove(b'_').remove(b'-'); const STREAM_BUFFER_BYTES: usize = 1024 * 1024; +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct ArtifactKey { + pub stage_id: StageId, + pub retry: u32, + pub relative_path: String, +} + +impl ArtifactKey { + #[must_use] + pub fn new(stage_id: StageId, retry: u32, relative_path: impl Into) -> Self { + Self { + stage_id, + retry, + relative_path: relative_path.into(), + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] pub struct NodeArtifact { pub node: StageId, + pub retry: u32, + pub filename: String, + pub size: u64, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)] +pub struct StageArtifactEntry { + pub retry: u32, pub filename: String, pub size: u64, } @@ -46,22 +72,16 @@ impl ArtifactStore { } } - pub async fn put( - &self, - run_id: &RunId, - node: &StageId, - filename: &str, - data: &[u8], - ) -> Result<()> { - let path = self.artifact_path(run_id, node, filename)?; + pub async fn put(&self, run_id: &RunId, key: &ArtifactKey, data: &[u8]) -> Result<()> { + let path = self.artifact_path(run_id, key)?; self.object_store .put(&path, Bytes::copy_from_slice(data).into()) .await?; Ok(()) } - pub fn writer(&self, run_id: &RunId, node: &StageId, filename: &str) -> Result { - let path = self.artifact_path(run_id, node, filename)?; + pub fn writer(&self, run_id: &RunId, key: &ArtifactKey) -> Result { + let path = self.artifact_path(run_id, key)?; Ok(BufWriter::with_capacity( Arc::clone(&self.object_store), path, @@ -72,14 +92,13 @@ impl ArtifactStore { pub async fn put_stream( &self, run_id: &RunId, - node: &StageId, - filename: &str, + key: &ArtifactKey, mut stream: S, ) -> Result<()> where S: futures::Stream> + Unpin, { - let mut writer = self.writer(run_id, node, filename)?; + let mut writer = self.writer(run_id, key)?; while let Some(chunk) = stream.next().await { let chunk = chunk?; writer @@ -94,13 +113,8 @@ impl ArtifactStore { Ok(()) } - pub async fn get( - &self, - run_id: &RunId, - node: &StageId, - filename: &str, - ) -> Result> { - let path = self.artifact_path(run_id, node, filename)?; + pub async fn get(&self, run_id: &RunId, key: &ArtifactKey) -> Result> { + let path = self.artifact_path(run_id, key)?; match self.object_store.get(&path).await { Ok(result) => Ok(Some(result.bytes().await?)), Err(object_store::Error::NotFound { .. }) => Ok(None), @@ -123,15 +137,23 @@ impl ArtifactStore { Ok(artifacts) } - pub async fn list_for_node(&self, run_id: &RunId, node: &StageId) -> Result> { + pub async fn list_for_node( + &self, + run_id: &RunId, + node: &StageId, + ) -> Result> { let prefix = self.node_prefix(run_id, node)?; let mut stream = self.object_store.list(Some(&prefix)); - let mut filenames = Vec::new(); + let mut entries = Vec::new(); while let Some(meta) = stream.next().await.transpose()? { - filenames.push(decode_filename(&prefix, &meta.location)?); + entries.push(decode_stage_artifact_entry( + &prefix, + &meta.location, + meta.size, + )?); } - filenames.sort(); - Ok(filenames) + entries.sort(); + Ok(entries) } pub async fn delete_for_run(&self, run_id: &RunId) -> Result<()> { @@ -171,9 +193,18 @@ impl ArtifactStore { ) } - fn artifact_path(&self, run_id: &RunId, node: &StageId, filename: &str) -> Result { + fn retry_prefix(&self, run_id: &RunId, node: &StageId, retry: u32) -> Result { let mut raw = self.node_prefix(run_id, node)?.to_string(); - for segment in validate_filename_segments(filename)? { + raw.push('/'); + raw.push_str(&retry_storage_segment(retry)); + parse_object_path(&raw) + } + + fn artifact_path(&self, run_id: &RunId, key: &ArtifactKey) -> Result { + let mut raw = self + .retry_prefix(run_id, &key.stage_id, key.retry)? + .to_string(); + for segment in validate_filename_segments(&key.relative_path)? { raw.push('/'); raw.push_str(&encode_path_segment(segment)); } @@ -225,6 +256,13 @@ pub fn stage_storage_segment(node: &StageId) -> String { ) } +const RETRY_SEGMENT_PREFIX: &str = "retry-"; + +#[must_use] +pub fn retry_storage_segment(retry: u32) -> String { + format!("{RETRY_SEGMENT_PREFIX}{retry:04}") +} + fn decode_path_segment(kind: &str, value: &str) -> Result { percent_decode_str(value) .decode_utf8() @@ -258,29 +296,48 @@ fn decode_artifact_location( "artifact location {location} has an invalid visit number: {err}" )) })?; - let filename_segments = parts - .map(|part| decode_path_segment("artifact filename segment", part.as_ref())) - .collect::>>()?; - if filename_segments.is_empty() { - return Err(Error::Other(format!( - "artifact location {location} is missing a filename" - ))); - } + let (retry, filename) = decode_retry_and_filename(location, &mut parts)?; Ok(NodeArtifact { node: StageId::new(node_id, visit), - filename: filename_segments.join("/"), + retry, + filename, size, }) } -fn decode_filename(prefix: &ObjectPath, location: &ObjectPath) -> Result { +fn decode_stage_artifact_entry( + prefix: &ObjectPath, + location: &ObjectPath, + size: u64, +) -> Result { let mut parts = location.prefix_match(prefix).ok_or_else(|| { Error::Other(format!( "artifact location {location} does not match expected prefix {prefix}" )) })?; + let (retry, filename) = decode_retry_and_filename(location, &mut parts)?; + Ok(StageArtifactEntry { + retry, + filename, + size, + }) +} + +fn decode_retry_and_filename<'a, I, P>( + location: &ObjectPath, + parts: &mut I, +) -> Result<(u32, String)> +where + I: Iterator, + P: AsRef + 'a, +{ + let retry_part = parts.next().ok_or_else(|| { + Error::Other(format!( + "artifact location {location} is missing a retry segment" + )) + })?; + let retry = decode_retry_segment(location, retry_part.as_ref())?; let filename_segments = parts - .by_ref() .map(|part| decode_path_segment("artifact filename segment", part.as_ref())) .collect::>>()?; if filename_segments.is_empty() { @@ -288,7 +345,20 @@ fn decode_filename(prefix: &ObjectPath, location: &ObjectPath) -> Result "artifact location {location} is missing a filename" ))); } - Ok(filename_segments.join("/")) + Ok((retry, filename_segments.join("/"))) +} + +fn decode_retry_segment(location: &ObjectPath, segment: &str) -> Result { + let Some(value) = segment.strip_prefix(RETRY_SEGMENT_PREFIX) else { + return Err(Error::Other(format!( + "artifact location {location} has an invalid retry segment" + ))); + }; + value.parse::().map_err(|err| { + Error::Other(format!( + "artifact location {location} has an invalid retry number: {err}" + )) + }) } fn parse_object_path(raw: &str) -> Result { @@ -334,19 +404,25 @@ mod tests { let run_id = fixtures::RUN_1; let node = StageId::new("build/naive @ alpha/π", 12); let filename = "logs/unicode/naive file ☃.txt"; + let key = ArtifactKey::new(node.clone(), 3, filename); - store.put(&run_id, &node, filename, b"hello").await.unwrap(); + store.put(&run_id, &key, b"hello").await.unwrap(); assert_eq!( - store.get(&run_id, &node, filename).await.unwrap(), + store.get(&run_id, &key).await.unwrap(), Some(Bytes::from_static(b"hello")) ); assert_eq!(store.list_for_node(&run_id, &node).await.unwrap(), vec![ - filename.to_string() + StageArtifactEntry { + retry: 3, + filename: filename.to_string(), + size: 5, + } ]); assert_eq!(store.list_for_run(&run_id).await.unwrap(), vec![ NodeArtifact { node, + retry: 3, filename: filename.to_string(), size: 5, } @@ -359,12 +435,12 @@ mod tests { let run_id = fixtures::RUN_1; let node = StageId::new("build", 2); let filename = "logs/output.txt"; + let key = ArtifactKey::new(node, 1, filename); store .put_stream( &run_id, - &node, - filename, + &key, stream::iter(vec![ Ok(Bytes::from_static(b"hello ")), Ok(Bytes::from_static(b"world")), @@ -374,7 +450,7 @@ mod tests { .unwrap(); assert_eq!( - store.get(&run_id, &node, filename).await.unwrap(), + store.get(&run_id, &key).await.unwrap(), Some(Bytes::from_static(b"hello world")) ); } @@ -392,10 +468,8 @@ mod tests { "logs/./output.txt", r"logs\output.txt", ] { - let err = store - .put(&run_id, &node, filename, b"boom") - .await - .unwrap_err(); + let key = ArtifactKey::new(node.clone(), 1, filename); + let err = store.put(&run_id, &key, b"boom").await.unwrap_err(); assert!(err.to_string().contains("artifact filename")); } } @@ -407,13 +481,24 @@ mod tests { let other_run_id = fixtures::RUN_2; let node = StageId::new("build", 1); - store.put(&run_id, &node, "a.txt", b"a").await.unwrap(); store - .put(&run_id, &node, "nested/b.txt", b"b") + .put(&run_id, &ArtifactKey::new(node.clone(), 1, "a.txt"), b"a") .await .unwrap(); store - .put(&other_run_id, &node, "keep.txt", b"keep") + .put( + &run_id, + &ArtifactKey::new(node.clone(), 1, "nested/b.txt"), + b"b", + ) + .await + .unwrap(); + store + .put( + &other_run_id, + &ArtifactKey::new(node.clone(), 1, "keep.txt"), + b"keep", + ) .await .unwrap(); @@ -422,7 +507,77 @@ mod tests { assert!(store.list_for_run(&run_id).await.unwrap().is_empty()); assert_eq!( store.list_for_node(&other_run_id, &node).await.unwrap(), - vec!["keep.txt".to_string()] + vec![StageArtifactEntry { + retry: 1, + filename: "keep.txt".to_string(), + size: 4, + }] ); } + + #[tokio::test] + async fn preserves_same_filename_across_retries() { + let store = test_store(); + let run_id = fixtures::RUN_1; + let node = StageId::new("build", 1); + let first = ArtifactKey::new(node.clone(), 1, "logs/output.txt"); + let second = ArtifactKey::new(node.clone(), 2, "logs/output.txt"); + + store.put(&run_id, &first, b"first").await.unwrap(); + store.put(&run_id, &second, b"second").await.unwrap(); + + assert_eq!( + store.get(&run_id, &first).await.unwrap(), + Some(Bytes::from_static(b"first")) + ); + assert_eq!( + store.get(&run_id, &second).await.unwrap(), + Some(Bytes::from_static(b"second")) + ); + assert_eq!(store.list_for_node(&run_id, &node).await.unwrap(), vec![ + StageArtifactEntry { + retry: 1, + filename: "logs/output.txt".to_string(), + size: 5, + }, + StageArtifactEntry { + retry: 2, + filename: "logs/output.txt".to_string(), + size: 6, + }, + ]); + assert_eq!(store.list_for_run(&run_id).await.unwrap(), vec![ + NodeArtifact { + node: node.clone(), + retry: 1, + filename: "logs/output.txt".to_string(), + size: 5, + }, + NodeArtifact { + node, + retry: 2, + filename: "logs/output.txt".to_string(), + size: 6, + }, + ]); + } + + #[tokio::test] + async fn rejects_legacy_artifact_paths_without_retry_segment() { + let object_store: Arc = Arc::new(InMemory::new()); + let store = ArtifactStore::new(object_store.clone(), "artifacts"); + let run_id = fixtures::RUN_1; + object_store + .put( + &ObjectPath::from(format!("artifacts/{run_id}/build@0001/output.txt")), + Bytes::from_static(b"legacy").into(), + ) + .await + .unwrap(); + + let err = store.list_for_run(&run_id).await.unwrap_err(); + + assert!(err.to_string().contains("invalid retry segment")); + assert!(err.to_string().contains("build@0001/output.txt")); + } } diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 5a97b4b9c..8c0165b24 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -10,7 +10,10 @@ mod serializable_projection; mod slate; mod types; -pub use artifact_store::{ArtifactStore, NodeArtifact, stage_storage_segment}; +pub use artifact_store::{ + ArtifactKey, ArtifactStore, NodeArtifact, StageArtifactEntry, retry_storage_segment, + stage_storage_segment, +}; pub use error::{Error, Result}; pub use fabro_types::{ EventEnvelope, PendingInterviewRecord, RunBlobId, RunProjection, RunSummary, StageId, diff --git a/lib/crates/fabro-workflow/src/artifact_upload.rs b/lib/crates/fabro-workflow/src/artifact_upload.rs index 2fb9d3c9a..265e2ad63 100644 --- a/lib/crates/fabro-workflow/src/artifact_upload.rs +++ b/lib/crates/fabro-workflow/src/artifact_upload.rs @@ -11,6 +11,7 @@ pub trait StageArtifactUploader: Send + Sync { async fn upload_stage_artifacts( &self, stage_id: &StageId, + retry: u32, artifact_capture_dir: &Path, artifacts: &[ArtifactUpload], ) -> Result<()>; diff --git a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs index 128f3f5a2..f65fcee86 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/artifact.rs @@ -8,7 +8,7 @@ use fabro_core::graph::NodeSpec; use fabro_core::lifecycle::{AttemptContext, AttemptResultContext, NodeDecision, RunLifecycle}; use fabro_core::outcome::NodeResult; use fabro_core::state::ExecutionState; -use fabro_store::ArtifactStore; +use fabro_store::{ArtifactKey, ArtifactStore}; use fabro_types::{ArtifactUpload, RunId, StageId}; use tokio::fs; use tokio::time::sleep; @@ -18,7 +18,7 @@ use crate::artifact_snapshot::collect_artifacts; use crate::artifact_upload::ArtifactSink; use crate::event::{Emitter, Event, RunNoticeLevel}; use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::lifecycle::event::stage_scope_for; +use crate::lifecycle::event::{stage_scope_for, stage_visit}; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; @@ -95,7 +95,7 @@ impl RunLifecycle for ArtifactLifecycle { } let epoch = self.attempt_start_epoch.lock().unwrap().unwrap_or(0.0); let node_id = ctx.node.id(); - let visit = state.node_visits.get(node_id).copied().unwrap_or(1); + let visit = stage_visit(state, node_id); let node_slug = if visit <= 1 { node_id.to_string() } else { @@ -113,10 +113,11 @@ impl RunLifecycle for ArtifactLifecycle { .await { Ok(summary) if summary.files_copied > 0 => { - let stage_id = StageId::new(node_id.to_string(), ctx.attempt); + let stage_id = StageId::new(node_id.to_string(), visit); if let Err(err) = self .persist_artifacts( &stage_id, + ctx.attempt, artifact_capture_dir.path(), &summary.captured_assets, ) @@ -199,6 +200,7 @@ impl ArtifactLifecycle { async fn persist_artifacts( &self, stage_id: &StageId, + retry: u32, artifact_capture_dir: &std::path::Path, artifacts: &[ArtifactUpload], ) -> Result<()> { @@ -209,7 +211,7 @@ impl ArtifactLifecycle { let mut last_error = None; for attempt in 0..=ARTIFACT_UPLOAD_RETRY_DELAYS.len() { match self - .persist_artifacts_once(sink, stage_id, artifact_capture_dir, artifacts) + .persist_artifacts_once(sink, stage_id, retry, artifact_capture_dir, artifacts) .await { Ok(()) => return Ok(()), @@ -228,17 +230,18 @@ impl ArtifactLifecycle { &self, sink: &ArtifactSink, stage_id: &StageId, + retry: u32, artifact_capture_dir: &std::path::Path, artifacts: &[ArtifactUpload], ) -> Result<()> { match sink { ArtifactSink::Store(store) => { - self.store_artifacts(store, stage_id, artifact_capture_dir, artifacts) + self.store_artifacts(store, stage_id, retry, artifact_capture_dir, artifacts) .await } ArtifactSink::Uploader(uploader) => { uploader - .upload_stage_artifacts(stage_id, artifact_capture_dir, artifacts) + .upload_stage_artifacts(stage_id, retry, artifact_capture_dir, artifacts) .await } } @@ -248,6 +251,7 @@ impl ArtifactLifecycle { &self, store: &ArtifactStore, stage_id: &StageId, + retry: u32, artifact_capture_dir: &std::path::Path, artifacts: &[ArtifactUpload], ) -> Result<()> { @@ -257,7 +261,11 @@ impl ArtifactLifecycle { .await .with_context(|| format!("failed to read artifact {}", local_path.display()))?; store - .put(&self.run_id, stage_id, &artifact.path, &bytes) + .put( + &self.run_id, + &ArtifactKey::new(stage_id.clone(), retry, artifact.path.clone()), + &bytes, + ) .await .map_err(anyhow::Error::new)?; } diff --git a/lib/crates/fabro-workflow/src/lifecycle/event.rs b/lib/crates/fabro-workflow/src/lifecycle/event.rs index 0c8cc9880..647ef2abd 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/event.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/event.rs @@ -86,9 +86,9 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } -fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { +pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { let visits = state.node_visits.get(node_id).copied().unwrap_or(1); - u32::try_from(visits.max(1)).unwrap_or(u32::MAX) + u32::try_from(visits).unwrap_or(u32::MAX) } pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope { diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index c3fe3c215..7e48de0a0 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -68,7 +68,8 @@ pub(crate) async fn build_conclusion_from_store( run_duration_ms: u64, final_git_commit_sha: Option, ) -> Conclusion { - let projection = run_store.state().await.ok(); + let (state_result, events_result) = tokio::join!(run_store.state(), run_store.list_events()); + let projection = state_result.ok(); let projection_order = projection .as_ref() .map(stage_projection_order) @@ -76,9 +77,7 @@ pub(crate) async fn build_conclusion_from_store( let checkpoint = projection .as_ref() .and_then(|state| state.checkpoint.as_ref()); - let stage_durations = run_store - .list_events() - .await + let stage_durations = events_result .map(|events| crate::extract_stage_durations_from_events(&events)) .unwrap_or_default(); diff --git a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs index 4fc56ec7d..b8188a539 100644 --- a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs @@ -27,7 +27,7 @@ use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_llm::provider::Provider; use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfig}; use fabro_static::EnvVars; -use fabro_store::{ArtifactStore, Database}; +use fabro_store::{ArtifactKey, ArtifactStore, Database}; use fabro_types::{RunId, StageId, WorkflowSettings}; use fabro_workflow::artifact::sync_artifacts_to_env; use fabro_workflow::context::Context; @@ -1387,8 +1387,11 @@ async fn daytona_asset_collection() { test_artifact_store(dir.path()) .get( &run_options.run_id, - &StageId::new("create_assets", 1), - "test-results/report.xml", + &ArtifactKey::new( + StageId::new("create_assets", 1), + 1, + "test-results/report.xml", + ), ) .await .unwrap() diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 9b8916cd8..002d473da 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -31,7 +31,7 @@ use fabro_interview::{ QueueInterviewer, RecordingInterviewer, }; use fabro_llm::provider::Provider; -use fabro_store::{ArtifactStore, Database}; +use fabro_store::{ArtifactKey, ArtifactStore, Database}; use fabro_types::{CommandTermination, RunEvent, RunId, StageId, WorkflowSettings, parse_blob_ref}; use fabro_validate::{Severity, validate, validate_or_raise}; use fabro_workflow::context::Context; @@ -13015,8 +13015,11 @@ async fn asset_collection_local_sandbox_success() { artifact_store .get( &run_options.run_id, - &StageId::new("create_assets", 1), - "test-results/report.xml", + &ArtifactKey::new( + StageId::new("create_assets", 1), + 1, + "test-results/report.xml", + ), ) .await .unwrap() @@ -13137,8 +13140,11 @@ async fn asset_collection_local_sandbox_on_failure() { test_artifact_store(run_dir.path()) .get( &run_options.run_id, - &StageId::new("create_assets", 1), - "test-results/report.xml", + &ArtifactKey::new( + StageId::new("create_assets", 1), + 1, + "test-results/report.xml", + ), ) .await .unwrap() @@ -13241,8 +13247,11 @@ async fn asset_collection_docker_sandbox() { test_artifact_store(run_dir.path()) .get( &run_options.run_id, - &StageId::new("create_assets", 1), - "test-results/report.xml", + &ArtifactKey::new( + StageId::new("create_assets", 1), + 1, + "test-results/report.xml", + ), ) .await .unwrap() diff --git a/lib/packages/fabro-api-client/src/api/run-internals-api.ts b/lib/packages/fabro-api-client/src/api/run-internals-api.ts index 6bcf36c3a..0852b16f6 100644 --- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts +++ b/lib/packages/fabro-api-client/src/api/run-internals-api.ts @@ -290,16 +290,19 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. + * @param {number} retry Retry attempt number for the artifact. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - getStageArtifact: async (id: string, stageId: string, filename: string, options: RawAxiosRequestConfig = {}): Promise => { + getStageArtifact: async (id: string, stageId: string, filename: string, retry: number, options: RawAxiosRequestConfig = {}): Promise => { // verify required parameter 'id' is not null or undefined assertParamExists('getStageArtifact', 'id', id) // verify required parameter 'stageId' is not null or undefined assertParamExists('getStageArtifact', 'stageId', stageId) // verify required parameter 'filename' is not null or undefined assertParamExists('getStageArtifact', 'filename', filename) + // verify required parameter 'retry' is not null or undefined + assertParamExists('getStageArtifact', 'retry', retry) const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/artifacts/download` .replace(`{${"id"}}`, encodeURIComponent(String(id))) .replace(`{${"stageId"}}`, encodeURIComponent(String(stageId))); @@ -324,6 +327,10 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config localVarQueryParameter['filename'] = filename; } + if (retry !== undefined) { + localVarQueryParameter['retry'] = retry; + } + localVarHeaderParameter['Accept'] = 'application/octet-stream,application/json'; setSearchParams(localVarUrlObj, localVarQueryParameter); @@ -578,16 +585,19 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. + * @param {number} retry Retry attempt number for the artifact. * @param {File} body * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - putStageArtifact: async (id: string, stageId: string, body: File, filename?: string, options: RawAxiosRequestConfig = {}): Promise => { + putStageArtifact: async (id: string, stageId: string, retry: number, body: File, filename?: string, options: RawAxiosRequestConfig = {}): Promise => { // verify required parameter 'id' is not null or undefined assertParamExists('putStageArtifact', 'id', id) // verify required parameter 'stageId' is not null or undefined assertParamExists('putStageArtifact', 'stageId', stageId) + // verify required parameter 'retry' is not null or undefined + assertParamExists('putStageArtifact', 'retry', retry) // verify required parameter 'body' is not null or undefined assertParamExists('putStageArtifact', 'body', body) const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/artifacts` @@ -610,6 +620,10 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config // http bearer authentication required await setBearerAuthToObject(localVarHeaderParameter, configuration) + if (retry !== undefined) { + localVarQueryParameter['retry'] = retry; + } + if (filename !== undefined) { localVarQueryParameter['filename'] = filename; } @@ -882,11 +896,12 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. + * @param {number} retry Retry attempt number for the artifact. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async getStageArtifact(id: string, stageId: string, filename: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { - const localVarAxiosArgs = await localVarAxiosParamCreator.getStageArtifact(id, stageId, filename, options); + async getStageArtifact(id: string, stageId: string, filename: string, retry: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.getStageArtifact(id, stageId, filename, retry, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.getStageArtifact']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); @@ -969,13 +984,14 @@ export const RunInternalsApiFp = function(configuration?: Configuration) { * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. + * @param {number} retry Retry attempt number for the artifact. * @param {File} body * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - async putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { - const localVarAxiosArgs = await localVarAxiosParamCreator.putStageArtifact(id, stageId, body, filename, options); + async putStageArtifact(id: string, stageId: string, retry: number, body: File, filename?: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + const localVarAxiosArgs = await localVarAxiosParamCreator.putStageArtifact(id, stageId, retry, body, filename, options); const localVarOperationServerIndex = configuration?.serverIndex ?? 0; const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.putStageArtifact']?.[localVarOperationServerIndex]?.url; return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath); @@ -1105,11 +1121,12 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. + * @param {number} retry Retry attempt number for the artifact. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - getStageArtifact(id: string, stageId: string, filename: string, options?: RawAxiosRequestConfig): AxiosPromise { - return localVarFp.getStageArtifact(id, stageId, filename, options).then((request) => request(axios, basePath)); + getStageArtifact(id: string, stageId: string, filename: string, retry: number, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.getStageArtifact(id, stageId, filename, retry, options).then((request) => request(axios, basePath)); }, /** * Lists captured artifact files for a run. @@ -1174,13 +1191,14 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. + * @param {number} retry Retry attempt number for the artifact. * @param {File} body * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig): AxiosPromise { - return localVarFp.putStageArtifact(id, stageId, body, filename, options).then((request) => request(axios, basePath)); + putStageArtifact(id: string, stageId: string, retry: number, body: File, filename?: string, options?: RawAxiosRequestConfig): AxiosPromise { + return localVarFp.putStageArtifact(id, stageId, retry, body, filename, options).then((request) => request(axios, basePath)); }, /** * Reads a previously stored blob by identifier. @@ -1298,11 +1316,12 @@ export class RunInternalsApi extends BaseAPI { * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. * @param {string} filename Relative artifact path. `/` is allowed as a path separator. Backslash, empty segments, and traversal segments (`.` and `..`) are invalid. + * @param {number} retry Retry attempt number for the artifact. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - public getStageArtifact(id: string, stageId: string, filename: string, options?: RawAxiosRequestConfig) { - return RunInternalsApiFp(this.configuration).getStageArtifact(id, stageId, filename, options).then((request) => request(this.axios, this.basePath)); + public getStageArtifact(id: string, stageId: string, filename: string, retry: number, options?: RawAxiosRequestConfig) { + return RunInternalsApiFp(this.configuration).getStageArtifact(id, stageId, filename, retry, options).then((request) => request(this.axios, this.basePath)); } /** @@ -1373,13 +1392,14 @@ export class RunInternalsApi extends BaseAPI { * @summary Put Stage Artifact * @param {string} id Unique run identifier (ULID). * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`. + * @param {number} retry Retry attempt number for the artifact. * @param {File} body * @param {string} [filename] Relative artifact path for `application/octet-stream` uploads. Ignored for multipart uploads. * @param {*} [options] Override http request option. * @throws {RequiredError} */ - public putStageArtifact(id: string, stageId: string, body: File, filename?: string, options?: RawAxiosRequestConfig) { - return RunInternalsApiFp(this.configuration).putStageArtifact(id, stageId, body, filename, options).then((request) => request(this.axios, this.basePath)); + public putStageArtifact(id: string, stageId: string, retry: number, body: File, filename?: string, options?: RawAxiosRequestConfig) { + return RunInternalsApiFp(this.configuration).putStageArtifact(id, stageId, retry, body, filename, options).then((request) => request(this.axios, this.basePath)); } /** diff --git a/lib/packages/fabro-api-client/src/models/artifact-entry.ts b/lib/packages/fabro-api-client/src/models/artifact-entry.ts index 6fbd0e851..4bd667edc 100644 --- a/lib/packages/fabro-api-client/src/models/artifact-entry.ts +++ b/lib/packages/fabro-api-client/src/models/artifact-entry.ts @@ -15,12 +15,20 @@ /** - * A single artifact filename. + * A single artifact file for a stage. */ export interface ArtifactEntry { /** * Artifact filename. */ 'filename': string; + /** + * Retry attempt number. + */ + 'retry': number; + /** + * Artifact size in bytes. + */ + 'size': number; } diff --git a/lib/packages/fabro-api-client/src/models/artifact-list-response.ts b/lib/packages/fabro-api-client/src/models/artifact-list-response.ts index 708bb4ad5..dcf2bfc90 100644 --- a/lib/packages/fabro-api-client/src/models/artifact-list-response.ts +++ b/lib/packages/fabro-api-client/src/models/artifact-list-response.ts @@ -18,7 +18,7 @@ import type { ArtifactEntry } from './artifact-entry'; /** - * List of artifact filenames for a stage. + * List of artifact files for a stage. */ export interface ArtifactListResponse { 'data': Array; diff --git a/lib/packages/fabro-api-client/src/models/stage-projection.ts b/lib/packages/fabro-api-client/src/models/stage-projection.ts index 336c9b34b..41321ec78 100644 --- a/lib/packages/fabro-api-client/src/models/stage-projection.ts +++ b/lib/packages/fabro-api-client/src/models/stage-projection.ts @@ -28,11 +28,23 @@ export interface StageProjection { 'prompt'?: string | null; 'response'?: string | null; 'completion'?: StageCompletion | null; - 'provider_used'?: any; + /** + * Provider and model metadata recorded for the stage attempt. + */ + 'provider_used'?: object | null; 'diff'?: string | null; - 'script_invocation'?: any; - 'script_timing'?: any; - 'parallel_results'?: any; + /** + * Command and environment recorded when the stage script ran. + */ + 'script_invocation'?: object | null; + /** + * Wall-clock and step timing metadata for the stage script. + */ + 'script_timing'?: object | null; + /** + * Per-branch result objects produced by a parallel stage. + */ + 'parallel_results'?: Array | null; 'stdout'?: string | null; 'stderr'?: string | null; 'stdout_bytes'?: number | null;