From 664aa1d6d62eef00ea110ff8486f835fe78c84d1 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 21:34:46 -0400 Subject: [PATCH] Use store-backed status for attach and run lookup --- .../fabro-cli/src/commands/run/attach.rs | 48 +++++----- lib/crates/fabro-cli/tests/it/cmd/attach.rs | 54 +++++------ lib/crates/fabro-workflow/src/run_lookup.rs | 89 ++++++++++++++++++- 3 files changed, 140 insertions(+), 51 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 532ed545e..74a48e3a5 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -29,6 +29,7 @@ const ATTACH_STARTUP_GRACE: Duration = Duration::from_secs(3); const INTERVIEW_UNANSWERED_MESSAGE: &str = "Interview ended without an answer. The run is still waiting for input; reattach to answer it."; const JSON_INTERVIEW_MESSAGE: &str = "This run is waiting for human input, but --json is non-interactive. Reattach without --json to answer it."; +const ATTACH_FINAL_STATUS_GRACE: Duration = Duration::from_secs(2); /// Attach to a running (or finished) workflow run, rendering progress live. /// @@ -245,7 +246,6 @@ async fn attach_run_store( .await .ok() .flatten() - .or_else(|| read_status_record(&run_dir.join("status.json"))) .map(|record| record.status) .filter(|status| status.is_terminal()); @@ -284,7 +284,7 @@ async fn attach_run_store( finish_progress(&mut progress_ui, json_output); - Ok(determine_exit_code_with_store(run_store, run_dir).await) + Ok(determine_exit_code_with_store(run_store).await) } async fn attach_run_files( @@ -711,25 +711,33 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option ExitCode { - if let Ok(Some(conclusion)) = run_store.get_conclusion().await { - let success = matches!( - conclusion.status, - StageStatus::Success | StageStatus::PartialSuccess - ); - if success { - ExitCode::from(0) - } else { - ExitCode::from(1) +async fn determine_exit_code_with_store(run_store: &dyn RunStore) -> ExitCode { + let deadline = Instant::now() + ATTACH_FINAL_STATUS_GRACE; + loop { + if let Ok(Some(conclusion)) = run_store.get_conclusion().await { + let success = matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess + ); + return if success { + ExitCode::from(0) + } else { + ExitCode::from(1) + }; } - } else { - let status_path = run_dir.join("status.json"); - let conclusion_path = run_dir.join("conclusion.json"); - let status_record = match run_store.get_status().await { - Ok(record) => record.or_else(|| read_status_record(&status_path)), - Err(_) => read_status_record(&status_path), - }; - determine_exit_code(&conclusion_path, status_record) + + match run_store.get_status().await { + Ok(Some(record)) if matches!(record.status, RunStatus::Succeeded) => { + return ExitCode::from(0); + } + Ok(Some(record)) if record.status.is_terminal() => return ExitCode::from(1), + Ok(Some(_)) | Ok(None) | Err(_) => {} + } + + if Instant::now() >= deadline { + return ExitCode::from(1); + } + sleep(Duration::from_millis(100)).await; } } diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 750652c86..38b2d93d7 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -289,9 +289,31 @@ fn attach_json_errors_without_prompting_for_human_input() { fabro_json_snapshot!(context, &progress, @r#" [ { - "event": "run.created", "id": "[EVENT_ID]", + "ts": "[TIMESTAMP]", + "run_id": "[ULID]", + "event": "run.created", "properties": { + "workflow_slug": "human-gate", + "settings": { + "goal": "Wait for approval", + "llm": { + "fallbacks": null, + "model": "gpt-5.4", + "provider": "openai" + }, + "mode": "standalone", + "no_retro": true, + "sandbox": { + "daytona": null, + "devcontainer": null, + "env": null, + "local": null, + "preserve": null, + "provider": "local" + }, + "storage_dir": "[STORAGE_DIR]" + }, "graph": { "attrs": { "goal": { @@ -392,34 +414,12 @@ fn attach_json_errors_without_prompting_for_human_input() { } } }, - "host_repo_path": "[TEMP_DIR]", + "workflow_source": "digraph HumanGate {/n graph [goal=\"Wait for approval\"]/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n approve [shape=hexagon, label=\"Approve?\"]/n ship [shape=parallelogram, script=\"echo shipped\"]/n revise [shape=parallelogram, script=\"echo revised\"]/n start -> approve/n approve -> ship [label=\"[A] Approve\"]/n approve -> revise [label=\"[R] Revise\"]/n ship -> exit/n revise -> exit/n}/n", "labels": {}, "run_dir": "[STORAGE_DIR]/runs/20260401-[ULID]", - "settings": { - "goal": "Wait for approval", - "llm": { - "fallbacks": null, - "model": "gpt-5.4", - "provider": "openai" - }, - "mode": "standalone", - "no_retro": true, - "sandbox": { - "daytona": null, - "devcontainer": null, - "env": null, - "local": null, - "preserve": null, - "provider": "local" - }, - "storage_dir": "[STORAGE_DIR]" - }, - "workflow_slug": "human-gate", - "workflow_source": "digraph HumanGate {/n graph [goal=\"Wait for approval\"]/n start [shape=Mdiamond, label=\"Start\"]/n exit [shape=Msquare, label=\"Exit\"]/n approve [shape=hexagon, label=\"Approve?\"]/n ship [shape=parallelogram, script=\"echo shipped\"]/n revise [shape=parallelogram, script=\"echo revised\"]/n start -> approve/n approve -> ship [label=\"[A] Approve\"]/n approve -> revise [label=\"[R] Revise\"]/n ship -> exit/n revise -> exit/n}/n", - "working_directory": "[TEMP_DIR]" - }, - "run_id": "[ULID]", - "ts": "[TIMESTAMP]" + "working_directory": "[TEMP_DIR]", + "host_repo_path": "[TEMP_DIR]" + } }, { "event": "sandbox.initializing", diff --git a/lib/crates/fabro-workflow/src/run_lookup.rs b/lib/crates/fabro-workflow/src/run_lookup.rs index a6cd9c983..25cee66e2 100644 --- a/lib/crates/fabro-workflow/src/run_lookup.rs +++ b/lib/crates/fabro-workflow/src/run_lookup.rs @@ -64,6 +64,14 @@ pub fn default_runs_base() -> PathBuf { } pub fn scan_runs(base: &Path) -> Result> { + scan_runs_inner(base, true) +} + +fn scan_runs_without_status(base: &Path) -> Result> { + scan_runs_inner(base, false) +} + +fn scan_runs_inner(base: &Path, include_status: bool) -> Result> { let entries = match std::fs::read_dir(base) { Ok(entries) => entries, Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), @@ -88,7 +96,11 @@ pub fn scan_runs(base: &Path) -> Result> { let start_time = start_time_dt.to_rfc3339(); let workflow_name = record.workflow_name().to_string(); let goal = record.goal().to_string(); - let status_info = read_status(&path); + let status_info = if include_status { + read_status(&path) + } else { + StatusInfo::simple(RunStatus::Dead) + }; runs.push(RunInfo { run_id: record.run_id, @@ -124,8 +136,12 @@ pub fn scan_runs(base: &Path) -> Result> { continue; }; - let status_info = read_status(&path); - let is_orphan = matches!(status_info.status, RunStatus::Dead); + let status_info = if include_status { + read_status(&path) + } else { + StatusInfo::simple(RunStatus::Dead) + }; + let is_orphan = !include_status || matches!(status_info.status, RunStatus::Dead); runs.push(RunInfo { run_id, dir_name, @@ -157,7 +173,7 @@ pub fn scan_runs(base: &Path) -> Result> { } pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result> { - let mut runs_by_id: HashMap = scan_runs(base)? + let mut runs_by_id: HashMap = scan_runs_without_status(base)? .into_iter() .map(|run| (run.run_id, run)) .collect(); @@ -433,3 +449,68 @@ fn parse_run_id(value: &str) -> Option { fn run_id_matches(run_id: RunId, prefix: &str) -> bool { run_id.to_string().starts_with(prefix) } + +#[cfg(test)] +mod tests { + use std::collections::HashMap; + use std::path::PathBuf; + + use chrono::Utc; + use fabro_config::FabroSettings; + use fabro_graphviz::graph::Graph; + use fabro_store::{InMemoryStore, Store}; + use fabro_types::{RunStatus, RunStatusRecord, fixtures}; + + use super::scan_runs_combined; + use crate::records::{RunRecord, RunRecordExt}; + + fn sample_run_record() -> RunRecord { + RunRecord { + run_id: fixtures::RUN_1, + created_at: Utc::now(), + settings: FabroSettings::default(), + graph: Graph::new("test"), + workflow_slug: Some("test".to_string()), + working_directory: PathBuf::from("/tmp/project"), + host_repo_path: Some("/tmp/project".to_string()), + base_branch: Some("main".to_string()), + labels: HashMap::new(), + } + } + + #[tokio::test] + async fn scan_runs_combined_uses_store_status_without_status_json() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join(fixtures::RUN_1.to_string()); + std::fs::create_dir_all(&run_dir).unwrap(); + + let run_record = sample_run_record(); + run_record.save(&run_dir).unwrap(); + std::fs::write(run_dir.join("id.txt"), format!("{}\n", fixtures::RUN_1)).unwrap(); + + let store = InMemoryStore::default(); + let run_dir_string = run_dir.to_string_lossy().to_string(); + let run_store = store + .create_run( + &fixtures::RUN_1, + run_record.created_at, + Some(&run_dir_string), + ) + .await + .unwrap(); + run_store.put_run(&run_record).await.unwrap(); + run_store + .put_status(&RunStatusRecord::new(RunStatus::Submitted, None)) + .await + .unwrap(); + + let runs = scan_runs_combined(&store, temp.path()).await.unwrap(); + let run = runs + .iter() + .find(|run| run.run_id == fixtures::RUN_1) + .expect("run should be listed"); + + assert_eq!(run.status, RunStatus::Submitted); + assert!(!run.is_orphan); + } +}