From 98e93d22fc994fc3eaf72306dadb11e87b888b2e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 3 Apr 2026 23:49:24 -0700 Subject: [PATCH] refactor: remove workflow lifecycle disk checkpoints --- .../fabro-cli/tests/it/cmd/system_df.rs | 14 +- lib/crates/fabro-cli/tests/it/cmd/wait.rs | 1 - lib/crates/fabro-cli/tests/it/workflow/mod.rs | 52 ++- .../fabro-workflow/src/lifecycle/disk.rs | 103 ------ .../fabro-workflow/src/lifecycle/git.rs | 28 +- .../fabro-workflow/src/lifecycle/mod.rs | 15 - .../fabro-workflow/src/operations/start.rs | 2 +- lib/crates/fabro-workflow/src/run_lookup.rs | 12 +- lib/crates/fabro-workflow/src/test_support.rs | 69 +++- .../tests/it/daytona_integration.rs | 81 ++++- .../fabro-workflow/tests/it/integration.rs | 336 ++++++++++++------ 11 files changed, 451 insertions(+), 262 deletions(-) delete mode 100644 lib/crates/fabro-workflow/src/lifecycle/disk.rs diff --git a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs index 8e40b398f..2e4d4fa83 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/system_df.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/system_df.rs @@ -48,9 +48,9 @@ fn system_df_summarizes_runs_and_logs() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (100%) - Logs 1 - [SIZE] [SIZE] (100%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 1 - [SIZE] [SIZE] (100%) Data directory: [STORAGE_DIR] ----- stderr ----- @@ -79,13 +79,13 @@ fn system_df_verbose_lists_runs_with_reclaimable_marker() { success: true exit_code: 0 ----- stdout ----- - TYPE COUNT ACTIVE SIZE RECLAIMABLE - Runs 1 0 [SIZE] [SIZE] (100%) - Logs 0 - [SIZE] [SIZE] (0%) + TYPE COUNT ACTIVE SIZE RECLAIMABLE + Runs 1 0 [SIZE] [SIZE] (0%) + Logs 0 - [SIZE] [SIZE] (0%) Data directory: [STORAGE_DIR] - RUN ID WORKFLOW STATUS AGE SIZE + RUN ID WORKFLOW STATUS AGE SIZE [RUN_PREFIX] Simple succeeded [AGE] [SIZE] * * = reclaimable diff --git a/lib/crates/fabro-cli/tests/it/cmd/wait.rs b/lib/crates/fabro-cli/tests/it/cmd/wait.rs index b773827fa..8b55f369c 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/wait.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/wait.rs @@ -58,7 +58,6 @@ fn wait_completed_run_prints_success_summary() { fn wait_completed_run_reads_store_without_status_or_conclusion_files() { let context = test_context!(); let run = setup_completed_dry_run(&context); - let _ = std::fs::remove_file(run.run_dir.join("status.json")); let _ = std::fs::remove_file(run.run_dir.join("conclusion.json")); let mut filters = context.filters(); filters.push(( diff --git a/lib/crates/fabro-cli/tests/it/workflow/mod.rs b/lib/crates/fabro-cli/tests/it/workflow/mod.rs index 809a27bd7..ffd4f71d4 100644 --- a/lib/crates/fabro-cli/tests/it/workflow/mod.rs +++ b/lib/crates/fabro-cli/tests/it/workflow/mod.rs @@ -9,9 +9,13 @@ mod human_gate; mod real_cli; use std::path::{Path, PathBuf}; +use std::sync::Arc; use std::time::Duration; +use fabro_store::{SlateRunStore, SlateStore}; use fabro_test::TestContext; +use fabro_types::RunId; +use object_store::local::LocalFileSystem; use serde_json::Value; pub(super) fn fixture(name: &str) -> PathBuf { @@ -32,13 +36,10 @@ pub(super) fn read_conclusion(run_dir: &Path) -> Value { } pub(super) fn completed_nodes(run_dir: &Path) -> Vec { - let cp = read_json(&run_dir.join("checkpoint.json")); - cp["completed_nodes"] - .as_array() - .expect("completed_nodes should be an array") - .iter() - .map(|v| v.as_str().unwrap().to_string()) - .collect() + let cp = run_state(run_dir) + .checkpoint + .expect("run store checkpoint should exist"); + cp.completed_nodes } pub(super) fn has_event(run_dir: &Path, event_name: &str) -> bool { @@ -87,6 +88,43 @@ pub(super) fn find_run_dir(storage_dir: &Path) -> PathBuf { entries[0].path() } +fn infer_run_id(run_dir: &Path) -> RunId { + if let Ok(id) = std::fs::read_to_string(run_dir.join("id.txt")) { + return id.trim().parse().expect("run id should parse"); + } + run_dir + .file_name() + .map(|name| name.to_string_lossy().to_string()) + .and_then(|name| name.rsplit('-').next().map(ToOwned::to_owned)) + .filter(|value| !value.is_empty()) + .expect("run directory name should contain run id suffix") + .parse() + .expect("run id should parse") +} + +fn block_on(future: impl std::future::Future) -> T { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap() + .block_on(future) +} + +fn run_store(run_dir: &Path) -> SlateRunStore { + let runs_dir = run_dir.parent().expect("run dir should have parent"); + let storage_dir = runs_dir.parent().expect("runs dir should have parent"); + let object_store = Arc::new( + LocalFileSystem::new_with_prefix(storage_dir.join("store")) + .expect("test store path should be accessible"), + ); + let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1))); + block_on(store.open_run_reader(&infer_run_id(run_dir))).expect("run store should exist") +} + +fn run_state(run_dir: &Path) -> fabro_store::RunProjection { + block_on(run_store(run_dir).state()).expect("run store state should exist") +} + macro_rules! sandbox_tests { ($name:ident) => { sandbox_tests!($name, keys = []); diff --git a/lib/crates/fabro-workflow/src/lifecycle/disk.rs b/lib/crates/fabro-workflow/src/lifecycle/disk.rs deleted file mode 100644 index d142d7ca3..000000000 --- a/lib/crates/fabro-workflow/src/lifecycle/disk.rs +++ /dev/null @@ -1,103 +0,0 @@ -use std::path::PathBuf; -use std::sync::{Arc, Mutex}; - -use async_trait::async_trait; -use tokio::fs; - -use fabro_core::error::CoreError; -use fabro_core::error::Result as CoreResult; -use fabro_core::graph::NodeSpec; -use fabro_core::lifecycle::RunLifecycle; -use fabro_core::outcome::NodeResult; -use fabro_core::state::ExecutionState; - -use super::circuit_breaker::CircuitBreakerLifecycle; -use super::git::GitCheckpointResult; -use crate::graph::WorkflowGraph; -use crate::graph::WorkflowNode; -use crate::outcome::StageUsage; - -type WfRunState = ExecutionState>; -type WfNodeResult = NodeResult>; - -/// Sub-lifecycle responsible for emitting store-backed run lifecycle events. -pub(crate) struct DiskLifecycle { - pub run_dir: PathBuf, - pub checkpoint_git_result: Arc>>, - pub circuit_breaker: Arc, - pub checkpoint_enabled: bool, -} - -pub(super) fn build_checkpoint( - node: &WorkflowNode, - result: &WfNodeResult, - next_node_id: Option<&str>, - state: &WfRunState, - loop_failure_signatures: std::collections::HashMap, - restart_failure_signatures: std::collections::HashMap, - git_commit_sha: Option, -) -> fabro_types::Checkpoint { - let mut node_outcomes = state.node_outcomes.clone(); - node_outcomes.insert(node.id().to_string(), result.outcome.clone()); - - fabro_types::Checkpoint { - timestamp: chrono::Utc::now(), - current_node: node.id().to_string(), - completed_nodes: state.completed_nodes.clone(), - node_outcomes, - node_retries: state.node_retries.clone(), - context_values: state.context.snapshot(), - next_node_id: next_node_id.map(String::from), - git_commit_sha, - node_visits: state.node_visits.clone(), - loop_failure_signatures, - restart_failure_signatures, - } -} - -#[async_trait] -impl RunLifecycle for DiskLifecycle { - async fn after_node( - &self, - _node: &WorkflowNode, - _result: &mut WfNodeResult, - _state: &WfRunState, - ) -> CoreResult<()> { - Ok(()) - } - - async fn on_checkpoint( - &self, - node: &WorkflowNode, - result: &WfNodeResult, - next_node_id: Option<&str>, - state: &WfRunState, - ) -> CoreResult<()> { - if !self.checkpoint_enabled { - return Ok(()); - } - - let git_commit_sha = self - .checkpoint_git_result - .lock() - .unwrap() - .as_ref() - .and_then(|result| result.commit_sha.clone()); - let (loop_sigs, restart_sigs) = self.circuit_breaker.snapshot(); - let checkpoint = build_checkpoint( - node, - result, - next_node_id, - state, - loop_sigs, - restart_sigs, - git_commit_sha, - ); - let checkpoint_bytes = serde_json::to_vec_pretty(&checkpoint) - .map_err(|err| CoreError::Other(format!("failed to serialize checkpoint: {err}")))?; - fs::write(self.run_dir.join("checkpoint.json"), checkpoint_bytes) - .await - .map_err(|err| CoreError::Other(format!("failed to write checkpoint.json: {err}")))?; - Ok(()) - } -} diff --git a/lib/crates/fabro-workflow/src/lifecycle/git.rs b/lib/crates/fabro-workflow/src/lifecycle/git.rs index f48719a07..9f00c8ecc 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/git.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/git.rs @@ -18,7 +18,6 @@ use crate::event::{EventEmitter, RunNoticeLevel, WorkflowRunEvent}; use crate::git::MetadataStore; use crate::graph::WorkflowGraph; use crate::graph::WorkflowNode; -use crate::lifecycle::disk::build_checkpoint; use crate::outcome::{Outcome, StageStatus, StageUsage}; use crate::run_dump::RunDump; use crate::run_options::RunOptions; @@ -27,6 +26,33 @@ use crate::sandbox_git::{git_checkpoint, git_diff, git_push_host}; type WfRunState = ExecutionState>; type WfNodeResult = NodeResult>; +fn build_checkpoint( + node: &WorkflowNode, + result: &WfNodeResult, + next_node_id: Option<&str>, + state: &WfRunState, + loop_failure_signatures: std::collections::HashMap, + restart_failure_signatures: std::collections::HashMap, + git_commit_sha: Option, +) -> fabro_types::Checkpoint { + let mut node_outcomes = state.node_outcomes.clone(); + node_outcomes.insert(node.id().to_string(), result.outcome.clone()); + + fabro_types::Checkpoint { + timestamp: chrono::Utc::now(), + current_node: node.id().to_string(), + completed_nodes: state.completed_nodes.clone(), + node_outcomes, + node_retries: state.node_retries.clone(), + context_values: state.context.snapshot(), + next_node_id: next_node_id.map(String::from), + git_commit_sha, + node_visits: state.node_visits.clone(), + loop_failure_signatures, + restart_failure_signatures, + } +} + /// Result of a git checkpoint operation, shared with EventLifecycle. #[derive(Debug, Clone)] pub(crate) struct GitCheckpointResult { diff --git a/lib/crates/fabro-workflow/src/lifecycle/mod.rs b/lib/crates/fabro-workflow/src/lifecycle/mod.rs index cda6bd358..7df1f6bce 100644 --- a/lib/crates/fabro-workflow/src/lifecycle/mod.rs +++ b/lib/crates/fabro-workflow/src/lifecycle/mod.rs @@ -1,7 +1,6 @@ pub(crate) mod artifact; pub(crate) mod auto_status; pub(crate) mod circuit_breaker; -pub(crate) mod disk; pub(crate) mod event; pub(crate) mod fidelity; pub(crate) mod git; @@ -41,7 +40,6 @@ use fabro_sandbox::Sandbox; use self::artifact::ArtifactLifecycle; use self::auto_status::AutoStatusLifecycle; use self::circuit_breaker::CircuitBreakerLifecycle; -use self::disk::DiskLifecycle; use self::event::EventLifecycle; use self::fidelity::FidelityLifecycle; use self::git::{GitCheckpointResult, GitLifecycle}; @@ -60,7 +58,6 @@ pub(crate) struct WorkflowLifecycle { fidelity: FidelityLifecycle, auto_status: AutoStatusLifecycle, circuit_breaker: Arc, - disk: DiskLifecycle, git: GitLifecycle, artifact: ArtifactLifecycle, on_node: crate::OnNodeCallback, @@ -143,13 +140,6 @@ impl WorkflowLifecycle { let fidelity = FidelityLifecycle::new(Arc::clone(&graph)); - let disk = DiskLifecycle { - run_dir: run_dir.clone(), - checkpoint_git_result: Arc::clone(&checkpoint_git_result), - circuit_breaker: Arc::clone(&circuit_breaker), - checkpoint_enabled: true, - }; - let start_node_id = graph.find_start_node().map(|n| n.id.clone()); let git = GitLifecycle { @@ -181,7 +171,6 @@ impl WorkflowLifecycle { fidelity, auto_status: AutoStatusLifecycle, circuit_breaker, - disk, git, artifact, on_node, @@ -311,7 +300,6 @@ impl RunLifecycle for WorkflowLifecycle { self.circuit_breaker.after_node(node, result, state).await?; self.event.after_node(node, result, state).await?; self.hook.after_node(node, result, state).await?; - self.disk.after_node(node, result, state).await?; self.artifact.after_node(node, result, state).await?; Ok(()) } @@ -405,9 +393,6 @@ impl RunLifecycle for WorkflowLifecycle { self.git .on_checkpoint(node, result, next_node_id, state) .await?; - self.disk - .on_checkpoint(node, result, next_node_id, state) - .await?; self.event .on_checkpoint(node, result, next_node_id, state) .await?; diff --git a/lib/crates/fabro-workflow/src/operations/start.rs b/lib/crates/fabro-workflow/src/operations/start.rs index ab94042cb..07d856b51 100644 --- a/lib/crates/fabro-workflow/src/operations/start.rs +++ b/lib/crates/fabro-workflow/src/operations/start.rs @@ -88,7 +88,7 @@ pub async fn start(run_dir: &Path, services: StartServices) -> Result Result> { )); } - runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt)); + runs.sort_by(|a, b| { + b.start_time_dt + .cmp(&a.start_time_dt) + .then_with(|| b.run_id().cmp(&a.run_id())) + }); Ok(runs) } @@ -217,7 +221,11 @@ pub async fn scan_runs_combined(store: &SlateStore, base: &Path) -> Result = runs_by_id.into_values().collect(); - runs.sort_by(|a, b| b.start_time_dt.cmp(&a.start_time_dt)); + runs.sort_by(|a, b| { + b.start_time_dt + .cmp(&a.start_time_dt) + .then_with(|| b.run_id().cmp(&a.run_id())) + }); Ok(runs) } diff --git a/lib/crates/fabro-workflow/src/test_support.rs b/lib/crates/fabro-workflow/src/test_support.rs index 304baf5e5..e9aa2feb4 100644 --- a/lib/crates/fabro-workflow/src/test_support.rs +++ b/lib/crates/fabro-workflow/src/test_support.rs @@ -6,7 +6,7 @@ use std::time::Duration; use fabro_agent::Sandbox; use fabro_graphviz::graph::Graph as GvGraph; use fabro_store::{RunProjection, SlateStore}; -use object_store::memory::InMemory; +use object_store::local::LocalFileSystem; use crate::error::{FabroError, Result}; use crate::event::{EventEmitter, StoreProgressLogger, WorkflowRunEvent, append_workflow_event}; @@ -44,8 +44,18 @@ async fn initialized( options: InitializedOptions, ) -> InitializedState { std::fs::create_dir_all(&run_options.run_dir).expect("failed to create run dir"); + std::fs::create_dir_all(run_options.run_dir.join("store")) + .expect("failed to create local test run store dir"); + std::fs::write( + run_options.run_dir.join("id.txt"), + run_options.run_id.to_string(), + ) + .expect("failed to write run id marker"); let store = Arc::new(SlateStore::new( - Arc::new(InMemory::new()), + Arc::new( + LocalFileSystem::new_with_prefix(run_options.run_dir.join("store")) + .expect("failed to create local test run store"), + ), "", Duration::from_millis(1), )); @@ -249,6 +259,38 @@ pub async fn run_graph_from_checkpoint( executed.outcome } +pub async fn run_graph_from_checkpoint_with_state( + registry: HandlerRegistry, + emitter: Arc, + sandbox: Arc, + graph: &GvGraph, + run_options: &RunOptions, + checkpoint: &Checkpoint, +) -> Result<(Outcome, RunProjection)> { + let initialized = initialized( + registry, + emitter, + sandbox, + graph, + run_options, + InitializedOptions { + hook_runner: None, + env: HashMap::new(), + checkpoint: Some(checkpoint.clone()), + }, + ) + .await; + let executed = pipeline::execute(initialized.initialized).await; + let outcome = executed.outcome?; + initialized.store_logger.flush().await; + let state = executed + .run_store + .state() + .await + .map_err(|err| FabroError::engine(err.to_string()))?; + Ok((outcome, state)) +} + pub struct WorkflowRunner { registry: std::sync::Mutex>, emitter: Arc, @@ -329,4 +371,27 @@ impl WorkflowRunner { ) .await } + + pub async fn run_from_checkpoint_with_state( + &self, + graph: &GvGraph, + run_options: &RunOptions, + checkpoint: &Checkpoint, + ) -> Result<(Outcome, RunProjection)> { + let registry = self + .registry + .lock() + .unwrap() + .take() + .expect("WorkflowRunner may only be used once"); + Box::pin(run_graph_from_checkpoint_with_state( + registry, + Arc::clone(&self.emitter), + Arc::clone(&self.sandbox), + graph, + run_options, + checkpoint, + )) + .await + } } diff --git a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs index d434beab3..5fed2d954 100644 --- a/lib/crates/fabro-workflow/tests/it/daytona_integration.rs +++ b/lib/crates/fabro-workflow/tests/it/daytona_integration.rs @@ -21,7 +21,7 @@ use fabro_agent::Sandbox; use fabro_graphviz::graph::{AttrValue, Edge, Graph, Node}; use fabro_llm::provider::Provider; use fabro_sandbox::daytona::{DaytonaConfig, DaytonaSandbox, DaytonaSnapshotConfig}; -use fabro_store::RuntimeState; +use fabro_store::{RuntimeState, SlateStore}; use fabro_types::{RunId, Settings}; use fabro_workflow::artifact::sync_artifacts_to_env; use fabro_workflow::context::Context; @@ -34,6 +34,7 @@ use fabro_workflow::outcome::{Outcome, OutcomeExt, StageStatus}; use fabro_workflow::records::Checkpoint; use fabro_workflow::run_options::{GitCheckpointOptions, RunOptions}; use fabro_workflow::test_support::WorkflowRunner; +use object_store::local::LocalFileSystem; use ulid::Ulid; fn test_run_id(label: &str) -> RunId { @@ -43,6 +44,80 @@ fn test_run_id(label: &str) -> RunId { } fn load_checkpoint(path: &Path) -> Result> { + if !path.exists() + && path + .file_name() + .is_some_and(|name| name == "checkpoint.json") + { + let run_dir = path + .parent() + .ok_or("checkpoint path should have a parent")?; + let local_store_dir = run_dir.join("store"); + let (store_dir, run_id) = + if let Ok(run_id_text) = std::fs::read_to_string(run_dir.join("id.txt")) { + (local_store_dir, run_id_text.trim().parse()?) + } else { + let runs_dir = run_dir.parent().ok_or("run dir should have parent")?; + let storage_dir = runs_dir.parent().ok_or("runs dir should have parent")?; + let run_id: RunId = run_dir + .file_name() + .ok_or("run dir should have file name")? + .to_string_lossy() + .rsplit('-') + .next() + .ok_or("run dir should contain run id suffix")? + .parse()?; + (storage_dir.join("store"), run_id) + }; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?); + let store = Arc::new(SlateStore::new( + object_store, + "", + std::time::Duration::from_millis(1), + )); + let state = if tokio::runtime::Handle::try_current().is_ok() { + std::thread::spawn( + move || -> Result<_, Box> { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let run = runtime.block_on(store.open_run_reader(&run_id))?; + let state = runtime.block_on(async { + for attempt in 0..20 { + let state = run.state().await?; + if state.checkpoint.is_some() || attempt == 19 { + return Ok::<_, fabro_store::StoreError>(state); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + unreachable!() + })?; + Ok(state) + }, + ) + .join() + .map_err(|_| "checkpoint loader thread panicked")? + .map_err(|err| err.to_string())? + } else { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let run = runtime.block_on(store.open_run_reader(&run_id))?; + runtime.block_on(async { + for attempt in 0..20 { + let state = run.state().await?; + if state.checkpoint.is_some() || attempt == 19 { + return Ok::<_, fabro_store::StoreError>(state); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + unreachable!() + })? + }; + return state + .checkpoint + .ok_or_else(|| "checkpoint should exist in run store".into()); + } let data = std::fs::read_to_string(path)?; Ok(serde_json::from_str(&data)?) } @@ -637,10 +712,6 @@ async fn daytona_git_checkpoint_remote_emits_events() { ); } - // Assert diff.patch was written for the work node - let work_diff = dir.path().join("nodes").join("work").join("diff.patch"); - assert!(work_diff.exists(), "diff.patch should exist for work node"); - // Verify checkpoint.json has git_commit_sha let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).expect("checkpoint should load"); diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index cc42be568..08215c595 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/lib/crates/fabro-workflow/tests/it/integration.rs @@ -24,7 +24,7 @@ use fabro_interview::{ QueueInterviewer, RecordingInterviewer, }; use fabro_llm::provider::Provider; -use fabro_store::RuntimeState; +use fabro_store::{RuntimeState, SlateStore}; use fabro_types::{RunId, Settings}; use fabro_validate::{Severity, validate, validate_or_raise}; use fabro_workflow::context::Context; @@ -48,6 +48,7 @@ use fabro_workflow::transforms::stylesheet::{apply_stylesheet, parse_stylesheet} use fabro_workflow::transforms::{ StylesheetApplicationTransform, Transform, VariableExpansionTransform, }; +use object_store::local::LocalFileSystem; use ulid::Ulid; fn local_env() -> Arc { @@ -63,6 +64,76 @@ fn test_run_id(label: &str) -> RunId { } fn load_checkpoint(path: &Path) -> Result> { + if !path.exists() + && path + .file_name() + .is_some_and(|name| name == "checkpoint.json") + { + let run_dir = path + .parent() + .ok_or("checkpoint path should have a parent")?; + let local_store_dir = run_dir.join("store"); + let (store_dir, run_id) = + if let Ok(run_id_text) = std::fs::read_to_string(run_dir.join("id.txt")) { + (local_store_dir, run_id_text.trim().parse()?) + } else { + let runs_dir = run_dir.parent().ok_or("run dir should have parent")?; + let storage_dir = runs_dir.parent().ok_or("runs dir should have parent")?; + let run_id: RunId = run_dir + .file_name() + .ok_or("run dir should have file name")? + .to_string_lossy() + .rsplit('-') + .next() + .ok_or("run dir should contain run id suffix")? + .parse()?; + (storage_dir.join("store"), run_id) + }; + let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?); + let store = Arc::new(SlateStore::new(object_store, "", Duration::from_millis(1))); + let state = if tokio::runtime::Handle::try_current().is_ok() { + std::thread::spawn( + move || -> Result<_, Box> { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let run = runtime.block_on(store.open_run_reader(&run_id))?; + let state = runtime.block_on(async { + for attempt in 0..20 { + let state = run.state().await?; + if state.checkpoint.is_some() || attempt == 19 { + return Ok::<_, fabro_store::StoreError>(state); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + unreachable!() + })?; + Ok(state) + }, + ) + .join() + .map_err(|_| "checkpoint loader thread panicked")? + .map_err(|err| err.to_string())? + } else { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()?; + let run = runtime.block_on(store.open_run_reader(&run_id))?; + runtime.block_on(async { + for attempt in 0..20 { + let state = run.state().await?; + if state.checkpoint.is_some() || attempt == 19 { + return Ok::<_, fabro_store::StoreError>(state); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + unreachable!() + })? + }; + return state + .checkpoint + .ok_or_else(|| "checkpoint should exist in run store".into()); + } let data = std::fs::read_to_string(path)?; Ok(serde_json::from_str(&data)?) } @@ -241,11 +312,8 @@ async fn end_to_end_linear_pipeline() { .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // Checkpoint should exist - let checkpoint_path = dir.path().join("checkpoint.json"); - assert!(checkpoint_path.exists(), "checkpoint.json should exist"); - - let checkpoint = load_checkpoint(&checkpoint_path).expect("checkpoint should load"); + let checkpoint = + load_checkpoint(&dir.path().join("checkpoint.json")).expect("checkpoint should load"); assert!(checkpoint.completed_nodes.contains(&"start".to_string())); assert!( checkpoint @@ -368,13 +436,13 @@ async fn end_to_end_branching_pipeline() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -487,13 +555,13 @@ async fn end_to_end_human_gate_pipeline() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint.completed_nodes.contains(&"reject".to_string()), "should have traversed reject path" @@ -582,8 +650,8 @@ async fn human_gate_aborted_input_fails_closed_without_fail_route() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("engine should return Ok with fail outcome"); assert_eq!( @@ -599,7 +667,7 @@ async fn human_gate_aborted_input_fails_closed_without_fail_route() { "unexpected outcome: {outcome:?}" ); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint.node_outcomes.contains_key("gate"), "gate outcome should be checkpointed before termination" @@ -692,13 +760,13 @@ async fn human_gate_aborted_input_routes_via_outcome_fail_condition() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("aborted human gate should follow explicit fail route"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -924,13 +992,13 @@ async fn goal_gate_routes_to_retry_target_when_present() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should eventually succeed after retry"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); // gated_work should appear in completed nodes (at least twice -- first fail, then succeed) let gated_work_count = checkpoint .completed_nodes @@ -1309,13 +1377,13 @@ async fn pipeline_with_many_nodes() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("large pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); // All 10 step nodes should be in completed_nodes for name in &node_names { assert!( @@ -1730,13 +1798,13 @@ async fn end_to_end_parallel_fan_out_fan_in() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("parallel pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); // The parallel node (fan_out) and fan_in_node should be in completed_nodes. // Branch nodes run inside the parallel handler, so they are not recorded @@ -1842,14 +1910,14 @@ async fn resume_from_checkpoint_completes_pipeline() { host_repo_path: None, git: None, }; - let outcome = engine - .run_from_checkpoint(&graph, &run_options, &checkpoint) + let (outcome, state) = engine + .run_from_checkpoint_with_state(&graph, &run_options, &checkpoint) .await .expect("resume should succeed"); assert_eq!(outcome.status, StageStatus::Success); // Verify checkpoint written after resume contains step_b - let final_cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let final_cp = state.checkpoint.expect("checkpoint should be captured"); assert!( final_cp.completed_nodes.contains(&"step_b".to_string()), "step_b should have been executed after resume" @@ -1982,9 +2050,12 @@ async fn graph_goal_in_context() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert_eq!( cp.context_values.get("graph.goal"), Some(&serde_json::json!("Ship the widget")) @@ -2092,9 +2163,12 @@ async fn context_flow_between_stages() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert_eq!( cp.context_values.get("last_stage"), Some(&serde_json::json!("step_b")) @@ -2257,9 +2331,12 @@ async fn codergen_without_backend_simulated() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); let last_response = cp .context_values .get("last_response") @@ -2587,10 +2664,13 @@ async fn scenario_parallel_expert_review() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); let results = cp .context_values .get("parallel.results") @@ -2670,10 +2750,13 @@ async fn scenario_node_retries_on_retry_status() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); let retry_count = cp .node_retries .get("flaky") @@ -2798,10 +2881,13 @@ async fn scenario_bug_triage_router() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!( cp.completed_nodes.contains(&"critical".to_string()), "critical should be selected (highest weight)" @@ -2856,13 +2942,13 @@ async fn scenario_crash_recovery() { host_repo_path: None, git: None, }; - let outcome = engine - .run_from_checkpoint(&graph, &run_options, &checkpoint) + let (outcome, state) = engine + .run_from_checkpoint_with_state(&graph, &run_options, &checkpoint) .await .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); assert!(cp.completed_nodes.contains(&"b".to_string())); assert!(cp.completed_nodes.contains(&"c".to_string())); assert!(cp.completed_nodes.contains(&"a".to_string())); @@ -2964,9 +3050,12 @@ async fn manager_loop_stop_condition_satisfied_e2e() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); let manager_outcome = cp.node_outcomes.get("manager").expect("manager outcome"); assert_eq!(manager_outcome.status, StageStatus::Success); assert!( @@ -3042,9 +3131,12 @@ async fn manager_loop_max_cycles_exceeded_e2e() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); let manager_outcome = cp.node_outcomes.get("manager").expect("manager outcome"); assert_eq!(manager_outcome.status, StageStatus::Fail); assert!( @@ -3179,10 +3271,13 @@ async fn conditional_branching_success_fail_paths() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!(cp.completed_nodes.contains(&"fail_path".to_string())); assert!(!cp.completed_nodes.contains(&"success_path".to_string())); } @@ -3231,9 +3326,12 @@ async fn edge_selection_condition_match_wins_over_weight() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!(cp.completed_nodes.contains(&"cond_target".to_string())); assert!(!cp.completed_nodes.contains(&"weighted_target".to_string())); } @@ -3277,9 +3375,12 @@ async fn edge_selection_weight_breaks_ties() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!(cp.completed_nodes.contains(&"high".to_string())); assert!(!cp.completed_nodes.contains(&"low".to_string())); } @@ -3315,9 +3416,12 @@ async fn edge_selection_lexical_tiebreak() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!(cp.completed_nodes.contains(&"alpha".to_string())); assert!(!cp.completed_nodes.contains(&"beta".to_string())); } @@ -3372,9 +3476,12 @@ async fn context_updates_visible_across_nodes() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert!(cp.completed_nodes.contains(&"yes".to_string())); assert!(!cp.completed_nodes.contains(&"no".to_string())); } @@ -3470,9 +3577,12 @@ async fn custom_handler_registration_and_execution() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); assert_eq!( cp.context_values.get("custom.ran"), Some(&serde_json::json!("true")) @@ -3631,13 +3741,13 @@ async fn manager_loop_runs_child_engine_e2e() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("manager loop E2E should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -3764,11 +3874,14 @@ async fn manager_loop_context_flows_e2e() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.expect("run"); + let (outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); assert_eq!(outcome.status, StageStatus::Success); // Check that child's context updates were propagated through the manager - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); let sup_outcome = checkpoint.node_outcomes.get("supervisor").unwrap(); assert_eq!( sup_outcome.context_updates.get("review.result"), @@ -3937,13 +4050,13 @@ async fn import_e2e_through_engine() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("import E2E should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -4994,9 +5107,12 @@ async fn fidelity_stored_in_checkpoint_context() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); assert_eq!( cp.context_values.get("internal.fidelity"), Some(&serde_json::json!("summary:low")), @@ -5638,11 +5754,13 @@ async fn fidelity_checkpoint_roundtrip_preserves_fidelity() { host_repo_path: None, git: None, }; - engine.run(&graph, &run_options).await.expect("run"); + let (_outcome, state) = engine + .run_with_state(&graph, &run_options) + .await + .expect("run"); - // Load, save, load again to verify roundtrip - let checkpoint_path = dir.path().join("checkpoint.json"); - let cp1 = load_checkpoint(&checkpoint_path).expect("first load"); + // Save and load again to verify roundtrip + let cp1 = state.checkpoint.expect("checkpoint should be captured"); assert_eq!( cp1.context_values.get("internal.fidelity"), Some(&serde_json::json!("summary:high")), @@ -5793,8 +5911,8 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() { host_repo_path: None, git: None, }; - engine - .run_from_checkpoint(&graph, &run_options, &checkpoint) + let (_outcome, state) = engine + .run_from_checkpoint_with_state(&graph, &run_options, &checkpoint) .await .expect("resume should succeed"); @@ -5806,7 +5924,7 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() { ); // Verify the final checkpoint still has the fidelity - let final_cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let final_cp = state.checkpoint.expect("checkpoint should be captured"); assert_eq!( final_cp.context_values.get("internal.fidelity"), Some(&serde_json::json!("summary:low")), @@ -6463,13 +6581,13 @@ async fn human_gate_freeform_only_routes_text() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -6592,13 +6710,13 @@ async fn human_gate_freeform_with_fixed_choice_match() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint.completed_nodes.contains(&"approve".to_string()), "fixed choice match should route to approve" @@ -6706,13 +6824,13 @@ async fn human_gate_freeform_fallback_on_unmatched_text() { host_repo_path: None, git: None, }; - let outcome = engine - .run(&graph, &run_options) + let (outcome, state) = engine + .run_with_state(&graph, &run_options) .await .expect("run should succeed"); assert_eq!(outcome.status, StageStatus::Success); - let checkpoint = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let checkpoint = state.checkpoint.expect("checkpoint should be captured"); assert!( checkpoint .completed_nodes @@ -7303,7 +7421,7 @@ async fn hook_run_start_proceed_allows_run() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, _state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!(outcome.status, StageStatus::Success); } @@ -7484,7 +7602,7 @@ async fn hook_stage_start_matcher_no_match_proceeds() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, _state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!(outcome.status, StageStatus::Success); } @@ -8011,7 +8129,7 @@ async fn hook_json_proceed_explicit() { let dir = tempfile::tempdir().unwrap(); let run_options = make_run_options(dir.path()); - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, _state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!(outcome.status, StageStatus::Success); } @@ -10002,18 +10120,7 @@ async fn git_checkpoint_host_emits_events_and_diff_patch() { "all SHAs should be 40-char hex, got: {git_events:?}" ); - // 7. diff.patch is NOT written for the start node (git checkpoint skipped) - let start_diff = run_dir - .path() - .join("nodes") - .join("start") - .join("diff.patch"); - assert!( - !start_diff.exists(), - "diff.patch should not exist for start node (git checkpoint skipped)" - ); - - // 8. Verify checkpoint.json has git_commit_sha + // 7. Verify checkpoint has git_commit_sha let checkpoint = load_checkpoint(&run_dir.path().join("checkpoint.json")).expect("checkpoint should load"); assert!( @@ -10615,13 +10722,6 @@ async fn git_checkpoint_host_skips_empty_diff_patch() { .expect("pipeline should succeed"); assert_eq!(outcome.status, StageStatus::Success); - // diff.patch should NOT exist for the "work" node (no file changes) - let work_diff = run_dir.path().join("nodes").join("work").join("diff.patch"); - assert!( - !work_diff.exists(), - "diff.patch should not exist when there are no changes" - ); - // final.patch should NOT exist either let final_patch = run_dir.path().join("final.patch"); assert!( @@ -11217,13 +11317,13 @@ async fn e2e_failure_signature_persisted_in_context() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); // Pipeline reaches exit (terminal) with goal gates satisfied. // Per spec, reaching exit with satisfied goal gates returns SUCCESS. assert_eq!(outcome.status, StageStatus::Success); // Verify checkpoint has failure_signature in context - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); let sig_value = cp .context_values .get("failure_signature") @@ -11280,9 +11380,9 @@ async fn e2e_failure_signature_hint_overrides_reason_in_context() { host_repo_path: None, git: None, }; - let _outcome = engine.run(&graph, &run_options).await.unwrap(); + let (_outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should be captured"); let sig_str = cp .context_values .get("failure_signature") @@ -11335,11 +11435,11 @@ async fn e2e_signature_maps_persist_in_checkpoint() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!(outcome.status, StageStatus::Success); - // Load checkpoint and verify signature maps - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + // Verify signature maps persisted to the run state checkpoint. + let cp = state.checkpoint.expect("checkpoint should be captured"); // The pipeline had 3 deterministic failures at "work" before succeeding. // loop_failure_signatures should have recorded them. assert!( @@ -11525,7 +11625,7 @@ async fn e2e_circuit_breaker_does_not_fire_below_limit() { host_repo_path: None, git: None, }; - let outcome = engine.run(&graph, &run_options).await.unwrap(); + let (outcome, state) = engine.run_with_state(&graph, &run_options).await.unwrap(); assert_eq!( outcome.status, StageStatus::Success, @@ -11533,7 +11633,7 @@ async fn e2e_circuit_breaker_does_not_fire_below_limit() { ); // Verify signatures were tracked but didn't trigger abort - let cp = load_checkpoint(&dir.path().join("checkpoint.json")).unwrap(); + let cp = state.checkpoint.expect("checkpoint should exist"); let total_failures: usize = cp.loop_failure_signatures.values().sum(); assert_eq!( total_failures, 4,