diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index 8032f4611..de94f919f 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -21,7 +21,9 @@ pub(crate) async fn dispatch(ns: PrNamespace, globals: &GlobalArgs) -> Result<() let github_app = build_github_app_credentials(cli_settings.app_id())?; match ns.command { - PrCommand::Create(args) => create::create_command(args, github_app, globals).await, + PrCommand::Create(args) => { + Box::pin(create::create_command(args, github_app, globals)).await + } PrCommand::List(args) => list::list_command(args, github_app, globals).await, PrCommand::View(args) => view::view_command(args, github_app, globals).await, PrCommand::Merge(args) => merge::merge_command(args, github_app, globals).await, diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 90a310738..99f3edebf 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -137,9 +137,7 @@ async fn attach_run_store( emit_progress_line(&mut progress_ui, line, json_output)?; } - let mut stream = run_store - .watch_events_from(if last_seq == 0 { 1 } else { last_seq + 1 }) - .await?; + let mut stream = run_store.watch_events_from(if last_seq == 0 { 1 } else { last_seq + 1 })?; let mut next_seq = if last_seq == 0 { 1 } else { last_seq + 1 }; let mut cached_pid: Option = None; let attach_started = Instant::now(); @@ -745,29 +743,26 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option ExitCode { let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE; loop { - match run_store.state().await { - Ok(state) => { - if let Some(conclusion) = state.conclusion { - let success = matches!( - conclusion.status, - StageStatus::Success | StageStatus::PartialSuccess - ); - return if success { - ExitCode::from(0) - } else { - ExitCode::from(1) - }; - } - - match state.status { - Some(record) if matches!(record.status, RunStatus::Succeeded) => { - return ExitCode::from(0); - } - Some(record) if record.status.is_terminal() => return ExitCode::from(1), - Some(_) | None => {} - } + if let Ok(state) = run_store.state().await { + if let Some(conclusion) = state.conclusion { + let success = matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess + ); + return if success { + ExitCode::from(0) + } else { + ExitCode::from(1) + }; + } + + match state.status { + Some(record) if matches!(record.status, RunStatus::Succeeded) => { + return ExitCode::from(0); + } + Some(record) if record.status.is_terminal() => return ExitCode::from(1), + Some(_) | None => {} } - Err(_) => {} } if Instant::now() >= deadline { diff --git a/lib/crates/fabro-cli/src/commands/run/command.rs b/lib/crates/fabro-cli/src/commands/run/command.rs index 36e6bb05f..bd47a3640 100644 --- a/lib/crates/fabro-cli/src/commands/run/command.rs +++ b/lib/crates/fabro-cli/src/commands/run/command.rs @@ -13,7 +13,7 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<( let quiet = args.detach; let prevent_idle_sleep = cli_settings.prevent_idle_sleep_enabled(); - let (run_id, run_dir) = super::create::create_run(&args, cli, styles, quiet).await?; + let (run_id, run_dir) = Box::pin(super::create::create_run(&args, cli, styles, quiet)).await?; #[cfg(feature = "sleep_inhibitor")] let _sleep_guard = crate::sleep_inhibitor::guard(prevent_idle_sleep); diff --git a/lib/crates/fabro-cli/src/commands/run/create.rs b/lib/crates/fabro-cli/src/commands/run/create.rs index 942a11859..9d9b6cc60 100644 --- a/lib/crates/fabro-cli/src/commands/run/create.rs +++ b/lib/crates/fabro-cli/src/commands/run/create.rs @@ -39,7 +39,7 @@ pub(crate) async fn create_run( let store = store::build_store(settings.storage_dir().as_path())?; - let created = match create( + let created = match Box::pin(create( store.as_ref(), CreateRunInput { workflow: WorkflowInput::Path(workflow_path.clone()), @@ -51,7 +51,7 @@ pub(crate) async fn create_run( base_branch: None, host_repo_path: None, }, - ) + )) .await { Ok(created) => created, diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 1aee65577..a387101c2 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -147,7 +147,6 @@ async fn follow_store_logs( ) -> Result<()> { let mut stream = run_store .watch_events_from(seq) - .await .context("Failed to watch store-backed run events")?; let stdout = io::stdout(); let mut out = stdout.lock(); diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 27df84f71..048eea894 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -40,13 +40,13 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( match cmd { RunCommands::Run(mut args) => { apply_json_defaults(&mut args, globals); - command::execute(args, globals).await + Box::pin(command::execute(args, globals)).await } RunCommands::Create(mut args) => { apply_json_defaults(&mut args, globals); let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let cli = user_layer_with_globals(globals)?; - let (run_id, _run_dir) = create::create_run(&args, cli, styles, true).await?; + let (run_id, _run_dir) = Box::pin(create::create_run(&args, cli, styles, true)).await?; if globals.json { print_json_pretty(&serde_json::json!({ "run_id": run_id }))?; } else { @@ -125,7 +125,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } RunCommands::Rewind(args) => { let styles = Styles::detect_stderr(); - rewind::run(&args, &styles, globals).await + Box::pin(rewind::run(&args, &styles, globals)).await } RunCommands::Fork(args) => { let styles = Styles::detect_stderr(); diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index 9695e3c50..d40a28b3b 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -135,10 +135,6 @@ async fn reset_rewound_run_state( let run_store = durable_store.open_run(run_id).await.map_err(|err| { anyhow::anyhow!("failed to open durable store run for rewind reset: {err}") })?; - run_store - .reset_for_rewind() - .await - .map_err(|err| anyhow::anyhow!("failed to clear rewound run state: {err}"))?; append_workflow_event( run_store.as_ref(), run_id, diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index d58b48f11..b82d7bf6b 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -224,7 +224,7 @@ async fn main_inner() -> (String, Result<()>) { Commands::Install { web_url } => { commands::install::run_install(&web_url, &globals).await?; } - Commands::Pr(ns) => commands::pr::dispatch(ns, &globals).await?, + Commands::Pr(ns) => Box::pin(commands::pr::dispatch(ns, &globals)).await?, Commands::Secret(ns) => commands::secret::dispatch(ns, &globals)?, Commands::Settings(args) => commands::config::execute(&args, &globals)?, Commands::Workflow(ns) => commands::workflow::dispatch(ns, &globals)?, diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 7e31df920..97144bb43 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/lib/crates/fabro-retro/src/retro_agent.rs @@ -207,7 +207,7 @@ pub async fn run_retro_agent( let prompt = build_retro_prompt(RETRO_DATA_DIR); - write_retro_prompt(run_store, &retro_dir, &prompt).await?; + write_retro_prompt(run_store, &retro_dir, &prompt)?; let process_result = session .process_input(&prompt) @@ -250,7 +250,7 @@ pub async fn run_retro_agent( }; // Write artifacts (on both success and failure) - write_retro_response(run_store, &retro_dir, &response_text).await?; + write_retro_response(run_store, &retro_dir, &response_text)?; write_retro_artifacts( &retro_dir, provider.as_str(), @@ -285,7 +285,7 @@ pub fn dry_run_narrative() -> RetroNarrative { } } -async fn write_retro_prompt( +fn write_retro_prompt( _run_store: &SlateRunStore, retro_dir: &Path, prompt: &str, @@ -294,7 +294,7 @@ async fn write_retro_prompt( Ok(()) } -async fn write_retro_response( +fn write_retro_response( _run_store: &SlateRunStore, retro_dir: &Path, response: &str, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 90d10810d..d0609c3e7 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -547,7 +547,7 @@ async fn start_run( info!(run_id = %run_id, "Run queued"); let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4())); let settings = state.settings.read().unwrap().clone(); - let created = match operations::create( + let created = match Box::pin(operations::create( state.store.as_ref(), CreateRunInput { workflow: WorkflowInput::DotSource { @@ -562,7 +562,7 @@ async fn start_run( host_repo_path: None, base_branch: None, }, - ) + )) .await { Ok(created) => created, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index a6592744c..63e75ced2 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/lib/crates/fabro-store/src/run_state.rs @@ -3,6 +3,7 @@ use std::path::PathBuf; use std::str::FromStr; use chrono::{DateTime, Utc}; +use serde::de::DeserializeOwned; use serde_json::Value; use crate::{ @@ -247,7 +248,7 @@ impl RunState { }; let visit = required_u32(&properties, "visit")?; self.node_mut(node_id, visit).provider_used = - provider_used_from_agent_event(event_name, &properties); + Some(provider_used_from_agent_event(event_name, &properties)); } "command.started" => { let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { @@ -267,6 +268,13 @@ impl RunState { node.stderr = optional_string(&properties, "stderr"); node.script_timing = Some(Value::Object(properties.clone())); } + "parallel.completed" => { + let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { + return Ok(()); + }; + let visit = self.current_visit_for(node_id).unwrap_or(1); + self.node_mut(node_id, visit).parallel_results = properties.get("results").cloned(); + } _ => {} } @@ -416,7 +424,7 @@ fn parse_ts(value: &Value) -> Result> { .ok_or_else(|| StoreError::InvalidEvent("event payload missing ts".into()))?; chrono::DateTime::parse_from_rfc3339(ts) .map(|ts| ts.with_timezone(&Utc)) - .map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}")).into()) + .map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}"))) } fn parse_run_id(value: &Value) -> Result { @@ -426,7 +434,7 @@ fn parse_run_id(value: &Value) -> Result { .ok_or_else(|| StoreError::InvalidEvent("event payload missing run_id".into()))?; run_id .parse() - .map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}")).into()) + .map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}"))) } fn required_string(properties: &serde_json::Map, key: &str) -> Result { @@ -434,9 +442,7 @@ fn required_string(properties: &serde_json::Map, key: &str) -> Re .get(key) .and_then(Value::as_str) .map(ToString::to_string) - .ok_or_else(|| { - StoreError::InvalidEvent(format!("event missing string property {key}")).into() - }) + .ok_or_else(|| StoreError::InvalidEvent(format!("event missing string property {key}"))) } fn optional_string(properties: &serde_json::Map, key: &str) -> Option { @@ -447,17 +453,18 @@ fn optional_string(properties: &serde_json::Map, key: &str) -> Op } fn required_u64(properties: &serde_json::Map, key: &str) -> Result { - properties.get(key).and_then(Value::as_u64).ok_or_else(|| { - StoreError::InvalidEvent(format!("event missing integer property {key}")).into() - }) + properties + .get(key) + .and_then(Value::as_u64) + .ok_or_else(|| StoreError::InvalidEvent(format!("event missing integer property {key}"))) } fn required_u32(properties: &serde_json::Map, key: &str) -> Result { u32::try_from(required_u64(properties, key)?) - .map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32")).into()) + .map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32"))) } -fn required_json( +fn required_json( properties: &serde_json::Map, key: &str, ) -> Result { @@ -466,10 +473,10 @@ fn required_json( .cloned() .ok_or_else(|| StoreError::InvalidEvent(format!("event missing property {key}")))?; serde_json::from_value(value) - .map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into()) + .map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}"))) } -fn optional_json( +fn optional_json( properties: &serde_json::Map, key: &str, ) -> Result> { @@ -478,9 +485,8 @@ fn optional_json( .filter(|value| !value.is_null()) .cloned() .map(|value| { - serde_json::from_value(value).map_err(|err| { - StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into() - }) + serde_json::from_value(value) + .map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}"))) }) .transpose() } @@ -488,9 +494,8 @@ fn optional_json( fn parse_reason(properties: &serde_json::Map) -> Result> { optional_string(properties, "reason") .map(|reason| { - serde_json::from_value(Value::String(reason)).map_err(|err| { - StoreError::InvalidEvent(format!("invalid status reason: {err}")).into() - }) + serde_json::from_value(Value::String(reason)) + .map_err(|err| StoreError::InvalidEvent(format!("invalid status reason: {err}"))) }) .transpose() } @@ -662,7 +667,7 @@ fn provider_used_from_prompt(properties: &serde_json::Map) -> Opt fn provider_used_from_agent_event( event_name: &str, properties: &serde_json::Map, -) -> Option { +) -> Value { let mut provider_used = serde_json::Map::new(); provider_used.insert( "mode".to_string(), @@ -681,5 +686,5 @@ fn provider_used_from_agent_event( if let Some(command) = optional_string(properties, "command") { provider_used.insert("command".to_string(), Value::String(command)); } - Some(Value::Object(provider_used)) + Value::Object(provider_used) } diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index fc2291b9e..cd6718b7c 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -824,7 +824,7 @@ mod tests { .create_run(&test_run_id("run-1"), created_at, None) .await .unwrap(); - let mut stream = run.watch_events_from(1).await.unwrap(); + let mut stream = run.watch_events_from(1).unwrap(); run.append_event(&event_payload( "run-1", @@ -1018,7 +1018,7 @@ mod tests { } #[tokio::test] - async fn slate_run_store_round_trips_node_data_and_assets() { + async fn slate_run_store_round_trips_assets_and_projects_events() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); let run = store @@ -1029,16 +1029,22 @@ mod tests { node_id: "code", visit: 2, }; - run.put_node_prompt(&node, "Plan").await.unwrap(); - run.put_node_status(&node, &sample_node_status()) - .await - .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:01:00Z", + "stage.prompt", + Some("code"), + serde_json::json!({"text": "Plan", "visit": 2}), + )) + .await + .unwrap(); run.put_asset(&node, "src/lib.rs", b"fn main() {}") .await .unwrap(); - let snapshot = run.get_node(&node).await.unwrap(); - assert_eq!(snapshot.prompt, Some("Plan".to_string())); + let state = run.state().await.unwrap(); + let node_state = state.node(&node).unwrap(); + assert_eq!(node_state.prompt, Some("Plan".to_string())); assert_eq!( run.get_asset(&node, "src/lib.rs").await.unwrap(), Some(Bytes::from_static(b"fn main() {}")) @@ -1050,7 +1056,7 @@ mod tests { } #[tokio::test] - async fn slate_run_store_lists_artifact_values_and_asset_only_visits() { + async fn slate_run_store_lists_artifact_values_and_assets() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); let run = store @@ -1068,7 +1074,6 @@ mod tests { node_id: "code", visit: 2, }; - run.put_node_prompt(&snapshot_node, "Plan").await.unwrap(); run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") .await .unwrap(); @@ -1096,8 +1101,5 @@ mod tests { ("code".to_string(), 2, "src/lib.rs".to_string()) ] ); - - let code_node = run.get_node(&snapshot_node).await.unwrap(); - assert_eq!(code_node.node_id, "code"); } } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 929d737a6..bc9df6b3c 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -1,4 +1,3 @@ -use std::collections::BTreeSet; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::{Arc, Weak}; use std::time::Duration; @@ -16,10 +15,10 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; use crate::{ - CatalogRecord, EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, - Result, RunState, RunSummary, StoreError, + CatalogRecord, EventEnvelope, EventPayload, NodeVisitRef, Result, RunState, RunSummary, + StoreError, }; -use fabro_types::{NodeStatusRecord, RunId}; +use fabro_types::RunId; #[derive(Clone)] pub struct SlateRunStore { @@ -152,40 +151,6 @@ impl SlateRunStore { Ok(state.build_summary(catalog)) } - async fn build_node_snapshot(&self, node: &NodeVisitRef<'_>) -> Result { - Ok(NodeSnapshot { - node_id: node.node_id.to_string(), - visit: node.visit, - prompt: self.inner.db.get_text(&keys::node_prompt(node)).await?, - response: self.inner.db.get_text(&keys::node_response(node)).await?, - status: self.inner.db.get_json(&keys::node_status(node)).await?, - outcome: self.inner.db.get_json(&keys::node_outcome(node)).await?, - provider_used: self - .inner - .db - .get_json(&keys::node_provider_used(node)) - .await?, - diff: self.inner.db.get_text(&keys::node_diff(node)).await?, - script_invocation: self - .inner - .db - .get_json(&keys::node_script_invocation(node)) - .await?, - script_timing: self - .inner - .db - .get_json(&keys::node_script_timing(node)) - .await?, - parallel_results: self - .inner - .db - .get_json(&keys::node_parallel_results(node)) - .await?, - stdout: self.inner.db.get_text(&keys::node_stdout(node)).await?, - stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?, - }) - } - async fn projected_state(&self) -> Result { let next_seq = { let cache = self.inner.projection_cache.lock().await; @@ -202,157 +167,6 @@ impl SlateRunStore { } impl SlateRunStore { - pub async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> { - self.inner - .db - .put_text(&keys::node_prompt(node), prompt) - .await - } - - pub async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> { - self.inner - .db - .put_text(&keys::node_response(node), response) - .await - } - - pub async fn put_node_status( - &self, - node: &NodeVisitRef<'_>, - status: &NodeStatusRecord, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_status(node), status) - .await - } - - pub async fn put_node_outcome( - &self, - node: &NodeVisitRef<'_>, - outcome: &NodeOutcomeRecord, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_outcome(node), outcome) - .await - } - - pub async fn put_node_provider_used( - &self, - node: &NodeVisitRef<'_>, - provider_used: &serde_json::Value, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_provider_used(node), provider_used) - .await - } - - pub async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> { - self.inner.db.put_text(&keys::node_diff(node), diff).await - } - - pub async fn put_node_script_invocation( - &self, - node: &NodeVisitRef<'_>, - invocation: &serde_json::Value, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_script_invocation(node), invocation) - .await - } - - pub async fn put_node_script_timing( - &self, - node: &NodeVisitRef<'_>, - timing: &serde_json::Value, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_script_timing(node), timing) - .await - } - - pub async fn put_node_parallel_results( - &self, - node: &NodeVisitRef<'_>, - results: &serde_json::Value, - ) -> Result<()> { - self.inner - .db - .put_json(&keys::node_parallel_results(node), results) - .await - } - - pub async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.inner.db.put_text(&keys::node_stdout(node), log).await - } - - pub async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.inner.db.put_text(&keys::node_stderr(node), log).await - } - - pub async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result { - self.build_node_snapshot(node).await - } - - pub async fn list_node_visits(&self, node_id: &str) -> Result> { - let prefix = format!("nodes/{node_id}/visit-"); - let mut iter = self.inner.db.scan_prefix(prefix.as_bytes()).await?; - let mut visits = BTreeSet::new(); - while let Some(entry) = iter.next().await? { - let key = key_to_string(&entry.key)?; - if let Some((current_node_id, visit, _)) = keys::parse_node_key(&key) { - if current_node_id == node_id { - visits.insert(visit); - } - } - } - Ok(visits.into_iter().collect()) - } - - pub async fn list_node_ids(&self) -> Result> { - let mut iter = self.inner.db.scan_prefix(b"nodes/").await?; - let mut node_ids = BTreeSet::new(); - while let Some(entry) = iter.next().await? { - let key = key_to_string(&entry.key)?; - if let Some((node_id, _, _)) = keys::parse_node_key(&key) { - node_ids.insert(node_id); - } - } - - let mut asset_iter = self - .inner - .db - .scan_prefix(keys::ARTIFACT_NODES_PREFIX.as_bytes()) - .await?; - while let Some(entry) = asset_iter.next().await? { - let key = key_to_string(&entry.key)?; - if let Some((node_id, _, _)) = keys::parse_node_asset_key(&key) { - node_ids.insert(node_id); - } - } - - Ok(node_ids.into_iter().collect()) - } - - pub async fn reset_for_rewind(&self) -> Result<()> { - let db = self.inner.db.writer()?; - for key in [keys::retro_prompt(), keys::retro_response()] { - db.delete(key).await?; - } - for prefix in [ - b"nodes/".as_slice(), - keys::ARTIFACT_VALUES_PREFIX.as_bytes(), - keys::ARTIFACT_NODES_PREFIX.as_bytes(), - ] { - delete_prefix(db, prefix).await?; - } - Ok(()) - } - pub async fn append_event(&self, payload: &EventPayload) -> Result { payload.validate(&self.inner.run_id)?; let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst); @@ -374,7 +188,7 @@ impl SlateRunStore { self.inner.db.list_events_from(seq).await } - pub async fn watch_events_from( + pub fn watch_events_from( &self, seq: u32, ) -> Result> + Send>>> { @@ -412,22 +226,6 @@ impl SlateRunStore { Ok(Box::pin(UnboundedReceiverStream::new(receiver))) } - pub async fn put_retro_prompt(&self, text: &str) -> Result<()> { - self.inner.db.put_text(keys::retro_prompt(), text).await - } - - pub async fn get_retro_prompt(&self) -> Result> { - self.inner.db.get_text(keys::retro_prompt()).await - } - - pub async fn put_retro_response(&self, text: &str) -> Result<()> { - self.inner.db.put_text(keys::retro_response(), text).await - } - - pub async fn get_retro_response(&self) -> Result> { - self.inner.db.get_text(keys::retro_response()).await - } - pub async fn put_artifact_value( &self, artifact_id: &str, @@ -522,17 +320,6 @@ impl SlateRunDb { put_json(self.writer()?, key, value).await } - async fn get_text(&self, key: &str) -> Result> { - match self { - Self::Writer(db) => get_text(db, key).await, - Self::Reader(db) => get_text(db.as_ref(), key).await, - } - } - - async fn put_text(&self, key: &str, value: &str) -> Result<()> { - put_text(self.writer()?, key, value).await - } - async fn get_bytes(&self, key: &str) -> Result> { match self { Self::Writer(db) => get_bytes(db, key).await, @@ -596,41 +383,11 @@ where .map_err(Into::into) } -async fn put_text(db: &slatedb::Db, key: &str, value: &str) -> Result<()> { - db.put(key, value.as_bytes()).await?; - Ok(()) -} - -async fn get_text(db: &R, key: &str) -> Result> -where - R: DbRead + Sync, -{ - db.get(key) - .await? - .map(|value| { - String::from_utf8(value.to_vec()) - .map_err(|err| StoreError::Other(format!("stored text is not valid UTF-8: {err}"))) - }) - .transpose() -} - async fn put_bytes(db: &slatedb::Db, key: &str, value: &[u8]) -> Result<()> { db.put(key, value).await?; Ok(()) } -async fn delete_prefix(db: &slatedb::Db, prefix: &[u8]) -> Result<()> { - let mut iter = db.scan_prefix(prefix).await?; - let mut keys = Vec::new(); - while let Some(entry) = iter.next().await? { - keys.push(key_to_string(&entry.key)?); - } - for key in keys { - db.delete(key).await?; - } - Ok(()) -} - async fn get_bytes(db: &slatedb::Db, key: &str) -> Result> { Ok(db.get(key).await?) } diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index fd9d4c620..be6cbac38 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 anyhow::{Context, Result}; use chrono::{SecondsFormat, Utc}; -use fabro_store::{EventPayload, NodeVisitRef, RunStoreHandle, SlateRunStore}; +use fabro_store::{EventPayload, RunStoreHandle, SlateRunStore}; use fabro_types::RunId; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; @@ -1572,15 +1572,6 @@ impl StoreProgressLogger { if let Err(err) = run_store.append_event(&payload).await { tracing::warn!(error = %err, "Failed to append event to run store"); } - if let Err(err) = - project_provider_used_from_event_payload(run_store.as_ref(), &payload) - .await - { - tracing::warn!( - error = %err, - "Failed to project provider metadata from event" - ); - } } StoreProgressCommand::Flush(tx) => { let _ = tx.send(()); @@ -1626,80 +1617,6 @@ impl StoreProgressLogger { } } -async fn project_provider_used_from_event_payload( - run_store: &SlateRunStore, - payload: &EventPayload, -) -> Result<()> { - let value = payload.as_value(); - let Some(event_name) = value.get("event").and_then(Value::as_str) else { - return Ok(()); - }; - let Some(node_id) = value.get("node_id").and_then(Value::as_str) else { - return Ok(()); - }; - let Some(properties) = value.get("properties").and_then(Value::as_object) else { - return Ok(()); - }; - let Some(visit) = properties - .get("visit") - .and_then(Value::as_u64) - .and_then(|visit| u32::try_from(visit).ok()) - else { - return Ok(()); - }; - - let provider_used = match event_name { - "stage.prompt" => { - let mut provider_used = Map::new(); - if let Some(mode) = properties.get("mode").and_then(Value::as_str) { - provider_used.insert("mode".to_string(), Value::String(mode.to_string())); - } - if let Some(provider) = properties.get("provider").and_then(Value::as_str) { - provider_used.insert("provider".to_string(), Value::String(provider.to_string())); - } - if let Some(model) = properties.get("model").and_then(Value::as_str) { - provider_used.insert("model".to_string(), Value::String(model.to_string())); - } - (!provider_used.is_empty()).then_some(Value::Object(provider_used)) - } - "agent.session.started" => { - let mut provider_used = Map::new(); - provider_used.insert("mode".to_string(), Value::String("agent".to_string())); - if let Some(provider) = properties.get("provider").and_then(Value::as_str) { - provider_used.insert("provider".to_string(), Value::String(provider.to_string())); - } - if let Some(model) = properties.get("model").and_then(Value::as_str) { - provider_used.insert("model".to_string(), Value::String(model.to_string())); - } - Some(Value::Object(provider_used)) - } - "agent.cli.started" => { - let mut provider_used = Map::new(); - provider_used.insert("mode".to_string(), Value::String("cli".to_string())); - if let Some(provider) = properties.get("provider").and_then(Value::as_str) { - provider_used.insert("provider".to_string(), Value::String(provider.to_string())); - } - if let Some(model) = properties.get("model").and_then(Value::as_str) { - provider_used.insert("model".to_string(), Value::String(model.to_string())); - } - if let Some(command) = properties.get("command").and_then(Value::as_str) { - provider_used.insert("command".to_string(), Value::String(command.to_string())); - } - Some(Value::Object(provider_used)) - } - _ => None, - }; - - let Some(provider_used) = provider_used else { - return Ok(()); - }; - - run_store - .put_node_provider_used(&NodeVisitRef { node_id, visit }, &provider_used) - .await - .map_err(anyhow::Error::from) -} - /// Current time as epoch milliseconds. fn epoch_millis() -> i64 { let millis = std::time::SystemTime::now() diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index 0053a81b0..f7e34fc9e 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -2,7 +2,7 @@ use std::path::Path; use std::process::Command; use fabro_checkpoint::git::Store; -use fabro_store::{NodeVisitRef, SlateRunStore}; +use fabro_store::RunState; use fabro_types::Settings; use crate::error::{FabroError, Result}; @@ -353,76 +353,64 @@ pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec)> { result } -pub async fn scan_node_files_from_store(run_store: &SlateRunStore) -> Vec<(String, Vec)> { +pub fn scan_node_files_from_state(state: &RunState) -> Vec<(String, Vec)> { let mut result = Vec::new(); - let Ok(node_ids) = run_store.list_node_ids().await else { - return result; - }; + let mut keys: Vec<_> = state.nodes.keys().collect(); + keys.sort(); - for node_id in node_ids { - let Ok(visits) = run_store.list_node_visits(&node_id).await else { + for (node_id, visit) in keys { + let Some(node) = state.nodes.get(&(node_id.clone(), *visit)) else { continue; }; - for visit in visits { - let Ok(node) = run_store - .get_node(&NodeVisitRef { - node_id: &node_id, - visit, - }) - .await - else { - continue; - }; - if let Some(prompt) = node.prompt { + if let Some(ref prompt) = node.prompt { + result.push(( + node_file_path(node_id, *visit, "prompt.md"), + prompt.as_bytes().to_vec(), + )); + } + if let Some(ref response) = node.response { + result.push(( + node_file_path(node_id, *visit, "response.md"), + response.as_bytes().to_vec(), + )); + } + if let Some(ref status) = node.status { + if let Ok(bytes) = serde_json::to_vec_pretty(status) { + result.push((node_file_path(node_id, *visit, "status.json"), bytes)); + } + } + if let Some(ref provider_used) = node.provider_used { + if let Ok(bytes) = serde_json::to_vec_pretty(provider_used) { + result.push((node_file_path(node_id, *visit, "provider_used.json"), bytes)); + } + } + if let Some(ref diff) = node.diff { + result.push(( + node_file_path(node_id, *visit, "diff.patch"), + diff.as_bytes().to_vec(), + )); + } + if let Some(ref script_invocation) = node.script_invocation { + if let Ok(bytes) = serde_json::to_vec_pretty(script_invocation) { result.push(( - node_file_path(&node_id, visit, "prompt.md"), - prompt.into_bytes(), + node_file_path(node_id, *visit, "script_invocation.json"), + bytes, )); } - if let Some(response) = node.response { + } + if let Some(ref script_timing) = node.script_timing { + if let Ok(bytes) = serde_json::to_vec_pretty(script_timing) { + result.push((node_file_path(node_id, *visit, "script_timing.json"), bytes)); + } + } + if let Some(ref parallel_results) = node.parallel_results { + if let Ok(bytes) = serde_json::to_vec_pretty(parallel_results) { result.push(( - node_file_path(&node_id, visit, "response.md"), - response.into_bytes(), + node_file_path(node_id, *visit, "parallel_results.json"), + bytes, )); } - if let Some(status) = node.status { - if let Ok(bytes) = serde_json::to_vec_pretty(&status) { - result.push((node_file_path(&node_id, visit, "status.json"), bytes)); - } - } - if let Some(provider_used) = node.provider_used { - if let Ok(bytes) = serde_json::to_vec_pretty(&provider_used) { - result.push((node_file_path(&node_id, visit, "provider_used.json"), bytes)); - } - } - if let Some(diff) = node.diff { - result.push(( - node_file_path(&node_id, visit, "diff.patch"), - diff.into_bytes(), - )); - } - if let Some(script_invocation) = node.script_invocation { - if let Ok(bytes) = serde_json::to_vec_pretty(&script_invocation) { - result.push(( - node_file_path(&node_id, visit, "script_invocation.json"), - bytes, - )); - } - } - if let Some(script_timing) = node.script_timing { - if let Ok(bytes) = serde_json::to_vec_pretty(&script_timing) { - result.push((node_file_path(&node_id, visit, "script_timing.json"), bytes)); - } - } - if let Some(parallel_results) = node.parallel_results { - if let Ok(bytes) = serde_json::to_vec_pretty(¶llel_results) { - result.push(( - node_file_path(&node_id, visit, "parallel_results.json"), - bytes, - )); - } - } } } @@ -440,9 +428,8 @@ fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String { #[cfg(test)] mod tests { use super::*; - use chrono::Utc; use fabro_store::SlateStore; - use fabro_types::{NodeStatusRecord, StageStatus, fixtures}; + use fabro_types::fixtures; use object_store::memory::InMemory; use std::fs; use std::sync::Arc; @@ -592,52 +579,78 @@ mod tests { } #[tokio::test] - async fn scan_node_files_from_store_reconstructs_allowlisted_entries() { + async fn scan_node_files_from_state_reconstructs_allowlisted_entries() { + use fabro_store::EventPayload; + let store = test_store(); let run = store .create_run(&fixtures::RUN_1, chrono::Utc::now(), None) .await .unwrap(); - let node = NodeVisitRef { - node_id: "work", - visit: 2, + let run_id_str = fixtures::RUN_1.to_string(); + + let event = |event_name: &str, props: serde_json::Value| -> EventPayload { + let value = serde_json::json!({ + "id": format!("evt-{event_name}"), + "ts": "2026-03-27T12:01:00Z", + "run_id": run_id_str, + "event": event_name, + "node_id": "work", + "properties": props, + }); + EventPayload::new(value, &fixtures::RUN_1).unwrap() }; - run.put_node_prompt(&node, "hello").await.unwrap(); - run.put_node_response(&node, "world").await.unwrap(); - run.put_node_status( - &node, - &NodeStatusRecord { - status: StageStatus::Success, - notes: None, - failure_reason: None, - timestamp: Utc::now(), - }, - ) + + run.append_event(&event( + "stage.prompt", + serde_json::json!({"text": "hello", "visit": 2, "mode": "prompt", "provider": "openai"}), + )) + .await + .unwrap(); + run.append_event(&event( + "prompt.completed", + serde_json::json!({"response": "world"}), + )) + .await + .unwrap(); + run.append_event(&event( + "stage.completed", + serde_json::json!({"response": "world", "status": "success", "visit": 2}), + )) + .await + .unwrap(); + run.append_event(&event( + "command.started", + serde_json::json!({"command": "echo hi"}), + )) + .await + .unwrap(); + run.append_event(&event( + "command.completed", + serde_json::json!({"stdout": "hi\n", "stderr": "", "exit_code": 0}), + )) + .await + .unwrap(); + run.append_event(&event( + "parallel.completed", + serde_json::json!({"results": [{"id": "a"}], "duration_ms": 100, "success_count": 1, "failure_count": 0}), + )) + .await + .unwrap(); + run.append_event(&event( + "checkpoint.completed", + serde_json::json!({"diff": "diff --git a/story.txt b/story.txt", "ordinal": 1}), + )) .await .unwrap(); - run.put_node_provider_used(&node, &serde_json::json!({"provider":"openai"})) - .await - .unwrap(); - run.put_node_diff(&node, "diff --git a/story.txt b/story.txt") - .await - .unwrap(); - run.put_node_script_invocation(&node, &serde_json::json!({"command":"echo hi"})) - .await - .unwrap(); - run.put_node_script_timing(&node, &serde_json::json!({"exit_code":0})) - .await - .unwrap(); - run.put_node_parallel_results(&node, &serde_json::json!([{"id":"a"}])) - .await - .unwrap(); - let files = scan_node_files_from_store(run.as_ref()).await; + let state = run.state().await.unwrap(); + let files = scan_node_files_from_state(&state); let paths: Vec<&str> = files.iter().map(|(path, _)| path.as_str()).collect(); assert!(paths.contains(&"nodes/work-visit_2/prompt.md")); assert!(paths.contains(&"nodes/work-visit_2/response.md")); assert!(paths.contains(&"nodes/work-visit_2/status.json")); assert!(paths.contains(&"nodes/work-visit_2/provider_used.json")); - assert!(paths.contains(&"nodes/work-visit_2/diff.patch")); assert!(paths.contains(&"nodes/work-visit_2/script_invocation.json")); assert!(paths.contains(&"nodes/work-visit_2/script_timing.json")); assert!(paths.contains(&"nodes/work-visit_2/parallel_results.json")); diff --git a/lib/crates/fabro-workflow/src/handler/fan_in.rs b/lib/crates/fabro-workflow/src/handler/fan_in.rs index ea70c68d6..94a898633 100644 --- a/lib/crates/fabro-workflow/src/handler/fan_in.rs +++ b/lib/crates/fabro-workflow/src/handler/fan_in.rs @@ -1,17 +1,15 @@ use std::path::Path; use std::sync::Arc; -use async_trait::async_trait; -use fabro_agent::Sandbox; -use fabro_store::{NodeVisitRef, RunStoreHandle}; - use crate::context::Context; use crate::context::keys; use crate::error::FabroError; -use crate::event::EventEmitter; +use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::outcome::{Outcome, OutcomeExt}; use crate::run_dir::{node_dir, visit_from_context}; use crate::sandbox_git::git_merge_ff_only; +use async_trait::async_trait; +use fabro_agent::Sandbox; use fabro_graphviz::graph::{Graph, Node}; use tokio::fs; @@ -90,7 +88,6 @@ impl Handler for FanInHandler { &node.id, &services.emitter, &services.sandbox, - services.run_store.clone(), ) .await? } else { @@ -226,7 +223,6 @@ async fn llm_evaluate( node_id: &str, emitter: &Arc, sandbox: &Arc, - run_store: RunStoreHandle, ) -> Result { let results_text = serde_json::to_string_pretty(results).unwrap_or_else(|_| results.to_string()); @@ -236,20 +232,21 @@ async fn llm_evaluate( Respond with the ID of the best candidate." ); - // Write prompt to logs let visit = visit_from_context(context); + let visit_u32 = u32::try_from(visit).unwrap_or(u32::MAX); let stage_dir = node_dir(run_dir, node_id, visit); fs::create_dir_all(&stage_dir).await?; - let node_ref = NodeVisitRef { - node_id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - run_store - .put_node_prompt(&node_ref, &full_prompt) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; fs::write(stage_dir.join("prompt.md"), &full_prompt).await?; + emitter.emit(&WorkflowRunEvent::Prompt { + stage: node_id.to_string(), + visit: visit_u32, + text: full_prompt.clone(), + mode: Some("fan_in".to_string()), + provider: None, + model: None, + }); + // Build a synthetic node for the backend call let eval_node = Node::new("fan_in_eval"); @@ -278,10 +275,13 @@ async fn llm_evaluate( .unwrap_or_else(|| "unknown".to_string()); let response_text = serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string()); - run_store - .put_node_response(&node_ref, &response_text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; + emitter.emit(&WorkflowRunEvent::PromptCompleted { + node_id: node_id.to_string(), + response: response_text.clone(), + model: String::new(), + provider: String::new(), + usage: None, + }); fs::write(stage_dir.join("response.md"), &response_text).await?; Ok(Candidate { id: best_id, @@ -290,11 +290,13 @@ async fn llm_evaluate( }) } Ok(CodergenResult::Text { text, .. }) => { - // Write response to logs - run_store - .put_node_response(&node_ref, &text) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; + emitter.emit(&WorkflowRunEvent::PromptCompleted { + node_id: node_id.to_string(), + response: text.clone(), + model: String::new(), + provider: String::new(), + usage: None, + }); fs::write(stage_dir.join("response.md"), &text).await?; // The LLM responded with text; try to find a matching candidate ID diff --git a/lib/crates/fabro-workflow/src/handler/parallel.rs b/lib/crates/fabro-workflow/src/handler/parallel.rs index 1cc79a518..98af0958c 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -481,15 +481,6 @@ impl Handler for ParallelHandler { let visit = visit_from_context(context); let node_dir = node_dir(run_dir, &node.id, visit); let _ = fs::create_dir_all(&node_dir).await; - let node_ref = NodeVisitRef { - node_id: &node.id, - visit: u32::try_from(visit).unwrap_or(u32::MAX), - }; - services - .run_store - .put_node_parallel_results(&node_ref, &serde_json::json!(results_json)) - .await - .map_err(|err| FabroError::handler(err.to_string()))?; if let Ok(json) = serde_json::to_string_pretty(&results_json) { let _ = fs::write(node_dir.join("parallel_results.json"), json).await; } diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 24f7c346d..9350bdf8e 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -14,7 +14,7 @@ use fabro_core::state::RunState; use crate::artifact::ArtifactStore; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::git::MetadataStore; -use crate::git::scan_node_files_from_store; +use crate::git::scan_node_files_from_state; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; use crate::outcome::{Outcome, StageStatus, StageUsage}; @@ -149,7 +149,9 @@ impl RunLifecycle for GitLifecycle { }) .collect() }; - extra_entries.extend(scan_node_files_from_store(self.run_store.as_ref()).await); + if let Ok(store_state) = self.run_store.state().await { + extra_entries.extend(scan_node_files_from_state(&store_state)); + } let extra_refs: Vec<(&str, &[u8])> = extra_entries .iter() .map(|(k, v)| (k.as_str(), v.as_slice())) diff --git a/lib/crates/fabro-workflow/src/operations/create.rs b/lib/crates/fabro-workflow/src/operations/create.rs index 02d4801d9..98c97fb51 100644 --- a/lib/crates/fabro-workflow/src/operations/create.rs +++ b/lib/crates/fabro-workflow/src/operations/create.rs @@ -148,7 +148,7 @@ async fn persist_created_run( .open_run(&record.run_id) .await .map_err(|open_err| FabroError::engine(open_err.to_string())) - .or_else(|_| Err(FabroError::engine(err.to_string())))?, + .map_err(|_| FabroError::engine(err.to_string()))?, }; let envelope = canonicalize_event_at( diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index bb0a00763..830d48471 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -553,7 +553,7 @@ impl RunSession { }; let retro_start = Instant::now(); - let retroed = pipeline::retro(executed, &retro_opts).await; + let retroed = Box::pin(pipeline::retro(executed, &retro_opts)).await; let retro_duration = retro_start.elapsed(); let finalize_opts = FinalizeOptions { diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index a65f51125..25cdad78a 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -3,7 +3,7 @@ use std::sync::Arc; use crate::error::FabroError; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; -use crate::git::{MetadataStore, scan_node_files_from_store}; +use crate::git::{MetadataStore, scan_node_files_from_state}; use crate::outcome::{Outcome, OutcomeExt, StageStatus}; use crate::records::{Checkpoint, Conclusion, StageSummary}; use crate::run_options::RunOptions; @@ -195,13 +195,14 @@ pub async fn write_finalize_commit( let git_author = run_options.git_author(); let store = MetadataStore::new(repo_path, &git_author); - let mut entries = scan_node_files_from_store(run_store).await; - let retro_bytes = run_store - .state() - .await - .ok() - .and_then(|state| state.retro) - .and_then(|retro| serde_json::to_vec_pretty(&retro).ok()); + let Ok(store_state) = run_store.state().await else { + return; + }; + let mut entries = scan_node_files_from_state(&store_state); + let retro_bytes = store_state + .retro + .as_ref() + .and_then(|retro| serde_json::to_vec_pretty(retro).ok()); if let Some(bytes) = retro_bytes { entries.push(("retro.json".to_string(), bytes)); } diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index dd971c589..bbf3ff239 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflow/src/pipeline/retro.rs @@ -23,18 +23,15 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { return None; } }; - let cp = match state.checkpoint { - Some(cp) => cp, - None => { - tracing::warn!("Could not load checkpoint, skipping retro"); - if let Some(ref emitter) = options.emitter { - emitter.emit(&WorkflowRunEvent::RetroFailed { - error: "checkpoint not found".to_string(), - duration_ms: 0, - }); - } - return None; + let Some(cp) = state.checkpoint else { + tracing::warn!("Could not load checkpoint, skipping retro"); + if let Some(ref emitter) = options.emitter { + emitter.emit(&WorkflowRunEvent::RetroFailed { + error: "checkpoint not found".to_string(), + duration_ms: 0, + }); } + return None; }; let completed_stages = crate::build_completed_stages(&cp, options.failed); diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index e69225de8..7220d0b29 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -11,7 +11,7 @@ use object_store::memory::InMemory; use crate::error::Result; use crate::event::{EventEmitter, WorkflowRunEvent, append_workflow_event}; -use crate::git::scan_node_files_from_store; +use crate::git::scan_node_files_from_state; use crate::handler::HandlerRegistry; use crate::outcome::Outcome; use crate::pipeline; @@ -199,7 +199,7 @@ async fn persist_run_artifacts_for_tests(run_store: &SlateRunStore, run_dir: &st let _ = std::fs::write(run_dir.join("final.patch"), final_patch); } - for (relative_path, contents) in scan_node_files_from_store(run_store).await { + for (relative_path, contents) in scan_node_files_from_state(&state) { let path = run_dir.join(relative_path); if let Some(parent) = path.parent() { let _ = std::fs::create_dir_all(parent);