diff --git a/docs/agents/outputs.mdx b/docs/agents/outputs.mdx index 978f2e910..fb128488d 100644 --- a/docs/agents/outputs.mdx +++ b/docs/agents/outputs.mdx @@ -183,8 +183,8 @@ Artifact data is persisted on the Git [metadata branch](/execution/checkpoints#m ``` fabro/meta/{run_id} - manifest.json - graph.fabro + run.json + start.json checkpoint.json artifacts/ response.plan.json diff --git a/lib/crates/fabro-api/src/server.rs b/lib/crates/fabro-api/src/server.rs index 4bce2f7bf..566912a0d 100644 --- a/lib/crates/fabro-api/src/server.rs +++ b/lib/crates/fabro-api/src/server.rs @@ -525,7 +525,7 @@ 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) = { + let (cancel_rx, graph, created_at) = { 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, @@ -541,7 +541,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()) + (cancel_rx, managed_run.graph.clone(), managed_run.created_at) }; // Create interviewer, sandbox, engine (this is the "provisioning" phase) @@ -610,6 +610,32 @@ 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/create.rs b/lib/crates/fabro-cli/src/commands/create.rs index bdebc3d46..296c67e44 100644 --- a/lib/crates/fabro-cli/src/commands/create.rs +++ b/lib/crates/fabro-cli/src/commands/create.rs @@ -97,8 +97,8 @@ pub async fn create_run( None, ); - // Serialize the merged run config so the run dir is self-contained (debug artifact). - write_run_config_snapshot(&run_dir, prep.run_cfg.as_ref()).await?; + // Copy the original workflow TOML into the run dir as a debug artifact. + write_run_config_snapshot(&run_dir, prep.workflow_toml_path.as_deref()).await?; // Build normalized config and RunRecord let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")); diff --git a/lib/crates/fabro-cli/src/commands/resume.rs b/lib/crates/fabro-cli/src/commands/resume.rs index 8bf300f6c..d83ccf022 100644 --- a/lib/crates/fabro-cli/src/commands/resume.rs +++ b/lib/crates/fabro-cli/src/commands/resume.rs @@ -209,6 +209,7 @@ async fn prepare_from_checkpoint( let prepared_model = prepared.model; let prepared_provider = prepared.provider; let prepared_run_defaults = prepared.run_defaults; + let workflow_toml_path = prepared.workflow_toml_path; eprintln!( "{} {} from checkpoint {}", @@ -228,7 +229,7 @@ async fn prepare_from_checkpoint( let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; tokio::fs::write(run_dir.join("graph.fabro"), &source).await?; let run_cfg: Option = run_cfg; - write_run_config_snapshot(&run_dir, run_cfg.as_ref()).await?; + write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?; // Write RunRecord for the resumed run { @@ -653,7 +654,8 @@ async fn prepare_from_branch( let status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; tokio::fs::write(run_dir.join("graph.fabro"), &graph_source).await?; let run_cfg: Option = run_cfg; - write_run_config_snapshot(&run_dir, run_cfg.as_ref()).await?; + // Git-branch resume: no original TOML available, skip debug snapshot. + write_run_config_snapshot(&run_dir, None).await?; // Write RunRecord for the resumed run { diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs index aca671088..327f9c4e0 100644 --- a/lib/crates/fabro-cli/src/commands/run.rs +++ b/lib/crates/fabro-cli/src/commands/run.rs @@ -484,8 +484,8 @@ pub(crate) fn local_sandbox_with_callback( Arc::new(env) } -pub(crate) const RUN_GRAPH_FILE: &str = "graph.fabro"; -pub(crate) const RUN_CONFIG_FILE: &str = "run.toml"; +pub(crate) const RUN_GRAPH_FILE: &str = "workflow.fabro"; +pub(crate) const RUN_CONFIG_FILE: &str = "workflow.toml"; pub(crate) fn cached_graph_path(run_dir: &Path) -> PathBuf { run_dir.join(RUN_GRAPH_FILE) @@ -495,19 +495,18 @@ pub(crate) fn cached_run_config_path(run_dir: &Path) -> PathBuf { run_dir.join(RUN_CONFIG_FILE) } -fn serialize_run_config_snapshot(run_cfg: &FabroConfig) -> anyhow::Result { - let mut snapshot = run_cfg.clone(); - snapshot.graph = Some(RUN_GRAPH_FILE.to_string()); - toml::to_string_pretty(&snapshot).context("Failed to serialize run config") -} - +/// Copy the original workflow TOML into the run directory as a debug artifact. +/// Nothing reads this programmatically — execution uses RunRecord. pub(crate) async fn write_run_config_snapshot( run_dir: &Path, - run_cfg: Option<&FabroConfig>, + workflow_toml_path: Option<&Path>, ) -> anyhow::Result<()> { - if let Some(cfg) = run_cfg { - let toml_str = serialize_run_config_snapshot(cfg)?; - tokio::fs::write(cached_run_config_path(run_dir), toml_str).await?; + if let Some(toml_path) = workflow_toml_path { + if toml_path.is_file() { + tokio::fs::copy(toml_path, cached_run_config_path(run_dir)) + .await + .context("Failed to copy workflow TOML to run directory")?; + } } Ok(()) } @@ -550,6 +549,8 @@ pub(crate) struct PreparedWorkflow { pub provider: Option, pub workflow_slug: Option, pub run_defaults: FabroConfig, + /// Resolved TOML path (Some for TOML-based workflows, None for bare .fabro). + pub workflow_toml_path: Option, } impl PreparedWorkflow { @@ -715,6 +716,15 @@ pub(crate) fn prepare_workflow_with_project_config( validated.graph(), ); + let workflow_toml_path = if resolved_workflow_path + .extension() + .is_some_and(|ext| ext == "toml") + { + Some(resolved_workflow_path) + } else { + None + }; + Ok(PreparedWorkflow { validated, run_cfg, @@ -723,9 +733,99 @@ pub(crate) fn prepare_workflow_with_project_config( provider, workflow_slug, run_defaults, + workflow_toml_path, }) } +/// Pre-prepared run state, used to skip workflow preparation in `run_command_impl`. +struct RecordBasedRun { + graph: fabro_graphviz::graph::Graph, + source: String, + run_cfg: Option, + sandbox_provider: SandboxProvider, + model: String, + provider: Option, + workflow_slug: Option, + run_defaults: FabroConfig, + /// Original TOML path for debug snapshot (None for record-based or bare .fabro runs). + workflow_toml_path: Option, +} + +/// Execute a workflow run from a saved RunRecord, bypassing workflow preparation. +/// +/// Used by `run_engine_entrypoint` for detached runs that already have a RunRecord on disk. +pub async fn run_from_record( + record: fabro_workflows::run_record::RunRecord, + run_dir: PathBuf, + run_defaults: FabroConfig, + styles: &'static Styles, + github_app: Option, + git_author: fabro_workflows::git::GitAuthor, +) -> anyhow::Result<()> { + let sandbox_provider = record + .config + .sandbox + .as_ref() + .and_then(|s| s.provider.as_deref()) + .unwrap_or("local") + .parse() + .unwrap_or(SandboxProvider::Local); + let model = record + .config + .llm + .as_ref() + .and_then(|l| l.model.clone()) + .unwrap_or_default(); + let provider = record + .config + .llm + .as_ref() + .and_then(|l| l.provider.clone()) + .filter(|s| !s.is_empty()); + + let record_run = RecordBasedRun { + source: String::new(), // DOT source not needed — graph is already deserialized + graph: record.graph.clone(), + run_cfg: Some(record.config.clone()), + sandbox_provider, + model: model.clone(), + provider: provider.clone(), + workflow_slug: record.workflow_slug.clone(), + run_defaults, + workflow_toml_path: None, // No TOML to copy — config is in RunRecord + }; + + let args = RunArgs { + workflow: None, + run_dir: Some(run_dir), + dry_run: record.config.dry_run_enabled(), + preflight: false, + auto_approve: record.config.auto_approve_enabled(), + goal: record.config.goal.clone(), + goal_file: None, + model: Some(model), + provider, + verbose: record.config.verbose_enabled(), + sandbox: Some(CliSandboxProvider::from(sandbox_provider)), + label: record + .labels + .into_iter() + .map(|(k, v)| format!("{k}={v}")) + .collect(), + no_retro: record.config.no_retro_enabled(), + preserve_sandbox: record + .config + .sandbox + .as_ref() + .and_then(|s| s.preserve) + .unwrap_or(false), + detach: false, + run_id: Some(record.run_id), + }; + + run_command_impl(args, styles, github_app, git_author, Some(record_run)).await +} + /// Execute a full workflow run. /// /// # Errors @@ -740,16 +840,65 @@ pub async fn run_command( ) -> anyhow::Result<()> { let PreparedWorkflow { validated, + run_cfg, + sandbox_provider, + model, + provider, + workflow_slug, + run_defaults, + workflow_toml_path, + } = prepare_workflow(&args, run_defaults, styles, false)?; + let (graph, source, _diagnostics) = validated.into_parts(); + + let record_run = RecordBasedRun { + graph, + source, + run_cfg, + sandbox_provider, + model, + provider, + workflow_slug, + run_defaults, + workflow_toml_path, + }; + + run_command_impl(args, styles, github_app, git_author, Some(record_run)).await +} + +async fn run_command_impl( + args: RunArgs, + styles: &'static Styles, + github_app: Option, + git_author: fabro_workflows::git::GitAuthor, + record_run: Option, +) -> anyhow::Result<()> { + let ( + graph, + source, mut run_cfg, sandbox_provider, model, provider, - workflow_slug: prepared_workflow_slug, + prepared_workflow_slug, run_defaults, - } = prepare_workflow(&args, run_defaults, styles, false)?; - let (graph, source, _diagnostics) = validated.into_parts(); + workflow_toml_path, + ) = match record_run { + Some(rr) => ( + rr.graph, + rr.source, + rr.run_cfg, + rr.sandbox_provider, + rr.model, + rr.provider, + rr.workflow_slug, + rr.run_defaults, + rr.workflow_toml_path, + ), + None => unreachable!("run_command_impl always receives a RecordBasedRun"), + }; - let workflow_path = args.workflow.as_ref().unwrap(); // safe: prepare_workflow validated + // For record-based runs from run_from_record, workflow is None (preparation was skipped). + let from_record = args.workflow.is_none(); // Collect setup commands — they'll be run inside the sandbox let setup_commands: Vec = run_cfg @@ -798,7 +947,13 @@ pub async fn run_command( .run_dir .unwrap_or_else(|| default_run_dir(&run_id, dry_run_flag)); tokio::fs::create_dir_all(&run_dir).await?; - let cached_run_restart = is_cached_run_restart(workflow_path, &run_dir); + let cached_run_restart = if from_record { + // Record-based runs already have RunRecord on disk — skip re-writing. + true + } else { + let workflow_path = args.workflow.as_ref().unwrap(); + is_cached_run_restart(workflow_path, &run_dir) + }; let existing_record = if cached_run_restart { fabro_workflows::run_record::RunRecord::load(&run_dir).ok() } else { @@ -813,18 +968,15 @@ pub async fn run_command( }; fabro_util::run_log::activate(&run_dir.join("cli.log")) .context("Failed to activate per-run log")?; - tokio::fs::write(cached_graph_path(&run_dir), &source).await?; + if !from_record { + tokio::fs::write(cached_graph_path(&run_dir), &source).await?; + } let mut status_guard = DetachedRunBootstrapGuard::arm(&run_dir)?; - // Serialize the merged run config so the run dir is self-contained (debug artifact). - // Skip when the workflow path is already the cached run.toml (i.e. _run_engine - // restart) — create_run already wrote the correct snapshot and re-writing here - // would persist a double-merged config (merge_overlay ran again on load). - let is_cached_snapshot = workflow_path - .file_name() - .is_some_and(|f| f == RUN_CONFIG_FILE); - if !is_cached_snapshot { - write_run_config_snapshot(&run_dir, run_cfg.as_ref()).await?; + // Copy the original workflow TOML as a debug artifact. + // Skip for record-based runs (already exists) and cached restarts. + if !from_record && !cached_run_restart { + write_run_config_snapshot(&run_dir, workflow_toml_path.as_deref()).await?; } // Write RunRecord @@ -2779,26 +2931,28 @@ pub(crate) fn build_event_envelope( mod tests { use super::*; - #[test] - fn serialize_run_config_snapshot_rewrites_graph_path() { - let cfg = FabroConfig { - version: Some(1), - goal: Some("test".to_string()), - graph: Some("workflow.fabro".to_string()), - pull_request: Some(run_config::PullRequestConfig { - enabled: true, - ..Default::default() - }), - ..Default::default() - }; + #[tokio::test] + async fn write_run_config_snapshot_copies_toml_file() { + let dir = tempfile::tempdir().unwrap(); + let toml_content = "version = 1\ngoal = \"test\"\n"; + let toml_path = dir.path().join("original.toml"); + std::fs::write(&toml_path, toml_content).unwrap(); - let pr = cfg.pull_request.clone(); - let mut cfg = cfg; - let serialized = serialize_run_config_snapshot(&mut cfg).unwrap(); - let reparsed = run_config::parse_run_config(&serialized).unwrap(); + let run_dir = dir.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + write_run_config_snapshot(&run_dir, Some(toml_path.as_path())) + .await + .unwrap(); - assert_eq!(reparsed.graph.as_deref(), Some(RUN_GRAPH_FILE)); - assert_eq!(reparsed.pull_request, pr); + let copied = std::fs::read_to_string(run_dir.join(RUN_CONFIG_FILE)).unwrap(); + assert_eq!(copied, toml_content); + } + + #[tokio::test] + async fn write_run_config_snapshot_skips_when_none() { + let dir = tempfile::tempdir().unwrap(); + write_run_config_snapshot(dir.path(), None).await.unwrap(); + assert!(!dir.path().join(RUN_CONFIG_FILE).exists()); } #[test] diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index e22861b35..8128710f6 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -353,59 +353,18 @@ async fn run_engine_entrypoint( return Err(err); } - let workflow_path = { - let config_path = commands::run::cached_run_config_path(&run_dir); - if config_path.exists() { - config_path - } else { - commands::run::cached_graph_path(&run_dir) - } - }; - - let sandbox_provider_str = record - .config - .sandbox - .as_ref() - .and_then(|s| s.provider.as_deref()) - .unwrap_or("local"); - - let run_args = commands::run::RunArgs { - workflow: Some(workflow_path), - run_dir: Some(run_dir.clone()), - dry_run: record.config.dry_run_enabled(), - preflight: false, - auto_approve: record.config.auto_approve_enabled(), - goal: record.config.goal.clone(), - goal_file: None, - model: record.config.llm.as_ref().and_then(|l| l.model.clone()), - provider: record - .config - .llm - .as_ref() - .and_then(|l| l.provider.clone()) - .filter(|s| !s.is_empty()), - verbose: record.config.verbose_enabled(), - sandbox: sandbox_provider_str - .parse::() - .ok() - .map(commands::run::CliSandboxProvider::from), - label: record - .labels - .into_iter() - .map(|(k, v)| format!("{k}={v}")) - .collect(), - no_retro: record.config.no_retro_enabled(), - preserve_sandbox: record - .config - .sandbox - .as_ref() - .and_then(|s| s.preserve) - .unwrap_or(false), - detach: false, - run_id: Some(record.run_id), - }; - - match commands::run::run_command(run_args, cli_config, styles, github_app, git_author).await { + // Use run_from_record: loads config + graph directly from RunRecord, + // skipping prepare_workflow() entirely. No TOML/DOT re-parsing needed. + match commands::run::run_from_record( + record, + run_dir.clone(), + cli_config, + styles, + github_app, + git_author, + ) + .await + { Ok(()) => Ok(()), Err(err) => { let _ = commands::detached_support::persist_detached_failure(