Persist run.created events in the run store

Hydrate the durable run store immediately after create-time event emission so
store-backed readers see the initial run.created event instead of only the
on-disk progress log. Add a regression test covering create-time store
visibility and wire in the object_store dependency needed by that test.

Cargo.lock

@@ -2041,6 +2041,7 @@ dependencies = [
  "hex",
  "md5",
  "mime_guess",
+ "object_store",
  "predicates",
  "rand 0.8.5",
  "regex",

lib/crates/fabro-workflow/Cargo.toml

@@ -40,6 +40,7 @@ thiserror.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tokio.workspace = true
+object_store.workspace = true
 ulid.workspace = true
 uuid.workspace = true
 rand.workspace = true
@@ -64,6 +65,7 @@ base64.workspace = true
 toml.workspace = true
 fabro-mcp = { path = "../fabro-mcp" }
 tokio = { workspace = true, features = ["test-util", "macros"] }
+object_store.workspace = true
 tempfile = "3"
 assert_cmd = "2"
 predicates = "3"

lib/crates/fabro-workflow/src/operations/create.rs

@@ -1,13 +1,17 @@
-use std::collections::HashMap;
 use std::collections::BTreeMap;
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
 
 use chrono::{Local, Utc};
 use fabro_config::{FabroSettings, FabroSettingsExt};
 use fabro_graphviz::graph::{AttrValue, Graph};
 use fabro_model::{Catalog, Provider};
 use fabro_sandbox::SandboxProvider;
+use fabro_store::SlateStore;
 use fabro_types::RunId;
+use object_store::local::LocalFileSystem;
 
 use crate::error::FabroError;
 use crate::pipeline::types::PersistOptions;
@@ -17,9 +21,13 @@ use crate::run_lookup::default_runs_base;
 use crate::run_status::{RunStatus, write_run_status};
 use crate::transforms::{Transform, expand_vars};
 use fabro_sandbox::daytona::detect_repo_info;
+use tokio::runtime::Builder;
 
-use crate::event::{WorkflowRunEvent, append_progress_event, canonicalize_event_at};
+use super::hydrate::open_or_hydrate_run;
 use super::source::{ResolveWorkflowInput, WorkflowInput, resolve_workflow};
+use crate::event::{
+    WorkflowRunEvent, append_progress_event, canonicalize_event_at, normalize_json_value,
+};
 
 const RUN_CONFIG_FILE: &str = "workflow.toml";
 
@@ -123,6 +131,7 @@ pub fn create(request: CreateRunInput) -> Result<CreatedRun, FabroError> {
         &resolved.raw_source,
         resolved.workflow_toml_path.as_deref(),
     )?;
+    hydrate_created_run(&persisted)?;
 
     Ok(CreatedRun {
         persisted,
@@ -139,10 +148,11 @@ fn emit_run_created_event(
 ) -> Result<(), FabroError> {
     let record = persisted.run_record();
     let workflow_config = workflow_toml_path.and_then(|path| std::fs::read_to_string(path).ok());
-    let settings = sort_json_value(
-        serde_json::to_value(&record.settings).map_err(|err| FabroError::engine(err.to_string()))?,
+    let settings = normalize_json_value(
+        serde_json::to_value(&record.settings)
+            .map_err(|err| FabroError::engine(err.to_string()))?,
     );
-    let graph = sort_json_value(
+    let graph = normalize_json_value(
         serde_json::to_value(&record.graph).map_err(|err| FabroError::engine(err.to_string()))?,
     );
     let event = WorkflowRunEvent::RunCreated {
@@ -151,7 +161,11 @@ fn emit_run_created_event(
         graph,
         workflow_source: (!workflow_source.is_empty()).then(|| workflow_source.to_string()),
         workflow_config,
-        labels: record.labels.clone().into_iter().collect::<BTreeMap<_, _>>(),
+        labels: record
+            .labels
+            .clone()
+            .into_iter()
+            .collect::<BTreeMap<_, _>>(),
         run_dir: persisted.run_dir().display().to_string(),
         working_directory: record.working_directory.display().to_string(),
         host_repo_path: record.host_repo_path.clone(),
@@ -164,20 +178,44 @@ fn emit_run_created_event(
         .map_err(|err| FabroError::engine(err.to_string()))
 }
 
-fn sort_json_value(value: serde_json::Value) -> serde_json::Value {
-    match value {
-        serde_json::Value::Object(map) => serde_json::Value::Object(
-            map.into_iter()
-                .map(|(key, value)| (key, sort_json_value(value)))
-                .collect(),
-        ),
-        serde_json::Value::Array(values) => {
-            serde_json::Value::Array(values.into_iter().map(sort_json_value).collect())
-        }
-        other => other,
+fn hydrate_created_run(persisted: &Persisted) -> Result<(), FabroError> {
+    let storage_dir = persisted.run_record().settings.storage_dir();
+    let run_dir = persisted.run_dir().to_path_buf();
+    let join = std::thread::spawn(move || -> Result<(), FabroError> {
+        let runtime = Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .map_err(|err| FabroError::engine(err.to_string()))?;
+        runtime.block_on(async move {
+            let store = build_durable_store(&storage_dir)?;
+            open_or_hydrate_run(store.as_ref(), &run_dir)
+                .await
+                .map(|_| ())
+        })
+    });
+
+    match join.join() {
+        Ok(result) => result,
+        Err(_) => Err(FabroError::engine(
+            "store hydration thread panicked".to_string(),
+        )),
     }
 }
 
+fn build_durable_store(storage_dir: &Path) -> Result<Arc<SlateStore>, FabroError> {
+    let store_path = storage_dir.join("store");
+    std::fs::create_dir_all(&store_path)?;
+    let object_store = Arc::new(
+        LocalFileSystem::new_with_prefix(&store_path)
+            .map_err(|err| FabroError::engine(err.to_string()))?,
+    );
+    Ok(Arc::new(SlateStore::new(
+        object_store,
+        "",
+        Duration::from_millis(5),
+    )))
+}
+
 fn validate_sandbox_provider(settings: &FabroSettings) -> Result<(), FabroError> {
     if let Some(provider) = settings
         .sandbox_settings()
@@ -374,7 +412,11 @@ pub(crate) fn make_run_dir(runs_base: &Path, run_id: &str, dry_run: bool) -> Pat
 mod tests {
     use super::*;
     use fabro_graphviz::graph::AttrValue;
+    use fabro_store::{SlateStore, Store};
     use fabro_types::fixtures;
+    use object_store::local::LocalFileSystem;
+    use std::sync::Arc;
+    use std::time::Duration;
 
     use crate::operations::{ValidateInput, validate};
     use crate::run_status::RunStatusRecordExt;
@@ -745,4 +787,45 @@ mod tests {
             )
         );
     }
+
+    #[tokio::test]
+    async fn create_hydrates_run_created_event_into_store() {
+        let dir = tempfile::tempdir().unwrap();
+        let storage_dir = dir.path().join("storage");
+        let run_dir = dir.path().join("run");
+        let created = create(CreateRunInput {
+            workflow: WorkflowInput::DotSource {
+                source: MINIMAL_DOT.to_string(),
+                base_dir: None,
+            },
+            settings: FabroSettings {
+                storage_dir: Some(storage_dir.clone()),
+                dry_run: Some(true),
+                ..Default::default()
+            },
+            cwd: dir.path().to_path_buf(),
+            workflow_slug: Some("slug".to_string()),
+            run_dir: Some(run_dir.clone()),
+            run_id: Some(fixtures::RUN_3),
+            host_repo_path: None,
+            base_branch: None,
+        })
+        .unwrap();
+
+        std::fs::create_dir_all(storage_dir.join("store")).unwrap();
+        let object_store =
+            Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).unwrap());
+        let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5)));
+        let run_store = store
+            .open_run_reader(&created.run_id)
+            .await
+            .unwrap()
+            .unwrap();
+        let events = run_store.list_events().await.unwrap();
+
+        assert_eq!(
+            events.first().unwrap().payload.as_value()["event"],
+            "run.created"
+        );
+    }
 }