scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-08-28 05:27:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

feat(server): resolve run selectors on the server

Browse source 
Add a server-native run selector endpoint and migrate CLI single-run flows to
use it instead of local workflow-store heuristics. This also moves store dump
export assembly into the CLI, removes the production CLI dependency on
fabro_workflow run lookup and dump helpers, and records the remaining
cli-to-workflow coupling in an audit document.
This commit is contained in:
Bryan Helmkamp 2026-04-19 19:31:13 -04:00
parent 32b88d7833
commit 26db764e6c
No known key found for this signature in database
32 changed files with 1134 additions and 196 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

50
docs-internal/cli-workflow-coupling-audit.md Normal file
View file

@ -0,0 +1,50 @@
# CLI Workflow Coupling Audit
## Scope
- Production imports under `lib/crates/fabro-cli/src/**` that still reference `fabro_workflow::*` after the server-owned selector/export refactor.
- Test-only imports are listed separately so the remaining architectural debt is explicit.
## Completed In This Change
- Removed the production CLI dependency on `fabro_workflow::run_lookup`.
- Removed the production CLI dependency on `fabro_workflow::run_dump`.
- Added server-owned selector resolution via `GET /api/v1/runs/resolve` and migrated single-run selector flows to it.
## Remaining Production Couplings
| Path | Direct dependency | Why it still exists | Required remediation track |
| --- | --- | --- | --- |
| `lib/crates/fabro-cli/src/commands/run/fork.rs` | `operations::{ForkRunInput, RewindTarget, build_timeline_or_rebuild, fork}` | User-facing CLI still reconstructs run timelines and mutates rewind/fork metadata locally. | Replace with a server API for timeline inspection and fork execution. |
| `lib/crates/fabro-cli/src/commands/run/rewind.rs` | `git::MetadataStore`, `operations::{RewindInput, RewindTarget, RunTimeline, TimelineEntry, build_timeline_or_rebuild, rewind}` | User-facing CLI still performs rewind timeline resolution and metadata mutation locally. | Replace with a server API for rewind preview and rewind execution. |
| `lib/crates/fabro-cli/src/commands/pr/create.rs` | `outcome::StageStatus`, `pull_request::maybe_open_pull_request` | CLI still reconstructs store state and runs PR creation logic from the workflow pipeline directly. | Replace with a server API, or extract PR orchestration into a non-engine shared service crate plus API. |
| `lib/crates/fabro-cli/src/commands/run/runner.rs` | `artifact_snapshot::CapturedArtifactInfo`, `artifact_upload::{ArtifactSink, StageArtifactUploader}`, `event::{Emitter, RunEventSink}`, `operations::{self, StartServices}`, `run_control::RunControlState`, `runtime_store::{RunStoreBackend, RunStoreHandle}` | Hidden worker subprocess path still lives inside the CLI crate and embeds the workflow engine directly. | Re-home worker/runtime code outside the user CLI surface, ideally into a dedicated worker crate or binary. |
| `lib/crates/fabro-cli/src/manifest_builder.rs` | `git::{GitSyncStatus, head_sha, sync_status}` | Manifest submission still relies on git helper logic that happens to live in `fabro_workflow`. | Extract git-sync inspection helpers into a non-workflow shared crate/module. |
| `lib/crates/fabro-cli/src/server_client.rs` | `artifact_snapshot::CapturedArtifactInfo` | The upload client reuses a workflow-owned artifact snapshot DTO. | Extract shared artifact snapshot DTOs into `fabro-store`, `fabro-types`, or a dedicated shared crate. |
| `lib/crates/fabro-cli/src/commands/runs/inspect.rs` | `run_status::RunStatus` | CLI output types still depend on engine-owned run status enums. | Extract shared status types into `fabro-types` or switch to API-generated/public store types. |
| `lib/crates/fabro-cli/src/commands/runs/list.rs` | `run_status::RunStatus` | List rendering still depends on engine-owned run status enums. | Extract shared status types into `fabro-types` or switch to API-generated/public store types. |
| `lib/crates/fabro-cli/src/commands/run/attach.rs` | `outcome::StageStatus`, `run_status::RunStatus` | Attach/replay logic still formats engine-owned terminal status types directly. | Extract shared run/conclusion status types into `fabro-types`. |
| `lib/crates/fabro-cli/src/commands/run/output.rs` | `outcome::StageStatus`, `records::Conclusion` | Human-readable completion output still consumes workflow-owned conclusion/status records. | Extract shared conclusion/status DTOs into `fabro-types` or `fabro-store`. |
| `lib/crates/fabro-cli/src/commands/run/wait.rs` | `records::Conclusion`, `run_status::RunStatus` | Wait output still depends on workflow-owned status/conclusion records. | Extract shared conclusion/status DTOs into `fabro-types` or `fabro-store`. |
| `lib/crates/fabro-cli/src/commands/run/run_progress/stage_display.rs` | `outcome::{StageStatus, format_cost}` | Progress UI still depends on workflow-owned stage status and cost-formatting helper code. | Extract shared stage status types into `fabro-types` and move formatting helpers into `fabro-util`. |
| `lib/crates/fabro-cli/src/commands/run/run_progress/info_display.rs` | `event::RunNoticeLevel` | Progress UI still formats workflow-owned notice levels directly. | Extract shared notice/event enums into `fabro-types`. |
| `lib/crates/fabro-cli/src/commands/run/run_progress/event.rs` | `event::RunNoticeLevel` | Progress event translation still depends on workflow-owned notice levels. | Extract shared notice/event enums into `fabro-types`. |
## Test-Only Couplings
| Path | Direct dependency | Why it still exists | Suggested handling |
| --- | --- | --- | --- |
| `lib/crates/fabro-cli/src/commands/store/dump.rs` test module | `event::{Event, append_event}` | Unit tests synthesize workflow events directly. | Low priority; keep until a lighter-weight event fixture helper exists. |
| `lib/crates/fabro-cli/src/commands/run/wait.rs` test module | `outcome::StageStatus`, `records::Conclusion`, `run_status::RunStatusRecord` | Output tests construct workflow-owned records directly. | Replace with shared fixture builders once status/conclusion DTOs move out. |
| `lib/crates/fabro-cli/src/commands/run/run_progress/mod.rs` test module | `event::{Event, RunNoticeLevel, to_run_event, to_run_event_at}`, `outcome::billed_model_usage_from_llm` | Progress tests build engine events directly. | Replace with shared event fixture helpers after event DTO extraction. |
| `lib/crates/fabro-cli/src/commands/run/run_progress/event.rs` test module | `event::{Event, to_run_event}` | Event rendering tests depend on engine event constructors. | Replace with shared event fixture helpers after event DTO extraction. |
| `lib/crates/fabro-cli/src/commands/run/runner.rs` test module | `artifact_upload::StageArtifactUploader` | Worker tests still reach into workflow upload internals. | Keep with worker re-home work; not worth separating first. |
| `lib/crates/fabro-cli/tests/it/workflow/real_cli.rs` | `context::Context`, `event::Emitter`, `handler::agent::{CodergenBackend, CodergenResult}`, `handler::llm::cli::AgentCliBackend` | Integration test exercises the real workflow engine directly through CLI harnesses. | Accept as engine integration coverage or move under workflow-owned test support later. |
| `lib/crates/fabro-cli/tests/it/scenario/recovery.rs` | `operations::{RunTimeline, build_timeline}` | Scenario test inspects rewind timeline internals directly. | Replace after server-owned rewind/timeline APIs exist. |
## Follow-Up Order
1. Design server APIs for rewind/fork and PR creation so user-facing CLI commands stop importing workflow operations directly.
2. Decide whether the hidden worker path should move to a dedicated worker crate/binary or remain a CLI-internal implementation detail with a stricter boundary.
3. Extract shared status, conclusion, notice, and artifact snapshot types/helpers out of `fabro_workflow`.
4. Extract git sync helpers from `fabro_workflow` so manifest building no longer depends on engine code.

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

