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

Move attach and logs to store-backed event history

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-01 20:59:52 -04:00
parent 991c5eb956
commit 9bf4046a6b
No known key found for this signature in database
7 changed files with 180 additions and 147 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -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<ExitCode> {
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<String>,
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<std::process::Child>,
@ -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));
}

1
lib/crates/fabro-cli/src/commands/run/command.rs
View file

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

181
lib/crates/fabro-cli/src/commands/run/logs.rs
View file

@ -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::<Result<Vec<_>>>()?;
(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::<Result<Vec<_>>>()?;
(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<Vec<String>> {
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<Utc>>,
@ -193,47 +139,9 @@ fn try_parse_relative_duration(s: &str) -> Option<chrono::Duration> {
}
}
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<bool> {
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,

1
lib/crates/fabro-cli/src/commands/run/mod.rs
View file

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

1
lib/crates/fabro-cli/src/commands/run/resume.rs
View file

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

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

@ -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!();

37
lib/crates/fabro-cli/tests/it/cmd/logs.rs
View file

@ -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!();