scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-15 23:32:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Use store-backed status for attach and run lookup

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-01 21:34:46 -04:00
parent 0945b6b9ba
commit 664aa1d6d6
No known key found for this signature in database
3 changed files with 140 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
lib/crates/fabro-cli/src/commands/run/attach.rs
View file

@ -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<RunStatusRe
}
}
async fn determine_exit_code_with_store(run_store: &dyn RunStore, run_dir: &Path) -> 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;
}
}

54
lib/crates/fabro-cli/tests/it/cmd/attach.rs
View file

@ -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",

89
lib/crates/fabro-workflow/src/run_lookup.rs
View file

@ -64,6 +64,14 @@ pub fn default_runs_base() -> PathBuf {
}
pub fn scan_runs(base: &Path) -> Result<Vec<RunInfo>> {
scan_runs_inner(base, true)
}
fn scan_runs_without_status(base: &Path) -> Result<Vec<RunInfo>> {
scan_runs_inner(base, false)
}
fn scan_runs_inner(base: &Path, include_status: bool) -> Result<Vec<RunInfo>> {
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<Vec<RunInfo>> {
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<Vec<RunInfo>> {
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<Vec<RunInfo>> {
}
pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result<Vec<RunInfo>> {
let mut runs_by_id: HashMap<RunId, RunInfo> = scan_runs(base)?
let mut runs_by_id: HashMap<RunId, RunInfo> = 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<RunId> {
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);
}
}