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

feat(system): add server-backed system commands

Browse source
This commit is contained in:
Bryan Helmkamp 2026-04-06 16:09:35 -04:00
parent 8ba893bf68
commit 62f6f9b100
No known key found for this signature in database
36 changed files with 2293 additions and 427 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -1907,6 +1907,7 @@ dependencies = [
"tracing",
"ulid",
"uuid",
"walkdir",
"x509-parser",
]

285
docs/api-reference/fabro-api.yaml
View file

@ -33,6 +33,8 @@ tags:
description: Single-turn LLM completions
- name: Settings
description: Platform configuration
- name: System
description: Server runtime, maintenance, and event streaming
security:
- BearerAuth: []
@ -1649,6 +1651,91 @@ paths:
schema:
$ref: "#/components/schemas/AggregateUsage"
# ── System ───────────────────────────────────────────────────────────
/api/v1/attach:
get:
operationId: attachEvents
tags: [System]
summary: Attach Global Events
description: Opens a server-sent event stream for live run events across the server.
parameters:
- name: run_id
in: query
required: false
description: Optional comma-separated list of run IDs to include.
schema:
type: string
responses:
"200":
description: Server-sent event stream
content:
text/event-stream:
schema:
type: string
/api/v1/system/info:
get:
operationId: getSystemInfo
tags: [System]
summary: Retrieve System Info
description: Returns runtime details about the active Fabro server process.
responses:
"200":
description: System information
content:
application/json:
schema:
$ref: "#/components/schemas/SystemInfoResponse"
/api/v1/system/df:
get:
operationId: getSystemDiskUsage
tags: [System]
summary: Retrieve System Disk Usage
description: Returns disk usage for the server storage directory.
parameters:
- name: verbose
in: query
required: false
description: Include per-run disk usage rows.
schema:
type: boolean
default: false
responses:
"200":
description: Disk usage summary
content:
application/json:
schema:
$ref: "#/components/schemas/DiskUsageResponse"
/api/v1/system/prune/runs:
post:
operationId: pruneRuns
tags: [System]
summary: Prune Runs
description: Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled.
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/PruneRunsRequest"
responses:
"200":
description: Prune result
content:
application/json:
schema:
$ref: "#/components/schemas/PruneRunsResponse"
"400":
description: Invalid prune request
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
# ── Secrets ──────────────────────────────────────────────────────────
/api/v1/secrets:
@ -5808,6 +5895,204 @@ components:
type: string
description: Project fabro root directory.
SystemInfoResponse:
description: Runtime information for the active Fabro server process.
type: object
properties:
version:
type: string
description: Server version string.
git_sha:
type: string
nullable: true
description: Build git SHA when available.
build_date:
type: string
nullable: true
description: Build date when available.
os:
type: string
description: Target operating system.
arch:
type: string
description: Target CPU architecture.
storage_engine:
type: string
description: Backing run storage engine.
storage_dir:
type: string
description: Configured storage directory.
uptime_secs:
type: integer
format: int64
description: Seconds since this server process started.
runs:
$ref: "#/components/schemas/SystemRunCounts"
sandbox_provider:
type: string
description: Effective sandbox provider for launched runs.
SystemRunCounts:
description: Counts of known runs in the active server process.
type: object
properties:
total:
type: integer
format: int64
description: Total runs tracked by the server process.
active:
type: integer
format: int64
description: Runs currently queued or executing.
DiskUsageResponse:
description: Disk usage summary for server-managed data.
type: object
properties:
summary:
type: array
items:
$ref: "#/components/schemas/DiskUsageSummaryRow"
total_size_bytes:
type: integer
format: int64
description: Total size of all tracked system data.
total_reclaimable_bytes:
type: integer
format: int64
description: Total bytes reclaimable by deleting inactive runs and logs.
runs:
type: array
nullable: true
description: Per-run usage rows when verbose output is requested.
items:
$ref: "#/components/schemas/DiskUsageRunRow"
DiskUsageSummaryRow:
description: One top-level disk usage category.
type: object
properties:
type:
type: string
description: Category name, such as runs or logs.
count:
type: integer
format: int64
description: Number of items in the category.
active:
type: integer
format: int64
nullable: true
description: Number of active items when applicable.
size_bytes:
type: integer
format: int64
description: Total bytes used by the category.
reclaimable_bytes:
type: integer
format: int64
nullable: true
description: Bytes reclaimable by pruning the category.
DiskUsageRunRow:
description: Per-run disk usage information.
type: object
properties:
run_id:
type: string
description: Run identifier.
workflow_name:
type: string
description: Workflow display name.
status:
type: string
description: Current run status.
start_time:
type: string
description: Human-readable start timestamp.
size_bytes:
type: integer
format: int64
description: Size used by the run scratch directory.
reclaimable:
type: boolean
description: Whether the run is inactive and reclaimable.
PruneRunsRequest:
description: Filters for system run pruning.
type: object
properties:
dry_run:
type: boolean
description: Preview matching runs without deleting them.
default: true
before:
type: string
description: Include runs started before this YYYY-MM-DD prefix.
workflow:
type: string
description: Filter by workflow name substring.
labels:
type: object
additionalProperties:
type: string
description: Label filters applied with AND semantics.
orphans:
type: boolean
description: Include orphan run directories without run metadata.
default: false
older_than:
type: string
description: Include only runs older than this duration, such as 24h or 7d.
PruneRunsResponse:
description: Result of a prune preview or deletion.
type: object
properties:
dry_run:
type: boolean
description: Whether this response is a dry-run preview.
runs:
type: array
nullable: true
description: Matched runs when dry-run is enabled.
items:
$ref: "#/components/schemas/PruneRunEntry"
total_count:
type: integer
format: int64
description: Count of runs matching the prune filters.
total_size_bytes:
type: integer
format: int64
description: Total bytes of the matching runs.
deleted_count:
type: integer
format: int64
description: Number of runs deleted when dry-run is false.
freed_bytes:
type: integer
format: int64
description: Estimated freed bytes when deletion occurs.
PruneRunEntry:
description: One run matched by a prune preview.
type: object
properties:
run_id:
type: string
description: Run identifier.
dir_name:
type: string
description: Scratch directory name for the run.
workflow_name:
type: string
description: Workflow display name.
size_bytes:
type: integer
format: int64
description: Bytes used by the run scratch directory.
GitHubSettings:
description: GitHub App token injection configuration.
type: object

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

