scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-10-01 02:04:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge branch 'petri-followup-fork' into petri-integration

Browse source
This commit is contained in:
Bryan Helmkamp 2026-09-18 23:18:14 -04:00
commit 86ac13713b
No known key found for this signature in database
41 changed files with 3749 additions and 24 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

366
docs/public/api-reference/fabro-api.yaml
View file

@ -2327,6 +2327,197 @@ paths:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/retry:
post:
operationId: retryRun
tags: [Runs]
summary: Retry Run
description: >
Creates a new run from the terminal source run's last checkpoint and
starts it. When the source failed on a stage, that stage runs again on
the files of the stage before it; otherwise the new run continues from
the last checkpoint as it stands. The new run records `retried_from`
and `fork_source_ref`; the source run is left unchanged. Active and
archived runs are not retryable.
parameters:
- $ref: "#/components/parameters/RunId"
responses:
"201":
description: New retry run created and started
content:
application/json:
schema:
$ref: "#/components/schemas/Run"
"400":
description: The source has no checkpoint to retry from
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"404":
description: Run not found
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"409":
description: Source run is not retryable
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/rewind:
post:
operationId: rewindRun
tags: [Runs]
summary: Rewind Run
description: >
Creates a new run from a checkpoint of a terminal source run and
starts it, then archives the source run and records
`run.superseded_by` on it. Returns 207 when the new run was created
but the source archive step failed.
parameters:
- $ref: "#/components/parameters/RunId"
requestBody:
required: false
content:
application/json:
schema:
$ref: "#/components/schemas/RewindRequest"
responses:
"200":
description: Source archived and new run created
content:
application/json:
schema:
$ref: "#/components/schemas/RewindResponse"
"207":
description: New run created but source archive failed
content:
application/json:
schema:
$ref: "#/components/schemas/RewindResponse"
"400":
description: Invalid rewind target
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"404":
description: Run not found
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"409":
description: Source run is archived or is not terminal
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/fork:
post:
operationId: forkRun
tags: [Runs]
summary: Fork Run
description: >
Creates a new run from a checkpoint of the source run and starts it.
The new run holds the source's records up to the checkpoint's
position and continues from there in a fresh workspace restored to
the checkpoint's commit. The source run is left untouched. A
checkpoint inside a parallel branch cannot be forked at; fork at the
parallel stage instead.
parameters:
- $ref: "#/components/parameters/RunId"
requestBody:
required: false
content:
application/json:
schema:
$ref: "#/components/schemas/ForkRequest"
responses:
"200":
description: New run created and started
content:
application/json:
schema:
$ref: "#/components/schemas/ForkResponse"
"400":
description: Invalid fork target
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"404":
description: Run not found
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
"409":
description: Source run is archived
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/timeline:
get:
operationId: getRunTimeline
tags: [Runs]
summary: Get Run Timeline
description: >
Returns the run's checkpoints in the order they were recorded, each
at its Petri position with the commit it made and the stage it
belongs to, and where the run was forked from when it is a fork.
parameters:
- $ref: "#/components/parameters/RunId"
responses:
"200":
description: Run checkpoint timeline
content:
application/json:
schema:
$ref: "#/components/schemas/RunTimelineResponse"
"404":
description: Run not found
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
content:
application/json:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/unarchive:
post:
operationId: unarchiveRun
@ -12778,6 +12969,11 @@ components:
retried_from:
type: ["string", "null"]
description: Source run ID when this run was created by manual retry.
forked_from:
oneOf:
- $ref: "#/components/schemas/ForkOrigin"
- type: "null"
description: Where the run's records came from when it is a fork.
git_identity:
oneOf:
- $ref: "#/components/schemas/GitIdentity"
@ -13128,6 +13324,176 @@ components:
type: ["string", "null"]
format: uri
ForkOrigin:
description: >-
Where a forked run came from: the source run, the position (an
execution and a firing of its root invocation) the source's records
were kept up to, and whether that firing runs again in the fork.
type: object
required:
- source_run_id
- execution
- firing
- rerun_last
properties:
source_run_id:
type: string
execution:
type: integer
format: uint64
firing:
type: integer
format: uint64
rerun_last:
type: boolean
ForkRequest:
description: Request body for creating a new run from a source run checkpoint.
type: object
properties:
target:
type: ["string", "null"]
description: Optional checkpoint target such as `@2`, `build`, or `build@1`. Defaults to the latest checkpoint.
ForkResponse:
description: Response returned after creating and starting a forked run.
type: object
required:
- source_run_id
- new_run_id
- target
- checkpoint_sha
- execution
- firing
- rerun_last
properties:
source_run_id:
type: string
new_run_id:
type: string
target:
type: string
description: The checkpoint the fork was resolved to, as `@ordinal`.
checkpoint_sha:
type: string
description: The commit the new run's workspace starts on.
execution:
type: integer
format: uint64
description: The Petri execution of the fork position.
firing:
type: integer
format: uint64
description: The Petri firing of the fork position.
rerun_last:
type: boolean
description: Whether the position's stage runs again in the new run.
RewindRequest:
description: Request body for creating a replacement run from a source run checkpoint.
type: object
properties:
target:
type: ["string", "null"]
description: Optional checkpoint target such as `@2`, `build`, or `build@1`. Defaults to the latest checkpoint.
RewindResponse:
description: Response returned after rewind creates and starts a new run.
type: object
required:
- source_run_id
- new_run_id
- target
- checkpoint_sha
- execution
- firing
- archived
properties:
source_run_id:
type: string
new_run_id:
type: string
target:
type: string
description: The checkpoint the rewind was resolved to, as `@ordinal`.
checkpoint_sha:
type: string
execution:
type: integer
format: uint64
firing:
type: integer
format: uint64
archived:
type: boolean
archive_error:
type: ["string", "null"]
RunTimelineResponse:
description: The run's checkpoints in order, and its fork origin when it is a fork.
type: object
required:
- entries
properties:
entries:
type: array
items:
$ref: "#/components/schemas/TimelineEntryResponse"
forked_from:
oneOf:
- $ref: "#/components/schemas/ForkOrigin"
- type: "null"
TimelineEntryResponse:
description: One checkpoint of a run.
type: object
required:
- ordinal
- checkpoint_seq
- execution
- firing
- attempt
- node_name
- visit
properties:
ordinal:
type: integer
format: uint64
description: 1-based, in the order the checkpoints were recorded; the `@ordinal` a fork target names.
checkpoint_seq:
type: integer
format: uint64
description: The checkpoint record's position among the run's platform records.
execution:
type: integer
format: uint64
description: The Petri execution the checkpoint belongs to.
firing:
type: integer
format: uint64
description: The Petri firing the checkpoint belongs to.
attempt:
type: integer
format: uint32
description: The attempt of the firing whose files the commit holds, from 1.
stage:
type: ["string", "null"]
description: The stage id (`node@visit`) when the projection shows the stage.
node_name:
type: string
visit:
type: integer
format: uint32
workspace:
type: ["string", "null"]
description: The Petri workspace id the commit was made in.
run_commit_sha:
type: ["string", "null"]
diff_summary:
oneOf:
- $ref: "#/components/schemas/DiffSummary"
- type: "null"
BoardColumn:
description: |
Status bucket for a run, shared by list and kanban renderings and by

88
docs/public/reference/cli.mdx
View file

