scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Remove legacy run file projection layer

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-01 22:23:22 -07:00
parent ebe7db8e62
commit 93eab71892
No known key found for this signature in database
39 changed files with 488 additions and 2247 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -6,7 +6,6 @@ use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant};
use anyhow::Result;
use fabro_config::FabroSettingsExt;
use fabro_types::RunId;
use futures::StreamExt;
@ -14,7 +13,7 @@ use fabro_interview::{AnswerValue, ConsoleInterviewer};
use fabro_store::{EventEnvelope, RunStore, RuntimeState};
use fabro_util::terminal::Styles;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::records::{Conclusion, ConclusionExt, RunRecord, RunRecordExt};
use fabro_workflow::records::{Conclusion, ConclusionExt};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt};
use serde_json::{Map, Value};
use tokio::signal::ctrl_c;
@ -44,18 +43,10 @@ pub(crate) async fn attach_run(
engine_child: Option<std::process::Child>,
json_output: bool,
) -> Result<ExitCode> {
let run_record = RunRecord::load(run_dir).ok();
let inferred_storage_dir = infer_storage_dir(run_dir);
let fallback_storage_dir = run_record
.as_ref()
.map(|record| record.settings.storage_dir())
.or(inferred_storage_dir);
let inferred_run_id = infer_run_id(run_dir);
let storage_dir = storage_dir.map(Path::to_path_buf).or(fallback_storage_dir);
let run_id = run_id
.copied()
.or_else(|| run_record.as_ref().map(|record| record.run_id))
.or(inferred_run_id);
let storage_dir = storage_dir.map(Path::to_path_buf).or(inferred_storage_dir);
let run_id = run_id.copied().or(inferred_run_id);
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 {
@ -104,13 +95,9 @@ pub(crate) async fn attach_run(
}
}
let verbose = run_record
.as_ref()
.map(|record| record.settings.verbose_enabled())
.unwrap_or(false);
attach_run_files(
run_dir,
verbose,
false,
kill_on_detach,
styles,
engine_child,

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

@ -54,7 +54,7 @@ pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
}
async fn resolve_diff(
run_dir: &Path,
_run_dir: &Path,
run_store: &dyn fabro_store::RunStore,
args: &DiffArgs,
) -> Result<String> {
@ -73,11 +73,7 @@ async fn resolve_diff(
}
}
debug!(node_id, "Reading per-node diff");
let node_patch = run_dir.join("nodes").join(node_id).join("diff.patch");
return std::fs::read_to_string(&node_patch).with_context(|| {
format!("No diff found for node '{node_id}' — check the node ID and try again")
});
bail!("No diff found for node '{node_id}' — check the node ID and try again");
}
let start = run_store

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

