Decompose fabro run into create / start / attach (#116)

This PR decomposes `fabro run` into three composable primitives —
`create`, `start`, and `attach` — following the Docker-style lifecycle
model. Previously, `fabro run` performed everything in a single
monolithic function, and `--detach` was implemented by reconstructing
CLI argv to spawn a child process, which was brittle and hard to extend.
The new architecture cleanly separates concerns: `fabro create`
allocates the run directory and persists a `RunSpec` struct to
`spec.json`; `fabro start` spawns a detached `_run_engine` process (a
hidden internal command that reads `spec.json`) via `setsid`; and `fabro
attach` tails `progress.jsonl` with live rendering and handles
file-based interview IPC. `fabro run` is now a composition of these
three primitives, and `fabro run --detach` simply skips the attach step.

The main rendering work lives in a new `handle_json_line()` method on
`ProgressUI` that parses JSONL envelopes and dispatches to the same
internal rendering methods already used by the in-process event handler.
This preserves 100% rendering fidelity without duplicating
spinner/stage/tool-call logic — the attach loop just feeds file lines
into the same code paths. File-based interview IPC is handled in the
attach loop itself: it watches for `interview_request.json`, prompts the
user via `ConsoleInterviewer`, and writes `interview_response.json` back
for the engine to consume. The `hide_bars`/`show_bars` methods
previously private to `ProgressAwareInterviewer` are promoted to public
methods on `ProgressUI` and reused in both the attach loop and the
existing in-process interviewer.

The old `detach_run()` function in `main.rs`, which reconstructed argv
by string-scanning `std::env::args()`, is deleted entirely and replaced
by the `create` + `start` composition. New tests cover the
`handle_json_line` dispatch paths (stage started/completed, tool calls,
retro events, invalid input) and the CLI argument parsing for the new
command variants.

### Fabro Details

<details>
<summary>Ran 9 stages in 30m 55s for $8.55</summary>

| Stage | Duration | Cost | Retries |
|---|---|---|---|
| start | 0s | – | 0 |
| toolchain | 0s | – | 0 |
| preflight_compile | 1m 11s | – | 0 |
| preflight_lint | 12s | – | 0 |
| implement | 18m 15s | $5.28 | 0 |
| simplify_opus | 10m 31s | $3.27 | 0 |
| simplify_gpt | 0s | – | 0 |
| verify | 17s | – | 0 |
| fmt | 1s | – | 0 |
| **Total** | **30m 55s** | **$8.55** | **0** |

</details>

<details>
<summary>Ran <code>ImplementAndSimplify.fabro</code> (12 nodes and 15
edges)</summary>

```dot
digraph ImplementAndSimplify {
    graph [
        goal="Implement and simplify",
        model_stylesheet="
            * { backend: api; model: claude-opus-4-6;}
        "
    ]
    rankdir=LR

    start [shape=Mdiamond, label="Start"]
    exit  [shape=Msquare, label="Exit"]

    toolchain         [label="Toolchain", shape=parallelogram, script="command -v cargo >/dev/null || { curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y && sudo ln -sf $HOME/.cargo/bin/* /usr/local/bin/; }; cargo --version 2>&1", max_retries=0]
    preflight_compile [label="Preflight Compile", shape=parallelogram, script="cargo check -q --workspace 2>&1", max_retries=0]
    preflight_lint    [label="Preflight Lint", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1", max_retries=0]
    fix_lints         [label="Fix Lints", prompt="The preflight lint step failed. Read the build output from context and fix all clippy lint warnings.", max_visits=3]
    implement         [label="Implement", prompt="Read the plan file referenced in the goal and implement every step. Make all the code changes described in the plan. Use red/green TDD."]
    simplify_opus     [label="Simplify (Opus)", prompt="@prompts/simplify.md"]
    simplify_gpt      [label="Simplify (GPT-54)", prompt="@prompts/simplify.md", model="gpt-54"]
    verify            [label="Verify", shape=parallelogram, script="cargo clippy -q --workspace -- -D warnings 2>&1 && cargo nextest run --cargo-quiet --workspace --status-level fail 2>&1", goal_gate=true, retry_target="fixup"]
    fixup             [label="Fixup", prompt="The verify step failed. Read the build output from context and fix all clippy lint warnings and test failures.", max_visits=3]
    fmt               [label="Format", shape=parallelogram, script="cargo fmt --all 2>&1", max_retries=0]

    start -> toolchain
    toolchain -> preflight_compile [condition="outcome=success"]
    toolchain -> exit
    preflight_compile -> preflight_lint [condition="outcome=success"]
    preflight_compile -> exit
    preflight_lint -> implement [condition="outcome=success"]
    preflight_lint -> fix_lints
    fix_lints -> preflight_lint
    implement -> simplify_opus -> simplify_gpt -> verify
    verify -> fmt   [condition="outcome=success"]
    verify -> fixup
    fixup -> verify
    fmt -> exit
}

```

</details>

⚒️ Generated with [Fabro](https://fabro.sh)

---------

Co-authored-by: Fabro <noreply@fabro.sh>
Co-authored-by: Bryan Helmkamp <bryan@brynary.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
brynary-fabro[bot] 2026-03-20 14:08:54 -04:00 committed by GitHub
parent bbfea71bb9
commit 3680a57a25
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
14 changed files with 1768 additions and 175 deletions

1
Cargo.lock generated
View file

@ -1514,6 +1514,7 @@ dependencies = [
"serde_json",
"tempfile",
"tokio",
"tracing",
]
[[package]]

View file

@ -0,0 +1,221 @@
use std::io::{BufRead, BufReader, IsTerminal};
use std::path::Path;
use std::process::ExitCode;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use anyhow::{bail, Result};
use fabro_interview::ConsoleInterviewer;
use fabro_util::terminal::Styles;
use super::run_progress;
/// Attach to a running (or finished) workflow run, rendering progress live.
///
/// Returns exit code 0 for success/partial_success, 1 otherwise.
pub async fn attach_run(
run_dir: &Path,
kill_on_detach: bool,
styles: &'static Styles,
) -> Result<ExitCode> {
let progress_path = run_dir.join("progress.jsonl");
let conclusion_path = run_dir.join("conclusion.json");
let interview_request_path = run_dir.join("interview_request.json");
let interview_response_path = run_dir.join("interview_response.json");
let pid_path = run_dir.join("run.pid");
let is_tty = std::io::stderr().is_terminal();
let verbose = fabro_workflows::run_spec::RunSpec::load(run_dir)
.map(|spec| spec.verbose)
.unwrap_or(false);
let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose);
// Install Ctrl+C handler
let cancelled = Arc::new(AtomicBool::new(false));
{
let cancelled = Arc::clone(&cancelled);
tokio::spawn(async move {
let _ = tokio::signal::ctrl_c().await;
cancelled.store(true, Ordering::Relaxed);
});
}
// Wait for progress.jsonl to appear
let mut wait_count = 0;
while !progress_path.exists() {
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
wait_count += 1;
if wait_count > 100 {
bail!(
"Timed out waiting for progress.jsonl to appear in {}",
run_dir.display()
);
}
if cancelled.load(Ordering::Relaxed) {
return Ok(ExitCode::from(0));
}
}
let file = std::fs::File::open(&progress_path)?;
let mut reader = BufReader::new(file);
let mut line = String::new();
let mut cached_pid: Option<u32> = None;
loop {
if cancelled.load(Ordering::Relaxed) {
if kill_on_detach {
// Kill the engine process
kill_engine(&pid_path);
// Wait briefly for conclusion
for _ in 0..20 {
if conclusion_path.exists() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
} else {
eprintln!("Detached from run (engine continues in background)");
}
break;
}
// Read new lines from progress.jsonl
loop {
line.clear();
let bytes_read = reader.read_line(&mut line)?;
if bytes_read == 0 {
break;
}
let trimmed = line.trim();
if !trimmed.is_empty() {
progress_ui.handle_json_line(trimmed);
}
}
// Check for interview request
if interview_request_path.exists() {
if let Ok(request_data) = std::fs::read_to_string(&interview_request_path) {
// Delete the request file immediately to prevent re-prompting
let _ = std::fs::remove_file(&interview_request_path);
if let Ok(question) =
serde_json::from_str::<fabro_interview::Question>(&request_data)
{
// Hide progress bars during interview
progress_ui.hide_bars();
// Prompt user via ConsoleInterviewer
let interviewer = ConsoleInterviewer::new(styles);
let answer = fabro_interview::Interviewer::ask(&interviewer, question).await;
// Write response
if let Ok(response_json) = serde_json::to_string_pretty(&answer) {
let _ = std::fs::write(&interview_response_path, response_json);
}
// Show progress bars again
progress_ui.show_bars();
}
}
}
// Check if run is complete
if conclusion_path.exists() {
// Drain any remaining lines
drain_remaining(&mut reader, &mut line, &mut progress_ui);
break;
}
// Check if engine process is still alive (cache PID after first read)
let engine_alive = match cached_pid {
Some(pid) => process_alive(pid),
None => {
if let Ok(pid_str) = std::fs::read_to_string(&pid_path) {
if let Ok(pid) = pid_str.trim().parse::<u32>() {
cached_pid = Some(pid);
process_alive(pid)
} else {
true
}
} else {
true // no PID file yet, assume alive
}
}
};
if !engine_alive && !conclusion_path.exists() {
drain_remaining(&mut reader, &mut line, &mut progress_ui);
break;
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
// Finish progress bars
progress_ui.finish();
// Determine exit code from conclusion
if conclusion_path.exists() {
match fabro_workflows::conclusion::Conclusion::load(&conclusion_path) {
Ok(conclusion) => {
let success = matches!(
conclusion.status,
fabro_workflows::outcome::StageStatus::Success
| fabro_workflows::outcome::StageStatus::PartialSuccess
);
Ok(if success {
ExitCode::from(0)
} else {
ExitCode::from(1)
})
}
Err(_) => Ok(ExitCode::from(1)),
}
} else {
Ok(ExitCode::from(1))
}
}
fn drain_remaining(
reader: &mut BufReader<std::fs::File>,
line: &mut String,
progress_ui: &mut run_progress::ProgressUI,
) {
loop {
line.clear();
match reader.read_line(line) {
Ok(0) => break,
Ok(_) => {
let trimmed = line.trim();
if !trimmed.is_empty() {
progress_ui.handle_json_line(trimmed);
}
}
Err(_) => break,
}
}
}
fn kill_engine(pid_path: &Path) {
if let Ok(pid_str) = std::fs::read_to_string(pid_path) {
if let Ok(pid) = pid_str.trim().parse::<i32>() {
#[cfg(unix)]
unsafe {
libc::kill(pid, libc::SIGTERM);
}
let _ = pid;
}
}
}
fn process_alive(pid: u32) -> bool {
#[cfg(unix)]
{
unsafe { libc::kill(pid as i32, 0) == 0 }
}
#[cfg(not(unix))]
{
let _ = pid;
true
}
}

View file

@ -0,0 +1,92 @@
use std::path::PathBuf;
use chrono::Local;
use fabro_config::run::RunDefaults;
use fabro_workflows::run_spec::RunSpec;
use super::run::{prepare_workflow, RunArgs};
use fabro_util::terminal::Styles;
/// Create a workflow run: allocate run directory, persist spec, return (run_id, run_dir).
///
/// This does NOT execute the workflow — it only prepares the run directory.
pub async fn create_run(
args: &RunArgs,
run_defaults: RunDefaults,
styles: &Styles,
) -> anyhow::Result<(String, PathBuf)> {
let workflow_path = args
.workflow
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--workflow is required"))?;
let prep = prepare_workflow(args, run_defaults, styles)?;
let goal = prep.graph.goal();
// Create run directory
let run_id = ulid::Ulid::new().to_string();
let run_dir = args.run_dir.clone().unwrap_or_else(|| {
if args.dry_run {
std::env::temp_dir().join("fabro-dry-run").join(&run_id)
} else {
let base = dirs::home_dir()
.expect("could not determine home directory")
.join(".fabro")
.join("runs");
base.join(format!("{}-{}", Local::now().format("%Y%m%d"), run_id))
}
});
tokio::fs::create_dir_all(&run_dir).await?;
// Write essential files
tokio::fs::write(run_dir.join("graph.fabro"), &prep.source).await?;
tokio::fs::write(run_dir.join("id.txt"), &run_id).await?;
std::fs::File::create(run_dir.join("progress.jsonl"))?;
fabro_workflows::run_status::write_run_status(
&run_dir,
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
// Save TOML config alongside the run if present
if workflow_path.extension().is_some_and(|ext| ext == "toml") {
if let Ok(toml_contents) = tokio::fs::read(workflow_path).await {
tokio::fs::write(run_dir.join("run.toml"), toml_contents).await?;
}
}
// Build and save RunSpec
let working_directory = std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."));
let spec = RunSpec {
run_id: run_id.clone(),
workflow_path: std::fs::canonicalize(workflow_path).unwrap_or(workflow_path.clone()),
dot_source: prep.source,
working_directory,
goal: if goal.is_empty() {
None
} else {
Some(goal.to_string())
},
model: prep.model,
provider: prep.provider,
sandbox_provider: prep.sandbox_provider.to_string(),
labels: args
.label
.iter()
.filter_map(|s| s.split_once('='))
.map(|(k, v)| (k.to_string(), v.to_string()))
.collect(),
verbose: args.verbose,
no_retro: args.no_retro,
ssh: args.ssh,
preserve_sandbox: args.preserve_sandbox,
dry_run: args.dry_run,
auto_approve: args.auto_approve,
resume: args.resume.clone(),
run_branch: args.run_branch.clone(),
};
spec.save(&run_dir)?;
Ok((run_id, run_dir))
}

View file

@ -1,5 +1,7 @@
pub mod asset;
pub mod attach;
pub mod cp;
pub mod create;
pub mod diff;
pub mod fork;
pub mod graph;
@ -11,11 +13,12 @@ pub mod preview;
pub mod provider;
pub mod rewind;
pub mod run;
mod run_progress;
pub(crate) mod run_progress;
pub mod runs;
pub mod secret;
pub(crate) mod shared;
pub mod ssh;
pub mod start;
pub mod validate;
pub mod wait;
pub mod workflow;

View file

@ -10,7 +10,7 @@ use clap::{Args, ValueEnum};
use fabro_agent::{DockerSandbox, DockerSandboxConfig, LocalSandbox, Sandbox};
use fabro_config::run::{RunDefaults, WorkflowRunConfig};
use fabro_config::{project as project_config, run as run_config, sandbox as sandbox_config};
use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, Interviewer};
use fabro_interview::{AutoApproveInterviewer, ConsoleInterviewer, FileInterviewer, Interviewer};
use fabro_model::Provider;
use fabro_util::terminal::Styles;
use fabro_validate::Severity;
@ -57,6 +57,19 @@ impl From<CliSandboxProvider> for SandboxProvider {
}
}
impl From<SandboxProvider> for CliSandboxProvider {
fn from(value: SandboxProvider) -> Self {
match value {
SandboxProvider::Local => Self::Local,
SandboxProvider::Docker => Self::Docker,
SandboxProvider::Daytona => Self::Daytona,
#[cfg(feature = "exedev")]
SandboxProvider::Exe => Self::Exe,
SandboxProvider::Ssh => Self::Ssh,
}
}
}
#[derive(Args)]
pub struct RunArgs {
/// Path to a .fabro workflow file or .toml task config (not required with --run-branch)
@ -137,7 +150,7 @@ pub struct RunArgs {
}
/// Resolve goal from `--goal` string or `--goal-file` path.
fn resolve_cli_goal(
pub(crate) fn resolve_cli_goal(
goal: &Option<String>,
goal_file: &Option<PathBuf>,
) -> anyhow::Result<Option<String>> {
@ -156,7 +169,7 @@ fn resolve_cli_goal(
/// Apply goal to the graph from TOML config or CLI flag.
/// Precedence: CLI `--goal` / `--goal-file` > TOML `goal` > DOT `graph [goal="..."]`.
fn apply_goal_override(
pub(crate) fn apply_goal_override(
graph: &mut fabro_graphviz::graph::Graph,
cli_goal: Option<&str>,
toml_goal: Option<&str>,
@ -174,7 +187,7 @@ fn apply_goal_override(
/// Resolve model and provider through the full precedence chain:
/// CLI flag > TOML config > run defaults > DOT graph attrs > provider-specific defaults.
/// Then resolve through the catalog for alias expansion.
fn resolve_model_provider(
pub(crate) fn resolve_model_provider(
cli_model: Option<&str>,
cli_provider: Option<&str>,
run_cfg: Option<&WorkflowRunConfig>,
@ -221,7 +234,7 @@ fn resolve_model_provider(
}
/// Parse sandbox provider from an optional `SandboxConfig`.
fn parse_sandbox_provider(
pub(crate) fn parse_sandbox_provider(
sandbox: Option<&sandbox_config::SandboxConfig>,
) -> anyhow::Result<Option<SandboxProvider>> {
sandbox
@ -232,7 +245,7 @@ fn parse_sandbox_provider(
}
/// Resolve sandbox provider: CLI flag > TOML config > run defaults > default.
fn resolve_sandbox_provider(
pub(crate) fn resolve_sandbox_provider(
cli: Option<SandboxProvider>,
run_cfg: Option<&WorkflowRunConfig>,
run_defaults: &RunDefaults,
@ -420,30 +433,32 @@ struct CostAccumulator {
has_pricing: bool,
}
/// Execute a full workflow run.
///
/// # Errors
///
/// Returns an error if the workflow cannot be read, parsed, validated, or executed.
pub async fn run_command(
args: RunArgs,
mut run_defaults: RunDefaults,
styles: &'static Styles,
github_app: Option<fabro_github::GitHubAppCredentials>,
git_author: fabro_workflows::git::GitAuthor,
) -> anyhow::Result<()> {
// Handle --run-branch resume: read everything from git metadata
if let Some(branch) = args.run_branch.clone() {
return run_from_branch(args, &branch, styles, git_author, run_defaults, github_app).await;
}
/// Result of workflow preparation (shared between `create` and `run` commands).
pub(crate) struct PreparedWorkflow {
pub source: String,
pub graph: fabro_graphviz::graph::Graph,
pub run_cfg: Option<WorkflowRunConfig>,
pub sandbox_provider: SandboxProvider,
pub model: String,
pub provider: Option<String>,
pub run_defaults: RunDefaults,
}
/// Resolve config, parse/validate the workflow graph, and resolve sandbox + model.
///
/// Shared between `create_run` (which only persists the spec) and
/// `run_command` (which goes on to execute the workflow).
pub(crate) fn prepare_workflow(
args: &RunArgs,
mut run_defaults: RunDefaults,
styles: &Styles,
) -> anyhow::Result<PreparedWorkflow> {
let workflow_path = args
.workflow
.as_ref()
.ok_or_else(|| anyhow::anyhow!("--workflow is required unless --run-branch is provided"))?;
.ok_or_else(|| anyhow::anyhow!("--workflow is required"))?;
// Apply project-level config overrides (fabro.toml) on top of CLI defaults.
// Precedence: workflow.toml > fabro.toml > cli.toml/server.toml
if let Ok(Some((_config_path, project_config))) =
project_config::discover_project_config(&std::env::current_dir().unwrap_or_default())
{
@ -451,7 +466,7 @@ pub async fn run_command(
run_defaults.merge_overlay(project_config.into_run_defaults());
}
// 0. Resolve workflow arg, load run config if TOML, resolve DOT path, apply defaults
// Resolve workflow arg, load run config if TOML, apply defaults
let (dot_path, run_cfg) = {
let (dot, cfg) = project_config::resolve_workflow(workflow_path)?;
match cfg {
@ -463,18 +478,6 @@ pub async fn run_command(
}
};
// Extract workflow slug from the workflow path argument.
// If bare name (no extension, e.g. "smoke"), use it directly.
// Otherwise derive from the parent directory of the resolved .toml path.
let workflow_slug: Option<String> = if workflow_path.extension().is_none() {
Some(workflow_path.to_string_lossy().into_owned())
} else {
workflow_path
.parent()
.and_then(|p| p.file_name())
.map(|n| n.to_string_lossy().into_owned())
};
let directory = run_cfg
.as_ref()
.and_then(|c| c.work_dir.as_deref())
@ -484,15 +487,7 @@ pub async fn run_command(
.map_err(|e| anyhow::anyhow!("Failed to set working directory to {dir}: {e}"))?;
}
// Collect setup commands — they'll be run inside the sandbox
let setup_commands: Vec<String> = run_cfg
.as_ref()
.and_then(|c| c.setup.as_ref())
.or(run_defaults.setup.as_ref())
.map(|s| s.commands.clone())
.unwrap_or_default();
// 1. Parse and validate workflow
// Parse and validate workflow
let source = read_workflow_file(&dot_path)?;
let vars = run_cfg
.as_ref()
@ -550,8 +545,7 @@ pub async fn run_command(
bail!("Validation failed");
}
// 2. Pre-flight: check git cleanliness before creating any files
// (must happen before logs dir is created, which may be inside the repo)
// Resolve sandbox provider
let sandbox_provider = if args.dry_run {
SandboxProvider::Local
} else {
@ -561,6 +555,76 @@ pub async fn run_command(
&run_defaults,
)?
};
// Resolve model and provider
let (model, provider) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
&run_defaults,
&graph,
);
Ok(PreparedWorkflow {
source,
graph,
run_cfg,
sandbox_provider,
model,
provider,
run_defaults,
})
}
/// Execute a full workflow run.
///
/// # Errors
///
/// Returns an error if the workflow cannot be read, parsed, validated, or executed.
pub async fn run_command(
args: RunArgs,
run_defaults: RunDefaults,
styles: &'static Styles,
github_app: Option<fabro_github::GitHubAppCredentials>,
git_author: fabro_workflows::git::GitAuthor,
) -> anyhow::Result<()> {
// Handle --run-branch resume: read everything from git metadata
if let Some(branch) = args.run_branch.clone() {
return run_from_branch(args, &branch, styles, git_author, run_defaults, github_app).await;
}
let PreparedWorkflow {
source,
graph,
run_cfg,
sandbox_provider,
model,
provider,
run_defaults,
} = prepare_workflow(&args, run_defaults, styles)?;
// Extract workflow slug from the workflow path argument.
// If bare name (no extension, e.g. "smoke"), use it directly.
// Otherwise derive from the parent directory of the resolved .toml path.
let workflow_path = args.workflow.as_ref().unwrap(); // safe: prepare_workflow validated
let workflow_slug: Option<String> = if workflow_path.extension().is_none() {
Some(workflow_path.to_string_lossy().into_owned())
} else {
workflow_path
.parent()
.and_then(|p| p.file_name())
.map(|n| n.to_string_lossy().into_owned())
};
// Collect setup commands — they'll be run inside the sandbox
let setup_commands: Vec<String> = run_cfg
.as_ref()
.and_then(|c| c.setup.as_ref())
.or(run_defaults.setup.as_ref())
.map(|s| s.commands.clone())
.unwrap_or_default();
// Pre-flight: check git cleanliness before creating any files
let preserve_sandbox =
resolve_preserve_sandbox(args.preserve_sandbox, run_cfg.as_ref(), &run_defaults);
let original_cwd = std::env::current_dir()?;
@ -716,6 +780,10 @@ pub async fn run_command(
// 4. Build interviewer
let interviewer: Arc<dyn Interviewer> = if args.auto_approve {
Arc::new(AutoApproveInterviewer)
} else if !std::io::stdin().is_terminal() {
// Detached mode (stdin is /dev/null): use file-based IPC so the
// attach process can prompt the user on our behalf.
Arc::new(FileInterviewer::new(run_dir.clone()))
} else {
Arc::new(run_progress::ProgressAwareInterviewer::new(
ConsoleInterviewer::new(styles),
@ -1154,14 +1222,6 @@ pub async fn run_command(
}
};
let (model, provider) = resolve_model_provider(
args.model.as_deref(),
args.provider.as_deref(),
run_cfg.as_ref(),
&run_defaults,
&graph,
);
// Parse provider string to enum (defaults to best available from env)
let provider_enum: Provider = provider
.as_deref()

View file

@ -266,6 +266,20 @@ impl ProgressUI {
});
}
/// Hide indicatif progress bars (for interview prompts in attach mode).
pub fn hide_bars(&self) {
if let ProgressRenderer::Tty(tty) = &self.renderer {
tty.multi.set_draw_target(ProgressDrawTarget::hidden());
}
}
/// Show indicatif progress bars after an interview prompt.
pub fn show_bars(&self) {
if let ProgressRenderer::Tty(tty) = &self.renderer {
tty.multi.set_draw_target(ProgressDrawTarget::stderr());
}
}
/// Clear all active bars and release the terminal for normal stderr output.
pub fn finish(&mut self) {
for (_id, stage) in self.active_stages.drain() {
@ -610,6 +624,360 @@ impl ProgressUI {
}
}
// ── JSONL dispatch ────────────────────────────────────────────────
/// Parse a JSONL envelope line and dispatch to internal rendering methods.
/// Used by the attach loop to render events from progress.jsonl.
pub fn handle_json_line(&mut self, line: &str) {
let envelope: serde_json::Value = match serde_json::from_str(line) {
Ok(v) => v,
Err(_) => return,
};
let event_name = match envelope.get("event").and_then(|v| v.as_str()) {
Some(name) => name,
None => return,
};
let str_field = |key: &str| -> Option<&str> { envelope.get(key).and_then(|v| v.as_str()) };
let u64_field =
|key: &str| -> u64 { envelope.get(key).and_then(|v| v.as_u64()).unwrap_or(0) };
match event_name {
"Sandbox.Initializing" => {
let provider = str_field("sandbox_provider")
.unwrap_or("unknown")
.to_string();
self.on_sandbox_event(&fabro_agent::SandboxEvent::Initializing { provider });
}
"Sandbox.Ready" => {
let provider = str_field("sandbox_provider")
.unwrap_or("unknown")
.to_string();
let duration_ms = u64_field("duration_ms");
let name = str_field("name").map(String::from);
let cpu = envelope.get("cpu").and_then(|v| v.as_f64());
let memory = envelope.get("memory").and_then(|v| v.as_f64());
let url = str_field("url").map(String::from);
self.on_sandbox_event(&fabro_agent::SandboxEvent::Ready {
provider,
duration_ms,
name,
cpu,
memory,
url,
});
}
"SandboxInitialized" => {
if let Some(wd) = str_field("working_directory") {
self.set_working_directory(wd.to_string());
}
}
"SetupStarted" => {
let count = u64_field("command_count") as usize;
self.on_setup_started(count);
}
"SetupCompleted" => {
let duration_ms = u64_field("duration_ms");
self.on_setup_completed(duration_ms);
}
"StageStarted" => {
let node_id = str_field("node_id").unwrap_or("?");
let name = str_field("node_label").unwrap_or("?");
let script = str_field("script");
self.on_stage_started(node_id, name, script);
}
"StageCompleted" => {
let node_id = str_field("node_id").unwrap_or("?");
let name = str_field("node_label").unwrap_or("?");
let duration_ms = u64_field("duration_ms");
let status = str_field("status").unwrap_or("success");
let succeeded = matches!(status, "success" | "partial_success");
let dur = format_duration_ms(duration_ms);
// Parse usage for cost
let cost_str = envelope
.get("usage")
.and_then(|u| u.get("cost"))
.and_then(|c| c.as_f64())
.map(|c| format!("{} ", format_cost(c)))
.unwrap_or_default();
let stats_str = if self.verbose {
let counts = self.stage_counts.get(node_id);
let turn_count = counts.map_or(0, |c| c.0);
let tool_call_count = counts.map_or(0, |c| c.1);
let total_tokens = envelope
.get("usage")
.map(|u| {
u.get("input_tokens").and_then(|v| v.as_i64()).unwrap_or(0)
+ u.get("output_tokens").and_then(|v| v.as_i64()).unwrap_or(0)
})
.unwrap_or(0);
if turn_count > 0 || tool_call_count > 0 || total_tokens > 0 {
let dim = Style::new().dim();
format!(
" {}",
dim.apply_to(format!(
"({} turns, {} tools, {} toks)",
turn_count,
tool_call_count,
format_tokens_human(total_tokens),
))
)
} else {
String::new()
}
} else {
String::new()
};
let prefix = format!("{cost_str}{dur}{stats_str}");
let glyph = if succeeded {
green_check()
} else {
red_cross()
};
self.finish_stage(node_id, name, glyph, &prefix);
}
"StageFailed" => {
let node_id = str_field("node_id").unwrap_or("?");
let name = str_field("node_label").unwrap_or("?");
let message = str_field("error")
.or_else(|| str_field("failure_reason"))
.unwrap_or("unknown error");
self.finish_stage(node_id, name, red_cross(), "");
let red = Style::new().red();
let summary = last_line_truncated(message, 120);
self.insert_info_line(&format!("{} {}", red.apply_to("Error:"), summary));
}
"ParallelStarted" => {
self.parallel_parent = self
.active_stages
.keys()
.next()
.cloned()
.or_else(|| Some(String::new()));
}
"ParallelBranchStarted" => {
if let Some(branch) = str_field("node_id") {
self.on_parallel_branch_started(branch);
}
}
"ParallelBranchCompleted" => {
if let Some(branch) = str_field("node_id") {
let duration_ms = u64_field("duration_ms");
let status = str_field("status").unwrap_or("success");
self.on_parallel_branch_completed(branch, duration_ms, status);
}
}
"ParallelCompleted" => {
self.parallel_parent = None;
}
"Agent.ToolCallStarted" => {
let stage = str_field("node_id").unwrap_or("?");
let tool_name = str_field("tool_name").unwrap_or("?");
let tool_call_id = str_field("tool_call_id").unwrap_or("?");
let empty = serde_json::Value::Object(serde_json::Map::new());
let arguments = envelope.get("arguments").unwrap_or(&empty);
// Update tool_call count
if let Some(counts) = self.stage_counts.get_mut(stage) {
counts.1 += 1;
}
self.on_tool_call_started(stage, tool_name, tool_call_id, arguments);
}
"Agent.ToolCallCompleted" => {
let stage = str_field("node_id").unwrap_or("?");
let tool_call_id = str_field("tool_call_id").unwrap_or("?");
let is_error = envelope
.get("is_error")
.and_then(|v| v.as_bool())
.unwrap_or(false);
self.on_tool_call_completed(stage, tool_call_id, is_error);
}
"Agent.AssistantMessage" => {
let stage = str_field("node_id").unwrap_or("?");
let model = str_field("model").unwrap_or("?");
// Update turn count
if let Some(counts) = self.stage_counts.get_mut(stage) {
counts.0 += 1;
}
// Update model display on stage bar
if let ProgressRenderer::Tty(_) = &self.renderer {
if let Some(active_stage) = self.active_stages.get_mut(stage) {
if !active_stage.has_model {
active_stage.has_model = true;
let dim = Style::new().dim();
let suffix = format!(" {}", dim.apply_to(format!("[{model}]")));
active_stage
.spinner
.set_message(format!("{}{}", active_stage.display_name, suffix));
}
}
}
}
"Agent.CompactionStarted" => {
let stage = str_field("node_id").unwrap_or("?");
if let ProgressRenderer::Tty(tty) = &self.renderer {
if let Some(active_stage) = self.active_stages.get_mut(stage) {
if let Some(old) = active_stage.compaction_bar.take() {
old.finish_and_clear();
}
let bar = tty
.multi
.insert_after(active_stage.last_bar(), ProgressBar::new_spinner());
bar.set_style(style_tool_running());
bar.set_message("\u{27f3} compacting context\u{2026}");
bar.enable_steady_tick(Duration::from_millis(100));
active_stage.compaction_bar = Some(bar);
}
}
}
"Agent.CompactionCompleted" => {
let stage = str_field("node_id").unwrap_or("?");
let original = u64_field("original_turn_count");
let preserved = u64_field("preserved_turn_count");
let tracked = u64_field("tracked_file_count");
let msg = format!(
"\u{27f3} compaction: {original} \u{2192} {preserved} turns, {tracked} files"
);
match &self.renderer {
ProgressRenderer::Tty(_) => {
if let Some(bar) = self
.active_stages
.get_mut(stage)
.and_then(|s| s.compaction_bar.take())
{
bar.set_style(style_tool_done());
bar.finish_with_message(msg);
} else {
self.insert_info_line_for_stage(stage, &msg);
}
}
ProgressRenderer::Plain => {
eprintln!(" {msg}");
}
}
}
"SshAccessReady" => {
if let Some(cmd) = str_field("ssh_command") {
self.on_ssh_access_ready(cmd);
}
}
"RetroStarted" => {
self.on_stage_started("retro", "Retro", None);
}
"RetroCompleted" => {
let dur = format_duration_ms(u64_field("duration_ms"));
self.finish_stage("retro", "Retro", green_check(), &dur);
}
"RetroFailed" => {
let dur = format_duration_ms(u64_field("duration_ms"));
self.finish_stage("retro", "Retro", red_cross(), &dur);
}
"DevcontainerResolved" => {
let dockerfile_lines = u64_field("dockerfile_lines");
let environment_count = u64_field("environment_count");
let lifecycle_command_count = u64_field("lifecycle_command_count");
let workspace_folder = str_field("workspace_folder").unwrap_or("?").to_string();
let detail = format!(
"{dockerfile_lines} Dockerfile lines, {environment_count} env vars, \
{lifecycle_command_count} lifecycle cmds, {workspace_folder}"
);
match &self.renderer {
ProgressRenderer::Tty(tty) => {
let bar = tty.multi.add(ProgressBar::new_spinner());
bar.set_style(style_header_done());
bar.finish_with_message("Devcontainer: resolved".to_string());
let detail_bar = tty.multi.insert_after(&bar, ProgressBar::new_spinner());
detail_bar.set_style(style_sandbox_detail());
detail_bar.finish_with_message(detail);
}
ProgressRenderer::Plain => {
eprintln!(" Devcontainer: resolved");
eprintln!(" {detail}");
}
}
}
"DevcontainerLifecycleStarted" => {
let phase = str_field("phase").unwrap_or("?");
let command_count = u64_field("command_count") as usize;
self.devcontainer_command_count = command_count;
match &self.renderer {
ProgressRenderer::Tty(tty) => {
let bar = tty.multi.add(ProgressBar::new_spinner());
bar.set_style(style_header_running());
bar.set_message(format!(
"Running devcontainer {phase} ({command_count} commands)..."
));
bar.enable_steady_tick(Duration::from_millis(100));
self.devcontainer_bar = Some(bar);
}
ProgressRenderer::Plain => {
eprintln!(" Running devcontainer {phase} ({command_count} commands)...");
}
}
}
"DevcontainerLifecycleCompleted" => {
let phase = str_field("phase").unwrap_or("?");
let duration_ms = u64_field("duration_ms");
let dur = format_duration_ms(duration_ms);
match &self.renderer {
ProgressRenderer::Tty(_) => {
if let Some(bar) = self.devcontainer_bar.take() {
bar.set_style(style_header_done());
bar.set_prefix(dur);
bar.finish_with_message(format!("Devcontainer: {phase}"));
}
}
ProgressRenderer::Plain => {
eprintln!(" Devcontainer: {phase} ({dur})");
}
}
}
"DevcontainerLifecycleFailed" => {
let phase = str_field("phase").unwrap_or("?");
let command = str_field("command").unwrap_or("?");
let exit_code = u64_field("exit_code");
let stderr_text = str_field("stderr").unwrap_or("");
if let Some(bar) = self.devcontainer_bar.take() {
bar.abandon();
}
let red = Style::new().red();
let summary = if stderr_text.len() > 120 {
&stderr_text[..120]
} else {
stderr_text
};
self.insert_info_line(&format!(
"{} Devcontainer {phase} command failed (exit {exit_code}): {command}\n {summary}",
red.apply_to("Error:")
));
}
"CliEnsureStarted" => {
if let Some(cli_name) = str_field("cli_name") {
self.on_cli_ensure_started(cli_name);
}
}
"CliEnsureCompleted" => {
if let Some(cli_name) = str_field("cli_name") {
let already_installed = envelope
.get("already_installed")
.and_then(|v| v.as_bool())
.unwrap_or(false);
let duration_ms = u64_field("duration_ms");
self.on_cli_ensure_completed(cli_name, already_installed, duration_ms);
}
}
"CliEnsureFailed" => {
if let Some(cli_name) = str_field("cli_name") {
self.on_cli_ensure_failed(cli_name);
}
}
_ => {}
}
}
// ── Sandbox ─────────────────────────────────────────────────────────
fn on_sandbox_event(&mut self, event: &fabro_agent::SandboxEvent) {
@ -1271,43 +1639,33 @@ impl ProgressAwareInterviewer {
pub fn new(inner: ConsoleInterviewer, progress: Arc<Mutex<ProgressUI>>) -> Self {
Self { inner, progress }
}
fn hide_bars(&self) {
let ui = self.progress.lock().expect("progress lock poisoned");
if let ProgressRenderer::Tty(tty) = &ui.renderer {
tty.multi.set_draw_target(ProgressDrawTarget::hidden());
}
}
fn show_bars(&self) {
let ui = self.progress.lock().expect("progress lock poisoned");
if let ProgressRenderer::Tty(tty) = &ui.renderer {
tty.multi.set_draw_target(ProgressDrawTarget::stderr());
}
}
}
#[async_trait]
impl Interviewer for ProgressAwareInterviewer {
async fn ask(&self, question: Question) -> Answer {
{
let ui = self.progress.lock().expect("progress lock poisoned");
if let ProgressRenderer::Tty(tty) = &ui.renderer {
let sep = tty.multi.add(ProgressBar::new_spinner());
sep.set_style(style_empty());
sep.finish();
tty.multi.set_draw_target(ProgressDrawTarget::hidden());
}
}
self.progress
.lock()
.expect("progress lock poisoned")
.hide_bars();
let answer = self.inner.ask(question).await;
self.show_bars();
self.progress
.lock()
.expect("progress lock poisoned")
.show_bars();
answer
}
async fn inform(&self, message: &str, stage: &str) {
self.hide_bars();
self.progress
.lock()
.expect("progress lock poisoned")
.hide_bars();
self.inner.inform(message, stage).await;
self.show_bars();
self.progress
.lock()
.expect("progress lock poisoned")
.show_bars();
}
}
@ -1530,4 +1888,157 @@ mod tests {
});
assert!(ui.parallel_parent.is_none());
}
#[test]
fn handle_json_line_stage_started_and_completed() {
let mut ui = ProgressUI::new(false, false);
let started = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"plan","node_label":"Plan","stage_index":0,"script":null,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(started);
assert!(ui.stage_counts.contains_key("plan"));
let completed = r#"{"ts":"2026-01-01T12:00:10Z","event":"StageCompleted","node_id":"plan","node_label":"Plan","stage_index":0,"duration_ms":10000,"status":"success"}"#;
ui.handle_json_line(completed);
// In Plain mode, finish_stage just prints, so verify no panic
}
#[test]
fn handle_json_line_tool_call_round_trip() {
let mut ui = ProgressUI::new(false, true); // verbose
// Start a stage first
let started = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"code","node_label":"Code","stage_index":0,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(started);
let tc_start = r#"{"ts":"2026-01-01T12:00:01Z","event":"Agent.ToolCallStarted","node_id":"code","node_label":"code","tool_name":"read_file","tool_call_id":"tc1","arguments":{"path":"src/main.rs"}}"#;
ui.handle_json_line(tc_start);
assert_eq!(ui.stage_counts.get("code").map(|c| c.1), Some(1));
let tc_done = r#"{"ts":"2026-01-01T12:00:02Z","event":"Agent.ToolCallCompleted","node_id":"code","node_label":"code","tool_name":"read_file","tool_call_id":"tc1","is_error":false}"#;
ui.handle_json_line(tc_done);
}
#[test]
fn handle_json_line_retro_events() {
let mut ui = ProgressUI::new(false, false);
let retro_started = r#"{"ts":"2026-01-01T12:00:00Z","event":"RetroStarted"}"#;
ui.handle_json_line(retro_started);
let retro_completed =
r#"{"ts":"2026-01-01T12:00:05Z","event":"RetroCompleted","duration_ms":5000}"#;
ui.handle_json_line(retro_completed);
}
#[test]
fn handle_json_line_ignores_invalid_json() {
let mut ui = ProgressUI::new(false, false);
ui.handle_json_line("not valid json");
ui.handle_json_line("");
ui.handle_json_line("{}"); // no event field
}
// ── Bug regression tests (post-rename JSONL field names) ─────────
// Bug 1: handle_json_line reads pre-rename field names but real JSONL
// uses post-rename names from rename_fields(). These tests use the
// actual JSONL format produced by the engine.
#[test]
fn bug1_stage_started_uses_node_label_not_name() {
// Real JSONL: rename_fields renames "name" → "node_label"
let mut ui = ProgressUI::new(true, false);
let started = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"plan","node_label":"Plan","stage_index":0,"script":null,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(started);
let stage = ui
.active_stages
.get("plan")
.expect("stage should be tracked");
assert_eq!(
stage.display_name, "Plan",
"display name should come from node_label field, not be '?'"
);
}
#[test]
fn bug1_agent_tool_call_uses_node_id_not_stage() {
// Real JSONL: rename_fields renames "stage" → "node_id" for Agent.* events
let mut ui = ProgressUI::new(false, true);
// First create the stage
let started = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"code","node_label":"Code","stage_index":0,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(started);
assert_eq!(ui.stage_counts.get("code").map(|c| c.1), Some(0));
// Tool call with post-rename field: "node_id" instead of "stage"
let tc_start = r#"{"ts":"2026-01-01T12:00:01Z","event":"Agent.ToolCallStarted","node_id":"code","node_label":"code","tool_name":"read_file","tool_call_id":"tc1","arguments":{"path":"src/main.rs"}}"#;
ui.handle_json_line(tc_start);
assert_eq!(
ui.stage_counts.get("code").map(|c| c.1),
Some(1),
"tool call count should increment using node_id field"
);
}
#[test]
fn bug1_agent_assistant_message_uses_node_id_not_stage() {
// Real JSONL: rename_fields renames "stage" → "node_id"
let mut ui = ProgressUI::new(false, true);
let started = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"code","node_label":"Code","stage_index":0,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(started);
assert_eq!(ui.stage_counts.get("code").map(|c| c.0), Some(0));
// AssistantMessage with post-rename field: "node_id" instead of "stage"
let msg = r#"{"ts":"2026-01-01T12:00:01Z","event":"Agent.AssistantMessage","node_id":"code","node_label":"code","model":"claude-sonnet-4-20250514"}"#;
ui.handle_json_line(msg);
assert_eq!(
ui.stage_counts.get("code").map(|c| c.0),
Some(1),
"turn count should increment using node_id field"
);
}
#[test]
fn bug1_parallel_branch_uses_node_id_not_branch() {
// Real JSONL: rename_fields renames "branch" → "node_id"
let mut ui = ProgressUI::new(true, false);
// Set up a parent stage and start parallel
let parent = r#"{"ts":"2026-01-01T12:00:00Z","event":"StageStarted","node_id":"fork","node_label":"Fork","stage_index":0,"attempt":1,"max_attempts":1}"#;
ui.handle_json_line(parent);
let par = r#"{"ts":"2026-01-01T12:00:01Z","event":"ParallelStarted","branch_count":2,"join_policy":"wait_all","error_policy":"continue"}"#;
ui.handle_json_line(par);
assert!(ui.parallel_parent.is_some());
// ParallelBranchStarted with post-rename field: "node_id" instead of "branch"
let branch = r#"{"ts":"2026-01-01T12:00:02Z","event":"ParallelBranchStarted","node_id":"lint","node_label":"lint","branch_index":0}"#;
ui.handle_json_line(branch);
// Branch should have been registered as a tool_call entry on the parent
let parent_stage = ui.active_stages.get("fork").unwrap();
assert!(
!parent_stage.tool_calls.is_empty(),
"parallel branch should be registered using node_id field"
);
}
// Bug 5: start_run should write Starting status before spawning engine
// (tested in start.rs)
// Bug 6: handle_json_line is missing devcontainer event dispatch
#[test]
fn bug6_devcontainer_lifecycle_started_dispatched() {
let mut ui = ProgressUI::new(false, false);
let event = r#"{"ts":"2026-01-01T12:00:00Z","event":"DevcontainerLifecycleStarted","phase":"postCreate","command_count":2}"#;
ui.handle_json_line(event);
assert_eq!(
ui.devcontainer_command_count, 2,
"devcontainer_command_count should be set by DevcontainerLifecycleStarted"
);
}
}

View file

@ -0,0 +1,113 @@
use std::path::Path;
use anyhow::{bail, Result};
/// Spawn a detached engine process for the given run directory.
///
/// The engine process reads `spec.json` from the run directory and executes the
/// workflow. Returns the child process PID.
pub fn start_run(run_dir: &Path) -> Result<u32> {
// Validate status is Submitted
let status_path = run_dir.join("status.json");
match fabro_workflows::run_status::RunStatusRecord::load(&status_path) {
Ok(record) if record.status != fabro_workflows::run_status::RunStatus::Submitted => {
bail!(
"Cannot start run: status is {:?}, expected Submitted",
record.status
);
}
_ => {} // No status file or Submitted — proceed
}
// Validate spec.json is loadable
fabro_workflows::run_spec::RunSpec::load(run_dir)
.map_err(|e| anyhow::anyhow!("Cannot start run: failed to load spec.json: {e}"))?;
// Write Starting status before spawning to prevent duplicate engines
fabro_workflows::run_status::write_run_status(
run_dir,
fabro_workflows::run_status::RunStatus::Starting,
None,
);
let log_file = std::fs::File::create(run_dir.join("detach.log"))?;
let exe = std::env::current_exe()?;
let mut cmd = std::process::Command::new(&exe);
cmd.args(["_run_engine", "--run-dir"])
.arg(run_dir)
.stdout(log_file.try_clone()?)
.stderr(log_file)
.stdin(std::process::Stdio::null());
// Detach from the controlling terminal on unix
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
unsafe {
cmd.pre_exec(|| {
libc::setsid();
Ok(())
});
}
}
let child = cmd.spawn()?;
let pid = child.id();
// Write PID file
std::fs::write(run_dir.join("run.pid"), pid.to_string())?;
Ok(pid)
}
#[cfg(test)]
mod tests {
use super::*;
use fabro_workflows::run_spec::RunSpec;
use fabro_workflows::run_status::{write_run_status, RunStatus, RunStatusRecord};
use std::collections::HashMap;
use std::path::PathBuf;
fn sample_spec() -> RunSpec {
RunSpec {
run_id: "run-test123".to_string(),
workflow_path: PathBuf::from("/tmp/test-workflow.toml"),
dot_source: "digraph { a -> b }".to_string(),
working_directory: PathBuf::from("/tmp"),
goal: None,
model: "claude-sonnet-4-20250514".to_string(),
provider: Some("anthropic".to_string()),
sandbox_provider: "local".to_string(),
labels: HashMap::new(),
verbose: false,
no_retro: true,
ssh: false,
preserve_sandbox: false,
dry_run: false,
auto_approve: true,
resume: None,
run_branch: None,
}
}
// Bug 5: start_run should write Starting status before spawning engine
// to prevent duplicate engine processes from concurrent start calls.
#[test]
fn bug5_start_run_writes_starting_status_before_spawn() {
let dir = tempfile::tempdir().unwrap();
write_run_status(dir.path(), RunStatus::Submitted, None);
sample_spec().save(dir.path()).unwrap();
// start_run may fail on spawn (test binary != fabro), but we only
// care about the status file being updated before the spawn attempt.
let _ = start_run(dir.path());
let record = RunStatusRecord::load(&dir.path().join("status.json")).unwrap();
assert_eq!(
record.status,
RunStatus::Starting,
"start_run should write Starting status before spawning to prevent duplicate engines"
);
}
}

View file

@ -72,6 +72,25 @@ enum Command {
Exec(fabro_agent::cli::AgentArgs),
/// Launch a workflow run
Run(commands::run::RunArgs),
/// Create a workflow run (allocate run dir, persist spec)
Create(commands::run::RunArgs),
/// Start a created workflow run (spawn engine process)
Start {
/// Run ID prefix or workflow name
run: String,
},
/// Attach to a running or finished workflow run
Attach {
/// Run ID prefix or workflow name
run: String,
},
/// Internal: run the engine process (reads spec.json from run dir)
#[command(name = "_run_engine", hide = true)]
RunEngine {
/// Path to the run directory
#[arg(long)]
run_dir: PathBuf,
},
/// Validate a workflow
Validate(commands::validate::ValidateArgs),
/// Render a workflow graph as SVG or PNG
@ -292,90 +311,6 @@ pub(crate) fn build_github_app_credentials(
})
}
/// Fork the workflow as a background process, print the run ID, and exit.
fn detach_run(args: commands::run::RunArgs) -> Result<()> {
let run_id = ulid::Ulid::new().to_string();
let run_dir = args.run_dir.clone().unwrap_or_else(|| {
let base = dirs::home_dir()
.expect("could not determine home directory")
.join(".fabro")
.join("runs");
base.join(format!(
"{}-{}",
chrono::Local::now().format("%Y%m%d"),
run_id
))
});
std::fs::create_dir_all(&run_dir)?;
std::fs::write(run_dir.join("id.txt"), &run_id)?;
fabro_workflows::run_status::write_run_status(
&run_dir,
fabro_workflows::run_status::RunStatus::Submitted,
None,
);
std::fs::File::create(run_dir.join("progress.jsonl"))?;
let log_file = std::fs::File::create(run_dir.join("detach.log"))?;
// Rebuild argv: current exe + original args, stripping --detach/-d, injecting --run-id and --run-dir
let exe = std::env::current_exe()?;
let mut child_args: Vec<String> = Vec::new();
child_args.push("run".to_string());
let raw_args: Vec<String> = std::env::args().collect();
// Skip argv[0] (binary) and argv[1] ("run"), then filter out --detach / -d
let mut iter = raw_args.iter().skip(2).peekable();
while let Some(arg) = iter.next() {
if arg == "--detach" || arg == "-d" {
continue;
}
// Skip --run-dir and its value (we'll override it)
if arg == "--run-dir" {
iter.next(); // consume the value
continue;
}
if arg.starts_with("--run-dir=") {
continue;
}
// Skip --run-id and its value (we'll override it)
if arg == "--run-id" {
iter.next();
continue;
}
if arg.starts_with("--run-id=") {
continue;
}
child_args.push(arg.clone());
}
child_args.push("--run-id".to_string());
child_args.push(run_id.clone());
child_args.push("--run-dir".to_string());
child_args.push(run_dir.to_string_lossy().to_string());
let mut cmd = std::process::Command::new(&exe);
cmd.args(&child_args)
.stdout(log_file.try_clone()?)
.stderr(log_file)
.stdin(std::process::Stdio::null());
// Detach from the controlling terminal on unix
#[cfg(unix)]
{
use std::os::unix::process::CommandExt;
unsafe {
cmd.pre_exec(|| {
libc::setsid();
Ok(())
});
}
}
cmd.spawn()?;
println!("{run_id}");
Ok(())
}
#[tokio::main]
async fn main() {
fabro_telemetry::panic::install_panic_hook();
@ -458,6 +393,10 @@ async fn main_inner() -> (String, Result<()>) {
},
Command::Exec(_) => "exec",
Command::Run(_) => "run",
Command::Create(_) => "create",
Command::Start { .. } => "start",
Command::Attach { .. } => "attach",
Command::RunEngine { .. } => "_run_engine",
Command::Validate(_) => "validate",
Command::Graph(_) => "graph",
Command::Parse(_) => "parse",
@ -560,6 +499,7 @@ async fn main_inner() -> (String, Result<()>) {
let upgrade_handle = if matches!(
cli.command,
Command::Run(_)
| Command::Create(_)
| Command::Exec(_)
| Command::Repo { .. }
| Command::Init
@ -705,26 +645,124 @@ async fn main_inner() -> (String, Result<()>) {
}
}
Command::Run(mut args) => {
if args.detach {
return detach_run(args);
}
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let cli_config = cli_config::load_cli_config(None)?;
args.verbose = args.verbose || cli_config.verbose;
let github_app = build_github_app_credentials(cli_config.app_id());
if args.detach {
// Detach mode: create + start + print run ID
let (run_id, run_dir) =
commands::create::create_run(&args, cli_config.run_defaults, styles)
.await?;
commands::start::start_run(&run_dir)?;
println!("{run_id}");
} else {
// Foreground mode: use existing run_command
let github_app = build_github_app_credentials(cli_config.app_id());
let git_author = fabro_workflows::git::GitAuthor::from_options(
cli_config.git_author().and_then(|a| a.name.clone()),
cli_config.git_author().and_then(|a| a.email.clone()),
);
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = fabro_beastie::guard(cli_config.prevent_idle_sleep);
commands::run::run_command(
args,
cli_config.run_defaults,
styles,
github_app,
git_author,
)
.await?;
}
}
Command::Create(args) => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let cli_config = cli_config::load_cli_config(None)?;
let (run_id, _run_dir) =
commands::create::create_run(&args, cli_config.run_defaults, styles).await?;
println!("{run_id}");
}
Command::Start { run } => {
let base = fabro_workflows::run_lookup::default_runs_base();
let run_info = fabro_workflows::run_lookup::resolve_run(&base, &run)?;
let pid = commands::start::start_run(&run_info.path)?;
eprintln!("Started engine process (PID {pid})");
}
Command::Attach { run } => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let base = fabro_workflows::run_lookup::default_runs_base();
let run_info = fabro_workflows::run_lookup::resolve_run(&base, &run)?;
let exit_code = commands::attach::attach_run(&run_info.path, false, styles).await?;
if exit_code != std::process::ExitCode::SUCCESS {
std::process::exit(1);
}
}
Command::RunEngine { run_dir } => {
let styles: &'static fabro_util::terminal::Styles =
Box::leak(Box::new(fabro_util::terminal::Styles::detect_stderr()));
let cli_config = cli_config::load_cli_config(None)?;
let github_app = build_github_app_credentials(cli_config.app_id());
let git_author = fabro_workflows::git::GitAuthor::from_options(
cli_config.git_author().and_then(|a| a.name.clone()),
cli_config.git_author().and_then(|a| a.email.clone()),
);
#[cfg(feature = "sleep_inhibitor")]
let _sleep_guard = fabro_beastie::guard(cli_config.prevent_idle_sleep);
// Load spec and reconstruct RunArgs
let spec = fabro_workflows::run_spec::RunSpec::load(&run_dir)?;
// Restore the working directory captured at create time
std::env::set_current_dir(&spec.working_directory).map_err(|e| {
anyhow::anyhow!(
"Failed to set working directory to {}: {e}",
spec.working_directory.display()
)
})?;
// Use the cached graph snapshot instead of the original file
let cached_graph = run_dir.join("graph.fabro");
let workflow_path = if cached_graph.exists() {
cached_graph
} else {
spec.workflow_path
};
let run_args = commands::run::RunArgs {
workflow: Some(workflow_path),
run_dir: Some(run_dir),
dry_run: spec.dry_run,
preflight: false,
auto_approve: spec.auto_approve,
resume: spec.resume,
run_branch: spec.run_branch,
goal: spec.goal,
goal_file: None,
model: Some(spec.model),
provider: Some(spec.provider.unwrap_or_default()).filter(|s| !s.is_empty()),
verbose: spec.verbose,
sandbox: spec
.sandbox_provider
.parse::<fabro_workflows::sandbox_provider::SandboxProvider>()
.ok()
.map(commands::run::CliSandboxProvider::from),
label: spec
.labels
.into_iter()
.map(|(k, v)| format!("{k}={v}"))
.collect(),
no_retro: spec.no_retro,
ssh: spec.ssh,
preserve_sandbox: spec.preserve_sandbox,
detach: false,
run_id: Some(spec.run_id),
};
commands::run::run_command(
args,
run_args,
cli_config.run_defaults,
styles,
github_app,
@ -990,4 +1028,54 @@ mod tests {
let result = Cli::try_parse_from(["fabro", "provider", "login", "--provider", "bogus"]);
assert!(result.is_err(), "should fail with unknown provider");
}
#[test]
fn parse_create_command() {
let cli = Cli::try_parse_from(["fabro", "create", "my-workflow.toml", "--goal", "test"])
.expect("should parse");
match cli.command {
Command::Create(args) => {
assert_eq!(
args.workflow.as_deref(),
Some(std::path::Path::new("my-workflow.toml"))
);
assert_eq!(args.goal.as_deref(), Some("test"));
}
_ => panic!("unexpected command variant"),
}
}
#[test]
fn parse_start_command() {
let cli = Cli::try_parse_from(["fabro", "start", "ABC123"]).expect("should parse");
match cli.command {
Command::Start { run } => {
assert_eq!(run, "ABC123");
}
_ => panic!("unexpected command variant"),
}
}
#[test]
fn parse_attach_command() {
let cli = Cli::try_parse_from(["fabro", "attach", "ABC123"]).expect("should parse");
match cli.command {
Command::Attach { run } => {
assert_eq!(run, "ABC123");
}
_ => panic!("unexpected command variant"),
}
}
#[test]
fn parse_run_engine_command() {
let cli = Cli::try_parse_from(["fabro", "_run_engine", "--run-dir", "/tmp/runs/test"])
.expect("should parse");
match cli.command {
Command::RunEngine { run_dir } => {
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/runs/test"));
}
_ => panic!("unexpected command variant"),
}
}
}

View file

@ -527,3 +527,250 @@ fn detach_conflicts_with_resume() {
.failure()
.stderr(predicate::str::contains("cannot be used with"));
}
// == Bug regression: create/start/attach lifecycle ============================
/// Helper: create a minimal run directory that `resolve_run` can find.
/// Sets up manifest.json, status.json, spec.json, and progress.jsonl.
fn setup_run_dir(
home: &std::path::Path,
run_id: &str,
spec_overrides: serde_json::Value,
progress_lines: &[&str],
) -> std::path::PathBuf {
let run_dir = home.join(".fabro").join("runs").join(run_id);
std::fs::create_dir_all(&run_dir).unwrap();
// manifest.json for resolve_run
let manifest = serde_json::json!({
"run_id": run_id,
"workflow_name": "test",
"goal": "",
"start_time": "2026-01-01T00:00:00Z",
"node_count": 1,
"edge_count": 0
});
std::fs::write(
run_dir.join("manifest.json"),
serde_json::to_string(&manifest).unwrap(),
)
.unwrap();
// Merge spec defaults with overrides
let mut spec = serde_json::json!({
"run_id": run_id,
"workflow_path": "/tmp/test.fabro",
"dot_source": "digraph { start -> exit }",
"working_directory": "/tmp",
"goal": null,
"model": "test-model",
"provider": null,
"sandbox_provider": "local",
"labels": {},
"verbose": false,
"no_retro": true,
"ssh": false,
"preserve_sandbox": false,
"dry_run": true,
"auto_approve": true,
"resume": null,
"run_branch": null
});
if let (Some(base), Some(overrides)) = (spec.as_object_mut(), spec_overrides.as_object()) {
for (k, v) in overrides {
base.insert(k.clone(), v.clone());
}
}
std::fs::write(
run_dir.join("spec.json"),
serde_json::to_string(&spec).unwrap(),
)
.unwrap();
// progress.jsonl
std::fs::write(run_dir.join("progress.jsonl"), progress_lines.join("\n")).unwrap();
run_dir
}
// Bug 2: _run_engine should use cached graph.fabro, not spec.workflow_path.
// When the original workflow file is deleted between create and start,
// the engine should read the snapshot saved at create time.
#[test]
fn bug2_run_engine_uses_cached_graph_not_original_path() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let dot = "\
digraph G {
start [shape=Mdiamond, label=\"Start\"]
exit [shape=Msquare, label=\"Exit\"]
start -> exit
}";
// spec.json: workflow_path points to a file that no longer exists
let spec = serde_json::json!({
"run_id": "test-bug2",
"workflow_path": "/nonexistent/deleted-workflow.fabro",
"dot_source": dot,
"working_directory": run_dir.to_str().unwrap(),
"goal": null,
"model": "test-model",
"provider": null,
"sandbox_provider": "local",
"labels": {},
"verbose": false,
"no_retro": true,
"ssh": false,
"preserve_sandbox": false,
"dry_run": true,
"auto_approve": true,
"resume": null,
"run_branch": null
});
std::fs::write(
run_dir.join("spec.json"),
serde_json::to_string(&spec).unwrap(),
)
.unwrap();
// The cached graph snapshot saved by `fabro create`
std::fs::write(run_dir.join("graph.fabro"), dot).unwrap();
// _run_engine should use graph.fabro and never reference the deleted file.
// Bug: it reads spec.workflow_path → fails with file-not-found.
let output = arc()
.args(["_run_engine", "--run-dir", run_dir.to_str().unwrap()])
.env("NO_COLOR", "1")
.timeout(std::time::Duration::from_secs(15))
.output()
.expect("process should start");
let stderr = String::from_utf8(output.stderr).unwrap();
assert!(
!stderr.contains("deleted-workflow.fabro"),
"bug2: engine should use cached graph.fabro, not the original \
(deleted) workflow path.\nstderr: {stderr}"
);
}
// Bug 3: attach loop must delete interview_request.json after handling it
// to prevent re-prompting the user on the next poll iteration.
#[test]
fn bug3_attach_cleans_up_interview_request_after_handling() {
let home = tempfile::tempdir().unwrap();
let run_dir = setup_run_dir(
home.path(),
"bug3-test",
serde_json::json!({}),
&[
r#"{"ts":"2026-01-01T00:00:01Z","run_id":"bug3","event":"StageStarted","node_id":"gate","name":"Gate","index":0,"attempt":1,"max_attempts":1}"#,
],
);
// Status: running
std::fs::write(
run_dir.join("status.json"),
serde_json::json!({"status": "running", "updated_at": "2026-01-01T00:00:00Z"}).to_string(),
)
.unwrap();
// interview_request.json — a question the engine wrote
let question = serde_json::json!({
"text": "Approve?",
"question_type": "YesNo",
"options": [],
"allow_freeform": false,
"default": {"value": "Yes", "selected_option": null, "selected_options": [], "text": null},
"timeout_seconds": 1.0,
"stage": "gate",
"metadata": {}
});
std::fs::write(
run_dir.join("interview_request.json"),
serde_json::to_string(&question).unwrap(),
)
.unwrap();
// Dead engine so attach exits after one iteration
std::fs::write(run_dir.join("run.pid"), "99999999").unwrap();
// Pipe "y\n" so ConsoleInterviewer doesn't block on stdin
let _ = arc()
.env("HOME", home.path())
.env("NO_COLOR", "1")
.args(["attach", "bug3-test"])
.write_stdin("y\n")
.timeout(std::time::Duration::from_secs(5))
.output();
// Bug: interview_request.json is never deleted by the attach loop.
// After the fix it should be removed immediately after handling.
assert!(
!run_dir.join("interview_request.json").exists(),
"bug3: interview_request.json should be deleted after being handled by attach"
);
}
// Bug 4: attach should respect the verbose flag from spec.json.
// Currently ProgressUI is created with verbose=false regardless of spec.
#[test]
fn bug4_attach_respects_verbose_from_spec() {
let home = tempfile::tempdir().unwrap();
// Use pre-rename field names so handle_json_line can parse them
// (isolates this test from bug 1). With 2 turns and 1 tool call,
// verbose mode should display "(2 turns, 1 tools, …)" in the output.
let run_dir = setup_run_dir(
home.path(),
"bug4-test",
serde_json::json!({"verbose": true}),
&[
r#"{"ts":"2026-01-01T12:00:00Z","run_id":"bug4","event":"StageStarted","node_id":"code","name":"Code","index":0,"attempt":1,"max_attempts":1}"#,
r#"{"ts":"2026-01-01T12:00:01Z","run_id":"bug4","event":"Agent.AssistantMessage","stage":"code","model":"claude-sonnet"}"#,
r#"{"ts":"2026-01-01T12:00:02Z","run_id":"bug4","event":"Agent.AssistantMessage","stage":"code","model":"claude-sonnet"}"#,
r#"{"ts":"2026-01-01T12:00:03Z","run_id":"bug4","event":"Agent.ToolCallStarted","stage":"code","tool_name":"read_file","tool_call_id":"tc1","arguments":{}}"#,
r#"{"ts":"2026-01-01T12:00:04Z","run_id":"bug4","event":"Agent.ToolCallCompleted","stage":"code","tool_name":"read_file","tool_call_id":"tc1","is_error":false}"#,
r#"{"ts":"2026-01-01T12:00:10Z","run_id":"bug4","event":"StageCompleted","node_id":"code","name":"Code","index":0,"duration_ms":10000,"status":"success","usage":{"input_tokens":1000,"output_tokens":500}}"#,
],
);
// Succeeded status + conclusion so attach exits after reading events
std::fs::write(
run_dir.join("status.json"),
serde_json::json!({"status": "succeeded", "updated_at": "2026-01-01T12:00:10Z"})
.to_string(),
)
.unwrap();
std::fs::write(
run_dir.join("conclusion.json"),
serde_json::json!({
"timestamp": "2026-01-01T12:00:10Z",
"status": "success",
"duration_ms": 10000,
"stages": [],
"total_retries": 0
})
.to_string(),
)
.unwrap();
let output = arc()
.env("HOME", home.path())
.env("NO_COLOR", "1")
.args(["attach", "bug4-test"])
.timeout(std::time::Duration::from_secs(10))
.output()
.expect("process should start");
let stderr = String::from_utf8(output.stderr).unwrap();
// Bug: verbose is hardcoded false, so stats are suppressed.
// Fix: load spec.verbose and pass it to ProgressUI.
assert!(
stderr.contains("turns") && stderr.contains("tools"),
"bug4: attach should show verbose stats when spec.verbose=true.\nstderr: {stderr}"
);
}

View file

@ -13,6 +13,7 @@ serde.workspace = true
serde_json.workspace = true
async-trait.workspace = true
tokio.workspace = true
tracing.workspace = true
dialoguer.workspace = true
fabro-util = { path = "../fabro-util" }

View file

@ -0,0 +1,164 @@
use std::path::PathBuf;
use async_trait::async_trait;
use crate::{Answer, Interviewer, Question};
/// An interviewer that communicates via JSON files in the run directory.
///
/// The engine process writes `interview_request.json` and polls for
/// `interview_response.json`. The attach process watches for the request
/// file, prompts the user, and writes the response file.
pub struct FileInterviewer {
run_dir: PathBuf,
}
impl FileInterviewer {
pub fn new(run_dir: PathBuf) -> Self {
Self { run_dir }
}
fn request_path(&self) -> PathBuf {
self.run_dir.join("interview_request.json")
}
fn response_path(&self) -> PathBuf {
self.run_dir.join("interview_response.json")
}
}
#[async_trait]
impl Interviewer for FileInterviewer {
async fn ask(&self, question: Question) -> Answer {
let timeout_secs = question.timeout_seconds;
let default_answer = question.default.clone();
// Write the request file
let request_path = self.request_path();
let json = serde_json::to_string_pretty(&question).expect("Question serialization failed");
if let Err(e) = tokio::fs::write(&request_path, json).await {
tracing::warn!(error = %e, "Failed to write interview request");
return default_answer.unwrap_or_else(Answer::timeout);
}
// Poll for response with optional timeout
let poll = async {
let response_path = self.response_path();
loop {
match tokio::fs::read_to_string(&response_path).await {
Ok(data) => match serde_json::from_str::<Answer>(&data) {
Ok(answer) => {
// Clean up both files
let _ = tokio::fs::remove_file(&request_path).await;
let _ = tokio::fs::remove_file(&response_path).await;
return answer;
}
Err(e) => {
tracing::warn!(error = %e, "Failed to parse interview response, retrying");
// File might be partially written, wait and retry
}
},
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
// Not written yet, poll again
}
Err(e) => {
tracing::warn!(error = %e, "Failed to read interview response, retrying");
}
}
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
}
};
if let Some(secs) = timeout_secs {
let duration = std::time::Duration::from_secs_f64(secs);
match tokio::time::timeout(duration, poll).await {
Ok(answer) => answer,
Err(_) => {
// Clean up request file on timeout
let _ = tokio::fs::remove_file(&self.request_path()).await;
default_answer.unwrap_or_else(Answer::timeout)
}
}
} else {
poll.await
}
}
async fn inform(&self, _message: &str, _stage: &str) {
// No-op: inform messages are rendered by the attach process via progress.jsonl
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{AnswerValue, QuestionType};
#[tokio::test]
async fn write_request_poll_response() {
let dir = tempfile::tempdir().unwrap();
let run_dir = dir.path().to_path_buf();
let interviewer = FileInterviewer::new(run_dir.clone());
let question = Question::new("approve?", QuestionType::YesNo);
// Spawn the ask in a background task
let ask_handle = tokio::spawn(async move { interviewer.ask(question).await });
// Wait for the request file to appear
let request_path = run_dir.join("interview_request.json");
for _ in 0..50 {
if request_path.exists() {
break;
}
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
}
assert!(request_path.exists(), "interview_request.json should exist");
// Verify the request contains valid Question JSON
let request_data = tokio::fs::read_to_string(&request_path).await.unwrap();
let parsed: Question = serde_json::from_str(&request_data).unwrap();
assert_eq!(parsed.text, "approve?");
// Write a response
let answer = Answer::yes();
let response_json = serde_json::to_string_pretty(&answer).unwrap();
let response_path = run_dir.join("interview_response.json");
tokio::fs::write(&response_path, response_json)
.await
.unwrap();
// Wait for the ask to complete
let result = ask_handle.await.unwrap();
assert_eq!(result.value, AnswerValue::Yes);
// Both files should be cleaned up
assert!(!request_path.exists());
assert!(!response_path.exists());
}
#[tokio::test]
async fn timeout_returns_default() {
let dir = tempfile::tempdir().unwrap();
let interviewer = FileInterviewer::new(dir.path().to_path_buf());
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.1);
question.default = Some(Answer::no());
let answer = interviewer.ask(question).await;
assert_eq!(answer.value, AnswerValue::No);
}
#[tokio::test]
async fn timeout_without_default_returns_timeout() {
let dir = tempfile::tempdir().unwrap();
let interviewer = FileInterviewer::new(dir.path().to_path_buf());
let mut question = Question::new("approve?", QuestionType::YesNo);
question.timeout_seconds = Some(0.1);
let answer = interviewer.ask(question).await;
assert_eq!(answer.value, AnswerValue::Timeout);
}
}

View file

@ -1,6 +1,7 @@
mod auto_approve;
mod callback;
mod console;
pub mod file;
mod queue;
mod recording;
mod replay;
@ -203,6 +204,7 @@ pub trait Interviewer: Send + Sync {
pub use auto_approve::AutoApproveInterviewer;
pub use callback::CallbackInterviewer;
pub use console::ConsoleInterviewer;
pub use file::FileInterviewer;
pub use queue::QueueInterviewer;
pub use recording::RecordingInterviewer;
pub use replay::ReplayInterviewer;

View file

@ -116,6 +116,7 @@ pub mod pull_request;
pub mod run_fork;
pub mod run_lookup;
pub mod run_rewind;
pub mod run_spec;
pub mod run_status;
pub mod sandbox_provider;
pub mod sandbox_reconnect;

View file

@ -0,0 +1,89 @@
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunSpec {
pub run_id: String,
pub workflow_path: PathBuf,
pub dot_source: String,
pub working_directory: PathBuf,
pub goal: Option<String>,
pub model: String,
pub provider: Option<String>,
pub sandbox_provider: String,
pub labels: HashMap<String, String>,
pub verbose: bool,
pub no_retro: bool,
pub ssh: bool,
pub preserve_sandbox: bool,
pub dry_run: bool,
pub auto_approve: bool,
pub resume: Option<PathBuf>,
pub run_branch: Option<String>,
}
impl RunSpec {
pub fn save(&self, run_dir: &Path) -> anyhow::Result<()> {
let path = run_dir.join("spec.json");
let json = serde_json::to_string_pretty(self)?;
std::fs::write(path, json)?;
Ok(())
}
pub fn load(run_dir: &Path) -> anyhow::Result<Self> {
let path = run_dir.join("spec.json");
let json = std::fs::read_to_string(path)?;
let spec = serde_json::from_str(&json)?;
Ok(spec)
}
}
#[cfg(test)]
mod tests {
use super::*;
fn sample_spec() -> RunSpec {
let mut labels = HashMap::new();
labels.insert("env".to_string(), "test".to_string());
labels.insert("team".to_string(), "platform".to_string());
RunSpec {
run_id: "run-abc123".to_string(),
workflow_path: PathBuf::from("/home/user/workflows/deploy/workflow.toml"),
dot_source: "digraph { a -> b }".to_string(),
working_directory: PathBuf::from("/home/user/project"),
goal: Some("Deploy to staging".to_string()),
model: "claude-sonnet-4-20250514".to_string(),
provider: Some("anthropic".to_string()),
sandbox_provider: "local".to_string(),
labels,
verbose: true,
no_retro: false,
ssh: true,
preserve_sandbox: false,
dry_run: false,
auto_approve: true,
resume: Some(PathBuf::from("/tmp/checkpoint")),
run_branch: Some("fabro/run/abc123".to_string()),
}
}
#[test]
fn save_load_roundtrip() {
let dir = tempfile::tempdir().unwrap();
let spec = sample_spec();
spec.save(dir.path()).unwrap();
let loaded = RunSpec::load(dir.path()).unwrap();
assert_eq!(loaded, spec);
}
#[test]
fn load_nonexistent() {
let dir = PathBuf::from("/tmp/nonexistent-run-spec-dir-that-does-not-exist");
assert!(RunSpec::load(&dir).is_err());
}
}