diff --git a/apps/fabro-web/app/components/stage-popover.test.tsx b/apps/fabro-web/app/components/stage-popover.test.tsx index efdbd7785..4a8bbb12c 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: "review/security@1", + status: "running", + duration: "--", + }); + const tree = render( + + + , + ); + const text = textOf(tree); + expect(text).toContain("Resumed from"); + expect(text).toContain("review/security@1"); + expect(text).toContain("Graph visit"); + const json = JSON.stringify(tree.toJSON()); + expect(json).toContain("/runs/run-1/stages/review%2Fsecurity%401"); + }); + + test("stage without ordinal divergence hides the graph visit row", () => { + const stage = makeStage({ graphVisit: 1, status: "pending", duration: "--" }); + const tree = render(); + 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..8dce68fc3 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..c165adee0 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 post-checkpoint work was replayed after resume; null + * for stages recorded before execution identity was tracked. + */ + graphVisit: number | null; + /** StageId of the prior execution superseded by this resumed replay, if any. */ + resumedFromStageId: string | null; startedAt: string | null; providerUsed: StageModelUsage | null; } @@ -85,6 +94,8 @@ export function mapRunStagesToSidebarStages( handler: stage.handler, nodeId: stage.node_id, visit: stage.visit, + graphVisit: stage.graph_visit ?? null, + resumedFromStageId: stage.resumed_from_stage_id ?? null, status: stage.status, duration: stage.wall_time_ms != null ? formatDurationMs(stage.wall_time_ms) diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index e230b3858..b39b6bb72 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, @@ -1738,6 +1738,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 replay of post-checkpoint work after + resume. Automatic in-place retries do not increment it. example: 2 + graph_visit: + type: ["integer", "null"] + format: uint32 + minimum: 1 + description: >- + 1-based count of how many times workflow control entered this node + (drives `max_visits`). Differs from `visit` when a post-checkpoint + execution is replayed after resume. Absent for stages recorded + before execution identity was tracked. + example: 1 + resumed_from_stage_id: + oneOf: + - $ref: "#/components/schemas/StageId" + - type: "null" + description: >- + StageId of the prior post-checkpoint execution superseded by this + replay after the run was resumed. + example: verify@1 provider_used: oneOf: - $ref: "#/components/schemas/StageModelUsage" 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 24570dac4..c5ef114d3 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, } } @@ -591,10 +593,12 @@ mod tests { assert_eq!(ui.stage.parallel_parent.as_deref(), Some("fork1")); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); let stage = &ui.stage.active_stages["fork1"]; assert_eq!(stage.tool_calls.len(), 1); @@ -630,10 +634,12 @@ mod tests { branch_count: 1, }); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); let stage = &ui.stage.active_stages["fork1"]; @@ -1245,10 +1251,12 @@ mod tests { branch_count: 1, }); emit(&mut ui, Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fork1", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), - branch: "security".into(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fork1", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("fork1", 1), 0), + branch: "security".into(), + index: 0, }); emit(&mut ui, Event::ParallelBranchCompleted { parallel_group_id: StageId::new("fork1", 1), @@ -1277,12 +1285,14 @@ mod tests { let stage_started = serde_json::to_string(&to_run_event_at( &fixtures::RUN_1, &Event::StageStarted { - node_id: "code".into(), - name: "Code".into(), - index: 0, - handler_type: "agent".into(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "code".into(), + name: "Code".into(), + index: 0, + handler_type: "agent".into(), + attempt: 1, + max_attempts: 1, }, started_ts, None, diff --git a/lib/apps/fabro-cli/tests/it/cmd/attach.rs b/lib/apps/fabro-cli/tests/it/cmd/attach.rs index fff46c251..c1ff07165 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 @@ -1257,6 +1258,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "outcome": "succeeded" }, "current_node": "start", + "graph_visit": 1, "next_node_id": "approve", "node_outcomes": { "start": { @@ -1284,6 +1286,7 @@ fn attach_json_errors_without_prompting_for_human_input() { "node_label": "Approve?", "properties": { "attempt": 1, + "graph_visit": 1, "handler_type": "human", "index": 1, "max_attempts": 1 diff --git a/lib/apps/fabro-server/src/demo/mod.rs b/lib/apps/fabro-server/src/demo/mod.rs index 4b5bb204e..9fa69379b 100644 --- a/lib/apps/fabro-server/src/demo/mod.rs +++ b/lib/apps/fabro-server/src/demo/mod.rs @@ -1099,7 +1099,6 @@ mod runs { }; use super::ts; - use crate::server::run_stage_from_stage_id; static DEMO_PRINCIPAL: LazyLock = LazyLock::new(|| { Principal::user( @@ -1124,6 +1123,29 @@ mod runs { } } + fn stage( + stage_id: &StageId, + name: &str, + status: StageState, + wall_time_ms: Option, + handler: StageHandler, + ) -> RunStage { + RunStage { + id: stage_id.clone(), + name: name.to_owned(), + handler, + status, + wall_time_ms, + node_id: stage_id.node_id().to_owned(), + visit: std::num::NonZeroU32::new(stage_id.visit()) + .expect("StageId stores a non-zero visit"), + provider_used: None, + started_at: None, + graph_visit: None, + resumed_from_stage_id: None, + } + } + fn demo_run_ids() -> &'static [RunId; 7] { static IDS: OnceLock<[RunId; 7]> = OnceLock::new(); IDS.get_or_init(|| { @@ -1380,50 +1402,40 @@ mod runs { pub(super) fn stages() -> Vec { vec![ - run_stage_from_stage_id( + stage( &StageId::new("detect-drift", 1), "Detect Drift", StageState::Succeeded, Some(72_000), - None, StageHandler::Command, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("propose-changes", 1), "Propose Changes", StageState::Succeeded, Some(154_000), - None, StageHandler::Agent, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("review-changes", 1), "Review Changes", StageState::Succeeded, Some(45_000), - None, StageHandler::Agent, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("apply-changes", 1), "Apply Changes", StageState::Succeeded, Some(118_000), - None, StageHandler::Command, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("apply-changes", 2), "Apply Changes", StageState::Running, None, - None, StageHandler::Command, - None, ), ] } diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 2e577d8f6..11a9a0948 100644 --- a/lib/apps/fabro-server/src/server.rs +++ b/lib/apps/fabro-server/src/server.rs @@ -99,7 +99,7 @@ use fabro_types::{ AgentBackend, AskFabro, AskFabroUnavailableReason, EventBody, InterviewQuestionRecord, PairId, PairMessageId, PairTarget, PendingReason, Principal, PullRequestLink, QuestionType, RunBlobId, RunControlAction, RunEvent, RunId, RunRunnableSource, SandboxProviderKind, ServerSettings, - SessionCapability, StageModelUsage, + SessionCapability, }; use fabro_util::error::{ SharedError, collect_causes, render_compact_with_causes, render_with_causes, @@ -1332,29 +1332,6 @@ fn accumulate_billing_rollup( } } -pub(crate) fn run_stage_from_stage_id( - stage_id: &StageId, - name: impl Into, - status: StageState, - wall_time_ms: Option, - started_at: Option>, - handler: StageHandler, - provider_used: Option, -) -> RunStage { - RunStage { - id: stage_id.to_string(), - name: name.into(), - handler, - status, - wall_time_ms, - node_id: stage_id.node_id().to_string(), - visit: std::num::NonZeroU32::new(stage_id.visit()) - .expect("StageId stores a non-zero visit"), - provider_used, - started_at, - } -} - impl AppState { pub(crate) fn manifest_run_defaults(&self) -> Arc { Arc::clone( diff --git a/lib/apps/fabro-server/src/server/handler/billing.rs b/lib/apps/fabro-server/src/server/handler/billing.rs index ba7b71625..a1888713c 100644 --- a/lib/apps/fabro-server/src/server/handler/billing.rs +++ b/lib/apps/fabro-server/src/server/handler/billing.rs @@ -2,12 +2,14 @@ use std::collections::HashMap; use std::sync::Arc; use chrono::{DateTime, Utc}; -use fabro_types::{RunProjection, StageHandler, StageProjection, StageState, StageTiming}; +use fabro_types::{ + Graph, RunProjection, StageHandler, StageId, StageProjection, StageState, StageTiming, +}; use super::super::{ AppState, BillingByModel, BillingStageRef, IntoResponse, Json, ListResponse, PaginationParams, Path, Query, RequiredUser, Response, Router, RunBilling, RunBillingStage, RunBillingTotals, - RunId, State, StatusCode, get, parse_run_id_path, run_stage_from_stage_id, + RunId, RunStage, State, StatusCode, get, parse_run_id_path, }; pub(super) fn routes() -> Router> { @@ -16,6 +18,36 @@ pub(super) fn routes() -> Router> { .route("/runs/{id}/billing", get(get_run_billing)) } +fn run_stage_from_projection( + stage_id: &StageId, + stage: &StageProjection, + graph: &Graph, + now: DateTime, +) -> RunStage { + let handler = stage.handler.unwrap_or_else(|| { + StageHandler::from_handler_type( + graph + .nodes + .get(stage_id.node_id()) + .and_then(|node| node.handler_type()), + ) + }); + RunStage { + id: stage_id.clone(), + name: stage_id.node_id().to_owned(), + handler, + status: stage.effective_state(), + wall_time_ms: stage.live_wall_time_ms(now), + node_id: stage_id.node_id().to_owned(), + visit: std::num::NonZeroU32::new(stage_id.visit()) + .expect("StageId stores a non-zero visit"), + provider_used: stage.provider_used.clone(), + started_at: stage.started_at, + graph_visit: stage.graph_visit.and_then(std::num::NonZeroU32::new), + resumed_from_stage_id: stage.resumed_from_stage_id.clone(), + } +} + async fn list_run_stages( _auth: RequiredUser, State(state): State>, @@ -37,25 +69,7 @@ async fn list_run_stages( let graph = projection.spec().graph(); let stages = projection .iter_stages() - .map(|(stage_id, stage)| { - let handler = stage.handler.unwrap_or_else(|| { - StageHandler::from_handler_type( - graph - .nodes - .get(stage_id.node_id()) - .and_then(|n| n.handler_type()), - ) - }); - run_stage_from_stage_id( - stage_id, - stage_id.node_id().to_string(), - stage.effective_state(), - stage.live_wall_time_ms(now), - stage.started_at, - handler, - stage.provider_used.clone(), - ) - }) + .map(|(stage_id, stage)| run_stage_from_projection(stage_id, stage, graph, now)) .collect::>(); (StatusCode::OK, Json(ListResponse::new(stages))).into_response() diff --git a/lib/apps/fabro-server/src/server/tests.rs b/lib/apps/fabro-server/src/server/tests.rs index 2e4e14a1b..1a86ef59f 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, } } @@ -8169,12 +8301,14 @@ async fn get_run_stage_command_log_returns_scratch_slice() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandStarted { node_id: "script_node".to_string(), @@ -8236,12 +8370,14 @@ async fn get_run_stage_command_log_returns_cas_slice() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandCompleted { node_id: "script_node".to_string(), @@ -8300,12 +8436,14 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() { definition_blob: None, }, workflow_event::Event::StageStarted { - node_id: "script_node".to_string(), - name: "Script".to_string(), - index: 1, - handler_type: "command".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "script_node".to_string(), + name: "Script".to_string(), + index: 1, + handler_type: "command".to_string(), + attempt: 1, + max_attempts: 1, }, workflow_event::Event::CommandCompleted { node_id: "script_node".to_string(), @@ -9491,12 +9629,14 @@ async fn cache_backed_run_endpoints_reflect_events_appended_after_warmup() { "review", 1, &workflow_event::Event::StageStarted { - node_id: "review".to_string(), - name: "Review".to_string(), - index: 0, - handler_type: "human".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "review".to_string(), + name: "Review".to_string(), + index: 0, + handler_type: "human".to_string(), + attempt: 1, + max_attempts: 1, }, ) .await; @@ -11528,6 +11668,8 @@ async fn resume_cancelled_run_with_checkpoint_transitions_to_runnable() { workflow_event::Event::RunStarting, workflow_event::Event::RunRunning, workflow_event::Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: "succeeded".to_string(), current_node: checkpoint.current_node.clone(), diff --git a/lib/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index 1a63f2915..60f3eb877 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -227,35 +227,46 @@ impl RunProjectionReducer for RunProjection { } EventBody::CheckpointCompleted(props) => { let checkpoint = checkpoint_from_props(props, ts); - if let Some(node_id) = stored.node_id.as_deref() { - let visit = checkpoint - .node_visits - .get(node_id) - .and_then(|visit| u32::try_from(*visit).ok()) - .unwrap_or(1); - if let Some(diff) = props.diff.clone() { - self.stage_entry(node_id, visit, first_event_seq(event.seq)) - .diff = Some(diff); + if let Some(stage_id) = stored.stage_id.as_ref() { + // Envelope-first: the diff and any skipped-stage synthesis + // attach to the exact execution recorded on the event. + // Historical `node_outcomes` must not create or collide + // with a newer execution ordinal. + apply_checkpoint_to_stage(self, stage_id, props, &checkpoint, event.seq, ts); + } else { + // Legacy fallback for events without a stored stage id: + // resolve the visit from the checkpointed `node_visits` + // and synthesize skipped stages from historical outcomes. + if let Some(node_id) = stored.node_id.as_deref() { + let visit = checkpoint + .node_visits + .get(node_id) + .and_then(|visit| u32::try_from(*visit).ok()) + .unwrap_or(1); + if let Some(diff) = props.diff.clone() { + self.stage_entry(node_id, visit, first_event_seq(event.seq)) + .diff = Some(diff); + } } - } - for (node_id, outcome) in &checkpoint.node_outcomes { - if outcome.status != StageOutcome::Skipped { - continue; + for (node_id, outcome) in &checkpoint.node_outcomes { + if outcome.status != StageOutcome::Skipped { + continue; + } + let visit = checkpoint + .node_visits + .get(node_id) + .and_then(|visit| u32::try_from(*visit).ok()) + .unwrap_or(1); + if self + .stage(&fabro_types::StageId::new(node_id, visit)) + .is_some() + { + continue; + } + let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq)); + stage.completion = Some(stage_completion_from_outcome(outcome, ts)); + stage.state = StageState::Skipped; } - let visit = checkpoint - .node_visits - .get(node_id) - .and_then(|visit| u32::try_from(*visit).ok()) - .unwrap_or(1); - if self - .stage(&fabro_types::StageId::new(node_id, visit)) - .is_some() - { - continue; - } - let stage = self.stage_entry(node_id, visit, first_event_seq(event.seq)); - stage.completion = Some(stage_completion_from_outcome(outcome, ts)); - stage.state = StageState::Skipped; } self.checkpoints.push(CheckpointRecord { seq: event.seq, @@ -340,15 +351,23 @@ impl RunProjectionReducer for RunProjection { let Some(stage_id) = stored.stage_id.as_ref() else { return Ok(()); }; - let stage = self.stage_entry( - stage_id.node_id(), - stage_id.visit(), - first_event_seq(event.seq), - ); + // A `stage.started` for a new `StageId` creates a new + // projection, so an older execution's terminal projection + // stays immutable. `begin_attempt` on an existing entry + // remains the compatibility path for automatic retries and + // legacy histories that repeat one `StageId`. + let is_new = self.stage(stage_id).is_none(); + let stage = stage_at_stored_stage_id(self, stage_id, event.seq); stage.begin_attempt( ts, StageHandler::from_handler_type(Some(&props.handler_type)), ); + if is_new { + stage.graph_visit = props.graph_visit; + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } } EventBody::StageRetrying(_) => { let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { @@ -537,16 +556,26 @@ impl RunProjectionReducer for RunProjection { }; stage.parallel_results = Some(props.results.clone()); } - EventBody::ParallelBranchStarted(_) => { + EventBody::ParallelBranchStarted(props) => { // Branches bypass the engine's StageStarted/StageCompleted // lifecycle. Seed started_at so the branch stage drives a live // wall-clock timer while it runs (the entry is created Running). + let is_new = stored + .stage_id + .as_ref() + .is_none_or(|stage_id| self.stage(stage_id).is_none()); let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else { return Ok(()); }; if stage.started_at.is_none() { stage.started_at = Some(ts); } + if is_new { + stage.graph_visit = props.graph_visit; + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } stage.state = StageState::Running; } EventBody::ParallelBranchCompleted(props) => { @@ -947,6 +976,50 @@ fn stage_at_stored_or_current_visit<'a>( stage_at_current_visit(state, stored, seq) } +/// Apply a `checkpoint.completed` event to the exact execution named by the +/// envelope `StageId`: attach the checkpoint diff, and for a skipped +/// checkpoint finalize that execution as `Skipped`. A synthetic skipped +/// projection is created only for a first-attempt skip that never +/// materialized a stage; an existing non-terminal (e.g. `Retrying`) +/// projection keeps its identity and becomes terminal, while an older +/// terminal execution stays immutable. +fn apply_checkpoint_to_stage( + state: &mut RunProjection, + stage_id: &StageId, + props: &CheckpointCompletedProps, + checkpoint: &Checkpoint, + seq: u32, + ts: DateTime, +) { + let node_id = stage_id.node_id(); + let skipped_completion = checkpoint + .node_outcomes + .get(node_id) + .filter(|outcome| outcome.status == StageOutcome::Skipped) + .map(|outcome| stage_completion_from_outcome(outcome, ts)); + if props.diff.is_none() && skipped_completion.is_none() { + return; + } + + let is_new = state.stage(stage_id).is_none(); + let stage = stage_at_stored_stage_id(state, stage_id, seq); + if is_new { + stage.graph_visit = props.graph_visit; + stage + .resumed_from_stage_id + .clone_from(&props.resumed_from_stage_id); + } + if let Some(diff) = props.diff.clone() { + stage.diff = Some(diff); + } + if let Some(completion) = skipped_completion { + if !stage.state.is_terminal() { + stage.completion = Some(completion); + stage.state = StageState::Skipped; + } + } +} + fn stage_at_completed_visit<'a>( state: &'a mut RunProjection, stored: &RunEvent, @@ -2008,10 +2081,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2030,10 +2105,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2073,7 +2150,11 @@ mod tests { .apply_event(&test_stage_event_at( 3, "2026-04-07T12:00:00Z", - EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }), + EventBody::ParallelBranchStarted(ParallelBranchStartedProps { + index: 0, + graph_visit: None, + resumed_from_stage_id: None, + }), branch.clone(), )) .unwrap(); @@ -2110,7 +2191,11 @@ mod tests { state .apply_event(&test_stage_event( 3, - EventBody::ParallelBranchStarted(ParallelBranchStartedProps { index: 0 }), + EventBody::ParallelBranchStarted(ParallelBranchStartedProps { + index: 0, + graph_visit: None, + resumed_from_stage_id: None, + }), branch.clone(), )) .unwrap(); @@ -2144,10 +2229,12 @@ mod tests { .apply_event(&test_stage_event( 3, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2400,10 +2487,12 @@ mod tests { .apply_event(&test_stage_event( 2, EventBody::StageStarted(StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }), stage_id.clone(), )) @@ -2571,6 +2660,8 @@ mod tests { .apply_event(&test_event( 5, EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: None, + resumed_from_stage_id: None, status: "running".to_string(), current_node: "next".to_string(), completed_nodes: vec!["skip_me".to_string()], @@ -3827,10 +3918,12 @@ mod tests { fn started_props() -> StageStartedProps { StageStartedProps { - index: 0, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 3, + graph_visit: None, + resumed_from_stage_id: None, + index: 0, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 3, } } @@ -4330,6 +4423,420 @@ mod tests { ); } + #[test] + fn stage_started_records_execution_identity_metadata() { + let mut state = running_projection(); + let stage_id = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(StageId::new("work", 1)), + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!(stage.graph_visit, Some(1)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1))); + } + + #[test] + fn retry_stage_started_preserves_execution_identity_metadata() { + let mut state = running_projection(); + let stage_id = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(StageId::new("work", 1)), + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + // A malformed retry event for the same StageId cannot rewrite the + // first attempt's immutable execution identity. + state + .apply_event(&test_stage_event( + 5, + EventBody::StageStarted(StageStartedProps { + attempt: 2, + graph_visit: Some(99), + resumed_from_stage_id: Some(StageId::new("other", 7)), + ..started_props() + }), + stage_id.clone(), + )) + .unwrap(); + + let stage = state.stage(&stage_id).unwrap(); + assert_eq!(stage.graph_visit, Some(1)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("work", 1))); + } + + #[test] + fn cancelled_execution_stays_immutable_when_resumed_execution_starts() { + let mut state = running_projection(); + let first = StageId::new("work", 1); + let second = StageId::new("work", 2); + + state + .apply_event(&test_stage_event_at( + 4, + "2026-04-07T12:00:00Z", + EventBody::StageStarted(started_props()), + first.clone(), + )) + .unwrap(); + state + .apply_event(&test_raw_event_at( + 5, + "2026-04-07T12:00:05Z", + "run.failed", + &serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event( + 6, + "run.start_requested", + &json!({ "resume": true }), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event( + 7, + "run.runnable", + &json!({ "source": "start_requested" }), + None, + )) + .unwrap(); + state + .apply_event(&test_raw_event(8, "run.starting", &json!({}), None)) + .unwrap(); + state + .apply_event(&test_raw_event(9, "run.running", &json!({}), None)) + .unwrap(); + state + .apply_event(&test_stage_event( + 10, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + resumed_from_stage_id: Some(first.clone()), + ..started_props() + }), + second.clone(), + )) + .unwrap(); + + // The cancelled execution keeps its terminal projection untouched... + let cancelled = state.stage(&first).unwrap(); + assert_eq!(cancelled.state, StageState::Cancelled); + assert_eq!( + cancelled.timing, + Some(fabro_types::StageTiming::wall_only(5_000)) + ); + // ...while the reexecution runs as a distinct stage linked back to it. + let resumed = state.stage(&second).unwrap(); + assert_eq!(resumed.state, StageState::Running); + assert_eq!(resumed.graph_visit, Some(1)); + assert_eq!(resumed.resumed_from_stage_id, Some(first)); + } + + #[test] + fn run_failed_after_resume_preserves_earlier_terminal_executions() { + let mut state = running_projection(); + let done = StageId::new("verify", 1); + let active = StageId::new("work", 2); + + state + .apply_event(&test_stage_event_at( + 4, + "2026-04-07T12:00:00Z", + EventBody::StageStarted(started_props()), + done.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event_at( + 5, + "2026-04-07T12:00:03Z", + EventBody::StageCompleted(completed_props(3_000, StageOutcome::Succeeded)), + done.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event_at( + 6, + "2026-04-07T12:00:04Z", + EventBody::StageStarted(started_props()), + active.clone(), + )) + .unwrap(); + state + .apply_event(&test_raw_event_at( + 7, + "2026-04-07T12:00:09Z", + "run.failed", + &serde_json::to_value(run_failed_props(FailureReason::Cancelled)).unwrap(), + None, + )) + .unwrap(); + + let terminal = state.stage(&done).unwrap(); + assert_eq!(terminal.state, StageState::Succeeded); + assert_eq!( + terminal.timing, + Some(fabro_types::StageTiming::wall_only(3_000)) + ); + assert_eq!(state.stage(&active).unwrap().state, StageState::Cancelled); + } + + #[test] + fn checkpoint_completed_targets_envelope_stage_id_after_ordinal_divergence() { + let mut state = running_projection(); + let execution = StageId::new("work", 2); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(StageStartedProps { + graph_visit: Some(1), + ..started_props() + }), + execution.clone(), + )) + .unwrap(); + // Checkpointed graph state still says visit 1 for `work`, and carries + // a historical skipped outcome for another node. Neither may create + // or mutate a projection at a stale ordinal. + state + .apply_event(&test_stage_event( + 5, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "success".to_string(), + current_node: "work".to_string(), + completed_nodes: vec!["old_skip".to_string(), "work".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "old_skip".to_string(), + Outcome::skipped("historical skip"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([ + ("work".to_string(), 1usize), + ("old_skip".to_string(), 1usize), + ]), + diff: Some("diff for work@2".to_string()), + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + assert_eq!( + state.stage(&execution).unwrap().diff.as_deref(), + Some("diff for work@2") + ); + assert!(state.stage(&StageId::new("work", 1)).is_none()); + assert!(state.stage(&StageId::new("old_skip", 1)).is_none()); + } + + #[test] + fn skipped_checkpoint_with_stage_id_creates_synthetic_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 3); + + // First-attempt StageStart-hook skip: no stage.started was emitted, + // the checkpoint is the first stage-scoped event for this execution. + state + .apply_event(&test_stage_event( + 4, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(2), + resumed_from_stage_id: Some(StageId::new("gate", 2)), + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("skipped by StageStart hook"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 2usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Skipped); + assert_eq!(stage.graph_visit, Some(2)); + assert_eq!(stage.resumed_from_stage_id, Some(StageId::new("gate", 2))); + assert_eq!( + stage.completion.as_ref().unwrap().notes.as_deref(), + Some("skipped by StageStart hook") + ); + } + + #[test] + fn skipped_checkpoint_finalizes_existing_retrying_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 1); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(started_props()), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 5, + EventBody::StageRetrying(StageRetryingProps { + index: 0, + attempt: 2, + max_attempts: 3, + delay_ms: 0, + }), + execution.clone(), + )) + .unwrap(); + assert_eq!(state.stage(&execution).unwrap().state, StageState::Retrying); + + // StageStart hook skipped the retry; the checkpoint finalizes the + // existing execution instead of allocating a new projection. + state + .apply_event(&test_stage_event( + 6, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("skipped on retry"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 1usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Skipped); + assert_eq!(stage.first_event_seq, first_event_seq(4)); + assert!(state.stage(&StageId::new("gate", 2)).is_none()); + } + + #[test] + fn skipped_checkpoint_never_reopens_an_older_terminal_execution() { + let mut state = running_projection(); + let execution = StageId::new("gate", 1); + + state + .apply_event(&test_stage_event( + 4, + EventBody::StageStarted(started_props()), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 5, + EventBody::StageCompleted(completed_props(2_000, StageOutcome::Succeeded)), + execution.clone(), + )) + .unwrap(); + state + .apply_event(&test_stage_event( + 6, + EventBody::CheckpointCompleted(CheckpointCompletedProps { + graph_visit: Some(1), + resumed_from_stage_id: None, + status: "skipped".to_string(), + current_node: "gate".to_string(), + completed_nodes: vec!["gate".to_string()], + node_retries: BTreeMap::new(), + context_values: BTreeMap::new(), + node_outcomes: BTreeMap::from([( + "gate".to_string(), + Outcome::skipped("late skip"), + )]), + next_node_id: None, + git_commit_sha: None, + loop_failure_signatures: BTreeMap::new(), + restart_failure_signatures: BTreeMap::new(), + node_visits: BTreeMap::from([("gate".to_string(), 1usize)]), + diff: None, + diff_summary: None, + }), + execution.clone(), + )) + .unwrap(); + + let stage = state.stage(&execution).unwrap(); + assert_eq!(stage.state, StageState::Succeeded); + assert_eq!( + stage.completion.as_ref().unwrap().outcome, + StageOutcome::Succeeded + ); + } + + #[test] + fn legacy_stage_started_payload_without_identity_fields_deserializes() { + let event = test_raw_event( + 4, + "stage.started", + &json!({ + "index": 0, + "handler_type": "agent", + "attempt": 1, + "max_attempts": 3 + }), + Some("work"), + ); + + let EventBody::StageStarted(props) = &event.event.body else { + panic!("expected stage.started body"); + }; + assert_eq!(props.graph_visit, None); + assert_eq!(props.resumed_from_stage_id, None); + } + #[test] fn run_failed_non_cancelled_finalizes_running_stage_as_failed() { let mut state = running_projection(); diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 86c5ea79e..6e94368b2 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"; @@ -46,8 +53,11 @@ pub mod keys { pub const PARALLEL_BRANCH_COUNT: &str = "parallel.branch_count"; /// Runtime-only keys stripped from durable context projections. - pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = - &[CURRENT_PREAMBLE, INTERNAL_PARALLEL_BRANCH_PREAMBLES]; + pub(crate) const TRANSIENT_CONTEXT_KEYS: &[&str] = &[ + CURRENT_PREAMBLE, + INTERNAL_PARALLEL_BRANCH_PREAMBLES, + INTERNAL_STAGE_EXECUTION_ORDINAL, + ]; // --- Prefix constants (for filtering and dynamic keys) --- pub const GRAPH_PREFIX: &str = "graph."; @@ -195,7 +205,7 @@ pub trait WorkflowContext { fn parallel_group_id(&self) -> Option; fn parallel_branch_id(&self) -> Option; /// Build the stage-level emit scope from the currently-executing node and - /// its accumulated visit count. Returns `None` for run-level emissions + /// its execution ordinal. Returns `None` for run-level emissions /// where no stage is active (i.e., `CURRENT_NODE` is unset). fn current_stage_scope(&self) -> Option; } diff --git a/lib/components/fabro-workflow/src/event/convert.rs b/lib/components/fabro-workflow/src/event/convert.rs index b6aa6e339..2cbc34d07 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, @@ -376,11 +380,16 @@ fn event_body_from_event(event: &Event) -> EventBody { visit: *visit, branch_count: *branch_count, }), - Event::ParallelBranchStarted { index, .. } => { - EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps { - index: *index, - }) - } + Event::ParallelBranchStarted { + index, + graph_visit, + resumed_from_stage_id, + .. + } => EventBody::ParallelBranchStarted(fabro_types::ParallelBranchStartedProps { + index: *index, + graph_visit: *graph_visit, + resumed_from_stage_id: resumed_from_stage_id.clone(), + }), Event::ParallelBranchCompleted { index, duration_ms, @@ -476,6 +485,8 @@ fn event_body_from_event(event: &Event) -> EventBody { node_visits, diff, diff_summary, + graph_visit, + resumed_from_stage_id, .. } => EventBody::CheckpointCompleted(fabro_types::CheckpointCompletedProps { status: status.clone(), @@ -491,6 +502,8 @@ fn event_body_from_event(event: &Event) -> EventBody { node_visits: node_visits.clone(), diff: diff.clone(), diff_summary: *diff_summary, + graph_visit: *graph_visit, + resumed_from_stage_id: resumed_from_stage_id.clone(), }), Event::CheckpointFailed { error, @@ -1676,12 +1689,14 @@ mod tests { let stored = to_run_event_at( &fixtures::RUN_1, &Event::StageStarted { - node_id: "review".to_string(), - name: "review".to_string(), - index: 1, - handler_type: "agent".to_string(), - attempt: 1, - max_attempts: 1, + graph_visit: None, + resumed_from_stage_id: None, + node_id: "review".to_string(), + name: "review".to_string(), + index: 1, + handler_type: "agent".to_string(), + attempt: 1, + max_attempts: 1, }, Utc::now(), Some(&StageScope { @@ -1794,10 +1809,12 @@ mod tests { #[test] fn parallel_branch_started_populates_group_and_branch_ids() { let stored = to_run_event(&fixtures::RUN_1, &Event::ParallelBranchStarted { - parallel_group_id: StageId::new("fanout", 2), - parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1), - branch: "review".to_string(), - index: 1, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("fanout", 2), + parallel_branch_id: ParallelBranchId::new(StageId::new("fanout", 2), 1), + branch: "review".to_string(), + index: 1, }); assert_eq!(stored.parallel_group_id, Some(StageId::new("fanout", 2))); assert_eq!( diff --git a/lib/components/fabro-workflow/src/event/events.rs b/lib/components/fabro-workflow/src/event/events.rs index ed940c5b1..20803b1fa 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 post-checkpoint work is replayed + /// after resume. + #[serde(default, skip_serializing_if = "Option::is_none")] + graph_visit: Option, + /// Prior execution superseded by this resumed replay, for the first + /// execution reserved after a resume when the node had an + /// observable post-checkpoint execution. + #[serde(default, skip_serializing_if = "Option::is_none")] + resumed_from_stage_id: Option, }, StageCompleted { node_id: String, @@ -306,10 +316,14 @@ pub enum Event { branch_count: usize, }, ParallelBranchStarted { - parallel_group_id: StageId, - parallel_branch_id: ParallelBranchId, - branch: String, - index: usize, + parallel_group_id: StageId, + parallel_branch_id: ParallelBranchId, + branch: String, + index: usize, + #[serde(default, skip_serializing_if = "Option::is_none")] + graph_visit: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + resumed_from_stage_id: Option, }, ParallelBranchCompleted { parallel_group_id: StageId, @@ -393,6 +407,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 722c72373..67be1ab3a 100644 --- a/lib/components/fabro-workflow/src/event/names.rs +++ b/lib/components/fabro-workflow/src/event/names.rs @@ -167,10 +167,12 @@ mod tests { fn event_name_matches_new_dot_notation() { assert_eq!( event_name(&Event::ParallelBranchStarted { - parallel_group_id: StageId::new("plan", 1), - parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0), - branch: "fork".to_string(), - index: 0, + graph_visit: None, + resumed_from_stage_id: None, + parallel_group_id: StageId::new("plan", 1), + parallel_branch_id: ParallelBranchId::new(StageId::new("plan", 1), 0), + branch: "fork".to_string(), + index: 0, }), "parallel.branch.started" ); diff --git a/lib/components/fabro-workflow/src/git.rs b/lib/components/fabro-workflow/src/git.rs index a1f0adf99..c6dd0c75f 100644 --- a/lib/components/fabro-workflow/src/git.rs +++ b/lib/components/fabro-workflow/src/git.rs @@ -455,6 +455,8 @@ mod tests { .await .unwrap(); append_event(&run, &fixtures::RUN_1, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "work".into(), status: "succeeded".into(), current_node: "work".into(), diff --git a/lib/components/fabro-workflow/src/handler/manager_loop.rs b/lib/components/fabro-workflow/src/handler/manager_loop.rs index 13644e997..98804d5e9 100644 --- a/lib/components/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/components/fabro-workflow/src/handler/manager_loop.rs @@ -19,10 +19,9 @@ use crate::error::Error; use crate::operations::{ValidateInput, WorkflowInput, validate}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::pipeline::types::Initialized; -use crate::run_dir::visit_from_context; use crate::run_options::RunOptions; use crate::static_reference::{ReferenceKind, validate_static_reference}; -use crate::{ManifestPath, pipeline}; +use crate::{ManifestPath, pipeline, stage_scope}; /// Orchestrates a child workflow engine, polling for completion or stop /// conditions. @@ -182,8 +181,10 @@ impl Handler for SubWorkflowHandler { } }; - // Build child RunOptions - let visit = visit_from_context(context) as u64; + // Build child RunOptions. The stage directory follows the execution + // ordinal so a replayed manager loop keeps the prior execution's + // child logs intact. + let visit = u64::from(stage_scope::execution_ordinal_from_context(context)); let child_logs = run_dir.join(format!("stages/{}@{visit}/child", node.id)); let _ = fs::create_dir_all(&child_logs).await; diff --git a/lib/components/fabro-workflow/src/handler/parallel.rs b/lib/components/fabro-workflow/src/handler/parallel.rs index 3efd0e14d..ac57f5b16 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -1,6 +1,6 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::path::Path; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::Instant; use async_trait::async_trait; @@ -17,6 +17,7 @@ use crate::error::Error; use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::hook_context::set_hook_node; use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt}; +use crate::run_dir::visit_from_context; use crate::{artifact, millis_u64}; /// Fans out execution to multiple branches concurrently. @@ -32,7 +33,12 @@ struct BranchDispatch { index: usize, target_id: String, branch_id: ParallelBranchId, - scope: StageScope, + /// Scope reserved by the branch task right before its + /// `ParallelBranchStarted` becomes observable. Empty when the branch was + /// cancelled or failed before starting — no events exist to pair a + /// completion with, and emitting one under a guessed ordinal would + /// resurrect a prior execution's stage. + scope: Arc>, handle: JoinHandle>, } @@ -117,6 +123,7 @@ async fn run_branches( let max_parallel = usize::try_from(max_parallel).unwrap_or(4).max(1); let semaphore = Arc::new(Semaphore::new(max_parallel)); let shared_graph = Arc::new(graph.clone()); + let branch_graph_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); let branch_preambles = parse_branch_preambles( context.get(keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES), @@ -169,18 +176,13 @@ async fn run_branches( let run_dir = run_dir.to_path_buf(); let semaphore = Arc::clone(&semaphore); let group_id = parallel_group_id.clone(); - let branch_scope = StageScope::for_parallel_branch( - target_id.clone(), - 1, - group_id.clone(), - parallel_branch_id.clone(), - ); + let reserved_scope = Arc::new(OnceLock::new()); dispatches.push(BranchDispatch { index: branch_index, target_id: target_id.clone(), branch_id: parallel_branch_id.clone(), - scope: branch_scope.clone(), + scope: Arc::clone(&reserved_scope), handle: tokio::spawn(async move { let branch_start = Instant::now(); let task = async { @@ -195,12 +197,39 @@ async fn run_branches( permit = &mut permit => permit .map_err(|err| Error::handler_with_source("semaphore error", err))?, }; + // Only reserve once the branch is ready to become + // observable, so a branch cancelled while waiting on the + // semaphore never consumes an execution identity. + let execution = branch_services + .run + .stage_executions + .reserve(&target_id, branch_graph_visit); + branch_context.set( + keys::CURRENT_NODE, + serde_json::Value::String(target_id.clone()), + ); + branch_context.set( + keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.stage_id.visit()), + ); + let branch_scope = reserved_scope + .get_or_init(|| { + StageScope::for_parallel_branch( + target_id.clone(), + execution.stage_id.visit(), + group_id.clone(), + parallel_branch_id.clone(), + ) + }) + .clone(); branch_services.run.emitter.emit_scoped( &Event::ParallelBranchStarted { - parallel_group_id: group_id.clone(), - parallel_branch_id: parallel_branch_id.clone(), - branch: target_id.clone(), - index: branch_index, + parallel_group_id: group_id.clone(), + parallel_branch_id: parallel_branch_id.clone(), + branch: target_id.clone(), + index: branch_index, + graph_visit: Some(execution.graph_visit), + resumed_from_stage_id: execution.resumed_from.clone(), }, &branch_scope, ); @@ -255,15 +284,17 @@ async fn run_branches( Err(payload) => { let result = failed_branch_result(&target_id, super::format_panic_message(&payload)); - emit_branch_completed( - &branch_services.run.emitter, - &branch_scope, - group_id, - parallel_branch_id, - branch_index, - millis_u64(branch_start.elapsed()), - result.outcome.status, - ); + if let Some(scope) = reserved_scope.get() { + emit_branch_completed( + &branch_services.run.emitter, + scope, + group_id, + parallel_branch_id, + branch_index, + millis_u64(branch_start.elapsed()), + result.outcome.status, + ); + } Ok(result) } } @@ -295,15 +326,17 @@ async fn run_branches( ), }; if emit_completion { - emit_branch_completed( - &services.run.emitter, - &dispatch.scope, - parallel_group_id.clone(), - dispatch.branch_id, - dispatch.index, - 0, - result.outcome.status, - ); + if let Some(scope) = dispatch.scope.get() { + emit_branch_completed( + &services.run.emitter, + scope, + parallel_group_id.clone(), + dispatch.branch_id, + dispatch.index, + 0, + result.outcome.status, + ); + } } if result.outcome.failure_category() == Some(FailureCategory::Canceled) { cancelled = true; @@ -799,6 +832,7 @@ mod tests { logger.register(services.run.emitter.as_ref()); let (node, graph) = parallel_graph(); let context = test_context(); + context.set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(2)); let outcome = ParallelHandler .execute(&node, &context, &graph, Path::new("/tmp/test"), &services) @@ -825,7 +859,7 @@ mod tests { let state = run_store.state().await.unwrap(); assert_eq!( state - .stage(&StageId::new("par", 1)) + .stage(&StageId::new("par", 2)) .unwrap() .parallel_results .as_ref() @@ -833,6 +867,15 @@ mod tests { .len(), 2 ); + for branch in ["branch_a", "branch_b"] { + assert_eq!( + state + .stage(&StageId::new(branch, 1)) + .and_then(|stage| stage.graph_visit), + Some(2), + "parallel children should inherit the parent graph visit" + ); + } } #[tokio::test] diff --git a/lib/components/fabro-workflow/src/lib.rs b/lib/components/fabro-workflow/src/lib.rs index b4d3cc727..ee3ec4566 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(crate) 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..3531f7bbc 100644 --- a/lib/components/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/components/fabro-workflow/src/lifecycle/artifact.rs @@ -20,9 +20,10 @@ use crate::artifact_snapshot::{ArtifactCollectionSummary, collect_artifacts}; use crate::artifact_upload::ArtifactSink; use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel}; use crate::graph::{WorkflowGraph, WorkflowNode}; -use crate::lifecycle::event::{stage_scope_for, stage_visit}; +use crate::lifecycle::event::stage_scope_for; use crate::outcome::BilledModelUsage; use crate::runtime_store::RunStoreHandle; +use crate::stage_execution::StageExecutionTracker; type WfRunState = ExecutionState>; 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,10 @@ 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 scope = stage_scope_for(&self.stage_executions, state, node_id); + let visit = scope.visit; let node_slug = if visit <= 1 { node_id.to_string() } else { @@ -144,7 +152,7 @@ impl RunLifecycle for ArtifactLifecycle { return Ok(()); } - let stage_id = StageId::new(node_id.to_string(), visit); + let stage_id = scope.stage_id(); if let Err(err) = self .persist_artifacts( &stage_id, @@ -162,7 +170,6 @@ impl RunLifecycle for ArtifactLifecycle { return Ok(()); } self.record_captured_assets(&new_assets); - let scope = stage_scope_for(state, node_id); for asset in &new_assets { self.emitter.emit_scoped( &Event::ArtifactCaptured { diff --git a/lib/components/fabro-workflow/src/lifecycle/event.rs b/lib/components/fabro-workflow/src/lifecycle/event.rs index f547f5fe5..c4af83250 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::{StageExecution, 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( @@ -92,13 +95,16 @@ fn response_from_outcome(node_id: &str, outcome: &Outcome) -> Option { .and_then(|value| value.as_str().map(ToOwned::to_owned)) } -/// Context values for `StageCompleted` events. Unlike -/// `artifact::strip_transient_keys`, this keeps `CURRENT_PREAMBLE` — stage -/// events have always included the active preamble — and drops only the -/// parallel stash, which can embed every branch's rendered preamble. +/// Context values for `StageCompleted` events. Runtime-only keys are stripped, +/// except for `CURRENT_PREAMBLE`, which stage events have historically +/// included. fn stage_context_values(workflow_context: &Context) -> Option> { let mut snapshot = workflow_context.snapshot(); - snapshot.remove(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES); + let preamble = snapshot.get(context::keys::CURRENT_PREAMBLE).cloned(); + artifact::strip_transient_keys(&mut snapshot); + if let Some(preamble) = preamble { + snapshot.insert(context::keys::CURRENT_PREAMBLE.to_owned(), preamble); + } (!snapshot.is_empty()).then(|| snapshot.into_iter().collect()) } @@ -107,15 +113,40 @@ pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { u32::try_from(visits).unwrap_or(u32::MAX) } -pub(crate) fn stage_scope_for(state: &WfRunState, node_id: &str) -> StageScope { +fn stage_scope_from_execution( + execution: Option<&StageExecution>, + state: &WfRunState, + node_id: &str, +) -> StageScope { + let (node_id, visit) = execution.map_or_else( + || (node_id.to_owned(), stage_visit(state, node_id)), + |execution| { + ( + execution.stage_id.node_id().to_owned(), + execution.stage_id.visit(), + ) + }, + ); StageScope { - node_id: node_id.to_string(), - visit: stage_visit(state, node_id), - parallel_group_id: state.context.parallel_group_id(), + node_id, + visit, + parallel_group_id: state.context.parallel_group_id(), parallel_branch_id: state.context.parallel_branch_id(), } } +/// Build the emission scope for a node from its active stage execution. +/// Falls back to the graph visit for direct unit-test call sites that emit +/// without a reservation; the two are equal for a first execution. +pub(crate) fn stage_scope_for( + stage_executions: &StageExecutionTracker, + state: &WfRunState, + node_id: &str, +) -> StageScope { + let execution = stage_executions.active(node_id); + stage_scope_from_execution(execution.as_deref(), state, node_id) +} + #[async_trait] impl RunLifecycle for EventLifecycle { async fn on_run_start(&self, _graph: &WorkflowGraph, _state: &WfRunState) -> CoreResult<()> { @@ -160,17 +191,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_from_execution(Some(&execution), state, &gv.id); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); self.emitter.emit_scoped( &Event::StageStarted { - node_id: gv.id.clone(), - name: gv.label().to_string(), - index: stage_index, - handler_type: gv.handler_type().unwrap_or_default().to_string(), - attempt: 1, - max_attempts: 1, + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: 1, + max_attempts: 1, + graph_visit: Some(execution.graph_visit), + resumed_from_stage_id: execution.resumed_from.clone(), }, &scope, ); @@ -210,15 +248,23 @@ impl RunLifecycle 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_from_execution(execution.as_deref(), state, &gv.id); + let graph_visit = execution + .as_ref() + .map_or_else(|| stage_visit(state, &gv.id), |e| e.graph_visit); self.emitter.emit_scoped( &Event::StageStarted { - node_id: gv.id.clone(), - name: gv.label().to_string(), - index: state.stage_index, - handler_type: gv.handler_type().unwrap_or_default().to_string(), - attempt: ctx.attempt as usize, - max_attempts: ctx.max_attempts as usize, + node_id: gv.id.clone(), + name: gv.label().to_string(), + index: state.stage_index, + handler_type: gv.handler_type().unwrap_or_default().to_string(), + attempt: ctx.attempt as usize, + max_attempts: ctx.max_attempts as usize, + graph_visit: Some(graph_visit), + resumed_from_stage_id: execution + .as_ref() + .and_then(|execution| execution.resumed_from.clone()), }, &scope, ); @@ -234,7 +280,7 @@ impl RunLifecycle for EventLifecycle { let gv = ctx.node.inner(); let outcome = &ctx.result.outcome; let stage_index = state.stage_index; - let scope = stage_scope_for(state, &gv.id); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); let timing = node_result_timing(ctx.result); let failure = outcome.failure.clone().unwrap_or_else(|| { @@ -283,7 +329,7 @@ impl RunLifecycle for EventLifecycle { } let gv = node.inner(); let stage_index = state.stage_index; - let scope = stage_scope_for(state, &gv.id); + let scope = stage_scope_for(&self.stage_executions, state, &gv.id); let timing = node_result_timing(result); let (loop_failure_signatures, restart_failure_signatures) = snapshot_failure_signatures(&self.circuit_breaker); @@ -400,7 +446,11 @@ impl RunLifecycle for EventLifecycle { node_outcomes.insert(node.id().to_string(), result.outcome.clone()); artifact::normalize_durable_outcomes(&mut node_outcomes); - let scope = stage_scope_for(state, node.id()); + let execution = self.stage_executions.active(node.id()); + let scope = stage_scope_from_execution(execution.as_deref(), state, node.id()); + let graph_visit = execution + .as_ref() + .map_or_else(|| stage_visit(state, node.id()), |e| e.graph_visit); self.emitter.emit_scoped( &Event::CheckpointCompleted { node_id: node.id().to_string(), @@ -425,6 +475,10 @@ impl RunLifecycle for EventLifecycle { .collect::>(), diff, diff_summary, + graph_visit: Some(graph_visit), + resumed_from_stage_id: execution + .as_ref() + .and_then(|execution| execution.resumed_from.clone()), }, &scope, ); @@ -458,17 +512,30 @@ mod tests { use super::*; #[test] - fn stage_context_values_drops_parallel_branch_preambles() { + fn stage_context_values_drops_runtime_keys_but_keeps_current_preamble() { let workflow_context = Context::new(); workflow_context.set( context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES, serde_json::json!([{"fidelity": "summary:high", "preamble": "runtime only"}]), ); + workflow_context.set( + context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(2), + ); + workflow_context.set( + context::keys::CURRENT_PREAMBLE, + serde_json::json!("active preamble"), + ); workflow_context.set("response.work", serde_json::json!("durable")); let values = stage_context_values(&workflow_context).expect("snapshot should not be empty"); assert!(!values.contains_key(context::keys::INTERNAL_PARALLEL_BRANCH_PREAMBLES)); + assert!(!values.contains_key(context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL)); + assert_eq!( + values.get(context::keys::CURRENT_PREAMBLE), + Some(&serde_json::json!("active preamble")) + ); assert_eq!( values.get("response.work"), Some(&serde_json::json!("durable")) 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..8006c7abc 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.stage_id.visit()), + ); // Event emission self.event.before_attempt(ctx, state).await?; // Record epoch AFTER hook+event (engine.rs:968→1006) @@ -410,6 +437,17 @@ impl RunLifecycle for WorkflowLifecycle { next_node_id: Option<&str>, state: &WfRunState, ) -> CoreResult<()> { + // A StageStart hook can skip before any attempt reserved an execution + // scope. Ensure one exists so Git metadata-snapshot events and the + // `checkpoint.completed` envelope attach to a concrete execution; + // an existing reservation from the attempt path is reused as-is. + let execution = self + .stage_executions + .ensure(node.id(), event::stage_visit(state, node.id())); + state.context.set( + context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.stage_id.visit()), + ); self.git .on_checkpoint(node, result, next_node_id, state) .await?; 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..433947716 100644 --- a/lib/components/fabro-workflow/src/operations/resume.rs +++ b/lib/components/fabro-workflow/src/operations/resume.rs @@ -4,6 +4,7 @@ use super::start::{StartServices, Started, execute_persisted_run}; use crate::error::Error; use crate::event::{Event, append_event_to_sink}; use crate::outcome::StageOutcome; +use crate::pipeline::ResumeState; use crate::run_status::RunStatus; /// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. @@ -32,9 +33,7 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result Result, + resume: Option, services: StartServices, ) -> Result { let cancel_token = services.cancel_token.clone(); @@ -261,7 +262,7 @@ pub(super) async fn execute_persisted_run( cancel_token, ); let run_start = Instant::now(); - let started = Box::pin(session.run(persisted, checkpoint)).await; + let started = Box::pin(session.run(persisted, resume)).await; match started { Ok(started) => { @@ -797,7 +798,7 @@ impl RunSession { async fn run( self, persisted: Persisted, - checkpoint: Option, + resume: Option, ) -> Result { let on_node = self.on_node.clone(); @@ -878,7 +879,7 @@ impl RunSession { registry_override: self.registry_override, artifact_sink: self.artifact_sink, run_control: self.run_control, - checkpoint, + resume, seed_context: self.seed_context, fabro_run_tools: self.fabro_run_tools, }; @@ -2126,6 +2127,8 @@ reasoning = false { injected.store(true, Ordering::SeqCst); emitter_for_injection.emit(&Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: "start".to_string(), status: "succeeded".to_string(), current_node: "start".to_string(), @@ -2509,6 +2512,8 @@ reasoning = false &store.open_run(&fixtures::RUN_1).await.unwrap(), &services.run_id, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: checkpoint .node_outcomes @@ -2607,6 +2612,8 @@ reasoning = false }; let run_store = store.open_run(&fixtures::RUN_1).await.unwrap(); crate::event::append_event(&run_store, &fixtures::RUN_1, &Event::CheckpointCompleted { + graph_visit: None, + resumed_from_stage_id: None, node_id: checkpoint.current_node.clone(), status: "succeeded".to_string(), current_node: checkpoint.current_node.clone(), diff --git a/lib/components/fabro-workflow/src/pipeline/execute.rs b/lib/components/fabro-workflow/src/pipeline/execute.rs index cf02a736d..e89006ab0 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute.rs @@ -77,6 +77,7 @@ pub async fn execute(init: Initialized) -> Executed { checkpoint.is_some(), on_node, run_control, + engine.run.stage_executions.clone(), ); if let Some(ref cp) = checkpoint { diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 400050896..2322e6b71 100644 --- a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs +++ b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs @@ -31,7 +31,7 @@ use crate::handler::start::StartHandler; use crate::handler::{Handler as HandlerTrait, HandlerRegistry}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::pipeline::initialize; -use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, SandboxEnvSpec}; +use crate::pipeline::types::{InitOptions, LlmSpec, Persisted, ResumeState, SandboxEnvSpec}; use crate::records::RunSpec; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions, SetupCommand}; use crate::test_support::run_graph; @@ -291,7 +291,7 @@ async fn execute_test_run_with_options( run_control: None, registry_override, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }, @@ -353,7 +353,7 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }, @@ -375,6 +375,186 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { ); } +#[tokio::test] +async fn resumed_in_flight_node_starts_a_new_stage_execution() { + let temp = tempfile::tempdir().unwrap(); + let run_dir = temp.path().join("run"); + std::fs::create_dir_all(&run_dir).unwrap(); + + // start -> work -> exit; `work` resolves to the default (dry-run) handler. + let mut graph = Graph::new("resume_identity"); + let mut start = Node::new("start"); + start.attrs.insert( + "shape".to_string(), + AttrValue::String("Mdiamond".to_string()), + ); + graph.nodes.insert("start".to_string(), start); + graph.nodes.insert("work".to_string(), Node::new("work")); + let mut exit = Node::new("exit"); + exit.attrs.insert( + "shape".to_string(), + AttrValue::String("Msquare".to_string()), + ); + graph.nodes.insert("exit".to_string(), exit); + graph.edges.push(Edge::new("start", "work")); + graph.edges.push(Edge::new("work", "exit")); + + let run_options = test_run_options(&run_dir, "resume-identity"); + let run_id = run_options.run_id; + let run_store = test_run_store(&run_id).await; + seed_created_and_starting(&run_store, &run_options, &graph).await; + // Resume reconnects to the previously recorded sandbox. + append_event(&run_store, &run_id, &Event::SandboxInitialized { + working_directory: std::env::current_dir().unwrap().display().to_string(), + provider: fabro_types::SandboxProviderKind::Local, + id: "local".to_string(), + image: None, + snapshot: None, + repo_cloned: None, + clone_origin_url: None, + clone_branch: None, + workspace_root: None, + repos_root: None, + primary_repo_path: None, + primary_repo_link: None, + }) + .await + .unwrap(); + let emitter = test_emitter_arc("resume-identity"); + let events: Arc>> = Arc::default(); + { + let events = Arc::clone(&events); + emitter.on_event(move |event| { + events + .lock() + .expect("event capture mutex should not be poisoned") + .push(event.clone()); + }); + } + + // Simulate resuming after `work@1` was cancelled mid-flight: the selected + // checkpoint predates `work`, while the allocator seed carries the + // projection-observed high-water mark and provenance link. + let checkpoint = crate::records::Checkpoint { + timestamp: chrono::Utc::now(), + current_node: "start".to_string(), + completed_nodes: vec!["start".to_string()], + node_retries: HashMap::new(), + context_values: HashMap::new(), + node_outcomes: HashMap::new(), + next_node_id: Some("work".to_string()), + git_commit_sha: None, + loop_failure_signatures: HashMap::new(), + restart_failure_signatures: HashMap::new(), + node_visits: HashMap::from([("start".to_string(), 1usize)]), + }; + let seed = crate::stage_execution::StageExecutionSeed::test_with_high_water( + &fabro_types::StageId::new("work", 1), + Some(fabro_types::StageId::new("work", 1)), + ); + let resume = ResumeState::for_test(checkpoint, seed); + + let initialized = initialize( + persisted_workflow(graph, String::new(), &run_dir, run_id), + InitOptions { + run_store: run_store.into(), + dry_run: false, + emitter: emitter.clone(), + sandbox: SandboxSpec::Local { + working_directory: std::env::current_dir().unwrap(), + }, + llm: LlmSpec { + model: "test-model".to_string(), + provider_id: fabro_model::ProviderId::anthropic(), + fallback_chain: Vec::new(), + mcp_servers: Vec::new(), + model_controls: RunModelControls::default(), + dry_run: true, + }, + interviewer: Arc::new(AutoApproveInterviewer::engine()), + steering_hub: Arc::new(crate::steering_hub::SteeringHub::new(emitter.clone())), + catalog: test_catalog(), + lifecycle: LifecycleOptions { + setup_commands: vec![], + setup_command_timeout_ms: 1_000, + }, + run_options, + workflow_path: None, + workflow_bundle: None, + hooks: HookSettings { hooks: vec![] }, + sandbox_env: SandboxEnvSpec { + toml_env: HashMap::new(), + github_permissions: None, + origin_url: None, + }, + vault: None, + git: None, + run_control: None, + registry_override: Some(Arc::new(make_registry())), + artifact_sink: None, + resume: Some(resume), + seed_context: None, + fabro_run_tools: None, + }, + ) + .await + .unwrap(); + + let executed = execute(initialized).await; + assert_eq!(executed.outcome.unwrap().status, StageOutcome::Succeeded); + + let events = events + .lock() + .expect("event capture mutex should not be poisoned"); + let work_started = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::StageStarted(_)) + && event.node_id.as_deref() == Some("work") + }) + .expect("resumed run should emit stage.started for work"); + // The reexecution owns a fresh StageId while the graph visit stays at 1. + assert_eq!( + work_started.stage_id, + Some(fabro_types::StageId::new("work", 2)) + ); + let fabro_types::EventBody::StageStarted(props) = &work_started.body else { + panic!("expected stage.started body"); + }; + assert_eq!(props.graph_visit, Some(1)); + assert_eq!( + props.resumed_from_stage_id, + Some(fabro_types::StageId::new("work", 1)) + ); + + // Every later stage-scoped event from this invocation carries the same + // execution id, including the checkpoint envelope. + let work_checkpoint = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::CheckpointCompleted(_)) + && event.node_id.as_deref() == Some("work") + }) + .expect("resumed run should checkpoint work"); + assert_eq!( + work_checkpoint.stage_id, + Some(fabro_types::StageId::new("work", 2)) + ); + + // A node without a prior observable execution starts at ordinal 1. + let exit_started = events + .iter() + .find(|event| { + matches!(event.body, fabro_types::EventBody::StageStarted(_)) + && event.node_id.as_deref() == Some("exit") + }) + .expect("terminal node should emit its synthetic stage.started"); + assert_eq!( + exit_started.stage_id, + Some(fabro_types::StageId::new("exit", 1)) + ); +} + async fn run_with_lifecycle( registry: HandlerRegistry, emitter: Arc, @@ -423,7 +603,7 @@ async fn run_with_lifecycle( run_control: None, registry_override: Some(Arc::new(registry)), artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }, 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 cffb2fd86..60df59c4f 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::{StageExecutionSeed, StageExecutionTracker}; use crate::steering_hub::SteeringHub; type BuiltSandboxEnv = (HashMap, Option>); @@ -286,6 +287,13 @@ pub async fn initialize( mut options: InitOptions, ) -> Result { let (graph, source, _diagnostics, run_dir, run_spec) = persisted.into_parts(); + let (checkpoint, stage_executions) = options.resume.take().map_or_else( + || (None, StageExecutionSeed::default()), + |resume| { + let (checkpoint, stage_executions) = resume.into_parts(); + (Some(checkpoint), stage_executions) + }, + ); let host_source_dir = run_spec.source_directory.as_deref().map(PathBuf::from); options.run_options.run_dir = run_dir.clone(); options.run_options.git = options.git.clone(); @@ -306,7 +314,7 @@ pub async fn initialize( ))) }; - let attach_existing = options.checkpoint.is_some(); + let attach_existing = checkpoint.is_some(); options.run_options.display_base_sha = options .run_options .pre_run_git @@ -619,6 +627,7 @@ pub async fn initialize( sandbox_git, metadata_runtime, metadata_writer, + StageExecutionTracker::seeded(stage_executions), ); let engine = Arc::new(EngineServices { run: Arc::clone(&run_services), @@ -640,7 +649,7 @@ pub async fn initialize( graph, source, run_options: options.run_options, - checkpoint: options.checkpoint, + checkpoint, seed_context: options.seed_context, on_node: None, artifact_sink: options.artifact_sink, @@ -863,7 +872,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -944,7 +953,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1165,7 +1174,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1260,7 +1269,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1402,7 +1411,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) diff --git a/lib/components/fabro-workflow/src/pipeline/mod.rs b/lib/components/fabro-workflow/src/pipeline/mod.rs index ac77cc992..d0ba5ae1b 100644 --- a/lib/components/fabro-workflow/src/pipeline/mod.rs +++ b/lib/components/fabro-workflow/src/pipeline/mod.rs @@ -23,7 +23,7 @@ pub use pull_request::{ pub use transform::transform; pub use types::{ Concluded, Executed, FinalizeOptions, Finalized, InitOptions, Initialized, LlmSpec, Parsed, - Persisted, PullRequestOptions, SandboxEnvSpec, TEMPLATE_UNDEFINED_VARIABLE_RULE, + Persisted, PullRequestOptions, ResumeState, SandboxEnvSpec, TEMPLATE_UNDEFINED_VARIABLE_RULE, TransformOptions, Transformed, Validated, }; pub use validate::validate; diff --git a/lib/components/fabro-workflow/src/pipeline/types.rs b/lib/components/fabro-workflow/src/pipeline/types.rs index 1776074a2..425cba9aa 100644 --- a/lib/components/fabro-workflow/src/pipeline/types.rs +++ b/lib/components/fabro-workflow/src/pipeline/types.rs @@ -9,7 +9,7 @@ use fabro_model::{Catalog, FallbackTarget, ProviderId}; use fabro_sandbox::SandboxSpec; use fabro_template::TemplateContext; use fabro_types::settings::run::{PullRequestSettings, RunModelControls}; -use fabro_types::{ManifestPath, RunId}; +use fabro_types::{ManifestPath, RunId, RunProjection}; use fabro_validate::{Diagnostic, Severity}; use fabro_vault::Vault; use tokio::sync::RwLock as AsyncRwLock; @@ -26,6 +26,7 @@ use crate::run_control::RunControlState; use crate::run_options::{GitCheckpointOptions, LifecycleOptions, RunOptions}; use crate::runtime_store::RunStoreHandle; use crate::services::{EngineServices, FabroRunToolServices, RunServices}; +use crate::stage_execution::StageExecutionSeed; use crate::steering_hub::SteeringHub; use crate::transforms::{RenderMode, Transform}; use crate::workflow_bundle::WorkflowBundle; @@ -248,6 +249,41 @@ pub struct SandboxEnvSpec { pub origin_url: Option, } +/// Opaque, internally consistent state needed to resume from the latest +/// checkpoint in a run projection. +pub struct ResumeState { + checkpoint: Checkpoint, + stage_executions: StageExecutionSeed, +} + +impl ResumeState { + /// Build resume state from a projection's latest checkpoint and complete + /// stage history. + #[must_use] + pub fn from_projection(projection: &RunProjection) -> Option { + let checkpoint_record = projection.checkpoints.last()?; + Some(Self { + checkpoint: checkpoint_record.checkpoint.clone(), + stage_executions: StageExecutionSeed::from_projection( + projection, + checkpoint_record.seq, + ), + }) + } + + pub(crate) fn into_parts(self) -> (Checkpoint, StageExecutionSeed) { + (self.checkpoint, self.stage_executions) + } + + #[cfg(test)] + pub(crate) fn for_test(checkpoint: Checkpoint, stage_executions: StageExecutionSeed) -> Self { + Self { + checkpoint, + stage_executions, + } + } +} + pub struct InitOptions { pub run_store: RunStoreHandle, pub dry_run: bool, @@ -268,7 +304,7 @@ pub struct InitOptions { pub registry_override: Option>, pub artifact_sink: Option, pub run_control: Option>, - pub checkpoint: Option, + pub resume: Option, pub seed_context: Option, pub fabro_run_tools: Option, } diff --git a/lib/components/fabro-workflow/src/services.rs b/lib/components/fabro-workflow/src/services.rs index 506206f45..2af76a9c7 100644 --- a/lib/components/fabro-workflow/src/services.rs +++ b/lib/components/fabro-workflow/src/services.rs @@ -21,6 +21,7 @@ use crate::interview_runtime::RunInterviewBlocker; use crate::run_metadata::{RunMetadataRuntime, RunMetadataWriterHandle}; use crate::runtime_store::RunStoreHandle; use crate::sandbox_git_runtime::SandboxGitRuntime; +use crate::stage_execution::StageExecutionTracker; use crate::workflow_bundle::WorkflowBundle; #[derive(Clone, Debug, PartialEq, Eq)] @@ -106,6 +107,9 @@ pub struct RunServices { pub(crate) metadata_runtime: Arc, 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 { @@ -124,6 +128,7 @@ impl RunServices { sandbox_git: Arc, metadata_runtime: Arc, metadata_writer: Option, + stage_executions: StageExecutionTracker, ) -> Arc { Arc::new(Self { run_store, @@ -140,6 +145,7 @@ impl RunServices { metadata_runtime, metadata_writer, interview_blocker: Arc::new(RunInterviewBlocker::new()), + stage_executions, }) } @@ -320,6 +326,7 @@ impl EngineServices { Arc::new(SandboxGitRuntime::new()), Arc::new(RunMetadataRuntime::new()), None, + StageExecutionTracker::default(), ), registry: Arc::new(HandlerRegistry::new(Box::new(start::StartHandler))), interviewer: Arc::new(fabro_interview::AutoApproveInterviewer::engine()), 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..153382040 --- /dev/null +++ b/lib/components/fabro-workflow/src/stage_execution.rs @@ -0,0 +1,343 @@ +//! Run-scoped stage execution identity. +//! +//! A *stage execution* is one top-level handler invocation of a node that +//! became observable within a run. Its 1-based ordinal is the numeric +//! component of the external `StageId` (`node_id@N`). The ordinal is distinct +//! from the *graph visit* (how many times workflow control entered the node, +//! which drives `max_visits` and checkpoints) and from the *handler attempt* +//! (automatic retries inside one execution). +//! +//! The tracker is deliberately not checkpointed: its durable source of truth +//! is the append-only stage event history. On resume it is seeded from the +//! run projection's per-node maxima, so a reexecuted in-flight node allocates +//! the next unused ordinal instead of mutating the prior execution. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; + +use fabro_types::{RunProjection, StageId}; + +/// One reserved stage execution: the identity of a single resumable handler +/// invocation of a node. +#[derive(Debug, PartialEq, Eq)] +pub(crate) struct StageExecution { + /// Canonical external identity for this execution. + pub stage_id: StageId, + /// Graph visit that produced this execution. + pub graph_visit: u32, + /// Prior post-checkpoint execution superseded by this resumed execution. + pub resumed_from: Option, +} + +#[derive(Debug, Default)] +struct NodeExecutionState { + /// Highest execution ordinal observed or reserved for this node. + high_water: u32, + /// Pending provenance link, consumed by the next reservation. + resumed_from: Option, + /// Execution reserved since the latest node boundary. + active: 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(Debug, Default)] +pub(crate) struct StageExecutionSeed { + nodes: 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 replay supersedes. + #[must_use] + pub(crate) fn from_projection(projection: &RunProjection, checkpoint_seq: u32) -> Self { + let mut nodes = HashMap::new(); + for (stage_id, stage) in projection.iter_stages_unordered() { + let entry = nodes + .entry(stage_id.node_id().to_owned()) + .or_insert_with(NodeExecutionState::default); + entry.high_water = entry.high_water.max(stage_id.visit()); + if stage.first_event_seq.get() > checkpoint_seq { + let is_latest = entry + .resumed_from + .as_ref() + .is_none_or(|current| current.visit() < stage_id.visit()); + if is_latest { + entry.resumed_from = Some(stage_id.clone()); + } + } + } + Self { nodes } + } + + #[cfg(test)] + pub(crate) fn test_with_high_water( + high_water: &StageId, + resumed_from: Option, + ) -> Self { + let node_id = high_water.node_id().to_owned(); + Self { + nodes: HashMap::from([(node_id, NodeExecutionState { + high_water: high_water.visit(), + resumed_from, + active: None, + })]), + } + } +} + +/// Cloneable, run-scoped allocator for stage execution ordinals. Clones share +/// one synchronized state so the core lifecycle and direct-dispatch handlers +/// (parallel branches) allocate from the same sequence. +#[derive(Clone, Debug, Default)] +pub(crate) struct StageExecutionTracker { + state: Arc>>, +} + +impl StageExecutionTracker { + #[must_use] + pub(crate) fn seeded(seed: StageExecutionSeed) -> Self { + Self { + state: Arc::new(Mutex::new(seed.nodes)), + } + } + + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.state + .lock() + .expect("stage execution tracker mutex is never poisoned: no code panics while holding this lock") + } + + /// Clear the node's prior execution scope at the node boundary. The next + /// `reserve`/`ensure` call allocates a fresh ordinal; a reservation is not + /// made here so that a StageStart hook block or process exit before any + /// stage-scoped event leaves no phantom execution. + pub(crate) fn begin_node(&self, node_id: &str) { + if let Some(node) = self.lock().get_mut(node_id) { + node.active = None; + } + } + + /// The node's active execution scope, if one has been reserved since the + /// last node boundary. + pub(crate) fn active(&self, node_id: &str) -> Option> { + self.lock() + .get(node_id) + .and_then(|node| node.active.as_ref().map(Arc::clone)) + } + + fn reserve_locked( + state: &mut HashMap, + node_id: &str, + graph_visit: u32, + ) -> Arc { + let node = state.entry(node_id.to_owned()).or_default(); + node.high_water = node.high_water.saturating_add(1); + let execution = Arc::new(StageExecution { + stage_id: StageId::new(node_id, node.high_water), + graph_visit, + resumed_from: node.resumed_from.take(), + }); + node.active = Some(Arc::clone(&execution)); + execution + } + + /// Allocate the next execution ordinal for the node and make it the active + /// scope. Consumes the node's pending provenance link, if any. + pub(crate) fn reserve(&self, node_id: &str, graph_visit: u32) -> Arc { + let mut state = self.lock(); + Self::reserve_locked(&mut state, node_id, graph_visit) + } + + /// The active scope for the node, reserving one only when none exists. + /// Later attempts within one execution and checkpoint pre-steps reuse the + /// first attempt's reservation. + pub(crate) fn ensure(&self, node_id: &str, graph_visit: u32) -> Arc { + let mut state = self.lock(); + if let Some(execution) = state + .get(node_id) + .and_then(|node| node.active.as_ref().map(Arc::clone)) + { + return execution; + } + Self::reserve_locked(&mut state, node_id, graph_visit) + } +} + +#[cfg(test)] +mod tests { + use std::num::NonZeroU32; + + use chrono::Utc; + use fabro_types::{Graph, RunId, RunSpec, StageId, WorkflowSettings, test_support}; + + use super::*; + + fn projection_with_stages(stages: &[(&str, u32, u32)]) -> RunProjection { + let spec = RunSpec { + run_id: RunId::new(), + settings: WorkflowSettings::default(), + graph: Graph::new("test"), + graph_source: None, + workflow_slug: None, + automation: None, + source_directory: None, + labels: std::collections::HashMap::new(), + provenance: test_support::test_run_provenance(), + manifest_blob: None, + definition_blob: None, + git: None, + fork_source_ref: None, + }; + let mut projection = RunProjection::new(String::new(), spec, Utc::now()); + for (node_id, visit, seq) in stages { + projection.stage_entry( + node_id, + *visit, + NonZeroU32::new(*seq).expect("test seq must be non-zero"), + ); + } + projection + } + + #[test] + fn reserve_starts_at_one_and_allocates_monotonically_per_node() { + let tracker = StageExecutionTracker::default(); + + assert_eq!(tracker.reserve("work", 1).stage_id.visit(), 1); + tracker.begin_node("work"); + assert_eq!(tracker.reserve("work", 2).stage_id.visit(), 2); + assert_eq!(tracker.reserve("other", 1).stage_id.visit(), 1); + } + + #[test] + fn seeds_from_projection_maxima() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5), ("plan", 1, 3)]); + let seed = StageExecutionSeed::from_projection(&projection, 0); + let tracker = StageExecutionTracker::seeded(seed); + + assert_eq!(tracker.reserve("work", 1).stage_id.visit(), 3); + assert_eq!(tracker.reserve("plan", 1).stage_id.visit(), 2); + assert_eq!(tracker.reserve("new", 1).stage_id.visit(), 1); + } + + #[test] + fn graph_visit_and_ordinal_can_diverge() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 5)]); + let seed = StageExecutionSeed::from_projection(&projection, 0); + let tracker = StageExecutionTracker::seeded(seed); + + let execution = tracker.reserve("work", 2); + assert_eq!(execution.stage_id.visit(), 3); + assert_eq!(execution.graph_visit, 2); + } + + #[test] + fn ensure_reuses_active_reservation_across_attempts() { + let tracker = StageExecutionTracker::default(); + + let first = tracker.ensure("work", 1); + let second = tracker.ensure("work", 1); + assert_eq!(first, second); + assert_eq!(second.stage_id.visit(), 1); + + tracker.begin_node("work"); + assert_eq!(tracker.ensure("work", 2).stage_id.visit(), 2); + } + + #[test] + fn begin_node_clears_only_that_node() { + let tracker = StageExecutionTracker::default(); + tracker.reserve("work", 1); + tracker.reserve("verify", 1); + + tracker.begin_node("work"); + + assert_eq!(tracker.active("work"), None); + assert_eq!( + tracker + .active("verify") + .map(|execution| execution.stage_id.visit()), + Some(1) + ); + } + + #[test] + fn provenance_only_selects_stages_after_the_checkpoint() { + let projection = projection_with_stages(&[("work", 1, 2), ("work", 2, 8), ("plan", 1, 3)]); + let seed = StageExecutionSeed::from_projection(&projection, 5); + + assert_eq!( + seed.nodes + .get("work") + .and_then(|node| node.resumed_from.as_ref()), + Some(&StageId::new("work", 2)) + ); + assert_eq!( + seed.nodes + .get("plan") + .and_then(|node| node.resumed_from.as_ref()), + None + ); + } + + #[test] + fn first_reservation_consumes_provenance() { + let projection = projection_with_stages(&[("work", 1, 6)]); + let seed = StageExecutionSeed::from_projection(&projection, 5); + let tracker = StageExecutionTracker::seeded(seed); + + let first = tracker.reserve("work", 1); + assert_eq!(first.stage_id.visit(), 2); + assert_eq!(first.resumed_from, Some(StageId::new("work", 1))); + + tracker.begin_node("work"); + let second = tracker.reserve("work", 2); + assert_eq!(second.stage_id.visit(), 3); + assert_eq!(second.resumed_from, None); + } + + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_reservations_stay_unique_per_node() { + let tracker = StageExecutionTracker::default(); + let handles: Vec<_> = (0..8) + .map(|_| { + let tracker = tracker.clone(); + tokio::spawn(async move { tracker.reserve("branch", 1).stage_id.visit() }) + }) + .collect(); + + let mut ordinals = Vec::new(); + for handle in handles { + ordinals.push(handle.await.expect("reservation task panicked")); + } + ordinals.sort_unstable(); + assert_eq!(ordinals, (1..=8).collect::>()); + } + + #[tokio::test(flavor = "multi_thread")] + async fn concurrent_ensure_calls_reuse_one_reservation() { + let tracker = StageExecutionTracker::default(); + let barrier = Arc::new(tokio::sync::Barrier::new(16)); + let handles: Vec<_> = (0..16) + .map(|_| { + let tracker = tracker.clone(); + let barrier = Arc::clone(&barrier); + tokio::spawn(async move { + barrier.wait().await; + tracker.ensure("branch", 1).stage_id.visit() + }) + }) + .collect(); + + for handle in handles { + assert_eq!(handle.await.expect("ensure task panicked"), 1); + } + } +} diff --git a/lib/components/fabro-workflow/src/stage_scope.rs b/lib/components/fabro-workflow/src/stage_scope.rs index 396e84156..fe6d8b08a 100644 --- a/lib/components/fabro-workflow/src/stage_scope.rs +++ b/lib/components/fabro-workflow/src/stage_scope.rs @@ -1,11 +1,30 @@ use fabro_types::{ParallelBranchId, StageId}; -use crate::context::{Context as WfContext, WorkflowContext}; +use crate::context::{Context as WfContext, WorkflowContext, keys}; use crate::run_dir::visit_from_context; +/// Read the stage execution ordinal seeded by the workflow lifecycle (or a +/// parallel branch dispatch). Direct-handler call sites that skip the full +/// lifecycle fall back to the graph visit, which equals the ordinal for a +/// first execution. +pub(crate) fn execution_ordinal_from_context(context: &WfContext) -> u32 { + context + .get(keys::INTERNAL_STAGE_EXECUTION_ORDINAL) + .and_then(|value| value.as_u64()) + .map_or_else( + || u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), + |ordinal| u32::try_from(ordinal).unwrap_or(u32::MAX), + ) +} + /// Stage-level scope threaded through event emission to populate /// `stage_id` / `parallel_group_id` / `parallel_branch_id` on events /// that happen inside a concrete stage execution. +/// +/// `visit` is the 1-based stage execution ordinal — the numeric component of +/// the external `StageId`. It matches the graph visit for a first execution +/// and diverges when post-checkpoint work is replayed after +/// resume. #[derive(Clone, Debug)] pub struct StageScope { pub node_id: String, @@ -15,13 +34,14 @@ pub struct StageScope { } impl StageScope { - /// Build a scope from the given node id, sourcing visit count and parallel - /// ids from the current context. + /// Build a scope from the given node id, sourcing the execution ordinal + /// and parallel ids from the current context. pub fn from_context(context: &WfContext, node_id: impl Into) -> Self { + let visit = execution_ordinal_from_context(context); Self { - node_id: node_id.into(), - visit: u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX), - parallel_group_id: context.parallel_group_id(), + node_id: node_id.into(), + visit, + parallel_group_id: context.parallel_group_id(), parallel_branch_id: context.parallel_branch_id(), } } @@ -39,11 +59,10 @@ impl StageScope { /// Build scope for the branch-lifecycle events emitted by the parallel /// handler (`ParallelBranchStarted` and `ParallelBranchCompleted`). /// - /// `target_visit` is the visit count of `target_node_id` for this - /// particular branch dispatch. The parallel handler currently passes - /// `1` because branches haven't been re-entered yet at the point of - /// scope construction; a future change that loops a parallel node - /// must pass the actual visit so envelope `stage_id`s stay accurate. + /// `target_visit` is the branch target's stage execution ordinal for this + /// particular dispatch, reserved through the run's shared + /// `StageExecutionTracker` so a resumed fan-out gets a fresh child + /// identity instead of overwriting the prior dispatch's. #[must_use] pub fn for_parallel_branch( target_node_id: impl Into, diff --git a/lib/components/fabro-workflow/src/test_support.rs b/lib/components/fabro-workflow/src/test_support.rs index edf0d8af2..5ece3320c 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-api/build.rs b/lib/foundation/fabro-api/build.rs index 96665e176..8d9b615ee 100644 --- a/lib/foundation/fabro-api/build.rs +++ b/lib/foundation/fabro-api/build.rs @@ -344,6 +344,7 @@ fn main() { ("StageCompletion", "fabro_types::StageCompletion", &[]), ("Conclusion", "fabro_types::Conclusion", &[]), ("StageOutcome", "fabro_types::StageOutcome", &[]), + ("StageId", "fabro_types::StageId", &[]), ("StageHandler", "fabro_types::StageHandler", &[]), ("StageState", "fabro_types::StageState", &[]), ("AgentControlState", "fabro_types::AgentControlState", &[]), diff --git a/lib/foundation/fabro-api/src/lib.rs b/lib/foundation/fabro-api/src/lib.rs index 6bd8ea14e..43d71a480 100644 --- a/lib/foundation/fabro-api/src/lib.rs +++ b/lib/foundation/fabro-api/src/lib.rs @@ -67,7 +67,7 @@ pub mod types { SessionSummary, SessionTurn, SkillsProjection, StageCompletion, StageContextWindow, StageContextWindowBreakdownItem, StageContextWindowCategory, StageContextWindowCountMethod, StageContextWindowProjection, StageContextWindowStaleness, - StageContextWindowUnavailableReason, StageContextWindowWarning, StageHandler, + StageContextWindowUnavailableReason, StageContextWindowWarning, StageHandler, StageId, StageModelUsage, StageOutcome, StageProjection, StageState, SubAgentProjection, SubAgentStatus, SystemActorKind, SystemIntegrationStatus, SystemIntegrationsResponse, TodoListProjection, TurnId, UpdateVariableRequest, UserPrincipal, Variable, diff --git a/lib/foundation/fabro-api/tests/stage_id_round_trip.rs b/lib/foundation/fabro-api/tests/stage_id_round_trip.rs new file mode 100644 index 000000000..23708d2ca --- /dev/null +++ b/lib/foundation/fabro-api/tests/stage_id_round_trip.rs @@ -0,0 +1,31 @@ +use std::any::{TypeId, type_name}; + +use fabro_api::types::StageId as ApiStageId; +use fabro_types::StageId; +use serde_json::json; + +#[test] +fn stage_id_reuses_canonical_type() { + assert_same_type::(); +} + +#[test] +fn stage_id_round_trips_openapi_representation() { + let stage_id = StageId::new("verify", 2); + + assert_eq!(serde_json::to_value(&stage_id).unwrap(), json!("verify@2")); + assert_eq!( + serde_json::from_value::(json!("verify@2")).unwrap(), + stage_id + ); +} + +fn assert_same_type() { + assert_eq!( + TypeId::of::(), + TypeId::of::(), + "{} should be the same type as {}", + type_name::(), + type_name::() + ); +} diff --git a/lib/foundation/fabro-types/src/run_event/misc.rs b/lib/foundation/fabro-types/src/run_event/misc.rs index 84276a3bd..d3bf9c1c0 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -1,7 +1,7 @@ use serde::{Deserialize, Serialize}; use super::ExecOutputTail; -use crate::{CommandTermination, ParallelBranchResult, PullRequestLink, StageOutcome}; +use crate::{CommandTermination, ParallelBranchResult, PullRequestLink, StageId, StageOutcome}; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] pub struct InterviewOption { @@ -21,7 +21,15 @@ pub struct ParallelStartedProps { #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub struct ParallelBranchStartedProps { - pub index: usize, + pub index: usize, + /// Graph visit of the branch target for this dispatch. The envelope + /// `stage_id` ordinal counts executions, so a resumed fan-out's branches + /// keep visit metadata even though their ordinals advanced. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior branch execution superseded by this resumed replay. + #[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..18a5319b0 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 post-checkpoint work is replayed after + /// resume. Absent on events written before stage execution identity. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior execution superseded by this resumed replay. + #[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 16833546d..00065b0d6 100644 --- a/lib/foundation/fabro-types/src/run_projection.rs +++ b/lib/foundation/fabro-types/src/run_projection.rs @@ -319,59 +319,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 + /// post-checkpoint work is replayed after resume. Absent on + /// projections built from events written before stage execution identity. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub graph_visit: Option, + /// Prior execution superseded by this resumed replay. + #[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( @@ -485,6 +497,8 @@ impl StageProjection { termination: None, started_at: None, handler: None, + graph_visit: None, + resumed_from_stage_id: None, state: StageState::Running, } } @@ -518,14 +532,24 @@ impl StageProjection { self.timing.map(|timing| timing.wall_time_ms) } - /// Begin a new attempt (or visit) for this stage: clear every + /// Begin a new automatic attempt within this stage execution: clear every /// per-attempt field so prior-attempt data does not leak, then record /// `started_at` and `state = Running`. Preserves `first_event_seq` - /// (identity / sort key). + /// (identity / sort key) and the execution identity metadata + /// (`graph_visit`, `resumed_from_stage_id`). + /// + /// One stage projection represents one execution; a replay after resume + /// gets a new `StageId` and never flows through here. Replays of legacy + /// histories with duplicate `stage.started` events for one `StageId` + /// retain this last-attempt behavior. pub fn begin_attempt(&mut self, started_at: DateTime, handler: StageHandler) { + let graph_visit = self.graph_visit; + let resumed_from_stage_id = self.resumed_from_stage_id.take(); *self = Self::new(self.first_event_seq); self.started_at = Some(started_at); self.handler = Some(handler); + self.graph_visit = graph_visit; + self.resumed_from_stage_id = resumed_from_stage_id; self.state = StageState::Running; } } @@ -569,11 +593,18 @@ impl RunProjection { self.stages.get(stage) } + /// Iterate stages in unspecified order without allocating or sorting. + /// + /// Use this only for order-independent aggregation. Presentation and + /// serialization callers should use [`Self::iter_stages`] instead. + pub fn iter_stages_unordered(&self) -> impl Iterator { + self.stages.iter() + } + /// Iterate stages in `first_event_seq` order (the chronological order in /// which each stage's first lifecycle event was recorded). Internal - /// storage is a `HashMap`, so iteration would otherwise be - /// non-deterministic; every caller wants chronological order, so we sort - /// here once instead of asking each caller to remember. + /// storage is a `HashMap`, so presentation callers sort through this + /// helper instead of relying on non-deterministic map iteration. pub fn iter_stages(&self) -> impl Iterator { let mut entries: Vec<(&StageId, &StageProjection)> = self.stages.iter().collect(); entries.sort_by(|(left_id, left_stage), (right_id, right_stage)| { 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..aafd25f92 100644 --- a/lib/packages/fabro-api-client/src/models/run-stage.ts +++ b/lib/packages/fabro-api-client/src/models/run-stage.ts @@ -28,7 +28,7 @@ import type { StageState } from './stage-state'; */ export interface RunStage { /** - * StageId in \"node_id@visit\" form, e.g. verify@2. + * Canonical stage execution identifier in `node_id@visit` form. */ 'id': string; /** @@ -46,9 +46,17 @@ export interface RunStage { */ 'node_id': string; /** - * 1-based visit count; bumped each time the workflow re-enters this node. + * 1-based stage execution ordinal, the numeric component of `id`. It increments each time the node produces a new observable execution: graph re-entry (loops) and replay of post-checkpoint work after resume. Automatic in-place retries do not increment it. */ 'visit': number; + /** + * 1-based count of how many times workflow control entered this node (drives `max_visits`). Differs from `visit` when a post-checkpoint execution is replayed after resume. Absent for stages recorded before execution identity was tracked. + */ + 'graph_visit'?: number | null; + /** + * Canonical stage execution identifier in `node_id@visit` form. + */ + 'resumed_from_stage_id'?: string | null; 'provider_used'?: StageModelUsage | null; /** * Wall-clock time the latest attempt of this stage started, if known.