diff --git a/apps/fabro-web/app/components/stage-popover.test.tsx b/apps/fabro-web/app/components/stage-popover.test.tsx index efdbd7785..45a1a69a4 100644 --- a/apps/fabro-web/app/components/stage-popover.test.tsx +++ b/apps/fabro-web/app/components/stage-popover.test.tsx @@ -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 { function makeStage(overrides: Partial = {}): Stage { return { - id: "implement@1", - name: "implement", - handler: "agent", - nodeId: "implement", - visit: 1, - status: "succeeded", - duration: "1m 30s", - startedAt: "2026-05-24T11:58:30Z", - providerUsed: { mode: "policy", model: "claude-opus-4-7", reasoning_effort: "high" }, + id: "implement@1", + name: "implement", + handler: "agent", + nodeId: "implement", + visit: 1, + graphVisit: null, + resumedFromStageId: null, + status: "succeeded", + duration: "1m 30s", + startedAt: "2026-05-24T11:58:30Z", + providerUsed: { mode: "policy", model: "claude-opus-4-7", reasoning_effort: "high" }, ...overrides, }; } @@ -275,6 +278,36 @@ describe("StagePopover rendering", () => { expect(text).not.toContain("Reason"); }); + test("resumed stage links to the prior execution and shows a divergent graph visit", () => { + const stage = makeStage({ + id: "implement@2", + visit: 2, + graphVisit: 1, + resumedFromStageId: "implement@1", + status: "running", + duration: "--", + }); + const tree = render( + + + , + ); + const text = textOf(tree); + expect(text).toContain("Resumed from"); + expect(text).toContain("implement@1"); + expect(text).toContain("Graph visit"); + const json = JSON.stringify(tree.toJSON()); + expect(json).toContain("/runs/run-1/stages/implement@1"); + }); + + test("stage without ordinal divergence hides the graph visit row", () => { + const stage = makeStage({ graphVisit: 1, status: "pending", duration: "--" }); + const tree = render(); + 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( [ diff --git a/apps/fabro-web/app/components/stage-popover.tsx b/apps/fabro-web/app/components/stage-popover.tsx index c6722a71b..b8f03cc28 100644 --- a/apps/fabro-web/app/components/stage-popover.tsx +++ b/apps/fabro-web/app/components/stage-popover.tsx @@ -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) { {duration} )} + {stage.resumedFromStageId && ( + + + {stage.resumedFromStageId} + + + )} + {stage.graphVisit != null && stage.graphVisit !== stage.visit && ( + + {stage.graphVisit} + + )} diff --git a/apps/fabro-web/app/components/stage-sidebar.test.tsx b/apps/fabro-web/app/components/stage-sidebar.test.tsx index b25ba4d25..69cf0c097 100644 --- a/apps/fabro-web/app/components/stage-sidebar.test.tsx +++ b/apps/fabro-web/app/components/stage-sidebar.test.tsx @@ -11,6 +11,8 @@ function makeStage(overrides: Partial = {}): Stage { handler: "agent", nodeId: "implement", visit: 1, + graphVisit: null, + resumedFromStageId: null, status: "running", duration: "--", startedAt: null, diff --git a/apps/fabro-web/app/lib/stage-sidebar.test.ts b/apps/fabro-web/app/lib/stage-sidebar.test.ts index 10b17a318..a71323654 100644 --- a/apps/fabro-web/app/lib/stage-sidebar.test.ts +++ b/apps/fabro-web/app/lib/stage-sidebar.test.ts @@ -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: [ diff --git a/apps/fabro-web/app/lib/stage-sidebar.ts b/apps/fabro-web/app/lib/stage-sidebar.ts index 58f91db56..a0228495d 100644 --- a/apps/fabro-web/app/lib/stage-sidebar.ts +++ b/apps/fabro-web/app/lib/stage-sidebar.ts @@ -14,8 +14,17 @@ export interface Stage { handler: StageHandler; status: StageState; duration: string; - nodeId: string; + /** 1-based stage execution ordinal — the numeric component of `id`. */ visit: number; + nodeId: string; + /** + * How many times workflow control entered this node. Differs from `visit` + * when a cancelled or crashed execution was reexecuted after resume; null + * for stages recorded before execution identity was tracked. + */ + graphVisit: number | null; + /** StageId of the prior execution this stage resumes from, if any. */ + resumedFromStageId: string | null; startedAt: string | null; providerUsed: StageModelUsage | null; } @@ -85,6 +94,8 @@ export function mapRunStagesToSidebarStages( handler: stage.handler, nodeId: stage.node_id, visit: stage.visit, + graphVisit: stage.graph_visit ?? null, + resumedFromStageId: stage.resumed_from_stage_id ?? null, status: stage.status, duration: stage.wall_time_ms != null ? formatDurationMs(stage.wall_time_ms) diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index 94da46bde..2c497522b 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -1,5 +1,5 @@ import { useMemo, useReducer, useState } from "react"; -import { useParams } from "react-router"; +import { Link, useParams } from "react-router"; import { ArrowDownTrayIcon, ChevronDownIcon, @@ -1745,6 +1745,17 @@ function RunStageActivityStage({
+ {selectedStage.resumedFromStageId && ( +

+ Resumed from{" "} + + {selectedStage.resumedFromStageId} + +

+ )} - + 1-based stage execution ordinal, the numeric component of `id`. It + increments each time the node produces a new observable execution: + graph re-entry (loops) and reexecution after cancel or crash + recovery. Automatic in-place retries do not increment it. example: 2 + graph_visit: + type: ["integer", "null"] + format: uint32 + minimum: 1 + description: >- + 1-based count of how many times workflow control entered this node + (drives `max_visits`). Differs from `visit` when a cancelled or + crashed execution was reexecuted after resume. Absent for stages + recorded before execution identity was tracked. + example: 1 + resumed_from_stage_id: + type: ["string", "null"] + description: >- + StageId of the prior cancelled or interrupted execution this stage + resumes from, when the run was resumed after that execution became + observable. + example: verify@1 provider_used: oneOf: - $ref: "#/components/schemas/StageModelUsage" diff --git a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs index 6e1a6227b..4bead36e3 100644 --- a/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs +++ b/lib/apps/fabro-cli/src/commands/run/run_progress/mod.rs @@ -510,12 +510,14 @@ mod tests { fn stage_started(node_id: &str, name: &str) -> Event { Event::StageStarted { - node_id: node_id.into(), - name: name.into(), - index: 0, - handler_type: String::new(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: node_id.into(), + name: name.into(), + index: 0, + handler_type: String::new(), + attempt: 1, + max_attempts: 1, } } @@ -592,10 +594,12 @@ mod tests { assert_eq!(ui.stage.parallel_parent.as_deref(), Some("fork1")); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); let stage = &ui.stage.active_stages["fork1"]; assert_eq!(stage.tool_calls.len(), 1); @@ -633,10 +637,12 @@ mod tests { join_policy: "wait_all".into(), }); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); let stage = &ui.stage.active_stages["fork1"]; @@ -1249,10 +1255,12 @@ mod tests { join_policy: "wait_all".into(), }); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); emit(&mut ui, Event::ParallelBranchCompleted { parallel_group_id: StageId::new("fork1", 1), @@ -1282,12 +1290,14 @@ mod tests { let stage_started = serde_json::to_string(&to_run_event_at( &fixtures::RUN_1, &Event::StageStarted { - node_id: "code".into(), - name: "Code".into(), - index: 0, - handler_type: "agent".into(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "code".into(), + name: "Code".into(), + index: 0, + handler_type: "agent".into(), + attempt: 1, + max_attempts: 1, }, started_ts, None, diff --git a/lib/apps/fabro-cli/tests/it/cmd/attach.rs b/lib/apps/fabro-cli/tests/it/cmd/attach.rs index fff46c251..c801ead11 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/attach.rs @@ -1169,6 +1169,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "node_label": "Start", "properties": { "attempt": 1, + "graph_visit": 1, "handler_type": "start", "index": 0, "max_attempts": 1 @@ -1195,6 +1196,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "internal.fidelity": "compact", "internal.node_visit_count": 1, "internal.run_id": "[ULID]", + "internal.stage_execution_ordinal": 1, "internal.thread_id": null }, "index": 0, @@ -1257,6 +1259,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "outcome": "succeeded" }, "current_node": "start", + "graph_visit": 1, "next_node_id": "approve", "node_outcomes": { "start": { @@ -1284,6 +1287,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "node_label": "Approve?", "properties": { "attempt": 1, + "graph_visit": 1, "handler_type": "human", "index": 1, "max_attempts": 1 diff --git a/lib/apps/fabro-server/src/demo/mod.rs b/lib/apps/fabro-server/src/demo/mod.rs index 4b5bb204e..74288c809 100644 --- a/lib/apps/fabro-server/src/demo/mod.rs +++ b/lib/apps/fabro-server/src/demo/mod.rs @@ -1388,6 +1388,8 @@ mod runs { None, StageHandler::Command, None, + None, + None, ), run_stage_from_stage_id( &StageId::new("propose-changes", 1), @@ -1397,6 +1399,8 @@ mod runs { None, StageHandler::Agent, None, + None, + None, ), run_stage_from_stage_id( &StageId::new("review-changes", 1), @@ -1406,6 +1410,8 @@ mod runs { None, StageHandler::Agent, None, + None, + None, ), run_stage_from_stage_id( &StageId::new("apply-changes", 1), @@ -1415,6 +1421,8 @@ mod runs { None, StageHandler::Command, None, + None, + None, ), run_stage_from_stage_id( &StageId::new("apply-changes", 2), @@ -1424,6 +1432,8 @@ mod runs { None, StageHandler::Command, None, + None, + None, ), ] } diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index fac2dd51a..0bee9c04c 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -1340,6 +1340,8 @@ pub(crate) fn run_stage_from_stage_id( started_at: Option>, handler: StageHandler, provider_used: Option, + graph_visit: Option, + resumed_from_stage_id: Option<&StageId>, ) -> RunStage { RunStage { id: stage_id.to_string(), @@ -1352,6 +1354,8 @@ pub(crate) fn run_stage_from_stage_id( .expect("StageId stores a non-zero visit"), provider_used, started_at, + graph_visit: graph_visit.and_then(std::num::NonZeroU32::new), + resumed_from_stage_id: resumed_from_stage_id.map(StageId::to_string), } } diff --git a/lib/apps/fabro-server/src/server/handler/billing.rs b/lib/apps/fabro-server/src/server/handler/billing.rs index 557e1180a..6c0c315e6 100644 --- a/lib/apps/fabro-server/src/server/handler/billing.rs +++ b/lib/apps/fabro-server/src/server/handler/billing.rs @@ -58,6 +58,8 @@ async fn list_run_stages( stage.started_at, handler, stage.provider_used.clone(), + stage.graph_visit, + stage.resumed_from_stage_id.as_ref(), ) }) .collect::>(); diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 454455be0..1939bbb94 100644 --- a/lib/apps/fabro-server/src/server/tests.rs +++ b/lib/apps/fabro-server/src/server/tests.rs @@ -4090,12 +4090,14 @@ async fn create_durable_run_with_events( fn stage_started_event(node_id: &str, handler_type: &str) -> workflow_event::Event { workflow_event::Event::StageStarted { - node_id: node_id.to_string(), - name: node_id.to_string(), - index: 1, - handler_type: handler_type.to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: node_id.to_string(), + name: node_id.to_string(), + index: 1, + handler_type: handler_type.to_string(), + attempt: 1, + max_attempts: 1, } } @@ -4938,12 +4940,14 @@ async fn list_run_stages_projects_retrying_until_completion() { "setup", 1, &workflow_event::Event::StageStarted { - node_id: "setup".to_string(), - name: "Setup".to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "setup".to_string(), + name: "Setup".to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -4982,12 +4986,14 @@ async fn list_run_stages_projects_retrying_until_completion() { "work", 1, &workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 3, }, ) .await; @@ -5103,12 +5109,14 @@ async fn list_run_stages_projects_running_stage_as_cancelled_after_cancelled_run "work", 1, &workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 1, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 1, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -5174,12 +5182,14 @@ async fn list_run_stages_includes_stage_model_usage() { "prompt", 1, &workflow_event::Event::StageStarted { - node_id: "prompt".to_string(), - name: "Prompt".to_string(), - index: 0, - handler_type: "prompt".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "prompt".to_string(), + name: "Prompt".to_string(), + index: 0, + handler_type: "prompt".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -5294,12 +5304,14 @@ async fn list_run_stages_distinguishes_visits() { "verify", 1, &workflow_event::Event::StageStarted { - node_id: "verify".to_string(), - name: "Verify".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "verify".to_string(), + name: "Verify".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -5340,12 +5352,14 @@ async fn list_run_stages_distinguishes_visits() { "verify", 2, &workflow_event::Event::StageStarted { - node_id: "verify".to_string(), - name: "Verify".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "verify".to_string(), + name: "Verify".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -5384,6 +5398,110 @@ async fn list_run_stages_distinguishes_visits() { assert!(first.get("dot_id").is_none(), "dot_id should be removed"); } +#[tokio::test] +async fn list_run_stages_exposes_execution_identity_for_resumed_stage() { + let state = test_app_state_with_isolated_storage(); + let app = crate::test_support::build_test_router(Arc::clone(&state)); + let run_id = RunId::new(); + let mut graph = Graph::new("test"); + let mut work = Node::new("work"); + work.attrs + .insert("type".to_string(), AttrValue::String("agent".to_string())); + graph.nodes.insert("work".to_string(), work); + + create_durable_run_with_events(&state, run_id, &[ + workflow_event::Event::RunCreated { + run_id, + title: None, + settings: serde_json::to_value(fabro_types::WorkflowSettings::default()).unwrap(), + graph: serde_json::to_value(&graph).unwrap(), + workflow_source: None, + workflow_config: None, + labels: std::collections::BTreeMap::default(), + run_dir: String::new(), + source_directory: None, + workflow_slug: Some("test".to_string()), + automation: None, + db_prefix: None, + provenance: test_support::test_run_provenance(), + manifest_blob: None, + git: None, + fork_source_ref: None, + retried_from: None, + parent_id: None, + web_url: None, + }, + workflow_event::Event::RunStarting, + workflow_event::Event::RunRunning, + ]) + .await; + + // Legacy-shaped first execution without identity metadata. + append_scoped_stage_event( + &state, + run_id, + "work", + 1, + &workflow_event::Event::StageStarted { + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 1, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, + }, + ) + .await; + // Reexecution after cancel/resume: same graph visit, next ordinal. + append_scoped_stage_event( + &state, + run_id, + "work", + 2, + &workflow_event::Event::StageStarted { + graph_visit: Some(1), + resumed_from_stage_id: Some(fabro_types::StageId::new("work", 1)), + node_id: "work".to_string(), + name: "Work".to_string(), + index: 1, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, + }, + ) + .await; + + let response = app + .clone() + .oneshot( + Request::builder() + .method("GET") + .uri(api(&format!("/runs/{run_id}/stages"))) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + let body = response_json!(response, StatusCode::OK).await; + + let first = stage_entry(&body, "work@1"); + assert!( + first.get("graph_visit").is_none(), + "legacy stage should omit graph_visit" + ); + assert!( + first.get("resumed_from_stage_id").is_none(), + "legacy stage should omit resumed_from_stage_id" + ); + + let second = stage_entry(&body, "work@2"); + assert_eq!(second["visit"], 2); + assert_eq!(second["graph_visit"], 1); + assert_eq!(second["resumed_from_stage_id"], "work@1"); +} + /// `checkpoint.completed_nodes` records every visit, so a looped node appears /// once per re-entry. Billing must dedup so a retried node renders as one row /// and `runtime_secs` is summed across all visits exactly once. @@ -5471,6 +5589,8 @@ async fn run_billing_dedups_retried_nodes_and_sums_their_durations() { &run_store, &run_id, &workflow_event::Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "verify".to_string(), status: "running".to_string(), current_node: "verify".to_string(), @@ -5600,6 +5720,8 @@ async fn run_billing_sums_usage_across_retry_visits_and_uses_latest_model() { &run_store, &run_id, &workflow_event::Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "verify".to_string(), status: "running".to_string(), current_node: "verify".to_string(), @@ -5689,12 +5811,14 @@ async fn list_run_stages_shows_retrying_after_failed_event() { "work", 1, &workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 3, }, ) .await; @@ -5767,12 +5891,14 @@ async fn list_run_stages_shows_retrying_when_failed_will_retry() { "work", 1, &workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 3, }, ) .await; @@ -5824,12 +5950,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp workflow_event::Event::RunStarting, workflow_event::Event::RunRunning, workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 3, }, workflow_event::Event::StageFailed { node_id: "work".to_string(), @@ -5850,12 +5978,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp delay_ms: 0, }, workflow_event::Event::StageStarted { - node_id: "work".to_string(), - name: "Work".to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 2, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "work".to_string(), + name: "Work".to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 2, + max_attempts: 3, }, workflow_event::Event::StageCompleted { node_id: "work".to_string(), @@ -5910,12 +6040,14 @@ async fn run_billing_retried_node_then_succeeded_emits_one_row_with_final_attemp fn revisit_test_started(node_id: &str) -> workflow_event::Event { workflow_event::Event::StageStarted { - node_id: node_id.to_string(), - name: node_id.to_string(), - index: 0, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: node_id.to_string(), + name: node_id.to_string(), + index: 0, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, } } @@ -8140,12 +8272,14 @@ async fn get_run_stage_command_log_returns_scratch_slice() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandStarted { node_id: "script_node".to_string(), @@ -8207,12 +8341,14 @@ async fn get_run_stage_command_log_returns_cas_slice() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandCompleted { node_id: "script_node".to_string(), @@ -8271,12 +8407,14 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandCompleted { node_id: "script_node".to_string(), @@ -9462,12 +9600,14 @@ async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() { "review", 1, &workflow_event::Event::StageStarted { - node_id: "review".to_string(), - name: "Review".to_string(), - index: 0, - handler_type: "human".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "review".to_string(), + name: "Review".to_string(), + index: 0, + handler_type: "human".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -11408,6 +11548,8 @@ async fn resume_cancelled_run_with_checkpoint_transitions_to_runnable() { workflow_event::Event::RunStarting, workflow_event::Event::RunRunning, workflow_event::Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: "succeeded".to_string(), current_node: checkpoint.current_node.clone(), diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index b37903299..c512b0028 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -224,35 +224,46 @@ impl RunProjectionReducer for RunProjection { } EventBody::CheckpointCompleted(props) => { let checkpoint = checkpoint_from_props(props, ts); - if let Some(node_id) = stored.node_id.as_deref() { - let visit = checkpoint - .node_visits - .get(node_id) - .and_then(|visit| u32::try_from(*visit).ok()) - .unwrap_or(1); - if let Some(diff) = props.diff.clone() { - self.stage_entry(node_id, visit, first_event_seq(event.seq)) - .diff = Some(diff); + if let Some(stage_id) = stored.stage_id.clone() { + // Envelope-first: the diff and any skipped-stage synthesis + // attach to the exact execution recorded on the event. + // Historical `node_outcomes` must not create or collide + // with a newer execution ordinal. + apply_checkpoint_to_stage(self, &stage_id, props, &checkpoint, event.seq, ts); + } else { + // Legacy fallback for events without a stored stage id: + // resolve the visit from the checkpointed `node_visits` + // and synthesize skipped stages from historical outcomes. + if let Some(node_id) = stored.node_id.as_deref() { + let visit = checkpoint + .node_visits + .get(node_id) + .and_then(|visit| u32::try_from(*visit).ok()) + .unwrap_or(1); + if let Some(diff) = props.diff.clone() { + self.stage_entry(node_id, visit, first_event_seq(event.seq)) + .diff = Some(diff); + } } - } - for (node_id, outcome) in &checkpoint.node_outcomes { - if outcome.status != StageOutcome::Skipped { - continue; + for (node_id, outcome) in &checkpoint.node_outcomes { + if outcome.status != StageOutcome::Skipped { + continue; + } + let visit = checkpoint + .node_visits + .get(node_id) + .and_then(|visit| u32::try_from(*visit).ok()) + .unwrap_or(1); + if self + .stage(&fabro_types::StageId::new(node_id, visit)) + .is_some() + { + continue; + } + let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq)); + stage.completion = Some(stage_completion_from_outcome(outcome, ts)); + stage.state = StageState::Skipped; } - let visit = checkpoint - .node_visits - .get(node_id) - .and_then(|visit| u32::try_from(*visit).ok()) - .unwrap_or(1); - if self - .stage(&fabro_types::StageId::new(node_id, visit)) - .is_some() - { - continue; - } - let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq)); - stage.completion = Some(stage_completion_from_outcome(outcome, ts)); - stage.state = StageState::Skipped; } self.checkpoints.push(CheckpointRecord { seq: event.seq, @@ -337,6 +348,11 @@ impl RunProjectionReducer for RunProjection { let Some(stage_id) = stored.stage_id.as_ref() else { return Ok(()); }; + // A `stage.started` for a new `StageId` creates a new + // projection, so an older execution's terminal projection + // stays immutable. `begin_attempt` on an existing entry + // remains the compatibility path for automatic retries and + // legacy histories that repeat one `StageId`. let stage = self.stage_entry( stage_id.node_id(), stage_id.visit(), @@ -346,6 +362,14 @@ impl RunProjectionReducer for RunProjection { ts, StageHandler::from_handler_type(Some(&props.handler_type)), ); + if props.graph_visit.is_some() { + stage.graph_visit = props.graph_visit; + } + if props.resumed_from_stage_id.is_some() { + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } } EventBody::StageRetrying(_) => { let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { @@ -537,7 +561,7 @@ impl RunProjectionReducer for RunProjection { }; stage.parallel_results = Some(parallel_results); } - EventBody::ParallelBranchStarted(_) => { + EventBody::ParallelBranchStarted(props) => { // Branches bypass the engine's StageStarted/StageCompleted // lifecycle. Seed started_at so the branch stage drives a live // wall-clock timer while it runs (the entry is created Running). @@ -547,6 +571,14 @@ impl RunProjectionReducer for RunProjection { if stage.started_at.is_none() { stage.started_at = Some(ts); } + if props.graph_visit.is_some() { + stage.graph_visit = props.graph_visit; + } + if props.resumed_from_stage_id.is_some() { + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } stage.state = StageState::Running; } EventBody::ParallelBranchCompleted(props) => { @@ -951,6 +983,50 @@ fn stage_at_stored_or_current_visit<'a>( stage_at_current_visit(state, stored, seq) } +/// Apply a `checkpoint.completed` event to the exact execution named by the +/// envelope `StageId`: attach the checkpoint diff, and for a skipped +/// checkpoint finalize that execution as `Skipped`. A synthetic skipped +/// projection is created only for a first-attempt skip that never +/// materialized a stage; an existing non-terminal (e.g. `Retrying`) +/// projection keeps its identity and becomes terminal, while an older +/// terminal execution stays immutable. +fn apply_checkpoint_to_stage( + state: &mut RunProjection, + stage_id: &StageId, + props: &CheckpointCompletedProps, + checkpoint: &Checkpoint, + seq: u32, + ts: DateTime, +) { + let node_id = stage_id.node_id(); + let skipped_completion = checkpoint + .node_outcomes + .get(node_id) + .filter(|outcome| outcome.status == StageOutcome::Skipped) + .map(|outcome| stage_completion_from_outcome(outcome, ts)); + if props.diff.is_none() && skipped_completion.is_none() { + return; + } + + let is_new = state.stage(stage_id).is_none(); + let stage = state.stage_entry(node_id, stage_id.visit(), first_event_seq(seq)); + if is_new { + stage.graph_visit = props.graph_visit; + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } + if let Some(diff) = props.diff.clone() { + stage.diff = Some(diff); + } + if let Some(completion) = skipped_completion { + if !stage.state.is_terminal() { + stage.completion = Some(completion); + stage.state = StageState::Skipped; + } + } +} + fn stage_at_completed_visit<'a>( state: &'a mut RunProjection, stored: &RunEvent, @@ -2012,10 +2088,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2034,10 +2112,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2077,7 +2157,11 @@ mod tests { .apply_event(&test_stage_event_at( 3, "2026-04-07T12:00:00Z", - EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }), + EventBody::ParallelBranchStarted(ParallelBranchStartedProps { + index: 0, + graph_visit: None, + resumed_from_stage_id: None, + }), branch.clone(), )) .unwrap(); @@ -2115,7 +2199,11 @@ mod tests { state .apply_event(&test_stage_event( 3, - EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }), + EventBody::ParallelBranchStarted(ParallelBranchStartedProps { + index: 0, + graph_visit: None, + resumed_from_stage_id: None, + }), branch.clone(), )) .unwrap(); @@ -2148,10 +2236,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2404,10 +2494,12 @@ mod tests { .apply_event(&test_stage_event( 2, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2575,6 +2667,8 @@ mod tests { .apply_event(&test_event( 5, EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: None, + resumed_from_stage_id: None, status: "running".to_string(), current_node: "next".to_string(), completed_nodes: vec!["skip_me".to_string()], @@ -3831,10 +3925,12 @@ mod tests { fn started_props() -> StageStartedProps { StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 3, } } @@ -4334,6 +4430,418 @@ mod tests { ); } + #[test] + fn stage_started_records_execution_identity_metadata() { + let mut state = running_projection(); + let stage_id = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(StageId::new("work", 1)), + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!(stage.graph_visit, Some(1)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1))); + } + + #[test] + fn retry_stage_started_preserves_execution_identity_metadata() { + let mut state = running_projection(); + let stage_id = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(StageId::new("work", 1)), + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + // A legacy-shaped retry event for the same StageId omits the identity + // fields; the projection keeps the first attempt's metadata. + state + .apply_event(&test_stage_event( + 5, + EventBody::StageStarted(StageStartedProps { + attempt: 2, + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!(stage.graph_visit, Some(1)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1))); + } + + #[test] + fn cancelled_execution_stays_immutable_when_resumed_execution_starts() { + let mut state = running_projection(); + let first = StageId::new("work", 1); + let second = StageId::new("work", 2); + + state + .apply_event(&test_stage_event_at( + 4, + "2026-04-07T12:00:00Z", + EventBody::StageStarted(started_props()), + first.clone(), + )) + .unwrap(); + state + .apply_event(&test_raw_event_at( + 5, + "2026-04-07T12:00:05Z", + "run.failed", + &serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event( + 6, + "run.start_requested", + &json!({ "resume": true }), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event( + 7, + "run.runnable", + &json!({ "source": "start_requested" }), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event(8, "run.starting", &json!({}), None)) + .unwrap(); + state + .apply_event(&test_raw_event(9, "run.running", &json!({}), None)) + .unwrap(); + state + .apply_event(&test_stage_event( + 10, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(first.clone()), + ..started_props() + }), + second.clone(), + )) + .unwrap(); + + // The cancelled execution keeps its terminal projection untouched... + let cancelled = state.stage(&first).unwrap(); + assert_eq!(cancelled.state, StageState::Cancelled); + assert_eq!( + cancelled.timing, + Some(fabro_types::StageTiming::wall_only(5_000)) + ); + // ...while the reexecution runs as a distinct stage linked back to it. + let resumed = state.stage(&second).unwrap(); + assert_eq!(resumed.state, StageState::Running); + assert_eq!(resumed.graph_visit, Some(1)); + assert_eq!(resumed.resumed_from_stage_id, Some(first)); + } + + #[test] + fn run_failed_after_resume_preserves_earlier_terminal_executions() { + let mut state = running_projection(); + let done = StageId::new("verify", 1); + let active = StageId::new("work", 2); + + state + .apply_event(&test_stage_event_at( + 4, + "2026-04-07T12:00:00Z", + EventBody::StageStarted(started_props()), + done.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event_at( + 5, + "2026-04-07T12:00:03Z", + EventBody::StageCompleted(completed_props(3_000, StageOutcome::Succeeded)), + done.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event_at( + 6, + "2026-04-07T12:00:04Z", + EventBody::StageStarted(started_props()), + active.clone(), + )) + .unwrap(); + state + .apply_event(&test_raw_event_at( + 7, + "2026-04-07T12:00:09Z", + "run.failed", + &serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(), + None, + )) + .unwrap(); + + let terminal = state.stage(&done).unwrap(); + assert_eq!(terminal.state, StageState::Succeeded); + assert_eq!( + terminal.timing, + Some(fabro_types::StageTiming::wall_only(3_000)) + ); + assert_eq!(state.stage(&active).unwrap().state, StageState::Cancelled); + } + + #[test] + fn checkpoint_completed_targets_envelope_stage_id_after_ordinal_divergence() { + let mut state = running_projection(); + let execution = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + ..started_props() + }), + execution.clone(), + )) + .unwrap(); + // Checkpointed graph state still says visit 1 for `work`, and carries + // a historical skipped outcome for another node. Neither may create + // or mutate a projection at a stale ordinal. + state + .apply_event(&test_stage_event( + 5, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "success".to_string(), + current_node: "work".to_string(), + completed_nodes: vec!["old_skip".to_string(), "work".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "old_skip".to_string(), + Outcome::skipped("historical skip"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([ + ("work".to_string(), 1usize), + ("old_skip".to_string(), 1usize), + ]), + diff: Some("diff for work@2".to_string()), + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + assert_eq!( + state.stage(&execution).unwrap().diff.as_deref(), + Some("diff for work@2") + ); + assert!(state.stage(&StageId::new("work", 1)).is_none()); + assert!(state.stage(&StageId::new("old_skip", 1)).is_none()); + } + + #[test] + fn skipped_checkpoint_with_stage_id_creates_synthetic_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 3); + + // First-attempt StageStart-hook skip: no stage.started was emitted, + // the checkpoint is the first stage-scoped event for this execution. + state + .apply_event(&test_stage_event( + 4, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(2), + resumed_from_stage_id: Some(StageId::new("gate", 2)), + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("skipped by StageStart hook"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 2usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Skipped); + assert_eq!(stage.graph_visit, Some(2)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("gate", 2))); + assert_eq!( + stage.completion.as_ref().unwrap().notes.as_deref(), + Some("skipped by StageStart hook") + ); + } + + #[test] + fn skipped_checkpoint_finalizes_existing_retrying_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 1); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(started_props()), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 5, + EventBody::StageRetrying(StageRetryingProps { + index: 0, + attempt: 2, + max_attempts: 3, + delay_ms: 0, + }), + execution.clone(), + )) + .unwrap(); + assert_eq!(state.stage(&execution).unwrap().state, StageState::Retrying); + + // StageStart hook skipped the retry; the checkpoint finalizes the + // existing execution instead of allocating a new projection. + state + .apply_event(&test_stage_event( + 6, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("skipped on retry"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 1usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Skipped); + assert_eq!(stage.first_event_seq, first_event_seq(4)); + assert!(state.stage(&StageId::new("gate", 2)).is_none()); + } + + #[test] + fn skipped_checkpoint_never_reopens_an_older_terminal_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 1); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(started_props()), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 5, + EventBody::StageCompleted(completed_props(2_000, StageOutcome::Succeeded)), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 6, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("late skip"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 1usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Succeeded); + assert_eq!( + stage.completion.as_ref().unwrap().outcome, + StageOutcome::Succeeded + ); + } + + #[test] + fn legacy_stage_started_payload_without_identity_fields_deserializes() { + let event = test_raw_event( + 4, + "stage.started", + &json!({ + "index": 0, + "handler_type": "agent", + "attempt": 1, + "max_attempts": 3 + }), + Some("work"), + ); + + let EventBody::StageStarted(props) = &event.event.body else { + panic!("expected stage.started body"); + }; + assert_eq!(props.graph_visit, None); + assert_eq!(props.resumed_from_stage_id, None); + } + #[test] fn run_failed_non_cancelled_finalizes_running_stage_as_failed() { let mut state = running_projection(); diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index b02a8dc51..0df5db554 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -22,6 +22,13 @@ pub mod keys { pub const INTERNAL_FIDELITY: &str = "internal.fidelity"; pub const INTERNAL_THREAD_ID: &str = "internal.thread_id"; pub const INTERNAL_NODE_VISIT_COUNT: &str = "internal.node_visit_count"; + /// 1-based stage execution ordinal for the currently-executing node — the + /// numeric component of the external `StageId`. Runtime-only: reserved by + /// the lifecycle when a stage execution first becomes observable and + /// stripped from durable context snapshots, unlike + /// [`INTERNAL_NODE_VISIT_COUNT`], which remains the checkpointed graph + /// visit. + pub const INTERNAL_STAGE_EXECUTION_ORDINAL: &str = "internal.stage_execution_ordinal"; pub const INTERNAL_PARENT_PREAMBLE: &str = "internal.parent_preamble"; pub const INTERNAL_PARALLEL_GROUP_ID: &str = "internal.parallel_group_id"; pub const INTERNAL_PARALLEL_BRANCH_ID: &str = "internal.parallel_branch_id"; @@ -49,8 +56,11 @@ pub mod keys { pub const PARALLEL_FAN_IN_BEST_HEAD_SHA: &str = "parallel.fan_in.best_head_sha"; /// Runtime-only keys stripped from durable context projections. - pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = - &[CURRENT_PREAMBLE, INTERNAL_PARALLEL_BRANCH_PREAMBLES]; + pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = &[ + CURRENT_PREAMBLE, + INTERNAL_PARALLEL_BRANCH_PREAMBLES, + INTERNAL_STAGE_EXECUTION_ORDINAL, + ]; // --- Prefix constants (for filtering and dynamic keys) --- pub const GRAPH_PREFIX: &str = "graph."; diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index 2ba7e4661..292dd3b72 100644 --- a/lib/components/fabro-workflow/src/event/convert.rs +++ b/lib/components/fabro-workflow/src/event/convert.rs @@ -295,12 +295,16 @@ fn event_body_from_event(event: &Event) -> EventBody { handler_type, attempt, max_attempts, + graph_visit, + resumed_from_stage_id, .. } => EventBody::StageStarted(fabro_types::StageStartedProps { - index: *index, + index: *index, handler_type: handler_type.clone(), - attempt: *attempt, + attempt: *attempt, max_attempts: *max_attempts, + graph_visit: *graph_visit, + resumed_from_stage_id: resumed_from_stage_id.clone(), }), Event::StageCompleted { index, @@ -378,11 +382,16 @@ fn event_body_from_event(event: &Event) -> EventBody { branch_count: *branch_count, join_policy: join_policy.clone(), }), - Event::ParallelBranchStarted { index, .. } => { - EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps { - index: *index, - }) - } + Event::ParallelBranchStarted { + index, + graph_visit, + resumed_from_stage_id, + .. + } => EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps { + index: *index, + graph_visit: *graph_visit, + resumed_from_stage_id: resumed_from_stage_id.clone(), + }), Event::ParallelBranchCompleted { index, duration_ms, @@ -480,6 +489,8 @@ fn event_body_from_event(event: &Event) -> EventBody { node_visits, diff, diff_summary, + graph_visit, + resumed_from_stage_id, .. } => EventBody::CheckpointCompleted(fabro_types::CheckpointCompletedProps { status: status.clone(), @@ -495,6 +506,8 @@ fn event_body_from_event(event: &Event) -> EventBody { node_visits: node_visits.clone(), diff: diff.clone(), diff_summary: *diff_summary, + graph_visit: *graph_visit, + resumed_from_stage_id: resumed_from_stage_id.clone(), }), Event::CheckpointFailed { error, @@ -1693,12 +1706,14 @@ mod tests { let stored = to_run_event_at( &fixtures::RUN_1, &Event::StageStarted { - node_id: "review".to_string(), - name: "review".to_string(), - index: 1, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "review".to_string(), + name: "review".to_string(), + index: 1, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }, Utc::now(), Some(&StageScope { @@ -1730,10 +1745,12 @@ mod tests { #[test] fn parallel_branch_started_populates_group_and_branch_ids() { let stored = to_run_event(&fixtures::RUN_1, &Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fanout", 2), - parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1), - branch: "review".to_string(), - index: 1, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fanout", 2), + parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1), + branch: "review".to_string(), + index: 1, }); assert_eq!(stored.parallel_group_id, Some(StageId::new("fanout", 2))); assert_eq!( diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 0b5afb1c7..84822b4fb 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -244,12 +244,22 @@ pub enum Event { exec_output_tail: Option, }, StageStarted { - node_id: String, - name: String, - index: usize, - handler_type: String, - attempt: usize, - max_attempts: usize, + node_id: String, + name: String, + index: usize, + handler_type: String, + attempt: usize, + max_attempts: usize, + /// Graph visit that produced this stage execution. Diverges from the + /// envelope `StageId` ordinal when a cancelled or crashed invocation + /// is reexecuted after resume. + #[serde(default, skip_serializing_if = "Option::is_none")] + graph_visit: Option, + /// Prior execution this one resumes from, for the first execution + /// reserved after a resume when the node had an observable + /// post-checkpoint execution. + #[serde(default, skip_serializing_if = "Option::is_none")] + resumed_from_stage_id: Option, }, StageCompleted { node_id: String, @@ -307,10 +317,14 @@ pub enum Event { join_policy: String, }, ParallelBranchStarted { - parallel_group_id: StageId, - parallel_branch_id: ParallelBranchId, - branch: String, - index: usize, + parallel_group_id: StageId, + parallel_branch_id: ParallelBranchId, + branch: String, + index: usize, + #[serde(default, skip_serializing_if = "Option::is_none")] + graph_visit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + resumed_from_stage_id: Option, }, ParallelBranchCompleted { parallel_group_id: StageId, @@ -396,6 +410,12 @@ pub enum Event { diff: Option, #[serde(default, skip_serializing_if = "Option::is_none")] diff_summary: Option, + /// 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + resumed_from_stage_id: Option, }, CheckpointFailed { node_id: String, diff --git a/lib/components/fabro-workflow/src/event/names.rs b/lib/components/fabro-workflow/src/event/names.rs index 39a2bf5e3..ee071d4af 100644 --- a/lib/components/fabro-workflow/src/event/names.rs +++ b/lib/components/fabro-workflow/src/event/names.rs @@ -170,10 +170,12 @@ mod tests { fn event_name_matches_new_dot_notation() { assert_eq!( event_name(&Event::ParallelBranchStarted { - parallel_group_id: StageId::new("plan", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0), - branch: "fork".to_string(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("plan", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0), + branch: "fork".to_string(), + index: 0, }), "parallel.branch.started" ); diff --git a/lib/components/fabro-workflow/src/git.rs b/lib/components/fabro-workflow/src/git.rs index 8fa23a208..df5186726 100644 --- a/lib/components/fabro-workflow/src/git.rs +++ b/lib/components/fabro-workflow/src/git.rs @@ -555,6 +555,8 @@ mod tests { .await .unwrap(); append_event(&run, &fixtures::RUN_1, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "work".into(), status: "succeeded".into(), current_node: "work".into(), diff --git a/lib/components/fabro-workflow/src/handler/manager_loop.rs b/lib/components/fabro-workflow/src/handler/manager_loop.rs index d16291b68..bc3637c52 100644 --- a/lib/components/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/components/fabro-workflow/src/handler/manager_loop.rs @@ -16,10 +16,10 @@ use crate::artifact_upload::ArtifactSink; use crate::condition::evaluate_condition; use crate::context::{Context, WorkflowContext, keys}; use crate::error::Error; +use crate::event::StageScope; use crate::operations::{ValidateInput, WorkflowInput, validate}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::pipeline::types::Initialized; -use crate::run_dir::visit_from_context; use crate::run_options::RunOptions; use crate::static_reference::{ReferenceKind, validate_static_reference}; use crate::{ManifestPath, pipeline}; @@ -197,8 +197,10 @@ impl Handler for SubWorkflowHandler { } }; - // Build child RunOptions - let visit = visit_from_context(context) as u64; + // Build child RunOptions. The stage directory follows the execution + // ordinal so a reexecuted manager loop keeps the cancelled + // invocation's child logs intact. + let visit = u64::from(StageScope::for_handler(context, &node.id).visit); let child_logs = run_dir.join(format!("stages/{}@{visit}/child", node.id)); let _ = fs::create_dir_all(&child_logs).await; diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 32e4a8b10..47f929ad3 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -17,10 +17,10 @@ use crate::git::sanitize_ref_component; use crate::hook_context::set_hook_node; use crate::millis_u64; use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt, StageOutcome}; -use crate::run_dir::visit_from_context; use crate::sandbox_git::{ GIT_REMOTE, checked_git_checkpoint, git_merge_ff_only, git_remove_worktree, }; +use crate::stage_execution::StageExecution; /// Fans out execution to multiple branches concurrently. /// Each branch gets an isolated context clone and runs independently. @@ -161,6 +161,9 @@ impl Handler for ParallelHandler { branch_context: Context, sandbox: Arc, worktree_path: Option, + /// Child stage execution reserved through the run's shared + /// tracker, so a resumed fan-out gets fresh branch identities. + execution: StageExecution, } let parallel_start = Instant::now(); @@ -264,6 +267,19 @@ impl Handler for ParallelHandler { parallel_group_id.clone(), u32::try_from(branch_index).unwrap_or(u32::MAX), ); + // Reserve the child's stage execution through the shared tracker + // and seed the branch context with its explicit stage scope, so + // branch lifecycle events and nested handler events agree on the + // child's identity instead of inheriting the fork's. + let execution = services.run.stage_executions.reserve(&target_id, 1); + branch_context.set( + keys::CURRENT_NODE, + serde_json::Value::String(target_id.clone()), + ); + branch_context.set( + keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.ordinal), + ); branch_context.set( keys::INTERNAL_PARALLEL_GROUP_ID, serde_json::Value::String(parallel_group_id.to_string()), @@ -294,12 +310,14 @@ impl Handler for ParallelHandler { (&git_state, &base_sha) { let branch_key = &target_id; - let visit = visit_from_context(&branch_context); + // `pass{N}` derives from the parent's execution ordinal so a + // resumed fan-out does not recreate the cancelled attempt's + // branch names. let branch_name = format!( "fabro/run/parallel/{}/{}/pass{}/{}", gs.run_id, sanitize_ref_component(&node.id), - visit, + parallel_stage_scope.visit, sanitize_ref_component(branch_key), ); @@ -345,6 +363,7 @@ impl Handler for ParallelHandler { branch_context, sandbox: branch_sandbox, worktree_path, + execution, }); } @@ -376,7 +395,7 @@ impl Handler for ParallelHandler { let group_id = parallel_group_id.clone(); let branch_scope = StageScope::for_parallel_branch( setup.target_id.clone(), - 1, + setup.execution.ordinal, group_id.clone(), setup.parallel_branch_id.clone(), ); @@ -389,10 +408,12 @@ impl Handler for ParallelHandler { parent_run.emitter.emit_scoped( &Event::ParallelBranchStarted { - parallel_group_id: group_id.clone(), - parallel_branch_id: setup.parallel_branch_id.clone(), - branch: setup.target_id.clone(), - index: setup.branch_index, + parallel_group_id: group_id.clone(), + parallel_branch_id: setup.parallel_branch_id.clone(), + branch: setup.target_id.clone(), + index: setup.branch_index, + graph_visit: Some(setup.execution.graph_visit), + resumed_from_stage_id: setup.execution.resumed_from.clone(), }, &branch_scope, ); diff --git a/lib/components/fabro-workflow/src/lib.rs b/lib/components/fabro-workflow/src/lib.rs index b4d3cc727..2bb1c4977 100644 --- a/lib/components/fabro-workflow/src/lib.rs +++ b/lib/components/fabro-workflow/src/lib.rs @@ -329,6 +329,7 @@ pub mod runtime_store; pub mod sandbox_git; pub(crate) mod sandbox_git_runtime; pub mod services; +pub mod stage_execution; mod stage_scope; pub mod static_reference; pub mod steering_hub; diff --git a/lib/components/fabro-workflow/src/lifecycle/artifact.rs b/lib/components/fabro-workflow/src/lifecycle/artifact.rs index 6d8b8de75..f4d0e4d0e 100644 --- a/lib/components/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/components/fabro-workflow/src/lifecycle/artifact.rs @@ -23,6 +23,7 @@ use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::lifecycle::event::{stage_scope_for, stage_visit}; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; +use crate::stage_execution::StageExecutionTracker; type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; @@ -46,6 +47,8 @@ pub(crate) struct ArtifactLifecycle { /// Per-attempt state: epoch seconds when the attempt started. attempt_start_epoch: std::sync::Mutex>, captured_artifacts: std::sync::Mutex>, + /// 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, artifact_sink: Option, + stage_executions: StageExecutionTracker, ) -> Self { Self { sandbox, @@ -66,6 +70,7 @@ impl ArtifactLifecycle { artifact_sink, attempt_start_epoch: std::sync::Mutex::new(None), captured_artifacts: std::sync::Mutex::new(HashSet::new()), + stage_executions, } } } @@ -120,7 +125,12 @@ impl RunLifecycle for ArtifactLifecycle { .expect("artifact mutex should not be poisoned: no code panics while holding this lock") .unwrap_or(0.0); let node_id = ctx.node.id(); - let visit = stage_visit(state, node_id); + // Artifact identity follows the stage execution ordinal so a resumed + // reexecution stores its captures under the new `StageId`. + let visit = self.stage_executions.active(node_id).map_or_else( + || stage_visit(state, node_id), + |execution| execution.ordinal, + ); let node_slug = if visit <= 1 { node_id.to_string() } else { @@ -162,7 +172,7 @@ impl RunLifecycle for ArtifactLifecycle { return Ok(()); } self.record_captured_assets(&new_assets); - let scope = stage_scope_for(state, node_id); + let scope = stage_scope_for(&self.stage_executions, state, node_id); for asset in &new_assets { self.emitter.emit_scoped( &Event::ArtifactCaptured { diff --git a/lib/components/fabro-workflow/src/lifecycle/event.rs b/lib/components/fabro-workflow/src/lifecycle/event.rs index f547f5fe5..0770323fc 100644 --- a/lib/components/fabro-workflow/src/lifecycle/event.rs +++ b/lib/components/fabro-workflow/src/lifecycle/event.rs @@ -18,6 +18,7 @@ use crate::context::{Context, WorkflowContext}; use crate::event::{Emitter, Event, StageScope}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageOutcome}; +use crate::stage_execution::StageExecutionTracker; use crate::{artifact, context}; type WfRunState = ExecutionState>; @@ -46,6 +47,8 @@ pub(crate) struct EventLifecycle { /// EventLifecycle when emitting CheckpointCompleted). pub checkpoint_git_result: Arc>>, pub circuit_breaker: Arc, + /// Run-scoped stage execution allocator shared with `RunServices`. + pub stage_executions: StageExecutionTracker, } fn snapshot_failure_signatures( @@ -107,11 +110,22 @@ pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { u32::try_from(visits).unwrap_or(u32::MAX) } -pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope { +/// Build the emission scope for a node from its active stage execution. +/// Falls back to the graph visit for direct unit-test call sites that emit +/// without a reservation; the two are equal for a first execution. +pub(crate) fn stage_scope_for( + stage_executions: &StageExecutionTracker, + state: &WfRunState, + node_id: &str, +) -> StageScope { + let visit = stage_executions.active(node_id).map_or_else( + || stage_visit(state, node_id), + |execution| execution.ordinal, + ); StageScope { - node_id: node_id.to_string(), - visit: stage_visit(state, node_id), - parallel_group_id: state.context.parallel_group_id(), + node_id: node_id.to_string(), + visit, + parallel_group_id: state.context.parallel_group_id(), parallel_branch_id: state.context.parallel_branch_id(), } } @@ -160,17 +174,24 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; - let scope = stage_scope_for(state, &gv.id); + // Terminal nodes bypass `before_node`/`before_attempt`, so their + // synthetic paired events reserve an execution here. + let execution = self + .stage_executions + .reserve(&gv.id, stage_visit(state, &gv.id)); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); self.emitter.emit_scoped( &Event::StageStarted { - node_id: gv.id.clone(), - name: gv.label().to_string(), - index: stage_index, - handler_type: gv.handler_type().unwrap_or_default().to_string(), - attempt: 1, - max_attempts: 1, + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: 1, + max_attempts: 1, + graph_visit: Some(execution.graph_visit), + resumed_from_stage_id: execution.resumed_from, }, &scope, ); @@ -210,15 +231,21 @@ impl RunLifecycle for EventLifecycle { state: &WfRunState, ) -> CoreResult>> { let gv = ctx.node.inner(); - let scope = stage_scope_for(state, &gv.id); + let execution = self.stage_executions.active(&gv.id); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); + let graph_visit = execution + .as_ref() + .map_or_else(|| stage_visit(state, &gv.id), |e| e.graph_visit); self.emitter.emit_scoped( &Event::StageStarted { - node_id: gv.id.clone(), - name: gv.label().to_string(), - index: state.stage_index, - handler_type: gv.handler_type().unwrap_or_default().to_string(), - attempt: ctx.attempt as usize, - max_attempts: ctx.max_attempts as usize, + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: state.stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: ctx.attempt as usize, + max_attempts: ctx.max_attempts as usize, + graph_visit: Some(graph_visit), + resumed_from_stage_id: execution.and_then(|e| e.resumed_from), }, &scope, ); @@ -234,7 +261,7 @@ impl RunLifecycle for EventLifecycle { let gv = ctx.node.inner(); let outcome = &ctx.result.outcome; let stage_index = state.stage_index; - let scope = stage_scope_for(state, &gv.id); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); let timing = node_result_timing(ctx.result); let failure = outcome.failure.clone().unwrap_or_else(|| { @@ -283,7 +310,7 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; - let scope = stage_scope_for(state, &gv.id); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); let timing = node_result_timing(result); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); @@ -400,7 +427,11 @@ impl RunLifecycle for EventLifecycle { node_outcomes.insert(node.id().to_string(), result.outcome.clone()); artifact::normalize_durable_outcomes(&mut node_outcomes); - let scope = stage_scope_for(state, node.id()); + let execution = self.stage_executions.active(node.id()); + let scope = stage_scope_for(&self.stage_executions, state, node.id()); + let graph_visit = execution + .as_ref() + .map_or_else(|| stage_visit(state, node.id()), |e| e.graph_visit); self.emitter.emit_scoped( &Event::CheckpointCompleted { node_id: node.id().to_string(), @@ -425,6 +456,8 @@ impl RunLifecycle for EventLifecycle { .collect::>(), diff, diff_summary, + graph_visit: Some(graph_visit), + resumed_from_stage_id: execution.and_then(|e| e.resumed_from), }, &scope, ); diff --git a/lib/components/fabro-workflow/src/lifecycle/git.rs b/lib/components/fabro-workflow/src/lifecycle/git.rs index a1aab9bc7..4db74c904 100644 --- a/lib/components/fabro-workflow/src/lifecycle/git.rs +++ b/lib/components/fabro-workflow/src/lifecycle/git.rs @@ -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>; type WfNodeResult = NodeResult>; @@ -88,6 +89,8 @@ pub(crate) struct GitLifecycle { // Cross-lifecycle data (shared with EventLifecycle) pub checkpoint_git_result: Arc>>, pub last_git_sha: Arc>>, + /// Run-scoped stage execution allocator shared with `RunServices`. + pub stage_executions: StageExecutionTracker, } #[async_trait] @@ -197,7 +200,7 @@ impl RunLifecycle 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 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, ) -> 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(), diff --git a/lib/components/fabro-workflow/src/lifecycle/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index 2db1e8d75..e3facc0f8 100644 --- a/lib/components/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/components/fabro-workflow/src/lifecycle/mod.rs @@ -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>; type WfNodeResult = NodeResult>; @@ -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, run_id: RunId, @@ -97,6 +100,7 @@ impl WorkflowLifecycle { is_resume: bool, on_node: crate::OnNodeCallback, run_control: Option>, + stage_executions: StageExecutionTracker, ) -> Self { let restarted_from: Arc>> = 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 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 for WorkflowLifecycle { NodeDecision::Continue => {} decision => return Ok(decision), } + // Reserve the stage execution once per handler invocation: the first + // attempt allocates the ordinal and automatic retries reuse it. + let node_id = ctx.node.id(); + let execution = self + .stage_executions + .ensure(node_id, event::stage_visit(state, node_id)); + state.context.set( + context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.ordinal), + ); // Event emission self.event.before_attempt(ctx, state).await?; // Record epoch AFTER hook+event (engine.rs:968→1006) @@ -410,6 +437,17 @@ impl RunLifecycle for WorkflowLifecycle { next_node_id: Option<&str>, state: &WfRunState, ) -> CoreResult<()> { + // A StageStart hook can skip before any attempt reserved an execution + // scope. Ensure one exists so Git metadata-snapshot events and the + // `checkpoint.completed` envelope attach to a concrete execution; + // an existing reservation from the attempt path is reused as-is. + let execution = self + .stage_executions + .ensure(node.id(), event::stage_visit(state, node.id())); + state.context.set( + context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.ordinal), + ); self.git .on_checkpoint(node, result, next_node_id, state) .await?; diff --git a/lib/components/fabro-workflow/src/operations/fork.rs b/lib/components/fabro-workflow/src/operations/fork.rs index d2353bdfc..559df2746 100644 --- a/lib/components/fabro-workflow/src/operations/fork.rs +++ b/lib/components/fabro-workflow/src/operations/fork.rs @@ -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(), diff --git a/lib/components/fabro-workflow/src/operations/resume.rs b/lib/components/fabro-workflow/src/operations/resume.rs index beaad87a9..fd2b4d62b 100644 --- a/lib/components/fabro-workflow/src/operations/resume.rs +++ b/lib/components/fabro-workflow/src/operations/resume.rs @@ -5,6 +5,7 @@ use crate::error::Error; use crate::event::{Event, append_event_to_sink}; use crate::outcome::StageOutcome; use crate::run_status::RunStatus; +use crate::stage_execution::StageExecutionSeed; /// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. pub async fn resume(run_dir: &Path, services: StartServices) -> Result { @@ -32,10 +33,15 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result Result, + stage_executions: StageExecutionSeed, services: StartServices, ) -> Result { let cancel_token = services.cancel_token.clone(); @@ -261,7 +269,7 @@ pub(super) async fn execute_persisted_run( cancel_token, ); let run_start = Instant::now(); - let started = Box::pin(session.run(persisted, checkpoint)).await; + let started = Box::pin(session.run(persisted, checkpoint, stage_executions)).await; match started { Ok(started) => { @@ -797,6 +805,7 @@ impl RunSession { self, persisted: Persisted, checkpoint: Option, + stage_executions: StageExecutionSeed, ) -> Result { let on_node = self.on_node.clone(); @@ -879,6 +888,7 @@ impl RunSession { run_control: self.run_control, checkpoint, seed_context: self.seed_context, + stage_executions, fabro_run_tools: self.fabro_run_tools, }; let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?; @@ -2125,6 +2135,8 @@ reasoning = false { injected.store(true, Ordering::SeqCst); emitter_for_injection.emit(&Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "start".to_string(), status: "succeeded".to_string(), current_node: "start".to_string(), @@ -2508,6 +2520,8 @@ reasoning = false &store.open_run(&fixtures::RUN_1).await.unwrap(), &services.run_id, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: checkpoint .node_outcomes @@ -2606,6 +2620,8 @@ reasoning = false }; let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); crate::event::append_event(&run_store, &fixtures::RUN_1, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: "succeeded".to_string(), current_node: checkpoint.current_node.clone(), diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index 0be068e88..5a35d59fa 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -91,6 +91,7 @@ pub async fn execute(init: Initialized) -> Executed { checkpoint.is_some(), on_node, run_control, + engine.run.stage_executions.clone(), ); if let Some(ref cp) = checkpoint { diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 400050896..22b6233cb 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -256,6 +256,7 @@ async fn execute_test_run_with_options( let initialized = initialize( persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value), InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -316,6 +317,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { let initialized = initialize( persisted_workflow(graph, source, &run_dir, test_run_id("run-test")), InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: test_emitter_arc("run-test"), @@ -375,6 +377,186 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { ); } +#[tokio::test] +async fn resumed_in_flight_node_starts_a_new_stage_execution() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + + // start -> work -> exit; `work` resolves to the default (dry-run) handler. + let mut graph = Graph::new("resume_identity"); + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + graph.nodes.insert("start".to_string(), start); + graph.nodes.insert("work".to_string(), Node::new("work")); + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + graph.nodes.insert("exit".to_string(), exit); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let run_options = test_run_options(&run_dir, "resume-identity"); + let run_id = run_options.run_id; + let run_store = test_run_store(&run_id).await; + seed_created_and_starting(&run_store, &run_options, &graph).await; + // Resume reconnects to the previously recorded sandbox. + append_event(&run_store, &run_id, &Event::SandboxInitialized { + working_directory: std::env::current_dir().unwrap().display().to_string(), + provider: fabro_types::SandboxProviderKind::Local, + id: "local".to_string(), + image: None, + snapshot: None, + repo_cloned: None, + clone_origin_url: None, + clone_branch: None, + workspace_root: None, + repos_root: None, + primary_repo_path: None, + primary_repo_link: None, + }) + .await + .unwrap(); + let emitter = test_emitter_arc("resume-identity"); + let events: Arc>> = Arc::default(); + { + let events = Arc::clone(&events); + emitter.on_event(move |event| { + events + .lock() + .expect("event capture mutex should not be poisoned") + .push(event.clone()); + }); + } + + // Simulate resuming after `work@1` was cancelled mid-flight: the selected + // checkpoint predates `work`, while the allocator seed carries the + // projection-observed high-water mark and provenance link. + let checkpoint = crate::records::Checkpoint { + timestamp: chrono::Utc::now(), + current_node: "start".to_string(), + completed_nodes: vec!["start".to_string()], + node_retries: HashMap::new(), + context_values: HashMap::new(), + node_outcomes: HashMap::new(), + next_node_id: Some("work".to_string()), + git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), + node_visits: HashMap::from([("start".to_string(), 1usize)]), + }; + let seed = crate::stage_execution::StageExecutionSeed { + high_water: HashMap::from([("work".to_string(), 1)]), + resumed_from: HashMap::from([("work".to_string(), fabro_types::StageId::new("work", 1))]), + }; + + let initialized = initialize( + persisted_workflow(graph, String::new(), &run_dir, run_id), + InitOptions { + stage_executions: seed, + run_store: run_store.into(), + dry_run: false, + emitter: emitter.clone(), + sandbox: SandboxSpec::Local { + working_directory: std::env::current_dir().unwrap(), + }, + llm: LlmSpec { + model: "test-model".to_string(), + provider_id: fabro_model::ProviderId::anthropic(), + fallback_chain: Vec::new(), + mcp_servers: Vec::new(), + model_controls: RunModelControls::default(), + dry_run: true, + }, + interviewer: Arc::new(AutoApproveInterviewer::engine()), + steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())), + catalog: test_catalog(), + lifecycle: LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + }, + run_options, + workflow_path: None, + workflow_bundle: None, + hooks: HookSettings { hooks: vec![] }, + sandbox_env: SandboxEnvSpec { + toml_env: HashMap::new(), + github_permissions: None, + origin_url: None, + }, + vault: None, + git: None, + run_control: None, + registry_override: Some(Arc::new(make_registry())), + artifact_sink: None, + checkpoint: Some(checkpoint), + seed_context: None, + fabro_run_tools: None, + }, + ) + .await + .unwrap(); + + let executed = execute(initialized).await; + assert_eq!(executed.outcome.unwrap().status, StageOutcome::Succeeded); + + let events = events + .lock() + .expect("event capture mutex should not be poisoned"); + let work_started = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::StageStarted(_)) + && event.node_id.as_deref() == Some("work") + }) + .expect("resumed run should emit stage.started for work"); + // The reexecution owns a fresh StageId while the graph visit stays at 1. + assert_eq!( + work_started.stage_id, + Some(fabro_types::StageId::new("work", 2)) + ); + let fabro_types::EventBody::StageStarted(props) = &work_started.body else { + panic!("expected stage.started body"); + }; + assert_eq!(props.graph_visit, Some(1)); + assert_eq!( + props.resumed_from_stage_id, + Some(fabro_types::StageId::new("work", 1)) + ); + + // Every later stage-scoped event from this invocation carries the same + // execution id, including the checkpoint envelope. + let work_checkpoint = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::CheckpointCompleted(_)) + && event.node_id.as_deref() == Some("work") + }) + .expect("resumed run should checkpoint work"); + assert_eq!( + work_checkpoint.stage_id, + Some(fabro_types::StageId::new("work", 2)) + ); + + // A node without a prior observable execution starts at ordinal 1. + let exit_started = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::StageStarted(_)) + && event.node_id.as_deref() == Some("exit") + }) + .expect("terminal node should emit its synthetic stage.started"); + assert_eq!( + exit_started.stage_id, + Some(fabro_types::StageId::new("exit", 1)) + ); +} + async fn run_with_lifecycle( registry: HandlerRegistry, emitter: Arc, @@ -391,6 +573,7 @@ async fn run_with_lifecycle( let initialized = initialize( persisted_workflow(graph.clone(), String::new(), &run_dir, run_id), InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), diff --git a/lib/components/fabro-workflow/src/pipeline/finalize.rs b/lib/components/fabro-workflow/src/pipeline/finalize.rs index 9fdcad717..949578880 100644 --- a/lib/components/fabro-workflow/src/pipeline/finalize.rs +++ b/lib/components/fabro-workflow/src/pipeline/finalize.rs @@ -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"), diff --git a/lib/components/fabro-workflow/src/pipeline/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index e20b2fb61..c13c31b3c 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -33,6 +33,7 @@ use crate::sandbox_git_runtime::SandboxGitRuntime; use crate::services::{ EngineServices, FabroRunToolServices, RunLocations, RunServices, WorkflowToolEnvProvider, }; +use crate::stage_execution::StageExecutionTracker; use crate::steering_hub::SteeringHub; type BuiltSandboxEnv = (HashMap, Option>); @@ -619,6 +620,7 @@ pub async fn initialize( sandbox_git, metadata_runtime, metadata_writer, + StageExecutionTracker::seeded(options.stage_executions), ); let engine = Arc::new(EngineServices { run: Arc::clone(&run_services), @@ -825,6 +827,7 @@ mod tests { }); let result = initialize(persisted, InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); @@ -906,6 +909,7 @@ mod tests { let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let initialized = initialize(persisted, InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); @@ -1131,6 +1135,7 @@ mod tests { let store = memory_store(); let run_store = store.create_run(&test_run_id()).await.unwrap(); let initialized = initialize(test_persisted(graph, source, &run_dir), InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -1226,6 +1231,7 @@ mod tests { store_logger.register(&emitter); let initialized = initialize(persisted, InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -1364,6 +1370,7 @@ mod tests { let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let result = initialize(persisted, InitOptions { + stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); diff --git a/lib/components/fabro-workflow/src/pipeline/types.rs b/lib/components/fabro-workflow/src/pipeline/types.rs index 2b881a24b..97b852b30 100644 --- a/lib/components/fabro-workflow/src/pipeline/types.rs +++ b/lib/components/fabro-workflow/src/pipeline/types.rs @@ -26,6 +26,7 @@ use crate::run_control::RunControlState; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::runtime_store::RunStoreHandle; use crate::services::{EngineServices, FabroRunToolServices, RunServices}; +use crate::stage_execution::StageExecutionSeed; use crate::steering_hub::SteeringHub; use crate::transforms::{RenderMode, Transform}; use crate::workflow_bundle::WorkflowBundle; @@ -270,6 +271,10 @@ pub struct InitOptions { pub run_control: Option>, pub checkpoint: Option, pub seed_context: Option, + /// Allocator seed for stage execution ordinals. Empty for a fresh run; + /// resume passes projection-derived high-water marks and provenance so a + /// reexecuted in-flight node gets a new `StageId` ordinal. + pub stage_executions: StageExecutionSeed, pub fabro_run_tools: Option, } diff --git a/lib/components/fabro-workflow/src/services.rs b/lib/components/fabro-workflow/src/services.rs index 44e0eca21..4140d0c48 100644 --- a/lib/components/fabro-workflow/src/services.rs +++ b/lib/components/fabro-workflow/src/services.rs @@ -22,6 +22,7 @@ use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle}; use crate::runtime_store::RunStoreHandle; use crate::sandbox_git::GitState; use crate::sandbox_git_runtime::SandboxGitRuntime; +use crate::stage_execution::StageExecutionTracker; use crate::workflow_bundle::WorkflowBundle; #[derive(Clone, Debug, PartialEq, Eq)] @@ -107,6 +108,9 @@ pub struct RunServices { pub(crate) metadata_runtime: Arc, pub(crate) metadata_writer: Option, pub(crate) interview_blocker: Arc, + /// Run-scoped stage execution allocator, shared between the core + /// lifecycle and direct-dispatch handlers such as parallel branches. + pub(crate) stage_executions: StageExecutionTracker, } impl RunServices { @@ -125,6 +129,7 @@ impl RunServices { sandbox_git: Arc, metadata_runtime: Arc, metadata_writer: Option, + stage_executions: StageExecutionTracker, ) -> Arc { Arc::new(Self { run_store, @@ -141,6 +146,7 @@ impl RunServices { metadata_runtime, metadata_writer, interview_blocker: Arc::new(RunInterviewBlocker::new()), + stage_executions, }) } @@ -340,6 +346,7 @@ impl EngineServices { Arc::new(SandboxGitRuntime::new()), Arc::new(RunMetadataRuntime::new()), None, + StageExecutionTracker::default(), ), registry: Arc::new(HandlerRegistry::new(Box::new(start::StartHandler))), interviewer: Arc::new(fabro_interview::AutoApproveInterviewer::engine()), diff --git a/lib/components/fabro-workflow/src/stage_execution.rs b/lib/components/fabro-workflow/src/stage_execution.rs new file mode 100644 index 000000000..62604695c --- /dev/null +++ b/lib/components/fabro-workflow/src/stage_execution.rs @@ -0,0 +1,291 @@ +//! Run-scoped stage execution identity. +//! +//! A *stage execution* is one top-level handler invocation of a node that +//! became observable within a run. Its 1-based ordinal is the numeric +//! component of the external `StageId` (`node_id@N`). The ordinal is distinct +//! from the *graph visit* (how many times workflow control entered the node, +//! which drives `max_visits` and checkpoints) and from the *handler attempt* +//! (automatic retries inside one execution). +//! +//! The tracker is deliberately not checkpointed: its durable source of truth +//! is the append-only stage event history. On resume it is seeded from the +//! run projection's per-node maxima, so a reexecuted in-flight node allocates +//! the next unused ordinal instead of mutating the cancelled execution. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use fabro_types::{RunProjection, StageId}; + +/// One reserved stage execution: the identity of a single resumable handler +/// invocation of a node. +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct StageExecution { + /// 1-based execution ordinal; becomes the `@N` in the external `StageId`. + pub ordinal: u32, + /// Graph visit that produced this execution. + pub graph_visit: u32, + /// Prior execution this one resumes from, when the node had an observable + /// post-checkpoint execution before the run was interrupted. + pub resumed_from: Option, +} + +/// Seed data for the [`StageExecutionTracker`], derived from the run +/// projection when a run is resumed. A fresh run uses the default (empty) +/// seed; new run IDs own a new ordinal sequence. +#[derive(Clone, Debug, Default)] +pub struct StageExecutionSeed { + /// Highest execution ordinal already observable per node. + pub high_water: HashMap, + /// Latest post-checkpoint execution per node; the next reservation for + /// that node links back to it via `resumed_from_stage_id`. + pub resumed_from: HashMap, +} + +impl StageExecutionSeed { + /// Build the seed from the run projection at resume time. + /// + /// `checkpoint_seq` is the event sequence number of the selected + /// checkpoint. Only stages that first became observable *after* that + /// checkpoint are eligible provenance targets: an older execution with the + /// same node ID completed before the checkpoint and is not what the + /// resumed invocation continues from. + #[must_use] + pub fn from_projection(projection: &RunProjection, checkpoint_seq: u32) -> Self { + let mut high_water: HashMap = HashMap::new(); + let mut resumed_from: HashMap = HashMap::new(); + // `iter_stages` yields chronological `first_event_seq` order, so a + // later insert per node retains the latest post-checkpoint execution. + for (stage_id, stage) in projection.iter_stages() { + let node_id = stage_id.node_id(); + let entry = high_water.entry(node_id.to_string()).or_default(); + *entry = (*entry).max(stage_id.visit()); + if stage.first_event_seq.get() > checkpoint_seq { + resumed_from.insert(node_id.to_string(), stage_id.clone()); + } + } + Self { + high_water, + resumed_from, + } + } +} + +#[derive(Debug, Default)] +struct TrackerState { + /// Highest ordinal observed or reserved per node. + high_water: HashMap, + /// Pending provenance links, consumed by the first reservation per node. + resumed_from: HashMap, + /// Active execution scope per node. Cleared at the node boundary and + /// replaced by the next reservation. + active: HashMap, +} + +/// 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>, +} + +impl StageExecutionTracker { + #[must_use] + pub(crate) fn seeded(seed: StageExecutionSeed) -> Self { + Self { + state: Arc::new(Mutex::new(TrackerState { + high_water: seed.high_water, + resumed_from: seed.resumed_from, + active: HashMap::new(), + })), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, TrackerState> { + self.state + .lock() + .expect("stage execution tracker mutex is never poisoned: no code panics while holding this lock") + } + + /// Clear the node's prior execution scope at the node boundary. The next + /// `reserve`/`ensure` call allocates a fresh ordinal; a reservation is not + /// made here so that a StageStart hook block or process exit before any + /// stage-scoped event leaves no phantom execution. + pub(crate) fn begin_node(&self, node_id: &str) { + self.lock().active.remove(node_id); + } + + /// The node's active execution scope, if one has been reserved since the + /// last node boundary. + pub(crate) fn active(&self, node_id: &str) -> Option { + self.lock().active.get(node_id).cloned() + } + + /// Allocate the next execution ordinal for the node and make it the active + /// scope. Consumes the node's pending provenance link, if any. + pub(crate) fn reserve(&self, node_id: &str, graph_visit: u32) -> StageExecution { + let mut state = self.lock(); + let entry = state.high_water.entry(node_id.to_string()).or_default(); + *entry = entry.saturating_add(1); + let ordinal = *entry; + let resumed_from = state.resumed_from.remove(node_id); + let execution = StageExecution { + ordinal, + graph_visit, + resumed_from, + }; + state.active.insert(node_id.to_string(), execution.clone()); + execution + } + + /// The active scope for the node, reserving one only when none exists. + /// Later attempts within one execution and checkpoint pre-steps reuse the + /// first attempt's reservation. + pub(crate) fn ensure(&self, node_id: &str, graph_visit: u32) -> StageExecution { + if let Some(execution) = self.active(node_id) { + return execution; + } + self.reserve(node_id, graph_visit) + } +} + +#[cfg(test)] +mod tests { + use std::num::NonZeroU32; + + use chrono::Utc; + use fabro_types::{Graph, RunId, RunSpec, StageId, WorkflowSettings, test_support}; + + use super::*; + + fn projection_with_stages(stages: &[(&str, u32, u32)]) -> RunProjection { + let spec = RunSpec { + run_id: RunId::new(), + settings: WorkflowSettings::default(), + graph: Graph::new("test"), + graph_source: None, + workflow_slug: None, + automation: None, + source_directory: None, + labels: std::collections::HashMap::new(), + provenance: test_support::test_run_provenance(), + manifest_blob: None, + definition_blob: None, + git: None, + fork_source_ref: None, + }; + let mut projection = RunProjection::new(String::new(), spec, Utc::now()); + for (node_id, visit, seq) in stages { + projection.stage_entry( + node_id, + *visit, + NonZeroU32::new(*seq).expect("test seq must be non-zero"), + ); + } + projection + } + + #[test] + fn reserve_starts_at_one_and_allocates_monotonically_per_node() { + let tracker = StageExecutionTracker::default(); + + assert_eq!(tracker.reserve("work", 1).ordinal, 1); + tracker.begin_node("work"); + assert_eq!(tracker.reserve("work", 2).ordinal, 2); + assert_eq!(tracker.reserve("other", 1).ordinal, 1); + } + + #[test] + fn seeds_from_projection_maxima() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5), ("plan", 1, 3)]); + let seed = StageExecutionSeed::from_projection(&projection, 0); + let tracker = StageExecutionTracker::seeded(seed); + + assert_eq!(tracker.reserve("work", 1).ordinal, 3); + assert_eq!(tracker.reserve("plan", 1).ordinal, 2); + assert_eq!(tracker.reserve("new", 1).ordinal, 1); + } + + #[test] + fn graph_visit_and_ordinal_can_diverge() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5)]); + let seed = StageExecutionSeed::from_projection(&projection, 0); + let tracker = StageExecutionTracker::seeded(seed); + + let execution = tracker.reserve("work", 2); + assert_eq!(execution.ordinal, 3); + assert_eq!(execution.graph_visit, 2); + } + + #[test] + fn ensure_reuses_active_reservation_across_attempts() { + let tracker = StageExecutionTracker::default(); + + let first = tracker.ensure("work", 1); + let second = tracker.ensure("work", 1); + assert_eq!(first, second); + assert_eq!(second.ordinal, 1); + + tracker.begin_node("work"); + assert_eq!(tracker.ensure("work", 2).ordinal, 2); + } + + #[test] + fn begin_node_clears_only_that_node() { + let tracker = StageExecutionTracker::default(); + tracker.reserve("work", 1); + tracker.reserve("verify", 1); + + tracker.begin_node("work"); + + assert_eq!(tracker.active("work"), None); + assert_eq!(tracker.active("verify").map(|e| e.ordinal), Some(1)); + } + + #[test] + fn provenance_only_selects_stages_after_the_checkpoint() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 8), ("plan", 1, 3)]); + let seed = StageExecutionSeed::from_projection(&projection, 5); + + assert_eq!( + seed.resumed_from.get("work"), + Some(&StageId::new("work", 2)) + ); + assert_eq!(seed.resumed_from.get("plan"), None); + } + + #[test] + fn first_reservation_consumes_provenance() { + let projection = projection_with_stages(&[("work", 1, 6)]); + let seed = StageExecutionSeed::from_projection(&projection, 5); + let tracker = StageExecutionTracker::seeded(seed); + + let first = tracker.reserve("work", 1); + assert_eq!(first.ordinal, 2); + assert_eq!(first.resumed_from, Some(StageId::new("work", 1))); + + tracker.begin_node("work"); + let second = tracker.reserve("work", 2); + assert_eq!(second.ordinal, 3); + assert_eq!(second.resumed_from, None); + } + + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_reservations_stay_unique_per_node() { + let tracker = StageExecutionTracker::default(); + let handles: Vec<_> = (0..8) + .map(|_| { + let tracker = tracker.clone(); + tokio::spawn(async move { tracker.reserve("branch", 1).ordinal }) + }) + .collect(); + + let mut ordinals = Vec::new(); + for handle in handles { + ordinals.push(handle.await.expect("reservation task panicked")); + } + ordinals.sort_unstable(); + assert_eq!(ordinals, (1..=8).collect::>()); + } +} diff --git a/lib/components/fabro-workflow/src/stage_scope.rs b/lib/components/fabro-workflow/src/stage_scope.rs index fa309044b..f7fa28ddf 100644 --- a/lib/components/fabro-workflow/src/stage_scope.rs +++ b/lib/components/fabro-workflow/src/stage_scope.rs @@ -1,11 +1,28 @@ use fabro_types::{ParallelBranchId, StageId}; -use crate::context::{Context as WfContext, WorkflowContext}; +use crate::context::{Context as WfContext, WorkflowContext, keys}; use crate::run_dir::visit_from_context; +/// Read the stage execution ordinal seeded by the workflow lifecycle (or a +/// parallel branch dispatch). `None` when the current node has not reserved an +/// execution yet — direct-handler call sites (tests, etc.) that skip the full +/// lifecycle fall back to the graph visit, which equals the ordinal for a +/// first execution. +fn execution_ordinal_from_context(context: &WfContext) -> Option { + context + .get(keys::INTERNAL_STAGE_EXECUTION_ORDINAL) + .and_then(|value| value.as_u64()) + .map(|ordinal| u32::try_from(ordinal).unwrap_or(u32::MAX)) +} + /// Stage-level scope threaded through event emission to populate /// `stage_id` / `parallel_group_id` / `parallel_branch_id` on events /// that happen inside a concrete stage execution. +/// +/// `visit` is the 1-based stage execution ordinal — the numeric component of +/// the external `StageId`. It matches the graph visit for a first execution +/// and diverges when a cancelled or crashed invocation is reexecuted after +/// resume. #[derive(Clone, Debug)] pub struct StageScope { pub node_id: String, @@ -15,13 +32,15 @@ pub struct StageScope { } impl StageScope { - /// Build a scope from the given node id, sourcing visit count and parallel - /// ids from the current context. + /// Build a scope from the given node id, sourcing the execution ordinal + /// and parallel ids from the current context. pub fn from_context(context: &WfContext, node_id: impl Into) -> Self { + let visit = execution_ordinal_from_context(context) + .unwrap_or_else(|| u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)); Self { - node_id: node_id.into(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), - parallel_group_id: context.parallel_group_id(), + node_id: node_id.into(), + visit, + parallel_group_id: context.parallel_group_id(), parallel_branch_id: context.parallel_branch_id(), } } @@ -40,11 +59,10 @@ impl StageScope { /// handler (`ParallelBranchStarted`, `ParallelBranchCompleted`, and the /// pre-dispatch `GitCommit` for the branch worktree). /// - /// `target_visit` is the visit count of `target_node_id` for this - /// particular branch dispatch. The parallel handler currently passes - /// `1` because branches haven't been re-entered yet at the point of - /// scope construction; a future change that loops a parallel node - /// must pass the actual visit so envelope `stage_id`s stay accurate. + /// `target_visit` is the branch target's stage execution ordinal for this + /// particular dispatch, reserved through the run's shared + /// `StageExecutionTracker` so a resumed fan-out gets a fresh child + /// identity instead of overwriting the cancelled attempt's. #[must_use] pub fn for_parallel_branch( target_node_id: impl Into, diff --git a/lib/components/fabro-workflow/src/test_support.rs b/lib/components/fabro-workflow/src/test_support.rs index 4094b539d..1e2bf8adb 100644 --- a/lib/components/fabro-workflow/src/test_support.rs +++ b/lib/components/fabro-workflow/src/test_support.rs @@ -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()), diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index 3f88cdbd3..0f72849b3 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use super::ExecOutputTail; -use crate::{CommandTermination, PullRequestLink}; +use crate::{CommandTermination, PullRequestLink, StageId}; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] pub struct InterviewOption { @@ -23,7 +23,15 @@ pub struct ParallelStartedProps { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ParallelBranchStartedProps { - pub index: usize, + pub index: usize, + /// Graph visit of the branch target for this dispatch. The envelope + /// `stage_id` ordinal counts executions, so a resumed fan-out's branches + /// keep visit metadata even though their ordinals advanced. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior branch execution this one resumes from. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resumed_from_stage_id: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/lib/foundation/fabro-types/src/run_event/stage.rs b/lib/foundation/fabro-types/src/run_event/stage.rs index c3ba68318..7867380de 100644 --- a/lib/foundation/fabro-types/src/run_event/stage.rs +++ b/lib/foundation/fabro-types/src/run_event/stage.rs @@ -5,14 +5,25 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use super::ExecOutputTail; -use crate::{BilledModelUsage, DiffSummary, FailureDetail, Outcome, StageOutcome, StageTiming}; +use crate::{ + BilledModelUsage, DiffSummary, FailureDetail, Outcome, StageId, StageOutcome, StageTiming, +}; #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct StageStartedProps { - pub index: usize, - pub handler_type: String, - pub attempt: usize, - pub max_attempts: usize, + pub index: usize, + pub handler_type: String, + pub attempt: usize, + pub max_attempts: usize, + /// Graph visit that produced this stage execution. The envelope + /// `stage_id` ordinal counts executions, which diverges from the graph + /// visit when a cancelled or crashed invocation is reexecuted after + /// resume. Absent on events written before stage execution identity. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior execution this one resumes from. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resumed_from_stage_id: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -123,6 +134,12 @@ pub struct CheckpointCompletedProps { pub diff: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub diff_summary: Option, + /// 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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resumed_from_stage_id: Option, } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] diff --git a/lib/foundation/fabro-types/src/run_projection.rs b/lib/foundation/fabro-types/src/run_projection.rs index 2300927b7..d8f026442 100644 --- a/lib/foundation/fabro-types/src/run_projection.rs +++ b/lib/foundation/fabro-types/src/run_projection.rs @@ -320,59 +320,71 @@ impl StageContextWindow { #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] pub struct StageProjection { - pub first_event_seq: NonZeroU32, - pub prompt: Option, - pub response: Option, - pub completion: Option, - pub provider_used: Option, - pub diff: Option, - pub script_invocation: Option, - pub script_timing: Option, - pub parallel_results: Option, - pub output: Option, + pub first_event_seq: NonZeroU32, + pub prompt: Option, + pub response: Option, + pub completion: Option, + pub provider_used: Option, + pub diff: Option, + pub script_invocation: Option, + pub script_timing: Option, + pub parallel_results: Option, + pub output: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub output_bytes: Option, + pub output_bytes: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub live_streaming: Option, + pub live_streaming: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub termination: Option, + pub termination: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option>, + pub started_at: Option>, #[serde(default, skip_serializing_if = "Option::is_none")] - pub handler: Option, - /// Per-attempt timing breakdown for the latest terminal attempt. + pub handler: Option, + /// Graph visit that produced this stage execution. The `StageId` ordinal + /// counts executions, which diverges from the graph visit when a + /// cancelled or crashed invocation is reexecuted after resume. Absent on + /// projections built from events written before stage execution identity. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior execution this one resumes from. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub resumed_from_stage_id: Option, + /// 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, + pub timing: Option, #[serde(default)] - pub usage: BilledTokenCounts, + pub usage: BilledTokenCounts, #[serde(default, skip_serializing_if = "Option::is_none")] - pub model: Option, + pub model: Option, /// 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, + pub root_agent_todos: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub subagents: Vec, + pub subagents: Vec, #[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, + pub permission_level: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub agent_tools: Vec, + pub agent_tools: Vec, #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub mcp_servers: Vec, + pub mcp_servers: Vec, #[serde(default, skip_serializing_if = "Option::is_none")] - pub context_window: Option, + pub context_window: Option, #[serde(default)] - pub agent_control: AgentControlState, - pub state: StageState, + pub agent_control: AgentControlState, + pub state: StageState, } #[derive( @@ -486,6 +498,8 @@ impl StageProjection { termination: None, started_at: None, handler: None, + graph_visit: None, + resumed_from_stage_id: None, state: StageState::Running, } } @@ -519,14 +533,24 @@ impl StageProjection { self.timing.map(|timing| timing.wall_time_ms) } - /// Begin a new attempt (or visit) for this stage: clear every + /// Begin a new automatic attempt within this stage execution: clear every /// per-attempt field so prior-attempt data does not leak, then record /// `started_at` and `state = Running`. Preserves `first_event_seq` - /// (identity / sort key). + /// (identity / sort key) and the execution identity metadata + /// (`graph_visit`, `resumed_from_stage_id`). + /// + /// One stage projection represents one execution; a reexecution after + /// cancel or crash recovery gets a new `StageId` and never flows through + /// here. Replays of legacy histories with duplicate `stage.started` + /// events for one `StageId` retain this last-attempt behavior. pub fn begin_attempt(&mut self, started_at: DateTime, 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; } } diff --git a/lib/packages/fabro-api-client/src/models/run-stage.ts b/lib/packages/fabro-api-client/src/models/run-stage.ts index 2e25d4c55..4dd742f11 100644 --- a/lib/packages/fabro-api-client/src/models/run-stage.ts +++ b/lib/packages/fabro-api-client/src/models/run-stage.ts @@ -46,9 +46,17 @@ export interface RunStage { */ 'node_id': string; /** - * 1-based visit count; bumped each time the workflow re-enters this node. + * 1-based stage execution ordinal, the numeric component of `id`. It increments each time the node produces a new observable execution: graph re-entry (loops) and reexecution after cancel or crash recovery. Automatic in-place retries do not increment it. */ 'visit': number; + /** + * 1-based count of how many times workflow control entered this node (drives `max_visits`). Differs from `visit` when a cancelled or crashed execution was reexecuted after resume. Absent for stages recorded before execution identity was tracked. + */ + 'graph_visit'?: number | null; + /** + * StageId of the prior cancelled or interrupted execution this stage resumes from, when the run was resumed after that execution became observable. + */ + 'resumed_from_stage_id'?: string | null; 'provider_used'?: StageModelUsage | null; /** * Wall-clock time the latest attempt of this stage started, if known.