From fefc236ab955c7a18465b2d9a16df0875f84cd55 Mon Sep 17 00:00:00 2001 From: Scott Werner Date: Fri, 4 Sep 2026 16:26:42 -0400 Subject: [PATCH] Collect inline workflows at their exact entrypoint Inline workflow sources were routed through the checkout-selector collector, which rewrites any extensionless relative path to a .fabro/workflows//workflow.toml lookup. A supplied entrypoint such as "review" therefore failed with "workflow was not found" even though its bytes were in the file map. Add a dedicated inline collector in fabro-manifest that treats the entrypoint as an exact key, checks the file paths for filesystem collisions before staging anything, and stages the bytes in a private temporary root only for the duration of collection. The server adapter now delegates to it instead of staging files itself. Co-Authored-By: Claude Fable 5.1 --- lib/apps/fabro-server/src/run_tool_create.rs | 79 ++++++-------- lib/components/fabro-manifest/src/lib.rs | 1 + .../src/workflow_version_collector.rs | 103 ++++++++++++++++++ lib/foundation/fabro-types/src/lib.rs | 2 +- .../fabro-types/src/workflow_version.rs | 7 ++ 5 files changed, 147 insertions(+), 45 deletions(-) diff --git a/lib/apps/fabro-server/src/run_tool_create.rs b/lib/apps/fabro-server/src/run_tool_create.rs index 664bf5de3..8b329a709 100644 --- a/lib/apps/fabro-server/src/run_tool_create.rs +++ b/lib/apps/fabro-server/src/run_tool_create.rs @@ -3,7 +3,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use async_trait::async_trait; use fabro_manifest::{ - CollectedWorkflowClosure, ResolvedLocalWorkflowPackage, collect_workflow_versions, + CollectedWorkflowClosure, ResolvedLocalWorkflowPackage, collect_inline_workflow_versions, observe_git_run_target, resolve_local_workflow_package, }; use fabro_tool::{ @@ -11,7 +11,6 @@ use fabro_tool::{ }; use fabro_types::settings::run::EnvironmentProvider; use fabro_types::{DirtyStatus, RunTarget}; -use tokio::io::AsyncWriteExt; use tokio::{fs, task}; #[derive(Clone, Debug)] @@ -242,53 +241,15 @@ struct ResolvedTarget { warnings: Vec, } -/// Collect an inline workflow by staging its bytes in a private temporary -/// root; the collected closure owns every file, so the root is discarded on -/// return. +/// Collect an inline workflow from its supplied bytes. The entrypoint is an +/// exact key of the file map, never a checkout selector. async fn collect_inline_workflow( source: &fabro_tool::InlineWorkflowSource, ) -> Result { - let root = tempfile::tempdir().context("failed to create private inline workflow root")?; - for (path, content) in &source.files { - let destination = root.path().join(path.as_str()); - if let Some(parent) = destination.parent() { - fs::create_dir_all(parent).await.with_context(|| { - format!( - "failed to create inline workflow directory {}", - parent.display() - ) - })?; - } - let mut file = fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&destination) - .await - .with_context(|| { - format!( - "failed to create inline workflow file {}", - destination.display() - ) - })?; - file.write_all(content.as_bytes()).await.with_context(|| { - format!( - "failed to write inline workflow file {}", - destination.display() - ) - })?; - // Dropping a tokio File does not wait for queued writes; the - // collector below reads these files synchronously, so flush first. - file.flush().await.with_context(|| { - format!( - "failed to flush inline workflow file {}", - destination.display() - ) - })?; - } let entrypoint = source.entrypoint.clone(); + let files = source.files.clone(); task::spawn_blocking(move || { - collect_workflow_versions(Path::new(entrypoint.as_str()), root.path()) - .map_err(anyhow::Error::new) + collect_inline_workflow_versions(&entrypoint, &files).map_err(anyhow::Error::new) }) .await .context("inline workflow package collection task failed")? @@ -430,6 +391,36 @@ mod tests { ); } + #[tokio::test] + async fn workflow_version_inline_entrypoint_is_exact_even_without_an_extension() { + let server = MockServer::start_async().await; + let registered = Arc::new(Mutex::new(Vec::new())); + let registration = + dynamic_version_registration_mock(&server, Arc::clone(®istered)).await; + let client = no_proxy_client(&server.url("")); + let spec = validated_spec(&json!({ + "workflow": { + "kind": "inline", + "entrypoint": "review", + "files": { + "review": "digraph Review { start [shape=Mdiamond] exit [shape=Msquare] start -> exit }" + } + }, + "target": { "kind": "none" } + })); + let adapter = ServerRunCreateAdapter::worker(EnvironmentProvider::Docker, None, None); + + let prepared = adapter + .prepare(&client, &spec, Path::new("/host/that-must-not-be-read")) + .await + .expect("an extensionless inline entrypoint names a supplied file, not a selector"); + + registration.assert_calls_async(1).await; + let registered = registered.lock().unwrap(); + assert_eq!(prepared.workflow_version_id, registered[0].id().unwrap()); + assert_eq!(registered[0].entrypoint().as_str(), "review"); + } + #[tokio::test] async fn workflow_version_stored_create_skips_registration_and_inherits_exact_worker_target() { let client = no_proxy_client("http://127.0.0.1:9"); diff --git a/lib/components/fabro-manifest/src/lib.rs b/lib/components/fabro-manifest/src/lib.rs index 1441d36c9..9f78aa015 100644 --- a/lib/components/fabro-manifest/src/lib.rs +++ b/lib/components/fabro-manifest/src/lib.rs @@ -44,6 +44,7 @@ use crate::workflow_bundler::WorkflowBundler; pub use crate::workflow_version_collector::{ MAX_WORKFLOW_VERSION_DEPTH, WorkflowVersionCollectError, collect_workflow_versions, collect_workflow_versions_at_location, + collect_inline_workflow_versions, }; pub use crate::workflow_version_packager::SuppliedWorkflowVersionPackager; diff --git a/lib/components/fabro-manifest/src/workflow_version_collector.rs b/lib/components/fabro-manifest/src/workflow_version_collector.rs index 131f9990c..a32ef9bd9 100644 --- a/lib/components/fabro-manifest/src/workflow_version_collector.rs +++ b/lib/components/fabro-manifest/src/workflow_version_collector.rs @@ -186,6 +186,79 @@ pub fn collect_workflow_versions_at_location( VersionAssembler::new(collected).assemble() } +/// Package a workflow whose bytes arrive by value. `entrypoint` is an exact +/// key of `files`; unlike a checkout selector, an extensionless entrypoint +/// is never rewritten to a `.fabro/workflows//workflow.toml` lookup. +/// The files are staged in a private temporary root only for the duration of +/// collection, so the resulting closure's paths are rooted at the file map. +/// +/// # Errors +/// +/// Returns a collision error before touching the filesystem when two paths +/// cannot coexist on one filesystem; staging and collection failures are +/// reported against the entrypoint. +pub fn collect_inline_workflow_versions( + entrypoint: &WorkflowPath, + files: &BTreeMap, +) -> Result { + if !files.contains_key(entrypoint) { + return Err(WorkflowVersionCollectError::MissingWorkflow { + path: entrypoint.to_string(), + }); + } + fabro_types::validate_workflow_path_collisions(files.keys()).map_err(|error| match error { + WorkflowVersionShapeError::PathCollision { second, .. } => { + WorkflowVersionCollectError::PathCollision { + entrypoint: entrypoint.clone(), + path: second, + } + } + source => WorkflowVersionCollectError::InvalidShape { + entrypoint: entrypoint.clone(), + source, + }, + })?; + let collect_error = |source: anyhow::Error| WorkflowVersionCollectError::Collect { + path: PathBuf::from(entrypoint.as_str()), + source, + }; + let root = tempfile::tempdir().map_err(|source| { + collect_error( + anyhow::Error::new(source).context("failed to create private inline workflow root"), + ) + })?; + for (path, content) in files { + let destination = root.path().join(path.as_str()); + if let Some(parent) = destination.parent() { + std::fs::create_dir_all(parent).map_err(|source| { + collect_error(anyhow::Error::new(source).context(format!( + "failed to create inline workflow directory for `{path}`" + ))) + })?; + } + std::fs::write(&destination, content).map_err(|source| { + collect_error( + anyhow::Error::new(source) + .context(format!("failed to write inline workflow file `{path}`")), + ) + })?; + } + let package_root = root.path().canonicalize().map_err(|source| { + collect_error( + anyhow::Error::new(source).context("failed to canonicalize the inline workflow root"), + ) + })?; + let location = WorkflowLocation::from_exact_path(package_root.join(entrypoint.as_str())) + .map_err(|source| collect_error(source.into()))?; + let location = canonicalize_location(location, |path, source| { + collect_error(anyhow::Error::new(source).context(format!( + "failed to canonicalize inline workflow path {}", + path.display() + ))) + })?; + collect_workflow_versions_at_location(&location, &package_root, Path::new(entrypoint.as_str())) +} + fn repository_workflow_path(workflow: &Path) -> PathBuf { if workflow.is_relative() && workflow.extension().is_none() { Path::new(".fabro/workflows") @@ -425,6 +498,36 @@ dockerfile = { path = "Dockerfile" } ); } + #[test] + fn inline_collection_uses_the_exact_entrypoint_and_rejects_collisions_before_staging() { + let files = BTreeMap::from([ + ( + WorkflowPath::new("review").unwrap(), + "digraph Review {}".to_string(), + ), + ( + WorkflowPath::new("notes/detail.md").unwrap(), + "detail".to_string(), + ), + ]); + let closure = + collect_inline_workflow_versions(&WorkflowPath::new("review").unwrap(), &files) + .unwrap(); + let (_, root) = closure.versions().next().unwrap(); + assert_eq!(root.version().entrypoint().as_str(), "review"); + + let colliding = BTreeMap::from([ + (WorkflowPath::new("a").unwrap(), "digraph A {}".to_string()), + (WorkflowPath::new("a/b.md").unwrap(), "b".to_string()), + ]); + let error = collect_inline_workflow_versions(&WorkflowPath::new("a").unwrap(), &colliding) + .unwrap_err(); + assert!( + matches!(error, WorkflowVersionCollectError::PathCollision { .. }), + "unexpected error: {error:#}" + ); + } + #[test] fn packages_named_workflow_without_project_config() { let temp = tempfile::tempdir().unwrap(); diff --git a/lib/foundation/fabro-types/src/lib.rs b/lib/foundation/fabro-types/src/lib.rs index 2259c3814..3783f0bc3 100644 --- a/lib/foundation/fabro-types/src/lib.rs +++ b/lib/foundation/fabro-types/src/lib.rs @@ -198,6 +198,6 @@ 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, - validate_workflow_files, validate_workflow_source_paths, + validate_workflow_files, validate_workflow_source_paths, validate_workflow_path_collisions, }; 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 58fb8e0ad..b4532c470 100644 --- a/lib/foundation/fabro-types/src/workflow_version.rs +++ b/lib/foundation/fabro-types/src/workflow_version.rs @@ -194,6 +194,13 @@ pub fn validate_workflow_source_paths<'a>( }) } +/// Reject exact duplicate paths and file/directory ancestor collisions. +pub fn validate_workflow_path_collisions<'a>( + paths: impl IntoIterator, +) -> Result<(), WorkflowVersionShapeError> { + validate_path_collisions(paths, Cow::Borrowed) +} + /// Detect colliding paths under a comparison key: identical keys, or a key /// that names an ancestor directory of another. fn validate_path_collisions<'a>(