From 78fea736e39b68ca2e83a5c771bff5ebbcff0ee8 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 24 Jul 2026 09:37:05 -0400 Subject: [PATCH] fix: harden stage execution identity on resume --- .../app/components/stage-popover.test.tsx | 6 +- .../app/components/stage-popover.tsx | 2 +- apps/fabro-web/app/lib/stage-sidebar.ts | 4 +- apps/fabro-web/app/routes/run-stages.tsx | 2 +- docs/public/api-reference/fabro-api.yaml | 28 ++- lib/apps/fabro-cli/tests/it/cmd/attach.rs | 1 - lib/apps/fabro-server/src/demo/mod.rs | 54 ++--- lib/apps/fabro-server/src/server.rs | 29 +-- .../src/server/handler/billing.rs | 58 +++-- lib/components/fabro-store/src/run_state.rs | 31 ++- lib/components/fabro-workflow/src/context.rs | 2 +- .../fabro-workflow/src/event/events.rs | 10 +- .../src/handler/manager_loop.rs | 9 +- .../fabro-workflow/src/handler/parallel.rs | 56 +++-- lib/components/fabro-workflow/src/lib.rs | 2 +- .../fabro-workflow/src/lifecycle/artifact.rs | 11 +- .../fabro-workflow/src/lifecycle/event.rs | 80 +++++-- .../fabro-workflow/src/lifecycle/mod.rs | 4 +- .../fabro-workflow/src/operations/resume.rs | 19 +- .../fabro-workflow/src/operations/start.rs | 23 +- .../src/pipeline/execute/tests.rs | 23 +- .../fabro-workflow/src/pipeline/initialize.rs | 30 +-- .../fabro-workflow/src/pipeline/mod.rs | 2 +- .../fabro-workflow/src/pipeline/types.rs | 43 +++- .../fabro-workflow/src/stage_execution.rs | 206 +++++++++++------- .../fabro-workflow/src/stage_scope.rs | 17 +- lib/foundation/fabro-api/build.rs | 1 + lib/foundation/fabro-api/src/lib.rs | 2 +- .../fabro-api/tests/stage_id_round_trip.rs | 31 +++ .../fabro-types/src/run_event/misc.rs | 2 +- .../fabro-types/src/run_event/stage.rs | 4 +- .../fabro-types/src/run_projection.rs | 27 ++- .../fabro-api-client/src/models/run-stage.ts | 8 +- 33 files changed, 477 insertions(+), 350 deletions(-) create mode 100644 lib/foundation/fabro-api/tests/stage_id_round_trip.rs diff --git a/apps/fabro-web/app/components/stage-popover.test.tsx b/apps/fabro-web/app/components/stage-popover.test.tsx index 45a1a69a4..4a8bbb12c 100644 --- a/apps/fabro-web/app/components/stage-popover.test.tsx +++ b/apps/fabro-web/app/components/stage-popover.test.tsx @@ -283,7 +283,7 @@ describe("StagePopover rendering", () => { id: "implement@2", visit: 2, graphVisit: 1, - resumedFromStageId: "implement@1", + resumedFromStageId: "review/security@1", status: "running", duration: "--", }); @@ -294,10 +294,10 @@ describe("StagePopover rendering", () => { ); const text = textOf(tree); expect(text).toContain("Resumed from"); - expect(text).toContain("implement@1"); + expect(text).toContain("review/security@1"); expect(text).toContain("Graph visit"); const json = JSON.stringify(tree.toJSON()); - expect(json).toContain("/runs/run-1/stages/implement@1"); + expect(json).toContain("/runs/run-1/stages/review%2Fsecurity%401"); }); test("stage without ordinal divergence hides the graph visit row", () => { diff --git a/apps/fabro-web/app/components/stage-popover.tsx b/apps/fabro-web/app/components/stage-popover.tsx index b8f03cc28..8dce68fc3 100644 --- a/apps/fabro-web/app/components/stage-popover.tsx +++ b/apps/fabro-web/app/components/stage-popover.tsx @@ -229,7 +229,7 @@ export function StagePopover({ runId, stage, duration }: StagePopoverProps) { {stage.resumedFromStageId && ( {stage.resumedFromStageId} diff --git a/apps/fabro-web/app/lib/stage-sidebar.ts b/apps/fabro-web/app/lib/stage-sidebar.ts index a0228495d..c165adee0 100644 --- a/apps/fabro-web/app/lib/stage-sidebar.ts +++ b/apps/fabro-web/app/lib/stage-sidebar.ts @@ -19,11 +19,11 @@ export interface Stage { nodeId: string; /** * How many times workflow control entered this node. Differs from `visit` - * when a cancelled or crashed execution was reexecuted after resume; null + * 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 this stage resumes from, if any. */ + /** StageId of the prior execution superseded by this resumed replay, if any. */ resumedFromStageId: string | null; startedAt: string | null; providerUsed: StageModelUsage | null; diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index 2c497522b..7041e5f8a 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/apps/fabro-web/app/routes/run-stages.tsx @@ -1749,7 +1749,7 @@ function RunStageActivityStage({

Resumed from{" "} {selectedStage.resumedFromStageId} diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 4d5e68922..fdcf980c1 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/docs/public/api-reference/fabro-api.yaml @@ -12369,6 +12369,11 @@ components: - running - waiting_for_steer + StageId: + description: Canonical stage execution identifier in `node_id@visit` form. + type: string + example: verify@2 + StageState: description: Lifecycle projection state of a workflow stage. type: string @@ -12410,9 +12415,7 @@ components: - visit properties: id: - type: string - description: StageId in "node_id@visit" form, e.g. verify@2. - example: verify@2 + $ref: "#/components/schemas/StageId" name: type: string description: Human-readable stage name. @@ -12438,8 +12441,8 @@ components: description: >- 1-based stage execution ordinal, the numeric component of `id`. It increments each time the node produces a new observable execution: - graph re-entry (loops) and reexecution after cancel or crash - recovery. Automatic in-place retries do not increment it. + 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"] @@ -12447,16 +12450,17 @@ components: minimum: 1 description: >- 1-based count of how many times workflow control entered this node - (drives `max_visits`). Differs from `visit` when a cancelled or - crashed execution was reexecuted after resume. Absent for stages - recorded before execution identity was tracked. + (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: - type: ["string", "null"] + oneOf: + - $ref: "#/components/schemas/StageId" + - type: "null" description: >- - StageId of the prior cancelled or interrupted execution this stage - resumes from, when the run was resumed after that execution became - observable. + StageId of the prior post-checkpoint execution superseded by this + replay after the run was resumed. example: verify@1 provider_used: oneOf: diff --git a/lib/apps/fabro-cli/tests/it/cmd/attach.rs b/lib/apps/fabro-cli/tests/it/cmd/attach.rs index c801ead11..c1ff07165 100644 --- a/lib/apps/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/apps/fabro-cli/tests/it/cmd/attach.rs @@ -1196,7 +1196,6 @@ fn attach_json_errors_without_prompting_for_human_input() { "internal.fidelity": "compact", "internal.node_visit_count": 1, "internal.run_id": "[ULID]", - "internal.stage_execution_ordinal": 1, "internal.thread_id": null }, "index": 0, diff --git a/lib/apps/fabro-server/src/demo/mod.rs b/lib/apps/fabro-server/src/demo/mod.rs index 74288c809..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,60 +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, - None, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("propose-changes", 1), "Propose Changes", StageState::Succeeded, Some(154_000), - None, StageHandler::Agent, - None, - None, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("review-changes", 1), "Review Changes", StageState::Succeeded, Some(45_000), - None, StageHandler::Agent, - None, - None, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("apply-changes", 1), "Apply Changes", StageState::Succeeded, Some(118_000), - None, StageHandler::Command, - None, - None, - None, ), - run_stage_from_stage_id( + stage( &StageId::new("apply-changes", 2), "Apply Changes", StageState::Running, None, - None, StageHandler::Command, - None, - None, - None, ), ] } diff --git a/lib/apps/fabro-server/src/server.rs b/lib/apps/fabro-server/src/server.rs index 0bee9c04c..73955c629 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,33 +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, - graph_visit: Option, - resumed_from_stage_id: Option<&StageId>, -) -> 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, - graph_visit: graph_visit.and_then(std::num::NonZeroU32::new), - resumed_from_stage_id: resumed_from_stage_id.map(StageId::to_string), - } -} - 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 6c0c315e6..e934d9fb4 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::{ ApiError, 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, + RunBillingTotals, 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>, @@ -41,27 +73,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(), - stage.graph_visit, - stage.resumed_from_stage_id.as_ref(), - ) - }) + .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/components/fabro-store/src/run_state.rs b/lib/components/fabro-store/src/run_state.rs index c512b0028..345665333 100644 --- a/lib/components/fabro-store/src/run_state.rs +++ b/lib/components/fabro-store/src/run_state.rs @@ -224,12 +224,12 @@ impl RunProjectionReducer for RunProjection { } EventBody::CheckpointCompleted(props) => { let checkpoint = checkpoint_from_props(props, ts); - if let Some(stage_id) = stored.stage_id.clone() { + 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); + 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` @@ -353,19 +353,14 @@ impl RunProjectionReducer for RunProjection { // stays immutable. `begin_attempt` on an existing entry // remains the compatibility path for automatic retries and // legacy histories that repeat one `StageId`. - let stage = self.stage_entry( - stage_id.node_id(), - stage_id.visit(), - first_event_seq(event.seq), - ); + 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 props.graph_visit.is_some() { + if is_new { stage.graph_visit = props.graph_visit; - } - if props.resumed_from_stage_id.is_some() { stage .resumed_from_stage_id .clone_from(&props.resumed_from_stage_id); @@ -565,16 +560,18 @@ impl RunProjectionReducer for RunProjection { // 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 props.graph_visit.is_some() { + if is_new { stage.graph_visit = props.graph_visit; - } - if props.resumed_from_stage_id.is_some() { stage .resumed_from_stage_id .clone_from(&props.resumed_from_stage_id); @@ -1009,7 +1006,7 @@ fn apply_checkpoint_to_stage( } let is_new = state.stage(stage_id).is_none(); - let stage = state.stage_entry(node_id, stage_id.visit(), first_event_seq(seq)); + let stage = stage_at_stored_stage_id(state, stage_id, seq); if is_new { stage.graph_visit = props.graph_visit; stage @@ -4468,13 +4465,15 @@ mod tests { stage_id.clone(), )) .unwrap(); - // A legacy-shaped retry event for the same StageId omits the identity - // fields; the projection keeps the first attempt's metadata. + // 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(), diff --git a/lib/components/fabro-workflow/src/context.rs b/lib/components/fabro-workflow/src/context.rs index 0df5db554..acbd7cf0a 100644 --- a/lib/components/fabro-workflow/src/context.rs +++ b/lib/components/fabro-workflow/src/context.rs @@ -177,7 +177,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/events.rs b/lib/components/fabro-workflow/src/event/events.rs index 84822b4fb..adb8810d3 100644 --- a/lib/components/fabro-workflow/src/event/events.rs +++ b/lib/components/fabro-workflow/src/event/events.rs @@ -251,13 +251,13 @@ pub enum Event { attempt: usize, max_attempts: usize, /// Graph visit that produced this stage execution. Diverges from the - /// envelope `StageId` ordinal when a cancelled or crashed invocation - /// is reexecuted after resume. + /// envelope `StageId` ordinal when post-checkpoint work is replayed + /// after resume. #[serde(default, skip_serializing_if = "Option::is_none")] graph_visit: Option, - /// Prior execution this one resumes from, for the first execution - /// reserved after a resume when the node had an observable - /// post-checkpoint execution. + /// 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, }, diff --git a/lib/components/fabro-workflow/src/handler/manager_loop.rs b/lib/components/fabro-workflow/src/handler/manager_loop.rs index bc3637c52..22d140783 100644 --- a/lib/components/fabro-workflow/src/handler/manager_loop.rs +++ b/lib/components/fabro-workflow/src/handler/manager_loop.rs @@ -16,13 +16,12 @@ use crate::artifact_upload::ArtifactSink; use crate::condition::evaluate_condition; use crate::context::{Context, WorkflowContext, keys}; use crate::error::Error; -use crate::event::StageScope; use crate::operations::{ValidateInput, WorkflowInput, validate}; use crate::outcome::{Outcome, OutcomeExt, StageOutcome}; use crate::pipeline::types::Initialized; use crate::run_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. @@ -198,9 +197,9 @@ impl Handler for SubWorkflowHandler { }; // Build child RunOptions. The stage directory follows the execution - // ordinal so a reexecuted manager loop keeps the cancelled - // invocation's child logs intact. - let visit = u64::from(StageScope::for_handler(context, &node.id).visit); + // 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 47f929ad3..d7b1d5009 100644 --- a/lib/components/fabro-workflow/src/handler/parallel.rs +++ b/lib/components/fabro-workflow/src/handler/parallel.rs @@ -17,10 +17,10 @@ use crate::git::sanitize_ref_component; use crate::hook_context::set_hook_node; use crate::millis_u64; use crate::outcome::{FailureCategory, FailureDetail, Outcome, OutcomeExt, StageOutcome}; +use crate::run_dir::visit_from_context; use crate::sandbox_git::{ GIT_REMOTE, checked_git_checkpoint, git_merge_ff_only, git_remove_worktree, }; -use crate::stage_execution::StageExecution; /// Fans out execution to multiple branches concurrently. /// Each branch gets an isolated context clone and runs independently. @@ -161,9 +161,6 @@ impl Handler for ParallelHandler { branch_context: Context, sandbox: Arc, worktree_path: Option, - /// Child stage execution reserved through the run's shared - /// tracker, so a resumed fan-out gets fresh branch identities. - execution: StageExecution, } let parallel_start = Instant::now(); @@ -210,6 +207,7 @@ impl Handler for ParallelHandler { let semaphore = Arc::new(Semaphore::new(max_parallel)); let git_state = services.git_state(); + let branch_graph_visit = u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX); // --- Git isolation: checkpoint "parallel base" before fan-out --- let base_sha: Option = if let Some(ref gs) = git_state { @@ -267,19 +265,10 @@ impl Handler for ParallelHandler { parallel_group_id.clone(), u32::try_from(branch_index).unwrap_or(u32::MAX), ); - // Reserve the child's stage execution through the shared tracker - // and seed the branch context with its explicit stage scope, so - // branch lifecycle events and nested handler events agree on the - // child's identity instead of inheriting the fork's. - let execution = services.run.stage_executions.reserve(&target_id, 1); branch_context.set( keys::CURRENT_NODE, serde_json::Value::String(target_id.clone()), ); - branch_context.set( - keys::INTERNAL_STAGE_EXECUTION_ORDINAL, - serde_json::json!(execution.ordinal), - ); branch_context.set( keys::INTERNAL_PARALLEL_GROUP_ID, serde_json::Value::String(parallel_group_id.to_string()), @@ -311,7 +300,7 @@ impl Handler for ParallelHandler { { let branch_key = &target_id; // `pass{N}` derives from the parent's execution ordinal so a - // resumed fan-out does not recreate the cancelled attempt's + // resumed fan-out does not recreate the prior dispatch's // branch names. let branch_name = format!( "fabro/run/parallel/{}/{}/pass{}/{}", @@ -363,7 +352,6 @@ impl Handler for ParallelHandler { branch_context, sandbox: branch_sandbox, worktree_path, - execution, }); } @@ -393,12 +381,6 @@ impl Handler for ParallelHandler { .map(|gs| gs.checkpoint.clone()) .unwrap_or_default(); let group_id = parallel_group_id.clone(); - let branch_scope = StageScope::for_parallel_branch( - setup.target_id.clone(), - setup.execution.ordinal, - group_id.clone(), - setup.parallel_branch_id.clone(), - ); let handle = tokio::spawn(async move { let _permit = sem @@ -406,14 +388,30 @@ impl Handler for ParallelHandler { .await .map_err(|e| Error::handler_with_source("semaphore error", e))?; + // Only reserve once the branch is ready to become observable. + // This avoids consuming an execution identity for worktree + // setup failures or branches still waiting on the semaphore. + let execution = parent_run + .stage_executions + .reserve(&setup.target_id, branch_graph_visit); + setup.branch_context.set( + keys::INTERNAL_STAGE_EXECUTION_ORDINAL, + serde_json::json!(execution.stage_id.visit()), + ); + let branch_scope = StageScope::for_parallel_branch( + setup.target_id.clone(), + execution.stage_id.visit(), + group_id.clone(), + setup.parallel_branch_id.clone(), + ); parent_run.emitter.emit_scoped( &Event::ParallelBranchStarted { parallel_group_id: group_id.clone(), parallel_branch_id: setup.parallel_branch_id.clone(), branch: setup.target_id.clone(), index: setup.branch_index, - graph_visit: Some(setup.execution.graph_visit), - resumed_from_stage_id: setup.execution.resumed_from.clone(), + graph_visit: Some(execution.graph_visit), + resumed_from_stage_id: execution.resumed_from.clone(), }, &branch_scope, ); @@ -1041,6 +1039,7 @@ mod tests { AttrValue::String("component".to_string()), ); let context = test_context(); + context.set(keys::INTERNAL_NODE_VISIT_COUNT, serde_json::json!(2)); let mut graph = Graph::new("test"); graph.nodes.insert("par".to_string(), node.clone()); graph @@ -1067,13 +1066,22 @@ mod tests { assert!(results.is_some()); let state = run_store.state().await.unwrap(); - let node_state = state.stage(&StageId::new("par", 1)).unwrap(); + let node_state = state.stage(&StageId::new("par", 2)).unwrap(); let parsed = node_state.parallel_results.as_ref().unwrap(); assert!( parsed.is_array(), "parallel_results.json should be a JSON array" ); assert_eq!(parsed.as_array().unwrap().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 2bb1c4977..ee3ec4566 100644 --- a/lib/components/fabro-workflow/src/lib.rs +++ b/lib/components/fabro-workflow/src/lib.rs @@ -329,7 +329,7 @@ pub mod runtime_store; pub mod sandbox_git; pub(crate) mod sandbox_git_runtime; pub mod services; -pub mod stage_execution; +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 f4d0e4d0e..3531f7bbc 100644 --- a/lib/components/fabro-workflow/src/lifecycle/artifact.rs +++ b/lib/components/fabro-workflow/src/lifecycle/artifact.rs @@ -20,7 +20,7 @@ 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; @@ -127,10 +127,8 @@ impl RunLifecycle for ArtifactLifecycle { let node_id = ctx.node.id(); // Artifact identity follows the stage execution ordinal so a resumed // reexecution stores its captures under the new `StageId`. - let visit = self.stage_executions.active(node_id).map_or_else( - || stage_visit(state, node_id), - |execution| execution.ordinal, - ); + let 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 { @@ -154,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, @@ -172,7 +170,6 @@ impl RunLifecycle for ArtifactLifecycle { return Ok(()); } self.record_captured_assets(&new_assets); - let scope = stage_scope_for(&self.stage_executions, state, node_id); for asset in &new_assets { self.emitter.emit_scoped( &Event::ArtifactCaptured { diff --git a/lib/components/fabro-workflow/src/lifecycle/event.rs b/lib/components/fabro-workflow/src/lifecycle/event.rs index 0770323fc..c4af83250 100644 --- a/lib/components/fabro-workflow/src/lifecycle/event.rs +++ b/lib/components/fabro-workflow/src/lifecycle/event.rs @@ -18,7 +18,7 @@ use crate::context::{Context, WorkflowContext}; use crate::event::{Emitter, Event, StageScope}; use crate::graph::{WorkflowGraph, WorkflowNode}; use crate::outcome::{BilledModelUsage, FailureCategory, FailureDetail, Outcome, StageOutcome}; -use crate::stage_execution::StageExecutionTracker; +use crate::stage_execution::{StageExecution, StageExecutionTracker}; use crate::{artifact, context}; type WfRunState = ExecutionState>; @@ -95,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()) } @@ -110,6 +113,28 @@ pub(super) fn stage_visit(state: &WfRunState, node_id: &str) -> u32 { u32::try_from(visits).unwrap_or(u32::MAX) } +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, + 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. @@ -118,16 +143,8 @@ pub(crate) fn stage_scope_for( state: &WfRunState, node_id: &str, ) -> StageScope { - let visit = stage_executions.active(node_id).map_or_else( - || stage_visit(state, node_id), - |execution| execution.ordinal, - ); - StageScope { - node_id: node_id.to_string(), - visit, - parallel_group_id: state.context.parallel_group_id(), - parallel_branch_id: state.context.parallel_branch_id(), - } + let execution = stage_executions.active(node_id); + stage_scope_from_execution(execution.as_deref(), state, node_id) } #[async_trait] @@ -179,7 +196,7 @@ impl RunLifecycle for EventLifecycle { let execution = self .stage_executions .reserve(&gv.id, stage_visit(state, &gv.id)); - let scope = stage_scope_for(&self.stage_executions, state, &gv.id); + let 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( @@ -191,7 +208,7 @@ impl RunLifecycle for EventLifecycle { attempt: 1, max_attempts: 1, graph_visit: Some(execution.graph_visit), - resumed_from_stage_id: execution.resumed_from, + resumed_from_stage_id: execution.resumed_from.clone(), }, &scope, ); @@ -232,7 +249,7 @@ impl RunLifecycle for EventLifecycle { ) -> CoreResult>> { let gv = ctx.node.inner(); let execution = self.stage_executions.active(&gv.id); - let scope = stage_scope_for(&self.stage_executions, state, &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); @@ -245,7 +262,9 @@ impl RunLifecycle for EventLifecycle { attempt: ctx.attempt as usize, max_attempts: ctx.max_attempts as usize, graph_visit: Some(graph_visit), - resumed_from_stage_id: execution.and_then(|e| e.resumed_from), + resumed_from_stage_id: execution + .as_ref() + .and_then(|execution| execution.resumed_from.clone()), }, &scope, ); @@ -428,7 +447,7 @@ impl RunLifecycle for EventLifecycle { artifact::normalize_durable_outcomes(&mut node_outcomes); let execution = self.stage_executions.active(node.id()); - let scope = stage_scope_for(&self.stage_executions, state, 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); @@ -457,7 +476,9 @@ impl RunLifecycle for EventLifecycle { diff, diff_summary, graph_visit: Some(graph_visit), - resumed_from_stage_id: execution.and_then(|e| e.resumed_from), + resumed_from_stage_id: execution + .as_ref() + .and_then(|execution| execution.resumed_from.clone()), }, &scope, ); @@ -491,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/mod.rs b/lib/components/fabro-workflow/src/lifecycle/mod.rs index e3facc0f8..8006c7abc 100644 --- a/lib/components/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/components/fabro-workflow/src/lifecycle/mod.rs @@ -313,7 +313,7 @@ impl RunLifecycle for WorkflowLifecycle { .ensure(node_id, event::stage_visit(state, node_id)); state.context.set( context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, - serde_json::json!(execution.ordinal), + serde_json::json!(execution.stage_id.visit()), ); // Event emission self.event.before_attempt(ctx, state).await?; @@ -446,7 +446,7 @@ impl RunLifecycle for WorkflowLifecycle { .ensure(node.id(), event::stage_visit(state, node.id())); state.context.set( context::keys::INTERNAL_STAGE_EXECUTION_ORDINAL, - serde_json::json!(execution.ordinal), + serde_json::json!(execution.stage_id.visit()), ); self.git .on_checkpoint(node, result, next_node_id, state) diff --git a/lib/components/fabro-workflow/src/operations/resume.rs b/lib/components/fabro-workflow/src/operations/resume.rs index fd2b4d62b..433947716 100644 --- a/lib/components/fabro-workflow/src/operations/resume.rs +++ b/lib/components/fabro-workflow/src/operations/resume.rs @@ -4,8 +4,8 @@ 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; -use crate::stage_execution::StageExecutionSeed; /// Resume a workflow run from its checkpoint. Errors if no checkpoint is found. pub async fn resume(run_dir: &Path, services: StartServices) -> Result { @@ -33,15 +33,8 @@ pub async fn resume(run_dir: &Path, services: StartServices) -> Result Result Result, - stage_executions: StageExecutionSeed, + resume: Option, services: StartServices, ) -> Result { let cancel_token = services.cancel_token.clone(); @@ -269,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, stage_executions)).await; + let started = Box::pin(session.run(persisted, resume)).await; match started { Ok(started) => { @@ -804,8 +797,7 @@ impl RunSession { async fn run( self, persisted: Persisted, - checkpoint: Option, - stage_executions: StageExecutionSeed, + resume: Option, ) -> Result { let on_node = self.on_node.clone(); @@ -886,9 +878,8 @@ impl RunSession { registry_override: self.registry_override, artifact_sink: self.artifact_sink, run_control: self.run_control, - checkpoint, + resume, seed_context: self.seed_context, - stage_executions, fabro_run_tools: self.fabro_run_tools, }; let mut initialized = Box::pin(pipeline::initialize(persisted, init_options)).await?; diff --git a/lib/components/fabro-workflow/src/pipeline/execute/tests.rs b/lib/components/fabro-workflow/src/pipeline/execute/tests.rs index 22b6233cb..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; @@ -256,7 +256,6 @@ async fn execute_test_run_with_options( let initialized = initialize( persisted_workflow(graph, String::new(), &run_options.run_dir, run_id_value), InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -292,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, }, @@ -317,7 +316,6 @@ async fn execute_runs_start_to_exit_and_returns_final_context() { let initialized = initialize( persisted_workflow(graph, source, &run_dir, test_run_id("run-test")), InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: test_emitter_arc("run-test"), @@ -355,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, }, @@ -450,15 +448,15 @@ async fn resumed_in_flight_node_starts_a_new_stage_execution() { restart_failure_signatures: HashMap::new(), node_visits: HashMap::from([("start".to_string(), 1usize)]), }; - let seed = crate::stage_execution::StageExecutionSeed { - high_water: HashMap::from([("work".to_string(), 1)]), - resumed_from: HashMap::from([("work".to_string(), fabro_types::StageId::new("work", 1))]), - }; + let 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 { - stage_executions: seed, run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -494,7 +492,7 @@ async fn resumed_in_flight_node_starts_a_new_stage_execution() { run_control: None, registry_override: Some(Arc::new(make_registry())), artifact_sink: None, - checkpoint: Some(checkpoint), + resume: Some(resume), seed_context: None, fabro_run_tools: None, }, @@ -573,7 +571,6 @@ async fn run_with_lifecycle( let initialized = initialize( persisted_workflow(graph.clone(), String::new(), &run_dir, run_id), InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -606,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/initialize.rs b/lib/components/fabro-workflow/src/pipeline/initialize.rs index c13c31b3c..9674991a7 100644 --- a/lib/components/fabro-workflow/src/pipeline/initialize.rs +++ b/lib/components/fabro-workflow/src/pipeline/initialize.rs @@ -33,7 +33,7 @@ use crate::sandbox_git_runtime::SandboxGitRuntime; use crate::services::{ EngineServices, FabroRunToolServices, RunLocations, RunServices, WorkflowToolEnvProvider, }; -use crate::stage_execution::StageExecutionTracker; +use crate::stage_execution::{StageExecutionSeed, StageExecutionTracker}; use crate::steering_hub::SteeringHub; type BuiltSandboxEnv = (HashMap, Option>); @@ -287,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(); @@ -307,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 @@ -620,7 +627,7 @@ pub async fn initialize( sandbox_git, metadata_runtime, metadata_writer, - StageExecutionTracker::seeded(options.stage_executions), + StageExecutionTracker::seeded(stage_executions), ); let engine = Arc::new(EngineServices { run: Arc::clone(&run_services), @@ -643,7 +650,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, @@ -827,7 +834,6 @@ mod tests { }); let result = initialize(persisted, InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); @@ -867,7 +873,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -909,7 +915,6 @@ mod tests { let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let initialized = initialize(persisted, InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); @@ -949,7 +954,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1135,7 +1140,6 @@ mod tests { let store = memory_store(); let run_store = store.create_run(&test_run_id()).await.unwrap(); let initialized = initialize(test_persisted(graph, source, &run_dir), InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -1171,7 +1175,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1231,7 +1235,6 @@ mod tests { store_logger.register(&emitter); let initialized = initialize(persisted, InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: run_store.into(), dry_run: false, emitter: emitter.clone(), @@ -1267,7 +1270,7 @@ mod tests { run_control: None, registry_override: None, artifact_sink: None, - checkpoint: None, + resume: None, seed_context: None, fabro_run_tools: None, }) @@ -1370,7 +1373,6 @@ mod tests { let emitter = Arc::new(crate::event::Emitter::new(test_run_id())); let result = initialize(persisted, InitOptions { - stage_executions: crate::stage_execution::StageExecutionSeed::default(), run_store: { let store = memory_store(); let inner = store.create_run(&test_run_id()).await.unwrap(); @@ -1410,7 +1412,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 97b852b30..b1f3b4469 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; @@ -249,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, @@ -269,12 +304,8 @@ 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, - /// Allocator seed for stage execution ordinals. Empty for a fresh run; - /// resume passes projection-derived high-water marks and provenance so a - /// reexecuted in-flight node gets a new `StageId` ordinal. - pub stage_executions: StageExecutionSeed, pub fabro_run_tools: Option, } diff --git a/lib/components/fabro-workflow/src/stage_execution.rs b/lib/components/fabro-workflow/src/stage_execution.rs index 62604695c..153382040 100644 --- a/lib/components/fabro-workflow/src/stage_execution.rs +++ b/lib/components/fabro-workflow/src/stage_execution.rs @@ -10,7 +10,7 @@ //! The tracker is deliberately not checkpointed: its durable source of truth //! is the append-only stage event history. On resume it is seeded from the //! run projection's per-node maxima, so a reexecuted in-flight node allocates -//! the next unused ordinal instead of mutating the cancelled execution. +//! the next unused ordinal instead of mutating the prior execution. use std::collections::HashMap; use std::sync::{Arc, Mutex}; @@ -19,27 +19,32 @@ use fabro_types::{RunProjection, StageId}; /// One reserved stage execution: the identity of a single resumable handler /// invocation of a node. -#[derive(Clone, Debug, PartialEq, Eq)] +#[derive(Debug, PartialEq, Eq)] pub(crate) struct StageExecution { - /// 1-based execution ordinal; becomes the `@N` in the external `StageId`. - pub ordinal: u32, + /// Canonical external identity for this execution. + pub stage_id: StageId, /// Graph visit that produced this execution. pub graph_visit: u32, - /// Prior execution this one resumes from, when the node had an observable - /// post-checkpoint execution before the run was interrupted. + /// 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(Clone, Debug, Default)] -pub struct StageExecutionSeed { - /// Highest execution ordinal already observable per node. - pub high_water: HashMap, - /// Latest post-checkpoint execution per node; the next reservation for - /// that node links back to it via `resumed_from_stage_id`. - pub resumed_from: HashMap, +#[derive(Debug, Default)] +pub(crate) struct StageExecutionSeed { + nodes: HashMap, } impl StageExecutionSeed { @@ -49,60 +54,61 @@ impl StageExecutionSeed { /// checkpoint. Only stages that first became observable *after* that /// checkpoint are eligible provenance targets: an older execution with the /// same node ID completed before the checkpoint and is not what the - /// resumed invocation continues from. + /// resumed replay supersedes. #[must_use] - pub fn from_projection(projection: &RunProjection, checkpoint_seq: u32) -> Self { - let mut high_water: HashMap = HashMap::new(); - let mut resumed_from: HashMap = HashMap::new(); - // `iter_stages` yields chronological `first_event_seq` order, so a - // later insert per node retains the latest post-checkpoint execution. - for (stage_id, stage) in projection.iter_stages() { - let node_id = stage_id.node_id(); - let entry = high_water.entry(node_id.to_string()).or_default(); - *entry = (*entry).max(stage_id.visit()); + 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 { - resumed_from.insert(node_id.to_string(), stage_id.clone()); + 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 { - high_water, - resumed_from, + nodes: HashMap::from([(node_id, NodeExecutionState { + high_water: high_water.visit(), + resumed_from, + active: None, + })]), } } } -#[derive(Debug, Default)] -struct TrackerState { - /// Highest ordinal observed or reserved per node. - high_water: HashMap, - /// Pending provenance links, consumed by the first reservation per node. - resumed_from: HashMap, - /// Active execution scope per node. Cleared at the node boundary and - /// replaced by the next reservation. - active: HashMap, -} - /// Cloneable, run-scoped allocator for stage execution ordinals. Clones share /// one synchronized state so the core lifecycle and direct-dispatch handlers /// (parallel branches) allocate from the same sequence. #[derive(Clone, Debug, Default)] pub(crate) struct StageExecutionTracker { - state: Arc>, + state: Arc>>, } impl StageExecutionTracker { #[must_use] pub(crate) fn seeded(seed: StageExecutionSeed) -> Self { Self { - state: Arc::new(Mutex::new(TrackerState { - high_water: seed.high_water, - resumed_from: seed.resumed_from, - active: HashMap::new(), - })), + state: Arc::new(Mutex::new(seed.nodes)), } } - fn lock(&self) -> std::sync::MutexGuard<'_, TrackerState> { + 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") @@ -113,40 +119,54 @@ impl StageExecutionTracker { /// made here so that a StageStart hook block or process exit before any /// stage-scoped event leaves no phantom execution. pub(crate) fn begin_node(&self, node_id: &str) { - self.lock().active.remove(node_id); + 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().active.get(node_id).cloned() + 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) -> StageExecution { + pub(crate) fn reserve(&self, node_id: &str, graph_visit: u32) -> Arc { let mut state = self.lock(); - let entry = state.high_water.entry(node_id.to_string()).or_default(); - *entry = entry.saturating_add(1); - let ordinal = *entry; - let resumed_from = state.resumed_from.remove(node_id); - let execution = StageExecution { - ordinal, - graph_visit, - resumed_from, - }; - state.active.insert(node_id.to_string(), execution.clone()); - execution + 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) -> StageExecution { - if let Some(execution) = self.active(node_id) { + 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(node_id, graph_visit) + Self::reserve_locked(&mut state, node_id, graph_visit) } } @@ -190,10 +210,10 @@ mod tests { fn reserve_starts_at_one_and_allocates_monotonically_per_node() { let tracker = StageExecutionTracker::default(); - assert_eq!(tracker.reserve("work", 1).ordinal, 1); + assert_eq!(tracker.reserve("work", 1).stage_id.visit(), 1); tracker.begin_node("work"); - assert_eq!(tracker.reserve("work", 2).ordinal, 2); - assert_eq!(tracker.reserve("other", 1).ordinal, 1); + assert_eq!(tracker.reserve("work", 2).stage_id.visit(), 2); + assert_eq!(tracker.reserve("other", 1).stage_id.visit(), 1); } #[test] @@ -202,9 +222,9 @@ mod tests { let seed = StageExecutionSeed::from_projection(&projection, 0); let tracker = StageExecutionTracker::seeded(seed); - assert_eq!(tracker.reserve("work", 1).ordinal, 3); - assert_eq!(tracker.reserve("plan", 1).ordinal, 2); - assert_eq!(tracker.reserve("new", 1).ordinal, 1); + 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] @@ -214,7 +234,7 @@ mod tests { let tracker = StageExecutionTracker::seeded(seed); let execution = tracker.reserve("work", 2); - assert_eq!(execution.ordinal, 3); + assert_eq!(execution.stage_id.visit(), 3); assert_eq!(execution.graph_visit, 2); } @@ -225,10 +245,10 @@ mod tests { let first = tracker.ensure("work", 1); let second = tracker.ensure("work", 1); assert_eq!(first, second); - assert_eq!(second.ordinal, 1); + assert_eq!(second.stage_id.visit(), 1); tracker.begin_node("work"); - assert_eq!(tracker.ensure("work", 2).ordinal, 2); + assert_eq!(tracker.ensure("work", 2).stage_id.visit(), 2); } #[test] @@ -240,7 +260,12 @@ mod tests { tracker.begin_node("work"); assert_eq!(tracker.active("work"), None); - assert_eq!(tracker.active("verify").map(|e| e.ordinal), Some(1)); + assert_eq!( + tracker + .active("verify") + .map(|execution| execution.stage_id.visit()), + Some(1) + ); } #[test] @@ -249,10 +274,17 @@ mod tests { let seed = StageExecutionSeed::from_projection(&projection, 5); assert_eq!( - seed.resumed_from.get("work"), + seed.nodes + .get("work") + .and_then(|node| node.resumed_from.as_ref()), Some(&StageId::new("work", 2)) ); - assert_eq!(seed.resumed_from.get("plan"), None); + assert_eq!( + seed.nodes + .get("plan") + .and_then(|node| node.resumed_from.as_ref()), + None + ); } #[test] @@ -262,12 +294,12 @@ mod tests { let tracker = StageExecutionTracker::seeded(seed); let first = tracker.reserve("work", 1); - assert_eq!(first.ordinal, 2); + 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.ordinal, 3); + assert_eq!(second.stage_id.visit(), 3); assert_eq!(second.resumed_from, None); } @@ -277,7 +309,7 @@ mod tests { let handles: Vec<_> = (0..8) .map(|_| { let tracker = tracker.clone(); - tokio::spawn(async move { tracker.reserve("branch", 1).ordinal }) + tokio::spawn(async move { tracker.reserve("branch", 1).stage_id.visit() }) }) .collect(); @@ -288,4 +320,24 @@ mod tests { 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 f7fa28ddf..a625e24e8 100644 --- a/lib/components/fabro-workflow/src/stage_scope.rs +++ b/lib/components/fabro-workflow/src/stage_scope.rs @@ -4,15 +4,17 @@ use crate::context::{Context as WfContext, WorkflowContext, keys}; use crate::run_dir::visit_from_context; /// Read the stage execution ordinal seeded by the workflow lifecycle (or a -/// parallel branch dispatch). `None` when the current node has not reserved an -/// execution yet — direct-handler call sites (tests, etc.) that skip the full +/// 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. -fn execution_ordinal_from_context(context: &WfContext) -> Option { +pub(crate) fn execution_ordinal_from_context(context: &WfContext) -> u32 { context .get(keys::INTERNAL_STAGE_EXECUTION_ORDINAL) .and_then(|value| value.as_u64()) - .map(|ordinal| u32::try_from(ordinal).unwrap_or(u32::MAX)) + .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 @@ -21,7 +23,7 @@ fn execution_ordinal_from_context(context: &WfContext) -> Option { /// /// `visit` is the 1-based stage execution ordinal — the numeric component of /// the external `StageId`. It matches the graph visit for a first execution -/// and diverges when a cancelled or crashed invocation is reexecuted after +/// and diverges when post-checkpoint work is replayed after /// resume. #[derive(Clone, Debug)] pub struct StageScope { @@ -35,8 +37,7 @@ impl StageScope { /// Build a scope from the given node id, sourcing the execution ordinal /// and parallel ids from the current context. pub fn from_context(context: &WfContext, node_id: impl Into) -> Self { - let visit = execution_ordinal_from_context(context) - .unwrap_or_else(|| u32::try_from(visit_from_context(context)).unwrap_or(u32::MAX)); + let visit = execution_ordinal_from_context(context); Self { node_id: node_id.into(), visit, @@ -62,7 +63,7 @@ impl StageScope { /// `target_visit` is the branch target's stage execution ordinal for this /// particular dispatch, reserved through the run's shared /// `StageExecutionTracker` so a resumed fan-out gets a fresh child - /// identity instead of overwriting the cancelled attempt's. + /// identity instead of overwriting the prior dispatch's. #[must_use] pub fn for_parallel_branch( target_node_id: impl Into, diff --git a/lib/foundation/fabro-api/build.rs b/lib/foundation/fabro-api/build.rs index cdec7f07a..c55fb74e6 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 6d68d57b4..ecee866cc 100644 --- a/lib/foundation/fabro-api/src/lib.rs +++ b/lib/foundation/fabro-api/src/lib.rs @@ -66,7 +66,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 0f72849b3..5c0b29016 100644 --- a/lib/foundation/fabro-types/src/run_event/misc.rs +++ b/lib/foundation/fabro-types/src/run_event/misc.rs @@ -29,7 +29,7 @@ pub struct ParallelBranchStartedProps { /// keep visit metadata even though their ordinals advanced. #[serde(default, skip_serializing_if = "Option::is_none")] pub graph_visit: Option, - /// Prior branch execution this one resumes from. + /// Prior branch execution superseded by this resumed replay. #[serde(default, skip_serializing_if = "Option::is_none")] pub resumed_from_stage_id: Option, } diff --git a/lib/foundation/fabro-types/src/run_event/stage.rs b/lib/foundation/fabro-types/src/run_event/stage.rs index 7867380de..18a5319b0 100644 --- a/lib/foundation/fabro-types/src/run_event/stage.rs +++ b/lib/foundation/fabro-types/src/run_event/stage.rs @@ -17,11 +17,11 @@ pub struct StageStartedProps { pub max_attempts: usize, /// Graph visit that produced this stage execution. The envelope /// `stage_id` ordinal counts executions, which diverges from the graph - /// visit when a cancelled or crashed invocation is reexecuted after + /// 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 this one resumes from. + /// Prior execution superseded by this resumed replay. #[serde(default, skip_serializing_if = "Option::is_none")] pub resumed_from_stage_id: Option, } diff --git a/lib/foundation/fabro-types/src/run_projection.rs b/lib/foundation/fabro-types/src/run_projection.rs index d8f026442..c13aa4d20 100644 --- a/lib/foundation/fabro-types/src/run_projection.rs +++ b/lib/foundation/fabro-types/src/run_projection.rs @@ -341,12 +341,12 @@ pub struct StageProjection { #[serde(default, skip_serializing_if = "Option::is_none")] pub handler: Option, /// Graph visit that produced this stage execution. The `StageId` ordinal - /// counts executions, which diverges from the graph visit when a - /// cancelled or crashed invocation is reexecuted after resume. Absent on + /// 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 this one resumes from. + /// 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. @@ -539,10 +539,10 @@ impl StageProjection { /// (identity / sort key) and the execution identity metadata /// (`graph_visit`, `resumed_from_stage_id`). /// - /// One stage projection represents one execution; a reexecution after - /// cancel or crash recovery gets a new `StageId` and never flows through - /// here. Replays of legacy histories with duplicate `stage.started` - /// events for one `StageId` retain this last-attempt behavior. + /// 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(); @@ -594,11 +594,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 4dd742f11..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,15 +46,15 @@ export interface RunStage { */ 'node_id': string; /** - * 1-based stage execution ordinal, the numeric component of `id`. It increments each time the node produces a new observable execution: graph re-entry (loops) and reexecution after cancel or crash recovery. Automatic in-place retries do not increment it. + * 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 cancelled or crashed execution was reexecuted after resume. Absent for stages recorded before execution identity was tracked. + * 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; /** - * StageId of the prior cancelled or interrupted execution this stage resumes from, when the run was resumed after that execution became observable. + * Canonical stage execution identifier in `node_id@visit` form. */ 'resumed_from_stage_id'?: string | null; 'provider_used'?: StageModelUsage | null;