From 0945b6b9ba888d899294d06b662917eeb362b1df Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 21:29:30 -0400 Subject: [PATCH] Remove run hydration from production paths --- .../fabro-cli/src/commands/run/rewind.rs | 33 +- lib/crates/fabro-server/src/server.rs | 21 +- .../fabro-workflow/src/operations/hydrate.rs | 326 ------------------ .../fabro-workflow/src/operations/mod.rs | 2 - .../fabro-workflow/src/operations/start.rs | 41 ++- 5 files changed, 64 insertions(+), 359 deletions(-) delete mode 100644 lib/crates/fabro-workflow/src/operations/hydrate.rs diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index d165324e6..f8b297b87 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/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(()) } diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 45c4b70ea..59f6d6ee9 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -682,14 +682,25 @@ async fn execute_run(state: Arc, 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(); diff --git a/lib/crates/fabro-workflow/src/operations/hydrate.rs b/lib/crates/fabro-workflow/src/operations/hydrate.rs deleted file mode 100644 index 8b3b40857..000000000 --- a/lib/crates/fabro-workflow/src/operations/hydrate.rs +++ /dev/null @@ -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, 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::(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, 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, 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, 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, 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, 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, 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, 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()) - ); - } -} diff --git a/lib/crates/fabro-workflow/src/operations/mod.rs b/lib/crates/fabro-workflow/src/operations/mod.rs index abaa993d2..7414107ba 100644 --- a/lib/crates/fabro-workflow/src/operations/mod.rs +++ b/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, }; diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index cac4389fe..1b0eddceb 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/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, registry: Arc, ) -> 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;