From 59fee1476fca1e2304ef9ef97cdd32bd765bbdcc Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 29 Mar 2026 09:48:05 -0400 Subject: [PATCH] Add store dump CLI export command --- lib/crates/fabro-cli/src/args.rs | 27 + lib/crates/fabro-cli/src/commands/mod.rs | 1 + .../fabro-cli/src/commands/store/dump.rs | 591 ++++++++++++++++++ .../fabro-cli/src/commands/store/mod.rs | 11 + lib/crates/fabro-cli/src/main.rs | 21 +- lib/crates/fabro-store/src/keys.rs | 15 +- lib/crates/fabro-store/src/lib.rs | 2 + lib/crates/fabro-store/src/memory.rs | 90 +++ lib/crates/fabro-store/src/slate/mod.rs | 54 ++ lib/crates/fabro-store/src/slate/run_store.rs | 60 ++ 10 files changed, 870 insertions(+), 2 deletions(-) create mode 100644 lib/crates/fabro-cli/src/commands/store/dump.rs create mode 100644 lib/crates/fabro-cli/src/commands/store/mod.rs diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index dae4410c0..705395ff4 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -419,6 +419,16 @@ pub(crate) struct InspectArgs { pub(crate) run: String, } +#[derive(Args)] +pub(crate) struct StoreDumpArgs { + /// Run ID prefix or workflow name + pub(crate) run: String, + + /// Output directory (must not exist or be empty) + #[arg(long, short)] + pub(crate) output: PathBuf, +} + #[derive(Args)] pub(crate) struct SecretGetArgs { /// Name of the secret @@ -751,6 +761,8 @@ pub(crate) enum Commands { Parse(ParseArgs), /// Inspect and copy run assets (screenshots, reports, traces) Asset(AssetNamespace), + /// Export store-backed run state for debugging + Store(StoreNamespace), #[command(flatten)] RunsCmd(RunsCommands), /// List and test LLM models @@ -828,6 +840,9 @@ impl Commands { AssetCommand::List(_) => "asset list", AssetCommand::Cp(_) => "asset cp", }, + Self::Store(ns) => match &ns.command { + StoreCommand::Dump(_) => "store dump", + }, Self::Exec(_) => "exec", Self::RunCmd(cmd) => cmd.name(), Self::Preflight(_) => "preflight", @@ -922,6 +937,18 @@ pub(crate) enum AssetCommand { Cp(AssetCpArgs), } +#[derive(Args)] +pub(crate) struct StoreNamespace { + #[command(subcommand)] + pub(crate) command: StoreCommand, +} + +#[derive(Subcommand)] +pub(crate) enum StoreCommand { + /// Export a run's durable state to a directory + Dump(StoreDumpArgs), +} + #[derive(Args)] pub(crate) struct SecretNamespace { #[command(subcommand)] diff --git a/lib/crates/fabro-cli/src/commands/mod.rs b/lib/crates/fabro-cli/src/commands/mod.rs index 799f7ba30..10d1d8945 100644 --- a/lib/crates/fabro-cli/src/commands/mod.rs +++ b/lib/crates/fabro-cli/src/commands/mod.rs @@ -15,6 +15,7 @@ pub(crate) mod run; pub(crate) mod runs; pub(crate) mod secret; pub(crate) mod skill; +pub(crate) mod store; pub(crate) mod system; pub(crate) mod upgrade; pub(crate) mod validate; diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs new file mode 100644 index 000000000..d10c4c36c --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -0,0 +1,591 @@ +use std::io::{ErrorKind, Write}; +use std::path::Path; + +use anyhow::{Context, Result}; +use fabro_config::FabroSettingsExt; +use fabro_store::{NodeVisitRef, RunStore}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; +use serde::Serialize; + +use crate::args::StoreDumpArgs; +use crate::cli_config::load_cli_settings; +use crate::store; + +pub(crate) async fn dump_command(args: &StoreDumpArgs) -> Result<()> { + let cli_settings = load_cli_settings(None)?; + let base = runs_base(&cli_settings.storage_dir()); + let store = store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + .with_context(|| { + format!( + "run {} is not in the store (it may be a legacy filesystem-only run)", + run.run_id + ) + })?; + + prepare_output_dir(&args.output)?; + let file_count = export_run(run_store.as_ref(), &args.output).await?; + println!( + "Exported {file_count} files for run {} to {}", + run.run_id, + args.output.display() + ); + Ok(()) +} + +pub(crate) async fn export_run(run_store: &dyn RunStore, output_dir: &Path) -> Result { + let snapshot = run_store + .get_snapshot() + .await? + .context("run has no data in the store")?; + + std::fs::create_dir_all(output_dir)?; + + let mut file_count = 0; + + write_json_file(&output_dir.join("run.json"), &snapshot.run)?; + file_count += 1; + file_count += usize::from(write_optional_json_file( + &output_dir.join("start.json"), + snapshot.start.as_ref(), + )?); + file_count += usize::from(write_optional_json_file( + &output_dir.join("status.json"), + snapshot.status.as_ref(), + )?); + file_count += usize::from(write_optional_json_file( + &output_dir.join("checkpoint.json"), + snapshot.checkpoint.as_ref(), + )?); + file_count += usize::from(write_optional_json_file( + &output_dir.join("conclusion.json"), + snapshot.conclusion.as_ref(), + )?); + file_count += usize::from(write_optional_json_file( + &output_dir.join("retro.json"), + snapshot.retro.as_ref(), + )?); + file_count += usize::from(write_optional_text_file( + &output_dir.join("graph.fabro"), + snapshot.graph.as_deref(), + )?); + file_count += usize::from(write_optional_json_file( + &output_dir.join("sandbox.json"), + snapshot.sandbox.as_ref(), + )?); + + for node in &snapshot.nodes { + let base = output_dir + .join("nodes") + .join(&node.node_id) + .join(format!("visit-{}", node.visit)); + file_count += usize::from(write_optional_text_file( + &base.join("prompt.md"), + node.prompt.as_deref(), + )?); + file_count += usize::from(write_optional_text_file( + &base.join("response.md"), + node.response.as_deref(), + )?); + file_count += usize::from(write_optional_json_file( + &base.join("status.json"), + node.status.as_ref(), + )?); + file_count += usize::from(write_optional_text_file( + &base.join("stdout.log"), + node.stdout.as_deref(), + )?); + file_count += usize::from(write_optional_text_file( + &base.join("stderr.log"), + node.stderr.as_deref(), + )?); + } + + file_count += usize::from(write_optional_text_file( + &output_dir.join("retro").join("prompt.md"), + run_store.get_retro_prompt().await?.as_deref(), + )?); + file_count += usize::from(write_optional_text_file( + &output_dir.join("retro").join("response.md"), + run_store.get_retro_response().await?.as_deref(), + )?); + + write_events_jsonl( + &output_dir.join("events.jsonl"), + &run_store.list_events().await?, + )?; + file_count += 1; + + for (seq, checkpoint) in run_store.list_checkpoints().await? { + write_json_file( + &output_dir + .join("checkpoints") + .join(format!("{seq:04}.json")), + &checkpoint, + )?; + file_count += 1; + } + + for artifact_id in run_store.list_artifact_values().await? { + let value = run_store + .get_artifact_value(&artifact_id) + .await? + .with_context(|| format!("artifact value {artifact_id:?} is missing from the store"))?; + write_json_file( + &output_dir + .join("artifacts") + .join("values") + .join(format!("{artifact_id}.json")), + &value, + )?; + file_count += 1; + } + + for (node_id, visit, filename) in run_store.list_all_assets().await? { + let node = NodeVisitRef { + node_id: &node_id, + visit, + }; + let data = run_store + .get_asset(&node, &filename) + .await? + .with_context(|| { + format!("asset {filename:?} for node {node_id:?} visit {visit} is missing from the store") + })?; + write_bytes_file( + &output_dir + .join("artifacts") + .join("nodes") + .join(&node_id) + .join(format!("visit-{visit}")) + .join(&filename), + data.as_ref(), + )?; + file_count += 1; + } + + Ok(file_count) +} + +fn prepare_output_dir(path: &Path) -> Result<()> { + match std::fs::symlink_metadata(path) { + Ok(metadata) => { + if metadata.file_type().is_symlink() || !metadata.is_dir() { + return Err(output_dir_error(path)); + } + + let mut entries = std::fs::read_dir(path) + .with_context(|| format!("failed to read {}", path.display()))?; + if entries.next().transpose()?.is_some() { + return Err(output_dir_error(path)); + } + } + Err(err) if err.kind() == ErrorKind::NotFound => { + std::fs::create_dir_all(path) + .with_context(|| format!("failed to create {}", path.display()))?; + } + Err(err) => return Err(err.into()), + } + + Ok(()) +} + +fn output_dir_error(path: &Path) -> anyhow::Error { + anyhow::anyhow!( + "output path {} already exists and is not an empty directory; remove it first or choose a different path", + path.display() + ) +} + +fn write_optional_json_file(path: &Path, value: Option<&T>) -> Result +where + T: Serialize, +{ + match value { + Some(value) => { + write_json_file(path, value)?; + Ok(true) + } + None => Ok(false), + } +} + +fn write_json_file(path: &Path, value: &T) -> Result<()> +where + T: Serialize, +{ + ensure_parent_dir(path)?; + let bytes = serde_json::to_vec_pretty(value)?; + std::fs::write(path, bytes).with_context(|| format!("failed to write {}", path.display()))?; + Ok(()) +} + +fn write_optional_text_file(path: &Path, value: Option<&str>) -> Result { + match value { + Some(value) => { + write_text_file(path, value)?; + Ok(true) + } + None => Ok(false), + } +} + +fn write_text_file(path: &Path, value: &str) -> Result<()> { + write_bytes_file(path, value.as_bytes()) +} + +fn write_bytes_file(path: &Path, value: &[u8]) -> Result<()> { + ensure_parent_dir(path)?; + std::fs::write(path, value).with_context(|| format!("failed to write {}", path.display()))?; + Ok(()) +} + +fn write_events_jsonl(path: &Path, events: &[fabro_store::EventEnvelope]) -> Result<()> { + ensure_parent_dir(path)?; + let mut file = std::fs::File::create(path) + .with_context(|| format!("failed to create {}", path.display()))?; + for event in events { + serde_json::to_writer(&mut file, event)?; + file.write_all(b"\n")?; + } + Ok(()) +} + +fn ensure_parent_dir(path: &Path) -> Result<()> { + let parent = path + .parent() + .with_context(|| format!("path {} has no parent", path.display()))?; + std::fs::create_dir_all(parent) + .with_context(|| format!("failed to create {}", parent.display()))?; + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + use std::collections::HashMap; + use std::path::PathBuf; + + use chrono::{DateTime, Utc}; + use fabro_store::{EventEnvelope, EventPayload, InMemoryStore, Store as _}; + use fabro_types::{ + AggregateStats, AttrValue, Checkpoint, Conclusion, FabroSettings, Graph, NodeStatusRecord, + Retro, RunRecord, RunStatus, RunStatusRecord, SandboxRecord, StageStatus, StartRecord, + StatusReason, + }; + + fn dt(rfc3339: &str) -> DateTime { + DateTime::parse_from_rfc3339(rfc3339) + .unwrap() + .with_timezone(&Utc) + } + + fn sample_run_record(run_id: &str, created_at: DateTime) -> RunRecord { + let mut graph = Graph::new("night-sky"); + graph.attrs.insert( + "goal".to_string(), + AttrValue::String("map the constellations".to_string()), + ); + RunRecord { + run_id: run_id.to_string(), + 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) -> StartRecord { + StartRecord { + run_id: run_id.to_string(), + start_time: created_at + chrono::Duration::seconds(5), + run_branch: Some("fabro/run/demo".to_string()), + base_sha: Some("abc123".to_string()), + } + } + + fn sample_status() -> RunStatusRecord { + RunStatusRecord { + status: RunStatus::Running, + reason: Some(StatusReason::SandboxInitializing), + updated_at: dt("2026-03-27T12:05:00Z"), + } + } + + fn sample_checkpoint(current_node: &str, visit: u32) -> Checkpoint { + Checkpoint { + timestamp: dt("2026-03-27T12:10:00Z"), + current_node: current_node.to_string(), + completed_nodes: vec!["plan".to_string()], + node_retries: HashMap::from([(current_node.to_string(), visit.saturating_sub(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([(current_node.to_string(), visit as usize)]), + } + } + + 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: run_id.to_string(), + 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, + data_host: 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!({ + "ts": ts, + "run_id": run_id, + "event": event + }), + run_id, + ) + .unwrap() + } + + fn read_json(path: &Path) -> T { + serde_json::from_slice(&std::fs::read(path).unwrap()).unwrap() + } + + #[tokio::test] + async fn export_run_writes_expected_directory_tree() { + let store = InMemoryStore::default(); + let created_at = dt("2026-03-27T12:00:00Z"); + let run = store.create_run("run-1", created_at, None).await.unwrap(); + + run.put_run(&sample_run_record("run-1", created_at)) + .await + .unwrap(); + run.put_start(&sample_start_record("run-1", created_at)) + .await + .unwrap(); + run.put_status(&sample_status()).await.unwrap(); + run.append_checkpoint(&sample_checkpoint("plan", 1)) + .await + .unwrap(); + run.append_checkpoint(&sample_checkpoint("code", 2)) + .await + .unwrap(); + run.put_conclusion(&sample_conclusion()).await.unwrap(); + run.put_retro(&sample_retro("run-1")).await.unwrap(); + run.put_graph("digraph night_sky {}").await.unwrap(); + run.put_sandbox(&sample_sandbox()).await.unwrap(); + + let node = NodeVisitRef { + node_id: "code", + visit: 2, + }; + run.put_node_prompt(&node, "Plan the fix").await.unwrap(); + run.put_node_response(&node, "Implemented").await.unwrap(); + run.put_node_status(&node, &sample_node_status()) + .await + .unwrap(); + run.put_node_stdout(&node, "stdout line").await.unwrap(); + run.put_node_stderr(&node, "").await.unwrap(); + run.put_retro_prompt("How did it go?").await.unwrap(); + run.put_retro_response("Smooth enough").await.unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:00.000Z", + "WorkflowRunStarted", + )) + .await + .unwrap(); + run.append_event(&event_payload( + "run-1", + "2026-03-27T12:00:01.000Z", + "StageCompleted", + )) + .await + .unwrap(); + run.put_artifact_value("summary", &serde_json::json!({"done": true})) + .await + .unwrap(); + run.put_artifact_value("plan", &serde_json::json!({"steps": 3})) + .await + .unwrap(); + run.put_asset(&node, "src/lib.rs", b"fn main() {}") + .await + .unwrap(); + + let asset_only_node = NodeVisitRef { + node_id: "artifact-only", + visit: 7, + }; + run.put_asset(&asset_only_node, "logs/output.txt", b"hello") + .await + .unwrap(); + + let output = tempfile::tempdir().unwrap(); + let file_count = export_run(run.as_ref(), output.path()).await.unwrap(); + assert_eq!(file_count, 22); + + let exported_run: RunRecord = read_json(&output.path().join("run.json")); + assert_eq!(exported_run.run_id, "run-1"); + + let exported_start: StartRecord = read_json(&output.path().join("start.json")); + assert_eq!(exported_start.run_id, "run-1"); + + let exported_status: RunStatusRecord = read_json(&output.path().join("status.json")); + assert_eq!(exported_status.status, RunStatus::Running); + + let exported_checkpoint: Checkpoint = read_json(&output.path().join("checkpoint.json")); + assert_eq!(exported_checkpoint.current_node, "code"); + assert_eq!( + std::fs::read_to_string(output.path().join("graph.fabro")).unwrap(), + "digraph night_sky {}" + ); + + assert_eq!( + std::fs::read_to_string(output.path().join("nodes/code/visit-2/prompt.md")).unwrap(), + "Plan the fix" + ); + assert_eq!( + std::fs::read_to_string(output.path().join("nodes/code/visit-2/response.md")).unwrap(), + "Implemented" + ); + let node_status: NodeStatusRecord = + read_json(&output.path().join("nodes/code/visit-2/status.json")); + assert_eq!(node_status.status, StageStatus::PartialSuccess); + assert_eq!( + std::fs::read_to_string(output.path().join("nodes/code/visit-2/stdout.log")).unwrap(), + "stdout line" + ); + assert_eq!( + std::fs::read_to_string(output.path().join("nodes/code/visit-2/stderr.log")).unwrap(), + "" + ); + + assert_eq!( + std::fs::read_to_string(output.path().join("retro/prompt.md")).unwrap(), + "How did it go?" + ); + assert_eq!( + std::fs::read_to_string(output.path().join("retro/response.md")).unwrap(), + "Smooth enough" + ); + + let event_lines = std::fs::read_to_string(output.path().join("events.jsonl")).unwrap(); + let events: Vec = event_lines + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!(events.len(), 2); + assert_eq!(events[0].seq, 1); + assert_eq!(events[1].seq, 2); + + let first_checkpoint: Checkpoint = read_json(&output.path().join("checkpoints/0001.json")); + let second_checkpoint: Checkpoint = read_json(&output.path().join("checkpoints/0002.json")); + assert_eq!(first_checkpoint.current_node, "plan"); + assert_eq!(second_checkpoint.current_node, "code"); + + let exported_plan: serde_json::Value = + read_json(&output.path().join("artifacts/values/plan.json")); + let exported_summary: serde_json::Value = + read_json(&output.path().join("artifacts/values/summary.json")); + assert_eq!(exported_plan, serde_json::json!({"steps": 3})); + assert_eq!(exported_summary, serde_json::json!({"done": true})); + + assert_eq!( + std::fs::read( + output + .path() + .join("artifacts/nodes/code/visit-2/src/lib.rs") + ) + .unwrap(), + b"fn main() {}" + ); + assert_eq!( + std::fs::read( + output + .path() + .join("artifacts/nodes/artifact-only/visit-7/logs/output.txt") + ) + .unwrap(), + b"hello" + ); + assert!(!output.path().join("nodes/artifact-only").exists()); + } + + #[test] + fn prepare_output_dir_rejects_non_empty_directory() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("existing.txt"), "x").unwrap(); + + let err = prepare_output_dir(dir.path()).unwrap_err(); + assert!( + err.to_string() + .contains("already exists and is not an empty directory") + ); + } +} diff --git a/lib/crates/fabro-cli/src/commands/store/mod.rs b/lib/crates/fabro-cli/src/commands/store/mod.rs new file mode 100644 index 000000000..44ed9963b --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/store/mod.rs @@ -0,0 +1,11 @@ +mod dump; + +use anyhow::Result; + +use crate::args::{StoreCommand, StoreNamespace}; + +pub(crate) async fn dispatch(ns: StoreNamespace) -> Result<()> { + match ns.command { + StoreCommand::Dump(args) => dump::dump_command(&args).await, + } +} diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index a4da3e962..26663c408 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -179,6 +179,7 @@ async fn main_inner() -> (String, Result<()>) { commands::parse::run(&args)?; } Commands::Asset(ns) => commands::asset::dispatch(ns)?, + Commands::Store(ns) => commands::store::dispatch(ns).await?, Commands::RunsCmd(cmd) => commands::runs::dispatch(cmd).await?, Commands::Model { command } => commands::model::execute(command, &globals).await?, #[cfg(feature = "server")] @@ -243,7 +244,10 @@ async fn main_inner() -> (String, Result<()>) { #[cfg(test)] mod tests { use super::*; - use args::{ConfigCommand, ConfigNamespace, ProviderCommand, ProviderNamespace}; + use args::{ + ConfigCommand, ConfigNamespace, ProviderCommand, ProviderNamespace, StoreCommand, + StoreNamespace, + }; use clap::Parser; #[test] @@ -302,6 +306,21 @@ mod tests { } } + #[test] + fn parse_store_dump_command() { + let cli = Cli::try_parse_from(["fabro", "store", "dump", "ABC123", "-o", "./out"]) + .expect("should parse"); + match *cli.command { + Commands::Store(StoreNamespace { + command: StoreCommand::Dump(args), + }) => { + assert_eq!(args.run, "ABC123"); + assert_eq!(args.output, std::path::PathBuf::from("./out")); + } + _ => panic!("unexpected command variant"), + } + } + #[test] fn parse_start_command() { let cli = Cli::try_parse_from(["fabro", "start", "ABC123"]).expect("should parse"); diff --git a/lib/crates/fabro-store/src/keys.rs b/lib/crates/fabro-store/src/keys.rs index 6444d02c5..358cfe858 100644 --- a/lib/crates/fabro-store/src/keys.rs +++ b/lib/crates/fabro-store/src/keys.rs @@ -115,11 +115,16 @@ pub(crate) fn parse_checkpoint_seq(key: &str) -> Option { parse_seq(key, CHECKPOINTS_PREFIX) } +pub(crate) fn parse_artifact_value_id(key: &str) -> Option { + key.strip_prefix(ARTIFACT_VALUES_PREFIX) + .and_then(|s| s.strip_suffix(".json")) + .map(ToString::to_string) +} + pub(crate) fn parse_node_key(key: &str) -> Option<(String, u32, String)> { parse_visit_scoped_key(key, "nodes/") } -#[cfg(test)] pub(crate) fn parse_node_asset_key(key: &str) -> Option<(String, u32, String)> { parse_visit_scoped_key(key, ARTIFACT_NODES_PREFIX) } @@ -185,6 +190,10 @@ mod tests { fn parse_helpers_extract_sequences_and_node_visits() { assert_eq!(parse_event_seq("events/000007-123.json"), Some(7)); assert_eq!(parse_checkpoint_seq("checkpoints/0042-456.json"), Some(42)); + assert_eq!( + parse_artifact_value_id("artifacts/values/summary.json"), + Some("summary".to_string()) + ); assert_eq!( parse_node_key("nodes/plan/visit-3/status.json"), Some(("plan".to_string(), 3, "status.json".to_string())) @@ -199,6 +208,10 @@ mod tests { fn parse_helpers_reject_invalid_keys() { assert_eq!(parse_event_seq("events/not-a-seq.json"), None); assert_eq!(parse_checkpoint_seq("checkpoints/oops.json"), None); + assert_eq!( + parse_artifact_value_id("artifacts/values/summary.txt"), + None + ); assert_eq!(parse_node_key("nodes/plan/status.json"), None); assert_eq!( parse_node_asset_key("artifacts/nodes/code/status.json"), diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 36753921b..90dccb261 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/lib/crates/fabro-store/src/lib.rs @@ -102,10 +102,12 @@ pub trait RunStore: Send + Sync { async fn put_artifact_value(&self, artifact_id: &str, value: &serde_json::Value) -> Result<()>; async fn get_artifact_value(&self, artifact_id: &str) -> Result>; + async fn list_artifact_values(&self) -> Result>; async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()>; async fn get_asset(&self, node: &NodeVisitRef<'_>, filename: &str) -> Result>; async fn list_assets(&self, node: &NodeVisitRef<'_>) -> Result>; + async fn list_all_assets(&self) -> Result>; async fn get_snapshot(&self) -> Result>; } diff --git a/lib/crates/fabro-store/src/memory.rs b/lib/crates/fabro-store/src/memory.rs index 3f247fa88..d4af91e3a 100644 --- a/lib/crates/fabro-store/src/memory.rs +++ b/lib/crates/fabro-store/src/memory.rs @@ -163,6 +163,32 @@ impl InMemoryRunStore { Ok(checkpoints) } + async fn list_artifact_values_inner(&self) -> Result> { + let data = self.snapshot_data().await; + let mut artifact_ids = Vec::new(); + for key in data.keys() { + let Some(artifact_id) = keys::parse_artifact_value_id(key) else { + continue; + }; + artifact_ids.push(artifact_id); + } + artifact_ids.sort(); + Ok(artifact_ids) + } + + async fn list_all_assets_inner(&self) -> Result> { + let data = self.snapshot_data().await; + let mut assets = Vec::new(); + for key in data.keys() { + let Some(asset) = keys::parse_node_asset_key(key) else { + continue; + }; + assets.push(asset); + } + assets.sort(); + Ok(assets) + } + fn build_snapshot_from_data( &self, data: &BTreeMap>, @@ -501,6 +527,10 @@ impl RunStore for InMemoryRunStore { self.get_json(&keys::artifact_value(artifact_id)).await } + async fn list_artifact_values(&self) -> Result> { + self.list_artifact_values_inner().await + } + async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()> { self.put_bytes(keys::node_asset(node, filename), data).await; Ok(()) @@ -526,6 +556,10 @@ impl RunStore for InMemoryRunStore { Ok(assets) } + async fn list_all_assets(&self) -> Result> { + self.list_all_assets_inner().await + } + async fn get_snapshot(&self) -> Result> { let data = self.snapshot_data().await; self.build_snapshot_from_data(&data) @@ -861,6 +895,62 @@ mod tests { assert_eq!(snapshot_status.failure_reason, node_status.failure_reason); } + #[tokio::test] + async fn list_artifact_values_and_all_assets_include_asset_only_visits() { + let store = InMemoryStore::default(); + let created_at = dt("2026-03-27T12:00:00Z"); + let run = store.create_run("run-1", created_at, None).await.unwrap(); + run.put_run(&sample_run_record("run-1", created_at)) + .await + .unwrap(); + + run.put_artifact_value("summary", &serde_json::json!({"done": true})) + .await + .unwrap(); + run.put_artifact_value("plan", &serde_json::json!({"steps": 3})) + .await + .unwrap(); + + let snapshot_node = NodeVisitRef { + node_id: "code", + visit: 2, + }; + run.put_node_prompt(&snapshot_node, "Plan the fix") + .await + .unwrap(); + run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") + .await + .unwrap(); + + let asset_only_node = NodeVisitRef { + node_id: "artifact-only", + visit: 7, + }; + run.put_asset(&asset_only_node, "logs/output.txt", b"hello") + .await + .unwrap(); + + assert_eq!( + run.list_artifact_values().await.unwrap(), + vec!["plan".to_string(), "summary".to_string()] + ); + assert_eq!( + run.list_all_assets().await.unwrap(), + vec![ + ( + "artifact-only".to_string(), + 7, + "logs/output.txt".to_string() + ), + ("code".to_string(), 2, "src/lib.rs".to_string()) + ] + ); + + let snapshot = run.get_snapshot().await.unwrap().unwrap(); + assert_eq!(snapshot.nodes.len(), 1); + assert_eq!(snapshot.nodes[0].node_id, "code"); + } + #[tokio::test] async fn append_event_validates_payload_shape_and_run_id() { let store = InMemoryStore::default(); diff --git a/lib/crates/fabro-store/src/slate/mod.rs b/lib/crates/fabro-store/src/slate/mod.rs index 19a4d6ba4..9e9545d57 100644 --- a/lib/crates/fabro-store/src/slate/mod.rs +++ b/lib/crates/fabro-store/src/slate/mod.rs @@ -939,4 +939,58 @@ mod tests { vec!["src/lib.rs".to_string()] ); } + + #[tokio::test] + async fn slate_run_store_lists_artifact_values_and_asset_only_visits() { + let (_object_store, store) = make_store(); + let created_at = dt("2026-03-27T12:00:00Z"); + let run = store.create_run("run-1", created_at, None).await.unwrap(); + run.put_run(&sample_run_record("run-1", created_at)) + .await + .unwrap(); + + run.put_artifact_value("summary", &serde_json::json!({"done": true})) + .await + .unwrap(); + run.put_artifact_value("plan", &serde_json::json!({"steps": 3})) + .await + .unwrap(); + + let snapshot_node = NodeVisitRef { + node_id: "code", + visit: 2, + }; + run.put_node_prompt(&snapshot_node, "Plan").await.unwrap(); + run.put_asset(&snapshot_node, "src/lib.rs", b"fn main() {}") + .await + .unwrap(); + + let asset_only_node = NodeVisitRef { + node_id: "artifact-only", + visit: 7, + }; + run.put_asset(&asset_only_node, "logs/output.txt", b"hello") + .await + .unwrap(); + + assert_eq!( + run.list_artifact_values().await.unwrap(), + vec!["plan".to_string(), "summary".to_string()] + ); + assert_eq!( + run.list_all_assets().await.unwrap(), + vec![ + ( + "artifact-only".to_string(), + 7, + "logs/output.txt".to_string() + ), + ("code".to_string(), 2, "src/lib.rs".to_string()) + ] + ); + + let snapshot = run.get_snapshot().await.unwrap().unwrap(); + assert_eq!(snapshot.nodes.len(), 1); + assert_eq!(snapshot.nodes[0].node_id, "code"); + } } diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index b40b1b62b..1b431dcdd 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -435,6 +435,10 @@ impl RunStore for SlateRunStore { .await } + async fn list_artifact_values(&self) -> Result> { + self.inner.db.list_artifact_values().await + } + async fn put_asset(&self, node: &NodeVisitRef<'_>, filename: &str, data: &[u8]) -> Result<()> { self.inner .db @@ -463,6 +467,10 @@ impl RunStore for SlateRunStore { Ok(assets) } + async fn list_all_assets(&self) -> Result> { + self.inner.db.list_all_assets().await + } + async fn get_snapshot(&self) -> Result> { let Some(run) = self.get_run().await? else { return Ok(None); @@ -574,6 +582,20 @@ impl SlateRunDb { Self::Reader(db) => list_checkpoints(db.as_ref()).await, } } + + async fn list_artifact_values(&self) -> Result> { + match self { + Self::Writer(db) => list_artifact_values(db).await, + Self::Reader(db) => list_artifact_values(db.as_ref()).await, + } + } + + async fn list_all_assets(&self) -> Result> { + match self { + Self::Writer(db) => list_all_assets(db).await, + Self::Reader(db) => list_all_assets(db.as_ref()).await, + } + } } async fn put_json(db: &slatedb::Db, key: &str, value: &T) -> Result<()> { @@ -675,6 +697,44 @@ where Ok(checkpoints) } +async fn list_artifact_values(db: &R) -> Result> +where + R: DbRead + Sync, +{ + let mut iter = db + .scan_prefix(keys::ARTIFACT_VALUES_PREFIX.as_bytes()) + .await?; + let mut artifact_ids = Vec::new(); + while let Some(entry) = iter.next().await? { + let key = key_to_string(&entry.key)?; + let Some(artifact_id) = keys::parse_artifact_value_id(&key) else { + continue; + }; + artifact_ids.push(artifact_id); + } + artifact_ids.sort(); + Ok(artifact_ids) +} + +async fn list_all_assets(db: &R) -> Result> +where + R: DbRead + Sync, +{ + let mut iter = db + .scan_prefix(keys::ARTIFACT_NODES_PREFIX.as_bytes()) + .await?; + let mut assets = Vec::new(); + while let Some(entry) = iter.next().await? { + let key = key_to_string(&entry.key)?; + let Some(asset) = keys::parse_node_asset_key(&key) else { + continue; + }; + assets.push(asset); + } + assets.sort(); + Ok(assets) +} + fn key_to_string(key: &Bytes) -> Result { String::from_utf8(key.to_vec()) .map_err(|err| StoreError::Other(format!("stored key is not valid UTF-8: {err}")))