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/<name>/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 <noreply@anthropic.com>

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<String>,
 }
 
-/// 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<CollectedWorkflowClosure> {
-    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(&registered)).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");

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;
 

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/<name>/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<WorkflowPath, String>,
+) -> Result<CollectedWorkflowClosure, WorkflowVersionCollectError> {
+    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();

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};

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<Item = &'a WorkflowPath>,
+) -> 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>(