scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-08 22:21:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Wire workflows and CLI to fabro-store

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-28 22:57:49 -04:00
parent 1a0874e22e
commit 4301c34c81
No known key found for this signature in database
45 changed files with 2011 additions and 278 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
Cargo.lock generated
View file

@ -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",

2
lib/crates/fabro-api/Cargo.toml
View file

@ -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"] }

10
lib/crates/fabro-api/src/serve.rs
View file

@ -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);

95
lib/crates/fabro-api/src/server.rs
View file

@ -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<dyn Interviewer>) -> HandlerRegistry +
pub struct AppState {
runs: Mutex<HashMap<String, ManagedRun>>,
aggregate_usage: Mutex<AggregateUsageTotals>,
store: Arc<dyn Store>,
llm_spec_factory: Box<LlmSpecFactory>,
registry_factory_override: Option<Box<RegistryFactoryOverride>>,
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<fabro_hooks::HookDefinition>,
) -> Arc<AppState> {
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<fabro_hooks::HookDefinition>,
store: Arc<dyn Store>,
) -> Arc<AppState> {
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<fabro_hooks::HookDefinition>,
store: Arc<dyn Store>,
) -> Arc<AppState> {
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<AppState>, 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<AppState>, run_id: String) {
let interviewer = Arc::clone(&interviewer) as Arc<dyn Interviewer>;
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<AppState>, run_id: String) {
persisted,
InitOptions {
run_id,
run_store,
dry_run,
emitter,
sandbox,
@ -761,13 +803,26 @@ async fn execute_run(state: Arc<AppState>, 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<AppState>, 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(),
}
}
}
}

1
lib/crates/fabro-cli/Cargo.toml
View file

@ -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"

44
lib/crates/fabro-cli/src/commands/pr/create.rs
View file

@ -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<fabro_github::GitHubAppCredentials>,
) -> 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,
)

13
lib/crates/fabro-cli/src/commands/run/detached.rs
View file

@ -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<dyn Fn(&str) + Send + Sync>
});
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,

52
lib/crates/fabro-cli/src/commands/run/diff.rs
View file

@ -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<String> {
async fn resolve_diff(
run_dir: &Path,
run_store: Option<&dyn fabro_store::RunStore>,
args: &DiffArgs,
) -> Result<String> {
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<String> {
});
}
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<String> {
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<String> {
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?;

124
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -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::<Result<Vec<_>>>()?;
(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<String> {
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());

4
lib/crates/fabro-cli/src/commands/run/mod.rs
View file

@ -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
}
}
}

26
lib/crates/fabro-cli/src/commands/run/preview.rs
View file

@ -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")?;

26
lib/crates/fabro-cli/src/commands/run/ssh.rs
View file

@ -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")?;

35
lib/crates/fabro-cli/src/commands/run/wait.rs
View file

@ -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());

48
lib/crates/fabro-cli/src/commands/runs/inspect.rs
View file

@ -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<serde_json::Value>,
}
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()

7
lib/crates/fabro-cli/src/commands/runs/list.rs
View file

@ -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,

4
lib/crates/fabro-cli/src/commands/runs/mod.rs
View file

@ -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,
}
}

1
lib/crates/fabro-cli/src/main.rs
View file

@ -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};

23
lib/crates/fabro-cli/src/store.rs Normal file
View file

@ -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<Arc<SlateStore>> {
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<Option<Arc<dyn RunStore>>> {
build_store(storage_dir)?
.open_run_reader(run_id)
.await
.map_err(Into::into)
}

2
lib/crates/fabro-store/src/error.rs
View file

@ -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),
}

7
lib/crates/fabro-store/src/lib.rs
View file

