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

feat: treat resumed in-flight nodes as new stage executions

Browse source 
A node cancelled (or lost to a crash) mid-flight and then resumed now
starts a new stage execution with the next StageId ordinal (work@2)
instead of reusing and clearing the cancelled execution's projection.
The old execution stays immutable with its own events, session, output,
timing, billing, and termination state.

Engine:
- Add a run-scoped StageExecutionTracker on RunServices with per-node
  high-water marks. Ordinals are reserved after the StageStart hook
  passes on the first attempt (retries reuse the reservation), ensured
  at the composite checkpoint pre-step for hook-skips, and reserved in
  on_terminal_reached for terminal nodes' synthetic events.
- Keep three concepts distinct: graph visit (max_visits/checkpoints,
  unchanged), stage execution ordinal (the @N in StageId), and handler
  attempt. The tracker is not checkpointed; the append-only stage event
  history is its durable source of truth.
- resume() seeds the allocator from the run projection and computes a
  node -> StageId provenance map of executions observed after the
  selected checkpoint, threaded through execute_persisted_run,
  RunSession, and InitOptions.

Events and projections:
- stage.started, parallel.branch.started, and checkpoint.completed
  carry optional graph_visit and resumed_from_stage_id; StageProjection
  stores both. Old events deserialize with None and legacy duplicate
  stage.started replays keep last-attempt behavior.
- The CheckpointCompleted reducer is envelope-first: diffs and
  skipped-stage synthesis attach to the exact execution StageId, an
  existing Retrying projection finalizes as Skipped without losing
  identity, and historical node_outcomes no longer create or collide
  with newer ordinals (node_visits remains a legacy fallback).

Handlers:
- Parallel fan-out reserves child ordinals through the shared tracker,
  derives worktree pass{N} from the parent's execution ordinal, and
  seeds branch contexts with explicit child stage scopes so branch
  lifecycle and nested handler events agree.
- Artifact capture and manager-loop child logs follow the ordinal.

API and UI:
- RunStage documents visit as the execution ordinal and adds optional
  graph_visit and resumed_from_stage_id; Rust and TypeScript clients
  regenerated.
- The web sidebar lists both executions chronologically; resumed stages
  show a "Resumed from" link in the stage detail header and hover
  popover, with the graph visit surfaced when it diverges.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-07-24 09:00:37 -04:00
parent 9afd456c06
commit cd706646c6
No known key found for this signature in database
43 changed files with 1906 additions and 306 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
apps/fabro-web/app/components/stage-popover.test.tsx
View file

