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

Expand store-backed run metadata and detached startup

Browse source 
Add the missing run-store records for node metadata, final patches, and pull
request state, and extend the store snapshots/backends to round-trip them.

Also cut the detached startup path over to explicit run IDs and store-backed
status loading so start and detached execution no longer require run.json for
bootstrap.
This commit is contained in:
Bryan Helmkamp 2026-04-01 20:44:08 -04:00
parent 96359f95cb
commit c83cf3a577
18 changed files with 755 additions and 62 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
lib/crates/fabro-cli/src/args.rs
View file

@ -646,9 +646,12 @@ pub(crate) enum RunCommands {
/// Run ID prefix or workflow name
run: String,
},
/// Internal: run the engine process (reads run.json from run dir)
/// Internal: run the engine process
#[command(name = "__detached", hide = true)]
Detached {
/// Run ID
#[arg(long)]
run_id: fabro_types::RunId,
/// Run directory
#[arg(long)]
run_dir: PathBuf,

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

@ -1,4 +1,5 @@
use anyhow::Result;
use fabro_config::FabroSettingsExt;
use fabro_util::terminal::Styles;
use crate::args::{GlobalArgs, RunArgs};
@ -21,7 +22,8 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<(
#[cfg(not(feature = "sleep_inhibitor"))]
let _ = prevent_idle_sleep;
let child = super::start::start_run(&run_dir, false)?;
let child =
super::start::start_run(&run_dir, &run_id, &cli_settings.storage_dir(), false).await?;
if args.detach {
if globals.json {

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

@ -1,26 +1,43 @@
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::Result;
use anyhow::{Result, anyhow};
use fabro_config::FabroSettingsExt;
use fabro_interview::FileInterviewer;
use fabro_store::RuntimeState;
use fabro_store::{RuntimeState, Store};
use fabro_types::RunId;
use fabro_workflow::event::EventEmitter;
use fabro_workflow::operations::{
StartServices, open_or_hydrate_run, resume as resume_run, start as start_run,
};
use fabro_workflow::records::{RunRecord, RunRecordExt};
use fabro_workflow::operations::{StartServices, resume as resume_run, start as start_run};
use crate::shared;
use crate::store;
pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bool) -> Result<()> {
use crate::user_config::load_user_settings;
pub(crate) async fn execute(
run_id: RunId,
run_dir: PathBuf,
storage_dir: Option<PathBuf>,
launcher_path: PathBuf,
resume: bool,
) -> Result<()> {
let _ = fabro_proc::title_init();
let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| {
super::launcher::remove_launcher_record(&path);
});
let run_record = RunRecord::load(&run_dir)?;
let storage_dir = match storage_dir {
Some(storage_dir) => storage_dir,
None => load_user_settings()?.storage_dir(),
};
let store = store::build_store(&storage_dir)?;
let run_store = store
.open_run(&run_id)
.await?
.ok_or_else(|| anyhow!("Run {run_id} not found in store"))?;
let run_record = run_store
.get_run()
.await?
.ok_or_else(|| anyhow!("Run {run_id} has no run record in store"))?;
let on_node: fabro_workflow::OnNodeCallback = Some({
let run_id = run_record.run_id.to_string();
let short_id = super::short_run_id(&run_id).to_string();
@ -29,8 +46,6 @@ pub(crate) async fn execute(run_dir: PathBuf, launcher_path: PathBuf, resume: bo
fabro_proc::title_set(&format!("fabro: {short_id} {node_id}"));
}) as Arc<dyn Fn(&str) + Send + Sync>
});
let store = store::build_store(&run_record.settings.storage_dir())?;
let run_store = open_or_hydrate_run(store.as_ref(), &run_dir).await?;
let github_app = shared::github::build_github_app_credentials(run_record.settings.app_id())?;
let runtime_state = RuntimeState::new(&run_dir);

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

@ -51,7 +51,11 @@ pub(crate) fn remove_launcher_record(path: &Path) {
pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord> {
let run_record = RunRecord::load(run_dir).ok()?;
let path = launcher_record_path(&run_record.settings.storage_dir(), &run_record.run_id);
active_launcher_record(&run_record.settings.storage_dir(), &run_record.run_id)
}
pub(crate) fn active_launcher_record(storage_dir: &Path, run_id: &RunId) -> Option<LauncherRecord> {
let path = launcher_record_path(storage_dir, run_id);
let launcher = read_launcher_record(&path)?;
if launcher_record_is_running(&launcher) {
Some(launcher)
@ -141,6 +145,7 @@ mod tests {
.unwrap();
assert!(active_launcher_record_for_run(&run_dir).is_none());
assert!(active_launcher_record(&storage_dir, &fixtures::RUN_1).is_none());
assert!(!launcher_path.exists());
}

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

@ -60,7 +60,13 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
let base = runs_base(&cli_settings.storage_dir());
let store = store::build_store(&cli_settings.storage_dir())?;
let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?;
let child = start::start_run(&run_info.path, false)?;
let child = start::start_run(
&run_info.path,
&run_info.run_id,
&cli_settings.storage_dir(),
false,
)
.await?;
if globals.json {
print_json_pretty(&serde_json::json!({ "run_id": run_info.run_id }))?;
} else {
@ -89,10 +95,20 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
Ok(())
}
RunCommands::Detached {
run_id,
run_dir,
launcher_path,
resume,
} => detached::execute(run_dir, launcher_path, resume).await,
} => {
detached::execute(
run_id,
run_dir,
globals.storage_dir.clone(),
launcher_path,
resume,
)
.await
}
RunCommands::Diff(args) => diff::run(args, globals).await,
RunCommands::Logs(args) => {
let styles = Styles::detect_stdout();

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

@ -1,7 +1,6 @@
use anyhow::bail;
use fabro_config::FabroSettingsExt;
use fabro_util::terminal::Styles;
use fabro_workflow::records::{RunRecord, RunRecordExt};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use crate::args::{GlobalArgs, ResumeArgs};
@ -29,13 +28,14 @@ pub(crate) async fn resume_command(
if !run_dir.join("run.json").exists() {
bail!("run directory exists but has no run.json — cannot resume");
}
let run_id = RunRecord::load(&run_dir)?.run_id;
let run_id = run.run_id;
if launcher_pid_alive(&run_dir) {
bail!("an engine process is still running for this run — cannot resume");
}
let child = super::start::start_run(&run_dir, true)?;
let child =
super::start::start_run(&run_dir, &run_id, &cli_settings.storage_dir(), true).await?;
if args.detach {
if globals.json {

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

@ -2,30 +2,29 @@ use std::path::Path;
use anyhow::{Result, anyhow, bail};
use chrono::Utc;
use fabro_config::FabroSettingsExt;
use fabro_workflow::records::{RunRecord, RunRecordExt};
use fabro_workflow::run_status::{RunStatus, RunStatusRecord, RunStatusRecordExt};
use fabro_types::RunId;
use fabro_workflow::run_status::RunStatus;
use super::launcher::{
LauncherRecord, active_launcher_record_for_run, launcher_log_path, launcher_record_path,
LauncherRecord, active_launcher_record, launcher_log_path, launcher_record_path,
remove_launcher_record, write_launcher_record,
};
use crate::store;
/// Spawn a detached engine process for the given run directory.
/// Spawn a detached engine process for the given run.
///
/// The engine process reads `run.json` from the run directory and executes the
/// workflow. Returns the child process handle (use `.id()` for the PID).
pub(crate) fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Child> {
/// Returns the child process handle (use `.id()` for the PID).
pub(crate) async fn start_run(
run_dir: &Path,
run_id: &RunId,
storage_dir: &Path,
resume: bool,
) -> Result<std::process::Child> {
if !resume {
ensure_startable_run(run_dir)?;
ensure_startable_run(storage_dir, run_id).await?;
}
let record = RunRecord::load(run_dir)
.map_err(|e| anyhow!("Cannot start run: failed to load run.json: {e}"))?;
let storage_dir = record.settings.storage_dir();
let launcher_path = launcher_record_path(&storage_dir, &record.run_id);
let log_path = launcher_log_path(&storage_dir, &record.run_id);
let launcher_path = launcher_record_path(storage_dir, run_id);
let log_path = launcher_log_path(storage_dir, run_id);
if let Some(parent) = log_path.parent() {
std::fs::create_dir_all(parent)?;
@ -36,8 +35,12 @@ pub(crate) fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Ch
let exe = std::env::current_exe()?;
let mut cmd = std::process::Command::new(&exe);
cmd.args(["__detached", "--run-dir"])
cmd.args(["__detached", "--run-id"])
.arg(run_id.to_string())
.args(["--run-dir"])
.arg(run_dir)
.args(["--storage-dir"])
.arg(storage_dir)
.args(["--launcher-path"])
.arg(&launcher_path);
if resume {
@ -56,7 +59,7 @@ pub(crate) fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Ch
if let Err(err) = write_launcher_record(
&launcher_path,
&LauncherRecord {
run_id: record.run_id,
run_id: *run_id,
run_dir: run_dir.to_path_buf(),
pid: child.id(),
resume,
@ -75,13 +78,15 @@ pub(crate) fn start_run(run_dir: &Path, resume: bool) -> Result<std::process::Ch
Ok(child)
}
fn ensure_startable_run(run_dir: &Path) -> Result<()> {
if active_launcher_record_for_run(run_dir).is_some() {
async fn ensure_startable_run(storage_dir: &Path, run_id: &RunId) -> Result<()> {
if active_launcher_record(storage_dir, run_id).is_some() {
bail!("an engine process is still running for this run — cannot start");
}
let status_path = run_dir.join("status.json");
if let Ok(record) = RunStatusRecord::load(&status_path) {
let run_store = store::open_run_reader(storage_dir, run_id)
.await?
.ok_or_else(|| anyhow!("Cannot start run: run {run_id} not found in store"))?;
if let Some(record) = run_store.get_status().await? {
if !matches!(record.status, RunStatus::Submitted | RunStatus::Starting) {
bail!(
"cannot start run: status is {:?}, expected submitted",

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

@ -27,7 +27,6 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
info!(run_id = %run_info.run_id, "Waiting for run to complete");
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?;
let status_path = run_info.path.join("status.json");
let deadline = args
.timeout
@ -37,6 +36,8 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
let final_status = loop {
let load_file_status = || RunStatusRecord::load(&status_path).ok().map(|r| r.status);
let run_store =
store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?;
let status = match run_store.as_ref() {
Some(run_store) => match run_store.get_status().await {
Ok(Some(record)) => Some(record.status),
@ -72,6 +73,7 @@ pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs)
};
let conclusion_path = run_info.path.join("conclusion.json");
let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run_info.run_id).await?;
let conclusion = match run_store.as_ref() {
Some(run_store) => run_store
.get_conclusion()

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

@ -11,18 +11,19 @@ fn help() {
success: true
exit_code: 0
----- stdout -----
Internal: run the engine process (reads run.json from run dir)
Internal: run the engine process
Usage: fabro __detached [OPTIONS] --run-dir <RUN_DIR> --launcher-path <LAUNCHER_PATH>
Usage: fabro __detached [OPTIONS] --run-id <RUN_ID> --run-dir <RUN_DIR> --launcher-path <LAUNCHER_PATH>
Options:
--json Output as JSON [env: FABRO_JSON=]
--run-dir <RUN_DIR> Run directory
--run-id <RUN_ID> Run ID
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--run-dir <RUN_DIR> Run directory
--launcher-path <LAUNCHER_PATH> Launcher metadata path
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--resume Resume from checkpoint instead of fresh start
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--resume Resume from checkpoint instead of fresh start
--verbose Enable verbose output [env: FABRO_VERBOSE=]
--storage-dir <STORAGE_DIR> Storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
-h, --help Print help
@ -74,6 +75,8 @@ digraph CachedGraph {
.command()
.args([
"__detached",
"--run-id",
run_id,
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
@ -156,6 +159,8 @@ digraph GitHubApp {
cmd.env("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%");
cmd.args([
"__detached",
"--run-id",
run_id,
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
@ -171,6 +176,68 @@ digraph GitHubApp {
");
}
#[test]
fn detached_runs_without_run_json_when_run_id_is_explicit() {
let context = test_context!();
let run_id = "01ARZ3NDEKTSV4RRFFQ69G5FAJ";
let workflow_path = context.temp_dir.join("workflow.fabro");
context.write_temp(
"workflow.fabro",
"\
digraph DetachedStoreOnly {
start [shape=Mdiamond, label=\"Start\"]
exit [shape=Msquare, label=\"Exit\"]
start -> exit
}
",
);
context
.command()
.args([
"create",
"--dry-run",
"--auto-approve",
"--run-id",
run_id,
workflow_path.to_str().unwrap(),
])
.assert()
.success();
let run_dir = context.find_run_dir(run_id);
std::fs::remove_file(run_dir.join("run.json")).unwrap();
context
.command()
.args([
"__detached",
"--run-id",
run_id,
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
launcher_path(&context, run_id).to_str().unwrap(),
])
.timeout(std::time::Duration::from_secs(15))
.assert()
.success();
let conclusion = read_json(run_dir.join("conclusion.json"));
fabro_json_snapshot!(
context,
serde_json::json!({
"status": conclusion["status"],
}),
@r#"
{
"status": "success"
}
"#
);
}
#[test]
fn detached_resume_rejects_completed_run_without_mutating_it() {
let context = test_context!();
@ -235,6 +302,8 @@ digraph Test {
let mut cmd = context.command();
cmd.args([
"__detached",
"--run-id",
&run_id,
"--run-dir",
&run_dir,
"--launcher-path",

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

@ -78,6 +78,50 @@ fn start_by_run_id_starts_created_run() {
);
}
#[test]
fn start_by_run_id_starts_created_run_without_run_json_or_status_json() {
let context = test_context!();
let run_id = "01ARZ3NDEKTSV4RRFFQ69G5FAH";
context
.command()
.args([
"create",
"--dry-run",
"--auto-approve",
"--run-id",
run_id,
example_fixture("simple.fabro").to_str().unwrap(),
])
.assert()
.success();
let run_dir = context.find_run_dir(run_id);
std::fs::remove_file(run_dir.join("run.json")).unwrap();
std::fs::remove_file(run_dir.join("status.json")).unwrap();
context.command().args(["start", run_id]).assert().success();
context
.command()
.args(["wait", run_id])
.timeout(std::time::Duration::from_secs(10))
.assert()
.success();
let conclusion = read_json(run_dir.join("conclusion.json"));
fabro_json_snapshot!(
context,
serde_json::json!({
"conclusion_status": conclusion["status"],
}),
@r#"
{
"conclusion_status": "success"
}
"#
);
}
#[test]
fn start_by_workflow_name_prefers_newly_created_submitted_run() {
let context = test_context!();

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

@ -10,11 +10,12 @@ use futures::Stream;
use tracing::warn;
use crate::{
EventEnvelope, EventPayload, NodeSnapshot, NodeVisitRef, Result, RunSnapshot, RunStore,
EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef, Result,
RunSnapshot, RunStore,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, Retro, RunRecord, RunStatusRecord, SandboxRecord,
StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunRecord, RunStatusRecord,
SandboxRecord, StartRecord,
};
#[derive(Debug, Clone)]
@ -265,6 +266,84 @@ impl RunStore for DiskProjectingRunStore {
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(
@ -291,6 +370,30 @@ impl RunStore for DiskProjectingRunStore {
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

54
lib/crates/fabro-store/src/keys.rs
View file

@ -9,6 +9,8 @@ pub(crate) const CONCLUSION_KEY: &str = "conclusion.json";
pub(crate) const RETRO_KEY: &str = "retro.json";
pub(crate) const GRAPH_KEY: &str = "graph.fabro";
pub(crate) const SANDBOX_KEY: &str = "sandbox.json";
pub(crate) const FINAL_PATCH_KEY: &str = "final.patch";
pub(crate) const PULL_REQUEST_KEY: &str = "pull_request.json";
pub(crate) const RETRO_PROMPT_KEY: &str = "retro/prompt.md";
pub(crate) const RETRO_RESPONSE_KEY: &str = "retro/response.md";
pub(crate) const EVENTS_PREFIX: &str = "events/";
@ -52,6 +54,14 @@ pub(crate) fn sandbox() -> &'static str {
SANDBOX_KEY
}
pub(crate) fn final_patch() -> &'static str {
FINAL_PATCH_KEY
}
pub(crate) fn pull_request() -> &'static str {
PULL_REQUEST_KEY
}
pub(crate) fn node_visit_prefix(node: &NodeVisitRef<'_>) -> String {
format!("nodes/{}/visit-{}", node.node_id, node.visit)
}
@ -68,6 +78,30 @@ pub(crate) fn node_status(node: &NodeVisitRef<'_>) -> String {
format!("{}/status.json", node_visit_prefix(node))
}
pub(crate) fn node_outcome(node: &NodeVisitRef<'_>) -> String {
format!("{}/outcome.json", node_visit_prefix(node))
}
pub(crate) fn node_provider_used(node: &NodeVisitRef<'_>) -> String {
format!("{}/provider_used.json", node_visit_prefix(node))
}
pub(crate) fn node_diff(node: &NodeVisitRef<'_>) -> String {
format!("{}/diff.patch", node_visit_prefix(node))
}
pub(crate) fn node_script_invocation(node: &NodeVisitRef<'_>) -> String {
format!("{}/script_invocation.json", node_visit_prefix(node))
}
pub(crate) fn node_script_timing(node: &NodeVisitRef<'_>) -> String {
format!("{}/script_timing.json", node_visit_prefix(node))
}
pub(crate) fn node_parallel_results(node: &NodeVisitRef<'_>) -> String {
format!("{}/parallel_results.json", node_visit_prefix(node))
}
pub(crate) fn node_stdout(node: &NodeVisitRef<'_>) -> String {
format!("{}/stdout.log", node_visit_prefix(node))
}
@ -149,6 +183,8 @@ mod tests {
assert_eq!(init(), "_init.json");
assert_eq!(run(), "run.json");
assert_eq!(graph(), "graph.fabro");
assert_eq!(final_patch(), "final.patch");
assert_eq!(pull_request(), "pull_request.json");
assert_eq!(retro_prompt(), "retro/prompt.md");
assert_eq!(retro_response(), "retro/response.md");
}
@ -163,6 +199,24 @@ mod tests {
assert_eq!(node_prompt(&node), "nodes/plan/visit-3/prompt.md");
assert_eq!(node_response(&node), "nodes/plan/visit-3/response.md");
assert_eq!(node_status(&node), "nodes/plan/visit-3/status.json");
assert_eq!(node_outcome(&node), "nodes/plan/visit-3/outcome.json");
assert_eq!(
node_provider_used(&node),
"nodes/plan/visit-3/provider_used.json"
);
assert_eq!(node_diff(&node), "nodes/plan/visit-3/diff.patch");
assert_eq!(
node_script_invocation(&node),
"nodes/plan/visit-3/script_invocation.json"
);
assert_eq!(
node_script_timing(&node),
"nodes/plan/visit-3/script_timing.json"
);
assert_eq!(
node_parallel_results(&node),
"nodes/plan/visit-3/parallel_results.json"
);
assert_eq!(node_stdout(&node), "nodes/plan/visit-3/stdout.log");
assert_eq!(node_stderr(&node), "nodes/plan/visit-3/stderr.log");
}

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

@ -24,10 +24,12 @@ pub use types::{
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, Retro, RunId, RunRecord, RunStatusRecord,
SandboxRecord, StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, Outcome, PullRequestRecord, Retro, RunId, RunRecord,
RunStatusRecord, SandboxRecord, StageUsage, StartRecord,
};
pub type NodeOutcomeRecord = Outcome<Option<StageUsage>>;
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct ListRunsQuery {
pub start: Option<DateTime<Utc>>,
@ -83,11 +85,44 @@ pub trait RunStore: Send + Sync {
node: &NodeVisitRef<'_>,
status: &NodeStatusRecord,
) -> Result<()>;
async fn put_node_outcome(
&self,
node: &NodeVisitRef<'_>,
outcome: &NodeOutcomeRecord,
) -> Result<()>;
async fn put_node_provider_used(
&self,
node: &NodeVisitRef<'_>,
provider_used: &serde_json::Value,
) -> Result<()>;
async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()>;
async fn put_node_script_invocation(
&self,
node: &NodeVisitRef<'_>,
invocation: &serde_json::Value,
) -> Result<()>;
async fn put_node_script_timing(
&self,
node: &NodeVisitRef<'_>,
timing: &serde_json::Value,
) -> Result<()>;
async fn put_node_parallel_results(
&self,
node: &NodeVisitRef<'_>,
results: &serde_json::Value,
) -> Result<()>;
async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()>;
async fn put_node_stderr(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()>;
async fn get_node(&self, node: &NodeVisitRef<'_>) -> Result<NodeSnapshot>;
async fn list_node_visits(&self, node_id: &str) -> Result<Vec<u32>>;
async fn list_node_ids(&self) -> Result<Vec<String>>;
async fn put_final_patch(&self, patch: &str) -> Result<()>;
async fn get_final_patch(&self) -> Result<Option<String>>;
async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()>;
async fn get_pull_request(&self) -> Result<Option<PullRequestRecord>>;
async fn append_event(&self, payload: &EventPayload) -> Result<u32>;
async fn list_events(&self) -> Result<Vec<EventEnvelope>>;

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

@ -14,12 +14,12 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeSnapshot, NodeVisitRef, Result,
RunSnapshot, RunStore, RunSummary, Store, StoreError,
CatalogRecord, EventEnvelope, EventPayload, ListRunsQuery, NodeOutcomeRecord, NodeSnapshot,
NodeVisitRef, Result, RunSnapshot, RunStore, RunSummary, Store, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, Retro, RunId, RunRecord, RunStatusRecord,
SandboxRecord, StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
RunStatusRecord, SandboxRecord, StartRecord,
};
#[derive(Debug, Default)]
@ -126,11 +126,32 @@ impl InMemoryRunStore {
prompt: read_text(data, &keys::node_prompt(node))?,
response: read_text(data, &keys::node_response(node))?,
status: read_json(data, &keys::node_status(node))?,
outcome: read_json(data, &keys::node_outcome(node))?,
provider_used: read_json(data, &keys::node_provider_used(node))?,
diff: read_text(data, &keys::node_diff(node))?,
script_invocation: read_json(data, &keys::node_script_invocation(node))?,
script_timing: read_json(data, &keys::node_script_timing(node))?,
parallel_results: read_json(data, &keys::node_parallel_results(node))?,
stdout: read_text(data, &keys::node_stdout(node))?,
stderr: read_text(data, &keys::node_stderr(node))?,
})
}
async fn list_node_ids_inner(&self) -> Vec<String> {
let data = self.snapshot_data().await;
let mut node_ids = BTreeSet::new();
for key in data.keys() {
if let Some((node_id, _, _)) = keys::parse_node_key(key) {
node_ids.insert(node_id);
continue;
}
if let Some((node_id, _, _)) = keys::parse_node_asset_key(key) {
node_ids.insert(node_id);
}
}
node_ids.into_iter().collect()
}
async fn list_events_from_inner(&self, seq: u32) -> Result<Vec<EventEnvelope>> {
let data = self.snapshot_data().await;
let mut events = Vec::new();
@ -222,6 +243,8 @@ impl InMemoryRunStore {
retro: read_json(data, keys::retro())?,
graph: read_text(data, keys::graph())?,
sandbox: read_json(data, keys::sandbox())?,
final_patch: read_text(data, keys::final_patch())?,
pull_request: read_json(data, keys::pull_request())?,
nodes,
}))
}
@ -413,6 +436,54 @@ impl RunStore for InMemoryRunStore {
self.put_json(keys::node_status(node), status).await
}
async fn put_node_outcome(
&self,
node: &NodeVisitRef<'_>,
outcome: &NodeOutcomeRecord,
) -> Result<()> {
self.put_json(keys::node_outcome(node), outcome).await
}
async fn put_node_provider_used(
&self,
node: &NodeVisitRef<'_>,
provider_used: &serde_json::Value,
) -> Result<()> {
self.put_json(keys::node_provider_used(node), provider_used)
.await
}
async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> {
self.put_text(keys::node_diff(node), diff).await;
Ok(())
}
async fn put_node_script_invocation(
&self,
node: &NodeVisitRef<'_>,
invocation: &serde_json::Value,
) -> Result<()> {
self.put_json(keys::node_script_invocation(node), invocation)
.await
}
async fn put_node_script_timing(
&self,
node: &NodeVisitRef<'_>,
timing: &serde_json::Value,
) -> Result<()> {
self.put_json(keys::node_script_timing(node), timing).await
}
async fn put_node_parallel_results(
&self,
node: &NodeVisitRef<'_>,
results: &serde_json::Value,
) -> Result<()> {
self.put_json(keys::node_parallel_results(node), results)
.await
}
async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.put_text(keys::node_stdout(node), log).await;
Ok(())
@ -442,6 +513,28 @@ impl RunStore for InMemoryRunStore {
Ok(visits.into_iter().collect())
}
async fn list_node_ids(&self) -> Result<Vec<String>> {
Ok(self.list_node_ids_inner().await)
}
async fn put_final_patch(&self, patch: &str) -> Result<()> {
self.put_text(keys::final_patch().to_string(), patch).await;
Ok(())
}
async fn get_final_patch(&self) -> Result<Option<String>> {
self.get_text(keys::final_patch()).await
}
async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()> {
self.put_json(keys::pull_request().to_string(), record)
.await
}
async fn get_pull_request(&self) -> Result<Option<PullRequestRecord>> {
self.get_json(keys::pull_request()).await
}
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
payload.validate(&self.run_id)?;
@ -650,7 +743,8 @@ mod tests {
use chrono::Duration as ChronoDuration;
use fabro_types::{
AttrValue, FabroSettings, Graph, RunId, RunStatus, StageStatus, StatusReason, fixtures,
AttrValue, FabroSettings, Graph, PullRequestRecord, RunId, RunStatus, StageStatus,
StatusReason, fixtures,
};
use tokio::time::timeout;
@ -787,6 +881,27 @@ mod tests {
}
}
fn sample_node_outcome() -> NodeOutcomeRecord {
fabro_types::Outcome {
status: StageStatus::Success,
notes: Some("all good".to_string()),
files_touched: vec!["src/lib.rs".to_string()],
..Default::default()
}
}
fn sample_pull_request() -> PullRequestRecord {
PullRequestRecord {
html_url: "https://github.com/fabro-sh/fabro/pull/123".to_string(),
number: 123,
owner: "fabro-sh".to_string(),
repo: "fabro".to_string(),
base_branch: "main".to_string(),
head_branch: "fabro/run/demo".to_string(),
title: "Map the constellations".to_string(),
}
}
#[tokio::test]
async fn create_run_put_get_and_snapshot_round_trip() {
let store = InMemoryStore::default();
@ -809,6 +924,12 @@ mod tests {
visit: 2,
};
let node_status = sample_node_status();
let node_outcome = sample_node_outcome();
let provider_used = serde_json::json!({"provider": "openai", "model": "gpt-5.4"});
let script_invocation = serde_json::json!({"command": "cargo test"});
let script_timing = serde_json::json!({"duration_ms": 3210});
let parallel_results = serde_json::json!([{"node_id": "lint", "status": "success"}]);
let pull_request = sample_pull_request();
run.put_run(&run_record).await.unwrap();
run.put_start(&start_record).await.unwrap();
@ -821,8 +942,28 @@ mod tests {
run.put_node_prompt(&node, "Plan the fix").await.unwrap();
run.put_node_response(&node, "Implemented").await.unwrap();
run.put_node_status(&node, &node_status).await.unwrap();
run.put_node_outcome(&node, &node_outcome).await.unwrap();
run.put_node_provider_used(&node, &provider_used)
.await
.unwrap();
run.put_node_diff(&node, "diff --git a/src/lib.rs b/src/lib.rs")
.await
.unwrap();
run.put_node_script_invocation(&node, &script_invocation)
.await
.unwrap();
run.put_node_script_timing(&node, &script_timing)
.await
.unwrap();
run.put_node_parallel_results(&node, &parallel_results)
.await
.unwrap();
run.put_node_stdout(&node, "ok").await.unwrap();
run.put_node_stderr(&node, "").await.unwrap();
run.put_final_patch("diff --git a/src/lib.rs b/src/lib.rs\n")
.await
.unwrap();
run.put_pull_request(&pull_request).await.unwrap();
run.put_retro_prompt("How did it go?").await.unwrap();
run.put_retro_response("Smooth enough").await.unwrap();
run.put_artifact_value("summary", &serde_json::json!({"done": true}))
@ -881,6 +1022,15 @@ mod tests {
run.get_asset(&node, "src/lib.rs").await.unwrap(),
Some(Bytes::from_static(b"fn main() {}"))
);
assert_eq!(
run.get_final_patch().await.unwrap().as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(
run.get_pull_request().await.unwrap(),
Some(pull_request.clone())
);
assert_eq!(run.list_node_ids().await.unwrap(), vec!["code".to_string()]);
assert_eq!(
run.list_assets(&node).await.unwrap(),
vec!["src/lib.rs".to_string()]
@ -909,6 +1059,35 @@ mod tests {
let snapshot_status = snapshot.nodes[0].status.as_ref().unwrap();
assert_eq!(snapshot_status.status, node_status.status);
assert_eq!(snapshot_status.failure_reason, node_status.failure_reason);
assert_eq!(
snapshot.nodes[0].outcome.as_ref().unwrap().status,
StageStatus::Success
);
assert_eq!(
snapshot.nodes[0].provider_used.as_ref(),
Some(&provider_used)
);
assert_eq!(
snapshot.nodes[0].diff.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs")
);
assert_eq!(
snapshot.nodes[0].script_invocation.as_ref(),
Some(&script_invocation)
);
assert_eq!(
snapshot.nodes[0].script_timing.as_ref(),
Some(&script_timing)
);
assert_eq!(
snapshot.nodes[0].parallel_results.as_ref(),
Some(&parallel_results)
);
assert_eq!(
snapshot.final_patch.as_deref(),
Some("diff --git a/src/lib.rs b/src/lib.rs\n")
);
assert_eq!(snapshot.pull_request, Some(pull_request));
}
#[tokio::test]
@ -964,6 +1143,10 @@ mod tests {
("code".to_string(), 2, "src/lib.rs".to_string())
]
);
assert_eq!(
run.list_node_ids().await.unwrap(),
vec!["artifact-only".to_string(), "code".to_string()]
);
let snapshot = run.get_snapshot().await.unwrap().unwrap();
assert_eq!(snapshot.nodes.len(), 1);

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

@ -16,12 +16,12 @@ use tokio_stream::wrappers::UnboundedReceiverStream;
use crate::keys;
use crate::{
CatalogRecord, EventEnvelope, EventPayload, NodeSnapshot, NodeVisitRef, Result, RunSnapshot,
RunStore, RunSummary, StoreError,
CatalogRecord, EventEnvelope, EventPayload, NodeOutcomeRecord, NodeSnapshot, NodeVisitRef,
Result, RunSnapshot, RunStore, RunSummary, StoreError,
};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, Retro, RunId, RunRecord, RunStatusRecord,
SandboxRecord, StartRecord,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
RunStatusRecord, SandboxRecord, StartRecord,
};
#[derive(Clone)]
@ -204,6 +204,28 @@ impl SlateRunStore {
prompt: self.inner.db.get_text(&keys::node_prompt(node)).await?,
response: self.inner.db.get_text(&keys::node_response(node)).await?,
status: self.inner.db.get_json(&keys::node_status(node)).await?,
outcome: self.inner.db.get_json(&keys::node_outcome(node)).await?,
provider_used: self
.inner
.db
.get_json(&keys::node_provider_used(node))
.await?,
diff: self.inner.db.get_text(&keys::node_diff(node)).await?,
script_invocation: self
.inner
.db
.get_json(&keys::node_script_invocation(node))
.await?,
script_timing: self
.inner
.db
.get_json(&keys::node_script_timing(node))
.await?,
parallel_results: self
.inner
.db
.get_json(&keys::node_parallel_results(node))
.await?,
stdout: self.inner.db.get_text(&keys::node_stdout(node)).await?,
stderr: self.inner.db.get_text(&keys::node_stderr(node)).await?,
})
@ -319,6 +341,65 @@ impl RunStore for SlateRunStore {
.await
}
async fn put_node_outcome(
&self,
node: &NodeVisitRef<'_>,
outcome: &NodeOutcomeRecord,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_outcome(node), outcome)
.await
}
async fn put_node_provider_used(
&self,
node: &NodeVisitRef<'_>,
provider_used: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_provider_used(node), provider_used)
.await
}
async fn put_node_diff(&self, node: &NodeVisitRef<'_>, diff: &str) -> Result<()> {
self.inner.db.put_text(&keys::node_diff(node), diff).await
}
async fn put_node_script_invocation(
&self,
node: &NodeVisitRef<'_>,
invocation: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_script_invocation(node), invocation)
.await
}
async fn put_node_script_timing(
&self,
node: &NodeVisitRef<'_>,
timing: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_script_timing(node), timing)
.await
}
async fn put_node_parallel_results(
&self,
node: &NodeVisitRef<'_>,
results: &serde_json::Value,
) -> Result<()> {
self.inner
.db
.put_json(&keys::node_parallel_results(node), results)
.await
}
async fn put_node_stdout(&self, node: &NodeVisitRef<'_>, log: &str) -> Result<()> {
self.inner.db.put_text(&keys::node_stdout(node), log).await
}
@ -346,6 +427,47 @@ impl RunStore for SlateRunStore {
Ok(visits.into_iter().collect())
}
async fn list_node_ids(&self) -> Result<Vec<String>> {
let mut iter = self.inner.db.scan_prefix(b"nodes/").await?;
let mut node_ids = BTreeSet::new();
while let Some(entry) = iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((node_id, _, _)) = keys::parse_node_key(&key) {
node_ids.insert(node_id);
}
}
let mut asset_iter = self
.inner
.db
.scan_prefix(keys::ARTIFACT_NODES_PREFIX.as_bytes())
.await?;
while let Some(entry) = asset_iter.next().await? {
let key = key_to_string(&entry.key)?;
if let Some((node_id, _, _)) = keys::parse_node_asset_key(&key) {
node_ids.insert(node_id);
}
}
Ok(node_ids.into_iter().collect())
}
async fn put_final_patch(&self, patch: &str) -> Result<()> {
self.inner.db.put_text(keys::final_patch(), patch).await
}
async fn get_final_patch(&self) -> Result<Option<String>> {
self.inner.db.get_text(keys::final_patch()).await
}
async fn put_pull_request(&self, record: &PullRequestRecord) -> Result<()> {
self.inner.db.put_json(keys::pull_request(), record).await
}
async fn get_pull_request(&self) -> Result<Option<PullRequestRecord>> {
self.inner.db.get_json(keys::pull_request()).await
}
async fn append_event(&self, payload: &EventPayload) -> Result<u32> {
payload.validate(&self.inner.run_id)?;
let seq = self.inner.event_seq.fetch_add(1, Ordering::SeqCst);
@ -503,6 +625,8 @@ impl RunStore for SlateRunStore {
retro: self.get_retro().await?,
graph: self.get_graph().await?,
sandbox: self.get_sandbox().await?,
final_patch: self.get_final_patch().await?,
pull_request: self.get_pull_request().await?,
nodes,
}))
}

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

@ -3,10 +3,10 @@ use std::collections::HashMap;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::{Result, StoreError};
use crate::{NodeOutcomeRecord, Result, StoreError};
use fabro_types::{
Checkpoint, Conclusion, NodeStatusRecord, Retro, RunId, RunRecord, RunStatus, RunStatusRecord,
SandboxRecord, StartRecord, StatusReason,
Checkpoint, Conclusion, NodeStatusRecord, PullRequestRecord, Retro, RunId, RunRecord,
RunStatus, RunStatusRecord, SandboxRecord, StartRecord, StatusReason,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@ -51,6 +51,8 @@ pub struct RunSnapshot {
pub retro: Option<Retro>,
pub graph: Option<String>,
pub sandbox: Option<SandboxRecord>,
pub final_patch: Option<String>,
pub pull_request: Option<PullRequestRecord>,
pub nodes: Vec<NodeSnapshot>,
}
@ -61,6 +63,12 @@ pub struct NodeSnapshot {
pub prompt: Option<String>,
pub response: Option<String>,
pub status: Option<NodeStatusRecord>,
pub outcome: Option<NodeOutcomeRecord>,
pub provider_used: Option<serde_json::Value>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
}

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

@ -7,6 +7,7 @@ pub mod failure_signature;
pub mod graph;
pub mod node_status;
pub mod outcome;
pub mod pull_request;
pub mod retro;
pub mod run;
pub mod run_id;
@ -22,6 +23,7 @@ pub use failure_signature::FailureSignature;
pub use graph::{AttrValue, Edge, Graph, Node, is_llm_handler_type, shape_to_handler_type};
pub use node_status::NodeStatusRecord;
pub use outcome::{FailureCategory, FailureDetail, NodeResult, Outcome, OutcomeMeta, StageStatus};
pub use pull_request::PullRequestRecord;
pub use retro::{
AggregateStats, FrictionKind, FrictionPoint, Learning, LearningCategory, OpenItem,
OpenItemKind, Retro, RetroNarrative, SmoothnessRating, StageRetro,

23
lib/crates/fabro-types/src/pull_request.rs Normal file
View file

@ -0,0 +1,23 @@
use std::path::Path;
use serde::{Deserialize, Serialize};
/// Record of a pull request created for a workflow run.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PullRequestRecord {
pub html_url: String,
pub number: u64,
pub owner: String,
pub repo: String,
pub base_branch: String,
pub head_branch: String,
pub title: String,
}
impl PullRequestRecord {
pub fn save(&self, path: &Path) -> Result<(), String> {
let json = serde_json::to_string_pretty(self)
.map_err(|e| format!("Failed to serialize pull_request.json: {e}"))?;
std::fs::write(path, json).map_err(|e| format!("Failed to write pull_request.json: {e}"))
}
}