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

refactor(run): unify legacy dump projections

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-03 14:50:12 -07:00
parent a4e272bb34
commit 2ce613204e
19 changed files with 539 additions and 677 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -5,7 +5,6 @@ use fabro_model::Catalog;
use fabro_sandbox::daytona::detect_repo_info;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::pull_request::maybe_open_pull_request;
use fabro_workflow::records::RunRecordExt;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use tracing::info;
@ -108,7 +107,7 @@ async fn create_from(
&origin_url,
base_branch,
run_branch,
record.goal(),
record.graph.goal(),
&diff,
&model,
true,

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

@ -3,7 +3,6 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use fabro_types::RunId;
use fabro_workflow::records::{RunRecord, RunRecordExt};
use serde::{Deserialize, Serialize};
#[cfg(test)]
@ -61,13 +60,6 @@ pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option<LauncherR
}
pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord> {
if let Ok(run_record) = RunRecord::load(run_dir) {
return read_launcher_record(&launcher_record_path(
&run_record.settings.storage_dir(),
&run_record.run_id,
));
}
let storage_dir = run_dir.parent()?.parent()?;
let launchers_dir = launcher_dir(storage_dir);
let entries = std::fs::read_dir(&launchers_dir).ok()?;
@ -131,34 +123,15 @@ fn launcher_process_matches(_record: &LauncherRecord) -> bool {
mod tests {
use super::*;
use chrono::Utc;
use fabro_graphviz::graph::Graph;
use fabro_types::{Settings, fixtures};
use fabro_workflow::records::RunRecord;
use fabro_types::fixtures;
#[test]
fn active_launcher_record_for_run_removes_stale_record() {
let dir = tempfile::tempdir().unwrap();
let storage_dir = dir.path().join("storage");
let run_dir = dir.path().join("run");
let run_dir = storage_dir.join("runs").join("run");
std::fs::create_dir_all(&run_dir).unwrap();
RunRecord {
run_id: fixtures::RUN_1,
created_at: Utc::now(),
settings: Settings {
storage_dir: Some(storage_dir.clone()),
..Default::default()
},
graph: Graph::default(),
workflow_slug: None,
working_directory: dir.path().to_path_buf(),
host_repo_path: None,
base_branch: None,
labels: std::collections::HashMap::new(),
}
.save(&run_dir)
.unwrap();
let launcher_path = launcher_record_path(&storage_dir, &fixtures::RUN_1);
write_launcher_record(
&launcher_path,

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

@ -10,7 +10,6 @@ use fabro_workflow::operations::{
RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild,
find_run_id_by_prefix_or_store, rewind,
};
use fabro_workflow::records::{RunRecord, RunRecordExt};
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use git2::Repository;
use serde::Serialize;
@ -124,7 +123,6 @@ async fn reset_rewound_run_state(
let _run_record = state
.run
.or_else(|| RunRecord::load(run_dir).ok())
.context("failed to restore run record after rewind: missing run metadata")?;
let checkpoint = MetadataStore::read_checkpoint(git_store.repo_dir(), &run_id.to_string())?
.context("rewound metadata branch is missing checkpoint.json")?;

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

@ -140,7 +140,7 @@ mod tests {
use fabro_types::fixtures;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::records::Conclusion;
use fabro_workflow::run_status::{RunStatusRecord, RunStatusRecordExt};
use fabro_workflow::run_status::RunStatusRecord;
fn no_color_styles() -> Styles {
Styles::new(false)
@ -247,18 +247,25 @@ mod tests {
let dir = tempfile::tempdir().unwrap();
let status_path = dir.path().join("status.json");
let record = RunStatusRecord::new(RunStatus::Succeeded, None);
record.save(&status_path).unwrap();
std::fs::write(&status_path, serde_json::to_string_pretty(&record).unwrap()).unwrap();
// Simulate what the poll loop does
let status = RunStatusRecord::load(&status_path).unwrap().status;
let status = serde_json::from_str::<RunStatusRecord>(
&std::fs::read_to_string(&status_path).unwrap(),
)
.unwrap()
.status;
assert!(status.is_terminal());
assert_eq!(status, RunStatus::Succeeded);
}
#[test]
fn missing_status_treated_as_dead() {
let status = match RunStatusRecord::load(std::path::Path::new("/nonexistent/status.json")) {
Ok(record) => record.status,
let status = match std::fs::read_to_string(std::path::Path::new("/nonexistent/status.json"))
{
Ok(data) => serde_json::from_str::<RunStatusRecord>(&data)
.map(|record| record.status)
.unwrap_or(RunStatus::Dead),
Err(_) => RunStatus::Dead,
};
assert_eq!(status, RunStatus::Dead);

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

@ -1,10 +1,12 @@
use std::io::{ErrorKind, Write};
use std::path::{Component, Path, PathBuf};
use std::io::ErrorKind;
use std::path::Path;
use anyhow::{Context, Result, bail};
use fabro_store::{NodeVisitRef, RunSnapshot, RunState, SlateRunStore};
use anyhow::{Context, Result};
#[cfg(test)]
use fabro_store::NodeVisitRef;
use fabro_store::{RunState, SlateRunStore};
use fabro_workflow::run_dump::RunDump;
use fabro_workflow::run_lookup::{resolve_run_combined, runs_base};
use serde::Serialize;
#[cfg(test)]
use serde::de::DeserializeOwned;
@ -39,9 +41,7 @@ pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) ->
pub(crate) async fn export_run(run_store: &SlateRunStore, output_dir: &Path) -> Result<usize> {
let state = run_store.state().await?;
let snapshot = state
.to_snapshot()
.context("run has no data in the store")?;
anyhow::ensure!(state.run.is_some(), "run has no data in the store");
let output_state = inspect_output_dir(output_dir)?;
let staging_parent = output_parent_dir(output_dir);
@ -59,7 +59,7 @@ pub(crate) async fn export_run(run_store: &SlateRunStore, output_dir: &Path) ->
})?;
let staging_path = staging_dir.path().to_path_buf();
let file_count = export_run_to_dir(run_store, &state, &snapshot, &staging_path).await?;
let file_count = export_run_to_dir(run_store, &state, &staging_path).await?;
if matches!(output_state, OutputDirState::ExistingEmpty) {
std::fs::remove_dir(output_dir)
@ -80,139 +80,10 @@ pub(crate) async fn export_run(run_store: &SlateRunStore, output_dir: &Path) ->
async fn export_run_to_dir(
run_store: &SlateRunStore,
state: &RunState,
snapshot: &RunSnapshot,
output_dir: &Path,
) -> Result<usize> {
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 node_id = validate_single_path_segment("node id", &node.node_id)?;
let base = output_dir
.join("nodes")
.join(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"),
state.retro_prompt.as_deref(),
)?);
file_count += usize::from(write_optional_text_file(
&output_dir.join("retro").join("response.md"),
state.retro_response.as_deref(),
)?);
write_events_jsonl(
&output_dir.join("events.jsonl"),
&run_store.list_events().await?,
)?;
file_count += 1;
for (seq, checkpoint) in &state.checkpoints {
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 artifact_id_segment = validate_single_path_segment("artifact id", &artifact_id)?;
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!("{}.json", artifact_id_segment.display())),
&value,
)?;
file_count += 1;
}
for (node_id, visit, filename) in run_store.list_all_assets().await? {
let node_id_segment = validate_single_path_segment("node id", &node_id)?;
let filename_path = validate_relative_path("asset filename", &filename)?;
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_segment)
.join(format!("visit-{visit}"))
.join(filename_path),
data.as_ref(),
)?;
file_count += 1;
}
Ok(file_count)
let dump = RunDump::store_export(run_store, state).await?;
dump.write_to_dir(output_dir)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@ -255,94 +126,6 @@ fn output_parent_dir(path: &Path) -> &Path {
}
}
fn validate_single_path_segment(kind: &str, value: &str) -> Result<PathBuf> {
let path = validate_relative_path(kind, value)?;
if path.components().count() != 1 {
bail!("{kind} {value:?} must be a single path segment");
}
Ok(path)
}
fn validate_relative_path(kind: &str, value: &str) -> Result<PathBuf> {
let mut normalized = PathBuf::new();
for component in Path::new(value).components() {
match component {
Component::Normal(part) => normalized.push(part),
Component::CurDir => {}
Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
bail!("{kind} {value:?} must be a relative path without '..'");
}
}
}
if normalized.as_os_str().is_empty() {
bail!("{kind} {value:?} must not be empty");
}
Ok(normalized)
}
fn write_optional_json_file<T>(path: &Path, value: Option<&T>) -> Result<bool>
where
T: Serialize,
{
match value {
Some(value) => {
write_json_file(path, value)?;
Ok(true)
}
None => Ok(false),
}
}
fn write_json_file<T>(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<bool> {
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::*;

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

@ -2,7 +2,6 @@ use std::path::Path;
use std::process::Command;
use fabro_checkpoint::git::Store;
use fabro_store::RunState;
use fabro_types::Settings;
use crate::error::{FabroError, Result};
@ -353,81 +352,10 @@ pub fn scan_node_files(run_dir: &Path) -> Vec<(String, Vec<u8>)> {
result
}
pub fn scan_node_files_from_state(state: &RunState) -> Vec<(String, Vec<u8>)> {
let mut result = Vec::new();
let mut keys: Vec<_> = state.nodes.keys().collect();
keys.sort();
for (node_id, visit) in keys {
let Some(node) = state.nodes.get(&(node_id.clone(), *visit)) else {
continue;
};
if let Some(ref prompt) = node.prompt {
result.push((
node_file_path(node_id, *visit, "prompt.md"),
prompt.as_bytes().to_vec(),
));
}
if let Some(ref response) = node.response {
result.push((
node_file_path(node_id, *visit, "response.md"),
response.as_bytes().to_vec(),
));
}
if let Some(ref status) = node.status {
if let Ok(bytes) = serde_json::to_vec_pretty(status) {
result.push((node_file_path(node_id, *visit, "status.json"), bytes));
}
}
if let Some(ref provider_used) = node.provider_used {
if let Ok(bytes) = serde_json::to_vec_pretty(provider_used) {
result.push((node_file_path(node_id, *visit, "provider_used.json"), bytes));
}
}
if let Some(ref diff) = node.diff {
result.push((
node_file_path(node_id, *visit, "diff.patch"),
diff.as_bytes().to_vec(),
));
}
if let Some(ref script_invocation) = node.script_invocation {
if let Ok(bytes) = serde_json::to_vec_pretty(script_invocation) {
result.push((
node_file_path(node_id, *visit, "script_invocation.json"),
bytes,
));
}
}
if let Some(ref script_timing) = node.script_timing {
if let Ok(bytes) = serde_json::to_vec_pretty(script_timing) {
result.push((node_file_path(node_id, *visit, "script_timing.json"), bytes));
}
}
if let Some(ref parallel_results) = node.parallel_results {
if let Ok(bytes) = serde_json::to_vec_pretty(parallel_results) {
result.push((
node_file_path(node_id, *visit, "parallel_results.json"),
bytes,
));
}
}
}
result
}
fn node_file_path(node_id: &str, visit: u32, filename: &str) -> String {
if visit <= 1 {
format!("nodes/{node_id}/{filename}")
} else {
format!("nodes/{node_id}-visit_{visit}/{filename}")
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::run_dump::RunDump;
use fabro_store::SlateStore;
use fabro_types::fixtures;
use object_store::memory::InMemory;
@ -645,7 +573,7 @@ mod tests {
.unwrap();
let state = run.state().await.unwrap();
let files = scan_node_files_from_state(&state);
let files = RunDump::metadata_checkpoint(&state).git_entries().unwrap();
let paths: Vec<&str> = files.iter().map(|(path, _)| path.as_str()).collect();
assert!(paths.contains(&"nodes/work-visit_2/prompt.md"));
assert!(paths.contains(&"nodes/work-visit_2/response.md"));

1
lib/crates/fabro-workflow/src/lib.rs
View file

@ -143,6 +143,7 @@ pub mod pull_request;
pub mod records;
mod retry;
pub(crate) mod run_dir;
pub mod run_dump;
pub mod run_lookup;
pub mod run_options;
pub mod run_status;

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

@ -14,10 +14,10 @@ use fabro_core::state::RunState;
use crate::artifact::ArtifactStore;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::git::MetadataStore;
use crate::git::scan_node_files_from_state;
use crate::graph::WorkflowGraph;
use crate::graph::WorkflowNode;
use crate::outcome::{Outcome, StageStatus, StageUsage};
use crate::run_dump::RunDump;
use crate::run_options::RunOptions;
use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host};
@ -67,29 +67,16 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
let git_author = self.run_options.git_author();
let store = MetadataStore::new(repo_path, &git_author);
let state = self.run_store.state().await.ok();
let run_json = state
let init_dump = state.as_ref().map(RunDump::metadata_init);
let init_entries = init_dump
.as_ref()
.and_then(|state| state.run.as_ref())
.and_then(|record| serde_json::to_vec_pretty(record).ok());
let start_json = state
.as_ref()
.and_then(|state| state.start.as_ref())
.and_then(|record| serde_json::to_vec_pretty(record).ok());
let sandbox_json = state
.as_ref()
.and_then(|state| state.sandbox.as_ref())
.and_then(|record| serde_json::to_vec_pretty(record).ok());
let mut files: Vec<(&str, &[u8])> = Vec::new();
if let Some(ref data) = run_json {
files.push(("run.json", data));
}
if let Some(ref data) = start_json {
files.push(("start.json", data));
}
if let Some(ref data) = sandbox_json {
files.push(("sandbox.json", data));
}
if let Err(e) = store.init_run(&self.run_id.to_string(), &files) {
.and_then(|dump| dump.git_entries().ok())
.unwrap_or_default();
let refs: Vec<(&str, &[u8])> = init_entries
.iter()
.map(|(path, bytes)| (path.as_str(), bytes.as_slice()))
.collect();
if let Err(e) = store.init_run(&self.run_id.to_string(), &refs) {
tracing::warn!(
run_id = %self.run_id,
error = %e,
@ -150,7 +137,11 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
.collect()
};
if let Ok(store_state) = self.run_store.state().await {
extra_entries.extend(scan_node_files_from_state(&store_state));
if let Ok(mut dump_entries) =
RunDump::metadata_checkpoint(&store_state).git_entries()
{
extra_entries.append(&mut dump_entries);
}
}
let extra_refs: Vec<(&str, &[u8])> = extra_entries
.iter()

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

@ -835,7 +835,7 @@ mod tests {
use crate::handler::exit::ExitHandler;
use crate::handler::start::StartHandler;
use crate::operations::resume;
use crate::records::{CheckpointExt, ConclusionExt};
use crate::records::CheckpointExt;
const MINIMAL_DOT: &str = r#"digraph Test {
graph [goal="Build feature"]
@ -1019,17 +1019,19 @@ mod tests {
let services = test_start_services(&store, &run_dir, emitter, registry).await;
// Seed an authoritative checkpoint event so start() sees it
let checkpoint = Checkpoint::from_context(
&Context::new(),
"start",
vec!["start".to_string()],
HashMap::new(),
HashMap::new(),
Some("exit".to_string()),
HashMap::new(),
HashMap::new(),
HashMap::new(),
);
let checkpoint = Checkpoint {
timestamp: chrono::Utc::now(),
current_node: "start".into(),
completed_nodes: vec!["start".to_string()],
node_retries: HashMap::new(),
context_values: Context::new().snapshot(),
node_outcomes: HashMap::new(),
next_node_id: Some("exit".to_string()),
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::new(),
};
append_workflow_event(
services.run_store.as_ref(),
&services.run_id,
@ -1119,7 +1121,7 @@ mod tests {
);
checkpoint.save(&run_dir.join("checkpoint.json")).unwrap();
crate::records::Conclusion {
let conclusion = crate::records::Conclusion {
timestamp: Utc::now(),
status: StageStatus::Success,
duration_ms: 1,
@ -1134,8 +1136,11 @@ mod tests {
total_cache_write_tokens: 0,
total_reasoning_tokens: 0,
has_pricing: false,
}
.save(&run_dir.join("conclusion.json"))
};
std::fs::write(
run_dir.join("conclusion.json"),
serde_json::to_string_pretty(&conclusion).unwrap(),
)
.unwrap();
let result = resume(

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

@ -3,9 +3,10 @@ use std::sync::Arc;
use crate::error::FabroError;
use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent};
use crate::git::{MetadataStore, scan_node_files_from_state};
use crate::git::MetadataStore;
use crate::outcome::{Outcome, OutcomeExt, StageStatus};
use crate::records::{Checkpoint, Conclusion, StageSummary};
use crate::run_dump::RunDump;
use crate::run_options::RunOptions;
use crate::run_status::{RunStatus, StatusReason};
use crate::sandbox_git::git_push_host;
@ -198,19 +199,10 @@ pub async fn write_finalize_commit(
let Ok(store_state) = run_store.state().await else {
return;
};
let mut entries = scan_node_files_from_state(&store_state);
let retro_bytes = store_state
.retro
.as_ref()
.and_then(|retro| serde_json::to_vec_pretty(retro).ok());
if let Some(bytes) = retro_bytes {
entries.push(("retro.json".to_string(), bytes));
}
let refs: Vec<(&str, &[u8])> = entries
.iter()
.map(|(k, v)| (k.as_str(), v.as_slice()))
.collect();
if let Err(e) = store.write_files(&run_options.run_id.to_string(), &refs, "finalize run") {
let dump = RunDump::metadata_finalize(&store_state);
if let Err(e) =
dump.write_to_metadata_store(&store, &run_options.run_id.to_string(), "finalize run")
{
tracing::warn!(error = %e, "Failed to write finalize commit to metadata branch");
return;
}

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

@ -13,7 +13,7 @@ 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, RunRecord, RunRecordExt};
use crate::records::{Conclusion, RunRecord};
use fabro_retro::retro::Retro;
/// Derive a PR title from the workflow goal.
@ -144,9 +144,14 @@ fn format_arc_details_section(
// Workflow graph summary — prefer RunRecord's graph, fall back to DOT parsing
if let Some(record) = run_record {
let graph_name = format!("{}.fabro", record.workflow_name());
let node_count = record.node_count();
let edge_count = record.edge_count();
let workflow_name = if record.graph.name.is_empty() {
"unnamed"
} else {
&record.graph.name
};
let graph_name = format!("{workflow_name}.fabro");
let node_count = record.graph.nodes.len();
let edge_count = record.graph.edges.len();
parts.push(String::new());
parts.push(format!(

18
lib/crates/fabro-workflow/src/records/checkpoint.rs
View file

@ -1,17 +1,16 @@
use std::collections::HashMap;
use std::path::Path;
pub use fabro_types::checkpoint::Checkpoint;
use crate::context::Context;
use crate::error::{FabroError, FailureSignature, Result as CrateResult};
use crate::error::{FabroError, Result as CrateResult};
use crate::outcome::Outcome;
pub use fabro_types::checkpoint::Checkpoint;
use fabro_types::failure_signature::FailureSignature;
pub trait CheckpointExt {
#[allow(clippy::too_many_arguments)]
fn from_context(
context: &Context,
current_node: impl Into<String>,
current_node: &str,
completed_nodes: Vec<String>,
node_retries: HashMap<String, u32>,
node_outcomes: HashMap<String, Outcome>,
@ -19,9 +18,8 @@ pub trait CheckpointExt {
loop_failure_signatures: HashMap<FailureSignature, usize>,
restart_failure_signatures: HashMap<FailureSignature, usize>,
node_visits: HashMap<String, usize>,
) -> Self
where
Self: Sized;
) -> Self;
fn save(&self, path: &Path) -> CrateResult<()>;
fn load(path: &Path) -> CrateResult<Self>
where
@ -31,7 +29,7 @@ pub trait CheckpointExt {
impl CheckpointExt for Checkpoint {
fn from_context(
context: &Context,
current_node: impl Into<String>,
current_node: &str,
completed_nodes: Vec<String>,
node_retries: HashMap<String, u32>,
node_outcomes: HashMap<String, Outcome>,
@ -42,7 +40,7 @@ impl CheckpointExt for Checkpoint {
) -> Self {
Self {
timestamp: chrono::Utc::now(),
current_node: current_node.into(),
current_node: current_node.to_string(),
completed_nodes,
node_retries,
context_values: context.snapshot(),

17
lib/crates/fabro-workflow/src/records/conclusion.rs
View file

@ -1,18 +1 @@
use std::path::Path;
pub use fabro_types::conclusion::{Conclusion, StageSummary};
use crate::error::{FabroError, Result as CrateResult};
pub trait ConclusionExt {
fn save(&self, path: &Path) -> CrateResult<()>;
}
impl ConclusionExt for Conclusion {
fn save(&self, path: &Path) -> CrateResult<()> {
let json = serde_json::to_string_pretty(self)
.map_err(|e| FabroError::Checkpoint(format!("conclusion serialize failed: {e}")))?;
std::fs::write(path, json)?;
Ok(())
}
}

6
lib/crates/fabro-workflow/src/records/mod.rs
View file

@ -4,6 +4,6 @@ mod run;
mod start;
pub use checkpoint::{Checkpoint, CheckpointExt};
pub use conclusion::{Conclusion, ConclusionExt, StageSummary};
pub use run::{RunRecord, RunRecordExt};
pub use start::{StartRecord, StartRecordExt};
pub use conclusion::{Conclusion, StageSummary};
pub use run::RunRecord;
pub use start::StartRecord;

50
lib/crates/fabro-workflow/src/records/run.rs
View file

@ -1,51 +1 @@
use std::path::Path;
pub use fabro_types::run::RunRecord;
use crate::error::{FabroError, Result as CrateResult};
const FILE_NAME: &str = "run.json";
pub trait RunRecordExt {
fn save(&self, run_dir: &Path) -> CrateResult<()>;
fn load(run_dir: &Path) -> CrateResult<Self>
where
Self: Sized;
fn workflow_name(&self) -> &str;
fn goal(&self) -> &str;
fn node_count(&self) -> usize;
fn edge_count(&self) -> usize;
}
impl RunRecordExt for RunRecord {
fn save(&self, run_dir: &Path) -> CrateResult<()> {
let json = serde_json::to_string_pretty(self)
.map_err(|e| FabroError::Checkpoint(format!("run record serialize failed: {e}")))?;
std::fs::write(run_dir.join(FILE_NAME), json)?;
Ok(())
}
fn load(run_dir: &Path) -> CrateResult<Self> {
crate::load_json(&run_dir.join(FILE_NAME), "run record")
}
fn workflow_name(&self) -> &str {
if self.graph.name.is_empty() {
"unnamed"
} else {
&self.graph.name
}
}
fn goal(&self) -> &str {
self.graph.goal()
}
fn node_count(&self) -> usize {
self.graph.nodes.len()
}
fn edge_count(&self) -> usize {
self.graph.edges.len()
}
}

18
lib/crates/fabro-workflow/src/records/start.rs
View file

@ -1,19 +1 @@
use std::path::Path;
pub use fabro_types::start::StartRecord;
use crate::error::Result as CrateResult;
const FILE_NAME: &str = "start.json";
pub trait StartRecordExt {
fn load(run_dir: &Path) -> CrateResult<Self>
where
Self: Sized;
}
impl StartRecordExt for StartRecord {
fn load(run_dir: &Path) -> CrateResult<Self> {
crate::load_json(&run_dir.join(FILE_NAME), "start record")
}
}

413
lib/crates/fabro-workflow/src/run_dump.rs Normal file
View file

@ -0,0 +1,413 @@
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_store::{NodeVisitRef, RunState, SlateRunStore};
use crate::git::MetadataStore;
#[derive(Debug, Clone)]
pub struct RunDump {
entries: Vec<RunDumpEntry>,
}
#[derive(Debug, Clone)]
pub struct RunDumpEntry {
path: String,
contents: RunDumpContents,
}
#[derive(Debug, Clone)]
pub enum RunDumpContents {
Text(String),
Json(serde_json::Value),
Bytes(Vec<u8>),
}
impl RunDump {
#[must_use]
pub fn metadata_init(state: &RunState) -> Self {
let mut entries = Vec::new();
if let Some(record) = state.run.as_ref() {
push_json_entry(&mut entries, "run.json", record);
}
if let Some(record) = state.start.as_ref() {
push_json_entry(&mut entries, "start.json", record);
}
if let Some(record) = state.sandbox.as_ref() {
push_json_entry(&mut entries, "sandbox.json", record);
}
Self { entries }
}
#[must_use]
pub fn metadata_checkpoint(state: &RunState) -> Self {
let mut entries = Vec::new();
let mut keys: Vec<_> = state.nodes.keys().collect();
keys.sort();
for (node_id, visit) in keys {
let Some(node) = state.nodes.get(&(node_id.clone(), *visit)) else {
continue;
};
if let Some(prompt) = node.prompt.as_ref() {
entries.push(RunDumpEntry::text(
metadata_node_file_path(node_id, *visit, "prompt.md"),
prompt.clone(),
));
}
if let Some(response) = node.response.as_ref() {
entries.push(RunDumpEntry::text(
metadata_node_file_path(node_id, *visit, "response.md"),
response.clone(),
));
}
if let Some(status) = node.status.as_ref() {
push_json_entry_path(
&mut entries,
metadata_node_file_path(node_id, *visit, "status.json").into(),
status,
);
}
if let Some(provider_used) = node.provider_used.as_ref() {
entries.push(RunDumpEntry::json(
metadata_node_file_path(node_id, *visit, "provider_used.json"),
provider_used.clone(),
));
}
if let Some(diff) = node.diff.as_ref() {
entries.push(RunDumpEntry::text(
metadata_node_file_path(node_id, *visit, "diff.patch"),
diff.clone(),
));
}
if let Some(script_invocation) = node.script_invocation.as_ref() {
entries.push(RunDumpEntry::json(
metadata_node_file_path(node_id, *visit, "script_invocation.json"),
script_invocation.clone(),
));
}
if let Some(script_timing) = node.script_timing.as_ref() {
entries.push(RunDumpEntry::json(
metadata_node_file_path(node_id, *visit, "script_timing.json"),
script_timing.clone(),
));
}
if let Some(parallel_results) = node.parallel_results.as_ref() {
entries.push(RunDumpEntry::json(
metadata_node_file_path(node_id, *visit, "parallel_results.json"),
parallel_results.clone(),
));
}
}
Self { entries }
}
#[must_use]
pub fn metadata_finalize(state: &RunState) -> Self {
let mut dump = Self::metadata_checkpoint(state);
if let Some(retro) = state.retro.as_ref() {
push_json_entry(&mut dump.entries, "retro.json", retro);
}
dump
}
pub async fn store_export(run_store: &SlateRunStore, state: &RunState) -> Result<Self> {
let mut entries = Vec::new();
if let Some(record) = state.run.as_ref() {
push_json_entry(&mut entries, "run.json", record);
}
if let Some(record) = state.start.as_ref() {
push_json_entry(&mut entries, "start.json", record);
}
if let Some(record) = state.status.as_ref() {
push_json_entry(&mut entries, "status.json", record);
}
if let Some(record) = state.checkpoint.as_ref() {
push_json_entry(&mut entries, "checkpoint.json", record);
}
if let Some(record) = state.conclusion.as_ref() {
push_json_entry(&mut entries, "conclusion.json", record);
}
if let Some(record) = state.retro.as_ref() {
push_json_entry(&mut entries, "retro.json", record);
}
if let Some(graph_source) = state.graph_source.as_ref() {
entries.push(RunDumpEntry::text("graph.fabro", graph_source.clone()));
}
if let Some(record) = state.sandbox.as_ref() {
push_json_entry(&mut entries, "sandbox.json", record);
}
let mut node_keys: Vec<_> = state.nodes.keys().cloned().collect();
node_keys.sort();
for (node_id, visit) in &node_keys {
let node = &state.nodes[&(node_id.clone(), *visit)];
let node_id_segment = validate_single_path_segment("node id", node_id)?;
let base = PathBuf::from("nodes")
.join(node_id_segment)
.join(format!("visit-{visit}"));
if let Some(prompt) = node.prompt.as_ref() {
entries.push(RunDumpEntry::text_path(
base.join("prompt.md"),
prompt.clone(),
));
}
if let Some(response) = node.response.as_ref() {
entries.push(RunDumpEntry::text_path(
base.join("response.md"),
response.clone(),
));
}
if let Some(status) = node.status.as_ref() {
push_json_entry_path(&mut entries, base.join("status.json"), status);
}
if let Some(stdout) = node.stdout.as_ref() {
entries.push(RunDumpEntry::text_path(
base.join("stdout.log"),
stdout.clone(),
));
}
if let Some(stderr) = node.stderr.as_ref() {
entries.push(RunDumpEntry::text_path(
base.join("stderr.log"),
stderr.clone(),
));
}
}
if let Some(prompt) = state.retro_prompt.as_ref() {
entries.push(RunDumpEntry::text("retro/prompt.md", prompt.clone()));
}
if let Some(response) = state.retro_response.as_ref() {
entries.push(RunDumpEntry::text("retro/response.md", response.clone()));
}
let mut events_jsonl = Vec::new();
for event in run_store.list_events().await? {
serde_json::to_writer(&mut events_jsonl, &event)?;
events_jsonl.write_all(b"\n")?;
}
entries.push(RunDumpEntry::bytes("events.jsonl", events_jsonl));
for (seq, checkpoint) in &state.checkpoints {
push_json_entry_path(
&mut entries,
PathBuf::from("checkpoints").join(format!("{seq:04}.json")),
checkpoint,
);
}
for artifact_id in run_store.list_artifact_values().await? {
let artifact_id_segment = validate_single_path_segment("artifact id", &artifact_id)?;
let value = run_store
.get_artifact_value(&artifact_id)
.await?
.with_context(|| {
format!("artifact value {artifact_id:?} is missing from the store")
})?;
entries.push(RunDumpEntry::json_path(
PathBuf::from("artifacts")
.join("values")
.join(format!("{}.json", artifact_id_segment.display())),
value,
));
}
for (node_id, visit, filename) in run_store.list_all_assets().await? {
let node_id_segment = validate_single_path_segment("node id", &node_id)?;
let filename_path = validate_relative_path("asset filename", &filename)?;
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"
)
})?;
entries.push(RunDumpEntry::bytes_path(
PathBuf::from("artifacts")
.join("nodes")
.join(node_id_segment)
.join(format!("visit-{visit}"))
.join(filename_path),
data.to_vec(),
));
}
Ok(Self { entries })
}
pub fn entries(&self) -> &[RunDumpEntry] {
&self.entries
}
#[must_use]
pub fn file_count(&self) -> usize {
self.entries.len()
}
pub fn write_to_dir(&self, root: &Path) -> Result<usize> {
for entry in &self.entries {
entry.write_to_dir(root)?;
}
Ok(self.file_count())
}
pub fn write_to_metadata_store(
&self,
store: &MetadataStore,
run_id: &str,
message: &str,
) -> Result<()> {
let git_entries = self.git_entries()?;
let refs: Vec<(&str, &[u8])> = git_entries
.iter()
.map(|(path, bytes)| (path.as_str(), bytes.as_slice()))
.collect();
store.write_files(run_id, &refs, message)?;
Ok(())
}
pub fn git_entries(&self) -> Result<Vec<(String, Vec<u8>)>> {
self.entries
.iter()
.map(|entry| Ok((entry.path.clone(), entry.contents.to_bytes()?)))
.collect()
}
}
impl RunDumpEntry {
fn text(path: impl Into<String>, contents: String) -> Self {
Self {
path: path.into(),
contents: RunDumpContents::Text(contents),
}
}
fn text_path(path: PathBuf, contents: String) -> Self {
Self {
path: path_to_string(path),
contents: RunDumpContents::Text(contents),
}
}
fn json(path: impl Into<String>, contents: serde_json::Value) -> Self {
Self {
path: path.into(),
contents: RunDumpContents::Json(contents),
}
}
fn json_path(path: PathBuf, contents: serde_json::Value) -> Self {
Self {
path: path_to_string(path),
contents: RunDumpContents::Json(contents),
}
}
fn bytes(path: impl Into<String>, contents: Vec<u8>) -> Self {
Self {
path: path.into(),
contents: RunDumpContents::Bytes(contents),
}
}
fn bytes_path(path: PathBuf, contents: Vec<u8>) -> Self {
Self {
path: path_to_string(path),
contents: RunDumpContents::Bytes(contents),
}
}
fn write_to_dir(&self, root: &Path) -> Result<()> {
let relative = validate_relative_path("run dump path", &self.path)?;
let path = root.join(relative);
ensure_parent_dir(&path)?;
std::fs::write(&path, self.contents.to_bytes()?)
.with_context(|| format!("failed to write {}", path.display()))?;
Ok(())
}
}
impl RunDumpContents {
fn to_bytes(&self) -> Result<Vec<u8>> {
match self {
Self::Text(value) => Ok(value.as_bytes().to_vec()),
Self::Json(value) => Ok(serde_json::to_vec_pretty(value)?),
Self::Bytes(value) => Ok(value.clone()),
}
}
}
fn push_json_entry<T>(entries: &mut Vec<RunDumpEntry>, path: &str, value: &T)
where
T: serde::Serialize,
{
if let Ok(value) = serde_json::to_value(value) {
entries.push(RunDumpEntry::json(path, value));
}
}
fn push_json_entry_path<T>(entries: &mut Vec<RunDumpEntry>, path: PathBuf, value: &T)
where
T: serde::Serialize,
{
if let Ok(value) = serde_json::to_value(value) {
entries.push(RunDumpEntry::json_path(path, value));
}
}
fn metadata_node_file_path(node_id: &str, visit: u32, filename: &str) -> String {
if visit <= 1 {
format!("nodes/{node_id}/{filename}")
} else {
format!("nodes/{node_id}-visit_{visit}/{filename}")
}
}
fn path_to_string(path: PathBuf) -> String {
path.to_string_lossy().into_owned()
}
fn validate_single_path_segment(kind: &str, value: &str) -> Result<PathBuf> {
let path = validate_relative_path(kind, value)?;
if path.components().count() != 1 {
bail!("{kind} {value:?} must be a single path segment");
}
Ok(path)
}
fn validate_relative_path(kind: &str, value: &str) -> Result<PathBuf> {
let mut normalized = PathBuf::new();
for component in Path::new(value).components() {
match component {
Component::Normal(part) => normalized.push(part),
Component::CurDir => {}
Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
bail!("{kind} {value:?} must be a relative path without '..'");
}
}
}
if normalized.as_os_str().is_empty() {
bail!("{kind} {value:?} must not be empty");
}
Ok(normalized)
}
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(())
}

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

@ -7,7 +7,6 @@ use fabro_store::{ListRunsQuery, SlateStore};
use fabro_types::RunId;
use serde::Serialize;
use crate::records::{RunRecord, RunRecordExt, StartRecord, StartRecordExt};
use crate::run_status::{RunStatus, StatusReason};
#[derive(Debug, Clone, Serialize)]
@ -61,15 +60,7 @@ pub fn default_runs_base() -> PathBuf {
runs_base(&default_storage_dir())
}
pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
scan_runs_inner(base, true)
}
fn scan_runs_without_status(base: &Path) -> Result<Vec<RunInfo>> {
scan_runs_inner(base, false)
}
fn scan_runs_inner(base: &Path, include_status: bool) -> Result<Vec<RunInfo>> {
fn scan_orphan_runs(base: &Path) -> Result<Vec<RunInfo>> {
let entries = match std::fs::read_dir(base) {
Ok(entries) => entries,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
@ -85,85 +76,40 @@ fn scan_runs_inner(base: &Path, include_status: bool) -> Result<Vec<RunInfo>> {
}
let dir_name = entry.file_name().to_string_lossy().to_string();
let mtime_dt = entry
.metadata()
.ok()
.and_then(|m| m.modified().ok())
.map(|time| -> DateTime<Utc> { time.into() });
let mtime = mtime_dt.map(|dt| dt.to_rfc3339()).unwrap_or_default();
if let Ok(record) = RunRecord::load(&path) {
let created_at = record.created_at;
let start_time_dt = StartRecord::load(&path)
.map(|s| s.start_time)
.unwrap_or(created_at);
let start_time = start_time_dt.to_rfc3339();
let workflow_name = record.workflow_name().to_string();
let goal = record.goal().to_string();
let status_info = if include_status {
read_status(&path)
} else {
StatusInfo::simple(RunStatus::Dead)
};
let run_id = std::fs::read_to_string(path.join("id.txt"))
.ok()
.and_then(|s| parse_run_id(&s))
.or_else(|| parse_run_id(&dir_name));
let Some(run_id) = run_id else {
continue;
};
runs.push(RunInfo {
run_id: record.run_id,
dir_name,
workflow_name,
workflow_slug: record.workflow_slug,
status: status_info.status,
status_reason: status_info.reason,
start_time,
labels: record.labels,
duration_ms: status_info.duration_ms,
total_cost: status_info.total_cost,
host_repo_path: record.host_repo_path,
start_time_dt: Some(created_at),
end_time: status_info.end_time,
path,
goal,
is_orphan: false,
});
} else {
let mtime_dt = entry
.metadata()
.ok()
.and_then(|m| m.modified().ok())
.map(|time| -> DateTime<Utc> { time.into() });
let mtime = mtime_dt.map(|dt| dt.to_rfc3339()).unwrap_or_default();
let run_id = std::fs::read_to_string(path.join("id.txt"))
.ok()
.and_then(|s| parse_run_id(&s))
.or_else(|| parse_run_id(&dir_name));
let Some(run_id) = run_id else {
continue;
};
let status_info = if include_status {
read_status(&path)
} else {
StatusInfo::simple(RunStatus::Dead)
};
let is_orphan = !include_status || matches!(status_info.status, RunStatus::Dead);
runs.push(RunInfo {
run_id,
dir_name,
workflow_name: if is_orphan {
"[no run record]"
} else {
"[starting]"
}
.to_string(),
workflow_slug: None,
status: status_info.status,
status_reason: status_info.reason,
start_time: mtime,
labels: HashMap::new(),
duration_ms: status_info.duration_ms,
total_cost: status_info.total_cost,
host_repo_path: None,
start_time_dt: mtime_dt,
end_time: status_info.end_time,
path,
goal: String::new(),
is_orphan,
});
}
let status_info = StatusInfo::simple(RunStatus::Dead);
runs.push(RunInfo {
run_id,
dir_name,
workflow_name: "[no run record]".to_string(),
workflow_slug: None,
status: status_info.status,
status_reason: status_info.reason,
start_time: mtime,
labels: HashMap::new(),
duration_ms: status_info.duration_ms,
total_cost: status_info.total_cost,
host_repo_path: None,
start_time_dt: mtime_dt,
end_time: status_info.end_time,
path,
goal: String::new(),
is_orphan: true,
});
}
runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt));
@ -186,7 +132,7 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result<Vec<R
.keys()
.copied()
.collect::<std::collections::HashSet<_>>();
for run in scan_runs_without_status(base)?
for run in scan_orphan_runs(base)?
.into_iter()
.filter(|run| run.is_orphan && !store_run_ids.contains(&run.run_id))
{
@ -260,11 +206,6 @@ impl StatusInfo {
}
}
fn read_status(run_dir: &Path) -> StatusInfo {
let _ = run_dir;
StatusInfo::simple(RunStatus::Dead)
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StatusFilter {
RunningOnly,
@ -309,69 +250,6 @@ pub fn filter_runs(
.collect()
}
pub fn find_run_by_prefix(base: &Path, prefix: &str) -> Result<PathBuf> {
let runs = scan_runs(base).context("Failed to scan runs")?;
let matches: Vec<_> = runs
.iter()
.filter(|run| run_id_matches(run.run_id, prefix))
.collect();
match matches.len() {
0 => bail!("No run found matching prefix '{prefix}'"),
1 => Ok(matches[0].path.clone()),
count => {
let ids: Vec<String> = matches.iter().map(|run| run.run_id.to_string()).collect();
bail!(
"Ambiguous prefix '{prefix}': {count} runs match: {}",
ids.join(", ")
)
}
}
}
pub fn resolve_run(base: &Path, identifier: &str) -> Result<RunInfo> {
let runs = scan_runs(base).context("Failed to scan runs")?;
let id_matches: Vec<_> = runs
.iter()
.filter(|run| run_id_matches(run.run_id, identifier))
.collect();
match id_matches.len() {
1 => return Ok(id_matches[0].clone()),
count if count > 1 => {
let ids: Vec<String> = id_matches
.iter()
.map(|run| run.run_id.to_string())
.collect();
bail!(
"Ambiguous prefix '{identifier}': {count} runs match: {}",
ids.join(", ")
)
}
_ => {}
}
let id_lower = identifier.to_lowercase();
let id_collapsed = collapse_separators(&id_lower);
let workflow_match = runs.iter().filter(|run| !run.is_orphan).find(|run| {
if let Some(slug) = &run.workflow_slug {
if slug.to_lowercase() == id_lower {
return true;
}
}
let name_lower = run.workflow_name.to_lowercase();
name_lower.contains(&id_lower) || collapse_separators(&name_lower).contains(&id_collapsed)
});
match workflow_match {
Some(run) => Ok(run.clone()),
None => {
bail!("No run found matching '{identifier}' (tried run ID prefix and workflow name)")
}
}
}
pub async fn resolve_run_combined(
store: &SlateStore,
base: &Path,
@ -449,7 +327,7 @@ mod tests {
use super::scan_runs_combined;
use crate::event::{WorkflowRunEvent, append_workflow_event};
use crate::records::{RunRecord, RunRecordExt};
use crate::records::RunRecord;
fn memory_store() -> StoreHandle {
Arc::new(SlateStore::new(
@ -478,13 +356,11 @@ mod tests {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join(fixtures::RUN_1.to_string());
std::fs::create_dir_all(&run_dir).unwrap();
let run_record = sample_run_record();
run_record.save(&run_dir).unwrap();
std::fs::write(run_dir.join("id.txt"), format!("{}\n", fixtures::RUN_1)).unwrap();
let store = memory_store();
let run_dir_string = run_dir.to_string_lossy().to_string();
let run_record = sample_run_record();
let run_store = store
.create_run(
&fixtures::RUN_1,

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

@ -1,25 +1,3 @@
use std::path::Path;
pub use fabro_types::status::{
InvalidTransition, ParseRunStatusError, RunStatus, RunStatusRecord, StatusReason,
};
pub trait RunStatusRecordExt {
fn save(&self, path: &Path) -> std::io::Result<()>;
fn load(path: &Path) -> std::io::Result<Self>
where
Self: Sized;
}
impl RunStatusRecordExt for RunStatusRecord {
fn save(&self, path: &Path) -> std::io::Result<()> {
let json = serde_json::to_string_pretty(self).map_err(std::io::Error::other)?;
std::fs::write(path, json)
}
fn load(path: &Path) -> std::io::Result<Self> {
let data = std::fs::read_to_string(path)?;
serde_json::from_str(&data)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))
}
}