@ -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<Utc>,
) -> Result<Box<dyn RunStore>>;
async fn open_run(&self, run_id: &str) -> Result<Option<Box<dyn RunStore>>>;
run_dir: Option<&str>,
) -> Result<Arc<dyn RunStore>>;
async fn open_run(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>>;
async fn open_run_reader(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>>;
async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>>;
async fn delete_run(&self, run_id: &str) -> Result<()>;
}

52
lib/crates/fabro-store/src/memory.rs
View file

@ -44,11 +44,17 @@ struct InMemoryRunStore {
}
impl InMemoryRunStore {
fn new(run_id: &str, created_at: DateTime<Utc>, db_prefix: String) -> Result<Self> {
fn new(
run_id: &str,
created_at: DateTime<Utc>,
db_prefix: String,
run_dir: Option<String>,
) -> Result<Self> {
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<Utc>,
) -> Result<Box<dyn RunStore>> {
run_dir: Option<&str>,
) -> Result<Arc<dyn RunStore>> {
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<dyn RunStore>);
}
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<dyn RunStore>)
}
async fn open_run(&self, run_id: &str) -> Result<Option<Box<dyn RunStore>>> {
async fn open_run(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>> {
let runs = self.runs.lock().await;
Ok(runs
.get(run_id)
.map(|catalog| Box::new(Arc::clone(&catalog.run_store)) as Box<dyn RunStore>))
.map(|catalog| Arc::clone(&catalog.run_store) as Arc<dyn RunStore>))
}
async fn open_run_reader(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>> {
self.open_run(run_id).await
}
async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
@ -276,7 +289,7 @@ impl Store for InMemoryStore {
}
#[async_trait]
impl RunStore for Arc<InMemoryRunStore> {
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"),

2
lib/crates/fabro-store/src/slate/catalog.rs
View file

@ -14,11 +14,13 @@ pub(crate) async fn write_catalog(
run_id: &str,
created_at: DateTime<Utc>,
db_prefix: &str,
run_dir: Option<&str>,
) -> Result<CatalogRecord> {
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

84
lib/crates/fabro-store/src/slate/mod.rs
View file

@ -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<Option<SlateRunStore>> {
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<Utc>,
) -> Result<Box<dyn RunStore>> {
run_dir: Option<&str>,
) -> Result<Arc<dyn RunStore>> {
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<dyn RunStore>);
}
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<dyn RunStore>)
}
async fn open_run(&self, run_id: &str) -> Result<Option<Box<dyn RunStore>>> {
async fn open_run(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>> {
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<dyn RunStore>))
}
async fn open_run_reader(&self, run_id: &str) -> Result<Option<Arc<dyn RunStore>>> {
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<dyn RunStore>))
}
async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
@ -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();

