diff --git a/lib/crates/fabro-cli/src/commands/pr/create.rs b/lib/crates/fabro-cli/src/commands/pr/create.rs index 847da5b24..e69667937 100644 --- a/lib/crates/fabro-cli/src/commands/pr/create.rs +++ b/lib/crates/fabro-cli/src/commands/pr/create.rs @@ -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, diff --git a/lib/crates/fabro-cli/src/commands/run/launcher.rs b/lib/crates/fabro-cli/src/commands/run/launcher.rs index abdedc447..d8fc34606 100644 --- a/lib/crates/fabro-cli/src/commands/run/launcher.rs +++ b/lib/crates/fabro-cli/src/commands/run/launcher.rs @@ -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 Option { - 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, diff --git a/lib/crates/fabro-cli/src/commands/run/rewind.rs b/lib/crates/fabro-cli/src/commands/run/rewind.rs index d40a28b3b..2bf90a1a8 100644 --- a/lib/crates/fabro-cli/src/commands/run/rewind.rs +++ b/lib/crates/fabro-cli/src/commands/run/rewind.rs @@ -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")?; diff --git a/lib/crates/fabro-cli/src/commands/run/wait.rs b/lib/crates/fabro-cli/src/commands/run/wait.rs index 603ba280a..bb05069a8 100644 --- a/lib/crates/fabro-cli/src/commands/run/wait.rs +++ b/lib/crates/fabro-cli/src/commands/run/wait.rs @@ -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::( + &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::(&data) + .map(|record| record.status) + .unwrap_or(RunStatus::Dead), Err(_) => RunStatus::Dead, }; assert_eq!(status, RunStatus::Dead); diff --git a/lib/crates/fabro-cli/src/commands/store/dump.rs b/lib/crates/fabro-cli/src/commands/store/dump.rs index 3b264f7a3..848077e5f 100644 --- a/lib/crates/fabro-cli/src/commands/store/dump.rs +++ b/lib/crates/fabro-cli/src/commands/store/dump.rs @@ -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 { 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 { - 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 { - 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 { - 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(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::*; diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index 884c76601..c543ca966 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/lib/crates/fabro-workflow/src/git.rs @@ -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)> { result } -pub fn scan_node_files_from_state(state: &RunState) -> Vec<(String, Vec)> { - 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")); diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index ad78f076d..026b00bf9 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/lib/crates/fabro-workflow/src/lib.rs @@ -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; diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index 9350bdf8e..99a1b9a79 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -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 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 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() diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index 9cedb99b8..e2f69d139 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -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( diff --git a/lib/crates/fabro-workflow/src/pipeline/finalize.rs b/lib/crates/fabro-workflow/src/pipeline/finalize.rs index 25cdad78a..e1477c833 100644 --- a/lib/crates/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/crates/fabro-workflow/src/pipeline/finalize.rs @@ -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; } diff --git a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs index ffc84fef1..5784ad737 100644 --- a/lib/crates/fabro-workflow/src/pipeline/pull_request.rs +++ b/lib/crates/fabro-workflow/src/pipeline/pull_request.rs @@ -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!( diff --git a/lib/crates/fabro-workflow/src/records/checkpoint.rs b/lib/crates/fabro-workflow/src/records/checkpoint.rs index fee269bc5..7ccb5bf45 100644 --- a/lib/crates/fabro-workflow/src/records/checkpoint.rs +++ b/lib/crates/fabro-workflow/src/records/checkpoint.rs @@ -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, + current_node: &str, completed_nodes: Vec, node_retries: HashMap, node_outcomes: HashMap, @@ -19,9 +18,8 @@ pub trait CheckpointExt { loop_failure_signatures: HashMap, restart_failure_signatures: HashMap, node_visits: HashMap, - ) -> Self - where - Self: Sized; + ) -> Self; + fn save(&self, path: &Path) -> CrateResult<()>; fn load(path: &Path) -> CrateResult where @@ -31,7 +29,7 @@ pub trait CheckpointExt { impl CheckpointExt for Checkpoint { fn from_context( context: &Context, - current_node: impl Into, + current_node: &str, completed_nodes: Vec, node_retries: HashMap, node_outcomes: HashMap, @@ -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(), diff --git a/lib/crates/fabro-workflow/src/records/conclusion.rs b/lib/crates/fabro-workflow/src/records/conclusion.rs index 42027deb2..6b98dcf0f 100644 --- a/lib/crates/fabro-workflow/src/records/conclusion.rs +++ b/lib/crates/fabro-workflow/src/records/conclusion.rs @@ -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(()) - } -} diff --git a/lib/crates/fabro-workflow/src/records/mod.rs b/lib/crates/fabro-workflow/src/records/mod.rs index f1c1938de..9c2af89c1 100644 --- a/lib/crates/fabro-workflow/src/records/mod.rs +++ b/lib/crates/fabro-workflow/src/records/mod.rs @@ -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; diff --git a/lib/crates/fabro-workflow/src/records/run.rs b/lib/crates/fabro-workflow/src/records/run.rs index 415847820..eb2dd8789 100644 --- a/lib/crates/fabro-workflow/src/records/run.rs +++ b/lib/crates/fabro-workflow/src/records/run.rs @@ -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 - 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 { - 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() - } -} diff --git a/lib/crates/fabro-workflow/src/records/start.rs b/lib/crates/fabro-workflow/src/records/start.rs index 9a148a3f5..7968868bb 100644 --- a/lib/crates/fabro-workflow/src/records/start.rs +++ b/lib/crates/fabro-workflow/src/records/start.rs @@ -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 - where - Self: Sized; -} - -impl StartRecordExt for StartRecord { - fn load(run_dir: &Path) -> CrateResult { - crate::load_json(&run_dir.join(FILE_NAME), "start record") - } -} diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs new file mode 100644 index 000000000..d80452931 --- /dev/null +++ b/lib/crates/fabro-workflow/src/run_dump.rs @@ -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, +} + +#[derive(Debug, Clone)] +pub struct RunDumpEntry { + path: String, + contents: RunDumpContents, +} + +#[derive(Debug, Clone)] +pub enum RunDumpContents { + Text(String), + Json(serde_json::Value), + Bytes(Vec), +} + +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 { + 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 { + 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)>> { + self.entries + .iter() + .map(|entry| Ok((entry.path.clone(), entry.contents.to_bytes()?))) + .collect() + } +} + +impl RunDumpEntry { + fn text(path: impl Into, 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, 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, contents: Vec) -> Self { + Self { + path: path.into(), + contents: RunDumpContents::Bytes(contents), + } + } + + fn bytes_path(path: PathBuf, contents: Vec) -> 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> { + 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(entries: &mut Vec, 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(entries: &mut Vec, 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 { + 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 { + 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(()) +} diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index c7ff56fa4..f7c6dfdb5 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -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> { - scan_runs_inner(base, true) -} - -fn scan_runs_without_status(base: &Path) -> Result> { - scan_runs_inner(base, false) -} - -fn scan_runs_inner(base: &Path, include_status: bool) -> Result> { +fn scan_orphan_runs(base: &Path) -> Result> { 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> { } 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 { 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 { 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>(); - 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 { - 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 = 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 { - 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 = 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, diff --git a/lib/crates/fabro-workflow/src/run_status.rs b/lib/crates/fabro-workflow/src/run_status.rs index 3b6a1b5f9..b5e84c110 100644 --- a/lib/crates/fabro-workflow/src/run_status.rs +++ b/lib/crates/fabro-workflow/src/run_status.rs @@ -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 - 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 { - 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)) - } -}