@ -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] <RUN>
| `--since <since>` | Events since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") |
| `-n, --tail <tail>` | Lines from end (default: all) |
### `fabro fork`
Fork a workflow run from an earlier checkpoint into a new run
```bash
fabro fork [OPTIONS] <RUN_ID> [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 <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] <RUN>
| `-d, --detach` | Run in the background and print the run ID |
| `--server <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] <RUN_ID>
```
#### Arguments
| Name | Description |
| --- | --- |
| `RUN_ID` | Run ID (or unambiguous prefix) |
#### Options
| Option | Description |
| --- | --- |
| `--server <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] <RUN_ID> [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 <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 <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] <RUN_ID>
```
#### Arguments
| Name | Description |
| --- | --- |
| `RUN_ID` | Run ID (or unambiguous prefix) |
#### Options
| Option | Description |
| --- | --- |
| `--server <server>` | Fabro server target: http(s) URL or absolute Unix socket path |
### `fabro unarchive`
Restore archived runs to their prior terminal status

63
lib/apps/fabro-cli/src/args.rs
View file

@ -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<String>,
/// 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<String>,
/// 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",

162
lib/apps/fabro-cli/src/commands/run/checkpoints.rs Normal file
View file

@ -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<String>,
execution: u64,
firing: u64,
attempt: u32,
run_commit_sha: Option<String>,
files_changed: Option<i64>,
}
/// The timeline as `--json` prints it.
#[derive(Serialize)]
pub(crate) struct TimelineJson {
entries: Vec<TimelineEntryJson>,
forked_from: Option<ForkOriginJson>,
}
#[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<Vec<CellStruct>> = 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());
}
}
}

52
lib/apps/fabro-cli/src/commands/run/fork.rs Normal file
View file

@ -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(())
}

18
lib/apps/fabro-cli/src/commands/run/mod.rs
View file

@ -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

32
lib/apps/fabro-cli/src/commands/run/retry.rs Normal file
View file

@ -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(())
}

74
lib/apps/fabro-cli/src/commands/run/rewind.rs Normal file
View file

@ -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(())
}

26
lib/apps/fabro-cli/src/commands/run/timeline.rs Normal file
View file

@ -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(())
}

4
lib/apps/fabro-cli/tests/it/cmd/fabro.rs
View file

@ -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

1
lib/apps/fabro-cli/tests/it/scenario/mod.rs
View file

@ -11,6 +11,7 @@ mod lifecycle;
mod petri;
mod petri_controls;
mod petri_docker;
mod petri_fork;
mod petri_tools;
mod server_lifecycle;
mod smoke;

515
lib/apps/fabro-cli/tests/it/scenario/petri_fork.rs Normal file
View file

@ -0,0 +1,515 @@
//! Fork, rewind, retry and the timeline over Petri runs, through a real
//! server and its worker subprocess (the integration plan's F5.1).
//!
//! The harness is `petri.rs`'s: a foreground server on disk storage, a run
//! created and started with `fabro run --detach`, executed by the worker
//! the server launches over the HTTP run store. Every checkpoint of a run
//! is a commit on its run branch and a `checkpoint` platform record at its
//! Petri position; the timeline lists them, and a fork seeds a new run from
//! the source's records up to one of them, whose worker restores the
//! checkpoint's files into a fresh workspace and continues from there.
#![expect(
clippy::disallowed_methods,
reason = "these scenarios start a real server subprocess and read its workspaces on disk"
)]
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use fabro_test::test_context;
use super::petri::{
RunningServer, host_plugin, run_detached, run_json, wait_for_status, wait_for_success,
write_petri_workflow,
};
/// Three command stages that build on each other's files: `one` writes a
/// file, `two` another, `three` checks both and writes its own.
fn three_stage_dot() -> String {
"digraph Stages {\n graph [goal=\"Three stages\", default_max_retries=0]\n start \
[shape=Mdiamond]\n exit [shape=Msquare]\n one [shape=parallelogram, script=\"echo one > \
one.txt\"]\n two [shape=parallelogram, script=\"echo two > two.txt\"]\n three \
[shape=parallelogram, script=\"test \\\"$(cat one.txt)\\\" = one && test \\\"$(cat \
two.txt)\\\" = two && echo three > three.txt\"]\n start -> one -> two -> three -> exit\n}\n"
.to_string()
}
/// Three stages whose middle one fails the first time it runs and passes
/// the next: the marker outside the workspace remembers the first run. Its
/// failure ends the run (`on_failure=exit`) rather than routing on.
fn flaky_dot(marker: &Path) -> String {
format!(
"digraph Flaky {{\n graph [goal=\"A transient failure\", default_max_retries=0]\n \
start [shape=Mdiamond]\n exit [shape=Msquare]\n one [shape=parallelogram, \
script=\"echo one > one.txt\"]\n flaky [shape=parallelogram, max_retries=0, on_failure=exit, \
script=\"if [ ! -f {marker} ]; then touch {marker}; exit 1; fi; echo flaky > \
flaky.txt\"]\n three [shape=parallelogram, script=\"test \\\"$(cat one.txt)\\\" = one \
&& test \\\"$(cat flaky.txt)\\\" = flaky && echo three > three.txt\"]\n start -> one \
-> flaky -> three -> exit\n}}\n",
marker = marker.display()
)
}
/// A parallel node with two command branches and a join.
fn parallel_dot() -> String {
"digraph Branches {\n graph [goal=\"Two branches\", default_max_retries=0]\n start \
[shape=Mdiamond]\n exit [shape=Msquare]\n fan [shape=component]\n a \
[shape=parallelogram, script=\"echo a > a.txt\"]\n b [shape=parallelogram, script=\"echo \
b > b.txt\"]\n join [shape=tripleoctagon]\n done [shape=parallelogram, script=\"echo done \
> done.txt\"]\n start -> fan\n fan -> a\n fan -> b\n a -> join\n b -> join\n join -> \
done -> exit\n}\n"
.to_string()
}
/// Run a CLI command against the server; the caller judges the exit.
fn cli(context: &fabro_test::TestContext, server: &RunningServer, args: &[&str]) -> Output {
let target = server.target();
context
.command()
.args(args)
.args(["--server", &target])
.output()
.expect("the CLI command executes")
}
/// Run a CLI command that must succeed, and parse its stdout as JSON.
fn cli_json(
context: &fabro_test::TestContext,
server: &RunningServer,
args: &[&str],
) -> serde_json::Value {
let output = cli(context, server, args);
assert!(
output.status.success(),
"`fabro {}` failed\nstdout:\n{}\nstderr:\n{}",
args.join(" "),
String::from_utf8_lossy(&output.stdout),
String::from_utf8_lossy(&output.stderr)
);
serde_json::from_slice(&output.stdout).unwrap_or_else(|err| {
panic!(
"`fabro {}` printed no JSON: {err}\nstdout:\n{}",
args.join(" "),
String::from_utf8_lossy(&output.stdout)
)
})
}
async fn timeline(server: &RunningServer, run_id: &str) -> serde_json::Value {
run_json(server, &format!("runs/{run_id}/timeline")).await
}
/// The timeline's entries as `(node, execution, firing, attempt, sha)`.
fn entries(timeline: &serde_json::Value) -> Vec<(String, u64, u64, u64, String)> {
timeline["entries"]
.as_array()
.expect("the timeline has entries")
.iter()
.map(|entry| {
(
entry["node_name"].as_str().unwrap_or_default().to_string(),
entry["execution"].as_u64().expect("an execution"),
entry["firing"].as_u64().expect("a firing"),
entry["attempt"].as_u64().expect("an attempt"),
entry["run_commit_sha"]
.as_str()
.expect("a checkpoint commit")
.to_string(),
)
})
.collect()
}
fn nodes(timeline: &serde_json::Value) -> Vec<String> {
entries(timeline)
.into_iter()
.map(|(node, ..)| node)
.collect()
}
/// The one host workspace of the run.
fn workspace(server: &RunningServer, run_id: &str) -> PathBuf {
let scopes = server.petri_run_dir(run_id).join("scopes");
let mut workspaces: Vec<PathBuf> = std::fs::read_dir(&scopes)
.expect("the scopes directory lists")
.map(|entry| entry.expect("an entry reads").path().join("work"))
.collect();
assert_eq!(workspaces.len(), 1, "one workspace: {workspaces:?}");
workspaces.remove(0)
}
/// The subjects of the commits on the workspace's branch, oldest first.
fn commit_subjects(workspace: &Path) -> Vec<String> {
let output = Command::new("git")
.args(["log", "--reverse", "--format=%s"])
.current_dir(workspace)
.output()
.expect("git runs");
assert!(
output.status.success(),
"git log failed: {}",
String::from_utf8_lossy(&output.stderr)
);
String::from_utf8_lossy(&output.stdout)
.lines()
.map(ToOwned::to_owned)
.collect()
}
fn current_branch(workspace: &Path) -> String {
let output = Command::new("git")
.args(["rev-parse", "--abbrev-ref", "HEAD"])
.current_dir(workspace)
.output()
.expect("git runs");
String::from_utf8_lossy(&output.stdout).trim().to_string()
}
fn read(workspace: &Path, name: &str) -> String {
std::fs::read_to_string(workspace.join(name))
.unwrap_or_else(|err| panic!("{name} in {}: {err}", workspace.display()))
}
/// A three-stage run forked at its first stage's checkpoint continues with
/// the other two in a new run: the new run keeps the source's records up to
/// `one`, its workspace holds `one`'s file restored from the checkpoint,
/// and `two` and `three` run on it and commit on the new run branch. The
/// new run says where it came from.
#[tokio::test(flavor = "multi_thread")]
async fn a_fork_at_the_first_stage_continues_with_the_rest_on_its_files() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
let source = run_detached(&context, &server, &bundle);
wait_for_success(&server, &source).await;
let source_timeline = timeline(&server, &source).await;
assert_eq!(nodes(&source_timeline), [
"start", "one", "two", "three", "exit"
]);
let source_entries = entries(&source_timeline);
let (_, one_execution, one_firing, _, one_sha) = source_entries[1].clone();
let forked = cli_json(&context, &server, &["fork", &source, "one", "--json"]);
assert_eq!(forked["source_run_id"], source);
assert_eq!(forked["target"], "@2");
assert_eq!(forked["execution"].as_u64(), Some(one_execution));
assert_eq!(forked["firing"].as_u64(), Some(one_firing));
assert_eq!(forked["checkpoint_sha"], one_sha);
assert_eq!(forked["rerun_last"], false);
let fork = forked["new_run_id"]
.as_str()
.expect("the new run id")
.to_string();
assert_ne!(fork, source);
wait_for_success(&server, &fork).await;
// The fork's timeline: the source's checkpoints up to `one`, then its own.
let fork_timeline = timeline(&server, &fork).await;
assert_eq!(nodes(&fork_timeline), [
"start", "one", "two", "three", "exit"
]);
let fork_entries = entries(&fork_timeline);
assert_eq!(fork_entries[..2], source_entries[..2]);
assert_ne!(fork_entries[2].4, source_entries[2].4);
assert_eq!(fork_timeline["forked_from"]["source_run_id"], source);
assert_eq!(
fork_timeline["forked_from"]["execution"].as_u64(),
Some(one_execution)
);
assert_eq!(
fork_timeline["forked_from"]["firing"].as_u64(),
Some(one_firing)
);
assert_eq!(fork_timeline["forked_from"]["rerun_last"], false);
assert!(source_timeline["forked_from"].is_null());
// The fork's workspace: `one.txt` restored from the checkpoint, the
// rest made by the fork's own stages, on the fork's run branch after the
// source's commits.
let fork_workspace = workspace(&server, &fork);
assert_eq!(read(&fork_workspace, "one.txt"), "one\n");
assert_eq!(read(&fork_workspace, "two.txt"), "two\n");
assert_eq!(read(&fork_workspace, "three.txt"), "three\n");
assert_eq!(current_branch(&fork_workspace), format!("fabro/run/{fork}"));
assert_eq!(commit_subjects(&fork_workspace), [
format!("fabro({source}): start (success)"),
format!("fabro({source}): one (success)"),
format!("fabro({fork}): two (success)"),
format!("fabro({fork}): three (success)"),
format!("fabro({fork}): exit (success)"),
]);
// The projection names the origin on both sides.
let state = run_json(&server, &format!("runs/{fork}/state")).await;
assert_eq!(state["forked_from"]["source_run_id"], source);
assert_eq!(state["spec"]["fork_source_ref"]["source_run_id"], source);
assert_eq!(state["spec"]["fork_source_ref"]["checkpoint_sha"], one_sha);
assert!(state["retried_from"].is_null());
let summary = run_json(&server, &format!("runs/{source}")).await;
assert!(summary["superseded_by"].is_null());
assert_eq!(summary["lifecycle"]["archived"], false);
server.shutdown();
}
/// A retry of a run whose last stage failed reruns that stage: the failure
/// was transient, so the retry passes it on the files of the stage before
/// and finishes the run.
#[tokio::test(flavor = "multi_thread")]
async fn a_retry_reruns_the_failed_stage_and_succeeds_when_the_failure_was_transient() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let marker = context.temp_dir.join("flaky.marker");
let bundle = write_petri_workflow(&context, &flaky_dot(&marker));
let source = run_detached(&context, &server, &bundle);
let status = wait_for_status(&server, &source, &["succeeded", "failed"]).await;
assert_eq!(status, "failed", "the first run fails on `flaky`");
assert!(marker.exists(), "the first run left its marker");
assert_eq!(nodes(&timeline(&server, &source).await), [
"start", "one", "flaky"
]);
let retried = cli_json(&context, &server, &["retry", &source, "--json"]);
assert_eq!(retried["source_run_id"], source);
let retry = retried["run_id"]
.as_str()
.expect("the new run id")
.to_string();
wait_for_success(&server, &retry).await;
let retry_timeline = timeline(&server, &retry).await;
assert_eq!(nodes(&retry_timeline), [
"start", "one", "flaky", "three", "exit"
]);
assert_eq!(retry_timeline["forked_from"]["source_run_id"], source);
assert_eq!(retry_timeline["forked_from"]["rerun_last"], true);
let retry_workspace = workspace(&server, &retry);
assert_eq!(read(&retry_workspace, "one.txt"), "one\n");
assert_eq!(read(&retry_workspace, "flaky.txt"), "flaky\n");
assert_eq!(read(&retry_workspace, "three.txt"), "three\n");
assert_eq!(commit_subjects(&retry_workspace), [
format!("fabro({source}): start (success)"),
format!("fabro({source}): one (success)"),
format!("fabro({retry}): flaky (success)"),
format!("fabro({retry}): three (success)"),
format!("fabro({retry}): exit (success)"),
]);
let state = run_json(&server, &format!("runs/{retry}/state")).await;
assert_eq!(state["retried_from"], source);
assert_eq!(state["spec"]["fork_source_ref"]["source_run_id"], source);
// The source is untouched, and a second retry is refused for the
// running or archived cases alone: it is terminal, so it may retry again.
let summary = run_json(&server, &format!("runs/{source}")).await;
assert_eq!(summary["lifecycle"]["status"]["kind"], "failed");
assert!(summary["superseded_by"].is_null());
server.shutdown();
}
/// A rewind forks the run and supersedes it: the source is archived and
/// names the run that replaced it.
#[tokio::test(flavor = "multi_thread")]
async fn a_rewind_supersedes_its_source() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
let source = run_detached(&context, &server, &bundle);
wait_for_success(&server, &source).await;
// Without a target the command lists the timeline instead.
let listed = cli_json(&context, &server, &["rewind", &source, "--json"]);
assert_eq!(
listed["entries"]
.as_array()
.map(Vec::len)
.expect("a timeline"),
5
);
let rewound = cli_json(&context, &server, &["rewind", &source, "@3", "--json"]);
assert_eq!(rewound["source_run_id"], source);
assert_eq!(rewound["target"], "@3");
assert_eq!(rewound["archived"], true);
assert!(rewound["archive_error"].is_null());
assert_eq!(rewound["status"], 200);
let replacement = rewound["new_run_id"]
.as_str()
.expect("the new run id")
.to_string();
let summary = run_json(&server, &format!("runs/{source}")).await;
assert_eq!(summary["superseded_by"], replacement);
assert_eq!(summary["lifecycle"]["archived"], true);
let state = run_json(&server, &format!("runs/{source}/state")).await;
assert_eq!(state["superseded_by"], replacement);
wait_for_success(&server, &replacement).await;
assert_eq!(nodes(&timeline(&server, &replacement).await), [
"start", "one", "two", "three", "exit"
]);
let replacement_workspace = workspace(&server, &replacement);
assert_eq!(read(&replacement_workspace, "two.txt"), "two\n");
assert_eq!(commit_subjects(&replacement_workspace), [
format!("fabro({source}): start (success)"),
format!("fabro({source}): one (success)"),
format!("fabro({source}): two (success)"),
format!("fabro({replacement}): three (success)"),
format!("fabro({replacement}): exit (success)"),
]);
// An archived run is neither forked nor rewound again.
let refused = cli(&context, &server, &["fork", &source, "@2"]);
assert!(!refused.status.success());
assert!(
String::from_utf8_lossy(&refused.stderr).contains("archived"),
"stderr:\n{}",
String::from_utf8_lossy(&refused.stderr)
);
server.shutdown();
}
/// The timeline lists every checkpoint the run recorded, in order, with
/// its position and commit, as the CLI prints it and as the API serves it.
#[tokio::test(flavor = "multi_thread")]
async fn the_timeline_lists_every_checkpoint_with_its_commit() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &three_stage_dot());
let run_id = run_detached(&context, &server, &bundle);
wait_for_success(&server, &run_id).await;
let recorded = server.checkpoints(&run_id).await;
let listed = cli_json(&context, &server, &["timeline", &run_id, "--json"]);
let listed_entries: Vec<(u64, u64, u64, String)> = listed["entries"]
.as_array()
.expect("entries")
.iter()
.map(|entry| {
(
entry["execution"].as_u64().expect("execution"),
entry["firing"].as_u64().expect("firing"),
entry["attempt"].as_u64().expect("attempt"),
entry["run_commit_sha"].as_str().expect("sha").to_string(),
)
})
.collect();
let recorded_entries: Vec<(u64, u64, u64, String)> = recorded
.iter()
.map(|(key, sha)| {
(
key.execution,
key.firing,
u64::from(key.attempt),
sha.clone(),
)
})
.collect();
assert_eq!(listed_entries, recorded_entries);
assert_eq!(listed_entries.len(), 5);
let ordinals: Vec<u64> = listed["entries"]
.as_array()
.expect("entries")
.iter()
.map(|entry| entry["ordinal"].as_u64().expect("ordinal"))
.collect();
assert_eq!(ordinals, [1, 2, 3, 4, 5]);
let stages: Vec<&str> = listed["entries"]
.as_array()
.expect("entries")
.iter()
.map(|entry| entry["stage"].as_str().unwrap_or("-"))
.collect();
assert_eq!(stages, ["start@1", "one@1", "two@1", "three@1", "exit@1"]);
assert!(listed["forked_from"].is_null());
let table = cli(&context, &server, &["timeline", &run_id]);
assert!(table.status.success());
let stderr = String::from_utf8_lossy(&table.stderr);
for (ordinal, node) in ["start", "one", "two", "three", "exit"].iter().enumerate() {
assert!(
stderr.contains(&format!("@{}", ordinal + 1)) && stderr.contains(node),
"the table names @{} {node}:\n{stderr}",
ordinal + 1
);
}
let api = timeline(&server, &run_id).await;
assert_eq!(nodes(&api), ["start", "one", "two", "three", "exit"]);
server.shutdown();
}
/// A checkpoint inside a parallel branch is not a fork position: Petri
/// refuses it, and the refusal says why. The join, in the root, is.
#[tokio::test(flavor = "multi_thread")]
async fn a_fork_inside_a_parallel_branch_is_refused() {
if host_plugin().is_none() {
return;
}
let context = test_context!();
let server = RunningServer::start().await;
let bundle = write_petri_workflow(&context, &parallel_dot());
let source = run_detached(&context, &server, &bundle);
wait_for_success(&server, &source).await;
let listed = timeline(&server, &source).await;
let all = entries(&listed);
let branch = all
.iter()
.zip(1_u64..)
.find(|((node, execution, ..), _)| *execution != 0 && (node == "a" || node == "b"))
.map(|(_, ordinal)| ordinal)
.expect("a branch stage has a checkpoint in a child execution");
let refused = cli(&context, &server, &["fork", &source, &format!("@{branch}")]);
assert!(
!refused.status.success(),
"a fork inside a branch is refused"
);
let stderr = String::from_utf8_lossy(&refused.stderr);
assert!(
stderr.contains("inside a child invocation cannot be forked"),
"the refusal names the branch:\n{stderr}"
);
// Nothing was started for it.
let runs = run_json(&server, "runs").await;
let ids: Vec<&str> = runs["data"]
.as_array()
.or_else(|| runs.as_array())
.expect("a run list")
.iter()
.filter_map(|run| run["id"].as_str())
.collect();
assert!(ids.iter().all(|id| *id == source), "runs: {ids:?}");
// A fork at the stage after the join continues from the root.
let done = all
.iter()
.zip(1_u64..)
.find(|((node, ..), _)| node == "done")
.map(|(_, ordinal)| ordinal)
.expect("`done` has a checkpoint");
let forked = cli_json(&context, &server, &[
"fork",
&source,
&format!("@{done}"),
"--json",
]);
let fork = forked["new_run_id"]
.as_str()
.expect("the new run id")
.to_string();
wait_for_success(&server, &fork).await;
let fork_workspace = workspace(&server, &fork);
assert_eq!(read(&fork_workspace, "done.txt"), "done\n");
server.shutdown();
}

20
lib/apps/fabro-server/src/server/handler/lifecycle.rs
View file

@ -35,7 +35,7 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/unarchive", post(unarchive_run))
}
async fn run_response(state: &AppState, id: RunId, status: StatusCode) -> Response {
pub(super) async fn run_response(state: &AppState, id: RunId, status: StatusCode) -> Response {
match state.stores.run_summaries.get(&id, Utc::now()).await {
Ok(Some(summary)) => {
(status, Json(state.decorate_run_summary(summary).await)).into_response()
@ -116,7 +116,21 @@ pub(in crate::server) async fn queue_run_start(
));
}
}
queue_run(state, id, &run_state, resume, actor).await
}
/// Queue the run for the scheduler: record that its start was requested
/// and that it is runnable (or pending approval), and register it as a
/// managed run in start or resume mode. The caller has checked that the run
/// may be queued; a fork, seeded to resume, is queued here without the
/// checkpoint check a resume of an interrupted run makes.
pub(super) async fn queue_run(
state: &AppState,
id: RunId,
run_state: &fabro_store::RunProjection,
resume: bool,
actor: Principal,
) -> Result<(), ApiError> {
let run_dir = Storage::new(state.server_storage_dir())
.run_scratch(&id)
.root()
@ -810,7 +824,7 @@ async fn batch_delete_runs(
}
#[derive(Clone, Copy)]
enum ArchiveAction {
pub(super) enum ArchiveAction {
Archive,
Unarchive,
}
@ -986,7 +1000,7 @@ fn batch_result_failure(
/// Archive a terminal run, or unarchive one: idempotent either way, refused
/// with a precondition error when the run is not terminal.
async fn run_archive_operation(
pub(super) async fn run_archive_operation(
state: &AppState,
id: &RunId,
actor: Option<Principal>,

364
lib/apps/fabro-server/src/server/handler/lineage.rs Normal file
View file

@ -0,0 +1,364 @@
//! A run's checkpoint timeline, and the runs made from it: fork, rewind and
//! retry (the integration plan's F5.1).
//!
//! The timeline is the run's `checkpoint` platform records, labelled with
//! the stages the projector folded them onto. A fork resolves a target on
//! that timeline (`@ordinal`, a node, or `node@visit`; the latest checkpoint
//! by default), creates the new run's row (`fabro_workflow::operations`),
//! seeds its records, checkpoints, snapshots and run branch from the source
//! (`fabro_petri::fork`), and queues it in resume mode, so its worker
//! acquires a fresh workspace, restores the checkpoint's commit into it and
//! continues from the position. A rewind is a fork of a terminal run that
//! archives the source and records `run.superseded` on it; a retry is a
//! fork of a terminal run at its last checkpoint, with the failed stage run
//! again when the run failed on one.
use std::sync::Arc;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::routing::{get, post};
use axum::{Json, Router};
use fabro_api::types as api;
use fabro_config::Storage;
use fabro_petri::SqliteRunStore;
use fabro_petri::fork::{self as petri_fork, ForkError, ForkRequest};
use fabro_petri::petri::RunStore;
use fabro_petri::platform_records::SqlitePlatformRecords;
use fabro_store::{PlatformRecordKind, RunProjection};
use fabro_types::{FailureReason, Principal, RunId};
use fabro_util::error as error_util;
use fabro_workflow::Error as WorkflowError;
use fabro_workflow::operations::{self, ForkTarget, ResolvedForkTarget, RunTimeline};
use tracing::{error, warn};
use super::super::{
ApiError, AppState, RequireRunManagementTarget, RequiredUser, parse_run_id_path, run_records,
};
use super::lifecycle::{ArchiveAction, queue_run, run_archive_operation, run_response};
use super::runs::run_provenance;
pub(super) fn routes() -> Router<Arc<AppState>> {
Router::new()
.route("/runs/{id}/timeline", get(run_timeline))
.route("/runs/{id}/fork", post(fork_run))
.route("/runs/{id}/rewind", post(rewind_run))
.route("/runs/{id}/retry", post(retry_run))
}
/// Which operation a fork is made for: what it requires of the source and
/// what it records.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum ForkKind {
Fork,
Rewind,
Retry,
}
/// A fork made and queued.
struct ForkOutcome {
source_run_id: RunId,
new_run_id: RunId,
target: ResolvedForkTarget,
rerun_last: bool,
}
async fn run_timeline(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
Path(id): Path<String>,
) -> Response {
let id = match parse_run_id_path(&id) {
Ok(id) => id,
Err(response) => return response,
};
let projection = match run_records::require_projection(state.as_ref(), id).await {
Ok(projection) => projection,
Err(err) => return err.into_response(),
};
match timeline(state.as_ref(), id).await {
Ok(timeline) => Json(timeline_response(&timeline, &projection)).into_response(),
Err(err) => err.into_response(),
}
}
async fn fork_run(
RequireRunManagementTarget(id, actor): RequireRunManagementTarget,
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: Option<Json<api::ForkRequest>>,
) -> Response {
let target = match parse_fork_target(body.and_then(|Json(body)| body.target)) {
Ok(target) => target,
Err(err) => return err.into_response(),
};
match fork_at(state.as_ref(), id, actor, &headers, ForkKind::Fork, target).await {
Ok(outcome) => (
StatusCode::OK,
Json(api::ForkResponse {
source_run_id: outcome.source_run_id.to_string(),
new_run_id: outcome.new_run_id.to_string(),
target: outcome.target.response_target(),
checkpoint_sha: outcome.target.checkpoint_sha.clone(),
execution: outcome.target.position.execution,
firing: outcome.target.position.firing,
rerun_last: outcome.rerun_last,
}),
)
.into_response(),
Err(err) => err.into_response(),
}
}
async fn rewind_run(
RequireRunManagementTarget(id, actor): RequireRunManagementTarget,
State(state): State<Arc<AppState>>,
headers: HeaderMap,
body: Option<Json<api::RewindRequest>>,
) -> Response {
let target = match parse_fork_target(body.and_then(|Json(body)| body.target)) {
Ok(target) => target,
Err(err) => return err.into_response(),
};
let outcome = match fork_at(
state.as_ref(),
id,
actor.clone(),
&headers,
ForkKind::Rewind,
target,
)
.await
{
Ok(outcome) => outcome,
Err(err) => return err.into_response(),
};
// The source is replaced: archived, and marked with what replaced it.
// A failed archive leaves the new run in place and says so.
let archived = run_archive_operation(state.as_ref(), &id, Some(actor), ArchiveAction::Archive)
.await
.map(|_| ());
if archived.is_ok() {
let record = operations::superseded_record(outcome.new_run_id, &outcome.target);
if let Err(err) = run_records::append(state.as_ref(), id, record).await {
error!(
source_run_id = %id,
new_run_id = %outcome.new_run_id,
error = %err,
"the rewound run was archived but its superseded record was not written"
);
}
}
let (status, archive_error) = match archived {
Ok(()) => (StatusCode::OK, None),
Err(err) => (StatusCode::MULTI_STATUS, Some(err.to_string())),
};
(
status,
Json(api::RewindResponse {
source_run_id: outcome.source_run_id.to_string(),
new_run_id: outcome.new_run_id.to_string(),
target: outcome.target.response_target(),
checkpoint_sha: outcome.target.checkpoint_sha.clone(),
execution: outcome.target.position.execution,
firing: outcome.target.position.firing,
archived: archive_error.is_none(),
archive_error,
}),
)
.into_response()
}
async fn retry_run(
RequireRunManagementTarget(id, actor): RequireRunManagementTarget,
State(state): State<Arc<AppState>>,
headers: HeaderMap,
) -> Response {
match fork_at(state.as_ref(), id, actor, &headers, ForkKind::Retry, None).await {
Ok(outcome) => run_response(state.as_ref(), outcome.new_run_id, StatusCode::CREATED).await,
Err(err) => err.into_response(),
}
}
/// The run's timeline: its checkpoint records, labelled through the
/// projector's fold state.
async fn timeline(state: &AppState, id: RunId) -> Result<RunTimeline, ApiError> {
let checkpoints = state
.stores
.run_summaries
.platform_records()
.read_kind(&id, PlatformRecordKind::Checkpoint)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
let labels = petri_fork::stage_labels(&state.stores.run_summaries.pool(), id)
.await
.map_err(|err| ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()))?;
Ok(RunTimeline::build(&checkpoints, &labels))
}
fn timeline_response(
timeline: &RunTimeline,
projection: &RunProjection,
) -> api::RunTimelineResponse {
api::RunTimelineResponse {
entries: timeline
.entries
.iter()
.map(|entry| api::TimelineEntryResponse {
ordinal: u64::try_from(entry.ordinal).unwrap_or(u64::MAX),
checkpoint_seq: entry.checkpoint_seq,
execution: entry.position.execution,
firing: entry.position.firing,
attempt: entry.position.attempt,
stage: entry.stage_id.clone(),
node_name: entry.node_name.clone(),
visit: entry.visit,
workspace: entry.workspace.clone(),
run_commit_sha: entry.run_commit_sha.clone(),
diff_summary: entry.diff_summary.as_ref().map(|summary| api::DiffSummary {
files_changed: summary.files_changed,
additions: summary.additions,
deletions: summary.deletions,
}),
})
.collect(),
forked_from: projection.forked_from.as_ref().map(fork_origin_response),
}
}
fn fork_origin_response(origin: &fabro_types::ForkOrigin) -> api::ForkOrigin {
api::ForkOrigin {
source_run_id: origin.source_run_id.to_string(),
execution: origin.execution,
firing: origin.firing,
rerun_last: origin.rerun_last,
}
}
fn parse_fork_target(target: Option<String>) -> Result<Option<ForkTarget>, ApiError> {
target
.map(|target| {
target
.parse::<ForkTarget>()
.map_err(workflow_operation_error)
})
.transpose()
}
/// Make the fork: check the source, resolve the target, create the new
/// run's row, seed it from the source, and queue it in resume mode.
async fn fork_at(
state: &AppState,
id: RunId,
actor: Principal,
headers: &HeaderMap,
kind: ForkKind,
target: Option<ForkTarget>,
) -> Result<ForkOutcome, ApiError> {
let source = run_records::require_projection(state, id).await?;
match kind {
ForkKind::Fork => operations::ensure_forkable(&source, &id),
ForkKind::Rewind => operations::ensure_rewindable(&source, &id),
ForkKind::Retry => operations::ensure_retryable(&source, &id),
}
.map_err(workflow_operation_error)?;
let timeline = timeline(state, id).await?;
let entry = match kind {
ForkKind::Retry => timeline.latest(),
ForkKind::Fork | ForkKind::Rewind => timeline.resolve_or_latest(target.as_ref()),
}
.map_err(workflow_operation_error)?;
let resolved = ResolvedForkTarget::of(entry).map_err(workflow_operation_error)?;
let rerun_last = kind == ForkKind::Retry && operations::reruns_last(source.status);
// A position Petri would refuse is refused before the new run exists.
let position = petri_fork::position(resolved.position.execution, resolved.position.firing);
let store: Arc<dyn RunStore> = state
.petri_projector
.observe_store(Arc::new(SqliteRunStore::new(state.db_pool.clone())));
petri_fork::check(store.as_ref(), id, position)
.await
.map_err(|err| fork_error(&err))?;
let new_run_id = RunId::new();
let storage = Storage::new(state.server_storage_dir());
let source_run_dir = storage.run_scratch(&id).root().to_path_buf();
let run_dir = storage.run_scratch(&new_run_id).root().to_path_buf();
let provenance = (kind == ForkKind::Retry).then(|| run_provenance(headers, &actor));
operations::persist_forked_run(state.store_ref().as_ref(), &operations::ForkedRunInput {
source: &source,
new_run_id,
run_dir: run_dir.clone(),
checkpoint_sha: resolved.checkpoint_sha.clone(),
provenance,
web_url: state.run_web_url(&new_run_id),
retried_from: (kind == ForkKind::Retry).then_some(id),
})
.await
.map_err(workflow_operation_error)?;
let seeded = petri_fork::fork(ForkRequest {
source: id,
fork: new_run_id,
source_run_dir: source_run_dir.join("petri"),
fork_run_dir: run_dir.join("petri"),
store,
records: Arc::new(SqlitePlatformRecords::new(Arc::clone(
&state.stores.run_summaries,
))),
position,
rerun_last,
settings: source.spec.settings.run.clone(),
})
.await;
if let Err(err) = seeded {
// The new run's row exists and holds nothing to continue from: it
// is reported failed with the reason, rather than left submitted.
let message = error_util::collect_chain(&err).join(": ");
warn!(source_run_id = %id, new_run_id = %new_run_id, error = %message, "the fork could not be seeded");
if let Err(record_err) = run_records::lifecycle(
state,
new_run_id,
run_records::failed(FailureReason::WorkflowError, message.clone()),
)
.await
{
error!(new_run_id = %new_run_id, error = %record_err, "the fork's failure was not recorded");
}
return Err(fork_error(&err));
}
// The fork continues as a run left in flight does: queued in resume
// mode, on the projection its seeded records folded to.
let projection = run_records::require_projection(state, new_run_id).await?;
queue_run(state, new_run_id, &projection, true, actor).await?;
Ok(ForkOutcome {
source_run_id: id,
new_run_id,
target: resolved,
rerun_last,
})
}
/// A refused position is the caller's mistake; anything else is the
/// server's.
fn fork_error(err: &ForkError) -> ApiError {
let message = error_util::collect_chain(err).join(": ");
let status = match err {
ForkError::Refused(_) => StatusCode::BAD_REQUEST,
_ => StatusCode::INTERNAL_SERVER_ERROR,
};
ApiError::new(status, message)
}
fn workflow_operation_error(err: WorkflowError) -> ApiError {
match err {
WorkflowError::Parse(message) | WorkflowError::Validation(message) => {
ApiError::bad_request(message)
}
WorkflowError::Precondition(message) => ApiError::new(StatusCode::CONFLICT, message),
WorkflowError::RunNotFound(_) => ApiError::not_found("Run not found."),
err => ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string()),
}
}

2
lib/apps/fabro-server/src/server/handler/mod.rs
View file

@ -14,6 +14,7 @@ mod environments;
pub(in crate::server) mod events;
pub(in crate::server) mod graph;
pub(in crate::server) mod lifecycle;
mod lineage;
mod llm_sse;
mod mcp_servers;
mod models;
@ -214,6 +215,7 @@ pub(super) fn real_routes() -> Router<Arc<AppState>> {
.merge(sandbox::routes())
.merge(sandboxes::routes())
.merge(lifecycle::routes())
.merge(lineage::routes())
.merge(steer::routes())
.merge(pair::routes())
.merge(petri::routes())

19
lib/components/fabro-petri/src/checkpoint.rs
View file

@ -607,6 +607,15 @@ impl RunWorkspaces {
Ok(self.head(site).await?.as_deref() == Some(sha) && self.is_clean(site).await?)
}
/// Whether the host workspace's repository holds the commit `sha`, so a
/// reset can reach it; a directory that is no repository holds none.
pub async fn has_commit(&self, workspace: &str, sha: &str) -> Result<bool, CheckpointError> {
if !self.workspace_exists(workspace).await {
return Ok(false);
}
self.has_commit_at(&self.host(workspace), sha).await
}
/// Whether a sandbox workspace's repository holds the commit `sha`, so
/// a reset can reach it without a transfer.
pub async fn has_commit_in(
@ -614,16 +623,20 @@ impl RunWorkspaces {
env: &Arc<dyn ExecEnv>,
sha: &str,
) -> Result<bool, CheckpointError> {
let site = Site::Sandbox(Arc::clone(env));
self.has_commit_at(&Site::Sandbox(Arc::clone(env)), sha)
.await
}
async fn has_commit_at(&self, site: &Site, sha: &str) -> Result<bool, CheckpointError> {
if self
.git_status(&site, "rev-parse", &["rev-parse", "--git-dir"])
.git_status(site, "rev-parse", &["rev-parse", "--git-dir"])
.await?
.is_none()
{
return Ok(false);
}
Ok(self
.git_status(&site, "cat-file", &[
.git_status(site, "cat-file", &[
"cat-file",
"-e",
&format!("{sha}^{{commit}}"),

534
lib/components/fabro-petri/src/fork.rs Normal file
View file

@ -0,0 +1,534 @@
//! Forking a Fabro run at a checkpoint: the seam over Petri's
//! `host::fork_from` that rewind, fork and retry are built on (the
//! integration plan's F5.1).
//!
//! Fabro's checkpoint record ties a Petri position `(execution, firing)` to
//! a Git commit. A fork seeds a new run from the source's records up to such
//! a position and leaves it ready to resume, in three steps:
//!
//! 1. Petri's `fork_from` writes the new run's records into the server's store
//! under the new run id: the same graphs, the position execution's engine
//! log cut after the firing's routing (before its first record with
//! `rerun_last`), the finished children the kept firings called, and a
//! `run.started` whose `forked_from` names the source and the position. No
//! sandbox lease is carried over, so the resume acquires the position
//! execution's scopes fresh.
//! 2. The source's checkpoint records for every attempt the fork kept are
//! written again under the new run, at their positions, and the snapshot
//! repository of every workspace they name is seeded with those checkpoints'
//! refs alone, fetched from the source's repository under the source's run
//! scratch. That is what the resume's recovery plan
//! ([`crate::recovery::plan`]) reads: the last durable finish of the
//! position execution names the snapshot the fresh workspace is restored to
//! at `scope_acquired`, on the host and in a sandbox alike.
//! 3. The fork's `run.branch` record names the run branch the restore creates
//! (`fabro/run/<new id>`) and the commit it starts from, with the
//! `git.identity` beside it, both at the checkpoint's position, so the hooks
//! record nothing twice and the run's diff is measured from the fork point.
//!
//! The Fabro run row (`run.created`, the lifecycle records) and the launch in
//! resume mode are the caller's: `fabro_workflow::operations` shapes the
//! records and the server launches the worker.
use std::collections::{BTreeMap, BTreeSet};
use std::path::PathBuf;
use std::process::Stdio;
use std::sync::Arc;
use fabro_checkpoint::author::GitAuthor;
use fabro_db::DbPool;
use fabro_store::platform_records::{CheckpointRecord, GitIdentityRecord, RunBranchRecord};
use fabro_store::{PlatformRecord, PlatformRecordKind, StagePosition, StoredPlatformRecord};
use fabro_types::settings::run::RunNamespace;
use fabro_types::{GitIdentity, GitIdentitySource, RunId};
use fabro_workflow::operations::{StageLabel, StageLabels};
use petri_execution::host::{self, ForkOptions, ForkOrigin, ForkPosition, HostError};
use petri_execution::inspect::{self, InspectError};
use petri_execution::{
Access, CoordinatorEvent, ExecutionId, InvocationId, RunKey, RunStore,
StoreError as CoordinatorStoreError,
};
use petri_runtime::ir::FiringId;
use petri_runtime::{RunOptions, Runtime};
use petri_store::StoreError;
use tokio::fs;
use tokio::process::Command;
use tracing::{debug, info};
use crate::checkpoint::{CheckpointKey, RunWorkspaces};
use crate::platform_records::{PlatformRecordError, PlatformRecords};
use crate::projection::FoldState;
use crate::projector::ProjectError;
/// One fork to seed.
pub struct ForkRequest {
/// The run whose records are copied.
pub source: RunId,
/// The new run's id: its Petri run key and its own run scratch.
pub fork: RunId,
/// The source's Petri run directory (its scratch's `petri`), where its
/// snapshot repositories are.
pub source_run_dir: PathBuf,
/// The fork's Petri run directory, where its snapshot repositories go.
pub fork_run_dir: PathBuf,
/// The server's run store: the source is read from it, the fork is
/// written into it.
pub store: Arc<dyn RunStore>,
/// The platform records of both runs.
pub records: Arc<dyn PlatformRecords>,
/// The position the source's records are kept up to.
pub position: ForkPosition,
/// Whether the position's firing runs again (a retry of a failed
/// stage) instead of keeping its finish.
pub rerun_last: bool,
/// The run's settings, for its Git author and checkpoint settings.
pub settings: RunNamespace,
}
/// A seeded fork, not yet resumed.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Forked {
/// The source and position, as the fork's own run declaration records
/// them.
pub origin: ForkOrigin,
/// The checkpoints the fork kept, in the source's record order.
pub checkpoints: Vec<KeptCheckpoint>,
/// The snapshot the fork's workspace starts on, when the kept records
/// name one for the position execution's last durable finish.
pub start: Option<KeptCheckpoint>,
}
/// A source checkpoint the fork carries over.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct KeptCheckpoint {
pub key: CheckpointKey,
pub sha: String,
/// The Petri workspace id the commit was made in.
pub workspace: Option<String>,
}
/// Why the fork could not be seeded.
#[derive(Debug, thiserror::Error)]
pub enum ForkError {
#[error("the source run's record could not be opened")]
Open(#[source] StoreError),
/// Petri refused the position: an unknown execution, a firing whose
/// finish was not routed, or a position inside a child invocation (a
/// branch of a parallel node). The message says which.
#[error("{0}")]
Refused(String),
#[error("Petri could not seed the fork")]
Seed(#[source] HostError),
#[error("the fork's record could not be inspected")]
Inspect(#[source] InspectError),
#[error("the fork's coordinator log could not be read")]
Log(#[source] CoordinatorStoreError),
#[error("the checkpoint records could not be read or written")]
Records(#[source] PlatformRecordError),
#[error("the snapshot repository for `{workspace}` could not be seeded: {detail}")]
Snapshots {
workspace: String,
detail: String,
},
}
/// Refuse a position Petri would refuse, before anything is written for
/// the fork: an execution the source does not have, or one inside a child
/// invocation (a branch of a parallel node), whose caller's firing is live
/// at every position inside it. The messages are Petri's own.
pub async fn check(
store: &dyn RunStore,
source: RunId,
position: ForkPosition,
) -> Result<(), ForkError> {
let logs = store
.open(&RunKey::new(source.to_string()), Access::Read)
.await
.map_err(ForkError::Open)?;
let state = host::stored_state(&*logs).await.map_err(ForkError::Seed)?;
let Some(execution) = state.executions.get(&position.execution) else {
return Err(ForkError::Refused(format!(
"the source run has no execution {}",
position.execution
)));
};
let invocation = execution.declaration.invocation;
if invocation != InvocationId::ROOT {
return Err(ForkError::Refused(format!(
"execution {} belongs to invocation {invocation}, not the root: a position inside a \
child invocation cannot be forked",
position.execution
)));
}
Ok(())
}
/// Seed the fork: Petri's records, then the kept checkpoints, their
/// snapshots and the run branch. The new run must not exist in the store
/// yet.
pub async fn fork(request: ForkRequest) -> Result<Forked, ForkError> {
let source_key = RunKey::new(request.source.to_string());
let fork_key = RunKey::new(request.fork.to_string());
let source_logs = request
.store
.open(&source_key, Access::Read)
.await
.map_err(ForkError::Open)?;
let mut options = RunOptions::new(&request.fork_run_dir);
options.run_key = Some(fork_key.clone());
let runtime = Runtime::standard()
.options(options)
.store(Arc::clone(&request.store));
let forked = host::fork_from(&runtime, &*source_logs, request.position, ForkOptions {
rerun_last: request.rerun_last,
})
.await
.map_err(|error| match error {
HostError::Fork(refused) => ForkError::Refused(refused.to_string()),
other => ForkError::Seed(other),
})?;
drop(source_logs);
info!(
source = %request.source,
fork = %request.fork,
position = %request.position,
rerun_last = request.rerun_last,
"Petri seeded the fork's records"
);
// What the fork kept: every attempt with a durable finish in its
// records, and the position execution's last one.
let fork_logs = request
.store
.open(&fork_key, Access::Read)
.await
.map_err(ForkError::Open)?;
let inspection = inspect::inspect_run(&*fork_logs)
.await
.map_err(ForkError::Inspect)?;
drop(fork_logs);
let mut kept_keys = BTreeSet::new();
let mut start_key = None;
for execution in &inspection.executions {
let Some(engine) = execution.engine.as_ref() else {
continue;
};
for attempt in &engine.attempts {
let key = CheckpointKey {
execution: execution.execution.raw(),
firing: attempt.firing,
attempt: attempt.attempt,
};
kept_keys.insert(key);
if execution.execution == request.position.execution {
start_key = Some(key);
}
}
}
// The source's checkpoint records for the kept attempts, written again
// under the fork at their positions.
let source_checkpoints = request
.records
.read_kind(&request.source, PlatformRecordKind::Checkpoint)
.await
.map_err(ForkError::Records)?;
let mut checkpoints = Vec::new();
for stored in source_checkpoints {
let PlatformRecord::Checkpoint(record) = &stored.record else {
continue;
};
let Some(key) = checkpoint_key(record) else {
continue;
};
if !kept_keys.contains(&key) {
continue;
}
let Some(sha) = record.git_commit_sha.clone() else {
continue;
};
let mut copied = record.clone();
copied.attempt = Some(key.attempt);
copied.operation = Some(key.operation());
request
.records
.append(
&request.fork,
&PlatformRecord::Checkpoint(copied),
Some(StagePosition {
execution: key.execution,
firing: key.firing,
}),
)
.await
.map_err(ForkError::Records)?;
checkpoints.push(KeptCheckpoint {
key,
sha,
workspace: record.workspace.clone(),
});
}
// The snapshot repositories: one per workspace the kept checkpoints
// name, holding those checkpoints' refs alone.
let author = request
.settings
.git
.author
.as_ref()
.map(GitAuthor::from)
.unwrap_or_default();
let source_workspaces = RunWorkspaces::new(
request.source_run_dir.clone(),
request.source.to_string(),
author.clone(),
&request.settings.checkpoint,
);
let fork_workspaces = RunWorkspaces::new(
request.fork_run_dir.clone(),
request.fork.to_string(),
author.clone(),
&request.settings.checkpoint,
);
let mut by_workspace: BTreeMap<String, Vec<CheckpointKey>> = BTreeMap::new();
for kept in &checkpoints {
if let Some(workspace) = &kept.workspace {
by_workspace
.entry(workspace.clone())
.or_default()
.push(kept.key);
}
}
for (workspace, keys) in &by_workspace {
seed_snapshots(&source_workspaces, &fork_workspaces, workspace, keys).await?;
}
// The run branch the restore creates, from the checkpoint the fork
// starts on, and the identity that authors the fork's commits.
let start = start_key.and_then(|key| checkpoints.iter().find(|kept| kept.key == key).cloned());
if let Some(start) = &start {
let position = StagePosition {
execution: start.key.execution,
firing: start.key.firing,
};
let branch = PlatformRecord::RunBranch(RunBranchRecord {
run_branch: Some(fork_workspaces.run_branch()),
base_sha: Some(start.sha.clone()),
workspace: start.workspace.clone(),
});
request
.records
.append(&request.fork, &branch, Some(position))
.await
.map_err(ForkError::Records)?;
let identity = PlatformRecord::GitIdentity(GitIdentityRecord {
identity: GitIdentity {
name: author.name.clone(),
email: author.email.clone(),
source: if author.is_default() {
GitIdentitySource::Default
} else {
GitIdentitySource::Explicit
},
},
});
request
.records
.append(&request.fork, &identity, Some(position))
.await
.map_err(ForkError::Records)?;
info!(
fork = %request.fork,
sha = start.sha,
execution = start.key.execution,
firing = start.key.firing,
"the fork's run branch starts at the position's checkpoint"
);
} else {
debug!(
fork = %request.fork,
"the fork keeps no checkpoint; its workspace starts empty"
);
}
Ok(Forked {
origin: forked.origin,
checkpoints,
start,
})
}
/// The key a checkpoint record names: its operation identity, else its
/// position with the attempt it recorded.
fn checkpoint_key(record: &CheckpointRecord) -> Option<CheckpointKey> {
record
.operation
.as_ref()
.and_then(CheckpointKey::from_operation)
.or_else(|| {
Some(CheckpointKey {
execution: record.execution,
firing: record.firing,
attempt: record.attempt?,
})
})
}
/// Create the fork's bare snapshot repository for `workspace` and fetch the
/// kept checkpoints' refs into it from the source's.
async fn seed_snapshots(
source: &RunWorkspaces,
fork: &RunWorkspaces,
workspace: &str,
keys: &[CheckpointKey],
) -> Result<(), ForkError> {
let failed = |detail: String| ForkError::Snapshots {
workspace: workspace.to_string(),
detail,
};
let source_repository = source.snapshot_repository(workspace);
if !fs::try_exists(&source_repository).await.unwrap_or(false) {
return Err(failed(format!(
"the source run has no snapshot repository at {}",
source_repository.display()
)));
}
let repository = fork.snapshot_repository(workspace);
fs::create_dir_all(&repository).await.map_err(|error| {
failed(format!(
"{} could not be created: {error}",
repository.display()
))
})?;
git(&repository, &["init", "-q", "--bare"])
.await
.map_err(failed)?;
let mut args = vec![
"fetch".to_string(),
"-q".to_string(),
source_repository.to_string_lossy().into_owned(),
];
for key in keys {
let name = key.snapshot_ref();
args.push(format!("+{name}:{name}"));
}
git(&repository, &args).await.map_err(failed)?;
debug!(
workspace,
refs = keys.len(),
repository = %repository.display(),
"the fork's snapshot repository is seeded"
);
Ok(())
}
/// Run `git` in `repository`; a non-zero exit is the error's detail.
async fn git<S: AsRef<str>>(repository: &std::path::Path, args: &[S]) -> Result<(), String> {
let output = Command::new("git")
.args(args.iter().map(AsRef::as_ref))
.current_dir(repository)
.stdin(Stdio::null())
.output()
.await
.map_err(|error| format!("git could not run: {error}"))?;
if output.status.success() {
Ok(())
} else {
Err(format!(
"git {} failed ({}): {}",
args.first().map_or("", AsRef::as_ref),
output.status,
String::from_utf8_lossy(&output.stderr).trim()
))
}
}
/// The position a checkpoint's execution and firing name, in Petri's ids.
#[must_use]
pub fn position(execution: u64, firing: u64) -> ForkPosition {
ForkPosition {
execution: ExecutionId::new(execution),
firing: FiringId::new(firing),
}
}
/// A fork origin as Fabro's projection shows it.
#[must_use]
pub fn origin_view(origin: &ForkOrigin) -> Option<fabro_types::ForkOrigin> {
Some(fabro_types::ForkOrigin {
source_run_id: origin.source.to_string().parse().ok()?,
execution: origin.position.execution.raw(),
firing: origin.position.firing.raw(),
rerun_last: origin.rerun_last,
})
}
/// Where a run came from, when it is a fork: the `forked_from` of its run
/// declaration. `None` for a run that is not a fork, or that has no record
/// yet.
pub async fn origin_of(
store: &dyn RunStore,
run_id: RunId,
) -> Result<Option<fabro_types::ForkOrigin>, ForkError> {
let logs = match store
.open(&RunKey::new(run_id.to_string()), Access::Read)
.await
{
Ok(logs) => logs,
Err(StoreError::NotFound { .. }) => return Ok(None),
Err(error) => return Err(ForkError::Open(error)),
};
let records = petri_execution::read_coordinator_log(&*logs)
.await
.map_err(ForkError::Log)?;
Ok(records.first().and_then(|record| match &record.body {
CoordinatorEvent::RunStarted {
forked_from: Some(origin),
..
} => origin_view(origin),
_ => None,
}))
}
/// The stages of a run by `(execution, firing)`, as the projector's fold
/// state names them: what a timeline labels its checkpoints with, and what
/// a fork target such as `build@2` resolves through. `views` is the pool
/// the view tables live in.
pub async fn stage_labels(views: &DbPool, run_id: RunId) -> Result<StageLabels, ProjectError> {
let fold_json: Option<String> =
sqlx::query_scalar("SELECT fold_json FROM petri_projection WHERE run_id = ?")
.bind(run_id.to_string())
.fetch_optional(views)
.await
.map_err(ProjectError::Database)?;
let Some(fold_json) = fold_json else {
return Ok(BTreeMap::new());
};
let state: FoldState = serde_json::from_str(&fold_json).map_err(ProjectError::Encode)?;
Ok(state
.stages
.iter()
.filter_map(|(key, stage)| {
let (execution, firing) = key.split_once(':')?;
Some((
(execution.parse().ok()?, firing.parse().ok()?),
StageLabel {
stage_id: stage.shown.then(|| stage.stage_id.to_string()),
node_name: stage.node_name.clone(),
visit: stage.visit,
},
))
})
.collect())
}
/// The checkpoint records of a run, in seq order.
pub async fn checkpoints(
records: &dyn PlatformRecords,
run_id: RunId,
) -> Result<Vec<StoredPlatformRecord>, PlatformRecordError> {
records
.read_kind(&run_id, PlatformRecordKind::Checkpoint)
.await
}

47
lib/components/fabro-petri/src/hooks.rs
View file

@ -62,12 +62,14 @@
//! `git` inside the scope through it, and move the commit out as a bundle
//! into the same snapshot repository the host path pushes to. Artifacts are
//! read out through the same environment on every provider. The same
//! point is where a resumed run brings a sandbox workspace to the snapshot
//! its durable state names, before the first attempt runs in it: verified,
//! point is where a resumed run brings a workspace to the snapshot its
//! durable state names, before the first attempt runs in it: verified,
//! reset, or, in a fresh sandbox (Petri replaces a lost one on Fabro's
//! request), restored from a bundle of the checkpoint. The plan is
//! [`recovery::plan`], the one the server applied to host workspaces before
//! it relaunched the worker.
//! it relaunched the worker; a host workspace is verified here, unless the
//! run is a fork whose fresh workspace nothing restored yet
//! ([`crate::fork`]), which is restored from the seeded snapshot repository.
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::{Path, PathBuf};
@ -659,6 +661,37 @@ impl FabroHooks {
.await
}
/// Bring a host workspace to the snapshot the resumed run's durable
/// state names, once, at its first acquisition. After a restart the
/// server already brought it there, so this verifies; a fork's fresh
/// workspace is restored here from the snapshot repository the fork
/// seeded.
async fn restore_host(&self, workspace: &str) -> Result<(), ScopeAcquiredError> {
let targets = self.restore_targets().await?;
let target = lock(targets).remove(workspace);
let Some(target) = target else {
return Ok(());
};
let serialized = self.workspace_lock(workspace);
let _held = serialized.lock().await;
let action = recovery::bring_host_to(&self.workspaces, workspace, &target)
.await
.map_err(|error| {
ScopeAcquiredError::new(format!(
"the host workspace `{workspace}` could not be brought to its snapshot: {}",
collect_chain(&error).join(": ")
))
})?;
info!(
run_id = %self.run_id,
workspace,
sha = target.sha,
action = ?action,
"host workspace brought to its durable snapshot"
);
Ok(())
}
/// Bring a sandbox workspace to the snapshot the resumed run's durable
/// state names, once, at its first acquisition.
async fn restore_sandbox(
@ -1279,10 +1312,14 @@ impl ExecutionHooks for FabroHooks {
(context.execution, acquired.scope),
(workspace.clone(), Arc::clone(&acquired.env)),
);
if self.host_workspaces || !self.resumed {
if !self.resumed {
return Ok(());
}
self.restore_sandbox(&workspace, &acquired.env).await
if self.host_workspaces {
self.restore_host(&workspace).await
} else {
self.restore_sandbox(&workspace, &acquired.env).await
}
}
}

7
lib/components/fabro-petri/src/lib.rs
View file

@ -48,7 +48,11 @@
//! - [`host_tools`]: Fabro's run tools on every native agent session of a run,
//! through Petri's `HostTools` capability;
//! - [`controls`]: the controls Fabro drives on a live run (pause, unpause,
//! steer, cancel), over Petri's control service.
//! steer, cancel), over Petri's control service;
//! - [`fork`]: a run seeded from another's records up to a checkpoint's
//! position, over Petri's `host::fork_from`, with the kept checkpoints, their
//! snapshots and the run branch carried over: what rewind, fork and retry are
//! built on.
//!
//! The Petri packages are pinned by revision in the workspace `Cargo.toml`
//! under `petri_*` keys.
@ -59,6 +63,7 @@ pub mod check;
pub mod checkpoint;
pub mod controls;
pub mod engine;
pub mod fork;
pub mod hooks;
pub mod host_tools;
pub mod http_store;

15
lib/components/fabro-petri/src/projection.rs
View file

@ -383,10 +383,23 @@ impl RunView {
fn fold_coordinator(&mut self, record: &CoordinatorEvent, event: &RunEvent, at: DateTime<Utc>) {
match record {
CoordinatorEvent::RunStarted { root, .. } => {
CoordinatorEvent::RunStarted {
root, forked_from, ..
} => {
self.state.root = Some(root.raw());
self.state.started_at = Some(event.recorded_at);
if let Some(projection) = self.projection.as_mut() {
// A fork's declaration names its source; a parse failure
// means the source was not a Fabro run, which the
// projection cannot show.
projection.forked_from = forked_from.as_ref().and_then(|origin| {
Some(fabro_types::ForkOrigin {
source_run_id: origin.source.as_str().parse().ok()?,
execution: origin.position.execution.raw(),
firing: origin.position.firing.raw(),
rerun_last: origin.rerun_last,
})
});
apply_status(projection, RunStatus::Running, at);
projection.start = Some(StartRecord {
start_time: at,

19
lib/components/fabro-petri/src/recovery.rs
View file

@ -14,8 +14,8 @@
//! when the record was lost to the crash, the commit found by its key in the
//! workspace's snapshot repository or history, which is then recorded again;
//! - a workspace that survives is verified to sit on that commit, unchanged, or
//! reset to it; a workspace that is gone is restored from the run's snapshot
//! repository into a fresh directory;
//! reset to it; a workspace that is gone, or a fresh one with no history (a
//! fork's first acquisition), is restored from the run's snapshot repository;
//! - a durable finish with no snapshot fails the run with a named error rather
//! than resume it on stale files.
//!
@ -321,7 +321,7 @@ pub async fn recover(request: RecoveryRequest) -> Result<Recovery, RecoveryError
let mut recovered = Vec::new();
for (workspace, target) in targets {
let action = if request.host_workspaces {
bring_to(&workspaces, &workspace, &target).await?
bring_host_to(&workspaces, &workspace, &target).await?
} else {
WorkspaceAction::Deferred
};
@ -470,8 +470,11 @@ async fn newest(
Ok(chosen.clone())
}
/// Verify, reset or restore the host workspace onto its target.
async fn bring_to(
/// Verify, reset or restore the host workspace onto its target: a
/// workspace that still holds the commit is verified or reset in place; a
/// gone one, or a fresh directory with no history (a fork's first
/// acquisition), is restored from the snapshot repository.
pub async fn bring_host_to(
workspaces: &RunWorkspaces,
workspace: &str,
target: &RestoreTarget,
@ -480,7 +483,11 @@ async fn bring_to(
workspace: workspace.to_string(),
source,
};
if workspaces.workspace_exists(workspace).await {
if workspaces
.has_commit(workspace, &target.sha)
.await
.map_err(failed)?
{
if workspaces
.matches(workspace, &target.sha)
.await

255
lib/components/fabro-workflow/src/operations/fork.rs Normal file
View file

@ -0,0 +1,255 @@
//! Forking a run at a checkpoint: the Fabro run the fork becomes.
//!
//! A fork is a new run whose records Petri seeds from the source's up to a
//! checkpoint's position (`fabro_petri::fork`, over the timeline here). What
//! Fabro itself makes of it is a run row like any other: the `run.created`
//! record carrying the source's spec (its admission, settings and target)
//! under the new id, with `fork_source_ref` naming the source and the
//! checkpoint's commit, and `retried_from` when the fork is a retry; then
//! the `submitted` lifecycle transition. The run is then started in resume
//! mode, as a run left in flight is.
use std::path::PathBuf;
use fabro_store::Database;
use fabro_store::platform_records::{
PlatformRecord, RunCreatedRecord, RunLifecycleKind, RunLifecycleRecord,
};
use fabro_types::{ForkSourceRef, RunId, RunProjection, RunProvenance, RunStatus};
use tokio::fs;
use super::ensure_not_archived;
use super::timeline::{TimelineEntry, TimelinePosition};
use crate::error::Error;
/// The checkpoint a fork was resolved to, as the API reports it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ResolvedForkTarget {
pub checkpoint_ordinal: usize,
pub node_id: String,
pub visit: usize,
pub position: TimelinePosition,
pub checkpoint_sha: String,
}
impl ResolvedForkTarget {
/// The entry as a fork target, refused when it has no commit.
pub fn of(entry: &TimelineEntry) -> Result<Self, Error> {
let checkpoint_sha = entry.run_commit_sha.clone().ok_or_else(|| {
Error::Validation(format!(
"checkpoint @{} has no git_commit_sha; cannot fork",
entry.ordinal
))
})?;
Ok(Self {
checkpoint_ordinal: entry.ordinal,
node_id: entry.node_name.clone(),
visit: usize::try_from(entry.visit).unwrap_or(1),
position: entry.position,
checkpoint_sha,
})
}
#[must_use]
pub fn response_target(&self) -> String {
format!("@{}", self.checkpoint_ordinal)
}
}
/// The new run a fork creates.
#[derive(Debug)]
pub struct ForkedRunInput<'a> {
pub source: &'a RunProjection,
pub new_run_id: RunId,
/// The new run's scratch directory.
pub run_dir: PathBuf,
pub checkpoint_sha: String,
/// Who forked, when the fork records its own provenance; `None` keeps
/// the source's.
pub provenance: Option<RunProvenance>,
pub web_url: Option<String>,
/// The source, when the fork is a retry of it.
pub retried_from: Option<RunId>,
}
/// A run can be forked unless it is archived.
pub fn ensure_forkable(source: &RunProjection, run_id: &RunId) -> Result<(), Error> {
ensure_not_archived(source.archived_at.is_some(), run_id)
}
/// A run must be terminal to be rewound or retried: its records are
/// complete, and nothing is writing them.
pub fn ensure_terminal(source: &RunProjection, run_id: &RunId, verb: &str) -> Result<(), Error> {
let current = source.status;
if current.is_terminal() {
Ok(())
} else {
Err(Error::Precondition(format!(
"run {run_id} must be terminal (succeeded, failed, or dead) to {verb}; current status \
is {current}"
)))
}
}
/// The `run.created` record of the fork: the source's spec under the new
/// id, naming where it came from.
#[must_use]
pub fn forked_run_record(input: &ForkedRunInput<'_>) -> RunCreatedRecord {
let mut spec = input.source.spec.clone();
spec.run_id = input.new_run_id;
spec.fork_source_ref = Some(ForkSourceRef {
source_run_id: input.source.spec.run_id,
checkpoint_sha: input.checkpoint_sha.clone(),
});
if let Some(provenance) = &input.provenance {
spec.provenance = provenance.clone();
}
RunCreatedRecord {
spec,
title: Some(input.source.title().into_owned()),
parent_id: input.source.parent_id,
retried_from: input.retried_from,
web_url: input.web_url.clone(),
}
}
/// Create the fork's run: its scratch directory, then its first records
/// (`run.created` and the `submitted` transition), which wake its
/// projector.
pub async fn persist_forked_run(store: &Database, input: &ForkedRunInput<'_>) -> Result<(), Error> {
fs::create_dir_all(&input.run_dir).await.map_err(|err| {
Error::Io(format!(
"creating run directory {}: {err}",
input.run_dir.display()
))
})?;
let created = PlatformRecord::RunCreated(forked_run_record(input));
let submitted = PlatformRecord::RunLifecycle(
RunLifecycleRecord::new(RunLifecycleKind::Submitted).with_status(RunStatus::Submitted),
);
let summaries = store.run_summary_store();
let platform_records = summaries.platform_records();
for record in [created, submitted] {
platform_records
.append(&input.new_run_id, &record, None)
.await
.map_err(|err| Error::engine_with_source("run store operation failed", err))?;
}
summaries.notify_platform_record(input.new_run_id);
Ok(())
}
#[cfg(test)]
mod tests {
use chrono::Utc;
use fabro_types::{FailureReason, Graph, PetriAdmission, RunSpec, WorkflowSettings, fixtures};
use super::*;
fn source(status: RunStatus) -> RunProjection {
let mut projection = RunProjection::new(
"Source title".to_string(),
RunSpec {
run_id: fixtures::RUN_1,
settings: WorkflowSettings::default(),
graph: Graph::new("source"),
graph_source: Some("digraph source { start -> exit }".to_string()),
workflow_slug: Some("source".to_string()),
workflow_version_id: None,
target: None,
automation: None,
source_directory: None,
labels: std::collections::HashMap::new(),
provenance: fabro_types::test_support::test_run_provenance(),
definition_blob: None,
spec_blob: None,
git: None,
fork_source_ref: None,
admission: PetriAdmission::default(),
},
Utc::now(),
);
projection.status = status;
projection.parent_id = Some(fixtures::RUN_2);
projection
}
fn entry(ordinal: usize, sha: Option<&str>) -> TimelineEntry {
TimelineEntry {
ordinal,
checkpoint_seq: 3,
position: TimelinePosition {
execution: 0,
firing: 2,
attempt: 1,
},
stage_id: Some("build@1".to_string()),
node_name: "build".to_string(),
visit: 1,
workspace: None,
run_commit_sha: sha.map(ToOwned::to_owned),
diff_summary: None,
}
}
#[test]
fn the_record_carries_the_source_spec_under_the_new_id_and_names_the_source() {
let source = source(RunStatus::Succeeded {
reason: fabro_types::SuccessReason::Completed,
});
let record = forked_run_record(&ForkedRunInput {
source: &source,
new_run_id: fixtures::RUN_3,
run_dir: PathBuf::from("/tmp/unused"),
checkpoint_sha: "abc".to_string(),
provenance: None,
web_url: Some("http://localhost/runs/x".to_string()),
retried_from: Some(fixtures::RUN_1),
});
assert_eq!(record.spec.run_id, fixtures::RUN_3);
assert_eq!(
record.spec.fork_source_ref,
Some(ForkSourceRef {
source_run_id: fixtures::RUN_1,
checkpoint_sha: "abc".to_string(),
})
);
assert_eq!(record.spec.graph.name, "source");
assert_eq!(record.title.as_deref(), Some("Source title"));
assert_eq!(record.parent_id, Some(fixtures::RUN_2));
assert_eq!(record.retried_from, Some(fixtures::RUN_1));
assert_eq!(record.web_url.as_deref(), Some("http://localhost/runs/x"));
}
#[test]
fn a_target_needs_a_commit() {
let resolved = ResolvedForkTarget::of(&entry(2, Some("abc"))).unwrap();
assert_eq!(resolved.response_target(), "@2");
assert_eq!(resolved.checkpoint_sha, "abc");
assert!(matches!(
ResolvedForkTarget::of(&entry(2, None)),
Err(Error::Validation(message)) if message.contains("no git_commit_sha")
));
}
#[test]
fn a_rewind_or_retry_needs_a_terminal_source() {
let running = source(RunStatus::Running);
assert!(matches!(
ensure_terminal(&running, &fixtures::RUN_1, "rewind"),
Err(Error::Precondition(message)) if message.contains("must be terminal")
));
for status in [
RunStatus::Dead,
RunStatus::Failed {
reason: FailureReason::Cancelled,
},
RunStatus::Succeeded {
reason: fabro_types::SuccessReason::Completed,
},
] {
ensure_terminal(&source(status), &fixtures::RUN_1, "retry").unwrap();
}
ensure_forkable(&running, &fixtures::RUN_1).unwrap();
}
}

13
lib/components/fabro-workflow/src/operations/mod.rs
View file

@ -1,5 +1,9 @@
mod create;
mod fork;
mod retry;
mod rewind;
mod source;
mod timeline;
mod validate;
pub use create::{
@ -8,7 +12,16 @@ pub use create::{
make_run_dir, materialize_admitted_run, persist_create_run,
};
use fabro_types::RunId;
pub use fork::{
ForkedRunInput, ResolvedForkTarget, ensure_forkable, ensure_terminal, forked_run_record,
persist_forked_run,
};
pub use retry::{ensure_retryable, reruns_last};
pub use rewind::{ensure_rewindable, superseded_record};
pub use source::WorkflowInput;
pub use timeline::{
ForkTarget, RunTimeline, StageLabel, StageLabels, TimelineEntry, TimelinePosition,
};
pub use validate::{ValidateInput, validate};
pub use crate::error::Error;

47
lib/components/fabro-workflow/src/operations/retry.rs Normal file
View file

@ -0,0 +1,47 @@
//! Retrying a run: a fork from its last checkpoint.
//!
//! A retry forks a terminal run at its last checkpoint. When the run failed
//! on a stage (its last durable finish is the failed stage's), the position's
//! firing runs again, so the retry reruns the failed stage on the files of
//! the stage before it; a run that succeeded, was cancelled or died forks at
//! the last position as it stands, and continues from there.
use fabro_types::{FailureReason, RunId, RunProjection, RunStatus};
use super::fork::{ensure_forkable, ensure_terminal};
use crate::error::Error;
/// A run can be retried when it is terminal and not archived.
pub fn ensure_retryable(source: &RunProjection, run_id: &RunId) -> Result<(), Error> {
ensure_forkable(source, run_id)?;
ensure_terminal(source, run_id, "retry")
}
/// Whether the retry reruns the last checkpointed stage: it does when the
/// run failed on its own terms, since that stage's finish is the failure.
#[must_use]
pub fn reruns_last(status: RunStatus) -> bool {
matches!(
status,
RunStatus::Failed { reason } if reason != FailureReason::Cancelled
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_failed_run_reruns_its_last_stage_and_the_others_continue() {
assert!(reruns_last(RunStatus::Failed {
reason: FailureReason::WorkflowError,
}));
assert!(!reruns_last(RunStatus::Failed {
reason: FailureReason::Cancelled,
}));
assert!(!reruns_last(RunStatus::Dead));
assert!(!reruns_last(RunStatus::Succeeded {
reason: fabro_types::SuccessReason::Completed,
}));
}
}

30
lib/components/fabro-workflow/src/operations/rewind.rs Normal file
View file

@ -0,0 +1,30 @@
//! Rewinding a run: a fork that replaces its source.
//!
//! A rewind forks a terminal run at a checkpoint (`fork`), then archives the
//! source and records `run.superseded` on it, naming the new run and the
//! checkpoint it continues from. The archive is the server's own archive
//! operation; what this module holds is the precondition and the record.
use fabro_store::platform_records::{PlatformRecord, RunSupersededRecord};
use fabro_types::{RunId, RunProjection};
use super::fork::{ResolvedForkTarget, ensure_forkable, ensure_terminal};
use crate::error::Error;
/// A run can be rewound when it is terminal and not archived.
pub fn ensure_rewindable(source: &RunProjection, run_id: &RunId) -> Result<(), Error> {
ensure_forkable(source, run_id)?;
ensure_terminal(source, run_id, "rewind")
}
/// The record a rewound source carries: which run replaced it, from which
/// checkpoint.
#[must_use]
pub fn superseded_record(new_run_id: RunId, target: &ResolvedForkTarget) -> PlatformRecord {
PlatformRecord::RunSuperseded(RunSupersededRecord {
new_run_id,
target_checkpoint_ordinal: target.checkpoint_ordinal,
target_node_id: target.node_id.clone(),
target_visit: target.visit,
})
}

327
lib/components/fabro-workflow/src/operations/timeline.rs Normal file
View file

@ -0,0 +1,327 @@
//! The checkpoint timeline of a run: every checkpoint Fabro recorded, at
//! its Petri position, with the commit it made and the stage it belongs
//! to, and the targets a fork names one of them by.
use std::collections::BTreeMap;
use std::str::FromStr;
use fabro_store::platform_records::{CheckpointRecord, DecisionRef};
use fabro_store::{PlatformRecord, StoredPlatformRecord};
use fabro_types::DiffSummary;
use crate::error::Error;
/// How a caller names a checkpoint: by ordinal (`@2`), by the latest visit
/// of a node (`build`), or by one visit of it (`build@1`).
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ForkTarget {
Ordinal(usize),
LatestVisit(String),
SpecificVisit(String, usize),
}
impl FromStr for ForkTarget {
type Err = Error;
fn from_str(s: &str) -> Result<Self, Error> {
if let Some(rest) = s.strip_prefix('@') {
let n: usize = rest
.parse()
.map_err(|_| Error::Validation(format!("invalid ordinal: @{rest}")))?;
if n == 0 {
return Err(Error::Validation("ordinal must be >= 1".to_string()));
}
return Ok(Self::Ordinal(n));
}
if let Some((name, visit)) = s.rsplit_once('@') {
if !name.is_empty() && !visit.is_empty() {
if let Ok(visit) = visit.parse::<usize>() {
if visit == 0 {
return Err(Error::Validation("visit number must be >= 1".to_string()));
}
return Ok(Self::SpecificVisit(name.to_string(), visit));
}
}
}
if s.trim().is_empty() {
return Err(Error::Validation("a target names a checkpoint".to_string()));
}
Ok(Self::LatestVisit(s.to_string()))
}
}
/// A stage as the run's projection labels it, by its Petri position: what
/// the timeline shows beside a checkpoint and what a target such as
/// `build@2` resolves through.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct StageLabel {
/// The stage id (`node@visit`) when the projection shows the stage.
pub stage_id: Option<String>,
pub node_name: String,
pub visit: u32,
}
/// The stages of a run by `(execution, firing)`.
pub type StageLabels = BTreeMap<(u64, u64), StageLabel>;
/// The Petri position of a checkpoint: the attempt whose files it holds.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct TimelinePosition {
pub execution: u64,
pub firing: u64,
pub attempt: u32,
}
/// One checkpoint of the run.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TimelineEntry {
/// 1-based, in the order the checkpoints were recorded.
pub ordinal: usize,
/// The checkpoint record's seq among the run's platform records.
pub checkpoint_seq: u64,
pub position: TimelinePosition,
/// The stage id (`node@visit`) when the projection shows the stage.
pub stage_id: Option<String>,
pub node_name: String,
pub visit: u32,
/// The Petri workspace id the commit was made in.
pub workspace: Option<String>,
pub run_commit_sha: Option<String>,
pub diff_summary: Option<DiffSummary>,
}
/// The run's checkpoints in order.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct RunTimeline {
pub entries: Vec<TimelineEntry>,
}
impl RunTimeline {
/// The timeline of `checkpoints` (the run's `checkpoint` platform
/// records, in seq order), labelled through `labels`.
#[must_use]
pub fn build(checkpoints: &[StoredPlatformRecord], labels: &StageLabels) -> Self {
let mut entries = Vec::new();
for stored in checkpoints {
let PlatformRecord::Checkpoint(record) = &stored.record else {
continue;
};
let position = position_of(record);
let label = labels.get(&(position.execution, position.firing));
entries.push(TimelineEntry {
ordinal: entries.len() + 1,
checkpoint_seq: stored.seq,
position,
stage_id: label.and_then(|label| label.stage_id.clone()),
node_name: label
.map(|label| label.node_name.clone())
.unwrap_or_default(),
visit: label.map_or(1, |label| label.visit),
workspace: record.workspace.clone(),
run_commit_sha: record.git_commit_sha.clone(),
diff_summary: record.diff_summary,
});
}
Self { entries }
}
/// The latest checkpoint, the default target.
pub fn latest(&self) -> Result<&TimelineEntry, Error> {
self.entries
.last()
.ok_or_else(|| Error::Validation("the run has no checkpoint to fork at".to_string()))
}
/// The checkpoint `target` names.
pub fn resolve(&self, target: &ForkTarget) -> Result<&TimelineEntry, Error> {
match target {
ForkTarget::Ordinal(n) => self
.entries
.iter()
.find(|entry| entry.ordinal == *n)
.ok_or_else(|| {
Error::Validation(format!(
"ordinal @{n} out of range (max @{})",
self.entries.len()
))
}),
ForkTarget::LatestVisit(name) => self
.entries
.iter()
.rev()
.find(|entry| entry.node_name == *name)
.ok_or_else(|| Error::Validation(format!("no checkpoint found for node '{name}'"))),
ForkTarget::SpecificVisit(name, visit) => self
.entries
.iter()
.find(|entry| {
entry.node_name == *name && usize::try_from(entry.visit) == Ok(*visit)
})
.ok_or_else(|| {
Error::Validation(format!("no visit {visit} found for node '{name}'"))
}),
}
}
/// The checkpoint `target` names, or the latest one.
pub fn resolve_or_latest(&self, target: Option<&ForkTarget>) -> Result<&TimelineEntry, Error> {
match target {
Some(target) => self.resolve(target),
None => self.latest(),
}
}
}
/// The position a checkpoint record names: its operation identity's
/// attempt, else the attempt it recorded, else the first.
fn position_of(record: &CheckpointRecord) -> TimelinePosition {
let attempt = match record
.operation
.as_ref()
.map(|operation| &operation.decision)
{
Some(DecisionRef::AttemptStart { attempt, .. } | DecisionRef::Route { attempt, .. }) => {
*attempt
}
Some(DecisionRef::ExecutionStart) | None => record.attempt.unwrap_or(1),
};
TimelinePosition {
execution: record.execution,
firing: record.firing,
attempt,
}
}
#[cfg(test)]
mod tests {
use fabro_store::platform_records::OperationKey;
use super::*;
fn checkpoint(
seq: u64,
execution: u64,
firing: u64,
sha: Option<&str>,
) -> StoredPlatformRecord {
StoredPlatformRecord {
seq,
recorded_at: seq * 1_000,
record: PlatformRecord::Checkpoint(CheckpointRecord {
execution,
firing,
attempt: Some(1),
workspace: Some("invocation-0-scope-0".to_string()),
git_commit_sha: sha.map(ToOwned::to_owned),
diff_summary: None,
patch_blob: None,
operation: Some(OperationKey {
execution,
decision: DecisionRef::AttemptStart { firing, attempt: 1 },
effect: "checkpoint".to_string(),
}),
}),
position: None,
}
}
fn label(node: &str, visit: u32) -> StageLabel {
StageLabel {
stage_id: Some(format!("{node}@{visit}")),
node_name: node.to_string(),
visit,
}
}
fn timeline() -> RunTimeline {
let labels: StageLabels = [
((0, 1), label("start", 1)),
((0, 2), label("build", 1)),
((0, 3), label("build", 2)),
]
.into_iter()
.collect();
RunTimeline::build(
&[
checkpoint(7, 0, 1, Some("aaa")),
checkpoint(9, 0, 2, Some("bbb")),
checkpoint(11, 0, 3, Some("ccc")),
],
&labels,
)
}
#[test]
fn a_target_parses_as_an_ordinal_a_node_or_a_visit() {
assert_eq!("@4".parse::<ForkTarget>().unwrap(), ForkTarget::Ordinal(4));
assert_eq!(
"step2".parse::<ForkTarget>().unwrap(),
ForkTarget::LatestVisit("step2".to_string())
);
assert_eq!(
"build@2".parse::<ForkTarget>().unwrap(),
ForkTarget::SpecificVisit("build".to_string(), 2)
);
assert!("@0".parse::<ForkTarget>().is_err());
assert!("@x".parse::<ForkTarget>().is_err());
}
#[test]
fn the_timeline_orders_checkpoints_and_labels_them() {
let timeline = timeline();
let ordinals: Vec<_> = timeline
.entries
.iter()
.map(|entry| (entry.ordinal, entry.node_name.as_str(), entry.visit))
.collect();
assert_eq!(ordinals, [
(1, "start", 1),
(2, "build", 1),
(3, "build", 2)
]);
assert_eq!(timeline.entries[1].checkpoint_seq, 9);
assert_eq!(timeline.entries[1].position, TimelinePosition {
execution: 0,
firing: 2,
attempt: 1,
});
assert_eq!(timeline.entries[2].stage_id.as_deref(), Some("build@2"));
}
#[test]
fn a_target_resolves_to_its_entry() {
let timeline = timeline();
assert_eq!(
timeline.resolve(&ForkTarget::Ordinal(2)).unwrap().ordinal,
2
);
assert_eq!(
timeline
.resolve(&ForkTarget::LatestVisit("build".to_string()))
.unwrap()
.ordinal,
3
);
assert_eq!(
timeline
.resolve(&ForkTarget::SpecificVisit("build".to_string(), 1))
.unwrap()
.ordinal,
2
);
assert_eq!(timeline.resolve_or_latest(None).unwrap().ordinal, 3);
assert!(matches!(
timeline.resolve(&ForkTarget::Ordinal(4)),
Err(Error::Validation(message)) if message.contains("out of range")
));
assert!(matches!(
timeline.resolve(&ForkTarget::LatestVisit("test".to_string())),
Err(Error::Validation(message)) if message.contains("no checkpoint found")
));
}
#[test]
fn an_empty_timeline_has_no_latest() {
assert!(RunTimeline::default().latest().is_err());
}
}

75
lib/foundation/fabro-client/src/client.rs
View file

@ -78,6 +78,13 @@ pub struct SessionEventStream {
buffered_events: VecDeque<SessionEvent>,
}
/// What a rewind returned: the response, and the status that says whether
/// the source was archived.
pub struct RewindRunResult {
pub status: u16,
pub response: types::RewindResponse,
}
#[derive(Default)]
struct ListStoreRunsOptions {
parent_id: Option<RunId>,
@ -1424,6 +1431,74 @@ impl Client {
convert_type(response.into_inner())
}
/// The run's checkpoint timeline, and where it was forked from.
pub async fn run_timeline(&self, run_id: &RunId) -> Result<types::RunTimelineResponse> {
let response = self
.send_api(|client| async move {
client
.get_run_timeline()
.id(run_id.to_string())
.send()
.await
})
.await?;
Ok(response.into_inner())
}
/// Fork the run at a checkpoint into a new run, started in resume mode.
pub async fn fork_run(
&self,
run_id: &RunId,
request: types::ForkRequest,
) -> Result<types::ForkResponse> {
let response = self
.send_api(|client| async move {
client
.fork_run()
.id(run_id.to_string())
.body(request)
.send()
.await
})
.await?;
Ok(response.into_inner())
}
/// Rewind the run to a checkpoint: a fork that archives and supersedes
/// the source. The status says whether the archive succeeded (200) or
/// the new run was made without it (207).
pub async fn rewind_run(
&self,
run_id: &RunId,
request: types::RewindRequest,
) -> Result<RewindRunResult> {
let response = self
.send_api(|client| async move {
client
.rewind_run()
.id(run_id.to_string())
.body(request)
.send()
.await
})
.await?;
let status = response.status().as_u16();
Ok(RewindRunResult {
status,
response: response.into_inner(),
})
}
/// Retry a terminal run from its last checkpoint: the new run.
pub async fn retry_run(&self, run_id: &RunId) -> Result<Run> {
let response = self
.send_api(
|client| async move { client.retry_run().id(run_id.to_string()).send().await },
)
.await?;
convert_type(response.into_inner())
}
pub async fn resolve_run(&self, selector: &str) -> Result<Run> {
let response = self
.send_api(|client| async move {

4
lib/foundation/fabro-client/src/lib.rs
View file

@ -12,8 +12,8 @@ pub use auth_store::{
AuthEntry, AuthStore, AuthStoreError, DevTokenEntry, LockError, OAuthEntry, StoredSubject,
};
pub use client::{
Client, RunStreamItemStream, RunStreamPage, SessionEventStream, TransportConnector,
apply_bearer_token_auth,
Client, RewindRunResult, RunStreamItemStream, RunStreamPage, SessionEventStream,
TransportConnector, apply_bearer_token_auth,
};
pub use credential::{Credential, CredentialFallback};
pub use error::{

6
lib/foundation/fabro-types/src/lib.rs
View file

@ -148,9 +148,9 @@ pub use run_intent::{
TargetValidationError, ValidatedGitRunTarget, ValidatedRunTarget,
};
pub use run_projection::{
CheckpointRecord, PendingInterviewRecord, RunArtifact, RunProjection, StageContextWindow,
StageContextWindowUnavailableReason, StageInferenceProjection, StageModelUsage,
StageProjection, StageToolBatchProjection, first_event_seq,
CheckpointRecord, ForkOrigin, PendingInterviewRecord, RunArtifact, RunProjection,
StageContextWindow, StageContextWindowUnavailableReason, StageInferenceProjection,
StageModelUsage, StageProjection, StageToolBatchProjection, first_event_seq,
};
pub use run_sandbox::{
RunSandbox, RunSandboxFailure, RunSandboxInstance, RunSandboxKind, RunSandboxPlan,

17
lib/foundation/fabro-types/src/run_projection.rs
View file

@ -47,6 +47,11 @@ pub struct RunProjection {
pub superseded_by: Option<RunId>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub retried_from: Option<RunId>,
/// Where the run's records came from when it is a fork: the source run
/// and the position its records were kept up to, as Petri's own run
/// declaration names them.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub forked_from: Option<ForkOrigin>,
/// The Git author/committer identity the run resolved for its commits.
/// Absent until the run's first initialization resolves it.
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -80,6 +85,17 @@ pub struct PendingInterviewRecord {
pub started_at: DateTime<Utc>,
}
/// The source of a forked run: the run whose records were copied, the
/// position (a firing of one of its root executions) they were kept up to,
/// and whether that firing runs again in the fork.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct ForkOrigin {
pub source_run_id: RunId,
pub execution: u64,
pub firing: u64,
pub rerun_last: bool,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct CheckpointRecord {
pub seq: u32,
@ -712,6 +728,7 @@ impl RunProjection {
pull_request_creation: None,
superseded_by: None,
retried_from: None,
forked_from: None,
git_identity: None,
pending_interviews: BTreeMap::new(),
artifacts: Vec::new(),

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

@ -164,6 +164,9 @@ models/failure-reason.ts
models/file-checkpoint.ts
models/file-diff.ts
models/folder-run-target.ts
models/fork-origin.ts
models/fork-request.ts
models/fork-response.ts
models/fork-source-ref.ts
models/git-author-settings.ts
models/git-context.ts
@ -365,6 +368,8 @@ models/repository-ref.ts
models/resolved-automation-git-workflow-source.ts
models/review-target-kind.ts
models/review-target.ts
models/rewind-request.ts
models/rewind-response.ts
models/root-response-urls.ts
models/root-response.ts
models/run-agent-settings.ts
@ -442,6 +447,7 @@ models/run-stream-item-kind.ts
models/run-stream-item.ts
models/run-superseded-by-props.ts
models/run-target.ts
models/run-timeline-response.ts
models/run-timestamps.ts
models/run-timing.ts
models/run-usage-stage.ts
@ -539,6 +545,7 @@ models/system-repair-run-issue.ts
models/system-repair-runs-response.ts
models/system-resources-response.ts
models/system-run-counts.ts
models/timeline-entry-response.ts
models/tls-mode.ts
models/todo-list-kind.ts
models/todo-list-projection.ts

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

@ -42,6 +42,10 @@ import type { DenyRunRequest } from '../models';
// @ts-ignore
import type { ErrorResponse } from '../models';
// @ts-ignore
import type { ForkRequest } from '../models';
// @ts-ignore
import type { ForkResponse } from '../models';
// @ts-ignore
import type { LinkRunPullRequestRequest } from '../models';
// @ts-ignore
import type { MergeRunPullRequestRequest } from '../models';
@ -60,12 +64,18 @@ import type { PullRequestResponse } from '../models';
// @ts-ignore
import type { RenderWorkflowGraphRequest } from '../models';
// @ts-ignore
import type { RewindRequest } from '../models';
// @ts-ignore
import type { RewindResponse } from '../models';
// @ts-ignore
import type { Run } from '../models';
// @ts-ignore
import type { RunIntent } from '../models';
// @ts-ignore
import type { RunManifest } from '../models';
// @ts-ignore
import type { RunTimelineResponse } from '../models';
// @ts-ignore
import type { StartRunRequest } from '../models';
// @ts-ignore
import type { UpdateRunParentRequest } from '../models';
@ -535,6 +545,49 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
options: localVarRequestOptions,
};
},
/**
* Creates a new run from a checkpoint of the source run and starts it. The new run holds the source\'s records up to the checkpoint\'s position and continues from there in a fresh workspace restored to the checkpoint\'s commit. The source run is left untouched. A checkpoint inside a parallel branch cannot be forked at; fork at the parallel stage instead.
* @summary Fork Run
* @param {string} id Unique run identifier (ULID).
* @param {ForkRequest} [forkRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
forkRun: async (id: string, forkRequest?: ForkRequest, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('forkRun', 'id', id)
const localVarPath = `/api/v1/runs/{id}/fork`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// 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 SessionCookie required
// 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(forkRequest, localVarRequestOptions, configuration)
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Returns the stored pull request record for a run plus live GitHub details when available.
* @summary Get Run Pull Request
@ -615,6 +668,46 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
options: localVarRequestOptions,
};
},
/**
* Returns the run\'s checkpoints in the order they were recorded, each at its Petri position with the commit it made and the stage it belongs to, and where the run was forked from when it is a fork.
* @summary Get Run Timeline
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getRunTimeline: async (id: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('getRunTimeline', 'id', id)
const localVarPath = `/api/v1/runs/{id}/timeline`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// 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)
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,
};
},
/**
* Links a run under an orchestration parent. Parent links are mutable for all run states, including archived and terminal runs.
* @summary Link Run Parent
@ -1070,6 +1163,89 @@ export const RunsApiAxiosParamCreator = function (configuration?: Configuration)
options: localVarRequestOptions,
};
},
/**
* Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable.
* @summary Retry Run
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
retryRun: async (id: string, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('retryRun', 'id', id)
const localVarPath = `/api/v1/runs/{id}/retry`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// 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 SessionCookie required
// 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,
};
},
/**
* Creates a new run from a checkpoint of a terminal source run and starts it, then archives the source run and records `run.superseded_by` on it. Returns 207 when the new run was created but the source archive step failed.
* @summary Rewind Run
* @param {string} id Unique run identifier (ULID).
* @param {RewindRequest} [rewindRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
rewindRun: async (id: string, rewindRequest?: RewindRequest, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('rewindRun', 'id', id)
const localVarPath = `/api/v1/runs/{id}/rewind`
.replace(`{${"id"}}`, encodeURIComponent(String(id)));
// 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 SessionCookie required
// 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(rewindRequest, localVarRequestOptions, configuration)
return {
url: toPathString(localVarUrlObj),
options: localVarRequestOptions,
};
},
/**
* Validates runtime readiness for a workflow manifest without creating a run.
* @summary Validate Workflow Manifest
@ -1555,6 +1731,20 @@ export const RunsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunsApi.denyRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Creates a new run from a checkpoint of the source run and starts it. The new run holds the source\'s records up to the checkpoint\'s position and continues from there in a fresh workspace restored to the checkpoint\'s commit. The source run is left untouched. A checkpoint inside a parallel branch cannot be forked at; fork at the parallel stage instead.
* @summary Fork Run
* @param {string} id Unique run identifier (ULID).
* @param {ForkRequest} [forkRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async forkRun(id: string, forkRequest?: ForkRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<ForkResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.forkRun(id, forkRequest, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.forkRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns the stored pull request record for a run plus live GitHub details when available.
* @summary Get Run Pull Request
@ -1581,6 +1771,19 @@ export const RunsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunsApi.getRunPullRequestCreation']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Returns the run\'s checkpoints in the order they were recorded, each at its Petri position with the commit it made and the stage it belongs to, and where the run was forked from when it is a fork.
* @summary Get Run Timeline
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async getRunTimeline(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<RunTimelineResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.getRunTimeline(id, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.getRunTimeline']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Links a run under an orchestration parent. Parent links are mutable for all run states, including archived and terminal runs.
* @summary Link Run Parent
@ -1721,6 +1924,33 @@ export const RunsApiFp = function(configuration?: Configuration) {
const localVarOperationServerBasePath = operationServerMap['RunsApi.retrieveRunGraphSource']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable.
* @summary Retry Run
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async retryRun(id: string, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<Run>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.retryRun(id, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.retryRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Creates a new run from a checkpoint of a terminal source run and starts it, then archives the source run and records `run.superseded_by` on it. Returns 207 when the new run was created but the source archive step failed.
* @summary Rewind Run
* @param {string} id Unique run identifier (ULID).
* @param {RewindRequest} [rewindRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async rewindRun(id: string, rewindRequest?: RewindRequest, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<RewindResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.rewindRun(id, rewindRequest, options);
const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
const localVarOperationServerBasePath = operationServerMap['RunsApi.rewindRun']?.[localVarOperationServerIndex]?.url;
return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
},
/**
* Validates runtime readiness for a workflow manifest without creating a run.
* @summary Validate Workflow Manifest
@ -1949,6 +2179,17 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
denyRun(id: string, denyRunRequest?: DenyRunRequest, options?: RawAxiosRequestConfig): AxiosPromise<Run> {
return localVarFp.denyRun(id, denyRunRequest, options).then((request) => request(axios, basePath));
},
/**
* Creates a new run from a checkpoint of the source run and starts it. The new run holds the source\'s records up to the checkpoint\'s position and continues from there in a fresh workspace restored to the checkpoint\'s commit. The source run is left untouched. A checkpoint inside a parallel branch cannot be forked at; fork at the parallel stage instead.
* @summary Fork Run
* @param {string} id Unique run identifier (ULID).
* @param {ForkRequest} [forkRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
forkRun(id: string, forkRequest?: ForkRequest, options?: RawAxiosRequestConfig): AxiosPromise<ForkResponse> {
return localVarFp.forkRun(id, forkRequest, options).then((request) => request(axios, basePath));
},
/**
* Returns the stored pull request record for a run plus live GitHub details when available.
* @summary Get Run Pull Request
@ -1969,6 +2210,16 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
getRunPullRequestCreation(id: string, options?: RawAxiosRequestConfig): AxiosPromise<PullRequestCreation> {
return localVarFp.getRunPullRequestCreation(id, options).then((request) => request(axios, basePath));
},
/**
* Returns the run\'s checkpoints in the order they were recorded, each at its Petri position with the commit it made and the stage it belongs to, and where the run was forked from when it is a fork.
* @summary Get Run Timeline
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getRunTimeline(id: string, options?: RawAxiosRequestConfig): AxiosPromise<RunTimelineResponse> {
return localVarFp.getRunTimeline(id, options).then((request) => request(axios, basePath));
},
/**
* Links a run under an orchestration parent. Parent links are mutable for all run states, including archived and terminal runs.
* @summary Link Run Parent
@ -2079,6 +2330,27 @@ export const RunsApiFactory = function (configuration?: Configuration, basePath?
retrieveRunGraphSource(id: string, options?: RawAxiosRequestConfig): AxiosPromise<string> {
return localVarFp.retrieveRunGraphSource(id, options).then((request) => request(axios, basePath));
},
/**
* Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable.
* @summary Retry Run
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
retryRun(id: string, options?: RawAxiosRequestConfig): AxiosPromise<Run> {
return localVarFp.retryRun(id, options).then((request) => request(axios, basePath));
},
/**
* Creates a new run from a checkpoint of a terminal source run and starts it, then archives the source run and records `run.superseded_by` on it. Returns 207 when the new run was created but the source archive step failed.
* @summary Rewind Run
* @param {string} id Unique run identifier (ULID).
* @param {RewindRequest} [rewindRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
rewindRun(id: string, rewindRequest?: RewindRequest, options?: RawAxiosRequestConfig): AxiosPromise<RewindResponse> {
return localVarFp.rewindRun(id, rewindRequest, options).then((request) => request(axios, basePath));
},
/**
* Validates runtime readiness for a workflow manifest without creating a run.
* @summary Validate Workflow Manifest
@ -2292,6 +2564,18 @@ export class RunsApi extends BaseAPI {
return RunsApiFp(this.configuration).denyRun(id, denyRunRequest, options).then((request) => request(this.axios, this.basePath));
}
/**
* Creates a new run from a checkpoint of the source run and starts it. The new run holds the source\'s records up to the checkpoint\'s position and continues from there in a fresh workspace restored to the checkpoint\'s commit. The source run is left untouched. A checkpoint inside a parallel branch cannot be forked at; fork at the parallel stage instead.
* @summary Fork Run
* @param {string} id Unique run identifier (ULID).
* @param {ForkRequest} [forkRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public forkRun(id: string, forkRequest?: ForkRequest, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).forkRun(id, forkRequest, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns the stored pull request record for a run plus live GitHub details when available.
* @summary Get Run Pull Request
@ -2314,6 +2598,17 @@ export class RunsApi extends BaseAPI {
return RunsApiFp(this.configuration).getRunPullRequestCreation(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Returns the run\'s checkpoints in the order they were recorded, each at its Petri position with the commit it made and the stage it belongs to, and where the run was forked from when it is a fork.
* @summary Get Run Timeline
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public getRunTimeline(id: string, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).getRunTimeline(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Links a run under an orchestration parent. Parent links are mutable for all run states, including archived and terminal runs.
* @summary Link Run Parent
@ -2434,6 +2729,29 @@ export class RunsApi extends BaseAPI {
return RunsApiFp(this.configuration).retrieveRunGraphSource(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Creates a new run from the terminal source run\'s last checkpoint and starts it. When the source failed on a stage, that stage runs again on the files of the stage before it; otherwise the new run continues from the last checkpoint as it stands. The new run records `retried_from` and `fork_source_ref`; the source run is left unchanged. Active and archived runs are not retryable.
* @summary Retry Run
* @param {string} id Unique run identifier (ULID).
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public retryRun(id: string, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).retryRun(id, options).then((request) => request(this.axios, this.basePath));
}
/**
* Creates a new run from a checkpoint of a terminal source run and starts it, then archives the source run and records `run.superseded_by` on it. Returns 207 when the new run was created but the source archive step failed.
* @summary Rewind Run
* @param {string} id Unique run identifier (ULID).
* @param {RewindRequest} [rewindRequest]
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public rewindRun(id: string, rewindRequest?: RewindRequest, options?: RawAxiosRequestConfig) {
return RunsApiFp(this.configuration).rewindRun(id, rewindRequest, options).then((request) => request(this.axios, this.basePath));
}
/**
* Validates runtime readiness for a workflow manifest without creating a run.
* @summary Validate Workflow Manifest

25
lib/packages/fabro-api-client/src/models/fork-origin.ts generated Normal file
View file

@ -0,0 +1,25 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Where a forked run came from: the source run, the position (an execution and a firing of its root invocation) the source\'s records were kept up to, and whether that firing runs again in the fork.
*/
export interface ForkOrigin {
'source_run_id': string;
'execution': number;
'firing': number;
'rerun_last': boolean;
}

25
lib/packages/fabro-api-client/src/models/fork-request.ts generated Normal file
View file

@ -0,0 +1,25 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Request body for creating a new run from a source run checkpoint.
*/
export interface ForkRequest {
/**
* Optional checkpoint target such as `@2`, `build`, or `build@1`. Defaults to the latest checkpoint.
*/
'target'?: string | null;
}

43
lib/packages/fabro-api-client/src/models/fork-response.ts generated Normal file
View file

@ -0,0 +1,43 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Response returned after creating and starting a forked run.
*/
export interface ForkResponse {
'source_run_id': string;
'new_run_id': string;
/**
* The checkpoint the fork was resolved to, as `@ordinal`.
*/
'target': string;
/**
* The commit the new run\'s workspace starts on.
*/
'checkpoint_sha': string;
/**
* The Petri execution of the fork position.
*/
'execution': number;
/**
* The Petri firing of the fork position.
*/
'firing': number;
/**
* Whether the position\'s stage runs again in the new run.
*/
'rerun_last': boolean;
}

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

@ -135,6 +135,9 @@ export * from './failure-reason';
export * from './file-checkpoint';
export * from './file-diff';
export * from './folder-run-target';
export * from './fork-origin';
export * from './fork-request';
export * from './fork-response';
export * from './fork-source-ref';
export * from './git-author-settings';
export * from './git-context';
@ -335,6 +338,8 @@ export * from './repository-ref';
export * from './resolved-automation-git-workflow-source';
export * from './review-target';
export * from './review-target-kind';
export * from './rewind-request';
export * from './rewind-response';
export * from './root-response';
export * from './root-response-urls';
export * from './run';
@ -413,6 +418,7 @@ export * from './run-stream-item';
export * from './run-stream-item-kind';
export * from './run-superseded-by-props';
export * from './run-target';
export * from './run-timeline-response';
export * from './run-timestamps';
export * from './run-timing';
export * from './run-usage';
@ -509,6 +515,7 @@ export * from './system-repair-run-issue';
export * from './system-repair-runs-response';
export * from './system-resources-response';
export * from './system-run-counts';
export * from './timeline-entry-response';
export * from './tls-mode';
export * from './todo-list-kind';
export * from './todo-list-projection';

25
lib/packages/fabro-api-client/src/models/rewind-request.ts generated Normal file
View file

@ -0,0 +1,25 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Request body for creating a replacement run from a source run checkpoint.
*/
export interface RewindRequest {
/**
* Optional checkpoint target such as `@2`, `build`, or `build@1`. Defaults to the latest checkpoint.
*/
'target'?: string | null;
}

32
lib/packages/fabro-api-client/src/models/rewind-response.ts generated Normal file
View file

@ -0,0 +1,32 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.0
*
*
* NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
* https://openapi-generator.tech
* Do not edit the class manually.
*/
/**
* Response returned after rewind creates and starts a new run.
*/
export interface RewindResponse {
'source_run_id': string;
'new_run_id': string;
/**
* The checkpoint the rewind was resolved to, as `@ordinal`.
*/
'target': string;
'checkpoint_sha': string;
'execution': number;
'firing': number;
'archived': boolean;
'archive_error'?: string | null;
}

4
lib/packages/fabro-api-client/src/models/run-projection.ts generated
View file

@ -21,6 +21,9 @@ import type { CheckpointRecord } from './checkpoint-record';
import type { Conclusion } from './conclusion';
// May contain unused imports in some cases
// @ts-ignore
import type { ForkOrigin } from './fork-origin';
// May contain unused imports in some cases
// @ts-ignore
import type { GitIdentity } from './git-identity';
// May contain unused imports in some cases
// @ts-ignore
@ -86,6 +89,7 @@ export interface RunProjection {
* Source run ID when this run was created by manual retry.
*/
'retried_from'?: string | null;
'forked_from'?: ForkOrigin | null;
'git_identity'?: GitIdentity | null;
'pending_interviews': { [key: string]: PendingInterviewRecord; };
/**

29
lib/packages/fabro-api-client/src/models/run-timeline-response.ts generated Normal file
View file

@ -0,0 +1,29 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.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 { ForkOrigin } from './fork-origin';
// May contain unused imports in some cases
// @ts-ignore
import type { TimelineEntryResponse } from './timeline-entry-response';
/**
* The run\'s checkpoints in order, and its fork origin when it is a fork.
*/
export interface RunTimelineResponse {
'entries': Array<TimelineEntryResponse>;
'forked_from'?: ForkOrigin | null;
}

56
lib/packages/fabro-api-client/src/models/timeline-entry-response.ts generated Normal file
View file

@ -0,0 +1,56 @@
/* tslint:disable */
/* eslint-disable */
/**
* Fabro Run API
* HTTP API for managing Fabro workflow run executions.
*
* The version of the OpenAPI document: 0.2.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 { DiffSummary } from './diff-summary';
/**
* One checkpoint of a run.
*/
export interface TimelineEntryResponse {
/**
* 1-based, in the order the checkpoints were recorded; the `@ordinal` a fork target names.
*/
'ordinal': number;
/**
* The checkpoint record\'s position among the run\'s platform records.
*/
'checkpoint_seq': number;
/**
* The Petri execution the checkpoint belongs to.
*/
'execution': number;
/**
* The Petri firing the checkpoint belongs to.
*/
'firing': number;
/**
* The attempt of the firing whose files the commit holds, from 1.
*/
'attempt': number;
/**
* The stage id (`node@visit`) when the projection shows the stage.
*/
'stage'?: string | null;
'node_name': string;
'visit': number;
/**
* The Petri workspace id the commit was made in.
*/
'workspace'?: string | null;
'run_commit_sha'?: string | null;
'diff_summary'?: DiffSummary | null;
}