diff --git a/Cargo.lock b/Cargo.lock
index 685fc36f7..0c066819c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2827,11 +2827,13 @@ dependencies = [
  "fabro-test",
  "fabro-types",
  "fabro-workflow",
+ "fabro-workflow-version",
  "git2",
  "insta",
  "serde_json",
  "temp-env",
  "tempfile",
+ "thiserror 2.0.18",
  "toml 0.8.23",
 ]
 
diff --git a/lib/apps/fabro-server/src/automation_materializer.rs b/lib/apps/fabro-server/src/automation_materializer.rs
index 070205ade..ceddfaaf1 100644
--- a/lib/apps/fabro-server/src/automation_materializer.rs
+++ b/lib/apps/fabro-server/src/automation_materializer.rs
@@ -2,11 +2,13 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use async_trait::async_trait;
-use fabro_api::types::RunManifest;
 use fabro_automation::AutomationId;
-use fabro_config::{EnvironmentLayer, MergeMap};
-use fabro_manifest::ManifestBuildInput;
-use fabro_types::{GitHubRepositorySlug, GitRunTarget, RunId, RunTarget, TargetValidationError};
+use fabro_manifest::WorkflowVersionCollectError;
+use fabro_types::{
+    GitHubRepositorySlug, GitRunTarget, RunId, RunIntent, RunIntentArgs, RunTarget,
+    TargetValidationError, WorkflowVersionId,
+};
+use fabro_workflow_version::{WorkflowVersionStore, WorkflowVersionStoreError};
 use tokio::{fs, task};
 
 use crate::git_checkout::{
@@ -15,19 +17,33 @@ use crate::git_checkout::{
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub(crate) struct AutomationRunMaterializeInput {
-    pub automation_id:      AutomationId,
-    pub target:             GitRunTarget,
-    pub workflow:           String,
-    pub run_id:             RunId,
-    pub user_settings_path: PathBuf,
-    pub temp_root:          PathBuf,
+    pub automation_id: AutomationId,
+    pub target:        GitRunTarget,
+    pub workflow:      String,
+    pub run_id:        RunId,
+    pub temp_root:     PathBuf,
 }
 
 #[derive(Debug, Clone)]
 pub(crate) struct AutomationRunMaterialized {
-    pub manifest:                 RunManifest,
-    pub submitted_manifest_bytes: Vec<u8>,
-    pub target:                   GitRunTarget,
+    pub workflow_version_id: WorkflowVersionId,
+    pub target:              GitRunTarget,
+}
+
+impl AutomationRunMaterialized {
+    /// The admission request for an automation run: the packaged workflow
+    /// version at the exact checked-out target, with no caller overrides.
+    pub(crate) fn into_run_intent(self) -> RunIntent {
+        RunIntent {
+            workflow_version_id: self.workflow_version_id,
+            target:              RunTarget::Git(self.target),
+            args:                RunIntentArgs::default(),
+            environment_id:      None,
+            parent_id:           None,
+            title:               None,
+            goal:                None,
+        }
+    }
 }
 
 #[derive(thiserror::Error, Debug)]
@@ -48,25 +64,25 @@ pub(crate) enum RunMaterializeError {
         #[source]
         source: std::io::Error,
     },
-    #[error("failed to resolve automation workflow")]
+    #[error("automation workflow was not found")]
     WorkflowNotFound {
         #[source]
-        source: anyhow::Error,
+        source: WorkflowVersionCollectError,
     },
-    #[error("failed to build run manifest")]
-    Manifest {
+    #[error("failed to package automation workflow versions")]
+    Package {
         #[source]
-        source: anyhow::Error,
+        source: WorkflowVersionCollectError,
     },
-    #[error("manifest build task failed")]
-    ManifestTask {
+    #[error("workflow-version packaging task failed")]
+    PackageTask {
         #[source]
         source: task::JoinError,
     },
-    #[error("failed to serialize materialized run manifest")]
-    SerializeManifest {
+    #[error("failed to store automation workflow versions")]
+    VersionStore {
         #[source]
-        source: serde_json::Error,
+        source: WorkflowVersionStoreError,
     },
     #[error("failed to load GitHub credentials")]
     Credentials {
@@ -85,11 +101,11 @@ pub(crate) trait AutomationRunMaterializer: Send + Sync {
 
 #[derive(Clone)]
 pub(crate) struct ProductionAutomationRunMaterializer {
-    github_credentials:   Option<fabro_github::GitHubCredentials>,
-    github_api_base_url:  String,
-    http_client:          Option<fabro_http::HttpClient>,
-    environment_defaults: MergeMap<EnvironmentLayer>,
-    repo_cache:           Arc<GitRepoCache>,
+    github_credentials:  Option<fabro_github::GitHubCredentials>,
+    github_api_base_url: String,
+    http_client:         Option<fabro_http::HttpClient>,
+    repo_cache:          Arc<GitRepoCache>,
+    version_store:       WorkflowVersionStore,
 }
 
 impl ProductionAutomationRunMaterializer {
@@ -97,15 +113,15 @@ impl ProductionAutomationRunMaterializer {
         github_credentials: Option<fabro_github::GitHubCredentials>,
         github_api_base_url: String,
         http_client: Option<fabro_http::HttpClient>,
-        environment_defaults: MergeMap<EnvironmentLayer>,
         repo_cache: Arc<GitRepoCache>,
+        version_store: WorkflowVersionStore,
     ) -> Self {
         Self {
             github_credentials,
             github_api_base_url,
             http_client,
-            environment_defaults,
             repo_cache,
+            version_store,
         }
     }
 }
@@ -161,101 +177,78 @@ impl AutomationRunMaterializer for ProductionAutomationRunMaterializer {
         let mut exact_target = input.target;
         exact_target.sha = Some(checked_out_sha);
 
-        let manifest_input = ManifestFromCheckoutInput {
-            workflow: input.workflow,
-            user_settings_path: input.user_settings_path,
-            checkout_dir,
-            target: exact_target,
-            environment_defaults: self.environment_defaults.clone(),
-        };
-        task::spawn_blocking(move || build_manifest_from_checkout(manifest_input))
-            .await
-            .map_err(|source| RunMaterializeError::ManifestTask { source })?
+        let workflow = PathBuf::from(input.workflow);
+        let closure = task::spawn_blocking(move || {
+            fabro_manifest::collect_workflow_versions(&workflow, &checkout_dir)
+                .map_err(package_error)
+        })
+        .await
+        .map_err(|source| RunMaterializeError::PackageTask { source })??;
+
+        // Versions arrive dependency-first, and the store derives the same
+        // content hash the collector did, so the root ID is known up front.
+        for (expected, version) in closure.versions() {
+            let stored = self
+                .version_store
+                .put(version)
+                .await
+                .map_err(|source| RunMaterializeError::VersionStore { source })?;
+            debug_assert_eq!(
+                stored, expected,
+                "store and collector disagree on version ID"
+            );
+        }
+        Ok(AutomationRunMaterialized {
+            workflow_version_id: closure.root_id(),
+            target:              exact_target,
+        })
     }
 }
 
-#[derive(Debug)]
-pub(crate) struct ManifestFromCheckoutInput {
-    workflow:             String,
-    user_settings_path:   PathBuf,
-    checkout_dir:         PathBuf,
-    target:               GitRunTarget,
-    environment_defaults: MergeMap<EnvironmentLayer>,
-}
-
-fn build_manifest_from_checkout(
-    args: ManifestFromCheckoutInput,
-) -> Result<AutomationRunMaterialized, RunMaterializeError> {
-    let ManifestFromCheckoutInput {
-        workflow,
-        user_settings_path,
-        checkout_dir,
-        target,
-        environment_defaults,
-    } = args;
-    // Re-validating the exact target (now carrying the checked-out SHA) yields
-    // the same `GitContext` projection the run-intent path uses.
-    let validated = RunTarget::Git(target)
-        .validate()
-        .map_err(|source| RunMaterializeError::InvalidTarget { source })?;
-    let RunTarget::Git(target) = validated.target else {
-        unreachable!("validating a Git target yields a Git target");
-    };
-    let built = fabro_manifest::build_run_manifest(ManifestBuildInput {
-        workflow: workflow.into(),
-        cwd: checkout_dir,
-        user_settings_path: Some(user_settings_path),
-        environment_defaults,
-        ..ManifestBuildInput::default()
-    })
-    .map_err(manifest_build_error)?;
-
-    let mut manifest = built.manifest;
-    manifest.git = validated.git;
-    let submitted_manifest_bytes = serde_json::to_vec(&manifest)
-        .map_err(|source| RunMaterializeError::SerializeManifest { source })?;
-    Ok(AutomationRunMaterialized {
-        manifest,
-        submitted_manifest_bytes,
-        target,
-    })
-}
-
-fn manifest_build_error(error: anyhow::Error) -> RunMaterializeError {
-    if error.chain().any(|source| {
-        source
-            .downcast_ref::<fabro_config::Error>()
-            .is_some_and(|err| matches!(err, fabro_config::Error::WorkflowNotFound(_)))
-    }) {
+fn package_error(error: WorkflowVersionCollectError) -> RunMaterializeError {
+    if matches!(&error, WorkflowVersionCollectError::WorkflowNotFound { .. }) {
         RunMaterializeError::WorkflowNotFound { source: error }
     } else {
-        RunMaterializeError::Manifest { source: error }
+        RunMaterializeError::Package { source: error }
     }
 }
 
 #[cfg(any(test, feature = "test-support"))]
 #[derive(Clone)]
 pub struct TestAutomationRunMaterializer {
-    inner: std::sync::Arc<std::sync::Mutex<TestAutomationRunMaterializerState>>,
+    inner:         std::sync::Arc<std::sync::Mutex<TestAutomationRunMaterializerState>>,
+    version_store: Option<WorkflowVersionStore>,
 }
 
 #[cfg(any(test, feature = "test-support"))]
 struct TestAutomationRunMaterializerState {
     captured_inputs: Vec<AutomationRunMaterializeInput>,
-    response:        Result<Box<AutomationRunMaterialized>, TargetValidationError>,
+    response:        Result<Box<TestMaterializedWorkflow>, TargetValidationError>,
+}
+
+#[cfg(any(test, feature = "test-support"))]
+#[derive(Clone)]
+struct TestMaterializedWorkflow {
+    version: fabro_workflow_version::ValidatedWorkflowVersion,
+    target:  GitRunTarget,
+    store:   bool,
 }
 
 #[cfg(any(test, feature = "test-support"))]
 impl TestAutomationRunMaterializer {
-    pub fn succeed(
-        manifest: RunManifest,
-        submitted_manifest_bytes: Vec<u8>,
-        target: GitRunTarget,
-    ) -> Self {
-        Self::new(Ok(Box::new(AutomationRunMaterialized {
-            manifest,
-            submitted_manifest_bytes,
+    pub fn succeed(target: GitRunTarget) -> Self {
+        Self::new(Ok(Box::new(TestMaterializedWorkflow {
+            version: test_workflow_version(),
             target,
+            store: true,
+        })))
+    }
+
+    pub fn return_unstored_version(target: GitRunTarget) -> Self {
+        Self::new(Ok(Box::new(TestMaterializedWorkflow {
+            version: test_workflow_version(),
+            target,
+            store: false,
         })))
     }
 
@@ -263,12 +256,15 @@ impl TestAutomationRunMaterializer {
         Self::new(Err(TargetValidationError::Repository))
     }
 
-    fn new(response: Result<Box<AutomationRunMaterialized>, TargetValidationError>) -> Self {
+    fn new(response: Result<Box<TestMaterializedWorkflow>, TargetValidationError>) -> Self {
         Self {
-            inner: std::sync::Arc::new(std::sync::Mutex::new(TestAutomationRunMaterializerState {
-                captured_inputs: Vec::new(),
-                response,
-            })),
+            inner:         std::sync::Arc::new(std::sync::Mutex::new(
+                TestAutomationRunMaterializerState {
+                    captured_inputs: Vec::new(),
+                    response,
+                },
+            )),
+            version_store: None,
         }
     }
 
@@ -280,11 +276,35 @@ impl TestAutomationRunMaterializer {
             .clone()
     }
 
-    pub(crate) fn into_materializer(self) -> std::sync::Arc<dyn AutomationRunMaterializer> {
+    pub(crate) fn into_materializer(
+        mut self,
+        version_store: WorkflowVersionStore,
+    ) -> std::sync::Arc<dyn AutomationRunMaterializer> {
+        self.version_store = Some(version_store);
         std::sync::Arc::new(self)
     }
 }
 
+#[cfg(any(test, feature = "test-support"))]
+fn test_workflow_version() -> fabro_workflow_version::ValidatedWorkflowVersion {
+    use std::collections::BTreeMap;
+
+    let entrypoint = fabro_types::WorkflowPath::new("workflow.fabro")
+        .expect("test workflow entrypoint should be valid");
+    let version = fabro_types::WorkflowVersion::new(
+        entrypoint.clone(),
+        BTreeMap::from([(
+            entrypoint,
+            "digraph Test { graph [goal=\"Test\"] start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"
+                .to_string(),
+        )]),
+        BTreeMap::new(),
+    )
+    .expect("test workflow version should have a valid shape");
+    fabro_workflow_version::ValidatedWorkflowVersion::new(version)
+        .expect("test workflow version should validate")
+}
+
 #[cfg(any(test, feature = "test-support"))]
 #[async_trait]
 impl AutomationRunMaterializer for TestAutomationRunMaterializer {
@@ -292,16 +312,36 @@ impl AutomationRunMaterializer for TestAutomationRunMaterializer {
         &self,
         input: AutomationRunMaterializeInput,
     ) -> Result<AutomationRunMaterialized, RunMaterializeError> {
-        let mut guard = self
-            .inner
-            .lock()
-            .expect("test automation materializer lock poisoned");
-        guard.captured_inputs.push(input);
-        guard
-            .response
-            .clone()
-            .map(|materialized| *materialized)
-            .map_err(|source| RunMaterializeError::InvalidTarget { source })
+        let response = {
+            let mut guard = self
+                .inner
+                .lock()
+                .expect("test automation materializer lock poisoned");
+            guard.captured_inputs.push(input);
+            guard.response.clone()
+        };
+        let materialized =
+            *response.map_err(|source| RunMaterializeError::InvalidTarget { source })?;
+        let store = self
+            .version_store
+            .as_ref()
+            .expect("test materializer must be attached to a version store");
+        let workflow_version_id = if materialized.store {
+            store
+                .put(&materialized.version)
+                .await
+                .map_err(|source| RunMaterializeError::VersionStore { source })?
+        } else {
+            materialized
+                .version
+                .version()
+                .id()
+                .expect("validated test workflow version should serialize canonically")
+        };
+        Ok(AutomationRunMaterialized {
+            workflow_version_id,
+            target: materialized.target,
+        })
     }
 }
 
@@ -312,86 +352,57 @@ mod tests {
         reason = "Materializer unit tests write small temporary workflow fixtures synchronously."
     )]
 
-    use std::collections::HashMap;
     use std::fs;
+    use std::path::Path;
+    use std::time::Duration;
 
-    use fabro_types::DirtyStatus;
+    use object_store::memory::InMemory;
     use tempfile::TempDir;
 
     use super::*;
 
-    fn test_environment_defaults() -> MergeMap<EnvironmentLayer> {
-        MergeMap::from(HashMap::from([("default".to_string(), EnvironmentLayer {
-            provider: Some("local".to_string()),
-            ..EnvironmentLayer::default()
-        })]))
-    }
-
-    #[test]
-    fn manifest_builder_uses_checkout_for_workflow_and_separate_git_context() {
+    #[tokio::test]
+    async fn collected_closure_stores_dependency_first_and_idempotently() {
         let temp = TempDir::new().unwrap();
         let checkout = temp.path().join("checkout");
-        let workflow_dir = checkout.join(".fabro/workflows/demo");
+        let workflow_dir = checkout.join(".fabro/workflows/root");
         fs::create_dir_all(&workflow_dir).unwrap();
         fs::write(checkout.join(".fabro/project.toml"), "_version = 1\n").unwrap();
-        fs::write(
-            workflow_dir.join("workflow.fabro"),
-            r#"digraph Demo { graph [goal="Ship automation"] start [shape=Mdiamond] exit [shape=Msquare] start -> exit }"#,
-        )
-        .unwrap();
         fs::write(
             workflow_dir.join("workflow.toml"),
             "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
         )
         .unwrap();
-        let user_settings_path = temp.path().join("settings.toml");
-        fs::write(&user_settings_path, "_version = 1\n").unwrap();
-        let sha = "0123456789abcdef0123456789abcdef01234567".to_string();
+        fs::write(
+            workflow_dir.join("workflow.fabro"),
+            r#"digraph Root { child [stack.child_workflow="../child/workflow.fabro"] }"#,
+        )
+        .unwrap();
+        let child_dir = checkout.join(".fabro/workflows/child");
+        fs::create_dir_all(&child_dir).unwrap();
+        fs::write(child_dir.join("workflow.fabro"), "digraph Child {}").unwrap();
 
-        let materialized = build_manifest_from_checkout(ManifestFromCheckoutInput {
-            workflow:             "demo".to_string(),
-            user_settings_path:   user_settings_path.clone(),
-            checkout_dir:         checkout.clone(),
-            target:               GitRunTarget {
-                repo:   "workspace-org/app".to_string(),
-                branch: "release".to_string(),
-                tag:    Some("v1".to_string()),
-                sha:    Some(sha.clone()),
-            },
-            environment_defaults: test_environment_defaults(),
-        })
-        .expect("manifest should build from checkout");
+        let closure =
+            fabro_manifest::collect_workflow_versions(Path::new("root"), &checkout).unwrap();
+        let database = fabro_store::test_support::test_database(
+            Arc::new(InMemory::new()),
+            "",
+            Duration::from_millis(1),
+            None,
+        );
+        let store = WorkflowVersionStore::new(database.blobs());
 
-        assert_eq!(materialized.manifest.cwd, checkout.display().to_string());
-        assert_eq!(
-            materialized.manifest.target.path,
-            ".fabro/workflows/demo/workflow.fabro"
-        );
-        assert!(
-            materialized
-                .manifest
-                .configs
-                .iter()
-                .any(|config| config.path.as_deref() == Some(user_settings_path.to_str().unwrap()))
-        );
-        let git = materialized
-            .manifest
-            .git
-            .as_ref()
-            .expect("git context should be set");
-        assert_eq!(git.origin_url, "https://github.com/workspace-org/app");
-        assert_eq!(git.branch, "release");
-        assert_eq!(git.sha.as_deref(), Some(sha.as_str()));
-        assert_eq!(git.dirty, DirtyStatus::Clean);
-        assert_eq!(materialized.target.tag.as_deref(), Some("v1"));
-        assert_eq!(materialized.target.sha.as_deref(), Some(sha.as_str()));
-        let submitted_manifest: serde_json::Value =
-            serde_json::from_slice(&materialized.submitted_manifest_bytes)
-                .expect("submitted bytes should be a manifest");
-        assert!(submitted_manifest.get("run_id").is_none());
-        assert_eq!(
-            submitted_manifest,
-            serde_json::to_value(&materialized.manifest).unwrap()
-        );
+        for _ in 0..2 {
+            for (expected, version) in closure.versions() {
+                assert_eq!(store.put(version).await.unwrap(), expected);
+            }
+        }
+        let loaded = store
+            .get_closure(&closure.root_id())
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(loaded.root_id(), closure.root_id());
+        assert_eq!(loaded.versions().count(), 2);
     }
 }
diff --git a/lib/apps/fabro-server/src/run_intent.rs b/lib/apps/fabro-server/src/run_intent.rs
index 99453a6d9..425392159 100644
--- a/lib/apps/fabro-server/src/run_intent.rs
+++ b/lib/apps/fabro-server/src/run_intent.rs
@@ -303,8 +303,7 @@ fn mount_version(
         .get(version.entrypoint())
         .cloned()
         .expect("validated workflow versions contain their entrypoint file");
-    let config_local = WorkflowPath::new("workflow.toml")
-        .expect("the static workflow config path should be valid");
+    let config_local = version.config_path();
     let config_path = version.files().get(&config_local).map(|source| {
         rebase_path(
             version.entrypoint(),
@@ -564,10 +563,10 @@ mod tests {
                     "digraph Root { child [stack.child_workflow=\"../deps/run.fabro\"] }",
                 ),
                 (
-                    "workflow.toml",
+                    "flows/workflow.toml",
                     "_version = 1\n[run.goal]\nfile = \"goal.md\"\n",
                 ),
-                ("goal.md", "Ship {{ vars.owner }}"),
+                ("flows/goal.md", "Ship {{ vars.owner }}"),
             ],
             [("deps/run.fabro", child_id)],
         );
diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs
index 9d60b694c..1c67c054e 100644
--- a/lib/apps/fabro-server/src/server.rs
+++ b/lib/apps/fabro-server/src/server.rs
@@ -1210,8 +1210,8 @@ impl AppState {
             credentials,
             self.github_api_base_url.clone(),
             self.http_client.clone(),
-            (*self.stores.environments.catalog_layer()).clone(),
             Arc::clone(&self.automation_repo_cache),
+            fabro_workflow_version::WorkflowVersionStore::new(self.store_ref().blobs()),
         )
         .materialize(input)
         .await
diff --git a/lib/apps/fabro-server/src/server/automation_scheduler.rs b/lib/apps/fabro-server/src/server/automation_scheduler.rs
index 4dafca804..29ecf3a2c 100644
--- a/lib/apps/fabro-server/src/server/automation_scheduler.rs
+++ b/lib/apps/fabro-server/src/server/automation_scheduler.rs
@@ -8,7 +8,7 @@ use croner::errors::CronError;
 use fabro_automation::{
     Automation, AutomationId, AutomationRevision, AutomationTriggerId, parse_schedule_expression,
 };
-use fabro_types::{AutomationRef, Principal, RunId, RunTarget, SystemActorKind};
+use fabro_types::{AutomationRef, Principal, RunId, SystemActorKind};
 use tokio::time::sleep;
 use tracing::{Instrument, error, info, info_span, warn};
 
@@ -242,7 +242,6 @@ async fn fire_scheduled_automation_run(
             target,
             workflow: automation.workflow.clone(),
             run_id,
-            user_settings_path: state.active_config_path().to_path_buf(),
             temp_root: state.automation_temp_root(),
         })
         .await
@@ -258,7 +257,6 @@ async fn fire_scheduled_automation_run(
         }
     };
 
-    let explicit_title_supplied = materialized.manifest.title.is_some();
     let actor = Principal::System {
         system_kind: SystemActorKind::Engine,
     };
@@ -267,19 +265,16 @@ async fn fire_scheduled_automation_run(
         name:       Some(automation.name.clone()),
         trigger_id: Some(trigger_id.to_string()),
     };
-    // `create_run_from_manifest` produces a large future; box it to keep our
+    // RunIntent admission produces a large future; box it to keep our
     // stack frame small (matches handler/automations.rs).
-    let response = Box::pin(handler::runs::create_run_from_manifest(
+    let response = Box::pin(handler::runs::create_run_from_intent(
         Arc::clone(&state),
-        handler::runs::CreateRunFromManifestRequest {
-            manifest: materialized.manifest,
-            submitted_manifest_bytes: materialized.submitted_manifest_bytes,
+        handler::runs::CreateRunFromIntentRequest {
+            intent:          materialized.into_run_intent(),
             explicit_run_id: Some(run_id),
-            explicit_title_supplied,
-            actor: actor.clone(),
-            headers: HeaderMap::new(),
-            automation: Some(automation_ref),
-            target: Some(RunTarget::Git(materialized.target)),
+            actor:           actor.clone(),
+            headers:         HeaderMap::new(),
+            automation:      Some(automation_ref),
         },
     ))
     .await;
@@ -343,12 +338,10 @@ fn run_due_schedules_once<'a>(
 
 #[cfg(test)]
 mod tests {
-    use fabro_api::types::RunManifest;
     use fabro_automation::{AutomationDraft, AutomationTrigger, ScheduleTrigger};
     use fabro_static::EnvVars;
     use fabro_store::ListRunsQuery;
-    use fabro_types::{GitRunTarget, RunStatus};
-    use serde_json::json;
+    use fabro_types::{GitRunTarget, RunStatus, RunTarget};
 
     use super::*;
     use crate::test_support::{TestAppStateBuilder, TestAutomationRunMaterializer};
@@ -412,35 +405,10 @@ mod tests {
             .expect("test automation should be created")
     }
 
-    fn minimal_manifest() -> RunManifest {
-        serde_json::from_value(json!({
-            "version": 1,
-            "cwd": "/tmp",
-            "target": {
-                "path": "workflow.fabro",
-            },
-            "workflows": {
-                "workflow.fabro": {
-                    "source": r#"digraph Test {
-                        graph [goal="Test"]
-                        start [shape=Mdiamond]
-                        exit  [shape=Msquare]
-                        start -> exit
-                    }"#,
-                    "files": {},
-                },
-            },
-        }))
-        .expect("minimal manifest should deserialize")
-    }
-
     fn succeeding_materializer() -> TestAutomationRunMaterializer {
-        let manifest = minimal_manifest();
-        let submitted_manifest_bytes =
-            serde_json::to_vec(&manifest).expect("manifest should serialize");
         let mut exact_target = git_target();
         exact_target.sha = Some("0123456789abcdef0123456789abcdef01234567".to_string());
-        TestAutomationRunMaterializer::succeed(manifest, submitted_manifest_bytes, exact_target)
+        TestAutomationRunMaterializer::succeed(exact_target)
     }
 
     fn test_state_with_materializer(materializer: TestAutomationRunMaterializer) -> Arc<AppState> {
@@ -616,6 +584,28 @@ mod tests {
         assert_eq!(automation_ref.name.as_deref(), Some("Nightly"));
         assert_eq!(automation_ref.trigger_id.as_deref(), Some("schedule"));
         let run_id = runs[0].id;
+        let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
+        let projection = run_store.state().await.unwrap();
+        assert!(projection.spec.workflow_version_id.is_some());
+        assert_eq!(
+            projection.spec.target,
+            Some(RunTarget::Git(fabro_types::GitRunTarget {
+                repo:   "fabro-sh/fabro".to_string(),
+                branch: "main".to_string(),
+                tag:    None,
+                sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
+            }))
+        );
+        assert_eq!(
+            run_store
+                .list_events()
+                .await
+                .unwrap()
+                .iter()
+                .filter(|event| event.event.event_name() == "run.start_requested")
+                .count(),
+            1
+        );
         assert!(matches!(
             state
                 .runs
diff --git a/lib/apps/fabro-server/src/server/handler/automations.rs b/lib/apps/fabro-server/src/server/handler/automations.rs
index 167c0de63..6a0c7f7af 100644
--- a/lib/apps/fabro-server/src/server/handler/automations.rs
+++ b/lib/apps/fabro-server/src/server/handler/automations.rs
@@ -6,7 +6,7 @@ use fabro_automation::{
     Automation, AutomationDraft, AutomationId, AutomationReplace, AutomationStoreError,
 };
 use fabro_store::{RunSummaryListQuery, RunSummaryVisibility};
-use fabro_types::{AutomationRef, RunId, RunTarget};
+use fabro_types::{AutomationRef, RunId};
 use fabro_util::error as error_util;
 use serde::Serialize;
 
@@ -132,7 +132,6 @@ async fn create_automation_run(
             target,
             workflow: automation.workflow.clone(),
             run_id,
-            user_settings_path: state.active_config_path().to_path_buf(),
             temp_root: state.automation_temp_root(),
         })
         .await
@@ -143,24 +142,20 @@ async fn create_automation_run(
             return ApiError::new(StatusCode::UNPROCESSABLE_ENTITY, message).into_response();
         }
     };
-    let explicit_title_supplied = materialized.manifest.title.is_some();
     let automation_ref = AutomationRef {
         id:         automation.id.to_string(),
         name:       Some(automation.name.clone()),
         trigger_id: Some(api_trigger_id),
     };
 
-    let response = Box::pin(runs::create_run_from_manifest(
+    let response = Box::pin(runs::create_run_from_intent(
         Arc::clone(&state),
-        runs::CreateRunFromManifestRequest {
-            manifest: materialized.manifest,
-            submitted_manifest_bytes: materialized.submitted_manifest_bytes,
+        runs::CreateRunFromIntentRequest {
+            intent: materialized.into_run_intent(),
             explicit_run_id: Some(run_id),
-            explicit_title_supplied,
             actor: actor.clone(),
             headers,
             automation: Some(automation_ref),
-            target: Some(RunTarget::Git(materialized.target)),
         },
     ))
     .await;
diff --git a/lib/apps/fabro-server/src/server/handler/runs.rs b/lib/apps/fabro-server/src/server/handler/runs.rs
index f9bf551a4..07efbada8 100644
--- a/lib/apps/fabro-server/src/server/handler/runs.rs
+++ b/lib/apps/fabro-server/src/server/handler/runs.rs
@@ -541,7 +541,14 @@ async fn create_run(
     // round-trip would silently collapse duplicate keys to last-key-wins.
     let intent_error = match serde_json::from_slice::<RunIntent>(&body) {
         Ok(intent) => {
-            return Box::pin(create_run_from_intent(state, intent, actor, headers)).await;
+            return Box::pin(create_run_from_intent(state, CreateRunFromIntentRequest {
+                intent,
+                explicit_run_id: None,
+                actor,
+                headers,
+                automation: None,
+            }))
+            .await;
         }
         Err(err) => err,
     };
@@ -604,12 +611,28 @@ fn create_run_parse_error(
     ApiError::bad_request(manifest_error.to_string()).into_response()
 }
 
-async fn create_run_from_intent(
+pub(crate) struct CreateRunFromIntentRequest {
+    pub(crate) intent:          RunIntent,
+    /// Run ID preallocated by server-side automation code, never supplied by
+    /// an HTTP create body.
+    pub(crate) explicit_run_id: Option<RunId>,
+    pub(crate) actor:           Principal,
+    pub(crate) headers:         HeaderMap,
+    pub(crate) automation:      Option<AutomationRef>,
+}
+
+pub(crate) async fn create_run_from_intent(
     state: Arc<AppState>,
-    intent: RunIntent,
-    actor: Principal,
-    headers: HeaderMap,
+    request: CreateRunFromIntentRequest,
 ) -> Response {
+    let CreateRunFromIntentRequest {
+        intent,
+        explicit_run_id,
+        actor,
+        headers,
+        automation,
+    } = request;
+    let explicit_title_supplied = intent.title.is_some();
     // Validate the pure, in-memory request facts before paying for
     // blob-store reads and closure lowering.
     let ValidatedRunTarget { target, git } = match intent.target.validate() {
@@ -713,7 +736,7 @@ async fn create_run_from_intent(
         cli_overrides: None,
         input_overrides,
         inline_goal_override: intent.goal,
-        run_id: None,
+        run_id: explicit_run_id,
         title,
         parent_id: intent.parent_id,
         // Target identity and its Git projection are attached after provider
@@ -726,7 +749,7 @@ async fn create_run_from_intent(
         provenance: run_provenance(&headers, &actor),
         web_url: None,
         submitted_manifest_bytes: None,
-        automation: None,
+        automation,
     };
     let normalized = match run_compiler::normalize_source(raw_compiler_input) {
         Ok(normalized) => normalized,
@@ -764,7 +787,7 @@ async fn create_run_from_intent(
     finalize_created_run(
         state,
         prepared,
-        intent.title.is_some(),
+        explicit_title_supplied,
         entrypoint,
         CreatedRunErrorStyle::Intent,
     )
diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs
index 3a59afe8a..e934a02e9 100644
--- a/lib/apps/fabro-server/src/server/tests.rs
+++ b/lib/apps/fabro-server/src/server/tests.rs
@@ -4502,6 +4502,78 @@ async fn create_run_from_manifest_helper_persists_automation_metadata_and_exact_
     assert_eq!(run_store.state().await.unwrap().spec.target, Some(target));
 }
 
+#[tokio::test]
+async fn create_run_from_intent_helper_persists_automation_version_and_exact_target() {
+    let state = TestAppStateBuilder::new()
+        .env_lookup(|_| None)
+        .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
+        .build();
+    let workflow_version_id = store_workflow_version(&state, MINIMAL_DOT, None).await;
+    let run_id = RunId::new();
+    let automation = fabro_types::AutomationRef {
+        id:         "nightly".to_string(),
+        name:       Some("Nightly".to_string()),
+        trigger_id: Some("schedule".to_string()),
+    };
+    let target = RunTarget::Git(GitRunTarget {
+        repo:   "fabro-sh/fabro".to_string(),
+        branch: "main".to_string(),
+        tag:    Some("v1.2.3".to_string()),
+        sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
+    });
+
+    let response = Box::pin(handler::runs::create_run_from_intent(
+        Arc::clone(&state),
+        handler::runs::CreateRunFromIntentRequest {
+            intent:          fabro_api::types::RunIntent {
+                workflow_version_id,
+                target: target.clone(),
+                args: fabro_api::types::RunIntentArgs::default(),
+                environment_id: None,
+                parent_id: None,
+                title: None,
+                goal: None,
+            },
+            explicit_run_id: Some(run_id),
+            actor:           Principal::System {
+                system_kind: SystemActorKind::Engine,
+            },
+            headers:         HeaderMap::new(),
+            automation:      Some(automation.clone()),
+        },
+    ))
+    .await;
+
+    let body = response_json!(response, StatusCode::CREATED).await;
+    assert_eq!(body["automation"]["id"], automation.id);
+    let summary = state
+        .stores
+        .runs
+        .get_cached_summary(&run_id, Utc::now())
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(summary.automation, Some(automation.clone()));
+    let run_store = state.stores.runs.open_run_reader(&run_id).await.unwrap();
+    let projection = run_store.state().await.unwrap();
+    assert_eq!(
+        projection.spec.workflow_version_id,
+        Some(workflow_version_id)
+    );
+    assert_eq!(projection.spec.target, Some(target));
+    assert_eq!(projection.spec.automation, Some(automation));
+    assert_eq!(
+        run_store
+            .list_events()
+            .await
+            .unwrap()
+            .iter()
+            .map(|event| event.event.event_name())
+            .collect::<Vec<_>>(),
+        ["run.created", "run.submitted"]
+    );
+}
+
 #[tokio::test]
 async fn create_run_from_manifest_pins_compiled_and_persisted_behavior() {
     let state = TestAppStateBuilder::new()
@@ -4854,24 +4926,17 @@ async fn create_run_from_manifest_resolves_generated_id_after_variable_snapshot(
 }
 
 #[tokio::test]
-async fn fake_automation_materializer_injection_captures_input_and_returns_manifest() {
-    let materialized_manifest: RunManifest =
-        serde_json::from_value(minimal_manifest_json(MINIMAL_DOT)).unwrap();
-    let fake = TestAutomationRunMaterializer::succeed(
-        materialized_manifest.clone(),
-        b"{\"fake\":true}".to_vec(),
-        GitRunTarget {
-            repo:   "fabro-sh/fabro".to_string(),
-            branch: "main".to_string(),
-            tag:    None,
-            sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
-        },
-    );
+async fn fake_automation_materializer_injection_captures_input_and_returns_version() {
+    let fake = TestAutomationRunMaterializer::succeed(GitRunTarget {
+        repo:   "fabro-sh/fabro".to_string(),
+        branch: "main".to_string(),
+        tag:    None,
+        sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
+    });
     let state = TestAppStateBuilder::new()
         .automation_materializer(fake.clone())
         .build();
     let run_id = RunId::new();
-    let user_settings_path = PathBuf::from("/tmp/fabro/settings.toml");
     let temp_root = PathBuf::from("/tmp/fabro/automation");
     let target = GitRunTarget {
         repo:   "fabro-sh/fabro".to_string(),
@@ -4886,24 +4951,22 @@ async fn fake_automation_materializer_injection_captures_input_and_returns_manif
             target: target.clone(),
             workflow: "demo".to_string(),
             run_id,
-            user_settings_path: user_settings_path.clone(),
             temp_root: temp_root.clone(),
         })
         .await
         .expect("fake materializer should succeed");
 
-    assert_eq!(
-        serde_json::to_value(&output.manifest).unwrap(),
-        serde_json::to_value(&materialized_manifest).unwrap()
-    );
-    assert_eq!(output.submitted_manifest_bytes, b"{\"fake\":true}".to_vec());
+    let stored = fabro_workflow_version::WorkflowVersionStore::new(state.store_ref().blobs())
+        .get(&output.workflow_version_id)
+        .await
+        .unwrap();
+    assert!(stored.is_some());
     let captured = fake.captured_inputs();
     assert_eq!(captured.len(), 1);
     assert_eq!(captured[0].automation_id.as_str(), "nightly");
     assert_eq!(captured[0].target, target);
     assert_eq!(captured[0].workflow, "demo");
     assert_eq!(captured[0].run_id, run_id);
-    assert_eq!(captured[0].user_settings_path, user_settings_path);
     assert_eq!(captured[0].temp_root, temp_root);
 }
 
diff --git a/lib/apps/fabro-server/src/test_support.rs b/lib/apps/fabro-server/src/test_support.rs
index a7a9f1432..8b0dd2185 100644
--- a/lib/apps/fabro-server/src/test_support.rs
+++ b/lib/apps/fabro-server/src/test_support.rs
@@ -33,7 +33,6 @@ use tokio::runtime::Builder as TokioRuntimeBuilder;
 use tokio_util::sync::CancellationToken;
 use ulid::Ulid;
 
-use crate::automation_materializer::AutomationRunMaterializer;
 pub use crate::automation_materializer::TestAutomationRunMaterializer;
 use crate::interp::process_env_var;
 use crate::jwt_auth::{AuthMode, ConfiguredAuth};
@@ -102,7 +101,7 @@ pub struct TestAppStateBuilder {
     default_environment_provider: Option<EnvironmentProvider>,
     env_lookup:                   EnvLookup,
     llm_catalog_settings:         LlmCatalogSettings,
-    automation_materializer:      Option<Arc<dyn AutomationRunMaterializer>>,
+    automation_materializer:      Option<TestAutomationRunMaterializer>,
     #[cfg(test)]
     worker_runtime:               Option<Arc<dyn WorkerRuntime>>,
 }
@@ -184,7 +183,7 @@ impl TestAppStateBuilder {
     }
 
     pub fn automation_materializer(mut self, materializer: TestAutomationRunMaterializer) -> Self {
-        self.automation_materializer = Some(materializer.into_materializer());
+        self.automation_materializer = Some(materializer);
         self
     }
 
@@ -282,6 +281,11 @@ impl TestAppStateBuilder {
             server::automation_dir_for_active_config(&active_config_path),
         )?;
         let preloaded_vault = test_secret_snapshot(db_pool.clone())?;
+        let automation_materializer_override = self.automation_materializer.map(|materializer| {
+            materializer.into_materializer(fabro_workflow_version::WorkflowVersionStore::new(
+                store.blobs(),
+            ))
+        });
         build_app_state(AppStateConfig {
             resolved_settings: resolved_runtime_settings_for_tests(
                 self.server_settings,
@@ -307,7 +311,7 @@ impl TestAppStateBuilder {
             worker_control_bus: None,
             #[cfg(test)]
             worker_runtime: self.worker_runtime,
-            automation_materializer_override: self.automation_materializer,
+            automation_materializer_override,
         })
     }
 }
diff --git a/lib/apps/fabro-server/tests/it/api/automations.rs b/lib/apps/fabro-server/tests/it/api/automations.rs
index 1c78e0af8..ba38b0bd6 100644
--- a/lib/apps/fabro-server/tests/it/api/automations.rs
+++ b/lib/apps/fabro-server/tests/it/api/automations.rs
@@ -13,10 +13,7 @@ use serde_json::{Value, json};
 use sqlx::Row as _;
 use tower::ServiceExt;
 
-use crate::helpers::{
-    MINIMAL_DOT, api, checked_response, minimal_manifest_json, response_json, response_status,
-    run_json,
-};
+use crate::helpers::{api, checked_response, response_json, response_status, run_json};
 
 fn automation_body(id: &str, name: &str) -> Value {
     json!({
@@ -78,29 +75,26 @@ fn automation_app() -> (axum::Router, tempfile::TempDir, PathBuf) {
 }
 
 fn automation_app_with_fake_materializer() -> (axum::Router, tempfile::TempDir, PathBuf) {
+    automation_app_with_materializer(TestAutomationRunMaterializer::succeed(GitRunTarget {
+        repo:   "fabro-sh/fabro".to_string(),
+        branch: "main".to_string(),
+        tag:    None,
+        sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
+    }))
+}
+
+fn automation_app_with_materializer(
+    materializer: TestAutomationRunMaterializer,
+) -> (axum::Router, tempfile::TempDir, PathBuf) {
     let temp_dir = tempfile::tempdir().expect("automation test tempdir should be created");
     let active_config_path = temp_dir.path().join("settings.toml");
     let vault_path = temp_dir.path().join("secrets.json");
     let sqlite_path = Storage::new(temp_dir.path()).sqlite_path();
-    let materialized_manifest: fabro_api::types::RunManifest =
-        serde_json::from_value(minimal_manifest_json(MINIMAL_DOT))
-            .expect("minimal run manifest fixture should deserialize");
-    let submitted_manifest_bytes =
-        serde_json::to_vec(&materialized_manifest).expect("minimal run manifest should serialize");
     let state = TestAppStateBuilder::new()
         .active_config_path(active_config_path)
         .vault_path(vault_path)
         .vault_entries([(EnvVars::OPENAI_API_KEY, "test-openai-api-key")])
-        .automation_materializer(TestAutomationRunMaterializer::succeed(
-            materialized_manifest,
-            submitted_manifest_bytes,
-            GitRunTarget {
-                repo:   "fabro-sh/fabro".to_string(),
-                branch: "main".to_string(),
-                tag:    None,
-                sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
-            },
-        ))
+        .automation_materializer(materializer)
         .build();
     (build_test_router(state), temp_dir, sqlite_path)
 }
@@ -920,6 +914,25 @@ async fn successful_api_triggered_automation_run_persists_automation_metadata()
     assert_eq!(retrieved["automation"], created["automation"]);
 }
 
+#[tokio::test]
+async fn api_triggered_automation_with_missing_version_does_not_create_or_start_a_run() {
+    let materializer = TestAutomationRunMaterializer::return_unstored_version(GitRunTarget {
+        repo:   "fabro-sh/fabro".to_string(),
+        branch: "main".to_string(),
+        tag:    None,
+        sha:    Some("0123456789abcdef0123456789abcdef01234567".to_string()),
+    });
+    let (app, _temp_dir, _automation_dir) = automation_app_with_materializer(materializer);
+    create_automation(&app, "nightly", "Nightly").await;
+
+    let error = create_automation_run(&app, "nightly", StatusCode::NOT_FOUND).await;
+
+    assert_eq!(error["errors"][0]["code"], "workflow_version_not_found");
+    let runs = list_automation_runs(&app, "/automations/nightly/runs").await;
+    assert_eq!(runs["meta"]["total"], 0);
+    assert_eq!(runs["data"], json!([]));
+}
+
 #[tokio::test]
 async fn automation_run_listing_includes_only_runs_for_that_automation() {
     let (app, _temp_dir, _automation_dir) = automation_app_with_fake_materializer();
diff --git a/lib/components/fabro-manifest/Cargo.toml b/lib/components/fabro-manifest/Cargo.toml
index 36b9d44c1..ab27375c4 100644
--- a/lib/components/fabro-manifest/Cargo.toml
+++ b/lib/components/fabro-manifest/Cargo.toml
@@ -21,7 +21,9 @@ fabro-graphviz = { path = "../fabro-graphviz" }
 fabro-template = { path = "../../foundation/fabro-template" }
 fabro-types = { path = "../../foundation/fabro-types" }
 fabro-workflow = { path = "../fabro-workflow" }
+fabro-workflow-version = { path = "../fabro-workflow-version" }
 git2.workspace = true
+thiserror.workspace = true
 toml.workspace = true
 
 [dev-dependencies]
diff --git a/lib/components/fabro-manifest/src/lib.rs b/lib/components/fabro-manifest/src/lib.rs
index 35891f419..b4612df28 100644
--- a/lib/components/fabro-manifest/src/lib.rs
+++ b/lib/components/fabro-manifest/src/lib.rs
@@ -4,6 +4,7 @@
 )]
 
 mod workflow_bundler;
+mod workflow_version_collector;
 
 use std::collections::HashMap;
 use std::path::{Component, Path, PathBuf};
@@ -29,6 +30,9 @@ use fabro_workflow::git::{
 };
 
 use crate::workflow_bundler::WorkflowBundler;
+pub use crate::workflow_version_collector::{
+    CollectedWorkflowClosure, WorkflowVersionCollectError, collect_workflow_versions,
+};
 
 #[derive(Debug, Default)]
 pub struct ManifestBuildInput {
@@ -126,13 +130,7 @@ pub fn build_sparse_run_overrides(input: RunOverrideInput<'_>) -> Option<RunLaye
 }
 
 pub fn build_run_manifest(input: ManifestBuildInput) -> Result<BuiltManifest> {
-    let root_location = WorkflowLocation::resolve(&input.workflow, &input.cwd)?;
-    if root_location.toml.is_none() && !root_location.graph.is_file() {
-        return Err(fabro_config::Error::WorkflowNotFound(
-            root_location.graph.display().to_string(),
-        )
-        .into());
-    }
+    let root_location = resolve_existing_workflow_location(&input.workflow, &input.cwd)?;
     let project_config = discover_project_config(&root_location.dir)?;
     let project_config_source = project_config
         .as_ref()
@@ -396,6 +394,19 @@ fn push_manifest_branch_best_effort(
     let _ = push_branch_noninteractive(repo_path, "origin", branch);
 }
 
+/// Resolve a workflow reference and reject it when neither its config nor
+/// its graph exists on disk.
+/// A missing workflow surfaces as `fabro_config::Error::WorkflowNotFound`.
+fn resolve_existing_workflow_location(workflow: &Path, cwd: &Path) -> Result<WorkflowLocation> {
+    let location = WorkflowLocation::resolve(workflow, cwd)?;
+    if location.toml.is_none() && !location.graph.is_file() {
+        return Err(
+            fabro_config::Error::WorkflowNotFound(location.graph.display().to_string()).into(),
+        );
+    }
+    Ok(location)
+}
+
 fn normalize_absolute_path(base_dir: &Path, reference: &str) -> Option<PathBuf> {
     let path = Path::new(reference);
     if path.is_absolute() || reference.starts_with('~') {
diff --git a/lib/components/fabro-manifest/src/workflow_bundler.rs b/lib/components/fabro-manifest/src/workflow_bundler.rs
index e1907c37b..bbf7454be 100644
--- a/lib/components/fabro-manifest/src/workflow_bundler.rs
+++ b/lib/components/fabro-manifest/src/workflow_bundler.rs
@@ -1,11 +1,13 @@
-use std::collections::{HashMap, HashSet};
+use std::collections::{BTreeSet, HashMap, HashSet};
 use std::path::{Component, Path, PathBuf};
 use std::sync::Arc;
 
 use anyhow::{Context as _, Result, anyhow};
 use fabro_api::types;
 use fabro_config::project::WorkflowLocation;
-use fabro_config::{EnvironmentDockerfileLayer, EnvironmentImageLayer, SettingsLayer};
+use fabro_config::{
+    EnvironmentDockerfileLayer, EnvironmentImageLayer, RunGoalLayer, SettingsLayer,
+};
 use fabro_graphviz::parser;
 use fabro_template::{
     BundleTemplateStore, FilesystemTemplateStore, GraphPosition, GraphReference,
@@ -18,11 +20,22 @@ use fabro_types::graph::ReferenceKind;
 use crate::{manifest_path_from_absolute, normalize_absolute_path};
 
 pub(super) struct WorkflowBundler<'a> {
-    cwd:               &'a Path,
-    inputs:            &'a HashMap<String, toml::Value>,
-    template_store:    FilesystemTemplateStore,
-    workflows:         HashMap<String, types::ManifestWorkflow>,
+    cwd: &'a Path,
+    inputs: &'a HashMap<String, toml::Value>,
+    template_store: FilesystemTemplateStore,
+    workflows: HashMap<String, CollectedWorkflowSource>,
     visited_workflows: HashSet<String>,
+    workflow_version_projection: bool,
+}
+
+pub(super) struct CollectedWorkflowSources {
+    pub(super) root_key:  String,
+    pub(super) workflows: HashMap<String, CollectedWorkflowSource>,
+}
+
+pub(super) struct CollectedWorkflowSource {
+    pub(super) workflow:        types::ManifestWorkflow,
+    pub(super) dependency_keys: BTreeSet<String>,
 }
 
 impl<'a> WorkflowBundler<'a> {
@@ -33,6 +46,7 @@ impl<'a> WorkflowBundler<'a> {
             template_store: FilesystemTemplateStore::new(cwd),
             workflows: HashMap::new(),
             visited_workflows: HashSet::new(),
+            workflow_version_projection: false,
         }
     }
 
@@ -48,11 +62,29 @@ impl<'a> WorkflowBundler<'a> {
                 .workflows
                 .remove(&root_key)
                 .ok_or_else(|| anyhow!("root workflow missing from manifest bundle"))?;
-            self.collect_config_dockerfile(config_path, source, &mut root.files)?;
+            let entrypoint = ManifestPath::from_wire(&root_key)
+                .ok_or_else(|| anyhow!("invalid root workflow path: {root_key}"))?;
+            self.collect_config_files(config_path, source, &entrypoint, &mut root.workflow.files)?;
             self.workflows.insert(root_key, root);
         }
 
-        Ok(self.workflows)
+        Ok(self
+            .workflows
+            .into_iter()
+            .map(|(key, source)| (key, source.workflow))
+            .collect())
+    }
+
+    pub(super) fn collect_versions(
+        mut self,
+        root: &WorkflowLocation,
+    ) -> Result<CollectedWorkflowSources> {
+        self.workflow_version_projection = true;
+        let root_key = self.collect_workflow_location(root)?;
+        Ok(CollectedWorkflowSources {
+            root_key,
+            workflows: self.workflows,
+        })
     }
 
     /// Collects the workflow at `location` and returns its manifest key.
@@ -77,28 +109,33 @@ impl<'a> WorkflowBundler<'a> {
 
         let scan = WorkflowScanInput {
             absolute_dot_path: location.graph.clone(),
-            dot_path,
-            source: source.clone(),
+            dot_path:          dot_path.clone(),
+            source:            source.clone(),
         };
         let mut files = HashMap::new();
         let mut visited_imports = HashSet::new();
+        let mut dependency_keys = BTreeSet::new();
         if let Some(config) = config.as_ref() {
             let config_path = ManifestPath::from_wire(&config.path)
                 .ok_or_else(|| anyhow!("invalid manifest workflow config path: {}", config.path))?;
-            self.collect_config_dockerfile(&config_path, &config.source, &mut files)?;
+            self.collect_config_files(&config_path, &config.source, &dot_path, &mut files)?;
         }
         self.collect_workflow_files(
             &scan,
             &mut files,
             &mut visited_imports,
+            &mut dependency_keys,
             GraphPosition::Entrypoint,
         )?;
 
         self.workflows
-            .insert(dot_key.clone(), types::ManifestWorkflow {
-                config,
-                files,
-                source,
+            .insert(dot_key.clone(), CollectedWorkflowSource {
+                workflow: types::ManifestWorkflow {
+                    config,
+                    files,
+                    source,
+                },
+                dependency_keys,
             });
 
         Ok(dot_key)
@@ -128,6 +165,7 @@ impl<'a> WorkflowBundler<'a> {
         workflow: &WorkflowScanInput,
         files: &mut HashMap<String, types::ManifestFileEntry>,
         visited_imports: &mut HashSet<String>,
+        dependency_keys: &mut BTreeSet<String>,
         position: GraphPosition,
     ) -> Result<()> {
         let graph = parser::parse(&workflow.source)
@@ -136,7 +174,11 @@ impl<'a> WorkflowBundler<'a> {
             .absolute_dot_path
             .parent()
             .unwrap_or_else(|| Path::new("."));
-        let workflow_template_root = manifest_parent_or_dot(&workflow.dot_path)?;
+        let workflow_template_root = if self.workflow_version_projection {
+            workflow_package_root()
+        } else {
+            manifest_parent_or_dot(&workflow.dot_path)?
+        };
 
         // Imports and child workflows require a mutable borrow of self, so
         // collect them during the walk and recurse after the visitor returns.
@@ -224,12 +266,15 @@ impl<'a> WorkflowBundler<'a> {
                     &imported_scan,
                     files,
                     visited_imports,
+                    dependency_keys,
                     GraphPosition::Imported,
                 )?;
             }
         }
         for child in children {
-            self.collect_workflow_entry(Path::new(child), workflow_base_dir)?;
+            let dependency_key =
+                self.collect_workflow_entry(Path::new(child), workflow_base_dir)?;
+            dependency_keys.insert(dependency_key);
         }
 
         Ok(())
@@ -320,10 +365,11 @@ impl<'a> WorkflowBundler<'a> {
         Ok(())
     }
 
-    fn collect_config_dockerfile(
+    fn collect_config_files(
         &self,
         config_path: &ManifestPath,
         source: &str,
+        entrypoint: &ManifestPath,
         files: &mut HashMap<String, types::ManifestFileEntry>,
     ) -> Result<()> {
         let layer = source
@@ -337,9 +383,49 @@ impl<'a> WorkflowBundler<'a> {
         for image in layer.environment_images() {
             self.collect_environment_dockerfile(files, base_dir, config_path, image)?;
         }
+        if self.workflow_version_projection {
+            self.collect_config_goal_files(files, base_dir, config_path, entrypoint, &layer)?;
+        }
         Ok(())
     }
 
+    fn collect_config_goal_files(
+        &self,
+        files: &mut HashMap<String, types::ManifestFileEntry>,
+        base_dir: &Path,
+        config_path: &ManifestPath,
+        entrypoint: &ManifestPath,
+        layer: &SettingsLayer,
+    ) -> Result<()> {
+        let Some(goal) = layer.run.as_ref().and_then(|run| run.goal.as_ref()) else {
+            return Ok(());
+        };
+        let content = match goal {
+            RunGoalLayer::Inline(goal) => goal.as_source(),
+            RunGoalLayer::File { file } => {
+                let reference = file.as_source();
+                let bundled = self.collect_bundled_file(
+                    files,
+                    base_dir,
+                    &reference,
+                    types::ManifestFileRefType::FileInline,
+                    ReferenceKind::RunGoalFile,
+                    Some(config_path.clone()),
+                )?;
+                files
+                    .get(&bundled.path.to_string())
+                    .expect("collect_bundled_file inserts the goal file it returns")
+                    .content
+                    .clone()
+            }
+        };
+        self.collect_template_include_files(
+            files,
+            TemplateSource::new(entrypoint.clone(), workflow_package_root(), content),
+            Some(config_path),
+        )
+    }
+
     fn collect_environment_dockerfile(
         &self,
         files: &mut HashMap<String, types::ManifestFileEntry>,
@@ -413,6 +499,10 @@ fn manifest_parent_or_dot(path: &ManifestPath) -> Result<ManifestPath> {
         .ok_or_else(|| anyhow!("invalid manifest parent path for {path}: {parent}"))
 }
 
+fn workflow_package_root() -> ManifestPath {
+    ManifestPath::from_wire(".").expect("the workflow package root must be a valid manifest path")
+}
+
 fn template_root_for_bundled_file(
     path: &ManifestPath,
     workflow_template_root: &ManifestPath,
diff --git a/lib/components/fabro-manifest/src/workflow_version_collector.rs b/lib/components/fabro-manifest/src/workflow_version_collector.rs
new file mode 100644
index 000000000..fc6a97cb4
--- /dev/null
+++ b/lib/components/fabro-manifest/src/workflow_version_collector.rs
@@ -0,0 +1,481 @@
+use std::collections::{BTreeMap, HashMap, HashSet};
+use std::path::{Path, PathBuf};
+
+use fabro_api::types;
+use fabro_types::{
+    WorkflowPath, WorkflowPathParseError, WorkflowVersion, WorkflowVersionId,
+    WorkflowVersionShapeError,
+};
+use fabro_workflow_version::{ValidatedWorkflowVersion, WorkflowVersionError};
+use thiserror::Error;
+
+use crate::workflow_bundler::{CollectedWorkflowSource, CollectedWorkflowSources, WorkflowBundler};
+
+/// One locally packaged workflow-version closure in dependency-first order.
+#[derive(Debug)]
+pub struct CollectedWorkflowClosure {
+    root_id:  WorkflowVersionId,
+    versions: Vec<(WorkflowVersionId, ValidatedWorkflowVersion)>,
+}
+
+impl CollectedWorkflowClosure {
+    #[must_use]
+    pub fn root_id(&self) -> WorkflowVersionId {
+        self.root_id
+    }
+
+    /// Iterate over every unique version with dependencies before parents.
+    pub fn versions(
+        &self,
+    ) -> impl Iterator<Item = (WorkflowVersionId, &ValidatedWorkflowVersion)> + '_ {
+        self.versions.iter().map(|(id, version)| (*id, version))
+    }
+}
+
+#[derive(Debug, Error)]
+pub enum WorkflowVersionCollectError {
+    #[error("workflow `{path}` was not found")]
+    WorkflowNotFound { path: PathBuf },
+    #[error("failed to collect workflow `{path}`")]
+    Collect {
+        path:   PathBuf,
+        #[source]
+        source: anyhow::Error,
+    },
+    #[error("collected workflow path `{path}` is invalid")]
+    InvalidPath {
+        path:   String,
+        #[source]
+        source: WorkflowPathParseError,
+    },
+    #[error("collected workflow `{entrypoint}` has conflicting content at `{path}`")]
+    PathCollision {
+        entrypoint: WorkflowPath,
+        path:       WorkflowPath,
+    },
+    #[error("collected workflow `{entrypoint}` has an invalid shape")]
+    InvalidShape {
+        entrypoint: WorkflowPath,
+        #[source]
+        source:     WorkflowVersionShapeError,
+    },
+    #[error("collected workflow `{entrypoint}` is invalid")]
+    InvalidVersion {
+        entrypoint: WorkflowPath,
+        #[source]
+        source:     WorkflowVersionError,
+    },
+    #[error("workflow dependency cycle reaches `{path}`")]
+    DependencyCycle { path: WorkflowPath },
+    #[error("collected workflow dependency `{path}` is missing")]
+    MissingWorkflow { path: String },
+}
+
+/// Package one workflow and every separately runnable dependency from a local
+/// checkout. All paths are rooted at `checkout_root`, so moving the physical
+/// checkout does not change canonical version bytes or IDs.
+pub fn collect_workflow_versions(
+    workflow: &Path,
+    checkout_root: &Path,
+) -> Result<CollectedWorkflowClosure, WorkflowVersionCollectError> {
+    let location =
+        crate::resolve_existing_workflow_location(workflow, checkout_root).map_err(|source| {
+            if matches!(
+                source.downcast_ref::<fabro_config::Error>(),
+                Some(fabro_config::Error::WorkflowNotFound(_))
+            ) {
+                WorkflowVersionCollectError::WorkflowNotFound {
+                    path: workflow.to_path_buf(),
+                }
+            } else {
+                WorkflowVersionCollectError::Collect {
+                    path: workflow.to_path_buf(),
+                    source,
+                }
+            }
+        })?;
+
+    let inputs = HashMap::new();
+    let collected = WorkflowBundler::new(checkout_root, &inputs)
+        .collect_versions(&location)
+        .map_err(|source| WorkflowVersionCollectError::Collect {
+            path: workflow.to_path_buf(),
+            source,
+        })?;
+    VersionAssembler::new(collected).assemble()
+}
+
+struct VersionAssembler {
+    root_key: String,
+    /// Sources still waiting to be assembled; each is removed once visited.
+    pending:  HashMap<String, CollectedWorkflowSource>,
+    visiting: HashSet<String>,
+    ids:      HashMap<String, WorkflowVersionId>,
+    versions: Vec<(WorkflowVersionId, ValidatedWorkflowVersion)>,
+}
+
+impl VersionAssembler {
+    fn new(collected: CollectedWorkflowSources) -> Self {
+        Self {
+            root_key: collected.root_key,
+            pending:  collected.workflows,
+            visiting: HashSet::new(),
+            ids:      HashMap::new(),
+            versions: Vec::new(),
+        }
+    }
+
+    fn assemble(mut self) -> Result<CollectedWorkflowClosure, WorkflowVersionCollectError> {
+        let root_key = std::mem::take(&mut self.root_key);
+        let root_id = self.assemble_one(&root_key)?;
+        Ok(CollectedWorkflowClosure {
+            root_id,
+            versions: self.versions,
+        })
+    }
+
+    fn assemble_one(
+        &mut self,
+        key: &str,
+    ) -> Result<WorkflowVersionId, WorkflowVersionCollectError> {
+        if let Some(id) = self.ids.get(key) {
+            return Ok(*id);
+        }
+        if !self.visiting.insert(key.to_owned()) {
+            return Err(WorkflowVersionCollectError::DependencyCycle {
+                path: workflow_path(key)?,
+            });
+        }
+        let source = self.pending.remove(key).ok_or_else(|| {
+            WorkflowVersionCollectError::MissingWorkflow {
+                path: key.to_owned(),
+            }
+        })?;
+
+        let mut dependencies = BTreeMap::new();
+        for dependency_key in &source.dependency_keys {
+            let dependency_id = self.assemble_one(dependency_key)?;
+            dependencies.insert(workflow_path(dependency_key)?, dependency_id);
+        }
+
+        let entrypoint = workflow_path(key)?;
+        let files = workflow_files(&entrypoint, source.workflow)?;
+        let version =
+            WorkflowVersion::new(entrypoint.clone(), files, dependencies).map_err(|source| {
+                WorkflowVersionCollectError::InvalidShape {
+                    entrypoint: entrypoint.clone(),
+                    source,
+                }
+            })?;
+        let validated = ValidatedWorkflowVersion::new(version).map_err(|source| {
+            WorkflowVersionCollectError::InvalidVersion {
+                entrypoint: entrypoint.clone(),
+                source,
+            }
+        })?;
+        let id = validated.version().id().map_err(|source| {
+            WorkflowVersionCollectError::InvalidShape {
+                entrypoint: entrypoint.clone(),
+                source,
+            }
+        })?;
+
+        self.visiting.remove(key);
+        // Keys are distinct entrypoints and the entrypoint is part of the
+        // canonical bytes, so each key yields a distinct ID.
+        self.ids.insert(key.to_owned(), id);
+        self.versions.push((id, validated));
+        Ok(id)
+    }
+}
+
+fn workflow_files(
+    entrypoint: &WorkflowPath,
+    workflow: types::ManifestWorkflow,
+) -> Result<BTreeMap<WorkflowPath, String>, WorkflowVersionCollectError> {
+    let mut files = BTreeMap::new();
+    insert_file(&mut files, entrypoint, entrypoint.clone(), workflow.source)?;
+    if let Some(config) = workflow.config {
+        insert_file(
+            &mut files,
+            entrypoint,
+            workflow_path(&config.path)?,
+            config.source,
+        )?;
+    }
+    for (path, file) in workflow.files {
+        insert_file(&mut files, entrypoint, workflow_path(&path)?, file.content)?;
+    }
+    Ok(files)
+}
+
+fn insert_file(
+    files: &mut BTreeMap<WorkflowPath, String>,
+    entrypoint: &WorkflowPath,
+    path: WorkflowPath,
+    content: String,
+) -> Result<(), WorkflowVersionCollectError> {
+    if let Some(existing) = files.get(&path) {
+        if existing == &content {
+            return Ok(());
+        }
+        return Err(WorkflowVersionCollectError::PathCollision {
+            entrypoint: entrypoint.clone(),
+            path,
+        });
+    }
+    files.insert(path, content);
+    Ok(())
+}
+
+fn workflow_path(value: &str) -> Result<WorkflowPath, WorkflowVersionCollectError> {
+    WorkflowPath::new(value).map_err(|source| WorkflowVersionCollectError::InvalidPath {
+        path: value.to_owned(),
+        source,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    #![expect(
+        clippy::disallowed_methods,
+        reason = "collector tests write small temporary workflow fixtures synchronously"
+    )]
+
+    use std::fs;
+
+    use super::*;
+
+    fn write(root: &Path, path: &str, content: &str) {
+        let path = root.join(path);
+        fs::create_dir_all(path.parent().expect("fixture path should have a parent")).unwrap();
+        fs::write(path, content).unwrap();
+    }
+
+    fn write_complete_fixture(root: &Path) {
+        write(root, ".fabro/project.toml", "_version = 1\n");
+        write(
+            root,
+            ".fabro/workflows/root/workflow.toml",
+            r#"_version = 1
+[workflow]
+graph = "workflow.fabro"
+[run.goal]
+file = "goal.md"
+[run.environment.image]
+dockerfile = { path = "Dockerfile" }
+"#,
+        );
+        write(
+            root,
+            ".fabro/workflows/root/workflow.fabro",
+            r#"digraph Root {
+                graph [goal="@graph-goal.md"]
+                prompt [prompt="@prompts/task.md"]
+                child [stack.child_workflow="../child/workflow.fabro"]
+            }"#,
+        );
+        write(
+            root,
+            ".fabro/workflows/root/goal.md",
+            r#"Ship it. {% include "shared.md" %}"#,
+        );
+        write(root, ".fabro/workflows/root/shared.md", "shared");
+        write(root, ".fabro/workflows/root/Dockerfile", "FROM alpine\n");
+        write(root, ".fabro/workflows/root/graph-goal.md", "graph goal");
+        write(
+            root,
+            ".fabro/workflows/root/prompts/task.md",
+            r#"Task {% include "detail.md" %}"#,
+        );
+        write(root, ".fabro/workflows/root/prompts/detail.md", "detail");
+        write(
+            root,
+            ".fabro/workflows/child/workflow.fabro",
+            "digraph Child {}",
+        );
+    }
+
+    #[test]
+    fn packages_complete_checkout_relative_dependency_closure() {
+        let temp = tempfile::tempdir().unwrap();
+        write_complete_fixture(temp.path());
+
+        let closure = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
+        let versions = closure.versions().collect::<Vec<_>>();
+
+        assert_eq!(versions.len(), 2);
+        let (child_id, child) = versions[0];
+        assert_eq!(
+            child.version().entrypoint().as_str(),
+            ".fabro/workflows/child/workflow.fabro"
+        );
+        let (root_id, root) = versions[1];
+        assert_eq!(root_id, closure.root_id());
+        assert_eq!(
+            root.version().workflow_dependencies(),
+            &BTreeMap::from([(
+                WorkflowPath::new(".fabro/workflows/child/workflow.fabro").unwrap(),
+                child_id,
+            )])
+        );
+        for path in [
+            ".fabro/workflows/root/workflow.fabro",
+            ".fabro/workflows/root/workflow.toml",
+            ".fabro/workflows/root/goal.md",
+            ".fabro/workflows/root/shared.md",
+            ".fabro/workflows/root/Dockerfile",
+            ".fabro/workflows/root/graph-goal.md",
+            ".fabro/workflows/root/prompts/task.md",
+            ".fabro/workflows/root/prompts/detail.md",
+        ] {
+            assert!(
+                root.version()
+                    .files()
+                    .contains_key(&WorkflowPath::new(path).unwrap()),
+                "missing {path}"
+            );
+        }
+    }
+
+    #[test]
+    fn package_identity_does_not_depend_on_checkout_location() {
+        let first = tempfile::tempdir().unwrap();
+        let second = tempfile::tempdir().unwrap();
+        write_complete_fixture(first.path());
+        write_complete_fixture(second.path());
+
+        let first = collect_workflow_versions(Path::new("root"), first.path()).unwrap();
+        let second = collect_workflow_versions(Path::new("root"), second.path()).unwrap();
+
+        assert_eq!(first.root_id(), second.root_id());
+        assert_eq!(
+            first
+                .versions()
+                .map(|(id, version)| (id, version.version().canonical_bytes().unwrap()))
+                .collect::<Vec<_>>(),
+            second
+                .versions()
+                .map(|(id, version)| (id, version.version().canonical_bytes().unwrap()))
+                .collect::<Vec<_>>()
+        );
+    }
+
+    #[test]
+    fn packages_nested_imported_and_diamond_dependencies_deterministically() {
+        let temp = tempfile::tempdir().unwrap();
+        write(temp.path(), ".fabro/project.toml", "_version = 1\n");
+        write(
+            temp.path(),
+            ".fabro/workflows/root/workflow.toml",
+            "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/root/workflow.fabro",
+            r#"digraph Root {
+                imported [import="imports/extension.fabro"]
+                left [stack.child_workflow="../left/workflow.fabro"]
+                right [stack.child_workflow="../right/workflow.fabro"]
+            }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/root/imports/extension.fabro",
+            r#"digraph Extension {
+                child [stack.child_workflow="../../imported-child/workflow.fabro"]
+            }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/imported-child/workflow.fabro",
+            "digraph ImportedChild {}",
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/left/workflow.fabro",
+            r#"digraph Left { nested [stack.child_workflow="../nested/workflow.fabro"] }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/nested/workflow.fabro",
+            r#"digraph Nested { shared [stack.child_workflow="../shared/workflow.fabro"] }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/right/workflow.fabro",
+            r#"digraph Right { shared [stack.child_workflow="../shared/workflow.fabro"] }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/shared/workflow.fabro",
+            "digraph Shared {}",
+        );
+
+        let first = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
+        let second = collect_workflow_versions(Path::new("root"), temp.path()).unwrap();
+        let entrypoint_order = |closure: &CollectedWorkflowClosure| {
+            closure
+                .versions()
+                .map(|(_, version)| version.version().entrypoint().as_str().to_owned())
+                .collect::<Vec<_>>()
+        };
+
+        let expected = vec![
+            ".fabro/workflows/imported-child/workflow.fabro",
+            ".fabro/workflows/shared/workflow.fabro",
+            ".fabro/workflows/nested/workflow.fabro",
+            ".fabro/workflows/left/workflow.fabro",
+            ".fabro/workflows/right/workflow.fabro",
+            ".fabro/workflows/root/workflow.fabro",
+        ];
+        assert_eq!(entrypoint_order(&first), expected);
+        assert_eq!(entrypoint_order(&second), expected);
+        assert_eq!(first.root_id(), second.root_id());
+        assert_eq!(
+            entrypoint_order(&first)
+                .iter()
+                .filter(|path| path.as_str() == ".fabro/workflows/shared/workflow.fabro")
+                .count(),
+            1
+        );
+        let root = first.versions().last().unwrap().1;
+        assert!(root.version().files().contains_key(
+            &WorkflowPath::new(".fabro/workflows/root/imports/extension.fabro").unwrap()
+        ));
+        assert_eq!(root.version().workflow_dependencies().len(), 3);
+    }
+
+    #[test]
+    fn rejects_dependency_cycles() {
+        let temp = tempfile::tempdir().unwrap();
+        write(temp.path(), ".fabro/project.toml", "_version = 1\n");
+        write(
+            temp.path(),
+            ".fabro/workflows/root/workflow.toml",
+            "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/child/workflow.toml",
+            "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n",
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/root/workflow.fabro",
+            r#"digraph Root { child [stack.child_workflow="../child/workflow.fabro"] }"#,
+        );
+        write(
+            temp.path(),
+            ".fabro/workflows/child/workflow.fabro",
+            r#"digraph Child { root [stack.child_workflow="../root/workflow.fabro"] }"#,
+        );
+
+        let error = collect_workflow_versions(Path::new("root"), temp.path()).unwrap_err();
+
+        assert!(
+            matches!(&error, WorkflowVersionCollectError::DependencyCycle { .. }),
+            "unexpected error: {error:?}"
+        );
+    }
+}
diff --git a/lib/components/fabro-workflow-version/src/lib.rs b/lib/components/fabro-workflow-version/src/lib.rs
index ff0bd1e1a..bf0d89a44 100644
--- a/lib/components/fabro-workflow-version/src/lib.rs
+++ b/lib/components/fabro-workflow-version/src/lib.rs
@@ -116,8 +116,7 @@ impl ValidatedWorkflowVersion {
     /// run-admission time read the same bytes validation proved present.
     #[must_use]
     pub fn resolved_goal_file_content(&self) -> Option<&str> {
-        let config_path = WorkflowPath::new("workflow.toml")
-            .expect("the static workflow config path must be valid");
+        let config_path = self.0.config_path();
         let source = self.0.files().get(&config_path)?;
         let layer: SettingsLayer = source.parse().expect("validated workflow.toml must parse");
         let RunGoalLayer::File { file } = layer.run.as_ref().and_then(|run| run.goal.as_ref())?
@@ -164,8 +163,7 @@ fn validate_config(
     version: &WorkflowVersion,
     template_roots: &mut TemplateRoots,
 ) -> Result<(), WorkflowVersionError> {
-    let config_path =
-        WorkflowPath::new("workflow.toml").expect("the static workflow config path must be valid");
+    let config_path = version.config_path();
     let Some(source) = version.files().get(&config_path) else {
         return Ok(());
     };
@@ -649,7 +647,7 @@ mod tests {
                     BTreeMap::from([
                         (path("graphs/main.fabro"), "digraph W {}".to_owned()),
                         (
-                            path("workflow.toml"),
+                            path("graphs/workflow.toml"),
                             "_version = 1\n[run]\ngoal = \"{% include \\\"shared.md\\\" %}\"\n"
                                 .to_owned(),
                         ),
@@ -677,6 +675,38 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn discovers_workflow_config_beside_a_nested_entrypoint() {
+        let version = ValidatedWorkflowVersion::new(
+            WorkflowVersion::new(
+                path(".fabro/workflows/demo/workflow.fabro"),
+                BTreeMap::from([
+                    (
+                        path(".fabro/workflows/demo/workflow.fabro"),
+                        "digraph W {}".to_owned(),
+                    ),
+                    (
+                        path(".fabro/workflows/demo/workflow.toml"),
+                        "_version = 1\n[workflow]\ngraph = \"workflow.fabro\"\n[run.goal]\nfile = \"goal.md\"\n"
+                            .to_owned(),
+                    ),
+                    (
+                        path(".fabro/workflows/demo/goal.md"),
+                        "Ship the nested workflow".to_owned(),
+                    ),
+                ]),
+                BTreeMap::new(),
+            )
+            .expect("nested workflow version should be structurally valid"),
+        )
+        .expect("entrypoint-adjacent workflow config should validate");
+
+        assert_eq!(
+            version.resolved_goal_file_content(),
+            Some("Ship the nested workflow")
+        );
+    }
+
     #[test]
     fn rejects_non_static_or_nonportable_workflow_goal_file_references() {
         for reference in ["{{ vars.NAME }}", "{% include \"goal.md\" %}"] {
diff --git a/lib/foundation/fabro-types/src/workflow_version.rs b/lib/foundation/fabro-types/src/workflow_version.rs
index cba3a04c1..125544aa3 100644
--- a/lib/foundation/fabro-types/src/workflow_version.rs
+++ b/lib/foundation/fabro-types/src/workflow_version.rs
@@ -6,7 +6,7 @@ use serde::de::{Error as _, MapAccess, Visitor};
 use serde::{Deserialize, Deserializer, Serialize};
 use thiserror::Error;
 
-use crate::{WorkflowPath, WorkflowVersionId};
+use crate::{BlobHash, WorkflowPath, WorkflowVersionId};
 
 pub const MAX_WORKFLOW_VERSION_FILES: usize = 512;
 pub const MAX_WORKFLOW_VERSION_DEPENDENCIES: usize = 512;
@@ -86,6 +86,22 @@ impl WorkflowVersion {
         &self.workflow_dependencies
     }
 
+    /// Path of the optional `workflow.toml` that configures this version. It
+    /// always sits beside the entrypoint graph.
+    #[must_use]
+    pub fn config_path(&self) -> WorkflowPath {
+        self.entrypoint
+            .resolve_reference("workflow.toml")
+            .expect("the static workflow config path must resolve beside a valid entrypoint")
+    }
+
+    /// Content-addressed identity: the hash of the canonical wire form.
+    pub fn id(&self) -> Result<WorkflowVersionId, WorkflowVersionShapeError> {
+        Ok(WorkflowVersionId::from(BlobHash::new(
+            &self.canonical_bytes()?,
+        )))
+    }
+
     /// Serialize to the canonical wire form.
     ///
     /// Structural validity is guaranteed by construction, so this only