@ -74,6 +74,15 @@ impl ServerTargetArgs {
}
}
#[derive(Args, Debug, Clone, Default)]
pub(crate) struct ServerConnectionArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
#[command(flatten)]
pub(crate) target: ServerTargetArgs,
}
#[derive(Debug, Clone, Copy, ValueEnum)]
pub(crate) enum CliSandboxProvider {
Local,
@ -579,10 +588,16 @@ pub(crate) struct ProviderLoginArgs {
pub(crate) provider: fabro_model::Provider,
}
#[derive(Args)]
pub(crate) struct SystemInfoArgs {
#[command(flatten)]
pub(crate) connection: ServerConnectionArgs,
}
#[derive(Args)]
pub(crate) struct RunsPruneArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) connection: ServerConnectionArgs,
#[command(flatten)]
pub(crate) filter: RunFilterArgs,
@ -603,13 +618,23 @@ pub(crate) struct RunsPruneArgs {
#[derive(Args)]
pub(crate) struct DfArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) connection: ServerConnectionArgs,
/// Show per-run breakdown
#[arg(short, long)]
pub(crate) verbose: bool,
}
#[derive(Args)]
pub(crate) struct SystemEventsArgs {
#[command(flatten)]
pub(crate) connection: ServerConnectionArgs,
/// Filter by run ID (repeatable)
#[arg(long = "run-id")]
pub(crate) run_ids: Vec<String>,
}
#[derive(Args)]
pub(crate) struct SettingsArgs {
#[command(flatten)]
@ -1032,8 +1057,10 @@ impl Commands {
Self::Sandbox { command } => command.name(),
Self::Completion(_) => "completion",
Self::System(ns) => match &ns.command {
SystemCommand::Info(_) => "system info",
SystemCommand::Prune(_) => "system prune",
SystemCommand::Df(_) => "system df",
SystemCommand::Events(_) => "system events",
},
Self::SendAnalytics { .. } => "__send_analytics",
Self::SendPanic { .. } => "__send_panic",
@ -1184,10 +1211,14 @@ pub(crate) struct SystemNamespace {
#[derive(Subcommand)]
pub(crate) enum SystemCommand {
/// Show server runtime information
Info(SystemInfoArgs),
/// Delete old workflow runs
Prune(RunsPruneArgs),
/// Show disk usage
Df(DfArgs),
/// Stream run events from the server
Events(SystemEventsArgs),
}
#[derive(Args)]

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

@ -1,12 +1,7 @@
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 tracing::warn;
use crate::args::{GlobalArgs, RunsRemoveArgs};
use crate::server_client;
use crate::server_client::RunProjection;
use crate::server_runs::{
ServerRunSummaryInfo, ServerSummaryLookup, resolve_server_run_from_summaries,
};
@ -94,70 +89,6 @@ async fn remove_from(
Ok(())
}
pub(crate) async fn remove_run_with_cleanup(
client: &server_client::ServerStoreClient,
run: &RunInfo,
) -> Result<()> {
remove_run_dir_with_cleanup(client, run).await?;
delete_run_store_state(client, run).await
}
async fn remove_run_dir_with_cleanup(
client: &server_client::ServerStoreClient,
run: &RunInfo,
) -> Result<()> {
let run_id = run.run_id();
let run_state = match client.get_run_state(&run_id).await {
Ok(run_state) => Some(run_state),
Err(err) => {
warn!(
run_id = %run_id,
error = %err,
"failed to open run store during removal"
);
None
}
};
if run_state.is_some() {
let run_event = to_run_event(&run_id, &Event::RunRemoving { reason: None });
if let Err(err) = client.append_run_event(&run_id, &run_event).await {
warn!(
run_id = %run_id,
error = %err,
"failed to append removing status event"
);
}
}
if let Some(record) = load_sandbox_record(run_state.as_ref()) {
if record.provider != "local" {
match reconnect_sandbox(&record).await {
Ok(sandbox) => {
if let Err(err) = sandbox.cleanup().await {
warn!(run_id = %run_id, error = %err, "sandbox cleanup failed");
}
}
Err(err) => {
warn!(run_id = %run_id, error = %err, "sandbox reconnect failed");
}
}
}
}
std::fs::remove_dir_all(&run.path)
.with_context(|| format!("failed to delete {}", run.path.display()))
}
async fn delete_run_store_state(
client: &server_client::ServerStoreClient,
run: &RunInfo,
) -> Result<()> {
client
.delete_store_run(&run.run_id())
.await
.with_context(|| format!("failed to delete store state for {}", run.run_id()))
}
async fn delete_server_run(
client: &server_client::ServerStoreClient,
run: &ServerRunSummaryInfo,
@ -167,10 +98,3 @@ async fn delete_server_run(
.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();
}
None
}

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

@ -1,126 +1,70 @@
use std::path::Path;
use anyhow::Result;
use chrono::{DateTime, Utc};
use cli_table::format::{Border, Justify, Separator};
use cli_table::{Cell, CellStruct, Style, Table};
use serde::Serialize;
use fabro_config::Storage;
use fabro_workflow::run_lookup::{scan_runs_with_summaries, scratch_base};
use fabro_workflow::run_status::RunStatus;
use fabro_api::types;
use crate::args::{DfArgs, GlobalArgs};
use crate::server_runs::ServerRunLookup;
use crate::server_client;
use crate::shared::{format_size, print_json_pretty};
use crate::user_config::load_settings_with_storage_dir;
#[derive(Serialize)]
struct SummaryRow {
r#type: String,
count: u64,
#[serde(skip_serializing_if = "Option::is_none")]
active: Option<u64>,
size_bytes: u64,
#[serde(skip_serializing_if = "Option::is_none")]
reclaimable_bytes: Option<u64>,
}
#[derive(Serialize)]
struct RunSizeRow {
run_id: String,
workflow_name: String,
status: RunStatus,
start_time: String,
size_bytes: u64,
reclaimable: bool,
}
#[derive(Serialize)]
struct DfOutput {
summary: Vec<SummaryRow>,
total_size_bytes: u64,
total_reclaimable_bytes: u64,
#[serde(skip_serializing_if = "Option::is_none")]
runs: Option<Vec<RunSizeRow>>,
}
pub(super) async fn df_command(args: &DfArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_settings_with_storage_dir(args.storage_dir.as_deref())?;
let data_dir = cli_settings.storage_dir();
let scratch_base_dir = scratch_base(&data_dir);
let logs_base_dir = Storage::new(&data_dir).logs_dir();
let lookup = ServerRunLookup::connect(&data_dir).await?;
df_from(
args,
lookup.summaries(),
&data_dir,
&scratch_base_dir,
&logs_base_dir,
globals,
let client = server_client::connect_server_backed_api_client_with_storage_dir(
&args.connection.target,
args.connection.storage_dir.as_deref(),
)
.await?;
let output = client
.get_system_disk_usage()
.verbose(args.verbose)
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
let storage_dir = if globals.json {
None
} else {
client
.get_system_info()
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner()
.storage_dir
};
df_from(&output, storage_dir.as_deref(), globals)
}
#[allow(clippy::print_stdout)]
fn df_from(
args: &DfArgs,
summaries: &[fabro_store::RunSummary],
data_dir: &Path,
scratch_base: &Path,
logs_base: &Path,
output: &types::DiskUsageResponse,
storage_dir: Option<&str>,
globals: &GlobalArgs,
) -> Result<()> {
struct RunSizeInfo {
run_id: String,
workflow_name: String,
status: RunStatus,
start_time: String,
start_time_dt: Option<DateTime<Utc>>,
size: u64,
}
let runs_summary = output
.summary
.iter()
.find(|row| row.type_.as_deref() == Some("runs"));
let logs_summary = output
.summary
.iter()
.find(|row| row.type_.as_deref() == Some("logs"));
let runs = scan_runs_with_summaries(summaries, scratch_base)?;
let mut active_count = 0u64;
let mut total_run_size = 0u64;
let mut reclaimable_run_size = 0u64;
let run_count = runs_summary.and_then(|row| row.count).map_or(0, as_u64);
let active_count = runs_summary.and_then(|row| row.active).map_or(0, as_u64);
let total_run_size = runs_summary
.and_then(|row| row.size_bytes)
.map_or(0, as_u64);
let reclaimable_run_size = runs_summary
.and_then(|row| row.reclaimable_bytes)
.map_or(0, as_u64);
let mut run_details = Vec::new();
for run in &runs {
let size = dir_size(&run.path);
total_run_size += size;
if run.status().is_active() {
active_count += 1;
} else {
reclaimable_run_size += size;
}
if args.verbose {
run_details.push(RunSizeInfo {
run_id: run.run_id().to_string(),
workflow_name: run.workflow_name(),
status: run.status(),
start_time: run.start_time(),
start_time_dt: run.start_time_dt,
size,
});
}
}
let mut log_count = 0u64;
let mut total_log_size = 0u64;
if let Ok(entries) = std::fs::read_dir(logs_base) {
for entry in entries.flatten() {
let path = entry.path();
if !path.is_file() {
continue;
}
if path.extension().is_some_and(|ext| ext == "log") {
if let Ok(meta) = path.metadata() {
log_count += 1;
total_log_size += meta.len();
}
}
}
}
let log_count = logs_summary.and_then(|row| row.count).map_or(0, as_u64);
let total_log_size = logs_summary
.and_then(|row| row.size_bytes)
.map_or(0, as_u64);
let run_reclaim_pct = if total_run_size > 0 {
#[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)]
@ -134,41 +78,7 @@ fn df_from(
let log_reclaim_pct = if total_log_size > 0 { 100 } else { 0 };
if globals.json {
let summary = vec![
SummaryRow {
r#type: "runs".to_string(),
count: runs.len().try_into().unwrap(),
active: Some(active_count),
size_bytes: total_run_size,
reclaimable_bytes: Some(reclaimable_run_size),
},
SummaryRow {
r#type: "logs".to_string(),
count: log_count,
active: None,
size_bytes: total_log_size,
reclaimable_bytes: Some(total_log_size),
},
];
let runs = args.verbose.then(|| {
run_details
.iter()
.map(|detail| RunSizeRow {
run_id: detail.run_id.clone(),
workflow_name: detail.workflow_name.clone(),
status: detail.status,
start_time: detail.start_time.clone(),
size_bytes: detail.size,
reclaimable: !detail.status.is_active(),
})
.collect::<Vec<_>>()
});
print_json_pretty(&DfOutput {
summary,
total_size_bytes: total_run_size + total_log_size,
total_reclaimable_bytes: reclaimable_run_size + total_log_size,
runs,
})?;
print_json_pretty(output)?;
return Ok(());
}
@ -189,7 +99,7 @@ fn df_from(
let summary_rows: Vec<Vec<CellStruct>> = vec![
vec![
"Runs".cell(),
runs.len().cell().justify(Justify::Right),
run_count.cell().justify(Justify::Right),
active_count.cell().justify(Justify::Right),
format_size(total_run_size).cell().justify(Justify::Right),
format!("{} ({run_reclaim_pct}%)", format_size(reclaimable_run_size))
@ -214,13 +124,15 @@ fn df_from(
.separator(Separator::builder().build());
println!("{}", summary_table.display()?);
println!();
println!("Data directory: {}", data_dir.display());
if !args.verbose {
return Ok(());
if let Some(storage_dir) = storage_dir {
println!();
println!("Data directory: {storage_dir}");
}
let Some(run_rows) = output.runs.as_ref() else {
return Ok(());
};
println!();
let verbose_title = vec![
"RUN ID".cell().bold(use_color),
@ -231,30 +143,36 @@ fn df_from(
];
let now = Utc::now();
let verbose_rows: Vec<Vec<CellStruct>> = run_details
let verbose_rows: Vec<Vec<CellStruct>> = run_rows
.iter()
.map(|detail| {
let age = if let Some(dt) = detail.start_time_dt {
let dur = now.signed_duration_since(dt);
if dur.num_days() > 0 {
format!("{}d", dur.num_days())
} else if dur.num_hours() > 0 {
format!("{}h", dur.num_hours())
} else {
format!("{}m", dur.num_minutes().max(1))
}
let age = detail
.start_time
.as_deref()
.and_then(parse_start_time)
.map_or_else(
|| "-".to_string(),
|dt| {
let dur = now.signed_duration_since(dt);
if dur.num_days() > 0 {
format!("{}d", dur.num_days())
} else if dur.num_hours() > 0 {
format!("{}h", dur.num_hours())
} else {
format!("{}m", dur.num_minutes().max(1))
}
},
);
let size = detail.size_bytes.map_or(0, as_u64);
let size_display = if detail.reclaimable.unwrap_or(false) {
format!("{} *", format_size(size))
} else {
"-".to_string()
};
let size_display = if detail.status.is_active() {
format_size(detail.size)
} else {
format!("{} *", format_size(detail.size))
format_size(size)
};
vec![
short_run_id(&detail.run_id).cell(),
truncate_str(&detail.workflow_name, 16).cell(),
detail.status.to_string().cell(),
short_run_id(detail.run_id.as_deref().unwrap_or("-")).cell(),
truncate_str(detail.workflow_name.as_deref().unwrap_or("-"), 16).cell(),
detail.status.as_deref().unwrap_or("-").cell(),
age.cell().justify(Justify::Right),
size_display.cell().justify(Justify::Right),
]
@ -277,6 +195,16 @@ fn short_run_id(id: &str) -> &str {
if id.len() > 12 { &id[..12] } else { id }
}
fn as_u64(value: i64) -> u64 {
value.try_into().unwrap_or_default()
}
fn parse_start_time(value: &str) -> Option<DateTime<Utc>> {
chrono::DateTime::parse_from_rfc3339(value)
.ok()
.map(|dt| dt.with_timezone(&Utc))
}
fn truncate_str(s: &str, max_len: usize) -> String {
let char_count = s.chars().count();
if char_count <= max_len {
@ -285,13 +213,3 @@ fn truncate_str(s: &str, max_len: usize) -> String {
let truncated: String = s.chars().take(max_len - 3).collect();
format!("{truncated}...")
}
fn dir_size(path: &Path) -> u64 {
walkdir::WalkDir::new(path)
.into_iter()
.filter_map(std::result::Result::ok)
.filter_map(|entry| entry.metadata().ok())
.filter(std::fs::Metadata::is_file)
.map(|metadata| metadata.len())
.sum()
}

85
lib/crates/fabro-cli/src/commands/system/events.rs Normal file
View file

@ -0,0 +1,85 @@
use anyhow::Result;
use futures::StreamExt;
use crate::args::{GlobalArgs, SystemEventsArgs};
use crate::server_client;
pub(super) async fn events_command(args: &SystemEventsArgs, globals: &GlobalArgs) -> Result<()> {
let client = server_client::connect_server_backed_api_client_with_storage_dir(
&args.connection.target,
args.connection.storage_dir.as_deref(),
)
.await?;
let mut request = client.attach_events();
if !args.run_ids.is_empty() {
request = request.run_id(args.run_ids.join(","));
}
let response = request.send().await.map_err(server_client::map_api_error)?;
let mut stream = response.into_inner();
let mut pending = Vec::new();
while let Some(chunk) = stream.next().await {
let chunk = chunk.map_err(|err| anyhow::anyhow!("{err}"))?;
pending.extend_from_slice(&chunk);
drain_sse_lines(&mut pending, globals.json)?;
}
if !pending.is_empty() {
drain_sse_lines(&mut pending, globals.json)?;
}
Ok(())
}
fn drain_sse_lines(buffer: &mut Vec<u8>, json_output: bool) -> Result<()> {
while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') {
let line = buffer.drain(..=pos).collect::<Vec<_>>();
let line = String::from_utf8_lossy(&line);
let line = line.trim_end_matches(['\r', '\n']);
if let Some(data) = line.strip_prefix("data:") {
render_sse_payload(data.trim(), json_output)?;
}
}
Ok(())
}
fn render_sse_payload(data: &str, json_output: bool) -> Result<()> {
if json_output {
#[allow(clippy::print_stdout)]
{
println!("{data}");
}
return Ok(());
}
let value: serde_json::Value = serde_json::from_str(data)?;
let payload = value
.get("payload")
.and_then(serde_json::Value::as_object)
.cloned()
.unwrap_or_default();
let ts = payload
.get("ts")
.and_then(serde_json::Value::as_str)
.unwrap_or("-");
let run_id = payload
.get("run_id")
.and_then(serde_json::Value::as_str)
.unwrap_or("-");
let event = payload
.get("event")
.and_then(serde_json::Value::as_str)
.unwrap_or("-");
#[allow(clippy::print_stdout)]
{
println!("{ts} {} {event}", short_run_id(run_id));
}
Ok(())
}
fn short_run_id(run_id: &str) -> &str {
run_id.get(..12).unwrap_or(run_id)
}

70
lib/crates/fabro-cli/src/commands/system/info.rs Normal file
View file

@ -0,0 +1,70 @@
use anyhow::Result;
use crate::args::{GlobalArgs, SystemInfoArgs};
use crate::server_client;
use crate::shared::print_json_pretty;
pub(super) async fn info_command(args: &SystemInfoArgs, globals: &GlobalArgs) -> Result<()> {
let client = server_client::connect_server_backed_api_client_with_storage_dir(
&args.connection.target,
args.connection.storage_dir.as_deref(),
)
.await?;
let response = client
.get_system_info()
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
if globals.json {
print_json_pretty(&response)?;
return Ok(());
}
#[allow(clippy::print_stdout)]
{
println!(
"Version: {}",
response
.version
.as_deref()
.unwrap_or(env!("CARGO_PKG_VERSION"))
);
println!(
"Build: {} {}",
response.git_sha.as_deref().unwrap_or("unknown"),
response.build_date.as_deref().unwrap_or("unknown")
);
println!(
"Platform: {}/{}",
response.os.as_deref().unwrap_or("unknown"),
response.arch.as_deref().unwrap_or("unknown")
);
println!(
"Storage: {} ({})",
response.storage_dir.as_deref().unwrap_or("unknown"),
response.storage_engine.as_deref().unwrap_or("unknown")
);
println!(
"Runs: total={} active={}",
response
.runs
.as_ref()
.and_then(|runs| runs.total)
.unwrap_or_default(),
response
.runs
.as_ref()
.and_then(|runs| runs.active)
.unwrap_or_default()
);
println!(
"Sandbox: {}",
response.sandbox_provider.as_deref().unwrap_or("unknown")
);
println!("Uptime: {}s", response.uptime_secs.unwrap_or_default());
}
Ok(())
}

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

@ -1,4 +1,6 @@
mod df;
mod events;
mod info;
mod prune;
use anyhow::Result;
@ -9,7 +11,9 @@ pub(crate) use prune::parse_duration;
pub(crate) async fn dispatch(ns: SystemNamespace, globals: &GlobalArgs) -> Result<()> {
match ns.command {
SystemCommand::Info(args) => info::info_command(&args, globals).await,
SystemCommand::Prune(args) => prune::prune_command(&args, globals).await,
SystemCommand::Df(args) => df::df_command(&args, globals).await,
SystemCommand::Events(args) => events::events_command(&args, globals).await,
}
}

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

@ -1,34 +1,35 @@
use std::path::Path;
use std::collections::HashMap;
use anyhow::{Context, Result, bail};
use chrono::Utc;
use serde::Serialize;
use tracing::{debug, info};
use fabro_workflow::run_lookup::{
StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base,
};
use fabro_api::types;
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_settings_with_storage_dir;
#[derive(Serialize)]
struct PruneRunRow {
run_id: String,
dir_name: String,
workflow_name: String,
size_bytes: u64,
}
pub(super) async fn prune_command(args: &RunsPruneArgs, globals: &GlobalArgs) -> Result<()> {
let cli_settings = load_settings_with_storage_dir(args.storage_dir.as_deref())?;
let base = scratch_base(&cli_settings.storage_dir());
let lookup = ServerRunLookup::connect(&cli_settings.storage_dir()).await?;
prune_from(args, lookup.client(), lookup.summaries(), &base, globals).await
let client = server_client::connect_server_backed_api_client_with_storage_dir(
&args.connection.target,
args.connection.storage_dir.as_deref(),
)
.await?;
let response = client
.prune_runs()
.body(types::PruneRunsRequest {
before: args.filter.before.clone(),
dry_run: !args.yes,
labels: parse_label_filters(&args.filter.label),
older_than: args.older_than.map(format_duration),
orphans: args.filter.orphans,
workflow: args.filter.workflow.clone(),
})
.send()
.await
.map_err(server_client::map_api_error)?
.into_inner();
prune_from(&response, globals)
}
pub(crate) fn parse_duration(s: &str) -> Result<chrono::Duration> {
@ -47,123 +48,56 @@ pub(crate) fn parse_duration(s: &str) -> Result<chrono::Duration> {
}
}
async fn prune_from(
args: &RunsPruneArgs,
client: &server_client::ServerStoreClient,
summaries: &[fabro_store::RunSummary],
base: &Path,
globals: &GlobalArgs,
) -> Result<()> {
let runs = scan_runs_with_summaries(summaries, base)?;
let label_filters = parse_label_filters(&args.filter.label);
let mut filtered = filter_runs(
&runs,
args.filter.before.as_deref(),
args.filter.workflow.as_deref(),
&label_filters,
args.filter.orphans,
StatusFilter::All,
fn prune_from(response: &types::PruneRunsResponse, globals: &GlobalArgs) -> Result<()> {
let total_count = response.total_count.unwrap_or_default();
let total_size_bytes = response.total_size_bytes.unwrap_or_default();
info!(
count = total_count,
bytes = total_size_bytes,
dry_run = response.dry_run.unwrap_or(true),
"pruning runs"
);
let has_explicit_filters =
args.filter.before.is_some() || args.filter.workflow.is_some() || !label_filters.is_empty();
let staleness_threshold = if let Some(duration) = args.older_than {
Some(duration)
} else if !has_explicit_filters {
Some(chrono::Duration::hours(24))
} else {
None
};
if let Some(threshold) = staleness_threshold {
let cutoff = Utc::now() - threshold;
filtered.retain(|run| {
run.end_time
.or(run.start_time_dt)
.is_some_and(|time| time < cutoff)
});
}
filtered.retain(|run| !run.status().is_active());
if filtered.is_empty() {
if globals.json {
if args.yes {
print_json_pretty(&serde_json::json!({
"dry_run": false,
"deleted_count": 0,
"freed_bytes": 0,
}))?;
} else {
print_json_pretty(&serde_json::json!({
"dry_run": true,
"runs": Vec::<PruneRunRow>::new(),
"total_count": 0,
"total_size_bytes": 0,
}))?;
}
} else {
eprintln!("No matching runs to prune.");
}
if globals.json {
print_json_pretty(response)?;
return Ok(());
}
let rows: Vec<PruneRunRow> = filtered
.iter()
.map(|run| PruneRunRow {
run_id: run.run_id().to_string(),
dir_name: run.dir_name.clone(),
workflow_name: run.workflow_name(),
size_bytes: dir_size(&run.path),
})
.collect();
let total_bytes: u64 = rows.iter().map(|row| row.size_bytes).sum();
info!(count = filtered.len(), bytes = total_bytes, "pruning runs");
if total_count == 0 {
eprintln!("No matching runs to prune.");
return Ok(());
}
if args.yes {
for run in &filtered {
info!(run_id = %run.run_id(), path = %run.path.display(), "deleting run");
remove_run_with_cleanup(client, run).await?;
}
if globals.json {
print_json_pretty(&serde_json::json!({
"dry_run": false,
"deleted_count": filtered.len(),
"freed_bytes": total_bytes,
}))?;
} else {
eprintln!(
"{} run(s) deleted ({} freed).",
filtered.len(),
format_size(total_bytes)
if response.dry_run.unwrap_or(true) {
for run in response.runs.as_deref().unwrap_or(&[]) {
debug!(
run_id = run.run_id.as_deref().unwrap_or("-"),
"would delete run (dry-run)"
);
println!(
"would delete: {} ({})",
run.dir_name.as_deref().unwrap_or("-"),
run.workflow_name.as_deref().unwrap_or("-")
);
}
eprintln!(
"\n{} run(s) would be deleted ({} freed). Pass --yes to confirm.",
total_count,
format_size(as_u64(total_size_bytes))
);
return Ok(());
}
if globals.json {
print_json_pretty(&serde_json::json!({
"dry_run": true,
"runs": rows,
"total_count": filtered.len(),
"total_size_bytes": total_bytes,
}))?;
return Ok(());
}
for run in &filtered {
debug!(run_id = %run.run_id(), "would delete run (dry-run)");
println!("would delete: {} ({})", run.dir_name, run.workflow_name());
}
eprintln!(
"\n{} run(s) would be deleted ({} freed). Pass --yes to confirm.",
filtered.len(),
format_size(total_bytes)
"{} run(s) deleted ({} freed).",
response.deleted_count.unwrap_or(total_count),
format_size(as_u64(response.freed_bytes.unwrap_or(total_size_bytes)))
);
Ok(())
}
fn parse_label_filters(label_args: &[String]) -> Vec<(String, String)> {
fn parse_label_filters(label_args: &[String]) -> HashMap<String, String> {
label_args
.iter()
.filter_map(|s| s.split_once('='))
@ -171,12 +105,14 @@ fn parse_label_filters(label_args: &[String]) -> Vec<(String, String)> {
.collect()
}
fn dir_size(path: &Path) -> u64 {
walkdir::WalkDir::new(path)
.into_iter()
.filter_map(std::result::Result::ok)
.filter_map(|entry| entry.metadata().ok())
.filter(std::fs::Metadata::is_file)
.map(|metadata| metadata.len())
.sum()
fn format_duration(duration: chrono::Duration) -> String {
if duration.num_hours() % 24 == 0 {
format!("{}d", duration.num_days())
} else {
format!("{}h", duration.num_hours())
}
}
fn as_u64(value: i64) -> u64 {
value.try_into().unwrap_or_default()
}

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

@ -164,7 +164,14 @@ async fn connect_target_api_client(
pub(crate) async fn connect_server_backed_api_client(
args: &ServerTargetArgs,
) -> Result<fabro_api::Client> {
let settings = user_config::load_settings()?;
connect_server_backed_api_client_with_storage_dir(args, None).await
}
pub(crate) async fn connect_server_backed_api_client_with_storage_dir(
args: &ServerTargetArgs,
storage_dir: Option<&Path>,
) -> Result<fabro_api::Client> {
let settings = user_config::load_settings_with_storage_dir(storage_dir)?;
let target = user_config::resolve_server_target(args, &settings)?;
let runtime = LocalServerRuntime {
active_config_path: user_config::active_settings_path(None),
@ -552,7 +559,7 @@ impl ServerStoreClient {
}
}
fn map_api_error<E>(err: progenitor_client::Error<E>) -> anyhow::Error
pub(crate) fn map_api_error<E>(err: progenitor_client::Error<E>) -> anyhow::Error
where
E: serde::Serialize + std::fmt::Debug,
{

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

@ -37,10 +37,6 @@ impl ServerRunLookup {
&self.client
}
pub(crate) fn summaries(&self) -> &[RunSummary] {
&self.summaries
}
pub(crate) fn resolve(&self, selector: &str) -> Result<RunInfo> {
resolve_run_from_summaries(&self.summaries, &self.scratch_base, selector)
}

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

@ -57,6 +57,8 @@ mod store_dump;
pub(crate) mod support;
mod system;
mod system_df;
mod system_events;
mod system_info;
mod system_prune;
mod test_panic;
mod top_level;

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

@ -358,7 +358,7 @@ pub(crate) fn setup_artifact_run(context: &TestContext) -> WorkspaceRunSetup {
start [shape=Mdiamond]
exit [shape=Msquare]
create_assets [shape=parallelogram, script="mkdir -p assets/shared assets/node_a && printf one > assets/shared/report.txt && printf alpha > assets/node_a/summary.txt", max_retries=0]
retry_assets [shape=parallelogram, script="mkdir -p assets/retry && touch -c -t 200001010000 assets/shared/report.txt assets/node_a/summary.txt && if [ ! -f .retry-sentinel ]; then printf first > assets/retry/report.txt && touch .retry-sentinel && sleep 0.2; else printf second > assets/retry/report.txt; fi", retry_policy="linear", timeout="150ms"]
retry_assets [shape=parallelogram, script="mkdir -p assets/retry && touch -c -t 200001010000 assets/shared/report.txt assets/node_a/summary.txt && if [ ! -f .retry-sentinel ]; then printf first > assets/retry/report.txt && touch .retry-sentinel && sleep 1; else printf second > assets/retry/report.txt; fi", retry_policy="linear", timeout="500ms"]
create_colliding [shape=parallelogram, script="mkdir -p assets/other assets/retry && touch -c -t 200001010000 assets/shared/report.txt assets/node_a/summary.txt assets/retry/report.txt && printf beta > assets/other/summary.txt && printf second > assets/retry/report.txt", max_retries=0]
start -> create_assets -> retry_assets -> create_colliding -> exit
}

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

@ -14,9 +14,11 @@ fn help() {
Usage: fabro system [OPTIONS] <COMMAND>
Commands:
prune Delete old workflow runs
df Show disk usage
help Print this message or the help of the given subcommand(s)
info Show server runtime information
prune Delete old workflow runs
df Show disk usage
events Stream run events from the server
help Print this message or the help of the given subcommand(s)
Options:
--json Output as JSON [env: FABRO_JSON=]

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

@ -20,8 +20,9 @@ fn help() {
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-v, --verbose Show per-run breakdown
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
-v, --verbose Show per-run breakdown
--quiet Suppress non-essential output [env: FABRO_QUIET=]
-h, --help Print help
----- stderr -----

28
lib/crates/fabro-cli/tests/it/cmd/system_events.rs Normal file
View file

@ -0,0 +1,28 @@
use fabro_test::{fabro_snapshot, test_context};
#[test]
fn help() {
let context = test_context!();
let mut cmd = context.command();
cmd.args(["system", "events", "--help"]);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
Stream run events from the server
Usage: fabro system events [OPTIONS]
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--run-id <RUN_IDS> Filter by run ID (repeatable)
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
----- stderr -----
");
}

50
lib/crates/fabro-cli/tests/it/cmd/system_info.rs Normal file
View file

@ -0,0 +1,50 @@
use fabro_test::{fabro_snapshot, test_context};
use serde_json::Value;
#[test]
fn help() {
let context = test_context!();
let mut cmd = context.command();
cmd.args(["system", "info", "--help"]);
fabro_snapshot!(context.filters(), cmd, @"
success: true
exit_code: 0
----- stdout -----
Show server runtime information
Usage: fabro system info [OPTIONS]
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--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 -----
");
}
#[test]
fn system_info_json_reports_runtime_fields() {
let context = test_context!();
let output = context
.command()
.args(["--json", "system", "info"])
.output()
.expect("command should run");
assert!(output.status.success(), "system info failed");
let value: Value =
serde_json::from_slice(&output.stdout).expect("system info JSON should parse");
assert!(value["version"].is_string());
assert_eq!(
value["storage_dir"],
context.storage_dir.display().to_string()
);
assert!(value["uptime_secs"].is_number());
assert!(value["runs"]["total"].is_number());
}

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

@ -18,14 +18,15 @@ fn help() {
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [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=]
--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)
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--workflow <WORKFLOW> Filter by workflow name (substring match)
--label <KEY=VALUE> Filter by label (KEY=VALUE, repeatable, AND semantics)
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--orphans Include orphan directories (no run.json)
--verbose Enable verbose output [env: FABRO_VERBOSE=]
--orphans Include orphan directories (no run.json)
--older-than <DURATION> Only prune runs older than this duration (e.g. 24h, 7d). Default: 24h when no explicit filters are set
--yes Actually delete (default is dry-run)
-h, --help Print help

1
lib/crates/fabro-server/Cargo.toml
View file

@ -69,6 +69,7 @@ mime_guess.workspace = true
rust-embed.workspace = true
regex.workspace = true
semver.workspace = true
walkdir.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }

131
lib/crates/fabro-server/src/demo/mod.rs
View file

@ -7,6 +7,7 @@ use std::sync::Arc;
use axum::Json;
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::response::sse::{Event, Sse};
use axum::response::{IntoResponse, Response};
use fabro_api::types::{
RunArtifactListResponse, RunStatus, RunStatusResponse, SessionTurn, SmoothnessRating,
@ -741,6 +742,136 @@ pub(crate) async fn get_server_settings(
(StatusCode::OK, Json(settings::server_settings())).into_response()
}
// ── System ────────────────────────────────────────────────────────────
pub(crate) async fn attach_events_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
) -> Response {
let events = vec![
Ok::<_, std::convert::Infallible>(
Event::default().data(
json!({
"seq": 1,
"payload": {
"id": "evt_demo_1",
"ts": "2026-04-06T15:00:00Z",
"run_id": "01JQ0000000000000000000001",
"event": "run.started"
}
})
.to_string(),
),
),
Ok::<_, std::convert::Infallible>(
Event::default().data(
json!({
"seq": 2,
"payload": {
"id": "evt_demo_2",
"ts": "2026-04-06T15:00:01Z",
"run_id": "01JQ0000000000000000000001",
"event": "stage.started"
}
})
.to_string(),
),
),
];
Sse::new(tokio_stream::iter(events)).into_response()
}
pub(crate) async fn get_system_info(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
) -> Response {
(
StatusCode::OK,
Json(json!({
"version": env!("CARGO_PKG_VERSION"),
"git_sha": option_env!("FABRO_GIT_SHA"),
"build_date": option_env!("FABRO_BUILD_DATE"),
"os": std::env::consts::OS,
"arch": std::env::consts::ARCH,
"storage_engine": "slatedb",
"storage_dir": "/demo/fabro/storage",
"uptime_secs": 42,
"runs": { "total": 3, "active": 1 },
"sandbox_provider": "local"
})),
)
.into_response()
}
pub(crate) async fn get_system_disk_usage(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
Query(params): Query<crate::server::DfParams>,
) -> Response {
let runs = params.verbose.then(|| {
json!([
{
"run_id": "01JQ0000000000000000000001",
"workflow_name": "Demo Workflow",
"status": "succeeded",
"start_time": "2026-04-06T15:00:00Z",
"size_bytes": 1024,
"reclaimable": true
}
])
});
(
StatusCode::OK,
Json(json!({
"summary": [
{
"type": "runs",
"count": 1,
"active": 0,
"size_bytes": 1024,
"reclaimable_bytes": 1024
},
{
"type": "logs",
"count": 1,
"active": null,
"size_bytes": 256,
"reclaimable_bytes": 256
}
],
"total_size_bytes": 1280,
"total_reclaimable_bytes": 1280,
"runs": runs
})),
)
.into_response()
}
pub(crate) async fn prune_runs(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
) -> Response {
(
StatusCode::OK,
Json(json!({
"dry_run": true,
"runs": [
{
"run_id": "01JQ0000000000000000000001",
"dir_name": "20260406-01JQ0000000000000000000001",
"workflow_name": "Demo Workflow",
"size_bytes": 1024
}
],
"total_count": 1,
"total_size_bytes": 1024,
"deleted_count": 0,
"freed_bytes": 0
})),
)
.into_response()
}
// ── Usage ──────────────────────────────────────────────────────────────
pub(crate) async fn get_aggregate_usage(

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

@ -1,4 +1,4 @@
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::path::{Component, PathBuf};
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, Ordering};
@ -40,10 +40,12 @@ use tokio::fs;
use tokio::sync::Notify;
use tokio::sync::RwLock as AsyncRwLock;
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::oneshot;
use tokio::task::spawn_blocking;
use tokio::time::sleep;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::BroadcastStream;
use tower::{ServiceExt, service_fn};
use ulid::Ulid;
@ -67,6 +69,9 @@ use fabro_workflow::event::{self as workflow_event, Emitter};
use fabro_workflow::operations::{self};
use fabro_workflow::pipeline::Persisted;
use fabro_workflow::records::Checkpoint;
use fabro_workflow::run_lookup::{
RunInfo, StatusFilter, filter_runs, scan_runs_with_summaries, scratch_base,
};
use fabro_workflow::run_status::RunStatus as WorkflowRunStatus;
use fabro_workflow::run_status::StatusReason as WorkflowStatusReason;
@ -75,13 +80,15 @@ pub use fabro_api::types::{
AggregateUsage, ApiQuestion, ApiQuestionOption, AppendEventResponse, ArtifactEntry,
ArtifactListResponse, CompletionContentPart, CompletionMessage, CompletionMessageRole,
CompletionResponse, CompletionToolChoiceMode, CompletionUsage, CreateCompletionRequest,
EventEnvelope as ApiEventEnvelope, ModelReference, PaginatedEventList, PaginatedRunList,
PaginationMeta, PreflightResponse, PreviewUrlRequest, PreviewUrlResponse,
DiskUsageResponse, DiskUsageRunRow, DiskUsageSummaryRow, EventEnvelope as ApiEventEnvelope,
ModelReference, PaginatedEventList, PaginatedRunList, PaginationMeta, PreflightResponse,
PreviewUrlRequest, PreviewUrlResponse, PruneRunEntry, PruneRunsRequest, PruneRunsResponse,
QuestionType as ApiQuestionType, RenderWorkflowGraphDirection, RenderWorkflowGraphFormat,
RenderWorkflowGraphRequest, RunArtifactEntry, RunArtifactListResponse, RunError,
RunEvent as ApiRunEvent, RunManifest, RunStatus, RunStatusResponse, SandboxFileEntry,
SandboxFileListResponse, ServerSettings, SetSecretRequest, SshAccessRequest, SshAccessResponse,
StartRunRequest, SubmitAnswerRequest, TokenUsage, UsageByModel, WriteBlobResponse,
StartRunRequest, SubmitAnswerRequest, SystemInfoResponse, SystemRunCounts, TokenUsage,
UsageByModel, WriteBlobResponse,
};
use fabro_graphviz::render::GraphFormat;
@ -139,6 +146,18 @@ struct AttachParams {
since_seq: Option<u32>,
}
#[derive(serde::Deserialize)]
pub(crate) struct DfParams {
#[serde(default)]
pub(crate) verbose: bool,
}
#[derive(serde::Deserialize)]
struct GlobalAttachParams {
#[serde(default)]
run_id: Option<String>,
}
#[derive(serde::Deserialize)]
struct ArtifactFilenameParams {
#[serde(default)]
@ -226,8 +245,10 @@ pub struct AppState {
aggregate_usage: Mutex<UsageAccumulator>,
store: Arc<Database>,
artifact_store: ArtifactStore,
started_at: Instant,
max_concurrent_runs: usize,
scheduler_notify: Notify,
global_event_tx: broadcast::Sender<EventEnvelope>,
pub sessions: SessionStore,
pub(crate) secret_store: AsyncRwLock<SecretStore>,
pub(crate) settings: Arc<RwLock<Settings>>,
@ -365,6 +386,7 @@ fn demo_routes() -> Router<Arc<AppState>> {
.route("/runs", get(demo::list_runs).post(demo::create_run_stub))
.route("/preflight", post(run_preflight))
.route("/graph/render", post(render_graph_from_manifest))
.route("/attach", get(demo::attach_events_stub))
.route("/runs/{id}", get(demo::get_run_status))
.route("/runs/{id}/questions", get(demo::get_questions_stub))
.route("/runs/{id}/questions/{qid}/answer", post(demo::answer_stub))
@ -466,6 +488,9 @@ fn demo_routes() -> Router<Arc<AppState>> {
.route("/health/diagnostics", post(demo::run_diagnostics))
.route("/completions", post(create_completion))
.route("/settings", get(demo::get_server_settings))
.route("/system/info", get(demo::get_system_info))
.route("/system/df", get(demo::get_system_disk_usage))
.route("/system/prune/runs", post(demo::prune_runs))
.route("/usage", get(demo::get_aggregate_usage))
}
@ -474,6 +499,7 @@ fn real_routes() -> Router<Arc<AppState>> {
.route("/runs", get(list_runs).post(create_run))
.route("/preflight", post(run_preflight))
.route("/graph/render", post(render_graph_from_manifest))
.route("/attach", get(attach_events))
.route("/boards/runs", get(list_board_runs))
.route("/runs/{id}", get(get_run_status).delete(delete_run))
.route("/runs/{id}/questions", get(get_questions))
@ -558,6 +584,9 @@ fn real_routes() -> Router<Arc<AppState>> {
.route("/health/diagnostics", post(run_diagnostics))
.route("/completions", post(create_completion))
.route("/settings", get(get_server_settings))
.route("/system/info", get(get_system_info))
.route("/system/df", get(get_system_df))
.route("/system/prune/runs", post(prune_runs))
.route("/usage", get(get_aggregate_usage))
}
@ -611,6 +640,412 @@ fn strip_nulls(value: &mut serde_json::Value) {
}
}
async fn get_system_info(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
) -> Response {
let settings = state.settings.read().unwrap().clone();
let (total_runs, active_runs) = {
let runs = state.runs.lock().expect("runs lock poisoned");
let active = runs
.values()
.filter(|run| {
matches!(
run.status,
RunStatus::Queued
| RunStatus::Starting
| RunStatus::Running
| RunStatus::Paused
)
})
.count();
(runs.len(), active)
};
let response = SystemInfoResponse {
version: Some(FABRO_VERSION.to_string()),
git_sha: option_env!("FABRO_GIT_SHA").map(str::to_string),
build_date: option_env!("FABRO_BUILD_DATE").map(str::to_string),
os: Some(std::env::consts::OS.to_string()),
arch: Some(std::env::consts::ARCH.to_string()),
storage_engine: Some("slatedb".to_string()),
storage_dir: Some(settings.storage_dir().display().to_string()),
uptime_secs: Some(to_i64(state.started_at.elapsed().as_secs())),
runs: Some(SystemRunCounts {
total: Some(to_i64(total_runs)),
active: Some(to_i64(active_runs)),
}),
sandbox_provider: Some(system_sandbox_provider(&settings)),
};
(StatusCode::OK, Json(response)).into_response()
}
async fn get_system_df(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Query(params): Query<DfParams>,
) -> Response {
let storage_dir = state.settings.read().unwrap().storage_dir();
let summaries = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(summaries) => summaries,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let response = match spawn_blocking(move || {
build_disk_usage_response(&summaries, &storage_dir, params.verbose)
})
.await
{
Ok(Ok(response)) => response,
Ok(Err(err)) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
(StatusCode::OK, Json(response)).into_response()
}
async fn prune_runs(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Json(body): Json<PruneRunsRequest>,
) -> Response {
let storage_dir = state.settings.read().unwrap().storage_dir();
let summaries = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(summaries) => summaries,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
let dry_run = body.dry_run;
let body_for_plan = body.clone();
let prune_plan =
match spawn_blocking(move || build_prune_plan(&body_for_plan, &summaries, &storage_dir))
.await
{
Ok(Ok(plan)) => plan,
Ok(Err(err)) => {
return ApiError::new(StatusCode::BAD_REQUEST, err.to_string()).into_response();
}
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
if dry_run {
return (
StatusCode::OK,
Json(PruneRunsResponse {
dry_run: Some(true),
runs: Some(prune_plan.rows),
total_count: Some(to_i64(prune_plan.run_ids.len())),
total_size_bytes: Some(to_i64(prune_plan.total_size_bytes)),
deleted_count: Some(0),
freed_bytes: Some(0),
}),
)
.into_response();
}
for run_id in &prune_plan.run_ids {
if let Err(response) = delete_run_internal(&state, *run_id).await {
return response;
}
}
(
StatusCode::OK,
Json(PruneRunsResponse {
dry_run: Some(false),
runs: None,
total_count: Some(to_i64(prune_plan.run_ids.len())),
total_size_bytes: Some(to_i64(prune_plan.total_size_bytes)),
deleted_count: Some(to_i64(prune_plan.run_ids.len())),
freed_bytes: Some(to_i64(prune_plan.total_size_bytes)),
}),
)
.into_response()
}
async fn attach_events(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Query(params): Query<GlobalAttachParams>,
) -> Response {
let run_filter = match parse_global_run_filter(params.run_id.as_deref()) {
Ok(filter) => filter,
Err(err) => return ApiError::new(StatusCode::BAD_REQUEST, err).into_response(),
};
let stream =
BroadcastStream::new(state.global_event_tx.subscribe()).filter_map(move |result| {
match result {
Ok(event) => {
if !event_matches_run_filter(&event, run_filter.as_ref()) {
return None;
}
sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>)
}
Err(_) => None,
}
});
Sse::new(stream)
.keep_alive(KeepAlive::default())
.into_response()
}
struct PrunePlan {
run_ids: Vec<RunId>,
rows: Vec<PruneRunEntry>,
total_size_bytes: u64,
}
fn build_disk_usage_response(
summaries: &[fabro_store::RunSummary],
storage_dir: &std::path::Path,
verbose: bool,
) -> anyhow::Result<DiskUsageResponse> {
let scratch_base_dir = scratch_base(storage_dir);
let logs_base_dir = Storage::new(storage_dir).logs_dir();
let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;
let mut active_count = 0u64;
let mut total_run_size = 0u64;
let mut reclaimable_run_size = 0u64;
let mut run_rows = Vec::new();
for run in &runs {
let size = dir_size(&run.path);
total_run_size += size;
if run.status().is_active() {
active_count += 1;
} else {
reclaimable_run_size += size;
}
if verbose {
run_rows.push(DiskUsageRunRow {
run_id: Some(run.run_id().to_string()),
workflow_name: Some(run.workflow_name()),
status: Some(run.status().to_string()),
start_time: Some(run.start_time()),
size_bytes: Some(to_i64(size)),
reclaimable: Some(!run.status().is_active()),
});
}
}
let mut log_count = 0u64;
let mut total_log_size = 0u64;
if let Ok(entries) = std::fs::read_dir(logs_base_dir) {
for entry in entries.flatten() {
let path = entry.path();
if !path.is_file() || path.extension().is_none_or(|ext| ext != "log") {
continue;
}
if let Ok(metadata) = path.metadata() {
log_count += 1;
total_log_size += metadata.len();
}
}
}
Ok(DiskUsageResponse {
summary: vec![
DiskUsageSummaryRow {
type_: Some("runs".to_string()),
count: Some(to_i64(runs.len())),
active: Some(to_i64(active_count)),
size_bytes: Some(to_i64(total_run_size)),
reclaimable_bytes: Some(to_i64(reclaimable_run_size)),
},
DiskUsageSummaryRow {
type_: Some("logs".to_string()),
count: Some(to_i64(log_count)),
active: None,
size_bytes: Some(to_i64(total_log_size)),
reclaimable_bytes: Some(to_i64(total_log_size)),
},
],
total_size_bytes: Some(to_i64(total_run_size + total_log_size)),
total_reclaimable_bytes: Some(to_i64(reclaimable_run_size + total_log_size)),
runs: verbose.then_some(run_rows),
})
}
fn build_prune_plan(
request: &PruneRunsRequest,
summaries: &[fabro_store::RunSummary],
storage_dir: &std::path::Path,
) -> anyhow::Result<PrunePlan> {
let scratch_base_dir = scratch_base(storage_dir);
let runs = scan_runs_with_summaries(summaries, &scratch_base_dir)?;
let label_filters = request
.labels
.iter()
.map(|(key, value)| (key.clone(), value.clone()))
.collect::<Vec<_>>();
let mut filtered = filter_runs(
&runs,
request.before.as_deref(),
request.workflow.as_deref(),
&label_filters,
request.orphans,
StatusFilter::All,
);
let has_explicit_filters =
request.before.is_some() || request.workflow.is_some() || !label_filters.is_empty();
let staleness_threshold = if let Some(duration) = request.older_than.as_deref() {
Some(parse_system_duration(duration)?)
} else if !has_explicit_filters {
Some(chrono::Duration::hours(24))
} else {
None
};
if let Some(threshold) = staleness_threshold {
let cutoff = chrono::Utc::now() - threshold;
filtered.retain(|run| {
run.end_time
.or(run.start_time_dt)
.is_some_and(|time| time < cutoff)
});
}
filtered.retain(|run| !run.status().is_active());
let rows = filtered
.iter()
.map(|run| PruneRunEntry {
run_id: Some(run.run_id().to_string()),
dir_name: Some(run.dir_name.clone()),
workflow_name: Some(run.workflow_name()),
size_bytes: Some(to_i64(dir_size(&run.path))),
})
.collect::<Vec<_>>();
let total_size_bytes = rows
.iter()
.map(|row| row.size_bytes.unwrap_or_default())
.sum::<i64>()
.max(0)
.try_into()
.unwrap_or_default();
Ok(PrunePlan {
run_ids: filtered.iter().map(RunInfo::run_id).collect(),
rows,
total_size_bytes,
})
}
fn system_sandbox_provider(settings: &Settings) -> String {
settings
.sandbox_settings()
.and_then(|sandbox| sandbox.provider.clone())
.unwrap_or_else(|| SandboxProvider::default().to_string())
}
fn parse_system_duration(raw: &str) -> anyhow::Result<chrono::Duration> {
let raw = raw.trim();
anyhow::ensure!(!raw.is_empty(), "empty duration string");
let (num_str, unit) = raw.split_at(raw.len().saturating_sub(1));
let amount = num_str.parse::<u64>()?;
match unit {
"h" => Ok(chrono::Duration::hours(
i64::try_from(amount).unwrap_or(i64::MAX),
)),
"d" => Ok(chrono::Duration::days(
i64::try_from(amount).unwrap_or(i64::MAX),
)),
_ => anyhow::bail!("invalid duration unit '{unit}' in '{raw}' (expected 'h' or 'd')"),
}
}
fn parse_global_run_filter(raw: Option<&str>) -> Result<Option<HashSet<RunId>>, String> {
let Some(raw) = raw else {
return Ok(None);
};
let mut run_ids = HashSet::new();
for part in raw
.split(',')
.map(str::trim)
.filter(|part| !part.is_empty())
{
let run_id = part
.parse::<RunId>()
.map_err(|err| format!("invalid run_id '{part}': {err}"))?;
run_ids.insert(run_id);
}
if run_ids.is_empty() {
Ok(None)
} else {
Ok(Some(run_ids))
}
}
fn event_matches_run_filter(event: &EventEnvelope, run_filter: Option<&HashSet<RunId>>) -> bool {
let Some(run_filter) = run_filter else {
return true;
};
let Some(run_id) = event
.payload
.as_value()
.get("run_id")
.and_then(serde_json::Value::as_str)
.and_then(|value| value.parse::<RunId>().ok())
else {
return false;
};
run_filter.contains(&run_id)
}
fn sse_event_from_store(event: &EventEnvelope) -> Option<Event> {
let event = api_event_envelope_from_store(event).ok()?;
let data = serde_json::to_string(&event).ok()?;
let data = redact_jsonl_line(&data);
Some(Event::default().data(data))
}
fn dir_size(path: &std::path::Path) -> u64 {
walkdir::WalkDir::new(path)
.into_iter()
.filter_map(std::result::Result::ok)
.filter_map(|entry| entry.metadata().ok())
.filter(std::fs::Metadata::is_file)
.map(|metadata| metadata.len())
.sum()
}
fn to_i64<T>(value: T) -> i64
where
i64: TryFrom<T>,
{
i64::try_from(value).unwrap_or(i64::MAX)
}
async fn list_secrets(_auth: AuthenticatedService, State(state): State<Arc<AppState>>) -> Response {
let data = state.secret_store.read().await.list();
(StatusCode::OK, Json(serde_json::json!({ "data": data }))).into_response()
@ -989,13 +1424,16 @@ pub(crate) fn build_app_state_with_path(
local_daemon_mode: bool,
) -> anyhow::Result<Arc<AppState>> {
let secret_store = SecretStore::load(secret_store_path)?;
let (global_event_tx, _) = broadcast::channel(4096);
Ok(Arc::new(AppState {
runs: Mutex::new(HashMap::new()),
aggregate_usage: Mutex::new(UsageAccumulator::default()),
store,
artifact_store,
started_at: Instant::now(),
max_concurrent_runs,
scheduler_notify: Notify::new(),
global_event_tx,
sessions: new_session_store(),
secret_store: AsyncRwLock::new(secret_store),
settings,
@ -1070,6 +1508,13 @@ async fn delete_run(
Err(response) => return response,
};
match delete_run_internal(&state, id).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(response) => response,
}
}
async fn delete_run_internal(state: &Arc<AppState>, id: RunId) -> Result<(), Response> {
let managed_run = if let Ok(mut runs) = state.runs.lock() {
runs.remove(&id)
} else {
@ -1087,31 +1532,29 @@ async fn delete_run(
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();
}
remove_run_dir(&run_dir).map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?;
}
} else {
let storage = Storage::new(state.settings.read().unwrap().storage_dir());
let run_dir = storage.run_scratch(&id).root().to_path_buf();
if let Err(err) = remove_run_dir(&run_dir) {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
remove_run_dir(&run_dir).map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?;
}
match state.store.delete_run(&id).await {
Ok(()) => match state.artifact_store.delete_for_run(&id).await {
Ok(()) => StatusCode::NO_CONTENT.into_response(),
Err(err) => {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
},
Err(err) => {
state.store.delete_run(&id).await.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
})?;
state
.artifact_store
.delete_for_run(&id)
.await
.map_err(|err| {
ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
})?;
Ok(())
}
fn remove_run_dir(run_dir: &std::path::Path) -> std::io::Result<()> {
@ -1252,6 +1695,21 @@ async fn persist_cancelled_run_status(state: &AppState, run_id: RunId) -> anyhow
.await
}
async fn forward_run_events_to_global(
mut run_events: broadcast::Receiver<EventEnvelope>,
global_event_tx: broadcast::Sender<EventEnvelope>,
) {
loop {
match run_events.recv().await {
Ok(event) => {
let _ = global_event_tx.send(event);
}
Err(RecvError::Lagged(_)) => {}
Err(RecvError::Closed) => break,
}
}
}
fn managed_run(
dot_source: String,
status: RunStatus,
@ -1597,6 +2055,10 @@ async fn execute_run(state: Arc<AppState>, run_id: RunId) {
return;
}
};
tokio::spawn(forward_run_events_to_global(
run_store.subscribe(),
state.global_event_tx.clone(),
));
let persisted = match Persisted::load_from_store(&run_store, &run_dir).await {
Ok(persisted) => persisted,
Err(e) => {
@ -2063,14 +2525,7 @@ async fn attach_run_events(
}
};
let stream = stream.filter_map(|result| match result {
Ok(event) => {
let event = api_event_envelope_from_store(&event).ok()?;
let data = serde_json::to_string(&event).ok()?;
let data = redact_jsonl_line(&data);
Some(Ok::<Event, std::convert::Infallible>(
Event::default().data(data),
))
}
Ok(event) => sse_event_from_store(&event).map(Ok::<Event, std::convert::Infallible>),
Err(_) => None,
});

1
lib/crates/fabro-server/tests/it/api/mod.rs
View file

@ -2,3 +2,4 @@
mod mtls;
mod routing;
mod settings;
mod system;

201
lib/crates/fabro-server/tests/it/api/system.rs Normal file
View file

@ -0,0 +1,201 @@
use std::time::Duration;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use fabro_config::Storage;
use fabro_server::server::create_app_state_with_options;
use fabro_types::{RunId, Settings};
use http_body_util::BodyExt;
use tempfile::tempdir;
use tokio::time::timeout;
use tower::ServiceExt;
use crate::helpers::{
MINIMAL_DOT, api, body_json, minimal_manifest_json_with_dry_run, test_app_with_scheduler,
test_settings, wait_for_run_status,
};
fn temp_storage_settings() -> (tempfile::TempDir, Settings) {
let temp = tempdir().expect("tempdir should create");
let mut settings = test_settings();
settings.dry_run = Some(true);
settings.storage_dir = Some(temp.path().join("storage"));
(temp, settings)
}
async fn create_run(app: &axum::Router, manifest: serde_json::Value) -> String {
let request = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(Body::from(serde_json::to_vec(&manifest).unwrap()))
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
let body = body_json(response.into_body()).await;
body["id"].as_str().unwrap().to_string()
}
async fn start_run(app: &axum::Router, run_id: &str) {
let request = Request::builder()
.method("POST")
.uri(api(&format!("/runs/{run_id}/start")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[tokio::test]
async fn get_system_info_returns_runtime_fields() {
let (_temp, settings) = temp_storage_settings();
let expected_storage_dir = settings.storage_dir.clone().unwrap();
let app = fabro_server::server::build_router(
create_app_state_with_options(settings, 5),
fabro_server::jwt_auth::AuthMode::Disabled,
);
let request = Request::builder()
.method("GET")
.uri(api("/system/info"))
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert!(body["version"].as_str().is_some());
assert_eq!(body["storage_engine"], "slatedb");
assert_eq!(
body["storage_dir"],
expected_storage_dir.display().to_string()
);
assert_eq!(body["runs"]["total"], 0);
assert_eq!(body["runs"]["active"], 0);
assert!(body["uptime_secs"].as_i64().is_some());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn get_system_disk_usage_returns_summary_and_verbose_rows() {
let (_temp, settings) = temp_storage_settings();
let storage_dir = settings.storage_dir.clone().unwrap();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
start_run(&app, &run_id).await;
let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
assert_eq!(status, "succeeded");
let logs_dir = storage_dir.join("logs");
std::fs::create_dir_all(&logs_dir).unwrap();
std::fs::write(logs_dir.join("server.log"), b"log line\n").unwrap();
let request = Request::builder()
.method("GET")
.uri(api("/system/df?verbose=true"))
.body(Body::empty())
.unwrap();
let response = app.oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert!(body["summary"].is_array());
assert!(body["total_size_bytes"].as_i64().unwrap_or_default() > 0);
assert!(
body["runs"]
.as_array()
.is_some_and(|runs| runs.iter().any(|entry| entry["run_id"] == run_id))
);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn prune_runs_supports_dry_run_and_deletion() {
let (_temp, settings) = temp_storage_settings();
let storage_dir = settings.storage_dir.clone().unwrap();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let run_id = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
start_run(&app, &run_id).await;
let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
assert_eq!(status, "succeeded");
let run_id_parsed: RunId = run_id.parse().unwrap();
let run_dir = Storage::new(&storage_dir)
.run_scratch(&run_id_parsed)
.root()
.to_path_buf();
assert!(run_dir.exists());
let dry_run_request = Request::builder()
.method("POST")
.uri(api("/system/prune/runs"))
.header("content-type", "application/json")
.body(Body::from(r#"{"before":"9999"}"#))
.unwrap();
let dry_run_response = app.clone().oneshot(dry_run_request).await.unwrap();
assert_eq!(dry_run_response.status(), StatusCode::OK);
let dry_run_body = body_json(dry_run_response.into_body()).await;
assert_eq!(dry_run_body["dry_run"], true);
assert_eq!(dry_run_body["total_count"], 1);
assert_eq!(dry_run_body["runs"][0]["run_id"], run_id);
assert!(run_dir.exists());
let delete_request = Request::builder()
.method("POST")
.uri(api("/system/prune/runs"))
.header("content-type", "application/json")
.body(Body::from(r#"{"dry_run":false,"before":"9999"}"#))
.unwrap();
let delete_response = app.clone().oneshot(delete_request).await.unwrap();
assert_eq!(delete_response.status(), StatusCode::OK);
let delete_body = body_json(delete_response.into_body()).await;
assert_eq!(delete_body["dry_run"], false);
assert_eq!(delete_body["deleted_count"], 1);
assert!(!run_dir.exists());
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn attach_events_streams_only_matching_run_ids() {
let (_temp, settings) = temp_storage_settings();
let app = test_app_with_scheduler(create_app_state_with_options(settings, 5));
let run_one = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
let run_two = create_run(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT)).await;
let request = Request::builder()
.method("GET")
.uri(api(&format!("/attach?run_id={run_one}")))
.body(Body::empty())
.unwrap();
let response = app.clone().oneshot(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let content_type = response
.headers()
.get("content-type")
.expect("content-type should be present")
.to_str()
.unwrap();
assert!(content_type.contains("text/event-stream"));
start_run(&app, &run_one).await;
start_run(&app, &run_two).await;
let mut body = response.into_body();
let mut sse_data = String::new();
while let Ok(Some(Ok(frame))) = timeout(Duration::from_secs(2), body.frame()).await {
if let Some(data) = frame.data_ref() {
sse_data.push_str(&String::from_utf8_lossy(data));
if sse_data.contains(&run_one) {
break;
}
}
}
assert!(
sse_data.contains(&run_one),
"expected filtered stream data: {sse_data}"
);
assert!(
!sse_data.contains(&run_two),
"filtered stream should exclude non-matching run ids: {sse_data}"
);
}

4
lib/crates/fabro-store/src/slate/run_store.rs
View file

@ -114,6 +114,10 @@ impl RunDatabase {
self.inner.run_id
}
pub fn subscribe(&self) -> broadcast::Receiver<EventEnvelope> {
self.inner.event_tx.subscribe()
}
pub(crate) fn matches_run(&self, run_id: &RunId) -> bool {
self.inner.run_id == *run_id
}

9
lib/packages/fabro-api-client/src/.openapi-generator/FILES
View file

@ -12,6 +12,7 @@ api/runs-api.ts
api/secrets-api.ts
api/sessions-api.ts
api/settings-api.ts
api/system-api.ts
api/usage-api.ts
api/verification-api.ts
api/workflows-api.ts
@ -61,6 +62,9 @@ models/diagnostics-report.ts
models/diagnostics-section.ts
models/diff-file.ts
models/diff-stats.ts
models/disk-usage-response.ts
models/disk-usage-run-row.ts
models/disk-usage-summary-row.ts
models/error-response-entry.ts
models/error-response.ts
models/event-envelope.ts
@ -132,6 +136,9 @@ models/preflight-response.ts
models/preflight-workflow-summary.ts
models/preview-url-request.ts
models/preview-url-response.ts
models/prune-run-entry.ts
models/prune-runs-request.ts
models/prune-runs-response.ts
models/pull-request-settings.ts
models/question-type.ts
models/recent-control-result.ts
@ -202,6 +209,8 @@ models/status-reason.ts
models/steer-request.ts
models/store-run-summary.ts
models/submit-answer-request.ts
models/system-info-response.ts
models/system-run-counts.ts
models/system-stage-turn.ts
models/tls-settings.ts
models/token-usage.ts

1
lib/packages/fabro-api-client/src/api.ts
View file

@ -27,6 +27,7 @@ export * from './api/runs-api';
export * from './api/secrets-api';
export * from './api/sessions-api';
export * from './api/settings-api';
export * from './api/system-api';
export * from './api/usage-api';
export * from './api/verification-api';
export * from './api/workflows-api';

360
lib/packages/fabro-api-client/src/api/system-api.ts Normal file
View file

@ -0,0 +1,360 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
import type { Configuration } from '../configuration';
import type { AxiosPromise, AxiosInstance, RawAxiosRequestConfig } from 'axios';
import globalAxios from 'axios';
// Some imports not used depending on template conditions
// @ts-ignore
import { DUMMY_BASE_URL, assertParamExists, setApiKeyToObject, setBasicAuthToObject, setBearerAuthToObject, setOAuthToObject, setSearchParams, serializeDataIfNeeded, toPathString, createRequestFunction, replaceWithSerializableTypeIfNeeded } from '../common';
// @ts-ignore
import { BASE_PATH, COLLECTION_FORMATS, type RequestArgs, BaseAPI, RequiredError, operationServerMap } from '../base';
// @ts-ignore
import type { DiskUsageResponse } from '../models';
// @ts-ignore
import type { ErrorResponse } from '../models';
// @ts-ignore
import type { PruneRunsRequest } from '../models';
// @ts-ignore
import type { PruneRunsResponse } from '../models';
// @ts-ignore
import type { SystemInfoResponse } from '../models';
/**
* SystemApi - axios parameter creator
*/
export const SystemApiAxiosParamCreator = function (configuration?: Configuration) {
return {
/**
* Opens a server-sent event stream for live run events across the server.
* @summary Attach Global Events
* @param {string} [runId] Optional comma-separated list of run IDs to include.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
attachEvents: async (runId?: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
const localVarPath = `/api/v1/attach`;
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
if (runId !== undefined) {
localVarQueryParameter['run_id'] = runId;
}
localVarHeaderParameter['Accept'] = 'text/event-stream';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Returns disk usage for the server storage directory.
* @summary Retrieve System Disk Usage
* @param {boolean} [verbose] Include per-run disk usage rows.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getSystemDiskUsage: async (verbose?: boolean, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
const localVarPath = `/api/v1/system/df`;
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
if (verbose !== undefined) {
localVarQueryParameter['verbose'] = verbose;
}
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Returns runtime details about the active Fabro server process.
* @summary Retrieve System Info
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getSystemInfo: async (options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
const localVarPath = `/api/v1/system/info`;
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled.
* @summary Prune Runs
* @param {PruneRunsRequest} pruneRunsRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
pruneRuns: async (pruneRunsRequest: PruneRunsRequest, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'pruneRunsRequest' is not null or undefined
assertParamExists('pruneRuns', 'pruneRunsRequest', pruneRunsRequest)
const localVarPath = `/api/v1/system/prune/runs`;
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
if (configuration) {
baseOptions = configuration.baseOptions;
}
const localVarRequestOptions = { method: 'POST', ...baseOptions, ...options};
const localVarHeaderParameter = {} as any;
const localVarQueryParameter = {} as any;
// authentication mTLS required
await setApiKeyToObject(localVarHeaderParameter, "X-mTLS-Client-CN", configuration)
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
localVarHeaderParameter['Content-Type'] = 'application/json';
localVarHeaderParameter['Accept'] = 'application/json';
setSearchParams(localVarUrlObj, localVarQueryParameter);
let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
localVarRequestOptions.data = serializeDataIfNeeded(pruneRunsRequest, localVarRequestOptions, configuration)
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
}
};
/**
* SystemApi - functional programming interface
*/
export const SystemApiFp = function(configuration?: Configuration) {
const localVarAxiosParamCreator = SystemApiAxiosParamCreator(configuration)
return {
/**
* Opens a server-sent event stream for live run events across the server.
* @summary Attach Global Events
* @param {string} [runId] Optional comma-separated list of run IDs to include.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async attachEvents(runId?: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<string>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.attachEvents(runId, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['SystemApi.attachEvents']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns disk usage for the server storage directory.
* @summary Retrieve System Disk Usage
* @param {boolean} [verbose] Include per-run disk usage rows.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async getSystemDiskUsage(verbose?: boolean, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<DiskUsageResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.getSystemDiskUsage(verbose, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['SystemApi.getSystemDiskUsage']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns runtime details about the active Fabro server process.
* @summary Retrieve System Info
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async getSystemInfo(options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<SystemInfoResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.getSystemInfo(options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['SystemApi.getSystemInfo']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled.
* @summary Prune Runs
* @param {PruneRunsRequest} pruneRunsRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async pruneRuns(pruneRunsRequest: PruneRunsRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<PruneRunsResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.pruneRuns(pruneRunsRequest, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['SystemApi.pruneRuns']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
}
};
/**
* SystemApi - factory interface
*/
export const SystemApiFactory = function (configuration?: Configuration, basePath?: string, axios?: AxiosInstance) {
const localVarFp = SystemApiFp(configuration)
return {
/**
* Opens a server-sent event stream for live run events across the server.
* @summary Attach Global Events
* @param {string} [runId] Optional comma-separated list of run IDs to include.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
attachEvents(runId?: string, options?: RawAxiosRequestConfig): AxiosPromise<string> {
return localVarFp.attachEvents(runId, options).then((request) => request(axios, basePath));
},
/**
* Returns disk usage for the server storage directory.
* @summary Retrieve System Disk Usage
* @param {boolean} [verbose] Include per-run disk usage rows.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getSystemDiskUsage(verbose?: boolean, options?: RawAxiosRequestConfig): AxiosPromise<DiskUsageResponse> {
return localVarFp.getSystemDiskUsage(verbose, options).then((request) => request(axios, basePath));
},
/**
* Returns runtime details about the active Fabro server process.
* @summary Retrieve System Info
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getSystemInfo(options?: RawAxiosRequestConfig): AxiosPromise<SystemInfoResponse> {
return localVarFp.getSystemInfo(options).then((request) => request(axios, basePath));
},
/**
* Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled.
* @summary Prune Runs
* @param {PruneRunsRequest} pruneRunsRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
pruneRuns(pruneRunsRequest: PruneRunsRequest, options?: RawAxiosRequestConfig): AxiosPromise<PruneRunsResponse> {
return localVarFp.pruneRuns(pruneRunsRequest, options).then((request) => request(axios, basePath));
},
};
};
/**
* SystemApi - object-oriented interface
*/
export class SystemApi extends BaseAPI {
/**
* Opens a server-sent event stream for live run events across the server.
* @summary Attach Global Events
* @param {string} [runId] Optional comma-separated list of run IDs to include.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public attachEvents(runId?: string, options?: RawAxiosRequestConfig) {
return SystemApiFp(this.configuration).attachEvents(runId, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns disk usage for the server storage directory.
* @summary Retrieve System Disk Usage
* @param {boolean} [verbose] Include per-run disk usage rows.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public getSystemDiskUsage(verbose?: boolean, options?: RawAxiosRequestConfig) {
return SystemApiFp(this.configuration).getSystemDiskUsage(verbose, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns runtime details about the active Fabro server process.
* @summary Retrieve System Info
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public getSystemInfo(options?: RawAxiosRequestConfig) {
return SystemApiFp(this.configuration).getSystemInfo(options).then((request) => request(this.axios, this.basePath));
}
/**
* Deletes completed runs matching the provided filters, or previews the deletion set when dry-run is enabled.
* @summary Prune Runs
* @param {PruneRunsRequest} pruneRunsRequest
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public pruneRuns(pruneRunsRequest: PruneRunsRequest, options?: RawAxiosRequestConfig) {
return SystemApiFp(this.configuration).pruneRuns(pruneRunsRequest, options).then((request) => request(this.axios, this.basePath));
}
}

41
lib/packages/fabro-api-client/src/models/disk-usage-response.ts Normal file
View file

@ -0,0 +1,41 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
// May contain unused imports in some cases
// @ts-ignore
import type { DiskUsageRunRow } from './disk-usage-run-row';
// May contain unused imports in some cases
// @ts-ignore
import type { DiskUsageSummaryRow } from './disk-usage-summary-row';
/**
* Disk usage summary for server-managed data.
*/
export interface DiskUsageResponse {
'summary'?: Array<DiskUsageSummaryRow>;
/**
* Total size of all tracked system data.
*/
'total_size_bytes'?: number;
/**
* Total bytes reclaimable by deleting inactive runs and logs.
*/
'total_reclaimable_bytes'?: number;
/**
* Per-run usage rows when verbose output is requested.
*/
'runs'?: Array<DiskUsageRunRow>;
}

46
lib/packages/fabro-api-client/src/models/disk-usage-run-row.ts Normal file
View file

@ -0,0 +1,46 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Per-run disk usage information.
*/
export interface DiskUsageRunRow {
/**
* Run identifier.
*/
'run_id'?: string;
/**
* Workflow display name.
*/
'workflow_name'?: string;
/**
* Current run status.
*/
'status'?: string;
/**
* Human-readable start timestamp.
*/
'start_time'?: string;
/**
* Size used by the run scratch directory.
*/
'size_bytes'?: number;
/**
* Whether the run is inactive and reclaimable.
*/
'reclaimable'?: boolean;
}

42
lib/packages/fabro-api-client/src/models/disk-usage-summary-row.ts Normal file
View file

@ -0,0 +1,42 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* One top-level disk usage category.
*/
export interface DiskUsageSummaryRow {
/**
* Category name, such as runs or logs.
*/
'type'?: string;
/**
* Number of items in the category.
*/
'count'?: number;
/**
* Number of active items when applicable.
*/
'active'?: number;
/**
* Total bytes used by the category.
*/
'size_bytes'?: number;
/**
* Bytes reclaimable by pruning the category.
*/
'reclaimable_bytes'?: number;
}

8
lib/packages/fabro-api-client/src/models/index.ts
View file

@ -40,6 +40,9 @@ export * from './diagnostics-report';
export * from './diagnostics-section';
export * from './diff-file';
export * from './diff-stats';
export * from './disk-usage-response';
export * from './disk-usage-run-row';
export * from './disk-usage-summary-row';
export * from './error-response';
export * from './error-response-entry';
export * from './event-envelope';
@ -110,6 +113,9 @@ export * from './preflight-response';
export * from './preflight-workflow-summary';
export * from './preview-url-request';
export * from './preview-url-response';
export * from './prune-run-entry';
export * from './prune-runs-request';
export * from './prune-runs-response';
export * from './pull-request-settings';
export * from './question-type';
export * from './recent-control-result';
@ -180,6 +186,8 @@ export * from './status-reason';
export * from './steer-request';
export * from './store-run-summary';
export * from './submit-answer-request';
export * from './system-info-response';
export * from './system-run-counts';
export * from './system-stage-turn';
export * from './tls-settings';
export * from './token-usage';

38
lib/packages/fabro-api-client/src/models/prune-run-entry.ts Normal file
View file

@ -0,0 +1,38 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* One run matched by a prune preview.
*/
export interface PruneRunEntry {
/**
* Run identifier.
*/
'run_id'?: string;
/**
* Scratch directory name for the run.
*/
'dir_name'?: string;
/**
* Workflow display name.
*/
'workflow_name'?: string;
/**
* Bytes used by the run scratch directory.
*/
'size_bytes'?: number;
}

46
lib/packages/fabro-api-client/src/models/prune-runs-request.ts Normal file
View file

@ -0,0 +1,46 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Filters for system run pruning.
*/
export interface PruneRunsRequest {
/**
* Preview matching runs without deleting them.
*/
'dry_run'?: boolean;
/**
* Include runs started before this YYYY-MM-DD prefix.
*/
'before'?: string;
/**
* Filter by workflow name substring.
*/
'workflow'?: string;
/**
* Label filters applied with AND semantics.
*/
'labels'?: { [key: string]: string; };
/**
* Include orphan run directories without run metadata.
*/
'orphans'?: boolean;
/**
* Include only runs older than this duration, such as 24h or 7d.
*/
'older_than'?: string;
}

49
lib/packages/fabro-api-client/src/models/prune-runs-response.ts Normal file
View file

@ -0,0 +1,49 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
// May contain unused imports in some cases
// @ts-ignore
import type { PruneRunEntry } from './prune-run-entry';
/**
* Result of a prune preview or deletion.
*/
export interface PruneRunsResponse {
/**
* Whether this response is a dry-run preview.
*/
'dry_run'?: boolean;
/**
* Matched runs when dry-run is enabled.
*/
'runs'?: Array<PruneRunEntry>;
/**
* Count of runs matching the prune filters.
*/
'total_count'?: number;
/**
* Total bytes of the matching runs.
*/
'total_size_bytes'?: number;
/**
* Number of runs deleted when dry-run is false.
*/
'deleted_count'?: number;
/**
* Estimated freed bytes when deletion occurs.
*/
'freed_bytes'?: number;
}

62
lib/packages/fabro-api-client/src/models/system-info-response.ts Normal file
View file

@ -0,0 +1,62 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
// May contain unused imports in some cases
// @ts-ignore
import type { SystemRunCounts } from './system-run-counts';
/**
* Runtime information for the active Fabro server process.
*/
export interface SystemInfoResponse {
/**
* Server version string.
*/
'version'?: string;
/**
* Build git SHA when available.
*/
'git_sha'?: string;
/**
* Build date when available.
*/
'build_date'?: string;
/**
* Target operating system.
*/
'os'?: string;
/**
* Target CPU architecture.
*/
'arch'?: string;
/**
* Backing run storage engine.
*/
'storage_engine'?: string;
/**
* Configured storage directory.
*/
'storage_dir'?: string;
/**
* Seconds since this server process started.
*/
'uptime_secs'?: number;
'runs'?: SystemRunCounts;
/**
* Effective sandbox provider for launched runs.
*/
'sandbox_provider'?: string;
}

30
lib/packages/fabro-api-client/src/models/system-run-counts.ts Normal file
View file

@ -0,0 +1,30 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.1.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Counts of known runs in the active server process.
*/
export interface SystemRunCounts {
/**
* Total runs tracked by the server process.
*/
'total'?: number;
/**
* Runs currently queued or executing.
*/
'active'?: number;
}