diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index 6b3f4ec76..dbabb5a2e 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -77,6 +77,7 @@ fabro [OPTIONS] [COMMAND] | `fabro doctor` | Check environment and integration health | | `fabro dump` | Export a run's durable state to a directory | | `fabro events` | View the event log of a workflow run | +| `fabro fork` | Fork a workflow run from an earlier checkpoint into a new run | | `fabro graph` | Render a workflow graph as SVG | | `fabro inspect` | Show detailed information about a workflow run | | `fabro install` | Set up the Fabro environment (LLMs, certs, GitHub) | @@ -89,6 +90,8 @@ fabro [OPTIONS] [COMMAND] | `fabro provider` | Provider operations | | `fabro repo` | Repository commands | | `fabro resume` | Resume an interrupted workflow run | +| `fabro retry` | Retry a finished workflow run from its last checkpoint in a new run | +| `fabro rewind` | Rewind a workflow run to an earlier checkpoint, replacing it | | `fabro rm` | Remove one or more workflow runs | | `fabro run` | Register a workflow version, create a run, and start it | | `fabro sandbox` | Sandbox operations (cp, ssh, preview) | @@ -98,6 +101,7 @@ fabro [OPTIONS] [COMMAND] | `fabro start` | Start a created workflow run on the server | | `fabro steer` | Steer a running agent mid-execution | | `fabro system` | System maintenance commands | +| `fabro timeline` | Show the checkpoint timeline of a workflow run | | `fabro unarchive` | Restore archived runs to their prior terminal status | | `fabro uninstall` | Uninstall Fabro from this machine | | `fabro upgrade` | Upgrade fabro to the latest version | @@ -464,6 +468,28 @@ fabro events [OPTIONS] | `--since ` | Events since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") | | `-n, --tail ` | Lines from end (default: all) | +### `fabro fork` + +Fork a workflow run from an earlier checkpoint into a new run + +```bash +fabro fork [OPTIONS] [TARGET] +``` + +#### Arguments + +| Name | Description | +| --- | --- | +| `RUN_ID` | Run ID (or unambiguous prefix) | +| `TARGET` | Target checkpoint: node name, node@visit, or @ordinal (omit to fork from latest) | + +#### Options + +| Option | Description | +| --- | --- | +| `--list` | Show the checkpoint timeline instead of forking | +| `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | + ### `fabro graph` Render a workflow graph as SVG @@ -998,6 +1024,48 @@ fabro resume [OPTIONS] | `-d, --detach` | Run in the background and print the run ID | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | +### `fabro retry` + +Retry a finished workflow run from its last checkpoint in a new run + +```bash +fabro retry [OPTIONS] +``` + +#### Arguments + +| Name | Description | +| --- | --- | +| `RUN_ID` | Run ID (or unambiguous prefix) | + +#### Options + +| Option | Description | +| --- | --- | +| `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | + +### `fabro rewind` + +Rewind a workflow run to an earlier checkpoint, replacing it + +```bash +fabro rewind [OPTIONS] [TARGET] +``` + +#### Arguments + +| Name | Description | +| --- | --- | +| `RUN_ID` | Run ID (or unambiguous prefix) | +| `TARGET` | Target checkpoint: node name, node@visit, or @ordinal (omit with --list) | + +#### Options + +| Option | Description | +| --- | --- | +| `--list` | Show the checkpoint timeline instead of rewinding | +| `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | + ### `fabro rm` Remove one or more workflow runs @@ -1480,6 +1548,26 @@ fabro system repair runs [OPTIONS] | `--storage-dir ` | Local storage directory (default: ~/.fabro/storage) | | `--yes` | Actually delete unreadable runs (default is dry-run) | +### `fabro timeline` + +Show the checkpoint timeline of a workflow run + +```bash +fabro timeline [OPTIONS] +``` + +#### Arguments + +| Name | Description | +| --- | --- | +| `RUN_ID` | Run ID (or unambiguous prefix) | + +#### Options + +| Option | Description | +| --- | --- | +| `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | + ### `fabro unarchive` Restore archived runs to their prior terminal status diff --git a/lib/apps/fabro-cli/src/args.rs b/lib/apps/fabro-cli/src/args.rs index 5cca0f072..b78459f7a 100644 --- a/lib/apps/fabro-cli/src/args.rs +++ b/lib/apps/fabro-cli/src/args.rs @@ -762,6 +762,57 @@ pub(crate) struct ResumeArgs { pub(crate) detach: bool, } +#[derive(Args)] +pub(crate) struct RetryArgs { + #[command(flatten)] + pub(crate) server: ServerTargetArgs, + + /// Run ID (or unambiguous prefix) + pub(crate) run_id: String, +} + +#[derive(Args)] +pub(crate) struct ForkArgs { + #[command(flatten)] + pub(crate) server: ServerTargetArgs, + + /// Run ID (or unambiguous prefix) + pub(crate) run_id: String, + + /// Target checkpoint: node name, node@visit, or @ordinal (omit to fork from + /// latest) + pub(crate) target: Option, + + /// Show the checkpoint timeline instead of forking + #[arg(long)] + pub(crate) list: bool, +} + +#[derive(Args)] +pub(crate) struct RewindArgs { + #[command(flatten)] + pub(crate) server: ServerTargetArgs, + + /// Run ID (or unambiguous prefix) + pub(crate) run_id: String, + + /// Target checkpoint: node name, node@visit, or @ordinal (omit with --list) + pub(crate) target: Option, + + /// Show the checkpoint timeline instead of rewinding + #[arg(long)] + pub(crate) list: bool, +} + +#[derive(Args)] +pub(crate) struct TimelineArgs { + #[command(flatten)] + pub(crate) server: ServerTargetArgs, + + /// Run ID (or unambiguous prefix) + pub(crate) run_id: String, +} + #[derive(Args)] pub(crate) struct WaitArgs { #[command(flatten)] @@ -1276,6 +1327,14 @@ pub(crate) enum RunCommands { Logs(LogsArgs), /// Resume an interrupted workflow run Resume(ResumeArgs), + /// Retry a finished workflow run from its last checkpoint in a new run + Retry(RetryArgs), + /// Fork a workflow run from an earlier checkpoint into a new run + Fork(ForkArgs), + /// Rewind a workflow run to an earlier checkpoint, replacing it + Rewind(RewindArgs), + /// Show the checkpoint timeline of a workflow run + Timeline(TimelineArgs), /// Block until a workflow run completes Wait(WaitArgs), /// Steer a running agent mid-execution @@ -1296,6 +1355,10 @@ impl RunCommands { Self::Events(_) => "events", Self::Logs(_) => "logs", Self::Resume(_) => "resume", + Self::Retry(_) => "retry", + Self::Fork(_) => "fork", + Self::Rewind(_) => "rewind", + Self::Timeline(_) => "timeline", Self::Steer(_) => "steer", Self::Ask(_) => "ask", Self::Wait(_) => "wait", diff --git a/lib/apps/fabro-cli/src/commands/run/checkpoints.rs b/lib/apps/fabro-cli/src/commands/run/checkpoints.rs new file mode 100644 index 000000000..8fb9d0d77 --- /dev/null +++ b/lib/apps/fabro-cli/src/commands/run/checkpoints.rs @@ -0,0 +1,162 @@ +//! The checkpoint timeline as the CLI shows it: what `fabro timeline` +//! prints, and what `fabro fork --list` and `fabro rewind --list` show +//! before a target is chosen. + +use cli_table::format::{Border, Separator}; +use cli_table::{Cell, CellStruct, Color, Style, Table}; +use fabro_api::types::{RunTimelineResponse, TimelineEntryResponse}; +use fabro_util::printer::Printer; +use fabro_util::terminal::Styles; +use serde::Serialize; + +use crate::shared::color_if; + +/// One timeline entry as `--json` prints it. +#[derive(Serialize)] +pub(crate) struct TimelineEntryJson { + ordinal: u64, + node_name: String, + visit: u32, + stage: Option, + execution: u64, + firing: u64, + attempt: u32, + run_commit_sha: Option, + files_changed: Option, +} + +/// The timeline as `--json` prints it. +#[derive(Serialize)] +pub(crate) struct TimelineJson { + entries: Vec, + forked_from: Option, +} + +#[derive(Serialize)] +pub(crate) struct ForkOriginJson { + source_run_id: String, + execution: u64, + firing: u64, + rerun_last: bool, +} + +pub(crate) fn timeline_json(timeline: &RunTimelineResponse) -> TimelineJson { + TimelineJson { + entries: timeline.entries.iter().map(entry_json).collect(), + forked_from: timeline.forked_from.as_ref().map(|origin| ForkOriginJson { + source_run_id: origin.source_run_id.clone(), + execution: origin.execution, + firing: origin.firing, + rerun_last: origin.rerun_last, + }), + } +} + +fn entry_json(entry: &TimelineEntryResponse) -> TimelineEntryJson { + TimelineEntryJson { + ordinal: entry.ordinal, + node_name: entry.node_name.clone(), + visit: entry.visit, + stage: entry.stage.clone(), + execution: entry.execution, + firing: entry.firing, + attempt: entry.attempt, + run_commit_sha: entry.run_commit_sha.clone(), + files_changed: entry + .diff_summary + .as_ref() + .map(|summary| summary.files_changed), + } +} + +pub(crate) fn short_id(run_id: &str) -> &str { + &run_id[..8.min(run_id.len())] +} + +/// Print the timeline as a table on stderr, with where the run was forked +/// from when it is a fork. +pub(crate) fn print_timeline(timeline: &RunTimelineResponse, styles: &Styles, printer: Printer) { + if let Some(origin) = &timeline.forked_from { + fabro_util::printerr!( + printer, + "Forked from {} at execution {} firing {}{}", + short_id(&origin.source_run_id), + origin.execution, + origin.firing, + if origin.rerun_last { + " (that stage runs again)" + } else { + "" + } + ); + } + if timeline.entries.is_empty() { + fabro_util::printerr!(printer, "No checkpoints found."); + return; + } + + let use_color = styles.use_color; + let title = vec![ + "@".cell().bold(use_color), + "Node".cell().bold(use_color), + "Commit".cell().bold(use_color), + "Details".cell().bold(use_color), + ]; + + let rows: Vec> = timeline + .entries + .iter() + .map(|entry| { + let mut details = Vec::new(); + if entry.visit > 1 { + details.push(format!("visit {}, loop", entry.visit)); + } + if entry.attempt > 1 { + details.push(format!("attempt {}", entry.attempt)); + } + if let Some(summary) = &entry.diff_summary { + details.push(format!( + "{} files, +{} -{}", + summary.files_changed, summary.additions, summary.deletions + )); + } + let commit = entry.run_commit_sha.as_deref().map_or_else( + || "no run commit".to_string(), + |sha| short_id(sha).to_string(), + ); + let detail_str = if details.is_empty() { + String::new() + } else { + format!("({})", details.join(", ")) + }; + + vec![ + format!("@{}", entry.ordinal) + .cell() + .foreground_color(color_if(use_color, Color::Cyan)), + entry.node_name.clone().cell(), + commit.cell(), + detail_str + .cell() + .foreground_color(color_if(use_color, Color::Ansi256(8))), + ] + }) + .collect(); + + let color_choice = if use_color { + cli_table::ColorChoice::Auto + } else { + cli_table::ColorChoice::Never + }; + let table = rows + .table() + .title(title) + .color_choice(color_choice) + .border(Border::builder().build()) + .separator(Separator::builder().build()); + if let Ok(display) = table.display() { + for line in display.to_string().lines() { + fabro_util::printerr!(printer, "{}", line.trim_end()); + } + } +} diff --git a/lib/apps/fabro-cli/src/commands/run/fork.rs b/lib/apps/fabro-cli/src/commands/run/fork.rs new file mode 100644 index 000000000..cb6f16d28 --- /dev/null +++ b/lib/apps/fabro-cli/src/commands/run/fork.rs @@ -0,0 +1,52 @@ +use anyhow::Result; +use fabro_api::types::ForkRequest; +use fabro_util::terminal::Styles; + +use super::checkpoints::{print_timeline, short_id, timeline_json}; +use crate::args::ForkArgs; +use crate::command_context::CommandContext; +use crate::shared::print_json_pretty; + +/// Fork a run at a checkpoint into a new run, which starts at once. +pub(crate) async fn run(args: &ForkArgs, styles: &Styles, base_ctx: &CommandContext) -> Result<()> { + let printer = base_ctx.printer(); + let ctx = base_ctx.with_target(&args.server)?; + let client = ctx.server().await?; + let run_id = client.resolve_run(&args.run_id).await?.id; + + if args.list { + let timeline = client.run_timeline(&run_id).await?; + if ctx.json_output() { + print_json_pretty(&timeline_json(&timeline))?; + } else { + print_timeline(&timeline, styles, printer); + } + return Ok(()); + } + + let response = client + .fork_run(&run_id, ForkRequest { + target: args.target.clone(), + }) + .await?; + + if ctx.json_output() { + print_json_pretty(&response)?; + } else { + fabro_util::printerr!( + printer, + "\nForked run {} -> {} at {} ({})", + short_id(&response.source_run_id), + short_id(&response.new_run_id), + response.target, + short_id(&response.checkpoint_sha) + ); + fabro_util::printerr!( + printer, + "To follow: fabro attach {}", + short_id(&response.new_run_id) + ); + } + + Ok(()) +} diff --git a/lib/apps/fabro-cli/src/commands/run/mod.rs b/lib/apps/fabro-cli/src/commands/run/mod.rs index dbede919a..14c72e710 100644 --- a/lib/apps/fabro-cli/src/commands/run/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/mod.rs @@ -10,11 +10,13 @@ use crate::sleep_inhibitor; pub(crate) mod ask; pub(crate) mod attach; +pub(crate) mod checkpoints; pub(crate) mod command; pub(crate) mod cp; pub(crate) mod create; pub(crate) mod diff; pub(crate) mod events; +pub(crate) mod fork; pub(crate) mod logs; pub(crate) mod output; pub(crate) mod overrides; @@ -24,6 +26,8 @@ pub(crate) mod preview; mod remote_workflow; mod resolution; pub(crate) mod resume; +pub(crate) mod retry; +pub(crate) mod rewind; pub(crate) mod run_progress; pub(crate) mod runner; mod selection; @@ -32,6 +36,7 @@ pub(crate) mod start; pub(crate) mod steer; #[cfg(test)] pub(crate) mod test_support; +pub(crate) mod timeline; pub(crate) mod wait; pub(crate) async fn dispatch( @@ -136,6 +141,19 @@ pub(crate) async fn dispatch( }; Box::pin(resume::resume_command(args, styles, base_ctx)).await } + RunCommands::Retry(args) => retry::run(&args, base_ctx).await, + RunCommands::Fork(args) => { + let styles = Styles::detect_stderr(); + Box::pin(fork::run(&args, &styles, base_ctx)).await + } + RunCommands::Rewind(args) => { + let styles = Styles::detect_stderr(); + Box::pin(rewind::run(&args, &styles, base_ctx)).await + } + RunCommands::Timeline(args) => { + let styles = Styles::detect_stderr(); + timeline::run(&args, &styles, base_ctx).await + } RunCommands::Wait(args) => { let styles = Styles::detect_stderr(); wait::run(&args, &styles, base_ctx).await diff --git a/lib/apps/fabro-cli/src/commands/run/retry.rs b/lib/apps/fabro-cli/src/commands/run/retry.rs new file mode 100644 index 000000000..2fae3e865 --- /dev/null +++ b/lib/apps/fabro-cli/src/commands/run/retry.rs @@ -0,0 +1,32 @@ +use anyhow::Result; + +use super::checkpoints::short_id; +use crate::args::RetryArgs; +use crate::command_context::CommandContext; +use crate::shared::print_json_pretty; + +/// Retry a finished run from its last checkpoint in a new run, which starts +/// at once. +pub(crate) async fn run(args: &RetryArgs, base_ctx: &CommandContext) -> Result<()> { + let printer = base_ctx.printer(); + let ctx = base_ctx.with_target(&args.server)?; + let client = ctx.server().await?; + let run_id = client.resolve_run(&args.run_id).await?.id; + let new_run = client.retry_run(&run_id).await?; + + if ctx.json_output() { + print_json_pretty(&serde_json::json!({ + "source_run_id": run_id, + "run_id": new_run.id, + }))?; + } else { + fabro_util::printerr!( + printer, + "Retrying {} as {}", + short_id(&run_id.to_string()), + short_id(&new_run.id.to_string()) + ); + fabro_util::printout!(printer, "{}", new_run.id); + } + Ok(()) +} diff --git a/lib/apps/fabro-cli/src/commands/run/rewind.rs b/lib/apps/fabro-cli/src/commands/run/rewind.rs new file mode 100644 index 000000000..5749fc618 --- /dev/null +++ b/lib/apps/fabro-cli/src/commands/run/rewind.rs @@ -0,0 +1,74 @@ +use anyhow::Result; +use fabro_api::types::RewindRequest; +use fabro_util::terminal::Styles; + +use super::checkpoints::{print_timeline, short_id, timeline_json}; +use crate::args::RewindArgs; +use crate::command_context::CommandContext; +use crate::shared::print_json_pretty; + +/// Rewind a run to a checkpoint: a new run replaces it and starts at once. +pub(crate) async fn run( + args: &RewindArgs, + styles: &Styles, + base_ctx: &CommandContext, +) -> Result<()> { + let printer = base_ctx.printer(); + let ctx = base_ctx.with_target(&args.server)?; + let client = ctx.server().await?; + let run_id = client.resolve_run(&args.run_id).await?.id; + + if args.list || args.target.is_none() { + let timeline = client.run_timeline(&run_id).await?; + if ctx.json_output() { + print_json_pretty(&timeline_json(&timeline))?; + } else { + print_timeline(&timeline, styles, printer); + } + return Ok(()); + } + + let result = client + .rewind_run(&run_id, RewindRequest { + target: args.target.clone(), + }) + .await?; + let response = result.response; + + if ctx.json_output() { + print_json_pretty(&serde_json::json!({ + "source_run_id": response.source_run_id, + "new_run_id": response.new_run_id, + "target": response.target, + "checkpoint_sha": response.checkpoint_sha, + "execution": response.execution, + "firing": response.firing, + "archived": response.archived, + "archive_error": response.archive_error, + "status": result.status, + }))?; + } else { + fabro_util::printerr!( + printer, + "\nRewound {} to {}; new run {}", + short_id(&response.source_run_id), + response.target, + short_id(&response.new_run_id) + ); + fabro_util::printerr!( + printer, + "To follow: fabro attach {}", + short_id(&response.new_run_id) + ); + if !response.archived { + let archive_error = response.archive_error.as_deref().unwrap_or("unknown error"); + fabro_util::printerr!( + printer, + "Warning: source not archived: {archive_error}. Run `fabro archive {}` to finish.", + short_id(&response.source_run_id) + ); + } + } + + Ok(()) +} diff --git a/lib/apps/fabro-cli/src/commands/run/timeline.rs b/lib/apps/fabro-cli/src/commands/run/timeline.rs new file mode 100644 index 000000000..4e1dcb2bc --- /dev/null +++ b/lib/apps/fabro-cli/src/commands/run/timeline.rs @@ -0,0 +1,26 @@ +use anyhow::Result; +use fabro_util::terminal::Styles; + +use super::checkpoints::{print_timeline, timeline_json}; +use crate::args::TimelineArgs; +use crate::command_context::CommandContext; +use crate::shared::print_json_pretty; + +/// Show the checkpoint timeline of a run. +pub(crate) async fn run( + args: &TimelineArgs, + styles: &Styles, + base_ctx: &CommandContext, +) -> Result<()> { + let printer = base_ctx.printer(); + let ctx = base_ctx.with_target(&args.server)?; + let client = ctx.server().await?; + let run_id = client.resolve_run(&args.run_id).await?.id; + let timeline = client.run_timeline(&run_id).await?; + if ctx.json_output() { + print_json_pretty(&timeline_json(&timeline))?; + } else { + print_timeline(&timeline, styles, printer); + } + Ok(()) +} diff --git a/lib/apps/fabro-cli/tests/it/cmd/fabro.rs b/lib/apps/fabro-cli/tests/it/cmd/fabro.rs index 808f4e895..c28cc5ed8 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/fabro.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/fabro.rs @@ -19,6 +19,10 @@ fn help() { events View the event log of a workflow run logs View the raw worker tracing log of a workflow run resume Resume an interrupted workflow run + retry Retry a finished workflow run from its last checkpoint in a new run + fork Fork a workflow run from an earlier checkpoint into a new run + rewind Rewind a workflow run to an earlier checkpoint, replacing it + timeline Show the checkpoint timeline of a workflow run wait Block until a workflow run completes steer Steer a running agent mid-execution ask Ask Fabro a read-only question about a run