diff --git a/Cargo.lock b/Cargo.lock index 12cc71e8a..ed34a2c98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1907,6 +1907,7 @@ dependencies = [ "tracing", "ulid", "uuid", + "walkdir", "x509-parser", ] diff --git a/docs/api-reference/fabro-api.yaml b/docs/api-reference/fabro-api.yaml index 3c332e46e..2383637a6 100644 --- a/docs/api-reference/fabro-api.yaml +++ b/docs/api-reference/fabro-api.yaml @@ -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 diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index ee8c917b6..72c71a341 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -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, +} + #[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)] diff --git a/lib/crates/fabro-cli/src/commands/runs/rm.rs b/lib/crates/fabro-cli/src/commands/runs/rm.rs index 0dc336235..36e435fe2 100644 --- a/lib/crates/fabro-cli/src/commands/runs/rm.rs +++ b/lib/crates/fabro-cli/src/commands/runs/rm.rs @@ -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 { - if let Some(run_state) = run_state { - return run_state.sandbox.clone(); - } - None -} diff --git a/lib/crates/fabro-cli/src/commands/system/df.rs b/lib/crates/fabro-cli/src/commands/system/df.rs index 9e3e3e070..9defef074 100644 --- a/lib/crates/fabro-cli/src/commands/system/df.rs +++ b/lib/crates/fabro-cli/src/commands/system/df.rs @@ -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, - size_bytes: u64, - #[serde(skip_serializing_if = "Option::is_none")] - reclaimable_bytes: Option, -} - -#[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, - total_size_bytes: u64, - total_reclaimable_bytes: u64, - #[serde(skip_serializing_if = "Option::is_none")] - runs: Option>, -} 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>, - 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::>() - }); - 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![ 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> = run_details + let verbose_rows: Vec> = 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> { + 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() -} diff --git a/lib/crates/fabro-cli/src/commands/system/events.rs b/lib/crates/fabro-cli/src/commands/system/events.rs new file mode 100644 index 000000000..8be8519c8 --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/system/events.rs @@ -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, json_output: bool) -> Result<()> { + while let Some(pos) = buffer.iter().position(|byte| *byte == b'\n') { + let line = buffer.drain(..=pos).collect::>(); + 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) +} diff --git a/lib/crates/fabro-cli/src/commands/system/info.rs b/lib/crates/fabro-cli/src/commands/system/info.rs new file mode 100644 index 000000000..e0a4b9003 --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/system/info.rs @@ -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(()) +} diff --git a/lib/crates/fabro-cli/src/commands/system/mod.rs b/lib/crates/fabro-cli/src/commands/system/mod.rs index 09062e017..62c596c92 100644 --- a/lib/crates/fabro-cli/src/commands/system/mod.rs +++ b/lib/crates/fabro-cli/src/commands/system/mod.rs @@ -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, } } diff --git a/lib/crates/fabro-cli/src/commands/system/prune.rs b/lib/crates/fabro-cli/src/commands/system/prune.rs index f129b93ef..b86329751 100644 --- a/lib/crates/fabro-cli/src/commands/system/prune.rs +++ b/lib/crates/fabro-cli/src/commands/system/prune.rs @@ -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 { @@ -47,123 +48,56 @@ pub(crate) fn parse_duration(s: &str) -> Result { } } -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::::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 = 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 { 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() } diff --git a/lib/crates/fabro-cli/src/server_client.rs b/lib/crates/fabro-cli/src/server_client.rs index 24558aa12..fd3646836 100644 --- a/lib/crates/fabro-cli/src/server_client.rs +++ b/lib/crates/fabro-cli/src/server_client.rs @@ -164,7 +164,14 @@ async fn connect_target_api_client( pub(crate) async fn connect_server_backed_api_client( args: &ServerTargetArgs, ) -> Result { - 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 { + 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(err: progenitor_client::Error) -> anyhow::Error +pub(crate) fn map_api_error(err: progenitor_client::Error) -> anyhow::Error where E: serde::Serialize + std::fmt::Debug, { diff --git a/lib/crates/fabro-cli/src/server_runs.rs b/lib/crates/fabro-cli/src/server_runs.rs index f819398c7..9bbfb077d 100644 --- a/lib/crates/fabro-cli/src/server_runs.rs +++ b/lib/crates/fabro-cli/src/server_runs.rs @@ -37,10 +37,6 @@ impl ServerRunLookup { &self.client } - pub(crate) fn summaries(&self) -> &[RunSummary] { - &self.summaries - } - pub(crate) fn resolve(&self, selector: &str) -> Result { resolve_run_from_summaries(&self.summaries, &self.scratch_base, selector) } diff --git a/lib/crates/fabro-cli/tests/it/cmd/mod.rs b/lib/crates/fabro-cli/tests/it/cmd/mod.rs index 3f7b9d3e3..0d3616d8f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mod.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mod.rs @@ -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; diff --git a/lib/crates/fabro-cli/tests/it/cmd/support.rs b/lib/crates/fabro-cli/tests/it/cmd/support.rs index 4922b277d..0a7632710 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/support.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/support.rs @@ -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 } diff --git a/lib/crates/fabro-cli/tests/it/cmd/system.rs b/lib/crates/fabro-cli/tests/it/cmd/system.rs index 808b3a839..7b5e3d09c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system.rs @@ -14,9 +14,11 @@ fn help() { Usage: fabro system [OPTIONS] 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=] diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs index d4a6ba2da..14475ae8b 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs @@ -20,8 +20,9 @@ fn help() { --json Output as JSON [env: FABRO_JSON=] --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 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 ----- diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_events.rs b/lib/crates/fabro-cli/tests/it/cmd/system_events.rs new file mode 100644 index 000000000..b13612539 --- /dev/null +++ b/lib/crates/fabro-cli/tests/it/cmd/system_events.rs @@ -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 Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]] + --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] + --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 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 ----- + "); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_info.rs b/lib/crates/fabro-cli/tests/it/cmd/system_info.rs new file mode 100644 index 000000000..c99c185c0 --- /dev/null +++ b/lib/crates/fabro-cli/tests/it/cmd/system_info.rs @@ -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 Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]] + --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] + --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()); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs index f0bc8bc9e..b3493a540 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_prune.rs @@ -18,14 +18,15 @@ fn help() { Options: --json Output as JSON [env: FABRO_JSON=] --storage-dir Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=[STORAGE_DIR]] - --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 Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=] + --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 Filter by workflow name (substring match) --label 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 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 diff --git a/lib/crates/fabro-server/Cargo.toml b/lib/crates/fabro-server/Cargo.toml index 8df204c20..d44916d69 100644 --- a/lib/crates/fabro-server/Cargo.toml +++ b/lib/crates/fabro-server/Cargo.toml @@ -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"] } diff --git a/lib/crates/fabro-server/src/demo/mod.rs b/lib/crates/fabro-server/src/demo/mod.rs index 599e6bc45..82208efd8 100644 --- a/lib/crates/fabro-server/src/demo/mod.rs +++ b/lib/crates/fabro-server/src/demo/mod.rs @@ -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>, +) -> 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>, +) -> 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>, + Query(params): Query, +) -> 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>, +) -> 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( diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index cb33dcddc..dc2d9e942 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/lib/crates/fabro-server/src/server.rs @@ -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, } +#[derive(serde::Deserialize)] +pub(crate) struct DfParams { + #[serde(default)] + pub(crate) verbose: bool, +} + +#[derive(serde::Deserialize)] +struct GlobalAttachParams { + #[serde(default)] + run_id: Option, +} + #[derive(serde::Deserialize)] struct ArtifactFilenameParams { #[serde(default)] @@ -226,8 +245,10 @@ pub struct AppState { aggregate_usage: Mutex, store: Arc, artifact_store: ArtifactStore, + started_at: Instant, max_concurrent_runs: usize, scheduler_notify: Notify, + global_event_tx: broadcast::Sender, pub sessions: SessionStore, pub(crate) secret_store: AsyncRwLock, pub(crate) settings: Arc>, @@ -365,6 +386,7 @@ fn demo_routes() -> Router> { .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> { .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> { .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> { .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>, +) -> 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>, + Query(params): Query, +) -> 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>, + Json(body): Json, +) -> 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>, + Query(params): Query, +) -> 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::) + } + Err(_) => None, + } + }); + + Sse::new(stream) + .keep_alive(KeepAlive::default()) + .into_response() +} + +struct PrunePlan { + run_ids: Vec, + rows: Vec, + total_size_bytes: u64, +} + +fn build_disk_usage_response( + summaries: &[fabro_store::RunSummary], + storage_dir: &std::path::Path, + verbose: bool, +) -> anyhow::Result { + 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 { + 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::>(); + + 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::>(); + let total_size_bytes = rows + .iter() + .map(|row| row.size_bytes.unwrap_or_default()) + .sum::() + .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 { + 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::()?; + 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>, 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::() + .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>) -> 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::().ok()) + else { + return false; + }; + run_filter.contains(&run_id) +} + +fn sse_event_from_store(event: &EventEnvelope) -> Option { + 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(value: T) -> i64 +where + i64: TryFrom, +{ + i64::try_from(value).unwrap_or(i64::MAX) +} + async fn list_secrets(_auth: AuthenticatedService, State(state): State>) -> 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> { 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, 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, + global_event_tx: broadcast::Sender, +) { + 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, 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::default().data(data), - )) - } + Ok(event) => sse_event_from_store(&event).map(Ok::), Err(_) => None, }); diff --git a/lib/crates/fabro-server/tests/it/api/mod.rs b/lib/crates/fabro-server/tests/it/api/mod.rs index 62180f615..5f65bdc48 100644 --- a/lib/crates/fabro-server/tests/it/api/mod.rs +++ b/lib/crates/fabro-server/tests/it/api/mod.rs @@ -2,3 +2,4 @@ mod mtls; mod routing; mod settings; +mod system; diff --git a/lib/crates/fabro-server/tests/it/api/system.rs b/lib/crates/fabro-server/tests/it/api/system.rs new file mode 100644 index 000000000..e915ab468 --- /dev/null +++ b/lib/crates/fabro-server/tests/it/api/system.rs @@ -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}" + ); +} diff --git a/lib/crates/fabro-store/src/slate/run_store.rs b/lib/crates/fabro-store/src/slate/run_store.rs index 2381c8f75..f0fd91fc3 100644 --- a/lib/crates/fabro-store/src/slate/run_store.rs +++ b/lib/crates/fabro-store/src/slate/run_store.rs @@ -114,6 +114,10 @@ impl RunDatabase { self.inner.run_id } + pub fn subscribe(&self) -> broadcast::Receiver { + self.inner.event_tx.subscribe() + } + pub(crate) fn matches_run(&self, run_id: &RunId) -> bool { self.inner.run_id == *run_id } diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index 541f1bfb0..2f521ef08 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/lib/packages/fabro-api-client/src/.openapi-generator/FILES @@ -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 diff --git a/lib/packages/fabro-api-client/src/api.ts b/lib/packages/fabro-api-client/src/api.ts index ed2649846..e4eaa0f55 100644 --- a/lib/packages/fabro-api-client/src/api.ts +++ b/lib/packages/fabro-api-client/src/api.ts @@ -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'; diff --git a/lib/packages/fabro-api-client/src/api/system-api.ts b/lib/packages/fabro-api-client/src/api/system-api.ts new file mode 100644 index 000000000..f4206c40b --- /dev/null +++ b/lib/packages/fabro-api-client/src/api/system-api.ts @@ -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 => { + 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 => { + 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 => { + 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 => { + // 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> { + 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> { + 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> { + 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> { + 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 { + 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 { + 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 { + 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 { + 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)); + } +} + diff --git a/lib/packages/fabro-api-client/src/models/disk-usage-response.ts b/lib/packages/fabro-api-client/src/models/disk-usage-response.ts new file mode 100644 index 000000000..3d0c0a8bb --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/disk-usage-response.ts @@ -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; + /** + * 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; +} + diff --git a/lib/packages/fabro-api-client/src/models/disk-usage-run-row.ts b/lib/packages/fabro-api-client/src/models/disk-usage-run-row.ts new file mode 100644 index 000000000..8b3a64208 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/disk-usage-run-row.ts @@ -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; +} + diff --git a/lib/packages/fabro-api-client/src/models/disk-usage-summary-row.ts b/lib/packages/fabro-api-client/src/models/disk-usage-summary-row.ts new file mode 100644 index 000000000..bd41f1f43 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/disk-usage-summary-row.ts @@ -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; +} + diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index 939479942..fb7143d25 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/lib/packages/fabro-api-client/src/models/index.ts @@ -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'; diff --git a/lib/packages/fabro-api-client/src/models/prune-run-entry.ts b/lib/packages/fabro-api-client/src/models/prune-run-entry.ts new file mode 100644 index 000000000..d1521ff08 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/prune-run-entry.ts @@ -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; +} + diff --git a/lib/packages/fabro-api-client/src/models/prune-runs-request.ts b/lib/packages/fabro-api-client/src/models/prune-runs-request.ts new file mode 100644 index 000000000..7da7e2a3c --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/prune-runs-request.ts @@ -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; +} + diff --git a/lib/packages/fabro-api-client/src/models/prune-runs-response.ts b/lib/packages/fabro-api-client/src/models/prune-runs-response.ts new file mode 100644 index 000000000..4382b5afe --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/prune-runs-response.ts @@ -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; + /** + * 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; +} + diff --git a/lib/packages/fabro-api-client/src/models/system-info-response.ts b/lib/packages/fabro-api-client/src/models/system-info-response.ts new file mode 100644 index 000000000..a53c9f6c2 --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/system-info-response.ts @@ -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; +} + diff --git a/lib/packages/fabro-api-client/src/models/system-run-counts.ts b/lib/packages/fabro-api-client/src/models/system-run-counts.ts new file mode 100644 index 000000000..8fe9780be --- /dev/null +++ b/lib/packages/fabro-api-client/src/models/system-run-counts.ts @@ -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; +} +