@ -437,6 +437,34 @@ paths:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/resolve:
get:
operationId: resolveRun
tags: [Runs]
summary: Resolve Run Selector
description: Resolves a run selector to one durable run summary using server-owned selector semantics.
parameters:
- $ref: "#/components/parameters/RunSelector"
responses:
"200":
description: Durable run summary
content:
application/json:
schema:
$ref: "#/components/schemas/StoreRunSummary"
"400":
description: Selector is invalid or ambiguous
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"404":
description: No run matched the selector
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/preflight:
post:
operationId: runPreflight
@ -1822,6 +1850,15 @@ components:
type: string
example: 01JNQVR7M0EJ5GKAT2SC4ERS1Z
RunSelector:
name: selector
in: query
required: true
description: Run selector, such as a run ID prefix, workflow slug, or workflow name.
schema:
type: string
example: nightly-build
SettingsView:
name: view
in: query

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

@ -496,7 +496,7 @@ pub(crate) struct InspectArgs {
#[derive(Args)]
pub(crate) struct StoreDumpArgs {
#[command(flatten)]
pub(crate) storage_dir: StorageDirArgs,
pub(crate) server: ServerTargetArgs,
/// Run ID prefix or workflow name
pub(crate) run: String,

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

@ -10,7 +10,6 @@ use fabro_util::printer::Printer;
use crate::args::{ArtifactCommand, ArtifactNamespace, ServerTargetArgs};
use crate::command_context::CommandContext;
use crate::server_client::ServerStoreClient;
use crate::server_runs::ServerSummaryLookup;
#[derive(Clone, Debug, serde::Serialize)]
pub(super) struct ArtifactEntry {
@ -32,11 +31,10 @@ pub(super) async fn resolve_artifacts(
printer: Printer,
) -> Result<(RunId, ServerStoreClient, Vec<ArtifactEntry>)> {
let ctx = CommandContext::for_target(server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(run_selector)?;
let run_id = run.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(run_selector).await?.run_id;
let mut entries = Vec::new();
for entry in lookup.client().list_run_artifacts(&run_id).await? {
for entry in client.list_run_artifacts(&run_id).await? {
if node.is_some_and(|value| entry.node_slug != value) {
continue;
}
@ -62,8 +60,7 @@ pub(super) async fn resolve_artifacts(
.then_with(|| a.relative_path.cmp(&b.relative_path))
});
let client = lookup.client().clone_for_reuse();
Ok((run_id, client, entries))
Ok((run_id, client.clone_for_reuse(), entries))
}
pub(crate) async fn dispatch(

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

@ -17,7 +17,6 @@ use tracing::info;
use crate::args::PrCreateArgs;
use crate::command_context::CommandContext;
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
use crate::shared::repo::ensure_matching_repo_origin;
use crate::user_config;
@ -29,10 +28,9 @@ pub(super) async fn create_command(
printer: Printer,
) -> Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run_id).await?.run_id;
let events = client.list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let state = run_store.state().await?;

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

@ -14,7 +14,6 @@ use fabro_util::printer::Printer;
use crate::args::{PrCommand, PrNamespace, ServerTargetArgs};
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::github::build_github_credentials;
use crate::user_config;
@ -81,10 +80,9 @@ pub(crate) async fn load_pr_record(
printer: Printer,
) -> Result<(PullRequestRecord, fabro_types::RunId)> {
let ctx = CommandContext::for_target(server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(run_id)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(run_id).await?.run_id;
let state = client.get_run_state(&run_id).await?;
let record = state.pull_request.with_context(|| {
format!("No pull request found in store. Create one first with: fabro pr create {run_id}")
})?;

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

@ -10,7 +10,6 @@ use tracing::{debug, info};
use crate::args::{CpArgs, ServerTargetArgs};
use crate::command_context::CommandContext;
use crate::server_client::ServerStoreClient;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::{print_json_pretty, split_run_path};
#[derive(Debug)]
@ -134,9 +133,9 @@ async fn resolve_client_and_run_id(
printer: Printer,
) -> Result<(ServerStoreClient, fabro_types::RunId)> {
let ctx = CommandContext::for_target(server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(run_prefix)?;
Ok((lookup.client().clone_for_reuse(), run.run_id()))
let client = ctx.server().await?;
let run_id = client.resolve_run(run_prefix).await?.run_id;
Ok((client.clone_for_reuse(), run_id))
}
async fn write_sandbox_file(

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

@ -18,7 +18,6 @@ use tracing::{debug, info};
use crate::args::DiffArgs;
use crate::command_context::CommandContext;
use crate::server_client::RunProjection;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
pub(crate) async fn run(
@ -29,10 +28,9 @@ pub(crate) async fn run(
) -> Result<()> {
info!(run_id = %args.run, "Showing diff");
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
let state = client.get_run_state(&run_id).await?;
let patch = resolve_diff(&state, &args)?;

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

@ -10,7 +10,6 @@ use git2::Repository;
use crate::args::ForkArgs;
use crate::command_context::CommandContext;
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
use crate::shared::repo::ensure_matching_repo_origin;
@ -23,14 +22,13 @@ pub(crate) async fn run(
) -> Result<()> {
let repo = Repository::discover(".").context("not in a git repository")?;
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run_id).await?.run_id;
let state = client.get_run_state(&run_id).await?;
let record = state.run.context("Failed to load run record from store")?;
ensure_matching_repo_origin(record.repo_origin_url.as_deref(), "fork")?;
let store = Store::new(repo);
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let events = client.list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;

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

@ -25,7 +25,6 @@ use tracing::{debug, info};
use crate::args::LogsArgs;
use crate::command_context::CommandContext;
use crate::server_client;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::format_usd_micros;
const FOLLOW_TERMINAL_GRACE: Duration = Duration::from_millis(500);
@ -38,11 +37,8 @@ pub(crate) async fn run(
printer: Printer,
) -> Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let client = lookup.client();
let run_id = run.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
info!(run_id = %run_id, "Showing logs");
let since_cutoff = match &args.since {
@ -78,7 +74,7 @@ pub(crate) async fn run(
if args.follow {
follow_store_logs(
client,
client.as_ref(),
&run_id,
if last_seq == 0 { 1 } else { last_seq + 1 },
pretty,

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

@ -6,7 +6,6 @@ use fabro_util::terminal::Styles;
use crate::args::{AttachArgs, RunCommands, RunWorkerArgs, StartArgs};
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
use crate::user_config::settings_layer_with_storage_dir;
@ -59,10 +58,9 @@ pub(crate) async fn dispatch(
}
RunCommands::Start(StartArgs { server, run }) => {
let ctx = CommandContext::for_target(&server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run_info = lookup.resolve(&run)?;
let run_id = run_info.run_id();
start::start_run_with_client(lookup.client(), &run_id, false).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&run).await?.run_id;
start::start_run_with_client(client.as_ref(), &run_id, false).await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&serde_json::json!({ "run_id": run_id }))?;
}
@ -71,11 +69,10 @@ pub(crate) async fn dispatch(
RunCommands::Attach(AttachArgs { server, run }) => {
let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr()));
let ctx = CommandContext::for_target(&server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run_info = lookup.resolve(&run)?;
let run_id = run_info.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(&run).await?.run_id;
let exit_code = attach::attach_run_with_client(
lookup.client(),
client.as_ref(),
&run_id,
false,
styles,

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

@ -6,7 +6,6 @@ use tracing::info;
use crate::args::PreviewArgs;
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
pub(crate) async fn run(
@ -17,13 +16,11 @@ pub(crate) async fn run(
printer: Printer,
) -> Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
let expires_in_secs =
u64::try_from(args.ttl).map_err(|_| anyhow::anyhow!("--ttl must be positive"))?;
let response = lookup
.client()
let response = client
.generate_preview_url(
&run_id,
args.port,

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

@ -5,7 +5,6 @@ use fabro_util::terminal::Styles;
use crate::args::ResumeArgs;
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
/// Resume an interrupted workflow run.
@ -21,11 +20,10 @@ pub(crate) async fn resume_command(
printer: Printer,
) -> anyhow::Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
super::start::start_run_with_client(lookup.client(), &run_id, true).await?;
super::start::start_run_with_client(client.as_ref(), &run_id, true).await?;
let json = cli.output.format == OutputFormat::Json;
if args.detach {
@ -36,7 +34,7 @@ pub(crate) async fn resume_command(
}
} else {
let exit_code = super::attach::attach_run_with_client(
lookup.client(),
client.as_ref(),
&run_id,
true,
styles,
@ -46,7 +44,7 @@ pub(crate) async fn resume_command(
.await?;
if !json {
super::output::print_run_summary_with_client(
lookup.client(),
client.as_ref(),
&run_id,
None,
styles,

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

@ -19,7 +19,6 @@ use crate::args::RewindArgs;
use crate::command_context::CommandContext;
use crate::commands::store::rebuild::rebuild_run_store;
use crate::server_client::ServerStoreClient;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::repo::ensure_matching_repo_origin;
use crate::shared::{color_if, print_json_pretty};
@ -40,14 +39,13 @@ pub(crate) async fn run(
) -> Result<()> {
let repo = Repository::discover(".").context("not in a git repository")?;
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run_id)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run_id).await?.run_id;
let state = client.get_run_state(&run_id).await?;
let record = state.run.context("Failed to load run record from store")?;
ensure_matching_repo_origin(record.repo_origin_url.as_deref(), "rewind")?;
let store = Store::new(repo);
let events = lookup.client().list_run_events(&run_id, None, None).await?;
let events = client.list_run_events(&run_id, None, None).await?;
let run_store = rebuild_run_store(&run_id, &events).await?;
let timeline = build_timeline_or_rebuild(&store, Some(&run_store), &run_id).await?;
@ -69,7 +67,7 @@ pub(crate) async fn run(
push: !args.no_push,
})?;
let entry = timeline.resolve(&target)?;
reset_rewound_run_state(lookup.client(), &store, &run_id, entry).await?;
reset_rewound_run_state(client.as_ref(), &store, &run_id, entry).await?;
let run_id_string = run_id.to_string();

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

@ -6,7 +6,6 @@ use tracing::info;
use crate::args::{SshArgs, require_no_json_override};
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::print_json_pretty;
pub(crate) async fn run(
@ -21,13 +20,9 @@ pub(crate) async fn run(
}
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let ssh = lookup
.client()
.create_run_ssh_access(&run_id, args.ttl)
.await?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
let ssh = client.create_run_ssh_access(&run_id, args.ttl).await?;
info!(run_id = %args.run, ttl_minutes = args.ttl, "Creating SSH access");

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

@ -22,7 +22,6 @@ use tracing::info;
use crate::args::WaitArgs;
use crate::command_context::CommandContext;
use crate::server_runs::ServerSummaryLookup;
use crate::shared::{format_duration_ms, format_usd_micros};
pub(crate) async fn run(
@ -33,11 +32,8 @@ pub(crate) async fn run(
printer: Printer,
) -> Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run_info = lookup.resolve(&args.run)?;
let client = lookup.client();
let run_id = run_info.run_id();
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
info!(run_id = %run_id, "Waiting for run to complete");
let deadline = args

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

@ -8,7 +8,7 @@ use serde::Serialize;
use crate::args::InspectArgs;
use crate::command_context::CommandContext;
use crate::server_client::RunProjection;
use crate::server_runs::{ServerRunSummaryInfo, ServerSummaryLookup};
use crate::server_runs::ServerRunSummaryInfo;
#[derive(Debug, Serialize)]
pub(crate) struct InspectOutput {
@ -28,10 +28,10 @@ pub(crate) async fn run(
printer: Printer,
) -> Result<()> {
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let lookup = ServerSummaryLookup::from_client(ctx.server().await?).await?;
let run = lookup.resolve(&args.run)?;
let client = ctx.server().await?;
let run = ServerRunSummaryInfo::from_summary(client.resolve_run(&args.run).await?);
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let state = client.get_run_state(&run_id).await?;
let output = inspect_run_state(&run, state);
let json = serde_json::to_string_pretty(&[output])?;
fabro_util::printout!(printer, "{json}");

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

@ -12,31 +12,31 @@ use bytes::Bytes;
use fabro_store::{ArtifactStore, RunDatabase};
use fabro_store::{EventEnvelope, RunProjection, StageId};
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::OutputFormat;
use fabro_types::settings::cli::{CliLayer, OutputFormat};
use fabro_types::{RunBlobId, RunId};
use fabro_util::printer::Printer;
use fabro_workflow::run_dump::RunDump;
use futures::future::BoxFuture;
#[cfg(test)]
use serde::de::DeserializeOwned;
use tokio::task::spawn_blocking;
use super::run_export::StoreRunExport;
use crate::args::StoreDumpArgs;
use crate::command_context::CommandContext;
use crate::server_client::ServerStoreClient;
use crate::server_runs::ServerRunLookup;
use crate::shared::{absolute_or_current, print_json_pretty};
use crate::user_config::{load_settings_with_storage_dir, storage_dir};
pub(crate) async fn dump_command(
args: &StoreDumpArgs,
cli: &CliSettings,
cli_layer: &CliLayer,
printer: Printer,
) -> Result<()> {
let cli_settings = load_settings_with_storage_dir(args.storage_dir.as_deref())?;
let lookup = ServerRunLookup::connect(&storage_dir(&cli_settings)?).await?;
let run = lookup.resolve(&args.run)?;
let run_id = run.run_id();
let state = lookup.client().get_run_state(&run_id).await?;
let source = ServerDumpSource::new(lookup.client(), &run_id);
let ctx = CommandContext::for_target(&args.server, printer, cli.clone(), cli_layer)?;
let client = ctx.server().await?;
let run_id = client.resolve_run(&args.run).await?.run_id;
let state = client.get_run_state(&run_id).await?;
let source = ServerDumpSource::new(client.as_ref(), &run_id);
let file_count = export_run_from_source(&source, &state, &args.output).await?;
if cli.output.format == OutputFormat::Json {
print_json_pretty(&serde_json::json!({
@ -248,7 +248,7 @@ async fn write_run_dump(
output_dir: &Path,
) -> Result<usize> {
let events = source.list_events().await?;
let mut dump = RunDump::from_store_state_and_events(state, &events)?;
let mut dump = StoreRunExport::from_store_state_and_events(state, &events)?;
dump.hydrate_referenced_blobs_with_reader(|blob_id| source.read_blob(blob_id))
.await?;
@ -257,7 +257,10 @@ async fn write_run_dump(
dump.add_artifact_bytes(&artifact.stage_id, &artifact.relative_path, artifact.data)?;
}
dump.write_to_dir(output_dir)
let output_dir = output_dir.to_path_buf();
spawn_blocking(move || dump.write_to_dir(&output_dir))
.await
.map_err(|err| anyhow::anyhow!("run dump write task failed: {err}"))?
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]

5
lib/crates/fabro-cli/src/commands/store/mod.rs
View file

@ -1,8 +1,10 @@
pub(crate) mod dump;
pub(crate) mod rebuild;
mod run_export;
use anyhow::Result;
use fabro_types::settings::CliSettings;
use fabro_types::settings::cli::CliLayer;
use fabro_util::printer::Printer;
use crate::args::{StoreCommand, StoreNamespace};
@ -10,9 +12,10 @@ use crate::args::{StoreCommand, StoreNamespace};
pub(crate) async fn dispatch(
ns: StoreNamespace,
cli: &CliSettings,
cli_layer: &CliLayer,
printer: Printer,
) -> Result<()> {
match ns.command {
StoreCommand::Dump(args) => dump::dump_command(&args, cli, printer).await,
StoreCommand::Dump(args) => dump::dump_command(&args, cli, cli_layer, printer).await,
}
}

372
lib/crates/fabro-cli/src/commands/store/run_export.rs Normal file
View file

@ -0,0 +1,372 @@
#![expect(
clippy::disallowed_methods,
reason = "CLI-owned export writer uses sync std::fs for final local materialization"
)]
use std::collections::HashMap;
#[expect(
clippy::disallowed_types,
reason = "in-memory Vec<u8>::write_all for jsonl serialization; no filesystem or network I/O"
)]
use std::io::Write;
use std::path::{Component, Path, PathBuf};
use anyhow::{Context, Result, bail};
use bytes::Bytes;
use fabro_store::{EventEnvelope, RunProjection, StageId};
use fabro_types::{RunBlobId, parse_blob_ref, parse_legacy_blob_file_ref};
use futures::future::BoxFuture;
#[derive(Debug, Clone)]
pub(super) struct StoreRunExport {
entries: Vec<StoreRunExportEntry>,
}
#[derive(Debug, Clone)]
struct StoreRunExportEntry {
path: String,
contents: StoreRunExportContents,
}
#[derive(Debug, Clone)]
enum StoreRunExportContents {
Text(String),
Json(serde_json::Value),
Bytes(Vec<u8>),
}
impl StoreRunExport {
pub(super) fn from_store_state_and_events(
state: &RunProjection,
events: &[EventEnvelope],
) -> Result<Self> {
let mut entries = Vec::new();
if let Some(record) = state.run.as_ref() {
push_json_entry(&mut entries, "run.json", record);
}
if let Some(record) = state.start.as_ref() {
push_json_entry(&mut entries, "start.json", record);
}
if let Some(record) = state.status.as_ref() {
push_json_entry(&mut entries, "status.json", record);
}
if let Some(record) = state.checkpoint.as_ref() {
push_json_entry(&mut entries, "checkpoint.json", record);
}
if let Some(record) = state.conclusion.as_ref() {
push_json_entry(&mut entries, "conclusion.json", record);
}
if let Some(record) = state.retro.as_ref() {
push_json_entry(&mut entries, "retro.json", record);
}
if let Some(graph_source) = state.graph_source.as_ref() {
entries.push(StoreRunExportEntry::text(
"graph.fabro",
graph_source.clone(),
));
}
if let Some(record) = state.sandbox.as_ref() {
push_json_entry(&mut entries, "sandbox.json", record);
}
let mut node_keys: Vec<_> = state.iter_nodes().map(|(node, _)| node.clone()).collect();
node_keys.sort();
for node_key in &node_keys {
let node = state
.node(node_key)
.with_context(|| format!("missing node {node_key:?} in projection"))?;
let node_id_segment = validate_single_path_segment("node id", node_key.node_id())?;
let base = PathBuf::from("nodes")
.join(node_id_segment)
.join(format!("visit-{}", node_key.visit()));
if let Some(prompt) = node.prompt.as_ref() {
entries.push(StoreRunExportEntry::text_path(
&base.join("prompt.md"),
prompt.clone(),
));
}
if let Some(response) = node.response.as_ref() {
entries.push(StoreRunExportEntry::text_path(
&base.join("response.md"),
response.clone(),
));
}
if let Some(status) = node.status.as_ref() {
push_json_entry_path(&mut entries, &base.join("status.json"), status);
}
if let Some(stdout) = node.stdout.as_ref() {
entries.push(StoreRunExportEntry::text_path(
&base.join("stdout.log"),
stdout.clone(),
));
}
if let Some(stderr) = node.stderr.as_ref() {
entries.push(StoreRunExportEntry::text_path(
&base.join("stderr.log"),
stderr.clone(),
));
}
}
if let Some(prompt) = state.retro_prompt.as_ref() {
entries.push(StoreRunExportEntry::text("retro/prompt.md", prompt.clone()));
}
if let Some(response) = state.retro_response.as_ref() {
entries.push(StoreRunExportEntry::text(
"retro/response.md",
response.clone(),
));
}
let mut events_jsonl = Vec::new();
for event in events {
serde_json::to_writer(&mut events_jsonl, event)?;
events_jsonl.write_all(b"\n")?;
}
entries.push(StoreRunExportEntry::bytes("events.jsonl", events_jsonl));
for (seq, checkpoint) in &state.checkpoints {
push_json_entry_path(
&mut entries,
&PathBuf::from("checkpoints").join(format!("{seq:04}.json")),
checkpoint,
);
}
Ok(Self { entries })
}
pub(super) fn add_artifact_bytes(
&mut self,
stage_id: &StageId,
filename: &str,
data: Vec<u8>,
) -> Result<()> {
let path = artifact_dump_path(stage_id, filename)?;
self.entries
.push(StoreRunExportEntry::bytes_path(&path, data));
Ok(())
}
pub(super) async fn hydrate_referenced_blobs_with_reader<'a, F>(
&mut self,
mut read_blob: F,
) -> Result<()>
where
F: FnMut(RunBlobId) -> BoxFuture<'a, Result<Option<Bytes>>>,
{
let mut cache = HashMap::new();
for entry in &mut self.entries {
if let StoreRunExportContents::Json(value) = &mut entry.contents {
let mut blob_ids = Vec::new();
collect_blob_refs_in_value(value, &mut blob_ids);
for blob_id in blob_ids {
if cache.contains_key(&blob_id) {
continue;
}
let blob = read_blob(blob_id)
.await?
.with_context(|| format!("blob {blob_id:?} is missing from the store"))?;
let hydrated: serde_json::Value = serde_json::from_slice(&blob)
.with_context(|| format!("blob {blob_id:?} is not valid JSON"))?;
cache.insert(blob_id, hydrated);
}
replace_blob_refs_in_value(value, &cache)?;
}
}
Ok(())
}
pub(super) fn write_to_dir(&self, root: &Path) -> Result<usize> {
for entry in &self.entries {
entry.write_to_dir(root)?;
}
Ok(self.entries.len())
}
}
impl StoreRunExportEntry {
fn text(path: impl Into<String>, contents: String) -> Self {
Self {
path: path.into(),
contents: StoreRunExportContents::Text(contents),
}
}
fn text_path(path: &Path, contents: String) -> Self {
Self {
path: path_to_string(path),
contents: StoreRunExportContents::Text(contents),
}
}
fn json(path: impl Into<String>, contents: serde_json::Value) -> Self {
Self {
path: path.into(),
contents: StoreRunExportContents::Json(contents),
}
}
fn json_path(path: &Path, contents: serde_json::Value) -> Self {
Self {
path: path_to_string(path),
contents: StoreRunExportContents::Json(contents),
}
}
fn bytes(path: impl Into<String>, contents: Vec<u8>) -> Self {
Self {
path: path.into(),
contents: StoreRunExportContents::Bytes(contents),
}
}
fn bytes_path(path: &Path, contents: Vec<u8>) -> Self {
Self {
path: path_to_string(path),
contents: StoreRunExportContents::Bytes(contents),
}
}
fn write_to_dir(&self, root: &Path) -> Result<()> {
let relative = validate_relative_path("run dump path", &self.path)?;
let path = root.join(relative);
ensure_parent_dir(&path)?;
std::fs::write(&path, self.contents.to_bytes()?)
.with_context(|| format!("failed to write {}", path.display()))?;
Ok(())
}
}
impl StoreRunExportContents {
fn to_bytes(&self) -> Result<Vec<u8>> {
match self {
Self::Text(value) => Ok(value.as_bytes().to_vec()),
Self::Json(value) => Ok(serde_json::to_vec_pretty(value)?),
Self::Bytes(value) => Ok(value.clone()),
}
}
}
fn push_json_entry<T>(entries: &mut Vec<StoreRunExportEntry>, path: &str, value: &T)
where
T: serde::Serialize,
{
if let Ok(value) = serde_json::to_value(value) {
entries.push(StoreRunExportEntry::json(path, value));
}
}
fn push_json_entry_path<T>(entries: &mut Vec<StoreRunExportEntry>, path: &Path, value: &T)
where
T: serde::Serialize,
{
if let Ok(value) = serde_json::to_value(value) {
entries.push(StoreRunExportEntry::json_path(path, value));
}
}
fn path_to_string(path: &Path) -> String {
path.to_string_lossy().into_owned()
}
fn validate_single_path_segment(kind: &str, value: &str) -> Result<PathBuf> {
let path = validate_relative_path(kind, value)?;
if path.components().count() != 1 {
bail!("{kind} {value:?} must be a single path segment");
}
Ok(path)
}
fn validate_relative_path(kind: &str, value: &str) -> Result<PathBuf> {
let mut normalized = PathBuf::new();
for component in Path::new(value).components() {
match component {
Component::Normal(part) => normalized.push(part),
Component::CurDir => {}
Component::ParentDir | Component::RootDir | Component::Prefix(_) => {
bail!("{kind} {value:?} must be a relative path without '..'");
}
}
}
if normalized.as_os_str().is_empty() {
bail!("{kind} {value:?} must not be empty");
}
Ok(normalized)
}
fn collect_blob_refs_in_value(value: &serde_json::Value, blob_ids: &mut Vec<RunBlobId>) {
match value {
serde_json::Value::String(current) => {
if let Some(blob_id) =
parse_blob_ref(current).or_else(|| parse_legacy_blob_file_ref(current))
{
blob_ids.push(blob_id);
}
}
serde_json::Value::Array(items) => {
for item in items {
collect_blob_refs_in_value(item, blob_ids);
}
}
serde_json::Value::Object(map) => {
for item in map.values() {
collect_blob_refs_in_value(item, blob_ids);
}
}
serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => {}
}
}
fn replace_blob_refs_in_value(
value: &mut serde_json::Value,
cache: &HashMap<RunBlobId, serde_json::Value>,
) -> Result<()> {
match value {
serde_json::Value::String(current) => {
let Some(blob_id) =
parse_blob_ref(current).or_else(|| parse_legacy_blob_file_ref(current))
else {
return Ok(());
};
let hydrated = cache
.get(&blob_id)
.cloned()
.with_context(|| format!("blob {blob_id:?} is missing from the hydration cache"))?;
*value = hydrated;
}
serde_json::Value::Array(items) => {
for item in items {
replace_blob_refs_in_value(item, cache)?;
}
}
serde_json::Value::Object(map) => {
for item in map.values_mut() {
replace_blob_refs_in_value(item, cache)?;
}
}
serde_json::Value::Null | serde_json::Value::Bool(_) | serde_json::Value::Number(_) => {}
}
Ok(())
}
fn artifact_dump_path(stage_id: &StageId, filename: &str) -> Result<PathBuf> {
let node_id_segment = validate_single_path_segment("node id", stage_id.node_id())?;
let filename_path = validate_relative_path("artifact filename", filename)?;
Ok(PathBuf::from("artifacts")
.join("nodes")
.join(node_id_segment)
.join(format!("visit-{}", stage_id.visit()))
.join(filename_path))
}
fn ensure_parent_dir(path: &Path) -> Result<()> {
let parent = path
.parent()
.with_context(|| format!("path {} has no parent", path.display()))?;
std::fs::create_dir_all(parent)
.with_context(|| format!("failed to create {}", parent.display()))?;
Ok(())
}

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

@ -246,7 +246,9 @@ async fn main_inner() -> (String, Result<()>) {
Commands::Artifact(ns) => {
commands::artifact::dispatch(ns, &cli_settings, &cli_layer, printer).await?;
}
Commands::Store(ns) => commands::store::dispatch(ns, &cli_settings, printer).await?,
Commands::Store(ns) => {
commands::store::dispatch(ns, &cli_settings, &cli_layer, printer).await?;
}
Commands::RunsCmd(cmd) => {
commands::runs::dispatch(cmd, &cli_settings, &cli_layer, printer).await?;
}

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

@ -85,6 +85,7 @@ impl RunAttachEventStream {
pub(crate) use fabro_store::RunProjection;
#[cfg(test)]
pub(crate) async fn connect_server(storage_dir: &Path) -> Result<ServerStoreClient> {
connect_api_client_bundle(storage_dir).await
}
@ -637,6 +638,17 @@ impl ServerStoreClient {
convert_type(response.into_inner())
}
pub(crate) async fn resolve_run(&self, selector: &str) -> Result<RunSummary> {
let response = self
.client
.resolve_run()
.selector(selector.to_string())
.send()
.await
.map_err(map_api_error)?;
convert_type(response.into_inner())
}
pub(crate) async fn get_run_state(&self, run_id: &RunId) -> Result<RunProjection> {
let response = self
.client

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

@ -1,45 +1,12 @@
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Result, bail};
use chrono::{DateTime, Utc};
use fabro_store::RunSummary;
use fabro_types::{RunId, RunStatus, StatusReason};
use fabro_workflow::run_lookup::{RunInfo, resolve_run_from_summaries, scratch_base};
use crate::server_client::{self, ServerStoreClient};
pub(crate) struct ServerRunLookup {
client: ServerStoreClient,
scratch_base: PathBuf,
summaries: Vec<RunSummary>,
}
impl ServerRunLookup {
pub(crate) async fn connect(storage_dir: &Path) -> Result<Self> {
Self::connect_from_scratch_base(&scratch_base(storage_dir)).await
}
pub(crate) async fn connect_from_scratch_base(scratch_base: &Path) -> Result<Self> {
let storage_dir = scratch_base.parent().unwrap_or(scratch_base);
let client = server_client::connect_server(storage_dir).await?;
let summaries = client.list_store_runs().await?;
Ok(Self {
client,
scratch_base: scratch_base.to_path_buf(),
summaries,
})
}
pub(crate) fn client(&self) -> &ServerStoreClient {
&self.client
}
pub(crate) fn resolve(&self, selector: &str) -> Result<RunInfo> {
resolve_run_from_summaries(&self.summaries, &self.scratch_base, selector)
}
}
use crate::server_client::ServerStoreClient;
#[derive(Debug, Clone)]
pub(crate) struct ServerRunSummaryInfo {
@ -47,6 +14,10 @@ pub(crate) struct ServerRunSummaryInfo {
}
impl ServerRunSummaryInfo {
pub(crate) fn from_summary(summary: RunSummary) -> Self {
Self { summary }
}
pub(crate) fn run_id(&self) -> RunId {
self.summary.run_id
}
@ -113,7 +84,7 @@ impl ServerSummaryLookup {
let summaries = client.list_store_runs().await?;
let mut runs = summaries
.into_iter()
.map(|summary| ServerRunSummaryInfo { summary })
.map(ServerRunSummaryInfo::from_summary)
.collect::<Vec<_>>();
runs.sort_by(|a, b| {
b.start_time_dt()
@ -130,10 +101,6 @@ impl ServerSummaryLookup {
pub(crate) fn runs(&self) -> &[ServerRunSummaryInfo] {
&self.runs
}
pub(crate) fn resolve(&self, selector: &str) -> Result<ServerRunSummaryInfo> {
resolve_server_run_from_infos(&self.runs, selector)
}
}
pub(crate) fn resolve_server_run_from_summaries(

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

@ -1,10 +1,35 @@
use fabro_test::{fabro_snapshot, test_context};
use httpmock::MockServer;
use insta::assert_snapshot;
use serde_json::json;
use super::support::{
compact_git_inspect, compact_inspect, run_success, setup_completed_fast_dry_run,
setup_created_fast_dry_run, setup_git_backed_changed_run,
};
use crate::support::unique_run_id;
fn remote_run_summary(run_id: &str, status: &str) -> serde_json::Value {
json!({
"run_id": run_id,
"workflow_name": "Nightly Build",
"workflow_slug": "nightly-build",
"goal": "Inspect remote state",
"title": "Inspect remote state",
"labels": {},
"host_repo_path": "/srv/repo",
"repository": { "name": "repo" },
"start_time": "2026-04-19T12:00:00Z",
"created_at": "2026-04-19T12:00:00Z",
"status": status,
"status_reason": null,
"blocked_reason": null,
"pending_control": null,
"duration_ms": null,
"elapsed_secs": null,
"total_usd_micros": null
})
}
#[test]
fn help() {
@ -34,6 +59,59 @@ fn help() {
");
}
#[test]
fn inspect_resolves_selector_via_server_endpoint() {
let context = test_context!();
let server = MockServer::start();
let run_id = unique_run_id();
let summary = remote_run_summary(run_id.as_str(), "succeeded");
let resolve_run = server.mock(|when, then| {
when.method("GET")
.path("/api/v1/runs/resolve")
.query_param("selector", "nightly-build");
then.status(200)
.header("content-type", "application/json")
.body(summary.to_string());
});
let run_state = server.mock(|when, then| {
when.method("GET")
.path(format!("/api/v1/runs/{}/state", run_id.as_str()));
then.status(200)
.header("content-type", "application/json")
.body(r#"{"nodes": {}}"#);
});
let mut cmd = context.command();
cmd.args([
"inspect",
"--server",
&format!("{}/api/v1", server.base_url()),
"nightly-build",
]);
fabro_snapshot!(context.filters(), cmd, @r###"
success: true
exit_code: 0
----- stdout -----
[
{
"run_id": "[ULID]",
"status": "succeeded",
"run_record": null,
"start_record": null,
"conclusion": null,
"checkpoint": null,
"sandbox": null
}
]
----- stderr -----
"###);
resolve_run.assert();
run_state.assert();
}
#[test]
fn inspect_created_run_shows_run_record_without_start_or_conclusion() {
let context = test_context!();

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

@ -9,8 +9,8 @@ use std::time::Duration;
use fabro_test::{fabro_snapshot, test_context};
use insta::assert_snapshot;
use super::support::setup_completed_dry_run;
use crate::support::unique_run_id;
use super::support::{local_dev_token, server_target, setup_completed_dry_run};
use crate::support::{LightweightCli, unique_run_id};
#[test]
fn help() {
@ -29,18 +29,50 @@ fn help() {
<RUN> Run ID prefix or workflow name
Options:
--json Output as JSON [env: FABRO_JSON=]
--storage-dir <STORAGE_DIR> Local storage directory (default: ~/.fabro/storage) [env: FABRO_STORAGE_DIR=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-o, --output <OUTPUT> Output directory (must not exist or be empty)
--no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true]
--quiet Suppress non-essential output [env: FABRO_QUIET=]
--verbose Enable verbose output [env: FABRO_VERBOSE=]
-h, --help Print help
--json Output as JSON [env: FABRO_JSON=]
--server <SERVER> Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=]
--debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=]
-o, --output <OUTPUT> Output directory (must not exist or be empty)
--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 store_dump_accepts_server_target_from_separate_home() {
let context = test_context!();
let run = setup_completed_dry_run(&context);
let cli = LightweightCli::new();
let output_dir = context.temp_dir.join("remote-export");
let server = server_target(&context.storage_dir);
let mut cmd = cli.command();
cmd.args([
"store",
"dump",
"--server",
&server,
"--output",
output_dir.to_str().unwrap(),
&run.run_id,
]);
if let Some(dev_token) = local_dev_token(&context.storage_dir) {
cmd.env("FABRO_DEV_TOKEN", dev_token);
}
let output = cmd.output().expect("store dump should execute");
assert!(
output.status.success(),
"store dump via remote server target failed\nstdout:\n{}\nstderr:\n{}",
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
assert!(output_dir.join("checkpoint.json").is_file());
}
#[test]
fn store_dump_exports_large_command_output_backed_by_blob_refs() {
let context = test_context!();

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

@ -145,11 +145,13 @@ fn wait_blocked_run_times_out_without_treating_it_as_terminal() {
let server = MockServer::start();
let summary = remote_run_summary(run_id.as_str(), "blocked");
let list_runs = server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
let resolve_run = server.mock(|when, then| {
when.method("GET")
.path("/api/v1/runs/resolve")
.query_param("selector", run_id.as_str());
then.status(200)
.header("content-type", "application/json")
.body(json!({ "data": [summary.clone()], "meta": { "has_more": false } }).to_string());
.body(summary.clone().to_string());
});
let retrieve_run = server.mock(|when, then| {
when.method("GET")
@ -185,7 +187,7 @@ fn wait_blocked_run_times_out_without_treating_it_as_terminal() {
----- stderr -----
error: Timed out after 1s waiting for run '[ULID]'
");
list_runs.assert();
resolve_run.assert();
assert!(
retrieve_run.calls() > 0,
"wait should keep polling the blocked run summary until timeout"

80
lib/crates/fabro-cli/tests/it/scenario/smoke.rs
View file

@ -233,29 +233,31 @@ fn attach_smoke_covers_arg_validation_and_remote_server_behaviors() {
let success_server = MockServer::start();
let success_run_id = unique_run_id();
let list_mock = success_server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
let resolve_mock = success_server.mock(|when, then| {
when.method("GET")
.path("/api/v1/runs/resolve")
.query_param("selector", success_run_id.as_str());
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"data": [{
"run_id": success_run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"title": "Remote output",
"labels": {},
"host_repo_path": null,
"repository": { "name": "unknown" },
"start_time": "2026-04-05T12:00:00Z",
"created_at": "2026-04-05T12:00:00Z",
"status": "running",
"status_reason": null,
"duration_ms": 12,
"total_usd_micros": null
}],
"meta": { "has_more": false }
"run_id": success_run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"title": "Remote output",
"labels": {},
"host_repo_path": null,
"repository": { "name": "unknown" },
"start_time": "2026-04-05T12:00:00Z",
"created_at": "2026-04-05T12:00:00Z",
"status": "running",
"status_reason": null,
"blocked_reason": null,
"pending_control": null,
"duration_ms": 12,
"elapsed_secs": 0,
"total_usd_micros": null
})
.to_string(),
);
@ -324,7 +326,7 @@ fn attach_smoke_covers_arg_validation_and_remote_server_behaviors() {
String::from_utf8_lossy(&success_output.stdout),
String::from_utf8_lossy(&success_output.stderr)
);
list_mock.assert();
resolve_mock.assert();
attach_mock.assert();
let success_stdout = String::from_utf8(success_output.stdout).expect("stdout should be UTF-8");
assert!(
@ -336,28 +338,30 @@ fn attach_smoke_covers_arg_validation_and_remote_server_behaviors() {
let eof_run_id = unique_run_id();
eof_server.mock(|when, then| {
when.method("GET").path("/api/v1/runs");
when.method("GET")
.path("/api/v1/runs/resolve")
.query_param("selector", eof_run_id.as_str());
then.status(200)
.header("Content-Type", "application/json")
.body(
serde_json::json!({
"data": [{
"run_id": eof_run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"title": "Remote output",
"labels": {},
"host_repo_path": null,
"repository": { "name": "unknown" },
"start_time": "2026-04-05T12:00:00Z",
"created_at": "2026-04-05T12:00:00Z",
"status": "running",
"status_reason": null,
"duration_ms": 12,
"total_usd_micros": null
}],
"meta": { "has_more": false }
"run_id": eof_run_id,
"workflow_name": "Remote Workflow",
"workflow_slug": "remote-workflow",
"goal": "Remote output",
"title": "Remote output",
"labels": {},
"host_repo_path": null,
"repository": { "name": "unknown" },
"start_time": "2026-04-05T12:00:00Z",
"created_at": "2026-04-05T12:00:00Z",
"status": "running",
"status_reason": null,
"blocked_reason": null,
"pending_control": null,
"duration_ms": 12,
"elapsed_secs": 0,
"total_usd_micros": null
})
.to_string(),
);

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

@ -15,6 +15,7 @@ use serde_json::json;
use crate::error::ApiError;
use crate::jwt_auth::AuthenticatedService;
use crate::run_selector::{ResolveRunError, resolve_run_by_selector};
use crate::server::{AppState, PaginationParams};
use crate::settings_view;
@ -77,6 +78,30 @@ pub(crate) async fn create_run_stub(
.into_response()
}
pub(crate) async fn resolve_run(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
Query(params): Query<ResolveRunParams>,
) -> Response {
let runs = runs::summaries();
match resolve_run_by_selector(
&runs,
&params.selector,
|run| run.run_id.clone(),
|run| run.workflow_slug.clone(),
|run| run.workflow_name.clone(),
|run| run.created_at,
) {
Ok(run) => (StatusCode::OK, Json(run.clone())).into_response(),
Err(ResolveRunError::InvalidSelector | ResolveRunError::AmbiguousPrefix { .. }) => {
ApiError::bad_request("Run selector could not be resolved.").into_response()
}
Err(ResolveRunError::NotFound { .. }) => {
ApiError::not_found("Run not found.").into_response()
}
}
}
pub(crate) async fn start_run_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,
@ -205,6 +230,11 @@ pub(crate) async fn get_run_status(
}
}
#[derive(Debug, serde::Deserialize)]
pub(crate) struct ResolveRunParams {
selector: String,
}
pub(crate) async fn get_questions_stub(
_auth: AuthenticatedService,
State(_state): State<Arc<AppState>>,

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

@ -14,6 +14,7 @@ pub mod install;
pub mod ip_allowlist;
pub mod jwt_auth;
mod run_manifest;
mod run_selector;
pub mod security_headers;
pub mod serve;
pub mod server;

87
lib/crates/fabro-server/src/run_selector.rs Normal file
View file

@ -0,0 +1,87 @@
use std::fmt;
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ResolveRunError {
InvalidSelector,
AmbiguousPrefix {
selector: String,
matches: Vec<String>,
},
NotFound {
selector: String,
},
}
impl fmt::Display for ResolveRunError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidSelector => write!(f, "Run selector must not be empty."),
Self::AmbiguousPrefix { selector, matches } => write!(
f,
"Ambiguous prefix '{selector}': {} runs match: {}",
matches.len(),
matches.join(", ")
),
Self::NotFound { selector } => write!(
f,
"No run found matching '{selector}' (tried run ID prefix and workflow name)"
),
}
}
}
pub(crate) fn resolve_run_by_selector<'a, T, FRunId, FWorkflowSlug, FWorkflowName, FCreatedAt, K>(
runs: &'a [T],
selector: &str,
run_id: FRunId,
workflow_slug: FWorkflowSlug,
workflow_name: FWorkflowName,
created_at: FCreatedAt,
) -> Result<&'a T, ResolveRunError>
where
FRunId: Fn(&T) -> String,
FWorkflowSlug: Fn(&T) -> Option<String>,
FWorkflowName: Fn(&T) -> Option<String>,
FCreatedAt: Fn(&T) -> K,
K: Ord,
{
let selector = selector.trim();
if selector.is_empty() {
return Err(ResolveRunError::InvalidSelector);
}
let id_matches: Vec<_> = runs
.iter()
.filter(|run| run_id(run).starts_with(selector))
.collect();
match id_matches.len() {
1 => return Ok(id_matches[0]),
count if count > 1 => {
return Err(ResolveRunError::AmbiguousPrefix {
selector: selector.to_string(),
matches: id_matches.iter().map(|run| run_id(run)).collect(),
});
}
_ => {}
}
let selector_lower = selector.to_lowercase();
let selector_collapsed = collapse_separators(&selector_lower);
runs.iter()
.filter(|run| {
workflow_slug(run).is_some_and(|slug| slug.to_lowercase() == selector_lower)
|| workflow_name(run).is_some_and(|name| {
let name_lower = name.to_lowercase();
name_lower.contains(&selector_lower)
|| collapse_separators(&name_lower).contains(&selector_collapsed)
})
})
.max_by_key(|run| created_at(run))
.ok_or_else(|| ResolveRunError::NotFound {
selector: selector.to_string(),
})
}
fn collapse_separators(value: &str) -> String {
value.chars().filter(|c| *c != '-' && *c != '_').collect()
}

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

@ -115,6 +115,7 @@ use crate::ip_allowlist::{IpAllowlistConfig, ip_allowlist_middleware};
use crate::jwt_auth::{
AuthMode, AuthenticatedService, AuthenticatedSubject, authenticate_service_parts,
};
use crate::run_selector::{ResolveRunError, resolve_run_by_selector};
use crate::server_secrets::{
LlmClientResult, ProviderCredentials, ServerSecrets, auth_issue_message,
};
@ -1007,6 +1008,7 @@ pub fn build_router_with_options(
fn demo_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(demo::list_runs).post(demo::create_run_stub))
.route("/runs/resolve", get(demo::resolve_run))
.route("/boards/runs", get(demo::list_board_runs))
.route("/preflight", post(run_preflight))
.route("/graph/render", post(render_graph_from_manifest))
@ -1087,6 +1089,7 @@ fn demo_routes() -> Router<Arc<AppState>> {
fn real_routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs", get(list_runs).post(create_run))
.route("/runs/resolve", get(resolve_run))
.route("/preflight", post(run_preflight))
.route("/graph/render", post(render_graph_from_manifest))
.route("/attach", get(attach_events))
@ -2807,6 +2810,50 @@ async fn list_runs(
}
}
#[derive(Debug, serde::Deserialize)]
struct ResolveRunQuery {
selector: String,
}
async fn resolve_run(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
Query(query): Query<ResolveRunQuery>,
) -> Response {
let runs = match state
.store
.list_runs(&fabro_store::ListRunsQuery::default())
.await
{
Ok(runs) => runs,
Err(err) => {
return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
.into_response();
}
};
match resolve_run_by_selector(
&runs,
&query.selector,
|run| run.run_id.to_string(),
|run| run.workflow_slug.clone(),
|run| run.workflow_name.clone(),
|run| run.run_id.created_at(),
) {
Ok(run) => (
StatusCode::OK,
Json(summary_to_api_run_summary(run.clone())),
)
.into_response(),
Err(err @ (ResolveRunError::InvalidSelector | ResolveRunError::AmbiguousPrefix { .. })) => {
ApiError::bad_request(err.to_string()).into_response()
}
Err(err @ ResolveRunError::NotFound { .. }) => {
ApiError::not_found(err.to_string()).into_response()
}
}
}
async fn delete_run(
_auth: AuthenticatedService,
State(state): State<Arc<AppState>>,
@ -7286,16 +7333,16 @@ type = "http"
assert_eq!(answer.value, AnswerValue::Cancelled);
}
fn minimal_manifest_json(dot_source: &str) -> serde_json::Value {
fn manifest_json(target_path: &str, dot_source: &str) -> serde_json::Value {
serde_json::json!({
"version": 1,
"cwd": "/tmp",
"target": {
"identifier": "workflow.fabro",
"path": "workflow.fabro",
"identifier": target_path,
"path": target_path,
},
"workflows": {
"workflow.fabro": {
target_path: {
"source": dot_source,
"files": {},
},
@ -7303,10 +7350,18 @@ type = "http"
})
}
fn minimal_manifest_json(dot_source: &str) -> serde_json::Value {
manifest_json("workflow.fabro", dot_source)
}
fn manifest_body(dot_source: &str) -> Body {
Body::from(serde_json::to_string(&minimal_manifest_json(dot_source)).unwrap())
}
fn manifest_body_for(target_path: &str, dot_source: &str) -> Body {
Body::from(serde_json::to_string(&manifest_json(target_path, dot_source)).unwrap())
}
async fn create_run(app: &Router, dot_source: &str) -> String {
let req = Request::builder()
.method("POST")
@ -7319,6 +7374,29 @@ type = "http"
body["id"].as_str().unwrap().to_string()
}
async fn create_run_for_target(app: &Router, target_path: &str, dot_source: &str) -> String {
let req = Request::builder()
.method("POST")
.uri(api("/runs"))
.header("content-type", "application/json")
.body(manifest_body_for(target_path, dot_source))
.unwrap();
let response = app.clone().oneshot(req).await.unwrap();
let body = body_json(response.into_body()).await;
body["id"].as_str().unwrap().to_string()
}
fn named_workflow_dot(name: &str, goal: &str) -> String {
format!(
r#"digraph {name} {{
graph [goal="{goal}"]
start [shape=Mdiamond]
exit [shape=Msquare]
start -> exit
}}"#
)
}
fn multipart_body(
boundary: &str,
manifest: &serde_json::Value,
@ -7692,6 +7770,144 @@ slug = "fabro"
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn resolve_run_returns_unique_run_id_prefix_match() {
let app = test_app_with();
let run_id = create_run(&app, MINIMAL_DOT).await;
let selector = &run_id[..8];
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri(api(&format!("/runs/resolve?selector={selector}")))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert_eq!(body["run_id"], run_id);
}
#[tokio::test]
async fn resolve_run_returns_bad_request_for_ambiguous_prefix() {
let app = test_app_with();
let run_id_a = create_run(&app, MINIMAL_DOT).await;
let run_id_b = create_run(&app, MINIMAL_DOT).await;
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri(api("/runs/resolve?selector=0"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
let body = body_json(response.into_body()).await;
let detail = body["errors"][0]["detail"]
.as_str()
.expect("error detail should be present");
assert!(
detail.contains(&run_id_a),
"detail should mention first run: {detail}"
);
assert!(
detail.contains(&run_id_b),
"detail should mention second run: {detail}"
);
}
#[tokio::test]
async fn resolve_run_prefers_most_recent_exact_workflow_slug_match() {
let app = test_app_with();
let older_id = create_run_for_target(
&app,
"ship-feature.fabro",
&named_workflow_dot("ShipFeatureAlpha", "older"),
)
.await;
let newer_id = create_run_for_target(
&app,
"ship-feature.fabro",
&named_workflow_dot("ShipFeatureBeta", "newer"),
)
.await;
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri(api("/runs/resolve?selector=ship-feature"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert_eq!(body["run_id"], newer_id);
assert_ne!(body["run_id"], older_id);
}
#[tokio::test]
async fn resolve_run_prefers_most_recent_collapsed_workflow_name_match() {
let app = test_app_with();
let older_id = create_run_for_target(
&app,
"nightly-alpha.fabro",
&named_workflow_dot("Nightly_Build", "older"),
)
.await;
let newer_id = create_run_for_target(
&app,
"nightly-beta.fabro",
&named_workflow_dot("Nightly_Build", "newer"),
)
.await;
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri(api("/runs/resolve?selector=nightlybuild"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let body = body_json(response.into_body()).await;
assert_eq!(body["run_id"], newer_id);
assert_ne!(body["run_id"], older_id);
}
#[tokio::test]
async fn resolve_run_returns_not_found_for_unknown_selector() {
let app = test_app_with();
let response = app
.oneshot(
Request::builder()
.method("GET")
.uri(api("/runs/resolve?selector=missing-run"))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
assert_eq!(response.status(), StatusCode::NOT_FOUND);
}
#[tokio::test]
async fn get_questions_returns_empty_list() {
let state = create_app_state();

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

@ -338,6 +338,49 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
options: localVarRequestOptions,
};
},
/**
* Resolves a run selector to one durable run summary using server-owned selector semantics.
* @summary Resolve Run Selector
* @param {string} selector Run selector, such as a run ID prefix, workflow slug, or workflow name.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
resolveRun: async (selector: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'selector' is not null or undefined
assertParamExists('resolveRun', 'selector', selector)
const localVarPath = `/api/v1/runs/resolve`;
// 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 SessionCookie required
// authentication BearerAuth required
// http bearer authentication required
await setBearerAuthToObject(localVarHeaderParameter, configuration)
if (selector !== undefined) {
localVarQueryParameter['selector'] = selector;
}
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 the durable run summary for a run.
* @summary Retrieve Run
@ -644,6 +687,19 @@ export const RunsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunsApi.renderWorkflowGraph']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Resolves a run selector to one durable run summary using server-owned selector semantics.
* @summary Resolve Run Selector
* @param {string} selector Run selector, such as a run ID prefix, workflow slug, or workflow name.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async resolveRun(selector: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<StoreRunSummary>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.resolveRun(selector, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.resolveRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns the durable run summary for a run.
* @summary Retrieve Run
@ -791,6 +847,16 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
renderWorkflowGraph(renderWorkflowGraphRequest: RenderWorkflowGraphRequest, options?: RawAxiosRequestConfig): AxiosPromise<File> {
return localVarFp.renderWorkflowGraph(renderWorkflowGraphRequest, options).then((request) => request(axios, basePath));
},
/**
* Resolves a run selector to one durable run summary using server-owned selector semantics.
* @summary Resolve Run Selector
* @param {string} selector Run selector, such as a run ID prefix, workflow slug, or workflow name.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
resolveRun(selector: string, options?: RawAxiosRequestConfig): AxiosPromise<StoreRunSummary> {
return localVarFp.resolveRun(selector, options).then((request) => request(axios, basePath));
},
/**
* Returns the durable run summary for a run.
* @summary Retrieve Run
@ -928,6 +994,17 @@ export class RunsApi extends BaseAPI {
return RunsApiFp(this.configuration).renderWorkflowGraph(renderWorkflowGraphRequest, options).then((request) => request(this.axios, this.basePath));
}
/**
* Resolves a run selector to one durable run summary using server-owned selector semantics.
* @summary Resolve Run Selector
* @param {string} selector Run selector, such as a run ID prefix, workflow slug, or workflow name.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public resolveRun(selector: string, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).resolveRun(selector, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns the durable run summary for a run.
* @summary Retrieve Run