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 ae4b7d608..946c9b653 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -208,7 +208,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-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 2a16a2974..c384f7424 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -463,6 +463,13 @@ fn attach_json_errors_without_prompting_for_human_input() { "run_id": "[ULID]", "ts": "[TIMESTAMP]" }, + { + "event": "run.running", + "id": "[EVENT_ID]", + "properties": {}, + "run_id": "[ULID]", + "ts": "[TIMESTAMP]" + }, { "event": "sandbox.initialized", "id": "[EVENT_ID]", diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs index f53fd4fbf..67d013274 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_start.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_start.rs @@ -1,3 +1,4 @@ +#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs index 1a9785060..13039a893 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_status.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_status.rs @@ -1,3 +1,4 @@ +#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs index 42cabafaf..49a6f5c65 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/server_stop.rs @@ -1,3 +1,4 @@ +#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] diff --git a/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs b/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs index 1d23ae7a2..88bdd536d 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/store_dump.rs @@ -56,22 +56,25 @@ fn store_dump_exports_completed_run_snapshot() { ----- stderr ----- "); - assert_snapshot!(dump_file_summary(&output_dir), @r###" + assert_snapshot!(dump_file_summary(&output_dir), @" checkpoint.json - checkpoints/0001.json - checkpoints/0002.json - checkpoints/0003.json + checkpoints/0012.json + checkpoints/0016.json + checkpoints/0020.json conclusion.json events.jsonl graph.fabro + nodes/exit/visit-1/status.json + nodes/report/visit-1/response.md nodes/report/visit-1/status.json + nodes/run_tests/visit-1/response.md nodes/run_tests/visit-1/status.json nodes/start/visit-1/status.json run.json sandbox.json start.json status.json - "###); + "); } #[test] diff --git a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs index 051c4e024..83677825e 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/server_lifecycle.rs @@ -1,3 +1,4 @@ +#[cfg(feature = "server")] use fabro_test::{fabro_snapshot, test_context}; #[test] 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/keys.rs b/lib/crates/fabro-store/src/keys.rs index 302daaa1c..0c8d4b6bf 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -1,8 +1,6 @@ use crate::NodeVisitRef; pub(crate) const INIT_KEY: &str = "_init.json"; -pub(crate) const RETRO_PROMPT_KEY: &str = "retro/prompt.md"; -pub(crate) const RETRO_RESPONSE_KEY: &str = "retro/response.md"; pub(crate) const EVENTS_PREFIX: &str = "events/"; pub(crate) const ARTIFACT_VALUES_PREFIX: &str = "artifacts/values/"; pub(crate) const ARTIFACT_NODES_PREFIX: &str = "artifacts/nodes/"; @@ -11,62 +9,6 @@ pub(crate) fn init() -> &'static str { INIT_KEY } -pub(crate) fn node_visit_prefix(node: &NodeVisitRef<'_>) -> String { - format!("nodes/{}/visit-{}", node.node_id, node.visit) -} - -pub(crate) fn node_prompt(node: &NodeVisitRef<'_>) -> String { - format!("{}/prompt.md", node_visit_prefix(node)) -} - -pub(crate) fn node_response(node: &NodeVisitRef<'_>) -> String { - format!("{}/response.md", node_visit_prefix(node)) -} - -pub(crate) fn node_status(node: &NodeVisitRef<'_>) -> String { - format!("{}/status.json", node_visit_prefix(node)) -} - -pub(crate) fn node_outcome(node: &NodeVisitRef<'_>) -> String { - format!("{}/outcome.json", node_visit_prefix(node)) -} - -pub(crate) fn node_provider_used(node: &NodeVisitRef<'_>) -> String { - format!("{}/provider_used.json", node_visit_prefix(node)) -} - -pub(crate) fn node_diff(node: &NodeVisitRef<'_>) -> String { - format!("{}/diff.patch", node_visit_prefix(node)) -} - -pub(crate) fn node_script_invocation(node: &NodeVisitRef<'_>) -> String { - format!("{}/script_invocation.json", node_visit_prefix(node)) -} - -pub(crate) fn node_script_timing(node: &NodeVisitRef<'_>) -> String { - format!("{}/script_timing.json", node_visit_prefix(node)) -} - -pub(crate) fn node_parallel_results(node: &NodeVisitRef<'_>) -> String { - format!("{}/parallel_results.json", node_visit_prefix(node)) -} - -pub(crate) fn node_stdout(node: &NodeVisitRef<'_>) -> String { - format!("{}/stdout.log", node_visit_prefix(node)) -} - -pub(crate) fn node_stderr(node: &NodeVisitRef<'_>) -> String { - format!("{}/stderr.log", node_visit_prefix(node)) -} - -pub(crate) fn retro_prompt() -> &'static str { - RETRO_PROMPT_KEY -} - -pub(crate) fn retro_response() -> &'static str { - RETRO_RESPONSE_KEY -} - pub(crate) fn event_key(seq: u32, epoch_ms: i64) -> String { format!("{EVENTS_PREFIX}{seq:06}-{epoch_ms}.json") } @@ -96,10 +38,6 @@ pub(crate) fn parse_artifact_value_id(key: &str) -> Option { .map(ToString::to_string) } -pub(crate) fn parse_node_key(key: &str) -> Option<(String, u32, String)> { - parse_visit_scoped_key(key, "nodes/") -} - pub(crate) fn parse_node_asset_key(key: &str) -> Option<(String, u32, String)> { parse_visit_scoped_key(key, ARTIFACT_NODES_PREFIX) } @@ -123,40 +61,6 @@ mod tests { fn top_level_keys_match_spec() { assert_eq!(init(), "_init.json"); assert_eq!(event_key(7, 123), "events/000007-123.json"); - assert_eq!(retro_prompt(), "retro/prompt.md"); - assert_eq!(retro_response(), "retro/response.md"); - } - - #[test] - fn node_keys_match_spec() { - let node = NodeVisitRef { - node_id: "plan", - visit: 3, - }; - assert_eq!(node_visit_prefix(&node), "nodes/plan/visit-3"); - assert_eq!(node_prompt(&node), "nodes/plan/visit-3/prompt.md"); - assert_eq!(node_response(&node), "nodes/plan/visit-3/response.md"); - assert_eq!(node_status(&node), "nodes/plan/visit-3/status.json"); - assert_eq!(node_outcome(&node), "nodes/plan/visit-3/outcome.json"); - assert_eq!( - node_provider_used(&node), - "nodes/plan/visit-3/provider_used.json" - ); - assert_eq!(node_diff(&node), "nodes/plan/visit-3/diff.patch"); - assert_eq!( - node_script_invocation(&node), - "nodes/plan/visit-3/script_invocation.json" - ); - assert_eq!( - node_script_timing(&node), - "nodes/plan/visit-3/script_timing.json" - ); - assert_eq!( - node_parallel_results(&node), - "nodes/plan/visit-3/parallel_results.json" - ); - assert_eq!(node_stdout(&node), "nodes/plan/visit-3/stdout.log"); - assert_eq!(node_stderr(&node), "nodes/plan/visit-3/stderr.log"); } #[test] @@ -184,10 +88,6 @@ mod tests { parse_artifact_value_id("artifacts/values/summary.json"), Some("summary".to_string()) ); - assert_eq!( - parse_node_key("nodes/plan/visit-3/status.json"), - Some(("plan".to_string(), 3, "status.json".to_string())) - ); assert_eq!( parse_node_asset_key("artifacts/nodes/code/visit-2/src/main.rs"), Some(("code".to_string(), 2, "src/main.rs".to_string())) @@ -201,7 +101,6 @@ mod tests { parse_artifact_value_id("artifacts/values/summary.txt"), None ); - assert_eq!(parse_node_key("nodes/plan/status.json"), None); assert_eq!( parse_node_asset_key("artifacts/nodes/code/status.json"), None diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index a720792e1..fe77181ad 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -1,4 +1,4 @@ -use std::collections::{BTreeMap, BTreeSet, HashMap}; +use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; use std::sync::atomic::{AtomicU32, Ordering}; @@ -14,10 +14,10 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use crate::keys; use crate::run_state::EventProjectionCache; use crate::{ - CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot, - NodeVisitRef, Result, RunState, RunSummary, StoreError, + CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeVisitRef, Result, RunState, + RunSummary, StoreError, }; -use fabro_types::{NodeStatusRecord, RunId}; +use fabro_types::RunId; #[derive(Debug, Default)] pub struct InMemoryStore { @@ -79,24 +79,6 @@ impl InMemoryRunStore { .map_err(Into::into) } - async fn put_text(&self, key: String, value: &str) { - self.data - .lock() - .await - .insert(key, value.as_bytes().to_vec()); - } - - async fn get_text(&self, key: &str) -> Result> { - let bytes = self.data.lock().await.get(key).cloned(); - bytes - .map(|value| { - String::from_utf8(value).map_err(|err| { - StoreError::Other(format!("stored text is not valid UTF-8: {err}")) - }) - }) - .transpose() - } - async fn put_bytes(&self, key: String, value: &[u8]) { self.data.lock().await.insert(key, value.to_vec()); } @@ -109,44 +91,6 @@ impl InMemoryRunStore { self.data.lock().await.clone() } - #[allow(clippy::unused_self)] - fn build_node_snapshot_from_data( - &self, - data: &BTreeMap>, - node: &NodeVisitRef<'_>, - ) -> Result { - Ok(NodeSnapshot { - node_id: node.node_id.to_string(), - visit: node.visit, - prompt: read_text(data, &keys::node_prompt(node))?, - response: read_text(data, &keys::node_response(node))?, - status: read_json(data, &keys::node_status(node))?, - outcome: read_json(data, &keys::node_outcome(node))?, - provider_used: read_json(data, &keys::node_provider_used(node))?, - diff: read_text(data, &keys::node_diff(node))?, - script_invocation: read_json(data, &keys::node_script_invocation(node))?, - script_timing: read_json(data, &keys::node_script_timing(node))?, - parallel_results: read_json(data, &keys::node_parallel_results(node))?, - stdout: read_text(data, &keys::node_stdout(node))?, - stderr: read_text(data, &keys::node_stderr(node))?, - }) - } - - async fn list_node_ids_inner(&self) -> Vec { - let data = self.snapshot_data().await; - let mut node_ids = BTreeSet::new(); - for key in data.keys() { - if let Some((node_id, _, _)) = keys::parse_node_key(key) { - node_ids.insert(node_id); - continue; - } - if let Some((node_id, _, _)) = keys::parse_node_asset_key(key) { - node_ids.insert(node_id); - } - } - node_ids.into_iter().collect() - } - async fn list_events_from_inner(&self, seq: u32) -> Result> { let data = self.snapshot_data().await; let mut events = Vec::new(); @@ -283,111 +227,6 @@ impl InMemoryStore { } impl InMemoryRunStore { - pub async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> { - self.put_text(keys::node_prompt(node), prompt).await; - Ok(()) - } - - pub async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> { - self.put_text(keys::node_response(node), response).await; - Ok(()) - } - - pub async fn put_node_status( - &self, - node: &NodeVisitRef<'_>, - status: &NodeStatusRecord, - ) -> Result<()> { - self.put_json(keys::node_status(node), status).await - } - - pub async fn put_node_outcome( - &self, - node: &NodeVisitRef<'_>, - outcome: &NodeOutcomeRecord, - ) -> Result<()> { - self.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.put_json(keys::node_provider_used(node), provider_used) - .await - } - - pub async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> { - self.put_text(keys::node_diff(node), diff).await; - Ok(()) - } - - pub async fn put_node_script_invocation( - &self, - node: &NodeVisitRef<'_>, - invocation: &serde_json::Value, - ) -> Result<()> { - self.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.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.put_json(keys::node_parallel_results(node), results) - .await - } - - pub async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.put_text(keys::node_stdout(node), log).await; - Ok(()) - } - - pub async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - self.put_text(keys::node_stderr(node), log).await; - Ok(()) - } - - pub async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result { - let data = self.snapshot_data().await; - self.build_node_snapshot_from_data(&data, node) - } - - pub async fn list_node_visits(&self, node_id: &str) -> Result> { - let data = self.snapshot_data().await; - let mut visits = BTreeSet::new(); - for key in data.keys() { - let Some((current_node_id, visit, _)) = keys::parse_node_key(key) else { - continue; - }; - if current_node_id == node_id { - visits.insert(visit); - } - } - Ok(visits.into_iter().collect()) - } - - pub async fn list_node_ids(&self) -> Result> { - Ok(self.list_node_ids_inner().await) - } - - pub async fn reset_for_rewind(&self) -> Result<()> { - let mut data = self.data.lock().await; - data.retain(|key, _| key == keys::init() || key.starts_with(keys::EVENTS_PREFIX)); - Ok(()) - } - pub async fn append_event(&self, payload: &EventPayload) -> Result { payload.validate(&self.run_id)?; @@ -445,25 +284,6 @@ impl InMemoryRunStore { Ok(Box::pin(UnboundedReceiverStream::new(receiver).map(Ok))) } - pub async fn put_retro_prompt(&self, text: &str) -> Result<()> { - self.put_text(keys::retro_prompt().to_string(), text).await; - Ok(()) - } - - pub async fn get_retro_prompt(&self) -> Result> { - self.get_text(keys::retro_prompt()).await - } - - pub async fn put_retro_response(&self, text: &str) -> Result<()> { - self.put_text(keys::retro_response().to_string(), text) - .await; - Ok(()) - } - - pub async fn get_retro_response(&self) -> Result> { - self.get_text(keys::retro_response()).await - } - pub async fn put_artifact_value( &self, artifact_id: &str, @@ -538,25 +358,6 @@ fn matches_query(created_at: &DateTime, query: &ListRunsQuery) -> bool { true } -fn read_json( - data: &BTreeMap>, - key: &str, -) -> Result> { - data.get(key) - .map(|value| serde_json::from_slice(value)) - .transpose() - .map_err(Into::into) -} - -fn read_text(data: &BTreeMap>, key: &str) -> Result> { - data.get(key) - .map(|value| { - String::from_utf8(value.clone()) - .map_err(|err| StoreError::Other(format!("stored text is not valid UTF-8: {err}"))) - }) - .transpose() -} - #[cfg(test)] mod tests { use super::*; @@ -696,24 +497,6 @@ mod tests { } } - fn sample_node_status() -> NodeStatusRecord { - NodeStatusRecord { - status: StageStatus::PartialSuccess, - notes: Some("captured output".to_string()), - failure_reason: Some("minor lint".to_string()), - timestamp: dt("2026-03-27T12:12:00Z"), - } - } - - fn sample_node_outcome() -> NodeOutcomeRecord { - fabro_types::Outcome { - status: StageStatus::Success, - notes: Some("all good".to_string()), - files_touched: vec!["src/lib.rs".to_string()], - ..Default::default() - } - } - fn sample_pull_request() -> PullRequestRecord { PullRequestRecord { html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(), @@ -767,12 +550,6 @@ mod tests { node_id: "code", visit: 2, }; - let node_status = sample_node_status(); - let node_outcome = sample_node_outcome(); - let provider_used = serde_json::json!({"provider": "openai", "model": "gpt-5.4"}); - let script_invocation = serde_json::json!({"command": "cargo test"}); - let script_timing = serde_json::json!({"duration_ms": 3210}); - let parallel_results = serde_json::json!([{"node_id": "lint", "status": "success"}]); let pull_request = sample_pull_request(); run.append_event(&event_payload( @@ -852,27 +629,106 @@ mod tests { )) .await .unwrap(); - run.put_node_prompt(&node, "Plan the fix").await.unwrap(); - run.put_node_response(&node, "Implemented").await.unwrap(); - run.put_node_status(&node, &node_status).await.unwrap(); - run.put_node_outcome(&node, &node_outcome).await.unwrap(); - run.put_node_provider_used(&node, &provider_used) - .await - .unwrap(); - run.put_node_diff(&node, "diff --git a/src/lib.rs b/src/lib.rs") - .await - .unwrap(); - run.put_node_script_invocation(&node, &script_invocation) - .await - .unwrap(); - run.put_node_script_timing(&node, &script_timing) - .await - .unwrap(); - run.put_node_parallel_results(&node, ¶llel_results) - .await - .unwrap(); - run.put_node_stdout(&node, "ok").await.unwrap(); - run.put_node_stderr(&node, "").await.unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.1Z", + "stage.prompt", + Some("code"), + serde_json::json!({ + "visit": 2, + "text": "Plan the fix", + "mode": "prompt", + "provider": "openai", + "model": "gpt-5.4" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.2Z", + "command.started", + Some("code"), + serde_json::json!({ + "visit": 2, + "command": "cargo test" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.3Z", + "command.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "stdout": "ok", + "stderr": "", + "exit_code": 0 + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.4Z", + "checkpoint.completed", + Some("code"), + serde_json::json!({ + "status": "success", + "ordinal": 2, + "current_node": checkpoint.current_node, + "completed_nodes": checkpoint.completed_nodes, + "node_retries": checkpoint.node_retries, + "context_values": checkpoint.context_values, + "node_outcomes": checkpoint.node_outcomes, + "next_node_id": checkpoint.next_node_id, + "git_commit_sha": checkpoint.git_commit_sha, + "node_visits": checkpoint.node_visits, + "diff": "diff --git a/src/lib.rs b/src/lib.rs" + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.5Z", + "parallel.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "results": [{"node_id": "lint", "status": "success"}] + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.6Z", + "stage.completed", + Some("code"), + serde_json::json!({ + "visit": 2, + "status": "success", + "notes": "all good", + "response": "Implemented", + "files_touched": ["src/lib.rs"] + }), + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:08.7Z", + "retro.started", + None, + serde_json::json!({ + "prompt": "How did it go?" + }), + )) + .await + .unwrap(); run.append_event(&event_payload( "run-1", "2026-03-27T12:00:09Z", @@ -917,8 +773,6 @@ mod tests { )) .await .unwrap(); - run.put_retro_prompt("How did it go?").await.unwrap(); - run.put_retro_response("Smooth enough").await.unwrap(); run.put_artifact_value("summary", &serde_json::json!({"done": true})) .await .unwrap(); @@ -957,14 +811,8 @@ mod tests { let stored_sandbox = state.sandbox.as_ref().unwrap(); assert_eq!(stored_sandbox.provider, sandbox.provider); assert_eq!(stored_sandbox.working_directory, sandbox.working_directory); - assert_eq!( - run.get_retro_prompt().await.unwrap(), - Some("How did it go?".to_string()) - ); - assert_eq!( - run.get_retro_response().await.unwrap(), - Some("Smooth enough".to_string()) - ); + assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?")); + assert_eq!(state.retro_response.as_deref(), Some("Smooth enough")); assert_eq!( run.get_artifact_value("summary").await.unwrap(), Some(serde_json::json!({"done": true})) @@ -978,7 +826,26 @@ mod tests { Some("diff --git a/src/lib.rs b/src/lib.rs\n") ); assert_eq!(state.pull_request, Some(pull_request.clone())); - assert_eq!(run.list_node_ids().await.unwrap(), vec!["code".to_string()]); + assert_eq!(state.list_node_ids(), vec!["code".to_string()]); + let node_state = state + .node(&node) + .expect("node state should exist for code:2"); + assert_eq!(node_state.prompt.as_deref(), Some("Plan the fix")); + assert_eq!(node_state.response.as_deref(), Some("Implemented")); + assert_eq!(node_state.stdout.as_deref(), Some("ok")); + assert_eq!(node_state.stderr.as_deref(), Some("")); + assert_eq!( + node_state.diff.as_deref(), + Some("diff --git a/src/lib.rs b/src/lib.rs") + ); + assert_eq!( + node_state + .provider_used + .as_ref() + .and_then(|v| v.get("provider")) + .and_then(|v| v.as_str()), + Some("openai") + ); assert_eq!( run.list_assets(&node).await.unwrap(), vec!["src/lib.rs".to_string()] @@ -1358,9 +1225,6 @@ mod tests { node_id: "code", visit: 2, }; - run.put_node_prompt(&snapshot_node, "Plan the fix") - .await - .unwrap(); run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") .await .unwrap(); @@ -1388,13 +1252,6 @@ mod tests { ("code".to_string(), 2, "src/lib.rs".to_string()) ] ); - assert_eq!( - run.list_node_ids().await.unwrap(), - vec!["artifact-only".to_string(), "code".to_string()] - ); - - let code_node = run.get_node(&snapshot_node).await.unwrap(); - assert_eq!(code_node.node_id, "code"); } #[tokio::test] @@ -1517,33 +1374,6 @@ mod tests { ); } - #[tokio::test] - async fn node_visit_storage_round_trips() { - let store = InMemoryStore::default(); - let run = store - .create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None) - .await - .unwrap(); - - let first = NodeVisitRef { - node_id: "code", - visit: 1, - }; - let second = NodeVisitRef { - node_id: "code", - visit: 2, - }; - run.put_node_prompt(&first, "first").await.unwrap(); - run.put_node_prompt(&second, "second").await.unwrap(); - run.put_node_status(&second, &sample_node_status()) - .await - .unwrap(); - - let node = run.get_node(&second).await.unwrap(); - assert_eq!(node.prompt, Some("second".to_string())); - assert_eq!(run.list_node_visits("code").await.unwrap(), vec![1, 2]); - } - #[tokio::test] async fn list_runs_filters_dates_and_tolerates_missing_status() { let store = InMemoryStore::default(); 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..962156946 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -383,8 +383,7 @@ mod tests { use bytes::Bytes; use fabro_types::{ - AttrValue, Graph, NodeStatusRecord, RunId, RunRecord, RunStatus, Settings, StageStatus, - StatusReason, fixtures, + AttrValue, Graph, RunId, RunRecord, RunStatus, Settings, StatusReason, fixtures, }; use object_store::memory::InMemory; use slatedb::config::Settings as SlateSettings; @@ -432,15 +431,6 @@ mod tests { } } - fn sample_node_status() -> NodeStatusRecord { - NodeStatusRecord { - status: StageStatus::Success, - notes: Some("done".to_string()), - failure_reason: None, - timestamp: dt("2026-03-27T12:12:00Z"), - } - } - fn event_payload( run_id: &str, ts: &str, @@ -824,7 +814,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 +1008,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 +1019,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 +1046,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 +1064,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 +1091,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..884c76601 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, "current_node": "work", "node_visits": {"work": 2}}), + )) .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/agent.rs b/lib/crates/fabro-workflow/src/handler/agent.rs index f34f2d90a..ac511d751 100644 --- a/lib/crates/fabro-workflow/src/handler/agent.rs +++ b/lib/crates/fabro-workflow/src/handler/agent.rs @@ -719,14 +719,17 @@ mod tests { .unwrap(); logger.flush().await; - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&NodeVisitRef { node_id: "step", visit: 1, }) - .await .unwrap(); - assert_eq!(snapshot.provider_used.unwrap()["provider"], "openai"); + assert_eq!( + node_state.provider_used.as_ref().unwrap()["provider"], + "openai" + ); } #[test] 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..4c4872584 100644 --- a/lib/crates/fabro-workflow/src/handler/parallel.rs +++ b/lib/crates/fabro-workflow/src/handler/parallel.rs @@ -4,7 +4,6 @@ use std::time::Instant; use async_trait::async_trait; use fabro_agent::{Sandbox, WorktreeOptions, WorktreeSandbox}; -use fabro_store::NodeVisitRef; use fabro_types::RunId; use tokio::sync::Semaphore; @@ -481,15 +480,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; } @@ -722,14 +712,14 @@ mod tests { .await .unwrap(); - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&fabro_store::NodeVisitRef { node_id: "par", visit: 1, }) - .await .unwrap(); - let results = snapshot.parallel_results.unwrap(); + let results = node_state.parallel_results.as_ref().unwrap(); assert!(results.is_array()); assert_eq!(results.as_array().unwrap().len(), 2); } diff --git a/lib/crates/fabro-workflow/src/handler/prompt.rs b/lib/crates/fabro-workflow/src/handler/prompt.rs index 8c78d6b03..6c5de318f 100644 --- a/lib/crates/fabro-workflow/src/handler/prompt.rs +++ b/lib/crates/fabro-workflow/src/handler/prompt.rs @@ -377,14 +377,14 @@ mod tests { .unwrap(); logger.flush().await; - let snapshot = run_store - .get_node(&NodeVisitRef { + let state = run_store.state().await.unwrap(); + let node_state = state + .node(&NodeVisitRef { node_id: "classify", visit: 1, }) - .await .unwrap(); - assert_eq!(snapshot.provider_used.unwrap()["mode"], "prompt"); + assert_eq!(node_state.provider_used.as_ref().unwrap()["mode"], "prompt"); } struct OneShotCapturingBackend { 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);