From 87345604dd77b3acc1667de720e3daae1344cf6a Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sat, 28 Mar 2026 23:23:38 -0400 Subject: [PATCH] Fix remaining CLI store migration gaps --- lib/crates/fabro-cli/src/commands/pr/close.rs | 2 +- lib/crates/fabro-cli/src/commands/pr/list.rs | 10 +- lib/crates/fabro-cli/src/commands/pr/merge.rs | 2 +- lib/crates/fabro-cli/src/commands/pr/mod.rs | 13 +- lib/crates/fabro-cli/src/commands/pr/view.rs | 2 +- .../fabro-cli/src/commands/run/attach.rs | 400 ++++++++++++++---- .../fabro-cli/src/commands/run/command.rs | 3 +- lib/crates/fabro-cli/src/commands/run/logs.rs | 87 ++-- lib/crates/fabro-cli/src/commands/run/mod.rs | 12 +- .../fabro-cli/src/commands/run/resume.rs | 9 +- lib/crates/fabro-cli/src/commands/runs/rm.rs | 29 +- .../fabro-cli/src/commands/system/df.rs | 24 +- .../fabro-cli/src/commands/system/mod.rs | 6 +- .../fabro-cli/src/commands/system/prune.rs | 16 +- lib/crates/fabro-cli/src/main.rs | 2 +- lib/crates/fabro-workflows/src/run_lookup.rs | 3 + 16 files changed, 479 insertions(+), 141 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/pr/close.rs b/lib/crates/fabro-cli/src/commands/pr/close.rs index 04d28618b..b7b4cd831 100644 --- a/lib/crates/fabro-cli/src/commands/pr/close.rs +++ b/lib/crates/fabro-cli/src/commands/pr/close.rs @@ -22,7 +22,7 @@ async fn close_from( args: PrCloseArgs, github_app: Option, ) -> Result<()> { - let (record, _run_dir) = super::load_pr_record(base, &args.run_id)?; + let (record, _run_dir) = super::load_pr_record(base, &args.run_id).await?; let creds = github_app.context( "GitHub App credentials required — set GITHUB_APP_PRIVATE_KEY and configure app_id", diff --git a/lib/crates/fabro-cli/src/commands/pr/list.rs b/lib/crates/fabro-cli/src/commands/pr/list.rs index 950f36bad..f12b1fb5c 100644 --- a/lib/crates/fabro-cli/src/commands/pr/list.rs +++ b/lib/crates/fabro-cli/src/commands/pr/list.rs @@ -3,7 +3,7 @@ use std::path::Path; use anyhow::{Context, Result}; use fabro_config::FabroSettingsExt; use fabro_workflows::pull_request::PullRequestRecord; -use fabro_workflows::run_lookup::{runs_base, scan_runs}; +use fabro_workflows::run_lookup::{runs_base, scan_runs_combined}; use futures::future::join_all; use tracing::info; @@ -16,10 +16,12 @@ pub(super) async fn list_command( ) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - list_from(&base, args, github_app).await + let store = crate::store::build_store(&cli_settings.storage_dir())?; + list_from(store.as_ref(), &base, args, github_app).await } async fn list_from( + store: &dyn fabro_store::Store, base: &Path, args: PrListArgs, github_app: Option, @@ -36,7 +38,9 @@ async fn list_from( "GitHub App credentials required — set GITHUB_APP_PRIVATE_KEY and configure app_id", )?; - let runs = scan_runs(base).context("Failed to scan runs")?; + let runs = scan_runs_combined(store, base) + .await + .context("Failed to scan runs")?; let mut entries: Vec<(String, PullRequestRecord)> = Vec::new(); for run in &runs { diff --git a/lib/crates/fabro-cli/src/commands/pr/merge.rs b/lib/crates/fabro-cli/src/commands/pr/merge.rs index b68294636..ab6652986 100644 --- a/lib/crates/fabro-cli/src/commands/pr/merge.rs +++ b/lib/crates/fabro-cli/src/commands/pr/merge.rs @@ -23,7 +23,7 @@ async fn merge_from( args: PrMergeArgs, github_app: Option, ) -> Result<()> { - let (record, _run_dir) = super::load_pr_record(base, &args.run_id)?; + let (record, _run_dir) = super::load_pr_record(base, &args.run_id).await?; let creds = github_app.context( "GitHub App credentials required — set GITHUB_APP_PRIVATE_KEY and configure app_id", diff --git a/lib/crates/fabro-cli/src/commands/pr/mod.rs b/lib/crates/fabro-cli/src/commands/pr/mod.rs index 55cdc609d..6d33b3f7c 100644 --- a/lib/crates/fabro-cli/src/commands/pr/mod.rs +++ b/lib/crates/fabro-cli/src/commands/pr/mod.rs @@ -9,7 +9,7 @@ use std::path::{Path, PathBuf}; use anyhow::{Context, Result}; use fabro_workflows::pull_request::PullRequestRecord; -use fabro_workflows::run_lookup::resolve_run; +use fabro_workflows::run_lookup::resolve_run_combined; use crate::args::{PrCommand, PrNamespace}; use crate::cli_config::load_cli_settings; @@ -28,8 +28,15 @@ pub(crate) async fn dispatch(ns: PrNamespace) -> Result<()> { } } -pub(crate) fn load_pr_record(base: &Path, run_id: &str) -> Result<(PullRequestRecord, PathBuf)> { - let run_dir = resolve_run(base, run_id)?.path; +pub(crate) async fn load_pr_record( + base: &Path, + run_id: &str, +) -> Result<(PullRequestRecord, PathBuf)> { + let storage_dir = base.parent().unwrap_or(base); + let store = crate::store::build_store(storage_dir)?; + let run_dir = resolve_run_combined(store.as_ref(), base, run_id) + .await? + .path; let pr_path = run_dir.join("pull_request.json"); let content = std::fs::read_to_string(&pr_path).with_context(|| { format!( diff --git a/lib/crates/fabro-cli/src/commands/pr/view.rs b/lib/crates/fabro-cli/src/commands/pr/view.rs index cfda851da..340cf1c94 100644 --- a/lib/crates/fabro-cli/src/commands/pr/view.rs +++ b/lib/crates/fabro-cli/src/commands/pr/view.rs @@ -23,7 +23,7 @@ async fn view_from( args: PrViewArgs, github_app: Option, ) -> Result<()> { - let (record, _run_dir) = super::load_pr_record(base, &args.run_id)?; + let (record, _run_dir) = super::load_pr_record(base, &args.run_id).await?; let creds = github_app.context( "GitHub App credentials required — set GITHUB_APP_PRIVATE_KEY and configure app_id", diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index eae061c3a..257d84d35 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -6,9 +6,11 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, Instant}; use anyhow::{Result, bail}; +use fabro_config::FabroSettingsExt; +use futures::StreamExt; use fabro_interview::{AnswerValue, ConsoleInterviewer}; -use fabro_store::RuntimeState; +use fabro_store::{EventEnvelope, RunStore, RuntimeState}; use fabro_util::terminal::Styles; use fabro_workflows::outcome::StageStatus; use fabro_workflows::records::{Conclusion, ConclusionExt, RunRecord, RunRecordExt}; @@ -30,13 +32,67 @@ const INTERVIEW_UNANSWERED_MESSAGE: &str = /// Returns exit code 0 for success/partial_success, 1 otherwise. pub(crate) async fn attach_run( run_dir: &Path, + run_id: Option<&str>, + kill_on_detach: bool, + styles: &'static Styles, + engine_child: Option, +) -> Result { + let run_record = RunRecord::load(run_dir).ok(); + if let (Some(storage_dir), Some(run_id)) = ( + run_record + .as_ref() + .map(|record| record.settings.storage_dir()), + run_id.or_else(|| run_record.as_ref().map(|record| record.run_id.as_str())), + ) { + match crate::store::open_run_reader(&storage_dir, run_id).await { + Ok(Some(run_store)) => match run_store.list_events().await { + Ok(events) => { + let event_lines = events + .iter() + .map(event_payload_line) + .collect::>>()?; + return attach_run_store( + run_dir, + run_store.as_ref(), + event_lines, + events.last().map(|event| event.seq).unwrap_or(0), + kill_on_detach, + styles, + engine_child, + ) + .await; + } + Err(err) => { + tracing::warn!( + run_id, + error = %err, + "Failed to list events from store; falling back to filesystem attach" + ); + } + }, + Ok(None) => {} + Err(err) => { + tracing::warn!( + run_id, + error = %err, + "Failed to open store reader; falling back to filesystem attach" + ); + } + } + } + + attach_run_files(run_dir, kill_on_detach, styles, engine_child).await +} + +async fn attach_run_store( + run_dir: &Path, + run_store: &dyn RunStore, + existing_events: Vec, + last_seq: u32, kill_on_detach: bool, styles: &'static Styles, engine_child: Option, ) -> Result { - let progress_path = run_dir.join("progress.jsonl"); - let conclusion_path = run_dir.join("conclusion.json"); - let status_path = run_dir.join("status.json"); let runtime_state = RuntimeState::new(run_dir); let runtime_interview_paths = InterviewPaths::from_runtime_state(&runtime_state); @@ -58,68 +114,13 @@ pub(crate) async fn attach_run( }); } - // Wait for progress.jsonl to appear. - // If the engine dies during early init (before any event is emitted), - // progress.jsonl may never be created. Check for terminal status or - // engine death so we surface the real failure instead of timing out. - let mut wait_count = 0; - while !progress_path.exists() { - sleep(std::time::Duration::from_millis(100)).await; - wait_count += 1; - - // Check if engine died before writing any progress - if let Some(record) = read_status_record(&status_path) { - if record.status.is_terminal() { - progress_ui.finish(); - return Ok(determine_exit_code(&conclusion_path, Some(record))); - } - } - - if let Some(guard) = engine_guard.as_mut() { - if let Some(child) = guard.inner() { - if matches!(child.try_wait(), Ok(Some(_))) { - // Engine exited without writing progress.jsonl - progress_ui.finish(); - return Ok(determine_exit_code( - &conclusion_path, - read_status_record(&status_path), - )); - } - } - } - - if let Some(pid) = read_launcher_pid(run_dir) { - if !process_alive(pid) && wait_count > 5 { - progress_ui.finish(); - return Ok(determine_exit_code( - &conclusion_path, - read_status_record(&status_path), - )); - } - } - - if wait_count > 100 { - // Guard's Drop kills+waits on the engine child - drop(engine_guard.take()); - bail!( - "Timed out waiting for progress.jsonl to appear in {}", - run_dir.display() - ); - } - if cancelled.load(Ordering::Relaxed) { - if !kill_on_detach { - if let Some(guard) = engine_guard.as_mut() { - guard.defuse(); - } - } - // Guard's Drop kills+waits when kill_on_detach is true - return Ok(ExitCode::from(1)); - } + for line in &existing_events { + progress_ui.handle_json_line(line); } - let file = std::fs::File::open(&progress_path)?; - let mut reader = BufReader::new(file); - let mut line = String::new(); + let mut stream = run_store + .watch_events_from(if last_seq == 0 { 1 } else { last_seq + 1 }) + .await?; let mut cached_pid: Option = None; let attach_started = Instant::now(); @@ -135,8 +136,12 @@ pub(crate) async fn attach_run( } // Wait briefly for a terminal status or conclusion for _ in 0..20 { - if conclusion_path.exists() - || read_status_record(&status_path) + if run_store.get_conclusion().await.ok().flatten().is_some() + || run_store + .get_status() + .await + .ok() + .flatten() .is_some_and(|record| record.status.is_terminal()) { break; @@ -152,17 +157,15 @@ pub(crate) async fn attach_run( 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); + let mut saw_event = false; + match tokio::time::timeout(Duration::from_millis(100), stream.next()).await { + Ok(Some(Ok(event))) => { + let line = event_payload_line(&event)?; + progress_ui.handle_json_line(&line); + saw_event = true; } + Ok(Some(Err(err))) => return Err(err.into()), + Ok(None) | Err(_) => {} } // Check for interview request @@ -206,7 +209,11 @@ pub(crate) async fn attach_run( } } - let terminal_status = read_status_record(&status_path) + let terminal_status = run_store + .get_status() + .await + .ok() + .flatten() .map(|record| record.status) .filter(|status| status.is_terminal()); @@ -217,6 +224,211 @@ pub(crate) async fn attach_run( }) }); + if let Some(child_alive) = child_alive_via_handle { + if !child_alive && !saw_event { + break; + } + } else { + if terminal_status.is_some() && !saw_event { + break; + } + + let engine_alive = match cached_pid { + Some(pid) => process_alive(pid), + None => { + if let Some(pid) = read_launcher_pid(run_dir) { + cached_pid = Some(pid); + process_alive(pid) + } else { + attach_started.elapsed() < ATTACH_STARTUP_GRACE || last_seq > 0 + } + } + }; + if !engine_alive { + break; + } + } + } + + progress_ui.finish(); + + Ok(determine_exit_code_with_store(run_store, run_dir).await) +} + +async fn attach_run_files( + run_dir: &Path, + kill_on_detach: bool, + styles: &'static Styles, + engine_child: Option, +) -> Result { + let progress_path = run_dir.join("progress.jsonl"); + let conclusion_path = run_dir.join("conclusion.json"); + let status_path = run_dir.join("status.json"); + let runtime_state = RuntimeState::new(run_dir); + let runtime_interview_paths = InterviewPaths::from_runtime_state(&runtime_state); + + let mut engine_guard = engine_child.map(EngineChildGuard::new); + + let is_tty = std::io::stderr().is_terminal(); + let verbose = RunRecord::load(run_dir) + .map(|record| record.settings.verbose_enabled()) + .unwrap_or(false); + let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose); + + let cancelled = Arc::new(AtomicBool::new(false)); + { + let cancelled = Arc::clone(&cancelled); + tokio::spawn(async move { + let _ = ctrl_c().await; + cancelled.store(true, Ordering::Relaxed); + }); + } + + let mut wait_count = 0; + while !progress_path.exists() { + sleep(std::time::Duration::from_millis(100)).await; + wait_count += 1; + + if let Some(record) = read_status_record(&status_path) { + if record.status.is_terminal() { + progress_ui.finish(); + return Ok(determine_exit_code(&conclusion_path, Some(record))); + } + } + + if let Some(guard) = engine_guard.as_mut() { + if let Some(child) = guard.inner() { + if matches!(child.try_wait(), Ok(Some(_))) { + progress_ui.finish(); + return Ok(determine_exit_code( + &conclusion_path, + read_status_record(&status_path), + )); + } + } + } + + if let Some(pid) = read_launcher_pid(run_dir) { + if !process_alive(pid) && wait_count > 5 { + progress_ui.finish(); + return Ok(determine_exit_code( + &conclusion_path, + read_status_record(&status_path), + )); + } + } + + if wait_count > 100 { + drop(engine_guard.take()); + bail!( + "Timed out waiting for progress.jsonl to appear in {}", + run_dir.display() + ); + } + if cancelled.load(Ordering::Relaxed) { + if !kill_on_detach { + if let Some(guard) = engine_guard.as_mut() { + guard.defuse(); + } + } + return Ok(ExitCode::from(1)); + } + } + + 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; + let attach_started = Instant::now(); + + loop { + if cancelled.load(Ordering::Relaxed) { + if kill_on_detach { + if let Some(guard) = engine_guard.as_mut() { + if let Some(child) = guard.inner() { + let _ = child.kill(); + } + } else { + kill_engine(run_dir); + } + for _ in 0..20 { + if conclusion_path.exists() + || read_status_record(&status_path) + .is_some_and(|record| record.status.is_terminal()) + { + break; + } + sleep(Duration::from_millis(100)).await; + } + } else { + if let Some(guard) = engine_guard.as_mut() { + guard.defuse(); + } + eprintln!("Detached from run (engine continues in background)"); + } + break; + } + + 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); + } + } + + if runtime_interview_paths.request_path.exists() { + let interview_paths = &runtime_interview_paths; + if !interview_paths.response_path.exists() { + if let Some(_claim_guard) = + InterviewClaimGuard::acquire(&interview_paths.claim_path) + { + if let Ok(request_data) = std::fs::read_to_string(&interview_paths.request_path) + { + if let Ok(question) = + serde_json::from_str::(&request_data) + { + progress_ui.hide_bars(); + + let interviewer = ConsoleInterviewer::new(styles); + let answer = + fabro_interview::Interviewer::ask(&interviewer, question).await; + + progress_ui.show_bars(); + + if answer_requires_reattach(&answer) { + if let Some(guard) = engine_guard.as_mut() { + guard.defuse(); + } + eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}"); + return Ok(ExitCode::from(1)); + } + + write_interview_response_atomically( + &interview_paths.response_path, + &answer, + )?; + } + } + } + } + } + + let terminal_status = read_status_record(&status_path) + .map(|record| record.status) + .filter(|status| status.is_terminal()); + + let child_alive_via_handle = engine_guard.as_mut().and_then(|guard| { + guard.inner().map(|child| match child.try_wait() { + Ok(None) => true, + Ok(Some(_)) | Err(_) => false, + }) + }); + if let Some(child_alive) = child_alive_via_handle { if !child_alive { drain_remaining(&mut reader, &mut line, &mut progress_ui); @@ -249,7 +461,6 @@ pub(crate) async fn attach_run( sleep(Duration::from_millis(100)).await; } - // Finish progress bars progress_ui.finish(); Ok(determine_exit_code( @@ -277,6 +488,10 @@ fn drain_remaining( } } +fn event_payload_line(event: &EventEnvelope) -> Result { + serde_json::to_string(event.payload.as_value()).map_err(Into::into) +} + fn read_status_record(path: &Path) -> Option { RunStatusRecord::load(path).ok() } @@ -426,6 +641,31 @@ fn determine_exit_code(conclusion_path: &Path, status_record: Option ExitCode { + match run_store.get_conclusion().await { + Ok(Some(conclusion)) => { + let success = matches!( + conclusion.status, + StageStatus::Success | StageStatus::PartialSuccess + ); + if success { + ExitCode::from(0) + } else { + ExitCode::from(1) + } + } + Ok(None) | Err(_) => { + 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) + } + } +} + #[allow(unsafe_code)] fn kill_engine(run_dir: &Path) { if let Some(pid) = read_launcher_pid(run_dir).map(|pid| i32::try_from(pid).unwrap()) { @@ -497,7 +737,7 @@ mod tests { .unwrap(); let started = Instant::now(); - let exit = attach_run(dir.path(), false, no_color_styles(), Some(child)) + let exit = attach_run(dir.path(), None, false, no_color_styles(), Some(child)) .await .unwrap(); @@ -518,7 +758,7 @@ mod tests { Some(StatusReason::LaunchFailed), ); - let exit = attach_run(dir.path(), false, no_color_styles(), None) + let exit = attach_run(dir.path(), None, false, no_color_styles(), None) .await .unwrap(); diff --git a/lib/crates/fabro-cli/src/commands/run/command.rs b/lib/crates/fabro-cli/src/commands/run/command.rs index d6d03a522..8cbb1dff4 100644 --- a/lib/crates/fabro-cli/src/commands/run/command.rs +++ b/lib/crates/fabro-cli/src/commands/run/command.rs @@ -25,7 +25,8 @@ pub(crate) async fn execute(mut args: RunArgs, _globals: &GlobalArgs) -> Result< if args.detach { println!("{run_id}"); } else { - let exit_code = super::attach::attach_run(&run_dir, true, styles, Some(child)).await?; + let exit_code = + super::attach::attach_run(&run_dir, Some(&run_id), true, styles, Some(child)).await?; super::output::print_run_summary(&run_dir, &run_id, styles); if exit_code != std::process::ExitCode::SUCCESS { std::process::exit(1); diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 4e52723d8..223b8674f 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -9,7 +9,7 @@ use fabro_store::RunStore; use fabro_util::terminal::Styles; use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use futures::StreamExt; -use tracing::{debug, info}; +use tracing::{debug, info, warn}; use crate::args::LogsArgs; use crate::cli_config::load_cli_settings; @@ -28,25 +28,34 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles) -> Result<()> { }; let run_store = crate::store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; - let (all_lines, last_seq) = match run_store.as_ref() { - Some(run_store) => { - let events = run_store - .list_events() - .await - .context("Failed to list store-backed run events")?; - let last_seq = events.last().map(|event| event.seq).unwrap_or(0); - let lines = events - .iter() - .map(event_payload_line) - .collect::>>()?; - (lines, last_seq) - } + let progress_path = run.path.join("progress.jsonl"); + let (all_lines, last_seq, use_store_follow) = match run_store.as_ref() { + Some(run_store) => match run_store.list_events().await { + Ok(events) => { + let last_seq = events.last().map(|event| event.seq).unwrap_or(0); + let lines = events + .iter() + .map(event_payload_line) + .collect::>>()?; + (lines, last_seq, true) + } + Err(err) => { + if !progress_path.exists() { + return Err(err).context("Failed to list store-backed run events"); + } + warn!( + run_id = %run.run_id, + error = %err, + "Failed to read events from store; falling back to progress.jsonl" + ); + (read_lines(&progress_path)?, 0, false) + } + }, None => { - let progress_path = run.path.join("progress.jsonl"); if !progress_path.exists() { bail!("No progress.jsonl found for run '{}'", run.run_id); } - (read_lines(&progress_path)?, 0) + (read_lines(&progress_path)?, 0, false) } }; let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail); @@ -66,18 +75,44 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles) -> Result<()> { } if args.follow { - if let Some(run_store) = run_store.as_ref() { - follow_store_logs( - run_store.as_ref(), - if last_seq == 0 { 1 } else { last_seq + 1 }, - args.pretty, - styles, - is_tty, - ) - .await?; + if use_store_follow { + if let Some(run_store) = run_store.as_ref() { + match follow_store_logs( + run_store.as_ref(), + if last_seq == 0 { 1 } else { last_seq + 1 }, + args.pretty, + styles, + is_tty, + ) + .await + { + Ok(()) => {} + Err(err) => { + if !progress_path.exists() { + return Err(err); + } + warn!( + run_id = %run.run_id, + error = %err, + "Failed to follow store events; falling back to progress.jsonl" + ); + let lines_seen = read_lines(&progress_path)?.len(); + follow_logs( + &progress_path, + &run.path, + lines_seen, + args.pretty, + styles, + is_tty, + )?; + } + } + } else { + unreachable!("store follow requested without a run store"); + } } else { follow_logs( - &run.path.join("progress.jsonl"), + &progress_path, &run.path, all_lines.len(), args.pretty, diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 14adf7abc..956d71d60 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -2,7 +2,7 @@ use anyhow::Result; use fabro_config::FabroSettingsExt; use fabro_config::cli::load_cli_config; use fabro_util::terminal::Styles; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use crate::args::{GlobalArgs, RunCommands}; use crate::cli_config::load_cli_settings; @@ -43,7 +43,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( RunCommands::Start { run } => { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_info = resolve_run(&base, &run)?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?; let child = start::start_run(&run_info.path, false)?; eprintln!("Started engine process (PID {})", child.id()); Ok(()) @@ -52,8 +53,11 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_info = resolve_run(&base, &run)?; - let exit_code = attach::attach_run(&run_info.path, false, styles, None).await?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?; + let exit_code = + attach::attach_run(&run_info.path, Some(&run_info.run_id), false, styles, None) + .await?; if exit_code != std::process::ExitCode::SUCCESS { std::process::exit(1); } diff --git a/lib/crates/fabro-cli/src/commands/run/resume.rs b/lib/crates/fabro-cli/src/commands/run/resume.rs index ecc7fa469..6093db95c 100644 --- a/lib/crates/fabro-cli/src/commands/run/resume.rs +++ b/lib/crates/fabro-cli/src/commands/run/resume.rs @@ -2,7 +2,7 @@ use anyhow::bail; use fabro_config::FabroSettingsExt; use fabro_util::terminal::Styles; use fabro_workflows::records::{RunRecord, RunRecordExt}; -use fabro_workflows::run_lookup::{find_run_by_prefix, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use crate::args::ResumeArgs; use crate::cli_config::load_cli_settings; @@ -18,7 +18,9 @@ pub(crate) async fn resume_command( ) -> anyhow::Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - let run_dir = find_run_by_prefix(&base, &args.run)?; + let store = crate::store::build_store(&cli_settings.storage_dir())?; + let run = resolve_run_combined(store.as_ref(), &base, &args.run).await?; + let run_dir = run.path; // find_run_by_prefix can match orphan directories (no run.json). if !run_dir.join("run.json").exists() { @@ -35,7 +37,8 @@ pub(crate) async fn resume_command( if args.detach { println!("{run_id}"); } else { - let exit_code = super::attach::attach_run(&run_dir, true, styles, Some(child)).await?; + let exit_code = + super::attach::attach_run(&run_dir, Some(&run_id), true, styles, Some(child)).await?; super::output::print_run_summary(&run_dir, &run_id, styles); if exit_code != std::process::ExitCode::SUCCESS { std::process::exit(1); diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 0bc68d059..701ee1c06 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -3,10 +3,11 @@ use std::path::Path; use anyhow::{Context, Result, bail}; use fabro_config::FabroSettingsExt; use fabro_sandbox::SandboxRecordExt; +use fabro_store::Store; use tracing::warn; use fabro_sandbox::reconnect::reconnect as reconnect_sandbox; -use fabro_workflows::run_lookup::{resolve_run, runs_base}; +use fabro_workflows::run_lookup::{resolve_run_combined, runs_base}; use fabro_workflows::run_status::{RunStatus, write_run_status}; use crate::args::RunsRemoveArgs; @@ -17,14 +18,15 @@ use super::short_run_id; pub(crate) async fn remove_command(args: &RunsRemoveArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - remove_from(args, &base).await + let store = crate::store::build_store(&cli_settings.storage_dir())?; + remove_from(args, store.as_ref(), &base).await } -async fn remove_from(args: &RunsRemoveArgs, base: &Path) -> Result<()> { +async fn remove_from(args: &RunsRemoveArgs, store: &dyn Store, base: &Path) -> Result<()> { let mut had_errors = false; for identifier in &args.runs { - let run = match resolve_run(base, identifier) { + let run = match resolve_run_combined(store, base, identifier).await { Ok(run) => run, Err(err) => { eprintln!("error: {identifier}: {err}"); @@ -44,6 +46,21 @@ async fn remove_from(args: &RunsRemoveArgs, base: &Path) -> Result<()> { } write_run_status(&run.path, RunStatus::Removing, None); + if let Ok(Some(run_store)) = store.open_run_reader(&run.run_id).await { + if let Err(err) = run_store + .put_status(&fabro_workflows::run_status::RunStatusRecord::new( + RunStatus::Removing, + None, + )) + .await + { + warn!( + run_id = %run.run_id, + error = %err, + "failed to save removing status to store" + ); + } + } let sandbox_path = run.path.join("sandbox.json"); if let Ok(record) = fabro_sandbox::SandboxRecord::load(&sandbox_path) { @@ -63,6 +80,10 @@ async fn remove_from(args: &RunsRemoveArgs, base: &Path) -> Result<()> { std::fs::remove_dir_all(&run.path) .with_context(|| format!("failed to delete {}", run.path.display()))?; + store + .delete_run(&run.run_id) + .await + .with_context(|| format!("failed to delete store state for {}", run.run_id))?; eprintln!("{}", short_run_id(&run.run_id)); } diff --git a/lib/crates/fabro-cli/src/commands/system/df.rs b/lib/crates/fabro-cli/src/commands/system/df.rs index 0d0e85b14..6d41ee587 100644 --- a/lib/crates/fabro-cli/src/commands/system/df.rs +++ b/lib/crates/fabro-cli/src/commands/system/df.rs @@ -6,22 +6,36 @@ use cli_table::format::{Border, Justify, Separator}; use cli_table::{Cell, CellStruct, Style, Table, print_stdout}; use fabro_config::FabroSettingsExt; -use fabro_workflows::run_lookup::{logs_base, runs_base, scan_runs}; +use fabro_workflows::run_lookup::{logs_base, runs_base, scan_runs_combined}; use fabro_workflows::run_status::RunStatus; use crate::args::DfArgs; use crate::cli_config::load_cli_settings; use crate::shared::format_size; -pub(super) fn df_command(args: &DfArgs) -> Result<()> { +pub(super) async fn df_command(args: &DfArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let data_dir = cli_settings.storage_dir(); let runs_base_dir = runs_base(&data_dir); let logs_base_dir = logs_base(&data_dir); - df_from(args, &data_dir, &runs_base_dir, &logs_base_dir) + let store = crate::store::build_store(&data_dir)?; + df_from( + args, + store.as_ref(), + &data_dir, + &runs_base_dir, + &logs_base_dir, + ) + .await } -fn df_from(args: &DfArgs, data_dir: &Path, runs_base: &Path, logs_base: &Path) -> Result<()> { +async fn df_from( + args: &DfArgs, + store: &dyn fabro_store::Store, + data_dir: &Path, + runs_base: &Path, + logs_base: &Path, +) -> Result<()> { struct RunSizeInfo { run_id: String, workflow_name: String, @@ -30,7 +44,7 @@ fn df_from(args: &DfArgs, data_dir: &Path, runs_base: &Path, logs_base: &Path) - size: u64, } - let runs = scan_runs(runs_base)?; + let runs = scan_runs_combined(store, runs_base).await?; let mut active_count = 0u64; let mut total_run_size = 0u64; let mut reclaimable_run_size = 0u64; diff --git a/lib/crates/fabro-cli/src/commands/system/mod.rs b/lib/crates/fabro-cli/src/commands/system/mod.rs index b83323dbc..d41d33057 100644 --- a/lib/crates/fabro-cli/src/commands/system/mod.rs +++ b/lib/crates/fabro-cli/src/commands/system/mod.rs @@ -7,9 +7,9 @@ use crate::args::{SystemCommand, SystemNamespace}; pub(crate) use prune::parse_duration; -pub(crate) fn dispatch(ns: SystemNamespace) -> Result<()> { +pub(crate) async fn dispatch(ns: SystemNamespace) -> Result<()> { match ns.command { - SystemCommand::Prune(args) => prune::prune_command(&args), - SystemCommand::Df(args) => df::df_command(&args), + SystemCommand::Prune(args) => prune::prune_command(&args).await, + SystemCommand::Df(args) => df::df_command(&args).await, } } diff --git a/lib/crates/fabro-cli/src/commands/system/prune.rs b/lib/crates/fabro-cli/src/commands/system/prune.rs index f11044de5..bec2d6491 100644 --- a/lib/crates/fabro-cli/src/commands/system/prune.rs +++ b/lib/crates/fabro-cli/src/commands/system/prune.rs @@ -3,18 +3,20 @@ use std::path::Path; use anyhow::{Context, Result, bail}; use chrono::Utc; use fabro_config::FabroSettingsExt; +use fabro_store::Store; use tracing::{debug, info}; -use fabro_workflows::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs}; +use fabro_workflows::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs_combined}; use crate::args::RunsPruneArgs; use crate::cli_config::load_cli_settings; use crate::shared::format_size; -pub(super) fn prune_command(args: &RunsPruneArgs) -> Result<()> { +pub(super) async fn prune_command(args: &RunsPruneArgs) -> Result<()> { let cli_settings = load_cli_settings(None)?; let base = runs_base(&cli_settings.storage_dir()); - prune_from(args, &base) + let store = crate::store::build_store(&cli_settings.storage_dir())?; + prune_from(args, store.as_ref(), &base).await } pub(crate) fn parse_duration(s: &str) -> Result { @@ -33,8 +35,8 @@ pub(crate) fn parse_duration(s: &str) -> Result { } } -fn prune_from(args: &RunsPruneArgs, base: &Path) -> Result<()> { - let runs = scan_runs(base)?; +async fn prune_from(args: &RunsPruneArgs, store: &dyn Store, base: &Path) -> Result<()> { + let runs = scan_runs_combined(store, base).await?; let label_filters = parse_label_filters(&args.filter.label); let mut filtered = filter_runs( &runs, @@ -79,6 +81,10 @@ fn prune_from(args: &RunsPruneArgs, base: &Path) -> Result<()> { for run in &filtered { info!(run_id = %run.run_id, path = %run.path.display(), "deleting run"); std::fs::remove_dir_all(&run.path)?; + store + .delete_run(&run.run_id) + .await + .with_context(|| format!("failed to delete store state for {}", run.run_id))?; } eprintln!( "{} run(s) deleted ({} freed).", diff --git a/lib/crates/fabro-cli/src/main.rs b/lib/crates/fabro-cli/src/main.rs index 385740a98..a4da3e962 100644 --- a/lib/crates/fabro-cli/src/main.rs +++ b/lib/crates/fabro-cli/src/main.rs @@ -215,7 +215,7 @@ async fn main_inner() -> (String, Result<()>) { commands::upgrade::run_upgrade(args).await?; } Commands::Provider(ns) => commands::provider::dispatch(ns).await?, - Commands::System(ns) => commands::system::dispatch(ns)?, + Commands::System(ns) => commands::system::dispatch(ns).await?, Commands::SendAnalytics { path } => { let result = sender::upload(&path).await; let _ = std::fs::remove_file(&path); diff --git a/lib/crates/fabro-workflows/src/run_lookup.rs b/lib/crates/fabro-workflows/src/run_lookup.rs index bb314eb01..50018e2a0 100644 --- a/lib/crates/fabro-workflows/src/run_lookup.rs +++ b/lib/crates/fabro-workflows/src/run_lookup.rs @@ -162,6 +162,9 @@ pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result