From bafdd880f5880d7596637c0962e9dd1a0c5ffd80 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 11 Sep 2026 14:23:45 -0600 Subject: [PATCH] Simplify workflow version registration tool layering Move supplied-content packaging into fabro-manifest beside the checkout collector, and narrow the injected seam to a packager that returns the dependency-ordered closure so ClientBackend registers versions with the client it already owns. Validate the tool input once through a ValidatedWorkflowVersionCreate newtype, matching the other tools, instead of re-validating at three layers. Reuse the fabro-types unique-map deserializer and the shared "not available" error helper, derive budget messages from the limit constants, and render the tool result through the shared summary+JSON path used by sibling tools. Share one extension dispatch between WorkflowLocation::resolve and from_exact_path, compute the bundler's normalized reference once, key path-collision checks by a Cow so the canonical exact check no longer allocates, and log the full packaging error chain before returning the curated tool message. Replace the hand-rolled axum test server with httpmock and declare the new unicode dependencies at the workspace. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 1 + Cargo.toml | 2 + lib/apps/fabro-cli/src/commands/run/runner.rs | 4 +- lib/apps/fabro-mcp-server/src/server.rs | 22 +- lib/apps/fabro-server/src/server/tests.rs | 6 +- .../fabro-server/src/workflow_version_tool.rs | 484 +++--------------- lib/components/fabro-manifest/Cargo.toml | 2 +- lib/components/fabro-manifest/src/lib.rs | 2 + .../fabro-manifest/src/supplied_workflow.rs | 276 ++++++++++ .../fabro-manifest/src/workflow_bundler.rs | 40 +- lib/components/fabro-tool/Cargo.toml | 1 + lib/components/fabro-tool/src/common.rs | 13 +- lib/components/fabro-tool/src/fabro_client.rs | 157 +++++- lib/components/fabro-tool/src/lib.rs | 4 +- .../fabro-tool/src/workflow_version.rs | 185 +++---- lib/foundation/fabro-config/src/project.rs | 24 +- lib/foundation/fabro-types/Cargo.toml | 4 +- lib/foundation/fabro-types/src/lib.rs | 2 +- .../fabro-types/src/workflow_version.rs | 36 +- 19 files changed, 693 insertions(+), 572 deletions(-) create mode 100644 lib/components/fabro-manifest/src/supplied_workflow.rs diff --git a/Cargo.lock b/Cargo.lock index b7c953564..8923ac91a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3169,6 +3169,7 @@ dependencies = [ "fabro-types", "fabro-util", "futures", + "httpmock", "schemars 1.2.1", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 8bbbfb212..20d1ecf54 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,6 +65,8 @@ console = "0.15" dialoguer = "0.12" git2 = { version = "0.20", default-features = false, features = ["vendored-libgit2", "vendored-openssl", "https"] } tracing = "0.1" +unicase = "2" +unicode-normalization = "0.1" tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } tracing-appender = "0.2" rmcp = { version = "1.4", default-features = false } diff --git a/lib/apps/fabro-cli/src/commands/run/runner.rs b/lib/apps/fabro-cli/src/commands/run/runner.rs index df7b2f531..36f149e3a 100644 --- a/lib/apps/fabro-cli/src/commands/run/runner.rs +++ b/lib/apps/fabro-cli/src/commands/run/runner.rs @@ -16,7 +16,7 @@ use fabro_interview::{ WorkerControlMessage, }; use fabro_server::run_tool_manifest; -use fabro_server::workflow_version_tool::ServerWorkflowVersionCreateAdapter; +use fabro_server::workflow_version_tool::ServerWorkflowVersionPackager; use fabro_store::{EventEnvelope, RunProjection, RunProjectionReducer}; use fabro_tool::fabro_client::ClientBackend; use fabro_types::settings::run::{RunMode, RunNamespace}; @@ -238,7 +238,7 @@ fn build_fabro_run_tool_services( } let backend = ClientBackend::new(Arc::new(client)) .with_manifest_builder(Arc::new(WorkerRunManifestBuilder)) - .with_workflow_version_create_adapter(Arc::new(ServerWorkflowVersionCreateAdapter)); + .with_workflow_version_packager(Arc::new(ServerWorkflowVersionPackager)); Some(FabroRunToolServices { backend: Arc::new(backend), current_run_id, diff --git a/lib/apps/fabro-mcp-server/src/server.rs b/lib/apps/fabro-mcp-server/src/server.rs index 794da973a..914833318 100644 --- a/lib/apps/fabro-mcp-server/src/server.rs +++ b/lib/apps/fabro-mcp-server/src/server.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use std::time::Duration; use anyhow::Result; -use fabro_server::workflow_version_tool::ServerWorkflowVersionCreateAdapter; +use fabro_server::workflow_version_tool::ServerWorkflowVersionPackager; use fabro_tool::fabro_client::ClientBackend; use fabro_tool::{self as run_tools, FabroToolBackend}; use fabro_util::version::FABRO_VERSION; @@ -99,14 +99,15 @@ impl FabroMcpServer { &self, params: Parameters, ) -> Result { - if let Err(err) = params.0.validate() { - return Ok(error_result(&err)); - } + let source = match run_tools::ValidatedWorkflowVersionCreate::try_from(params.0) { + Ok(source) => source, + Err(err) => return Ok(error_result(&err)), + }; let backend = match self.backend().await { Ok(backend) => backend, Err(err) => return Ok(error_result(&err)), }; - match run_tools::create_workflow_version(backend, params.0).await { + match run_tools::create_workflow_version(backend, source).await { Ok(result) => success_result(&result, run_tools::workflow_version_create_text(&result)), Err(err) => Ok(error_result(&err)), } @@ -275,8 +276,8 @@ impl FabroMcpServer { Arc::new( ClientBackend::new(Arc::new(client)) .with_manifest_builder(Arc::new(McpRunManifestBuilder)) - .with_workflow_version_create_adapter(Arc::new( - ServerWorkflowVersionCreateAdapter, + .with_workflow_version_packager(Arc::new( + ServerWorkflowVersionPackager, )), ) as Arc }) @@ -360,7 +361,12 @@ mod tests { actual.as_object_mut().unwrap().remove("$schema"); assert_eq!(actual, expected); assert_eq!(tool.description.as_deref(), Some(definition.description)); - let result = server.fabro_workflow_version_create(Parameters(serde_json::from_value(serde_json::json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}})).unwrap())).await.unwrap(); + let params = + serde_json::json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}}); + let result = server + .fabro_workflow_version_create(Parameters(serde_json::from_value(params).unwrap())) + .await + .unwrap(); assert_eq!( result.structured_content, Some(serde_json::json!({"workflow_version_id":id})) diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index e81a3b576..a0a4533f6 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -19743,7 +19743,11 @@ fn validate_github_slug_rejects_overlong() { async fn workflow_version_registration_requires_user_or_run_tools_capability() { let (state, app) = jwt_auth_app(); let run_id = RunId::new(); - let body = json!({"entrypoint":"workflow.fabro","files":{"workflow.fabro":"digraph W {}"},"workflow_dependencies":{}}); + let body = json!({ + "entrypoint": "workflow.fabro", + "files": {"workflow.fabro": "digraph W {}"}, + "workflow_dependencies": {}, + }); for (token, expected) in [ (issue_test_user_jwt(), StatusCode::CREATED), ( diff --git a/lib/apps/fabro-server/src/workflow_version_tool.rs b/lib/apps/fabro-server/src/workflow_version_tool.rs index 27081eea0..816c526b5 100644 --- a/lib/apps/fabro-server/src/workflow_version_tool.rs +++ b/lib/apps/fabro-server/src/workflow_version_tool.rs @@ -1,99 +1,50 @@ -use std::path::Path; - -use anyhow::Context as _; use async_trait::async_trait; -use fabro_client::Client; -use fabro_config::project::WorkflowLocation; -use fabro_manifest::CollectedWorkflowClosure; -use fabro_tool::{FabroWorkflowVersionCreateParams, ToolError, WorkflowVersionCreateAdapter}; -use fabro_types::WorkflowVersionId; +use fabro_tool::{ + PackagedWorkflowVersions, ToolError, ValidatedWorkflowVersionCreate, WorkflowVersionPackager, +}; use tokio::task; +use tracing::warn; -/// Content-only registration shared by standalone MCP and capable run workers. -pub struct ServerWorkflowVersionCreateAdapter; +/// Packages supplied workflow contents for standalone MCP and capable run +/// workers; the backend that owns the API client performs registration. +pub struct ServerWorkflowVersionPackager; + +const PACKAGING_FAILED: &str = "workflow source could not be packaged; check configuration, \ + syntax, local references, and package limits"; #[async_trait] -impl WorkflowVersionCreateAdapter for ServerWorkflowVersionCreateAdapter { - async fn create_workflow_version( +impl WorkflowVersionPackager for ServerWorkflowVersionPackager { + async fn package( &self, - params: FabroWorkflowVersionCreateParams, - client: &Client, - ) -> anyhow::Result { - params.validate()?; + source: ValidatedWorkflowVersionCreate, + ) -> anyhow::Result { let closure = task::spawn_blocking(move || { - let staging = tempfile::Builder::new().prefix("fabro-workflow-version-").tempdir()?; - collect_supplied_workflow(¶ms, staging) + fabro_manifest::collect_supplied_workflow_versions(&source.entrypoint, &source.files) + }) + .await + .map_err(|err| anyhow::anyhow!("workflow packaging task failed: {err}"))? + .map_err(|err| { + // Parser diagnostics may quote supplied source, so the chain stays + // in the log and only a generic message crosses the tool boundary. + warn!(error = %format!("{err:#}"), "workflow version packaging failed"); + ToolError::message(PACKAGING_FAILED) + })?; + Ok(PackagedWorkflowVersions { + root_id: closure.root_id(), + versions: closure + .versions() + .map(|(_, version)| version.version().clone()) + .collect(), }) - .await - .context("workflow packaging task failed")? - // Parser diagnostics may contain supplied source. Keep them off the - // tool result boundary, including nested error chains. - .map_err(|_| ToolError::message("workflow source could not be packaged; check configuration, syntax, local references, and package limits"))?; - let versions = closure - .versions() - .map(|(_, version)| version.version()) - .collect::>(); - client.register_workflow_versions(versions).await?; - Ok(closure.root_id()) } } -#[expect( - clippy::disallowed_methods, - reason = "bounded file staging and collection run on spawn_blocking" -)] -fn collect_supplied_workflow( - params: &FabroWorkflowVersionCreateParams, - staging: tempfile::TempDir, -) -> anyhow::Result { - // TempDir owns cleanup on every return path, including collection errors. - let root = staging.path().canonicalize()?; - for (path, contents) in ¶ms.files { - let destination = root.join(path.as_str()); - if let Some(parent) = destination.parent() { - std::fs::create_dir_all(parent)?; - } - std::fs::write(destination, contents)?; - } - let location = WorkflowLocation::from_exact_path(Path::new(params.entrypoint.as_str()), &root)?; - let closure = fabro_manifest::collect_workflow_versions_at_location( - &location, - &root, - Path::new(params.entrypoint.as_str()), - )?; - // Collection validates the whole closure, including serialized request budgets. - // A case-insensitive host must not satisfy a missing exact source key. - for (_, version) in closure.versions() { - for (path, content) in version.version().files() { - anyhow::ensure!( - params.files.get(path) == Some(content), - "collected file does not match supplied source" - ); - } - } - drop(staging); - Ok(closure) -} - #[cfg(test)] mod tests { - #![expect( - clippy::disallowed_methods, - reason = "hermetic temporary source fixtures" - )] - use std::path::Path; - use std::sync::{Arc, Mutex}; - - use axum::http::StatusCode; - use axum::routing::post; - use axum::{Json, Router}; - use fabro_types::WorkflowVersion; - use serde_json::json; - use super::*; - fn params(entrypoint: &str, files: &[(&str, &str)]) -> FabroWorkflowVersionCreateParams { - FabroWorkflowVersionCreateParams { + fn source(entrypoint: &str, files: &[(&str, &str)]) -> ValidatedWorkflowVersionCreate { + ValidatedWorkflowVersionCreate { entrypoint: entrypoint.parse().unwrap(), files: files .iter() @@ -102,8 +53,8 @@ mod tests { } } - fn fixture() -> FabroWorkflowVersionCreateParams { - params("workflow.toml", &[ + fn fixture() -> ValidatedWorkflowVersionCreate { + source("workflow.toml", &[ ( "workflow.toml", "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n", @@ -112,350 +63,77 @@ mod tests { "workflow.fabro", r#"digraph W { p [prompt="@prompt.md"] child [stack.child_workflow="child.fabro"] }"#, ), - ( - "prompt.md", - "Keep {{ secrets.TEST }} and {{ env.TEST }} for runtime.", - ), + ("prompt.md", "Review the implementation."), ("child.fabro", "digraph Child {}"), ]) } - fn collect(params: &FabroWorkflowVersionCreateParams) -> CollectedWorkflowClosure { - params.validate().unwrap(); - collect_supplied_workflow(params, tempfile::tempdir().unwrap()).unwrap() - } - - #[test] - fn workflow_version_content_matches_existing_collector_and_cleans_staging() { - for input in [ - params("workflow.fabro", &[("workflow.fabro", "digraph W {}")]), - fixture(), - ] { - let source = tempfile::tempdir().unwrap(); - for (path, content) in &input.files { - std::fs::write(source.path().join(path.as_str()), content).unwrap(); - } - let expected = fabro_manifest::collect_workflow_versions( - Path::new(input.entrypoint.as_str()), - source.path(), - ) + #[tokio::test] + async fn packager_returns_dependencies_before_root() { + let packaged = ServerWorkflowVersionPackager + .package(fixture()) + .await .unwrap(); - let staging = tempfile::tempdir().unwrap(); - let path = staging.path().to_owned(); - let actual = collect_supplied_workflow(&input, staging).unwrap(); - assert!(!path.exists()); - assert_eq!(actual.root_id(), expected.root_id()); - assert_eq!( - actual - .versions() - .map(|(_, v)| v.version()) - .collect::>(), - expected - .versions() - .map(|(_, v)| v.version()) - .collect::>() - ); - } - } - - #[test] - fn workflow_version_exact_extensionless_entrypoint_and_child_ignore_selectors() { - let input = params("workflow", &[ - ( - "workflow", - r#"digraph W { child [stack.child_workflow="child"] }"#, - ), - ("child", "digraph Child {}"), - ( - ".fabro/project.toml", - "malformed project config must not be read", - ), - ( - ".fabro/workflows/workflow/workflow.toml", - "misleading named workflow", - ), - ]); - let closure = collect(&input); - let versions = closure.versions().collect::>(); - assert_eq!(versions.len(), 2); - assert_eq!(versions[1].1.version().entrypoint().as_str(), "workflow"); - assert_eq!(versions[0].1.version().entrypoint().as_str(), "child"); - } - - #[test] - fn workflow_version_rejects_missing_and_escaping_references_and_cleans_failure() { - let parent = tempfile::tempdir().unwrap(); - std::fs::write( - parent.path().join("outside.md"), - "host content must never satisfy a reference", - ) - .unwrap(); - std::fs::write(parent.path().join("child.fabro"), "digraph Host {}").unwrap(); - for (index, input) in [ - params("workflow.fabro", &[ - ("workflow.fabro", r#"digraph W { p [prompt="@prompt.md"] }"#), - ("Prompt.md", "wrong case"), - ]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [prompt="@../outside.md"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [prompt="@sub/../../outside.md"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [prompt="@outside.md"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [output_schema="@../outside.md"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [stack.child_workflow="../child.fabro"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [stack.child_workflow="sub/../../child.fabro"] }"#, - )]), - params("workflow.fabro", &[( - "workflow.fabro", - r#"digraph W { p [stack.child_workflow="missing"] }"#, - )]), - params("workflow.toml", &[( - "workflow.toml", - "_version = 1\n[workflow]\ngraph = \"../child.fabro\"\n", - )]), - params("workflow.fabro", &[( - "workflow.fabro", - "invalid source PRIVATE_CONTENT", - )]), - ] - .into_iter() - .enumerate() - { - let staging = tempfile::tempdir_in(parent.path()).unwrap(); - let path = staging.path().to_owned(); - assert!( - collect_supplied_workflow(&input, staging).is_err(), - "accepted invalid fixture {index}" - ); - assert!(!path.exists()); - } - } - - #[test] - fn workflow_version_registration_preserves_literal_scripts_without_executing() { - let directory = tempfile::tempdir().unwrap(); - let marker = directory.path().join("must-not-exist"); - // Script is literal command text in Fabro, not an @file import. - let graph = format!( - "digraph W {{ command [script=\"touch {}\"] }}", - marker.display() - ); - let input = params("workflow", &[("workflow", &graph)]); - let closure = collect(&input); - let root = closure.versions().last().unwrap().1.version(); - assert_eq!(root.files()[&"workflow".parse().unwrap()], graph); - assert!(!marker.exists()); - } - - #[test] - fn workflow_version_map_order_is_irrelevant_and_reachable_changes_change_ids() { - let input = fixture(); - let first = collect(&input); - let mut reordered = input.clone(); - reordered.files = input.files.into_iter().rev().collect(); - assert_eq!(first.root_id(), collect(&reordered).root_id()); - reordered - .files - .insert("prompt.md".parse().unwrap(), "changed".into()); - let changed = collect(&reordered); - assert_ne!(first.root_id(), changed.root_id()); + assert_eq!(packaged.versions.len(), 2); + assert_eq!(packaged.versions[0].entrypoint().as_str(), "child.fabro"); assert_eq!( - first.versions().next().unwrap().0, - changed.versions().next().unwrap().0 + packaged.versions[1].id().unwrap(), + packaged.root_id, + "root version must be last" ); - reordered - .files - .insert("child.fabro".parse().unwrap(), "digraph Changed {}".into()); - let changed_child = collect(&reordered); - assert_ne!(changed.root_id(), changed_child.root_id()); - assert_ne!( - changed.versions().next().unwrap().0, - changed_child.versions().next().unwrap().0 + let child_id = packaged.versions[0].id().unwrap(); + assert!( + packaged.versions[1] + .workflow_dependencies() + .values() + .any(|id| *id == child_id) ); } #[tokio::test] - async fn workflow_version_uploads_dependencies_first_and_retries_immutable_content() { - let uploads = Arc::new(Mutex::new(Vec::::new())); - let seen = uploads.clone(); - let app = Router::new().route( - "/api/v1/workflow-versions", - post(move |Json(version): Json| { - let seen = seen.clone(); - async move { - let mut seen = seen.lock().unwrap(); - for id in version.workflow_dependencies().values() { - assert!( - seen.iter().any(|prior| prior.id().unwrap() == *id), - "dependency must be registered first" - ); - } - let id = version.id().unwrap(); - seen.push(version); - (StatusCode::CREATED, Json(json!({"workflow_version_id":id}))) - } - }), - ); - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let client = - Client::new_no_proxy(&format!("http://{}", listener.local_addr().unwrap())).unwrap(); - let server = tokio::spawn(async move { - axum::serve(listener, app).await.unwrap(); - }); - let expected = collect(&fixture()).root_id(); - for _ in 0..2 { - let id = ServerWorkflowVersionCreateAdapter - .create_workflow_version(fixture(), &client) - .await - .unwrap(); - assert_eq!(id, expected); - } - assert_eq!(uploads.lock().unwrap().len(), 4); - server.abort(); - assert!(server.await.unwrap_err().is_cancelled()); - } - - #[tokio::test] - async fn workflow_version_invalid_closure_has_no_uploads_or_source_in_errors() { - let server = httpmock::MockServer::start_async().await; - let upload = server - .mock_async(|when, then| { - when.method(httpmock::Method::POST); - then.status(500); - }) - .await; - let client = Client::new_no_proxy(&server.url("")).unwrap(); - let mut invalid = fixture(); + async fn packaging_errors_never_quote_supplied_source() { + let mut invalid_root = fixture(); // Child is valid, but the root fails after its dependency is assembled. - invalid.files.insert("workflow.toml".parse().unwrap(), "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"missing.md\"".into()); + invalid_root.files.insert( + "workflow.toml".parse().unwrap(), + "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"missing.md\"" + .into(), + ); let mut oversized = fixture(); - let huge_prompt = "\u{1}".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES - 1); - oversized - .files - .insert("prompt.md".parse().unwrap(), huge_prompt); + oversized.files.insert( + "prompt.md".parse().unwrap(), + "\u{1}".repeat(fabro_types::MAX_WORKFLOW_VERSION_FILE_BYTES - 1), + ); for input in [ - invalid, + invalid_root, oversized, - params("workflow", &[( + source("workflow", &[( "workflow", "PRIVATE_CONTENT invalid source", )]), ] { - let error = ServerWorkflowVersionCreateAdapter - .create_workflow_version(input, &client) + let error = ServerWorkflowVersionPackager + .package(input) .await .unwrap_err(); - assert!(!format!("{error:#}").contains("PRIVATE_CONTENT")); + let rendered = format!("{error:#}"); + assert!(!rendered.contains("PRIVATE_CONTENT"), "{rendered}"); + assert_eq!(rendered, PACKAGING_FAILED); } - upload.assert_calls_async(0).await; - } - - #[tokio::test] - async fn workflow_version_root_upload_failure_leaves_child_for_safe_retry() { - let server = httpmock::MockServer::start_async().await; - let closure = collect(&fixture()); - let child = closure.versions().next().unwrap().1.version(); - let root = closure.versions().last().unwrap().1.version(); - let child_upload = server - .mock_async(|when, then| { - when.method(httpmock::Method::POST) - .path("/api/v1/workflow-versions") - .json_body_obj(child); - then.status(201) - .json_body(json!({"workflow_version_id":child.id().unwrap()})); - }) - .await; - let failed_root = server - .mock_async(|when, then| { - when.method(httpmock::Method::POST) - .path("/api/v1/workflow-versions") - .json_body_obj(root); - then.status(400); - }) - .await; - let deletion = server - .mock_async(|when, then| { - when.method(httpmock::Method::DELETE); - then.status(500); - }) - .await; - let client = Client::new_no_proxy(&server.url("")).unwrap(); + let mut wrong_case = fixture(); + let prompt = wrong_case + .files + .remove(&"prompt.md".parse().unwrap()) + .unwrap(); + wrong_case + .files + .insert("Prompt.md".parse().unwrap(), prompt); assert!( - ServerWorkflowVersionCreateAdapter - .create_workflow_version(fixture(), &client) + ServerWorkflowVersionPackager + .package(wrong_case) .await - .is_err() + .is_err(), + "a case-insensitive host must not satisfy an exact reference" ); - child_upload.assert_calls_async(1).await; - failed_root.assert_calls_async(1).await; - failed_root.delete_async().await; - let root_upload = server - .mock_async(|when, then| { - when.method(httpmock::Method::POST) - .path("/api/v1/workflow-versions") - .json_body_obj(root); - then.status(201) - .json_body(json!({"workflow_version_id":root.id().unwrap()})); - }) - .await; - assert_eq!( - ServerWorkflowVersionCreateAdapter - .create_workflow_version(fixture(), &client) - .await - .unwrap(), - closure.root_id() - ); - child_upload.assert_calls_async(2).await; - root_upload.assert_calls_async(1).await; - deletion.assert_calls_async(0).await; - } - - #[tokio::test] - async fn workflow_version_failed_upload_and_wrong_server_id_never_return_success() { - for wrong_id in [false, true] { - let server = httpmock::MockServer::start_async().await; - let closure = collect(&fixture()); - let child = closure.versions().next().unwrap().1.version(); - let upload = server.mock_async(|when, then| { - when.method(httpmock::Method::POST).path("/api/v1/workflow-versions").json_body_obj(child); - if wrong_id { - then.status(201).json_body(json!({"workflow_version_id": WorkflowVersionId::from(fabro_types::BlobHash::new(b"wrong"))})); - } else { then.status(400); } - }).await; - let root = closure.versions().last().unwrap().1.version(); - let root_upload = server - .mock_async(|when, then| { - when.method(httpmock::Method::POST) - .path("/api/v1/workflow-versions") - .json_body_obj(root); - then.status(201) - .json_body(json!({"workflow_version_id":root.id().unwrap()})); - }) - .await; - let client = Client::new_no_proxy(&server.url("")).unwrap(); - assert!( - ServerWorkflowVersionCreateAdapter - .create_workflow_version(fixture(), &client) - .await - .is_err() - ); - upload.assert_calls_async(1).await; - root_upload.assert_calls_async(0).await; - } } } diff --git a/lib/components/fabro-manifest/Cargo.toml b/lib/components/fabro-manifest/Cargo.toml index 1ac7e1eca..dcb395052 100644 --- a/lib/components/fabro-manifest/Cargo.toml +++ b/lib/components/fabro-manifest/Cargo.toml @@ -23,6 +23,7 @@ fabro-types = { path = "../../foundation/fabro-types" } fabro-workflow = { path = "../fabro-workflow" } fabro-workflow-version = { path = "../fabro-workflow-version" } git2.workspace = true +tempfile = "3" thiserror.workspace = true toml.workspace = true @@ -31,5 +32,4 @@ fabro-test.workspace = true fabro-util = { path = "../../foundation/fabro-util" } insta.workspace = true serde_json.workspace = true -tempfile = "3" temp-env = "0.3" diff --git a/lib/components/fabro-manifest/src/lib.rs b/lib/components/fabro-manifest/src/lib.rs index a3f36c3e4..9f46f2be7 100644 --- a/lib/components/fabro-manifest/src/lib.rs +++ b/lib/components/fabro-manifest/src/lib.rs @@ -4,6 +4,7 @@ )] mod local_workflow_package; +mod supplied_workflow; mod workflow_bundler; mod workflow_version_collector; @@ -34,6 +35,7 @@ use fabro_workflow::git::{self, GitSyncStatus}; pub use crate::local_workflow_package::{ LocalWorkflowPackageError, ResolvedLocalWorkflowPackage, resolve_local_workflow_package, }; +pub use crate::supplied_workflow::collect_supplied_workflow_versions; use crate::workflow_bundler::WorkflowBundler; pub use crate::workflow_version_collector::{ CollectedWorkflowClosure, WorkflowVersionCollectError, collect_workflow_versions, diff --git a/lib/components/fabro-manifest/src/supplied_workflow.rs b/lib/components/fabro-manifest/src/supplied_workflow.rs new file mode 100644 index 000000000..3d0d7ab5b --- /dev/null +++ b/lib/components/fabro-manifest/src/supplied_workflow.rs @@ -0,0 +1,276 @@ +//! Package workflow versions from caller-supplied file contents instead of a +//! checkout on disk. + +use std::collections::BTreeMap; +use std::path::Path; + +use anyhow::Result; +use fabro_config::project::WorkflowLocation; +use fabro_types::WorkflowPath; +use tempfile::TempDir; + +use crate::CollectedWorkflowClosure; + +/// Stage `files` in a private temporary directory and collect the workflow +/// closure rooted at `entrypoint` with the same collector used for checkouts. +/// Only supplied files can satisfy references; the staging directory is +/// removed on every return path. Every dependency is validated before this +/// returns and nothing is registered. +pub fn collect_supplied_workflow_versions( + entrypoint: &WorkflowPath, + files: &BTreeMap, +) -> Result { + let staging = tempfile::Builder::new() + .prefix("fabro-workflow-version-") + .tempdir()?; + collect_in_staging(entrypoint, files, &staging) +} + +fn collect_in_staging( + entrypoint: &WorkflowPath, + files: &BTreeMap, + staging: &TempDir, +) -> Result { + let root = staging.path().canonicalize()?; + for (path, contents) in files { + let destination = root.join(path.as_str()); + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent)?; + } + std::fs::write(destination, contents)?; + } + let entrypoint = Path::new(entrypoint.as_str()); + let location = WorkflowLocation::from_exact_path(entrypoint, &root)?; + let closure = crate::collect_workflow_versions_at_location(&location, &root, entrypoint)?; + // A case-insensitive host must not satisfy a reference that is missing + // from the supplied tree under its exact key. + for (_, version) in closure.versions() { + for path in version.version().files().keys() { + anyhow::ensure!( + files.contains_key(path), + "collected file `{path}` was not supplied" + ); + } + } + Ok(closure) +} + +#[cfg(test)] +mod tests { + use std::path::Path; + + use super::*; + + struct Supplied { + entrypoint: WorkflowPath, + files: BTreeMap, + } + + fn supplied(entrypoint: &str, files: &[(&str, &str)]) -> Supplied { + Supplied { + entrypoint: entrypoint.parse().unwrap(), + files: files + .iter() + .map(|(path, content)| (path.parse().unwrap(), (*content).to_string())) + .collect(), + } + } + + fn fixture() -> Supplied { + supplied("workflow.toml", &[ + ( + "workflow.toml", + "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n", + ), + ( + "workflow.fabro", + r#"digraph W { p [prompt="@prompt.md"] child [stack.child_workflow="child.fabro"] }"#, + ), + ( + "prompt.md", + "Keep {{ secrets.TEST }} and {{ env.TEST }} for runtime.", + ), + ("child.fabro", "digraph Child {}"), + ]) + } + + fn collect(input: &Supplied) -> CollectedWorkflowClosure { + collect_supplied_workflow_versions(&input.entrypoint, &input.files).unwrap() + } + + /// Consumes `staging` so the tests can assert cleanup after return. + fn collect_with_staging( + input: &Supplied, + staging: TempDir, + ) -> Result { + let result = collect_in_staging(&input.entrypoint, &input.files, &staging); + drop(staging); + result + } + + #[test] + fn supplied_content_matches_checkout_collector_and_cleans_staging() { + for input in [ + supplied("workflow.fabro", &[("workflow.fabro", "digraph W {}")]), + fixture(), + ] { + let source = tempfile::tempdir().unwrap(); + for (path, content) in &input.files { + std::fs::write(source.path().join(path.as_str()), content).unwrap(); + } + let expected = crate::collect_workflow_versions( + Path::new(input.entrypoint.as_str()), + source.path(), + ) + .unwrap(); + let staging = tempfile::tempdir().unwrap(); + let path = staging.path().to_owned(); + let actual = collect_with_staging(&input, staging).unwrap(); + assert!(!path.exists()); + assert_eq!(actual.root_id(), expected.root_id()); + assert_eq!( + actual + .versions() + .map(|(_, v)| v.version()) + .collect::>(), + expected + .versions() + .map(|(_, v)| v.version()) + .collect::>() + ); + } + } + + #[test] + fn exact_extensionless_entrypoint_and_child_ignore_selectors() { + let input = supplied("workflow", &[ + ( + "workflow", + r#"digraph W { child [stack.child_workflow="child"] }"#, + ), + ("child", "digraph Child {}"), + ( + ".fabro/project.toml", + "malformed project config must not be read", + ), + ( + ".fabro/workflows/workflow/workflow.toml", + "misleading named workflow", + ), + ]); + let closure = collect(&input); + let versions = closure.versions().collect::>(); + assert_eq!(versions.len(), 2); + assert_eq!(versions[1].1.version().entrypoint().as_str(), "workflow"); + assert_eq!(versions[0].1.version().entrypoint().as_str(), "child"); + } + + #[test] + fn rejects_missing_and_escaping_references_and_cleans_failure() { + let parent = tempfile::tempdir().unwrap(); + std::fs::write( + parent.path().join("outside.md"), + "host content must never satisfy a reference", + ) + .unwrap(); + std::fs::write(parent.path().join("child.fabro"), "digraph Host {}").unwrap(); + for (index, input) in [ + supplied("workflow.fabro", &[ + ("workflow.fabro", r#"digraph W { p [prompt="@prompt.md"] }"#), + ("Prompt.md", "wrong case"), + ]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@../outside.md"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@sub/../../outside.md"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [prompt="@outside.md"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [output_schema="@../outside.md"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="../child.fabro"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="sub/../../child.fabro"] }"#, + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + r#"digraph W { p [stack.child_workflow="missing"] }"#, + )]), + supplied("workflow.toml", &[( + "workflow.toml", + "_version = 1\n[workflow]\ngraph = \"../child.fabro\"\n", + )]), + supplied("workflow.fabro", &[( + "workflow.fabro", + "invalid source PRIVATE_CONTENT", + )]), + ] + .into_iter() + .enumerate() + { + let staging = tempfile::tempdir_in(parent.path()).unwrap(); + let path = staging.path().to_owned(); + assert!( + collect_with_staging(&input, staging).is_err(), + "accepted invalid fixture {index}" + ); + assert!(!path.exists()); + } + } + + #[test] + fn preserves_literal_scripts_without_executing() { + let directory = tempfile::tempdir().unwrap(); + let marker = directory.path().join("must-not-exist"); + // Script is literal command text in Fabro, not an @file import. + let graph = format!( + "digraph W {{ command [script=\"touch {}\"] }}", + marker.display() + ); + let input = supplied("workflow", &[("workflow", &graph)]); + let closure = collect(&input); + let root = closure.versions().last().unwrap().1.version(); + assert_eq!(root.files()[&"workflow".parse().unwrap()], graph); + assert!(!marker.exists()); + } + + #[test] + fn map_order_is_irrelevant_and_reachable_changes_change_ids() { + let input = fixture(); + let first = collect(&input); + let mut reordered = Supplied { + entrypoint: input.entrypoint.clone(), + files: input.files.into_iter().rev().collect(), + }; + assert_eq!(first.root_id(), collect(&reordered).root_id()); + reordered + .files + .insert("prompt.md".parse().unwrap(), "changed".into()); + let changed = collect(&reordered); + assert_ne!(first.root_id(), changed.root_id()); + assert_eq!( + first.versions().next().unwrap().0, + changed.versions().next().unwrap().0 + ); + reordered + .files + .insert("child.fabro".parse().unwrap(), "digraph Changed {}".into()); + let changed_child = collect(&reordered); + assert_ne!(changed.root_id(), changed_child.root_id()); + assert_ne!( + changed.versions().next().unwrap().0, + changed_child.versions().next().unwrap().0 + ); + } +} diff --git a/lib/components/fabro-manifest/src/workflow_bundler.rs b/lib/components/fabro-manifest/src/workflow_bundler.rs index 4aaa989f5..6e256d6a3 100644 --- a/lib/components/fabro-manifest/src/workflow_bundler.rs +++ b/lib/components/fabro-manifest/src/workflow_bundler.rs @@ -143,28 +143,28 @@ impl<'a> WorkflowBundler<'a> { /// Relative workflow references with an extension are lexically /// normalized (`..` segments resolved without consulting the filesystem, /// `~` rejected) before resolution, so the file read matches the manifest - /// key. Returns the collected workflow's manifest key. + /// key. Workflow-version projection normalizes every reference and + /// resolves it as an exact path inside the package root, with no + /// workflow-name lookup. Returns the collected workflow's manifest key. fn collect_workflow_entry(&mut self, workflow: &Path, resolve_from: &Path) -> Result { - let location = if self.workflow_version_projection { - let exact = normalize_absolute_path(resolve_from, &workflow.to_string_lossy()) - .ok_or_else(|| anyhow!("unsupported workflow reference"))?; - // Check containment before location resolution can read a config. - manifest_path_from_absolute(&exact, self.package_root)?; - WorkflowLocation::from_exact_path(&exact, self.package_root)? + let normalize = self.workflow_version_projection + || (workflow.extension().is_some() && workflow.is_relative()); + let normalized = if normalize { + normalize_absolute_path(resolve_from, &workflow.to_string_lossy()).ok_or_else(|| { + anyhow!( + "unsupported manifest workflow reference: {}", + workflow.display() + ) + })? } else { - let normalized_workflow = if workflow.extension().is_some() && workflow.is_relative() { - normalize_absolute_path(resolve_from, &workflow.to_string_lossy()).ok_or_else( - || { - anyhow!( - "unsupported manifest workflow reference: {}", - workflow.display() - ) - }, - )? - } else { - workflow.to_path_buf() - }; - WorkflowLocation::resolve(&normalized_workflow, resolve_from)? + workflow.to_path_buf() + }; + let location = if self.workflow_version_projection { + // Check containment before location resolution can read a config. + manifest_path_from_absolute(&normalized, self.package_root)?; + WorkflowLocation::from_exact_path(&normalized, self.package_root)? + } else { + WorkflowLocation::resolve(&normalized, resolve_from)? }; self.collect_workflow_location(&location) } diff --git a/lib/components/fabro-tool/Cargo.toml b/lib/components/fabro-tool/Cargo.toml index 0d3a2862d..892b7a6a7 100644 --- a/lib/components/fabro-tool/Cargo.toml +++ b/lib/components/fabro-tool/Cargo.toml @@ -30,4 +30,5 @@ toml.workspace = true [dev-dependencies] fabro-types = { path = "../../foundation/fabro-types", features = ["test-support"] } +httpmock = "0.8" tempfile = "3" diff --git a/lib/components/fabro-tool/src/common.rs b/lib/components/fabro-tool/src/common.rs index f88249fa2..f154759b8 100644 --- a/lib/components/fabro-tool/src/common.rs +++ b/lib/components/fabro-tool/src/common.rs @@ -50,9 +50,9 @@ pub type ToolResult = Result; pub trait FabroToolBackend: Send + Sync { async fn create_workflow_version( &self, - _params: crate::FabroWorkflowVersionCreateParams, + _source: crate::ValidatedWorkflowVersionCreate, ) -> anyhow::Result { - anyhow::bail!("fabro_workflow_version_create is not available") + Err(workflow_version_tool_unavailable_error()) } async fn create_run_from_spec( @@ -142,6 +142,13 @@ fn pair_tool_unavailable_error() -> anyhow::Error { ToolError::message(format!("{FABRO_RUN_PAIR_TOOL_NAME} is not available")).into() } +pub(crate) fn workflow_version_tool_unavailable_error() -> anyhow::Error { + ToolError::message(format!( + "{FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME} is not available" + )) + .into() +} + pub trait RunManifestBuilder: Send + Sync { fn build_run_manifest( &self, @@ -350,7 +357,7 @@ mod tests { fn workflow_version_create_has_strict_content_schema() { let definition = tool_definitions() .iter() - .find(|definition| definition.name == "fabro_workflow_version_create") + .find(|definition| definition.name == FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME) .expect("workflow version creation should be in the shared catalog"); let schema = &definition.parameters; assert_eq!(schema["additionalProperties"], false); diff --git a/lib/components/fabro-tool/src/fabro_client.rs b/lib/components/fabro-tool/src/fabro_client.rs index eb5599c5e..d01ebb1bd 100644 --- a/lib/components/fabro-tool/src/fabro_client.rs +++ b/lib/components/fabro-tool/src/fabro_client.rs @@ -8,14 +8,14 @@ use fabro_types::{ PairTranscriptResponse, Run, RunId, RunPairStatusResponse, RunProjection, StageId, }; -use crate::{FabroToolBackend, RunManifestBuilder, ToolError}; +use crate::{FabroToolBackend, RunManifestBuilder, ToolError, common}; #[derive(Clone)] pub struct ClientBackend { - client: Arc<::fabro_client::Client>, - manifest_builder: Option>, - run_scope: Option, - workflow_version_create_adapter: Option>, + client: Arc<::fabro_client::Client>, + manifest_builder: Option>, + run_scope: Option, + workflow_version_packager: Option>, } impl ClientBackend { @@ -25,7 +25,7 @@ impl ClientBackend { client, manifest_builder: None, run_scope: None, - workflow_version_create_adapter: None, + workflow_version_packager: None, } } @@ -36,11 +36,11 @@ impl ClientBackend { } #[must_use] - pub fn with_workflow_version_create_adapter( + pub fn with_workflow_version_packager( mut self, - adapter: Arc, + packager: Arc, ) -> Self { - self.workflow_version_create_adapter = Some(adapter); + self.workflow_version_packager = Some(packager); self } @@ -66,19 +66,26 @@ impl ClientBackend { #[async_trait] impl FabroToolBackend for ClientBackend { + /// Package the supplied tree, then register dependencies before parents. + /// Versions are immutable and content-addressed, so a failed upload can be + /// retried with the same contents without cleanup. async fn create_workflow_version( &self, - params: crate::FabroWorkflowVersionCreateParams, + source: crate::ValidatedWorkflowVersionCreate, ) -> anyhow::Result { anyhow::ensure!( self.run_scope.is_none(), "workflow version creation is outside this tool session's run scope" ); - let adapter = self - .workflow_version_create_adapter + let packager = self + .workflow_version_packager .as_ref() - .ok_or_else(|| anyhow::anyhow!("fabro_workflow_version_create is not available"))?; - adapter.create_workflow_version(params, &self.client).await + .ok_or_else(common::workflow_version_tool_unavailable_error)?; + let packaged = packager.package(source).await?; + self.client + .register_workflow_versions(&packaged.versions) + .await?; + Ok(packaged.root_id) } async fn create_run_from_spec( @@ -278,3 +285,125 @@ impl FabroToolBackend for ClientBackend { .await } } + +#[cfg(test)] +mod tests { + use std::collections::BTreeMap; + + use async_trait::async_trait; + use fabro_types::{WorkflowVersion, WorkflowVersionId}; + use serde_json::json; + + use super::*; + use crate::{ + PackagedWorkflowVersions, ValidatedWorkflowVersionCreate, WorkflowVersionPackager, + }; + + struct FixedPackager(PackagedWorkflowVersions); + + #[async_trait] + impl WorkflowVersionPackager for FixedPackager { + async fn package( + &self, + _: ValidatedWorkflowVersionCreate, + ) -> anyhow::Result { + Ok(self.0.clone()) + } + } + + fn version( + entrypoint: &str, + dependencies: BTreeMap, + ) -> WorkflowVersion { + WorkflowVersion::new( + entrypoint.parse().unwrap(), + BTreeMap::from([( + entrypoint.parse().unwrap(), + format!("digraph {entrypoint} {{}}"), + )]), + dependencies, + ) + .unwrap() + } + + fn source() -> ValidatedWorkflowVersionCreate { + ValidatedWorkflowVersionCreate { + entrypoint: "root".parse().unwrap(), + files: BTreeMap::new(), + } + } + + #[tokio::test] + async fn create_workflow_version_registers_packaged_closure_and_retries_after_failure() { + let server = httpmock::MockServer::start_async().await; + let child = version("child", BTreeMap::new()); + let child_id = child.id().unwrap(); + let root = version( + "root", + BTreeMap::from([("child".parse().unwrap(), child_id)]), + ); + let root_id = root.id().unwrap(); + let child_upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(&child); + then.status(201) + .json_body(json!({"workflow_version_id": child_id})); + }) + .await; + let failed_root = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(&root); + then.status(400); + }) + .await; + let client = ::fabro_client::Client::new_no_proxy(&server.url("")).unwrap(); + let backend = ClientBackend::new(Arc::new(client)).with_workflow_version_packager( + Arc::new(FixedPackager(PackagedWorkflowVersions { + root_id, + versions: vec![child, root.clone()], + })), + ); + + assert!(backend.create_workflow_version(source()).await.is_err()); + child_upload.assert_calls_async(1).await; + failed_root.assert_calls_async(1).await; + + // Immutable content: retrying re-sends the child and completes the root. + failed_root.delete_async().await; + let root_upload = server + .mock_async(|when, then| { + when.method(httpmock::Method::POST) + .path("/api/v1/workflow-versions") + .json_body_obj(&root); + then.status(201) + .json_body(json!({"workflow_version_id": root_id})); + }) + .await; + assert_eq!( + backend.create_workflow_version(source()).await.unwrap(), + root_id + ); + child_upload.assert_calls_async(2).await; + root_upload.assert_calls_async(1).await; + } + + #[tokio::test] + async fn create_workflow_version_without_packager_is_unavailable() { + let client = ::fabro_client::Client::new_no_proxy("http://127.0.0.1:1").unwrap(); + let error = ClientBackend::new(Arc::new(client)) + .create_workflow_version(source()) + .await + .unwrap_err(); + assert_eq!( + error.to_string(), + format!( + "{} is not available", + crate::FABRO_WORKFLOW_VERSION_CREATE_TOOL_NAME + ) + ); + } +} diff --git a/lib/components/fabro-tool/src/lib.rs b/lib/components/fabro-tool/src/lib.rs index 64ac87f27..d3ade2e3e 100644 --- a/lib/components/fabro-tool/src/lib.rs +++ b/lib/components/fabro-tool/src/lib.rs @@ -49,6 +49,6 @@ pub use search::{ search_runs, search_runs_text, }; pub use workflow_version::{ - FabroWorkflowVersionCreateParams, WorkflowVersionCreateAdapter, create_workflow_version, - workflow_version_create_text, + FabroWorkflowVersionCreateParams, PackagedWorkflowVersions, ValidatedWorkflowVersionCreate, + WorkflowVersionPackager, create_workflow_version, workflow_version_create_text, }; diff --git a/lib/components/fabro-tool/src/workflow_version.rs b/lib/components/fabro-tool/src/workflow_version.rs index 11567df8a..8adb193f8 100644 --- a/lib/components/fabro-tool/src/workflow_version.rs +++ b/lib/components/fabro-tool/src/workflow_version.rs @@ -5,10 +5,9 @@ use async_trait::async_trait; use fabro_api::types::CreateWorkflowVersionResponse; use fabro_types::{ MAX_WORKFLOW_VERSION_BYTES, MAX_WORKFLOW_VERSION_FILE_BYTES, MAX_WORKFLOW_VERSION_FILES, - WorkflowPath, WorkflowVersionId, + WorkflowPath, WorkflowVersion, WorkflowVersionId, }; use schemars::JsonSchema; -use serde::de::{Error as _, MapAccess, Visitor}; use serde::{Deserialize, Serialize}; use crate::{FabroToolBackend, ToolError, ToolResult}; @@ -23,80 +22,83 @@ pub struct FabroWorkflowVersionCreateParams { pub entrypoint: WorkflowPath, /// All local dependencies, keyed by package-relative path. Values are text /// contents. - #[serde(deserialize_with = "deserialize_files")] + #[serde(deserialize_with = "fabro_types::deserialize_unique_map")] #[schemars(with = "BTreeMap")] pub files: BTreeMap, } -fn deserialize_files<'de, D: serde::Deserializer<'de>>( - deserializer: D, -) -> Result, D::Error> { - struct FilesVisitor; - impl<'de> Visitor<'de> for FilesVisitor { - type Value = BTreeMap; - fn expecting(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - formatter.write_str("workflow files with unique path keys and text contents") - } - fn visit_map>(self, mut map: A) -> Result { - let mut files = BTreeMap::new(); - while let Some((path, content)) = map.next_entry()? { - if files.insert(path, content).is_some() { - return Err(A::Error::custom("duplicate workflow file key")); - } - } - Ok(files) - } - } - deserializer.deserialize_map(FilesVisitor) +/// A supplied source tree whose entrypoint, budgets, and portable path +/// collisions have been checked, so it is safe to stage on a filesystem. +#[derive(Clone, Debug)] +pub struct ValidatedWorkflowVersionCreate { + pub entrypoint: WorkflowPath, + pub files: BTreeMap, } -impl FabroWorkflowVersionCreateParams { - /// Validate the complete supplied tree before any filesystem writes. - pub fn validate(&self) -> ToolResult<()> { - if !self.files.contains_key(&self.entrypoint) { +impl TryFrom for ValidatedWorkflowVersionCreate { + type Error = ToolError; + + fn try_from(params: FabroWorkflowVersionCreateParams) -> Result { + let FabroWorkflowVersionCreateParams { entrypoint, files } = params; + if !files.contains_key(&entrypoint) { return Err(ToolError::message( "entrypoint must be an exact supplied file key", )); } - if self.files.len() > MAX_WORKFLOW_VERSION_FILES { - return Err(ToolError::message("workflow source exceeds 512 files")); + if files.len() > MAX_WORKFLOW_VERSION_FILES { + return Err(ToolError::message(format!( + "workflow source exceeds {MAX_WORKFLOW_VERSION_FILES} files" + ))); } let mut total = 0; - for content in self.files.values() { + for content in files.values() { if content.len() > MAX_WORKFLOW_VERSION_FILE_BYTES { - return Err(ToolError::message("workflow source file exceeds 512 KiB")); + return Err(ToolError::message(format!( + "workflow source file exceeds {} KiB", + MAX_WORKFLOW_VERSION_FILE_BYTES / 1024 + ))); } total += content.len(); } if total > MAX_WORKFLOW_VERSION_BYTES { - return Err(ToolError::message("workflow source exceeds 2 MiB")); + return Err(ToolError::message(format!( + "workflow source exceeds {} MiB", + MAX_WORKFLOW_VERSION_BYTES / (1024 * 1024) + ))); } - fabro_types::validate_workflow_source_paths(self.files.keys()) + fabro_types::validate_workflow_source_paths(files.keys()) .map_err(|_| ToolError::message("workflow source paths collide"))?; - Ok(()) + Ok(Self { entrypoint, files }) } } -/// Application seam for packaging and registering content without a dependency -/// cycle. Implementations must validate before staging, confine reads to -/// supplied files, validate the entire closure before uploading, and register -/// dependencies first. +/// The complete validated closure for one supplied source tree. +#[derive(Clone, Debug)] +pub struct PackagedWorkflowVersions { + pub root_id: WorkflowVersionId, + /// Every version in the closure, dependencies before the versions that + /// reference them, so callers can register them in this order. + pub versions: Vec, +} + +/// Application seam for packaging supplied content. The manifest crates that +/// own collection depend on this crate, so the packager is injected instead. +/// Implementations confine reads to supplied files and validate the entire +/// closure before returning. #[async_trait] -pub trait WorkflowVersionCreateAdapter: Send + Sync { - async fn create_workflow_version( +pub trait WorkflowVersionPackager: Send + Sync { + async fn package( &self, - params: FabroWorkflowVersionCreateParams, - client: &fabro_client::Client, - ) -> anyhow::Result; + source: ValidatedWorkflowVersionCreate, + ) -> anyhow::Result; } pub async fn create_workflow_version( backend: Arc, - params: FabroWorkflowVersionCreateParams, + source: ValidatedWorkflowVersionCreate, ) -> ToolResult { - params.validate()?; let workflow_version_id = backend - .create_workflow_version(params) + .create_workflow_version(source) .await .map_err(|err| ToolError::from_anyhow(&err))?; Ok(CreateWorkflowVersionResponse { @@ -106,7 +108,7 @@ pub async fn create_workflow_version( #[must_use] pub fn workflow_version_create_text(result: &CreateWorkflowVersionResponse) -> String { - serde_json::to_string(result).expect("workflow version response should serialize") + format!("Registered workflow version {}", result.workflow_version_id) } #[cfg(test)] @@ -116,12 +118,15 @@ mod tests { use super::*; use crate::fabro_client::ClientBackend; + fn validate(value: serde_json::Value) -> ToolResult { + let params: FabroWorkflowVersionCreateParams = serde_json::from_value(value).unwrap(); + ValidatedWorkflowVersionCreate::try_from(params) + } + #[test] fn workflow_version_request_rejects_unknown_fields_and_invalid_paths() { let valid = json!({"entrypoint": "workflow", "files": {"workflow": "digraph W {}"}}); - let params: FabroWorkflowVersionCreateParams = - serde_json::from_value(valid.clone()).unwrap(); - params.validate().unwrap(); + validate(valid.clone()).unwrap(); assert!( serde_json::from_str::( r#"{"entrypoint":"workflow","files":{"workflow":"a","workflow":"b"}}"# @@ -160,75 +165,75 @@ mod tests { #[test] fn workflow_version_source_enforces_presence_collisions_and_budgets() { + assert!( + validate(json!({"entrypoint":"missing","files":{"workflow":"digraph W {}"}})).is_err() + ); for files in [ - json!({}), json!({"A":"x","a":"y"}), json!({"A":"x","a/b.md":"y"}), json!({"a":"x","A/b.md":"y"}), ] { let mut files = files.as_object().unwrap().clone(); files.insert("workflow".into(), json!("digraph W {}")); - let mut params: FabroWorkflowVersionCreateParams = - serde_json::from_value(json!({"entrypoint":"workflow","files":files})).unwrap(); - if params.files.len() == 1 { - params.entrypoint = "missing".parse().unwrap(); - } - assert!(params.validate().is_err()); + assert!(validate(json!({"entrypoint":"workflow","files":files})).is_err()); } - let mut params = FabroWorkflowVersionCreateParams { + let oversized_file = FabroWorkflowVersionCreateParams { entrypoint: "workflow".parse().unwrap(), files: BTreeMap::from([( "workflow".parse().unwrap(), "x".repeat(MAX_WORKFLOW_VERSION_FILE_BYTES + 1), )]), }; - assert!(params.validate().is_err()); - params.files = (0..MAX_WORKFLOW_VERSION_FILES) - .map(|i| (format!("file{i}").parse().unwrap(), String::new())) - .collect(); - params + assert!(ValidatedWorkflowVersionCreate::try_from(oversized_file).is_err()); + let mut too_many_files = FabroWorkflowVersionCreateParams { + entrypoint: "workflow".parse().unwrap(), + files: (0..MAX_WORKFLOW_VERSION_FILES) + .map(|i| (format!("file{i}").parse().unwrap(), String::new())) + .collect(), + }; + too_many_files .files - .insert(params.entrypoint.clone(), String::new()); - assert!(params.validate().is_err()); - params.files = (0..5) - .map(|i| { - ( - format!("file{i}").parse().unwrap(), - "x".repeat(MAX_WORKFLOW_VERSION_FILE_BYTES), - ) - }) - .collect(); - params + .insert(too_many_files.entrypoint.clone(), String::new()); + assert!(ValidatedWorkflowVersionCreate::try_from(too_many_files).is_err()); + let mut oversized_total = FabroWorkflowVersionCreateParams { + entrypoint: "workflow".parse().unwrap(), + files: (0..5) + .map(|i| { + ( + format!("file{i}").parse().unwrap(), + "x".repeat(MAX_WORKFLOW_VERSION_FILE_BYTES), + ) + }) + .collect(), + }; + oversized_total .files - .insert(params.entrypoint.clone(), String::new()); - assert!(params.validate().is_err()); + .insert(oversized_total.entrypoint.clone(), String::new()); + assert!(ValidatedWorkflowVersionCreate::try_from(oversized_total).is_err()); } #[tokio::test] - async fn workflow_version_same_run_backend_denies_before_adapter() { + async fn workflow_version_same_run_backend_denies_before_packaging() { let client = fabro_client::Client::new_no_proxy("http://127.0.0.1:1").unwrap(); let backend = ClientBackend::new(Arc::new(client)) - .with_workflow_version_create_adapter(Arc::new(UnreachableAdapter)) + .with_workflow_version_packager(Arc::new(UnreachablePackager)) .with_run_scope("01KRBZW4DW0000000000000002".parse().unwrap()); - let params = serde_json::from_value( - json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}}), - ) - .unwrap(); - let error = create_workflow_version(Arc::new(backend), params) + let source = + validate(json!({"entrypoint":"workflow","files":{"workflow":"digraph W {}"}})).unwrap(); + let error = create_workflow_version(Arc::new(backend), source) .await .unwrap_err(); assert!(error.as_str().contains("run scope")); } - struct UnreachableAdapter; + struct UnreachablePackager; #[async_trait] - impl WorkflowVersionCreateAdapter for UnreachableAdapter { - async fn create_workflow_version( + impl WorkflowVersionPackager for UnreachablePackager { + async fn package( &self, - _: FabroWorkflowVersionCreateParams, - _: &fabro_client::Client, - ) -> anyhow::Result { - panic!("scoped backend must not invoke the adapter") + _: ValidatedWorkflowVersionCreate, + ) -> anyhow::Result { + panic!("scoped backend must not invoke the packager") } } } diff --git a/lib/foundation/fabro-config/src/project.rs b/lib/foundation/fabro-config/src/project.rs index c33f3fc79..f61ae2bff 100644 --- a/lib/foundation/fabro-config/src/project.rs +++ b/lib/foundation/fabro-config/src/project.rs @@ -46,23 +46,19 @@ impl WorkflowLocation { /// graph file (e.g. `workflow.fabro`); the three forms produce the same /// shape. pub fn resolve(arg: &Path, cwd: &Path) -> Result { - let resolved = resolve_workflow_arg_from(arg, cwd)?; - if resolved.extension().is_some_and(|ext| ext == "toml") { - Self::from_toml(resolved) - } else { - Ok(Self::from_graph(resolved)) - } + Self::from_resolved_path(resolve_workflow_arg_from(arg, cwd)?) } - /// Resolve an exact file path without workflow-name or ambient config - /// lookup. Relative paths are interpreted against the supplied - /// directory only. + /// Resolve an exact file path without workflow-name lookup. Relative paths + /// are interpreted against the supplied directory only. pub fn from_exact_path(path: &Path, directory: &Path) -> Result { - let path = directory.join(path); - if path - .extension() - .is_some_and(|extension| extension == "toml") - { + Self::from_resolved_path(directory.join(path)) + } + + /// Dispatch a resolved file path on its extension: `workflow.toml` loads + /// run config, anything else is treated as a graph file. + fn from_resolved_path(path: PathBuf) -> Result { + if path.extension().is_some_and(|ext| ext == "toml") { Self::from_toml(path) } else { Ok(Self::from_graph(path)) diff --git a/lib/foundation/fabro-types/Cargo.toml b/lib/foundation/fabro-types/Cargo.toml index 005fe228f..1fb90fb25 100644 --- a/lib/foundation/fabro-types/Cargo.toml +++ b/lib/foundation/fabro-types/Cargo.toml @@ -33,8 +33,8 @@ strum.workspace = true thiserror.workspace = true toml.workspace = true ulid.workspace = true -unicase = "2" -unicode-normalization = "0.1" +unicase.workspace = true +unicode-normalization.workspace = true url.workspace = true [dev-dependencies] diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index e678a8a1f..a398b827b 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -197,7 +197,7 @@ pub use workflow_path::{ }; pub use workflow_version::{ MAX_WORKFLOW_VERSION_BYTES, MAX_WORKFLOW_VERSION_DEPENDENCIES, MAX_WORKFLOW_VERSION_FILE_BYTES, - MAX_WORKFLOW_VERSION_FILES, WorkflowVersion, WorkflowVersionShapeError, + MAX_WORKFLOW_VERSION_FILES, WorkflowVersion, WorkflowVersionShapeError, deserialize_unique_map, validate_workflow_source_paths, }; pub use workflow_version_id::{WorkflowVersionId, WorkflowVersionIdParseError}; diff --git a/lib/foundation/fabro-types/src/workflow_version.rs b/lib/foundation/fabro-types/src/workflow_version.rs index 996a26444..8b448eaee 100644 --- a/lib/foundation/fabro-types/src/workflow_version.rs +++ b/lib/foundation/fabro-types/src/workflow_version.rs @@ -1,3 +1,4 @@ +use std::borrow::Cow; use std::collections::{BTreeMap, HashMap}; use std::fmt; use std::marker::PhantomData; @@ -153,33 +154,35 @@ impl WorkflowVersion { fn validate_path_collisions(&self) -> Result<(), WorkflowVersionShapeError> { validate_path_collisions( self.files.keys().chain(self.workflow_dependencies.keys()), - false, + Cow::Borrowed, ) } } /// Reject file and directory aliases before materializing a portable source /// tree, including Unicode case folding and normalization. Canonical versions -/// themselves retain their exact, case-sensitive -/// semantics. +/// themselves retain their exact, case-sensitive semantics. pub fn validate_workflow_source_paths<'a>( paths: impl IntoIterator, ) -> Result<(), WorkflowVersionShapeError> { - validate_path_collisions(paths, true) + validate_path_collisions(paths, |text| { + if text.is_ascii() { + Cow::Owned(text.to_ascii_lowercase()) + } else { + Cow::Owned(UniCase::unicode(text).to_folded_case().nfc().collect()) + } + }) } +/// Detect colliding paths under a comparison key: identical keys, or a key +/// that names an ancestor directory of another. fn validate_path_collisions<'a>( paths: impl IntoIterator, - case_insensitive: bool, + key: impl Fn(&'a str) -> Cow<'a, str>, ) -> Result<(), WorkflowVersionShapeError> { let mut by_text = HashMap::new(); for path in paths { - let text = if case_insensitive { - UniCase::new(path.as_str()).to_folded_case().nfc().collect() - } else { - path.as_str().to_owned() - }; - if let Some(existing) = by_text.insert(text, path) { + if let Some(existing) = by_text.insert(key(path.as_str()), path) { return Err(WorkflowVersionShapeError::PathCollision { first: existing.clone(), second: path.clone(), @@ -218,6 +221,17 @@ impl<'de> Deserialize<'de> for WorkflowVersion { } } +/// Deserialize a map while rejecting duplicate keys, which serde would +/// otherwise silently collapse to the last value. +pub fn deserialize_unique_map<'de, D, K, V>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, + K: Deserialize<'de> + Ord + fmt::Display, + V: Deserialize<'de>, +{ + UniqueBTreeMap::deserialize(deserializer).map(|map| map.0) +} + struct UniqueBTreeMap(BTreeMap); impl<'de, K, V> Deserialize<'de> for UniqueBTreeMap