diff --git a/docs/execution/checkpoints.mdx b/docs/execution/checkpoints.mdx index 94a6ea96b..891fe474c 100644 --- a/docs/execution/checkpoints.mdx +++ b/docs/execution/checkpoints.mdx @@ -45,7 +45,6 @@ The metadata branch (`fabro/meta/{run_id}`) is an orphan branch that stores stru - **`run.json`** — Run record: run ID, created_at, config, graph, workflow slug, working directory, host repo path, base branch, labels - **`start.json`** — Start record: run ID, start time, run branch, base SHA -- **`graph.fabro`** — The workflow Graphviz source as it was parsed After each node, the metadata branch is updated with: @@ -115,7 +114,7 @@ This reads the checkpoint, run record, and Graphviz graph from the metadata bran 1. Fabro reads `checkpoint.json` from the metadata branch -2. Reads `run.json` and `graph.fabro` to reconstruct the workflow +2. Reads `run.json` to reconstruct the workflow graph and config 3. Creates a fresh worktree attached to the existing run branch 4. Restores the full context, completed node list, retry counts, and failure signatures 5. If the checkpointed node used `full` fidelity, downgrades the first resumed node to `summary:high` (since the original conversation thread no longer exists in memory) @@ -198,4 +197,4 @@ It is skipped when: - The working directory has uncommitted changes - The working directory is not a Git repository -- The run uses `--dry-run` \ No newline at end of file +- The run uses `--dry-run` diff --git a/docs/reference/run-directory.mdx b/docs/reference/run-directory.mdx index 48b1115ca..d54da5686 100644 --- a/docs/reference/run-directory.mdx +++ b/docs/reference/run-directory.mdx @@ -21,9 +21,9 @@ The naming format is `YYYYMMDD-{run_id}`, where `run_id` is the ULID assigned to |---|---|---|---| | `run.json` | JSON | Run create | Run metadata — `run_id`, `created_at`, `config` (FabroConfig), `graph` (Graph), `workflow_slug`, `working_directory`, `host_repo_path`, `base_branch`, `labels` | | `start.json` | JSON | Run start | Start metadata — `run_id`, `start_time`, `run_branch`, `base_sha` | -| `graph.fabro` | Graphviz | Run start | Copy of the workflow graph | +| `workflow.fabro` | Graphviz | Run create | Copy of the original workflow graph when the raw DOT source is available | | `run.pid` | Text | Run start | Process ID of the running CLI process. Presence indicates the run is active; an orphaned file indicates a crash. | -| `run.toml` | TOML | Run start | Copy of the original workflow file (only when the workflow is defined in TOML) | +| `workflow.toml` | TOML | Run create | Copy of the original workflow file (only when the workflow is defined in TOML) | | `progress.jsonl` | JSONL | Continuous | Event stream — one JSON object per line for every significant event (stage starts, completions, tool calls, retries, etc.). See [Observability](/execution/observability) for the full event catalog. | | `live.json` | JSON | Continuous | Current execution state snapshot, overwritten on each event. Used for live monitoring. | | `checkpoint.json` | JSON | After each node | Crash recovery state — `current_node`, `completed_nodes`, `node_retries`, `context_values`, `node_outcomes`, `next_node_id`, `git_commit_sha`, failure signatures. See [Checkpoints](/execution/checkpoints). | @@ -88,9 +88,9 @@ fabro ps --filter workflow=my-workflow ├── 20260307-01JQXYZ123ABC456DEF789/ # One directory per run │ ├── run.json │ ├── start.json -│ ├── graph.fabro +│ ├── workflow.fabro │ ├── run.pid -│ ├── run.toml +│ ├── workflow.toml │ ├── progress.jsonl │ ├── live.json │ ├── checkpoint.json diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index 566912a0d..d71259fbe 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -485,6 +485,42 @@ async fn start_run( info!(run_id = %run_id, "Run queued"); let created_at = chrono::Utc::now(); + let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4())); + if let Err(err) = std::fs::create_dir_all(&run_dir) { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to create run directory: {err}"), + ) + .into_response(); + } + + let run_record = fabro_workflows::run_record::RunRecord { + run_id: run_id.clone(), + created_at, + config: fabro_config::config::FabroConfig { + dry_run: Some(state.dry_run), + hooks: state.hooks.clone(), + sandbox: Some(fabro_config::sandbox::SandboxConfig { + provider: Some("local".to_string()), + ..Default::default() + }), + ..Default::default() + }, + graph: graph.clone(), + workflow_slug: None, + working_directory: std::env::current_dir() + .unwrap_or_else(|_| std::path::PathBuf::from(".")), + host_repo_path: None, + base_branch: None, + labels: std::collections::HashMap::new(), + }; + if let Err(err) = run_record.save(&run_dir) { + return ApiError::new( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Failed to persist run record: {err}"), + ) + .into_response(); + } { let mut runs = state.runs.lock().expect("runs lock poisoned"); @@ -502,7 +538,7 @@ async fn start_run( checkpoint: None, cancel_tx: None, cancel_token: None, - run_dir: None, + run_dir: Some(run_dir), }, ); } @@ -525,12 +561,16 @@ async fn start_run( /// Execute a single run: transitions queued → starting → running → completed/failed/cancelled. async fn execute_run(state: Arc, run_id: String) { // Transition to Starting and set up cancel infrastructure - let (cancel_rx, graph, created_at) = { + let (cancel_rx, graph, run_dir) = { let mut runs = state.runs.lock().expect("runs lock poisoned"); let managed_run = match runs.get_mut(&run_id) { Some(r) if r.status == RunStatus::Queued => r, _ => return, }; + let run_dir = match managed_run.run_dir.clone() { + Some(path) => path, + None => return, + }; let (cancel_tx, cancel_rx) = tokio::sync::oneshot::channel::<()>(); let cancel_token = Arc::new(AtomicBool::new(false)); @@ -541,7 +581,7 @@ async fn execute_run(state: Arc, run_id: String) { managed_run.cancel_token = Some(Arc::clone(&cancel_token)); managed_run.event_tx = Some(event_tx); - (cancel_rx, managed_run.graph.clone(), managed_run.created_at) + (cancel_rx, managed_run.graph.clone(), run_dir) }; // Create interviewer, sandbox, engine (this is the "provisioning" phase) @@ -608,34 +648,6 @@ async fn execute_run(state: Arc, run_id: String) { } } - let run_dir = std::env::temp_dir().join(format!("fabro-{}", uuid::Uuid::new_v4())); - std::fs::create_dir_all(&run_dir).expect("failed to create run directory"); - - // Write RunRecord for observability (enables `fabro ps` / `fabro inspect` for API runs). - { - let record_config = fabro_config::config::FabroConfig { - dry_run: Some(state.dry_run), - sandbox: Some(fabro_config::sandbox::SandboxConfig { - provider: Some("local".to_string()), - ..Default::default() - }), - ..Default::default() - }; - let run_record = fabro_workflows::run_record::RunRecord { - run_id: run_id.clone(), - created_at, - config: record_config, - graph: graph.clone(), - workflow_slug: None, - working_directory: std::env::current_dir() - .unwrap_or_else(|_| std::path::PathBuf::from(".")), - host_repo_path: None, - base_branch: None, - labels: std::collections::HashMap::new(), - }; - let _ = run_record.save(&run_dir); - } - let config = RunConfig { run_dir, cancel_token: Some(cancel_token), diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index d83ccf022..0be5a6fd4 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -22,17 +22,16 @@ use indicatif::HumanDuration; use super::detached_support::{DetachedRunBootstrapGuard, DetachedRunCompletionGuard}; use super::run::{ - build_conclusion, build_event_envelope, classify_engine_result, default_run_dir, - emit_run_notice, generate_retro, local_sandbox_with_callback, mint_github_token, - persist_terminal_outcome, prepare_workflow_with_project_config, print_assets, - print_final_output, resolve_daytona_config, resolve_fallback_chain, resolve_model_provider, - resolve_sandbox_provider, resolve_ssh_clone_params, resolve_ssh_config, write_finalize_commit, + build_conclusion, build_event_envelope, cached_graph_path, classify_engine_result, + default_run_dir, emit_run_notice, generate_retro, local_sandbox_with_callback, + mint_github_token, persist_terminal_outcome, prepare_workflow_with_project_config, + print_assets, print_final_output, resolve_daytona_config, resolve_fallback_chain, + resolve_model_provider, resolve_ssh_clone_params, resolve_ssh_config, write_finalize_commit, write_run_config_snapshot, CliSandboxProvider, RunArgs, }; -use crate::commands::shared::{print_diagnostics, tilde_path}; +use crate::commands::shared::tilde_path; use fabro_config::project as project_config; use fabro_config::run as run_config; -use fabro_validate::Severity; use fabro_workflows::devcontainer_bridge; use std::collections::HashMap; use tracing::debug; @@ -227,7 +226,7 @@ async fn prepare_from_checkpoint( fabro_util::run_log::activate(&run_dir.join("cli.log")) .context("Failed to activate per-run log")?; let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; - tokio::fs::write(run_dir.join("graph.fabro"), &source).await?; + tokio::fs::write(cached_graph_path(&run_dir), &source).await?; let run_cfg: Option = run_cfg; write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?; @@ -606,27 +605,7 @@ async fn prepare_from_branch( rec.workflow_slug.clone(), ) } else { - // Fallback: read DOT source from metadata branch - let source = - fabro_workflows::git::MetadataStore::read_graph_dot(&resume_repo_path, &run_id)? - .ok_or_else(|| { - anyhow::anyhow!( - "no run.json or graph.fabro found on metadata branch for run {run_id}" - ) - })?; - let (graph, diagnostics) = - fabro_workflows::workflow::WorkflowBuilder::new().prepare(&source)?; - print_diagnostics(&diagnostics, styles); - if diagnostics.iter().any(|d| d.severity == Severity::Error) { - bail!("Validation failed"); - } - - let sandbox_provider = if args.dry_run { - SandboxProvider::Local - } else { - resolve_sandbox_provider(args.sandbox.map(Into::into), None, run_defaults)? - }; - (graph, source.clone(), None, sandbox_provider, None) + bail!("no run.json found on metadata branch for run {run_id}"); }; eprintln!( @@ -652,7 +631,9 @@ async fn prepare_from_branch( fabro_util::run_log::activate(&run_dir.join("cli.log")) .context("Failed to activate per-run log")?; let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; - tokio::fs::write(run_dir.join("graph.fabro"), &graph_source).await?; + if !graph_source.is_empty() { + tokio::fs::write(cached_graph_path(&run_dir), &graph_source).await?; + } let run_cfg: Option = run_cfg; // Git-branch resume: no original TOML available, skip debug snapshot. write_run_config_snapshot(&run_dir, None).await?; diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index 327f9c4e0..3aa626cba 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -543,6 +543,7 @@ pub(crate) fn resolve_workflow_source( /// Result of workflow preparation (shared between `create` and `run` commands). pub(crate) struct PreparedWorkflow { pub validated: fabro_workflows::pipeline::Validated, + pub raw_source: String, pub run_cfg: Option, pub sandbox_provider: SandboxProvider, pub model: String, @@ -558,9 +559,9 @@ impl PreparedWorkflow { pub fn graph(&self) -> &fabro_graphviz::graph::Graph { self.validated.graph() } - /// Read-through to validated source. + /// Original DOT source as authored on disk, before runtime var expansion. pub fn source(&self) -> &str { - self.validated.source() + &self.raw_source } } @@ -624,14 +625,14 @@ pub(crate) fn prepare_workflow_with_project_config( } // Parse and transform workflow using pipeline functions - let source = read_workflow_file(&dot_path)?; + let raw_source = read_workflow_file(&dot_path)?; let vars = run_cfg .as_ref() .and_then(|c| c.vars.as_ref()) .or(run_defaults.vars.as_ref()); let source = match vars { - Some(vars) => fabro_workflows::vars::expand_vars(&source, vars)?, - None => source, + Some(vars) => fabro_workflows::vars::expand_vars(&raw_source, vars)?, + None => raw_source.clone(), }; let dot_dir = dot_path.parent().unwrap_or(std::path::Path::new(".")); @@ -727,6 +728,7 @@ pub(crate) fn prepare_workflow_with_project_config( Ok(PreparedWorkflow { validated, + raw_source, run_cfg, sandbox_provider, model, @@ -740,7 +742,7 @@ pub(crate) fn prepare_workflow_with_project_config( /// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`. struct RecordBasedRun { graph: fabro_graphviz::graph::Graph, - source: String, + raw_source: String, run_cfg: Option, sandbox_provider: SandboxProvider, model: String, @@ -784,7 +786,7 @@ pub async fn run_from_record( .filter(|s| !s.is_empty()); let record_run = RecordBasedRun { - source: String::new(), // DOT source not needed — graph is already deserialized + raw_source: String::new(), // Raw DOT provenance is best-effort for record-based runs graph: record.graph.clone(), run_cfg: Some(record.config.clone()), sandbox_provider, @@ -840,6 +842,7 @@ pub async fn run_command( ) -> anyhow::Result<()> { let PreparedWorkflow { validated, + raw_source, run_cfg, sandbox_provider, model, @@ -848,11 +851,11 @@ pub async fn run_command( run_defaults, workflow_toml_path, } = prepare_workflow(&args, run_defaults, styles, false)?; - let (graph, source, _diagnostics) = validated.into_parts(); + let (graph, _source, _diagnostics) = validated.into_parts(); let record_run = RecordBasedRun { graph, - source, + raw_source, run_cfg, sandbox_provider, model, @@ -874,7 +877,7 @@ async fn run_command_impl( ) -> anyhow::Result<()> { let ( graph, - source, + raw_source, mut run_cfg, sandbox_provider, model, @@ -885,7 +888,7 @@ async fn run_command_impl( ) = match record_run { Some(rr) => ( rr.graph, - rr.source, + rr.raw_source, rr.run_cfg, rr.sandbox_provider, rr.model, @@ -968,8 +971,8 @@ async fn run_command_impl( }; fabro_util::run_log::activate(&run_dir.join("cli.log")) .context("Failed to activate per-run log")?; - if !from_record { - tokio::fs::write(cached_graph_path(&run_dir), &source).await?; + if !from_record && !raw_source.is_empty() { + tokio::fs::write(cached_graph_path(&run_dir), &raw_source).await?; } let mut status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; diff --git a/lib/crates/fabro-workflows/src/git.rs b/lib/crates/fabro-workflows/src/git.rs index 5e464875f..e281d93f8 100644 --- a/lib/crates/fabro-workflows/src/git.rs +++ b/lib/crates/fabro-workflows/src/git.rs @@ -511,18 +511,6 @@ impl MetadataStore { } } - /// Read the graph source from the metadata branch. Tries `graph.fabro` first, - /// then falls back to `graph.dot` for backward compatibility. Returns `None` if not found. - pub fn read_graph_dot(repo_path: &Path, run_id: &str) -> Result> { - if let Some(bytes) = Self::read_file(repo_path, run_id, "graph.fabro")? { - return Ok(Some(String::from_utf8_lossy(&bytes).to_string())); - } - match Self::read_file(repo_path, run_id, "graph.dot")? { - Some(bytes) => Ok(Some(String::from_utf8_lossy(&bytes).to_string())), - None => Ok(None), - } - } - /// Read an artifact from the metadata branch. Returns `None` if not found. pub fn read_artifact(repo_path: &Path, run_id: &str, key: &str) -> Result>> { Self::read_file(repo_path, run_id, &format!("artifacts/{key}.json")) @@ -627,21 +615,13 @@ mod tests { let store = MetadataStore::new(dir.path(), &GitAuthor::default()); let run_record = br#"{"run_id":"RUN1","created_at":"2025-01-01T00:00:00Z","config":{},"graph":{"name":"test","nodes":{},"edges":[],"attrs":{}},"working_directory":"/tmp"}"#; - let dot = b"digraph { start -> end }"; - store - .init_run("RUN1", &[("run.json", run_record), ("graph.fabro", dot)]) - .unwrap(); + store.init_run("RUN1", &[("run.json", run_record)]).unwrap(); let read_record = MetadataStore::read_run_record(dir.path(), "RUN1") .unwrap() .unwrap(); assert_eq!(read_record.run_id, "RUN1"); assert_eq!(read_record.workflow_name(), "test"); - - let read_dot = MetadataStore::read_graph_dot(dir.path(), "RUN1") - .unwrap() - .unwrap(); - assert_eq!(read_dot, "digraph { start -> end }"); } #[test] diff --git a/lib/crates/fabro-workflows/src/pull_request.rs b/lib/crates/fabro-workflows/src/pull_request.rs index b04f36a39..b18a3aef6 100644 --- a/lib/crates/fabro-workflows/src/pull_request.rs +++ b/lib/crates/fabro-workflows/src/pull_request.rs @@ -211,11 +211,18 @@ fn parse_dot_summary(dot: &str) -> (String, usize, usize) { } } -/// Read the workflow graph source from `run_dir/graph.fabro` (or `graph.dot` fallback). +/// Read the workflow graph source from `run_dir/workflow.fabro`. +/// Falls back to `graph.fabro` / `graph.dot` for older runs. fn read_dot_source(run_dir: &Path) -> Option { - let fabro_path = run_dir.join("graph.fabro"); - if let Ok(content) = std::fs::read_to_string(&fabro_path) { - debug!(path = %fabro_path.display(), "Read workflow graph for PR body"); + let workflow_fabro_path = run_dir.join("workflow.fabro"); + if let Ok(content) = std::fs::read_to_string(&workflow_fabro_path) { + debug!(path = %workflow_fabro_path.display(), "Read workflow graph for PR body"); + return Some(content); + } + + let legacy_fabro_path = run_dir.join("graph.fabro"); + if let Ok(content) = std::fs::read_to_string(&legacy_fabro_path) { + debug!(path = %legacy_fabro_path.display(), "Read workflow graph for PR body (legacy)"); return Some(content); } let dot_path = run_dir.join("graph.dot"); @@ -828,11 +835,19 @@ mod tests { #[test] fn read_dot_source_found() { let tmp = tempfile::tempdir().unwrap(); - std::fs::write(tmp.path().join("graph.fabro"), "digraph test {}").unwrap(); + std::fs::write(tmp.path().join("workflow.fabro"), "digraph test {}").unwrap(); let result = read_dot_source(tmp.path()); assert_eq!(result, Some("digraph test {}".to_string())); } + #[test] + fn read_dot_source_legacy_fabro_fallback() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("graph.fabro"), "digraph legacy {}").unwrap(); + let result = read_dot_source(tmp.path()); + assert_eq!(result, Some("digraph legacy {}".to_string())); + } + #[test] fn read_dot_source_dot_fallback() { let tmp = tempfile::tempdir().unwrap(); diff --git a/lib/crates/fabro-workflows/src/run_fork.rs b/lib/crates/fabro-workflows/src/run_fork.rs index 107202b36..b53b7f9ce 100644 --- a/lib/crates/fabro-workflows/src/run_fork.rs +++ b/lib/crates/fabro-workflows/src/run_fork.rs @@ -49,28 +49,24 @@ pub fn execute_fork( .ensure_branch() .map_err(|e| anyhow::anyhow!("failed to create metadata branch: {e}"))?; - // Read run record, start record, graph, and sandbox from source + // Read run record, start record, and sandbox from source metadata. let source_entries = source_bs - .read_entries(&["run.json", "start.json", "graph.fabro", "sandbox.json"]) + .read_entries(&["run.json", "start.json", "sandbox.json"]) .map_err(|e| anyhow::anyhow!("failed to read source metadata: {e}"))?; let mut run_record_bytes = None; let mut start_record_bytes = None; - let mut graph_bytes = None; let mut sandbox_bytes = None; for (path, data) in source_entries { match path { "run.json" => run_record_bytes = Some(data), "start.json" => start_record_bytes = Some(data), - "graph.fabro" => graph_bytes = Some(data), "sandbox.json" => sandbox_bytes = Some(data), _ => {} } } let run_record_bytes = run_record_bytes.ok_or_else(|| anyhow::anyhow!("source run has no run.json"))?; - let graph_bytes = - graph_bytes.ok_or_else(|| anyhow::anyhow!("source run has no graph.fabro"))?; let now = chrono::Utc::now(); @@ -112,7 +108,6 @@ pub fn execute_fork( // Write all entries to the new metadata branch in a single commit let mut file_entries: Vec<(&str, &[u8])> = vec![ ("run.json", &new_run_record_bytes), - ("graph.fabro", &graph_bytes), ("checkpoint.json", &checkpoint_bytes), ]; if let Some(ref start_record) = new_start_record_bytes { @@ -260,16 +255,11 @@ mod tests { let bs = BranchStore::new(store, &meta_branch, &sig); bs.ensure_branch().unwrap(); - // Write run record, start record, and graph + // Write run record and start record let run_record = make_run_record_json(run_id); let start_record = make_start_record_json(run_id); - let graph = b"digraph { start -> build -> test }"; bs.write_entries( - &[ - ("run.json", &run_record), - ("start.json", &start_record), - ("graph.fabro", graph), - ], + &[("run.json", &run_record), ("start.json", &start_record)], "init run", ) .unwrap(); @@ -330,10 +320,6 @@ mod tests { Some(format!("{}{new_run_id}", crate::git::RUN_BRANCH_PREFIX).as_str()) ); - // Check graph exists - let graph_bytes = bs.read_entry("graph.fabro").unwrap().unwrap(); - assert_eq!(graph_bytes, b"digraph { start -> build -> test }"); - // Check checkpoint matches target (@1 = start) let cp_bytes = bs.read_entry("checkpoint.json").unwrap().unwrap(); let cp: serde_json::Value = serde_json::from_slice(&cp_bytes).unwrap();