diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs
index ecd9f7e05..e85242f59 100644
--- a/lib/crates/fabro-cli/src/commands/run/rewind.rs
+++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs
@@ -3,7 +3,7 @@ use anyhow::Result;
 use cli_table::format::{Border, Separator};
 use cli_table::{Cell, CellStruct, Color, Style, Table};
 use fabro_checkpoint::git::Store;
-use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunStatusTransitionProps};
+use fabro_types::run_event::{CheckpointCompletedProps, RunRewoundProps, RunSubmittedProps};
 use fabro_types::{EventBody, RunEvent};
 use fabro_util::terminal::Styles;
 use fabro_workflow::git::MetadataStore;
@@ -104,6 +104,7 @@ async fn reset_rewound_run_state(
         anyhow::anyhow!("failed to load durable store state before rewind: {err}")
     })?;
 
+    let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
     let _run_record = state
         .run
         .context("failed to restore run record after rewind: missing run metadata")?;
@@ -138,7 +139,10 @@ async fn reset_rewound_run_state(
             &run_event(
                 *run_id,
                 None,
-                EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
+                EventBody::RunSubmitted(RunSubmittedProps {
+                    reason: None,
+                    definition_blob,
+                }),
             ),
         )
         .await
diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs
index 7c19b0ca9..43faa3515 100644
--- a/lib/crates/fabro-cli/src/commands/store/dump.rs
+++ b/lib/crates/fabro-cli/src/commands/store/dump.rs
@@ -344,6 +344,8 @@ mod tests {
             labels: HashMap::from([("team".to_string(), "infra".to_string())]),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -490,6 +492,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs
index b4a8a961d..9b8b8589b 100644
--- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs
@@ -352,6 +352,23 @@ fn attach_json_errors_without_prompting_for_human_input() {
         .lines()
         .filter(|line| !line.trim().is_empty())
         .map(|line| serde_json::from_str(line).expect("attach JSON output should be JSONL"))
+        .map(|mut event: Value| {
+            if let Some(properties) = event.get_mut("properties").and_then(Value::as_object_mut) {
+                if properties.contains_key("manifest_blob") {
+                    properties.insert(
+                        "manifest_blob".to_string(),
+                        Value::String("[BLOB_ID]".to_string()),
+                    );
+                }
+                if properties.contains_key("definition_blob") {
+                    properties.insert(
+                        "definition_blob".to_string(),
+                        Value::String("[BLOB_ID]".to_string()),
+                    );
+                }
+            }
+            event
+        })
         .collect();
     fabro_json_snapshot!(context, &progress, @r#"
     [
@@ -461,6 +478,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
             }
           },
           "host_repo_path": "[TEMP_DIR]",
+          "manifest_blob": "[BLOB_ID]",
           "provenance": {
             "client": {
               "name": "fabro-cli",
@@ -503,7 +521,9 @@ fn attach_json_errors_without_prompting_for_human_input() {
       {
         "event": "run.submitted",
         "id": "[EVENT_ID]",
-        "properties": {},
+        "properties": {
+          "definition_blob": "[BLOB_ID]"
+        },
         "run_id": "[ULID]",
         "ts": "[TIMESTAMP]"
       },
diff --git a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs
index 7b1022448..54b8dc71b 100644
--- a/lib/crates/fabro-cli/tests/it/cmd/rewind.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/rewind.rs
@@ -174,6 +174,11 @@ fn rewind_preserves_event_history_and_clears_terminal_snapshot_state() {
         after_events[before_events.len() + 2].payload.as_value()["event"],
         "run.submitted"
     );
+    assert!(
+        after_events[before_events.len() + 2].payload.as_value()["properties"]["definition_blob"]
+            .is_string(),
+        "rewind should re-emit run.submitted with the definition_blob"
+    );
 
     let state = run_state(&setup.run.run_dir);
     assert_eq!(
diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs
index 62445ba66..cc3be197c 100644
--- a/lib/crates/fabro-cli/tests/it/cmd/run.rs
+++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs
@@ -625,6 +625,20 @@ fn json_run_implies_auto_approve_for_human_gates() {
         .map(|line| serde_json::from_str(line).expect("run JSON output should be JSONL"))
         .collect();
     for event in &mut progress {
+        if let Some(properties) = event.get_mut("properties").and_then(Value::as_object_mut) {
+            if properties.contains_key("manifest_blob") {
+                properties.insert(
+                    "manifest_blob".to_string(),
+                    Value::String("[BLOB_ID]".to_string()),
+                );
+            }
+            if properties.contains_key("definition_blob") {
+                properties.insert(
+                    "definition_blob".to_string(),
+                    Value::String("[BLOB_ID]".to_string()),
+                );
+            }
+        }
         let Some(llm) = event.pointer_mut("/properties/settings/llm") else {
             continue;
         };
@@ -748,6 +762,7 @@ fn json_run_implies_auto_approve_for_human_gates() {
             }
           },
           "host_repo_path": "[TEMP_DIR]",
+          "manifest_blob": "[BLOB_ID]",
           "provenance": {
             "client": {
               "name": "fabro-cli",
@@ -791,7 +806,9 @@ fn json_run_implies_auto_approve_for_human_gates() {
       {
         "event": "run.submitted",
         "id": "[EVENT_ID]",
-        "properties": {},
+        "properties": {
+          "definition_blob": "[BLOB_ID]"
+        },
         "run_id": "[ULID]",
         "ts": "[TIMESTAMP]"
       },
diff --git a/lib/crates/fabro-cli/tests/it/support/mod.rs b/lib/crates/fabro-cli/tests/it/support/mod.rs
index 48fa25e72..5e965f312 100644
--- a/lib/crates/fabro-cli/tests/it/support/mod.rs
+++ b/lib/crates/fabro-cli/tests/it/support/mod.rs
@@ -18,6 +18,14 @@ macro_rules! fabro_json_snapshot {
             r#""duration_ms":\s*\d+"#.to_string(),
             r#""duration_ms": "[DURATION_MS]""#.to_string(),
         ));
+        filters.push((
+            r#""manifest_blob":\s*"[0-9a-f]{64}""#.to_string(),
+            r#""manifest_blob": "[BLOB_ID]""#.to_string(),
+        ));
+        filters.push((
+            r#""definition_blob":\s*"[0-9a-f]{64}""#.to_string(),
+            r#""definition_blob": "[BLOB_ID]""#.to_string(),
+        ));
         filters.push((
             r#""run_dir":\s*"\[STORAGE_DIR\]/scratch/\d{8}-\[ULID\]""#.to_string(),
             r#""run_dir": "[RUN_DIR]""#.to_string(),
diff --git a/lib/crates/fabro-server/src/run_manifest.rs b/lib/crates/fabro-server/src/run_manifest.rs
index 9f8bfce7a..21458d727 100644
--- a/lib/crates/fabro-server/src/run_manifest.rs
+++ b/lib/crates/fabro-server/src/run_manifest.rs
@@ -124,6 +124,7 @@ pub(crate) fn create_run_input(prepared: PreparedManifest) -> CreateRunInput {
         workflow_slug: None,
         workflow_path: Some(prepared.target_path),
         workflow_bundle: Some(prepared.workflow_bundle),
+        submitted_manifest_bytes: None,
         run_id: prepared.run_id,
         host_repo_path: Some(prepared.working_directory.display().to_string()),
         repo_origin_url: prepared
diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs
index 4af46f9d3..2b9ef819c 100644
--- a/lib/crates/fabro-server/src/server.rs
+++ b/lib/crates/fabro-server/src/server.rs
@@ -3103,8 +3103,12 @@ async fn create_run(
     subject: AuthenticatedSubject,
     State(state): State<Arc<AppState>>,
     headers: HeaderMap,
-    Json(req): Json<RunManifest>,
+    body: Bytes,
 ) -> Response {
+    let req = match serde_json::from_slice::<RunManifest>(&body) {
+        Ok(req) => req,
+        Err(err) => return ApiError::bad_request(err.to_string()).into_response(),
+    };
     let prepared = match run_manifest::prepare_manifest_with_mode(
         &state.settings.read().unwrap(),
         &req,
@@ -3120,6 +3124,7 @@ async fn create_run(
     create_input.run_id = Some(run_id);
     create_input.artifact_storage = Some(RunArtifactStorage::ObjectStoreV1);
     create_input.provenance = Some(run_provenance(&headers, &subject));
+    create_input.submitted_manifest_bytes = Some(body.to_vec());
 
     let created = match Box::pin(operations::create(state.store.as_ref(), create_input)).await {
         Ok(created) => created,
@@ -5810,13 +5815,15 @@ async fn get_graph(
 
 #[cfg(test)]
 mod tests {
+    use std::path::PathBuf;
+
     use super::*;
     use axum::body::Body;
     use axum::http::Request;
     use fabro_config::server::{
         AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings,
     };
-    use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, fixtures};
+    use fabro_types::{InterviewQuestionRecord, InterviewQuestionType, RunBlobId, RunId, fixtures};
     #[cfg(unix)]
     use std::process::Stdio;
     use tower::ServiceExt;
@@ -5944,6 +5951,7 @@ mod tests {
                 workflow_slug: None,
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: None,
                 host_repo_path: None,
                 repo_origin_url: None,
@@ -6477,6 +6485,72 @@ mod tests {
         assert!(body["run"]["provenance"]["subject"]["login"].is_null());
     }
 
+    #[tokio::test]
+    async fn create_run_persists_manifest_and_definition_blobs_without_bundle_file() {
+        let state = create_app_state();
+        let app = build_router(Arc::clone(&state), AuthMode::Disabled);
+        let raw_manifest =
+            serde_json::to_string_pretty(&minimal_manifest_json(MINIMAL_DOT)).unwrap();
+
+        let req = Request::builder()
+            .method("POST")
+            .uri(api("/runs"))
+            .header("content-type", "application/json")
+            .body(Body::from(raw_manifest.clone()))
+            .unwrap();
+
+        let response = app.clone().oneshot(req).await.unwrap();
+        assert_eq!(response.status(), StatusCode::CREATED);
+        let body = body_json(response.into_body()).await;
+        let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
+
+        let run_store = state.store.open_run_reader(&run_id).await.unwrap();
+        let events = run_store.list_events().await.unwrap();
+        let created = events[0].payload.as_value();
+        let submitted = events[1].payload.as_value();
+        let manifest_blob = created["properties"]["manifest_blob"]
+            .as_str()
+            .expect("run.created should carry manifest_blob")
+            .parse::<RunBlobId>()
+            .unwrap();
+        let definition_blob = submitted["properties"]["definition_blob"]
+            .as_str()
+            .expect("run.submitted should carry definition_blob")
+            .parse::<RunBlobId>()
+            .unwrap();
+
+        let submitted_manifest_bytes = run_store
+            .read_blob(&manifest_blob)
+            .await
+            .unwrap()
+            .expect("submitted manifest blob should exist");
+        assert_eq!(submitted_manifest_bytes.as_ref(), raw_manifest.as_bytes());
+
+        let accepted_definition_bytes = run_store
+            .read_blob(&definition_blob)
+            .await
+            .unwrap()
+            .expect("accepted definition blob should exist");
+        let accepted_definition: serde_json::Value =
+            serde_json::from_slice(&accepted_definition_bytes).unwrap();
+        assert!(
+            accepted_definition.get("version").is_none(),
+            "accepted run definition should not carry compatibility versioning"
+        );
+        assert_eq!(accepted_definition["workflow_path"], "workflow.fabro");
+        assert!(accepted_definition["workflows"]["workflow.fabro"].is_object());
+
+        let run_dir = PathBuf::from(
+            created["properties"]["run_dir"]
+                .as_str()
+                .expect("run.created should include run_dir"),
+        );
+        assert!(
+            !run_dir.join("workflow_bundle.json").exists(),
+            "run scratch should no longer persist workflow_bundle.json"
+        );
+    }
+
     #[tokio::test]
     async fn list_run_events_returns_paginated_json() {
         let state = create_app_state();
@@ -7612,14 +7686,20 @@ mod tests {
         create_durable_run_with_events(
             &state,
             fixtures::RUN_1,
-            &[workflow_event::Event::RunSubmitted { reason: None }],
+            &[workflow_event::Event::RunSubmitted {
+                reason: None,
+                definition_blob: None,
+            }],
         )
         .await;
         create_durable_run_with_events(
             &state,
             fixtures::RUN_2,
             &[
-                workflow_event::Event::RunSubmitted { reason: None },
+                workflow_event::Event::RunSubmitted {
+                    reason: None,
+                    definition_blob: None,
+                },
                 workflow_event::Event::RunStarting { reason: None },
                 workflow_event::Event::RunRunning { reason: None },
             ],
@@ -7629,7 +7709,10 @@ mod tests {
             &state,
             fixtures::RUN_3,
             &[
-                workflow_event::Event::RunSubmitted { reason: None },
+                workflow_event::Event::RunSubmitted {
+                    reason: None,
+                    definition_blob: None,
+                },
                 workflow_event::Event::RunStarting { reason: None },
                 workflow_event::Event::RunRunning { reason: None },
                 workflow_event::Event::RunPaused,
@@ -7687,7 +7770,10 @@ mod tests {
             &state,
             run_id,
             &[
-                workflow_event::Event::RunSubmitted { reason: None },
+                workflow_event::Event::RunSubmitted {
+                    reason: None,
+                    definition_blob: None,
+                },
                 workflow_event::Event::RunStarting { reason: None },
                 workflow_event::Event::RunRunning { reason: None },
             ],
diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs
index c517b8dff..5aeb2b438 100644
--- a/lib/crates/fabro-store/src/keys.rs
+++ b/lib/crates/fabro-store/src/keys.rs
@@ -2,8 +2,7 @@ use fabro_types::{RunBlobId, RunId};
 
 const RUNS_PREFIX: &str = "runs#";
 const RUNS_INDEX_BY_START_PREFIX: &str = "runs#_index#by-start#";
-const BLOBS_PREFIX: &str = "blobs#";
-const GLOBAL_BLOBS_PREFIX: &str = "blobs#sha256#";
+const BLOBS_PREFIX: &str = "blobs#sha256#";
 
 pub(crate) fn runs_index_by_start_prefix() -> &'static str {
     RUNS_INDEX_BY_START_PREFIX
@@ -28,18 +27,12 @@ pub(crate) fn run_event_key(run_id: &RunId, seq: u32, epoch_ms: i64) -> String {
     format!("{}{seq:06}-{epoch_ms}", run_events_prefix(run_id))
 }
 
-pub(crate) fn blobs_prefix(run_id: &RunId) -> String {
-    let _ = run_id;
-    GLOBAL_BLOBS_PREFIX.to_string()
+pub(crate) fn blobs_prefix() -> &'static str {
+    BLOBS_PREFIX
 }
 
-pub(crate) fn blob_key(run_id: &RunId, id: &RunBlobId) -> String {
-    let _ = run_id;
-    format!("{GLOBAL_BLOBS_PREFIX}{id}")
-}
-
-pub(crate) fn legacy_blob_key(run_id: &RunId, id: &RunBlobId) -> String {
-    format!("{BLOBS_PREFIX}{run_id}#{id}")
+pub(crate) fn blob_key(id: &RunBlobId) -> String {
+    format!("{BLOBS_PREFIX}{id}")
 }
 
 pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
@@ -52,13 +45,7 @@ pub(crate) fn parse_event_seq(key: &str) -> Option<u32> {
 }
 
 pub(crate) fn parse_blob_id(key: &str) -> Option<RunBlobId> {
-    if let Some(blob_id) = key.strip_prefix(GLOBAL_BLOBS_PREFIX) {
-        return blob_id.parse().ok();
-    }
-
-    let rest = key.strip_prefix(BLOBS_PREFIX)?;
-    let (_, blob_id) = rest.split_once('#')?;
-    blob_id.parse().ok()
+    key.strip_prefix(BLOBS_PREFIX)?.parse().ok()
 }
 
 pub(crate) fn parse_run_id_from_index_key(key: &str) -> Option<RunId> {
@@ -99,12 +86,8 @@ mod tests {
 
     #[test]
     fn blob_keys_match_spec() {
-        let run_id: RunId = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
         let blob_id = RunBlobId::new(b"summary");
-        assert_eq!(
-            blob_key(&run_id, &blob_id),
-            format!("blobs#sha256#{blob_id}")
-        );
+        assert_eq!(blob_key(&blob_id), format!("blobs#sha256#{blob_id}"));
     }
 
     #[test]
@@ -118,10 +101,6 @@ mod tests {
             parse_blob_id(&format!("blobs#sha256#{blob_id}")),
             Some(blob_id)
         );
-        assert_eq!(
-            parse_blob_id(&format!("blobs#01JT56VE4Z5NZ814GZN2JZD65A#{blob_id}")),
-            Some(blob_id)
-        );
         assert_eq!(
             parse_run_id_from_index_key(
                 "runs#_index#by-start#2026-03-27#01JT56VE4Z5NZ814GZN2JZD65A"
@@ -134,5 +113,9 @@ mod tests {
     fn parse_helpers_reject_invalid_keys() {
         assert_eq!(parse_event_seq("runs#not-a-run#events#not-a-seq"), None);
         assert_eq!(parse_blob_id("blobs#not-a-uuid"), None);
+        assert_eq!(
+            parse_blob_id("blobs#01JT56VE4Z5NZ814GZN2JZD65A#not-a-blob"),
+            None
+        );
     }
 }
diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs
index 4d793e18e..260d285ff 100644
--- a/lib/crates/fabro-store/src/run_state.rs
+++ b/lib/crates/fabro-store/src/run_state.rs
@@ -95,6 +95,8 @@ impl RunProjection {
                     labels,
                     artifact_storage: props.artifact_storage,
                     provenance: props.provenance.clone(),
+                    manifest_blob: props.manifest_blob,
+                    definition_blob: None,
                 });
                 self.graph_source.clone_from(&props.workflow_source);
             }
@@ -107,6 +109,9 @@ impl RunProjection {
                 });
             }
             EventBody::RunSubmitted(props) => {
+                if let Some(run) = self.run.as_mut() {
+                    run.definition_blob = props.definition_blob;
+                }
                 self.status = Some(run_status_record(RunStatus::Submitted, props.reason, ts));
             }
             EventBody::RunStarting(props) => {
@@ -590,12 +595,14 @@ mod tests {
     use std::collections::HashMap;
 
     use chrono::Utc;
+    use serde_json::json;
 
     use super::{NodeState, RunProjection};
     use crate::{EventEnvelope, EventPayload, StageId};
     use fabro_types::run_event::{InterviewCompletedProps, InterviewOption, InterviewStartedProps};
     use fabro_types::{
-        Checkpoint, EventBody, InterviewQuestionType, RunControlAction, RunEvent, fixtures,
+        Checkpoint, EventBody, InterviewQuestionType, RunBlobId, RunControlAction, RunEvent,
+        Settings, fixtures,
     };
 
     fn test_event(seq: u32, body: EventBody, node_id: Option<&str>) -> EventEnvelope {
@@ -786,4 +793,67 @@ mod tests {
             "completed interview should clear pending state"
         );
     }
+
+    #[test]
+    fn projection_serialization_includes_manifest_and_definition_blob_refs() {
+        let manifest_blob = RunBlobId::new(br#"{"version":1}"#).to_string();
+        let definition_blob =
+            RunBlobId::new(br#"{"version":1,"workflow_path":"workflow.fabro"}"#).to_string();
+        let events = vec![
+            EventEnvelope {
+                seq: 1,
+                payload: EventPayload::new(
+                    json!({
+                        "id": "evt-run-created",
+                        "ts": "2026-04-07T12:00:00Z",
+                        "run_id": fixtures::RUN_1,
+                        "event": "run.created",
+                        "properties": {
+                            "settings": Settings::default(),
+                            "graph": {
+                                "name": "test",
+                                "nodes": {},
+                                "edges": [],
+                                "attrs": {}
+                            },
+                            "labels": {},
+                            "run_dir": "/tmp/run",
+                            "working_directory": "/tmp/run",
+                            "manifest_blob": manifest_blob
+                        }
+                    }),
+                    &fixtures::RUN_1,
+                )
+                .unwrap(),
+            },
+            EventEnvelope {
+                seq: 2,
+                payload: EventPayload::new(
+                    json!({
+                        "id": "evt-run-submitted",
+                        "ts": "2026-04-07T12:00:01Z",
+                        "run_id": fixtures::RUN_1,
+                        "event": "run.submitted",
+                        "properties": {
+                            "definition_blob": definition_blob
+                        }
+                    }),
+                    &fixtures::RUN_1,
+                )
+                .unwrap(),
+            },
+        ];
+
+        let state = RunProjection::apply_events(&events).unwrap();
+        let value = serde_json::to_value(&state).unwrap();
+
+        assert_eq!(
+            value["run"]["manifest_blob"],
+            events[0].payload.as_value()["properties"]["manifest_blob"]
+        );
+        assert_eq!(
+            value["run"]["definition_blob"],
+            events[1].payload.as_value()["properties"]["definition_blob"]
+        );
+    }
 }
diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs
index cecafad7c..9618394cc 100644
--- a/lib/crates/fabro-store/src/slate/mod.rs
+++ b/lib/crates/fabro-store/src/slate/mod.rs
@@ -173,7 +173,7 @@ impl Database {
 
         let db = self.open_db().await?;
         let mut keys_to_delete = Vec::new();
-        for prefix in [keys::run_data_prefix(run_id), keys::blobs_prefix(run_id)] {
+        for prefix in [keys::run_data_prefix(run_id)] {
             let mut iter = db.scan_prefix(prefix.as_bytes()).await?;
             while let Some(entry) = iter.next().await? {
                 keys_to_delete.push(String::from_utf8(entry.key.to_vec()).map_err(|err| {
@@ -290,6 +290,8 @@ mod tests {
             labels: std::collections::HashMap::from([("team".to_string(), "infra".to_string())]),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -403,6 +405,24 @@ mod tests {
         assert!(!list_paths(object_store, "runs/db").await.is_empty());
     }
 
+    #[tokio::test]
+    async fn delete_run_keeps_global_cas_blobs() {
+        let (_object_store, store) = make_store();
+        let run_1 = store.create_run(&test_run_id("run-1")).await.unwrap();
+        let run_2 = store.create_run(&test_run_id("run-2")).await.unwrap();
+        append_created(&run_1, "run-1", dt("2026-03-27T12:00:00Z")).await;
+        append_created(&run_2, "run-2", dt("2026-03-27T12:00:10Z")).await;
+
+        let shared_blob = br#"{"summary":"shared"}"#;
+        let shared_blob_id = run_1.write_blob(shared_blob).await.unwrap();
+
+        store.delete_run(&test_run_id("run-1")).await.unwrap();
+
+        let reopened = store.open_run(&test_run_id("run-2")).await.unwrap();
+        let read = reopened.read_blob(&shared_blob_id).await.unwrap();
+        assert_eq!(read.as_deref(), Some(shared_blob.as_slice()));
+    }
+
     #[tokio::test]
     async fn open_run_reader_is_read_only() {
         let (_object_store, store) = make_store();
diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs
index 433d53069..f36efb5bf 100644
--- a/lib/crates/fabro-store/src/slate/run_store.rs
+++ b/lib/crates/fabro-store/src/slate/run_store.rs
@@ -15,7 +15,6 @@ use crate::{EventEnvelope, EventPayload, Result, RunProjection, RunSummary, Stor
 use fabro_types::{RunBlobId, RunId};
 
 const DEFAULT_EVENT_TAIL_LIMIT: usize = 1024;
-
 #[derive(Clone)]
 pub struct RunDatabase {
     inner: Arc<RunDatabaseInner>,
@@ -297,32 +296,20 @@ impl RunDatabase {
             return Err(StoreError::ReadOnly);
         }
         let id = RunBlobId::new(data);
-        self.inner
-            .db
-            .put(keys::blob_key(&self.inner.run_id, &id), data)
-            .await?;
+        self.inner.db.put(keys::blob_key(&id), data).await?;
         Ok(id)
     }
 
     pub async fn read_blob(&self, id: &RunBlobId) -> Result<Option<Bytes>> {
-        let global = self
-            .inner
-            .db
-            .get(keys::blob_key(&self.inner.run_id, id))
-            .await?;
+        let global = self.inner.db.get(keys::blob_key(id)).await?;
         if global.is_some() {
             return Ok(global);
         }
-
-        Ok(self
-            .inner
-            .db
-            .get(keys::legacy_blob_key(&self.inner.run_id, id))
-            .await?)
+        Ok(None)
     }
 
     pub async fn list_blobs(&self) -> Result<Vec<RunBlobId>> {
-        list_blobs(&self.inner.db, &self.inner.run_id).await
+        list_blobs(&self.inner.db).await
     }
 
     pub async fn state(&self) -> Result<RunProjection> {
@@ -384,13 +371,11 @@ where
     Ok(events)
 }
 
-async fn list_blobs<R>(db: &R, run_id: &RunId) -> Result<Vec<RunBlobId>>
+async fn list_blobs<R>(db: &R) -> Result<Vec<RunBlobId>>
 where
     R: DbRead + Sync,
 {
-    let mut iter = db
-        .scan_prefix(keys::blobs_prefix(run_id).as_bytes())
-        .await?;
+    let mut iter = db.scan_prefix(keys::blobs_prefix().as_bytes()).await?;
     let mut blob_ids = Vec::new();
     while let Some(entry) = iter.next().await? {
         let key = key_to_string(&entry.key)?;
@@ -416,28 +401,6 @@ mod tests {
     use object_store::memory::InMemory;
 
     use crate::Database;
-    use crate::keys;
-
-    #[tokio::test]
-    async fn read_blob_falls_back_to_legacy_run_scoped_key() {
-        let object_store = Arc::new(InMemory::new());
-        let store = Database::new(object_store, "", Duration::from_millis(1));
-        let run_id = "01JT56VE4Z5NZ814GZN2JZD65A".parse().unwrap();
-        let run = store.create_run(&run_id).await.unwrap();
-        let blob = br#"{"legacy":true}"#;
-        let blob_id = fabro_types::RunBlobId::new(blob);
-
-        run.inner
-            .db
-            .put(keys::legacy_blob_key(&run_id, &blob_id), blob.as_slice())
-            .await
-            .unwrap();
-
-        let read = run.read_blob(&blob_id).await.unwrap().unwrap();
-
-        assert_eq!(read.as_ref(), blob);
-    }
-
     #[tokio::test]
     async fn list_blobs_reads_global_cas_namespace() {
         let object_store = Arc::new(InMemory::new());
diff --git a/lib/crates/fabro-types/src/run.rs b/lib/crates/fabro-types/src/run.rs
index e25e31f92..f4b76c2f6 100644
--- a/lib/crates/fabro-types/src/run.rs
+++ b/lib/crates/fabro-types/src/run.rs
@@ -4,6 +4,7 @@ use std::path::PathBuf;
 use serde::{Deserialize, Serialize};
 
 use crate::graph::Graph;
+use crate::run_blob_id::RunBlobId;
 use crate::run_id::RunId;
 use crate::settings::Settings;
 
@@ -74,6 +75,10 @@ pub struct RunRecord {
     pub artifact_storage: Option<RunArtifactStorage>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub provenance: Option<RunProvenance>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub manifest_blob: Option<RunBlobId>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub definition_blob: Option<RunBlobId>,
 }
 
 impl RunRecord {
diff --git a/lib/crates/fabro-types/src/run_event/mod.rs b/lib/crates/fabro-types/src/run_event/mod.rs
index ed6663073..194759b27 100644
--- a/lib/crates/fabro-types/src/run_event/mod.rs
+++ b/lib/crates/fabro-types/src/run_event/mod.rs
@@ -48,7 +48,7 @@ pub enum EventBody {
     #[serde(rename = "run.started")]
     RunStarted(RunStartedProps),
     #[serde(rename = "run.submitted")]
-    RunSubmitted(RunStatusTransitionProps),
+    RunSubmitted(RunSubmittedProps),
     #[serde(rename = "run.starting")]
     RunStarting(RunStatusTransitionProps),
     #[serde(rename = "run.running")]
@@ -682,7 +682,7 @@ mod tests {
 
     use serde_json::json;
 
-    use crate::{Edge, Graph, Node, Settings, fixtures};
+    use crate::{Edge, Graph, Node, RunBlobId, Settings, fixtures};
 
     use super::*;
 
@@ -765,6 +765,53 @@ mod tests {
         assert!(matches!(parsed.body, EventBody::RunCreated(_)));
     }
 
+    #[test]
+    fn run_created_round_trip_preserves_manifest_blob() {
+        let line = json!({
+            "id": "evt_created_blob",
+            "ts": "2026-04-04T12:00:00.000Z",
+            "run_id": fixtures::RUN_1,
+            "event": "run.created",
+            "properties": {
+                "settings": Settings::default(),
+                "graph": Graph::new("test"),
+                "labels": {},
+                "run_dir": "/tmp/run",
+                "working_directory": "/tmp/run",
+                "manifest_blob": RunBlobId::new(br#"{"version":1}"#).to_string()
+            }
+        });
+
+        let parsed = RunEvent::from_value(line.clone()).unwrap();
+        let serialized = parsed.to_value().unwrap();
+
+        assert_eq!(
+            serialized["properties"]["manifest_blob"],
+            line["properties"]["manifest_blob"]
+        );
+    }
+
+    #[test]
+    fn run_submitted_round_trip_preserves_definition_blob() {
+        let line = json!({
+            "id": "evt_submitted_blob",
+            "ts": "2026-04-04T12:00:00.000Z",
+            "run_id": fixtures::RUN_1,
+            "event": "run.submitted",
+            "properties": {
+                "definition_blob": RunBlobId::new(br#"{"workflow_path":"workflow.fabro"}"#).to_string()
+            }
+        });
+
+        let parsed = RunEvent::from_value(line.clone()).unwrap();
+        let serialized = parsed.to_value().unwrap();
+
+        assert_eq!(
+            serialized["properties"]["definition_blob"],
+            line["properties"]["definition_blob"]
+        );
+    }
+
     #[test]
     fn event_body_event_name_matches_wire_name() {
         let body = EventBody::StageCompleted(StageCompletedProps {
diff --git a/lib/crates/fabro-types/src/run_event/run.rs b/lib/crates/fabro-types/src/run_event/run.rs
index bf5fcd2a0..f4467c7b0 100644
--- a/lib/crates/fabro-types/src/run_event/run.rs
+++ b/lib/crates/fabro-types/src/run_event/run.rs
@@ -2,7 +2,9 @@ use std::collections::BTreeMap;
 
 use serde::{Deserialize, Serialize};
 
-use crate::{Graph, RunArtifactStorage, RunControlAction, RunProvenance, Settings, StatusReason};
+use crate::{
+    Graph, RunArtifactStorage, RunBlobId, RunControlAction, RunProvenance, Settings, StatusReason,
+};
 
 use super::{BilledTokenCounts, RunNoticeLevel};
 
@@ -32,6 +34,8 @@ pub struct RunCreatedProps {
     pub artifact_storage: Option<RunArtifactStorage>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub provenance: Option<RunProvenance>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub manifest_blob: Option<RunBlobId>,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@@ -55,6 +59,14 @@ pub struct RunStatusTransitionProps {
     pub reason: Option<StatusReason>,
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct RunSubmittedProps {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub reason: Option<StatusReason>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub definition_blob: Option<RunBlobId>,
+}
+
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct RunControlRequestedProps {
     pub action: RunControlAction,
diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs
index 518b052b1..3013c7aac 100644
--- a/lib/crates/fabro-workflow/src/event.rs
+++ b/lib/crates/fabro-workflow/src/event.rs
@@ -5,7 +5,7 @@ use std::sync::atomic::{AtomicI64, Ordering};
 
 use ::fabro_types::run_event as fabro_types;
 use ::fabro_types::{
-    BilledTokenCounts, RunControlAction, RunEvent, RunId, StageStatus, StatusReason,
+    BilledTokenCounts, RunBlobId, RunControlAction, RunEvent, RunId, StageStatus, StatusReason,
 };
 use anyhow::{Context, Result};
 use chrono::Utc;
@@ -57,6 +57,8 @@ pub enum Event {
         artifact_storage: Option<::fabro_types::RunArtifactStorage>,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         provenance: Option<::fabro_types::RunProvenance>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        manifest_blob: Option<RunBlobId>,
     },
     WorkflowRunStarted {
         name: String,
@@ -75,6 +77,8 @@ pub enum Event {
     RunSubmitted {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         reason: Option<StatusReason>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        definition_blob: Option<RunBlobId>,
     },
     RunStarting {
         #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -551,8 +555,11 @@ impl Event {
             Self::WorkflowRunStarted { name, run_id, .. } => {
                 info!(workflow = name.as_str(), run_id = %run_id, "Workflow run started");
             }
-            Self::RunSubmitted { reason } => {
-                info!(?reason, "Run submitted");
+            Self::RunSubmitted {
+                reason,
+                definition_blob,
+            } => {
+                info!(?reason, ?definition_blob, "Run submitted");
             }
             Self::RunStarting { reason } => {
                 info!(?reason, "Run starting");
@@ -1392,6 +1399,7 @@ fn event_body_from_event(event: &Event) -> EventBody {
             db_prefix,
             artifact_storage,
             provenance,
+            manifest_blob,
             ..
         } => EventBody::RunCreated(fabro_types::RunCreatedProps {
             settings: serde_json::from_value(settings.clone()).expect("run.created settings"),
@@ -1408,6 +1416,7 @@ fn event_body_from_event(event: &Event) -> EventBody {
             db_prefix: db_prefix.clone(),
             artifact_storage: *artifact_storage,
             provenance: provenance.clone(),
+            manifest_blob: *manifest_blob,
         }),
         Event::WorkflowRunStarted {
             name,
@@ -1425,9 +1434,13 @@ fn event_body_from_event(event: &Event) -> EventBody {
             worktree_dir: worktree_dir.clone(),
             goal: goal.clone(),
         }),
-        Event::RunSubmitted { reason } => {
-            EventBody::RunSubmitted(fabro_types::RunStatusTransitionProps { reason: *reason })
-        }
+        Event::RunSubmitted {
+            reason,
+            definition_blob,
+        } => EventBody::RunSubmitted(fabro_types::RunSubmittedProps {
+            reason: *reason,
+            definition_blob: *definition_blob,
+        }),
         Event::RunStarting { reason } => {
             EventBody::RunStarting(fabro_types::RunStatusTransitionProps { reason: *reason })
         }
diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs
index f311f2baf..987440f9a 100644
--- a/lib/crates/fabro-workflow/src/operations/create.rs
+++ b/lib/crates/fabro-workflow/src/operations/create.rs
@@ -16,7 +16,7 @@ use crate::pipeline::{self, Persisted, TransformOptions, Validated};
 use crate::records::RunRecord;
 use crate::run_lookup::default_scratch_base;
 use crate::transforms::{Transform, expand_vars};
-use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
+use crate::workflow_bundle::{RunDefinition, WorkflowBundle};
 use fabro_sandbox::daytona::detect_repo_info;
 use fabro_util::json::normalize_json_value;
 
@@ -31,6 +31,7 @@ pub struct CreateRunInput {
     pub workflow_slug: Option<String>,
     pub workflow_path: Option<PathBuf>,
     pub workflow_bundle: Option<WorkflowBundle>,
+    pub submitted_manifest_bytes: Option<Vec<u8>>,
     pub run_id: Option<RunId>,
     pub host_repo_path: Option<String>,
     pub repo_origin_url: Option<String>,
@@ -81,6 +82,7 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
         workflow_slug,
         workflow_path,
         workflow_bundle,
+        submitted_manifest_bytes,
         run_id,
         host_repo_path,
         repo_origin_url,
@@ -111,6 +113,13 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
     let goal_override = resolved.goal_override.clone();
     let current_dir = resolved.current_dir.clone();
     let file_resolver = resolved.file_resolver.clone();
+    let accepted_definition = match (&workflow_path, &workflow_bundle) {
+        (Some(workflow_path), Some(workflow_bundle)) => Some(RunDefinition::new(
+            workflow_path.clone(),
+            workflow_bundle.clone(),
+        )),
+        _ => None,
+    };
 
     let persisted = create_from_source(
         &resolved.raw_source,
@@ -136,10 +145,15 @@ pub async fn create(store: &Database, request: CreateRunInput) -> Result<Created
         .workflow_toml_path
         .as_deref()
         .and_then(|path| std::fs::read_to_string(path).ok());
-    persist_created_run(store, &persisted, &resolved.raw_source, workflow_config).await?;
-    if let (Some(workflow_path), Some(workflow_bundle)) = (workflow_path, workflow_bundle) {
-        persist_workflow_bundle(persisted.run_dir(), workflow_path, workflow_bundle)?;
-    }
+    persist_created_run(
+        store,
+        &persisted,
+        &resolved.raw_source,
+        workflow_config,
+        submitted_manifest_bytes.as_deref(),
+        accepted_definition.as_ref(),
+    )
+    .await?;
 
     Ok(CreatedRun {
         persisted,
@@ -154,6 +168,8 @@ async fn persist_created_run(
     persisted: &Persisted,
     workflow_source: &str,
     workflow_config: Option<String>,
+    submitted_manifest_bytes: Option<&[u8]>,
+    accepted_definition: Option<&RunDefinition>,
 ) -> Result<(), FabroError> {
     let record = persisted.run_record();
     let run_store = match store.create_run(&record.run_id).await {
@@ -164,6 +180,18 @@ async fn persist_created_run(
             .map_err(|open_err| FabroError::engine(open_err.to_string()))
             .map_err(|_| FabroError::engine(err.to_string()))?,
     };
+    let manifest_blob = match submitted_manifest_bytes {
+        Some(bytes) => Some(run_store.write_blob(bytes).await.map_err(store_error)?),
+        None => None,
+    };
+    let definition_blob = match accepted_definition {
+        Some(definition) => {
+            let bytes = serde_json::to_vec(definition)
+                .map_err(|err| FabroError::engine(err.to_string()))?;
+            Some(run_store.write_blob(&bytes).await.map_err(store_error)?)
+        }
+        None => None,
+    };
 
     let stored = to_run_event_at(
         &record.run_id,
@@ -193,6 +221,7 @@ async fn persist_created_run(
             db_prefix: None,
             artifact_storage: record.artifact_storage,
             provenance: record.provenance.clone(),
+            manifest_blob,
         },
         record.run_id.created_at(),
     );
@@ -209,7 +238,10 @@ async fn persist_created_run(
     append_event(
         &run_store,
         &record.run_id,
-        &Event::RunSubmitted { reason: None },
+        &Event::RunSubmitted {
+            reason: None,
+            definition_blob,
+        },
     )
     .await
     .map_err(store_error)
@@ -219,18 +251,6 @@ fn store_error(err: impl std::fmt::Display) -> FabroError {
     FabroError::engine(err.to_string())
 }
 
-fn persist_workflow_bundle(
-    run_dir: &Path,
-    workflow_path: PathBuf,
-    workflow_bundle: WorkflowBundle,
-) -> Result<(), FabroError> {
-    let path = run_dir.join("workflow_bundle.json");
-    let payload =
-        serde_json::to_string_pretty(&StoredWorkflowBundle::new(workflow_path, workflow_bundle))
-            .map_err(|err| FabroError::engine(err.to_string()))?;
-    std::fs::write(&path, payload).map_err(|err| FabroError::Io(err.to_string()))
-}
-
 fn validate_sandbox_provider(settings: &Settings) -> Result<(), FabroError> {
     if let Some(provider) = settings
         .sandbox_settings()
@@ -345,6 +365,8 @@ fn persist_validated(
         labels,
         artifact_storage,
         provenance,
+        manifest_blob: None,
+        definition_blob: None,
     };
 
     pipeline::persist(
@@ -699,6 +721,7 @@ mod tests {
                 workflow_slug: None,
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: None,
                 host_repo_path: None,
                 repo_origin_url: None,
@@ -748,6 +771,7 @@ mod tests {
                 workflow_slug: Some("slug".to_string()),
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_1),
                 host_repo_path: Some(dir.path().display().to_string()),
                 repo_origin_url: None,
@@ -829,6 +853,7 @@ mod tests {
                 workflow_slug: None,
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_2),
                 host_repo_path: None,
                 repo_origin_url: None,
@@ -873,6 +898,7 @@ mod tests {
                 workflow_slug: None,
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_2),
                 host_repo_path: None,
                 repo_origin_url: Some("https://github.com/acme/widgets".to_string()),
@@ -914,6 +940,7 @@ mod tests {
                 workflow_slug: Some("slug".to_string()),
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_3),
                 host_repo_path: None,
                 repo_origin_url: None,
@@ -957,6 +984,7 @@ mod tests {
                 workflow_slug: Some("slug".to_string()),
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_64),
                 host_repo_path: None,
                 repo_origin_url: None,
diff --git a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
index 6c78bb1d7..9e7fa196f 100644
--- a/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
+++ b/lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
@@ -380,6 +380,8 @@ mod tests {
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -455,6 +457,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
diff --git a/lib/crates/fabro-workflow/src/operations/resume.rs b/lib/crates/fabro-workflow/src/operations/resume.rs
index 681b90fe4..2d51bc256 100644
--- a/lib/crates/fabro-workflow/src/operations/resume.rs
+++ b/lib/crates/fabro-workflow/src/operations/resume.rs
@@ -36,12 +36,16 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
     let checkpoint = state
         .checkpoint
         .ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
+    let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
 
     cleanup_resume_artifacts(run_dir);
     append_event_to_sink(
         &services.event_sink,
         &services.run_id,
-        &Event::RunSubmitted { reason: None },
+        &Event::RunSubmitted {
+            reason: None,
+            definition_blob,
+        },
     )
     .await
     .map_err(|err| FabroError::engine(err.to_string()))?;
diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs
index 44790cae8..4ae8122ae 100644
--- a/lib/crates/fabro-workflow/src/operations/start.rs
+++ b/lib/crates/fabro-workflow/src/operations/start.rs
@@ -30,7 +30,7 @@ use crate::run_control::RunControlState;
 use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
 use crate::run_status::{RunStatus, StatusReason};
 use crate::runtime_store::RunStoreHandle;
-use crate::workflow_bundle::{StoredWorkflowBundle, WorkflowBundle};
+use crate::workflow_bundle::{RunDefinition, WorkflowBundle};
 use fabro_config::run::PullRequestSettings;
 use fabro_retro::retro::Retro;
 use fabro_sandbox::daytona::DaytonaConfig;
@@ -274,12 +274,18 @@ impl RunSession {
                 meta_branch: Some(MetadataStore::branch_name(&record.run_id.to_string())),
             })
         });
-        let stored_workflow_bundle = StoredWorkflowBundle::load_from_run_dir(persisted.run_dir())?;
-        let workflow_path = stored_workflow_bundle
+        let definition_blob = state.run.as_ref().and_then(|run| run.definition_blob);
+        let accepted_definition = match definition_blob {
+            Some(blob_id) => {
+                Some(load_accepted_run_definition(&services.run_store, blob_id).await?)
+            }
+            None => None,
+        };
+        let workflow_path = accepted_definition
             .as_ref()
-            .map(|bundle| bundle.workflow_path.clone());
+            .map(|definition| definition.workflow_path.clone());
         let workflow_bundle =
-            stored_workflow_bundle.map(|bundle| Arc::new(bundle.workflow_bundle()));
+            accepted_definition.map(|definition| Arc::new(definition.workflow_bundle()));
 
         if let Some(env) = settings
             .sandbox
@@ -413,6 +419,22 @@ impl RunSession {
     }
 }
 
+async fn load_accepted_run_definition(
+    run_store: &RunStoreHandle,
+    blob_id: fabro_types::RunBlobId,
+) -> Result<RunDefinition, FabroError> {
+    let bytes = run_store
+        .read_blob(&blob_id)
+        .await
+        .map_err(|err| FabroError::engine(err.to_string()))?
+        .ok_or_else(|| {
+            FabroError::engine(format!(
+                "run definition blob is missing from the run store: {blob_id}"
+            ))
+        })?;
+    serde_json::from_slice(&bytes).map_err(|err| FabroError::Parse(err.to_string()))
+}
+
 fn resolve_sandbox_provider(settings: &Settings) -> Result<SandboxProvider, FabroError> {
     settings
         .sandbox_settings()
@@ -803,9 +825,11 @@ mod tests {
     use crate::event::Emitter;
     use crate::handler::HandlerRegistry;
     use crate::handler::exit::ExitHandler;
+    use crate::handler::manager_loop::SubWorkflowHandler;
     use crate::handler::start::StartHandler;
     use crate::operations::resume;
     use crate::records::CheckpointExt;
+    use crate::workflow_bundle::{BundledWorkflow, WorkflowBundle};
 
     const MINIMAL_DOT: &str = r#"digraph Test {
         graph [goal="Build feature"]
@@ -842,6 +866,7 @@ mod tests {
                 workflow_slug: Some("test".to_string()),
                 workflow_path: None,
                 workflow_bundle: None,
+                submitted_manifest_bytes: None,
                 run_id: Some(fixtures::RUN_1),
                 host_repo_path: None,
                 repo_origin_url: None,
@@ -859,6 +884,7 @@ mod tests {
         let mut registry = HandlerRegistry::new(Box::new(StartHandler));
         registry.register("start", Box::new(StartHandler));
         registry.register("exit", Box::new(ExitHandler));
+        registry.register("stack.manager_loop", Box::new(SubWorkflowHandler));
         registry
     }
 
@@ -958,6 +984,94 @@ mod tests {
         assert!(run_store.state().await.unwrap().conclusion.is_some());
     }
 
+    #[tokio::test]
+    async fn start_can_run_bundle_backed_child_workflow_without_workflow_bundle_json() {
+        let temp = tempfile::tempdir().unwrap();
+        let run_dir = temp.path().join("run");
+        let emitter = Arc::new(Emitter::new(fixtures::RUN_1));
+        let registry = Arc::new(test_registry());
+        let store = memory_store();
+        let workflow_bundle = WorkflowBundle::new(HashMap::from([
+            (
+                PathBuf::from("workflow.fabro"),
+                BundledWorkflow {
+                    logical_path: PathBuf::from("workflow.fabro"),
+                    source: r#"digraph Root {
+                        graph [goal="Bundle child"]
+                        start [shape=Mdiamond]
+                        manager [
+                            type="stack.manager_loop",
+                            stack.child_workflow="./children/review.fabro",
+                            manager.max_cycles=100,
+                            manager.poll_interval="10ms"
+                        ]
+                        exit [shape=Msquare]
+                        start -> manager -> exit
+                    }"#
+                    .to_string(),
+                    files: HashMap::new(),
+                },
+            ),
+            (
+                PathBuf::from("children/review.fabro"),
+                BundledWorkflow {
+                    logical_path: PathBuf::from("children/review.fabro"),
+                    source: r#"digraph Review {
+                        start [shape=Mdiamond]
+                        exit [shape=Msquare]
+                        start -> exit
+                    }"#
+                    .to_string(),
+                    files: HashMap::new(),
+                },
+            ),
+        ]));
+
+        let created = crate::operations::create(
+            &store,
+            crate::operations::CreateRunInput {
+                workflow: crate::operations::WorkflowInput::Bundled(
+                    workflow_bundle
+                        .workflow(Path::new("workflow.fabro"))
+                        .unwrap()
+                        .clone(),
+                ),
+                settings: Settings {
+                    dry_run: Some(true),
+                    ..Default::default()
+                },
+                cwd: temp.path().to_path_buf(),
+                workflow_slug: Some("bundle-child".to_string()),
+                workflow_path: Some(PathBuf::from("workflow.fabro")),
+                workflow_bundle: Some(workflow_bundle),
+                submitted_manifest_bytes: None,
+                run_id: Some(fixtures::RUN_1),
+                host_repo_path: None,
+                repo_origin_url: None,
+                base_branch: None,
+                artifact_storage: None,
+                provenance: None,
+            },
+        )
+        .await
+        .unwrap();
+
+        let bundle_file = created.run_dir.join("workflow_bundle.json");
+        assert!(
+            !bundle_file.exists(),
+            "run scratch should not persist workflow_bundle.json"
+        );
+
+        let started = start(
+            &run_dir,
+            test_start_services(&store, &run_dir, emitter, registry).await,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(started.finalized.conclusion.status, StageStatus::Success);
+    }
+
     #[tokio::test]
     async fn start_invokes_on_node_callback_before_execution() {
         let temp = tempfile::tempdir().unwrap();
diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
index a31765a83..d57682b99 100644
--- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs
@@ -146,6 +146,8 @@ fn persisted_workflow(graph: Graph, source: String, run_dir: &Path, run_id: RunI
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         },
     )
 }
diff --git a/lib/crates/fabro-workflow/src/pipeline/initialize.rs b/lib/crates/fabro-workflow/src/pipeline/initialize.rs
index 891b724d6..913e1754e 100644
--- a/lib/crates/fabro-workflow/src/pipeline/initialize.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/initialize.rs
@@ -751,6 +751,8 @@ mod tests {
                 labels: HashMap::new(),
                 artifact_storage: None,
                 provenance: None,
+                manifest_blob: None,
+                definition_blob: None,
             },
         )
     }
diff --git a/lib/crates/fabro-workflow/src/pipeline/persist.rs b/lib/crates/fabro-workflow/src/pipeline/persist.rs
index 9b74eeecc..d06b284cf 100644
--- a/lib/crates/fabro-workflow/src/pipeline/persist.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/persist.rs
@@ -135,6 +135,8 @@ mod tests {
             ]),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -160,6 +162,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: record.artifact_storage,
                 provenance: record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs
index eac9a82d5..0fa55212c 100644
--- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs
@@ -1092,6 +1092,8 @@ mod tests {
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         };
         append_event(
             &run_store,
@@ -1112,6 +1114,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
@@ -1164,6 +1167,8 @@ mod tests {
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         };
         append_event(
             &run_store,
@@ -1184,6 +1189,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
@@ -1389,6 +1395,8 @@ mod tests {
             labels: std::collections::HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         };
         append_event(
             &run_store,
@@ -1409,6 +1417,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs
index 9fdb40dc5..391501750 100644
--- a/lib/crates/fabro-workflow/src/pipeline/retro.rs
+++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs
@@ -243,6 +243,8 @@ mod tests {
             labels: std::collections::HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         };
         append_event(
             &run_store,
@@ -263,6 +265,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs
index d0a2852f3..75b5708ea 100644
--- a/lib/crates/fabro-workflow/src/run_lookup.rs
+++ b/lib/crates/fabro-workflow/src/run_lookup.rs
@@ -425,6 +425,8 @@ mod tests {
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -456,6 +458,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: run_record.artifact_storage,
                 provenance: run_record.provenance.clone(),
+                manifest_blob: None,
             },
         )
         .await
@@ -463,7 +466,10 @@ mod tests {
         append_event(
             &run_store,
             &fixtures::RUN_1,
-            &Event::RunSubmitted { reason: None },
+            &Event::RunSubmitted {
+                reason: None,
+                definition_blob: None,
+            },
         )
         .await
         .unwrap();
diff --git a/lib/crates/fabro-workflow/src/runtime_store.rs b/lib/crates/fabro-workflow/src/runtime_store.rs
index 5ce71b450..f4357c9a8 100644
--- a/lib/crates/fabro-workflow/src/runtime_store.rs
+++ b/lib/crates/fabro-workflow/src/runtime_store.rs
@@ -112,7 +112,7 @@ mod tests {
     use fabro_graphviz::graph::Graph;
     use fabro_store::Database;
     use fabro_types::fixtures;
-    use fabro_types::run_event::RunStatusTransitionProps;
+    use fabro_types::run_event::RunSubmittedProps;
     use fabro_types::{EventBody, RunEvent, Settings};
     use object_store::memory::InMemory;
 
@@ -142,6 +142,8 @@ mod tests {
             labels: HashMap::new(),
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
+            definition_blob: None,
         }
     }
 
@@ -168,6 +170,7 @@ mod tests {
                 db_prefix: None,
                 artifact_storage: None,
                 provenance: None,
+                manifest_blob: None,
             },
         )
         .await
@@ -194,7 +197,10 @@ mod tests {
             node_label: None,
             session_id: None,
             parent_session_id: None,
-            body: EventBody::RunSubmitted(RunStatusTransitionProps { reason: None }),
+            body: EventBody::RunSubmitted(RunSubmittedProps {
+                reason: None,
+                definition_blob: None,
+            }),
         };
         handle.append_run_event(&event).await.unwrap();
 
diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs
index 5a349d4e2..f37855312 100644
--- a/lib/crates/fabro-workflow/src/test_support.rs
+++ b/lib/crates/fabro-workflow/src/test_support.rs
@@ -99,6 +99,7 @@ async fn initialized(
             db_prefix: None,
             artifact_storage: None,
             provenance: None,
+            manifest_blob: None,
         },
     )
     .await
diff --git a/lib/crates/fabro-workflow/src/workflow_bundle.rs b/lib/crates/fabro-workflow/src/workflow_bundle.rs
index 5601e4b73..fd7afe078 100644
--- a/lib/crates/fabro-workflow/src/workflow_bundle.rs
+++ b/lib/crates/fabro-workflow/src/workflow_bundle.rs
@@ -4,7 +4,6 @@ use std::sync::Arc;
 
 use serde::{Deserialize, Serialize};
 
-use crate::error::FabroError;
 use crate::file_resolver::{BundleFileResolver, FileResolver, normalize_logical_path};
 
 #[derive(Clone, Debug, Default, Serialize, Deserialize)]
@@ -62,12 +61,12 @@ impl WorkflowBundle {
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
-pub struct StoredWorkflowBundle {
+pub struct RunDefinition {
     pub workflow_path: PathBuf,
     pub workflows: HashMap<PathBuf, BundledWorkflow>,
 }
 
-impl StoredWorkflowBundle {
+impl RunDefinition {
     #[must_use]
     pub fn new(workflow_path: PathBuf, bundle: WorkflowBundle) -> Self {
         Self {
@@ -80,17 +79,4 @@ impl StoredWorkflowBundle {
     pub fn workflow_bundle(&self) -> WorkflowBundle {
         WorkflowBundle::new(self.workflows.clone())
     }
-
-    pub fn load_from_run_dir(run_dir: &Path) -> Result<Option<Self>, FabroError> {
-        let path = run_dir.join("workflow_bundle.json");
-        if !path.exists() {
-            return Ok(None);
-        }
-
-        let payload =
-            std::fs::read_to_string(&path).map_err(|err| FabroError::Io(err.to_string()))?;
-        serde_json::from_str(&payload)
-            .map(Some)
-            .map_err(|err| FabroError::Parse(err.to_string()))
-    }
 }