@ -10,7 +10,7 @@ use fabro_util::text::strip_goal_decoration;
use fabro_workflow::asset_snapshot::collect_asset_paths;
use fabro_workflow::outcome::{StageStatus, format_cost};
use fabro_workflow::pipeline::{Persisted, Validated};
use fabro_workflow::records::{Checkpoint, CheckpointExt, Conclusion};
use fabro_workflow::records::Conclusion;
use indicatif::HumanDuration;
use crate::shared::{format_tokens_human, print_diagnostics, relative_path, tilde_path};
@ -204,12 +204,12 @@ pub(crate) fn print_run_conclusion(
pub(crate) async fn print_final_output(
run_store: Option<&dyn fabro_store::RunStore>,
run_dir: &Path,
_run_dir: &Path,
styles: &Styles,
) {
let checkpoint = match run_store {
Some(run_store) => run_store.get_checkpoint().await.ok().flatten(),
None => Checkpoint::load(&run_dir.join("checkpoint.json")).ok(),
None => None,
};
let Some(checkpoint) = checkpoint else {
return;

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

@ -10,7 +10,6 @@ use fabro_workflow::operations::{
RewindInput, RewindTarget, RunTimeline, build_timeline_or_rebuild,
find_run_id_by_prefix_or_store, rewind,
};
use fabro_workflow::records::CheckpointExt;
use fabro_workflow::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use fabro_workflow::run_status::RunStatus;
@ -132,16 +131,8 @@ async fn reset_rewound_run_state(
.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")?;
checkpoint.save(&run_dir.join("checkpoint.json"))?;
for name in [
"conclusion.json",
"pull_request.json",
"detached_failure.json",
"progress.jsonl",
"retro.json",
"final.patch",
] {
for name in ["detached_failure.json"] {
let _ = std::fs::remove_file(run_dir.join(name));
}
@ -164,9 +155,7 @@ async fn reset_rewound_run_state(
.await
.map_err(|err| anyhow::anyhow!("failed to restore start record after rewind: {err}"))?;
}
if let Some(dot_source) =
store_graph.or_else(|| std::fs::read_to_string(run_dir.join("workflow.fabro")).ok())
{
if let Some(dot_source) = store_graph {
run_store
.put_graph(&dot_source)
.await

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

@ -2,14 +2,13 @@ use std::path::Path;
use anyhow::{Context, Result, bail};
use fabro_config::FabroSettingsExt;
use fabro_sandbox::SandboxRecordExt;
use fabro_store::Store;
use tracing::warn;
use fabro_sandbox::reconnect::reconnect as reconnect_sandbox;
use fabro_workflow::run_lookup::RunInfo;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord, write_run_status};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord};
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::shared::print_json_pretty;
@ -116,7 +115,6 @@ 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<()> {
write_run_status(&run.path, RunStatus::Removing, None);
let run_store = match store.open_run_reader(&run.run_id).await {
Ok(run_store) => run_store,
Err(err) => {
@ -168,7 +166,7 @@ async fn delete_run_store_state(store: &dyn Store, run: &RunInfo) -> Result<()>
}
async fn load_sandbox_record(
run_dir: &Path,
_run_dir: &Path,
run_store: Option<&dyn fabro_store::RunStore>,
) -> Option<fabro_sandbox::SandboxRecord> {
if let Some(run_store) = run_store {
@ -180,7 +178,5 @@ async fn load_sandbox_record(
}
}
}
let sandbox_path = run_dir.join("sandbox.json");
fabro_sandbox::SandboxRecord::load(&sandbox_path).ok()
None
}

4
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -125,8 +125,8 @@ fn attach_replays_from_store_without_run_json_or_progress_jsonl() {
.success();
let run = resolve_run(&context, run_id);
std::fs::remove_file(run.run_dir.join("run.json")).unwrap();
std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap();
let _ = std::fs::remove_file(run.run_dir.join("run.json"));
let _ = std::fs::remove_file(run.run_dir.join("progress.jsonl"));
let mut cmd = context.command();
cmd.args(["attach", run_id]);

55
lib/crates/fabro-cli/tests/it/cmd/create.rs
View file

@ -3,9 +3,9 @@ use serde_json::json;
use fabro_test::{fabro_snapshot, test_context};
use crate::support::{fabro_json_snapshot, read_json};
use crate::support::fabro_json_snapshot;
use super::support::{fixture, output_stdout, resolve_run};
use super::support::{fixture, output_stdout, resolve_run, run_snapshot};
#[test]
fn help() {
@ -77,14 +77,13 @@ digraph BarBaz {
.success();
let run_dir = context.find_run_dir(run_id);
let run_record = read_json(run_dir.join("run.json"));
let cached_graph = std::fs::read_to_string(run_dir.join("workflow.fabro")).unwrap();
let snapshot = run_snapshot(&run_dir);
fabro_json_snapshot!(
context,
serde_json::json!({
"workflow_slug": run_record["workflow_slug"],
"graph_name": run_record["graph"]["name"],
"cached_graph_lines": cached_graph.lines().collect::<Vec<_>>(),
"workflow_slug": snapshot.run.workflow_slug,
"graph_name": snapshot.run.graph.name,
"cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::<Vec<_>>(),
}),
@r#"
{
@ -133,14 +132,13 @@ digraph FooWorkflow {
.success();
let run_dir = context.find_run_dir(run_id);
let run_record = read_json(run_dir.join("run.json"));
let cached_graph = std::fs::read_to_string(run_dir.join("workflow.fabro")).unwrap();
let snapshot = run_snapshot(&run_dir);
fabro_json_snapshot!(
context,
serde_json::json!({
"workflow_slug": run_record["workflow_slug"],
"graph_name": run_record["graph"]["name"],
"cached_graph_lines": cached_graph.lines().collect::<Vec<_>>(),
"workflow_slug": snapshot.run.workflow_slug,
"graph_name": snapshot.run.graph.name,
"cached_graph_lines": snapshot.graph.expect("graph should exist").lines().collect::<Vec<_>>(),
}),
@r#"
{
@ -159,7 +157,7 @@ digraph FooWorkflow {
}
#[test]
fn create_persists_requested_overrides_into_run_json() {
fn create_persists_requested_overrides_into_store() {
let context = test_context!();
let workflow = fixture("simple.fabro");
let mut cmd = context.command();
@ -200,26 +198,26 @@ fn create_persists_requested_overrides_into_run_json() {
.expect("create should print a run ID")
.to_string();
let run = resolve_run(&context, &run_id);
let run_json = read_json(run.run_dir.join("run.json"));
let snapshot = run_snapshot(&run.run_dir);
let labels = json!({
"env": run_json.pointer("/labels/env"),
"team": run_json.pointer("/labels/team"),
"env": snapshot.run.labels.get("env"),
"team": snapshot.run.labels.get("team"),
});
let compact = json!({
"workflow_slug": run_json["workflow_slug"],
"workflow_slug": snapshot.run.workflow_slug,
"settings": {
"goal": run_json.pointer("/settings/goal"),
"dry_run": run_json.pointer("/settings/dry_run"),
"auto_approve": run_json.pointer("/settings/auto_approve"),
"no_retro": run_json.pointer("/settings/no_retro"),
"verbose": run_json.pointer("/settings/verbose"),
"goal": snapshot.run.settings.goal,
"dry_run": snapshot.run.settings.dry_run,
"auto_approve": snapshot.run.settings.auto_approve,
"no_retro": snapshot.run.settings.no_retro,
"verbose": snapshot.run.settings.verbose,
"llm": {
"model": run_json.pointer("/settings/llm/model"),
"provider": run_json.pointer("/settings/llm/provider"),
"model": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.model.clone()),
"provider": snapshot.run.settings.llm.as_ref().and_then(|llm| llm.provider.clone()),
},
"sandbox": {
"provider": run_json.pointer("/settings/sandbox/provider"),
"preserve": run_json.pointer("/settings/sandbox/preserve"),
"provider": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.provider.clone()),
"preserve": snapshot.run.settings.sandbox.as_ref().and_then(|sandbox| sandbox.preserve),
},
},
"labels": labels,
@ -274,11 +272,10 @@ fn create_json_implies_auto_approve() {
.as_str()
.expect("create JSON should include run_id");
let run = resolve_run(&context, run_id);
let run_json = read_json(run.run_dir.join("run.json"));
assert_eq!(
run_json.pointer("/settings/auto_approve"),
Some(&json!(true))
run_snapshot(&run.run_dir).run.settings.auto_approve,
Some(true)
);
}

23
lib/crates/fabro-cli/tests/it/cmd/detached.rs
View file

@ -1,6 +1,7 @@
use fabro_test::{fabro_snapshot, test_context};
use crate::support::{fabro_json_snapshot, read_json};
use super::support::run_snapshot;
use crate::support::fabro_json_snapshot;
#[test]
fn help() {
@ -86,7 +87,12 @@ digraph CachedGraph {
.assert()
.success();
let conclusion = read_json(run_dir.join("conclusion.json"));
let conclusion = serde_json::to_value(
run_snapshot(&run_dir)
.conclusion
.expect("conclusion should exist"),
)
.unwrap();
fabro_json_snapshot!(
context,
serde_json::json!({
@ -140,11 +146,11 @@ digraph GitHubApp {
.success();
let run_dir = context.find_run_dir(run_id);
let run_record = read_json(run_dir.join("run.json"));
let snapshot = run_snapshot(&run_dir);
fabro_json_snapshot!(
context,
serde_json::json!({
"app_id": run_record["settings"]["git"]["app_id"],
"app_id": snapshot.run.settings.git.and_then(|git| git.app_id),
}),
@r#"
{
@ -207,8 +213,6 @@ digraph DetachedStoreOnly {
.success();
let run_dir = context.find_run_dir(run_id);
std::fs::remove_file(run_dir.join("run.json")).unwrap();
context
.command()
.args([
@ -224,7 +228,12 @@ digraph DetachedStoreOnly {
.assert()
.success();
let conclusion = read_json(run_dir.join("conclusion.json"));
let conclusion = serde_json::to_value(
run_snapshot(&run_dir)
.conclusion
.expect("conclusion should exist"),
)
.unwrap();
fabro_json_snapshot!(
context,
serde_json::json!({

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

@ -115,8 +115,14 @@ fn diff_completed_run_reads_store_final_patch_without_disk_file() {
let context = test_context!();
let setup = setup_git_backed_changed_run(&context);
let run_id: RunId = setup.run.run_id.parse().unwrap();
let patch = std::fs::read_to_string(setup.run.run_dir.join("final.patch")).unwrap();
std::fs::remove_file(setup.run.run_dir.join("final.patch")).unwrap();
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();
run_store.get_final_patch().await.unwrap().unwrap()
})
});
let _ = std::fs::remove_file(setup.run.run_dir.join("final.patch"));
with_runtime(|runtime| {
runtime.block_on(async {
@ -172,9 +178,22 @@ fn diff_node_reads_store_patch_without_disk_file() {
let context = test_context!();
let setup = setup_git_backed_changed_run(&context);
let run_id: RunId = setup.run.run_id.parse().unwrap();
let patch =
std::fs::read_to_string(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap();
std::fs::remove_file(setup.run.run_dir.join("nodes/step_one/diff.patch")).unwrap();
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();
run_store
.get_node(&fabro_store::NodeVisitRef {
node_id: "step_one",
visit: 1,
})
.await
.unwrap()
.diff
.unwrap()
})
});
let _ = std::fs::remove_file(setup.run.run_dir.join("nodes/step_one/diff.patch"));
with_runtime(|runtime| {
runtime.block_on(async {

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

@ -116,7 +116,7 @@ fn inspect_completed_run_reads_store_without_disk_metadata_files() {
"checkpoint.json",
"sandbox.json",
] {
std::fs::remove_file(run.run_dir.join(name)).unwrap();
let _ = std::fs::remove_file(run.run_dir.join(name));
}
let output = run_success(&context, &["inspect", &run.run_id]);

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

@ -92,7 +92,7 @@ fn logs_completed_run_outputs_raw_ndjson() {
fn logs_completed_run_reads_store_without_progress_jsonl() {
let context = test_context!();
let run = setup_completed_dry_run(&context);
std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap();
let _ = std::fs::remove_file(run.run_dir.join("progress.jsonl"));
let mut filters = context.filters();
filters.push((

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

@ -71,7 +71,7 @@ fn resume_rewound_run_succeeds() {
&setup.repo_dir,
&["rev-parse", &format!("fabro/run/{}", setup.run.run_id)],
);
std::fs::remove_file(setup.run.run_dir.join("run.json")).unwrap();
let _ = std::fs::remove_file(setup.run.run_dir.join("run.json"));
let mut resume_cmd = context.command();
resume_cmd.current_dir(&setup.repo_dir);

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

@ -85,7 +85,7 @@ fn rewind_target_updates_metadata_and_resume_hint() {
let mut cmd = context.command();
cmd.current_dir(&setup.repo_dir);
std::fs::remove_file(setup.run.run_dir.join("run.json")).unwrap();
let _ = std::fs::remove_file(setup.run.run_dir.join("run.json"));
cmd.args(["rewind", &setup.run.run_id, "@1", "--no-push"]);
let (snapshot, output) = run_and_format(&mut cmd, &git_filters(&context));

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

@ -122,7 +122,7 @@ fn rm_force_deletes_submitted_run() {
fn rm_force_deletes_run_without_sandbox_json_when_store_has_sandbox() {
let context = test_context!();
let setup = setup_local_sandbox_run(&context);
std::fs::remove_file(setup.run.run_dir.join("sandbox.json")).unwrap();
let _ = std::fs::remove_file(setup.run.run_dir.join("sandbox.json"));
let mut filters = context.filters();
filters.push((

576
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -1,9 +1,7 @@
use fabro_test::{fabro_snapshot, test_context};
use serde_json::Value;
use crate::support::{
example_fixture, fabro_json_snapshot, read_json, read_jsonl, run_output_filters,
};
use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters};
#[test]
fn help() {
@ -77,7 +75,7 @@ fn dry_run_simple() {
}
#[test]
fn dry_run_writes_jsonl_and_live_json() {
fn dry_run_persists_event_history_in_store() {
let context = test_context!();
let run_id = "01ARZ3NDEKTSV4RRFFQ69G5FB8";
@ -96,521 +94,61 @@ fn dry_run_writes_jsonl_and_live_json() {
.assert()
.success();
let run_dir = context.find_run_dir(run_id);
let jsonl_path = run_dir.join("progress.jsonl");
let progress = read_jsonl(&jsonl_path);
context.find_run_dir(run_id);
let output = context
.command()
.args(["logs", run_id])
.output()
.expect("logs command should execute");
assert!(
output.status.success(),
"logs failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
let progress: Vec<Value> = String::from_utf8(output.stdout)
.expect("stdout should be UTF-8")
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str(line).expect("logs output should be JSONL"))
.collect();
assert!(
!progress.is_empty(),
"progress.jsonl should have at least one line"
"store-backed event history should have at least one line"
);
assert_eq!(
progress.first().and_then(|event| event["event"].as_str()),
Some("run.created")
);
assert_eq!(
progress
.first()
.and_then(|event| event.pointer("/properties/settings/auto_approve"))
.and_then(Value::as_bool),
Some(true)
);
assert_eq!(
progress.last().and_then(|event| event["event"].as_str()),
Some("sandbox.cleanup.completed")
);
fabro_json_snapshot!(context, &progress, @r#"
[
{
"id": "[EVENT_ID]",
"ts": "[TIMESTAMP]",
"run_id": "[ULID]",
"event": "run.created",
"properties": {
"workflow_slug": "simple",
"settings": {
"auto_approve": true,
"dry_run": true,
"fabro": {
"root": "fabro/"
},
"features": {
"retros": false,
"session_sandboxes": false
},
"goal": "Run tests and report results",
"hooks": [
{
"blocking": true,
"command": "cargo fmt",
"event": "post_tool_use",
"matcher": "write_file|edit_file|apply_patch",
"name": "cargo-fmt",
"sandbox": null,
"timeout_ms": null
}
],
"llm": {
"fallbacks": null,
"model": "claude-sonnet-4-6",
"provider": "anthropic"
},
"mode": "standalone",
"pull_request": {
"auto_merge": false,
"draft": false,
"enabled": true,
"merge_strategy": "squash"
},
"sandbox": {
"daytona": {
"auto_stop_interval": 30,
"labels": {
"repo": "fabro-sh/fabro"
},
"network": null,
"skip_clone": false,
"snapshot": {
"cpu": 4,
"disk": 20,
"dockerfile": "FROM ubuntu:24.04/n/nRUN apt-get update && apt-get install -y --no-install-recommends curl git ca-certificates build-essential pkg-config libssl-dev unzip python3 && rm -rf /var/lib/apt/lists/*/n/n# GitHub CLI/nRUN curl -fsSL https://cli.github.com/packages/githubcli-archive-keyring.gpg | dd of=/usr/share/keyrings/githubcli-archive-keyring.gpg && echo \"deb [arch=$(dpkg --print-architecture) signed-by=/usr/share/keyrings/githubcli-archive-keyring.gpg] https://cli.github.com/packages stable main\" | tee /etc/apt/sources.list.d/github-cli.list > /dev/null && apt-get update && apt-get install -y --no-install-recommends gh && rm -rf /var/lib/apt/lists/*/n/n# Rust/nRUN curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y/nENV PATH=\"/root/.cargo/bin:${PATH}\"/nRUN cargo install cargo-nextest --locked/nENV CARGO_INCREMENTAL=0/n/n# Bun/nRUN curl -fsSL https://bun.sh/install | bash/nENV PATH=\"/root/.bun/bin:${PATH}\"/n/nWORKDIR /root/n",
"memory": 8,
"name": "fabro-v6"
}
},
"devcontainer": null,
"env": null,
"local": null,
"preserve": null,
"provider": "local"
},
"storage_dir": "[STORAGE_DIR]",
"version": 1
},
"graph": {
"attrs": {
"goal": {
"String": "Run tests and report results"
},
"rankdir": {
"String": "LR"
}
},
"edges": [
{
"attrs": {},
"from": "start",
"to": "run_tests"
},
{
"attrs": {},
"from": "run_tests",
"to": "report"
},
{
"attrs": {},
"from": "report",
"to": "exit"
}
],
"name": "Simple",
"nodes": {
"exit": {
"attrs": {
"label": {
"String": "Exit"
},
"shape": {
"String": "Msquare"
}
},
"id": "exit"
},
"report": {
"attrs": {
"label": {
"String": "Report"
},
"prompt": {
"String": "Summarize the test results"
}
},
"id": "report"
},
"run_tests": {
"attrs": {
"label": {
"String": "Run Tests"
},
"prompt": {
"String": "Run the test suite and report results"
}
},
"id": "run_tests"
},
"start": {
"attrs": {
"label": {
"String": "Start"
},
"shape": {
"String": "Mdiamond"
}
},
"id": "start"
}
}
},
"workflow_source": "digraph Simple {/n graph [goal=\"Run tests and report results\"]/n rankdir=LR/n/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n/n run_tests [label=\"Run Tests\", prompt=\"Run the test suite and report results\"]/n report [label=\"Report\", prompt=\"Summarize the test results\"]/n/n start -> run_tests -> report -> exit/n}/n",
"labels": {},
"run_dir": "[DRY_RUN_DIR]",
"working_directory": "[TEMP_DIR]",
"host_repo_path": "[TEMP_DIR]"
}
},
{
"event": "sandbox.initializing",
"id": "[EVENT_ID]",
"properties": {
"provider": "local"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.ready",
"id": "[EVENT_ID]",
"properties": {
"cpu": null,
"duration_ms": "[DURATION_MS]",
"memory": null,
"name": null,
"provider": "local",
"url": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.initialized",
"id": "[EVENT_ID]",
"properties": {
"provider": "local",
"working_directory": "[TEMP_DIR]"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.started",
"id": "[EVENT_ID]",
"properties": {
"goal": "Run tests and report results",
"name": "Simple"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "Start",
"properties": {
"attempt": 1,
"handler_type": "start",
"index": 0,
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.completed",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "Start",
"properties": {
"attempt": 1,
"context_values": {
"current.preamble": "Goal: Run tests and report results/n",
"current_node": "start",
"graph.goal": "Run tests and report results",
"graph.rankdir": "LR",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.run_id": "[ULID]",
"internal.thread_id": null
},
"duration_ms": "[DURATION_MS]",
"files_touched": [],
"index": 0,
"max_attempts": 1,
"node_visits": {
"start": 1
},
"notes": "[Simulated] start",
"preferred_label": null,
"status": "success",
"suggested_next_ids": [],
"usage": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "edge.selected",
"id": "[EVENT_ID]",
"properties": {
"condition": null,
"from_node": "start",
"is_jump": false,
"label": null,
"reason": "unconditional",
"stage_status": "success",
"to_node": "run_tests"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "checkpoint.completed",
"id": "[EVENT_ID]",
"node_id": "start",
"node_label": "start",
"properties": {
"status": "success"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "run_tests",
"node_label": "Run Tests",
"properties": {
"attempt": 1,
"handler_type": "agent",
"index": 1,
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.completed",
"id": "[EVENT_ID]",
"node_id": "run_tests",
"node_label": "Run Tests",
"properties": {
"attempt": 1,
"context_updates": {
"last_response": "[Simulated] Response for stage: run_tests",
"last_stage": "run_tests",
"response.run_tests": "[Simulated] Response for stage: run_tests"
},
"context_values": {
"current.preamble": "Goal: Run tests and report results/n",
"current_node": "run_tests",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Run tests and report results",
"graph.rankdir": "LR",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": "start",
"outcome": "success",
"thread.start.current_node": "run_tests"
},
"duration_ms": "[DURATION_MS]",
"files_touched": [],
"index": 1,
"max_attempts": 1,
"node_visits": {
"run_tests": 1,
"start": 1
},
"notes": "[Simulated] run_tests",
"preferred_label": null,
"status": "success",
"suggested_next_ids": [],
"usage": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "edge.selected",
"id": "[EVENT_ID]",
"properties": {
"condition": null,
"from_node": "run_tests",
"is_jump": false,
"label": null,
"reason": "unconditional",
"stage_status": "success",
"to_node": "report"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "checkpoint.completed",
"id": "[EVENT_ID]",
"node_id": "run_tests",
"node_label": "run_tests",
"properties": {
"status": "success"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "report",
"node_label": "Report",
"properties": {
"attempt": 1,
"handler_type": "agent",
"index": 2,
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.completed",
"id": "[EVENT_ID]",
"node_id": "report",
"node_label": "Report",
"properties": {
"attempt": 1,
"context_updates": {
"last_response": "[Simulated] Response for stage: report",
"last_stage": "report",
"response.report": "[Simulated] Response for stage: report"
},
"context_values": {
"current.preamble": "Goal: Run tests and report results/n/n## Completed stages/n- **run_tests**: success/n",
"current_node": "report",
"failure_class": "",
"failure_signature": "",
"graph.goal": "Run tests and report results",
"graph.rankdir": "LR",
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.retry_count.run_tests": 0,
"internal.retry_count.start": 0,
"internal.run_id": "[ULID]",
"internal.thread_id": "run_tests",
"last_response": "[Simulated] Response for stage: run_tests",
"last_stage": "run_tests",
"outcome": "success",
"response.run_tests": "[Simulated] Response for stage: run_tests",
"thread.run_tests.current_node": "report",
"thread.start.current_node": "run_tests"
},
"duration_ms": "[DURATION_MS]",
"files_touched": [],
"index": 2,
"max_attempts": 1,
"node_visits": {
"report": 1,
"run_tests": 1,
"start": 1
},
"notes": "[Simulated] report",
"preferred_label": null,
"status": "success",
"suggested_next_ids": [],
"usage": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "edge.selected",
"id": "[EVENT_ID]",
"properties": {
"condition": null,
"from_node": "report",
"is_jump": false,
"label": null,
"reason": "unconditional",
"stage_status": "success",
"to_node": "exit"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "checkpoint.completed",
"id": "[EVENT_ID]",
"node_id": "report",
"node_label": "report",
"properties": {
"status": "success"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.started",
"id": "[EVENT_ID]",
"node_id": "exit",
"node_label": "Exit",
"properties": {
"attempt": 1,
"handler_type": "exit",
"index": 3,
"max_attempts": 1
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "stage.completed",
"id": "[EVENT_ID]",
"node_id": "exit",
"node_label": "Exit",
"properties": {
"attempt": 1,
"duration_ms": "[DURATION_MS]",
"files_touched": [],
"index": 3,
"max_attempts": 1,
"notes": null,
"preferred_label": null,
"status": "success",
"suggested_next_ids": [],
"usage": null
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "run.completed",
"id": "[EVENT_ID]",
"properties": {
"artifact_count": 0,
"duration_ms": "[DURATION_MS]",
"status": "success"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.cleanup.started",
"id": "[EVENT_ID]",
"properties": {
"provider": "local"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
},
{
"event": "sandbox.cleanup.completed",
"id": "[EVENT_ID]",
"properties": {
"duration_ms": "[DURATION_MS]",
"provider": "local"
},
"run_id": "[ULID]",
"ts": "[TIMESTAMP]"
}
]
"#);
let live_path = run_dir.join("live.json");
let live_content = read_json(&live_path);
let tail_output = context
.command()
.args(["logs", "--tail", "1", run_id])
.output()
.expect("tail logs command should execute");
assert!(
tail_output.status.success(),
"tail logs failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&tail_output.stdout),
String::from_utf8_lossy(&tail_output.stderr)
);
let live_content: Value = String::from_utf8(tail_output.stdout)
.expect("stdout should be UTF-8")
.lines()
.find(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str(line).expect("tail logs output should be JSON"))
.expect("tail logs should include the latest event");
fabro_json_snapshot!(context, &live_content, @r#"
{
"event": "sandbox.cleanup.completed",
@ -645,9 +183,7 @@ fn run_id_passthrough_uses_provided_ulid() {
.assert()
.success();
let run_dir = context.find_run_dir(run_id);
let run_record = read_json(run_dir.join("run.json"));
assert_eq!(run_record["run_id"].as_str(), Some(run_id));
context.find_run_dir(run_id);
}
#[test]
@ -1219,10 +755,8 @@ fn json_run_implies_auto_approve_for_human_gates() {
]
"#);
let run = context.single_run_dir();
let run_json = read_json(run.join("run.json"));
assert_eq!(
run_json.pointer("/settings/auto_approve"),
progress[0].pointer("/properties/settings/auto_approve"),
Some(&serde_json::json!(true))
);
}

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

@ -97,7 +97,7 @@ fn start_by_run_id_starts_created_run_without_run_json_or_status_json() {
.success();
let run_dir = context.find_run_dir(run_id);
std::fs::remove_file(run_dir.join("run.json")).unwrap();
let _ = std::fs::remove_file(run_dir.join("run.json"));
context.command().args(["start", run_id]).assert().success();
let output = context

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

@ -1,9 +1,13 @@
use std::collections::BTreeSet;
use std::path::{Path, PathBuf};
use std::process::Output;
use std::sync::Arc;
use std::time::{Duration, Instant};
use fabro_store::{RunSnapshot, RunStore, SlateStore, Store};
use fabro_test::TestContext;
use fabro_types::RunId;
use object_store::local::LocalFileSystem;
use serde_json::Value;
use shlex::try_quote;
@ -162,10 +166,13 @@ 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.run_dir.join("progress.jsonl").exists() {
while run_store(&run.run_dir)
.and_then(|store| block_on(store.list_events()).ok())
.is_none_or(|events| events.is_empty())
{
assert!(
Instant::now() < deadline,
"timed out waiting for progress.jsonl for {run_id}"
"timed out waiting for store events for {run_id}"
);
std::thread::sleep(Duration::from_millis(50));
}
@ -281,8 +288,10 @@ worktree_mode = "never"
let run = run_local_workflow(context, &workspace_dir, "run.toml");
assert!(
run.run_dir.join("sandbox.json").exists(),
"setup_local_sandbox_run should persist sandbox.json"
run_store(&run.run_dir)
.and_then(|store| block_on(store.get_sandbox()).ok())
.flatten()
.is_some()
);
WorkspaceRunSetup { run, workspace_dir }
@ -369,8 +378,10 @@ 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) = read_json_if_exists(&run_dir.join("status.json"))
.and_then(|value| value["status"].as_str().map(ToOwned::to_owned))
if let Some(status) = run_store(run_dir)
.and_then(|store| block_on(store.get_status()).ok())
.flatten()
.map(|record| record.status.to_string())
{
if expected.iter().any(|candidate| *candidate == status) {
return status;
@ -401,10 +412,7 @@ pub(crate) fn only_run(context: &TestContext) -> RunSetup {
runs_dir.display()
);
let run_dir = entries[0].clone();
let run_id = read_json(&run_dir.join("run.json"))["run_id"]
.as_str()
.expect("run.json should include run_id")
.to_string();
let run_id = infer_run_id(&run_dir);
RunSetup { run_id, run_dir }
}
@ -457,6 +465,42 @@ pub(crate) fn find_run_dir(storage_dir: &Path, run_id: &str) -> Option<PathBuf>
})
}
fn infer_run_id(run_dir: &Path) -> String {
if let Ok(id) = std::fs::read_to_string(run_dir.join("id.txt")) {
return id.trim().to_string();
}
run_dir
.file_name()
.map(|name| name.to_string_lossy().to_string())
.and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned))
.filter(|value| !value.is_empty())
.expect("run directory name should contain run id suffix")
}
fn block_on<T>(future: impl std::future::Future<Output = T>) -> T {
tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.unwrap()
.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()?);
let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(5)));
block_on(store.open_run_reader(&run_id)).ok().flatten()
}
pub(crate) fn run_snapshot(run_dir: &Path) -> RunSnapshot {
run_store(run_dir)
.and_then(|store| block_on(store.get_snapshot()).ok())
.flatten()
.expect("run store snapshot should exist")
}
pub(crate) fn git_stdout(repo_dir: &Path, args: &[&str]) -> String {
stdout(&git_success(repo_dir, args))
}
@ -724,7 +768,12 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git
}
let run = only_run(context);
let start = read_json(&run.run_dir.join("start.json"));
let start = serde_json::to_value(
run_snapshot(&run.run_dir)
.start
.expect("start record should exist"),
)
.unwrap();
assert_eq!(
start["run_branch"].as_str(),
Some(format!("fabro/run/{}", run.run_id).as_str())
@ -733,22 +782,27 @@ fn setup_git_backed_run(context: &TestContext, workflow: GitWorkflowKind) -> Git
match workflow {
GitWorkflowKind::Changed => {
assert!(
run.run_dir.join("final.patch").exists(),
"changed git-backed run should emit final.patch"
run_snapshot(&run.run_dir).final_patch.is_some(),
"changed git-backed run should persist final patch in store"
);
let snapshot = run_snapshot(&run.run_dir);
assert!(
snapshot
.nodes
.iter()
.any(|node| node.node_id == "step_one" && node.diff.is_some())
);
assert!(
run.run_dir.join("nodes/step_one/diff.patch").exists(),
"changed git-backed run should emit a diff for step_one"
);
assert!(
run.run_dir.join("nodes/step_two/diff.patch").exists(),
"changed git-backed run should emit a diff for step_two"
snapshot
.nodes
.iter()
.any(|node| node.node_id == "step_two" && node.diff.is_some())
);
}
GitWorkflowKind::Noop => {
assert!(
!run.run_dir.join("final.patch").exists(),
"no-op git-backed run should not emit final.patch"
run_snapshot(&run.run_dir).final_patch.is_none(),
"no-op git-backed run should not persist final.patch"
);
}
}

2
lib/crates/fabro-retro/src/lib.rs
View file

@ -1,4 +1,2 @@
pub mod retro;
pub mod retro_agent;
pub use retro::RetroExt;

54
lib/crates/fabro-retro/src/retro.rs
View file

@ -1,11 +1,9 @@
use std::collections::HashMap;
use std::path::Path;
use fabro_types::RunId;
pub use fabro_types::retro::{
AggregateStats, FrictionKind, FrictionPoint, Learning, LearningCategory, OpenItem,
OpenItemKind, Retro, RetroNarrative, SmoothnessRating, StageRetro,
};
use std::collections::HashMap;
#[derive(Debug, Clone)]
pub struct CompletedStage {
@ -20,56 +18,6 @@ pub struct CompletedStage {
pub files_touched: Vec<String>,
}
pub trait RetroExt {
fn save(&self, run_dir: &Path) -> anyhow::Result<()>;
fn load(run_dir: &Path) -> anyhow::Result<Self>
where
Self: Sized;
}
impl RetroExt for Retro {
fn save(&self, run_dir: &Path) -> anyhow::Result<()> {
let json = serde_json::to_string_pretty(self)
.map_err(|e| anyhow::anyhow!("retro serialize failed: {e}"))?;
std::fs::write(run_dir.join("retro.json"), json)?;
Ok(())
}
fn load(run_dir: &Path) -> anyhow::Result<Self> {
let data = std::fs::read_to_string(run_dir.join("retro.json"))?;
serde_json::from_str(&data).map_err(|e| anyhow::anyhow!("retro deserialize failed: {e}"))
}
}
pub fn extract_stage_durations(run_dir: &Path) -> HashMap<String, u64> {
let mut durations = HashMap::new();
let jsonl_path = run_dir.join("progress.jsonl");
let Ok(data) = std::fs::read_to_string(&jsonl_path) else {
return durations;
};
for line in data.lines() {
let Ok(envelope) = serde_json::from_str::<serde_json::Value>(line) else {
continue;
};
if envelope.get("event").and_then(|v| v.as_str()) != Some("stage.completed") {
continue;
}
let Some(name) = envelope.get("node_id").and_then(|v| v.as_str()) else {
continue;
};
let Some(duration_ms) = envelope
.get("properties")
.and_then(serde_json::Value::as_object)
.and_then(|properties| properties.get("duration_ms"))
.and_then(serde_json::Value::as_u64)
else {
continue;
};
durations.insert(name.to_string(), duration_ms);
}
durations
}
pub fn derive_retro(
run_id: RunId,
workflow_name: &str,

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

@ -22,8 +22,8 @@ const RETRO_SYSTEM_PROMPT: &str = r"You are a workflow run retrospective analyst
You have access to the run's data files:
- `progress.jsonl` — the full event stream (stage starts/completions, agent tool calls, errors, retries)
- `checkpoint.json` — final execution state with node outcomes
- `run.json` — run record with config, graph, and metadata (if available)
- `start.json` — start record with start time and git info (if available)
- `run.json` — run record with config, graph, and metadata
- `start.json` — start record with start time and git info
## Your task
@ -379,7 +379,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_dir: &Path,
_run_dir: &Path,
target_dir: &str,
) -> anyhow::Result<()> {
// Create target directory
@ -388,37 +388,23 @@ async fn upload_data_files(
.await
.map_err(|e| anyhow::anyhow!("Failed to create retro data dir: {e}"))?;
// progress.jsonl — try store first, fall back to filesystem
let progress_content = if let Some(store) = run_store {
match store.list_events().await {
Ok(envelopes) => {
let lines: Vec<String> = envelopes
.into_iter()
.filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
.collect();
if lines.is_empty() {
None
} else {
Some(lines.join("\n") + "\n")
}
}
Err(e) => {
tracing::debug!(error = %e, "Could not read events from store, falling back to filesystem");
None
}
}
} else {
None
let Some(store) = run_store else {
anyhow::bail!("retro analysis now requires a run store");
};
let progress_content = if progress_content.is_some() {
progress_content
} else {
let source = run_dir.join("progress.jsonl");
if source.exists() {
Some(std::fs::read_to_string(&source)?)
} else {
None
let progress_content = match store.list_events().await {
Ok(envelopes) => {
let lines: Vec<String> = envelopes
.into_iter()
.filter_map(|env| serde_json::to_string(env.payload.as_value()).ok())
.collect();
if lines.is_empty() {
None
} else {
Some(lines.join("\n") + "\n")
}
}
Err(e) => return Err(anyhow::anyhow!("Failed to load events from store: {e}")),
};
if let Some(content) = progress_content {
sandbox
@ -427,80 +413,39 @@ async fn upload_data_files(
.map_err(|e| anyhow::anyhow!("Failed to upload progress.jsonl: {e}"))?;
}
// checkpoint.json — try store first, fall back to filesystem
let checkpoint_content = if let Some(store) = run_store {
match store.get_checkpoint().await {
Ok(Some(cp)) => serde_json::to_string_pretty(&cp).ok(),
Ok(None) => None,
Err(e) => {
tracing::debug!(error = %e, "Could not read checkpoint from store, falling back to filesystem");
None
}
}
} else {
None
};
upload_file_with_fallback(
sandbox,
run_dir,
target_dir,
"checkpoint.json",
checkpoint_content,
)
.await?;
let checkpoint_content = store
.get_checkpoint()
.await
.map_err(|e| anyhow::anyhow!("Failed to load checkpoint from store: {e}"))?
.map(|cp| serde_json::to_string_pretty(&cp))
.transpose()?;
upload_file(sandbox, target_dir, "checkpoint.json", checkpoint_content).await?;
// run.json — try store first, fall back to filesystem
let run_content = if let Some(store) = run_store {
match store.get_run().await {
Ok(Some(run)) => serde_json::to_string_pretty(&run).ok(),
Ok(None) => None,
Err(e) => {
tracing::debug!(error = %e, "Could not read run from store, falling back to filesystem");
None
}
}
} else {
None
};
upload_file_with_fallback(sandbox, run_dir, target_dir, "run.json", run_content).await?;
let run_content = store
.get_run()
.await
.map_err(|e| anyhow::anyhow!("Failed to load run metadata from store: {e}"))?
.map(|run| serde_json::to_string_pretty(&run))
.transpose()?;
upload_file(sandbox, target_dir, "run.json", run_content).await?;
// start.json — try store first, fall back to filesystem
let start_content = if let Some(store) = run_store {
match store.get_start().await {
Ok(Some(start)) => serde_json::to_string_pretty(&start).ok(),
Ok(None) => None,
Err(e) => {
tracing::debug!(error = %e, "Could not read start from store, falling back to filesystem");
None
}
}
} else {
None
};
upload_file_with_fallback(sandbox, run_dir, target_dir, "start.json", start_content).await?;
let start_content = store
.get_start()
.await
.map_err(|e| anyhow::anyhow!("Failed to load start metadata from store: {e}"))?
.map(|start| serde_json::to_string_pretty(&start))
.transpose()?;
upload_file(sandbox, target_dir, "start.json", start_content).await?;
Ok(())
}
/// Upload a single file to the sandbox. If `store_content` is `Some`, use it directly;
/// otherwise fall back to reading from `run_dir/filename` on the filesystem.
async fn upload_file_with_fallback(
async fn upload_file(
sandbox: &Arc<dyn Sandbox>,
run_dir: &Path,
target_dir: &str,
filename: &str,
store_content: Option<String>,
content: Option<String>,
) -> anyhow::Result<()> {
let content = if store_content.is_some() {
store_content
} else {
let source = run_dir.join(filename);
if source.exists() {
Some(std::fs::read_to_string(&source)?)
} else {
None
}
};
if let Some(content) = content {
sandbox
.write_file(&format!("{target_dir}/{filename}"), &content)

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

@ -20,7 +20,6 @@ use fabro_llm::types::{
ContentPart, FinishReason, Message as LlmMessage, Request as LlmRequest,
Response as LlmResponse, Role, StreamEvent, ToolChoice, ToolDefinition, Usage,
};
use fabro_retro::retro::{Retro, extract_stage_durations};
use fabro_store::{InMemoryStore, Store};
use fabro_types::RunId;
use fabro_util::redact::redact_jsonl_line;
@ -47,7 +46,6 @@ use crate::sessions as sessions_mod;
use crate::sessions::{SessionStore, new_session_store};
use crate::web_auth;
use fabro_interview::{Answer, Interviewer, QuestionType, WebInterviewer};
use fabro_retro::RetroExt;
use fabro_workflow::context::Context;
use fabro_workflow::event::{EventEmitter, RunEventEnvelope};
use fabro_workflow::operations::{self, CreateRunInput, WorkflowInput};
@ -771,7 +769,7 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
Ok(events) => fabro_workflow::extract_stage_durations_from_events(&events),
Err(err) => {
tracing::warn!(run_id = %run_id, error = %err, "Failed to load run events from store");
extract_stage_durations(&run_dir)
Default::default()
}
};
let mut agg = state
@ -1559,43 +1557,26 @@ async fn get_retro(
Ok(id) => id,
Err(response) => return response,
};
let run_dir = {
{
let runs = state.runs.lock().expect("runs lock poisoned");
match runs.get(&id) {
Some(managed_run) => managed_run.run_dir.clone(),
None => return ApiError::not_found("Run not found.").into_response(),
if !runs.contains_key(&id) {
return ApiError::not_found("Run not found.").into_response();
}
};
let Some(run_dir) = run_dir else {
return (StatusCode::OK, Json(serde_json::json!(null))).into_response();
};
}
match state.store.open_run_reader(&id).await {
Ok(Some(run_store)) => match run_store.get_retro().await {
Ok(Some(retro)) => (StatusCode::OK, Json(retro)).into_response(),
Ok(None) => match Retro::load(&run_dir) {
Ok(retro) => (StatusCode::OK, Json(retro)).into_response(),
Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
},
Ok(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");
match Retro::load(&run_dir) {
Ok(retro) => (StatusCode::OK, Json(retro)).into_response(),
Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
}
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
},
Ok(None) => match Retro::load(&run_dir) {
Ok(retro) => (StatusCode::OK, Json(retro)).into_response(),
Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
},
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");
match Retro::load(&run_dir) {
Ok(retro) => (StatusCode::OK, Json(retro)).into_response(),
Err(_) => (StatusCode::OK, Json(serde_json::json!(null))).into_response(),
}
(StatusCode::OK, Json(serde_json::json!(null))).into_response()
}
}
}
@ -1643,7 +1624,6 @@ mod tests {
AuthProvider, AuthSettings, GitAuthorSettings, GitProvider, GitSettings, WebSettings,
};
use fabro_types::fixtures;
use fabro_workflow::records::{RunRecord, RunRecordExt};
use tower::ServiceExt;
const MINIMAL_DOT: &str = r#"digraph Test {
@ -2507,13 +2487,22 @@ mod tests {
let body = body_json(response.into_body()).await;
let run_id = body["id"].as_str().unwrap().parse::<RunId>().unwrap();
let run_dir = {
let _run_dir = {
let runs = state.runs.lock().expect("runs lock poisoned");
runs.get(&run_id)
.and_then(|run| run.run_dir.clone())
.expect("run_dir should be recorded")
};
let run_record = RunRecord::load(&run_dir).unwrap();
let run_record = state
.store
.open_run_reader(&run_id)
.await
.unwrap()
.expect("run store should exist")
.get_run()
.await
.unwrap()
.expect("run record should exist");
let mut expected_settings = settings;
expected_settings.goal = Some("Test".to_string());

955
lib/crates/fabro-store/src/disk_projecting.rs
View file

@ -1,955 +0,0 @@
use std::fs::{self, OpenOptions};
use std::io::Write;
use std::path::{Path, PathBuf};
use std::pin::Pin;
use std::sync::Arc;
use async_trait::async_trait;
use bytes::Bytes;
use futures::Stream;
use tracing::warn;
use crate::{
EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result,
RunSnapshot, RunStore,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord,
SandboxRecord, StartRecord,
};
#[derive(Debug, Clone)]
pub struct ProjectionError {
pub path: PathBuf,
pub critical: bool,
pub error: String,
}
pub struct DiskProjectingRunStore {
inner: Arc<dyn RunStore>,
run_dir: PathBuf,
on_projection_error: Option<Arc<dyn Fn(ProjectionError) + Send + Sync>>,
}
impl DiskProjectingRunStore {
#[must_use]
pub fn new(inner: Arc<dyn RunStore>, run_dir: PathBuf) -> Self {
Self {
inner,
run_dir,
on_projection_error: None,
}
}
#[must_use]
pub fn on_projection_error(
mut self,
callback: Arc<dyn Fn(ProjectionError) + Send + Sync>,
) -> Self {
self.on_projection_error = Some(callback);
self
}
fn report_projection_error(&self, path: &Path, err: &std::io::Error, critical: bool) {
if critical {
warn!(
path = %path.display(),
error = %err,
"Critical disk projection failed"
);
} else {
warn!(path = %path.display(), error = %err, "Disk projection failed");
}
if let Some(ref callback) = self.on_projection_error {
callback(ProjectionError {
path: path.to_path_buf(),
critical,
error: err.to_string(),
});
}
}
fn write_json_critical<T: serde::Serialize>(&self, path: &Path, value: &T) {
if let Err(err) = write_json(path, value) {
self.report_projection_error(path, &err, true);
}
}
fn write_json_best_effort<T: serde::Serialize>(&self, path: &Path, value: &T) {
if let Err(err) = write_json(path, value) {
self.report_projection_error(path, &err, false);
}
}
fn write_text_best_effort(&self, path: &Path, value: &str) {
if let Err(err) = write_text(path, value) {
self.report_projection_error(path, &err, false);
}
}
fn append_jsonl_critical(&self, payload: &EventPayload) {
let progress_path = self.run_dir.join("progress.jsonl");
if let Err(err) = append_jsonl(&progress_path, payload) {
self.report_projection_error(&progress_path, &err, true);
}
let live_path = self.run_dir.join("live.json");
if let Err(err) = write_live_json(&live_path, payload) {
self.report_projection_error(&live_path, &err, true);
}
}
}
/// Map store node visits onto the legacy on-disk layout used by workflow logs.
///
/// The store key layout uses `nodes/{id}/visit-{N}/...`, but existing disk readers
/// expect first visits at `nodes/{id}/...` and later visits at
/// `nodes/{id}-visit_{N}/...`.
fn disk_node_dir(run_dir: &Path, node_id: &str, visit: u32) -> PathBuf {
if visit <= 1 {
run_dir.join("nodes").join(node_id)
} else {
run_dir
.join("nodes")
.join(format!("{node_id}-visit_{visit}"))
}
}
fn ensure_parent_dir(path: &Path) -> std::io::Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
Ok(())
}
fn write_json<T: serde::Serialize>(path: &Path, value: &T) -> std::io::Result<()> {
ensure_parent_dir(path)?;
let json = serde_json::to_string_pretty(value).map_err(std::io::Error::other)?;
fs::write(path, json)
}
fn write_text(path: &Path, value: &str) -> std::io::Result<()> {
ensure_parent_dir(path)?;
fs::write(path, value)
}
fn append_jsonl(path: &Path, payload: &EventPayload) -> std::io::Result<()> {
ensure_parent_dir(path)?;
let line = serde_json::to_string(payload.as_value()).map_err(std::io::Error::other)?;
let mut file = OpenOptions::new().create(true).append(true).open(path)?;
writeln!(file, "{line}")
}
fn write_live_json(path: &Path, payload: &EventPayload) -> std::io::Result<()> {
ensure_parent_dir(path)?;
let json = serde_json::to_string_pretty(payload.as_value()).map_err(std::io::Error::other)?;
fs::write(path, json)
}
#[async_trait]
impl RunStore for DiskProjectingRunStore {
async fn put_run(&self, record: &RunRecord) -> Result<()> {
self.inner.put_run(record).await?;
self.write_json_best_effort(&self.run_dir.join("run.json"), record);
Ok(())
}
async fn get_run(&self) -> Result<Option<RunRecord>> {
self.inner.get_run().await
}
async fn put_start(&self, record: &StartRecord) -> Result<()> {
self.inner.put_start(record).await?;
self.write_json_best_effort(&self.run_dir.join("start.json"), record);
Ok(())
}
async fn get_start(&self) -> Result<Option<StartRecord>> {
self.inner.get_start().await
}
async fn put_status(&self, record: &RunStatusRecord) -> Result<()> {
self.write_json_critical(&self.run_dir.join("status.json"), record);
self.inner.put_status(record).await
}
async fn get_status(&self) -> Result<Option<RunStatusRecord>> {
self.inner.get_status().await
}
async fn put_checkpoint(&self, record: &Checkpoint) -> Result<()> {
self.inner.put_checkpoint(record).await?;
self.write_json_best_effort(&self.run_dir.join("checkpoint.json"), record);
Ok(())
}
async fn get_checkpoint(&self) -> Result<Option<Checkpoint>> {
self.inner.get_checkpoint().await
}
async fn append_checkpoint(&self, record: &Checkpoint) -> Result<u32> {
self.inner.append_checkpoint(record).await
}
async fn list_checkpoints(&self) -> Result<Vec<(u32, Checkpoint)>> {
self.inner.list_checkpoints().await
}
async fn put_conclusion(&self, record: &Conclusion) -> Result<()> {
self.write_json_critical(&self.run_dir.join("conclusion.json"), record);
self.inner.put_conclusion(record).await
}
async fn get_conclusion(&self) -> Result<Option<Conclusion>> {
self.inner.get_conclusion().await
}
async fn put_retro(&self, retro: &Retro) -> Result<()> {
self.inner.put_retro(retro).await?;
self.write_json_best_effort(&self.run_dir.join("retro.json"), retro);
Ok(())
}
async fn get_retro(&self) -> Result<Option<Retro>> {
self.inner.get_retro().await
}
async fn put_graph(&self, dot_source: &str) -> Result<()> {
self.inner.put_graph(dot_source).await?;
self.write_text_best_effort(&self.run_dir.join("workflow.fabro"), dot_source);
Ok(())
}
async fn get_graph(&self) -> Result<Option<String>> {
self.inner.get_graph().await
}
async fn put_sandbox(&self, record: &SandboxRecord) -> Result<()> {
self.inner.put_sandbox(record).await?;
self.write_json_best_effort(&self.run_dir.join("sandbox.json"), record);
Ok(())
}
async fn get_sandbox(&self) -> Result<Option<SandboxRecord>> {
self.inner.get_sandbox().await
}
async fn put_node_prompt(&self, node: &NodeVisitRef<'_>, prompt: &str) -> Result<()> {
self.inner.put_node_prompt(node, prompt).await?;
self.write_text_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("prompt.md"),
prompt,
);
Ok(())
}
async fn put_node_response(&self, node: &NodeVisitRef<'_>, response: &str) -> Result<()> {
self.inner.put_node_response(node, response).await?;
self.write_text_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("response.md"),
response,
);
Ok(())
}
async fn put_node_status(
&self,
node: &NodeVisitRef<'_>,
status: &NodeStatusRecord,
) -> Result<()> {
self.inner.put_node_status(node, status).await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("status.json"),
status,
);
Ok(())
}
async fn put_node_outcome(
&self,
node: &NodeVisitRef<'_>,
outcome: &NodeOutcomeRecord,
) -> Result<()> {
self.inner.put_node_outcome(node, outcome).await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("outcome.json"),
outcome,
);
Ok(())
}
async fn put_node_provider_used(
&self,
node: &NodeVisitRef<'_>,
provider_used: &serde_json::Value,
) -> Result<()> {
self.inner
.put_node_provider_used(node, provider_used)
.await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("provider_used.json"),
provider_used,
);
Ok(())
}
async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> {
self.inner.put_node_diff(node, diff).await?;
self.write_text_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("diff.patch"),
diff,
);
Ok(())
}
async fn put_node_script_invocation(
&self,
node: &NodeVisitRef<'_>,
invocation: &serde_json::Value,
) -> Result<()> {
self.inner
.put_node_script_invocation(node, invocation)
.await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("script_invocation.json"),
invocation,
);
Ok(())
}
async fn put_node_script_timing(
&self,
node: &NodeVisitRef<'_>,
timing: &serde_json::Value,
) -> Result<()> {
self.inner.put_node_script_timing(node, timing).await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("script_timing.json"),
timing,
);
Ok(())
}
async fn put_node_parallel_results(
&self,
node: &NodeVisitRef<'_>,
results: &serde_json::Value,
) -> Result<()> {
self.inner.put_node_parallel_results(node, results).await?;
self.write_json_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("parallel_results.json"),
results,
);
Ok(())
}
async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.inner.put_node_stdout(node, log).await?;
self.write_text_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("stdout.log"),
log,
);
Ok(())
}
async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.inner.put_node_stderr(node, log).await?;
self.write_text_best_effort(
&disk_node_dir(&self.run_dir, node.node_id, node.visit).join("stderr.log"),
log,
);
Ok(())
}
async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot> {
self.inner.get_node(node).await
}
async fn list_node_visits(&self, node_id: &str) -> Result<Vec<u32>> {
self.inner.list_node_visits(node_id).await
}
async fn list_node_ids(&self) -> Result<Vec<String>> {
self.inner.list_node_ids().await
}
async fn put_final_patch(&self, patch: &str) -> Result<()> {
self.inner.put_final_patch(patch).await?;
self.write_text_best_effort(&self.run_dir.join("final.patch"), patch);
Ok(())
}
async fn get_final_patch(&self) -> Result<Option<String>> {
self.inner.get_final_patch().await
}
async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()> {
self.inner.put_pull_request(record).await?;
self.write_json_best_effort(&self.run_dir.join("pull_request.json"), record);
Ok(())
}
async fn get_pull_request(&self) -> Result<Option<PullRequestRecord>> {
self.inner.get_pull_request().await
}
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
self.append_jsonl_critical(payload);
self.inner.append_event(payload).await
}
async fn list_events(&self) -> Result<Vec<EventEnvelope>> {
self.inner.list_events().await
}
async fn list_events_from(&self, seq: u32) -> Result<Vec<EventEnvelope>> {
self.inner.list_events_from(seq).await
}
async fn watch_events_from(
&self,
seq: u32,
) -> Result<Pin<Box<dyn Stream<Item = Result<EventEnvelope>> + Send>>> {
self.inner.watch_events_from(seq).await
}
async fn put_retro_prompt(&self, text: &str) -> Result<()> {
self.inner.put_retro_prompt(text).await?;
self.write_text_best_effort(&self.run_dir.join("retro").join("prompt.md"), text);
Ok(())
}
async fn get_retro_prompt(&self) -> Result<Option<String>> {
self.inner.get_retro_prompt().await
}
async fn put_retro_response(&self, text: &str) -> Result<()> {
self.inner.put_retro_response(text).await?;
self.write_text_best_effort(&self.run_dir.join("retro").join("response.md"), text);
Ok(())
}
async fn get_retro_response(&self) -> Result<Option<String>> {
self.inner.get_retro_response().await
}
async fn put_artifact_value(&self, artifact_id: &str, value: &serde_json::Value) -> Result<()> {
self.inner.put_artifact_value(artifact_id, value).await
}
async fn get_artifact_value(&self, artifact_id: &str) -> Result<Option<serde_json::Value>> {
self.inner.get_artifact_value(artifact_id).await
}
async fn list_artifact_values(&self) -> Result<Vec<String>> {
self.inner.list_artifact_values().await
}
async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()> {
self.inner.put_asset(node, filename, data).await
}
async fn get_asset(&self, node: &NodeVisitRef<'_>, filename: &str) -> Result<Option<Bytes>> {
self.inner.get_asset(node, filename).await
}
async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result<Vec<String>> {
self.inner.list_assets(node).await
}
async fn list_all_assets(&self) -> Result<Vec<(String, u32, String)>> {
self.inner.list_all_assets().await
}
async fn get_snapshot(&self) -> Result<Option<RunSnapshot>> {
self.inner.get_snapshot().await
}
}
#[cfg(test)]
mod tests {
use std::collections::HashMap;
use std::path::PathBuf;
use chrono::{DateTime, Duration as ChronoDuration, Utc};
use tempfile::TempDir;
use super::*;
use crate::{InMemoryStore, Store};
use fabro_types::{
AggregateStats, AttrValue, FabroSettings, Graph, RunId, RunStatus, StageStatus,
StatusReason, fixtures,
};
fn dt(rfc3339: &str) -> DateTime<Utc> {
DateTime::parse_from_rfc3339(rfc3339)
.unwrap()
.with_timezone(&Utc)
}
fn test_run_id(label: &str) -> RunId {
match label {
"run-1" => fixtures::RUN_1,
_ => panic!("unknown test run id: {label}"),
}
}
fn sample_run_record(run_id: &str, created_at: DateTime<Utc>) -> RunRecord {
let mut graph = Graph::new("night-sky");
graph.attrs.insert(
"goal".to_string(),
AttrValue::String("map the constellations".to_string()),
);
RunRecord {
run_id: test_run_id(run_id),
created_at,
settings: FabroSettings::default(),
graph,
workflow_slug: Some("night-sky".to_string()),
working_directory: PathBuf::from("/tmp/night-sky"),
host_repo_path: Some("github.com/fabro-sh/fabro".to_string()),
base_branch: Some("main".to_string()),
labels: HashMap::from([("team".to_string(), "infra".to_string())]),
}
}
fn sample_start_record(run_id: &str, created_at: DateTime<Utc>) -> StartRecord {
StartRecord {
run_id: test_run_id(run_id),
start_time: created_at + ChronoDuration::seconds(5),
run_branch: Some("fabro/run/demo".to_string()),
base_sha: Some("abc123".to_string()),
}
}
fn sample_status(status: RunStatus, reason: Option<StatusReason>) -> RunStatusRecord {
RunStatusRecord {
status,
reason,
updated_at: dt("2026-03-27T12:05:00Z"),
}
}
fn sample_checkpoint() -> Checkpoint {
Checkpoint {
timestamp: dt("2026-03-27T12:10:00Z"),
current_node: "code".to_string(),
completed_nodes: vec!["plan".to_string()],
node_retries: HashMap::from([("code".to_string(), 1)]),
context_values: HashMap::from([(
"artifact".to_string(),
serde_json::json!({"kind": "summary"}),
)]),
node_outcomes: HashMap::new(),
next_node_id: Some("review".to_string()),
git_commit_sha: Some("def456".to_string()),
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("code".to_string(), 2)]),
}
}
fn sample_conclusion() -> Conclusion {
Conclusion {
timestamp: dt("2026-03-27T12:15:00Z"),
status: StageStatus::Success,
duration_ms: 3210,
failure_reason: None,
final_git_commit_sha: Some("feedbeef".to_string()),
stages: Vec::new(),
total_cost: Some(1.25),
total_retries: 2,
total_input_tokens: 10,
total_output_tokens: 20,
total_cache_read_tokens: 30,
total_cache_write_tokens: 40,
total_reasoning_tokens: 50,
has_pricing: true,
}
}
fn sample_retro(run_id: &str) -> Retro {
Retro {
run_id: test_run_id(run_id),
workflow_name: "night-sky".to_string(),
goal: "map the constellations".to_string(),
timestamp: dt("2026-03-27T12:20:00Z"),
smoothness: None,
stages: Vec::new(),
stats: AggregateStats {
total_duration_ms: 3210,
total_cost: Some(1.25),
total_retries: 2,
files_touched: vec!["src/lib.rs".to_string()],
stages_completed: 3,
stages_failed: 0,
},
intent: Some("ship the fix".to_string()),
outcome: Some("done".to_string()),
learnings: None,
friction_points: None,
open_items: None,
}
}
fn sample_sandbox() -> SandboxRecord {
SandboxRecord {
provider: "local".to_string(),
working_directory: "/tmp/night-sky".to_string(),
identifier: Some("sandbox-1".to_string()),
host_working_directory: Some("/tmp/night-sky".to_string()),
container_mount_point: None,
}
}
fn sample_node_status() -> NodeStatusRecord {
NodeStatusRecord {
status: StageStatus::PartialSuccess,
notes: Some("captured output".to_string()),
failure_reason: Some("minor lint".to_string()),
timestamp: dt("2026-03-27T12:12:00Z"),
}
}
fn event_payload(run_id: &str, ts: &str, event: &str) -> EventPayload {
EventPayload::new(
serde_json::json!({
"id": format!("evt-{run_id}-{event}"),
"ts": ts,
"run_id": test_run_id(run_id).to_string(),
"event": event,
}),
&test_run_id(run_id),
)
.unwrap()
}
async fn make_store(
run_dir: &Path,
created_at: DateTime<Utc>,
) -> (Arc<dyn RunStore>, DiskProjectingRunStore) {
let inner = InMemoryStore::default()
.create_run(
&test_run_id("run-1"),
created_at,
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
let projected = DiskProjectingRunStore::new(Arc::clone(&inner), run_dir.to_path_buf());
(inner, projected)
}
#[tokio::test]
async fn put_methods_project_expected_files() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let (_inner, store) = make_store(temp.path(), created_at).await;
let run = sample_run_record("run-1", created_at);
let start = sample_start_record("run-1", created_at);
let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
let checkpoint = sample_checkpoint();
let conclusion = sample_conclusion();
let retro = sample_retro("run-1");
let sandbox = sample_sandbox();
let node_status = sample_node_status();
store.put_run(&run).await.unwrap();
store.put_start(&start).await.unwrap();
store.put_status(&status).await.unwrap();
store.put_checkpoint(&checkpoint).await.unwrap();
store.put_conclusion(&conclusion).await.unwrap();
store.put_retro(&retro).await.unwrap();
store.put_graph("digraph night_sky {}").await.unwrap();
store.put_sandbox(&sandbox).await.unwrap();
let visit_one = NodeVisitRef {
node_id: "code",
visit: 1,
};
store
.put_node_response(&visit_one, "Applied the fix")
.await
.unwrap();
store
.put_node_status(&visit_one, &node_status)
.await
.unwrap();
store.put_node_stdout(&visit_one, "stdout").await.unwrap();
store.put_node_stderr(&visit_one, "stderr").await.unwrap();
let visit_two = NodeVisitRef {
node_id: "code",
visit: 2,
};
store
.put_node_prompt(&visit_two, "Plan the fix")
.await
.unwrap();
store.put_retro_prompt("How did it go?").await.unwrap();
store.put_retro_response("Smooth enough").await.unwrap();
assert_eq!(
serde_json::to_value(
serde_json::from_str::<RunRecord>(
&fs::read_to_string(temp.path().join("run.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(run).unwrap()
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<StartRecord>(
&fs::read_to_string(temp.path().join("start.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(start).unwrap()
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<RunStatusRecord>(
&fs::read_to_string(temp.path().join("status.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(status).unwrap()
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<Checkpoint>(
&fs::read_to_string(temp.path().join("checkpoint.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(checkpoint).unwrap()
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<Conclusion>(
&fs::read_to_string(temp.path().join("conclusion.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(conclusion).unwrap()
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<Retro>(
&fs::read_to_string(temp.path().join("retro.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(retro).unwrap()
);
assert_eq!(
fs::read_to_string(temp.path().join("workflow.fabro")).unwrap(),
"digraph night_sky {}"
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<SandboxRecord>(
&fs::read_to_string(temp.path().join("sandbox.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(sandbox).unwrap()
);
assert_eq!(
fs::read_to_string(temp.path().join("nodes/code/response.md")).unwrap(),
"Applied the fix"
);
assert_eq!(
serde_json::to_value(
serde_json::from_str::<NodeStatusRecord>(
&fs::read_to_string(temp.path().join("nodes/code/status.json")).unwrap()
)
.unwrap()
)
.unwrap(),
serde_json::to_value(node_status).unwrap()
);
assert_eq!(
fs::read_to_string(temp.path().join("nodes/code/stdout.log")).unwrap(),
"stdout"
);
assert_eq!(
fs::read_to_string(temp.path().join("nodes/code/stderr.log")).unwrap(),
"stderr"
);
assert_eq!(
fs::read_to_string(temp.path().join("nodes/code-visit_2/prompt.md")).unwrap(),
"Plan the fix"
);
assert_eq!(
fs::read_to_string(temp.path().join("retro/prompt.md")).unwrap(),
"How did it go?"
);
assert_eq!(
fs::read_to_string(temp.path().join("retro/response.md")).unwrap(),
"Smooth enough"
);
}
#[tokio::test]
async fn critical_projections_write_files_in_isolation() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let (_inner, store) = make_store(temp.path(), created_at).await;
let status = sample_status(RunStatus::Failed, Some(StatusReason::WorkflowError));
let conclusion = sample_conclusion();
store.put_status(&status).await.unwrap();
store.put_conclusion(&conclusion).await.unwrap();
assert!(temp.path().join("status.json").exists());
assert!(temp.path().join("conclusion.json").exists());
}
#[tokio::test]
async fn append_event_projects_progress_and_live_files() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let (_inner, store) = make_store(temp.path(), created_at).await;
let first = event_payload("run-1", "2026-03-27T12:00:00Z", "Started");
let second = event_payload("run-1", "2026-03-27T12:00:01Z", "Completed");
store.append_event(&first).await.unwrap();
store.append_event(&second).await.unwrap();
let progress = fs::read_to_string(temp.path().join("progress.jsonl")).unwrap();
let lines: Vec<&str> = progress.lines().collect();
assert_eq!(lines.len(), 2);
let first_value: serde_json::Value = serde_json::from_str(lines[0]).unwrap();
let second_value: serde_json::Value = serde_json::from_str(lines[1]).unwrap();
assert_eq!(first_value, first.as_value().clone());
assert_eq!(second_value, second.as_value().clone());
let live: serde_json::Value =
serde_json::from_str(&fs::read_to_string(temp.path().join("live.json")).unwrap())
.unwrap();
assert_eq!(live, second.as_value().clone());
let events = store.list_events().await.unwrap();
assert_eq!(events.len(), 2);
assert_eq!(events[0].payload, first);
assert_eq!(events[1].payload, second);
}
#[tokio::test]
async fn get_methods_read_from_inner_store_not_disk() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let (inner, store) = make_store(temp.path(), created_at).await;
let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
inner.put_status(&status).await.unwrap();
fs::write(
temp.path().join("status.json"),
serde_json::to_string_pretty(&sample_status(
RunStatus::Succeeded,
Some(StatusReason::Completed),
))
.unwrap(),
)
.unwrap();
assert_eq!(
serde_json::to_value(store.get_status().await.unwrap().unwrap()).unwrap(),
serde_json::to_value(status).unwrap()
);
}
#[tokio::test]
async fn disk_failures_do_not_block_store_writes() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let (inner, store) = make_store(temp.path(), created_at).await;
let mut permissions = fs::metadata(temp.path()).unwrap().permissions();
permissions.set_readonly(true);
fs::set_permissions(temp.path(), permissions).unwrap();
let status = sample_status(RunStatus::Running, Some(StatusReason::SandboxInitializing));
let node = NodeVisitRef {
node_id: "code",
visit: 1,
};
store.put_status(&status).await.unwrap();
store.put_node_prompt(&node, "Plan the fix").await.unwrap();
let stored_status = inner.get_status().await.unwrap();
let stored_node = inner.get_node(&node).await.unwrap();
assert_eq!(
serde_json::to_value(stored_status.unwrap()).unwrap(),
serde_json::to_value(status).unwrap()
);
assert_eq!(stored_node.prompt.as_deref(), Some("Plan the fix"));
}
#[tokio::test]
async fn projection_error_callback_runs_on_disk_failure() {
let temp = TempDir::new().unwrap();
let created_at = dt("2026-03-27T12:00:00Z");
let inner = InMemoryStore::default()
.create_run(
&test_run_id("run-1"),
created_at,
Some(temp.path().to_string_lossy().as_ref()),
)
.await
.unwrap();
let seen = Arc::new(std::sync::Mutex::new(Vec::<ProjectionError>::new()));
let seen_clone = Arc::clone(&seen);
let store = DiskProjectingRunStore::new(inner, temp.path().to_path_buf())
.on_projection_error(Arc::new(move |error| {
seen_clone.lock().unwrap().push(error);
}));
let mut permissions = fs::metadata(temp.path()).unwrap().permissions();
permissions.set_readonly(true);
fs::set_permissions(temp.path(), permissions).unwrap();
store
.put_status(&sample_status(
RunStatus::Running,
Some(StatusReason::SandboxInitializing),
))
.await
.unwrap();
let seen = seen.lock().unwrap();
assert_eq!(seen.len(), 1);
assert!(seen[0].critical);
assert_eq!(seen[0].path, temp.path().join("status.json"));
assert!(!seen[0].error.is_empty());
}
#[test]
fn disk_node_dir_matches_legacy_layout() {
let run_dir = Path::new("/tmp/fabro-run");
assert_eq!(
disk_node_dir(run_dir, "build", 1),
run_dir.join("nodes/build")
);
assert_eq!(
disk_node_dir(run_dir, "build", 2),
run_dir.join("nodes/build-visit_2")
);
}
}

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

@ -6,7 +6,6 @@ use bytes::Bytes;
use chrono::{DateTime, Utc};
use futures::Stream;
mod disk_projecting;
mod error;
mod keys;
mod memory;
@ -14,7 +13,6 @@ mod runtime;
mod slate;
mod types;
pub use disk_projecting::{DiskProjectingRunStore, ProjectionError};
pub use error::{Result, StoreError};
pub use memory::InMemoryStore;
pub use runtime::RuntimeState;

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

@ -331,9 +331,6 @@ impl Handler for AgentHandler {
match result {
Ok(CodergenResult::Full(outcome)) => {
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
let status_json = serde_json::to_string_pretty(&outcome)
.unwrap_or_else(|_| "{}".to_string());
fs::write(stage_dir.join("status.json"), &status_json).await?;
return Ok(outcome);
}
Ok(CodergenResult::Text {
@ -418,10 +415,6 @@ impl Handler for AgentHandler {
outcome.usage = stage_usage;
outcome.files_touched = backend_files_touched;
let status_json =
serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
fs::write(stage_dir.join("status.json"), &status_json).await?;
Ok(outcome)
}
}

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

@ -116,9 +116,6 @@ impl Handler for PromptHandler {
match result {
Ok(CodergenResult::Full(outcome)) => {
sync_provider_used_to_store(&stage_dir, &node_ref, services).await?;
let status_json = serde_json::to_string_pretty(&outcome)
.unwrap_or_else(|_| "{}".to_string());
fs::write(stage_dir.join("status.json"), &status_json).await?;
return Ok(outcome);
}
Ok(CodergenResult::Text {
@ -191,10 +188,6 @@ impl Handler for PromptHandler {
outcome.usage = stage_usage;
outcome.files_touched = backend_files_touched;
let status_json =
serde_json::to_string_pretty(&outcome).unwrap_or_else(|_| "{}".to_string());
fs::write(stage_dir.join("status.json"), &status_json).await?;
Ok(outcome)
}
}

6
lib/crates/fabro-workflow/src/lifecycle/git.rs
View file

@ -18,7 +18,6 @@ use crate::git::scan_node_files_from_store;
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::outcome::{Outcome, StageStatus, StageUsage};
use crate::run_dir::node_dir;
use crate::run_options::RunOptions;
use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
@ -289,8 +288,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
.and_then(|g| g.base_sha.clone())
})
.unwrap_or_else(|| sha.clone());
let diff_dest = node_dir(&self.run_dir, node_id, visit).join("diff.patch");
match git_diff(&*self.sandbox, &prev).await {
Ok(patch) if !patch.is_empty() => {
let node_ref = fabro_store::NodeVisitRef {
@ -309,7 +306,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
"failed to persist node diff for '{node_id}': {err}"
)));
}
let _ = std::fs::write(&diff_dest, &patch);
git_result.diff = Some(patch);
}
Ok(_) => {}
@ -352,7 +348,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
.as_ref()
.and_then(|g| g.base_sha.clone())
{
let diff_dest = self.run_dir.join("final.patch");
match git_diff(&*self.sandbox, &base_sha).await {
Ok(patch) if !patch.is_empty() => {
if let Err(err) = self.run_store.put_final_patch(&patch).await {
@ -365,7 +360,6 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
});
return;
}
let _ = std::fs::write(&diff_dest, patch);
}
Ok(_) => {}
Err(err) => {

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

@ -1,15 +1,13 @@
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use chrono::{Local, Utc};
use fabro_config::{FabroSettings, FabroSettingsExt};
use fabro_graphviz::graph::{AttrValue, Graph};
use fabro_model::{Catalog, Provider};
use fabro_sandbox::SandboxProvider;
use fabro_store::{DiskProjectingRunStore, RunStore, Store};
use fabro_store::Store;
use fabro_types::RunId;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use crate::error::FabroError;
use crate::pipeline::types::PersistOptions;
@ -141,7 +139,7 @@ async fn persist_created_run(
) -> Result<(), FabroError> {
let record = persisted.run_record();
let run_dir_string = persisted.run_dir().to_string_lossy().to_string();
let inner_run_store = match store
let run_store = match store
.create_run(&record.run_id, record.created_at, Some(&run_dir_string))
.await
{
@ -152,10 +150,6 @@ async fn persist_created_run(
.map_err(|open_err| FabroError::engine(open_err.to_string()))?
.ok_or_else(|| FabroError::engine(err.to_string()))?,
};
let run_store: Arc<dyn RunStore> = Arc::new(DiskProjectingRunStore::new(
inner_run_store,
persisted.run_dir().to_path_buf(),
));
run_store.put_run(record).await.map_err(store_error)?;
if !workflow_source.is_empty() {

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

@ -68,12 +68,7 @@ fn cleanup_resume_artifacts(run_dir: &Path) {
let _ = std::fs::remove_file(path);
}
for name in [
"conclusion.json",
"pull_request.json",
"detached_failure.json",
"progress.jsonl",
] {
for name in ["detached_failure.json"] {
let _ = std::fs::remove_file(run_dir.join(name));
}
}

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

@ -11,16 +11,15 @@ use fabro_config::{project as project_config, run as run_config, sandbox as sand
use fabro_interview::{AutoApproveInterviewer, Interviewer};
use fabro_model::{Catalog, FallbackTarget, Provider};
use fabro_sandbox::{SandboxProvider, SandboxSpec};
use fabro_store::{DiskProjectingRunStore, ProjectionError, RunStore};
use fabro_store::RunStore;
use fabro_types::RunId;
use serde::Serialize;
use crate::context::Context;
use crate::error::FabroError;
use crate::event::{
EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, append_progress_event,
append_progress_event_with_line, canonicalize_event, event_payload_from_redacted_json,
redacted_event_json,
EventEmitter, RunNoticeLevel, StoreProgressLogger, WorkflowRunEvent, canonicalize_event,
event_payload_from_redacted_json, redacted_event_json,
};
use crate::git::MetadataStore;
use crate::handler::HandlerRegistry;
@ -116,38 +115,10 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, F
pub(super) async fn execute_persisted_run(
run_dir: &Path,
checkpoint: Option<Checkpoint>,
mut services: StartServices,
services: StartServices,
) -> Result<Started, FabroError> {
let cancel_token = services.cancel_token.clone();
let run_id = services.run_id;
let inner_store = Arc::clone(&services.run_store);
let projection_run_dir = run_dir.to_path_buf();
services.run_store = Arc::new(
DiskProjectingRunStore::new(inner_store, run_dir.to_path_buf()).on_projection_error(
Arc::new(move |projection_error: ProjectionError| {
// Write directly to progress.jsonl/live.json so projection failures do not
// recurse back through the decorated store.append_event() path.
let envelope = canonicalize_event(
&run_id,
&WorkflowRunEvent::RunNotice {
level: RunNoticeLevel::Warn,
code: "disk_projection_failed".to_string(),
message: format!(
"{}disk projection failed for {}: {}",
if projection_error.critical {
"critical "
} else {
""
},
projection_error.path.display(),
projection_error.error
),
},
);
let _ = append_progress_event(&projection_run_dir, &envelope);
}),
),
);
let run_store = Arc::clone(&services.run_store);
if let Err(err) = run_store
.put_status(&run_status::RunStatusRecord::new(
@ -208,7 +179,7 @@ pub(super) async fn execute_persisted_run(
bootstrap_guard.defuse();
let mut completion_guard =
DetachedRunCompletionGuard::arm(run_dir, run_id, Arc::clone(&run_store), cancel_token);
DetachedRunCompletionGuard::arm(run_id, Arc::clone(&run_store), cancel_token);
let run_start = Instant::now();
let started = Box::pin(session.run(persisted, checkpoint)).await;
@ -228,7 +199,7 @@ pub(super) async fn execute_persisted_run(
async fn persist_terminal_engine_failure(
run_store: &dyn RunStore,
run_dir: &Path,
_run_dir: &Path,
error: &FabroError,
duration: Duration,
) {
@ -237,7 +208,6 @@ async fn persist_terminal_engine_failure(
classify_engine_result(&engine_result);
let conclusion = build_conclusion_from_store(
run_store,
run_dir,
final_status,
failure_reason,
u64::try_from(duration.as_millis()).unwrap(),
@ -655,7 +625,6 @@ const POSTRUN_ABORTED_MESSAGE: &str = "Run aborted before post-run finalization
const POSTRUN_CANCELLED_MESSAGE: &str = "Run cancelled before post-run finalization completed.";
struct DetachedRunCompletionGuard {
run_dir: PathBuf,
run_store: Arc<dyn RunStore>,
run_id: RunId,
cancel_token: Option<Arc<AtomicBool>>,
@ -664,13 +633,11 @@ struct DetachedRunCompletionGuard {
impl DetachedRunCompletionGuard {
fn arm(
run_dir: &Path,
run_id: RunId,
run_store: Arc<dyn RunStore>,
cancel_token: Option<Arc<AtomicBool>>,
) -> Self {
Self {
run_dir: run_dir.to_path_buf(),
run_store,
run_id,
cancel_token,
@ -727,11 +694,6 @@ impl Drop for DetachedRunCompletionGuard {
};
if line.is_empty() {
None
} else if let Err(err) =
append_progress_event_with_line(&self.run_dir, &envelope, &line)
{
tracing::warn!(error = %err, "Failed to append post-run abort event");
None
} else {
Some((self.run_id, line))
}
@ -836,8 +798,6 @@ async fn persist_detached_failure(
};
let envelope = canonicalize_event(&run_id, &event);
let line = redacted_event_json(&envelope).map_err(|err| FabroError::Io(err.to_string()))?;
append_progress_event_with_line(run_dir, &envelope, &line)
.map_err(|err| FabroError::Io(err.to_string()))?;
match event_payload_from_redacted_json(&line, &run_id) {
Ok(payload) => {
if let Err(err) = run_store.append_event(&payload).await {

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

@ -26,7 +26,7 @@ use crate::handler::{Handler as HandlerTrait, HandlerRegistry};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::pipeline::initialize;
use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec};
use crate::records::{Checkpoint, CheckpointExt, RunRecord, StartRecordExt};
use crate::records::RunRecord;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::test_support::run_graph;
@ -165,6 +165,62 @@ async fn test_run_store(_run_dir: &Path, run_id: &RunId) -> Arc<dyn fabro_store:
.unwrap()
}
async fn execute_test_run(run_dir: &Path, graph: Graph, run_id: &str) -> Executed {
execute_test_run_with_options(test_run_options(run_dir, run_id), graph, None).await
}
async fn execute_test_run_with_options(
run_options: RunOptions,
graph: Graph,
registry_override: Option<Arc<HandlerRegistry>>,
) -> Executed {
let run_id_value = run_options.run_id;
let git_options = run_options.git.clone();
let initialized = initialize(
persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value),
InitOptions {
run_id: run_id_value,
run_store: test_run_store(&run_options.run_dir, &run_id_value).await,
dry_run: false,
emitter: test_emitter_arc("test-run"),
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
llm: LlmSpec {
model: "test-model".to_string(),
provider: fabro_llm::Provider::Anthropic,
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
dry_run: true,
},
interviewer: Arc::new(AutoApproveInterviewer),
lifecycle: LifecycleOptions {
setup_commands: vec![],
setup_command_timeout_ms: 1_000,
devcontainer_phases: vec![],
},
run_options,
hooks: HookConfig { hooks: vec![] },
sandbox_env: SandboxEnvSpec {
devcontainer_env: HashMap::new(),
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
devcontainer: None,
git: git_options,
worktree_mode: None,
registry_override,
checkpoint: None,
seed_context: None,
},
)
.await
.unwrap();
execute(initialized).await
}
#[tokio::test]
async fn execute_runs_start_to_exit_and_returns_final_context() {
let temp = tempfile::tempdir().unwrap();
@ -447,16 +503,8 @@ async fn execute_runs_simple_workflow() {
#[tokio::test]
async fn execute_saves_checkpoint() {
let dir = tempfile::tempdir().unwrap();
run_graph(
make_registry(),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&test_run_options(dir.path(), "test-run"),
)
.await
.unwrap();
assert!(dir.path().join("checkpoint.json").exists());
let executed = execute_test_run(dir.path(), simple_graph(), "test-run").await;
assert!(executed.run_store.get_checkpoint().await.unwrap().is_some());
}
#[tokio::test]
@ -499,17 +547,8 @@ async fn execute_error_when_no_start_node() {
#[tokio::test]
async fn execute_mirrors_graph_goal_to_context() {
let dir = tempfile::tempdir().unwrap();
run_graph(
make_registry(),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&test_run_options(dir.path(), "test-run"),
)
.await
.unwrap();
let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap();
let executed = execute_test_run(dir.path(), simple_graph(), "test-run").await;
let cp = executed.run_store.get_checkpoint().await.unwrap().unwrap();
assert_eq!(
cp.context_values.get(context::keys::GRAPH_GOAL),
Some(&serde_json::json!("Run tests"))
@ -548,17 +587,8 @@ async fn execute_conditional_routing_uses_unconditional_success_path() {
g.edges.push(Edge::new("path_a", "exit"));
g.edges.push(Edge::new("path_b", "exit"));
run_graph(
make_registry(),
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
)
.await
.unwrap();
let cp = Checkpoint::load(&dir.path().join("checkpoint.json")).unwrap();
let executed = execute_test_run(dir.path(), g, "test-run").await;
let cp = executed.run_store.get_checkpoint().await.unwrap().unwrap();
assert!(cp.completed_nodes.contains(&"path_b".to_string()));
assert!(!cp.completed_nodes.contains(&"path_a".to_string()));
}
@ -573,17 +603,8 @@ async fn execute_writes_start_json_and_node_status() {
meta_branch: None,
});
run_graph(
make_registry(),
test_emitter_arc("test-run"),
local_env(),
&simple_graph(),
&run_options,
)
.await
.unwrap();
let start = crate::records::StartRecord::load(dir.path()).unwrap();
let executed = execute_test_run_with_options(run_options, simple_graph(), None).await;
let start = executed.run_store.get_start().await.unwrap().unwrap();
assert_eq!(start.run_id, test_run_id("test-run"));
assert_eq!(
start.run_branch.as_deref(),
@ -591,12 +612,19 @@ async fn execute_writes_start_json_and_node_status() {
);
assert_eq!(start.base_sha.as_deref(), Some("abc123"));
let status_path = dir.path().join("nodes").join("start").join("status.json");
assert!(status_path.exists());
let node = executed
.run_store
.get_node(&fabro_store::NodeVisitRef {
node_id: "start",
visit: 1,
})
.await
.unwrap();
assert_eq!(node.status.unwrap().status, StageStatus::Success);
}
#[tokio::test]
async fn timeout_causes_fail_status_json() {
async fn timeout_causes_fail_status_record() {
let dir = tempfile::tempdir().unwrap();
let mut g = Graph::new("timeout_test");
@ -635,20 +663,23 @@ async fn timeout_causes_fail_status_json() {
let mut registry = make_registry();
registry.register("slow", Box::new(SlowHandler { sleep_ms: 500 }));
run_graph(
registry,
test_emitter_arc("test-run"),
local_env(),
&g,
&test_run_options(dir.path(), "test-run"),
let executed = execute_test_run_with_options(
test_run_options(dir.path(), "test-run"),
g,
Some(Arc::new(registry)),
)
.await
.unwrap();
let status_path = dir.path().join("nodes").join("work").join("status.json");
let status: serde_json::Value =
serde_json::from_str(&std::fs::read_to_string(&status_path).unwrap()).unwrap();
assert_eq!(status["status"], "fail");
.await;
let status = executed
.run_store
.get_node(&fabro_store::NodeVisitRef {
node_id: "work",
visit: 1,
})
.await
.unwrap()
.status
.unwrap();
assert_eq!(status.status, StageStatus::Fail);
}
#[tokio::test]

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

@ -5,12 +5,11 @@ use crate::error::FabroError;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::git::{MetadataStore, scan_node_files_from_store};
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::records::{Checkpoint, CheckpointExt, Conclusion, StageSummary};
use crate::records::{Checkpoint, Conclusion, StageSummary};
use crate::run_options::RunOptions;
use crate::run_status::{RunStatus, StatusReason};
use crate::sandbox_git::git_push_host;
use fabro_hooks::{HookContext, HookEvent, HookRunner};
use fabro_retro::retro::extract_stage_durations;
use fabro_store::RunStore;
use super::types::{Concluded, FinalizeOptions, Retroed};
@ -63,105 +62,19 @@ pub fn classify_engine_result(
}
}
pub fn build_conclusion(
run_dir: &Path,
status: StageStatus,
failure_reason: Option<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> Conclusion {
let checkpoint = Checkpoint::load(&run_dir.join("checkpoint.json")).ok();
let stage_durations = extract_stage_durations(run_dir);
let mut total_input_tokens: i64 = 0;
let mut total_output_tokens: i64 = 0;
let mut total_cache_read_tokens: i64 = 0;
let mut total_cache_write_tokens: i64 = 0;
let mut total_reasoning_tokens: i64 = 0;
let mut has_pricing = false;
let (stages, total_cost, total_retries) = if let Some(ref cp) = checkpoint {
let mut stages = Vec::new();
let mut cost_sum: Option<f64> = None;
let mut retries_sum: u32 = 0;
for node_id in &cp.completed_nodes {
let outcome = cp.node_outcomes.get(node_id);
let retries = cp
.node_retries
.get(node_id)
.copied()
.unwrap_or(1)
.saturating_sub(1);
retries_sum += retries;
let cost = outcome.and_then(|o| o.usage.as_ref()).and_then(|u| u.cost);
if let Some(c) = cost {
*cost_sum.get_or_insert(0.0) += c;
has_pricing = true;
}
if let Some(usage) = outcome.and_then(|o| o.usage.as_ref()) {
total_input_tokens += usage.input_tokens;
total_output_tokens += usage.output_tokens;
total_cache_read_tokens += usage.cache_read_tokens.unwrap_or(0);
total_cache_write_tokens += usage.cache_write_tokens.unwrap_or(0);
total_reasoning_tokens += usage.reasoning_tokens.unwrap_or(0);
}
stages.push(StageSummary {
stage_id: node_id.clone(),
stage_label: node_id.clone(),
duration_ms: stage_durations.get(node_id).copied().unwrap_or(0),
cost,
retries,
});
}
(stages, cost_sum, retries_sum)
} else {
(vec![], None, 0)
};
Conclusion {
timestamp: chrono::Utc::now(),
status,
duration_ms: run_duration_ms,
failure_reason,
final_git_commit_sha,
stages,
total_cost,
total_retries,
total_input_tokens,
total_output_tokens,
total_cache_read_tokens,
total_cache_write_tokens,
total_reasoning_tokens,
has_pricing,
}
}
pub(crate) async fn build_conclusion_from_store(
run_store: &dyn RunStore,
run_dir: &Path,
status: StageStatus,
failure_reason: Option<String>,
run_duration_ms: u64,
final_git_commit_sha: Option<String>,
) -> Conclusion {
let checkpoint = match run_store.get_checkpoint().await {
Ok(checkpoint) => checkpoint,
Err(err) => {
tracing::warn!(error = %err, "Failed to load checkpoint from store while building conclusion");
Checkpoint::load(&run_dir.join("checkpoint.json")).ok()
}
};
let stage_durations = match run_store.list_events().await {
Ok(events) => crate::extract_stage_durations_from_events(&events),
Err(err) => {
tracing::warn!(error = %err, "Failed to load events from store while building conclusion");
extract_stage_durations(run_dir)
}
};
let checkpoint = run_store.get_checkpoint().await.ok().flatten();
let stage_durations = run_store
.list_events()
.await
.map(|events| crate::extract_stage_durations_from_events(&events))
.unwrap_or_default();
build_conclusion_from_parts(
checkpoint.as_ref(),
@ -360,7 +273,6 @@ pub async fn finalize(
classify_engine_result(&outcome);
let conclusion = build_conclusion_from_store(
options.run_store.as_ref(),
&options.run_dir,
final_status,
failure_reason,
duration_ms,
@ -479,9 +391,7 @@ mod tests {
)
.await
.unwrap();
let run_store: Arc<dyn fabro_store::RunStore> = Arc::new(
fabro_store::DiskProjectingRunStore::new(inner_store, run_dir.clone()),
);
let run_store: Arc<dyn fabro_store::RunStore> = inner_store;
let retroed = Retroed {
graph: Graph::new("test"),
outcome: Ok(Outcome::success()),

12
lib/crates/fabro-workflow/src/pipeline/initialize.rs
View file

@ -771,10 +771,7 @@ mod tests {
)
.await
.unwrap();
Arc::new(fabro_store::DiskProjectingRunStore::new(
inner,
run_dir.clone(),
))
inner
},
dry_run: false,
emitter,
@ -849,10 +846,7 @@ mod tests {
)
.await
.unwrap();
Arc::new(fabro_store::DiskProjectingRunStore::new(
inner,
run_dir.clone(),
))
inner
},
dry_run: false,
emitter,
@ -891,7 +885,7 @@ mod tests {
.await
.unwrap();
assert!(run_dir.join("sandbox.json").exists());
assert!(initialized.run_store.get_sandbox().await.unwrap().is_some());
assert_eq!(initialized.run_options.run_dir, run_dir);
}
}

3
lib/crates/fabro-workflow/src/pipeline/mod.rs
View file

@ -13,8 +13,7 @@ pub use execute::execute;
pub use fabro_types::PullRequestRecord;
pub(crate) use finalize::build_conclusion_from_store;
pub use finalize::{
build_conclusion, classify_engine_result, finalize, persist_terminal_outcome,
write_finalize_commit,
classify_engine_result, finalize, persist_terminal_outcome, write_finalize_commit,
};
pub use initialize::initialize;
pub use parse::parse;

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

@ -3,16 +3,10 @@ use std::path::Path;
use fabro_store::RunStore;
use crate::error::FabroError;
use crate::records::{RunRecord, RunRecordExt};
use super::types::{PersistOptions, Persisted, Validated};
const GRAPH_FILE_NAME: &str = "workflow.fabro";
const LEGACY_GRAPH_FILE_NAME: &str = "graph.fabro";
/// PERSIST phase: create run directory, write workflow.fabro and run.json to disk.
///
/// Overwrites `run_record.graph` with the validated graph before saving.
/// PERSIST phase: create the run directory and return durable metadata for store persistence.
pub(crate) fn persist(
validated: Validated,
mut options: PersistOptions,
@ -21,10 +15,6 @@ pub(crate) fn persist(
options.run_record.graph = graph.clone();
std::fs::create_dir_all(&options.run_dir)?;
if !source.is_empty() {
std::fs::write(options.run_dir.join(GRAPH_FILE_NAME), &source)?;
}
options.run_record.save(&options.run_dir)?;
Ok(Persisted::new(
graph,
@ -35,34 +25,6 @@ pub(crate) fn persist(
))
}
/// Load a previously persisted run from disk.
///
/// `run.json` is authoritative for graph + config; `workflow.fabro` provides the
/// original DOT source string when present.
pub(crate) fn load(run_dir: &Path) -> Result<Persisted, FabroError> {
let run_record = RunRecord::load(run_dir)?;
let graph = run_record.graph.clone();
let source = match std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)) {
Ok(source) => source,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
match std::fs::read_to_string(run_dir.join(LEGACY_GRAPH_FILE_NAME)) {
Ok(source) => source,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => String::new(),
Err(err) => return Err(err.into()),
}
}
Err(err) => return Err(err.into()),
};
Ok(Persisted::new(
graph,
source,
Vec::new(),
run_dir.to_path_buf(),
run_record,
))
}
pub(crate) async fn load_from_store(
run_store: &dyn RunStore,
run_dir: &Path,
@ -96,6 +58,7 @@ mod tests {
use chrono::Utc;
use fabro_config::FabroSettings;
use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node};
use fabro_store::{InMemoryStore, Store};
use fabro_types::fixtures;
use super::*;
@ -166,59 +129,33 @@ mod tests {
}
}
async fn seeded_store(
run_dir: &Path,
record: &RunRecord,
source: Option<&str>,
) -> std::sync::Arc<dyn RunStore> {
let store = InMemoryStore::default();
let run_store = store
.create_run(
&record.run_id,
record.created_at,
Some(run_dir.to_string_lossy().as_ref()),
)
.await
.unwrap();
run_store.put_run(record).await.unwrap();
if let Some(source) = source {
run_store.put_graph(source).await.unwrap();
}
run_store
}
#[test]
fn persist_creates_run_dir_and_writes_graph_and_record() {
fn persist_creates_run_dir_without_writing_legacy_files() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let persisted = persist(
Validated::new(graph.clone(), source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
assert!(run_dir.is_dir());
assert_eq!(
std::fs::read_to_string(run_dir.join(GRAPH_FILE_NAME)).unwrap(),
source
);
assert!(run_dir.join(RunRecord::file_name()).exists());
assert_eq!(persisted.run_dir(), run_dir.as_path());
assert_eq!(
serde_json::to_value(persisted.run_record().graph.clone()).unwrap(),
serde_json::to_value(graph).unwrap()
);
}
#[test]
fn persist_skips_graph_file_when_source_is_empty() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, _source) = graph_and_source();
persist(
Validated::new(graph, String::new(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
assert!(!run_dir.join(GRAPH_FILE_NAME).exists());
assert!(run_dir.join(RunRecord::file_name()).exists());
}
#[test]
fn persist_overwrites_run_record_graph_with_validated_graph() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
persist(
Validated::new(graph.clone(), source, vec![]),
PersistOptions {
run_dir: run_dir.clone(),
@ -227,17 +164,41 @@ mod tests {
)
.unwrap();
let saved = RunRecord::load(&run_dir).unwrap();
assert_eq!(saved.graph.name, graph.name);
assert!(saved.graph.nodes.contains_key("exit"));
assert!(run_dir.is_dir());
assert!(!run_dir.join("workflow.fabro").exists());
assert!(!run_dir.join("run.json").exists());
assert_eq!(persisted.run_dir(), run_dir.as_path());
assert_eq!(
serde_json::to_value(saved.graph).unwrap(),
serde_json::to_value(persisted.run_record().graph.clone()).unwrap(),
serde_json::to_value(graph).unwrap()
);
}
#[test]
fn persist_roundtrips_full_run_record_fields_through_load() {
fn persist_overwrites_run_record_graph_with_validated_graph() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let persisted = persist(
Validated::new(graph.clone(), source, vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
assert_eq!(persisted.run_record().graph.name, graph.name);
assert!(persisted.run_record().graph.nodes.contains_key("exit"));
assert_eq!(
serde_json::to_value(persisted.run_record().graph.clone()).unwrap(),
serde_json::to_value(graph).unwrap()
);
}
#[tokio::test]
async fn load_from_store_roundtrips_full_run_record_fields() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
@ -245,7 +206,7 @@ mod tests {
expected.graph = graph.clone();
persist(
Validated::new(graph, source, vec![]),
Validated::new(graph, source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: expected.clone(),
@ -253,12 +214,14 @@ mod tests {
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
let run_store = seeded_store(&run_dir, &expected, Some(&source)).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
assert_eq!(
serde_json::to_value(loaded.run_record()).unwrap(),
serde_json::to_value(expected).unwrap()
);
assert_eq!(loaded.source(), source);
assert!(loaded.diagnostics().is_empty());
}
@ -281,92 +244,38 @@ mod tests {
assert!(matches!(err, FabroError::Io(_)));
}
#[test]
fn load_roundtrips_persisted_workflow() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
let mut expected = sample_record(different_graph());
expected.graph = graph.clone();
let persisted = persist(
Validated::new(graph, source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: expected.clone(),
},
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(loaded.source(), source);
assert_eq!(loaded.run_dir(), run_dir.as_path());
assert_eq!(
serde_json::to_value(loaded.run_record()).unwrap(),
serde_json::to_value(expected).unwrap()
);
assert_eq!(
serde_json::to_value(loaded.graph()).unwrap(),
serde_json::to_value(persisted.graph()).unwrap()
);
}
#[test]
fn load_uses_empty_source_when_graph_file_is_missing() {
#[tokio::test]
async fn load_from_store_uses_empty_source_when_graph_missing() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, _source) = graph_and_source();
let mut record = sample_record(different_graph());
record.graph = graph;
record.save(&run_dir).unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
let run_store = seeded_store(&run_dir, &record, None).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
assert!(loaded.source().is_empty());
}
#[test]
fn load_reads_graph_from_run_json_and_source_from_graph_file() {
#[tokio::test]
async fn load_from_store_reads_graph_from_run_record_and_source_from_store() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let (graph, _) = graph_and_source();
let (graph, source) = graph_and_source();
let mut record = sample_record(different_graph());
record.graph = graph.clone();
record.save(&run_dir).unwrap();
std::fs::write(
run_dir.join(GRAPH_FILE_NAME),
"digraph mismatch { a -> b; }",
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
let run_store = seeded_store(&run_dir, &record, Some(&source)).await;
let loaded = load_from_store(run_store.as_ref(), &run_dir).await.unwrap();
assert_eq!(
serde_json::to_value(loaded.graph()).unwrap(),
serde_json::to_value(graph).unwrap()
);
assert_eq!(loaded.source(), "digraph mismatch { a -> b; }");
}
#[test]
fn load_reads_graph_source_from_graph_file() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
let (graph, source) = graph_and_source();
persist(
Validated::new(graph, source.clone(), vec![]),
PersistOptions {
run_dir: run_dir.clone(),
run_record: sample_record(different_graph()),
},
)
.unwrap();
let loaded = Persisted::load(&run_dir).unwrap();
assert_eq!(loaded.source(), source);
}
}

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

@ -8,16 +8,13 @@ use tracing::{debug, info};
use fabro_github::{self as github_app, GitHubAppCredentials, ssh_url_to_https};
use fabro_graphviz::parser;
use fabro_llm::generate::{GenerateParams, generate};
use fabro_retro::RetroExt;
use fabro_util::text::strip_goal_decoration;
use super::types::{Concluded, Finalized, PullRequestOptions};
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::outcome::{StageStatus, format_cost as outcome_format_cost};
use crate::records::{Conclusion, ConclusionExt, RunRecord, RunRecordExt};
use crate::records::{Conclusion, RunRecord, RunRecordExt};
use fabro_retro::retro::Retro;
use tokio::fs::read_to_string;
use super::types::{Concluded, Finalized, PullRequestOptions};
/// Derive a PR title from the workflow goal.
///
@ -199,8 +196,7 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) {
}
}
/// Read the workflow graph source from `run_dir/workflow.fabro`.
/// Falls back to `graph.fabro` / `graph.dot` for older runs.
#[cfg(test)]
fn read_dot_source(run_dir: &Path) -> Option<String> {
let workflow_fabro_path = run_dir.join("workflow.fabro");
if let Ok(content) = std::fs::read_to_string(&workflow_fabro_path) {
@ -307,9 +303,10 @@ async fn load_pull_request_diff(run_store: Option<&dyn RunStore>, run_dir: &Path
.ok()
.flatten()
.unwrap_or_default(),
None => read_to_string(run_dir.join("final.patch"))
.await
.unwrap_or_default(),
None => {
let _ = run_dir;
String::new()
}
}
}
@ -336,7 +333,7 @@ pub async fn build_pr_body(
})
.ok()
.flatten(),
None => Conclusion::load(&run_dir.join("conclusion.json")).ok(),
None => None,
}
} else {
None
@ -351,7 +348,7 @@ pub async fn build_pr_body(
})
.ok()
.flatten(),
None => Retro::load(run_dir).ok(),
None => None,
};
let run_record = match run_store {
Some(run_store) => run_store
@ -362,7 +359,7 @@ pub async fn build_pr_body(
})
.ok()
.flatten(),
None => RunRecord::load(run_dir).ok(),
None => None,
};
let dot_source = match run_store {
Some(run_store) => run_store
@ -373,7 +370,7 @@ pub async fn build_pr_body(
})
.ok()
.flatten(),
None => read_dot_source(run_dir),
None => None,
};
// Build LLM prompt

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

@ -1,7 +1,7 @@
use std::sync::Arc;
use fabro_agent::SessionEvent;
use fabro_retro::retro::{Retro, derive_retro, extract_stage_durations};
use fabro_retro::retro::{Retro, derive_retro};
use fabro_retro::retro_agent::{dry_run_narrative, run_retro_agent};
use super::types::{Executed, RetroOptions, Retroed};
@ -36,8 +36,8 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
let stage_durations = match options.run_store.list_events().await {
Ok(events) => crate::extract_stage_durations_from_events(&events),
Err(err) => {
tracing::warn!(error = %err, "Could not load events from store, falling back to disk");
extract_stage_durations(&options.run_dir)
tracing::warn!(error = %err, "Could not load events from store, skipping stage durations");
Default::default()
}
};
let mut retro = derive_retro(
@ -185,7 +185,7 @@ mod tests {
fixtures::RUN_1
}
fn write_checkpoint(run_dir: &std::path::Path) -> Checkpoint {
fn build_checkpoint() -> Checkpoint {
let context = Context::new();
context.set("response.work", serde_json::json!("done"));
let mut outcomes = HashMap::new();
@ -201,7 +201,6 @@ mod tests {
HashMap::new(),
HashMap::new(),
);
checkpoint.save(&run_dir.join("checkpoint.json")).unwrap();
checkpoint
}
@ -217,9 +216,7 @@ mod tests {
)
.await
.unwrap();
let run_store: Arc<dyn fabro_store::RunStore> = Arc::new(
fabro_store::DiskProjectingRunStore::new(inner, run_dir.to_path_buf()),
);
let run_store: Arc<dyn fabro_store::RunStore> = inner;
run_store.put_checkpoint(checkpoint).await.unwrap();
run_store
}
@ -245,7 +242,8 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let checkpoint = write_checkpoint(&run_dir);
let checkpoint = build_checkpoint();
let run_store = test_run_store(&run_dir, &checkpoint).await;
let emitter = Arc::new(EventEmitter::default());
let sandbox: Arc<dyn fabro_agent::Sandbox> = Arc::new(fabro_agent::LocalSandbox::new(
@ -255,7 +253,7 @@ mod tests {
graph: Graph::new("test"),
outcome: Ok(crate::outcome::Outcome::success()),
run_options: test_run_options(&run_dir),
run_store: test_run_store(&run_dir, &checkpoint).await,
run_store: Arc::clone(&run_store),
hook_runner: None,
emitter: Arc::clone(&emitter),
sandbox: Arc::clone(&sandbox),
@ -270,7 +268,7 @@ mod tests {
executed,
&RetroOptions {
run_id: test_run_id(),
run_store: test_run_store(&run_dir, &checkpoint).await,
run_store,
workflow_name: "test".to_string(),
goal: "Ship it".to_string(),
run_dir: run_dir.clone(),
@ -286,7 +284,7 @@ mod tests {
)
.await;
assert!(run_dir.join("retro.json").exists());
assert!(retroed.run_store.get_retro().await.unwrap().is_some());
assert!(retroed.retro.is_some());
}
@ -295,7 +293,7 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let checkpoint = write_checkpoint(&run_dir);
let checkpoint = build_checkpoint();
let emitter = Arc::new(EventEmitter::default());
let seen = Arc::new(Mutex::new(Vec::new()));

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

@ -194,11 +194,6 @@ impl Persisted {
)
}
/// Load a previously persisted run from disk.
pub fn load(run_dir: &Path) -> Result<Self, FabroError> {
super::persist::load(run_dir)
}
pub async fn load_from_store(
run_store: &dyn RunStore,
run_dir: &Path,

28
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -7,10 +7,8 @@ use fabro_store::{ListRunsQuery, Store};
use fabro_types::RunId;
use serde::Serialize;
use crate::records::{
Conclusion, ConclusionExt, RunRecord, RunRecordExt, StartRecord, StartRecordExt,
};
use crate::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt, StatusReason};
use crate::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt};
use crate::run_status::{RunStatus, StatusReason};
#[derive(Debug, Clone, Serialize)]
pub struct RunInfo {
@ -263,27 +261,7 @@ impl StatusInfo {
}
fn read_status(run_dir: &Path) -> StatusInfo {
if let Ok(record) = RunStatusRecord::load(&run_dir.join("status.json")) {
if record.status.is_terminal() {
if let Ok(conclusion) = Conclusion::load(&run_dir.join("conclusion.json")) {
return StatusInfo {
status: record.status,
reason: record.reason,
end_time: Some(conclusion.timestamp),
duration_ms: Some(conclusion.duration_ms),
total_cost: conclusion.total_cost,
};
}
}
return StatusInfo {
status: record.status,
reason: record.reason,
end_time: None,
duration_ms: None,
total_cost: None,
};
}
let _ = run_dir;
StatusInfo::simple(RunStatus::Dead)
}

7
lib/crates/fabro-workflow/src/test_support.rs
View file

@ -5,7 +5,7 @@ use std::sync::Arc;
use chrono::Utc;
use fabro_agent::Sandbox;
use fabro_graphviz::graph::Graph as GvGraph;
use fabro_store::{DiskProjectingRunStore, InMemoryStore, RunStore, Store};
use fabro_store::{InMemoryStore, Store};
use fabro_types::run::RunRecord;
use crate::error::Result;
@ -48,10 +48,7 @@ async fn initialized(
)
.await
.expect("failed to create in-memory run store");
let run_store = Arc::new(DiskProjectingRunStore::new(
inner_store,
run_options.run_dir.clone(),
));
let run_store = inner_store;
run_store
.put_run(&RunRecord {
run_id: run_options.run_id,