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

refactor(cli): simplify run lifecycle around server ownership

Browse source 
Rename the hidden detached worker command to __runner, remove launcher
bookkeeping, and resolve runs through a shared server-backed lookup.
This leaves attach, resume, logs, and related commands using the server
as the source of truth and updates the integration coverage around the
new lifecycle.
This commit is contained in:
Bryan Helmkamp 2026-04-05 12:11:55 -04:00
parent e012a351ce
commit 65bf139a09
No known key found for this signature in database
36 changed files with 335 additions and 827 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
lib/crates/fabro-cli/src/args.rs
View file

@ -635,7 +635,7 @@ pub(crate) enum RunCommands {
Run(RunArgs),
/// Create a workflow run (allocate run dir, persist spec)
Create(RunArgs),
/// Start a created workflow run (spawn engine process)
/// Start a created workflow run on the server
Start {
/// Run ID prefix or workflow name
run: String,
@ -645,18 +645,12 @@ pub(crate) enum RunCommands {
/// Run ID prefix or workflow name
run: String,
},
/// Internal: run the engine process
#[command(name = "__detached", hide = true)]
Detached {
/// Internal: queue or resume a workflow run via the server
#[command(name = "__runner", hide = true)]
Runner {
/// Run ID
#[arg(long)]
run_id: fabro_types::RunId,
/// Run directory
#[arg(long)]
run_dir: PathBuf,
/// Launcher metadata path
#[arg(long)]
launcher_path: PathBuf,
/// Resume from checkpoint instead of fresh start
#[arg(long)]
resume: bool,
@ -683,7 +677,7 @@ impl RunCommands {
Self::Create(_) => "create",
Self::Start { .. } => "start",
Self::Attach { .. } => "attach",
Self::Detached { .. } => "__detached",
Self::Runner { .. } => "__runner",
Self::Diff(_) => "diff",
Self::Logs(_) => "logs",
Self::Resume(_) => "resume",

9
lib/crates/fabro-cli/src/commands/artifact/cp.rs
View file

@ -3,20 +3,17 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_store::RuntimeState;
use fabro_workflow::artifacts::{ArtifactEntry, scan_artifacts};
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use crate::args::{ArtifactCpArgs, GlobalArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{print_json_pretty, split_run_path};
use crate::user_config::load_user_settings_with_globals;
pub(super) async fn cp_command(args: &ArtifactCpArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let (run_id, asset_path) = parse_source(&args.source);
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, run_id)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(run_id)?;
let runtime_state = RuntimeState::new(&run.path);
let entries = scan_artifacts(
&runtime_state.artifacts_dir(),

9
lib/crates/fabro-cli/src/commands/artifact/list.rs
View file

@ -1,19 +1,16 @@
use anyhow::Result;
use fabro_store::RuntimeState;
use fabro_workflow::artifacts::scan_artifacts;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use crate::args::{ArtifactListArgs, GlobalArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::format_size;
use crate::user_config::load_user_settings_with_globals;
pub(super) async fn list_command(args: &ArtifactListArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run_id)?;
let runtime_state = RuntimeState::new(&run.path);
let entries = scan_artifacts(
&runtime_state.artifacts_dir(),

12
lib/crates/fabro-cli/src/commands/pr/create.rs
View file

@ -5,12 +5,12 @@ use fabro_model::Catalog;
use fabro_sandbox::daytona::detect_repo_info;
use fabro_workflow::outcome::StageStatus;
use fabro_workflow::pull_request::maybe_open_pull_request;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use fabro_workflow::run_lookup::runs_base;
use tracing::info;
use crate::args::{GlobalArgs, PrCreateArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
@ -30,12 +30,10 @@ async fn create_from(
github_app: Option<fabro_github::GitHubAppCredentials>,
globals: &GlobalArgs,
) -> Result<()> {
let storage_dir = base.parent().unwrap_or(base);
let client = server_client::connect_server(storage_dir).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, base, &args.run_id)?;
let lookup = ServerRunLookup::connect_from_runs_base(base).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let events = client.list_run_events(&run_id, None, None).await?;
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let state = run_store.state().await?;

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

@ -9,6 +9,7 @@ use tracing::info;
use crate::args::{GlobalArgs, PrListArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
@ -28,9 +29,16 @@ pub(super) async fn list_command(
) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
list_from(&client, &summaries, &base, args, github_app, globals).await
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
list_from(
lookup.client(),
lookup.summaries(),
&base,
args,
github_app,
globals,
)
.await
}
async fn list_from(

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

@ -9,10 +9,9 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use fabro_types::PullRequestRecord;
use fabro_workflow::run_lookup::resolve_run_from_summaries;
use crate::args::{GlobalArgs, PrCommand, PrNamespace};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::github::build_github_app_credentials;
use crate::user_config::load_user_settings_with_globals;
@ -35,13 +34,11 @@ pub(crate) async fn load_pr_record(
base: &Path,
run_id: &str,
) -> Result<(PullRequestRecord, PathBuf)> {
let storage_dir = base.parent().unwrap_or(base);
let client = server_client::connect_server(storage_dir).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, base, run_id)?;
let lookup = ServerRunLookup::connect_from_runs_base(base).await?;
let run = lookup.resolve(run_id)?;
let run_id = run.run_id();
let run_dir = run.path;
let state = client.get_run_state(&run_id).await?;
let state = lookup.client().get_run_state(&run_id).await?;
let record = state.pull_request.with_context(|| {
format!("No pull request found in store. Create one first with: fabro pr create {run_id}")
})?;

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

@ -21,10 +21,6 @@ use tokio::time::sleep;
use super::run_progress;
use crate::server_client;
#[cfg(test)]
const ATTACH_STARTUP_GRACE: Duration = Duration::from_millis(200);
#[cfg(not(test))]
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.";
@ -42,7 +38,6 @@ pub(crate) async fn attach_run(
run_id: Option<&RunId>,
kill_on_detach: bool,
styles: &'static Styles,
engine_child: Option<std::process::Child>,
json_output: bool,
) -> Result<ExitCode> {
let inferred_storage_dir = infer_storage_dir(run_dir);
@ -64,7 +59,6 @@ pub(crate) async fn attach_run(
.collect::<Result<Vec<_>>>()?;
let initial_exit_code = events.iter().rev().find_map(event_exit_code);
return attach_run_server(
run_dir,
&client,
run_id,
verbose,
@ -73,7 +67,6 @@ pub(crate) async fn attach_run(
initial_exit_code,
kill_on_detach,
styles,
engine_child,
json_output,
)
.await;
@ -85,7 +78,6 @@ pub(crate) async fn attach_run(
}
async fn attach_run_server(
run_dir: &Path,
client: &server_client::ServerStoreClient,
run_id: &RunId,
verbose: bool,
@ -94,11 +86,8 @@ async fn attach_run_server(
initial_exit_code: Option<ExitCode>,
kill_on_detach: bool,
styles: &'static Styles,
engine_child: Option<std::process::Child>,
json_output: bool,
) -> Result<ExitCode> {
let mut engine_guard = engine_child.map(EngineChildGuard::new);
let is_tty = std::io::stderr().is_terminal();
let mut progress_ui = run_progress::ProgressUI::new(is_tty, verbose);
@ -117,30 +106,18 @@ async fn attach_run_server(
}
if json_output && !client.list_run_questions(run_id).await?.is_empty() {
defuse_engine_child(&mut engine_guard);
eprintln!("{JSON_INTERVIEW_MESSAGE}");
return Ok(ExitCode::from(1));
}
let mut next_seq = if last_seq == 0 { 1 } else { last_seq + 1 };
let mut cached_pid: Option<u32> = None;
let attach_started = Instant::now();
let mut terminal_exit_code = initial_exit_code;
let mut terminal_event_seen_at = initial_exit_code.map(|_| Instant::now());
loop {
let server_owned = engine_guard.is_none() && read_launcher_pid(run_dir).is_none();
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 if server_owned {
let _ = client.cancel_run(run_id).await;
} else {
kill_engine(run_dir);
}
let _ = client.cancel_run(run_id).await;
// Wait briefly for a terminal status or conclusion
for _ in 0..20 {
if client
@ -159,9 +136,6 @@ async fn attach_run_server(
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;
@ -197,7 +171,6 @@ async fn attach_run_server(
// Check for server-backed interview request
if let Some(question) = client.list_run_questions(run_id).await?.into_iter().next() {
if json_output {
defuse_engine_child(&mut engine_guard);
eprintln!("{JSON_INTERVIEW_MESSAGE}");
return Ok(ExitCode::from(1));
}
@ -212,9 +185,6 @@ async fn attach_run_server(
show_progress(&mut progress_ui, json_output);
if answer_requires_reattach(&answer) {
if let Some(guard) = engine_guard.as_mut() {
guard.defuse();
}
eprintln!("{INTERVIEW_UNANSWERED_MESSAGE}");
return Ok(ExitCode::from(1));
}
@ -230,64 +200,10 @@ async fn attach_run_server(
.and_then(|state| state.status.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, // still running
Ok(Some(_)) | Err(_) => false, // exited or error
})
});
if let Some(child_alive) = child_alive_via_handle {
if !child_alive && !saw_event {
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
if terminal_status.is_some() && !saw_event {
flush_remaining_server_events(client, run_id, next_seq, &mut progress_ui, json_output)
.await?;
break;
}
} else {
if terminal_status.is_some() && !saw_event {
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
.await?;
break;
}
let engine_alive = if server_owned {
true
} else {
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
}
}
}
};
if !engine_alive {
flush_remaining_server_events(
client,
run_id,
next_seq,
&mut progress_ui,
json_output,
)
.await?;
break;
}
break;
}
if !saw_event {
@ -450,10 +366,6 @@ fn restore_empty_run_properties(value: &mut serde_json::Value) {
}
}
fn read_launcher_pid(run_dir: &Path) -> Option<u32> {
super::launcher::active_launcher_record_for_run(run_dir).map(|record| record.pid)
}
fn infer_storage_dir(run_dir: &Path) -> Option<PathBuf> {
let runs_dir = run_dir.parent()?;
let storage_dir = runs_dir.parent()?;
@ -461,124 +373,17 @@ fn infer_storage_dir(run_dir: &Path) -> Option<PathBuf> {
}
fn infer_run_id(run_dir: &Path) -> Option<RunId> {
super::launcher::launcher_record_for_run(run_dir)
.map(|record| record.run_id)
.or_else(|| {
std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|run_id| run_id.trim().to_string())
.filter(|run_id| !run_id.is_empty())
.and_then(|run_id| run_id.parse().ok())
})
}
#[cfg(test)]
struct InterviewClaimGuard {
claim_path: PathBuf,
}
#[cfg(test)]
impl InterviewClaimGuard {
fn acquire(claim_path: &Path) -> Option<Self> {
if try_claim_interview_request(claim_path) {
Some(Self {
claim_path: claim_path.to_path_buf(),
})
} else {
None
}
}
}
#[cfg(test)]
impl Drop for InterviewClaimGuard {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.claim_path);
}
}
struct EngineChildGuard {
child: Option<std::process::Child>,
}
impl EngineChildGuard {
fn new(child: std::process::Child) -> Self {
Self { child: Some(child) }
}
fn inner(&mut self) -> Option<&mut std::process::Child> {
self.child.as_mut()
}
fn defuse(&mut self) {
self.child.take();
}
}
fn defuse_engine_child(engine_guard: &mut Option<EngineChildGuard>) {
if let Some(guard) = engine_guard.as_mut() {
guard.defuse();
}
}
impl Drop for EngineChildGuard {
fn drop(&mut self) {
if let Some(mut child) = self.child.take() {
let _ = child.kill();
let _ = child.wait();
}
}
}
#[cfg(test)]
fn try_claim_interview_request(claim_path: &Path) -> bool {
if let Some(parent) = claim_path.parent() {
if std::fs::create_dir_all(parent).is_err() {
return false;
}
}
if let Ok(existing) = std::fs::read_to_string(claim_path) {
if let Ok(pid) = existing.trim().parse::<u32>() {
if process_alive(pid) {
return pid == std::process::id();
}
}
let _ = std::fs::remove_file(claim_path);
}
match std::fs::OpenOptions::new()
.create_new(true)
.write(true)
.open(claim_path)
{
Ok(mut file) => {
let _ = writeln!(file, "{}", std::process::id());
true
}
Err(_) => false,
}
std::fs::read_to_string(run_dir.join("id.txt"))
.ok()
.map(|run_id| run_id.trim().to_string())
.filter(|run_id| !run_id.is_empty())
.and_then(|run_id| run_id.parse().ok())
}
fn answer_requires_reattach(answer: &fabro_interview::Answer) -> bool {
matches!(answer.value, AnswerValue::Aborted | AnswerValue::Skipped)
}
#[cfg(test)]
fn write_interview_response_atomically(
response_path: &Path,
answer: &fabro_interview::Answer,
) -> Result<()> {
let response_json = serde_json::to_string_pretty(answer)?;
if let Some(parent) = response_path.parent() {
std::fs::create_dir_all(parent)?;
}
let temp_path = response_path.with_extension("json.tmp");
std::fs::write(&temp_path, response_json)?;
std::fs::rename(temp_path, response_path)?;
Ok(())
}
async fn determine_exit_code_with_server(
client: &server_client::ServerStoreClient,
run_id: &RunId,
@ -614,16 +419,6 @@ async fn determine_exit_code_with_server(
}
}
fn kill_engine(run_dir: &Path) {
if let Some(pid) = read_launcher_pid(run_dir) {
fabro_proc::sigterm(pid);
}
}
fn process_alive(pid: u32) -> bool {
fabro_proc::process_alive(pid)
}
fn event_exit_code(event: &EventEnvelope) -> Option<ExitCode> {
let run_event = RunEvent::try_from(&event.payload).ok()?;
match run_event.body {
@ -642,8 +437,6 @@ fn event_exit_code(event: &EventEnvelope) -> Option<ExitCode> {
#[cfg(test)]
mod tests {
use super::*;
use crate::commands::run::launcher;
use chrono::Utc;
use fabro_interview::{Answer, AnswerValue};
use fabro_util::terminal::Styles;
@ -655,17 +448,9 @@ mod tests {
async fn attach_errors_without_store_context() {
let dir = tempfile::tempdir().unwrap();
let err = attach_run(
dir.path(),
None,
None,
false,
no_color_styles(),
None,
false,
)
.await
.unwrap_err();
let err = attach_run(dir.path(), None, None, false, no_color_styles(), false)
.await
.unwrap_err();
assert!(
err.to_string()
@ -690,55 +475,20 @@ mod tests {
}
#[test]
fn infer_run_id_uses_launcher_record_without_run_json() {
fn infer_run_id_reads_id_txt() {
let dir = tempfile::tempdir().unwrap();
let storage_dir = dir.path().join("storage");
let run_dir = storage_dir.join("runs").join("20260401-test");
std::fs::create_dir_all(&run_dir).unwrap();
launcher::write_launcher_record(
&launcher::launcher_record_path(&storage_dir, &fabro_types::fixtures::RUN_1),
&launcher::LauncherRecord {
run_id: fabro_types::fixtures::RUN_1,
run_dir: run_dir.clone(),
pid: u32::MAX,
resume: false,
log_path: dir.path().join("launcher.log"),
started_at: Utc::now(),
},
std::fs::write(
run_dir.join("id.txt"),
format!("{}\n", fabro_types::fixtures::RUN_1),
)
.unwrap();
assert_eq!(infer_run_id(&run_dir), Some(fabro_types::fixtures::RUN_1));
}
#[test]
fn try_claim_interview_request_reclaims_stale_claim() {
let dir = tempfile::tempdir().unwrap();
let claim_path = dir.path().join("runtime").join("interview_request.claim");
std::fs::create_dir_all(claim_path.parent().unwrap()).unwrap();
std::fs::write(&claim_path, "999999\n").unwrap();
assert!(try_claim_interview_request(&claim_path));
assert_eq!(
std::fs::read_to_string(claim_path).unwrap(),
format!("{}\n", std::process::id())
);
}
#[test]
fn interview_claim_guard_releases_claim_on_drop() {
let dir = tempfile::tempdir().unwrap();
let claim_path = dir.path().join("runtime").join("interview_request.claim");
{
let _guard = InterviewClaimGuard::acquire(&claim_path).unwrap();
assert!(claim_path.exists());
}
assert!(!claim_path.exists());
}
#[test]
fn answer_requires_reattach_for_aborted_and_skipped_answers() {
let aborted = Answer {
@ -757,65 +507,4 @@ mod tests {
assert!(answer_requires_reattach(&skipped));
assert!(!answer_requires_reattach(&answered));
}
#[test]
fn engine_child_guard_kills_on_drop() {
let child = std::process::Command::new("sleep")
.arg("60")
.spawn()
.unwrap();
let pid = child.id();
{
let _guard = EngineChildGuard::new(child);
}
// Process should be dead after guard is dropped
assert!(
!process_alive(pid),
"process should be dead after guard drop"
);
}
#[test]
fn engine_child_guard_defuse_keeps_alive() {
let child = std::process::Command::new("sleep")
.arg("60")
.spawn()
.unwrap();
let pid = child.id();
{
let mut guard = EngineChildGuard::new(child);
guard.defuse();
}
// Process should still be alive after defused guard is dropped
assert!(
process_alive(pid),
"process should still be alive after defused guard drop"
);
// Clean up
#[cfg(unix)]
fabro_proc::sigkill(pid);
}
#[test]
fn write_interview_response_atomically_persists_answer() {
let dir = tempfile::tempdir().unwrap();
let response_path = dir.path().join("interview_response.json");
let answer = Answer {
value: AnswerValue::Text("ship it".to_string()),
selected_option: None,
text: Some("ship it".to_string()),
};
write_interview_response_atomically(&response_path, &answer).unwrap();
let saved: Answer =
serde_json::from_str(&std::fs::read_to_string(&response_path).unwrap()).unwrap();
assert_eq!(saved.text.as_deref(), Some("ship it"));
assert!(!response_path.with_extension("json.tmp").exists());
}
}

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

@ -36,7 +36,6 @@ pub(crate) async fn execute(mut args: RunArgs, globals: &GlobalArgs) -> Result<(
Some(&run_id),
true,
styles,
None,
globals.json,
)
.await?;

22
lib/crates/fabro-cli/src/commands/run/cp.rs
View file

@ -3,12 +3,11 @@ use std::path::{Path, PathBuf};
use anyhow::{Context, Result, bail};
use fabro_agent::sandbox::Sandbox;
use fabro_sandbox::reconnect::reconnect;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tokio::fs;
use tracing::{debug, info};
use crate::args::{CpArgs, GlobalArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{print_json_pretty, split_run_path};
use crate::user_config::load_user_settings_with_globals;
@ -28,7 +27,6 @@ enum CopyDirection {
pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()> {
let direction = parse_direction(&args.src, &args.dst)?;
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
match direction {
CopyDirection::Download {
@ -36,7 +34,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()>
remote_path,
local_path,
} => {
let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?;
let sandbox = load_sandbox(&cli_settings.storage_dir(), &run_prefix).await?;
let file_count = if args.recursive {
Some(download_recursive(&*sandbox, &remote_path, &local_path).await?)
@ -69,7 +67,7 @@ pub(crate) async fn cp_command(args: CpArgs, globals: &GlobalArgs) -> Result<()>
run_prefix,
remote_path,
} => {
let sandbox = load_sandbox(&cli_settings.storage_dir(), &base, &run_prefix).await?;
let sandbox = load_sandbox(&cli_settings.storage_dir(), &run_prefix).await?;
let file_count = if args.recursive {
Some(upload_recursive(&*sandbox, &local_path, &remote_path).await?)
@ -123,15 +121,11 @@ fn parse_direction(src: &str, dst: &str) -> Result<CopyDirection> {
}
}
async fn load_sandbox(
storage_dir: &Path,
base: &Path,
run_prefix: &str,
) -> Result<Box<dyn Sandbox>> {
let client = server_client::connect_server(storage_dir).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, base, run_prefix)?;
let record = client
async fn load_sandbox(storage_dir: &Path, run_prefix: &str) -> Result<Box<dyn Sandbox>> {
let lookup = ServerRunLookup::connect(storage_dir).await?;
let run = lookup.resolve(run_prefix)?;
let record = lookup
.client()
.get_run_state(&run.run_id())
.await?
.sandbox

11
lib/crates/fabro-cli/src/commands/run/diff.rs
View file

@ -3,25 +3,22 @@ use std::path::Path;
use anyhow::{Context, Result, bail};
use fabro_sandbox::reconnect::reconnect;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use fabro_workflow::sandbox_git::GIT_REMOTE;
use tracing::{debug, info};
use crate::args::{DiffArgs, GlobalArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: DiffArgs, globals: &GlobalArgs) -> Result<()> {
info!(run_id = %args.run, "Showing diff");
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let state = client.get_run_state(&run_id).await?;
let state = lookup.client().get_run_state(&run_id).await?;
let patch = resolve_diff(&run.path, &state, &args).await?;

11
lib/crates/fabro-cli/src/commands/run/fork.rs
View file

@ -3,25 +3,22 @@ use anyhow::Result;
use fabro_checkpoint::git::Store;
use fabro_util::terminal::Styles;
use fabro_workflow::operations::{ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork};
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use git2::Repository;
use crate::args::{ForkArgs, GlobalArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: &ForkArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let repo = Repository::discover(".").context("not in a git repository")?;
let cli_settings = load_user_settings_with_globals(globals)?;
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let base = runs_base(&cli_settings.storage_dir());
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let store = Store::new(repo);
let events = client.list_run_events(&run_id, None, None).await?;
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;

217
lib/crates/fabro-cli/src/commands/run/launcher.rs
View file

@ -1,217 +0,0 @@
use std::path::{Path, PathBuf};
#[cfg(test)]
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use fabro_types::RunId;
use serde::{Deserialize, Serialize};
#[cfg(test)]
use crate::commands::run::short_run_id;
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct LauncherRecord {
pub run_id: RunId,
pub run_dir: PathBuf,
pub pid: u32,
pub resume: bool,
pub log_path: PathBuf,
pub started_at: DateTime<Utc>,
}
pub(crate) fn launcher_dir(storage_dir: &Path) -> PathBuf {
storage_dir.join("launchers")
}
pub(crate) fn launcher_record_path(storage_dir: &Path, run_id: &RunId) -> PathBuf {
launcher_dir(storage_dir).join(format!("{run_id}.json"))
}
#[cfg(test)]
pub(crate) fn write_launcher_record(path: &Path, record: &LauncherRecord) -> Result<()> {
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)?;
}
std::fs::write(path, serde_json::to_string_pretty(record)?)
.with_context(|| format!("Failed to write launcher metadata to {}", path.display()))
}
pub(crate) fn read_launcher_record(path: &Path) -> Option<LauncherRecord> {
let content = std::fs::read_to_string(path).ok()?;
serde_json::from_str(&content).ok()
}
pub(crate) fn remove_launcher_record(path: &Path) {
let _ = std::fs::remove_file(path);
}
pub(crate) fn active_launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord> {
let launcher = launcher_record_for_run(run_dir)?;
let storage_dir = run_dir.parent()?.parent()?;
let path = launcher_record_path(storage_dir, &launcher.run_id);
if launcher_record_is_running(&launcher) {
Some(launcher)
} else {
remove_launcher_record(&path);
None
}
}
pub(crate) fn launcher_record_for_run(run_dir: &Path) -> Option<LauncherRecord> {
let storage_dir = run_dir.parent()?.parent()?;
let launchers_dir = launcher_dir(storage_dir);
let entries = std::fs::read_dir(&launchers_dir).ok()?;
for entry in entries.filter_map(std::result::Result::ok) {
let path = entry.path();
let Some(launcher) = read_launcher_record(&path) else {
continue;
};
if launcher.run_dir == run_dir {
return Some(launcher);
}
}
None
}
#[cfg(test)]
pub(crate) fn active_launcher_record(storage_dir: &Path, run_id: &RunId) -> Option<LauncherRecord> {
let path = launcher_record_path(storage_dir, run_id);
let launcher = read_launcher_record(&path)?;
if launcher_record_is_running(&launcher) {
Some(launcher)
} else {
remove_launcher_record(&path);
None
}
}
pub(crate) fn launcher_record_is_running(record: &LauncherRecord) -> bool {
fabro_proc::process_alive(record.pid) && launcher_process_matches(record)
}
#[cfg(unix)]
fn launcher_process_matches(record: &LauncherRecord) -> bool {
let output = match std::process::Command::new("ps")
.args(["-ww", "-o", "command=", "-p", &record.pid.to_string()])
.output()
{
Ok(output) if output.status.success() => output,
_ => return false,
};
let command = String::from_utf8_lossy(&output.stdout);
command_matches_launcher(record, &command)
}
#[cfg(unix)]
fn command_matches_launcher(record: &LauncherRecord, command: &str) -> bool {
let run_dir = record.run_dir.to_string_lossy();
let old_match = command.contains("__detached") && command.contains(run_dir.as_ref());
let run_id = record.run_id.to_string();
let new_match = command.contains(&format!("fabro: {}", super::short_run_id(&run_id)));
old_match || new_match
}
#[cfg(not(unix))]
fn launcher_process_matches(_record: &LauncherRecord) -> bool {
true
}
#[cfg(test)]
mod tests {
use super::*;
use chrono::Utc;
use fabro_types::fixtures;
#[test]
fn active_launcher_record_for_run_removes_stale_record() {
let dir = tempfile::tempdir().unwrap();
let storage_dir = dir.path().join("storage");
let run_dir = storage_dir.join("runs").join("run");
std::fs::create_dir_all(&run_dir).unwrap();
let launcher_path = launcher_record_path(&storage_dir, &fixtures::RUN_1);
write_launcher_record(
&launcher_path,
&LauncherRecord {
run_id: fixtures::RUN_1,
run_dir: run_dir.clone(),
pid: u32::MAX,
resume: false,
log_path: dir.path().join("launcher.log"),
started_at: Utc::now(),
},
)
.unwrap();
assert!(active_launcher_record_for_run(&run_dir).is_none());
assert!(active_launcher_record(&storage_dir, &fixtures::RUN_1).is_none());
assert!(!launcher_path.exists());
}
#[test]
fn active_launcher_record_for_run_falls_back_without_run_json() {
let dir = tempfile::tempdir().unwrap();
let storage_dir = dir.path().join("storage");
let run_dir = storage_dir.join("runs").join("20260401-test");
std::fs::create_dir_all(&run_dir).unwrap();
let launcher_path = launcher_record_path(&storage_dir, &fixtures::RUN_1);
write_launcher_record(
&launcher_path,
&LauncherRecord {
run_id: fixtures::RUN_1,
run_dir: run_dir.clone(),
pid: u32::MAX,
resume: false,
log_path: dir.path().join("launcher.log"),
started_at: Utc::now(),
},
)
.unwrap();
assert!(active_launcher_record_for_run(&run_dir).is_none());
assert!(!launcher_path.exists());
}
#[cfg(unix)]
#[test]
fn command_matches_launcher_accepts_old_detached_format() {
let dir = tempfile::tempdir().unwrap();
let record = LauncherRecord {
run_id: fixtures::RUN_2,
run_dir: dir.path().join("run"),
pid: 42,
resume: false,
log_path: dir.path().join("launcher.log"),
started_at: Utc::now(),
};
let command = format!(
"/usr/local/bin/fabro __detached --run-dir {} --launcher-path /tmp/launcher.json",
record.run_dir.display()
);
assert!(command_matches_launcher(&record, &command));
}
#[cfg(unix)]
#[test]
fn command_matches_launcher_accepts_new_title_format() {
let dir = tempfile::tempdir().unwrap();
let record = LauncherRecord {
run_id: fixtures::RUN_3,
run_dir: dir.path().join("run"),
pid: 42,
resume: false,
log_path: dir.path().join("launcher.log"),
started_at: Utc::now(),
};
assert!(command_matches_launcher(
&record,
&format!("fabro: {} plan", short_run_id(&record.run_id.to_string()))
));
}
}

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

@ -7,20 +7,19 @@ use chrono::{DateTime, Utc};
use fabro_util::json::normalize_json_value;
use fabro_util::redact::redact_jsonl_line;
use fabro_util::terminal::Styles;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tokio::time;
use tracing::{debug, info};
use crate::args::{GlobalArgs, LogsArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let client = lookup.client();
let run_id = run.run_id();
info!(run_id = %run_id, "Showing logs");
@ -64,7 +63,7 @@ pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs)
return Ok(());
}
follow_store_logs(
&client,
client,
&run_id,
if last_seq == 0 { 1 } else { last_seq + 1 },
pretty,

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

@ -1,9 +1,8 @@
use anyhow::Result;
use fabro_util::terminal::Styles;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use crate::args::{GlobalArgs, RunArgs, RunCommands};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::{load_user_settings_with_globals, user_layer_with_globals};
@ -11,10 +10,8 @@ pub(crate) mod attach;
pub(crate) mod command;
pub(crate) mod cp;
pub(crate) mod create;
pub(crate) mod detached;
pub(crate) mod diff;
pub(crate) mod fork;
pub(crate) mod launcher;
pub(crate) mod logs;
pub(crate) mod output;
pub(crate) mod overrides;
@ -22,14 +19,11 @@ pub(crate) mod preview;
pub(crate) mod resume;
pub(crate) mod rewind;
pub(crate) mod run_progress;
pub(crate) mod runner;
pub(crate) mod ssh;
pub(crate) mod start;
pub(crate) mod wait;
pub(super) fn short_run_id(id: &str) -> &str {
if id.len() > 12 { &id[..12] } else { id }
}
fn apply_json_defaults(args: &mut RunArgs, globals: &GlobalArgs) {
if globals.json {
args.auto_approve = true;
@ -56,10 +50,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
}
RunCommands::Start { run } => {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run_info = resolve_run_from_summaries(&summaries, &base, &run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run_info = lookup.resolve(&run)?;
let run_id = run_info.run_id();
start::start_run(&run_id, &cli_settings.storage_dir(), false).await?;
if globals.json {
@ -70,10 +62,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
RunCommands::Attach { run } => {
let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr()));
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run_info = resolve_run_from_summaries(&summaries, &base, &run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run_info = lookup.resolve(&run)?;
let run_id = run_info.run_id();
let exit_code = attach::attach_run(
&run_info.path,
@ -81,7 +71,6 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
Some(&run_id),
false,
styles,
None,
globals.json,
)
.await?;
@ -90,20 +79,8 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
}
Ok(())
}
RunCommands::Detached {
run_id,
run_dir,
launcher_path,
resume,
} => {
detached::execute(
run_id,
run_dir,
globals.storage_dir.clone(),
launcher_path,
resume,
)
.await
RunCommands::Runner { run_id, resume } => {
runner::execute(run_id, globals.storage_dir.clone(), resume).await
}
RunCommands::Diff(args) => diff::run(args, globals).await,
RunCommands::Logs(args) => {

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

@ -1,20 +1,18 @@
use anyhow::{Context, Result};
use fabro_sandbox::daytona::DaytonaSandbox;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tracing::info;
use crate::args::{GlobalArgs, PreviewArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{print_json_pretty, validate_daytona_provider};
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn run(args: PreviewArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let record = client
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let record = lookup
.client()
.get_run_state(&run.run_id())
.await?
.sandbox

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

@ -1,15 +1,14 @@
use fabro_util::terminal::Styles;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use crate::args::{GlobalArgs, ResumeArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
/// Resume an interrupted workflow run.
///
/// Looks up the run by ID prefix, validates a checkpoint exists, cleans stale
/// artifacts from the previous execution, then spawns an engine subprocess
/// artifacts from the previous execution, then asks the server to resume it
/// (identical to `fabro run`'s create→start→attach flow).
pub(crate) async fn resume_command(
args: ResumeArgs,
@ -17,10 +16,8 @@ pub(crate) async fn resume_command(
globals: &GlobalArgs,
) -> anyhow::Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let run_dir = run.path;
@ -39,7 +36,6 @@ pub(crate) async fn resume_command(
Some(&run_id),
true,
styles,
None,
globals.json,
)
.await?;

13
lib/crates/fabro-cli/src/commands/run/rewind.rs
View file

@ -10,14 +10,13 @@ use fabro_workflow::git::MetadataStore;
use fabro_workflow::operations::{
RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind,
};
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use git2::Repository;
use serde::Serialize;
use crate::args::{GlobalArgs, RewindArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client;
use crate::server_client::ServerStoreClient;
use crate::server_runs::ServerRunLookup;
use crate::shared::{color_if, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
@ -32,13 +31,11 @@ pub(crate) struct TimelineEntryJson {
pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let repo = Repository::discover(".").context("not in a git repository")?;
let cli_settings = load_user_settings_with_globals(globals)?;
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let base = runs_base(&cli_settings.storage_dir());
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run_id)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let store = Store::new(repo);
let events = client.list_run_events(&run_id, None, None).await?;
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;
@ -63,7 +60,7 @@ pub(crate) async fn run(args: &RewindArgs, styles: &Styles, globals: &GlobalArgs
},
)?;
let entry = timeline.resolve(&target)?;
reset_rewound_run_state(&client, &store, &run_id, &run.path, entry).await?;
reset_rewound_run_state(lookup.client(), &store, &run_id, &run.path, entry).await?;
let run_id_string = run_id.to_string();

6
lib/crates/fabro-cli/src/commands/run/detached.rs → lib/crates/fabro-cli/src/commands/run/runner.rs
View file

@ -10,17 +10,11 @@ use crate::user_config::load_user_settings;
pub(crate) async fn execute(
run_id: RunId,
_run_dir: PathBuf,
storage_dir: Option<PathBuf>,
launcher_path: PathBuf,
resume: bool,
) -> Result<()> {
let _ = fabro_proc::title_init();
let _launcher_guard = scopeguard::guard(launcher_path.clone(), |path| {
super::launcher::remove_launcher_record(&path);
});
let storage_dir = match storage_dir {
Some(storage_dir) => storage_dir,
None => load_user_settings()?.storage_dir(),

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

@ -1,10 +1,9 @@
use anyhow::{Context, Result, bail};
use fabro_sandbox::daytona::DaytonaSandbox;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use tracing::info;
use crate::args::{GlobalArgs, SshArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{print_json_pretty, validate_daytona_provider};
use crate::user_config::load_user_settings_with_globals;
@ -14,12 +13,11 @@ pub(crate) async fn run(args: SshArgs, globals: &GlobalArgs) -> Result<()> {
}
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let record = client
let record = lookup
.client()
.get_run_state(&run_id)
.await?
.sandbox

10
lib/crates/fabro-cli/src/commands/run/wait.rs
View file

@ -4,12 +4,11 @@ use anyhow::{Result, bail};
use fabro_types::RunId;
use fabro_util::terminal::Styles;
use fabro_workflow::records::Conclusion;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use fabro_workflow::run_status::RunStatus;
use tracing::info;
use crate::args::{GlobalArgs, WaitArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::format_duration_ms;
use crate::user_config::load_user_settings_with_globals;
@ -20,10 +19,9 @@ const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_secs(3
pub(crate) async fn run(args: &WaitArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run_info = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run_info = lookup.resolve(&args.run)?;
let client = lookup.client();
let run_id = run_info.run_id();
info!(run_id = %run_id, "Waiting for run to complete");

11
lib/crates/fabro-cli/src/commands/runs/inspect.rs
View file

@ -4,12 +4,11 @@ use anyhow::Result;
use fabro_types::RunId;
use serde::Serialize;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, InspectArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::server_runs::ServerRunLookup;
use crate::user_config::load_user_settings_with_globals;
#[derive(Debug, Serialize)]
@ -26,12 +25,10 @@ pub(crate) struct InspectOutput {
pub(crate) async fn run(args: &InspectArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let state = client.get_run_state(&run_id).await?;
let state = lookup.client().get_run_state(&run_id).await?;
let output = inspect_run_state(&run_id, &run.path, run.status(), state);
let json = serde_json::to_string_pretty(&[output])?;
println!("{json}");

7
lib/crates/fabro-cli/src/commands/runs/list.rs
View file

@ -11,7 +11,7 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs
use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, RunsListArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{color_if, format_duration_ms, tilde_path};
use crate::user_config::load_user_settings_with_globals;
@ -25,9 +25,8 @@ pub(crate) async fn list_command(
) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let runs = scan_runs_with_summaries(&summaries, &base)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let runs = scan_runs_with_summaries(lookup.summaries(), &base)?;
let label_filters = parse_label_filters(&args.filter.label);
let filtered = filter_runs(
&runs,

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

@ -4,12 +4,13 @@ use anyhow::{Context, Result, bail};
use fabro_sandbox::reconnect::reconnect as reconnect_sandbox;
use fabro_workflow::event::{Event, to_run_event};
use fabro_workflow::run_lookup::RunInfo;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
use fabro_workflow::run_lookup::resolve_run_from_summaries;
use tracing::warn;
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::server_runs::ServerRunLookup;
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_globals;
@ -17,10 +18,15 @@ use super::short_run_id;
pub(crate) async fn remove_command(args: &RunsRemoveArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
remove_from(args, &client, &summaries, &base, globals).await
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
remove_from(
args,
lookup.client(),
lookup.summaries(),
lookup.runs_base(),
globals,
)
.await
}
async fn remove_from(

11
lib/crates/fabro-cli/src/commands/store/dump.rs
View file

@ -3,7 +3,6 @@ use anyhow::{Context, Result};
use fabro_store::StageId;
use fabro_store::{RunProjection, SlateRunStore};
use fabro_workflow::run_dump::RunDump;
use fabro_workflow::run_lookup::{resolve_run_from_summaries, runs_base};
#[cfg(test)]
use serde::de::DeserializeOwned;
use std::io::ErrorKind;
@ -11,18 +10,16 @@ use std::path::Path;
use crate::args::{GlobalArgs, StoreDumpArgs};
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{absolute_or_current, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
pub(crate) async fn dump_command(args: &StoreDumpArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
let run = resolve_run_from_summaries(&summaries, &base, &args.run)?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let events = client.list_run_events(&run_id, None, None).await?;
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let file_count = export_run(&run_store, &args.output).await?;

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

@ -10,7 +10,7 @@ use fabro_workflow::run_lookup::{logs_base, runs_base, scan_runs_with_summaries}
use fabro_workflow::run_status::RunStatus;
use crate::args::{DfArgs, GlobalArgs};
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{format_size, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
@ -49,11 +49,10 @@ pub(super) async fn df_command(args: &DfArgs, globals: &GlobalArgs) -> Result<()
let data_dir = cli_settings.storage_dir();
let runs_base_dir = runs_base(&data_dir);
let logs_base_dir = logs_base(&data_dir);
let client = server_client::connect_server(&data_dir).await?;
let summaries = client.list_store_runs().await?;
let lookup = ServerRunLookup::connect(&data_dir).await?;
df_from(
args,
&summaries,
lookup.summaries(),
&data_dir,
&runs_base_dir,
&logs_base_dir,

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

@ -10,6 +10,7 @@ use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs
use crate::args::{GlobalArgs, RunsPruneArgs};
use crate::commands::runs::rm::remove_run_with_cleanup;
use crate::server_client;
use crate::server_runs::ServerRunLookup;
use crate::shared::{format_size, print_json_pretty};
use crate::user_config::load_user_settings_with_globals;
@ -24,9 +25,8 @@ struct PruneRunRow {
pub(super) async fn prune_command(args: &RunsPruneArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_globals(globals)?;
let base = runs_base(&cli_settings.storage_dir());
let client = server_client::connect_server(&cli_settings.storage_dir()).await?;
let summaries = client.list_store_runs().await?;
prune_from(args, &client, &summaries, &base, globals).await
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
prune_from(args, lookup.client(), lookup.summaries(), &base, globals).await
}
pub(crate) fn parse_duration(s: &str) -> Result<chrono::Duration> {

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

@ -4,6 +4,7 @@ mod args;
mod commands;
mod logging;
mod server_client;
mod server_runs;
mod shared;
#[cfg(feature = "sleep_inhibitor")]
mod sleep_inhibitor;
@ -421,31 +422,17 @@ mod tests {
}
#[test]
fn parse_detached_command() {
fn parse_runner_command() {
let cli = Cli::try_parse_from([
"fabro",
"__detached",
"__runner",
"--run-id",
"01ARZ3NDEKTSV4RRFFQ69G5FAV",
"--run-dir",
"/tmp/fabro/runs/01ABC",
"--launcher-path",
"/tmp/fabro/launchers/01ABC.json",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Detached {
run_id,
run_dir,
launcher_path,
resume,
}) => {
Commands::RunCmd(RunCommands::Runner { run_id, resume }) => {
assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap());
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC"));
assert_eq!(
launcher_path,
std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json")
);
assert!(!resume);
}
_ => panic!("unexpected command variant"),
@ -453,32 +440,18 @@ mod tests {
}
#[test]
fn parse_detached_with_resume() {
fn parse_runner_with_resume() {
let cli = Cli::try_parse_from([
"fabro",
"__detached",
"__runner",
"--run-id",
"01ARZ3NDEKTSV4RRFFQ69G5FAV",
"--run-dir",
"/tmp/fabro/runs/01ABC",
"--launcher-path",
"/tmp/fabro/launchers/01ABC.json",
"--resume",
])
.expect("should parse");
match *cli.command {
Commands::RunCmd(RunCommands::Detached {
run_id,
run_dir,
launcher_path,
resume,
}) => {
Commands::RunCmd(RunCommands::Runner { run_id, resume }) => {
assert_eq!(run_id, "01ARZ3NDEKTSV4RRFFQ69G5FAV".parse().unwrap());
assert_eq!(run_dir, std::path::PathBuf::from("/tmp/fabro/runs/01ABC"));
assert_eq!(
launcher_path,
std::path::PathBuf::from("/tmp/fabro/launchers/01ABC.json")
);
assert!(resume);
}
_ => panic!("unexpected command variant"),

46
lib/crates/fabro-cli/src/server_runs.rs Normal file
View file

@ -0,0 +1,46 @@
use std::path::{Path, PathBuf};
use anyhow::Result;
use fabro_store::RunSummary;
use fabro_workflow::run_lookup::{RunInfo, resolve_run_from_summaries, runs_base};
use crate::server_client::{self, ServerStoreClient};
pub(crate) struct ServerRunLookup {
client: ServerStoreClient,
runs_base: PathBuf,
summaries: Vec<RunSummary>,
}
impl ServerRunLookup {
pub(crate) async fn connect(storage_dir: &Path) -> Result<Self> {
Self::connect_from_runs_base(&runs_base(storage_dir)).await
}
pub(crate) async fn connect_from_runs_base(runs_base: &Path) -> Result<Self> {
let storage_dir = runs_base.parent().unwrap_or(runs_base);
let client = server_client::connect_server(storage_dir).await?;
let summaries = client.list_store_runs().await?;
Ok(Self {
client,
runs_base: runs_base.to_path_buf(),
summaries,
})
}
pub(crate) fn client(&self) -> &ServerStoreClient {
&self.client
}
pub(crate) fn runs_base(&self) -> &Path {
&self.runs_base
}
pub(crate) fn summaries(&self) -> &[RunSummary] {
&self.summaries
}
pub(crate) fn resolve(&self, selector: &str) -> Result<RunInfo> {
resolve_run_from_summaries(&self.summaries, &self.runs_base, selector)
}
}

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

@ -14,7 +14,7 @@ fn help() {
Commands:
run Launch a workflow run
create Create a workflow run (allocate run dir, persist spec)
start Start a created workflow run (spawn engine process)
start Start a created workflow run on the server
attach Attach to a running or finished workflow run
logs View the event log of a workflow run
resume Resume an interrupted workflow run

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

@ -8,7 +8,9 @@ fn parse_ndjson(stdout: &[u8]) -> Vec<Value> {
.expect("stdout should be valid UTF-8")
.lines()
.filter(|line| !line.trim().is_empty())
.map(|line| serde_json::from_str::<Value>(line).expect("logs output should be valid NDJSON"))
.map(|line| {
serde_json::from_str::<Value>(line).expect("logs output should be valid NDJSON")
})
.collect()
}
@ -22,7 +24,8 @@ fn assert_event_sequence_contains(events: &[Value], expected: &[&str]) {
for expected_name in expected {
let Some(found_at) = event_names[cursor..]
.iter()
.position(|name| name == expected_name) else {
.position(|name| name == expected_name)
else {
panic!("missing event {expected_name} in sequence: {event_names:?}");
};
cursor += found_at + 1;

2
lib/crates/fabro-cli/tests/it/cmd/mod.rs
View file

@ -5,7 +5,6 @@ mod attach;
mod completion;
mod config;
mod create;
mod detached;
mod diff;
mod discord;
mod docs;
@ -41,6 +40,7 @@ mod resume;
mod rewind;
mod rm;
mod run;
mod runner;
mod sandbox_cp;
mod sandbox_preview;
mod sandbox_ssh;

43
lib/crates/fabro-cli/tests/it/cmd/resume.rs
View file

@ -2,6 +2,8 @@ use fabro_test::{fabro_snapshot, test_context};
use super::support::{git_stdout, output_stderr, setup_git_backed_changed_run};
const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
#[test]
fn help() {
let context = test_context!();
@ -115,3 +117,44 @@ fn resume_rewound_run_succeeds() {
);
assert_ne!(resumed_head.trim(), rewound_head.trim());
}
#[test]
fn resume_detached_does_not_create_launcher_record() {
let context = test_context!();
let setup = setup_git_backed_changed_run(&context);
context
.command()
.current_dir(&setup.repo_dir)
.args(["rewind", &setup.run.run_id, "@1", "--no-push"])
.assert()
.success();
let mut resume_cmd = context.command();
resume_cmd.current_dir(&setup.repo_dir);
resume_cmd.env("OPENAI_API_KEY", "test");
resume_cmd.args(["resume", "--detach", &setup.run.run_id]);
let resume_output = resume_cmd.output().expect("resume should execute");
assert!(
resume_output.status.success(),
"resume should succeed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&resume_output.stdout),
output_stderr(&resume_output)
);
assert!(
!context
.storage_dir
.join("launchers")
.join(format!("{}.json", setup.run.run_id))
.exists(),
"server-owned resume should not create a launcher record"
);
context
.command()
.args(["wait", &setup.run.run_id])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
}

78
lib/crates/fabro-cli/tests/it/cmd/run.rs
View file

@ -1,8 +1,15 @@
use fabro_test::{fabro_snapshot, test_context};
use fabro_types::StatusReason;
use serde_json::Value;
use super::support::{
only_run, output_stderr, run_count_for_test_case, run_state, wait_for_status,
write_gated_workflow,
};
use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id};
const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
#[test]
fn help() {
let context = test_context!();
@ -242,7 +249,10 @@ fn json_run_implies_auto_approve_for_human_gates() {
let Some(llm) = llm.as_object_mut() else {
continue;
};
llm.insert("model".to_string(), Value::String("[LLM_MODEL]".to_string()));
llm.insert(
"model".to_string(),
Value::String("[LLM_MODEL]".to_string()),
);
llm.insert(
"provider".to_string(),
Value::String("[LLM_PROVIDER]".to_string()),
@ -961,3 +971,69 @@ fn detach_creates_run_dir_with_detach_log() {
"#
);
}
#[test]
fn ctrl_c_cancels_active_run_via_server() {
let context = test_context!();
let _gate = write_gated_workflow(&context.temp_dir.join("slow.fabro"), "slow", "Run slowly");
let mut run_cmd = std::process::Command::new(env!("CARGO_BIN_EXE_fabro"));
run_cmd.current_dir(&context.temp_dir);
run_cmd.env("NO_COLOR", "1");
run_cmd.env("HOME", &context.home_dir);
run_cmd.env("FABRO_NO_UPGRADE_CHECK", "true");
run_cmd.env("FABRO_STORAGE_DIR", &context.storage_dir);
run_cmd.env("FABRO_SERVER_MAX_CONCURRENT_RUNS", "64");
run_cmd.env("OPENAI_API_KEY", "test");
run_cmd.args([
"run",
"--label",
&context.test_run_label(),
"--label",
&context.test_case_label(),
"--provider",
"openai",
"--sandbox",
"local",
"--no-retro",
"slow.fabro",
]);
let child = run_cmd.spawn().expect("run should spawn");
let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT;
while run_count_for_test_case(&context) == 0 {
assert!(
std::time::Instant::now() < deadline,
"timed out waiting for run directory"
);
std::thread::sleep(std::time::Duration::from_millis(50));
}
let run = only_run(&context);
wait_for_status(&run.run_dir, &["running"]);
let kill_status = std::process::Command::new("kill")
.args(["-INT", &child.id().to_string()])
.status()
.expect("kill should execute");
assert!(kill_status.success(), "kill -INT should succeed");
let output = child
.wait_with_output()
.expect("run should exit after SIGINT");
assert!(
!output.status.success(),
"run should exit non-zero after cancellation\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
output_stderr(&output)
);
let final_status = wait_for_status(&run.run_dir, &["failed"]);
assert_eq!(final_status, "failed");
assert_eq!(
run_state(&run.run_dir)
.status
.and_then(|record| record.reason),
Some(StatusReason::Cancelled)
);
}

85
lib/crates/fabro-cli/tests/it/cmd/detached.rs → lib/crates/fabro-cli/tests/it/cmd/runner.rs
View file

@ -9,41 +9,32 @@ const SHARED_DAEMON_TIMEOUT: std::time::Duration = std::time::Duration::from_sec
fn help() {
let context = test_context!();
let mut cmd = context.command();
cmd.args(["__detached", "--help"]);
cmd.args(["__runner", "--help"]);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
Internal: run the engine process
Internal: queue or resume a workflow run via the server
Usage: fabro __detached [OPTIONS] --run-id <RUN_ID> --run-dir <RUN_DIR> --launcher-path <LAUNCHER_PATH>
Usage: fabro __runner [OPTIONS] --run-id <RUN_ID>
Options:
--json Output as JSON [env: FABRO_JSON=]
--run-id <RUN_ID> Run ID
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--run-dir <RUN_DIR> Run directory
--launcher-path <LAUNCHER_PATH> Launcher metadata path
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--resume Resume from checkpoint instead of fresh start
--verbose Enable verbose output [env: FABRO_VERBOSE=]
--storage-dir <STORAGE_DIR> Storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--server-url <SERVER_URL> Server URL (overrides server.base_url from user.toml) [env: FABRO_SERVER_URL=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--run-id <RUN_ID> Run ID
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--resume Resume from checkpoint instead of fresh start
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--verbose Enable verbose output [env: FABRO_VERBOSE=]
--storage-dir <STORAGE_DIR> Storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--server-url <SERVER_URL> Server URL (overrides server.base_url from user.toml) [env: FABRO_SERVER_URL=]
-h, --help Print help
----- stderr -----
");
}
fn launcher_path(context: &fabro_test::TestContext, run_id: &str) -> std::path::PathBuf {
context
.storage_dir
.join("launchers")
.join(format!("{run_id}.json"))
}
#[test]
fn detached_uses_cached_graph_after_source_deleted() {
fn runner_uses_cached_graph_after_source_deleted() {
let context = test_context!();
let run_id = unique_run_id();
let workflow_path = context.temp_dir.join("workflow.fabro");
@ -77,15 +68,7 @@ digraph CachedGraph {
context
.command()
.args([
"__detached",
"--run-id",
run_id.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
launcher_path(&context, &run_id).to_str().unwrap(),
])
.args(["__runner", "--run-id", run_id.as_str()])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
@ -110,7 +93,7 @@ digraph CachedGraph {
}
#[test]
fn detached_uses_snapshotted_app_id_for_github_credentials() {
fn runner_uses_snapshotted_app_id_for_github_credentials() {
let context = test_context!();
let run_id = unique_run_id();
let workflow_path = context.temp_dir.join("workflow.fabro");
@ -167,15 +150,7 @@ digraph GitHubApp {
let mut cmd = context.command();
cmd.env("GITHUB_APP_PRIVATE_KEY", "%%%not-base64%%%");
cmd.args([
"__detached",
"--run-id",
run_id.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
launcher_path(&context, &run_id).to_str().unwrap(),
]);
cmd.args(["__runner", "--run-id", run_id.as_str()]);
cmd.timeout(SHARED_DAEMON_TIMEOUT);
fabro_snapshot!(context.filters(), cmd, @"
success: true
@ -186,7 +161,7 @@ digraph GitHubApp {
}
#[test]
fn detached_runs_without_run_json_when_run_id_is_explicit() {
fn runner_runs_without_run_json_when_run_id_is_explicit() {
let context = test_context!();
let run_id = unique_run_id();
let workflow_path = context.temp_dir.join("workflow.fabro");
@ -218,15 +193,7 @@ digraph DetachedStoreOnly {
let run_dir = context.find_run_dir(&run_id);
context
.command()
.args([
"__detached",
"--run-id",
run_id.as_str(),
"--run-dir",
run_dir.to_str().unwrap(),
"--launcher-path",
launcher_path(&context, &run_id).to_str().unwrap(),
])
.args(["__runner", "--run-id", run_id.as_str()])
.timeout(SHARED_DAEMON_TIMEOUT)
.assert()
.success();
@ -251,7 +218,7 @@ digraph DetachedStoreOnly {
}
#[test]
fn detached_resume_rejects_completed_run_without_mutating_it() {
fn runner_resume_rejects_completed_run_without_mutating_it() {
let context = test_context!();
context.write_temp(
"workflow.fabro",
@ -301,7 +268,6 @@ digraph Test {
"conclusion_timestamp": before[0]["conclusion"]["timestamp"],
"conclusion_status": before[0]["conclusion"]["status"],
});
let run_dir = before_summary["run_dir"].as_str().unwrap().to_string();
fabro_json_snapshot!(context, &before_summary, @r#"
{
"run_dir": "[RUN_DIR]",
@ -312,16 +278,7 @@ digraph Test {
"#);
let mut cmd = context.command();
cmd.args([
"__detached",
"--run-id",
&run_id,
"--run-dir",
&run_dir,
"--launcher-path",
launcher_path(&context, &run_id).to_str().unwrap(),
"--resume",
]);
cmd.args(["__runner", "--run-id", &run_id, "--resume"]);
cmd.timeout(SHARED_DAEMON_TIMEOUT);
fabro_snapshot!(context.filters(), cmd, @"
success: true

2
lib/crates/fabro-cli/tests/it/cmd/start.rs
View file

@ -15,7 +15,7 @@ fn help() {
success: true
exit_code: 0
----- stdout -----
Start a created workflow run (spawn engine process)
Start a created workflow run on the server
Usage: fabro start [OPTIONS] <RUN>

7
lib/crates/fabro-cli/tests/it/scenario/artifacts.rs
View file

@ -165,7 +165,12 @@ fn artifact_commands_share_populated_run_fixture() {
let flat_dest = context.temp_dir.join("artifact-flat");
let mut cp_flat = context.command();
cp_flat.args(["artifact", "cp", &setup.run.run_id, flat_dest.to_str().unwrap()]);
cp_flat.args([
"artifact",
"cp",
&setup.run.run_id,
flat_dest.to_str().unwrap(),
]);
fabro_snapshot!(context.filters(), cp_flat, @"
success: false
exit_code: 1