scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Require a durable run store throughout execution

Browse source 
Make run lookup fail with RunNotFound instead of returning Option, thread a required RunStore through workflow and retro paths, and update CLI, server, and test callers to match. Also treat null optional event properties as absent during store-backed replay so event-sourced state stays robust.
This commit is contained in:
Bryan Helmkamp 2026-04-02 15:08:21 -07:00
parent f786f91fe7
commit 031d4cdbb6
47 changed files with 1724 additions and 666 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -34,23 +34,19 @@ async fn create_from(
let store = 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 = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let run_store = store::open_run_reader(storage_dir, &run.run_id).await?;
let state = run_store.state().await?;
let record = run_store
.get_run()
.await?
let record = state
.run
.context("Failed to load run record from store")?;
let start = run_store
.get_start()
.await?
let start = state
.start
.context("Failed to load start record from store")?;
let conclusion = run_store
.get_conclusion()
.await?
let conclusion = state
.conclusion
.context("Failed to load conclusion from store — is the run finished?")?;
match conclusion.status {
@ -63,9 +59,8 @@ async fn create_from(
.as_deref()
.context("Run has no run_branch — was it run with git push enabled?")?;
let diff = run_store
.get_final_patch()
.await?
let diff = state
.final_patch
.context("Failed to load final patch from store — no diff available")?;
if diff.trim().is_empty() {
bail!("final.patch is empty — nothing to create a PR for");
@ -120,7 +115,7 @@ async fn create_from(
&model,
true,
None,
Some(run_store.as_ref()),
run_store.as_ref(),
&run_dir,
None,
)

8
lib/crates/fabro-cli/src/commands/pr/list.rs
View file

@ -49,9 +49,11 @@ async fn list_from(
let mut entries: Vec<(String, PullRequestRecord)> = Vec::new();
for run in &runs {
if let Ok(Some(run_store)) = store.open_run_reader(&run.run_id).await {
if let Ok(Some(record)) = run_store.get_pull_request().await {
entries.push((run.run_id.to_string(), record));
if let Ok(run_store) = store.open_run_reader(&run.run_id).await {
if let Ok(state) = run_store.state().await {
if let Some(record) = state.pull_request {
entries.push((run.run_id.to_string(), record));
}
}
}
}

7
lib/crates/fabro-cli/src/commands/pr/mod.rs
View file

@ -37,10 +37,9 @@ pub(crate) async fn load_pr_record(
let store = store::build_store(storage_dir)?;
let run = resolve_run_combined(store.as_ref(), base, run_id).await?;
let run_dir = run.path;
let run_store = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let record = run_store.get_pull_request().await?.with_context(|| {
let run_store = store::open_run_reader(storage_dir, &run.run_id).await?;
let state = run_store.state().await?;
let record = state.pull_request.with_context(|| {
format!("No pull request found in store. Create one first with: fabro pr create {run_id}")
})?;
Ok((record, run_dir))

64
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -50,13 +50,13 @@ pub(crate) async fn attach_run(
if let (Some(storage_dir), Some(run_id)) = (storage_dir.as_deref(), run_id.as_ref()) {
match store::open_run_reader(storage_dir, run_id).await {
Ok(Some(run_store)) => match run_store.list_events().await {
Ok(run_store) => match run_store.list_events().await {
Ok(events) => {
let verbose = run_store
.get_run()
.state()
.await
.ok()
.flatten()
.and_then(|state| state.run)
.is_some_and(|record| record.settings.verbose_enabled());
let event_lines = events
.iter()
@ -83,7 +83,6 @@ pub(crate) async fn attach_run(
);
}
},
Ok(None) => {}
Err(err) => {
tracing::warn!(
run_id = %run_id,
@ -157,13 +156,14 @@ async fn attach_run_store(
}
// Wait briefly for a terminal status or conclusion
for _ in 0..20 {
if run_store.get_conclusion().await.ok().flatten().is_some()
|| run_store
.get_status()
.await
.ok()
.flatten()
.is_some_and(|record| record.status.is_terminal())
if run_store
.state()
.await
.ok()
.is_some_and(|state| {
state.conclusion.is_some()
|| state.status.is_some_and(|record| record.status.is_terminal())
})
{
break;
}
@ -237,11 +237,10 @@ async fn attach_run_store(
}
let terminal_status = run_store
.get_status()
.state()
.await
.ok()
.flatten()
.map(|record| record.status)
.and_then(|state| state.status.map(|record| record.status))
.filter(|status| status.is_terminal());
let child_alive_via_handle = engine_guard.as_mut().and_then(|guard| {
@ -749,24 +748,29 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option<RunStatusRe
async fn determine_exit_code_with_store(run_store: &dyn RunStore) -> ExitCode {
let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE;
loop {
if let Ok(Some(conclusion)) = run_store.get_conclusion().await {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
};
}
match run_store.state().await {
Ok(state) => {
if let Some(conclusion) = state.conclusion {
let success = matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess
);
return if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
};
}
match run_store.get_status().await {
Ok(Some(record)) if matches!(record.status, RunStatus::Succeeded) => {
return ExitCode::from(0);
match state.status {
Some(record) if matches!(record.status, RunStatus::Succeeded) => {
return ExitCode::from(0);
}
Some(record) if record.status.is_terminal() => return ExitCode::from(1),
Some(_) | None => {}
}
}
Ok(Some(record)) if record.status.is_terminal() => return ExitCode::from(1),
Ok(Some(_) | None) | Err(_) => {}
Err(_) => {}
}
if Instant::now() >= deadline {

7
lib/crates/fabro-cli/src/commands/run/cp.rs
View file

@ -130,12 +130,11 @@ async fn load_sandbox(
) -> Result<Box<dyn Sandbox>> {
let store = store::build_store(storage_dir)?;
let run = resolve_run_combined(store.as_ref(), base, run_prefix).await?;
let run_store = store::open_run_reader(storage_dir, &run.run_id)
.await?
.context("Failed to open run store")?;
let run_store = store::open_run_reader(storage_dir, &run.run_id).await?;
let record = run_store
.get_sandbox()
.state()
.await?
.sandbox
.context("Failed to load sandbox record from store")?;
info!(run_id = %run_prefix, provider = %record.provider, "Connecting to sandbox");

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

@ -29,10 +29,7 @@ pub(crate) async fn execute(
None => load_user_settings()?.storage_dir(),
};
let store = store::build_store(&storage_dir)?;
let run_store = store
.open_run(&run_id)
.await?
.ok_or_else(|| anyhow!("Run {run_id} not found in store"))?;
let run_store = store.open_run(&run_id).await?;
let run_record = run_store
.get_run()
.await?

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

@ -18,9 +18,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.context("Failed to open run store")?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let patch = resolve_diff(&run.path, run_store.as_ref(), &args).await?;
@ -57,17 +55,13 @@ async fn resolve_diff(
run_store: &dyn fabro_store::RunStore,
args: &DiffArgs,
) -> Result<String> {
let state = run_store.state().await?;
if let Some(ref node_id) = args.node {
if let Ok(visits) = run_store.list_node_visits(node_id).await {
if let Some(visit) = visits.into_iter().max() {
if let Ok(node) = run_store
.get_node(&fabro_store::NodeVisitRef { node_id, visit })
.await
{
if let Some(patch) = node.diff {
debug!(node_id, visit, "Reading per-node diff from store");
return Ok(patch);
}
if let Some(visit) = state.list_node_visits(node_id).into_iter().max() {
if let Some(node) = state.node(&fabro_store::NodeVisitRef { node_id, visit }) {
if let Some(patch) = node.diff.clone() {
debug!(node_id, visit, "Reading per-node diff from projected state");
return Ok(patch);
}
}
}
@ -75,9 +69,8 @@ async fn resolve_diff(
bail!("No diff found for node '{node_id}' — check the node ID and try again");
}
let start = run_store
.get_start()
.await?
let start = state
.start
.context("Failed to load start record from store")?;
let base_sha = start
@ -85,12 +78,12 @@ async fn resolve_diff(
.as_deref()
.ok_or_else(|| anyhow::anyhow!("This run was not git-checkpointed; no diff available"))?;
if let Ok(Some(patch)) = run_store.get_final_patch().await {
if let Some(patch) = state.final_patch {
debug!("Reading final.patch from store");
return Ok(patch);
}
let run_concluded = run_store.get_conclusion().await?.is_some();
let run_concluded = state.conclusion.is_some();
if run_concluded {
bail!(
"Run completed but no final.patch exists — the run may not have produced any changes"
@ -98,9 +91,8 @@ async fn resolve_diff(
}
debug!("No final.patch found; attempting live diff from sandbox");
let record = run_store
.get_sandbox()
.await?
let record = state
.sandbox
.context("Failed to load sandbox record from store")?;
info!(provider = %record.provider, "Reconnecting to sandbox for live diff");

2
lib/crates/fabro-cli/src/commands/run/fork.rs
View file

@ -21,7 +21,7 @@ pub(crate) async fn run(args: &ForkArgs, styles: &Styles, globals: &GlobalArgs)
let store = Store::new(repo);
let run_store = open_run_reader(&cli_settings.storage_dir(), &run_id).await?;
let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?;
let timeline = build_timeline_or_rebuild(&store, Some(run_store.as_ref()), &run_id).await?;
if args.list {
if globals.json {

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

@ -31,9 +31,7 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
None => None,
};
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.with_context(|| format!("Run '{}' not found in store", run.run_id))?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let (all_lines, last_seq) = match run_store.list_events().await {
Ok(events) => {
let last_seq = events.last().map_or(0, |event| event.seq);
@ -190,20 +188,11 @@ async fn follow_store_logs(
}
async fn run_concluded(run_store: &dyn RunStore, _run_dir: &Path) -> Result<bool> {
if run_store
.get_conclusion()
let state = run_store
.state()
.await
.context("Failed to read conclusion from store while following logs")?
.is_some()
{
return Ok(true);
}
Ok(run_store
.get_status()
.await
.context("Failed to read status from store while following logs")?
.is_some_and(|record| record.status.is_terminal()))
.context("Failed to read run state from store while following logs")?;
Ok(state.conclusion.is_some() || state.status.is_some_and(|record| record.status.is_terminal()))
}
async fn flush_remaining_store_events(

21
lib/crates/fabro-cli/src/commands/run/output.rs
View file

@ -81,18 +81,13 @@ pub(crate) async fn print_run_summary(
let (run_store, conclusion, pr_url) = match run_id.parse() {
Ok(parsed_run_id) => {
let run_store = store::open_run_reader(storage_dir, &parsed_run_id).await?;
let conclusion = match run_store.as_deref() {
Some(run_store) => run_store.get_conclusion().await?,
None => None,
};
let pr_url = match run_store.as_deref() {
Some(run_store) => run_store
.get_pull_request()
.await?
.map(|record: PullRequestRecord| record.html_url),
None => None,
};
(run_store, conclusion, pr_url)
let run_state = run_store.state().await?;
let conclusion = run_state.conclusion.clone();
let pr_url = run_state
.pull_request
.as_ref()
.map(|record: &PullRequestRecord| record.html_url.clone());
(Some(run_store), conclusion, pr_url)
}
Err(_) => (None, None, None),
};
@ -208,7 +203,7 @@ pub(crate) async fn print_final_output(
styles: &Styles,
) {
let checkpoint = match run_store {
Some(run_store) => run_store.get_checkpoint().await.ok().flatten(),
Some(run_store) => run_store.state().await.ok().and_then(|state| state.checkpoint),
None => None,
};
let Some(checkpoint) = checkpoint else {

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

@ -13,12 +13,11 @@ pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> {
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.context("Failed to open run store")?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let record = run_store
.get_sandbox()
.state()
.await?
.sandbox
.context("Failed to load sandbox record from store")?;
validate_daytona_provider(&record, "Preview URLs")?;

39
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -45,7 +45,7 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs
.await
.ok();
let timeline = build_timeline_or_rebuild(&store, run_store.as_deref(), &run_id).await?;
let timeline = build_timeline_or_rebuild(&store, Some(run_store.as_ref()), &run_id).await?;
if args.list || args.target.is_none() {
if globals.json {
@ -113,45 +113,28 @@ async fn reset_rewound_run_state(
.open_run_reader(run_id)
.await
.map_err(|err| anyhow::anyhow!("failed to open durable store run before rewind: {err}"))?;
let store_run_record = if let Some(run_store) = existing_run_store.as_ref() {
run_store.get_run().await.ok().flatten()
} else {
None
};
let store_start_record = if let Some(run_store) = existing_run_store.as_ref() {
run_store.get_start().await.ok().flatten()
} else {
None
};
let store_graph = if let Some(run_store) = existing_run_store.as_ref() {
run_store.get_graph().await.ok().flatten()
} else {
None
};
let store_run_record = existing_run_store.get_run().await.ok().flatten();
let store_start_record = existing_run_store.get_start().await.ok().flatten();
let store_graph = existing_run_store.get_graph().await.ok().flatten();
let run_record = store_run_record
.or_else(|| RunRecord::load(run_dir).ok())
.context("failed to restore run record after rewind: missing run metadata")?;
let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())?
.context("rewound metadata branch is missing checkpoint.json")?;
let previous_status = if let Some(run_store) = existing_run_store.as_ref() {
run_store
.get_status()
.await
.ok()
.flatten()
.map(|status| status.status.to_string())
} else {
None
};
let previous_status = existing_run_store
.get_status()
.await
.ok()
.flatten()
.map(|status| status.status.to_string());
let _ = std::fs::remove_file(run_dir.join("detached_failure.json"));
let run_store = durable_store
.open_run(run_id)
.await
.map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))?
.context("failed to reset durable store run after rewind: missing run store")?;
.map_err(|err| anyhow::anyhow!("failed to open durable store run for rewind reset: {err}"))?;
run_store
.reset_for_rewind()
.await

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

@ -17,12 +17,11 @@ pub(crate) async fn run(args: SshArgs, globals: &GlobalArgs) -> Result<()> {
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.context("Failed to open run store")?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let record = run_store
.get_sandbox()
.state()
.await?
.sandbox
.context("Failed to load sandbox record from store")?;
validate_daytona_provider(&record, "SSH access")?;

8
lib/crates/fabro-cli/src/commands/run/start.rs
View file

@ -1,6 +1,6 @@
use std::path::Path;
use anyhow::{Result, anyhow, bail};
use anyhow::{Result, bail};
use chrono::Utc;
use fabro_types::RunId;
use fabro_workflow::run_status::RunStatus;
@ -83,10 +83,8 @@ async fn ensure_startable_run(storage_dir: &Path, run_id: &RunId) -> Result<()>
bail!("an engine process is still running for this run — cannot start");
}
let run_store = store::open_run_reader(storage_dir, run_id)
.await?
.ok_or_else(|| anyhow!("Cannot start run: run {run_id} not found in store"))?;
if let Some(record) = run_store.get_status().await? {
let run_store = store::open_run_reader(storage_dir, run_id).await?;
if let Some(record) = run_store.state().await?.status {
if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) {
bail!(
"cannot start run: status is {:?}, expected submitted",

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

@ -34,9 +34,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
let final_status = loop {
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id)
.await?
.ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?;
let status = run_store.get_status().await?.map(|record| record.status);
.await?;
let status = run_store.state().await?.status.map(|record| record.status);
let status = status.unwrap_or_else(|| {
if started_waiting_at.elapsed() < WAIT_STARTUP_GRACE {
RunStatus::Submitted
@ -65,9 +64,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
};
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id)
.await?
.ok_or_else(|| anyhow::anyhow!("Run {} not found in store", run_info.run_id))?;
let conclusion = run_store.get_conclusion().await?;
.await?;
let conclusion = run_store.state().await?.conclusion;
if globals.json {
let json_value = build_json_output(final_status, &run_info.run_id, conclusion.as_ref());

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

@ -1,6 +1,6 @@
use std::path::{Path, PathBuf};
use anyhow::{Result, anyhow};
use anyhow::Result;
use fabro_types::RunId;
use serde::Serialize;
@ -28,9 +28,7 @@ pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()>
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.ok_or_else(|| anyhow!("Run {} not found in store", run.run_id))?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let output = inspect_run_store(&run.run_id, &run.path, run.status, run_store.as_ref()).await;
let json = serde_json::to_string_pretty(&[output])?;
println!("{json}");
@ -43,6 +41,29 @@ async fn inspect_run_store(
status: RunStatus,
run_store: &dyn fabro_store::RunStore,
) -> InspectOutput {
if let Ok(state) = run_store.state().await {
if let Some(snapshot) = state.to_snapshot() {
return InspectOutput {
run_id: run_id.to_string(),
run_dir: run_dir.to_path_buf(),
status: state.status.as_ref().map_or(status, |record| record.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()),
};
}
}
if let Ok(Some(snapshot)) = run_store.get_snapshot().await {
return InspectOutput {
run_id: run_id.to_string(),

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

@ -116,7 +116,7 @@ pub(crate) async fn remove_run_with_cleanup(store: &dyn Store, run: &RunInfo) ->
async fn remove_run_dir_with_cleanup(store: &dyn Store, run: &RunInfo) -> Result<()> {
let run_store = match store.open_run_reader(&run.run_id).await {
Ok(run_store) => run_store,
Ok(run_store) => Some(run_store),
Err(err) => {
warn!(
run_id = %run.run_id,
@ -183,9 +183,8 @@ async fn load_sandbox_record(
run_store: Option<&dyn fabro_store::RunStore>,
) -> Option<fabro_sandbox::SandboxRecord> {
if let Some(run_store) = run_store {
match run_store.get_sandbox().await {
Ok(Some(record)) => return Some(record),
Ok(None) => {}
match run_store.state().await {
Ok(state) => return state.sandbox,
Err(err) => {
warn!(error = %err, "failed to load sandbox record from store");
}

28
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -2,7 +2,7 @@ use std::io::{ErrorKind, Write};
use std::path::{Component, Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_store::{NodeVisitRef, RunSnapshot, RunStore};
use fabro_store::{NodeVisitRef, RunSnapshot, RunState, RunStore};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use serde::Serialize;
#[cfg(test)]
@ -18,14 +18,7 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) ->
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id)
.await?
.with_context(|| {
format!(
"run {} is not in the store (it may be a legacy filesystem-only run)",
run.run_id
)
})?;
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?;
let file_count = export_run(run_store.as_ref(), &args.output).await?;
if globals.json {
@ -45,9 +38,9 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) ->
}
pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> Result<usize> {
let snapshot = run_store
.get_snapshot()
.await?
let state = run_store.state().await?;
let snapshot = state
.to_snapshot()
.context("run has no data in the store")?;
let output_state = inspect_output_dir(output_dir)?;
@ -66,7 +59,7 @@ pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> R
})?;
let staging_path = staging_dir.path().to_path_buf();
let file_count = export_run_to_dir(run_store, &snapshot, &staging_path).await?;
let file_count = export_run_to_dir(run_store, &state, &snapshot, &staging_path).await?;
if matches!(output_state, OutputDirState::ExistingEmpty) {
std::fs::remove_dir(output_dir)
@ -86,6 +79,7 @@ pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> R
async fn export_run_to_dir(
run_store: &dyn RunStore,
state: &RunState,
snapshot: &RunSnapshot,
output_dir: &Path,
) -> Result<usize> {
@ -152,11 +146,11 @@ async fn export_run_to_dir(
file_count += usize::from(write_optional_text_file(
&output_dir.join("retro").join("prompt.md"),
run_store.get_retro_prompt().await?.as_deref(),
state.retro_prompt.as_deref(),
)?);
file_count += usize::from(write_optional_text_file(
&output_dir.join("retro").join("response.md"),
run_store.get_retro_response().await?.as_deref(),
state.retro_response.as_deref(),
)?);
write_events_jsonl(
@ -165,12 +159,12 @@ async fn export_run_to_dir(
)?;
file_count += 1;
for (seq, checkpoint) in run_store.list_checkpoints().await? {
for (seq, checkpoint) in &state.checkpoints {
write_json_file(
&output_dir
.join("checkpoints")
.join(format!("{seq:04}.json")),
&checkpoint,
checkpoint,
)?;
file_count += 1;
}

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

@ -21,7 +21,7 @@ pub(crate) fn build_store(storage_dir: &Path) -> Result<Arc<SlateStore>> {
pub(crate) async fn open_run_reader(
storage_dir: &Path,
run_id: &RunId,
) -> Result<Option<Arc<dyn RunStore>>> {
) -> Result<Arc<dyn RunStore>> {
build_store(storage_dir)?
.open_run_reader(run_id)
.await

8
lib/crates/fabro-cli/tests/it/cmd/diff.rs
View file

@ -117,7 +117,7 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() {
let patch = with_runtime(|runtime| {
runtime.block_on(async {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap().unwrap();
let run_store = store.open_run(&run_id).await.unwrap();
run_store.get_final_patch().await.unwrap().unwrap()
})
});
@ -126,7 +126,7 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() {
with_runtime(|runtime| {
runtime.block_on(async {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap().unwrap();
let run_store = store.open_run(&run_id).await.unwrap();
run_store.put_final_patch(&patch).await.unwrap();
});
});
@ -180,7 +180,7 @@ fn diff_node_reads_store_patch_without_disk_file() {
let patch = with_runtime(|runtime| {
runtime.block_on(async {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap().unwrap();
let run_store = store.open_run(&run_id).await.unwrap();
run_store
.get_node(&fabro_store::NodeVisitRef {
node_id: "step_one",
@ -197,7 +197,7 @@ fn diff_node_reads_store_patch_without_disk_file() {
with_runtime(|runtime| {
runtime.block_on(async {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap().unwrap();
let run_store = store.open_run(&run_id).await.unwrap();
run_store
.put_node_diff(
&fabro_store::NodeVisitRef {

2
lib/crates/fabro-cli/tests/it/cmd/pr_view.rs
View file

@ -76,7 +76,7 @@ fn pr_view_reads_pull_request_from_store_without_pull_request_json() {
with_runtime(|runtime| {
runtime.block_on(async {
let store = build_store(&context.storage_dir);
let run_store = store.open_run(&run_id).await.unwrap().unwrap();
let run_store = store.open_run(&run_id).await.unwrap();
run_store
.put_pull_request(&PullRequestRecord {
html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),

44
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -159,10 +159,12 @@ pub(crate) fn setup_detached_dry_run(context: &TestContext) -> RunSetup {
.to_string();
let run = resolve_run(context, &run_id);
let deadline = Instant::now() + COMMAND_TIMEOUT;
while run_store(&run.run_dir)
.and_then(|store| block_on(store.list_events()).ok())
.is_none_or(|events| events.is_empty())
{
while {
let store = run_store(&run.run_dir);
block_on(store.list_events())
.ok()
.is_none_or(|events| events.is_empty())
} {
assert!(
Instant::now() < deadline,
"timed out waiting for store events for {run_id}"
@ -280,9 +282,10 @@ worktree_mode = "never"
);
let run = run_local_workflow(context, &workspace_dir, "run.toml");
let store = run_store(&run.run_dir);
assert!(
run_store(&run.run_dir)
.and_then(|store| block_on(store.get_sandbox()).ok())
block_on(store.get_sandbox())
.ok()
.flatten()
.is_some()
);
@ -371,8 +374,8 @@ pub(crate) fn write_gated_workflow(path: &Path, name: &str, goal: &str) -> Workf
pub(crate) fn wait_for_status(run_dir: &Path, expected: &[&str]) -> String {
let deadline = Instant::now() + COMMAND_TIMEOUT;
loop {
if let Some(status) = run_store(run_dir)
.and_then(|store| block_on(store.get_status()).ok())
if let Some(status) = block_on(run_store(run_dir).get_status())
.ok()
.flatten()
.map(|record| record.status.to_string())
{
@ -478,25 +481,30 @@ fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
.block_on(future)
}
fn run_store(run_dir: &Path) -> Option<Arc<dyn RunStore>> {
let runs_dir = run_dir.parent()?;
let storage_dir = runs_dir.parent()?;
let run_id: RunId = infer_run_id(run_dir).parse().ok()?;
let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).ok()?);
fn run_store(run_dir: &Path) -> Arc<dyn RunStore> {
let runs_dir = run_dir.parent().expect("run dir should have parent");
let storage_dir = runs_dir.parent().expect("runs dir should have parent");
let run_id: RunId = infer_run_id(run_dir).parse().expect("run id should parse");
let object_store = Arc::new(
LocalFileSystem::new_with_prefix(storage_dir.join("store"))
.expect("test store path should be accessible"),
);
let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1)));
block_on(store.open_run_reader(&run_id)).ok().flatten()
block_on(store.open_run_reader(&run_id)).expect("run store should exist")
}
pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
run_store(run_dir)
.and_then(|store| block_on(store.get_snapshot()).ok())
let store = run_store(run_dir);
block_on(store.get_snapshot())
.ok()
.flatten()
.expect("run store snapshot should exist")
}
pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
run_store(run_dir)
.and_then(|store| block_on(store.list_events()).ok())
let store = run_store(run_dir);
block_on(store.list_events())
.ok()
.expect("run store events should exist")
}

23
lib/crates/fabro-cli/tests/it/scenario/mod.rs
View file

@ -23,9 +23,9 @@ fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
.block_on(future)
}
fn run_store(run_dir: &Path) -> Option<Arc<dyn RunStore>> {
let runs_dir = run_dir.parent()?;
let storage_dir = runs_dir.parent()?;
fn run_store(run_dir: &Path) -> Arc<dyn RunStore> {
let runs_dir = run_dir.parent().expect("run dir should have parent");
let storage_dir = runs_dir.parent().expect("runs dir should have parent");
let run_id: RunId = std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|id| id.trim().to_string())
@ -34,17 +34,22 @@ fn run_store(run_dir: &Path) -> Option<Arc<dyn RunStore>> {
.file_name()
.map(|name| name.to_string_lossy().to_string())
.and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned))
})?
})
.expect("run dir should contain resolvable run id")
.parse()
.ok()?;
let object_store = Arc::new(LocalFileSystem::new_with_prefix(storage_dir.join("store")).ok()?);
.expect("run id should parse");
let object_store = Arc::new(
LocalFileSystem::new_with_prefix(storage_dir.join("store"))
.expect("test store path should be accessible"),
);
let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1)));
block_on(store.open_run_reader(&run_id)).ok().flatten()
block_on(store.open_run_reader(&run_id)).expect("run store should exist")
}
pub(super) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
run_store(run_dir)
.and_then(|store| block_on(store.get_snapshot()).ok())
let store = run_store(run_dir);
block_on(store.get_snapshot())
.ok()
.flatten()
.expect("run store snapshot should exist")
}

36
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -137,7 +137,7 @@ pub fn build_retro_prompt(retro_data_dir: &str) -> String {
/// files via tool access, then calls `submit_retro` with its analysis.
pub async fn run_retro_agent(
sandbox: &Arc<dyn Sandbox>,
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
run_dir: &Path,
llm_client: &Client,
provider: Provider,
@ -286,32 +286,24 @@ pub fn dry_run_narrative() -> RetroNarrative {
}
async fn write_retro_prompt(
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
retro_dir: &Path,
prompt: &str,
) -> anyhow::Result<()> {
if let Some(store) = run_store {
if let Err(err) = store.put_retro_prompt(prompt).await {
tracing::warn!(error = %err, "Failed to save retro prompt to store");
std::fs::write(retro_dir.join("prompt.md"), prompt)?;
}
} else {
if let Err(err) = run_store.put_retro_prompt(prompt).await {
tracing::warn!(error = %err, "Failed to save retro prompt to store");
std::fs::write(retro_dir.join("prompt.md"), prompt)?;
}
Ok(())
}
async fn write_retro_response(
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
retro_dir: &Path,
response: &str,
) -> anyhow::Result<()> {
if let Some(store) = run_store {
if let Err(err) = store.put_retro_response(response).await {
tracing::warn!(error = %err, "Failed to save retro response to store");
std::fs::write(retro_dir.join("response.md"), response)?;
}
} else {
if let Err(err) = run_store.put_retro_response(response).await {
tracing::warn!(error = %err, "Failed to save retro response to store");
std::fs::write(retro_dir.join("response.md"), response)?;
}
Ok(())
@ -393,7 +385,7 @@ fn build_profile(provider: Provider, model: &str) -> Box<dyn AgentProfile> {
async fn upload_data_files(
sandbox: &Arc<dyn Sandbox>,
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
_run_dir: &Path,
target_dir: &str,
) -> anyhow::Result<()> {
@ -403,11 +395,7 @@ async fn upload_data_files(
.await
.map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?;
let Some(store) = run_store else {
anyhow::bail!("retro analysis now requires a run store");
};
let progress_content = match store.list_events().await {
let progress_content = match run_store.list_events().await {
Ok(envelopes) => {
let lines: Vec<String> = envelopes
.into_iter()
@ -428,7 +416,7 @@ async fn upload_data_files(
.map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?;
}
let checkpoint_content = store
let checkpoint_content = run_store
.get_checkpoint()
.await
.map_err(|e| anyhow::anyhow!("Failed to load checkpoint from store: {e}"))?
@ -436,7 +424,7 @@ async fn upload_data_files(
.transpose()?;
upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?;
let run_content = store
let run_content = run_store
.get_run()
.await
.map_err(|e| anyhow::anyhow!("Failed to load run metadata from store: {e}"))?
@ -444,7 +432,7 @@ async fn upload_data_files(
.transpose()?;
upload_file(sandbox, target_dir, "run.json", run_content).await?;
let start_content = store
let start_content = run_store
.get_start()
.await
.map_err(|e| anyhow::anyhow!("Failed to load start metadata from store: {e}"))?

79
lib/crates/fabro-server/src/server.rs
View file

@ -680,18 +680,7 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
}
let run_store = match state.store.open_run(&run_id).await {
Ok(Some(run_store)) => run_store,
Ok(None) => {
tracing::error!(run_id = %run_id, "Run store missing");
let mut runs = state.runs.lock().expect("runs lock poisoned");
if let Some(managed_run) = runs.get_mut(&run_id) {
managed_run.status = RunStatus::Failed;
managed_run.error = Some("Run store missing".to_string());
clear_live_run_state(managed_run);
}
state.scheduler_notify.notify_one();
return;
}
Ok(run_store) => run_store,
Err(e) => {
tracing::error!(run_id = %run_id, error = %e, "Failed to open run store");
let mut runs = state.runs.lock().expect("runs lock poisoned");
@ -754,10 +743,10 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
};
// Save final checkpoint
let checkpoint = match run_store.get_checkpoint().await {
Ok(checkpoint) => checkpoint,
let checkpoint = match run_store.state().await {
Ok(state) => state.checkpoint,
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Failed to load checkpoint from store");
tracing::warn!(run_id = %run_id, error = %err, "Failed to load run state from store");
None
}
};
@ -1069,13 +1058,32 @@ async fn get_checkpoint(
Ok(id) => id,
Err(response) => return response,
};
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => match &managed_run.checkpoint {
Some(cp) => (StatusCode::OK, Json(cp.clone())).into_response(),
None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
let live_checkpoint = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => managed_run.checkpoint.clone(),
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if let Some(cp) = live_checkpoint {
return (StatusCode::OK, Json(cp)).into_response();
}
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => match run_state.checkpoint {
Some(cp) => (StatusCode::OK, Json(cp)).into_response(),
None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
},
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to load checkpoint state from store");
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
},
None => ApiError::not_found("Run not found.").into_response(),
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader");
ApiError::not_found("Run not found.").into_response()
}
}
}
@ -1564,18 +1572,19 @@ async fn get_retro(
}
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) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
Ok(run_store) => match run_store.state().await {
Ok(run_state) => match run_state.retro {
Some(retro) => (StatusCode::OK, Json(retro)).into_response(),
None => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
},
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to load retro from store");
tracing::warn!(run_id = %id, error = %err, "Failed to load retro state from store");
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
},
Ok(None) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
Err(err) => {
tracing::warn!(run_id = %id, error = %err, "Failed to open run store reader");
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
ApiError::not_found("Run not found.").into_response()
}
}
}
@ -1603,15 +1612,27 @@ async fn get_graph(
Ok(id) => id,
Err(response) => return response,
};
let dot_source = {
let live_dot_source = {
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => managed_run.dot_source.clone(),
None => return ApiError::not_found("Run not found.").into_response(),
}
};
if !live_dot_source.is_empty() {
return render_dot_svg(&live_dot_source).await;
}
render_dot_svg(&dot_source).await
match state.store.open_run_reader(&id).await {
Ok(run_store) => match run_store.state().await {
Ok(run_state) => match run_state.graph_source {
Some(dot_source) => render_dot_svg(&dot_source).await,
None => ApiError::new(StatusCode::NOT_FOUND, "Graph not found.").into_response(),
},
Err(err) => ApiError::new(StatusCode::BAD_GATEWAY, err.to_string()).into_response(),
},
Err(_) => ApiError::new(StatusCode::NOT_FOUND, "Run not found.").into_response(),
}
}
#[cfg(test)]

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

@ -9,12 +9,14 @@ use futures::Stream;
mod error;
mod keys;
mod memory;
mod run_state;
mod runtime;
mod slate;
mod types;
pub use error::{Result, StoreError};
pub use memory::InMemoryStore;
pub use run_state::{NodeState, RunState};
pub use runtime::RuntimeState;
pub use slate::SlateStore;
pub use types::{
@ -42,8 +44,8 @@ pub trait Store: Send + Sync {
created_at: DateTime<Utc>,
run_dir: Option<&str>,
) -> Result<Arc<dyn RunStore>>;
async fn open_run(&self, run_id: &RunId) -> Result<Option<Arc<dyn RunStore>>>;
async fn open_run_reader(&self, run_id: &RunId) -> Result<Option<Arc<dyn RunStore>>>;
async fn open_run(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>>;
async fn open_run_reader(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>>;
async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>>;
async fn delete_run(&self, run_id: &RunId) -> Result<()>;
}
@ -146,5 +148,6 @@ pub trait RunStore: Send + Sync {
async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result<Vec<String>>;
async fn list_all_assets(&self) -> Result<Vec<(String, u32, String)>>;
async fn state(&self) -> Result<RunState>;
async fn get_snapshot(&self) -> Result<Option<RunSnapshot>>;
}

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

@ -13,9 +13,10 @@ use tokio_stream::StreamExt as _;
use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot,
NodeVisitRef, Result, RunSnapshot, RunStore, RunSummary, Store, StoreError,
NodeVisitRef, Result, RunSnapshot, RunState, RunStore, RunSummary, Store, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
@ -41,6 +42,7 @@ struct InMemoryRunStore {
event_seq: AtomicU32,
checkpoint_seq: AtomicU32,
watchers: Mutex<Vec<mpsc::UnboundedSender<EventEnvelope>>>,
projection_cache: Mutex<EventProjectionCache>,
}
impl InMemoryRunStore {
@ -65,6 +67,7 @@ impl InMemoryRunStore {
event_seq: AtomicU32::new(1),
checkpoint_seq: AtomicU32::new(1),
watchers: Mutex::new(Vec::new()),
projection_cache: Mutex::new(EventProjectionCache::default()),
})
}
@ -264,6 +267,20 @@ impl InMemoryRunStore {
}
Ok(())
}
async fn projected_state(&self) -> Result<RunState> {
let next_seq = {
let cache = self.projection_cache.lock().await;
cache.last_seq.saturating_add(1)
};
let events = self.list_events_from_inner(next_seq).await?;
let mut cache = self.projection_cache.lock().await;
for event in &events {
cache.state.apply_event(event)?;
cache.last_seq = event.seq;
}
Ok(cache.state.clone())
}
}
#[async_trait]
@ -302,14 +319,14 @@ impl Store for InMemoryStore {
Ok(run_store as Arc<dyn RunStore>)
}
async fn open_run(&self, run_id: &RunId) -> Result<Option<Arc<dyn RunStore>>> {
async fn open_run(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>> {
let runs = self.runs.lock().await;
Ok(runs
.get(run_id)
.map(|catalog| Arc::clone(&catalog.run_store) as Arc<dyn RunStore>))
runs.get(run_id)
.map(|catalog| Arc::clone(&catalog.run_store) as Arc<dyn RunStore>)
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))
}
async fn open_run_reader(&self, run_id: &RunId) -> Result<Option<Arc<dyn RunStore>>> {
async fn open_run_reader(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>> {
self.open_run(run_id).await
}
@ -324,8 +341,7 @@ impl Store for InMemoryStore {
if !matches_query(&catalog.record.created_at, query) {
continue;
}
let data = catalog.run_store.snapshot_data().await;
summaries.push(build_run_summary(&catalog.record, &data)?);
summaries.push(catalog.run_store.state().await?.build_summary(&catalog.record));
}
summaries.sort_by(|a, b| b.created_at.cmp(&a.created_at));
Ok(summaries)
@ -665,9 +681,21 @@ impl RunStore for InMemoryRunStore {
self.list_all_assets_inner().await
}
async fn get_snapshot(&self) -> Result<Option<RunSnapshot>> {
async fn state(&self) -> Result<RunState> {
let mut state = self.projected_state().await?;
let data = self.snapshot_data().await;
self.build_snapshot_from_data(&data)
state.merge_legacy(
self.build_snapshot_from_data(&data)?,
read_text(&data, keys::graph())?,
read_text(&data, keys::retro_prompt())?,
read_text(&data, keys::retro_response())?,
self.list_checkpoints_inner().await?,
);
Ok(state)
}
async fn get_snapshot(&self) -> Result<Option<RunSnapshot>> {
self.state().await.map(|state| state.to_snapshot())
}
}
@ -704,48 +732,6 @@ fn read_text(data: &BTreeMap<String, Vec<u8>>, key: &str) -> Result<Option<Strin
.transpose()
}
fn build_run_summary(
record: &CatalogRecord,
data: &BTreeMap<String, Vec<u8>>,
) -> Result<RunSummary> {
let run = read_json::<RunRecord>(data, keys::run())?;
let start = read_json::<StartRecord>(data, keys::start())?;
let status = read_json::<RunStatusRecord>(data, keys::status())?;
let conclusion = read_json::<Conclusion>(data, keys::conclusion())?;
let workflow_name = run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
} else {
run.graph.name.clone()
}
});
let goal = run.as_ref().and_then(|run| {
let goal = run.graph.goal();
(!goal.is_empty()).then(|| goal.to_string())
});
Ok(RunSummary {
run_id: record.run_id,
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,
labels: run
.as_ref()
.map(|run| run.labels.clone())
.unwrap_or_default(),
host_repo_path: run.as_ref().and_then(|run| run.host_repo_path.clone()),
start_time: start.map(|start| start.start_time),
status: status.as_ref().map(|status| status.status),
status_reason: status.and_then(|status| status.reason),
duration_ms: conclusion.as_ref().map(|conclusion| conclusion.duration_ms),
total_cost: conclusion.and_then(|conclusion| conclusion.total_cost),
})
}
#[cfg(test)]
mod tests {
use super::*;
@ -914,6 +900,26 @@ mod tests {
}
}
fn event_payload(
run_id: &str,
ts: &str,
event: &str,
node_id: Option<&str>,
properties: serde_json::Value,
) -> EventPayload {
let mut value = serde_json::json!({
"id": format!("evt-{event}-{ts}"),
"ts": ts,
"run_id": test_run_id(run_id).to_string(),
"event": event,
"properties": properties,
});
if let Some(node_id) = node_id {
value["node_id"] = serde_json::Value::String(node_id.to_string());
}
EventPayload::new(value, &test_run_id(run_id)).unwrap()
}
#[tokio::test]
async fn create_run_put_get_and_snapshot_round_trip() {
let store = InMemoryStore::default();
@ -1102,6 +1108,317 @@ mod tests {
assert_eq!(snapshot.pull_request, Some(pull_request));
}
#[tokio::test]
async fn state_projects_event_stream_and_compat_fields() {
let store = InMemoryStore::default();
let created_at = dt("2026-03-27T12:00:00Z");
let run = store
.create_run(&test_run_id("run-1"), created_at, None)
.await
.unwrap();
let run_record = sample_run_record("run-1", created_at);
let retro = sample_retro("run-1");
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": run_record.settings,
"graph": run_record.graph,
"workflow_source": "digraph night_sky {}",
"workflow_slug": run_record.workflow_slug,
"working_directory": run_record.working_directory,
"host_repo_path": run_record.host_repo_path,
"base_branch": run_record.base_branch,
"labels": run_record.labels,
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"run.started",
None,
serde_json::json!({
"run_branch": "fabro/run/demo",
"base_sha": "abc123"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.running",
None,
serde_json::json!({}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:07Z",
"stage.prompt",
Some("code"),
serde_json::json!({
"visit": 2,
"text": "Plan the fix",
"mode": "prompt",
"provider": "openai",
"model": "gpt-5.4"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:08Z",
"stage.completed",
Some("code"),
serde_json::json!({
"status": "success",
"notes": "all good",
"response": "Implemented",
"files_touched": ["src/lib.rs"],
"node_visits": {"code": 2}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:09Z",
"checkpoint.completed",
Some("code"),
serde_json::json!({
"status": "success",
"current_node": "code",
"completed_nodes": ["plan"],
"context_values": {"artifact": {"kind": "summary"}},
"next_node_id": "review",
"git_commit_sha": "def456",
"node_visits": {"code": 2},
"diff": "diff --git a/src/lib.rs b/src/lib.rs"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:10Z",
"sandbox.initialized",
None,
serde_json::json!({
"provider": "local",
"working_directory": "/tmp/night-sky",
"identifier": "sandbox-1",
"host_working_directory": "/tmp/night-sky"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:11Z",
"retro.started",
None,
serde_json::json!({
"prompt": "How did it go?"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:12Z",
"retro.completed",
None,
serde_json::json!({
"response": "Smooth enough",
"retro": retro
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:13Z",
"pull_request.created",
None,
serde_json::json!({
"pr_url": "https://github.com/fabro-sh/fabro/pull/123",
"pr_number": 123,
"owner": "fabro-sh",
"repo": "fabro",
"base_branch": "main",
"head_branch": "fabro/run/demo",
"title": "Map the constellations",
"draft": false
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:15Z",
"run.completed",
None,
serde_json::json!({
"duration_ms": 3210,
"artifact_count": 1,
"status": "success",
"total_cost": 1.25,
"final_git_commit_sha": "feedbeef",
"final_patch": "diff --git a/src/lib.rs b/src/lib.rs\n"
}),
))
.await
.unwrap();
let state = run.state().await.unwrap();
assert_eq!(state.run.as_ref().map(|run| run.run_id), Some(test_run_id("run-1")));
assert_eq!(state.graph_source.as_deref(), Some("digraph night_sky {}"));
assert_eq!(state.start.as_ref().and_then(|start| start.run_branch.as_deref()), Some("fabro/run/demo"));
assert_eq!(state.status.as_ref().map(|status| status.status), Some(RunStatus::Succeeded));
assert_eq!(state.checkpoint.as_ref().map(|checkpoint| checkpoint.current_node.as_str()), Some("code"));
assert_eq!(state.checkpoints.len(), 1);
assert_eq!(state.final_patch.as_deref(), Some("diff --git a/src/lib.rs b/src/lib.rs\n"));
assert_eq!(state.retro_prompt.as_deref(), Some("How did it go?"));
assert_eq!(state.retro_response.as_deref(), Some("Smooth enough"));
assert_eq!(state.pull_request.as_ref().map(|pr| pr.number), Some(123));
assert_eq!(state.sandbox.as_ref().map(|sandbox| sandbox.provider.as_str()), Some("local"));
assert_eq!(state.list_node_visits("code"), vec![2]);
let node = state
.node(&NodeVisitRef {
node_id: "code",
visit: 2,
})
.unwrap();
assert_eq!(node.prompt.as_deref(), Some("Plan the fix"));
assert_eq!(node.response.as_deref(), Some("Implemented"));
assert_eq!(node.diff.as_deref(), Some("diff --git a/src/lib.rs b/src/lib.rs"));
assert_eq!(
node.provider_used.as_ref().and_then(|value| value.get("provider")).and_then(|value| value.as_str()),
Some("openai")
);
}
#[tokio::test]
async fn state_rewind_keeps_active_projection_only() {
let store = InMemoryStore::default();
let run = store
.create_run(&test_run_id("run-1"), dt("2026-03-27T12:00:00Z"), None)
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:00Z",
"run.created",
None,
serde_json::json!({
"settings": Settings::default(),
"graph": Graph::new("night-sky"),
"working_directory": "/tmp/night-sky",
"labels": {}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:01Z",
"stage.prompt",
Some("code"),
serde_json::json!({
"visit": 1,
"text": "before rewind"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:02Z",
"pull_request.created",
None,
serde_json::json!({
"pr_url": "https://github.com/fabro-sh/fabro/pull/123",
"pr_number": 123,
"owner": "fabro-sh",
"repo": "fabro",
"base_branch": "main",
"head_branch": "fabro/run/demo",
"title": "Map the constellations",
"draft": false
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:03Z",
"run.completed",
None,
serde_json::json!({
"duration_ms": 10,
"artifact_count": 0,
"status": "success",
"final_patch": "old patch"
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:04Z",
"run.rewound",
None,
serde_json::json!({
"target_checkpoint_ordinal": 1,
"target_node_id": "plan",
"target_visit": 1
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:05Z",
"checkpoint.completed",
Some("plan"),
serde_json::json!({
"status": "success",
"current_node": "plan",
"completed_nodes": [],
"node_visits": {"plan": 1}
}),
))
.await
.unwrap();
run.append_event(&event_payload(
"run-1",
"2026-03-27T12:00:06Z",
"run.submitted",
None,
serde_json::json!({}),
))
.await
.unwrap();
let state = run.state().await.unwrap();
assert_eq!(state.status.as_ref().map(|status| status.status), Some(RunStatus::Submitted));
assert!(state.conclusion.is_none());
assert!(state.final_patch.is_none());
assert!(state.pull_request.is_none());
assert_eq!(state.checkpoints.len(), 1);
assert_eq!(state.checkpoint.as_ref().map(|checkpoint| checkpoint.current_node.as_str()), Some("plan"));
assert!(state.list_node_ids().is_empty());
}
#[tokio::test]
async fn list_artifact_values_and_all_assets_include_asset_only_visits() {
let store = InMemoryStore::default();
@ -1379,11 +1696,10 @@ mod tests {
store.delete_run(&test_run_id("run-1")).await.unwrap();
store.delete_run(&test_run_id("run-1")).await.unwrap();
assert!(
store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.is_none()
matches!(
store.open_run(&test_run_id("run-1")).await,
Err(StoreError::RunNotFound(_))
)
);
}

734
lib/crates/fabro-store/src/run_state.rs Normal file
View file

@ -0,0 +1,734 @@
use std::collections::{BTreeMap, HashMap};
use std::path::PathBuf;
use std::str::FromStr;
use chrono::{DateTime, Utc};
use serde_json::Value;
use crate::{
CatalogRecord, EventEnvelope, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result,
RunSnapshot, RunSummary, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, FailureSignature, NodeStatusRecord, Outcome, PullRequestRecord, Retro,
RunId, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StartRecord,
StatusReason,
};
#[derive(Debug, Clone, Default)]
pub struct RunState {
pub run: Option<RunRecord>,
pub graph_source: Option<String>,
pub start: Option<StartRecord>,
pub status: Option<RunStatusRecord>,
pub checkpoint: Option<Checkpoint>,
pub checkpoints: Vec<(u32, Checkpoint)>,
pub conclusion: Option<Conclusion>,
pub retro: Option<Retro>,
pub retro_prompt: Option<String>,
pub retro_response: Option<String>,
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub nodes: HashMap<(String, u32), NodeState>,
pub last_git_sha: Option<String>,
}
#[derive(Debug, Clone, Default)]
pub struct NodeState {
pub prompt: Option<String>,
pub response: Option<String>,
pub status: Option<NodeStatusRecord>,
pub outcome: Option<NodeOutcomeRecord>,
pub provider_used: Option<serde_json::Value>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
}
#[derive(Debug, Clone, Default)]
pub(crate) struct EventProjectionCache {
pub last_seq: u32,
pub state: RunState,
}
impl RunState {
pub fn apply_events(events: &[EventEnvelope]) -> Result<Self> {
let mut state = Self::default();
for event in events {
state.apply_event(event)?;
}
Ok(state)
}
pub fn apply_event(&mut self, event: &EventEnvelope) -> Result<()> {
let value = event.payload.as_value();
let ts = parse_ts(value)?;
let event_name = value
.get("event")
.and_then(Value::as_str)
.ok_or_else(|| StoreError::InvalidEvent("event payload missing event name".into()))?;
let run_id = parse_run_id(value)?;
let properties = value
.get("properties")
.and_then(Value::as_object)
.cloned()
.unwrap_or_default();
match event_name {
"run.created" => {
let settings = required_json::<fabro_types::Settings>(&properties, "settings")?;
let graph = required_json::<fabro_types::Graph>(&properties, "graph")?;
let working_directory =
required_string(&properties, "working_directory").map(PathBuf::from)?;
let labels = optional_json::<BTreeMap<String, String>>(&properties, "labels")?
.unwrap_or_default()
.into_iter()
.collect::<HashMap<_, _>>();
self.run = Some(RunRecord {
run_id,
created_at: ts,
settings,
graph,
workflow_slug: optional_string(&properties, "workflow_slug"),
working_directory,
host_repo_path: optional_string(&properties, "host_repo_path"),
base_branch: optional_string(&properties, "base_branch"),
labels,
});
self.graph_source = optional_string(&properties, "workflow_source");
}
"run.started" => {
self.start = Some(StartRecord {
run_id,
start_time: ts,
run_branch: optional_string(&properties, "run_branch"),
base_sha: optional_string(&properties, "base_sha"),
});
}
"run.submitted" => {
self.status = Some(run_status_record(RunStatus::Submitted, &properties, ts)?);
}
"run.starting" => {
self.status = Some(run_status_record(RunStatus::Starting, &properties, ts)?);
}
"run.running" => {
self.status = Some(run_status_record(RunStatus::Running, &properties, ts)?);
}
"run.removing" => {
self.status = Some(run_status_record(RunStatus::Removing, &properties, ts)?);
}
"run.completed" => {
self.status = Some(run_status_record(RunStatus::Succeeded, &properties, ts)?);
self.conclusion = Some(conclusion_from_completed(&properties, ts)?);
self.final_patch = optional_string(&properties, "final_patch");
self.last_git_sha = optional_string(&properties, "final_git_commit_sha")
.or_else(|| self.last_git_sha.clone());
}
"run.failed" => {
self.status = Some(run_status_record(RunStatus::Failed, &properties, ts)?);
self.conclusion = Some(conclusion_from_failed(&properties, ts));
self.last_git_sha =
optional_string(&properties, "git_commit_sha").or_else(|| self.last_git_sha.clone());
}
"run.rewound" => {
self.reset_for_rewind();
self.last_git_sha =
optional_string(&properties, "run_commit_sha").or_else(|| self.last_git_sha.clone());
}
"checkpoint.completed" => {
let checkpoint = checkpoint_from_properties(&properties, ts)?;
self.last_git_sha = checkpoint.git_commit_sha.clone().or_else(|| self.last_git_sha.clone());
if let Some(node_id) = value.get("node_id").and_then(Value::as_str) {
let visit = checkpoint
.node_visits
.get(node_id)
.and_then(|visit| u32::try_from(*visit).ok())
.unwrap_or(1);
if let Some(diff) = optional_string(&properties, "diff") {
self.node_mut(node_id, visit).diff = Some(diff);
}
}
self.checkpoint = Some(checkpoint.clone());
self.checkpoints.push((event.seq, checkpoint));
}
"sandbox.initialized" => {
self.sandbox = Some(SandboxRecord {
provider: required_string(&properties, "provider")?,
working_directory: required_string(&properties, "working_directory")?,
identifier: optional_string(&properties, "identifier"),
host_working_directory: optional_string(&properties, "host_working_directory"),
container_mount_point: optional_string(&properties, "container_mount_point"),
});
}
"retro.started" => {
self.retro_prompt = optional_string(&properties, "prompt");
}
"retro.completed" => {
self.retro_response = optional_string(&properties, "response");
self.retro = optional_json::<Retro>(&properties, "retro")?;
}
"pull_request.created" => {
self.pull_request = Some(PullRequestRecord {
html_url: required_string(&properties, "pr_url")?,
number: required_u64(&properties, "pr_number")?,
owner: required_string(&properties, "owner")?,
repo: required_string(&properties, "repo")?,
base_branch: required_string(&properties, "base_branch")?,
head_branch: required_string(&properties, "head_branch")?,
title: required_string(&properties, "title")?,
});
}
"stage.prompt" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = required_u32(&properties, "visit")?;
self.node_mut(node_id, visit).prompt = optional_string(&properties, "text");
self.node_mut(node_id, visit).provider_used =
provider_used_from_prompt(&properties);
}
"prompt.completed" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = self.current_visit_for(node_id).unwrap_or(1);
self.node_mut(node_id, visit).response = optional_string(&properties, "response");
}
"stage.completed" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = stage_visit(node_id, &properties, self).unwrap_or(1);
let response = optional_string(&properties, "response");
let outcome = stage_outcome_from_properties(&properties)?;
let status = node_status_from_outcome(&outcome, ts);
let node = self.node_mut(node_id, visit);
node.response = response;
node.status = Some(status);
node.outcome = Some(outcome);
}
"stage.failed" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = self.current_visit_for(node_id).unwrap_or(1);
let failure = optional_json::<fabro_types::FailureDetail>(&properties, "failure")?;
let failure_reason = failure.as_ref().map(|detail| detail.message.clone());
let node = self.node_mut(node_id, visit);
node.status = Some(NodeStatusRecord {
status: StageStatus::Fail,
notes: None,
failure_reason: failure_reason.clone(),
timestamp: ts,
});
node.outcome = Some(Outcome {
status: StageStatus::Fail,
preferred_label: None,
suggested_next_ids: Vec::new(),
context_updates: HashMap::new(),
jump_to_node: None,
notes: None,
failure,
usage: None,
files_touched: Vec::new(),
duration_ms: None,
});
}
"agent.session.started" | "agent.cli.started" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = required_u32(&properties, "visit")?;
self.node_mut(node_id, visit).provider_used =
provider_used_from_agent_event(event_name, &properties);
}
"command.started" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = self.current_visit_for(node_id).unwrap_or(1);
self.node_mut(node_id, visit).script_invocation =
Some(Value::Object(properties.clone()));
}
"command.completed" => {
let Some(node_id) = value.get("node_id").and_then(Value::as_str) else {
return Ok(());
};
let visit = self.current_visit_for(node_id).unwrap_or(1);
let node = self.node_mut(node_id, visit);
node.stdout = optional_string(&properties, "stdout");
node.stderr = optional_string(&properties, "stderr");
node.script_timing = Some(Value::Object(properties.clone()));
}
_ => {}
}
Ok(())
}
pub fn merge_legacy(
&mut self,
snapshot: Option<RunSnapshot>,
graph_source: Option<String>,
retro_prompt: Option<String>,
retro_response: Option<String>,
checkpoints: Vec<(u32, Checkpoint)>,
) {
let Some(snapshot) = snapshot else {
self.graph_source = self.graph_source.clone().or(graph_source);
self.retro_prompt = self.retro_prompt.clone().or(retro_prompt);
self.retro_response = self.retro_response.clone().or(retro_response);
if self.checkpoints.is_empty() {
self.checkpoints = checkpoints;
}
if self.checkpoint.is_none() {
self.checkpoint = self.checkpoints.last().map(|(_, checkpoint)| checkpoint.clone());
}
return;
};
if self.run.is_none() {
self.run = Some(snapshot.run);
}
self.start = self.start.clone().or(snapshot.start);
self.status = self.status.clone().or(snapshot.status);
self.checkpoint = self.checkpoint.clone().or(snapshot.checkpoint);
self.conclusion = self.conclusion.clone().or(snapshot.conclusion);
self.retro = self.retro.clone().or(snapshot.retro);
self.graph_source = self.graph_source.clone().or(snapshot.graph).or(graph_source);
self.sandbox = self.sandbox.clone().or(snapshot.sandbox);
self.final_patch = self.final_patch.clone().or(snapshot.final_patch);
self.pull_request = self.pull_request.clone().or(snapshot.pull_request);
self.retro_prompt = self.retro_prompt.clone().or(retro_prompt);
self.retro_response = self.retro_response.clone().or(retro_response);
if self.checkpoints.is_empty() {
self.checkpoints = checkpoints;
}
if self.checkpoint.is_none() {
self.checkpoint = self.checkpoints.last().map(|(_, checkpoint)| checkpoint.clone());
}
for node in snapshot.nodes {
let entry = self
.nodes
.entry((node.node_id.clone(), node.visit))
.or_default();
merge_node_snapshot(entry, node);
}
}
pub fn node(&self, node: &NodeVisitRef<'_>) -> Option<&NodeState> {
self.nodes.get(&(node.node_id.to_string(), node.visit))
}
pub fn list_node_ids(&self) -> Vec<String> {
let mut ids = self
.nodes
.keys()
.map(|(node_id, _)| node_id.clone())
.collect::<Vec<_>>();
ids.sort();
ids.dedup();
ids
}
pub fn list_node_visits(&self, node_id: &str) -> Vec<u32> {
let mut visits = self
.nodes
.keys()
.filter(|(current_node_id, _)| current_node_id == node_id)
.map(|(_, visit)| *visit)
.collect::<Vec<_>>();
visits.sort_unstable();
visits.dedup();
visits
}
pub fn to_snapshot(&self) -> Option<RunSnapshot> {
let run = self.run.clone()?;
let mut node_keys = self.nodes.keys().cloned().collect::<Vec<_>>();
node_keys.sort();
let nodes = node_keys
.into_iter()
.filter_map(|(node_id, visit)| {
self.nodes.get(&(node_id.clone(), visit)).map(|node| NodeSnapshot {
node_id,
visit,
prompt: node.prompt.clone(),
response: node.response.clone(),
status: node.status.clone(),
outcome: node.outcome.clone(),
provider_used: node.provider_used.clone(),
diff: node.diff.clone(),
script_invocation: node.script_invocation.clone(),
script_timing: node.script_timing.clone(),
parallel_results: node.parallel_results.clone(),
stdout: node.stdout.clone(),
stderr: node.stderr.clone(),
})
})
.collect();
Some(RunSnapshot {
run,
start: self.start.clone(),
status: self.status.clone(),
checkpoint: self.checkpoint.clone(),
conclusion: self.conclusion.clone(),
retro: self.retro.clone(),
graph: self.graph_source.clone(),
sandbox: self.sandbox.clone(),
final_patch: self.final_patch.clone(),
pull_request: self.pull_request.clone(),
nodes,
})
}
pub fn build_summary(&self, catalog: &CatalogRecord) -> RunSummary {
let workflow_name = self.run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
} else {
run.graph.name.clone()
}
});
let goal = self.run.as_ref().and_then(|run| {
let goal = run.graph.goal();
(!goal.is_empty()).then(|| goal.to_string())
});
RunSummary {
run_id: catalog.run_id,
created_at: catalog.created_at,
db_prefix: catalog.db_prefix.clone(),
run_dir: catalog.run_dir.clone(),
workflow_name,
workflow_slug: self.run.as_ref().and_then(|run| run.workflow_slug.clone()),
goal,
labels: self
.run
.as_ref()
.map(|run| run.labels.clone())
.unwrap_or_default(),
host_repo_path: self.run.as_ref().and_then(|run| run.host_repo_path.clone()),
start_time: self.start.as_ref().map(|start| start.start_time),
status: self.status.as_ref().map(|status| status.status),
status_reason: self.status.as_ref().and_then(|status| status.reason),
duration_ms: self.conclusion.as_ref().map(|conclusion| conclusion.duration_ms),
total_cost: self
.conclusion
.as_ref()
.and_then(|conclusion| conclusion.total_cost),
}
}
fn node_mut(&mut self, node_id: &str, visit: u32) -> &mut NodeState {
self.nodes
.entry((node_id.to_string(), visit))
.or_default()
}
fn current_visit_for(&self, node_id: &str) -> Option<u32> {
self.nodes
.keys()
.filter(|(current_node_id, _)| current_node_id == node_id)
.map(|(_, visit)| *visit)
.max()
}
fn reset_for_rewind(&mut self) {
self.status = None;
self.checkpoint = None;
self.checkpoints.clear();
self.conclusion = None;
self.retro = None;
self.retro_prompt = None;
self.retro_response = None;
self.sandbox = None;
self.final_patch = None;
self.pull_request = None;
self.nodes.clear();
}
}
fn merge_node_snapshot(node: &mut NodeState, snapshot: NodeSnapshot) {
node.prompt = node.prompt.clone().or(snapshot.prompt);
node.response = node.response.clone().or(snapshot.response);
node.status = node.status.clone().or(snapshot.status);
node.outcome = node.outcome.clone().or(snapshot.outcome);
node.provider_used = node.provider_used.clone().or(snapshot.provider_used);
node.diff = node.diff.clone().or(snapshot.diff);
node.script_invocation = node.script_invocation.clone().or(snapshot.script_invocation);
node.script_timing = node.script_timing.clone().or(snapshot.script_timing);
node.parallel_results = node.parallel_results.clone().or(snapshot.parallel_results);
node.stdout = node.stdout.clone().or(snapshot.stdout);
node.stderr = node.stderr.clone().or(snapshot.stderr);
}
fn parse_ts(value: &Value) -> Result<DateTime<Utc>> {
let ts = value
.get("ts")
.and_then(Value::as_str)
.ok_or_else(|| StoreError::InvalidEvent("event payload missing ts".into()))?;
chrono::DateTime::parse_from_rfc3339(ts)
.map(|ts| ts.with_timezone(&Utc))
.map_err(|err| StoreError::InvalidEvent(format!("invalid event ts: {err}")).into())
}
fn parse_run_id(value: &Value) -> Result<RunId> {
let run_id = value
.get("run_id")
.and_then(Value::as_str)
.ok_or_else(|| StoreError::InvalidEvent("event payload missing run_id".into()))?;
run_id
.parse()
.map_err(|err| StoreError::InvalidEvent(format!("invalid run_id: {err}")).into())
}
fn required_string(properties: &serde_json::Map<String, Value>, key: &str) -> Result<String> {
properties
.get(key)
.and_then(Value::as_str)
.map(ToString::to_string)
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing string property {key}")).into())
}
fn optional_string(properties: &serde_json::Map<String, Value>, key: &str) -> Option<String> {
properties
.get(key)
.and_then(Value::as_str)
.map(ToString::to_string)
}
fn required_u64(properties: &serde_json::Map<String, Value>, key: &str) -> Result<u64> {
properties
.get(key)
.and_then(Value::as_u64)
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing integer property {key}")).into())
}
fn required_u32(properties: &serde_json::Map<String, Value>, key: &str) -> Result<u32> {
u32::try_from(required_u64(properties, key)?)
.map_err(|_| StoreError::InvalidEvent(format!("property {key} does not fit in u32")).into())
}
fn required_json<T: serde::de::DeserializeOwned>(
properties: &serde_json::Map<String, Value>,
key: &str,
) -> Result<T> {
let value = properties
.get(key)
.cloned()
.ok_or_else(|| StoreError::InvalidEvent(format!("event missing property {key}")))?;
serde_json::from_value(value)
.map_err(|err| StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into())
}
fn optional_json<T: serde::de::DeserializeOwned>(
properties: &serde_json::Map<String, Value>,
key: &str,
) -> Result<Option<T>> {
properties
.get(key)
.filter(|value| !value.is_null())
.cloned()
.map(|value| {
serde_json::from_value(value).map_err(|err| {
StoreError::InvalidEvent(format!("invalid property {key}: {err}")).into()
})
})
.transpose()
}
fn parse_reason(properties: &serde_json::Map<String, Value>) -> Result<Option<StatusReason>> {
optional_string(properties, "reason")
.map(|reason| {
serde_json::from_value(Value::String(reason)).map_err(|err| {
StoreError::InvalidEvent(format!("invalid status reason: {err}")).into()
})
})
.transpose()
}
fn run_status_record(
status: RunStatus,
properties: &serde_json::Map<String, Value>,
updated_at: DateTime<Utc>,
) -> Result<RunStatusRecord> {
Ok(RunStatusRecord {
status,
reason: parse_reason(properties)?,
updated_at,
})
}
fn checkpoint_from_properties(
properties: &serde_json::Map<String, Value>,
timestamp: DateTime<Utc>,
) -> Result<Checkpoint> {
let loop_failure_signatures =
optional_json::<HashMap<String, usize>>(properties, "loop_failure_signatures")?
.unwrap_or_default()
.into_iter()
.map(|(key, value)| (FailureSignature(key), value))
.collect();
let restart_failure_signatures =
optional_json::<HashMap<String, usize>>(properties, "restart_failure_signatures")?
.unwrap_or_default()
.into_iter()
.map(|(key, value)| (FailureSignature(key), value))
.collect();
Ok(Checkpoint {
timestamp,
current_node: required_string(properties, "current_node")?,
completed_nodes: optional_json(properties, "completed_nodes")?.unwrap_or_default(),
node_retries: optional_json(properties, "node_retries")?.unwrap_or_default(),
context_values: optional_json(properties, "context_values")?.unwrap_or_default(),
node_outcomes: optional_json(properties, "node_outcomes")?.unwrap_or_default(),
next_node_id: optional_string(properties, "next_node_id"),
git_commit_sha: optional_string(properties, "git_commit_sha"),
loop_failure_signatures,
restart_failure_signatures,
node_visits: optional_json(properties, "node_visits")?.unwrap_or_default(),
})
}
fn conclusion_from_completed(
properties: &serde_json::Map<String, Value>,
timestamp: DateTime<Utc>,
) -> Result<Conclusion> {
let usage = optional_json::<fabro_types::StageUsage>(properties, "usage")?;
Ok(Conclusion {
timestamp,
status: StageStatus::from_str(&required_string(properties, "status")?).map_err(
|err| StoreError::InvalidEvent(format!("invalid completed stage status: {err}")),
)?,
duration_ms: required_u64(properties, "duration_ms")?,
failure_reason: None,
final_git_commit_sha: optional_string(properties, "final_git_commit_sha"),
stages: Vec::new(),
total_cost: properties.get("total_cost").and_then(Value::as_f64),
total_retries: 0,
total_input_tokens: usage.as_ref().map_or(0, |usage| usage.input_tokens),
total_output_tokens: usage.as_ref().map_or(0, |usage| usage.output_tokens),
total_cache_read_tokens: usage
.as_ref()
.and_then(|usage| usage.cache_read_tokens)
.unwrap_or(0),
total_cache_write_tokens: usage
.as_ref()
.and_then(|usage| usage.cache_write_tokens)
.unwrap_or(0),
total_reasoning_tokens: usage
.as_ref()
.and_then(|usage| usage.reasoning_tokens)
.unwrap_or(0),
has_pricing: usage.as_ref().is_some_and(|usage| usage.cost.is_some()),
})
}
fn conclusion_from_failed(
properties: &serde_json::Map<String, Value>,
timestamp: DateTime<Utc>,
) -> Conclusion {
Conclusion {
timestamp,
status: StageStatus::Fail,
duration_ms: properties
.get("duration_ms")
.and_then(Value::as_u64)
.unwrap_or_default(),
failure_reason: optional_string(properties, "error"),
final_git_commit_sha: optional_string(properties, "git_commit_sha"),
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,
}
}
fn stage_visit(node_id: &str, properties: &serde_json::Map<String, Value>, state: &RunState) -> Option<u32> {
properties
.get("node_visits")
.and_then(|value| serde_json::from_value::<HashMap<String, usize>>(value.clone()).ok())
.and_then(|visits| visits.get(node_id).copied())
.and_then(|visit| u32::try_from(visit).ok())
.or_else(|| state.current_visit_for(node_id))
}
fn stage_outcome_from_properties(
properties: &serde_json::Map<String, Value>,
) -> Result<NodeOutcomeRecord> {
let status = StageStatus::from_str(&required_string(properties, "status")?).map_err(|err| {
StoreError::InvalidEvent(format!("invalid stage status: {err}"))
})?;
Ok(Outcome {
status,
preferred_label: optional_string(properties, "preferred_label"),
suggested_next_ids: optional_json(properties, "suggested_next_ids")?.unwrap_or_default(),
context_updates: optional_json(properties, "context_updates")?.unwrap_or_default(),
jump_to_node: optional_string(properties, "jump_to_node"),
notes: optional_string(properties, "notes"),
failure: optional_json(properties, "failure")?,
usage: optional_json(properties, "usage")?,
files_touched: optional_json(properties, "files_touched")?.unwrap_or_default(),
duration_ms: properties.get("duration_ms").and_then(Value::as_u64),
})
}
fn node_status_from_outcome(outcome: &NodeOutcomeRecord, timestamp: DateTime<Utc>) -> NodeStatusRecord {
NodeStatusRecord {
status: outcome.status.clone(),
notes: outcome.notes.clone(),
failure_reason: outcome.failure.as_ref().map(|failure| failure.message.clone()),
timestamp,
}
}
fn provider_used_from_prompt(properties: &serde_json::Map<String, Value>) -> Option<Value> {
let mut provider_used = serde_json::Map::new();
if let Some(mode) = optional_string(properties, "mode") {
provider_used.insert("mode".to_string(), Value::String(mode));
}
if let Some(provider) = optional_string(properties, "provider") {
provider_used.insert("provider".to_string(), Value::String(provider));
}
if let Some(model) = optional_string(properties, "model") {
provider_used.insert("model".to_string(), Value::String(model));
}
(!provider_used.is_empty()).then_some(Value::Object(provider_used))
}
fn provider_used_from_agent_event(
event_name: &str,
properties: &serde_json::Map<String, Value>,
) -> Option<Value> {
let mut provider_used = serde_json::Map::new();
provider_used.insert(
"mode".to_string(),
Value::String(if event_name == "agent.cli.started" {
"cli".to_string()
} else {
"agent".to_string()
}),
);
if let Some(provider) = optional_string(properties, "provider") {
provider_used.insert("provider".to_string(), Value::String(provider));
}
if let Some(model) = optional_string(properties, "model") {
provider_used.insert("model".to_string(), Value::String(model));
}
if let Some(command) = optional_string(properties, "command") {
provider_used.insert("command".to_string(), Value::String(command));
}
Some(Value::Object(provider_used))
}

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

@ -236,30 +236,28 @@ impl Store for SlateStore {
Ok(Arc::new(run_store) as Arc<dyn RunStore>)
}
async fn open_run(&self, run_id: &RunId) -> 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);
};
async fn open_run(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>> {
let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
let Some(run_store) = self.open_run_store(&locator).await? else {
return Ok(None);
};
Ok(Some(Arc::new(run_store) as Arc<dyn RunStore>))
let run_store = self
.open_run_store(&locator)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(Arc::new(run_store) as Arc<dyn RunStore>)
}
async fn open_run_reader(&self, run_id: &RunId) -> 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);
};
async fn open_run_reader(&self, run_id: &RunId) -> Result<Arc<dyn RunStore>> {
let locator = catalog::read_locator(self.object_store.clone(), &self.base_prefix, run_id)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
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>))
let run_store = self
.open_run_reader_store(&locator)
.await?
.ok_or_else(|| StoreError::RunNotFound(run_id.to_string()))?;
Ok(Arc::new(run_store) as Arc<dyn RunStore>)
}
async fn list_runs(&self, query: &ListRunsQuery) -> Result<Vec<RunSummary>> {
@ -584,7 +582,6 @@ mod tests {
let reopened = store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.unwrap();
let stored = reopened.get_run().await.unwrap().unwrap();
assert_eq!(stored.run_id, test_run_id("run-1"));
@ -594,8 +591,7 @@ mod tests {
store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.is_none()
.is_err()
);
assert!(!object_exists(object_store.clone(), &by_id).await);
assert!(!object_exists(object_store.clone(), &by_start).await);
@ -634,8 +630,7 @@ mod tests {
store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.is_some()
.is_ok()
);
assert!(
store
@ -680,7 +675,6 @@ mod tests {
let reopened = store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.unwrap();
let next_event = reopened
.append_event(&event_payload(
@ -730,8 +724,7 @@ mod tests {
store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.is_none()
.is_err()
);
assert!(
store
@ -783,7 +776,6 @@ mod tests {
let reopened = store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.unwrap();
let first_event = run
.append_event(&event_payload("run-1", "2026-03-27T12:00:00Z", "Started"))
@ -884,8 +876,7 @@ mod tests {
store
.open_run(&test_run_id("run-1"))
.await
.unwrap()
.is_none()
.is_err()
);
assert!(list_paths(object_store, "runs").await.is_empty());
}

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

@ -15,9 +15,10 @@ use tokio::time;
use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::run_state::EventProjectionCache;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef,
Result, RunSnapshot, RunStore, RunSummary, StoreError,
Result, RunSnapshot, RunState, RunStore, RunSummary, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
@ -38,6 +39,7 @@ pub(crate) struct SlateRunStoreInner {
event_seq: AtomicU32,
checkpoint_seq: AtomicU32,
close_lock: Mutex<()>,
projection_cache: Mutex<EventProjectionCache>,
}
enum SlateRunDb {
@ -60,6 +62,7 @@ impl SlateRunStore {
event_seq: AtomicU32::new(event_seq),
checkpoint_seq: AtomicU32::new(checkpoint_seq),
close_lock: Mutex::new(()),
projection_cache: Mutex::new(EventProjectionCache::default()),
}),
})
}
@ -78,6 +81,7 @@ impl SlateRunStore {
event_seq: AtomicU32::new(event_seq),
checkpoint_seq: AtomicU32::new(checkpoint_seq),
close_lock: Mutex::new(()),
projection_cache: Mutex::new(EventProjectionCache::default()),
}),
})
}
@ -143,42 +147,74 @@ impl SlateRunStore {
where
R: DbRead + Sync,
{
let run = get_json::<_, RunRecord>(db, keys::run()).await?;
let start = get_json::<_, StartRecord>(db, keys::start()).await?;
let status = get_json::<_, RunStatusRecord>(db, keys::status()).await?;
let conclusion = get_json::<_, Conclusion>(db, keys::conclusion()).await?;
let workflow_name = run.as_ref().map(|run| {
if run.graph.name.is_empty() {
"unnamed".to_string()
} else {
run.graph.name.clone()
let events = list_events_from(db, 1).await?;
let mut state = RunState::apply_events(&events)?;
let mut nodes = BTreeSet::new();
let mut iter = db.scan_prefix(b"nodes/").await?;
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((node_id, visit, _)) = keys::parse_node_key(&key) {
nodes.insert((node_id, visit));
}
});
let goal = run.as_ref().and_then(|run| {
let goal = run.graph.goal();
(!goal.is_empty()).then(|| goal.to_string())
});
Ok(RunSummary {
run_id: catalog.run_id,
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,
labels: run
.as_ref()
.map(|run| run.labels.clone())
.unwrap_or_default(),
host_repo_path: run.as_ref().and_then(|run| run.host_repo_path.clone()),
start_time: start.map(|start| start.start_time),
status: status.as_ref().map(|status| status.status),
status_reason: status.and_then(|status| status.reason),
duration_ms: conclusion.as_ref().map(|conclusion| conclusion.duration_ms),
total_cost: conclusion.and_then(|conclusion| conclusion.total_cost),
})
}
let snapshot = if let Some(run) = get_json::<_, RunRecord>(db, keys::run()).await? {
let mut snapshot_nodes = Vec::new();
for (node_id, visit) in nodes {
let node = NodeVisitRef {
node_id: &node_id,
visit,
};
let prompt_key = keys::node_prompt(&node);
let response_key = keys::node_response(&node);
let status_key = keys::node_status(&node);
let outcome_key = keys::node_outcome(&node);
let provider_key = keys::node_provider_used(&node);
let diff_key = keys::node_diff(&node);
let invocation_key = keys::node_script_invocation(&node);
let timing_key = keys::node_script_timing(&node);
let results_key = keys::node_parallel_results(&node);
let stdout_key = keys::node_stdout(&node);
let stderr_key = keys::node_stderr(&node);
snapshot_nodes.push(NodeSnapshot {
node_id: node_id.clone(),
visit,
prompt: get_text(db, &prompt_key).await?,
response: get_text(db, &response_key).await?,
status: get_json(db, &status_key).await?,
outcome: get_json(db, &outcome_key).await?,
provider_used: get_json(db, &provider_key).await?,
diff: get_text(db, &diff_key).await?,
script_invocation: get_json(db, &invocation_key).await?,
script_timing: get_json(db, &timing_key).await?,
parallel_results: get_json(db, &results_key).await?,
stdout: get_text(db, &stdout_key).await?,
stderr: get_text(db, &stderr_key).await?,
});
}
Some(RunSnapshot {
run,
start: get_json::<_, StartRecord>(db, keys::start()).await?,
status: get_json::<_, RunStatusRecord>(db, keys::status()).await?,
checkpoint: get_json::<_, Checkpoint>(db, keys::checkpoint()).await?,
conclusion: get_json::<_, Conclusion>(db, keys::conclusion()).await?,
retro: get_json::<_, Retro>(db, keys::retro()).await?,
graph: get_text(db, keys::graph()).await?,
sandbox: get_json::<_, SandboxRecord>(db, keys::sandbox()).await?,
final_patch: get_text(db, keys::final_patch()).await?,
pull_request: get_json::<_, PullRequestRecord>(db, keys::pull_request()).await?,
nodes: snapshot_nodes,
})
} else {
None
};
state.merge_legacy(
snapshot,
get_text(db, keys::graph()).await?,
get_text(db, keys::retro_prompt()).await?,
get_text(db, keys::retro_response()).await?,
list_checkpoints(db).await?,
);
Ok(state.build_summary(catalog))
}
fn validate_run_record(&self, record: &RunRecord) -> Result<()> {
@ -230,6 +266,20 @@ impl SlateRunStore {
stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?,
})
}
async fn projected_state(&self) -> Result<RunState> {
let next_seq = {
let cache = self.inner.projection_cache.lock().await;
cache.last_seq.saturating_add(1)
};
let events = self.inner.db.list_events_from(next_seq).await?;
let mut cache = self.inner.projection_cache.lock().await;
for event in &events {
cache.state.apply_event(event)?;
cache.last_seq = event.seq;
}
Ok(cache.state.clone())
}
}
#[async_trait]
@ -619,7 +669,25 @@ impl RunStore for SlateRunStore {
self.inner.db.list_all_assets().await
}
async fn state(&self) -> Result<RunState> {
let mut state = self.projected_state().await?;
state.merge_legacy(
self.get_snapshot_legacy().await?,
self.get_graph().await?,
self.get_retro_prompt().await?,
self.get_retro_response().await?,
self.list_checkpoints().await?,
);
Ok(state)
}
async fn get_snapshot(&self) -> Result<Option<RunSnapshot>> {
self.state().await.map(|state| state.to_snapshot())
}
}
impl SlateRunStore {
async fn get_snapshot_legacy(&self) -> Result<Option<RunSnapshot>> {
let Some(run) = self.get_run().await? else {
return Ok(None);
};

4
lib/crates/fabro-workflow/src/event.rs
View file

@ -1510,7 +1510,9 @@ pub(crate) fn normalize_json_value(value: Value) -> Value {
}
pub fn event_payload_from_redacted_json(line: &str, run_id: &RunId) -> Result<EventPayload> {
let value = serde_json::from_str(line).context("Failed to parse redacted event payload")?;
let value = normalize_json_value(
serde_json::from_str(line).context("Failed to parse redacted event payload")?,
);
EventPayload::new(value, run_id).map_err(anyhow::Error::from)
}

27
lib/crates/fabro-workflow/src/handler/agent.rs
View file

@ -4,7 +4,6 @@ use std::sync::Arc;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_store::NodeVisitRef;
use fabro_types::RunId;
use crate::context::keys;
@ -257,18 +256,7 @@ impl Handler for AgentHandler {
let visit = visit_from_context(context);
let stage_dir = node_dir(run_dir, &node.id, visit);
fs::create_dir_all(&stage_dir).await?;
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
if let Some(ref store) = services.run_store {
store
.put_node_prompt(&node_ref, &prompt)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("prompt.md"), &prompt).await?;
}
fs::write(stage_dir.join("prompt.md"), &prompt).await?;
// 3. Call LLM backend (agent loop)
let thread_id = context.thread_id();
@ -326,14 +314,7 @@ impl Handler for AgentHandler {
};
// 4. Write response to logs
if let Some(ref store) = services.run_store {
store
.put_node_response(&node_ref, &response_text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("response.md"), &response_text).await?;
}
fs::write(stage_dir.join("response.md"), &response_text).await?;
// 7. Build and write status
let mut outcome = Outcome::success();
outcome.notes = Some(format!("Stage completed: {}", node.id));
@ -392,7 +373,7 @@ mod tests {
use super::*;
use crate::event::EventEmitter;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{InMemoryStore, RunStore, Store};
use fabro_store::{InMemoryStore, NodeVisitRef, RunStore, Store};
use fabro_types::fixtures;
use std::sync::Arc;
use tempfile::TempDir;
@ -412,7 +393,7 @@ mod tests {
.await
.unwrap();
let services = EngineServices {
run_store: Some(Arc::clone(&run_store)),
run_store: Arc::clone(&run_store),
..EngineServices::test_default()
};
let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store));

78
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -1,8 +1,6 @@
use std::path::Path;
use async_trait::async_trait;
use fabro_store::NodeVisitRef;
use crate::context::Context;
use crate::context::keys;
use crate::error::FabroError;
@ -92,22 +90,11 @@ impl Handler for CommandHandler {
let visit = visit_from_context(context);
let stage_dir = node_dir(run_dir, &node.id, visit);
fs::create_dir_all(&stage_dir).await?;
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
let invocation = serde_json::json!({
"command": script,
"language": language,
"timeout_ms": timeout_ms(node),
});
if let Some(ref store) = services.run_store {
store
.put_node_script_invocation(&node_ref, &invocation)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
}
fs::write(
stage_dir.join("script_invocation.json"),
serde_json::to_string_pretty(&invocation).unwrap(),
@ -142,31 +129,14 @@ impl Handler for CommandHandler {
.await
.map_err(|e| FabroError::handler(format!("Failed to spawn script: {e}")))?;
if let Some(ref store) = services.run_store {
store
.put_node_stdout(&node_ref, &result.stdout)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
store
.put_node_stderr(&node_ref, &result.stderr)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("stdout.log"), &result.stdout).await?;
fs::write(stage_dir.join("stderr.log"), &result.stderr).await?;
}
fs::write(stage_dir.join("stdout.log"), &result.stdout).await?;
fs::write(stage_dir.join("stderr.log"), &result.stderr).await?;
let timing = serde_json::json!({
"duration_ms": result.duration_ms,
"exit_code": if result.timed_out { serde_json::Value::Null } else { serde_json::json!(result.exit_code) },
"timed_out": result.timed_out,
});
if let Some(ref store) = services.run_store {
store
.put_node_script_timing(&node_ref, &timing)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
}
fs::write(
stage_dir.join("script_timing.json"),
serde_json::to_string_pretty(&timing).unwrap(),
@ -230,6 +200,7 @@ mod tests {
use crate::outcome::StageStatus;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{InMemoryStore, NodeVisitRef, RunStore, Store};
use fabro_types::fixtures;
use std::sync::Arc;
use std::time::Duration;
@ -237,6 +208,26 @@ mod tests {
EngineServices::test_default()
}
async fn make_services_with_run_store() -> (
EngineServices,
Arc<dyn RunStore>,
crate::event::StoreProgressLogger,
) {
let store = InMemoryStore::default();
let run_store = store
.create_run(&fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let services = EngineServices {
emitter: Arc::new(crate::event::EventEmitter::new(fixtures::RUN_1)),
run_store: Arc::clone(&run_store),
..EngineServices::test_default()
};
let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store));
logger.register(services.emitter.as_ref());
(services, run_store, logger)
}
#[tokio::test]
async fn script_handler_no_script() {
let handler = CommandHandler;
@ -586,31 +577,28 @@ mod tests {
let context = Context::new();
let graph = Graph::new("test");
let run_dir = tempfile::tempdir().unwrap();
let store = Arc::new(InMemoryStore::default());
let run_store = store
.create_run(&fabro_types::fixtures::RUN_1, chrono::Utc::now(), None)
.await
.unwrap();
let services = EngineServices {
run_store: Some(Arc::clone(&run_store) as Arc<dyn RunStore>),
..EngineServices::test_default()
};
let (services, run_store, logger) = make_services_with_run_store().await;
handler
.execute(&node, &context, &graph, run_dir.path(), &services)
.await
.unwrap();
logger.flush().await;
let snapshot = run_store
.get_node(&NodeVisitRef {
.state()
.await
.unwrap();
let node = snapshot
.node(&NodeVisitRef {
node_id: "script_node",
visit: 1,
})
.await
.cloned()
.unwrap();
assert_eq!(snapshot.script_invocation.unwrap()["command"], "echo hello");
assert_eq!(snapshot.script_timing.unwrap()["exit_code"], 0);
assert_eq!(node.script_invocation.unwrap()["script"], "echo hello");
assert_eq!(node.script_timing.unwrap()["exit_code"], 0);
}
#[tokio::test]

41
lib/crates/fabro-workflow/src/handler/fan_in.rs
View file

@ -226,7 +226,7 @@ async fn llm_evaluate(
node_id: &str,
emitter: &Arc<EventEmitter>,
sandbox: &Arc<dyn Sandbox>,
run_store: Option<Arc<dyn RunStore>>,
run_store: Arc<dyn RunStore>,
) -> Result<Candidate, FabroError> {
let results_text =
serde_json::to_string_pretty(results).unwrap_or_else(|_| results.to_string());
@ -244,14 +244,11 @@ async fn llm_evaluate(
node_id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
if let Some(ref store) = run_store {
store
.put_node_prompt(&node_ref, &full_prompt)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
}
run_store
.put_node_prompt(&node_ref, &full_prompt)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
fs::write(stage_dir.join("prompt.md"), &full_prompt).await?;
// Build a synthetic node for the backend call
let eval_node = Node::new("fan_in_eval");
@ -281,14 +278,11 @@ async fn llm_evaluate(
.unwrap_or_else(|| "unknown".to_string());
let response_text =
serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
if let Some(ref store) = run_store {
store
.put_node_response(&node_ref, &response_text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("response.md"), &response_text).await?;
}
run_store
.put_node_response(&node_ref, &response_text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
fs::write(stage_dir.join("response.md"), &response_text).await?;
Ok(Candidate {
id: best_id,
status: outcome.status.to_string(),
@ -297,14 +291,11 @@ async fn llm_evaluate(
}
Ok(CodergenResult::Text { text, .. }) => {
// Write response to logs
if let Some(ref store) = run_store {
store
.put_node_response(&node_ref, &text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("response.md"), &text).await?;
}
run_store
.put_node_response(&node_ref, &text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
fs::write(stage_dir.join("response.md"), &text).await?;
// The LLM responded with text; try to find a matching candidate ID
let text = text.trim().to_string();

11
lib/crates/fabro-workflow/src/handler/mod.rs
View file

@ -19,6 +19,8 @@ use std::sync::Arc;
use async_trait::async_trait;
use fabro_agent::Sandbox;
use fabro_store::RunStore;
#[cfg(test)]
use fabro_store::Store;
use crate::context::Context;
use crate::error::FabroError;
@ -34,7 +36,7 @@ pub struct EngineServices {
pub registry: Arc<HandlerRegistry>,
pub emitter: Arc<EventEmitter>,
pub sandbox: Arc<dyn Sandbox>,
pub run_store: Option<Arc<dyn RunStore>>,
pub run_store: Arc<dyn RunStore>,
/// Git state for the current run. Set via `set_git_state` at the start of
/// `run_via_core` and read by parallel/fan-in handlers.
pub(crate) git_state: std::sync::RwLock<Option<Arc<GitState>>>,
@ -75,7 +77,12 @@ impl EngineServices {
sandbox: Arc::new(fabro_agent::LocalSandbox::new(
std::env::current_dir().unwrap_or_else(|_| std::path::PathBuf::from(".")),
)),
run_store: None,
run_store: futures::executor::block_on(async {
fabro_store::InMemoryStore::default()
.create_run(&fabro_types::RunId::new(), chrono::Utc::now(), None)
.await
.expect("in-memory test run store should initialize")
}),
git_state: std::sync::RwLock::new(None),
hook_runner: None,
env: HashMap::new(),

21
lib/crates/fabro-workflow/src/handler/parallel.rs
View file

@ -481,16 +481,15 @@ impl Handler for ParallelHandler {
let visit = visit_from_context(context);
let node_dir = node_dir(run_dir, &node.id, visit);
let _ = fs::create_dir_all(&node_dir).await;
if let Some(ref store) = services.run_store {
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
store
.put_node_parallel_results(&node_ref, &serde_json::json!(results_json))
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
}
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
services
.run_store
.put_node_parallel_results(&node_ref, &serde_json::json!(results_json))
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
if let Ok(json) = serde_json::to_string_pretty(&results_json) {
let _ = fs::write(node_dir.join("parallel_results.json"), json).await;
}
@ -687,7 +686,7 @@ mod tests {
.await
.unwrap();
let services = EngineServices {
run_store: Some(Arc::clone(&run_store) as Arc<dyn RunStore>),
run_store: Arc::clone(&run_store) as Arc<dyn RunStore>,
..EngineServices::test_default()
};
let mut node = Node::new("par");

26
lib/crates/fabro-workflow/src/handler/prompt.rs
View file

@ -3,8 +3,6 @@ use std::path::Path;
use async_trait::async_trait;
use fabro_model::Provider;
use fabro_store::NodeVisitRef;
use crate::context::keys;
use crate::context::{Context, WorkflowContext};
use crate::error::FabroError;
@ -93,18 +91,7 @@ impl Handler for PromptHandler {
let visit = visit_from_context(context);
let stage_dir = node_dir(run_dir, &node.id, visit);
fs::create_dir_all(&stage_dir).await?;
let node_ref = NodeVisitRef {
node_id: &node.id,
visit: u32::try_from(visit).unwrap_or(u32::MAX),
};
if let Some(ref store) = services.run_store {
store
.put_node_prompt(&node_ref, &prompt)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("prompt.md"), &prompt).await?;
}
fs::write(stage_dir.join("prompt.md"), &prompt).await?;
let prompt_provider = node
.provider()
@ -169,14 +156,7 @@ impl Handler for PromptHandler {
});
// 4. Write response to logs
if let Some(ref store) = services.run_store {
store
.put_node_response(&node_ref, &response_text)
.await
.map_err(|err| FabroError::handler(err.to_string()))?;
} else {
fs::write(stage_dir.join("response.md"), &response_text).await?;
}
fs::write(stage_dir.join("response.md"), &response_text).await?;
// 5. Build and write status
let mut outcome = Outcome::success();
@ -225,7 +205,7 @@ mod tests {
.await
.unwrap();
let services = EngineServices {
run_store: Some(Arc::clone(&run_store)),
run_store: Arc::clone(&run_store),
..EngineServices::test_default()
};
let logger = crate::event::StoreProgressLogger::new(Arc::clone(&run_store));

12
lib/crates/fabro-workflow/src/operations/create.rs
View file

@ -148,8 +148,8 @@ async fn persist_created_run(
Err(err) => store
.open_run(&record.run_id)
.await
.map_err(|open_err| FabroError::engine(open_err.to_string()))?
.ok_or_else(|| FabroError::engine(err.to_string()))?,
.map_err(|open_err| FabroError::engine(open_err.to_string()))
.or_else(|_| Err(FabroError::engine(err.to_string())))?,
};
run_store.put_run(record).await.map_err(store_error)?;
@ -717,7 +717,7 @@ mod tests {
created.persisted.run_record().workflow_slug.as_deref(),
Some("slug")
);
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap().unwrap();
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap();
assert_eq!(
run_store.get_status().await.unwrap().unwrap().status,
crate::run_status::RunStatus::Submitted
@ -839,11 +839,7 @@ mod tests {
)
.await
.unwrap();
let run_store = store
.open_run_reader(&created.run_id)
.await
.unwrap()
.unwrap();
let run_store = store.open_run_reader(&created.run_id).await.unwrap();
let events = run_store.list_events().await.unwrap();
assert_eq!(

20
lib/crates/fabro-workflow/src/operations/rebuild_meta.rs
View file

@ -26,9 +26,13 @@ pub async fn rebuild_metadata_branch(
bail!("metadata branch already exists for run {run_id}");
}
let run_record = run_store
.get_run()
let state = run_store
.state()
.await?
;
let run_record = state
.run
.clone()
.ok_or_else(|| anyhow::anyhow!("run record not found for {run_id}"))?;
let sig = Signature::now("Fabro", "noreply@fabro.sh")?;
@ -43,10 +47,10 @@ pub async fn rebuild_metadata_branch(
"run.json".to_string(),
serde_json::to_vec_pretty(&run_record)?,
));
if let Some(start) = run_store.get_start().await? {
if let Some(start) = state.start.clone() {
init_entries.push(("start.json".to_string(), serde_json::to_vec_pretty(&start)?));
}
if let Some(sandbox) = run_store.get_sandbox().await? {
if let Some(sandbox) = state.sandbox.clone() {
init_entries.push((
"sandbox.json".to_string(),
serde_json::to_vec_pretty(&sandbox)?,
@ -54,7 +58,7 @@ pub async fn rebuild_metadata_branch(
}
write_entries(&bs, &init_entries, "init run")?;
let mut checkpoints = run_store.list_checkpoints().await?;
let mut checkpoints = state.checkpoints.clone();
backfill_missing_checkpoint_shas(git_store, run_id, &mut checkpoints);
for (_seq, checkpoint) in checkpoints {
@ -69,7 +73,9 @@ pub async fn rebuild_metadata_branch(
for visit in 1..=max_visit {
let visit = u32::try_from(visit)
.with_context(|| format!("visit {visit} for node {node_id} exceeds u32"))?;
let node = run_store.get_node(&NodeVisitRef { node_id, visit }).await?;
let Some(node) = state.node(&NodeVisitRef { node_id, visit }).cloned() else {
continue;
};
if let Some(prompt) = node.prompt {
entries.push((
@ -125,7 +131,7 @@ pub async fn rebuild_metadata_branch(
write_entries(&bs, &entries, "checkpoint")?;
}
if let Some(retro) = run_store.get_retro().await? {
if let Some(retro) = state.retro.clone() {
let entries = vec![("retro.json".to_string(), serde_json::to_vec_pretty(&retro)?)];
write_entries(&bs, &entries, "finalize run")?;
}

23
lib/crates/fabro-workflow/src/operations/resume.rs
View file

@ -11,24 +11,20 @@ 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 Some(record) = services
let state = services
.run_store
.get_status()
.state()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
{
.map_err(|err| FabroError::engine(err.to_string()))?;
if let Some(record) = state.status {
if record.status == RunStatus::Succeeded {
return Err(FabroError::Precondition(
"run already finished successfully — nothing to resume".to_string(),
));
}
}
if let Some(conclusion) = services
.run_store
.get_conclusion()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
{
if let Some(conclusion) = state.conclusion {
if matches!(
conclusion.status,
StageStatus::Success | StageStatus::PartialSuccess | StageStatus::Skipped
@ -39,11 +35,8 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
}
}
let checkpoint = services
.run_store
.get_checkpoint()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
let checkpoint = state
.checkpoint
.ok_or_else(|| FabroError::Precondition("no checkpoint to resume from".to_string()))?;
cleanup_resume_artifacts(run_dir);

22
lib/crates/fabro-workflow/src/operations/start.rs
View file

@ -82,24 +82,18 @@ 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 services
let state = services
.run_store
.get_checkpoint()
.state()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
.is_some()
{
.map_err(|err| FabroError::engine(err.to_string()))?;
if state.checkpoint.is_some() {
return Err(FabroError::Precondition(
"checkpoint.json exists in run directory — did you mean to resume?".to_string(),
));
}
if let Some(record) = services
.run_store
.get_status()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
{
if let Some(record) = state.status {
if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) {
return Err(FabroError::Precondition(format!(
"cannot start run: status is {:?}, expected submitted",
@ -580,7 +574,7 @@ impl RunSession {
};
let pr_opts = PullRequestOptions {
run_dir: retroed.run_options.run_dir.clone(),
run_store: Some(Arc::clone(&retroed.run_store)),
run_store: Arc::clone(&retroed.run_store),
pr_config: self.pr_config,
github_app: self.pr_github_app,
origin_url: self.pr_origin_url,
@ -976,7 +970,7 @@ mod tests {
cancel_token: None,
emitter,
interviewer: Arc::new(fabro_interview::AutoApproveInterviewer),
run_store: store.open_run(&fixtures::RUN_1).await.unwrap().unwrap(),
run_store: store.open_run(&fixtures::RUN_1).await.unwrap(),
github_app: None,
on_node: None,
registry_override: Some(registry),
@ -1052,7 +1046,7 @@ mod tests {
.unwrap();
assert_eq!(started.finalized.conclusion.status, StageStatus::Success);
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap().unwrap();
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap();
assert!(run_store.get_conclusion().await.unwrap().is_some());
}

2
lib/crates/fabro-workflow/src/pipeline/execute.rs
View file

@ -73,7 +73,7 @@ pub async fn execute(init: Initialized) -> Executed {
registry,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),
run_store: Some(Arc::clone(&run_store)),
run_store: Arc::clone(&run_store),
git_state: std::sync::RwLock::new(git_state),
hook_runner: hook_runner.clone(),
env,

16
lib/crates/fabro-workflow/src/pipeline/finalize.rs
View file

@ -69,7 +69,11 @@ pub(crate) async fn build_conclusion_from_store(
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> Conclusion {
let checkpoint = run_store.get_checkpoint().await.ok().flatten();
let checkpoint = run_store
.state()
.await
.ok()
.and_then(|state| state.checkpoint);
let stage_durations = run_store
.list_events()
.await
@ -192,10 +196,12 @@ pub async fn write_finalize_commit(
let git_author = run_options.git_author();
let store = MetadataStore::new(repo_path, &git_author);
let mut entries = scan_node_files_from_store(run_store).await;
let retro_bytes = match run_store.get_retro().await {
Ok(Some(retro)) => serde_json::to_vec_pretty(&retro).ok(),
_ => None,
};
let retro_bytes = run_store
.state()
.await
.ok()
.and_then(|state| state.retro)
.and_then(|retro| serde_json::to_vec_pretty(&retro).ok());
if let Some(bytes) = retro_bytes {
entries.push(("retro.json".to_string(), bytes));
}

14
lib/crates/fabro-workflow/src/pipeline/persist.rs
View file

@ -29,17 +29,15 @@ pub(crate) async fn load_from_store(
run_store: &dyn RunStore,
run_dir: &Path,
) -> Result<Persisted, FabroError> {
let run_record = run_store
.get_run()
let state = run_store
.state()
.await
.map_err(|err| FabroError::engine(err.to_string()))?
.map_err(|err| FabroError::engine(err.to_string()))?;
let run_record = state
.run
.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();
let source = state.graph_source.unwrap_or_default();
Ok(Persisted::new(
graph,

150
lib/crates/fabro-workflow/src/pipeline/pull_request.rs
View file

@ -292,21 +292,17 @@ fn emit_run_notice(
});
}
async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path) -> String {
if let Some(run_store) = run_store {
run_store
.get_final_patch()
.await
.inspect_err(|err| {
tracing::warn!(error = %err, "Failed to load final patch from store for PR");
})
.ok()
.flatten()
.unwrap_or_default()
} else {
let _ = run_dir;
String::new()
}
async fn load_pull_request_diff(run_store: &dyn RunStore, run_dir: &Path) -> String {
let _ = run_dir;
run_store
.state()
.await
.inspect_err(|err| {
tracing::warn!(error = %err, "Failed to load final patch from store for PR");
})
.ok()
.and_then(|state| state.final_patch)
.unwrap_or_default()
}
/// Build a complete PR body by combining LLM-generated narrative with
@ -315,7 +311,7 @@ pub async fn build_pr_body(
diff: &str,
goal: &str,
model: &str,
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
run_dir: &Path,
conclusion: Option<&Conclusion>,
) -> Result<String, String> {
@ -323,54 +319,28 @@ pub async fn build_pr_body(
let plan_text = read_plan_text(run_dir);
let loaded_conclusion = if conclusion.is_none() {
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(),
None => None,
}
run_store
.state()
.await
.inspect_err(|err| {
tracing::warn!(error = %err, "Failed to load conclusion from store for PR body");
})
.ok()
.and_then(|state| state.conclusion)
} else {
None
};
let conclusion = conclusion.or(loaded_conclusion.as_ref());
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(),
None => None,
};
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(),
None => None,
};
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(),
None => None,
};
let run_state = run_store
.state()
.await
.inspect_err(|err| {
tracing::warn!(error = %err, "Failed to load run state from store for PR body");
})
.ok();
let retro = run_state.as_ref().and_then(|state| state.retro.clone());
let run_record = run_state.as_ref().and_then(|state| state.run.clone());
let dot_source = run_state.as_ref().and_then(|state| state.graph_source.clone());
// Build LLM prompt
let system = if plan_text.is_some() {
@ -448,7 +418,7 @@ pub async fn maybe_open_pull_request(
model: &str,
draft: bool,
auto_merge: Option<AutoMergeOptions>,
run_store: Option<&dyn RunStore>,
run_store: &dyn RunStore,
run_dir: &Path,
conclusion: Option<&Conclusion>,
) -> Result<Option<PullRequestRecord>, String> {
@ -519,12 +489,10 @@ pub async fn maybe_open_pull_request(
title,
};
if let Some(run_store) = run_store {
run_store
.put_pull_request(&record)
.await
.map_err(|err| format!("failed to persist pull request in run store: {err}"))?;
}
run_store
.put_pull_request(&record)
.await
.map_err(|err| format!("failed to persist pull request in run store: {err}"))?;
Ok(Some(record))
}
@ -554,8 +522,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) ->
result.status,
StageStatus::Success | StageStatus::PartialSuccess
) {
let diff =
load_pull_request_diff(options.run_store.as_deref(), &options.run_dir).await;
let diff = load_pull_request_diff(options.run_store.as_ref(), &options.run_dir).await;
if let (Some(base_branch), Some(run_branch), Some(creds), Some(origin)) = (
&run_options.base_branch,
pushed_branch.as_deref(),
@ -580,7 +547,7 @@ pub async fn pull_request(concluded: Concluded, options: &PullRequestOptions) ->
&options.model,
pr_cfg.draft,
auto_merge,
options.run_store.as_deref(),
options.run_store.as_ref(),
&options.run_dir,
Some(&conclusion),
)
@ -1067,12 +1034,21 @@ mod tests {
install_mock_llm();
let tmp = tempfile::tempdir().unwrap();
let store = InMemoryStore::default();
let run_store = store
.create_run(
&fixtures::RUN_1,
Utc::now(),
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let conclusion = make_test_conclusion();
let body = build_pr_body(
"diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
"Implement feature",
"mock-model",
None,
run_store.as_ref(),
tmp.path(),
Some(&conclusion),
)
@ -1126,7 +1102,7 @@ mod tests {
"diff --git a/src/lib.rs b/src/lib.rs\n+fn new_feature() {}\n",
"Implement feature",
"mock-model",
Some(run_store.as_ref()),
run_store.as_ref(),
tmp.path(),
Some(&conclusion),
)
@ -1316,6 +1292,15 @@ mod tests {
#[tokio::test]
async fn empty_diff_returns_none() {
let tmp = tempfile::tempdir().unwrap();
let store = InMemoryStore::default();
let run_store = store
.create_run(
&fixtures::RUN_1,
Utc::now(),
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
let creds = GitHubAppCredentials {
app_id: "123".to_string(),
private_key_pem: "unused".to_string(),
@ -1330,7 +1315,7 @@ mod tests {
"claude-sonnet-4-20250514",
false,
None,
None,
run_store.as_ref(),
tmp.path(),
None,
)
@ -1343,20 +1328,35 @@ mod tests {
async fn load_pull_request_diff_uses_store_without_disk_patch() {
let tmp = tempfile::tempdir().unwrap();
let store = InMemoryStore::default();
let created_at = Utc::now();
let run_store = store
.create_run(
&fixtures::RUN_1,
Utc::now(),
created_at,
Some(&tmp.path().display().to_string()),
)
.await
.unwrap();
run_store
.put_run(&RunRecord {
run_id: fixtures::RUN_1,
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,
working_directory: tmp.path().to_path_buf(),
host_repo_path: None,
base_branch: None,
labels: std::collections::HashMap::new(),
})
.await
.unwrap();
run_store
.put_final_patch("diff --git a/src/lib.rs b/src/lib.rs\n+fn from_store() {}\n")
.await
.unwrap();
let diff = load_pull_request_diff(Some(run_store.as_ref()), tmp.path()).await;
let diff = load_pull_request_diff(run_store.as_ref(), tmp.path()).await;
assert!(diff.contains("from_store"));
}

32
lib/crates/fabro-workflow/src/pipeline/retro.rs
View file

@ -8,12 +8,14 @@ use fabro_retro::retro_agent::{
use super::types::{Executed, RetroOptions, Retroed};
use crate::event::WorkflowRunEvent;
#[cfg(test)]
use crate::records::RunRecord;
pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
let cp = match options.run_store.get_checkpoint().await {
Ok(Some(cp)) => cp,
let state = match options.run_store.state().await {
Ok(state) => state,
Err(e) => {
tracing::warn!(error = %e, "Could not load checkpoint, skipping retro");
tracing::warn!(error = %e, "Could not load run state, skipping retro");
if let Some(ref emitter) = options.emitter {
emitter.emit(&WorkflowRunEvent::RetroFailed {
error: e.to_string(),
@ -22,7 +24,10 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
}
return None;
}
Ok(None) => {
};
let cp = match state.checkpoint {
Some(cp) => cp,
None => {
tracing::warn!("Could not load checkpoint, skipping retro");
if let Some(ref emitter) = options.emitter {
emitter.emit(&WorkflowRunEvent::RetroFailed {
@ -86,7 +91,7 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
});
run_retro_agent(
&options.sandbox,
Some(&*options.run_store),
options.run_store.as_ref(),
&options.run_dir,
client,
options.provider,
@ -213,15 +218,30 @@ mod tests {
run_dir: &std::path::Path,
checkpoint: &Checkpoint,
) -> Arc<dyn fabro_store::RunStore> {
let created_at = Utc::now();
let inner = InMemoryStore::default()
.create_run(
&test_run_id(),
Utc::now(),
created_at,
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let run_store: Arc<dyn fabro_store::RunStore> = inner;
run_store
.put_run(&RunRecord {
run_id: test_run_id(),
created_at,
settings: Settings::default(),
graph: Graph::new("test"),
workflow_slug: None,
working_directory: run_dir.to_path_buf(),
host_repo_path: None,
base_branch: None,
labels: std::collections::HashMap::new(),
})
.await
.unwrap();
run_store.put_checkpoint(checkpoint).await.unwrap();
run_store
}

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

@ -356,7 +356,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 run_store: Arc<dyn RunStore>,
pub pr_config: Option<PullRequestSettings>,
pub github_app: Option<fabro_github::GitHubAppCredentials>,
pub origin_url: Option<String>,