242
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -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<Utc>,
db_prefix: String,
db: slatedb::Db,
run_dir: Option<String>,
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<Self> {
pub(crate) async fn open_writer(record: CatalogRecord, db: slatedb::Db) -> Result<Self> {
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<Self> {
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<Utc> {
@ -93,7 +120,10 @@ impl SlateRunStore {
}
pub(crate) async fn snapshot(&self) -> Result<Arc<slatedb::DbSnapshot>> {
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<R>(db: &R, expected: &CatalogRecord) -> Result<bool>
@ -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<Option<RunRecord>> {
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<Option<StartRecord>> {
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<Option<RunStatusRecord>> {
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<Option<Checkpoint>> {
get_json(&self.inner.db, keys::checkpoint()).await
self.inner.db.get_json(keys::checkpoint()).await
}
async fn append_checkpoint(&self, record: &Checkpoint) -> Result<u32> {
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<Vec<(u32, Checkpoint)>> {
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<Option<Conclusion>> {
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<Option<Retro>> {
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<Option<String>> {
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<Option<SandboxRecord>> {
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<NodeSnapshot> {
@ -308,28 +349,29 @@ impl RunStore for SlateRunStore {
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
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<Vec<EventEnvelope>> {
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<Vec<EventEnvelope>> {
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<std::pin::Pin<Box<dyn Stream<Item = Result<EventEnvelope>> + 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<Option<String>> {
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<Option<String>> {
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<Option<serde_json::Value>> {
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<Option<Bytes>> {
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<Vec<String>> {
@ -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<T: DeserializeOwned>(&self, key: &str) -> Result<Option<T>> {
match self {
Self::Writer(db) => get_json(db, key).await,
Self::Reader(db) => get_json(db, key).await,
}
}
async fn put_json<T: Serialize>(&self, key: &str, value: &T) -> Result<()> {
put_json(self.writer()?, key, value).await
}
async fn get_text(&self, key: &str) -> Result<Option<String>> {
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<Option<Bytes>> {
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<P>(
&self,
prefix: P,
) -> std::result::Result<slatedb::DbIterator, slatedb::Error>
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<Vec<EventEnvelope>> {
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<Vec<(u32, Checkpoint)>> {
match self {
Self::Writer(db) => list_checkpoints(db).await,
Self::Reader(db) => list_checkpoints(db).await,
}
}
}
async fn put_json<T: Serialize>(db: &slatedb::Db, key: &str, value: &T) -> Result<()> {
db.put(key, serde_json::to_vec(value)?).await?;
Ok(())

2
lib/crates/fabro-store/src/types.rs
View file

@ -20,6 +20,7 @@ pub struct CatalogRecord {
pub run_id: String,
pub created_at: DateTime<Utc>,
pub db_prefix: String,
pub run_dir: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@ -27,6 +28,7 @@ pub struct RunSummary {
pub run_id: String,
pub created_at: DateTime<Utc>,
pub db_prefix: String,
pub run_dir: Option<String>,
pub workflow_name: Option<String>,
pub workflow_slug: Option<String>,
pub goal: Option<String>,

90
lib/crates/fabro-workflows/src/event.rs
View file

@ -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<EventPayload> {
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<StoreProgressCommand>,
run_id: Arc<std::sync::Mutex<String>>,
}
impl StoreProgressLogger {
#[must_use]
pub fn new(run_store: Arc<dyn RunStore>, run_id: impl Into<String>) -> 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<String, serde_json::Value>) {
let serde_json::Value::Object(mut agent_fields) = inner else {
return ("Agent".to_string(), serde_json::Map::new());

11
lib/crates/fabro-workflows/src/handler/manager_loop.rs
View file

@ -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,

20
lib/crates/fabro-workflows/src/lib.rs
View file

@ -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<String, u64> {
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;

62
lib/crates/fabro-workflows/src/lifecycle/disk.rs
View file

@ -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<Option<StageUsage>>;
pub(crate) struct DiskLifecycle {
pub run_dir: PathBuf,
pub run_id: String,
pub run_store: Arc<dyn RunStore>,
pub graph: Arc<GvGraph>,
pub run_options: Arc<RunOptions>,
pub emitter: Arc<EventEmitter>,
@ -38,9 +41,27 @@ pub(crate) struct DiskLifecycle {
impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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<WorkflowGraph> 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(())
}

121
lib/crates/fabro-workflows/src/lifecycle/git.rs
View file

@ -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<EventEmitter>,
pub run_dir: PathBuf,
pub run_id: String,
pub run_store: Arc<dyn RunStore>,
pub run_options: Arc<RunOptions>,
pub start_node_id: Option<String>,
// Cross-lifecycle data (shared with EventLifecycle)
@ -61,9 +63,30 @@ impl RunLifecycle<WorkflowGraph> 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<WorkflowGraph> 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<u8>)> = 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<u8>)> = 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<WorkflowGraph> 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() {

4
lib/crates/fabro-workflows/src/lifecycle/mod.rs
View file

@ -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<dyn Sandbox>,
graph: Arc<GvGraph>,
run_dir: &PathBuf,
run_store: Arc<dyn RunStore>,
run_options: &Arc<RunOptions>,
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),

301
lib/crates/fabro-workflows/src/operations/hydrate.rs Normal file
View file

@ -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<Arc<dyn RunStore>, FabroError> {
let record = RunRecord::load(run_dir)?;
if let Some(run_store) = store.open_run(&record.run_id).await.map_err(store_error)? {
return Ok(run_store);
}
let run_dir_string = run_dir.to_string_lossy().to_string();
let run_store = store
.create_run(&record.run_id, record.created_at, Some(&run_dir_string))
.await
.map_err(store_error)?;
run_store.put_run(&record).await.map_err(store_error)?;
if let Some(dot_source) = load_graph_source(run_dir)? {
run_store
.put_graph(&dot_source)
.await
.map_err(store_error)?;
}
if let Some(status) = load_status_record(run_dir)? {
run_store.put_status(&status).await.map_err(store_error)?;
}
if let Some(start) = load_start_record(run_dir)? {
run_store.put_start(&start).await.map_err(store_error)?;
}
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::<serde_json::Value>(trimmed) {
Ok(value) => value,
Err(err) => {
warn!(
path = %progress_path.display(),
line_number = line_number + 1,
error = %err,
"Skipping malformed progress event during hydration"
);
continue;
}
};
let payload = match EventPayload::new(value, run_id) {
Ok(payload) => payload,
Err(err) => {
warn!(
path = %progress_path.display(),
line_number = line_number + 1,
error = %err,
"Skipping invalid progress event during hydration"
);
continue;
}
};
run_store
.append_event(&payload)
.await
.map_err(store_error)?;
}
Ok(())
}
fn load_graph_source(run_dir: &Path) -> Result<Option<String>, FabroError> {
for name in [GRAPH_FILE_NAME, LEGACY_GRAPH_FILE_NAME] {
match std::fs::read_to_string(run_dir.join(name)) {
Ok(source) => return Ok(Some(source)),
Err(err) if err.kind() == ErrorKind::NotFound => {}
Err(err) => return Err(err.into()),
}
}
Ok(None)
}
fn load_status_record(run_dir: &Path) -> Result<Option<RunStatusRecord>, FabroError> {
let path = run_dir.join("status.json");
if !path.exists() {
return Ok(None);
}
Ok(Some(
RunStatusRecord::load(&path).map_err(|err| FabroError::Io(err.to_string()))?,
))
}
fn load_start_record(run_dir: &Path) -> Result<Option<StartRecord>, FabroError> {
let path = run_dir.join("start.json");
if !path.exists() {
return Ok(None);
}
Ok(Some(StartRecord::load(run_dir)?))
}
fn load_checkpoint(run_dir: &Path) -> Result<Option<Checkpoint>, FabroError> {
let path = run_dir.join("checkpoint.json");
if !path.exists() {
return Ok(None);
}
Ok(Some(Checkpoint::load(&path)?))
}
fn load_conclusion(run_dir: &Path) -> Result<Option<Conclusion>, FabroError> {
let path = run_dir.join("conclusion.json");
if !path.exists() {
return Ok(None);
}
Ok(Some(Conclusion::load(&path)?))
}
fn load_retro(run_dir: &Path) -> Result<Option<Retro>, FabroError> {
let path = run_dir.join("retro.json");
if !path.exists() {
return Ok(None);
}
Retro::load(run_dir)
.map(Some)
.map_err(|err| FabroError::Io(err.to_string()))
}
fn load_sandbox_record(run_dir: &Path) -> Result<Option<SandboxRecord>, FabroError> {
let path = run_dir.join("sandbox.json");
if !path.exists() {
return Ok(None);
}
SandboxRecord::load(&path)
.map(Some)
.map_err(|err| FabroError::Io(err.to_string()))
}
fn store_error(err: impl std::fmt::Display) -> FabroError {
FabroError::engine(err.to_string())
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use chrono::Utc;
use fabro_config::FabroSettings;
use fabro_graphviz::graph::Graph;
use fabro_store::{InMemoryStore, Store};
use fabro_types::{Conclusion, RunStatus, RunStatusRecord, StageStatus};
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())
);
}
}

2
lib/crates/fabro-workflows/src/operations/mod.rs
View file

@ -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,

34
lib/crates/fabro-workflows/src/operations/resume.rs
View file

@ -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<Started, FabroError> {
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<Started,
}
}
let cp_path = run_dir.join("checkpoint.json");
let checkpoint = Checkpoint::load(&cp_path)
.map_err(|e| FabroError::Precondition(format!("no checkpoint to resume from: {e}")))?;
let checkpoint = services
.run_store
.get_checkpoint()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
.ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
cleanup_resume_artifacts(run_dir);
run_status::write_run_status(run_dir, RunStatus::Submitted, None);
services
.run_store
.put_status(&run_status::RunStatusRecord::new(
RunStatus::Submitted,
None,
))
.await
.map_err(|err| FabroError::engine(err.to_string()))?;
Box::pin(execute_persisted_run(run_dir, Some(checkpoint), services)).await
}

253
lib/crates/fabro-workflows/src/operations/start.rs
View file

@ -11,12 +11,14 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
use fabro_interview::{AutoApproveInterviewer, Interviewer};
use fabro_model::{Catalog, FallbackTarget, Provider};
use fabro_sandbox::{SandboxProvider, SandboxSpec, detect_clone_params};
use fabro_store::RunStore;
use serde::Serialize;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{
EventEmitter, ProgressLogger, RunNoticeLevel, WorkflowRunEvent, append_progress_event,
EventEmitter, ProgressLogger, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent,
append_progress_event, build_redacted_event_payload,
};
use crate::git::GitAuthor;
use crate::handler::HandlerRegistry;
@ -28,7 +30,7 @@ use crate::pipeline::{
};
use crate::records::{Checkpoint, Conclusion, ConclusionExt, RunRecord, RunRecordExt};
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::run_status::{self, RunStatus, RunStatusRecordExt, StatusReason};
use crate::run_status::{self, RunStatus, StatusReason};
use fabro_config::run::PullRequestSettings;
use fabro_retro::retro::Retro;
use fabro_sandbox::daytona::DaytonaConfig;
@ -48,6 +50,7 @@ struct RunSession {
sandbox_env: SandboxEnvSpec,
devcontainer: Option<DevcontainerSpec>,
seed_context: Option<Context>,
run_store: Arc<dyn RunStore>,
git_author: GitAuthor,
git: Option<GitCheckpointOptions>,
github_app: Option<fabro_github::GitHubAppCredentials>,
@ -65,6 +68,7 @@ pub struct StartServices {
pub cancel_token: Option<Arc<AtomicBool>>,
pub emitter: Arc<EventEmitter>,
pub interviewer: Arc<dyn Interviewer>,
pub run_store: Arc<dyn RunStore>,
pub git_author: GitAuthor,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
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<Started, FabroError> {
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<Checkpoint>,
services: StartServices,
) -> Result<Started, FabroError> {
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<Outcome, FabroError> = 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<dyn RunStore>,
active: bool,
}
impl DetachedRunBootstrapGuard {
fn arm(run_dir: &Path) -> Self {
fn arm(run_dir: &Path, run_store: Arc<dyn RunStore>) -> 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<dyn RunStore>,
run_id: Option<String>,
active: bool,
}
impl DetachedRunCompletionGuard {
fn arm(run_dir: &Path) -> Self {
fn arm(run_dir: &Path, run_store: Arc<dyn RunStore>) -> 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<String> {
})
}
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<StatusReason>,
) -> Result<(), FabroError> {
) -> Result<Conclusion, FabroError> {
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<EventEmitter>,
registry: Arc<HandlerRegistry>,
) -> 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(_))),

6
lib/crates/fabro-workflows/src/pipeline/execute.rs
View file

@ -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,

9
lib/crates/fabro-workflows/src/pipeline/execute/tests.rs
View file

@ -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<String>) -> LifecycleOptions {
}
}
async fn test_run_store(run_dir: &Path) -> Arc<dyn fabro_store::RunStore> {
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 {

141
lib/crates/fabro-workflows/src/pipeline/finalize.rs
View file

@ -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<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> 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<String, u64>,
status: StageStatus,
failure_reason: Option<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> 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<f64> = 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,

24
lib/crates/fabro-workflows/src/pipeline/initialize.rs
View file

@ -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 {

27
lib/crates/fabro-workflows/src/pipeline/persist.rs
View file

@ -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<Persisted, FabroError> {
))
}
pub(crate) async fn load_from_store(
run_store: &dyn RunStore,
run_dir: &Path,
) -> Result<Persisted, FabroError> {
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;

60
lib/crates/fabro-workflows/src/pipeline/pull_request.rs
View file

@ -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<String, String> {
@ -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<AutoMergeOptions>,
run_store: Option<&dyn RunStore>,
run_dir: &Path,
conclusion: Option<&Conclusion>,
) -> Result<Option<PullRequestRecord>, 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,
)

64
lib/crates/fabro-workflows/src/pipeline/retro.rs
View file

@ -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<Retro> {
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<Retro> {
}
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<Retro> {
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<Retro> {
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<dyn fabro_store::RunStore> {
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<dyn fabro_agent::Sandbox> = 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(),

15
lib/crates/fabro-workflows/src/pipeline/types.rs
View file

@ -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<Self, FabroError> {
super::persist::load(run_dir)
}
pub async fn load_from_store(
run_store: &dyn RunStore,
run_dir: &Path,
) -> Result<Self, FabroError> {
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<dyn RunStore>,
pub dry_run: bool,
pub emitter: Arc<EventEmitter>,
pub sandbox: SandboxSpec,
@ -246,6 +255,7 @@ pub struct Initialized {
pub graph: Graph,
pub source: String,
pub run_options: RunOptions,
pub run_store: Arc<dyn RunStore>,
pub(crate) checkpoint: Option<Checkpoint>,
pub(crate) seed_context: Option<Context>,
pub emitter: Arc<EventEmitter>,
@ -266,6 +276,7 @@ pub struct Executed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: Arc<dyn RunStore>,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -282,6 +293,7 @@ pub struct Retroed {
pub graph: Graph,
pub outcome: Result<Outcome, FabroError>,
pub run_options: RunOptions,
pub run_store: Arc<dyn RunStore>,
pub hook_runner: Option<Arc<HookRunner>>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
@ -320,6 +332,7 @@ pub struct TransformOptions {
/// Options for the RETRO phase.
pub struct RetroOptions {
pub run_id: String,
pub run_store: Arc<dyn RunStore>,
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<dyn RunStore>,
pub workflow_name: String,
pub hook_runner: Option<Arc<HookRunner>>,
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<Arc<dyn RunStore>>,
pub pr_config: Option<PullRequestSettings>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub origin_url: Option<String>,

108
lib/crates/fabro-workflows/src/run_lookup.rs
View file

@ -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<Vec<RunInfo>> {
Ok(runs)
}
pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result<Vec<RunInfo>> {
let mut runs_by_id: HashMap<String, RunInfo> = 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<StatusReason>,
@ -298,6 +360,52 @@ pub fn resolve_run(base: &Path, identifier: &str) -> Result<RunInfo> {
}
}
pub async fn resolve_run_combined(
store: &dyn Store,
base: &Path,
identifier: &str,
) -> Result<RunInfo> {
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()
}

28
lib/crates/fabro-workflows/src/test_support.rs
View file

@ -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<Checkpoint>,
}
fn initialized(
async fn initialized(
registry: HandlerRegistry,
emitter: Arc<EventEmitter>,
sandbox: Arc<dyn Sandbox>,
@ -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<Outcome> {
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<fabro_hooks::HookRunner>,
env: Option<HashMap<String, String>>,
) -> Result<Outcome> {
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<Outcome> {
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
}