scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-11 22:53:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge branch 'main' into feat/backward-event-pagination

Browse source
This commit is contained in:
Bryan Helmkamp 2026-07-24 10:14:57 -04:00 • committed by GitHub
commit c9900b2bfa
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
47 changed files with 2172 additions and 431 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: "review/security@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("review/security@1");
expect(text).toContain("Graph visit");
const json = JSON.stringify(tree.toJSON());
expect(json).toContain("/runs/run-1/stages/review%2Fsecurity%401");
});
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/${encodeURIComponent(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 post-checkpoint work was replayed after resume; null
* for stages recorded before execution identity was tracked.
*/
graphVisit: number | null;
/** StageId of the prior execution superseded by this resumed replay, 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,
@ -1738,6 +1738,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/${encodeURIComponent(selectedStage.resumedFromStageId)}`}
className="font-mono text-teal-500 hover:underline"
>
{selectedStage.resumedFromStageId}
</Link>
</p>
)}
<EventsToolbar
tab={effectiveTab}
renderer={renderer}

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

@ -12465,6 +12465,11 @@ components:
- running
- waiting_for_steer
StageId:
description: Canonical stage execution identifier in `node_id@visit` form.
type: string
example: verify@2
StageState:
description: Lifecycle projection state of a workflow stage.
type: string
@ -12506,9 +12511,7 @@ components:
- visit
properties:
id:
type: string
description: StageId in "node_id@visit" form, e.g. verify@2.
example: verify@2
$ref: "#/components/schemas/StageId"
name:
type: string
description: Human-readable stage name.
@ -12531,8 +12534,30 @@ 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 replay of post-checkpoint work after
resume. 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 post-checkpoint
execution is replayed after resume. Absent for stages recorded
before execution identity was tracked.
example: 1
resumed_from_stage_id:
oneOf:
- $ref: "#/components/schemas/StageId"
- type: "null"
description: >-
StageId of the prior post-checkpoint execution superseded by this
replay after the run was resumed.
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,
}
}
@ -591,10 +593,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);
@ -630,10 +634,12 @@ mod tests {
branch_count: 1,
});
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"];
@ -1245,10 +1251,12 @@ mod tests {
branch_count: 1,
});
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),
@ -1277,12 +1285,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,

3
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
@ -1257,6 +1258,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 +1286,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

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

@ -1099,7 +1099,6 @@ mod runs {
};
use super::ts;
use crate::server::run_stage_from_stage_id;
static DEMO_PRINCIPAL: LazyLock<Principal> = LazyLock::new(|| {
Principal::user(
@ -1124,6 +1123,29 @@ mod runs {
}
}
fn stage(
stage_id: &StageId,
name: &str,
status: StageState,
wall_time_ms: Option<u64>,
handler: StageHandler,
) -> RunStage {
RunStage {
id: stage_id.clone(),
name: name.to_owned(),
handler,
status,
wall_time_ms,
node_id: stage_id.node_id().to_owned(),
visit: std::num::NonZeroU32::new(stage_id.visit())
.expect("StageId stores a non-zero visit"),
provider_used: None,
started_at: None,
graph_visit: None,
resumed_from_stage_id: None,
}
}
fn demo_run_ids() -> &'static [RunId; 7] {
static IDS: OnceLock<[RunId; 7]> = OnceLock::new();
IDS.get_or_init(|| {
@ -1380,50 +1402,40 @@ mod runs {
pub(super) fn stages() -> Vec<RunStage> {
vec![
run_stage_from_stage_id(
stage(
&StageId::new("detect-drift", 1),
"Detect Drift",
StageState::Succeeded,
Some(72_000),
None,
StageHandler::Command,
None,
),
run_stage_from_stage_id(
stage(
&StageId::new("propose-changes", 1),
"Propose Changes",
StageState::Succeeded,
Some(154_000),
None,
StageHandler::Agent,
None,
),
run_stage_from_stage_id(
stage(
&StageId::new("review-changes", 1),
"Review Changes",
StageState::Succeeded,
Some(45_000),
None,
StageHandler::Agent,
None,
),
run_stage_from_stage_id(
stage(
&StageId::new("apply-changes", 1),
"Apply Changes",
StageState::Succeeded,
Some(118_000),
None,
StageHandler::Command,
None,
),
run_stage_from_stage_id(
stage(
&StageId::new("apply-changes", 2),
"Apply Changes",
StageState::Running,
None,
None,
StageHandler::Command,
None,
),
]
}

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

@ -99,7 +99,7 @@ use fabro_types::{
AgentBackend, AskFabro, AskFabroUnavailableReason, EventBody, InterviewQuestionRecord, PairId,
PairMessageId, PairTarget, PendingReason, Principal, PullRequestLink, QuestionType, RunBlobId,
RunControlAction, RunEvent, RunId, RunRunnableSource, SandboxProviderKind, ServerSettings,
SessionCapability, StageModelUsage,
SessionCapability,
};
use fabro_util::error::{
SharedError, collect_causes, render_compact_with_causes, render_with_causes,
@ -1332,29 +1332,6 @@ fn accumulate_billing_rollup(
}
}
pub(crate) fn run_stage_from_stage_id(
stage_id: &StageId,
name: impl Into<String>,
status: StageState,
wall_time_ms: Option<u64>,
started_at: Option<chrono::DateTime<chrono::Utc>>,
handler: StageHandler,
provider_used: Option<StageModelUsage>,
) -> RunStage {
RunStage {
id: stage_id.to_string(),
name: name.into(),
handler,
status,
wall_time_ms,
node_id: stage_id.node_id().to_string(),
visit: std::num::NonZeroU32::new(stage_id.visit())
.expect("StageId stores a non-zero visit"),
provider_used,
started_at,
}
}
impl AppState {
pub(crate) fn manifest_run_defaults(&self) -> Arc<RunLayer> {
Arc::clone(

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

@ -2,12 +2,14 @@ use std::collections::HashMap;
use std::sync::Arc;
use chrono::{DateTime, Utc};
use fabro_types::{RunProjection, StageHandler, StageProjection, StageState, StageTiming};
use fabro_types::{
Graph, RunProjection, StageHandler, StageId, StageProjection, StageState, StageTiming,
};
use super::super::{
AppState, BillingByModel, BillingStageRef, IntoResponse, Json, ListResponse, PaginationParams,
Path, Query, RequiredUser, Response, Router, RunBilling, RunBillingStage, RunBillingTotals,
RunId, State, StatusCode, get, parse_run_id_path, run_stage_from_stage_id,
RunId, RunStage, State, StatusCode, get, parse_run_id_path,
};
pub(super) fn routes() -> Router<Arc<AppState>> {
@ -16,6 +18,36 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/billing", get(get_run_billing))
}
fn run_stage_from_projection(
stage_id: &StageId,
stage: &StageProjection,
graph: &Graph,
now: DateTime<Utc>,
) -> RunStage {
let handler = stage.handler.unwrap_or_else(|| {
StageHandler::from_handler_type(
graph
.nodes
.get(stage_id.node_id())
.and_then(|node| node.handler_type()),
)
});
RunStage {
id: stage_id.clone(),
name: stage_id.node_id().to_owned(),
handler,
status: stage.effective_state(),
wall_time_ms: stage.live_wall_time_ms(now),
node_id: stage_id.node_id().to_owned(),
visit: std::num::NonZeroU32::new(stage_id.visit())
.expect("StageId stores a non-zero visit"),
provider_used: stage.provider_used.clone(),
started_at: stage.started_at,
graph_visit: stage.graph_visit.and_then(std::num::NonZeroU32::new),
resumed_from_stage_id: stage.resumed_from_stage_id.clone(),
}
}
async fn list_run_stages(
_auth: RequiredUser,
State(state): State<Arc<AppState>>,
@ -37,25 +69,7 @@ async fn list_run_stages(
let graph = projection.spec().graph();
let stages = projection
.iter_stages()
.map(|(stage_id, stage)| {
let handler = stage.handler.unwrap_or_else(|| {
StageHandler::from_handler_type(
graph
.nodes
.get(stage_id.node_id())
.and_then(|n| n.handler_type()),
)
});
run_stage_from_stage_id(
stage_id,
stage_id.node_id().to_string(),
stage.effective_state(),
stage.live_wall_time_ms(now),
stage.started_at,
handler,
stage.provider_used.clone(),
)
})
.map(|(stage_id, stage)| run_stage_from_projection(stage_id, stage, graph, now))
.collect::<Vec<_>>();
(StatusCode::OK, Json(ListResponse::new(stages))).into_response()

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,
}
}
@ -8169,12 +8301,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(),
@ -8236,12 +8370,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(),
@ -8300,12 +8436,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(),
@ -9491,12 +9629,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;
@ -11528,6 +11668,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(),

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

@ -227,35 +227,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.as_ref() {
// 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,
@ -340,15 +351,23 @@ impl RunProjectionReducer for RunProjection {
let Some(stage_id) = stored.stage_id.as_ref() else {
return Ok(());
};
let stage = self.stage_entry(
stage_id.node_id(),
stage_id.visit(),
first_event_seq(event.seq),
);
// 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 is_new = self.stage(stage_id).is_none();
let stage = stage_at_stored_stage_id(self, stage_id, event.seq);
stage.begin_attempt(
ts,
StageHandler::from_handler_type(Some(&props.handler_type)),
);
if is_new {
stage.graph_visit = props.graph_visit;
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,16 +556,26 @@ impl RunProjectionReducer for RunProjection {
};
stage.parallel_results = Some(props.results.clone());
}
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).
let is_new = stored
.stage_id
.as_ref()
.is_none_or(|stage_id| self.stage(stage_id).is_none());
let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else {
return Ok(());
};
if stage.started_at.is_none() {
stage.started_at = Some(ts);
}
if is_new {
stage.graph_visit = props.graph_visit;
stage
.resumed_from_stage_id
.clone_from(&props.resumed_from_stage_id);
}
stage.state = StageState::Running;
}
EventBody::ParallelBranchCompleted(props) => {
@ -947,6 +976,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 = stage_at_stored_stage_id(state, stage_id, 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,
@ -2008,10 +2081,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(),
))
@ -2030,10 +2105,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(),
))
@ -2073,7 +2150,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();
@ -2110,7 +2191,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();
@ -2144,10 +2229,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(),
))
@ -2400,10 +2487,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(),
))
@ -2571,6 +2660,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()],
@ -3827,10 +3918,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,
}
}
@ -4330,6 +4423,420 @@ 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 malformed retry event for the same StageId cannot rewrite the
// first attempt's immutable execution identity.
state
.apply_event(&test_stage_event(
5,
EventBody::StageStarted(StageStartedProps {
attempt: 2,
graph_visit: Some(99),
resumed_from_stage_id: Some(StageId::new("other", 7)),
..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();

16
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";
@ -46,8 +53,11 @@ pub mod keys {
pub const PARALLEL_BRANCH_COUNT: &str = "parallel.branch_count";
/// 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.";
@ -195,7 +205,7 @@ pub trait WorkflowContext {
fn parallel_group_id(&self) -> Option<StageId>;
fn parallel_branch_id(&self) -> Option<ParallelBranchId>;
/// Build the stage-level emit scope from the currently-executing node and
/// its accumulated visit count. Returns `None` for run-level emissions
/// its execution ordinal. Returns `None` for run-level emissions
/// where no stage is active (i.e., `CURRENT_NODE` is unset).
fn current_stage_scope(&self) -> Option<StageScope>;
}

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,
@ -376,11 +380,16 @@ fn event_body_from_event(event: &Event) -> EventBody {
visit: *visit,
branch_count: *branch_count,
}),
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,
@ -476,6 +485,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(),
@ -491,6 +502,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,
@ -1676,12 +1689,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 {
@ -1794,10 +1809,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 post-checkpoint work is replayed
/// after resume.
#[serde(default, skip_serializing_if = "Option::is_none")]
graph_visit: Option<u32>,
/// Prior execution superseded by this resumed replay, 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,
@ -306,10 +316,14 @@ pub enum Event {
branch_count: usize,
},
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,
@ -393,6 +407,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

@ -167,10 +167,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

@ -455,6 +455,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(),

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

@ -19,10 +19,9 @@ use crate::error::Error;
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};
use crate::{ManifestPath, pipeline, stage_scope};
/// Orchestrates a child workflow engine, polling for completion or stop
/// conditions.
@ -182,8 +181,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 replayed manager loop keeps the prior execution's
// child logs intact.
let visit = u64::from(stage_scope::execution_ordinal_from_context(context));
let child_logs = run_dir.join(format!("stages/{}@{visit}/child", node.id));
let _ = fs::create_dir_all(&child_logs).await;

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

@ -1,6 +1,6 @@
use std::collections::{BTreeMap, HashMap, HashSet};
use std::path::Path;
use std::sync::Arc;
use std::sync::{Arc, OnceLock};
use std::time::Instant;
use async_trait::async_trait;
@ -17,6 +17,7 @@ use crate::error::Error;
use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope};
use crate::hook_context::set_hook_node;
use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt};
use crate::run_dir::visit_from_context;
use crate::{artifact, millis_u64};
/// Fans out execution to multiple branches concurrently.
@ -32,7 +33,12 @@ struct BranchDispatch {
index: usize,
target_id: String,
branch_id: ParallelBranchId,
scope: StageScope,
/// Scope reserved by the branch task right before its
/// `ParallelBranchStarted` becomes observable. Empty when the branch was
/// cancelled or failed before starting — no events exist to pair a
/// completion with, and emitting one under a guessed ordinal would
/// resurrect a prior execution's stage.
scope: Arc<OnceLock<StageScope>>,
handle: JoinHandle<Result<BranchResult, Error>>,
}
@ -117,6 +123,7 @@ async fn run_branches(
let max_parallel = usize::try_from(max_parallel).unwrap_or(4).max(1);
let semaphore = Arc::new(Semaphore::new(max_parallel));
let shared_graph = Arc::new(graph.clone());
let branch_graph_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX);
let branch_preambles = parse_branch_preambles(
context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES),
@ -169,18 +176,13 @@ async fn run_branches(
let run_dir = run_dir.to_path_buf();
let semaphore = Arc::clone(&semaphore);
let group_id = parallel_group_id.clone();
let branch_scope = StageScope::for_parallel_branch(
target_id.clone(),
1,
group_id.clone(),
parallel_branch_id.clone(),
);
let reserved_scope = Arc::new(OnceLock::new());
dispatches.push(BranchDispatch {
index: branch_index,
target_id: target_id.clone(),
branch_id: parallel_branch_id.clone(),
scope: branch_scope.clone(),
scope: Arc::clone(&reserved_scope),
handle: tokio::spawn(async move {
let branch_start = Instant::now();
let task = async {
@ -195,12 +197,39 @@ async fn run_branches(
permit = &mut permit => permit
.map_err(|err| Error::handler_with_source("semaphore error", err))?,
};
// Only reserve once the branch is ready to become
// observable, so a branch cancelled while waiting on the
// semaphore never consumes an execution identity.
let execution = branch_services
.run
.stage_executions
.reserve(&target_id, branch_graph_visit);
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.stage_id.visit()),
);
let branch_scope = reserved_scope
.get_or_init(|| {
StageScope::for_parallel_branch(
target_id.clone(),
execution.stage_id.visit(),
group_id.clone(),
parallel_branch_id.clone(),
)
})
.clone();
branch_services.run.emitter.emit_scoped(
&Event::ParallelBranchStarted {
parallel_group_id: group_id.clone(),
parallel_branch_id: parallel_branch_id.clone(),
branch: target_id.clone(),
index: branch_index,
parallel_group_id: group_id.clone(),
parallel_branch_id: parallel_branch_id.clone(),
branch: target_id.clone(),
index: branch_index,
graph_visit: Some(execution.graph_visit),
resumed_from_stage_id: execution.resumed_from.clone(),
},
&branch_scope,
);
@ -255,15 +284,17 @@ async fn run_branches(
Err(payload) => {
let result =
failed_branch_result(&target_id, super::format_panic_message(&payload));
emit_branch_completed(
&branch_services.run.emitter,
&branch_scope,
group_id,
parallel_branch_id,
branch_index,
millis_u64(branch_start.elapsed()),
result.outcome.status,
);
if let Some(scope) = reserved_scope.get() {
emit_branch_completed(
&branch_services.run.emitter,
scope,
group_id,
parallel_branch_id,
branch_index,
millis_u64(branch_start.elapsed()),
result.outcome.status,
);
}
Ok(result)
}
}
@ -295,15 +326,17 @@ async fn run_branches(
),
};
if emit_completion {
emit_branch_completed(
&services.run.emitter,
&dispatch.scope,
parallel_group_id.clone(),
dispatch.branch_id,
dispatch.index,
0,
result.outcome.status,
);
if let Some(scope) = dispatch.scope.get() {
emit_branch_completed(
&services.run.emitter,
scope,
parallel_group_id.clone(),
dispatch.branch_id,
dispatch.index,
0,
result.outcome.status,
);
}
}
if result.outcome.failure_category() == Some(FailureCategory::Canceled) {
cancelled = true;
@ -799,6 +832,7 @@ mod tests {
logger.register(services.run.emitter.as_ref());
let (node, graph) = parallel_graph();
let context = test_context();
context.set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(2));
let outcome = ParallelHandler
.execute(&node, &context, &graph, Path::new("/tmp/test"), &services)
@ -825,7 +859,7 @@ mod tests {
let state = run_store.state().await.unwrap();
assert_eq!(
state
.stage(&StageId::new("par", 1))
.stage(&StageId::new("par", 2))
.unwrap()
.parallel_results
.as_ref()
@ -833,6 +867,15 @@ mod tests {
.len(),
2
);
for branch in ["branch_a", "branch_b"] {
assert_eq!(
state
.stage(&StageId::new(branch, 1))
.and_then(|stage| stage.graph_visit),
Some(2),
"parallel children should inherit the parent graph visit"
);
}
}
#[tokio::test]

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(crate) mod stage_execution;
mod stage_scope;
pub mod static_reference;
pub mod steering_hub;

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

@ -20,9 +20,10 @@ use crate::artifact_snapshot::{ArtifactCollectionSummary, collect_artifacts};
use crate::artifact_upload::ArtifactSink;
use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel};
use crate::graph::{WorkflowGraph, WorkflowNode};
use crate::lifecycle::event::{stage_scope_for, stage_visit};
use crate::lifecycle::event::stage_scope_for;
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,10 @@ 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 scope = stage_scope_for(&self.stage_executions, state, node_id);
let visit = scope.visit;
let node_slug = if visit <= 1 {
node_id.to_string()
} else {
@ -144,7 +152,7 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
return Ok(());
}
let stage_id = StageId::new(node_id.to_string(), visit);
let stage_id = scope.stage_id();
if let Err(err) = self
.persist_artifacts(
&stage_id,
@ -162,7 +170,6 @@ impl RunLifecycle<WorkflowGraph> for ArtifactLifecycle {
return Ok(());
}
self.record_captured_assets(&new_assets);
let scope = stage_scope_for(state, node_id);
for asset in &new_assets {
self.emitter.emit_scoped(
&Event::ArtifactCaptured {

121
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::{StageExecution, 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(
@ -92,13 +95,16 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option<String> {
.and_then(|value| value.as_str().map(ToOwned::to_owned))
}
/// Context values for `StageCompleted` events. Unlike
/// `artifact::strip_transient_keys`, this keeps `CURRENT_PREAMBLE` — stage
/// events have always included the active preamble — and drops only the
/// parallel stash, which can embed every branch's rendered preamble.
/// Context values for `StageCompleted` events. Runtime-only keys are stripped,
/// except for `CURRENT_PREAMBLE`, which stage events have historically
/// included.
fn stage_context_values(workflow_context: &Context) -> Option<BTreeMap<String, serde_json::Value>> {
let mut snapshot = workflow_context.snapshot();
snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES);
let preamble = snapshot.get(context::keys::CURRENT_PREAMBLE).cloned();
artifact::strip_transient_keys(&mut snapshot);
if let Some(preamble) = preamble {
snapshot.insert(context::keys::CURRENT_PREAMBLE.to_owned(), preamble);
}
(!snapshot.is_empty()).then(|| snapshot.into_iter().collect())
}
@ -107,15 +113,40 @@ 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 {
fn stage_scope_from_execution(
execution: Option<&StageExecution>,
state: &WfRunState,
node_id: &str,
) -> StageScope {
let (node_id, visit) = execution.map_or_else(
|| (node_id.to_owned(), stage_visit(state, node_id)),
|execution| {
(
execution.stage_id.node_id().to_owned(),
execution.stage_id.visit(),
)
},
);
StageScope {
node_id: node_id.to_string(),
visit: stage_visit(state, node_id),
parallel_group_id: state.context.parallel_group_id(),
node_id,
visit,
parallel_group_id: state.context.parallel_group_id(),
parallel_branch_id: state.context.parallel_branch_id(),
}
}
/// 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 execution = stage_executions.active(node_id);
stage_scope_from_execution(execution.as_deref(), state, node_id)
}
#[async_trait]
impl RunLifecycle<WorkflowGraph> for EventLifecycle {
async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> {
@ -160,17 +191,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_from_execution(Some(&execution), 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.clone(),
},
&scope,
);
@ -210,15 +248,23 @@ 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_from_execution(execution.as_deref(), 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
.as_ref()
.and_then(|execution| execution.resumed_from.clone()),
},
&scope,
);
@ -234,7 +280,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 +329,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 +446,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_from_execution(execution.as_deref(), 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 +475,10 @@ impl RunLifecycle<WorkflowGraph> for EventLifecycle {
.collect::<BTreeMap<_, _>>(),
diff,
diff_summary,
graph_visit: Some(graph_visit),
resumed_from_stage_id: execution
.as_ref()
.and_then(|execution| execution.resumed_from.clone()),
},
&scope,
);
@ -458,17 +512,30 @@ mod tests {
use super::*;
#[test]
fn stage_context_values_drops_parallel_branch_preambles() {
fn stage_context_values_drops_runtime_keys_but_keeps_current_preamble() {
let workflow_context = Context::new();
workflow_context.set(
context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES,
serde_json::json!([{"fidelity": "summary:high", "preamble": "runtime only"}]),
);
workflow_context.set(
context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL,
serde_json::json!(2),
);
workflow_context.set(
context::keys::CURRENT_PREAMBLE,
serde_json::json!("active preamble"),
);
workflow_context.set("response.work", serde_json::json!("durable"));
let values = stage_context_values(&workflow_context).expect("snapshot should not be empty");
assert!(!values.contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES));
assert!(!values.contains_key(context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL));
assert_eq!(
values.get(context::keys::CURRENT_PREAMBLE),
Some(&serde_json::json!("active preamble"))
);
assert_eq!(
values.get("response.work"),
Some(&serde_json::json!("durable"))

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.stage_id.visit()),
);
// 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.stage_id.visit()),
);
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(),

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

@ -4,6 +4,7 @@ use super::start::{StartServices, Started, execute_persisted_run};
use crate::error::Error;
use crate::event::{Event, append_event_to_sink};
use crate::outcome::StageOutcome;
use crate::pipeline::ResumeState;
use crate::run_status::RunStatus;
/// Resume a workflow run from its checkpoint. Errors if no checkpoint is found.
@ -32,9 +33,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result<Started,
}
}
let checkpoint = state
.current_checkpoint()
.cloned()
let resume_state = ResumeState::from_projection(&state)
.ok_or_else(|| Error::Precondition("no checkpoint to resume from".to_string()))?;
let definition_blob = state.spec.definition_blob;
@ -47,7 +46,7 @@ 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(resume_state), 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(),

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

@ -38,8 +38,9 @@ use crate::handler::HandlerRegistry;
use crate::outcome::{Outcome, StageOutcome};
use crate::pipeline::{
self, FinalizeOptions, Finalized, InitOptions, LlmSpec, Persisted, PullRequestOptions,
SandboxEnvSpec, build_conclusion_from_store, classify_engine_result,
ResumeState, SandboxEnvSpec, build_conclusion_from_store, classify_engine_result,
};
#[cfg(test)]
use crate::records::Checkpoint;
use crate::run_control::RunControlState;
use crate::run_materialization::resolve_run_model;
@ -174,7 +175,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result<Started, E
pub(super) async fn execute_persisted_run(
run_dir: &Path,
checkpoint: Option<Checkpoint>,
resume: Option<ResumeState>,
services: StartServices,
) -> Result<Started, Error> {
let cancel_token = services.cancel_token.clone();
@ -261,7 +262,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, resume)).await;
match started {
Ok(started) => {
@ -797,7 +798,7 @@ impl RunSession {
async fn run(
self,
persisted: Persisted,
checkpoint: Option<Checkpoint>,
resume: Option<ResumeState>,
) -> Result<Started, Error> {
let on_node = self.on_node.clone();
@ -878,7 +879,7 @@ impl RunSession {
registry_override: self.registry_override,
artifact_sink: self.artifact_sink,
run_control: self.run_control,
checkpoint,
resume,
seed_context: self.seed_context,
fabro_run_tools: self.fabro_run_tools,
};
@ -2126,6 +2127,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(),
@ -2509,6 +2512,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
@ -2607,6 +2612,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

@ -77,6 +77,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 {

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

@ -31,7 +31,7 @@ use crate::handler::start::StartHandler;
use crate::handler::{Handler as HandlerTrait, HandlerRegistry};
use crate::outcome::{Outcome, OutcomeExt, StageOutcome};
use crate::pipeline::initialize;
use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec};
use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, ResumeState, SandboxEnvSpec};
use crate::records::RunSpec;
use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions, SetupCommand};
use crate::test_support::run_graph;
@ -291,7 +291,7 @@ async fn execute_test_run_with_options(
run_control: None,
registry_override,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
},
@ -353,7 +353,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
},
@ -375,6 +375,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::test_with_high_water(
&fabro_types::StageId::new("work", 1),
Some(fabro_types::StageId::new("work", 1)),
);
let resume = ResumeState::for_test(checkpoint, seed);
let initialized = initialize(
persisted_workflow(graph, String::new(), &run_dir, run_id),
InitOptions {
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,
resume: Some(resume),
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>,
@ -423,7 +603,7 @@ async fn run_with_lifecycle(
run_control: None,
registry_override: Some(Arc::new(registry)),
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
},

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"),

23
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::{StageExecutionSeed, StageExecutionTracker};
use crate::steering_hub::SteeringHub;
type BuiltSandboxEnv = (HashMap<String, String>, Option<Arc<GitHubTokenSource>>);
@ -286,6 +287,13 @@ pub async fn initialize(
mut options: InitOptions,
) -> Result<Initialized, Error> {
let (graph, source, _diagnostics, run_dir, run_spec) = persisted.into_parts();
let (checkpoint, stage_executions) = options.resume.take().map_or_else(
|| (None, StageExecutionSeed::default()),
|resume| {
let (checkpoint, stage_executions) = resume.into_parts();
(Some(checkpoint), stage_executions)
},
);
let host_source_dir = run_spec.source_directory.as_deref().map(PathBuf::from);
options.run_options.run_dir = run_dir.clone();
options.run_options.git = options.git.clone();
@ -306,7 +314,7 @@ pub async fn initialize(
)))
};
let attach_existing = options.checkpoint.is_some();
let attach_existing = checkpoint.is_some();
options.run_options.display_base_sha = options
.run_options
.pre_run_git
@ -619,6 +627,7 @@ pub async fn initialize(
sandbox_git,
metadata_runtime,
metadata_writer,
StageExecutionTracker::seeded(stage_executions),
);
let engine = Arc::new(EngineServices {
run: Arc::clone(&run_services),
@ -640,7 +649,7 @@ pub async fn initialize(
graph,
source,
run_options: options.run_options,
checkpoint: options.checkpoint,
checkpoint,
seed_context: options.seed_context,
on_node: None,
artifact_sink: options.artifact_sink,
@ -863,7 +872,7 @@ mod tests {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
})
@ -944,7 +953,7 @@ mod tests {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
})
@ -1165,7 +1174,7 @@ mod tests {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
})
@ -1260,7 +1269,7 @@ mod tests {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
})
@ -1402,7 +1411,7 @@ mod tests {
run_control: None,
registry_override: None,
artifact_sink: None,
checkpoint: None,
resume: None,
seed_context: None,
fabro_run_tools: None,
})

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

@ -23,7 +23,7 @@ pub use pull_request::{
pub use transform::transform;
pub use types::{
Concluded, Executed, FinalizeOptions, Finalized, InitOptions, Initialized, LlmSpec, Parsed,
Persisted, PullRequestOptions, SandboxEnvSpec, TEMPLATE_UNDEFINED_VARIABLE_RULE,
Persisted, PullRequestOptions, ResumeState, SandboxEnvSpec, TEMPLATE_UNDEFINED_VARIABLE_RULE,
TransformOptions, Transformed, Validated,
};
pub use validate::validate;

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

@ -9,7 +9,7 @@ use fabro_model::{Catalog, FallbackTarget, ProviderId};
use fabro_sandbox::SandboxSpec;
use fabro_template::TemplateContext;
use fabro_types::settings::run::{PullRequestSettings, RunModelControls};
use fabro_types::{ManifestPath, RunId};
use fabro_types::{ManifestPath, RunId, RunProjection};
use fabro_validate::{Diagnostic, Severity};
use fabro_vault::Vault;
use tokio::sync::RwLock as AsyncRwLock;
@ -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;
@ -248,6 +249,41 @@ pub struct SandboxEnvSpec {
pub origin_url: Option<String>,
}
/// Opaque, internally consistent state needed to resume from the latest
/// checkpoint in a run projection.
pub struct ResumeState {
checkpoint: Checkpoint,
stage_executions: StageExecutionSeed,
}
impl ResumeState {
/// Build resume state from a projection's latest checkpoint and complete
/// stage history.
#[must_use]
pub fn from_projection(projection: &RunProjection) -> Option<Self> {
let checkpoint_record = projection.checkpoints.last()?;
Some(Self {
checkpoint: checkpoint_record.checkpoint.clone(),
stage_executions: StageExecutionSeed::from_projection(
projection,
checkpoint_record.seq,
),
})
}
pub(crate) fn into_parts(self) -> (Checkpoint, StageExecutionSeed) {
(self.checkpoint, self.stage_executions)
}
#[cfg(test)]
pub(crate) fn for_test(checkpoint: Checkpoint, stage_executions: StageExecutionSeed) -> Self {
Self {
checkpoint,
stage_executions,
}
}
}
pub struct InitOptions {
pub run_store: RunStoreHandle,
pub dry_run: bool,
@ -268,7 +304,7 @@ pub struct InitOptions {
pub registry_override: Option<Arc<HandlerRegistry>>,
pub artifact_sink: Option<ArtifactSink>,
pub run_control: Option<Arc<RunControlState>>,
pub checkpoint: Option<Checkpoint>,
pub resume: Option<ResumeState>,
pub seed_context: Option<Context>,
pub fabro_run_tools: Option<FabroRunToolServices>,
}

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

@ -21,6 +21,7 @@ use crate::interview_runtime::RunInterviewBlocker;
use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle};
use crate::runtime_store::RunStoreHandle;
use crate::sandbox_git_runtime::SandboxGitRuntime;
use crate::stage_execution::StageExecutionTracker;
use crate::workflow_bundle::WorkflowBundle;
#[derive(Clone, Debug, PartialEq, Eq)]
@ -106,6 +107,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 {
@ -124,6 +128,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,
@ -140,6 +145,7 @@ impl RunServices {
metadata_runtime,
metadata_writer,
interview_blocker: Arc::new(RunInterviewBlocker::new()),
stage_executions,
})
}
@ -320,6 +326,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()),

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

@ -0,0 +1,343 @@
//! 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 prior 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(Debug, PartialEq, Eq)]
pub(crate) struct StageExecution {
/// Canonical external identity for this execution.
pub stage_id: StageId,
/// Graph visit that produced this execution.
pub graph_visit: u32,
/// Prior post-checkpoint execution superseded by this resumed execution.
pub resumed_from: Option<StageId>,
}
#[derive(Debug, Default)]
struct NodeExecutionState {
/// Highest execution ordinal observed or reserved for this node.
high_water: u32,
/// Pending provenance link, consumed by the next reservation.
resumed_from: Option<StageId>,
/// Execution reserved since the latest node boundary.
active: Option<Arc<StageExecution>>,
}
/// 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(Debug, Default)]
pub(crate) struct StageExecutionSeed {
nodes: HashMap<String, NodeExecutionState>,
}
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 replay supersedes.
#[must_use]
pub(crate) fn from_projection(projection: &RunProjection, checkpoint_seq: u32) -> Self {
let mut nodes = HashMap::new();
for (stage_id, stage) in projection.iter_stages_unordered() {
let entry = nodes
.entry(stage_id.node_id().to_owned())
.or_insert_with(NodeExecutionState::default);
entry.high_water = entry.high_water.max(stage_id.visit());
if stage.first_event_seq.get() > checkpoint_seq {
let is_latest = entry
.resumed_from
.as_ref()
.is_none_or(|current| current.visit() < stage_id.visit());
if is_latest {
entry.resumed_from = Some(stage_id.clone());
}
}
}
Self { nodes }
}
#[cfg(test)]
pub(crate) fn test_with_high_water(
high_water: &StageId,
resumed_from: Option<StageId>,
) -> Self {
let node_id = high_water.node_id().to_owned();
Self {
nodes: HashMap::from([(node_id, NodeExecutionState {
high_water: high_water.visit(),
resumed_from,
active: None,
})]),
}
}
}
/// 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<HashMap<String, NodeExecutionState>>>,
}
impl StageExecutionTracker {
#[must_use]
pub(crate) fn seeded(seed: StageExecutionSeed) -> Self {
Self {
state: Arc::new(Mutex::new(seed.nodes)),
}
}
fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, NodeExecutionState>> {
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) {
if let Some(node) = self.lock().get_mut(node_id) {
node.active = None;
}
}
/// 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<Arc<StageExecution>> {
self.lock()
.get(node_id)
.and_then(|node| node.active.as_ref().map(Arc::clone))
}
fn reserve_locked(
state: &mut HashMap<String, NodeExecutionState>,
node_id: &str,
graph_visit: u32,
) -> Arc<StageExecution> {
let node = state.entry(node_id.to_owned()).or_default();
node.high_water = node.high_water.saturating_add(1);
let execution = Arc::new(StageExecution {
stage_id: StageId::new(node_id, node.high_water),
graph_visit,
resumed_from: node.resumed_from.take(),
});
node.active = Some(Arc::clone(&execution));
execution
}
/// 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) -> Arc<StageExecution> {
let mut state = self.lock();
Self::reserve_locked(&mut state, node_id, graph_visit)
}
/// 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) -> Arc<StageExecution> {
let mut state = self.lock();
if let Some(execution) = state
.get(node_id)
.and_then(|node| node.active.as_ref().map(Arc::clone))
{
return execution;
}
Self::reserve_locked(&mut state, 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).stage_id.visit(), 1);
tracker.begin_node("work");
assert_eq!(tracker.reserve("work", 2).stage_id.visit(), 2);
assert_eq!(tracker.reserve("other", 1).stage_id.visit(), 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).stage_id.visit(), 3);
assert_eq!(tracker.reserve("plan", 1).stage_id.visit(), 2);
assert_eq!(tracker.reserve("new", 1).stage_id.visit(), 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.stage_id.visit(), 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.stage_id.visit(), 1);
tracker.begin_node("work");
assert_eq!(tracker.ensure("work", 2).stage_id.visit(), 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(|execution| execution.stage_id.visit()),
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.nodes
.get("work")
.and_then(|node| node.resumed_from.as_ref()),
Some(&StageId::new("work", 2))
);
assert_eq!(
seed.nodes
.get("plan")
.and_then(|node| node.resumed_from.as_ref()),
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.stage_id.visit(), 2);
assert_eq!(first.resumed_from, Some(StageId::new("work", 1)));
tracker.begin_node("work");
let second = tracker.reserve("work", 2);
assert_eq!(second.stage_id.visit(), 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).stage_id.visit() })
})
.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<_>>());
}
#[tokio::test(flavor = "multi_thread")]
async fn concurrent_ensure_calls_reuse_one_reservation() {
let tracker = StageExecutionTracker::default();
let barrier = Arc::new(tokio::sync::Barrier::new(16));
let handles: Vec<_> = (0..16)
.map(|_| {
let tracker = tracker.clone();
let barrier = Arc::clone(&barrier);
tokio::spawn(async move {
barrier.wait().await;
tracker.ensure("branch", 1).stage_id.visit()
})
})
.collect();
for handle in handles {
assert_eq!(handle.await.expect("ensure task panicked"), 1);
}
}
}

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

@ -1,11 +1,30 @@
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). Direct-handler call sites that skip the full
/// lifecycle fall back to the graph visit, which equals the ordinal for a
/// first execution.
pub(crate) fn execution_ordinal_from_context(context: &WfContext) -> u32 {
context
.get(keys::INTERNAL_STAGE_EXECUTION_ORDINAL)
.and_then(|value| value.as_u64())
.map_or_else(
|| u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX),
|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 post-checkpoint work is replayed after
/// resume.
#[derive(Clone, Debug)]
pub struct StageScope {
pub node_id: String,
@ -15,13 +34,14 @@ 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);
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(),
}
}
@ -39,11 +59,10 @@ impl StageScope {
/// Build scope for the branch-lifecycle events emitted by the parallel
/// handler (`ParallelBranchStarted` and `ParallelBranchCompleted`).
///
/// `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 prior dispatch'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()),

1
lib/foundation/fabro-api/build.rs
View file

@ -344,6 +344,7 @@ fn main() {
("StageCompletion", "fabro_types::StageCompletion", &[]),
("Conclusion", "fabro_types::Conclusion", &[]),
("StageOutcome", "fabro_types::StageOutcome", &[]),
("StageId", "fabro_types::StageId", &[]),
("StageHandler", "fabro_types::StageHandler", &[]),
("StageState", "fabro_types::StageState", &[]),
("AgentControlState", "fabro_types::AgentControlState", &[]),

2
lib/foundation/fabro-api/src/lib.rs
View file

@ -67,7 +67,7 @@ pub mod types {
SessionSummary, SessionTurn, SkillsProjection, StageCompletion, StageContextWindow,
StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod,
StageContextWindowProjection, StageContextWindowStaleness,
StageContextWindowUnavailableReason, StageContextWindowWarning, StageHandler,
StageContextWindowUnavailableReason, StageContextWindowWarning, StageHandler, StageId,
StageModelUsage, StageOutcome, StageProjection, StageState, SubAgentProjection,
SubAgentStatus, SystemActorKind, SystemIntegrationStatus, SystemIntegrationsResponse,
TodoListProjection, TurnId, UpdateVariableRequest, UserPrincipal, Variable,

31
lib/foundation/fabro-api/tests/stage_id_round_trip.rs Normal file
View file

@ -0,0 +1,31 @@
use std::any::{TypeId, type_name};
use fabro_api::types::StageId as ApiStageId;
use fabro_types::StageId;
use serde_json::json;
#[test]
fn stage_id_reuses_canonical_type() {
assert_same_type::<ApiStageId, StageId>();
}
#[test]
fn stage_id_round_trips_openapi_representation() {
let stage_id = StageId::new("verify", 2);
assert_eq!(serde_json::to_value(&stage_id).unwrap(), json!("verify@2"));
assert_eq!(
serde_json::from_value::<ApiStageId>(json!("verify@2")).unwrap(),
stage_id
);
}
fn assert_same_type<T: 'static, U: 'static>() {
assert_eq!(
TypeId::of::<T>(),
TypeId::of::<U>(),
"{} should be the same type as {}",
type_name::<T>(),
type_name::<U>()
);
}

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

@ -1,7 +1,7 @@
use serde::{Deserialize, Serialize};
use super::ExecOutputTail;
use crate::{CommandTermination, ParallelBranchResult, PullRequestLink, StageOutcome};
use crate::{CommandTermination, ParallelBranchResult, PullRequestLink, StageId, StageOutcome};
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
pub struct InterviewOption {
@ -21,7 +21,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 superseded by this resumed replay.
#[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 post-checkpoint work is replayed 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 superseded by this resumed replay.
#[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)]

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

@ -319,59 +319,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<Vec<crate::ParallelBranchResult>>,
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<Vec<crate::ParallelBranchResult>>,
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
/// post-checkpoint work is replayed 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 superseded by this resumed replay.
#[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(
@ -485,6 +497,8 @@ impl StageProjection {
termination: None,
started_at: None,
handler: None,
graph_visit: None,
resumed_from_stage_id: None,
state: StageState::Running,
}
}
@ -518,14 +532,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 replay after resume
/// 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;
}
}
@ -569,11 +593,18 @@ impl RunProjection {
self.stages.get(stage)
}
/// Iterate stages in unspecified order without allocating or sorting.
///
/// Use this only for order-independent aggregation. Presentation and
/// serialization callers should use [`Self::iter_stages`] instead.
pub fn iter_stages_unordered(&self) -> impl Iterator<Item = (&StageId, &StageProjection)> {
self.stages.iter()
}
/// Iterate stages in `first_event_seq` order (the chronological order in
/// which each stage's first lifecycle event was recorded). Internal
/// storage is a `HashMap`, so iteration would otherwise be
/// non-deterministic; every caller wants chronological order, so we sort
/// here once instead of asking each caller to remember.
/// storage is a `HashMap`, so presentation callers sort through this
/// helper instead of relying on non-deterministic map iteration.
pub fn iter_stages(&self) -> impl Iterator<Item = (&StageId, &StageProjection)> {
let mut entries: Vec<(&StageId, &StageProjection)> = self.stages.iter().collect();
entries.sort_by(|(left_id, left_stage), (right_id, right_stage)| {

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

@ -28,7 +28,7 @@ import type { StageState } from './stage-state';
*/
export interface RunStage {
/**
* StageId in \"node_id@visit\" form, e.g. verify@2.
* Canonical stage execution identifier in `node_id@visit` form.
*/
'id': string;
/**
@ -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 replay of post-checkpoint work after resume. 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 post-checkpoint execution is replayed after resume. Absent for stages recorded before execution identity was tracked.
*/
'graph_visit'?: number | null;
/**
* Canonical stage execution identifier in `node_id@visit` form.
*/
'resumed_from_stage_id'?: string | null;
'provider_used'?: StageModelUsage | null;
/**
* Wall-clock time the latest attempt of this stage started, if known.