From 3680a57a25d969593ae3c7c3bb199228ea788073 Mon Sep 17 00:00:00 2001
From: "brynary-fabro[bot]"
<265161896+brynary-fabro[bot]@users.noreply.github.com>
Date: Fri, 20 Mar 2026 14:08:54 -0400
Subject: [PATCH] Decompose `fabro run` into `create` / `start` / `attach`
(#116)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
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
Ran 9 stages in 30m 55s for $8.55
| 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** |
Ran ImplementAndSimplify.fabro (12 nodes and 15
edges)
```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
}
```
⚒️ Generated with [Fabro](https://fabro.sh)
---------
Co-authored-by: Fabro
Co-authored-by: Bryan Helmkamp
Co-authored-by: Claude Opus 4.6 (1M context)
---
Cargo.lock | 1 +
lib/crates/fabro-cli/src/commands/attach.rs | 221 +++++++
lib/crates/fabro-cli/src/commands/create.rs | 92 +++
lib/crates/fabro-cli/src/commands/mod.rs | 5 +-
lib/crates/fabro-cli/src/commands/run.rs | 172 ++++--
.../fabro-cli/src/commands/run_progress.rs | 563 +++++++++++++++++-
lib/crates/fabro-cli/src/commands/start.rs | 113 ++++
lib/crates/fabro-cli/src/main.rs | 272 ++++++---
lib/crates/fabro-cli/tests/cli.rs | 247 ++++++++
lib/crates/fabro-interview/Cargo.toml | 1 +
lib/crates/fabro-interview/src/file.rs | 164 +++++
lib/crates/fabro-interview/src/lib.rs | 2 +
lib/crates/fabro-workflows/src/lib.rs | 1 +
lib/crates/fabro-workflows/src/run_spec.rs | 89 +++
14 files changed, 1768 insertions(+), 175 deletions(-)
create mode 100644 lib/crates/fabro-cli/src/commands/attach.rs
create mode 100644 lib/crates/fabro-cli/src/commands/create.rs
create mode 100644 lib/crates/fabro-cli/src/commands/start.rs
create mode 100644 lib/crates/fabro-interview/src/file.rs
create mode 100644 lib/crates/fabro-workflows/src/run_spec.rs
diff --git a/Cargo.lock b/Cargo.lock
index d14944a41..7f91c51ce 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1514,6 +1514,7 @@ dependencies = [
"serde_json",
"tempfile",
"tokio",
+ "tracing",
]
[[package]]
diff --git a/lib/crates/fabro-cli/src/commands/attach.rs b/lib/crates/fabro-cli/src/commands/attach.rs
new file mode 100644
index 000000000..f500b513f
--- /dev/null
+++ b/lib/crates/fabro-cli/src/commands/attach.rs
@@ -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 {
+ 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 = 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::(&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::() {
+ 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,
+ 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::() {
+ #[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
+ }
+}
diff --git a/lib/crates/fabro-cli/src/commands/create.rs b/lib/crates/fabro-cli/src/commands/create.rs
new file mode 100644
index 000000000..a226f2c0e
--- /dev/null
+++ b/lib/crates/fabro-cli/src/commands/create.rs
@@ -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))
+}
diff --git a/lib/crates/fabro-cli/src/commands/mod.rs b/lib/crates/fabro-cli/src/commands/mod.rs
index 6c4535f57..64dea84b1 100644
--- a/lib/crates/fabro-cli/src/commands/mod.rs
+++ b/lib/crates/fabro-cli/src/commands/mod.rs
@@ -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;
diff --git a/lib/crates/fabro-cli/src/commands/run.rs b/lib/crates/fabro-cli/src/commands/run.rs
index 8bfa9d51a..a5ba4dc05 100644
--- a/lib/crates/fabro-cli/src/commands/run.rs
+++ b/lib/crates/fabro-cli/src/commands/run.rs
@@ -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 for SandboxProvider {
}
}
+impl From 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,
goal_file: &Option,
) -> anyhow::Result> {
@@ -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> {
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,
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,
- 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,
+ pub sandbox_provider: SandboxProvider,
+ pub model: String,
+ pub provider: Option,
+ 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 {
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 = 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 = 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,
+ 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 = 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 = 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 = 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()
diff --git a/lib/crates/fabro-cli/src/commands/run_progress.rs b/lib/crates/fabro-cli/src/commands/run_progress.rs
index 5154c3f4a..8e8214277 100644
--- a/lib/crates/fabro-cli/src/commands/run_progress.rs
+++ b/lib/crates/fabro-cli/src/commands/run_progress.rs
@@ -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>) -> 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"
+ );
+ }
}
diff --git a/lib/crates/fabro-cli/src/commands/start.rs b/lib/crates/fabro-cli/src/commands/start.rs
new file mode 100644
index 000000000..b5791df2e
--- /dev/null
+++ b/lib/crates/fabro-cli/src/commands/start.rs
@@ -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 {
+ // 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"
+ );
+ }
+}
diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs
index f43707989..7a97ddf4f 100644
--- a/lib/crates/fabro-cli/src/main.rs
+++ b/lib/crates/fabro-cli/src/main.rs
@@ -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 = Vec::new();
- child_args.push("run".to_string());
-
- let raw_args: Vec = 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::()
+ .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"),
+ }
+ }
}
diff --git a/lib/crates/fabro-cli/tests/cli.rs b/lib/crates/fabro-cli/tests/cli.rs
index 60fdd6abc..302c3a51d 100644
--- a/lib/crates/fabro-cli/tests/cli.rs
+++ b/lib/crates/fabro-cli/tests/cli.rs
@@ -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}"
+ );
+}
diff --git a/lib/crates/fabro-interview/Cargo.toml b/lib/crates/fabro-interview/Cargo.toml
index 38b2b3d69..ef57afbe6 100644
--- a/lib/crates/fabro-interview/Cargo.toml
+++ b/lib/crates/fabro-interview/Cargo.toml
@@ -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" }
diff --git a/lib/crates/fabro-interview/src/file.rs b/lib/crates/fabro-interview/src/file.rs
new file mode 100644
index 000000000..2a7e41dc0
--- /dev/null
+++ b/lib/crates/fabro-interview/src/file.rs
@@ -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::(&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);
+ }
+}
diff --git a/lib/crates/fabro-interview/src/lib.rs b/lib/crates/fabro-interview/src/lib.rs
index 12ac75daa..31879dd2a 100644
--- a/lib/crates/fabro-interview/src/lib.rs
+++ b/lib/crates/fabro-interview/src/lib.rs
@@ -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;
diff --git a/lib/crates/fabro-workflows/src/lib.rs b/lib/crates/fabro-workflows/src/lib.rs
index 012c6f739..f77355620 100644
--- a/lib/crates/fabro-workflows/src/lib.rs
+++ b/lib/crates/fabro-workflows/src/lib.rs
@@ -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;
diff --git a/lib/crates/fabro-workflows/src/run_spec.rs b/lib/crates/fabro-workflows/src/run_spec.rs
new file mode 100644
index 000000000..df5dbac52
--- /dev/null
+++ b/lib/crates/fabro-workflows/src/run_spec.rs
@@ -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,
+ pub model: String,
+ pub provider: Option,
+ pub sandbox_provider: String,
+ pub labels: HashMap,
+ 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,
+ pub run_branch: Option,
+}
+
+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 {
+ 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());
+ }
+}