scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-06 08:18:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Fix remaining CLI store migration gaps

Browse source
This commit is contained in:
Bryan Helmkamp 2026-03-28 23:23:38 -04:00
parent f6f5c25498
commit 9c66dcb6f0
16 changed files with 479 additions and 141 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
lib/crates/fabro-cli/src/commands/pr/close.rs
View file

@ -22,7 +22,7 @@ async fn close_from(
args: PrCloseArgs,
github_app: Option<fabro_github::GitHubAppCredentials>,
) -> 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",

10
lib/crates/fabro-cli/src/commands/pr/list.rs
View file

@ -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<fabro_github::GitHubAppCredentials>,
@ -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 {

2
lib/crates/fabro-cli/src/commands/pr/merge.rs
View file

@ -23,7 +23,7 @@ async fn merge_from(
args: PrMergeArgs,
github_app: Option<fabro_github::GitHubAppCredentials>,
) -> 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",

13
lib/crates/fabro-cli/src/commands/pr/mod.rs
View file

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

2
lib/crates/fabro-cli/src/commands/pr/view.rs
View file

@ -23,7 +23,7 @@ async fn view_from(
args: PrViewArgs,
github_app: Option<fabro_github::GitHubAppCredentials>,
) -> 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",

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

@ -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<std::process::Child>,
) -> 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.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::<Result<Vec<_>>>()?;
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<String>,
last_seq: u32,
kill_on_detach: bool,
styles: &'static Styles,
engine_child: Option<std::process::Child>,
) -> Result<ExitCode> {
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<u32> = 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<std::process::Child>,
) -> Result<ExitCode> {
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<u32> = 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::<fabro_interview::Question>(&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<String> {
serde_json::to_string(event.payload.as_value()).map_err(Into::into)
}
fn read_status_record(path: &Path) -> Option<RunStatusRecord> {
RunStatusRecord::load(path).ok()
}
@ -426,6 +641,31 @@ 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 {
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();

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

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

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

@ -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::<Result<Vec<_>>>()?;
(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::<Result<Vec<_>>>()?;
(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,

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

@ -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);
}

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

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

29
lib/crates/fabro-cli/src/commands/runs/rm.rs
View file

@ -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));
}

24
lib/crates/fabro-cli/src/commands/system/df.rs
View file

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

6
lib/crates/fabro-cli/src/commands/system/mod.rs
View file

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

16
lib/crates/fabro-cli/src/commands/system/prune.rs
View file

@ -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<chrono::Duration> {
@ -33,8 +35,8 @@ pub(crate) fn parse_duration(s: &str) -> Result<chrono::Duration> {
}
}
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).",

2
lib/crates/fabro-cli/src/main.rs
View file

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

3
lib/crates/fabro-workflows/src/run_lookup.rs
View file

@ -162,6 +162,9 @@ pub async fn scan_runs_combined(store: &dyn Store, base: &Path) -> Result<Vec<Ru
continue;
};
let path = PathBuf::from(run_dir);
if !path.exists() {
continue;
}
let Some(dir_name) = path
.file_name()
.map(|name| name.to_string_lossy().to_string())