diff --git a/Cargo.lock b/Cargo.lock index 4d7f58fee..c6d6bb193 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1465,6 +1465,7 @@ dependencies = [ "fabro-model", "fabro-retro", "fabro-sandbox", + "fabro-store", "fabro-types", "fabro-util", "fabro-workflows", @@ -1475,6 +1476,7 @@ dependencies = [ "hyper", "hyper-util", "jsonwebtoken", + "object_store", "openapiv3", "reqwest", "rustls", @@ -1560,6 +1562,7 @@ dependencies = [ "insta", "jsonwebtoken", "libc", + "object_store", "open", "paste", "predicates", diff --git a/lib/crates/fabro-api/Cargo.toml b/lib/crates/fabro-api/Cargo.toml index 342d9f009..3f17e6b62 100644 --- a/lib/crates/fabro-api/Cargo.toml +++ b/lib/crates/fabro-api/Cargo.toml @@ -27,6 +27,7 @@ fabro-types = { path = "../fabro-types", features = ["exedev"] } fabro-util = { path = "../fabro-util" } fabro-db = { path = "../fabro-db" } fabro-api-types = { path = "../fabro-api-types" } +fabro-store = { path = "../fabro-store" } chrono.workspace = true futures-util.workspace = true axum = "0.8" @@ -59,6 +60,7 @@ sha2.workspace = true hex.workspace = true reqwest.workspace = true bytes = "1" +object_store.workspace = true [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/lib/crates/fabro-api/src/serve.rs b/lib/crates/fabro-api/src/serve.rs index fc7b33a3a..491b256f5 100644 --- a/lib/crates/fabro-api/src/serve.rs +++ b/lib/crates/fabro-api/src/serve.rs @@ -6,6 +6,7 @@ use fabro_config::server::{load_server_settings, resolve_storage_dir}; use fabro_model::{Catalog, Provider}; use fabro_util::terminal::Styles; use fabro_workflows::git::GitAuthor; +use object_store::local::LocalFileSystem; use tokio::net::TcpListener; use tokio::time::interval; use tracing::{error, info, warn}; @@ -16,7 +17,7 @@ use fabro_config::FabroSettings; use crate::github_webhooks::WebhookManager; use crate::jwt_auth::{AuthMode, AuthStrategy, decode_pem_env, resolve_auth_mode}; -use crate::server::{build_router, create_app_state_with_options, spawn_scheduler}; +use crate::server::{build_router, create_app_state_with_store, spawn_scheduler}; use crate::tls::{ClientAuth, build_rustls_config, serve_tls}; use fabro_llm::client::Client as LlmClient; use fabro_sandbox::SandboxProvider; @@ -148,13 +149,18 @@ pub async fn serve_command(args: ServeArgs, styles: &'static Styles) -> anyhow:: let cfg = shared_settings.read().expect("config lock poisoned"); cfg.hooks.clone() }; - let state = create_app_state_with_options( + let store_path = data_dir.join("store"); + std::fs::create_dir_all(&store_path)?; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?); + let store = Arc::new(fabro_store::SlateStore::new(object_store, "")); + let state = create_app_state_with_store( db, factory, dry_run_mode, max_concurrent_runs, git_author, hooks, + store, ); spawn_scheduler(Arc::clone(&state)); let router = build_router(state, auth_mode); diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index a9ce0e287..c4e6b1fdd 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -16,7 +16,8 @@ use fabro_llm::types::{ ContentPart, FinishReason, Message as LlmMessage, Request as LlmRequest, Response as LlmResponse, Role, StreamEvent, ToolChoice, ToolDefinition, Usage, }; -use fabro_retro::retro::{Retro, derive_retro, extract_stage_durations}; +use fabro_retro::retro::{Retro, derive_retro}; +use fabro_store::{InMemoryStore, Store}; use fabro_util::redact::redact_jsonl_line; use fabro_workflows::error::FabroError; use fabro_workflows::git::GitAuthor; @@ -123,6 +124,7 @@ type RegistryFactoryOverride = dyn Fn(Arc) -> HandlerRegistry + pub struct AppState { runs: Mutex>, aggregate_usage: Mutex, + store: Arc, llm_spec_factory: Box, registry_factory_override: Option>, pub dry_run: bool, @@ -406,6 +408,7 @@ pub fn create_app_state_with_registry_factory( 5, GitAuthor::default(), Vec::new(), + Arc::new(InMemoryStore::default()), ) } @@ -417,6 +420,26 @@ pub fn create_app_state_with_options( max_concurrent_runs: usize, git_author: GitAuthor, hooks: Vec, +) -> Arc { + create_app_state_with_store( + db, + llm_spec_factory, + dry_run, + max_concurrent_runs, + git_author, + hooks, + Arc::new(InMemoryStore::default()), + ) +} + +pub fn create_app_state_with_store( + db: sqlx::SqlitePool, + llm_spec_factory: impl Fn() -> LlmSpec + Send + Sync + 'static, + dry_run: bool, + max_concurrent_runs: usize, + git_author: GitAuthor, + hooks: Vec, + store: Arc, ) -> Arc { build_app_state( db, @@ -426,6 +449,7 @@ pub fn create_app_state_with_options( max_concurrent_runs, git_author, hooks, + store, ) } @@ -437,10 +461,12 @@ fn build_app_state( max_concurrent_runs: usize, git_author: GitAuthor, hooks: Vec, + store: Arc, ) -> Arc { Arc::new(AppState { runs: Mutex::new(HashMap::new()), aggregate_usage: Mutex::new(AggregateUsageTotals::default()), + store, llm_spec_factory, registry_factory_override, dry_run, @@ -671,7 +697,21 @@ async fn execute_run(state: Arc, run_id: String) { } } - let persisted = match Persisted::load(&run_dir) { + let run_store = match operations::open_or_hydrate_run(state.store.as_ref(), &run_dir).await { + Ok(run_store) => run_store, + Err(e) => { + tracing::error!(run_id = %run_id, error = %e, "Failed to open or hydrate run store"); + let mut runs = state.runs.lock().expect("runs lock poisoned"); + if let Some(managed_run) = runs.get_mut(&run_id) { + managed_run.status = RunStatus::Failed; + managed_run.error = Some(format!("Failed to open or hydrate run store: {e}")); + managed_run.event_tx = None; + } + state.scheduler_notify.notify_one(); + return; + } + }; + let persisted = match Persisted::load_from_store(run_store.as_ref(), &run_dir).await { Ok(persisted) => persisted, Err(e) => { tracing::error!(run_id = %run_id, error = %e, "Failed to load persisted run"); @@ -706,6 +746,7 @@ async fn execute_run(state: Arc, run_id: String) { let interviewer = Arc::clone(&interviewer) as Arc; let run_id = run_id.clone(); let run_options = run_options.clone(); + let run_store = Arc::clone(&run_store); let hooks = state.hooks.clone(); let dry_run = state.dry_run; async move { @@ -713,6 +754,7 @@ async fn execute_run(state: Arc, run_id: String) { persisted, InitOptions { run_id, + run_store, dry_run, emitter, sandbox, @@ -761,13 +803,26 @@ async fn execute_run(state: Arc, run_id: String) { }; // Save final checkpoint - let checkpoint = Checkpoint::load(&run_options.run_dir.join("checkpoint.json")).ok(); + let checkpoint = match run_store.get_checkpoint().await { + Ok(checkpoint) => checkpoint + .or_else(|| Checkpoint::load(&run_options.run_dir.join("checkpoint.json")).ok()), + Err(err) => { + tracing::warn!(run_id = %run_id, error = %err, "Failed to load checkpoint from store"); + Checkpoint::load(&run_options.run_dir.join("checkpoint.json")).ok() + } + }; // Auto-derive retro and accumulate aggregate usage if let Some(ref cp) = checkpoint { let failed = result.is_err(); let completed_stages = fabro_workflows::build_completed_stages(cp, failed); - let stage_durations = extract_stage_durations(&run_options.run_dir); + let stage_durations = match run_store.list_events().await { + Ok(events) => fabro_workflows::extract_stage_durations_from_events(&events), + Err(err) => { + tracing::warn!(run_id = %run_id, error = %err, "Failed to load run events from store"); + fabro_retro::retro::extract_stage_durations(&run_options.run_dir) + } + }; let retro = derive_retro( &run_id, "workflow", @@ -777,6 +832,9 @@ async fn execute_run(state: Arc, run_id: String) { &stage_durations, ); let _ = retro.save(&run_options.run_dir); + if let Err(err) = run_store.put_retro(&retro).await { + tracing::warn!(run_id = %run_id, error = %err, "Failed to save retro to store"); + } // Accumulate aggregate usage let mut agg = state @@ -1523,9 +1581,32 @@ async fn get_retro( return (StatusCode::OK, Json(serde_json::json!(null))).into_response(); }; - match Retro::load(&run_dir) { - Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), - Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + match state.store.open_run_reader(&id).await { + Ok(Some(run_store)) => match run_store.get_retro().await { + Ok(Some(retro)) => (StatusCode::OK, Json(retro)).into_response(), + Ok(None) => match Retro::load(&run_dir) { + Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), + Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + }, + Err(err) => { + tracing::warn!(run_id = %id, error = %err, "Failed to load retro from store"); + match Retro::load(&run_dir) { + Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), + Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + } + } + }, + Ok(None) => match Retro::load(&run_dir) { + Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), + Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + }, + Err(err) => { + tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader"); + match Retro::load(&run_dir) { + Ok(retro) => (StatusCode::OK, Json(retro)).into_response(), + Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(), + } + } } } diff --git a/lib/crates/fabro-cli/Cargo.toml b/lib/crates/fabro-cli/Cargo.toml index 9d7d1c854..3ee033b55 100644 --- a/lib/crates/fabro-cli/Cargo.toml +++ b/lib/crates/fabro-cli/Cargo.toml @@ -79,6 +79,7 @@ tempfile = "3" sha2.workspace = true shlex = "1" walkdir.workspace = true +object_store.workspace = true [target.'cfg(unix)'.dependencies] libc = "0.2" diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index 39c8bcf1f..e39b4495b 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -9,7 +9,7 @@ use fabro_workflows::pull_request::maybe_open_pull_request; use fabro_workflows::records::{ Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord, StartRecordExt, }; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use tracing::info; use crate::args::PrCreateArgs; @@ -29,14 +29,45 @@ async fn create_from( args: PrCreateArgs, github_app: Option, ) -> Result<()> { - let run_dir = resolve_run(base, &args.run_id)?.path; + let storage_dir = base.parent().unwrap_or(base); + let store = crate::store::build_store(storage_dir)?; + let run = resolve_run_combined(store.as_ref(), base, &args.run_id).await?; + let run_dir = run.path.clone(); + let run_store = crate::store::open_run_reader(storage_dir, &run.run_id).await?; - let record = RunRecord::load(&run_dir).context("Failed to load run.json")?; + let record = match run_store.as_ref() { + Some(run_store) => run_store + .get_run() + .await + .ok() + .flatten() + .or_else(|| RunRecord::load(&run_dir).ok()) + .context("Failed to load run.json")?, + None => RunRecord::load(&run_dir).context("Failed to load run.json")?, + }; - let start = StartRecord::load(&run_dir).context("Failed to load start.json")?; + let start = match run_store.as_ref() { + Some(run_store) => run_store + .get_start() + .await + .ok() + .flatten() + .or_else(|| StartRecord::load(&run_dir).ok()) + .context("Failed to load start.json")?, + None => StartRecord::load(&run_dir).context("Failed to load start.json")?, + }; - let conclusion = Conclusion::load(&run_dir.join("conclusion.json")) - .context("Failed to load conclusion.json — is the run finished?")?; + let conclusion = match run_store.as_ref() { + Some(run_store) => run_store + .get_conclusion() + .await + .ok() + .flatten() + .or_else(|| Conclusion::load(&run_dir.join("conclusion.json")).ok()) + .context("Failed to load conclusion.json — is the run finished?")?, + None => Conclusion::load(&run_dir.join("conclusion.json")) + .context("Failed to load conclusion.json — is the run finished?")?, + }; match conclusion.status { StageStatus::Success | StageStatus::PartialSuccess => {} @@ -103,6 +134,7 @@ async fn create_from( &model, true, None, + run_store.as_deref(), &run_dir, None, ) diff --git a/lib/crates/fabro-cli/src/commands/run/detached.rs b/lib/crates/fabro-cli/src/commands/run/detached.rs index 979ea62ca..f705f1911 100644 --- a/lib/crates/fabro-cli/src/commands/run/detached.rs +++ b/lib/crates/fabro-cli/src/commands/run/detached.rs @@ -2,11 +2,14 @@ use std::path::PathBuf; use std::sync::Arc; use anyhow::Result; +use fabro_config::FabroSettingsExt; use fabro_interview::FileInterviewer; use fabro_store::RuntimeState; use fabro_workflows::event::EventEmitter; use fabro_workflows::git::GitAuthor; -use fabro_workflows::operations::{StartServices, resume as resume_run, start as start_run}; +use fabro_workflows::operations::{ + StartServices, open_or_hydrate_run, resume as resume_run, start as start_run, +}; use fabro_workflows::records::{RunRecord, RunRecordExt}; use crate::cli_config; @@ -19,14 +22,17 @@ pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bo super::launcher::remove_launcher_record(&path); }); + let run_record = RunRecord::load(&run_dir)?; let cli_settings = cli_config::load_cli_settings(None)?; - let on_node: fabro_workflows::OnNodeCallback = RunRecord::load(&run_dir).ok().map(|record| { - let short_id = super::short_run_id(&record.run_id).to_string(); + let on_node: fabro_workflows::OnNodeCallback = Some({ + let short_id = super::short_run_id(&run_record.run_id).to_string(); fabro_proctitle::set(&format!("fabro: {short_id}")); Arc::new(move |node_id: &str| { fabro_proctitle::set(&format!("fabro: {short_id} {node_id}")); }) as Arc }); + let store = crate::store::build_store(&run_record.settings.storage_dir())?; + let run_store = open_or_hydrate_run(store.as_ref(), &run_dir).await?; let github_app = shared::github::build_github_app_credentials(cli_settings.app_id()); let git_author = GitAuthor::from_options( @@ -43,6 +49,7 @@ pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bo runtime_state.interview_response_path(), runtime_state.interview_claim_path(), )), + run_store, git_author, github_app, on_node, diff --git a/lib/crates/fabro-cli/src/commands/run/diff.rs b/lib/crates/fabro-cli/src/commands/run/diff.rs index 93ccaaea8..7b8992c60 100644 --- a/lib/crates/fabro-cli/src/commands/run/diff.rs +++ b/lib/crates/fabro-cli/src/commands/run/diff.rs @@ -6,7 +6,7 @@ use fabro_config::FabroSettingsExt; use fabro_sandbox::SandboxRecordExt; use fabro_sandbox::reconnect::reconnect; use fabro_workflows::records::{StartRecord, StartRecordExt}; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use fabro_workflows::sandbox_git::GIT_REMOTE; use tracing::{debug, info}; @@ -17,9 +17,11 @@ pub(crate) async fn run(args: DiffArgs) -> Result<()> { info!(run_id = %args.run, "Showing diff"); let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_dir = resolve_run(&base, &args.run)?.path; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let run_store = crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; - let patch = resolve_diff(&run_dir, &args).await?; + let patch = resolve_diff(&run.path, run_store.as_deref(), &args).await?; let is_tty = io::stdout().is_terminal(); let mut stdout = io::stdout().lock(); @@ -33,7 +35,11 @@ pub(crate) async fn run(args: DiffArgs) -> Result<()> { Ok(()) } -async fn resolve_diff(run_dir: &Path, args: &DiffArgs) -> Result { +async fn resolve_diff( + run_dir: &Path, + run_store: Option<&dyn fabro_store::RunStore>, + args: &DiffArgs, +) -> Result { if let Some(ref node_id) = args.node { debug!(node_id, "Reading per-node diff"); let node_patch = run_dir.join("nodes").join(node_id).join("diff.patch"); @@ -42,7 +48,16 @@ async fn resolve_diff(run_dir: &Path, args: &DiffArgs) -> Result { }); } - let start = StartRecord::load(run_dir).context("Failed to load start.json")?; + let start = match run_store { + Some(run_store) => run_store + .get_start() + .await + .ok() + .flatten() + .or_else(|| StartRecord::load(run_dir).ok()) + .context("Failed to load start.json")?, + None => StartRecord::load(run_dir).context("Failed to load start.json")?, + }; let base_sha = start .base_sha @@ -55,8 +70,14 @@ async fn resolve_diff(run_dir: &Path, args: &DiffArgs) -> Result { return std::fs::read_to_string(&final_patch_path).context("Failed to read final.patch"); } - let conclusion_path = run_dir.join("conclusion.json"); - if conclusion_path.exists() { + let run_concluded = match run_store { + Some(run_store) => { + run_store.get_conclusion().await.ok().flatten().is_some() + || run_dir.join("conclusion.json").exists() + } + None => run_dir.join("conclusion.json").exists(), + }; + if run_concluded { bail!( "Run completed but no final.patch exists — the run may not have produced any changes" ); @@ -64,9 +85,20 @@ async fn resolve_diff(run_dir: &Path, args: &DiffArgs) -> Result { debug!("No final.patch found; attempting live diff from sandbox"); let sandbox_json = run_dir.join("sandbox.json"); - let record = fabro_sandbox::SandboxRecord::load(&sandbox_json).context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?; + let record = match run_store { + Some(run_store) => run_store + .get_sandbox() + .await + .ok() + .flatten() + .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) + .context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + }; info!(provider = %record.provider, "Reconnecting to sandbox for live diff"); let sandbox = reconnect(&record).await?; diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index e703f3064..4e52723d8 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -1,34 +1,54 @@ use std::io::{self, BufRead, IsTerminal, Write}; use std::path::Path; +use std::time::Duration; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; use fabro_config::FabroSettingsExt; +use fabro_store::RunStore; use fabro_util::terminal::Styles; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; +use futures::StreamExt; use tracing::{debug, info}; use crate::args::LogsArgs; use crate::cli_config::load_cli_settings; -pub(crate) fn run(args: &LogsArgs, styles: &Styles) -> Result<()> { +pub(crate) async fn run(args: &LogsArgs, styles: &Styles) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run = resolve_run(&base, &args.run)?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; info!(run_id = %run.run_id, "Showing logs"); - let progress_path = run.path.join("progress.jsonl"); - if !progress_path.exists() { - bail!("No progress.jsonl found for run '{}'", run.run_id); - } - let since_cutoff = match &args.since { Some(value) => Some(parse_since(value)?), None => None, }; - let all_lines = read_lines(&progress_path)?; + let run_store = crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; + let (all_lines, last_seq) = match run_store.as_ref() { + Some(run_store) => { + let events = run_store + .list_events() + .await + .context("Failed to list store-backed run events")?; + let last_seq = events.last().map(|event| event.seq).unwrap_or(0); + let lines = events + .iter() + .map(event_payload_line) + .collect::>>()?; + (lines, last_seq) + } + None => { + let progress_path = run.path.join("progress.jsonl"); + if !progress_path.exists() { + bail!("No progress.jsonl found for run '{}'", run.run_id); + } + (read_lines(&progress_path)?, 0) + } + }; let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail); let stdout = io::stdout(); @@ -46,14 +66,25 @@ pub(crate) fn run(args: &LogsArgs, styles: &Styles) -> Result<()> { } if args.follow { - follow_logs( - &progress_path, - &run.path, - all_lines.len(), - args.pretty, - styles, - is_tty, - )?; + if let Some(run_store) = run_store.as_ref() { + follow_store_logs( + run_store.as_ref(), + if last_seq == 0 { 1 } else { last_seq + 1 }, + args.pretty, + styles, + is_tty, + ) + .await?; + } else { + follow_logs( + &run.path.join("progress.jsonl"), + &run.path, + all_lines.len(), + args.pretty, + styles, + is_tty, + )?; + } } Ok(()) @@ -171,6 +202,65 @@ fn follow_logs( Ok(()) } +async fn follow_store_logs( + run_store: &dyn RunStore, + seq: u32, + pretty: bool, + styles: &Styles, + _is_tty: bool, +) -> 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(); + + loop { + match tokio::time::timeout(Duration::from_millis(200), stream.next()).await { + Ok(Some(Ok(event))) => { + let line = event_payload_line(&event)?; + if pretty { + if let Some(formatted) = format_event_pretty(&line, styles) { + writeln!(out, "{formatted}")?; + } + } else { + writeln!(out, "{line}")?; + } + out.flush()?; + } + Ok(Some(Err(err))) => return Err(err.into()), + Ok(None) => break, + Err(_) => { + if run_store + .get_conclusion() + .await + .context("Failed to read conclusion from store while following logs")? + .is_some() + { + debug!("Run concluded, stopping follow"); + break; + } + if run_store + .get_status() + .await + .context("Failed to read status from store while following logs")? + .is_some_and(|record| record.status.is_terminal()) + { + debug!("Run reached terminal status, stopping follow"); + break; + } + } + } + } + + Ok(()) +} + +fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result { + serde_json::to_string(event.payload.as_value()).map_err(Into::into) +} + fn render_indented_markdown(styles: &Styles, text: &str, indent: &str) -> String { let term_width = Styles::terminal_width(); let wrap_width = term_width.saturating_sub(indent.len()); diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 278b368c6..14adf7abc 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -70,7 +70,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( RunCommands::Diff(args) => diff::run(args).await, RunCommands::Logs(args) => { let styles = Styles::detect_stdout(); - logs::run(&args, &styles) + logs::run(&args, &styles).await } RunCommands::Resume(args) => { let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); @@ -91,7 +91,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( } RunCommands::Wait(args) => { let styles = Styles::detect_stderr(); - wait::run(&args, &styles) + wait::run(&args, &styles).await } } } diff --git a/lib/crates/fabro-cli/src/commands/run/preview.rs b/lib/crates/fabro-cli/src/commands/run/preview.rs index c12033ac9..0a1125748 100644 --- a/lib/crates/fabro-cli/src/commands/run/preview.rs +++ b/lib/crates/fabro-cli/src/commands/run/preview.rs @@ -2,7 +2,7 @@ use anyhow::{Context, Result}; use fabro_config::FabroSettingsExt; use fabro_sandbox::SandboxRecordExt; use fabro_sandbox::daytona::DaytonaSandbox; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use tracing::info; use crate::args::PreviewArgs; @@ -12,11 +12,25 @@ use crate::shared::validate_daytona_provider; pub(crate) async fn run(args: PreviewArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_dir = resolve_run(&base, &args.run)?.path; - let sandbox_json = run_dir.join("sandbox.json"); - let record = fabro_sandbox::SandboxRecord::load(&sandbox_json).context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let sandbox_json = run.path.join("sandbox.json"); + let record = match crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + { + Some(run_store) => run_store + .get_sandbox() + .await + .ok() + .flatten() + .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) + .context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + }; validate_daytona_provider(&record, "Preview URLs")?; diff --git a/lib/crates/fabro-cli/src/commands/run/ssh.rs b/lib/crates/fabro-cli/src/commands/run/ssh.rs index 1592a6f12..84dccad78 100644 --- a/lib/crates/fabro-cli/src/commands/run/ssh.rs +++ b/lib/crates/fabro-cli/src/commands/run/ssh.rs @@ -2,7 +2,7 @@ use anyhow::{Context, Result, bail}; use fabro_config::FabroSettingsExt; use fabro_sandbox::SandboxRecordExt; use fabro_sandbox::daytona::DaytonaSandbox; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use tracing::info; use crate::args::SshArgs; @@ -12,11 +12,25 @@ use crate::shared::validate_daytona_provider; pub(crate) async fn run(args: SshArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_dir = resolve_run(&base, &args.run)?.path; - let sandbox_json = run_dir.join("sandbox.json"); - let record = fabro_sandbox::SandboxRecord::load(&sandbox_json).context( - "Failed to load sandbox.json — was this run started with a recent version of arc?", - )?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let sandbox_json = run.path.join("sandbox.json"); + let record = match crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + { + Some(run_store) => run_store + .get_sandbox() + .await + .ok() + .flatten() + .or_else(|| fabro_sandbox::SandboxRecord::load(&sandbox_json).ok()) + .context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + None => fabro_sandbox::SandboxRecord::load(&sandbox_json).context( + "Failed to load sandbox.json — was this run started with a recent version of arc?", + )?, + }; validate_daytona_provider(&record, "SSH access")?; diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index 914b95a43..9d05766a8 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -4,7 +4,7 @@ use anyhow::{Result, bail}; use fabro_config::FabroSettingsExt; use fabro_util::terminal::Styles; use fabro_workflows::records::{Conclusion, ConclusionExt}; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use fabro_workflows::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt}; use tracing::info; @@ -12,13 +12,16 @@ use crate::args::WaitArgs; use crate::cli_config::load_cli_settings; use crate::shared::format_duration_ms; -pub(crate) fn run(args: &WaitArgs, styles: &Styles) -> Result<()> { +pub(crate) async fn run(args: &WaitArgs, styles: &Styles) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_info = resolve_run(&base, &args.run)?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run_info = resolve_run_combined(store.as_ref(), &base, &args.run).await?; info!(run_id = %run_info.run_id, "Waiting for run to complete"); + let run_store = + crate::store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?; let status_path = run_info.path.join("status.json"); let deadline = args .timeout @@ -26,9 +29,19 @@ pub(crate) fn run(args: &WaitArgs, styles: &Styles) -> Result<()> { let interval = std::time::Duration::from_millis(args.interval); let final_status = loop { - let status = match RunStatusRecord::load(&status_path) { - Ok(record) => record.status, - Err(_) => RunStatus::Dead, + let status = match run_store.as_ref() { + Some(run_store) => match run_store.get_status().await { + Ok(Some(record)) => record.status, + Ok(None) => RunStatus::Dead, + Err(_) => match RunStatusRecord::load(&status_path) { + Ok(record) => record.status, + Err(_) => RunStatus::Dead, + }, + }, + None => match RunStatusRecord::load(&status_path) { + Ok(record) => record.status, + Err(_) => RunStatus::Dead, + }, }; if status.is_terminal() { @@ -51,7 +64,15 @@ pub(crate) fn run(args: &WaitArgs, styles: &Styles) -> Result<()> { }; let conclusion_path = run_info.path.join("conclusion.json"); - let conclusion = Conclusion::load(&conclusion_path).ok(); + let conclusion = match run_store.as_ref() { + Some(run_store) => run_store + .get_conclusion() + .await + .ok() + .flatten() + .or_else(|| Conclusion::load(&conclusion_path).ok()), + None => Conclusion::load(&conclusion_path).ok(), + }; if args.json { let json_value = build_json_output(final_status, &run_info.run_id, conclusion.as_ref()); diff --git a/lib/crates/fabro-cli/src/commands/runs/inspect.rs b/lib/crates/fabro-cli/src/commands/runs/inspect.rs index dc43a3de9..e66409321 100644 --- a/lib/crates/fabro-cli/src/commands/runs/inspect.rs +++ b/lib/crates/fabro-cli/src/commands/runs/inspect.rs @@ -7,7 +7,7 @@ use fabro_workflows::records::{CheckpointExt, ConclusionExt, RunRecordExt, Start use serde::Serialize; use fabro_workflows::records::{Checkpoint, Conclusion, RunRecord, StartRecord}; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use fabro_workflows::run_status::RunStatus; use crate::args::InspectArgs; @@ -25,16 +25,56 @@ pub(crate) struct InspectOutput { pub sandbox: Option, } -pub(crate) fn run(args: &InspectArgs) -> Result<()> { +pub(crate) async fn run(args: &InspectArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run = resolve_run(&base, &args.run)?; - let output = inspect_run_dir(&run.run_id, &run.path, run.status); + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let output = + match crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await? { + Some(run_store) => { + inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await + } + None => inspect_run_dir(&run.run_id, &run.path, run.status), + }; let json = serde_json::to_string_pretty(&[output])?; println!("{json}"); Ok(()) } +async fn inspect_run_store( + run_id: &str, + run_dir: &Path, + status: RunStatus, + run_store: &dyn fabro_store::RunStore, +) -> InspectOutput { + match run_store.get_snapshot().await { + Ok(Some(snapshot)) => InspectOutput { + run_id: run_id.to_string(), + run_dir: run_dir.to_path_buf(), + status: snapshot + .status + .as_ref() + .map(|record| record.status) + .unwrap_or(status), + run_record: serde_json::to_value(snapshot.run).ok(), + start_record: snapshot + .start + .and_then(|record| serde_json::to_value(record).ok()), + conclusion: snapshot + .conclusion + .and_then(|record| serde_json::to_value(record).ok()), + checkpoint: snapshot + .checkpoint + .and_then(|record| serde_json::to_value(record).ok()), + sandbox: snapshot + .sandbox + .and_then(|record| serde_json::to_value(record).ok()), + }, + _ => inspect_run_dir(run_id, run_dir, status), + } +} + fn inspect_run_dir(run_id: &str, run_dir: &Path, status: RunStatus) -> InspectOutput { let run_record = RunRecord::load(run_dir) .ok() diff --git a/lib/crates/fabro-cli/src/commands/runs/list.rs b/lib/crates/fabro-cli/src/commands/runs/list.rs index 33fb74b68..f9b47dc02 100644 --- a/lib/crates/fabro-cli/src/commands/runs/list.rs +++ b/lib/crates/fabro-cli/src/commands/runs/list.rs @@ -8,7 +8,7 @@ use fabro_config::FabroSettingsExt; use fabro_util::terminal::Styles; use fabro_util::text::strip_goal_decoration; -use fabro_workflows::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs}; +use fabro_workflows::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs_combined}; use fabro_workflows::run_status::RunStatus; use crate::args::RunsListArgs; @@ -17,10 +17,11 @@ use crate::shared::{color_if, format_duration_ms, tilde_path}; use super::short_run_id; -pub(crate) fn list_command(args: &RunsListArgs, styles: &Styles) -> Result<()> { +pub(crate) async fn list_command(args: &RunsListArgs, styles: &Styles) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let runs = scan_runs(&base)?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let runs = scan_runs_combined(store.as_ref(), &base).await?; let label_filters = parse_label_filters(&args.filter.label); let filtered = filter_runs( &runs, diff --git a/lib/crates/fabro-cli/src/commands/runs/mod.rs b/lib/crates/fabro-cli/src/commands/runs/mod.rs index 5edacae7e..ed3354e08 100644 --- a/lib/crates/fabro-cli/src/commands/runs/mod.rs +++ b/lib/crates/fabro-cli/src/commands/runs/mod.rs @@ -11,10 +11,10 @@ pub(crate) async fn dispatch(cmd: RunsCommands) -> Result<()> { match cmd { RunsCommands::Ps(args) => { let styles = Styles::detect_stdout(); - list::list_command(&args, &styles) + list::list_command(&args, &styles).await } RunsCommands::Rm(args) => rm::remove_command(&args).await, - RunsCommands::Inspect(args) => inspect::run(&args), + RunsCommands::Inspect(args) => inspect::run(&args).await, } } diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index bd5773840..385740a98 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -7,6 +7,7 @@ mod logging; mod shared; #[cfg(feature = "sleep_inhibitor")] mod sleep_inhibitor; +mod store; use anyhow::Result; use args::{Commands, GlobalArgs, LONG_VERSION, RunCommands}; diff --git a/lib/crates/fabro-cli/src/store.rs b/lib/crates/fabro-cli/src/store.rs new file mode 100644 index 000000000..f1de4a0fa --- /dev/null +++ b/lib/crates/fabro-cli/src/store.rs @@ -0,0 +1,23 @@ +use std::path::Path; +use std::sync::Arc; + +use anyhow::Result; +use fabro_store::{RunStore, SlateStore, Store}; +use object_store::local::LocalFileSystem; + +pub(crate) fn build_store(storage_dir: &Path) -> Result> { + let store_path = storage_dir.join("store"); + std::fs::create_dir_all(&store_path)?; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(&store_path)?); + Ok(Arc::new(SlateStore::new(object_store, ""))) +} + +pub(crate) async fn open_run_reader( + storage_dir: &Path, + run_id: &str, +) -> Result>> { + build_store(storage_dir)? + .open_run_reader(run_id) + .await + .map_err(Into::into) +} diff --git a/lib/crates/fabro-store/src/error.rs b/lib/crates/fabro-store/src/error.rs index 0ea00d1ff..5dccf318f 100644 --- a/lib/crates/fabro-store/src/error.rs +++ b/lib/crates/fabro-store/src/error.rs @@ -14,6 +14,8 @@ pub enum StoreError { RunNotFound(String), #[error("Run already exists: {0}")] RunAlreadyExists(String), + #[error("run store is read-only")] + ReadOnly, #[error("{0}")] Other(String), } diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 232fe99da..36753921b 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -1,4 +1,5 @@ use std::pin::Pin; +use std::sync::Arc; use async_trait::async_trait; use bytes::Bytes; @@ -37,8 +38,10 @@ pub trait Store: Send + Sync { &self, run_id: &str, created_at: DateTime, - ) -> Result>; - async fn open_run(&self, run_id: &str) -> Result>>; + run_dir: Option<&str>, + ) -> Result>; + async fn open_run(&self, run_id: &str) -> Result>>; + async fn open_run_reader(&self, run_id: &str) -> Result>>; async fn list_runs(&self, query: &ListRunsQuery) -> Result>; async fn delete_run(&self, run_id: &str) -> Result<()>; } diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index 9eaf8a2c4..3f247fa88 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -44,11 +44,17 @@ struct InMemoryRunStore { } impl InMemoryRunStore { - fn new(run_id: &str, created_at: DateTime, db_prefix: String) -> Result { + fn new( + run_id: &str, + created_at: DateTime, + db_prefix: String, + run_dir: Option, + ) -> Result { let record = CatalogRecord { run_id: run_id.to_string(), created_at, db_prefix, + run_dir, }; let mut data = BTreeMap::new(); data.insert(keys::init().to_string(), serde_json::to_vec(&record)?); @@ -217,11 +223,12 @@ impl Store for InMemoryStore { &self, run_id: &str, created_at: DateTime, - ) -> Result> { + run_dir: Option<&str>, + ) -> Result> { let mut runs = self.runs.lock().await; if let Some(existing) = runs.get(run_id) { if existing.record.created_at == created_at { - return Ok(Box::new(Arc::clone(&existing.run_store))); + return Ok(Arc::clone(&existing.run_store) as Arc); } return Err(StoreError::RunAlreadyExists(run_id.to_string())); } @@ -231,24 +238,30 @@ impl Store for InMemoryStore { run_id, created_at, db_prefix.clone(), + run_dir.map(ToOwned::to_owned), )?); let catalog = InMemoryCatalog { record: CatalogRecord { run_id: run_id.to_string(), created_at, db_prefix, + run_dir: run_dir.map(ToOwned::to_owned), }, run_store: Arc::clone(&run_store), }; runs.insert(run_id.to_string(), catalog); - Ok(Box::new(run_store)) + Ok(run_store as Arc) } - async fn open_run(&self, run_id: &str) -> Result>> { + async fn open_run(&self, run_id: &str) -> Result>> { let runs = self.runs.lock().await; Ok(runs .get(run_id) - .map(|catalog| Box::new(Arc::clone(&catalog.run_store)) as Box)) + .map(|catalog| Arc::clone(&catalog.run_store) as Arc)) + } + + async fn open_run_reader(&self, run_id: &str) -> Result>> { + self.open_run(run_id).await } async fn list_runs(&self, query: &ListRunsQuery) -> Result> { @@ -276,7 +289,7 @@ impl Store for InMemoryStore { } #[async_trait] -impl RunStore for Arc { +impl RunStore for InMemoryRunStore { async fn put_run(&self, record: &RunRecord) -> Result<()> { self.validate_run_record(record)?; self.put_json(keys::run().to_string(), record).await @@ -577,6 +590,7 @@ fn build_run_summary( run_id: record.run_id.clone(), created_at: record.created_at, db_prefix: record.db_prefix.clone(), + run_dir: record.run_dir.clone(), workflow_name, workflow_slug: run.as_ref().and_then(|run| run.workflow_slug.clone()), goal, @@ -730,7 +744,7 @@ mod tests { async fn create_run_put_get_and_snapshot_round_trip() { let store = InMemoryStore::default(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); let run_record = sample_run_record("run-1", created_at); let start_record = sample_start_record("run-1", created_at); @@ -851,7 +865,7 @@ mod tests { async fn append_event_validates_payload_shape_and_run_id() { let store = InMemoryStore::default(); let run = store - .create_run("run-1", dt("2026-03-27T12:00:00Z")) + .create_run("run-1", dt("2026-03-27T12:00:00Z"), None) .await .unwrap(); @@ -876,7 +890,7 @@ mod tests { async fn put_run_rejects_created_at_mismatch() { let store = InMemoryStore::default(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); let err = run .put_run(&sample_run_record( "run-1", @@ -891,7 +905,7 @@ mod tests { async fn watch_events_from_receives_existing_and_live_events() { let store = InMemoryStore::default(); let run = store - .create_run("run-1", dt("2026-03-27T12:00:00Z")) + .create_run("run-1", dt("2026-03-27T12:00:00Z"), None) .await .unwrap(); let first = EventPayload::new( @@ -942,7 +956,7 @@ mod tests { async fn checkpoint_history_round_trips() { let store = InMemoryStore::default(); let run = store - .create_run("run-1", dt("2026-03-27T12:00:00Z")) + .create_run("run-1", dt("2026-03-27T12:00:00Z"), None) .await .unwrap(); let checkpoint = sample_checkpoint(); @@ -961,7 +975,7 @@ mod tests { async fn node_visit_storage_round_trips() { let store = InMemoryStore::default(); let run = store - .create_run("run-1", dt("2026-03-27T12:00:00Z")) + .create_run("run-1", dt("2026-03-27T12:00:00Z"), None) .await .unwrap(); @@ -990,13 +1004,13 @@ mod tests { let early = dt("2026-03-27T10:00:00Z"); let late = dt("2026-03-27T12:00:00Z"); - let early_run = store.create_run("run-early", early).await.unwrap(); + let early_run = store.create_run("run-early", early, None).await.unwrap(); early_run .put_run(&sample_run_record("run-early", early)) .await .unwrap(); - let late_run = store.create_run("run-late", late).await.unwrap(); + let late_run = store.create_run("run-late", late, None).await.unwrap(); late_run .put_run(&sample_run_record("run-late", late)) .await @@ -1043,7 +1057,7 @@ mod tests { async fn delete_run_is_idempotent() { let store = InMemoryStore::default(); store - .create_run("run-1", dt("2026-03-27T12:00:00Z")) + .create_run("run-1", dt("2026-03-27T12:00:00Z"), None) .await .unwrap(); store.delete_run("run-1").await.unwrap(); @@ -1057,14 +1071,14 @@ mod tests { let ts = dt("2026-03-27T12:00:00Z"); // First create succeeds. - store.create_run("run-1", ts).await.unwrap(); + store.create_run("run-1", ts, None).await.unwrap(); // Retry with exact same created_at succeeds (idempotent). - store.create_run("run-1", ts).await.unwrap(); + store.create_run("run-1", ts, None).await.unwrap(); // Different created_at for the same run_id is rejected. let different_ts = dt("2026-03-27T12:00:01Z"); - match store.create_run("run-1", different_ts).await { + match store.create_run("run-1", different_ts, None).await { Err(StoreError::RunAlreadyExists(_)) => {} // expected Err(other) => panic!("expected RunAlreadyExists, got: {other:?}"), Ok(_) => panic!("expected RunAlreadyExists, but create_run succeeded"), diff --git a/lib/crates/fabro-store/src/slate/catalog.rs b/lib/crates/fabro-store/src/slate/catalog.rs index 99bfd0a92..51332bfd4 100644 --- a/lib/crates/fabro-store/src/slate/catalog.rs +++ b/lib/crates/fabro-store/src/slate/catalog.rs @@ -14,11 +14,13 @@ pub(crate) async fn write_catalog( run_id: &str, created_at: DateTime, db_prefix: &str, + run_dir: Option<&str>, ) -> Result { let record = CatalogRecord { run_id: run_id.to_string(), created_at, db_prefix: db_prefix.to_string(), + run_dir: run_dir.map(ToOwned::to_owned), }; let bytes = serde_json::to_vec(&record)?; store diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 4ea776fbf..55712eada 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -113,11 +113,32 @@ impl SlateStore { let _ = db.close().await; return Ok(None); } - let run_store = SlateRunStore::open(record.clone(), db).await?; + let run_store = SlateRunStore::open_writer(record.clone(), db).await?; self.cache_active_run(&run_store).await; Ok(Some(run_store)) } + async fn open_run_reader_store(&self, record: &CatalogRecord) -> Result> { + if !self.db_prefix_has_objects(&record.db_prefix).await? { + return Ok(None); + } + let reader = self.open_reader(&record.db_prefix).await?; + let has_init = match SlateRunStore::validate_init(&reader, record).await { + Ok(has_init) => has_init, + Err(err) => { + let _ = reader.close().await; + return Err(err); + } + }; + if !has_init { + let _ = reader.close().await; + return Ok(None); + } + SlateRunStore::open_reader(record.clone(), reader) + .await + .map(Some) + } + async fn delete_db_prefix(&self, db_prefix: &str) -> Result<()> { let prefix = Path::from(db_prefix.to_string()); let metas = self @@ -138,7 +159,8 @@ impl Store for SlateStore { &self, run_id: &str, created_at: DateTime, - ) -> Result> { + run_dir: Option<&str>, + ) -> Result> { let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await?; if let Some(active) = self.get_active_run(run_id).await { @@ -156,9 +178,10 @@ impl Store for SlateStore { run_id, created_at, &record.db_prefix, + run_dir, ) .await?; - return Ok(Box::new(active)); + return Ok(Arc::new(active) as Arc); } let db_prefix = match locator { @@ -173,12 +196,13 @@ impl Store for SlateStore { run_id: run_id.to_string(), created_at, db_prefix: db_prefix.clone(), + run_dir: run_dir.map(ToOwned::to_owned), }; let db = self.open_db(&db_prefix).await?; SlateRunStore::validate_init(&db, &record).await?; db.put(keys::init(), serde_json::to_vec(&record)?).await?; - let run_store = SlateRunStore::open(record.clone(), db).await?; + let run_store = SlateRunStore::open_writer(record.clone(), db).await?; self.cache_active_run(&run_store).await; catalog::write_catalog( self.object_store.clone(), @@ -186,12 +210,13 @@ impl Store for SlateStore { run_id, created_at, &db_prefix, + run_dir, ) .await?; - Ok(Box::new(run_store)) + Ok(Arc::new(run_store) as Arc) } - async fn open_run(&self, run_id: &str) -> Result>> { + async fn open_run(&self, run_id: &str) -> Result>> { let Some(locator) = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? else { @@ -201,7 +226,20 @@ impl Store for SlateStore { let Some(run_store) = self.open_run_store(&locator).await? else { return Ok(None); }; - Ok(Some(Box::new(run_store))) + Ok(Some(Arc::new(run_store) as Arc)) + } + + async fn open_run_reader(&self, run_id: &str) -> Result>> { + let Some(locator) = + catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id).await? + else { + return Ok(None); + }; + + let Some(run_store) = self.open_run_reader_store(&locator).await? else { + return Ok(None); + }; + Ok(Some(Arc::new(run_store) as Arc)) } async fn list_runs(&self, query: &ListRunsQuery) -> Result> { @@ -475,7 +513,7 @@ mod tests { async fn create_open_list_and_delete_full_lifecycle() { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await @@ -523,6 +561,7 @@ mod tests { run_id: "run-1".to_string(), created_at, db_prefix: catalog::db_prefix("runs/", created_at, "run-1"), + run_dir: None, }; let db = seed_db(object_store.clone(), &record, true).await; @@ -567,7 +606,7 @@ mod tests { async fn reopen_recovers_event_and_checkpoint_sequences() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await .unwrap(); @@ -605,6 +644,7 @@ mod tests { run_id: "run-1".to_string(), created_at, db_prefix: catalog::db_prefix("runs/", created_at, "run-1"), + run_dir: None, }; let db = seed_db(object_store.clone(), &record, false).await; @@ -638,11 +678,11 @@ mod tests { async fn create_run_allows_idempotent_retry_and_rejects_conflict() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - store.create_run("run-1", created_at).await.unwrap(); - store.create_run("run-1", created_at).await.unwrap(); + store.create_run("run-1", created_at, None).await.unwrap(); + store.create_run("run-1", created_at, None).await.unwrap(); let conflict = store - .create_run("run-1", created_at + chrono::Duration::seconds(1)) + .create_run("run-1", created_at + chrono::Duration::seconds(1), None) .await; assert!(matches!(conflict, Err(StoreError::RunAlreadyExists(_)))); } @@ -651,7 +691,7 @@ mod tests { async fn list_runs_and_open_run_reuse_active_handle_without_fencing() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await .unwrap(); @@ -684,7 +724,7 @@ mod tests { async fn watch_events_from_polls_new_events() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); let mut stream = run.watch_events_from(1).await.unwrap(); run.append_event(&event_payload("run-1", "2026-03-27T12:00:00Z", "Started")) @@ -710,6 +750,7 @@ mod tests { run_id: "run-1".to_string(), created_at, db_prefix: catalog::db_prefix("runs/", created_at, "run-1"), + run_dir: None, }; let db = seed_db(object_store.clone(), &record, true).await; db.put( @@ -736,7 +777,7 @@ mod tests { async fn delete_run_closes_active_handles() { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await .unwrap(); @@ -757,7 +798,7 @@ mod tests { let (object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); let wrong_time = dt("2026-03-27T11:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await .unwrap(); @@ -789,6 +830,7 @@ mod tests { run_id: "run-1".to_string(), created_at: old_created_at, db_prefix: catalog::db_prefix("runs/", old_created_at, "run-1"), + run_dir: None, }; let new_prefix = catalog::db_prefix("runs/", new_created_at, "run-1"); assert_ne!(orphan.db_prefix, new_prefix); @@ -797,7 +839,10 @@ mod tests { db.put(keys::graph(), b"stale graph").await.unwrap(); db.close().await.unwrap(); - let run = store.create_run("run-1", new_created_at).await.unwrap(); + let run = store + .create_run("run-1", new_created_at, None) + .await + .unwrap(); assert_eq!(run.get_graph().await.unwrap(), None); let locator = catalog::read_locator(object_store, "runs/", "run-1") @@ -820,13 +865,14 @@ mod tests { run_id: "other-run".to_string(), created_at, db_prefix, + run_dir: None, }; db.put(keys::init(), serde_json::to_vec(&mismatched).unwrap()) .await .unwrap(); db.close().await.unwrap(); - let err = match store.create_run("run-1", created_at).await { + let err = match store.create_run("run-1", created_at, None).await { Ok(_) => panic!("expected create_run to reject mismatched _init.json"), Err(err) => err, }; @@ -840,7 +886,7 @@ mod tests { async fn slate_run_store_round_trips_node_data_and_assets() { let (_object_store, store) = make_store(); let created_at = dt("2026-03-27T12:00:00Z"); - let run = store.create_run("run-1", created_at).await.unwrap(); + let run = store.create_run("run-1", created_at, None).await.unwrap(); run.put_run(&sample_run_record("run-1", created_at)) .await .unwrap(); diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 3836de823..5d4726b93 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -9,7 +9,7 @@ use chrono::{DateTime, Utc}; use futures::Stream; use serde::Serialize; use serde::de::DeserializeOwned; -use slatedb::{CloseReason, DbRead, ErrorKind}; +use slatedb::{CloseReason, DbRead, DbReader, ErrorKind}; use tokio::sync::{Mutex, mpsc}; use tokio::time; use tokio_stream::wrappers::UnboundedReceiverStream; @@ -33,14 +33,20 @@ pub(crate) struct SlateRunStoreInner { run_id: String, created_at: DateTime, db_prefix: String, - db: slatedb::Db, + run_dir: Option, + db: SlateRunDb, event_seq: AtomicU32, checkpoint_seq: AtomicU32, close_lock: Mutex<()>, } +enum SlateRunDb { + Writer(slatedb::Db), + Reader(DbReader), +} + impl SlateRunStore { - pub(crate) async fn open(record: CatalogRecord, db: slatedb::Db) -> Result { + pub(crate) async fn open_writer(record: CatalogRecord, db: slatedb::Db) -> Result { let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; let checkpoint_seq = recover_next_seq(&db, keys::CHECKPOINTS_PREFIX, keys::parse_checkpoint_seq).await?; @@ -49,7 +55,26 @@ impl SlateRunStore { run_id: record.run_id, created_at: record.created_at, db_prefix: record.db_prefix, - db, + run_dir: record.run_dir, + db: SlateRunDb::Writer(db), + event_seq: AtomicU32::new(event_seq), + checkpoint_seq: AtomicU32::new(checkpoint_seq), + close_lock: Mutex::new(()), + }), + }) + } + + pub(crate) async fn open_reader(record: CatalogRecord, db: DbReader) -> Result { + let event_seq = recover_next_seq(&db, keys::EVENTS_PREFIX, keys::parse_event_seq).await?; + let checkpoint_seq = + recover_next_seq(&db, keys::CHECKPOINTS_PREFIX, keys::parse_checkpoint_seq).await?; + Ok(Self { + inner: Arc::new(SlateRunStoreInner { + run_id: record.run_id, + created_at: record.created_at, + db_prefix: record.db_prefix, + run_dir: record.run_dir, + db: SlateRunDb::Reader(db), event_seq: AtomicU32::new(event_seq), checkpoint_seq: AtomicU32::new(checkpoint_seq), close_lock: Mutex::new(()), @@ -70,6 +95,7 @@ impl SlateRunStore { run_id: self.inner.run_id.clone(), created_at: self.inner.created_at, db_prefix: self.inner.db_prefix.clone(), + run_dir: self.inner.run_dir.clone(), } } @@ -77,6 +103,7 @@ impl SlateRunStore { self.inner.run_id == record.run_id && self.inner.created_at == record.created_at && self.inner.db_prefix == record.db_prefix + && self.inner.run_dir == record.run_dir } pub(crate) fn created_at(&self) -> DateTime { @@ -93,7 +120,10 @@ impl SlateRunStore { } pub(crate) async fn snapshot(&self) -> Result> { - Ok(self.inner.db.snapshot().await?) + match &self.inner.db { + SlateRunDb::Writer(db) => Ok(db.snapshot().await?), + SlateRunDb::Reader(_) => Err(StoreError::ReadOnly), + } } pub(crate) async fn validate_init(db: &R, expected: &CatalogRecord) -> Result @@ -134,6 +164,7 @@ impl SlateRunStore { run_id: catalog.run_id.clone(), created_at: catalog.created_at, db_prefix: catalog.db_prefix.clone(), + run_dir: catalog.run_dir.clone(), workflow_name, workflow_slug: run.as_ref().and_then(|run| run.workflow_slug.clone()), goal, @@ -170,11 +201,11 @@ impl SlateRunStore { Ok(NodeSnapshot { node_id: node.node_id.to_string(), visit: node.visit, - prompt: get_text(&self.inner.db, &keys::node_prompt(node)).await?, - response: get_text(&self.inner.db, &keys::node_response(node)).await?, - status: get_json(&self.inner.db, &keys::node_status(node)).await?, - stdout: get_text(&self.inner.db, &keys::node_stdout(node)).await?, - stderr: get_text(&self.inner.db, &keys::node_stderr(node)).await?, + 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?, + stdout: self.inner.db.get_text(&keys::node_stdout(node)).await?, + stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?, }) } } @@ -183,91 +214,98 @@ impl SlateRunStore { impl RunStore for SlateRunStore { async fn put_run(&self, record: &RunRecord) -> Result<()> { self.validate_run_record(record)?; - put_json(&self.inner.db, keys::run(), record).await + self.inner.db.put_json(keys::run(), record).await } async fn get_run(&self) -> Result> { - get_json(&self.inner.db, keys::run()).await + self.inner.db.get_json(keys::run()).await } async fn put_start(&self, record: &StartRecord) -> Result<()> { - put_json(&self.inner.db, keys::start(), record).await + self.inner.db.put_json(keys::start(), record).await } async fn get_start(&self) -> Result> { - get_json(&self.inner.db, keys::start()).await + self.inner.db.get_json(keys::start()).await } async fn put_status(&self, record: &RunStatusRecord) -> Result<()> { - put_json(&self.inner.db, keys::status(), record).await + self.inner.db.put_json(keys::status(), record).await } async fn get_status(&self) -> Result> { - get_json(&self.inner.db, keys::status()).await + self.inner.db.get_json(keys::status()).await } async fn put_checkpoint(&self, record: &Checkpoint) -> Result<()> { - put_json(&self.inner.db, keys::checkpoint(), record).await + self.inner.db.put_json(keys::checkpoint(), record).await } async fn get_checkpoint(&self) -> Result> { - get_json(&self.inner.db, keys::checkpoint()).await + self.inner.db.get_json(keys::checkpoint()).await } async fn append_checkpoint(&self, record: &Checkpoint) -> Result { let seq = self.inner.checkpoint_seq.fetch_add(1, Ordering::SeqCst); self.put_checkpoint(record).await?; - put_json( - &self.inner.db, - &keys::checkpoint_history_key(seq, Utc::now().timestamp_millis()), - record, - ) - .await?; + self.inner + .db + .put_json( + &keys::checkpoint_history_key(seq, Utc::now().timestamp_millis()), + record, + ) + .await?; Ok(seq) } async fn list_checkpoints(&self) -> Result> { - list_checkpoints(&self.inner.db).await + self.inner.db.list_checkpoints().await } async fn put_conclusion(&self, record: &Conclusion) -> Result<()> { - put_json(&self.inner.db, keys::conclusion(), record).await + self.inner.db.put_json(keys::conclusion(), record).await } async fn get_conclusion(&self) -> Result> { - get_json(&self.inner.db, keys::conclusion()).await + self.inner.db.get_json(keys::conclusion()).await } async fn put_retro(&self, retro: &Retro) -> Result<()> { - put_json(&self.inner.db, keys::retro(), retro).await + self.inner.db.put_json(keys::retro(), retro).await } async fn get_retro(&self) -> Result> { - get_json(&self.inner.db, keys::retro()).await + self.inner.db.get_json(keys::retro()).await } async fn put_graph(&self, dot_source: &str) -> Result<()> { - put_text(&self.inner.db, keys::graph(), dot_source).await + self.inner.db.put_text(keys::graph(), dot_source).await } async fn get_graph(&self) -> Result> { - get_text(&self.inner.db, keys::graph()).await + self.inner.db.get_text(keys::graph()).await } async fn put_sandbox(&self, record: &SandboxRecord) -> Result<()> { - put_json(&self.inner.db, keys::sandbox(), record).await + self.inner.db.put_json(keys::sandbox(), record).await } async fn get_sandbox(&self) -> Result> { - get_json(&self.inner.db, keys::sandbox()).await + self.inner.db.get_json(keys::sandbox()).await } async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> { - put_text(&self.inner.db, &keys::node_prompt(node), prompt).await + self.inner + .db + .put_text(&keys::node_prompt(node), prompt) + .await } async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> { - put_text(&self.inner.db, &keys::node_response(node), response).await + self.inner + .db + .put_text(&keys::node_response(node), response) + .await } async fn put_node_status( @@ -275,15 +313,18 @@ impl RunStore for SlateRunStore { node: &NodeVisitRef<'_>, status: &NodeStatusRecord, ) -> Result<()> { - put_json(&self.inner.db, &keys::node_status(node), status).await + self.inner + .db + .put_json(&keys::node_status(node), status) + .await } async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - put_text(&self.inner.db, &keys::node_stdout(node), log).await + self.inner.db.put_text(&keys::node_stdout(node), log).await } async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> { - put_text(&self.inner.db, &keys::node_stderr(node), log).await + self.inner.db.put_text(&keys::node_stderr(node), log).await } async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result { @@ -308,28 +349,29 @@ impl RunStore for SlateRunStore { 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); - put_json( - &self.inner.db, - &keys::event_key(seq, Utc::now().timestamp_millis()), - payload, - ) - .await?; + self.inner + .db + .put_json( + &keys::event_key(seq, Utc::now().timestamp_millis()), + payload, + ) + .await?; Ok(seq) } async fn list_events(&self) -> Result> { - list_events_from(&self.inner.db, 1).await + self.inner.db.list_events_from(1).await } async fn list_events_from(&self, seq: u32) -> Result> { - list_events_from(&self.inner.db, seq).await + self.inner.db.list_events_from(seq).await } async fn watch_events_from( &self, seq: u32, ) -> Result> + Send>>> { - let db = self.inner.db.clone(); + let inner = Arc::clone(&self.inner); let (sender, receiver) = mpsc::unbounded_channel(); tokio::spawn(async move { @@ -339,7 +381,7 @@ impl RunStore for SlateRunStore { return; } - match list_events_from(&db, next_seq).await { + match inner.db.list_events_from(next_seq).await { Ok(events) => { if events.is_empty() { time::sleep(Duration::from_millis(100)).await; @@ -364,35 +406,47 @@ impl RunStore for SlateRunStore { } async fn put_retro_prompt(&self, text: &str) -> Result<()> { - put_text(&self.inner.db, keys::retro_prompt(), text).await + self.inner.db.put_text(keys::retro_prompt(), text).await } async fn get_retro_prompt(&self) -> Result> { - get_text(&self.inner.db, keys::retro_prompt()).await + self.inner.db.get_text(keys::retro_prompt()).await } async fn put_retro_response(&self, text: &str) -> Result<()> { - put_text(&self.inner.db, keys::retro_response(), text).await + self.inner.db.put_text(keys::retro_response(), text).await } async fn get_retro_response(&self) -> Result> { - get_text(&self.inner.db, keys::retro_response()).await + self.inner.db.get_text(keys::retro_response()).await } async fn put_artifact_value(&self, artifact_id: &str, value: &serde_json::Value) -> Result<()> { - put_json(&self.inner.db, &keys::artifact_value(artifact_id), value).await + self.inner + .db + .put_json(&keys::artifact_value(artifact_id), value) + .await } async fn get_artifact_value(&self, artifact_id: &str) -> Result> { - get_json(&self.inner.db, &keys::artifact_value(artifact_id)).await + self.inner + .db + .get_json(&keys::artifact_value(artifact_id)) + .await } async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()> { - put_bytes(&self.inner.db, &keys::node_asset(node, filename), data).await + self.inner + .db + .put_bytes(&keys::node_asset(node, filename), data) + .await } async fn get_asset(&self, node: &NodeVisitRef<'_>, filename: &str) -> Result> { - get_bytes(&self.inner.db, &keys::node_asset(node, filename)).await + self.inner + .db + .get_bytes(&keys::node_asset(node, filename)) + .await } async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result> { @@ -446,6 +500,82 @@ impl RunStore for SlateRunStore { } } +impl SlateRunDb { + fn writer(&self) -> Result<&slatedb::Db> { + match self { + Self::Writer(db) => Ok(db), + Self::Reader(_) => Err(StoreError::ReadOnly), + } + } + + async fn close(&self) -> std::result::Result<(), slatedb::Error> { + match self { + Self::Writer(db) => db.close().await, + Self::Reader(db) => db.close().await, + } + } + + async fn get_json(&self, key: &str) -> Result> { + match self { + Self::Writer(db) => get_json(db, key).await, + Self::Reader(db) => get_json(db, key).await, + } + } + + async fn put_json(&self, key: &str, value: &T) -> Result<()> { + 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, 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, + Self::Reader(db) => db.get(key).await.map_err(Into::into), + } + } + + async fn put_bytes(&self, key: &str, value: &[u8]) -> Result<()> { + put_bytes(self.writer()?, key, value).await + } + + async fn scan_prefix

( + &self, + prefix: P, + ) -> std::result::Result + where + P: AsRef<[u8]> + Send, + { + match self { + Self::Writer(db) => db.scan_prefix(prefix).await, + Self::Reader(db) => db.scan_prefix(prefix).await, + } + } + + async fn list_events_from(&self, start_seq: u32) -> Result> { + match self { + Self::Writer(db) => list_events_from(db, start_seq).await, + Self::Reader(db) => list_events_from(db, start_seq).await, + } + } + + async fn list_checkpoints(&self) -> Result> { + match self { + Self::Writer(db) => list_checkpoints(db).await, + Self::Reader(db) => list_checkpoints(db).await, + } + } +} + async fn put_json(db: &slatedb::Db, key: &str, value: &T) -> Result<()> { db.put(key, serde_json::to_vec(value)?).await?; Ok(()) diff --git a/lib/crates/fabro-store/src/types.rs b/lib/crates/fabro-store/src/types.rs index a6ecad640..f642e0c2e 100644 --- a/lib/crates/fabro-store/src/types.rs +++ b/lib/crates/fabro-store/src/types.rs @@ -20,6 +20,7 @@ pub struct CatalogRecord { pub run_id: String, pub created_at: DateTime, pub db_prefix: String, + pub run_dir: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -27,6 +28,7 @@ pub struct RunSummary { pub run_id: String, pub created_at: DateTime, pub db_prefix: String, + pub run_dir: Option, pub workflow_name: Option, pub workflow_slug: Option, pub goal: Option, diff --git a/lib/crates/fabro-workflows/src/event.rs b/lib/crates/fabro-workflows/src/event.rs index 7970e9db1..1e190dcb8 100644 --- a/lib/crates/fabro-workflows/src/event.rs +++ b/lib/crates/fabro-workflows/src/event.rs @@ -5,7 +5,9 @@ use std::sync::atomic::{AtomicI64, Ordering}; use anyhow::{Context, Result}; use chrono::{SecondsFormat, Utc}; +use fabro_store::{EventPayload, RunStore}; use serde::{Deserialize, Serialize}; +use tokio::sync::{mpsc, oneshot}; use crate::error::FabroError; use crate::outcome::{FailureDetail, StageUsage}; @@ -819,6 +821,17 @@ pub fn build_event_envelope(event: &WorkflowRunEvent, run_id: &str) -> serde_jso serde_json::Value::Object(envelope) } +pub fn build_redacted_event_payload( + event: &WorkflowRunEvent, + run_id: &str, +) -> Result { + let envelope = build_event_envelope(event, run_id); + let line = serde_json::to_string(&envelope)?; + let line = redact_jsonl_line(&line); + let value = serde_json::from_str(&line).context("Failed to parse redacted event payload")?; + EventPayload::new(value, run_id).map_err(anyhow::Error::from) +} + pub fn append_progress_event(run_dir: &Path, run_id: &str, event: &WorkflowRunEvent) -> Result<()> { let envelope = build_event_envelope(event, run_id); let line = serde_json::to_string(&envelope)?; @@ -873,6 +886,83 @@ impl ProgressLogger { } } +enum StoreProgressCommand { + Event(EventPayload), + Flush(oneshot::Sender<()>), +} + +#[derive(Clone)] +pub struct StoreProgressLogger { + tx: mpsc::UnboundedSender, + run_id: Arc>, +} + +impl StoreProgressLogger { + #[must_use] + pub fn new(run_store: Arc, run_id: impl Into) -> Self { + let (tx, mut rx) = mpsc::unbounded_channel(); + + tokio::spawn(async move { + while let Some(command) = rx.recv().await { + match command { + StoreProgressCommand::Event(payload) => { + if let Err(err) = run_store.append_event(&payload).await { + tracing::warn!(error = %err, "Failed to append event to run store"); + } + } + StoreProgressCommand::Flush(tx) => { + let _ = tx.send(()); + } + } + } + }); + + Self { + tx, + run_id: Arc::new(std::sync::Mutex::new(run_id.into())), + } + } + + pub fn register(&self, emitter: &EventEmitter) { + let tx = self.tx.clone(); + let run_id = Arc::clone(&self.run_id); + emitter.on_event(move |event| { + if let WorkflowRunEvent::WorkflowRunStarted { + run_id: started_run_id, + .. + } = event + { + (*run_id.lock().unwrap()).clone_from(started_run_id); + } + + let run_id = run_id.lock().unwrap().clone(); + match build_redacted_event_payload(event, &run_id) { + Ok(payload) => { + if tx.send(StoreProgressCommand::Event(payload)).is_err() { + tracing::warn!( + "Store progress logger channel closed while appending event" + ); + } + } + Err(err) => { + tracing::warn!(error = %err, "Failed to build store event payload"); + } + } + }); + } + + pub async fn flush(&self) { + let (tx, rx) = oneshot::channel(); + if self.tx.send(StoreProgressCommand::Flush(tx)).is_err() { + tracing::warn!("Store progress logger channel closed before flush"); + return; + } + if rx.await.is_err() { + tracing::warn!("Store progress logger flush dropped before completion"); + } + } +} + fn flatten_agent(inner: serde_json::Value) -> (String, serde_json::Map) { let serde_json::Value::Object(mut agent_fields) = inner else { return ("Agent".to_string(), serde_json::Map::new()); diff --git a/lib/crates/fabro-workflows/src/handler/manager_loop.rs b/lib/crates/fabro-workflows/src/handler/manager_loop.rs index 6b2474f2d..7e6817b41 100644 --- a/lib/crates/fabro-workflows/src/handler/manager_loop.rs +++ b/lib/crates/fabro-workflows/src/handler/manager_loop.rs @@ -5,7 +5,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use async_trait::async_trait; +use chrono::Utc; use fabro_config::FabroSettings; +use fabro_store::{InMemoryStore, Store}; use crate::condition::evaluate_condition; use crate::context::keys; @@ -196,6 +198,14 @@ impl Handler for SubWorkflowHandler { let hook_runner = services.hook_runner.clone(); let env = services.env.clone(); let dry_run = services.dry_run; + let run_store = InMemoryStore::default() + .create_run( + &child_run_options.run_id, + Utc::now(), + Some(child_run_options.run_dir.to_string_lossy().as_ref()), + ) + .await + .map_err(|err| FabroError::engine(err.to_string()))?; // Spawn child engine let mut child_handle = tokio::spawn(async move { @@ -203,6 +213,7 @@ impl Handler for SubWorkflowHandler { graph: child_graph, source: String::new(), run_options: child_run_options, + run_store, checkpoint: None, seed_context: Some(child_context), emitter, diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs index 5139e7209..c3e19e617 100644 --- a/lib/crates/fabro-workflows/src/lib.rs +++ b/lib/crates/fabro-workflows/src/lib.rs @@ -1,6 +1,8 @@ +use std::collections::HashMap; use std::sync::Arc; use fabro_retro::retro::CompletedStage; +use fabro_store::EventEnvelope; use serde::de::DeserializeOwned; /// Callback invoked when a workflow node starts executing. @@ -91,6 +93,24 @@ pub fn build_completed_stages(cp: &records::Checkpoint, run_failed: bool) -> Vec stages } +pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap { + let mut durations = HashMap::new(); + for envelope in events { + let value = envelope.payload.as_value(); + if value.get("event").and_then(serde_json::Value::as_str) != Some("StageCompleted") { + continue; + } + let Some(node_id) = value.get("node_id").and_then(serde_json::Value::as_str) else { + continue; + }; + let Some(duration_ms) = value.get("duration_ms").and_then(serde_json::Value::as_u64) else { + continue; + }; + durations.insert(node_id.to_string(), duration_ms); + } + durations +} + #[doc(hidden)] pub mod artifact; pub mod asset_snapshot; diff --git a/lib/crates/fabro-workflows/src/lifecycle/disk.rs b/lib/crates/fabro-workflows/src/lifecycle/disk.rs index bc2923825..c0dab2365 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/disk.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/disk.rs @@ -2,6 +2,8 @@ use std::path::PathBuf; use std::sync::Arc; use async_trait::async_trait; +use fabro_store::{NodeVisitRef, RunStore}; +use fabro_types::NodeStatusRecord; use fabro_core::error::Result as CoreResult; use fabro_core::graph::NodeSpec; @@ -13,7 +15,7 @@ use super::circuit_breaker::CircuitBreakerLifecycle; use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::outcome::StageUsage; +use crate::outcome::{OutcomeExt, StageUsage}; use crate::records::{Checkpoint, CheckpointExt}; use crate::run_dir::{write_node_status, write_start_record}; use crate::run_options::RunOptions; @@ -27,6 +29,7 @@ type WfNodeResult = NodeResult>; pub(crate) struct DiskLifecycle { pub run_dir: PathBuf, pub run_id: String, + pub run_store: Arc, pub graph: Arc, pub run_options: Arc, pub emitter: Arc, @@ -38,9 +41,27 @@ pub(crate) struct DiskLifecycle { impl RunLifecycle for DiskLifecycle { async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> { // Write start.json - write_start_record(&self.run_dir, &self.run_options); + let start_record = write_start_record(&self.run_dir, &self.run_options); // Write run status as Running write_run_status(&self.run_dir, RunStatus::Running, None); + if let Err(err) = self.run_store.put_start(&start_record).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "start_store_save_failed".to_string(), + message: format!("failed to save start record to store: {err}"), + }); + } + if let Err(err) = self + .run_store + .put_status(&fabro_types::RunStatusRecord::new(RunStatus::Running, None)) + .await + { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "status_store_save_failed".to_string(), + message: format!("failed to save running status to store: {err}"), + }); + } Ok(()) } @@ -53,6 +74,29 @@ impl RunLifecycle for DiskLifecycle { let gv = node.inner(); let visit = state.node_visits.get(gv.id.as_str()).copied().unwrap_or(1); write_node_status(&self.run_dir, &gv.id, visit, &result.outcome); + let node_status = NodeStatusRecord { + status: result.outcome.status.clone(), + notes: result.outcome.notes.clone(), + failure_reason: result.outcome.failure_reason().map(ToOwned::to_owned), + timestamp: chrono::Utc::now(), + }; + if let Err(err) = self + .run_store + .put_node_status( + &NodeVisitRef { + node_id: &gv.id, + visit: u32::try_from(visit).unwrap_or(u32::MAX), + }, + &node_status, + ) + .await + { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "node_status_store_save_failed".to_string(), + message: format!("[node: {}] node status store save failed: {err}", node.id()), + }); + } Ok(()) } @@ -95,6 +139,20 @@ impl RunLifecycle for DiskLifecycle { message: format!("[node: {}] checkpoint save failed: {e}", node.id()), }); } + if let Err(err) = self.run_store.put_checkpoint(&checkpoint).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "checkpoint_store_save_failed".to_string(), + message: format!("[node: {}] checkpoint store save failed: {err}", node.id()), + }); + } + if let Err(err) = self.run_store.append_checkpoint(&checkpoint).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "checkpoint_store_append_failed".to_string(), + message: format!("[node: {}] checkpoint append failed: {err}", node.id()), + }); + } Ok(()) } diff --git a/lib/crates/fabro-workflows/src/lifecycle/git.rs b/lib/crates/fabro-workflows/src/lifecycle/git.rs index 1846e91a3..61964db7c 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/git.rs @@ -2,6 +2,7 @@ use std::path::PathBuf; use std::sync::{Arc, Mutex}; use async_trait::async_trait; +use fabro_store::RunStore; use fabro_core::error::{CoreError, Result as CoreResult}; use fabro_core::graph::NodeSpec; @@ -38,6 +39,7 @@ pub(crate) struct GitLifecycle { pub emitter: Arc, pub run_dir: PathBuf, pub run_id: String, + pub run_store: Arc, pub run_options: Arc, pub start_node_id: Option, // Cross-lifecycle data (shared with EventLifecycle) @@ -61,9 +63,30 @@ impl RunLifecycle for GitLifecycle { self.run_options.host_repo_path.as_ref(), ) { let store = MetadataStore::new(repo_path, &self.run_options.git_author); - let run_json = std::fs::read(self.run_dir.join("run.json")).ok(); - let start_json = std::fs::read(self.run_dir.join("start.json")).ok(); - let sandbox_json = std::fs::read(self.run_dir.join("sandbox.json")).ok(); + let run_json = self + .run_store + .get_run() + .await + .ok() + .flatten() + .and_then(|record| serde_json::to_vec_pretty(&record).ok()) + .or_else(|| std::fs::read(self.run_dir.join("run.json")).ok()); + let start_json = self + .run_store + .get_start() + .await + .ok() + .flatten() + .and_then(|record| serde_json::to_vec_pretty(&record).ok()) + .or_else(|| std::fs::read(self.run_dir.join("start.json")).ok()); + let sandbox_json = self + .run_store + .get_sandbox() + .await + .ok() + .flatten() + .and_then(|record| serde_json::to_vec_pretty(&record).ok()) + .or_else(|| std::fs::read(self.run_dir.join("sandbox.json")).ok()); let mut files: Vec<(&str, &[u8])> = Vec::new(); if let Some(ref data) = run_json { files.push(("run.json", data)); @@ -111,39 +134,45 @@ impl RunLifecycle for GitLifecycle { ) { let store = MetadataStore::new(repo_path, &self.run_options.git_author); // Build checkpoint JSON for shadow branch - let checkpoint_path = self.run_dir.join("checkpoint.json"); - std::fs::read(&checkpoint_path).ok().and_then(|cp_json| { - let artifact_store = self.artifact_store.lock().unwrap(); - let mut extra_entries: Vec<(String, Vec)> = artifact_store - .list() - .iter() - .filter_map(|info| { - info.file_path.as_ref().and_then(|path| { - std::fs::read(path) - .ok() - .map(|data| (format!("artifacts/{}.json", info.id), data)) + self.run_store + .get_checkpoint() + .await + .ok() + .flatten() + .and_then(|checkpoint| serde_json::to_vec_pretty(&checkpoint).ok()) + .or_else(|| std::fs::read(self.run_dir.join("checkpoint.json")).ok()) + .and_then(|cp_json| { + let artifact_store = self.artifact_store.lock().unwrap(); + let mut extra_entries: Vec<(String, Vec)> = artifact_store + .list() + .iter() + .filter_map(|info| { + info.file_path.as_ref().and_then(|path| { + std::fs::read(path) + .ok() + .map(|data| (format!("artifacts/{}.json", info.id), data)) + }) }) - }) - .collect(); - extra_entries.extend(scan_node_files(&self.run_dir)); - let extra_refs: Vec<(&str, &[u8])> = extra_entries - .iter() - .map(|(k, v)| (k.as_str(), v.as_slice())) - .collect(); - match store.write_checkpoint(&self.run_id, &cp_json, &extra_refs) { - Ok(sha) => Some(sha), - Err(e) => { - self.emitter.emit(&WorkflowRunEvent::RunNotice { - level: RunNoticeLevel::Warn, - code: "checkpoint_metadata_write_failed".to_string(), - message: format!( - "[node: {node_id}] metadata checkpoint write failed: {e}" - ), - }); - None + .collect(); + extra_entries.extend(scan_node_files(&self.run_dir)); + let extra_refs: Vec<(&str, &[u8])> = extra_entries + .iter() + .map(|(k, v)| (k.as_str(), v.as_slice())) + .collect(); + match store.write_checkpoint(&self.run_id, &cp_json, &extra_refs) { + Ok(sha) => Some(sha), + Err(e) => { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "checkpoint_metadata_write_failed".to_string(), + message: format!( + "[node: {node_id}] metadata checkpoint write failed: {e}" + ), + }); + None + } } - } - }) + }) } else { None }; @@ -183,6 +212,30 @@ impl RunLifecycle for GitLifecycle { }); } } + match self.run_store.get_checkpoint().await { + Ok(Some(mut checkpoint)) => { + checkpoint.git_commit_sha = Some(sha.clone()); + if let Err(err) = self.run_store.put_checkpoint(&checkpoint).await { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "checkpoint_store_resave_failed".to_string(), + message: format!( + "[node: {node_id}] checkpoint store re-save with SHA failed: {err}" + ), + }); + } + } + Ok(None) => {} + Err(err) => { + self.emitter.emit(&WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Warn, + code: "checkpoint_store_load_failed".to_string(), + message: format!( + "[node: {node_id}] checkpoint store load failed: {err}" + ), + }); + } + } // Push run branch (skip in dry-run mode) if !self.run_options.dry_run_enabled() { diff --git a/lib/crates/fabro-workflows/src/lifecycle/mod.rs b/lib/crates/fabro-workflows/src/lifecycle/mod.rs index bc5d9d66d..61a57365a 100644 --- a/lib/crates/fabro-workflows/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflows/src/lifecycle/mod.rs @@ -14,6 +14,7 @@ use std::sync::{Arc, Mutex}; use std::time::Instant; use async_trait::async_trait; +use fabro_store::RunStore; use fabro_store::RuntimeState; use fabro_core::error::Result as CoreResult; @@ -83,6 +84,7 @@ impl WorkflowLifecycle { sandbox: &Arc, graph: Arc, run_dir: &PathBuf, + run_store: Arc, run_options: &Arc, is_resume: bool, on_node: crate::OnNodeCallback, @@ -140,6 +142,7 @@ impl WorkflowLifecycle { let disk = DiskLifecycle { run_dir: run_dir.clone(), run_id: run_options.run_id.clone(), + run_store: Arc::clone(&run_store), graph: Arc::clone(&graph), run_options: Arc::clone(run_options), emitter: Arc::clone(emitter), @@ -155,6 +158,7 @@ impl WorkflowLifecycle { emitter: Arc::clone(emitter), run_dir: run_dir.clone(), run_id: run_options.run_id.clone(), + run_store, run_options: Arc::clone(run_options), start_node_id, checkpoint_git_result: Arc::clone(&checkpoint_git_result), diff --git a/lib/crates/fabro-workflows/src/operations/hydrate.rs b/lib/crates/fabro-workflows/src/operations/hydrate.rs new file mode 100644 index 000000000..be70f1e85 --- /dev/null +++ b/lib/crates/fabro-workflows/src/operations/hydrate.rs @@ -0,0 +1,301 @@ +use std::io::{BufRead, ErrorKind}; +use std::path::Path; +use std::sync::Arc; + +use fabro_sandbox::{SandboxRecord, SandboxRecordExt}; +use fabro_store::{EventPayload, RunStore, Store}; +use tracing::warn; + +use crate::error::FabroError; +use crate::records::{ + Checkpoint, CheckpointExt, Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord, + StartRecordExt, +}; +use crate::run_status::{RunStatusRecord, RunStatusRecordExt}; +use fabro_retro::{RetroExt, retro::Retro}; + +const GRAPH_FILE_NAME: &str = "workflow.fabro"; +const LEGACY_GRAPH_FILE_NAME: &str = "graph.fabro"; + +pub async fn open_or_hydrate_run( + store: &dyn Store, + run_dir: &Path, +) -> Result, FabroError> { + let record = RunRecord::load(run_dir)?; + if let Some(run_store) = store.open_run(&record.run_id).await.map_err(store_error)? { + return Ok(run_store); + } + + let run_dir_string = run_dir.to_string_lossy().to_string(); + let run_store = store + .create_run(&record.run_id, record.created_at, Some(&run_dir_string)) + .await + .map_err(store_error)?; + + run_store.put_run(&record).await.map_err(store_error)?; + + if let Some(dot_source) = load_graph_source(run_dir)? { + run_store + .put_graph(&dot_source) + .await + .map_err(store_error)?; + } + + if let Some(status) = load_status_record(run_dir)? { + run_store.put_status(&status).await.map_err(store_error)?; + } + if let Some(start) = load_start_record(run_dir)? { + run_store.put_start(&start).await.map_err(store_error)?; + } + if let Some(checkpoint) = load_checkpoint(run_dir)? { + run_store + .put_checkpoint(&checkpoint) + .await + .map_err(store_error)?; + } + if let Some(conclusion) = load_conclusion(run_dir)? { + run_store + .put_conclusion(&conclusion) + .await + .map_err(store_error)?; + } + if let Some(retro) = load_retro(run_dir)? { + run_store.put_retro(&retro).await.map_err(store_error)?; + } + if let Some(sandbox) = load_sandbox_record(run_dir)? { + run_store.put_sandbox(&sandbox).await.map_err(store_error)?; + } + + hydrate_events(run_dir, &record.run_id, run_store.as_ref()).await?; + + Ok(run_store) +} + +async fn hydrate_events( + run_dir: &Path, + run_id: &str, + run_store: &dyn RunStore, +) -> Result<(), FabroError> { + let progress_path = run_dir.join("progress.jsonl"); + if !progress_path.exists() { + return Ok(()); + } + + let file = std::fs::File::open(&progress_path)?; + for (line_number, line_result) in std::io::BufReader::new(file).lines().enumerate() { + let line = line_result?; + let trimmed = line.trim(); + if trimmed.is_empty() { + continue; + } + + let value = match serde_json::from_str::(trimmed) { + Ok(value) => value, + Err(err) => { + warn!( + path = %progress_path.display(), + line_number = line_number + 1, + error = %err, + "Skipping malformed progress event during hydration" + ); + continue; + } + }; + + let payload = match EventPayload::new(value, run_id) { + Ok(payload) => payload, + Err(err) => { + warn!( + path = %progress_path.display(), + line_number = line_number + 1, + error = %err, + "Skipping invalid progress event during hydration" + ); + continue; + } + }; + + run_store + .append_event(&payload) + .await + .map_err(store_error)?; + } + + Ok(()) +} + +fn load_graph_source(run_dir: &Path) -> Result, FabroError> { + for name in [GRAPH_FILE_NAME, LEGACY_GRAPH_FILE_NAME] { + match std::fs::read_to_string(run_dir.join(name)) { + Ok(source) => return Ok(Some(source)), + Err(err) if err.kind() == ErrorKind::NotFound => {} + Err(err) => return Err(err.into()), + } + } + Ok(None) +} + +fn load_status_record(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("status.json"); + if !path.exists() { + return Ok(None); + } + Ok(Some( + RunStatusRecord::load(&path).map_err(|err| FabroError::Io(err.to_string()))?, + )) +} + +fn load_start_record(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("start.json"); + if !path.exists() { + return Ok(None); + } + Ok(Some(StartRecord::load(run_dir)?)) +} + +fn load_checkpoint(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("checkpoint.json"); + if !path.exists() { + return Ok(None); + } + Ok(Some(Checkpoint::load(&path)?)) +} + +fn load_conclusion(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("conclusion.json"); + if !path.exists() { + return Ok(None); + } + Ok(Some(Conclusion::load(&path)?)) +} + +fn load_retro(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("retro.json"); + if !path.exists() { + return Ok(None); + } + Retro::load(run_dir) + .map(Some) + .map_err(|err| FabroError::Io(err.to_string())) +} + +fn load_sandbox_record(run_dir: &Path) -> Result, FabroError> { + let path = run_dir.join("sandbox.json"); + if !path.exists() { + return Ok(None); + } + SandboxRecord::load(&path) + .map(Some) + .map_err(|err| FabroError::Io(err.to_string())) +} + +fn store_error(err: impl std::fmt::Display) -> FabroError { + FabroError::engine(err.to_string()) +} + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::{Path, PathBuf}; + + use chrono::Utc; + use fabro_config::FabroSettings; + use fabro_graphviz::graph::Graph; + use fabro_store::{InMemoryStore, Store}; + use fabro_types::{Conclusion, RunStatus, RunStatusRecord, StageStatus}; + + use super::open_or_hydrate_run; + use crate::event::{WorkflowRunEvent, append_progress_event}; + use crate::records::{Checkpoint, CheckpointExt, ConclusionExt, RunRecord, RunRecordExt}; + use crate::run_status::RunStatusRecordExt; + + fn write_run(run_dir: &Path) { + let record = RunRecord { + run_id: "run-123".to_string(), + created_at: Utc::now(), + settings: FabroSettings::default(), + graph: Graph::new("test"), + workflow_slug: Some("test".to_string()), + working_directory: PathBuf::from("/tmp/project"), + host_repo_path: Some("/tmp/project".to_string()), + base_branch: Some("main".to_string()), + labels: HashMap::new(), + }; + std::fs::create_dir_all(run_dir).unwrap(); + record.save(run_dir).unwrap(); + std::fs::write( + run_dir.join("workflow.fabro"), + "digraph test { start -> exit }", + ) + .unwrap(); + RunStatusRecord::new(RunStatus::Running, None) + .save(&run_dir.join("status.json")) + .unwrap(); + let checkpoint = Checkpoint::from_context( + &crate::context::Context::new(), + "start", + vec!["start".to_string()], + HashMap::new(), + HashMap::new(), + Some("exit".to_string()), + HashMap::new(), + HashMap::new(), + HashMap::new(), + ); + checkpoint.save(&run_dir.join("checkpoint.json")).unwrap(); + let conclusion = Conclusion { + timestamp: Utc::now(), + status: StageStatus::Success, + duration_ms: 5, + failure_reason: None, + final_git_commit_sha: None, + stages: Vec::new(), + total_cost: None, + total_retries: 0, + total_input_tokens: 0, + total_output_tokens: 0, + total_cache_read_tokens: 0, + total_cache_write_tokens: 0, + total_reasoning_tokens: 0, + has_pricing: false, + }; + conclusion.save(&run_dir.join("conclusion.json")).unwrap(); + append_progress_event( + run_dir, + "run-123", + &WorkflowRunEvent::RunNotice { + level: crate::event::RunNoticeLevel::Info, + code: "hydrated".to_string(), + message: "hello".to_string(), + }, + ) + .unwrap(); + } + + #[tokio::test] + async fn hydrates_run_records_into_store() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run-123"); + write_run(&run_dir); + + let store = InMemoryStore::default(); + let run_store = open_or_hydrate_run(&store, &run_dir).await.unwrap(); + + assert_eq!( + run_store.get_run().await.unwrap().unwrap().run_id, + "run-123" + ); + assert!(run_store.get_checkpoint().await.unwrap().is_some()); + assert!(run_store.get_conclusion().await.unwrap().is_some()); + assert_eq!(run_store.list_events().await.unwrap().len(), 1); + + let listed = store + .list_runs(&fabro_store::ListRunsQuery::default()) + .await + .unwrap(); + assert_eq!( + listed[0].run_dir.as_deref(), + Some(run_dir.to_string_lossy().as_ref()) + ); + } +} diff --git a/lib/crates/fabro-workflows/src/operations/mod.rs b/lib/crates/fabro-workflows/src/operations/mod.rs index 3d526df45..621cb79ef 100644 --- a/lib/crates/fabro-workflows/src/operations/mod.rs +++ b/lib/crates/fabro-workflows/src/operations/mod.rs @@ -1,5 +1,6 @@ mod create; mod fork; +mod hydrate; mod resume; mod rewind; mod source; @@ -11,6 +12,7 @@ mod validate; pub use crate::pipeline::{DevcontainerSpec, LlmSpec, SandboxEnvSpec}; pub use create::{CreateRunInput, CreatedRun, create}; pub use fork::{ForkRunInput, fork}; +pub use hydrate::open_or_hydrate_run; pub use resume::resume; pub use rewind::{ RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline, find_run_id_by_prefix, diff --git a/lib/crates/fabro-workflows/src/operations/resume.rs b/lib/crates/fabro-workflows/src/operations/resume.rs index 111433a61..8c13ae0ce 100644 --- a/lib/crates/fabro-workflows/src/operations/resume.rs +++ b/lib/crates/fabro-workflows/src/operations/resume.rs @@ -4,21 +4,30 @@ use fabro_store::RuntimeState; use crate::error::FabroError; use crate::outcome::StageStatus; -use crate::records::{Checkpoint, CheckpointExt, Conclusion, ConclusionExt}; -use crate::run_status::{self, RunStatus, RunStatusRecordExt}; +use crate::run_status::{self, RunStatus}; use super::start::{StartServices, Started, execute_persisted_run}; /// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. pub async fn resume(run_dir: &Path, services: StartServices) -> Result { - if let Ok(record) = run_status::RunStatusRecord::load(&run_dir.join("status.json")) { + if let Some(record) = services + .run_store + .get_status() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + { if record.status == RunStatus::Succeeded { return Err(FabroError::Precondition( "run already finished successfully — nothing to resume".to_string(), )); } } - if let Ok(conclusion) = Conclusion::load(&run_dir.join("conclusion.json")) { + if let Some(conclusion) = services + .run_store + .get_conclusion() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + { if matches!( conclusion.status, StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped @@ -29,12 +38,23 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result, seed_context: Option, + run_store: Arc, git_author: GitAuthor, git: Option, github_app: Option, @@ -65,6 +68,7 @@ pub struct StartServices { pub cancel_token: Option>, pub emitter: Arc, pub interviewer: Arc, + pub run_store: Arc, pub git_author: GitAuthor, pub github_app: Option, pub on_node: crate::OnNodeCallback, @@ -79,13 +83,24 @@ pub struct Started { /// Start a fresh workflow run. Errors if a checkpoint already exists (use `resume()` instead). pub async fn start(run_dir: &Path, services: StartServices) -> Result { - if run_dir.join("checkpoint.json").exists() { + if services + .run_store + .get_checkpoint() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + .is_some() + { return Err(FabroError::Precondition( "checkpoint.json exists in run directory — did you mean to resume?".to_string(), )); } - if let Ok(record) = run_status::RunStatusRecord::load(&run_dir.join("status.json")) { + if let Some(record) = services + .run_store + .get_status() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + { if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) { return Err(FabroError::Precondition(format!( "cannot start run: status is {:?}, expected submitted", @@ -102,13 +117,39 @@ pub(super) async fn execute_persisted_run( checkpoint: Option, services: StartServices, ) -> Result { - let mut bootstrap_guard = DetachedRunBootstrapGuard::arm(run_dir); + let run_store = Arc::clone(&services.run_store); + if let Err(err) = run_store + .put_status(&run_status::RunStatusRecord::new( + RunStatus::Starting, + Some(StatusReason::SandboxInitializing), + )) + .await + { + let error = FabroError::engine(err.to_string()); + let _ = persist_detached_failure( + run_store.as_ref(), + run_dir, + "bootstrap", + StatusReason::BootstrapFailed, + &error, + ) + .await; + return Err(error); + } - let persisted = match Persisted::load(run_dir) { + let mut bootstrap_guard = DetachedRunBootstrapGuard::arm(run_dir, Arc::clone(&run_store)); + + let persisted = match Persisted::load_from_store(services.run_store.as_ref(), run_dir).await { Ok(persisted) => persisted, Err(err) => { - let _ = - persist_detached_failure(run_dir, "bootstrap", StatusReason::BootstrapFailed, &err); + let _ = persist_detached_failure( + run_store.as_ref(), + run_dir, + "bootstrap", + StatusReason::BootstrapFailed, + &err, + ) + .await; bootstrap_guard.defuse(); return Err(err); } @@ -117,15 +158,21 @@ pub(super) async fn execute_persisted_run( let session = match RunSession::new(&persisted, services) { Ok(session) => session, Err(err) => { - let _ = - persist_detached_failure(run_dir, "bootstrap", StatusReason::BootstrapFailed, &err); + let _ = persist_detached_failure( + run_store.as_ref(), + run_dir, + "bootstrap", + StatusReason::BootstrapFailed, + &err, + ) + .await; bootstrap_guard.defuse(); return Err(err); } }; bootstrap_guard.defuse(); - let mut completion_guard = DetachedRunCompletionGuard::arm(run_dir); + let mut completion_guard = DetachedRunCompletionGuard::arm(run_dir, Arc::clone(&run_store)); let run_start = Instant::now(); let started = Box::pin(session.run(persisted, checkpoint)).await; @@ -135,14 +182,20 @@ pub(super) async fn execute_persisted_run( Ok(started) } Err(err) => { - persist_terminal_engine_failure(run_dir, &err, run_start.elapsed()); + persist_terminal_engine_failure(run_store.as_ref(), run_dir, &err, run_start.elapsed()) + .await; completion_guard.defuse(); Err(err) } } } -fn persist_terminal_engine_failure(run_dir: &Path, error: &FabroError, duration: Duration) { +async fn persist_terminal_engine_failure( + run_store: &dyn RunStore, + run_dir: &Path, + error: &FabroError, + duration: Duration, +) { let engine_result: Result = Err(error.clone()); let (final_status, failure_reason, run_status, status_reason) = classify_engine_result(&engine_result); @@ -154,6 +207,15 @@ fn persist_terminal_engine_failure(run_dir: &Path, error: &FabroError, duration: None, ); persist_terminal_outcome(run_dir, &conclusion, run_status, status_reason); + if let Err(err) = run_store.put_conclusion(&conclusion).await { + tracing::warn!(error = %err, "Failed to save terminal engine failure conclusion to store"); + } + if let Err(err) = run_store + .put_status(&run_status::RunStatusRecord::new(run_status, status_reason)) + .await + { + tracing::warn!(error = %err, "Failed to save terminal engine failure status to store"); + } } impl RunSession { @@ -298,6 +360,7 @@ impl RunSession { sandbox_env, devcontainer, seed_context: None, + run_store: services.run_store, git_author: services.git_author, git: None, github_app: services.github_app.clone(), @@ -414,9 +477,13 @@ impl RunSession { ProgressLogger::new(persisted.run_dir(), record.run_id.clone()) .register(self.emitter.as_ref()); + let store_progress_logger = + StoreProgressLogger::new(Arc::clone(&self.run_store), record.run_id.clone()); + store_progress_logger.register(self.emitter.as_ref()); let init_options = InitOptions { run_id: record.run_id.clone(), + run_store: Arc::clone(&self.run_store), dry_run: run_options.dry_run_enabled(), emitter: self.emitter, sandbox: self.sandbox, @@ -449,6 +516,7 @@ impl RunSession { }); let executed = pipeline::execute(initialized).await; + store_progress_logger.flush().await; let failed = !matches!( executed.outcome.as_ref().map(|outcome| &outcome.status), Ok(StageStatus::Success | StageStatus::PartialSuccess) @@ -456,6 +524,7 @@ impl RunSession { let retro_opts = RetroOptions { run_id: executed.run_options.run_id.clone(), + run_store: Arc::clone(&executed.run_store), workflow_name: executed.graph.name.clone(), goal: executed.graph.goal().to_string(), run_dir: executed.run_options.run_dir.clone(), @@ -476,6 +545,7 @@ impl RunSession { let finalize_opts = FinalizeOptions { run_dir: retroed.run_options.run_dir.clone(), run_id: retroed.run_options.run_id.clone(), + run_store: Arc::clone(&retroed.run_store), workflow_name: retroed.graph.name.clone(), hook_runner: retroed.hook_runner.clone(), preserve_sandbox: self.preserve_sandbox, @@ -483,6 +553,7 @@ impl RunSession { }; let pr_opts = PullRequestOptions { run_dir: retroed.run_options.run_dir.clone(), + run_store: Some(Arc::clone(&retroed.run_store)), pr_config: self.pr_config, github_app: self.pr_github_app, origin_url: self.pr_origin_url, @@ -492,6 +563,7 @@ impl RunSession { let retro = retroed.retro.clone(); let concluded = pipeline::finalize(retroed, &finalize_opts).await?; let finalized = pipeline::pull_request(concluded, &pr_opts).await; + store_progress_logger.flush().await; scopeguard::ScopeGuard::into_inner(cleanup_guard); @@ -505,11 +577,12 @@ impl RunSession { struct DetachedRunBootstrapGuard { run_dir: PathBuf, + run_store: Arc, active: bool, } impl DetachedRunBootstrapGuard { - fn arm(run_dir: &Path) -> Self { + fn arm(run_dir: &Path, run_store: Arc) -> Self { run_status::write_run_status( run_dir, RunStatus::Starting, @@ -517,6 +590,7 @@ impl DetachedRunBootstrapGuard { ); Self { run_dir: run_dir.to_path_buf(), + run_store, active: true, } } @@ -534,6 +608,17 @@ impl Drop for DetachedRunBootstrapGuard { RunStatus::Failed, Some(StatusReason::SandboxInitFailed), ); + let run_store = Arc::clone(&self.run_store); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let _ = run_store + .put_status(&run_status::RunStatusRecord::new( + RunStatus::Failed, + Some(StatusReason::SandboxInitFailed), + )) + .await; + }); + } } } } @@ -542,13 +627,17 @@ const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization struct DetachedRunCompletionGuard { run_dir: PathBuf, + run_store: Arc, + run_id: Option, active: bool, } impl DetachedRunCompletionGuard { - fn arm(run_dir: &Path) -> Self { + fn arm(run_dir: &Path, run_store: Arc) -> Self { Self { run_dir: run_dir.to_path_buf(), + run_store, + run_id: load_run_id(run_dir), active: true, } } @@ -587,6 +676,45 @@ impl Drop for DetachedRunCompletionGuard { }, ); } + let run_store = Arc::clone(&self.run_store); + let run_id = self.run_id.clone(); + if let Ok(handle) = Handle::try_current() { + handle.spawn(async move { + let _ = run_store + .put_status(&run_status::RunStatusRecord::new( + RunStatus::Failed, + Some(StatusReason::WorkflowError), + )) + .await; + if let Err(err) = run_store + .put_conclusion(&build_failure_conclusion(POSTRUN_ABORTED_MESSAGE)) + .await + { + tracing::warn!( + error = %err, + "Failed to save post-run abort conclusion to store" + ); + } + if let Some(run_id) = run_id { + let event = WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Error, + code: "postrun_aborted".to_string(), + message: POSTRUN_ABORTED_MESSAGE.to_string(), + }; + match build_redacted_event_payload(&event, &run_id) { + Ok(payload) => { + let _ = run_store.append_event(&payload).await; + } + Err(err) => { + tracing::warn!( + error = %err, + "Failed to build post-run abort event payload" + ); + } + } + } + }); + } } } @@ -603,7 +731,8 @@ fn load_run_id(run_dir: &Path) -> Option { }) } -fn persist_detached_failure( +async fn persist_detached_failure( + run_store: &dyn RunStore, run_dir: &Path, phase: &'static str, reason: StatusReason, @@ -631,8 +760,20 @@ fn persist_detached_failure( ) .map_err(|err| FabroError::Io(err.to_string()))?; - write_failure_conclusion(run_dir, &message, Some(reason))?; + let conclusion = write_failure_conclusion(run_dir, &message, Some(reason))?; run_status::write_run_status(run_dir, RunStatus::Failed, Some(reason)); + if let Err(err) = run_store.put_conclusion(&conclusion).await { + tracing::warn!(error = %err, "Failed to save detached failure conclusion to store"); + } + if let Err(err) = run_store + .put_status(&run_status::RunStatusRecord::new( + RunStatus::Failed, + Some(reason), + )) + .await + { + tracing::warn!(error = %err, "Failed to save detached failure status to store"); + } if let Some(run_id) = load_run_id(run_dir) { append_progress_event( @@ -641,10 +782,25 @@ fn persist_detached_failure( &WorkflowRunEvent::RunNotice { level: RunNoticeLevel::Error, code: format!("{phase}_failed"), - message, + message: message.clone(), }, ) .map_err(|err| FabroError::Io(err.to_string()))?; + let event = WorkflowRunEvent::RunNotice { + level: RunNoticeLevel::Error, + code: format!("{phase}_failed"), + message, + }; + match build_redacted_event_payload(&event, &run_id) { + Ok(payload) => { + if let Err(err) = run_store.append_event(&payload).await { + tracing::warn!(error = %err, "Failed to append detached failure event to store"); + } + } + Err(err) => { + tracing::warn!(error = %err, "Failed to build detached failure event payload"); + } + } } Ok(()) @@ -654,12 +810,18 @@ fn write_failure_conclusion( run_dir: &Path, message: &str, _reason: Option, -) -> Result<(), FabroError> { +) -> Result { if run_dir.join("conclusion.json").exists() { - return Ok(()); + return Conclusion::load(&run_dir.join("conclusion.json")).map_err(Into::into); } - let conclusion = Conclusion { + let conclusion = build_failure_conclusion(message); + conclusion.save(&run_dir.join("conclusion.json"))?; + Ok(conclusion) +} + +fn build_failure_conclusion(message: &str) -> Conclusion { + Conclusion { timestamp: Utc::now(), status: StageStatus::Fail, duration_ms: 0, @@ -674,9 +836,7 @@ fn write_failure_conclusion( total_cache_write_tokens: 0, total_reasoning_tokens: 0, has_pricing: false, - }; - conclusion.save(&run_dir.join("conclusion.json"))?; - Ok(()) + } } #[cfg(test)] @@ -686,6 +846,7 @@ mod tests { use chrono::Utc; use fabro_config::FabroSettings; + use fabro_store::InMemoryStore; use super::*; use crate::context::Context; @@ -734,7 +895,8 @@ mod tests { registry } - fn test_start_services( + async fn test_start_services( + run_dir: &Path, emitter: Arc, registry: Arc, ) -> StartServices { @@ -742,6 +904,9 @@ mod tests { cancel_token: None, emitter, interviewer: Arc::new(fabro_interview::AutoApproveInterviewer), + run_store: crate::operations::open_or_hydrate_run(&InMemoryStore::default(), run_dir) + .await + .unwrap(), git_author: crate::git::GitAuthor::default(), github_app: None, on_node: None, @@ -778,9 +943,12 @@ mod tests { } persisted_workflow(MINIMAL_DOT, &run_dir); - let started = start(&run_dir, test_start_services(emitter, registry)) - .await - .unwrap(); + let started = start( + &run_dir, + test_start_services(&run_dir, emitter, registry).await, + ) + .await + .unwrap(); assert_eq!( started.finalized.conclusion.final_git_commit_sha.as_deref(), @@ -799,9 +967,12 @@ mod tests { persisted_workflow(MINIMAL_DOT, &run_dir); - let started = start(&run_dir, test_start_services(emitter, registry)) - .await - .unwrap(); + let started = start( + &run_dir, + test_start_services(&run_dir, emitter, registry).await, + ) + .await + .unwrap(); assert_eq!(started.finalized.conclusion.status, StageStatus::Success); assert!(run_dir.join("conclusion.json").exists()); @@ -826,7 +997,7 @@ mod tests { visited.lock().unwrap().push(node_id.to_string()); } })), - ..test_start_services(emitter, registry) + ..test_start_services(&run_dir, emitter, registry).await }, ) .await @@ -846,7 +1017,11 @@ mod tests { persisted_workflow(MINIMAL_DOT, &run_dir); std::fs::write(run_dir.join("checkpoint.json"), "{}").unwrap(); - let result = start(&run_dir, test_start_services(emitter, registry)).await; + let result = start( + &run_dir, + test_start_services(&run_dir, emitter, registry).await, + ) + .await; assert!( matches!(&result, Err(crate::error::FabroError::Precondition(_))), @@ -864,7 +1039,11 @@ mod tests { persisted_workflow(MINIMAL_DOT, &run_dir); - let result = resume(&run_dir, test_start_services(emitter, registry)).await; + let result = resume( + &run_dir, + test_start_services(&run_dir, emitter, registry).await, + ) + .await; assert!( matches!(&result, Err(crate::error::FabroError::Precondition(_))), @@ -914,7 +1093,11 @@ mod tests { .save(&run_dir.join("conclusion.json")) .unwrap(); - let result = resume(&run_dir, test_start_services(emitter, registry)).await; + let result = resume( + &run_dir, + test_start_services(&run_dir, emitter, registry).await, + ) + .await; assert!( matches!(&result, Err(crate::error::FabroError::Precondition(_))), diff --git a/lib/crates/fabro-workflows/src/pipeline/execute.rs b/lib/crates/fabro-workflows/src/pipeline/execute.rs index 19c98539e..0da5fc54f 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute.rs @@ -38,6 +38,7 @@ pub async fn execute(init: Initialized) -> Executed { graph, source: _, run_options, + run_store, checkpoint, seed_context, emitter, @@ -91,6 +92,7 @@ pub async fn execute(init: Initialized) -> Executed { &sandbox, graph_arc, &run_options.run_dir, + Arc::clone(&run_store), &settings_arc, checkpoint.is_some(), on_node, @@ -146,6 +148,7 @@ pub async fn execute(init: Initialized) -> Executed { graph, outcome: Err(err), run_options, + run_store, hook_runner, emitter, sandbox, @@ -170,6 +173,7 @@ pub async fn execute(init: Initialized) -> Executed { graph, outcome: Err(err), run_options, + run_store, hook_runner, emitter, sandbox, @@ -189,6 +193,7 @@ pub async fn execute(init: Initialized) -> Executed { graph, outcome: Err(err), run_options, + run_store, hook_runner, emitter, sandbox, @@ -313,6 +318,7 @@ pub async fn execute(init: Initialized) -> Executed { graph, outcome, run_options, + run_store, hook_runner, emitter, sandbox, diff --git a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs index 887fc8c8c..e6d46b24c 100644 --- a/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs +++ b/lib/crates/fabro-workflows/src/pipeline/execute/tests.rs @@ -12,6 +12,7 @@ use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_hooks::HookConfig; use fabro_interview::AutoApproveInterviewer; use fabro_sandbox::SandboxSpec; +use fabro_store::InMemoryStore; use super::*; use crate::context::{self, Context}; @@ -139,6 +140,12 @@ fn test_lifecycle(setup_commands: Vec) -> LifecycleOptions { } } +async fn test_run_store(run_dir: &Path) -> Arc { + crate::operations::open_or_hydrate_run(&InMemoryStore::default(), run_dir) + .await + .unwrap() +} + #[tokio::test] async fn execute_runs_start_to_exit_and_returns_final_context() { let temp = tempfile::tempdir().unwrap(); @@ -149,6 +156,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { persisted_workflow(graph, source, &run_dir, "run-test"), InitOptions { run_id: "run-test".to_string(), + run_store: test_run_store(&run_dir).await, dry_run: false, emitter: Arc::new(crate::event::EventEmitter::new()), sandbox: SandboxSpec::Local { @@ -215,6 +223,7 @@ async fn run_with_lifecycle( persisted_workflow(graph.clone(), String::new(), &run_dir, &run_id), InitOptions { run_id, + run_store: test_run_store(&run_dir).await, dry_run: false, emitter, sandbox: SandboxSpec::Local { diff --git a/lib/crates/fabro-workflows/src/pipeline/finalize.rs b/lib/crates/fabro-workflows/src/pipeline/finalize.rs index 627ecc1f7..72c54a3c3 100644 --- a/lib/crates/fabro-workflows/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/finalize.rs @@ -11,6 +11,7 @@ use crate::run_status::{RunStatus, StatusReason, write_run_status}; use crate::sandbox_git::git_push_host; use fabro_hooks::{HookContext, HookEvent, HookRunner}; use fabro_retro::retro::extract_stage_durations; +use fabro_store::RunStore; use super::types::{Concluded, FinalizeOptions, Retroed}; @@ -139,6 +140,114 @@ pub fn build_conclusion( } } +pub(crate) async fn build_conclusion_from_store( + run_store: &dyn RunStore, + run_dir: &Path, + status: StageStatus, + failure_reason: Option, + run_duration_ms: u64, + final_git_commit_sha: Option, +) -> Conclusion { + let checkpoint = match run_store.get_checkpoint().await { + Ok(checkpoint) => checkpoint, + Err(err) => { + tracing::warn!(error = %err, "Failed to load checkpoint from store while building conclusion"); + Checkpoint::load(&run_dir.join("checkpoint.json")).ok() + } + }; + let stage_durations = match run_store.list_events().await { + Ok(events) => crate::extract_stage_durations_from_events(&events), + Err(err) => { + tracing::warn!(error = %err, "Failed to load events from store while building conclusion"); + extract_stage_durations(run_dir) + } + }; + + build_conclusion_from_parts( + checkpoint.as_ref(), + &stage_durations, + status, + failure_reason, + run_duration_ms, + final_git_commit_sha, + ) +} + +fn build_conclusion_from_parts( + checkpoint: Option<&Checkpoint>, + stage_durations: &std::collections::HashMap, + status: StageStatus, + failure_reason: Option, + run_duration_ms: u64, + final_git_commit_sha: Option, +) -> Conclusion { + let mut total_input_tokens: i64 = 0; + let mut total_output_tokens: i64 = 0; + let mut total_cache_read_tokens: i64 = 0; + let mut total_cache_write_tokens: i64 = 0; + let mut total_reasoning_tokens: i64 = 0; + let mut has_pricing = false; + + let (stages, total_cost, total_retries) = if let Some(cp) = checkpoint { + let mut stages = Vec::new(); + let mut cost_sum: Option = None; + let mut retries_sum: u32 = 0; + + for node_id in &cp.completed_nodes { + let outcome = cp.node_outcomes.get(node_id); + let retries = cp + .node_retries + .get(node_id) + .copied() + .unwrap_or(1) + .saturating_sub(1); + retries_sum += retries; + + let cost = outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost); + if let Some(c) = cost { + *cost_sum.get_or_insert(0.0) += c; + has_pricing = true; + } + + if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) { + total_input_tokens += usage.input_tokens; + total_output_tokens += usage.output_tokens; + total_cache_read_tokens += usage.cache_read_tokens.unwrap_or(0); + total_cache_write_tokens += usage.cache_write_tokens.unwrap_or(0); + total_reasoning_tokens += usage.reasoning_tokens.unwrap_or(0); + } + + stages.push(StageSummary { + stage_id: node_id.clone(), + stage_label: node_id.clone(), + duration_ms: stage_durations.get(node_id).copied().unwrap_or(0), + cost, + retries, + }); + } + (stages, cost_sum, retries_sum) + } else { + (vec![], None, 0) + }; + + Conclusion { + timestamp: chrono::Utc::now(), + status, + duration_ms: run_duration_ms, + failure_reason, + final_git_commit_sha, + stages, + total_cost, + total_retries, + total_input_tokens, + total_output_tokens, + total_cache_read_tokens, + total_cache_write_tokens, + total_reasoning_tokens, + has_pricing, + } +} + pub fn persist_terminal_outcome( run_dir: &Path, conclusion: &Conclusion, @@ -231,6 +340,7 @@ pub async fn finalize( graph, outcome, run_options, + run_store: _run_store, hook_runner, emitter, sandbox, @@ -240,13 +350,15 @@ pub async fn finalize( let (final_status, failure_reason, run_status, status_reason) = classify_engine_result(&outcome); - let conclusion = build_conclusion( + let conclusion = build_conclusion_from_store( + options.run_store.as_ref(), &options.run_dir, final_status, failure_reason, duration_ms, options.last_git_sha.clone(), - ); + ) + .await; write_finalize_commit(&run_options, &options.run_dir).await; @@ -287,6 +399,19 @@ pub async fn finalize( } persist_terminal_outcome(&options.run_dir, &conclusion, run_status, status_reason); + if let Err(err) = options.run_store.put_conclusion(&conclusion).await { + tracing::warn!(error = %err, "Failed to save conclusion to store"); + } + if let Err(err) = options + .run_store + .put_status(&fabro_types::RunStatusRecord::new( + run_status, + status_reason, + )) + .await + { + tracing::warn!(error = %err, "Failed to save terminal status to store"); + } Ok(Concluded { run_id: run_options.run_id.clone(), @@ -304,8 +429,10 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; + use chrono::Utc; use fabro_config::FabroSettings; use fabro_graphviz::graph::Graph; + use fabro_store::{InMemoryStore, Store}; use super::*; use crate::pipeline::types::Retroed; @@ -333,10 +460,19 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); + let run_store = InMemoryStore::default() + .create_run( + "run-test", + Utc::now(), + Some(run_dir.to_string_lossy().as_ref()), + ) + .await + .unwrap(); let retroed = Retroed { graph: Graph::new("test"), outcome: Ok(Outcome::success()), run_options: test_run_options(&run_dir), + run_store: Arc::clone(&run_store), hook_runner: None, emitter: Arc::new(EventEmitter::new()), sandbox: Arc::new(fabro_agent::LocalSandbox::new( @@ -351,6 +487,7 @@ mod tests { &FinalizeOptions { run_dir: run_dir.clone(), run_id: "run-test".to_string(), + run_store, workflow_name: "test".to_string(), hook_runner: None, preserve_sandbox: true, diff --git a/lib/crates/fabro-workflows/src/pipeline/initialize.rs b/lib/crates/fabro-workflows/src/pipeline/initialize.rs index 79524387e..7157420d9 100644 --- a/lib/crates/fabro-workflows/src/pipeline/initialize.rs +++ b/lib/crates/fabro-workflows/src/pipeline/initialize.rs @@ -495,13 +495,13 @@ pub async fn initialize( options.emitter.emit(&WorkflowRunEvent::SandboxInitialized { working_directory: sandbox.working_directory().to_string(), }); - if let Err(e) = options - .sandbox - .to_sandbox_record(&*sandbox) - .save(&run_dir.join("sandbox.json")) - { + let sandbox_record = options.sandbox.to_sandbox_record(&*sandbox); + if let Err(e) = sandbox_record.save(&run_dir.join("sandbox.json")) { tracing::warn!(error = %e, "Failed to save sandbox record"); } + if let Err(err) = options.run_store.put_sandbox(&sandbox_record).await { + tracing::warn!(error = %err, "Failed to save sandbox record to store"); + } let env = build_sandbox_env( &options.sandbox_env, @@ -630,6 +630,7 @@ pub async fn initialize( graph, source, run_options: options.run_options, + run_store: options.run_store, checkpoint: options.checkpoint, seed_context: options.seed_context, emitter: options.emitter, @@ -655,6 +656,7 @@ mod tests { use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_interview::AutoApproveInterviewer; use fabro_sandbox::SandboxSpec; + use fabro_store::InMemoryStore; use super::*; use crate::pipeline::types::InitOptions; @@ -735,6 +737,12 @@ mod tests { persisted, InitOptions { run_id: "run-test".to_string(), + run_store: crate::operations::open_or_hydrate_run( + &InMemoryStore::default(), + &run_dir, + ) + .await + .unwrap(), dry_run: false, emitter, sandbox: SandboxSpec::Local { @@ -798,6 +806,12 @@ mod tests { persisted, InitOptions { run_id: "run-test".to_string(), + run_store: crate::operations::open_or_hydrate_run( + &InMemoryStore::default(), + &run_dir, + ) + .await + .unwrap(), dry_run: false, emitter, sandbox: SandboxSpec::Local { diff --git a/lib/crates/fabro-workflows/src/pipeline/persist.rs b/lib/crates/fabro-workflows/src/pipeline/persist.rs index 41392e08a..3120da770 100644 --- a/lib/crates/fabro-workflows/src/pipeline/persist.rs +++ b/lib/crates/fabro-workflows/src/pipeline/persist.rs @@ -1,5 +1,7 @@ use std::path::Path; +use fabro_store::RunStore; + use crate::error::FabroError; use crate::records::{RunRecord, RunRecordExt}; @@ -61,6 +63,31 @@ pub(crate) fn load(run_dir: &Path) -> Result { )) } +pub(crate) async fn load_from_store( + run_store: &dyn RunStore, + run_dir: &Path, +) -> Result { + let run_record = run_store + .get_run() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + .ok_or_else(|| FabroError::Precondition("run record missing from store".to_string()))?; + let graph = run_record.graph.clone(); + let source = run_store + .get_graph() + .await + .map_err(|err| FabroError::engine(err.to_string()))? + .unwrap_or_default(); + + Ok(Persisted::new( + graph, + source, + Vec::new(), + run_dir.to_path_buf(), + run_record, + )) +} + #[cfg(test)] mod tests { use std::collections::HashMap; diff --git a/lib/crates/fabro-workflows/src/pipeline/pull_request.rs b/lib/crates/fabro-workflows/src/pipeline/pull_request.rs index 557d1f884..6ad4a5f4a 100644 --- a/lib/crates/fabro-workflows/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflows/src/pipeline/pull_request.rs @@ -1,6 +1,7 @@ use std::path::Path; use fabro_config::run::MergeStrategy; +use fabro_store::RunStore; use serde::{Deserialize, Serialize}; use tracing::{debug, info}; @@ -321,6 +322,7 @@ pub async fn build_pr_body( diff: &str, goal: &str, model: &str, + run_store: Option<&dyn RunStore>, run_dir: &Path, conclusion: Option<&Conclusion>, ) -> Result { @@ -328,14 +330,58 @@ pub async fn build_pr_body( let plan_text = read_plan_text(run_dir); let loaded_conclusion = if conclusion.is_none() { - Conclusion::load(&run_dir.join("conclusion.json")).ok() + match run_store { + Some(run_store) => run_store + .get_conclusion() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load conclusion from store for PR body"); + }) + .ok() + .flatten() + .or_else(|| Conclusion::load(&run_dir.join("conclusion.json")).ok()), + None => Conclusion::load(&run_dir.join("conclusion.json")).ok(), + } } else { None }; let conclusion = conclusion.or(loaded_conclusion.as_ref()); - let retro = Retro::load(run_dir).ok(); - let run_record = RunRecord::load(run_dir).ok(); - let dot_source = read_dot_source(run_dir); + let retro = match run_store { + Some(run_store) => run_store + .get_retro() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load retro from store for PR body"); + }) + .ok() + .flatten() + .or_else(|| Retro::load(run_dir).ok()), + None => Retro::load(run_dir).ok(), + }; + let run_record = match run_store { + Some(run_store) => run_store + .get_run() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load run record from store for PR body"); + }) + .ok() + .flatten() + .or_else(|| RunRecord::load(run_dir).ok()), + None => RunRecord::load(run_dir).ok(), + }; + let dot_source = match run_store { + Some(run_store) => run_store + .get_graph() + .await + .inspect_err(|err| { + tracing::warn!(error = %err, "Failed to load graph from store for PR body"); + }) + .ok() + .flatten() + .or_else(|| read_dot_source(run_dir)), + None => read_dot_source(run_dir), + }; // Build LLM prompt let system = if plan_text.is_some() { @@ -413,6 +459,7 @@ pub async fn maybe_open_pull_request( model: &str, draft: bool, auto_merge: Option, + run_store: Option<&dyn RunStore>, run_dir: &Path, conclusion: Option<&Conclusion>, ) -> Result, String> { @@ -424,7 +471,7 @@ pub async fn maybe_open_pull_request( let https_url = ssh_url_to_https(origin_url); let (owner, repo) = github_app::parse_github_owner_repo(&https_url)?; - let body = build_pr_body(diff, goal, model, run_dir, conclusion).await?; + let body = build_pr_body(diff, goal, model, run_store, run_dir, conclusion).await?; let body = truncate_pr_body(&body); let title = pr_title_from_goal(goal); @@ -528,6 +575,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) -> &options.model, pr_cfg.draft, auto_merge, + options.run_store.as_deref(), &options.run_dir, Some(&conclusion), ) @@ -1014,6 +1062,7 @@ mod tests { "diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n", "Implement feature", "mock-model", + None, tmp.path(), Some(&conclusion), ) @@ -1217,6 +1266,7 @@ mod tests { "claude-sonnet-4-20250514", false, None, + None, tmp.path(), None, ) diff --git a/lib/crates/fabro-workflows/src/pipeline/retro.rs b/lib/crates/fabro-workflows/src/pipeline/retro.rs index a3ed8f229..bb0618704 100644 --- a/lib/crates/fabro-workflows/src/pipeline/retro.rs +++ b/lib/crates/fabro-workflows/src/pipeline/retro.rs @@ -5,14 +5,12 @@ use fabro_retro::RetroExt; use fabro_retro::retro::{Retro, derive_retro, extract_stage_durations}; use fabro_retro::retro_agent::{dry_run_narrative, run_retro_agent}; -use crate::event::WorkflowRunEvent; -use crate::records::{Checkpoint, CheckpointExt}; - use super::types::{Executed, RetroOptions, Retroed}; +use crate::event::WorkflowRunEvent; pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { - let cp = match Checkpoint::load(&options.run_dir.join("checkpoint.json")) { - Ok(cp) => cp, + let cp = match options.run_store.get_checkpoint().await { + Ok(Some(cp)) => cp, Err(e) => { tracing::warn!(error = %e, "Could not load checkpoint, skipping retro"); if let Some(ref emitter) = options.emitter { @@ -23,10 +21,26 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { } return None; } + Ok(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 completed_stages = crate::build_completed_stages(&cp, options.failed); - let stage_durations = extract_stage_durations(&options.run_dir); + let stage_durations = match options.run_store.list_events().await { + Ok(events) => crate::extract_stage_durations_from_events(&events), + Err(err) => { + tracing::warn!(error = %err, "Could not load events from store, falling back to disk"); + extract_stage_durations(&options.run_dir) + } + }; let mut retro = derive_retro( &options.run_id, &options.workflow_name, @@ -39,6 +53,9 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { if let Err(e) = retro.save(&options.run_dir) { tracing::warn!(error = %e, "Failed to save initial retro"); } + if let Err(err) = options.run_store.put_retro(&retro).await { + tracing::warn!(error = %err, "Failed to save initial retro to store"); + } let retro_start = std::time::Instant::now(); if let Some(ref emitter) = options.emitter { @@ -101,6 +118,9 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option { if let Err(e) = retro.save(&options.run_dir) { tracing::warn!(error = %e, "Failed to save retro with narrative"); } + if let Err(err) = options.run_store.put_retro(&retro).await { + tracing::warn!(error = %err, "Failed to save retro with narrative to store"); + } } Err(e) => { tracing::debug!(error = %e, "Retro agent skipped"); @@ -119,6 +139,7 @@ pub async fn retro(executed: Executed, options: &RetroOptions) -> Retroed { graph, outcome, run_options, + run_store, hook_runner, emitter, sandbox, @@ -141,6 +162,7 @@ pub async fn retro(executed: Executed, options: &RetroOptions) -> Retroed { graph, outcome, run_options, + run_store, hook_runner, emitter, sandbox, @@ -154,17 +176,19 @@ mod tests { use std::collections::HashMap; use std::sync::{Arc, Mutex}; + use chrono::Utc; use fabro_config::FabroSettings; use fabro_graphviz::graph::Graph; + use fabro_store::{InMemoryStore, Store}; use super::*; use crate::context::Context; use crate::event::{EventEmitter, WorkflowRunEvent}; use crate::pipeline::types::Executed; - use crate::records::Checkpoint; + use crate::records::{Checkpoint, CheckpointExt}; use crate::run_options::RunOptions; - fn write_checkpoint(run_dir: &std::path::Path) { + fn write_checkpoint(run_dir: &std::path::Path) -> Checkpoint { let context = Context::new(); context.set("response.work", serde_json::json!("done")); let mut outcomes = HashMap::new(); @@ -181,6 +205,23 @@ mod tests { HashMap::new(), ); checkpoint.save(&run_dir.join("checkpoint.json")).unwrap(); + checkpoint + } + + async fn test_run_store( + run_dir: &std::path::Path, + checkpoint: &Checkpoint, + ) -> Arc { + let run_store = InMemoryStore::default() + .create_run( + "run-test", + Utc::now(), + Some(run_dir.to_string_lossy().as_ref()), + ) + .await + .unwrap(); + run_store.put_checkpoint(checkpoint).await.unwrap(); + run_store } fn test_run_options(run_dir: &std::path::Path) -> RunOptions { @@ -205,7 +246,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - write_checkpoint(&run_dir); + let checkpoint = write_checkpoint(&run_dir); let emitter = Arc::new(EventEmitter::new()); let sandbox: Arc = Arc::new(fabro_agent::LocalSandbox::new( @@ -215,6 +256,7 @@ mod tests { graph: Graph::new("test"), outcome: Ok(crate::outcome::Outcome::success()), run_options: test_run_options(&run_dir), + run_store: test_run_store(&run_dir, &checkpoint).await, hook_runner: None, emitter: Arc::clone(&emitter), sandbox: Arc::clone(&sandbox), @@ -229,6 +271,7 @@ mod tests { executed, &RetroOptions { run_id: "run-test".to_string(), + run_store: test_run_store(&run_dir, &checkpoint).await, workflow_name: "test".to_string(), goal: "Ship it".to_string(), run_dir: run_dir.clone(), @@ -253,7 +296,7 @@ mod tests { let temp = tempfile::tempdir().unwrap(); let run_dir = temp.path().join("run"); std::fs::create_dir_all(&run_dir).unwrap(); - write_checkpoint(&run_dir); + let checkpoint = write_checkpoint(&run_dir); let emitter = Arc::new(EventEmitter::new()); let seen = Arc::new(Mutex::new(Vec::new())); @@ -265,6 +308,7 @@ mod tests { let retro = run_retro( &RetroOptions { run_id: "run-test".to_string(), + run_store: test_run_store(&run_dir, &checkpoint).await, workflow_name: "test".to_string(), goal: "Ship it".to_string(), run_dir: run_dir.clone(), diff --git a/lib/crates/fabro-workflows/src/pipeline/types.rs b/lib/crates/fabro-workflows/src/pipeline/types.rs index e5c5b4ea1..03e0cb315 100644 --- a/lib/crates/fabro-workflows/src/pipeline/types.rs +++ b/lib/crates/fabro-workflows/src/pipeline/types.rs @@ -11,6 +11,7 @@ use fabro_llm::Provider; use fabro_mcp::config::McpServerConfig; use fabro_model::FallbackTarget; use fabro_sandbox::SandboxSpec; +use fabro_store::RunStore; use fabro_validate::Diagnostic; use crate::context::Context; @@ -196,6 +197,13 @@ impl Persisted { pub fn load(run_dir: &Path) -> Result { super::persist::load(run_dir) } + + pub async fn load_from_store( + run_store: &dyn RunStore, + run_dir: &Path, + ) -> Result { + super::persist::load_from_store(run_store, run_dir).await + } } #[derive(Clone)] @@ -223,6 +231,7 @@ pub struct DevcontainerSpec { pub struct InitOptions { pub run_id: String, + pub run_store: Arc, pub dry_run: bool, pub emitter: Arc, pub sandbox: SandboxSpec, @@ -246,6 +255,7 @@ pub struct Initialized { pub graph: Graph, pub source: String, pub run_options: RunOptions, + pub run_store: Arc, pub(crate) checkpoint: Option, pub(crate) seed_context: Option, pub emitter: Arc, @@ -266,6 +276,7 @@ pub struct Executed { pub graph: Graph, pub outcome: Result, pub run_options: RunOptions, + pub run_store: Arc, pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, @@ -282,6 +293,7 @@ pub struct Retroed { pub graph: Graph, pub outcome: Result, pub run_options: RunOptions, + pub run_store: Arc, pub hook_runner: Option>, pub emitter: Arc, pub sandbox: Arc, @@ -320,6 +332,7 @@ pub struct TransformOptions { /// Options for the RETRO phase. pub struct RetroOptions { pub run_id: String, + pub run_store: Arc, pub workflow_name: String, pub goal: String, pub run_dir: PathBuf, @@ -337,6 +350,7 @@ pub struct RetroOptions { pub struct FinalizeOptions { pub run_dir: PathBuf, pub run_id: String, + pub run_store: Arc, pub workflow_name: String, pub hook_runner: Option>, pub preserve_sandbox: bool, @@ -346,6 +360,7 @@ pub struct FinalizeOptions { /// Options for the PULL_REQUEST phase. pub struct PullRequestOptions { pub run_dir: PathBuf, + pub run_store: Option>, pub pr_config: Option, pub github_app: Option, pub origin_url: Option, diff --git a/lib/crates/fabro-workflows/src/run_lookup.rs b/lib/crates/fabro-workflows/src/run_lookup.rs index 995ece82f..bb314eb01 100644 --- a/lib/crates/fabro-workflows/src/run_lookup.rs +++ b/lib/crates/fabro-workflows/src/run_lookup.rs @@ -3,6 +3,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result, bail}; use chrono::{DateTime, Utc}; +use fabro_store::{ListRunsQuery, Store}; use serde::Serialize; use crate::records::{ @@ -149,6 +150,67 @@ pub fn scan_runs(base: &Path) -> Result> { Ok(runs) } +pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result> { + let mut runs_by_id: HashMap = scan_runs(base)? + .into_iter() + .map(|run| (run.run_id.clone(), run)) + .collect(); + + if let Ok(store_runs) = store.list_runs(&ListRunsQuery::default()).await { + for summary in store_runs { + let Some(run_dir) = summary.run_dir.as_deref() else { + continue; + }; + let path = PathBuf::from(run_dir); + let Some(dir_name) = path + .file_name() + .map(|name| name.to_string_lossy().to_string()) + else { + continue; + }; + let start_time_dt = summary.created_at; + let start_time = summary.start_time.unwrap_or(start_time_dt); + let end_time = if summary.status.is_some_and(|status| status.is_terminal()) { + summary.duration_ms.and_then(|duration_ms| { + Some( + start_time_dt + + chrono::Duration::milliseconds(i64::try_from(duration_ms).ok()?), + ) + }) + } else { + None + }; + runs_by_id.insert( + summary.run_id.clone(), + RunInfo { + run_id: summary.run_id, + dir_name, + workflow_name: summary + .workflow_name + .unwrap_or_else(|| "[starting]".to_string()), + workflow_slug: summary.workflow_slug, + status: summary.status.unwrap_or(RunStatus::Dead), + status_reason: summary.status_reason, + start_time: start_time.to_rfc3339(), + labels: summary.labels, + duration_ms: summary.duration_ms, + total_cost: summary.total_cost, + host_repo_path: summary.host_repo_path, + goal: summary.goal.unwrap_or_default(), + start_time_dt: Some(start_time_dt), + end_time, + path, + is_orphan: false, + }, + ); + } + } + + let mut runs: Vec<_> = runs_by_id.into_values().collect(); + runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt)); + Ok(runs) +} + struct StatusInfo { status: RunStatus, reason: Option, @@ -298,6 +360,52 @@ pub fn resolve_run(base: &Path, identifier: &str) -> Result { } } +pub async fn resolve_run_combined( + store: &dyn Store, + base: &Path, + identifier: &str, +) -> Result { + let runs = scan_runs_combined(store, base) + .await + .context("Failed to scan runs")?; + + let id_matches: Vec<_> = runs + .iter() + .filter(|run| run.run_id.starts_with(identifier)) + .collect(); + + match id_matches.len() { + 1 => return Ok(id_matches[0].clone()), + count if count > 1 => { + let ids: Vec<&str> = id_matches.iter().map(|run| run.run_id.as_str()).collect(); + bail!( + "Ambiguous prefix '{identifier}': {count} runs match: {}", + ids.join(", ") + ) + } + _ => {} + } + + let id_lower = identifier.to_lowercase(); + let id_collapsed = collapse_separators(&id_lower); + let workflow_match = runs.iter().filter(|run| !run.is_orphan).find(|run| { + if let Some(slug) = &run.workflow_slug { + if slug.to_lowercase() == id_lower { + return true; + } + } + let name_lower = run.workflow_name.to_lowercase(); + name_lower.contains(&id_lower) || collapse_separators(&name_lower).contains(&id_collapsed) + }); + + match workflow_match { + Some(run) => Ok(run.clone()), + None => { + bail!("No run found matching '{identifier}' (tried run ID prefix and workflow name)") + } + } +} + fn collapse_separators(s: &str) -> String { s.chars().filter(|c| *c != '-' && *c != '_').collect() } diff --git a/lib/crates/fabro-workflows/src/test_support.rs b/lib/crates/fabro-workflows/src/test_support.rs index eb17a8585..d860f3122 100644 --- a/lib/crates/fabro-workflows/src/test_support.rs +++ b/lib/crates/fabro-workflows/src/test_support.rs @@ -1,8 +1,10 @@ use std::collections::HashMap; use std::sync::Arc; +use chrono::Utc; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; +use fabro_store::{InMemoryStore, Store}; use crate::error::Result; use crate::event::EventEmitter; @@ -19,7 +21,7 @@ struct InitializedOptions { checkpoint: Option, } -fn initialized( +async fn initialized( registry: HandlerRegistry, emitter: Arc, sandbox: Arc, @@ -28,10 +30,19 @@ fn initialized( options: InitializedOptions, ) -> Initialized { std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir"); + let run_store = InMemoryStore::default() + .create_run( + &run_options.run_id, + Utc::now(), + Some(run_options.run_dir.to_string_lossy().as_ref()), + ) + .await + .expect("failed to create in-memory run store"); Initialized { graph: graph.clone(), source: String::new(), run_options: run_options.clone(), + run_store, checkpoint: options.checkpoint, seed_context: None, emitter, @@ -54,7 +65,7 @@ pub async fn run_graph( graph: &GvGraph, run_options: &RunOptions, ) -> Result { - let executed = pipeline::execute(initialized( + let initialized = initialized( registry, emitter, sandbox, @@ -65,8 +76,9 @@ pub async fn run_graph( env: HashMap::new(), checkpoint: None, }, - )) + ) .await; + let executed = pipeline::execute(initialized).await; executed.outcome } @@ -79,7 +91,7 @@ pub async fn run_graph_with_hooks( hook_runner: Arc, env: Option>, ) -> Result { - let executed = pipeline::execute(initialized( + let initialized = initialized( registry, emitter, sandbox, @@ -90,8 +102,9 @@ pub async fn run_graph_with_hooks( env: env.unwrap_or_default(), checkpoint: None, }, - )) + ) .await; + let executed = pipeline::execute(initialized).await; executed.outcome } @@ -103,7 +116,7 @@ pub async fn run_graph_from_checkpoint( run_options: &RunOptions, checkpoint: &Checkpoint, ) -> Result { - let executed = pipeline::execute(initialized( + let initialized = initialized( registry, emitter, sandbox, @@ -114,8 +127,9 @@ pub async fn run_graph_from_checkpoint( env: HashMap::new(), checkpoint: Some(checkpoint.clone()), }, - )) + ) .await; + let executed = pipeline::execute(initialized).await; executed.outcome }