scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-07 03:00:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

refactor(cli): make core run lifecycle server-only

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-05 21:57:27 -04:00
parent aa15e7a1be
commit c66ba89875
No known key found for this signature in database
19 changed files with 655 additions and 163 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -246,7 +246,7 @@ pub(crate) struct RunFilterArgs {
#[derive(Args)]
pub(crate) struct RunsListArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
#[command(flatten)]
pub(crate) filter: RunFilterArgs,
@ -263,7 +263,7 @@ pub(crate) struct RunsListArgs {
#[derive(Args)]
pub(crate) struct RunsRemoveArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run IDs or workflow names to remove
#[arg(required = true)]
@ -277,7 +277,7 @@ pub(crate) struct RunsRemoveArgs {
#[derive(Args)]
pub(crate) struct LogsArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name (most recent run)
pub(crate) run: String,
@ -484,7 +484,7 @@ pub(crate) struct DiffArgs {
#[derive(Args)]
pub(crate) struct InspectArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name (most recent run)
pub(crate) run: String,
@ -576,7 +576,7 @@ pub(crate) struct ForkArgs {
#[derive(Args)]
pub(crate) struct WaitArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name (most recent run)
pub(crate) run: String,
@ -735,7 +735,7 @@ pub(crate) struct PrCloseArgs {
#[derive(Args)]
pub(crate) struct StartArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name
pub(crate) run: String,
@ -744,7 +744,7 @@ pub(crate) struct StartArgs {
#[derive(Args)]
pub(crate) struct AttachArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name
pub(crate) run: String,

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

@ -12,12 +12,10 @@ 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_storage_dir;
use crate::server_runs::ServerSummaryLookup;
pub(crate) async fn run(args: &LogsArgs, styles: &Styles, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_storage_dir(args.storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let lookup = ServerSummaryLookup::connect(&args.server).await?;
let run = lookup.resolve(&args.run)?;
let client = lookup.client();

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

@ -2,9 +2,9 @@ use anyhow::Result;
use fabro_util::terminal::Styles;
use crate::args::{AttachArgs, GlobalArgs, RunArgs, RunCommands, RunnerArgs, StartArgs};
use crate::server_runs::ServerRunLookup;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
use crate::user_config::{load_user_settings_with_storage_dir, user_layer_with_storage_dir};
use crate::user_config::user_layer_with_storage_dir;
pub(crate) mod attach;
pub(crate) mod command;
@ -48,27 +48,24 @@ pub(crate) async fn dispatch(cmd: RunCommands, globals: &GlobalArgs) -> Result<(
}
Ok(())
}
RunCommands::Start(StartArgs { storage_dir, run }) => {
let cli_settings = load_user_settings_with_storage_dir(storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
RunCommands::Start(StartArgs { server, run }) => {
let lookup = ServerSummaryLookup::connect(&server).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?;
start::start_run_with_client(lookup.client(), &run_id, false).await?;
if globals.json {
print_json_pretty(&serde_json::json!({ "run_id": run_id }))?;
}
Ok(())
}
RunCommands::Attach(AttachArgs { storage_dir, run }) => {
RunCommands::Attach(AttachArgs { server, run }) => {
let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr()));
let cli_settings = load_user_settings_with_storage_dir(storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let lookup = ServerSummaryLookup::connect(&server).await?;
let run_info = lookup.resolve(&run)?;
let run_id = run_info.run_id();
let exit_code = attach::attach_run(
&run_info.path,
Some(cli_settings.storage_dir().as_path()),
Some(&run_id),
let exit_code = attach::attach_run_with_client(
lookup.client(),
&run_id,
false,
styles,
globals.json,

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

@ -8,9 +8,8 @@ use fabro_workflow::run_status::RunStatus;
use tracing::info;
use crate::args::{GlobalArgs, WaitArgs};
use crate::server_runs::ServerRunLookup;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::format_duration_ms;
use crate::user_config::load_user_settings_with_storage_dir;
#[cfg(test)]
const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_millis(500);
@ -18,8 +17,7 @@ const WAIT_STARTUP_GRACE: std::time::Duration = std::time::Duration::from_millis
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_storage_dir(args.storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let lookup = ServerSummaryLookup::connect(&args.server).await?;
let run_info = lookup.resolve(&args.run)?;
let client = lookup.client();

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

@ -1,20 +1,15 @@
use std::path::{Path, PathBuf};
use anyhow::Result;
use fabro_types::RunId;
use serde::Serialize;
use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, InspectArgs};
use crate::server_client::RunProjection;
use crate::server_runs::ServerRunLookup;
use crate::user_config::load_user_settings_with_storage_dir;
use crate::server_runs::{ServerRunSummaryInfo, ServerSummaryLookup};
#[derive(Debug, Serialize)]
pub(crate) struct InspectOutput {
pub run_id: String,
pub run_dir: PathBuf,
pub status: RunStatus,
pub run_record: Option<serde_json::Value>,
pub start_record: Option<serde_json::Value>,
@ -24,27 +19,23 @@ pub(crate) struct InspectOutput {
}
pub(crate) async fn run(args: &InspectArgs, _globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_storage_dir(args.storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let lookup = ServerSummaryLookup::connect(&args.server).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let output = inspect_run_state(&run_id, &run.path, run.status(), state);
let output = inspect_run_state(&run, state);
let json = serde_json::to_string_pretty(&[output])?;
println!("{json}");
Ok(())
}
fn inspect_run_state(
run_id: &RunId,
run_dir: &Path,
status: RunStatus,
state: RunProjection,
) -> InspectOutput {
fn inspect_run_state(run: &ServerRunSummaryInfo, state: RunProjection) -> InspectOutput {
InspectOutput {
run_id: run_id.to_string(),
run_dir: run_dir.to_path_buf(),
status: state.status.as_ref().map_or(status, |record| record.status),
run_id: run.run_id().to_string(),
status: state
.status
.as_ref()
.map_or(run.status(), |record| record.status),
run_record: state
.run
.and_then(|record| serde_json::to_value(record).ok()),

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

@ -7,13 +7,11 @@ use cli_table::{Cell, CellStruct, Color, Style, Table};
use fabro_util::terminal::Styles;
use fabro_util::text::strip_goal_decoration;
use fabro_workflow::run_lookup::{StatusFilter, filter_runs, runs_base, scan_runs_with_summaries};
use fabro_workflow::run_status::RunStatus;
use crate::args::{GlobalArgs, RunsListArgs};
use crate::server_runs::ServerRunLookup;
use crate::server_runs::{ServerSummaryLookup, filter_server_runs};
use crate::shared::{color_if, format_duration_ms, tilde_path};
use crate::user_config::load_user_settings_with_storage_dir;
use super::short_run_id;
@ -23,22 +21,14 @@ pub(crate) async fn list_command(
styles: &Styles,
globals: &GlobalArgs,
) -> Result<()> {
let cli_settings = load_user_settings_with_storage_dir(args.storage_dir.as_deref())?;
let base = runs_base(&cli_settings.storage_dir());
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
let runs = scan_runs_with_summaries(lookup.summaries(), &base)?;
let lookup = ServerSummaryLookup::connect(&args.server).await?;
let label_filters = parse_label_filters(&args.filter.label);
let filtered = filter_runs(
&runs,
let filtered = filter_server_runs(
lookup.runs(),
args.filter.before.as_deref(),
args.filter.workflow.as_deref(),
&label_filters,
args.filter.orphans,
if args.all {
StatusFilter::All
} else {
StatusFilter::RunningOnly
},
!args.all,
);
if globals.json {
@ -47,7 +37,6 @@ pub(crate) async fn list_command(
.map(|run| {
serde_json::json!({
"run_id": run.run_id(),
"dir_name": run.dir_name,
"workflow_name": run.workflow_name(),
"workflow_slug": run.workflow_slug(),
"status": run.status(),
@ -100,7 +89,7 @@ pub(crate) async fn list_command(
.map(|run| {
let duration_display = match run.duration_ms() {
Some(ms) => format_duration_ms(ms),
None => match run.start_time_dt {
None => match run.start_time_dt() {
Some(start) => {
let elapsed = now.signed_duration_since(start);
format_duration_ms(

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

@ -1,39 +1,28 @@
use std::path::Path;
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;
use tracing::warn;
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::server_runs::ServerRunLookup;
use crate::server_runs::{
ServerRunSummaryInfo, ServerSummaryLookup, resolve_server_run_from_summaries,
};
use crate::shared::print_json_pretty;
use crate::user_config::load_user_settings_with_storage_dir;
use super::short_run_id;
pub(crate) async fn remove_command(args: &RunsRemoveArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_user_settings_with_storage_dir(args.storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
remove_from(
args,
lookup.client(),
lookup.summaries(),
lookup.runs_base(),
globals,
)
.await
let lookup = ServerSummaryLookup::connect(&args.server).await?;
remove_from(args, lookup.client(), lookup.runs(), globals).await
}
async fn remove_from(
args: &RunsRemoveArgs,
client: &server_client::ServerStoreClient,
summaries: &[fabro_store::RunSummary],
base: &Path,
runs: &[ServerRunSummaryInfo],
globals: &GlobalArgs,
) -> Result<()> {
let mut had_errors = false;
@ -41,7 +30,7 @@ async fn remove_from(
let mut errors = Vec::new();
for identifier in &args.runs {
let run = match resolve_run_from_summaries(summaries, base, identifier) {
let run = match resolve_server_run_from_summaries(runs, identifier) {
Ok(run) => run,
Err(err) => {
if !globals.json {
@ -75,7 +64,7 @@ async fn remove_from(
}
let run_id = run.run_id().to_string();
if let Err(err) = remove_run_dir_with_cleanup(client, &run).await {
if let Err(err) = delete_server_run(client, &run).await {
if !globals.json {
eprintln!("error: {identifier}: {err}");
}
@ -90,16 +79,6 @@ async fn remove_from(
if !globals.json {
eprintln!("{}", short_run_id(&run_id));
}
if let Err(err) = delete_run_store_state(client, &run).await {
if !globals.json {
eprintln!("error: {identifier}: {err}");
}
errors.push(serde_json::json!({
"identifier": identifier,
"error": err.to_string(),
}));
had_errors = true;
}
}
if globals.json {
@ -179,6 +158,16 @@ async fn delete_run_store_state(
.with_context(|| format!("failed to delete store state for {}", run.run_id()))
}
async fn delete_server_run(
client: &server_client::ServerStoreClient,
run: &ServerRunSummaryInfo,
) -> Result<()> {
client
.delete_store_run(&run.run_id())
.await
.with_context(|| format!("failed to delete store state for {}", run.run_id()))
}
fn load_sandbox_record(run_state: Option<&RunProjection>) -> Option<fabro_sandbox::SandboxRecord> {
if let Some(run_state) = run_state {
return run_state.sandbox.clone();

9
lib/crates/fabro-cli/src/server_client.rs
View file

@ -15,7 +15,7 @@ use futures::StreamExt;
use serde::de::DeserializeOwned;
use tokio::time::sleep;
use crate::args::ServerConnectionArgs;
use crate::args::{ServerConnectionArgs, ServerTargetArgs};
use crate::commands::server::start;
use crate::user_config;
@ -122,6 +122,13 @@ pub(crate) async fn connect_server_connection(
})
}
pub(crate) async fn connect_server_only(args: &ServerTargetArgs) -> Result<ServerStoreClient> {
let storage_dir = std::env::var_os("FABRO_STORAGE_DIR").map(std::path::PathBuf::from);
let settings = user_config::load_user_settings_with_storage_dir(storage_dir.as_deref())?;
let connection = user_config::server_only_command_connection(args, &settings)?;
connect_server_connection(&connection).await
}
pub(crate) async fn connect_api_client(storage_dir: &Path) -> Result<fabro_api::Client> {
let bind = start::ensure_server_running(storage_dir)
.with_context(|| format!("Failed to start fabro server for {}", storage_dir.display()))?;

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

@ -1,9 +1,14 @@
use std::path::{Path, PathBuf};
use anyhow::Result;
use std::collections::HashMap;
use anyhow::{Result, bail};
use chrono::{DateTime, Utc};
use fabro_store::RunSummary;
use fabro_types::{RunId, RunStatus, StatusReason};
use fabro_workflow::run_lookup::{RunInfo, resolve_run_from_summaries, runs_base};
use crate::args::ServerTargetArgs;
use crate::server_client::{self, ServerStoreClient};
pub(crate) struct ServerRunLookup {
@ -32,10 +37,6 @@ impl ServerRunLookup {
&self.client
}
pub(crate) fn runs_base(&self) -> &Path {
&self.runs_base
}
pub(crate) fn summaries(&self) -> &[RunSummary] {
&self.summaries
}
@ -44,3 +45,189 @@ impl ServerRunLookup {
resolve_run_from_summaries(&self.summaries, &self.runs_base, selector)
}
}
#[derive(Debug, Clone)]
pub(crate) struct ServerRunSummaryInfo {
summary: RunSummary,
}
impl ServerRunSummaryInfo {
pub(crate) fn run_id(&self) -> RunId {
self.summary.run_id
}
pub(crate) fn workflow_name(&self) -> String {
self.summary
.workflow_name
.clone()
.unwrap_or_else(|| "[no run record]".to_string())
}
pub(crate) fn workflow_slug(&self) -> Option<&str> {
self.summary.workflow_slug.as_deref()
}
pub(crate) fn status(&self) -> RunStatus {
self.summary.status.unwrap_or(RunStatus::Dead)
}
pub(crate) fn status_reason(&self) -> Option<StatusReason> {
self.summary.status_reason
}
pub(crate) fn start_time(&self) -> String {
self.start_time_dt()
.map(|time| time.to_rfc3339())
.unwrap_or_default()
}
pub(crate) fn start_time_dt(&self) -> Option<DateTime<Utc>> {
self.summary
.start_time
.or(Some(self.summary.run_id.created_at()))
}
pub(crate) fn labels(&self) -> &HashMap<String, String> {
&self.summary.labels
}
pub(crate) fn duration_ms(&self) -> Option<u64> {
self.summary.duration_ms
}
pub(crate) fn total_cost(&self) -> Option<f64> {
self.summary.total_cost
}
pub(crate) fn host_repo_path(&self) -> Option<&str> {
self.summary.host_repo_path.as_deref()
}
pub(crate) fn goal(&self) -> String {
self.summary.goal.clone().unwrap_or_default()
}
}
pub(crate) struct ServerSummaryLookup {
client: ServerStoreClient,
runs: Vec<ServerRunSummaryInfo>,
}
impl ServerSummaryLookup {
pub(crate) async fn connect(args: &ServerTargetArgs) -> Result<Self> {
let client = server_client::connect_server_only(args).await?;
let summaries = client.list_store_runs().await?;
let mut runs = summaries
.into_iter()
.map(|summary| ServerRunSummaryInfo { summary })
.collect::<Vec<_>>();
runs.sort_by(|a, b| {
b.start_time_dt()
.cmp(&a.start_time_dt())
.then_with(|| b.run_id().cmp(&a.run_id()))
});
Ok(Self { client, runs })
}
pub(crate) fn client(&self) -> &ServerStoreClient {
&self.client
}
pub(crate) fn runs(&self) -> &[ServerRunSummaryInfo] {
&self.runs
}
pub(crate) fn resolve(&self, selector: &str) -> Result<ServerRunSummaryInfo> {
resolve_server_run_from_infos(&self.runs, selector)
}
}
pub(crate) fn resolve_server_run_from_summaries(
runs: &[ServerRunSummaryInfo],
selector: &str,
) -> Result<ServerRunSummaryInfo> {
resolve_server_run_from_infos(runs, selector)
}
pub(crate) fn filter_server_runs(
runs: &[ServerRunSummaryInfo],
before: Option<&str>,
workflow: Option<&str>,
labels: &[(String, String)],
running_only: bool,
) -> Vec<ServerRunSummaryInfo> {
runs.iter()
.filter(|run| !running_only || run.status().is_active())
.filter(|run| {
before.is_none_or(|before| {
let start_time = run.start_time();
start_time.is_empty() || start_time.as_str() < before
})
})
.filter(|run| workflow.is_none_or(|pattern| run.workflow_name().contains(pattern)))
.filter(|run| {
labels.iter().all(|(key, value)| {
run.labels()
.get(key)
.is_some_and(|current| current == value)
})
})
.cloned()
.collect()
}
fn resolve_server_run_from_infos(
runs: &[ServerRunSummaryInfo],
identifier: &str,
) -> Result<ServerRunSummaryInfo> {
let id_matches: Vec<_> = runs
.iter()
.filter(|run| run_id_matches(run.run_id(), identifier))
.collect();
match id_matches.len() {
1 => return Ok(id_matches[0].clone()),
count if count > 1 => {
let ids: Vec<String> = id_matches
.iter()
.map(|run| run.run_id().to_string())
.collect();
bail!(
"Ambiguous prefix '{identifier}': {count} runs match: {}",
ids.join(", ")
);
}
_ => {}
}
let id_lower = identifier.to_lowercase();
let id_collapsed = collapse_separators(&id_lower);
let workflow_match = runs
.iter()
.filter(|run| {
if let Some(slug) = run.workflow_slug() {
if slug.to_lowercase() == id_lower {
return true;
}
}
let name_lower = run.workflow_name().to_lowercase();
name_lower.contains(&id_lower)
|| collapse_separators(&name_lower).contains(&id_collapsed)
})
.max_by_key(|run| run.run_id().created_at());
match workflow_match {
Some(run) => Ok(run.clone()),
None => {
bail!("No run found matching '{identifier}' (tried run ID prefix and workflow name)")
}
}
}
fn collapse_separators(s: &str) -> String {
s.chars().filter(|c| *c != '-' && *c != '_').collect()
}
fn run_id_matches(run_id: RunId, prefix: &str) -> bool {
run_id.to_string().starts_with(prefix)
}

71
lib/crates/fabro-cli/src/user_config.rs
View file

@ -147,6 +147,23 @@ pub(crate) fn exec_server_target(
Ok(target)
}
pub(crate) fn server_only_command_connection(
args: &ServerTargetArgs,
settings: &Settings,
) -> Result<ServerConnection> {
let connection = if let Some(target) = explicit_server_target(args, settings)? {
ServerConnection::Target(target)
} else if let Some(target) = configured_server_target(settings)? {
ServerConnection::Target(target)
} else {
ServerConnection::Local {
storage_dir: settings.storage_dir(),
}
};
debug!(?connection, "Resolved server-only command connection");
Ok(connection)
}
pub(crate) fn model_server_connection(
args: &ServerConnectionArgs,
settings: &Settings,
@ -281,6 +298,60 @@ mod tests {
);
}
#[test]
fn server_only_command_uses_configured_server_target() {
let settings = Settings {
server: Some(ServerSettings {
target: Some("https://config.example.com".to_string()),
tls: None,
}),
..Settings::default()
};
assert_eq!(
server_only_command_connection(&server_target_args(None), &settings).unwrap(),
ServerConnection::Target(ServerTarget::HttpUrl {
api_url: "https://config.example.com".to_string(),
tls: None,
})
);
}
#[test]
fn server_only_command_explicit_target_overrides_config_target() {
let settings = Settings {
server: Some(ServerSettings {
target: Some("https://config.example.com".to_string()),
tls: None,
}),
..Settings::default()
};
assert_eq!(
server_only_command_connection(
&server_target_args(Some("https://cli.example.com")),
&settings,
)
.unwrap(),
ServerConnection::Target(ServerTarget::HttpUrl {
api_url: "https://cli.example.com".to_string(),
tls: None,
})
);
}
#[test]
fn server_only_command_defaults_to_local_storage_dir() {
let settings = Settings {
storage_dir: Some(PathBuf::from("/tmp/fabro")),
..Settings::default()
};
assert_eq!(
server_only_command_connection(&server_target_args(None), &settings).unwrap(),
ServerConnection::Local {
storage_dir: PathBuf::from("/tmp/fabro"),
}
);
}
#[test]
fn explicit_server_target_overrides_config_target() {
let settings = Settings {

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

@ -1,6 +1,7 @@
use std::time::Duration;
use fabro_test::{fabro_snapshot, run_and_format, test_context};
use httpmock::MockServer;
use serde_json::Value;
use crate::support::{example_fixture, fabro_json_snapshot, run_output_filters, unique_run_id};
@ -26,13 +27,13 @@ fn help() {
<RUN> Run ID prefix or workflow name
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
----- stderr -----
");
}
@ -50,12 +51,116 @@ fn attach_requires_run_arg() {
error: the following required arguments were not provided:
<RUN>
Usage: fabro attach --storage-dir <STORAGE_DIR> --no-upgrade-check <RUN>
Usage: fabro attach --no-upgrade-check <RUN>
For more information, try '--help'.
");
}
#[test]
fn attach_uses_configured_server_target_without_server_flag() {
let context = test_context!();
let server = MockServer::start();
let run_id = unique_run_id();
let list_mock = server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!([
{
"run_id": run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"labels": {},
"host_repo_path": null,
"start_time": "2026-04-05T12:00:00Z",
"status": "succeeded",
"status_reason": null,
"duration_ms": 12,
"total_cost": null
}
])
.to_string(),
);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/events"));
then.status(200)
.header("Content-Type", "application/json")
.body(r#"{"data":[],"meta":{"has_more":false}}"#);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/state"));
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"run": null,
"graph_source": null,
"start": null,
"status": {
"status": "succeeded",
"reason": null,
"updated_at": "2026-04-05T12:00:01Z"
},
"checkpoint": null,
"checkpoints": [],
"conclusion": {
"timestamp": "2026-04-05T12:00:01Z",
"status": "success",
"duration_ms": 12,
"stages": [],
"total_cost": null,
"total_retries": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
"total_cache_read_tokens": 0,
"total_cache_write_tokens": 0,
"total_reasoning_tokens": 0,
"has_pricing": false
},
"retro": null,
"retro_prompt": null,
"retro_response": null,
"sandbox": null,
"final_patch": null,
"pull_request": null,
"nodes": {}
})
.to_string(),
);
});
server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{run_id}/questions"));
then.status(200)
.header("Content-Type", "application/json")
.body(r#"{"data":[],"meta":{"has_more":false}}"#);
});
context.write_home(
".fabro/user.toml",
format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()),
);
let output = context
.command()
.args(["attach", &run_id])
.output()
.expect("attach should execute");
assert!(
output.status.success(),
"attach failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
list_mock.assert();
}
#[test]
fn attach_replays_completed_detached_run() {
let context = test_context!();

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

@ -24,13 +24,13 @@ fn help() {
<RUN> Run ID prefix or workflow name (most recent run)
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
----- stderr -----
");
}
@ -105,6 +105,23 @@ fn inspect_completed_run_shows_run_start_conclusion_checkpoint() {
"#);
}
#[test]
fn inspect_json_omits_run_dir() {
let context = test_context!();
let run = setup_completed_fast_dry_run(&context);
let output = run_success(&context, &["inspect", &run.run_id]);
let items: serde_json::Value =
serde_json::from_slice(&output.stdout).expect("inspect output should parse");
let first = items
.as_array()
.and_then(|items| items.first())
.expect("inspect output should contain one item");
assert!(
first.get("run_dir").is_none(),
"inspect JSON should not expose run_dir"
);
}
#[test]
fn inspect_completed_run_reads_store_without_disk_metadata_files() {
let context = test_context!();

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

@ -60,17 +60,17 @@ fn help() {
<RUN> Run ID prefix or workflow name (most recent run)
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-f, --follow Follow log output
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--since <SINCE> Logs since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z")
-n, --tail <TAIL> Lines from end (default: all)
--quiet Suppress non-essential output [env: FABRO_QUIET=]
-p, --pretty Formatted colored output with rendered assistant text
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-f, --follow Follow log output
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--since <SINCE> Logs since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z")
-n, --tail <TAIL> Lines from end (default: all)
--quiet Suppress non-essential output [env: FABRO_QUIET=]
-p, --pretty Formatted colored output with rendered assistant text
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
----- stderr -----
"#);
}

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

@ -1,6 +1,9 @@
use fabro_test::{fabro_snapshot, test_context};
use httpmock::MockServer;
use serde_json::Value;
use crate::support::unique_run_id;
use super::support::{fixture, setup_completed_fast_dry_run, setup_created_fast_dry_run};
#[test]
@ -17,18 +20,18 @@ fn help() {
Usage: fabro ps [OPTIONS]
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--before <BEFORE> Only include runs started before this date (YYYY-MM-DD prefix match)
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--workflow <WORKFLOW> Filter by workflow name (substring match)
--label <KEY=VALUE> Filter by label (KEY=VALUE, repeatable, AND semantics)
--orphans Include orphan directories (no run.json)
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-a, --all Show all runs, not just running (like docker ps -a)
-q, --quiet Only display run IDs
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--before <BEFORE> Only include runs started before this date (YYYY-MM-DD prefix match)
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--workflow <WORKFLOW> Filter by workflow name (substring match)
--label <KEY=VALUE> Filter by label (KEY=VALUE, repeatable, AND semantics)
--orphans Include orphan directories (no run.json)
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-a, --all Show all runs, not just running (like docker ps -a)
-q, --quiet Only display run IDs
-h, --help Print help
----- stderr -----
");
}
@ -165,3 +168,52 @@ fn ps_filters_by_workflow_and_label() {
assert_eq!(run["labels"]["fabro_test_case"], context.test_case_id());
assert_eq!(run["labels"]["fabro_test_run"], context.test_run_id());
}
#[test]
fn ps_uses_configured_server_target_without_server_flag() {
let context = test_context!();
let server = MockServer::start();
let run_id = unique_run_id();
let mock = server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!([
{
"run_id": run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote goal",
"labels": {
"suite": "remote"
},
"host_repo_path": "/srv/repo",
"start_time": "2026-04-05T12:00:00Z",
"status": "succeeded",
"status_reason": null,
"duration_ms": 123,
"total_cost": null
}
])
.to_string(),
);
});
context.write_home(
".fabro/user.toml",
format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()),
);
let output = context
.ps()
.args(["-a", "--json"])
.output()
.expect("ps should execute");
assert!(output.status.success(), "ps should succeed");
let runs: Vec<Value> = serde_json::from_slice(&output.stdout).expect("ps JSON should parse");
mock.assert();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0]["workflow_name"], "Remote Workflow");
assert_eq!(runs[0]["host_repo_path"], "/srv/repo");
}

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

@ -1,6 +1,9 @@
use fabro_test::{fabro_snapshot, test_context};
use httpmock::MockServer;
use serde_json::Value;
use crate::support::unique_run_id;
use super::support::{
setup_completed_fast_dry_run, setup_created_fast_dry_run, setup_local_sandbox_run,
};
@ -22,14 +25,14 @@ fn help() {
<RUNS>... Run IDs or workflow names to remove
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-f, --force Force removal of active runs
--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=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-f, --force Force removal of active runs
--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=]
-h, --help Print help
----- stderr -----
");
}
@ -200,3 +203,56 @@ fn rm_partial_failure_json_includes_removed_and_errors() {
"existing run should still be removed"
);
}
#[test]
fn rm_uses_configured_server_target_without_local_run_dir() {
let context = test_context!();
let run_id = unique_run_id();
let server = MockServer::start();
let list_mock = server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!([
{
"run_id": run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote goal",
"labels": {},
"host_repo_path": null,
"start_time": "2026-04-05T12:00:00Z",
"status": "succeeded",
"status_reason": null,
"duration_ms": 123,
"total_cost": null
}
])
.to_string(),
);
});
let delete_mock = server.mock(|when, then| {
when.method("DELETE").path(format!("/api/v1/runs/{run_id}"));
then.status(204);
});
context.write_home(
".fabro/user.toml",
format!("[server]\ntarget = \"{}/api/v1\"\n", server.base_url()),
);
let output = context
.command()
.args(["rm", &run_id])
.output()
.expect("rm should execute");
assert!(
output.status.success(),
"rm failed:\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
list_mock.assert();
delete_mock.assert();
}

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

@ -269,7 +269,7 @@ digraph Test {
});
fabro_json_snapshot!(context, &before_summary, @r#"
{
"run_dir": "[RUN_DIR]",
"run_dir": null,
"start_time": "[TIMESTAMP]",
"conclusion_timestamp": "[TIMESTAMP]",
"conclusion_status": "success"

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

@ -23,13 +23,13 @@ fn help() {
<RUN> Run ID prefix or workflow name
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--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=]
-h, --help Print help
----- stderr -----
");
}

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

@ -19,15 +19,15 @@ fn help() {
<RUN> Run ID prefix or workflow name (most recent run)
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--timeout <SECONDS> Maximum time to wait in seconds
--interval <MS> Poll interval in milliseconds [default: 1000]
--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=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--timeout <SECONDS> Maximum time to wait in seconds
--interval <MS> Poll interval in milliseconds [default: 1000]
--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=]
-h, --help Print help
----- stderr -----
");
}

39
lib/crates/fabro-server/src/server.rs
View file

@ -970,8 +970,35 @@ async fn delete_run(
Err(response) => return response,
};
if let Ok(mut runs) = state.runs.lock() {
runs.remove(&id);
let managed_run = if let Ok(mut runs) = state.runs.lock() {
runs.remove(&id)
} else {
None
};
if let Some(mut managed_run) = managed_run {
if let Some(token) = &managed_run.cancel_token {
token.store(true, Ordering::Relaxed);
}
if let Some(interviewer) = &managed_run.interviewer {
interviewer.abort_pending();
}
if let Some(cancel_tx) = managed_run.cancel_tx.take() {
let _ = cancel_tx.send(());
}
if let Some(run_dir) = managed_run.run_dir.take() {
if let Err(err) = remove_run_dir(&run_dir) {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
} else {
let storage_dir = state.settings.read().unwrap().storage_dir();
let run_dir = operations::make_run_dir(&storage_dir.join("runs"), &id);
if let Err(err) = remove_run_dir(&run_dir) {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
}
match state.store.delete_run(&id).await {
@ -982,6 +1009,14 @@ async fn delete_run(
}
}
fn remove_run_dir(run_dir: &std::path::Path) -> std::io::Result<()> {
match std::fs::remove_dir_all(run_dir) {
Ok(()) => Ok(()),
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(err),
}
}
fn compute_queue_positions(runs: &HashMap<RunId, ManagedRun>) -> HashMap<RunId, i64> {
let mut queued: Vec<(&RunId, &ManagedRun)> = runs
.iter()