@ -1,6 +1,7 @@
import { describe, expect, test } from "bun:test";
import type { ReactNode } from "react";
import TestRenderer, { act } from "react-test-renderer";
import { MemoryRouter } from "react-router";
import { SWRConfig } from "swr";
import type { EventEnvelope } from "@qltysh/fabro-api-client";
@ -22,15 +23,17 @@ function makeEvent(overrides: Partial<EventEnvelope>): EventEnvelope {
function makeStage(overrides: Partial<Stage> = {}): Stage {
return {
id: "implement@1",
name: "implement",
handler: "agent",
nodeId: "implement",
visit: 1,
status: "succeeded",
duration: "1m 30s",
startedAt: "2026-05-24T11:58:30Z",
providerUsed: { mode: "policy", model: "claude-opus-4-7", reasoning_effort: "high" },
id: "implement@1",
name: "implement",
handler: "agent",
nodeId: "implement",
visit: 1,
graphVisit: null,
resumedFromStageId: null,
status: "succeeded",
duration: "1m 30s",
startedAt: "2026-05-24T11:58:30Z",
providerUsed: { mode: "policy", model: "claude-opus-4-7", reasoning_effort: "high" },
...overrides,
};
}
@ -275,6 +278,36 @@ describe("StagePopover rendering", () => {
expect(text).not.toContain("Reason");
});
test("resumed stage links to the prior execution and shows a divergent graph visit", () => {
const stage = makeStage({
id: "implement@2",
visit: 2,
graphVisit: 1,
resumedFromStageId: "implement@1",
status: "running",
duration: "--",
});
const tree = render(
<MemoryRouter initialEntries={["/runs/run-1"]}>
<StagePopover runId="run-1" stage={stage} duration="--" />
</MemoryRouter>,
);
const text = textOf(tree);
expect(text).toContain("Resumed from");
expect(text).toContain("implement@1");
expect(text).toContain("Graph visit");
const json = JSON.stringify(tree.toJSON());
expect(json).toContain("/runs/run-1/stages/implement@1");
});
test("stage without ordinal divergence hides the graph visit row", () => {
const stage = makeStage({ graphVisit: 1, status: "pending", duration: "--" });
const tree = render(<StagePopover runId="run-1" stage={stage} duration="--" />);
const text = textOf(tree);
expect(text).not.toContain("Resumed from");
expect(text).not.toContain("Graph visit");
});
test("failed command stage shows exit code instead of model", async () => {
await withMockedStageEvents(
[

16
apps/fabro-web/app/components/stage-popover.tsx
View file

@ -1,4 +1,5 @@
import { useMemo } from "react";
import { Link } from "react-router";
import type { StageState } from "@qltysh/fabro-api-client";
import { formatTokenCount } from "../lib/format";
@ -225,6 +226,21 @@ export function StagePopover({ runId, stage, duration }: StagePopoverProps) {
<span className="font-mono tabular-nums">{duration}</span>
</PopoverRow>
)}
{stage.resumedFromStageId && (
<PopoverRow label="Resumed from">
<Link
to={`/runs/${runId}/stages/${stage.resumedFromStageId}`}
className="font-mono text-teal-500 hover:underline"
>
{stage.resumedFromStageId}
</Link>
</PopoverRow>
)}
{stage.graphVisit != null && stage.graphVisit !== stage.visit && (
<PopoverRow label="Graph visit">
<span className="font-mono tabular-nums">{stage.graphVisit}</span>
</PopoverRow>
)}
<StatusTail stage={stage} summary={summary} loading={loading} />
</PopoverRows>
</div>

2
apps/fabro-web/app/components/stage-sidebar.test.tsx
View file

@ -11,6 +11,8 @@ function makeStage(overrides: Partial<Stage> = {}): Stage {
handler: "agent",
nodeId: "implement",
visit: 1,
graphVisit: null,
resumedFromStageId: null,
status: "running",
duration: "--",
startedAt: null,

58
apps/fabro-web/app/lib/stage-sidebar.test.ts
View file

@ -11,6 +11,8 @@ function makeStage(nodeId: string, visit: number, status: StageState): Stage {
handler: "agent",
nodeId,
visit,
graphVisit: null,
resumedFromStageId: null,
status,
duration: "--",
startedAt: null,
@ -124,6 +126,62 @@ describe("mapRunStagesToSidebarStages", () => {
expect(mapRunStagesToSidebarStages(stages)[0].duration).toBe("--");
});
test("maps a resumed execution's identity fields and keeps both entries in order", () => {
const stages: PaginatedRunStageList = {
data: [
{
id: "work@1",
name: "work",
handler: "agent",
status: "cancelled",
node_id: "work",
visit: 1,
graph_visit: 1,
},
{
id: "work@2",
name: "work",
handler: "agent",
status: "running",
node_id: "work",
visit: 2,
graph_visit: 1,
resumed_from_stage_id: "work@1",
},
],
meta: { has_more: false },
};
const result = mapRunStagesToSidebarStages(stages);
expect(result.map((s) => s.id)).toEqual(["work@1", "work@2"]);
expect(result[0].status).toBe("cancelled");
expect(result[0].resumedFromStageId).toBeNull();
expect(result[1].status).toBe("running");
expect(result[1].graphVisit).toBe(1);
expect(result[1].resumedFromStageId).toBe("work@1");
expect(formatStageLabel(result[1])).toBe("work@2");
});
test("omits identity fields for stages recorded before execution tracking", () => {
const stages: PaginatedRunStageList = {
data: [
{
id: "verify@1",
name: "verify",
handler: "agent",
status: "succeeded",
node_id: "verify",
visit: 1,
},
],
meta: { has_more: false },
};
const result = mapRunStagesToSidebarStages(stages);
expect(result[0].graphVisit).toBeNull();
expect(result[0].resumedFromStageId).toBeNull();
});
test("preserves the authoritative handler for renderer dispatch", () => {
const stages: PaginatedRunStageList = {
data: [

13
apps/fabro-web/app/lib/stage-sidebar.ts
View file

@ -14,8 +14,17 @@ export interface Stage {
handler: StageHandler;
status: StageState;
duration: string;
nodeId: string;
/** 1-based stage execution ordinal — the numeric component of `id`. */
visit: number;
nodeId: string;
/**
* How many times workflow control entered this node. Differs from `visit`
* when a cancelled or crashed execution was reexecuted after resume; null
* for stages recorded before execution identity was tracked.
*/
graphVisit: number | null;
/** StageId of the prior execution this stage resumes from, if any. */
resumedFromStageId: string | null;
startedAt: string | null;
providerUsed: StageModelUsage | null;
}
@ -85,6 +94,8 @@ export function mapRunStagesToSidebarStages(
handler: stage.handler,
nodeId: stage.node_id,
visit: stage.visit,
graphVisit: stage.graph_visit ?? null,
resumedFromStageId: stage.resumed_from_stage_id ?? null,
status: stage.status,
duration: stage.wall_time_ms != null
? formatDurationMs(stage.wall_time_ms)

13
apps/fabro-web/app/routes/run-stages.tsx
View file

@ -1,5 +1,5 @@
import { useMemo, useReducer, useState } from "react";
import { useParams } from "react-router";
import { Link, useParams } from "react-router";
import {
ArrowDownTrayIcon,
ChevronDownIcon,
@ -1745,6 +1745,17 @@ function RunStageActivityStage({
<div className="flex min-h-0 min-w-0 flex-1 flex-col pt-3">
<div className="shrink-0 border-b border-line">
<div className="pl-3 pr-3">
{selectedStage.resumedFromStageId && (
<p className="pb-2 text-xs text-fg-muted">
Resumed from{" "}
<Link
to={`/runs/${runId}/stages/${selectedStage.resumedFromStageId}`}
className="font-mono text-teal-500 hover:underline"
>
{selectedStage.resumedFromStageId}
</Link>
</p>
)}
<EventsToolbar
tab={effectiveTab}
renderer={renderer}

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

@ -12435,8 +12435,29 @@ components:
type: integer
format: uint32
minimum: 1
description: 1-based visit count; bumped each time the workflow re-enters this node.
description: >-
1-based stage execution ordinal, the numeric component of `id`. It
increments each time the node produces a new observable execution:
graph re-entry (loops) and reexecution after cancel or crash
recovery. Automatic in-place retries do not increment it.
example: 2
graph_visit:
type: ["integer", "null"]
format: uint32
minimum: 1
description: >-
1-based count of how many times workflow control entered this node
(drives `max_visits`). Differs from `visit` when a cancelled or
crashed execution was reexecuted after resume. Absent for stages
recorded before execution identity was tracked.
example: 1
resumed_from_stage_id:
type: ["string", "null"]
description: >-
StageId of the prior cancelled or interrupted execution this stage
resumes from, when the run was resumed after that execution became
observable.
example: verify@1
provider_used:
oneOf:
- $ref: "#/components/schemas/StageModelUsage"

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

@ -510,12 +510,14 @@ mod tests {
fn stage_started(node_id: &str, name: &str) -> Event {
Event::StageStarted {
node_id: node_id.into(),
name: name.into(),
index: 0,
handler_type: String::new(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: node_id.into(),
name: name.into(),
index: 0,
handler_type: String::new(),
attempt: 1,
max_attempts: 1,
}
}
@ -592,10 +594,12 @@ mod tests {
assert_eq!(ui.stage.parallel_parent.as_deref(), Some("fork1"));
emit(&mut ui, Event::ParallelBranchStarted {
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
});
let stage = &ui.stage.active_stages["fork1"];
assert_eq!(stage.tool_calls.len(), 1);
@ -633,10 +637,12 @@ mod tests {
join_policy: "wait_all".into(),
});
emit(&mut ui, Event::ParallelBranchStarted {
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
});
let stage = &ui.stage.active_stages["fork1"];
@ -1249,10 +1255,12 @@ mod tests {
join_policy: "wait_all".into(),
});
emit(&mut ui, Event::ParallelBranchStarted {
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
parallel_group_id: StageId::new("fork1", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0),
branch: "security".into(),
index: 0,
});
emit(&mut ui, Event::ParallelBranchCompleted {
parallel_group_id: StageId::new("fork1", 1),
@ -1282,12 +1290,14 @@ mod tests {
let stage_started = serde_json::to_string(&to_run_event_at(
&fixtures::RUN_1,
&Event::StageStarted {
node_id: "code".into(),
name: "Code".into(),
index: 0,
handler_type: "agent".into(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "code".into(),
name: "Code".into(),
index: 0,
handler_type: "agent".into(),
attempt: 1,
max_attempts: 1,
},
started_ts,
None,

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

@ -1169,6 +1169,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"node_label": "Start",
"properties": {
"attempt": 1,
"graph_visit": 1,
"handler_type": "start",
"index": 0,
"max_attempts": 1
@ -1195,6 +1196,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"internal.fidelity": "compact",
"internal.node_visit_count": 1,
"internal.run_id": "[ULID]",
"internal.stage_execution_ordinal": 1,
"internal.thread_id": null
},
"index": 0,
@ -1257,6 +1259,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"outcome": "succeeded"
},
"current_node": "start",
"graph_visit": 1,
"next_node_id": "approve",
"node_outcomes": {
"start": {
@ -1284,6 +1287,7 @@ fn attach_json_errors_without_prompting_for_human_input() {
"node_label": "Approve?",
"properties": {
"attempt": 1,
"graph_visit": 1,
"handler_type": "human",
"index": 1,
"max_attempts": 1

10
lib/apps/fabro-server/src/demo/mod.rs
View file

@ -1388,6 +1388,8 @@ mod runs {
None,
StageHandler::Command,
None,
None,
None,
),
run_stage_from_stage_id(
&StageId::new("propose-changes", 1),
@ -1397,6 +1399,8 @@ mod runs {
None,
StageHandler::Agent,
None,
None,
None,
),
run_stage_from_stage_id(
&StageId::new("review-changes", 1),
@ -1406,6 +1410,8 @@ mod runs {
None,
StageHandler::Agent,
None,
None,
None,
),
run_stage_from_stage_id(
&StageId::new("apply-changes", 1),
@ -1415,6 +1421,8 @@ mod runs {
None,
StageHandler::Command,
None,
None,
None,
),
run_stage_from_stage_id(
&StageId::new("apply-changes", 2),
@ -1424,6 +1432,8 @@ mod runs {
None,
StageHandler::Command,
None,
None,
None,
),
]
}

4
lib/apps/fabro-server/src/server.rs
View file

@ -1340,6 +1340,8 @@ pub(crate) fn run_stage_from_stage_id(
started_at: Option<chrono::DateTime<chrono::Utc>>,
handler: StageHandler,
provider_used: Option<StageModelUsage>,
graph_visit: Option<u32>,
resumed_from_stage_id: Option<&StageId>,
) -> RunStage {
RunStage {
id: stage_id.to_string(),
@ -1352,6 +1354,8 @@ pub(crate) fn run_stage_from_stage_id(
.expect("StageId stores a non-zero visit"),
provider_used,
started_at,
graph_visit: graph_visit.and_then(std::num::NonZeroU32::new),
resumed_from_stage_id: resumed_from_stage_id.map(StageId::to_string),
}
}

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

@ -58,6 +58,8 @@ async fn list_run_stages(
stage.started_at,
handler,
stage.provider_used.clone(),
stage.graph_visit,
stage.resumed_from_stage_id.as_ref(),
)
})
.collect::<Vec<_>>();

334
lib/apps/fabro-server/src/server/tests.rs
View file

@ -4090,12 +4090,14 @@ async fn create_durable_run_with_events(
fn stage_started_event(node_id: &str, handler_type: &str) -> workflow_event::Event {
workflow_event::Event::StageStarted {
node_id: node_id.to_string(),
name: node_id.to_string(),
index: 1,
handler_type: handler_type.to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: node_id.to_string(),
name: node_id.to_string(),
index: 1,
handler_type: handler_type.to_string(),
attempt: 1,
max_attempts: 1,
}
}
@ -4938,12 +4940,14 @@ async fn list_run_stages_projects_retrying_until_completion() {
"setup",
1,
&workflow_event::Event::StageStarted {
node_id: "setup".to_string(),
name: "Setup".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "setup".to_string(),
name: "Setup".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -4982,12 +4986,14 @@ async fn list_run_stages_projects_retrying_until_completion() {
"work",
1,
&workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
},
)
.await;
@ -5103,12 +5109,14 @@ async fn list_run_stages_projects_running_stage_as_cancelled_after_cancelled_run
"work",
1,
&workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -5174,12 +5182,14 @@ async fn list_run_stages_includes_stage_model_usage() {
"prompt",
1,
&workflow_event::Event::StageStarted {
node_id: "prompt".to_string(),
name: "Prompt".to_string(),
index: 0,
handler_type: "prompt".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "prompt".to_string(),
name: "Prompt".to_string(),
index: 0,
handler_type: "prompt".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -5294,12 +5304,14 @@ async fn list_run_stages_distinguishes_visits() {
"verify",
1,
&workflow_event::Event::StageStarted {
node_id: "verify".to_string(),
name: "Verify".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "verify".to_string(),
name: "Verify".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -5340,12 +5352,14 @@ async fn list_run_stages_distinguishes_visits() {
"verify",
2,
&workflow_event::Event::StageStarted {
node_id: "verify".to_string(),
name: "Verify".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "verify".to_string(),
name: "Verify".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -5384,6 +5398,110 @@ async fn list_run_stages_distinguishes_visits() {
assert!(first.get("dot_id").is_none(), "dot_id should be removed");
}
#[tokio::test]
async fn list_run_stages_exposes_execution_identity_for_resumed_stage() {
let state = test_app_state_with_isolated_storage();
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = RunId::new();
let mut graph = Graph::new("test");
let mut work = Node::new("work");
work.attrs
.insert("type".to_string(), AttrValue::String("agent".to_string()));
graph.nodes.insert("work".to_string(), work);
create_durable_run_with_events(&state, run_id, &[
workflow_event::Event::RunCreated {
run_id,
title: None,
settings: serde_json::to_value(fabro_types::WorkflowSettings::default()).unwrap(),
graph: serde_json::to_value(&graph).unwrap(),
workflow_source: None,
workflow_config: None,
labels: std::collections::BTreeMap::default(),
run_dir: String::new(),
source_directory: None,
workflow_slug: Some("test".to_string()),
automation: None,
db_prefix: None,
provenance: test_support::test_run_provenance(),
manifest_blob: None,
git: None,
fork_source_ref: None,
retried_from: None,
parent_id: None,
web_url: None,
},
workflow_event::Event::RunStarting,
workflow_event::Event::RunRunning,
])
.await;
// Legacy-shaped first execution without identity metadata.
append_scoped_stage_event(
&state,
run_id,
"work",
1,
&workflow_event::Event::StageStarted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
// Reexecution after cancel/resume: same graph visit, next ordinal.
append_scoped_stage_event(
&state,
run_id,
"work",
2,
&workflow_event::Event::StageStarted {
graph_visit: Some(1),
resumed_from_stage_id: Some(fabro_types::StageId::new("work", 1)),
node_id: "work".to_string(),
name: "Work".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
let response = app
.clone()
.oneshot(
Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/stages")))
.body(Body::empty())
.unwrap(),
)
.await
.unwrap();
let body = response_json!(response, StatusCode::OK).await;
let first = stage_entry(&body, "work@1");
assert!(
first.get("graph_visit").is_none(),
"legacy stage should omit graph_visit"
);
assert!(
first.get("resumed_from_stage_id").is_none(),
"legacy stage should omit resumed_from_stage_id"
);
let second = stage_entry(&body, "work@2");
assert_eq!(second["visit"], 2);
assert_eq!(second["graph_visit"], 1);
assert_eq!(second["resumed_from_stage_id"], "work@1");
}
/// `checkpoint.completed_nodes` records every visit, so a looped node appears
/// once per re-entry. Billing must dedup so a retried node renders as one row
/// and `runtime_secs` is summed across all visits exactly once.
@ -5471,6 +5589,8 @@ async fn run_billing_dedups_retried_nodes_and_sums_their_durations() {
&run_store,
&run_id,
&workflow_event::Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "verify".to_string(),
status: "running".to_string(),
current_node: "verify".to_string(),
@ -5600,6 +5720,8 @@ async fn run_billing_sums_usage_across_retry_visits_and_uses_latest_model() {
&run_store,
&run_id,
&workflow_event::Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "verify".to_string(),
status: "running".to_string(),
current_node: "verify".to_string(),
@ -5689,12 +5811,14 @@ async fn list_run_stages_shows_retrying_after_failed_event() {
"work",
1,
&workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
},
)
.await;
@ -5767,12 +5891,14 @@ async fn list_run_stages_shows_retrying_when_failed_will_retry() {
"work",
1,
&workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
},
)
.await;
@ -5824,12 +5950,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp
workflow_event::Event::RunStarting,
workflow_event::Event::RunRunning,
workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 3,
},
workflow_event::Event::StageFailed {
node_id: "work".to_string(),
@ -5850,12 +5978,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp
delay_ms: 0,
},
workflow_event::Event::StageStarted {
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 2,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
name: "Work".to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 2,
max_attempts: 3,
},
workflow_event::Event::StageCompleted {
node_id: "work".to_string(),
@ -5910,12 +6040,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp
fn revisit_test_started(node_id: &str) -> workflow_event::Event {
workflow_event::Event::StageStarted {
node_id: node_id.to_string(),
name: node_id.to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: node_id.to_string(),
name: node_id.to_string(),
index: 0,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
}
}
@ -8140,12 +8272,14 @@ async fn get_run_stage_command_log_returns_scratch_slice() {
definition_blob: None,
},
workflow_event::Event::StageStarted {
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
workflow_event::Event::CommandStarted {
node_id: "script_node".to_string(),
@ -8207,12 +8341,14 @@ async fn get_run_stage_command_log_returns_cas_slice() {
definition_blob: None,
},
workflow_event::Event::StageStarted {
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
workflow_event::Event::CommandCompleted {
node_id: "script_node".to_string(),
@ -8271,12 +8407,14 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
definition_blob: None,
},
workflow_event::Event::StageStarted {
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "script_node".to_string(),
name: "Script".to_string(),
index: 1,
handler_type: "command".to_string(),
attempt: 1,
max_attempts: 1,
},
workflow_event::Event::CommandCompleted {
node_id: "script_node".to_string(),
@ -9462,12 +9600,14 @@ async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() {
"review",
1,
&workflow_event::Event::StageStarted {
node_id: "review".to_string(),
name: "Review".to_string(),
index: 0,
handler_type: "human".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "review".to_string(),
name: "Review".to_string(),
index: 0,
handler_type: "human".to_string(),
attempt: 1,
max_attempts: 1,
},
)
.await;
@ -11408,6 +11548,8 @@ async fn resume_cancelled_run_with_checkpoint_transitions_to_runnable() {
workflow_event::Event::RunStarting,
workflow_event::Event::RunRunning,
workflow_event::Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: checkpoint.current_node.clone(),
status: "succeeded".to_string(),
current_node: checkpoint.current_node.clone(),

608
lib/components/fabro-store/src/run_state.rs
View file

@ -224,35 +224,46 @@ impl RunProjectionReducer for RunProjection {
}
EventBody::CheckpointCompleted(props) => {
let checkpoint = checkpoint_from_props(props, ts);
if let Some(node_id) = stored.node_id.as_deref() {
let visit = checkpoint
.node_visits
.get(node_id)
.and_then(|visit| u32::try_from(*visit).ok())
.unwrap_or(1);
if let Some(diff) = props.diff.clone() {
self.stage_entry(node_id, visit, first_event_seq(event.seq))
.diff = Some(diff);
if let Some(stage_id) = stored.stage_id.clone() {
// Envelope-first: the diff and any skipped-stage synthesis
// attach to the exact execution recorded on the event.
// Historical `node_outcomes` must not create or collide
// with a newer execution ordinal.
apply_checkpoint_to_stage(self, &stage_id, props, &checkpoint, event.seq, ts);
} else {
// Legacy fallback for events without a stored stage id:
// resolve the visit from the checkpointed `node_visits`
// and synthesize skipped stages from historical outcomes.
if let Some(node_id) = stored.node_id.as_deref() {
let visit = checkpoint
.node_visits
.get(node_id)
.and_then(|visit| u32::try_from(*visit).ok())
.unwrap_or(1);
if let Some(diff) = props.diff.clone() {
self.stage_entry(node_id, visit, first_event_seq(event.seq))
.diff = Some(diff);
}
}
}
for (node_id, outcome) in &checkpoint.node_outcomes {
if outcome.status != StageOutcome::Skipped {
continue;
for (node_id, outcome) in &checkpoint.node_outcomes {
if outcome.status != StageOutcome::Skipped {
continue;
}
let visit = checkpoint
.node_visits
.get(node_id)
.and_then(|visit| u32::try_from(*visit).ok())
.unwrap_or(1);
if self
.stage(&fabro_types::StageId::new(node_id, visit))
.is_some()
{
continue;
}
let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq));
stage.completion = Some(stage_completion_from_outcome(outcome, ts));
stage.state = StageState::Skipped;
}
let visit = checkpoint
.node_visits
.get(node_id)
.and_then(|visit| u32::try_from(*visit).ok())
.unwrap_or(1);
if self
.stage(&fabro_types::StageId::new(node_id, visit))
.is_some()
{
continue;
}
let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq));
stage.completion = Some(stage_completion_from_outcome(outcome, ts));
stage.state = StageState::Skipped;
}
self.checkpoints.push(CheckpointRecord {
seq: event.seq,
@ -337,6 +348,11 @@ impl RunProjectionReducer for RunProjection {
let Some(stage_id) = stored.stage_id.as_ref() else {
return Ok(());
};
// A `stage.started` for a new `StageId` creates a new
// projection, so an older execution's terminal projection
// stays immutable. `begin_attempt` on an existing entry
// remains the compatibility path for automatic retries and
// legacy histories that repeat one `StageId`.
let stage = self.stage_entry(
stage_id.node_id(),
stage_id.visit(),
@ -346,6 +362,14 @@ impl RunProjectionReducer for RunProjection {
ts,
StageHandler::from_handler_type(Some(&props.handler_type)),
);
if props.graph_visit.is_some() {
stage.graph_visit = props.graph_visit;
}
if props.resumed_from_stage_id.is_some() {
stage
.resumed_from_stage_id
.clone_from(&props.resumed_from_stage_id);
}
}
EventBody::StageRetrying(_) => {
let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else {
@ -537,7 +561,7 @@ impl RunProjectionReducer for RunProjection {
};
stage.parallel_results = Some(parallel_results);
}
EventBody::ParallelBranchStarted(_) => {
EventBody::ParallelBranchStarted(props) => {
// Branches bypass the engine's StageStarted/StageCompleted
// lifecycle. Seed started_at so the branch stage drives a live
// wall-clock timer while it runs (the entry is created Running).
@ -547,6 +571,14 @@ impl RunProjectionReducer for RunProjection {
if stage.started_at.is_none() {
stage.started_at = Some(ts);
}
if props.graph_visit.is_some() {
stage.graph_visit = props.graph_visit;
}
if props.resumed_from_stage_id.is_some() {
stage
.resumed_from_stage_id
.clone_from(&props.resumed_from_stage_id);
}
stage.state = StageState::Running;
}
EventBody::ParallelBranchCompleted(props) => {
@ -951,6 +983,50 @@ fn stage_at_stored_or_current_visit<'a>(
stage_at_current_visit(state, stored, seq)
}
/// Apply a `checkpoint.completed` event to the exact execution named by the
/// envelope `StageId`: attach the checkpoint diff, and for a skipped
/// checkpoint finalize that execution as `Skipped`. A synthetic skipped
/// projection is created only for a first-attempt skip that never
/// materialized a stage; an existing non-terminal (e.g. `Retrying`)
/// projection keeps its identity and becomes terminal, while an older
/// terminal execution stays immutable.
fn apply_checkpoint_to_stage(
state: &mut RunProjection,
stage_id: &StageId,
props: &CheckpointCompletedProps,
checkpoint: &Checkpoint,
seq: u32,
ts: DateTime<Utc>,
) {
let node_id = stage_id.node_id();
let skipped_completion = checkpoint
.node_outcomes
.get(node_id)
.filter(|outcome| outcome.status == StageOutcome::Skipped)
.map(|outcome| stage_completion_from_outcome(outcome, ts));
if props.diff.is_none() && skipped_completion.is_none() {
return;
}
let is_new = state.stage(stage_id).is_none();
let stage = state.stage_entry(node_id, stage_id.visit(), first_event_seq(seq));
if is_new {
stage.graph_visit = props.graph_visit;
stage
.resumed_from_stage_id
.clone_from(&props.resumed_from_stage_id);
}
if let Some(diff) = props.diff.clone() {
stage.diff = Some(diff);
}
if let Some(completion) = skipped_completion {
if !stage.state.is_terminal() {
stage.completion = Some(completion);
stage.state = StageState::Skipped;
}
}
}
fn stage_at_completed_visit<'a>(
state: &'a mut RunProjection,
stored: &RunEvent,
@ -2012,10 +2088,12 @@ mod tests {
.apply_event(&test_stage_event(
3,
EventBody::StageStarted(StageStartedProps {
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
}),
stage_id.clone(),
))
@ -2034,10 +2112,12 @@ mod tests {
.apply_event(&test_stage_event(
3,
EventBody::StageStarted(StageStartedProps {
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
}),
stage_id.clone(),
))
@ -2077,7 +2157,11 @@ mod tests {
.apply_event(&test_stage_event_at(
3,
"2026-04-07T12:00:00Z",
EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }),
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
}),
branch.clone(),
))
.unwrap();
@ -2115,7 +2199,11 @@ mod tests {
state
.apply_event(&test_stage_event(
3,
EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }),
EventBody::ParallelBranchStarted(ParallelBranchStartedProps {
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
}),
branch.clone(),
))
.unwrap();
@ -2148,10 +2236,12 @@ mod tests {
.apply_event(&test_stage_event(
3,
EventBody::StageStarted(StageStartedProps {
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
}),
stage_id.clone(),
))
@ -2404,10 +2494,12 @@ mod tests {
.apply_event(&test_stage_event(
2,
EventBody::StageStarted(StageStartedProps {
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
}),
stage_id.clone(),
))
@ -2575,6 +2667,8 @@ mod tests {
.apply_event(&test_event(
5,
EventBody::CheckpointCompleted(CheckpointCompletedProps {
graph_visit: None,
resumed_from_stage_id: None,
status: "running".to_string(),
current_node: "next".to_string(),
completed_nodes: vec!["skip_me".to_string()],
@ -3831,10 +3925,12 @@ mod tests {
fn started_props() -> StageStartedProps {
StageStartedProps {
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 3,
graph_visit: None,
resumed_from_stage_id: None,
index: 0,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 3,
}
}
@ -4334,6 +4430,418 @@ mod tests {
);
}
#[test]
fn stage_started_records_execution_identity_metadata() {
let mut state = running_projection();
let stage_id = StageId::new("work", 2);
state
.apply_event(&test_stage_event(
4,
EventBody::StageStarted(StageStartedProps {
graph_visit: Some(1),
resumed_from_stage_id: Some(StageId::new("work", 1)),
..started_props()
}),
stage_id.clone(),
))
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.graph_visit, Some(1));
assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1)));
}
#[test]
fn retry_stage_started_preserves_execution_identity_metadata() {
let mut state = running_projection();
let stage_id = StageId::new("work", 2);
state
.apply_event(&test_stage_event(
4,
EventBody::StageStarted(StageStartedProps {
graph_visit: Some(1),
resumed_from_stage_id: Some(StageId::new("work", 1)),
..started_props()
}),
stage_id.clone(),
))
.unwrap();
// A legacy-shaped retry event for the same StageId omits the identity
// fields; the projection keeps the first attempt's metadata.
state
.apply_event(&test_stage_event(
5,
EventBody::StageStarted(StageStartedProps {
attempt: 2,
..started_props()
}),
stage_id.clone(),
))
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.graph_visit, Some(1));
assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1)));
}
#[test]
fn cancelled_execution_stays_immutable_when_resumed_execution_starts() {
let mut state = running_projection();
let first = StageId::new("work", 1);
let second = StageId::new("work", 2);
state
.apply_event(&test_stage_event_at(
4,
"2026-04-07T12:00:00Z",
EventBody::StageStarted(started_props()),
first.clone(),
))
.unwrap();
state
.apply_event(&test_raw_event_at(
5,
"2026-04-07T12:00:05Z",
"run.failed",
&serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(),
None,
))
.unwrap();
state
.apply_event(&test_raw_event(
6,
"run.start_requested",
&json!({ "resume": true }),
None,
))
.unwrap();
state
.apply_event(&test_raw_event(
7,
"run.runnable",
&json!({ "source": "start_requested" }),
None,
))
.unwrap();
state
.apply_event(&test_raw_event(8, "run.starting", &json!({}), None))
.unwrap();
state
.apply_event(&test_raw_event(9, "run.running", &json!({}), None))
.unwrap();
state
.apply_event(&test_stage_event(
10,
EventBody::StageStarted(StageStartedProps {
graph_visit: Some(1),
resumed_from_stage_id: Some(first.clone()),
..started_props()
}),
second.clone(),
))
.unwrap();
// The cancelled execution keeps its terminal projection untouched...
let cancelled = state.stage(&first).unwrap();
assert_eq!(cancelled.state, StageState::Cancelled);
assert_eq!(
cancelled.timing,
Some(fabro_types::StageTiming::wall_only(5_000))
);
// ...while the reexecution runs as a distinct stage linked back to it.
let resumed = state.stage(&second).unwrap();
assert_eq!(resumed.state, StageState::Running);
assert_eq!(resumed.graph_visit, Some(1));
assert_eq!(resumed.resumed_from_stage_id, Some(first));
}
#[test]
fn run_failed_after_resume_preserves_earlier_terminal_executions() {
let mut state = running_projection();
let done = StageId::new("verify", 1);
let active = StageId::new("work", 2);
state
.apply_event(&test_stage_event_at(
4,
"2026-04-07T12:00:00Z",
EventBody::StageStarted(started_props()),
done.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event_at(
5,
"2026-04-07T12:00:03Z",
EventBody::StageCompleted(completed_props(3_000, StageOutcome::Succeeded)),
done.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event_at(
6,
"2026-04-07T12:00:04Z",
EventBody::StageStarted(started_props()),
active.clone(),
))
.unwrap();
state
.apply_event(&test_raw_event_at(
7,
"2026-04-07T12:00:09Z",
"run.failed",
&serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(),
None,
))
.unwrap();
let terminal = state.stage(&done).unwrap();
assert_eq!(terminal.state, StageState::Succeeded);
assert_eq!(
terminal.timing,
Some(fabro_types::StageTiming::wall_only(3_000))
);
assert_eq!(state.stage(&active).unwrap().state, StageState::Cancelled);
}
#[test]
fn checkpoint_completed_targets_envelope_stage_id_after_ordinal_divergence() {
let mut state = running_projection();
let execution = StageId::new("work", 2);
state
.apply_event(&test_stage_event(
4,
EventBody::StageStarted(StageStartedProps {
graph_visit: Some(1),
..started_props()
}),
execution.clone(),
))
.unwrap();
// Checkpointed graph state still says visit 1 for `work`, and carries
// a historical skipped outcome for another node. Neither may create
// or mutate a projection at a stale ordinal.
state
.apply_event(&test_stage_event(
5,
EventBody::CheckpointCompleted(CheckpointCompletedProps {
graph_visit: Some(1),
resumed_from_stage_id: None,
status: "success".to_string(),
current_node: "work".to_string(),
completed_nodes: vec!["old_skip".to_string(), "work".to_string()],
node_retries: BTreeMap::new(),
context_values: BTreeMap::new(),
node_outcomes: BTreeMap::from([(
"old_skip".to_string(),
Outcome::skipped("historical skip"),
)]),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: BTreeMap::new(),
restart_failure_signatures: BTreeMap::new(),
node_visits: BTreeMap::from([
("work".to_string(), 1usize),
("old_skip".to_string(), 1usize),
]),
diff: Some("diff for work@2".to_string()),
diff_summary: None,
}),
execution.clone(),
))
.unwrap();
assert_eq!(
state.stage(&execution).unwrap().diff.as_deref(),
Some("diff for work@2")
);
assert!(state.stage(&StageId::new("work", 1)).is_none());
assert!(state.stage(&StageId::new("old_skip", 1)).is_none());
}
#[test]
fn skipped_checkpoint_with_stage_id_creates_synthetic_execution() {
let mut state = running_projection();
let execution = StageId::new("gate", 3);
// First-attempt StageStart-hook skip: no stage.started was emitted,
// the checkpoint is the first stage-scoped event for this execution.
state
.apply_event(&test_stage_event(
4,
EventBody::CheckpointCompleted(CheckpointCompletedProps {
graph_visit: Some(2),
resumed_from_stage_id: Some(StageId::new("gate", 2)),
status: "skipped".to_string(),
current_node: "gate".to_string(),
completed_nodes: vec!["gate".to_string()],
node_retries: BTreeMap::new(),
context_values: BTreeMap::new(),
node_outcomes: BTreeMap::from([(
"gate".to_string(),
Outcome::skipped("skipped by StageStart hook"),
)]),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: BTreeMap::new(),
restart_failure_signatures: BTreeMap::new(),
node_visits: BTreeMap::from([("gate".to_string(), 2usize)]),
diff: None,
diff_summary: None,
}),
execution.clone(),
))
.unwrap();
let stage = state.stage(&execution).unwrap();
assert_eq!(stage.state, StageState::Skipped);
assert_eq!(stage.graph_visit, Some(2));
assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("gate", 2)));
assert_eq!(
stage.completion.as_ref().unwrap().notes.as_deref(),
Some("skipped by StageStart hook")
);
}
#[test]
fn skipped_checkpoint_finalizes_existing_retrying_execution() {
let mut state = running_projection();
let execution = StageId::new("gate", 1);
state
.apply_event(&test_stage_event(
4,
EventBody::StageStarted(started_props()),
execution.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event(
5,
EventBody::StageRetrying(StageRetryingProps {
index: 0,
attempt: 2,
max_attempts: 3,
delay_ms: 0,
}),
execution.clone(),
))
.unwrap();
assert_eq!(state.stage(&execution).unwrap().state, StageState::Retrying);
// StageStart hook skipped the retry; the checkpoint finalizes the
// existing execution instead of allocating a new projection.
state
.apply_event(&test_stage_event(
6,
EventBody::CheckpointCompleted(CheckpointCompletedProps {
graph_visit: Some(1),
resumed_from_stage_id: None,
status: "skipped".to_string(),
current_node: "gate".to_string(),
completed_nodes: vec!["gate".to_string()],
node_retries: BTreeMap::new(),
context_values: BTreeMap::new(),
node_outcomes: BTreeMap::from([(
"gate".to_string(),
Outcome::skipped("skipped on retry"),
)]),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: BTreeMap::new(),
restart_failure_signatures: BTreeMap::new(),
node_visits: BTreeMap::from([("gate".to_string(), 1usize)]),
diff: None,
diff_summary: None,
}),
execution.clone(),
))
.unwrap();
let stage = state.stage(&execution).unwrap();
assert_eq!(stage.state, StageState::Skipped);
assert_eq!(stage.first_event_seq, first_event_seq(4));
assert!(state.stage(&StageId::new("gate", 2)).is_none());
}
#[test]
fn skipped_checkpoint_never_reopens_an_older_terminal_execution() {
let mut state = running_projection();
let execution = StageId::new("gate", 1);
state
.apply_event(&test_stage_event(
4,
EventBody::StageStarted(started_props()),
execution.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event(
5,
EventBody::StageCompleted(completed_props(2_000, StageOutcome::Succeeded)),
execution.clone(),
))
.unwrap();
state
.apply_event(&test_stage_event(
6,
EventBody::CheckpointCompleted(CheckpointCompletedProps {
graph_visit: Some(1),
resumed_from_stage_id: None,
status: "skipped".to_string(),
current_node: "gate".to_string(),
completed_nodes: vec!["gate".to_string()],
node_retries: BTreeMap::new(),
context_values: BTreeMap::new(),
node_outcomes: BTreeMap::from([(
"gate".to_string(),
Outcome::skipped("late skip"),
)]),
next_node_id: None,
git_commit_sha: None,
loop_failure_signatures: BTreeMap::new(),
restart_failure_signatures: BTreeMap::new(),
node_visits: BTreeMap::from([("gate".to_string(), 1usize)]),
diff: None,
diff_summary: None,
}),
execution.clone(),
))
.unwrap();
let stage = state.stage(&execution).unwrap();
assert_eq!(stage.state, StageState::Succeeded);
assert_eq!(
stage.completion.as_ref().unwrap().outcome,
StageOutcome::Succeeded
);
}
#[test]
fn legacy_stage_started_payload_without_identity_fields_deserializes() {
let event = test_raw_event(
4,
"stage.started",
&json!({
"index": 0,
"handler_type": "agent",
"attempt": 1,
"max_attempts": 3
}),
Some("work"),
);
let EventBody::StageStarted(props) = &event.event.body else {
panic!("expected stage.started body");
};
assert_eq!(props.graph_visit, None);
assert_eq!(props.resumed_from_stage_id, None);
}
#[test]
fn run_failed_non_cancelled_finalizes_running_stage_as_failed() {
let mut state = running_projection();

14
lib/components/fabro-workflow/src/context.rs
View file

@ -22,6 +22,13 @@ pub mod keys {
pub const INTERNAL_FIDELITY: &str = "internal.fidelity";
pub const INTERNAL_THREAD_ID: &str = "internal.thread_id";
pub const INTERNAL_NODE_VISIT_COUNT: &str = "internal.node_visit_count";
/// 1-based stage execution ordinal for the currently-executing node — the
/// numeric component of the external `StageId`. Runtime-only: reserved by
/// the lifecycle when a stage execution first becomes observable and
/// stripped from durable context snapshots, unlike
/// [`INTERNAL_NODE_VISIT_COUNT`], which remains the checkpointed graph
/// visit.
pub const INTERNAL_STAGE_EXECUTION_ORDINAL: &str = "internal.stage_execution_ordinal";
pub const INTERNAL_PARENT_PREAMBLE: &str = "internal.parent_preamble";
pub const INTERNAL_PARALLEL_GROUP_ID: &str = "internal.parallel_group_id";
pub const INTERNAL_PARALLEL_BRANCH_ID: &str = "internal.parallel_branch_id";
@ -49,8 +56,11 @@ pub mod keys {
pub const PARALLEL_FAN_IN_BEST_HEAD_SHA: &str = "parallel.fan_in.best_head_sha";
/// Runtime-only keys stripped from durable context projections.
pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] =
&[CURRENT_PREAMBLE, INTERNAL_PARALLEL_BRANCH_PREAMBLES];
pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = &[
CURRENT_PREAMBLE,
INTERNAL_PARALLEL_BRANCH_PREAMBLES,
INTERNAL_STAGE_EXECUTION_ORDINAL,
];
// --- Prefix constants (for filtering and dynamic keys) ---
pub const GRAPH_PREFIX: &str = "graph.";

51
lib/components/fabro-workflow/src/event/convert.rs
View file

@ -295,12 +295,16 @@ fn event_body_from_event(event: &Event) -> EventBody {
handler_type,
attempt,
max_attempts,
graph_visit,
resumed_from_stage_id,
..
} => EventBody::StageStarted(fabro_types::StageStartedProps {
index: *index,
index: *index,
handler_type: handler_type.clone(),
attempt: *attempt,
attempt: *attempt,
max_attempts: *max_attempts,
graph_visit: *graph_visit,
resumed_from_stage_id: resumed_from_stage_id.clone(),
}),
Event::StageCompleted {
index,
@ -378,11 +382,16 @@ fn event_body_from_event(event: &Event) -> EventBody {
branch_count: *branch_count,
join_policy: join_policy.clone(),
}),
Event::ParallelBranchStarted { index, .. } => {
EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps {
index: *index,
})
}
Event::ParallelBranchStarted {
index,
graph_visit,
resumed_from_stage_id,
..
} => EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps {
index: *index,
graph_visit: *graph_visit,
resumed_from_stage_id: resumed_from_stage_id.clone(),
}),
Event::ParallelBranchCompleted {
index,
duration_ms,
@ -480,6 +489,8 @@ fn event_body_from_event(event: &Event) -> EventBody {
node_visits,
diff,
diff_summary,
graph_visit,
resumed_from_stage_id,
..
} => EventBody::CheckpointCompleted(fabro_types::CheckpointCompletedProps {
status: status.clone(),
@ -495,6 +506,8 @@ fn event_body_from_event(event: &Event) -> EventBody {
node_visits: node_visits.clone(),
diff: diff.clone(),
diff_summary: *diff_summary,
graph_visit: *graph_visit,
resumed_from_stage_id: resumed_from_stage_id.clone(),
}),
Event::CheckpointFailed {
error,
@ -1693,12 +1706,14 @@ mod tests {
let stored = to_run_event_at(
&fixtures::RUN_1,
&Event::StageStarted {
node_id: "review".to_string(),
name: "review".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: None,
resumed_from_stage_id: None,
node_id: "review".to_string(),
name: "review".to_string(),
index: 1,
handler_type: "agent".to_string(),
attempt: 1,
max_attempts: 1,
},
Utc::now(),
Some(&StageScope {
@ -1730,10 +1745,12 @@ mod tests {
#[test]
fn parallel_branch_started_populates_group_and_branch_ids() {
let stored = to_run_event(&fixtures::RUN_1, &Event::ParallelBranchStarted {
parallel_group_id: StageId::new("fanout", 2),
parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1),
branch: "review".to_string(),
index: 1,
graph_visit: None,
resumed_from_stage_id: None,
parallel_group_id: StageId::new("fanout", 2),
parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1),
branch: "review".to_string(),
index: 1,
});
assert_eq!(stored.parallel_group_id, Some(StageId::new("fanout", 2)));
assert_eq!(

40
lib/components/fabro-workflow/src/event/events.rs
View file

@ -244,12 +244,22 @@ pub enum Event {
exec_output_tail: Option<fabro_types::ExecOutputTail>,
},
StageStarted {
node_id: String,
name: String,
index: usize,
handler_type: String,
attempt: usize,
max_attempts: usize,
node_id: String,
name: String,
index: usize,
handler_type: String,
attempt: usize,
max_attempts: usize,
/// Graph visit that produced this stage execution. Diverges from the
/// envelope `StageId` ordinal when a cancelled or crashed invocation
/// is reexecuted after resume.
#[serde(default, skip_serializing_if = "Option::is_none")]
graph_visit: Option<u32>,
/// Prior execution this one resumes from, for the first execution
/// reserved after a resume when the node had an observable
/// post-checkpoint execution.
#[serde(default, skip_serializing_if = "Option::is_none")]
resumed_from_stage_id: Option<StageId>,
},
StageCompleted {
node_id: String,
@ -307,10 +317,14 @@ pub enum Event {
join_policy: String,
},
ParallelBranchStarted {
parallel_group_id: StageId,
parallel_branch_id: ParallelBranchId,
branch: String,
index: usize,
parallel_group_id: StageId,
parallel_branch_id: ParallelBranchId,
branch: String,
index: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
graph_visit: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
resumed_from_stage_id: Option<StageId>,
},
ParallelBranchCompleted {
parallel_group_id: StageId,
@ -396,6 +410,12 @@ pub enum Event {
diff: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
diff_summary: Option<DiffSummary>,
/// Graph visit of the checkpointed stage execution; used when this
/// checkpoint is the event that first materializes a skipped stage.
#[serde(default, skip_serializing_if = "Option::is_none")]
graph_visit: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
resumed_from_stage_id: Option<StageId>,
},
CheckpointFailed {
node_id: String,

10
lib/components/fabro-workflow/src/event/names.rs
View file

@ -170,10 +170,12 @@ mod tests {
fn event_name_matches_new_dot_notation() {
assert_eq!(
event_name(&Event::ParallelBranchStarted {
parallel_group_id: StageId::new("plan", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0),
branch: "fork".to_string(),
index: 0,
graph_visit: None,
resumed_from_stage_id: None,
parallel_group_id: StageId::new("plan", 1),
parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0),
branch: "fork".to_string(),
index: 0,
}),
"parallel.branch.started"
);

2
lib/components/fabro-workflow/src/git.rs
View file

@ -555,6 +555,8 @@ mod tests {
.await
.unwrap();
append_event(&run, &fixtures::RUN_1, &Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".into(),
status: "succeeded".into(),
current_node: "work".into(),

8
lib/components/fabro-workflow/src/handler/manager_loop.rs
View file

@ -16,10 +16,10 @@ use crate::artifact_upload::ArtifactSink;
use crate::condition::evaluate_condition;
use crate::context::{Context, WorkflowContext, keys};
use crate::error::Error;
use crate::event::StageScope;
use crate::operations::{ValidateInput, WorkflowInput, validate};
use crate::outcome::{Outcome, OutcomeExt, StageOutcome};
use crate::pipeline::types::Initialized;
use crate::run_dir::visit_from_context;
use crate::run_options::RunOptions;
use crate::static_reference::{ReferenceKind, validate_static_reference};
use crate::{ManifestPath, pipeline};
@ -197,8 +197,10 @@ impl Handler for SubWorkflowHandler {
}
};
// Build child RunOptions
let visit = visit_from_context(context) as u64;
// Build child RunOptions. The stage directory follows the execution
// ordinal so a reexecuted manager loop keeps the cancelled
// invocation's child logs intact.
let visit = u64::from(StageScope::for_handler(context, &node.id).visit);
let child_logs = run_dir.join(format!("stages/{}@{visit}/child", node.id));
let _ = fs::create_dir_all(&child_logs).await;

37
lib/components/fabro-workflow/src/handler/parallel.rs
View file

@ -17,10 +17,10 @@ use crate::git::sanitize_ref_component;
use crate::hook_context::set_hook_node;
use crate::millis_u64;
use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt, StageOutcome};
use crate::run_dir::visit_from_context;
use crate::sandbox_git::{
GIT_REMOTE, checked_git_checkpoint, git_merge_ff_only, git_remove_worktree,
};
use crate::stage_execution::StageExecution;
/// Fans out execution to multiple branches concurrently.
/// Each branch gets an isolated context clone and runs independently.
@ -161,6 +161,9 @@ impl Handler for ParallelHandler {
branch_context: Context,
sandbox: Arc<dyn Sandbox>,
worktree_path: Option<PathBuf>,
/// Child stage execution reserved through the run's shared
/// tracker, so a resumed fan-out gets fresh branch identities.
execution: StageExecution,
}
let parallel_start = Instant::now();
@ -264,6 +267,19 @@ impl Handler for ParallelHandler {
parallel_group_id.clone(),
u32::try_from(branch_index).unwrap_or(u32::MAX),
);
// Reserve the child's stage execution through the shared tracker
// and seed the branch context with its explicit stage scope, so
// branch lifecycle events and nested handler events agree on the
// child's identity instead of inheriting the fork's.
let execution = services.run.stage_executions.reserve(&target_id, 1);
branch_context.set(
keys::CURRENT_NODE,
serde_json::Value::String(target_id.clone()),
);
branch_context.set(
keys::INTERNAL_STAGE_EXECUTION_ORDINAL,
serde_json::json!(execution.ordinal),
);
branch_context.set(
keys::INTERNAL_PARALLEL_GROUP_ID,
serde_json::Value::String(parallel_group_id.to_string()),
@ -294,12 +310,14 @@ impl Handler for ParallelHandler {
(&git_state, &base_sha)
{
let branch_key = &target_id;
let visit = visit_from_context(&branch_context);
// `pass{N}` derives from the parent's execution ordinal so a
// resumed fan-out does not recreate the cancelled attempt's
// branch names.
let branch_name = format!(
"fabro/run/parallel/{}/{}/pass{}/{}",
gs.run_id,
sanitize_ref_component(&node.id),
visit,
parallel_stage_scope.visit,
sanitize_ref_component(branch_key),
);
@ -345,6 +363,7 @@ impl Handler for ParallelHandler {
branch_context,
sandbox: branch_sandbox,
worktree_path,
execution,
});
}
@ -376,7 +395,7 @@ impl Handler for ParallelHandler {
let group_id = parallel_group_id.clone();
let branch_scope = StageScope::for_parallel_branch(
setup.target_id.clone(),
1,
setup.execution.ordinal,
group_id.clone(),
setup.parallel_branch_id.clone(),
);
@ -389,10 +408,12 @@ impl Handler for ParallelHandler {
parent_run.emitter.emit_scoped(
&Event::ParallelBranchStarted {
parallel_group_id: group_id.clone(),
parallel_branch_id: setup.parallel_branch_id.clone(),
branch: setup.target_id.clone(),
index: setup.branch_index,
parallel_group_id: group_id.clone(),
parallel_branch_id: setup.parallel_branch_id.clone(),
branch: setup.target_id.clone(),
index: setup.branch_index,
graph_visit: Some(setup.execution.graph_visit),
resumed_from_stage_id: setup.execution.resumed_from.clone(),
},
&branch_scope,
);

1
lib/components/fabro-workflow/src/lib.rs
View file

@ -329,6 +329,7 @@ pub mod runtime_store;
pub mod sandbox_git;
pub(crate) mod sandbox_git_runtime;
pub mod services;
pub mod stage_execution;
mod stage_scope;
pub mod static_reference;
pub mod steering_hub;

14
lib/components/fabro-workflow/src/lifecycle/artifact.rs
View file

@ -23,6 +23,7 @@ use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::{stage_scope_for, stage_visit};
use crate::outcome::BilledModelUsage;
use crate::runtime_store::RunStoreHandle;
use crate::stage_execution::StageExecutionTracker;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -46,6 +47,8 @@ pub(crate) struct ArtifactLifecycle {
/// Per-attempt state: epoch seconds when the attempt started.
attempt_start_epoch: std::sync::Mutex<Option<f64>>,
captured_artifacts: std::sync::Mutex<HashSet<ArtifactIdentity>>,
/// Run-scoped stage execution allocator shared with `RunServices`.
stage_executions: StageExecutionTracker,
}
impl ArtifactLifecycle {
@ -56,6 +59,7 @@ impl ArtifactLifecycle {
run_id: RunId,
artifact_globs: Vec<String>,
artifact_sink: Option<ArtifactSink>,
stage_executions: StageExecutionTracker,
) -> Self {
Self {
sandbox,
@ -66,6 +70,7 @@ impl ArtifactLifecycle {
artifact_sink,
attempt_start_epoch: std::sync::Mutex::new(None),
captured_artifacts: std::sync::Mutex::new(HashSet::new()),
stage_executions,
}
}
}
@ -120,7 +125,12 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
.expect("artifact mutex should not be poisoned: no code panics while holding this lock")
.unwrap_or(0.0);
let node_id = ctx.node.id();
let visit = stage_visit(state, node_id);
// Artifact identity follows the stage execution ordinal so a resumed
// reexecution stores its captures under the new `StageId`.
let visit = self.stage_executions.active(node_id).map_or_else(
|| stage_visit(state, node_id),
|execution| execution.ordinal,
);
let node_slug = if visit <= 1 {
node_id.to_string()
} else {
@ -162,7 +172,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
return Ok(());
}
self.record_captured_assets(&new_assets);
let scope = stage_scope_for(state, node_id);
let scope = stage_scope_for(&self.stage_executions, state, node_id);
for asset in &new_assets {
self.emitter.emit_scoped(
&Event::ArtifactCaptured {

75
lib/components/fabro-workflow/src/lifecycle/event.rs
View file

@ -18,6 +18,7 @@ use crate::context::{Context, WorkflowContext};
use crate::event::{Emitter, Event, StageScope};
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageOutcome};
use crate::stage_execution::StageExecutionTracker;
use crate::{artifact, context};
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
@ -46,6 +47,8 @@ pub(crate) struct EventLifecycle {
/// EventLifecycle when emitting CheckpointCompleted).
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub circuit_breaker: Arc<CircuitBreakerLifecycle>,
/// Run-scoped stage execution allocator shared with `RunServices`.
pub stage_executions: StageExecutionTracker,
}
fn snapshot_failure_signatures(
@ -107,11 +110,22 @@ pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 {
u32::try_from(visits).unwrap_or(u32::MAX)
}
pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope {
/// Build the emission scope for a node from its active stage execution.
/// Falls back to the graph visit for direct unit-test call sites that emit
/// without a reservation; the two are equal for a first execution.
pub(crate) fn stage_scope_for(
stage_executions: &StageExecutionTracker,
state: &WfRunState,
node_id: &str,
) -> StageScope {
let visit = stage_executions.active(node_id).map_or_else(
|| stage_visit(state, node_id),
|execution| execution.ordinal,
);
StageScope {
node_id: node_id.to_string(),
visit: stage_visit(state, node_id),
parallel_group_id: state.context.parallel_group_id(),
node_id: node_id.to_string(),
visit,
parallel_group_id: state.context.parallel_group_id(),
parallel_branch_id: state.context.parallel_branch_id(),
}
}
@ -160,17 +174,24 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
}
let gv = node.inner();
let stage_index = state.stage_index;
let scope = stage_scope_for(state, &gv.id);
// Terminal nodes bypass `before_node`/`before_attempt`, so their
// synthetic paired events reserve an execution here.
let execution = self
.stage_executions
.reserve(&gv.id, stage_visit(state, &gv.id));
let scope = stage_scope_for(&self.stage_executions, state, &gv.id);
let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker);
self.emitter.emit_scoped(
&Event::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().unwrap_or_default().to_string(),
attempt: 1,
max_attempts: 1,
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: stage_index,
handler_type: gv.handler_type().unwrap_or_default().to_string(),
attempt: 1,
max_attempts: 1,
graph_visit: Some(execution.graph_visit),
resumed_from_stage_id: execution.resumed_from,
},
&scope,
);
@ -210,15 +231,21 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
state: &WfRunState,
) -> CoreResult<NodeDecision<Option<BilledModelUsage>>> {
let gv = ctx.node.inner();
let scope = stage_scope_for(state, &gv.id);
let execution = self.stage_executions.active(&gv.id);
let scope = stage_scope_for(&self.stage_executions, state, &gv.id);
let graph_visit = execution
.as_ref()
.map_or_else(|| stage_visit(state, &gv.id), |e| e.graph_visit);
self.emitter.emit_scoped(
&Event::StageStarted {
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: state.stage_index,
handler_type: gv.handler_type().unwrap_or_default().to_string(),
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
node_id: gv.id.clone(),
name: gv.label().to_string(),
index: state.stage_index,
handler_type: gv.handler_type().unwrap_or_default().to_string(),
attempt: ctx.attempt as usize,
max_attempts: ctx.max_attempts as usize,
graph_visit: Some(graph_visit),
resumed_from_stage_id: execution.and_then(|e| e.resumed_from),
},
&scope,
);
@ -234,7 +261,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
let gv = ctx.node.inner();
let outcome = &ctx.result.outcome;
let stage_index = state.stage_index;
let scope = stage_scope_for(state, &gv.id);
let scope = stage_scope_for(&self.stage_executions, state, &gv.id);
let timing = node_result_timing(ctx.result);
let failure = outcome.failure.clone().unwrap_or_else(|| {
@ -283,7 +310,7 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
}
let gv = node.inner();
let stage_index = state.stage_index;
let scope = stage_scope_for(state, &gv.id);
let scope = stage_scope_for(&self.stage_executions, state, &gv.id);
let timing = node_result_timing(result);
let (loop_failure_signatures, restart_failure_signatures) =
snapshot_failure_signatures(&self.circuit_breaker);
@ -400,7 +427,11 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
node_outcomes.insert(node.id().to_string(), result.outcome.clone());
artifact::normalize_durable_outcomes(&mut node_outcomes);
let scope = stage_scope_for(state, node.id());
let execution = self.stage_executions.active(node.id());
let scope = stage_scope_for(&self.stage_executions, state, node.id());
let graph_visit = execution
.as_ref()
.map_or_else(|| stage_visit(state, node.id()), |e| e.graph_visit);
self.emitter.emit_scoped(
&Event::CheckpointCompleted {
node_id: node.id().to_string(),
@ -425,6 +456,8 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
.collect::<BTreeMap<_, _>>(),
diff,
diff_summary,
graph_visit: Some(graph_visit),
resumed_from_stage_id: execution.and_then(|e| e.resumed_from),
},
&scope,
);

9
lib/components/fabro-workflow/src/lifecycle/git.rs
View file

@ -25,6 +25,7 @@ use crate::sandbox_git::{
checked_git_checkpoint, git_diff, list_diff_numstat, summarize_diff_numstat,
};
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::stage_execution::StageExecutionTracker;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -88,6 +89,8 @@ pub(crate) struct GitLifecycle {
// Cross-lifecycle data (shared with EventLifecycle)
pub checkpoint_git_result: Arc<Mutex<Option<GitCheckpointResult>>>,
pub last_git_sha: Arc<Mutex<Option<String>>>,
/// Run-scoped stage execution allocator shared with `RunServices`.
pub stage_executions: StageExecutionTracker,
}
#[async_trait]
@ -197,7 +200,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
} else {
let phase = MetadataSnapshotPhase::Checkpoint;
let started = Instant::now();
let scope = stage_scope_for(state, node_id);
let scope = stage_scope_for(&self.stage_executions, state, node_id);
self.emit_metadata_snapshot_started(phase, &meta_branch, Some(&scope));
match self.run_store.state().await {
Ok(mut projection) => {
@ -401,7 +404,7 @@ impl RunLifecycle<WorkflowGraph> for GitLifecycle {
let exec_output_tail = fabro_sandbox::default_redacted_output_tail(&e);
let error = e.to_string();
// Emit CheckpointFailed and return error
let scope = stage_scope_for(state, node_id);
let scope = stage_scope_for(&self.stage_executions, state, node_id);
self.emitter.emit_scoped(
&Event::CheckpointFailed {
node_id: node_id.to_string(),
@ -788,6 +791,7 @@ mod tests {
metadata_writer: Option<RunMetadataWriterHandle>,
) -> GitLifecycle {
GitLifecycle {
stage_executions: StageExecutionTracker::default(),
sandbox: Arc::new(fabro_agent::LocalSandbox::new(repo.to_path_buf())),
emitter,
run_id: fixtures::RUN_1,
@ -1262,6 +1266,7 @@ mod tests {
Arc::new(SandboxGitRuntime::new()),
Arc::clone(&lifecycle.metadata_runtime),
lifecycle.metadata_writer.clone(),
crate::stage_execution::StageExecutionTracker::default(),
);
let conclusion = Conclusion {
timestamp: chrono::Utc::now(),

38
lib/components/fabro-workflow/src/lifecycle/mod.rs
View file

@ -44,6 +44,7 @@ use crate::run_options::RunOptions;
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::RunLocations;
use crate::stage_execution::StageExecutionTracker;
type WfRunState = ExecutionState<Option<BilledModelUsage>>;
type WfNodeResult = NodeResult<Option<BilledModelUsage>>;
@ -70,6 +71,8 @@ pub(crate) struct WorkflowLifecycle {
/// True when constructed with a checkpoint; cleared after first
/// on_run_start. Gates context seeding on initial resume.
is_initial_resume: AtomicBool,
/// Run-scoped stage execution allocator shared with `RunServices`.
stage_executions: StageExecutionTracker,
// Config needed for context seeding
graph: Arc<GvGraph>,
run_id: RunId,
@ -97,6 +100,7 @@ impl WorkflowLifecycle {
is_resume: bool,
on_node: crate::OnNodeCallback,
run_control: Option<Arc<RunControlState>>,
stage_executions: StageExecutionTracker,
) -> Self {
let restarted_from: Arc<Mutex<Option<(String, String)>>> = Arc::new(Mutex::new(None));
let loop_restart_signature_limit = graph.loop_restart_signature_limit();
@ -133,6 +137,7 @@ impl WorkflowLifecycle {
goal: (!graph.goal().is_empty()).then(|| graph.goal().to_string()),
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
circuit_breaker: Arc::clone(&circuit_breaker),
stage_executions: stage_executions.clone(),
};
let hook = HookLifecycle {
@ -164,6 +169,7 @@ impl WorkflowLifecycle {
start_node_id,
checkpoint_git_result: Arc::clone(&checkpoint_git_result),
last_git_sha,
stage_executions: stage_executions.clone(),
};
let artifact = ArtifactLifecycle::new(
@ -173,6 +179,7 @@ impl WorkflowLifecycle {
run_options.run_id,
run_options.artifact_globs(),
artifact_sink,
stage_executions.clone(),
);
Self {
@ -189,6 +196,7 @@ impl WorkflowLifecycle {
restarted_from,
checkpoint_git_result,
is_initial_resume: AtomicBool::new(is_resume),
stage_executions,
graph,
run_id: run_options.run_id,
sandbox_work_dir: run_branch_sandbox_work_dir,
@ -275,6 +283,15 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
if let Some(on_node) = &self.on_node {
on_node(node.id());
}
// Node boundary: clear the prior execution scope so the next
// observable attempt reserves a fresh ordinal. No reservation happens
// here — a hook block or process exit before any stage-scoped event
// must not consume an ordinal.
self.stage_executions.begin_node(node.id());
state.context.set(
context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL,
serde_json::Value::Null,
);
self.fidelity.before_node(node, state).await
}
@ -288,6 +305,16 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
NodeDecision::Continue => {}
decision => return Ok(decision),
}
// Reserve the stage execution once per handler invocation: the first
// attempt allocates the ordinal and automatic retries reuse it.
let node_id = ctx.node.id();
let execution = self
.stage_executions
.ensure(node_id, event::stage_visit(state, node_id));
state.context.set(
context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL,
serde_json::json!(execution.ordinal),
);
// Event emission
self.event.before_attempt(ctx, state).await?;
// Record epoch AFTER hook+event (engine.rs:968→1006)
@ -410,6 +437,17 @@ impl RunLifecycle<WorkflowGraph> for WorkflowLifecycle {
next_node_id: Option<&str>,
state: &WfRunState,
) -> CoreResult<()> {
// A StageStart hook can skip before any attempt reserved an execution
// scope. Ensure one exists so Git metadata-snapshot events and the
// `checkpoint.completed` envelope attach to a concrete execution;
// an existing reservation from the attempt path is reused as-is.
let execution = self
.stage_executions
.ensure(node.id(), event::stage_visit(state, node.id()));
state.context.set(
context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL,
serde_json::json!(execution.ordinal),
);
self.git
.on_checkpoint(node, result, next_node_id, state)
.await?;

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

@ -273,6 +273,8 @@ fn checkpoint_completed_event(checkpoint: &Checkpoint) -> Event {
node_visits: checkpoint.node_visits.clone().into_iter().collect(),
diff: None,
diff_summary: None,
graph_visit: None,
resumed_from_stage_id: None,
}
}
@ -428,6 +430,8 @@ mod tests {
.unwrap();
event::append_event(&source, &source_run_id, &Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
status: "succeeded".to_string(),
current_node: "work".to_string(),

20
lib/components/fabro-workflow/src/operations/resume.rs
View file

@ -5,6 +5,7 @@ use crate::error::Error;
use crate::event::{Event, append_event_to_sink};
use crate::outcome::StageOutcome;
use crate::run_status::RunStatus;
use crate::stage_execution::StageExecutionSeed;
/// Resume a workflow run from its checkpoint. Errors if no checkpoint is found.
pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started, Error> {
@ -32,10 +33,15 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
}
}
let checkpoint = state
.current_checkpoint()
.cloned()
let checkpoint_record = state
.checkpoints
.last()
.ok_or_else(|| Error::Precondition("no checkpoint to resume from".to_string()))?;
let checkpoint = checkpoint_record.checkpoint.clone();
// Seed the stage execution allocator from the projection so a node whose
// in-flight execution was cancelled or lost gets the next unused ordinal,
// and link it to the latest execution observed after this checkpoint.
let stage_executions = StageExecutionSeed::from_projection(&state, checkpoint_record.seq);
let definition_blob = state.spec.definition_blob;
cleanup_resume_artifacts(run_dir);
@ -47,7 +53,13 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
.await
.map_err(|err| Error::engine(err.to_string()))?;
Box::pin(execute_persisted_run(run_dir, Some(checkpoint), services)).await
Box::pin(execute_persisted_run(
run_dir,
Some(checkpoint),
stage_executions,
services,
))
.await
}
fn cleanup_resume_artifacts(run_dir: &Path) {

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

@ -299,6 +299,8 @@ mod tests {
.await
.unwrap();
event::append_event(&source_store, &source_run_id, &Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "work".to_string(),
status: "succeeded".to_string(),
current_node: "work".to_string(),

20
lib/components/fabro-workflow/src/operations/start.rs
View file

@ -48,6 +48,7 @@ use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions, Set
use crate::run_status::{FailureReason, RunStatus};
use crate::runtime_store::RunStoreHandle;
use crate::services::FabroRunToolServices;
use crate::stage_execution::StageExecutionSeed;
use crate::steering_hub::SteeringHub;
#[cfg(feature = "test-support")]
use crate::test_support as workflow_test_support;
@ -169,12 +170,19 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
.map_err(|err| Error::engine(err.to_string()))?;
}
Box::pin(execute_persisted_run(run_dir, None, services)).await
Box::pin(execute_persisted_run(
run_dir,
None,
StageExecutionSeed::default(),
services,
))
.await
}
pub(super) async fn execute_persisted_run(
run_dir: &Path,
checkpoint: Option<Checkpoint>,
stage_executions: StageExecutionSeed,
services: StartServices,
) -> Result<Started, Error> {
let cancel_token = services.cancel_token.clone();
@ -261,7 +269,7 @@ pub(super) async fn execute_persisted_run(
cancel_token,
);
let run_start = Instant::now();
let started = Box::pin(session.run(persisted, checkpoint)).await;
let started = Box::pin(session.run(persisted, checkpoint, stage_executions)).await;
match started {
Ok(started) => {
@ -797,6 +805,7 @@ impl RunSession {
self,
persisted: Persisted,
checkpoint: Option<Checkpoint>,
stage_executions: StageExecutionSeed,
) -> Result<Started, Error> {
let on_node = self.on_node.clone();
@ -879,6 +888,7 @@ impl RunSession {
run_control: self.run_control,
checkpoint,
seed_context: self.seed_context,
stage_executions,
fabro_run_tools: self.fabro_run_tools,
};
let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?;
@ -2125,6 +2135,8 @@ reasoning = false
{
injected.store(true, Ordering::SeqCst);
emitter_for_injection.emit(&Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: "start".to_string(),
status: "succeeded".to_string(),
current_node: "start".to_string(),
@ -2508,6 +2520,8 @@ reasoning = false
&store.open_run(&fixtures::RUN_1).await.unwrap(),
&services.run_id,
&Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: checkpoint.current_node.clone(),
status: checkpoint
.node_outcomes
@ -2606,6 +2620,8 @@ reasoning = false
};
let run_store = store.open_run(&fixtures::RUN_1).await.unwrap();
crate::event::append_event(&run_store, &fixtures::RUN_1, &Event::CheckpointCompleted {
graph_visit: None,
resumed_from_stage_id: None,
node_id: checkpoint.current_node.clone(),
status: "succeeded".to_string(),
current_node: checkpoint.current_node.clone(),

1
lib/components/fabro-workflow/src/pipeline/execute.rs
View file

@ -91,6 +91,7 @@ pub async fn execute(init: Initialized) -> Executed {
checkpoint.is_some(),
on_node,
run_control,
engine.run.stage_executions.clone(),
);
if let Some(ref cp) = checkpoint {

183
lib/components/fabro-workflow/src/pipeline/execute/tests.rs
View file

@ -256,6 +256,7 @@ async fn execute_test_run_with_options(
let initialized = initialize(
persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value),
InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: run_store.into(),
dry_run: false,
emitter: emitter.clone(),
@ -316,6 +317,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
let initialized = initialize(
persisted_workflow(graph, source, &run_dir, test_run_id("run-test")),
InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: run_store.into(),
dry_run: false,
emitter: test_emitter_arc("run-test"),
@ -375,6 +377,186 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
);
}
#[tokio::test]
async fn resumed_in_flight_node_starts_a_new_stage_execution() {
let temp = tempfile::tempdir().unwrap();
let run_dir = temp.path().join("run");
std::fs::create_dir_all(&run_dir).unwrap();
// start -> work -> exit; `work` resolves to the default (dry-run) handler.
let mut graph = Graph::new("resume_identity");
let mut start = Node::new("start");
start.attrs.insert(
"shape".to_string(),
AttrValue::String("Mdiamond".to_string()),
);
graph.nodes.insert("start".to_string(), start);
graph.nodes.insert("work".to_string(), Node::new("work"));
let mut exit = Node::new("exit");
exit.attrs.insert(
"shape".to_string(),
AttrValue::String("Msquare".to_string()),
);
graph.nodes.insert("exit".to_string(), exit);
graph.edges.push(Edge::new("start", "work"));
graph.edges.push(Edge::new("work", "exit"));
let run_options = test_run_options(&run_dir, "resume-identity");
let run_id = run_options.run_id;
let run_store = test_run_store(&run_id).await;
seed_created_and_starting(&run_store, &run_options, &graph).await;
// Resume reconnects to the previously recorded sandbox.
append_event(&run_store, &run_id, &Event::SandboxInitialized {
working_directory: std::env::current_dir().unwrap().display().to_string(),
provider: fabro_types::SandboxProviderKind::Local,
id: "local".to_string(),
image: None,
snapshot: None,
repo_cloned: None,
clone_origin_url: None,
clone_branch: None,
workspace_root: None,
repos_root: None,
primary_repo_path: None,
primary_repo_link: None,
})
.await
.unwrap();
let emitter = test_emitter_arc("resume-identity");
let events: Arc<std::sync::Mutex<Vec<fabro_types::RunEvent>>> = Arc::default();
{
let events = Arc::clone(&events);
emitter.on_event(move |event| {
events
.lock()
.expect("event capture mutex should not be poisoned")
.push(event.clone());
});
}
// Simulate resuming after `work@1` was cancelled mid-flight: the selected
// checkpoint predates `work`, while the allocator seed carries the
// projection-observed high-water mark and provenance link.
let checkpoint = crate::records::Checkpoint {
timestamp: chrono::Utc::now(),
current_node: "start".to_string(),
completed_nodes: vec!["start".to_string()],
node_retries: HashMap::new(),
context_values: HashMap::new(),
node_outcomes: HashMap::new(),
next_node_id: Some("work".to_string()),
git_commit_sha: None,
loop_failure_signatures: HashMap::new(),
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("start".to_string(), 1usize)]),
};
let seed = crate::stage_execution::StageExecutionSeed {
high_water: HashMap::from([("work".to_string(), 1)]),
resumed_from: HashMap::from([("work".to_string(), fabro_types::StageId::new("work", 1))]),
};
let initialized = initialize(
persisted_workflow(graph, String::new(), &run_dir, run_id),
InitOptions {
stage_executions: seed,
run_store: run_store.into(),
dry_run: false,
emitter: emitter.clone(),
sandbox: SandboxSpec::Local {
working_directory: std::env::current_dir().unwrap(),
},
llm: LlmSpec {
model: "test-model".to_string(),
provider_id: fabro_model::ProviderId::anthropic(),
fallback_chain: Vec::new(),
mcp_servers: Vec::new(),
model_controls: RunModelControls::default(),
dry_run: true,
},
interviewer: Arc::new(AutoApproveInterviewer::engine()),
steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())),
catalog: test_catalog(),
lifecycle: LifecycleOptions {
setup_commands: vec![],
setup_command_timeout_ms: 1_000,
},
run_options,
workflow_path: None,
workflow_bundle: None,
hooks: HookSettings { hooks: vec![] },
sandbox_env: SandboxEnvSpec {
toml_env: HashMap::new(),
github_permissions: None,
origin_url: None,
},
vault: None,
git: None,
run_control: None,
registry_override: Some(Arc::new(make_registry())),
artifact_sink: None,
checkpoint: Some(checkpoint),
seed_context: None,
fabro_run_tools: None,
},
)
.await
.unwrap();
let executed = execute(initialized).await;
assert_eq!(executed.outcome.unwrap().status, StageOutcome::Succeeded);
let events = events
.lock()
.expect("event capture mutex should not be poisoned");
let work_started = events
.iter()
.find(|event| {
matches!(event.body, fabro_types::EventBody::StageStarted(_))
&& event.node_id.as_deref() == Some("work")
})
.expect("resumed run should emit stage.started for work");
// The reexecution owns a fresh StageId while the graph visit stays at 1.
assert_eq!(
work_started.stage_id,
Some(fabro_types::StageId::new("work", 2))
);
let fabro_types::EventBody::StageStarted(props) = &work_started.body else {
panic!("expected stage.started body");
};
assert_eq!(props.graph_visit, Some(1));
assert_eq!(
props.resumed_from_stage_id,
Some(fabro_types::StageId::new("work", 1))
);
// Every later stage-scoped event from this invocation carries the same
// execution id, including the checkpoint envelope.
let work_checkpoint = events
.iter()
.find(|event| {
matches!(event.body, fabro_types::EventBody::CheckpointCompleted(_))
&& event.node_id.as_deref() == Some("work")
})
.expect("resumed run should checkpoint work");
assert_eq!(
work_checkpoint.stage_id,
Some(fabro_types::StageId::new("work", 2))
);
// A node without a prior observable execution starts at ordinal 1.
let exit_started = events
.iter()
.find(|event| {
matches!(event.body, fabro_types::EventBody::StageStarted(_))
&& event.node_id.as_deref() == Some("exit")
})
.expect("terminal node should emit its synthetic stage.started");
assert_eq!(
exit_started.stage_id,
Some(fabro_types::StageId::new("exit", 1))
);
}
async fn run_with_lifecycle(
registry: HandlerRegistry,
emitter: Arc<Emitter>,
@ -391,6 +573,7 @@ async fn run_with_lifecycle(
let initialized = initialize(
persisted_workflow(graph.clone(), String::new(), &run_dir, run_id),
InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: run_store.into(),
dry_run: false,
emitter: emitter.clone(),

2
lib/components/fabro-workflow/src/pipeline/finalize.rs
View file

@ -1025,6 +1025,7 @@ mod tests {
Arc::new(SandboxGitRuntime::new()),
metadata_runtime,
metadata_writer,
crate::stage_execution::StageExecutionTracker::default(),
)
}
@ -1057,6 +1058,7 @@ mod tests {
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
crate::stage_execution::StageExecutionTracker::default(),
);
let executed = test_executed(
Graph::new("test"),

7
lib/components/fabro-workflow/src/pipeline/initialize.rs
View file

@ -33,6 +33,7 @@ use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{
EngineServices, FabroRunToolServices, RunLocations, RunServices, WorkflowToolEnvProvider,
};
use crate::stage_execution::StageExecutionTracker;
use crate::steering_hub::SteeringHub;
type BuiltSandboxEnv = (HashMap<String, String>, Option<Arc<GitHubTokenSource>>);
@ -619,6 +620,7 @@ pub async fn initialize(
sandbox_git,
metadata_runtime,
metadata_writer,
StageExecutionTracker::seeded(options.stage_executions),
);
let engine = Arc::new(EngineServices {
run: Arc::clone(&run_services),
@ -825,6 +827,7 @@ mod tests {
});
let result = initialize(persisted, InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: {
let store = memory_store();
let inner = store.create_run(&test_run_id()).await.unwrap();
@ -906,6 +909,7 @@ mod tests {
let emitter = Arc::new(crate::event::Emitter::new(test_run_id()));
let initialized = initialize(persisted, InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: {
let store = memory_store();
let inner = store.create_run(&test_run_id()).await.unwrap();
@ -1131,6 +1135,7 @@ mod tests {
let store = memory_store();
let run_store = store.create_run(&test_run_id()).await.unwrap();
let initialized = initialize(test_persisted(graph, source, &run_dir), InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: run_store.into(),
dry_run: false,
emitter: emitter.clone(),
@ -1226,6 +1231,7 @@ mod tests {
store_logger.register(&emitter);
let initialized = initialize(persisted, InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: run_store.into(),
dry_run: false,
emitter: emitter.clone(),
@ -1364,6 +1370,7 @@ mod tests {
let emitter = Arc::new(crate::event::Emitter::new(test_run_id()));
let result = initialize(persisted, InitOptions {
stage_executions: crate::stage_execution::StageExecutionSeed::default(),
run_store: {
let store = memory_store();
let inner = store.create_run(&test_run_id()).await.unwrap();

5
lib/components/fabro-workflow/src/pipeline/types.rs
View file

@ -26,6 +26,7 @@ use crate::run_control::RunControlState;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions};
use crate::runtime_store::RunStoreHandle;
use crate::services::{EngineServices, FabroRunToolServices, RunServices};
use crate::stage_execution::StageExecutionSeed;
use crate::steering_hub::SteeringHub;
use crate::transforms::{RenderMode, Transform};
use crate::workflow_bundle::WorkflowBundle;
@ -270,6 +271,10 @@ pub struct InitOptions {
pub run_control: Option<Arc<RunControlState>>,
pub checkpoint: Option<Checkpoint>,
pub seed_context: Option<Context>,
/// Allocator seed for stage execution ordinals. Empty for a fresh run;
/// resume passes projection-derived high-water marks and provenance so a
/// reexecuted in-flight node gets a new `StageId` ordinal.
pub stage_executions: StageExecutionSeed,
pub fabro_run_tools: Option<FabroRunToolServices>,
}

7
lib/components/fabro-workflow/src/services.rs
View file

@ -22,6 +22,7 @@ use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git::GitState;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::stage_execution::StageExecutionTracker;
use crate::workflow_bundle::WorkflowBundle;
#[derive(Clone, Debug, PartialEq, Eq)]
@ -107,6 +108,9 @@ pub struct RunServices {
pub(crate) metadata_runtime: Arc<RunMetadataRuntime>,
pub(crate) metadata_writer: Option<RunMetadataWriterHandle>,
pub(crate) interview_blocker: Arc<RunInterviewBlocker>,
/// Run-scoped stage execution allocator, shared between the core
/// lifecycle and direct-dispatch handlers such as parallel branches.
pub(crate) stage_executions: StageExecutionTracker,
}
impl RunServices {
@ -125,6 +129,7 @@ impl RunServices {
sandbox_git: Arc<SandboxGitRuntime>,
metadata_runtime: Arc<RunMetadataRuntime>,
metadata_writer: Option<RunMetadataWriterHandle>,
stage_executions: StageExecutionTracker,
) -> Arc<Self> {
Arc::new(Self {
run_store,
@ -141,6 +146,7 @@ impl RunServices {
metadata_runtime,
metadata_writer,
interview_blocker: Arc::new(RunInterviewBlocker::new()),
stage_executions,
})
}
@ -340,6 +346,7 @@ impl EngineServices {
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
StageExecutionTracker::default(),
),
registry: Arc::new(HandlerRegistry::new(Box::new(start::StartHandler))),
interviewer: Arc::new(fabro_interview::AutoApproveInterviewer::engine()),

291
lib/components/fabro-workflow/src/stage_execution.rs Normal file
View file

@ -0,0 +1,291 @@
//! Run-scoped stage execution identity.
//!
//! A *stage execution* is one top-level handler invocation of a node that
//! became observable within a run. Its 1-based ordinal is the numeric
//! component of the external `StageId` (`node_id@N`). The ordinal is distinct
//! from the *graph visit* (how many times workflow control entered the node,
//! which drives `max_visits` and checkpoints) and from the *handler attempt*
//! (automatic retries inside one execution).
//!
//! The tracker is deliberately not checkpointed: its durable source of truth
//! is the append-only stage event history. On resume it is seeded from the
//! run projection's per-node maxima, so a reexecuted in-flight node allocates
//! the next unused ordinal instead of mutating the cancelled execution.
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use fabro_types::{RunProjection, StageId};
/// One reserved stage execution: the identity of a single resumable handler
/// invocation of a node.
#[derive(Clone, Debug, PartialEq, Eq)]
pub(crate) struct StageExecution {
/// 1-based execution ordinal; becomes the `@N` in the external `StageId`.
pub ordinal: u32,
/// Graph visit that produced this execution.
pub graph_visit: u32,
/// Prior execution this one resumes from, when the node had an observable
/// post-checkpoint execution before the run was interrupted.
pub resumed_from: Option<StageId>,
}
/// Seed data for the [`StageExecutionTracker`], derived from the run
/// projection when a run is resumed. A fresh run uses the default (empty)
/// seed; new run IDs own a new ordinal sequence.
#[derive(Clone, Debug, Default)]
pub struct StageExecutionSeed {
/// Highest execution ordinal already observable per node.
pub high_water: HashMap<String, u32>,
/// Latest post-checkpoint execution per node; the next reservation for
/// that node links back to it via `resumed_from_stage_id`.
pub resumed_from: HashMap<String, StageId>,
}
impl StageExecutionSeed {
/// Build the seed from the run projection at resume time.
///
/// `checkpoint_seq` is the event sequence number of the selected
/// checkpoint. Only stages that first became observable *after* that
/// checkpoint are eligible provenance targets: an older execution with the
/// same node ID completed before the checkpoint and is not what the
/// resumed invocation continues from.
#[must_use]
pub fn from_projection(projection: &RunProjection, checkpoint_seq: u32) -> Self {
let mut high_water: HashMap<String, u32> = HashMap::new();
let mut resumed_from: HashMap<String, StageId> = HashMap::new();
// `iter_stages` yields chronological `first_event_seq` order, so a
// later insert per node retains the latest post-checkpoint execution.
for (stage_id, stage) in projection.iter_stages() {
let node_id = stage_id.node_id();
let entry = high_water.entry(node_id.to_string()).or_default();
*entry = (*entry).max(stage_id.visit());
if stage.first_event_seq.get() > checkpoint_seq {
resumed_from.insert(node_id.to_string(), stage_id.clone());
}
}
Self {
high_water,
resumed_from,
}
}
}
#[derive(Debug, Default)]
struct TrackerState {
/// Highest ordinal observed or reserved per node.
high_water: HashMap<String, u32>,
/// Pending provenance links, consumed by the first reservation per node.
resumed_from: HashMap<String, StageId>,
/// Active execution scope per node. Cleared at the node boundary and
/// replaced by the next reservation.
active: HashMap<String, StageExecution>,
}
/// Cloneable, run-scoped allocator for stage execution ordinals. Clones share
/// one synchronized state so the core lifecycle and direct-dispatch handlers
/// (parallel branches) allocate from the same sequence.
#[derive(Clone, Debug, Default)]
pub(crate) struct StageExecutionTracker {
state: Arc<Mutex<TrackerState>>,
}
impl StageExecutionTracker {
#[must_use]
pub(crate) fn seeded(seed: StageExecutionSeed) -> Self {
Self {
state: Arc::new(Mutex::new(TrackerState {
high_water: seed.high_water,
resumed_from: seed.resumed_from,
active: HashMap::new(),
})),
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, TrackerState> {
self.state
.lock()
.expect("stage execution tracker mutex is never poisoned: no code panics while holding this lock")
}
/// Clear the node's prior execution scope at the node boundary. The next
/// `reserve`/`ensure` call allocates a fresh ordinal; a reservation is not
/// made here so that a StageStart hook block or process exit before any
/// stage-scoped event leaves no phantom execution.
pub(crate) fn begin_node(&self, node_id: &str) {
self.lock().active.remove(node_id);
}
/// The node's active execution scope, if one has been reserved since the
/// last node boundary.
pub(crate) fn active(&self, node_id: &str) -> Option<StageExecution> {
self.lock().active.get(node_id).cloned()
}
/// Allocate the next execution ordinal for the node and make it the active
/// scope. Consumes the node's pending provenance link, if any.
pub(crate) fn reserve(&self, node_id: &str, graph_visit: u32) -> StageExecution {
let mut state = self.lock();
let entry = state.high_water.entry(node_id.to_string()).or_default();
*entry = entry.saturating_add(1);
let ordinal = *entry;
let resumed_from = state.resumed_from.remove(node_id);
let execution = StageExecution {
ordinal,
graph_visit,
resumed_from,
};
state.active.insert(node_id.to_string(), execution.clone());
execution
}
/// The active scope for the node, reserving one only when none exists.
/// Later attempts within one execution and checkpoint pre-steps reuse the
/// first attempt's reservation.
pub(crate) fn ensure(&self, node_id: &str, graph_visit: u32) -> StageExecution {
if let Some(execution) = self.active(node_id) {
return execution;
}
self.reserve(node_id, graph_visit)
}
}
#[cfg(test)]
mod tests {
use std::num::NonZeroU32;
use chrono::Utc;
use fabro_types::{Graph, RunId, RunSpec, StageId, WorkflowSettings, test_support};
use super::*;
fn projection_with_stages(stages: &[(&str, u32, u32)]) -> RunProjection {
let spec = RunSpec {
run_id: RunId::new(),
settings: WorkflowSettings::default(),
graph: Graph::new("test"),
graph_source: None,
workflow_slug: None,
automation: None,
source_directory: None,
labels: std::collections::HashMap::new(),
provenance: test_support::test_run_provenance(),
manifest_blob: None,
definition_blob: None,
git: None,
fork_source_ref: None,
};
let mut projection = RunProjection::new(String::new(), spec, Utc::now());
for (node_id, visit, seq) in stages {
projection.stage_entry(
node_id,
*visit,
NonZeroU32::new(*seq).expect("test seq must be non-zero"),
);
}
projection
}
#[test]
fn reserve_starts_at_one_and_allocates_monotonically_per_node() {
let tracker = StageExecutionTracker::default();
assert_eq!(tracker.reserve("work", 1).ordinal, 1);
tracker.begin_node("work");
assert_eq!(tracker.reserve("work", 2).ordinal, 2);
assert_eq!(tracker.reserve("other", 1).ordinal, 1);
}
#[test]
fn seeds_from_projection_maxima() {
let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5), ("plan", 1, 3)]);
let seed = StageExecutionSeed::from_projection(&projection, 0);
let tracker = StageExecutionTracker::seeded(seed);
assert_eq!(tracker.reserve("work", 1).ordinal, 3);
assert_eq!(tracker.reserve("plan", 1).ordinal, 2);
assert_eq!(tracker.reserve("new", 1).ordinal, 1);
}
#[test]
fn graph_visit_and_ordinal_can_diverge() {
let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5)]);
let seed = StageExecutionSeed::from_projection(&projection, 0);
let tracker = StageExecutionTracker::seeded(seed);
let execution = tracker.reserve("work", 2);
assert_eq!(execution.ordinal, 3);
assert_eq!(execution.graph_visit, 2);
}
#[test]
fn ensure_reuses_active_reservation_across_attempts() {
let tracker = StageExecutionTracker::default();
let first = tracker.ensure("work", 1);
let second = tracker.ensure("work", 1);
assert_eq!(first, second);
assert_eq!(second.ordinal, 1);
tracker.begin_node("work");
assert_eq!(tracker.ensure("work", 2).ordinal, 2);
}
#[test]
fn begin_node_clears_only_that_node() {
let tracker = StageExecutionTracker::default();
tracker.reserve("work", 1);
tracker.reserve("verify", 1);
tracker.begin_node("work");
assert_eq!(tracker.active("work"), None);
assert_eq!(tracker.active("verify").map(|e| e.ordinal), Some(1));
}
#[test]
fn provenance_only_selects_stages_after_the_checkpoint() {
let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 8), ("plan", 1, 3)]);
let seed = StageExecutionSeed::from_projection(&projection, 5);
assert_eq!(
seed.resumed_from.get("work"),
Some(&StageId::new("work", 2))
);
assert_eq!(seed.resumed_from.get("plan"), None);
}
#[test]
fn first_reservation_consumes_provenance() {
let projection = projection_with_stages(&[("work", 1, 6)]);
let seed = StageExecutionSeed::from_projection(&projection, 5);
let tracker = StageExecutionTracker::seeded(seed);
let first = tracker.reserve("work", 1);
assert_eq!(first.ordinal, 2);
assert_eq!(first.resumed_from, Some(StageId::new("work", 1)));
tracker.begin_node("work");
let second = tracker.reserve("work", 2);
assert_eq!(second.ordinal, 3);
assert_eq!(second.resumed_from, None);
}
#[tokio::test(flavor = "multi_thread")]
async fn concurrent_reservations_stay_unique_per_node() {
let tracker = StageExecutionTracker::default();
let handles: Vec<_> = (0..8)
.map(|_| {
let tracker = tracker.clone();
tokio::spawn(async move { tracker.reserve("branch", 1).ordinal })
})
.collect();
let mut ordinals = Vec::new();
for handle in handles {
ordinals.push(handle.await.expect("reservation task panicked"));
}
ordinals.sort_unstable();
assert_eq!(ordinals, (1..=8).collect::<Vec<_>>());
}
}

40
lib/components/fabro-workflow/src/stage_scope.rs
View file

@ -1,11 +1,28 @@
use fabro_types::{ParallelBranchId, StageId};
use crate::context::{Context as WfContext, WorkflowContext};
use crate::context::{Context as WfContext, WorkflowContext, keys};
use crate::run_dir::visit_from_context;
/// Read the stage execution ordinal seeded by the workflow lifecycle (or a
/// parallel branch dispatch). `None` when the current node has not reserved an
/// execution yet — direct-handler call sites (tests, etc.) that skip the full
/// lifecycle fall back to the graph visit, which equals the ordinal for a
/// first execution.
fn execution_ordinal_from_context(context: &WfContext) -> Option<u32> {
context
.get(keys::INTERNAL_STAGE_EXECUTION_ORDINAL)
.and_then(|value| value.as_u64())
.map(|ordinal| u32::try_from(ordinal).unwrap_or(u32::MAX))
}
/// Stage-level scope threaded through event emission to populate
/// `stage_id` / `parallel_group_id` / `parallel_branch_id` on events
/// that happen inside a concrete stage execution.
///
/// `visit` is the 1-based stage execution ordinal — the numeric component of
/// the external `StageId`. It matches the graph visit for a first execution
/// and diverges when a cancelled or crashed invocation is reexecuted after
/// resume.
#[derive(Clone, Debug)]
pub struct StageScope {
pub node_id: String,
@ -15,13 +32,15 @@ pub struct StageScope {
}
impl StageScope {
/// Build a scope from the given node id, sourcing visit count and parallel
/// ids from the current context.
/// Build a scope from the given node id, sourcing the execution ordinal
/// and parallel ids from the current context.
pub fn from_context(context: &WfContext, node_id: impl Into<String>) -> Self {
let visit = execution_ordinal_from_context(context)
.unwrap_or_else(|| u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX));
Self {
node_id: node_id.into(),
visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX),
parallel_group_id: context.parallel_group_id(),
node_id: node_id.into(),
visit,
parallel_group_id: context.parallel_group_id(),
parallel_branch_id: context.parallel_branch_id(),
}
}
@ -40,11 +59,10 @@ impl StageScope {
/// handler (`ParallelBranchStarted`, `ParallelBranchCompleted`, and the
/// pre-dispatch `GitCommit` for the branch worktree).
///
/// `target_visit` is the visit count of `target_node_id` for this
/// particular branch dispatch. The parallel handler currently passes
/// `1` because branches haven't been re-entered yet at the point of
/// scope construction; a future change that loops a parallel node
/// must pass the actual visit so envelope `stage_id`s stay accurate.
/// `target_visit` is the branch target's stage execution ordinal for this
/// particular dispatch, reserved through the run's shared
/// `StageExecutionTracker` so a resumed fan-out gets a fresh child
/// identity instead of overwriting the cancelled attempt's.
#[must_use]
pub fn for_parallel_branch(
target_node_id: impl Into<String>,

2
lib/components/fabro-workflow/src/test_support.rs
View file

@ -27,6 +27,7 @@ use crate::run_metadata::RunMetadataRuntime;
use crate::run_options::RunOptions;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::services::{EngineServices, RunLocations, RunServices};
use crate::stage_execution::StageExecutionTracker;
#[cfg(feature = "test-support")]
pub(crate) fn test_configured_provider_ids(
@ -253,6 +254,7 @@ async fn initialized(
Arc::new(SandboxGitRuntime::new()),
Arc::new(RunMetadataRuntime::new()),
None,
StageExecutionTracker::default(),
),
registry: Arc::new(registry),
interviewer: Arc::new(AutoApproveInterviewer::engine()),

12
lib/foundation/fabro-types/src/run_event/misc.rs
View file

@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use super::ExecOutputTail;
use crate::{CommandTermination, PullRequestLink};
use crate::{CommandTermination, PullRequestLink, StageId};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct InterviewOption {
@ -23,7 +23,15 @@ pub struct ParallelStartedProps {
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ParallelBranchStartedProps {
pub index: usize,
pub index: usize,
/// Graph visit of the branch target for this dispatch. The envelope
/// `stage_id` ordinal counts executions, so a resumed fan-out's branches
/// keep visit metadata even though their ordinals advanced.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub graph_visit: Option<u32>,
/// Prior branch execution this one resumes from.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resumed_from_stage_id: Option<StageId>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

27
lib/foundation/fabro-types/src/run_event/stage.rs
View file

@ -5,14 +5,25 @@ use serde::{Deserialize, Serialize};
use serde_json::Value;
use super::ExecOutputTail;
use crate::{BilledModelUsage, DiffSummary, FailureDetail, Outcome, StageOutcome, StageTiming};
use crate::{
BilledModelUsage, DiffSummary, FailureDetail, Outcome, StageId, StageOutcome, StageTiming,
};
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct StageStartedProps {
pub index: usize,
pub handler_type: String,
pub attempt: usize,
pub max_attempts: usize,
pub index: usize,
pub handler_type: String,
pub attempt: usize,
pub max_attempts: usize,
/// Graph visit that produced this stage execution. The envelope
/// `stage_id` ordinal counts executions, which diverges from the graph
/// visit when a cancelled or crashed invocation is reexecuted after
/// resume. Absent on events written before stage execution identity.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub graph_visit: Option<u32>,
/// Prior execution this one resumes from.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resumed_from_stage_id: Option<StageId>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
@ -123,6 +134,12 @@ pub struct CheckpointCompletedProps {
pub diff: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub diff_summary: Option<DiffSummary>,
/// Graph visit of the checkpointed stage execution; used when this
/// checkpoint is the event that first materializes a skipped stage.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub graph_visit: Option<u32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resumed_from_stage_id: Option<StageId>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

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

@ -320,59 +320,71 @@ impl StageContextWindow {
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StageProjection {
pub first_event_seq: NonZeroU32,
pub prompt: Option<String>,
pub response: Option<String>,
pub completion: Option<StageCompletion>,
pub provider_used: Option<StageModelUsage>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub output: Option<String>,
pub first_event_seq: NonZeroU32,
pub prompt: Option<String>,
pub response: Option<String>,
pub completion: Option<StageCompletion>,
pub provider_used: Option<StageModelUsage>,
pub diff: Option<String>,
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub output: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub output_bytes: Option<u64>,
pub output_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub live_streaming: Option<bool>,
pub live_streaming: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub termination: Option<crate::CommandTermination>,
pub termination: Option<crate::CommandTermination>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub started_at: Option<DateTime<Utc>>,
pub started_at: Option<DateTime<Utc>>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub handler: Option<StageHandler>,
/// Per-attempt timing breakdown for the latest terminal attempt.
pub handler: Option<StageHandler>,
/// Graph visit that produced this stage execution. The `StageId` ordinal
/// counts executions, which diverges from the graph visit when a
/// cancelled or crashed invocation is reexecuted after resume. Absent on
/// projections built from events written before stage execution identity.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub graph_visit: Option<u32>,
/// Prior execution this one resumes from.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub resumed_from_stage_id: Option<StageId>,
/// Timing breakdown for this stage execution's latest terminal attempt.
/// One projection represents one execution, which may contain multiple
/// automatic attempts; earlier executions of the same node keep their own
/// immutable projections under their own `StageId`s.
///
/// `None` for stages still in flight (`started_at` is set but no terminal
/// event has been observed yet). For live wall-time ticking, the UI uses
/// `started_at`; once terminal this carries the finalized breakdown.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timing: Option<StageTiming>,
pub timing: Option<StageTiming>,
#[serde(default)]
pub usage: BilledTokenCounts,
pub usage: BilledTokenCounts,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub model: Option<ModelRef>,
pub model: Option<ModelRef>,
/// Todo/task list owned by the stage's root agent session.
///
/// OpenAI child sessions own separate per-session plans and do not appear
/// here. Anthropic task lists are root-scoped and shared with child
/// sessions, so child mutations of that shared list do appear here.
#[serde(default, rename = "todos", skip_serializing_if = "Option::is_none")]
pub root_agent_todos: Option<TodoListProjection>,
pub root_agent_todos: Option<TodoListProjection>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub subagents: Vec<SubAgentProjection>,
pub subagents: Vec<SubAgentProjection>,
#[serde(default, skip_serializing_if = "SkillsProjection::is_empty")]
pub skills: SkillsProjection,
pub skills: SkillsProjection,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub permission_level: Option<PermissionLevel>,
pub permission_level: Option<PermissionLevel>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub agent_tools: Vec<AgentToolSummary>,
pub agent_tools: Vec<AgentToolSummary>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub mcp_servers: Vec<McpServerProjection>,
pub mcp_servers: Vec<McpServerProjection>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub context_window: Option<StageContextWindowProjection>,
pub context_window: Option<StageContextWindowProjection>,
#[serde(default)]
pub agent_control: AgentControlState,
pub state: StageState,
pub agent_control: AgentControlState,
pub state: StageState,
}
#[derive(
@ -486,6 +498,8 @@ impl StageProjection {
termination: None,
started_at: None,
handler: None,
graph_visit: None,
resumed_from_stage_id: None,
state: StageState::Running,
}
}
@ -519,14 +533,24 @@ impl StageProjection {
self.timing.map(|timing| timing.wall_time_ms)
}
/// Begin a new attempt (or visit) for this stage: clear every
/// Begin a new automatic attempt within this stage execution: clear every
/// per-attempt field so prior-attempt data does not leak, then record
/// `started_at` and `state = Running`. Preserves `first_event_seq`
/// (identity / sort key).
/// (identity / sort key) and the execution identity metadata
/// (`graph_visit`, `resumed_from_stage_id`).
///
/// One stage projection represents one execution; a reexecution after
/// cancel or crash recovery gets a new `StageId` and never flows through
/// here. Replays of legacy histories with duplicate `stage.started`
/// events for one `StageId` retain this last-attempt behavior.
pub fn begin_attempt(&mut self, started_at: DateTime<Utc>, handler: StageHandler) {
let graph_visit = self.graph_visit;
let resumed_from_stage_id = self.resumed_from_stage_id.take();
*self = Self::new(self.first_event_seq);
self.started_at = Some(started_at);
self.handler = Some(handler);
self.graph_visit = graph_visit;
self.resumed_from_stage_id = resumed_from_stage_id;
self.state = StageState::Running;
}
}

10
lib/packages/fabro-api-client/src/models/run-stage.ts generated
View file

@ -46,9 +46,17 @@ export interface RunStage {
*/
'node_id': string;
/**
* 1-based visit count; bumped each time the workflow re-enters this node.
* 1-based stage execution ordinal, the numeric component of `id`. It increments each time the node produces a new observable execution: graph re-entry (loops) and reexecution after cancel or crash recovery. Automatic in-place retries do not increment it.
*/
'visit': number;
/**
* 1-based count of how many times workflow control entered this node (drives `max_visits`). Differs from `visit` when a cancelled or crashed execution was reexecuted after resume. Absent for stages recorded before execution identity was tracked.
*/
'graph_visit'?: number | null;
/**
* StageId of the prior cancelled or interrupted execution this stage resumes from, when the run was resumed after that execution became observable.
*/
'resumed_from_stage_id'?: string | null;
'provider_used'?: StageModelUsage | null;
/**
* Wall-clock time the latest attempt of this stage started, if known.