Remove run hydration from production paths

lib/crates/fabro-cli/src/commands/run/rewind.rs

@@ -8,9 +8,10 @@ use fabro_util::terminal::Styles;
 use fabro_workflow::git::MetadataStore;
 use fabro_workflow::operations::{
     RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild,
-    find_run_id_by_prefix_or_store, open_or_hydrate_run, rewind,
+    find_run_id_by_prefix_or_store, rewind,
 };
 use fabro_workflow::records::CheckpointExt;
+use fabro_workflow::records::{RunRecord, RunRecordExt};
 use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
 use fabro_workflow::run_status::{self, RunStatus};
 use git2::Repository;
@@ -106,6 +107,7 @@ async fn reset_rewound_run_state(
     run_id: &fabro_types::RunId,
     run_dir: &std::path::Path,
 ) -> Result<()> {
+    let run_record = RunRecord::load(run_dir)?;
     let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())?
         .context("rewound metadata branch is missing checkpoint.json")?;
     checkpoint.save(&run_dir.join("checkpoint.json"))?;
@@ -126,11 +128,32 @@ async fn reset_rewound_run_state(
         .delete_run(run_id)
         .await
         .map_err(|err| anyhow::anyhow!("failed to reset durable store run: {err}"))?;
-    open_or_hydrate_run(durable_store, run_dir)
+    let run_dir_string = run_dir.to_string_lossy().to_string();
+    let run_store = durable_store
+        .create_run(run_id, run_record.created_at, Some(&run_dir_string))
         .await
-        .map_err(|err| {
-            anyhow::anyhow!("failed to restore durable store run after rewind: {err}")
-        })?;
+        .map_err(|err| anyhow::anyhow!("failed to recreate durable store run: {err}"))?;
+    run_store
+        .put_run(&run_record)
+        .await
+        .map_err(|err| anyhow::anyhow!("failed to restore run record after rewind: {err}"))?;
+    if let Ok(dot_source) = std::fs::read_to_string(run_dir.join("workflow.fabro")) {
+        run_store
+            .put_graph(&dot_source)
+            .await
+            .map_err(|err| anyhow::anyhow!("failed to restore graph after rewind: {err}"))?;
+    }
+    run_store
+        .put_status(&fabro_types::RunStatusRecord::new(
+            RunStatus::Submitted,
+            None,
+        ))
+        .await
+        .map_err(|err| anyhow::anyhow!("failed to restore run status after rewind: {err}"))?;
+    run_store
+        .put_checkpoint(&checkpoint)
+        .await
+        .map_err(|err| anyhow::anyhow!("failed to restore checkpoint after rewind: {err}"))?;
     Ok(())
 }
 

lib/crates/fabro-server/src/server.rs

@@ -682,14 +682,25 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
         }
     }
 
