From 9bf4046a6b26ff05fa8958be657d7ed4f5cc233e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Wed, 1 Apr 2026 20:59:52 -0400 Subject: [PATCH] Move attach and logs to store-backed event history --- .../fabro-cli/src/commands/run/attach.rs | 59 ++++-- .../fabro-cli/src/commands/run/command.rs | 1 + lib/crates/fabro-cli/src/commands/run/logs.rs | 181 +++++------------- lib/crates/fabro-cli/src/commands/run/mod.rs | 1 + .../fabro-cli/src/commands/run/resume.rs | 1 + lib/crates/fabro-cli/tests/it/cmd/attach.rs | 47 +++++ lib/crates/fabro-cli/tests/it/cmd/logs.rs | 37 ++++ 7 files changed, 180 insertions(+), 147 deletions(-) diff --git a/lib/crates/fabro-cli/src/commands/run/attach.rs b/lib/crates/fabro-cli/src/commands/run/attach.rs index 3c5ba883b..532ed545e 100644 --- a/lib/crates/fabro-cli/src/commands/run/attach.rs +++ b/lib/crates/fabro-cli/src/commands/run/attach.rs @@ -35,6 +35,7 @@ const JSON_INTERVIEW_MESSAGE: &str = "This run is waiting for human input, but - /// Returns exit code 0 for success/partial_success, 1 otherwise. pub(crate) async fn attach_run( run_dir: &Path, + storage_dir: Option<&Path>, run_id: Option<&RunId>, kill_on_detach: bool, styles: &'static Styles, @@ -42,15 +43,23 @@ pub(crate) async fn attach_run( json_output: bool, ) -> 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)), - ) { + let fallback_storage_dir = run_record + .as_ref() + .map(|record| record.settings.storage_dir()); + let storage_dir = storage_dir.or(fallback_storage_dir.as_deref()); + let run_id = run_id.or_else(|| run_record.as_ref().map(|record| &record.run_id)); + + if let (Some(storage_dir), Some(run_id)) = (storage_dir, run_id) { match store::open_run_reader(&storage_dir, run_id).await { Ok(Some(run_store)) => match run_store.list_events().await { Ok(events) => { + let verbose = run_store + .get_run() + .await + .ok() + .flatten() + .map(|record| record.settings.verbose_enabled()) + .unwrap_or(false); let event_lines = events .iter() .map(event_payload_line) @@ -58,6 +67,7 @@ pub(crate) async fn attach_run( return attach_run_store( run_dir, run_store.as_ref(), + verbose, event_lines, events.last().map_or(0, |event| event.seq), kill_on_detach, @@ -86,12 +96,25 @@ pub(crate) async fn attach_run( } } - attach_run_files(run_dir, kill_on_detach, styles, engine_child, json_output).await + let verbose = run_record + .as_ref() + .map(|record| record.settings.verbose_enabled()) + .unwrap_or(false); + attach_run_files( + run_dir, + verbose, + kill_on_detach, + styles, + engine_child, + json_output, + ) + .await } async fn attach_run_store( run_dir: &Path, run_store: &dyn RunStore, + verbose: bool, existing_events: Vec, last_seq: u32, kill_on_detach: bool, @@ -105,9 +128,6 @@ async fn attach_run_store( 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); // Install Ctrl+C handler @@ -269,6 +289,7 @@ async fn attach_run_store( async fn attach_run_files( run_dir: &Path, + verbose: bool, kill_on_detach: bool, styles: &'static Styles, engine_child: Option, @@ -283,9 +304,6 @@ async fn attach_run_files( 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)); @@ -775,6 +793,7 @@ mod tests { let exit = attach_run( dir.path(), None, + None, false, no_color_styles(), Some(child), @@ -800,9 +819,17 @@ mod tests { Some(StatusReason::LaunchFailed), ); - let exit = attach_run(dir.path(), None, false, no_color_styles(), None, false) - .await - .unwrap(); + let exit = attach_run( + dir.path(), + None, + None, + false, + no_color_styles(), + None, + false, + ) + .await + .unwrap(); assert_eq!(exit, ExitCode::from(1)); } diff --git a/lib/crates/fabro-cli/src/commands/run/command.rs b/lib/crates/fabro-cli/src/commands/run/command.rs index 58ed3b454..22b29b8ba 100644 --- a/lib/crates/fabro-cli/src/commands/run/command.rs +++ b/lib/crates/fabro-cli/src/commands/run/command.rs @@ -34,6 +34,7 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<( } else { let exit_code = super::attach::attach_run( &run_dir, + Some(cli_settings.storage_dir().as_path()), Some(&run_id), true, styles, diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 7c2fa9a2f..a86411e49 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -1,5 +1,5 @@ use std::fmt::Write as _; -use std::io::{self, BufRead, IsTerminal, Write}; +use std::io::{self, IsTerminal, Write}; use std::path::Path; use std::time::Duration; @@ -12,7 +12,7 @@ use fabro_util::terminal::Styles; use fabro_workflow::run_lookup::{resolve_run_combined, runs_base}; use futures::StreamExt; use tokio::time; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; use crate::args::{GlobalArgs, LogsArgs}; use crate::store; @@ -31,26 +31,21 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) None => None, }; - let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id).await?; - let progress_path = run.path.join("progress.jsonl"); - let (all_lines, last_seq, use_store_follow) = if progress_path.exists() { - (read_lines(&progress_path)?, 0, false) - } else if let Some(run_store) = run_store.as_ref() { - match run_store.list_events().await { - Ok(events) => { - let last_seq = events.last().map_or(0, |event| event.seq); - let lines = events - .iter() - .map(event_payload_line) - .collect::>>()?; - (lines, last_seq, true) - } - Err(err) => { - return Err(err).context("Failed to list store-backed run events"); - } + let run_store = store::open_run_reader(&cli_settings.storage_dir(), &run.run_id) + .await? + .with_context(|| format!("Run '{}' not found in store", run.run_id))?; + let (all_lines, last_seq) = match run_store.list_events().await { + Ok(events) => { + let last_seq = events.last().map_or(0, |event| event.seq); + let lines = events + .iter() + .map(event_payload_line) + .collect::>>()?; + (lines, last_seq) + } + Err(err) => { + return Err(err).context("Failed to list store-backed run events"); } - } else { - bail!("No progress.jsonl found for run '{}'", run.run_id); }; let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail); @@ -70,69 +65,20 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) } if args.follow { - 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 }, - 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, - pretty, - styles, - is_tty, - )?; - } - } - } else { - unreachable!("store follow requested without a run store"); - } - } else { - follow_logs( - &progress_path, - &run.path, - all_lines.len(), - pretty, - styles, - is_tty, - )?; - } + follow_store_logs( + run_store.as_ref(), + &run.path, + if last_seq == 0 { 1 } else { last_seq + 1 }, + pretty, + styles, + is_tty, + ) + .await?; } Ok(()) } -fn read_lines(path: &Path) -> Result> { - let file = std::fs::File::open(path).context("Failed to open progress.jsonl")?; - let reader = io::BufReader::new(file); - let mut lines = Vec::new(); - for line in reader.lines() { - let line = line?; - if !line.trim().is_empty() { - lines.push(line); - } - } - Ok(lines) -} - fn apply_filters( lines: &[String], since: Option<&DateTime>, @@ -193,47 +139,9 @@ fn try_parse_relative_duration(s: &str) -> Option { } } -fn follow_logs( - progress_path: &Path, - run_dir: &Path, - mut lines_seen: usize, - pretty: bool, - styles: &Styles, - _is_tty: bool, -) -> Result<()> { - let conclusion_path = run_dir.join("conclusion.json"); - let stdout = io::stdout(); - let mut out = stdout.lock(); - - loop { - std::thread::sleep(std::time::Duration::from_millis(200)); - - let all_lines = read_lines(progress_path)?; - if all_lines.len() > lines_seen { - for line in &all_lines[lines_seen..] { - if pretty { - if let Some(formatted) = format_event_pretty(line, styles) { - writeln!(out, "{formatted}")?; - } - } else { - writeln!(out, "{line}")?; - } - } - out.flush()?; - lines_seen = all_lines.len(); - } - - if conclusion_path.exists() && all_lines.len() <= lines_seen { - debug!("Run concluded, stopping follow"); - break; - } - } - - Ok(()) -} - async fn follow_store_logs( run_store: &dyn RunStore, + run_dir: &Path, seq: u32, pretty: bool, styles: &Styles, @@ -262,20 +170,13 @@ async fn follow_store_logs( next_seq = event.seq.saturating_add(1); } Ok(Some(Err(err))) => return Err(err.into()), - Ok(None) => break, + Ok(None) => { + if run_concluded(run_store, run_dir).await? { + break; + } + } Err(_) => { - let concluded = run_store - .get_conclusion() - .await - .context("Failed to read conclusion from store while following logs")? - .is_some() - || run_store - .get_status() - .await - .context("Failed to read status from store while following logs")? - .is_some_and(|record| record.status.is_terminal()); - - if concluded { + if run_concluded(run_store, run_dir).await? { flush_remaining_store_events(run_store, next_seq, pretty, styles, &mut out) .await?; debug!("Run reached terminal status, stopping follow"); @@ -288,6 +189,24 @@ async fn follow_store_logs( Ok(()) } +async fn run_concluded(run_store: &dyn RunStore, run_dir: &Path) -> Result { + if run_store + .get_conclusion() + .await + .context("Failed to read conclusion from store while following logs")? + .is_some() + { + return Ok(true); + } + + Ok(run_store + .get_status() + .await + .context("Failed to read status from store while following logs")? + .is_some_and(|record| record.status.is_terminal()) + || run_dir.join("conclusion.json").exists()) +} + async fn flush_remaining_store_events( run_store: &dyn RunStore, next_seq: u32, diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index 00d689672..7a23f2c33 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -82,6 +82,7 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<( let run_info = resolve_run_combined(store.as_ref(), &base, &run).await?; let exit_code = attach::attach_run( &run_info.path, + Some(cli_settings.storage_dir().as_path()), Some(&run_info.run_id), false, styles, diff --git a/lib/crates/fabro-cli/src/commands/run/resume.rs b/lib/crates/fabro-cli/src/commands/run/resume.rs index 1a04ed544..1c42f67f0 100644 --- a/lib/crates/fabro-cli/src/commands/run/resume.rs +++ b/lib/crates/fabro-cli/src/commands/run/resume.rs @@ -46,6 +46,7 @@ pub(crate) async fn resume_command( } else { let exit_code = super::attach::attach_run( &run_dir, + Some(cli_settings.storage_dir().as_path()), Some(&run_id), true, styles, diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index ae312ea94..750652c86 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -97,6 +97,53 @@ fn attach_replays_completed_detached_run() { "); } +#[test] +fn attach_replays_from_store_without_run_json_or_progress_jsonl() { + let context = test_context!(); + let run_id = "01ARZ3NDEKTSV4RRFFQ69G5FAQ"; + + context + .command() + .args([ + "run", + "--dry-run", + "--auto-approve", + "--no-retro", + "--detach", + "--run-id", + run_id, + example_fixture("simple.fabro").to_str().unwrap(), + ]) + .assert() + .success(); + + context + .command() + .args(["wait", run_id]) + .timeout(std::time::Duration::from_secs(10)) + .assert() + .success(); + + let run = resolve_run(&context, run_id); + std::fs::remove_file(run.run_dir.join("run.json")).unwrap(); + std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap(); + + let mut cmd = context.command(); + cmd.args(["attach", run_id]); + cmd.timeout(std::time::Duration::from_secs(10)); + fabro_snapshot!(run_output_filters(&context), cmd, @" + success: true + exit_code: 0 + ----- stdout ----- + ----- stderr ----- + Sandbox: local (ready in [TIME]) + ✓ Start [TIME] + ✓ Run Tests [TIME] + ✓ Report [TIME] + ✓ Exit [TIME] + "); +} + #[test] fn attach_before_completion_streams_to_finished_state() { let context = test_context!(); diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index d61ff248a..315c0cef0 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -88,6 +88,43 @@ fn logs_completed_run_outputs_raw_ndjson() { "#); } +#[test] +fn logs_completed_run_reads_store_without_progress_jsonl() { + let context = test_context!(); + let run = setup_completed_dry_run(&context); + std::fs::remove_file(run.run_dir.join("progress.jsonl")).unwrap(); + + let mut filters = context.filters(); + filters.push(( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), + "[TIMESTAMP]".to_string(), + )); + filters.push(( + r#""duration_ms":\s*\d+"#.to_string(), + r#""duration_ms": [DURATION_MS]"#.to_string(), + )); + filters.push(( + r#""id":"[0-9a-f-]+""#.to_string(), + r#""id":"[EVENT_ID]""#.to_string(), + )); + filters.push(( + r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/runs/REDACTED)""#.to_string(), + r#""run_dir":"[RUN_DIR]""#.to_string(), + )); + + let mut cmd = context.command(); + cmd.args(["logs", "--tail", "2", &run.run_id]); + + fabro_snapshot!(filters, cmd, @r#" + success: true + exit_code: 0 + ----- stdout ----- + {"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + ----- stderr ----- + "#); +} + #[test] fn logs_tail_limits_output() { let context = test_context!();