-    let run_store = match operations::open_or_hydrate_run(state.store.as_ref(), &run_dir).await {
-        Ok(run_store) => run_store,
-        Err(e) => {
-            tracing::error!(run_id = %run_id, error = %e, "Failed to open or hydrate run store");
+    let run_store = match state.store.open_run(&run_id).await {
+        Ok(Some(run_store)) => run_store,
+        Ok(None) => {
+            tracing::error!(run_id = %run_id, "Run store missing");
             let mut runs = state.runs.lock().expect("runs lock poisoned");
             if let Some(managed_run) = runs.get_mut(&run_id) {
                 managed_run.status = RunStatus::Failed;
-                managed_run.error = Some(format!("Failed to open or hydrate run store: {e}"));
+                managed_run.error = Some("Run store missing".to_string());
+                clear_live_run_state(managed_run);
+            }
+            state.scheduler_notify.notify_one();
+            return;
+        }
+        Err(e) => {
+            tracing::error!(run_id = %run_id, error = %e, "Failed to open run store");
+            let mut runs = state.runs.lock().expect("runs lock poisoned");
+            if let Some(managed_run) = runs.get_mut(&run_id) {
+                managed_run.status = RunStatus::Failed;
+                managed_run.error = Some(format!("Failed to open run store: {e}"));
                 clear_live_run_state(managed_run);
             }
             state.scheduler_notify.notify_one();

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

@@ -1,326 +0,0 @@
-use std::io::{BufRead, ErrorKind};
-use std::path::Path;
-use std::sync::Arc;
-
-use fabro_sandbox::{SandboxRecord, SandboxRecordExt};
-use fabro_store::{EventPayload, RunStore, Store};
-use tracing::warn;
-
-use crate::error::FabroError;
-use crate::records::{
-    Checkpoint, CheckpointExt, Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord,
-    StartRecordExt,
-};
-use crate::run_status::{RunStatusRecord, RunStatusRecordExt};
-use fabro_retro::{RetroExt, retro::Retro};
-
-const GRAPH_FILE_NAME: &str = "workflow.fabro";
-const LEGACY_GRAPH_FILE_NAME: &str = "graph.fabro";
-
-pub async fn open_or_hydrate_run(
-    store: &dyn Store,
-    run_dir: &Path,
-) -> Result<Arc<dyn RunStore>, FabroError> {
-    let record = RunRecord::load(run_dir)?;
-    if let Some(run_store) = store.open_run(&record.run_id).await.map_err(store_error)? {
-        return Ok(run_store);
-    }
-
-    let run_dir_string = run_dir.to_string_lossy().to_string();
-    let run_store = store
-        .create_run(&record.run_id, record.created_at, Some(&run_dir_string))
-        .await
-        .map_err(store_error)?;
-
-    run_store.put_run(&record).await.map_err(store_error)?;
-
-    if let Some(dot_source) = load_graph_source(run_dir)? {
-        run_store
-            .put_graph(&dot_source)
-            .await
-            .map_err(store_error)?;
-    }
-
-    if let Some(status) = load_status_record(run_dir)? {
-        run_store.put_status(&status).await.map_err(store_error)?;
-    }
-    if let Some(start) = load_start_record(run_dir)? {
-        run_store.put_start(&start).await.map_err(store_error)?;
-    }
-    match load_checkpoint(run_dir) {
-        Ok(Some(checkpoint)) => {
-            run_store
-                .put_checkpoint(&checkpoint)
-                .await
-                .map_err(store_error)?;
-        }
-        Ok(None) => {}
-        Err(err) => {
-            tracing::warn!(error = %err, "Skipping malformed checkpoint.json during hydration");
-        }
-    }
-    match load_conclusion(run_dir) {
-        Ok(Some(conclusion)) => {
-            run_store
-                .put_conclusion(&conclusion)
-                .await
-                .map_err(store_error)?;
-        }
-        Ok(None) => {}
-        Err(err) => {
-            tracing::warn!(error = %err, "Skipping malformed conclusion.json during hydration");
-        }
-    }
-    match load_retro(run_dir) {
-        Ok(Some(retro)) => {
-            run_store.put_retro(&retro).await.map_err(store_error)?;
-        }
-        Ok(None) => {}
-        Err(err) => tracing::warn!(error = %err, "Skipping malformed retro.json during hydration"),
-    }
-    match load_sandbox_record(run_dir) {
-        Ok(Some(sandbox)) => {
-            run_store.put_sandbox(&sandbox).await.map_err(store_error)?;
-        }
-        Ok(None) => {}
-        Err(err) => {
-            tracing::warn!(error = %err, "Skipping malformed sandbox.json during hydration");
-        }
-    }
-
-    hydrate_events(run_dir, &record.run_id, run_store.as_ref()).await?;
-
-    Ok(run_store)
-}
-
-async fn hydrate_events(
-    run_dir: &Path,
-    run_id: &fabro_types::RunId,
-    run_store: &dyn RunStore,
-) -> Result<(), FabroError> {
-    let progress_path = run_dir.join("progress.jsonl");
-    if !progress_path.exists() {
-        return Ok(());
-    }
-
-    let file = std::fs::File::open(&progress_path)?;
-    for (line_number, line_result) in std::io::BufReader::new(file).lines().enumerate() {
-        let line = line_result?;
-        let trimmed = line.trim();
-        if trimmed.is_empty() {
-            continue;
-        }
-
-        let value = match serde_json::from_str::<serde_json::Value>(trimmed) {
-            Ok(value) => value,
-            Err(err) => {
-                warn!(
-                    path = %progress_path.display(),
-                    line_number = line_number + 1,
-                    error = %err,
-                    "Skipping malformed progress event during hydration"
-                );
-                continue;
-            }
-        };
-
-        let payload = match EventPayload::new(value, run_id) {
-            Ok(payload) => payload,
-            Err(err) => {
-                warn!(
-                    path = %progress_path.display(),
-                    line_number = line_number + 1,
-                    error = %err,
-                    "Skipping invalid progress event during hydration"
-                );
-                continue;
-            }
-        };
-
-        run_store
-            .append_event(&payload)
-            .await
-            .map_err(store_error)?;
-    }
-
-    Ok(())
-}
-
-fn load_graph_source(run_dir: &Path) -> Result<Option<String>, FabroError> {
-    for name in [GRAPH_FILE_NAME, LEGACY_GRAPH_FILE_NAME] {
-        match std::fs::read_to_string(run_dir.join(name)) {
-            Ok(source) => return Ok(Some(source)),
-            Err(err) if err.kind() == ErrorKind::NotFound => {}
-            Err(err) => return Err(err.into()),
-        }
-    }
-    Ok(None)
-}
-
-fn load_status_record(run_dir: &Path) -> Result<Option<RunStatusRecord>, FabroError> {
-    let path = run_dir.join("status.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    Ok(Some(
-        RunStatusRecord::load(&path).map_err(|err| FabroError::Io(err.to_string()))?,
-    ))
-}
-
-fn load_start_record(run_dir: &Path) -> Result<Option<StartRecord>, FabroError> {
-    let path = run_dir.join("start.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    Ok(Some(StartRecord::load(run_dir)?))
-}
-
-fn load_checkpoint(run_dir: &Path) -> Result<Option<Checkpoint>, FabroError> {
-    let path = run_dir.join("checkpoint.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    Ok(Some(Checkpoint::load(&path)?))
-}
-
-fn load_conclusion(run_dir: &Path) -> Result<Option<Conclusion>, FabroError> {
-    let path = run_dir.join("conclusion.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    Ok(Some(Conclusion::load(&path)?))
-}
-
-fn load_retro(run_dir: &Path) -> Result<Option<Retro>, FabroError> {
-    let path = run_dir.join("retro.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    Retro::load(run_dir)
-        .map(Some)
-        .map_err(|err| FabroError::Io(err.to_string()))
-}
-
-fn load_sandbox_record(run_dir: &Path) -> Result<Option<SandboxRecord>, FabroError> {
-    let path = run_dir.join("sandbox.json");
-    if !path.exists() {
-        return Ok(None);
-    }
-    SandboxRecord::load(&path)
-        .map(Some)
-        .map_err(|err| FabroError::Io(err.to_string()))
-}
-
-fn store_error(err: impl std::fmt::Display) -> FabroError {
-    FabroError::engine(err.to_string())
-}
-
-#[cfg(test)]
-mod tests {
-    use std::collections::HashMap;
-    use std::path::{Path, PathBuf};
-
-    use chrono::Utc;
-    use fabro_config::FabroSettings;
-    use fabro_graphviz::graph::Graph;
-    use fabro_store::{InMemoryStore, Store};
-    use fabro_types::{Conclusion, RunStatus, RunStatusRecord, StageStatus, fixtures};
-
-    use super::open_or_hydrate_run;
-    use crate::event::{WorkflowRunEvent, append_progress_event, canonicalize_event};
-    use crate::records::{Checkpoint, CheckpointExt, ConclusionExt, RunRecord, RunRecordExt};
-    use crate::run_status::RunStatusRecordExt;
-
-    fn test_run_id() -> fabro_types::RunId {
-        fixtures::RUN_1
-    }
-
-    fn write_run(run_dir: &Path) {
-        let record = RunRecord {
-            run_id: test_run_id(),
-            created_at: Utc::now(),
-            settings: FabroSettings::default(),
-            graph: Graph::new("test"),
-            workflow_slug: Some("test".to_string()),
-            working_directory: PathBuf::from("/tmp/project"),
-            host_repo_path: Some("/tmp/project".to_string()),
-            base_branch: Some("main".to_string()),
-            labels: HashMap::new(),
-        };
-        std::fs::create_dir_all(run_dir).unwrap();
-        record.save(run_dir).unwrap();
-        std::fs::write(
-            run_dir.join("workflow.fabro"),
-            "digraph test { start -> exit }",
-        )
-        .unwrap();
-        RunStatusRecord::new(RunStatus::Running, None)
-            .save(&run_dir.join("status.json"))
-            .unwrap();
-        let checkpoint = Checkpoint::from_context(
-            &crate::context::Context::new(),
-            "start",
-            vec!["start".to_string()],
-            HashMap::new(),
-            HashMap::new(),
-            Some("exit".to_string()),
-            HashMap::new(),
-            HashMap::new(),
-            HashMap::new(),
-        );
-        checkpoint.save(&run_dir.join("checkpoint.json")).unwrap();
-        let conclusion = Conclusion {
-            timestamp: Utc::now(),
-            status: StageStatus::Success,
-            duration_ms: 5,
-            failure_reason: None,
-            final_git_commit_sha: None,
-            stages: Vec::new(),
-            total_cost: None,
-            total_retries: 0,
-            total_input_tokens: 0,
-            total_output_tokens: 0,
-            total_cache_read_tokens: 0,
-            total_cache_write_tokens: 0,
-            total_reasoning_tokens: 0,
-            has_pricing: false,
-        };
-        conclusion.save(&run_dir.join("conclusion.json")).unwrap();
-        let envelope = canonicalize_event(
-            &test_run_id(),
-            &WorkflowRunEvent::RunNotice {
-                level: crate::event::RunNoticeLevel::Info,
-                code: "hydrated".to_string(),
-                message: "hello".to_string(),
-            },
-        );
-        append_progress_event(run_dir, &envelope).unwrap();
-    }
-
-    #[tokio::test]
-    async fn hydrates_run_records_into_store() {
-        let temp = tempfile::tempdir().unwrap();
-        let run_dir = temp.path().join("run-123");
-        write_run(&run_dir);
-
-        let store = InMemoryStore::default();
-        let run_store = open_or_hydrate_run(&store, &run_dir).await.unwrap();
-
-        assert_eq!(
-            run_store.get_run().await.unwrap().unwrap().run_id,
-            test_run_id()
-        );
-        assert!(run_store.get_checkpoint().await.unwrap().is_some());
-        assert!(run_store.get_conclusion().await.unwrap().is_some());
-        assert_eq!(run_store.list_events().await.unwrap().len(), 1);
-
-        let listed = store
-            .list_runs(&fabro_store::ListRunsQuery::default())
-            .await
-            .unwrap();
-        assert_eq!(
-            listed[0].run_dir.as_deref(),
-            Some(run_dir.to_string_lossy().as_ref())
-        );
-    }
-}

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

@@ -1,6 +1,5 @@
 mod create;
 mod fork;
-mod hydrate;
 mod rebuild_meta;
 mod resume;
 mod rewind;
@@ -13,7 +12,6 @@ mod validate;
 pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec};
 pub use create::{CreateRunInput, CreatedRun, create};
 pub use fork::{ForkRunInput, fork};
-pub use hydrate::open_or_hydrate_run;
 pub use rebuild_meta::{
     build_timeline_or_rebuild, find_run_id_by_prefix_or_store, rebuild_metadata_branch,
 };

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

@@ -912,7 +912,7 @@ mod tests {
 
     use chrono::Utc;
     use fabro_config::FabroSettings;
-    use fabro_store::InMemoryStore;
+    use fabro_store::{InMemoryStore, Store};
     use fabro_types::fixtures;
 
     use super::*;
@@ -931,9 +931,9 @@ mod tests {
         start -> exit
     }"#;
 
-    async fn persisted_workflow(dot: &str, run_dir: &Path) -> Persisted {
+    async fn persisted_workflow(dot: &str, run_dir: &Path) -> (Persisted, InMemoryStore) {
         let store = InMemoryStore::default();
-        crate::operations::create(
+        let created = crate::operations::create(
             &store,
             crate::operations::CreateRunInput {
                 workflow: crate::operations::WorkflowInput::DotSource {
@@ -956,8 +956,8 @@ mod tests {
             },
         )
         .await
-        .unwrap()
-        .persisted
+        .unwrap();
+        (created.persisted, store)
     }
 
     fn test_registry() -> HandlerRegistry {
@@ -968,7 +968,8 @@ mod tests {
     }
 
     async fn test_start_services(
-        run_dir: &Path,
+        store: &InMemoryStore,
+        _run_dir: &Path,
         emitter: Arc<EventEmitter>,
         registry: Arc<HandlerRegistry>,
     ) -> StartServices {
@@ -976,9 +977,7 @@ mod tests {
             cancel_token: None,
             emitter,
             interviewer: Arc::new(fabro_interview::AutoApproveInterviewer),
-            run_store: crate::operations::open_or_hydrate_run(&InMemoryStore::default(), run_dir)
-                .await
-                .unwrap(),
+            run_store: store.open_run(&fixtures::RUN_1).await.unwrap().unwrap(),
             github_app: None,
             on_node: None,
             registry_override: Some(registry),
@@ -1012,10 +1011,10 @@ mod tests {
             });
         }
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
         let started = start(
             &run_dir,
-            test_start_services(&run_dir, emitter, registry).await,
+            test_start_services(&store, &run_dir, emitter, registry).await,
         )
         .await
         .unwrap();
@@ -1035,11 +1034,11 @@ mod tests {
         let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
         let registry = Arc::new(test_registry());
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
 
         let started = start(
             &run_dir,
-            test_start_services(&run_dir, emitter, registry).await,
+            test_start_services(&store, &run_dir, emitter, registry).await,
         )
         .await
         .unwrap();
@@ -1056,7 +1055,7 @@ mod tests {
         let registry = Arc::new(test_registry());
         let visited = Arc::new(Mutex::new(Vec::new()));
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
 
         let started = start(
             &run_dir,
@@ -1067,7 +1066,7 @@ mod tests {
                         visited.lock().unwrap().push(node_id.to_string());
                     }
                 })),
-                ..test_start_services(&run_dir, emitter, registry).await
+                ..test_start_services(&store, &run_dir, emitter, registry).await
             },
         )
         .await
@@ -1084,8 +1083,8 @@ mod tests {
         let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
         let registry = Arc::new(test_registry());
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
-        let services = test_start_services(&run_dir, emitter, registry).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let services = test_start_services(&store, &run_dir, emitter, registry).await;
 
         // Write a checkpoint to the store (not disk) so start() sees it
         let checkpoint = Checkpoint::from_context(
@@ -1121,11 +1120,11 @@ mod tests {
         let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
         let registry = Arc::new(test_registry());
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
 
         let result = resume(
             &run_dir,
-            test_start_services(&run_dir, emitter, registry).await,
+            test_start_services(&store, &run_dir, emitter, registry).await,
         )
         .await;
 
@@ -1143,7 +1142,7 @@ mod tests {
         let emitter = Arc::new(EventEmitter::new(fixtures::RUN_1));
         let registry = Arc::new(test_registry());
 
-        persisted_workflow(MINIMAL_DOT, &run_dir).await;
+        let (_persisted, store) = persisted_workflow(MINIMAL_DOT, &run_dir).await;
 
         let checkpoint = Checkpoint::from_context(
             &Context::new(),
@@ -1179,7 +1178,7 @@ mod tests {
 
         let result = resume(
             &run_dir,
-            test_start_services(&run_dir, emitter, registry).await,
+            test_start_services(&store, &run_dir, emitter, registry).await,
         )
         .await;