From 96bcff05b321aaf7a94cba893be2cc30dba38469 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Sun, 8 Mar 2026 22:57:32 -0400 Subject: [PATCH] Add duration_ms, node_visits, and base_branch to shadow branch structs Enrich Outcome with stage duration, Checkpoint with persisted node visit counts (avoiding fragile reconstruction on resume), and Manifest with the base branch name for offline analysis without git lookups. Co-Authored-By: Claude Opus 4.6 --- crates/arc-workflows/src/checkpoint.rs | 9 +++++++++ crates/arc-workflows/src/engine.rs | 13 ++++++++++--- crates/arc-workflows/src/git.rs | 3 +++ crates/arc-workflows/src/manifest.rs | 3 +++ crates/arc-workflows/src/outcome.rs | 8 ++++++++ crates/arc-workflows/src/retro.rs | 2 ++ crates/arc-workflows/tests/integration.rs | 9 +++++++++ 7 files changed, 44 insertions(+), 3 deletions(-) diff --git a/crates/arc-workflows/src/checkpoint.rs b/crates/arc-workflows/src/checkpoint.rs index d3df58f2a..c4d741712 100644 --- a/crates/arc-workflows/src/checkpoint.rs +++ b/crates/arc-workflows/src/checkpoint.rs @@ -33,6 +33,9 @@ pub struct Checkpoint { /// Failure signature counts across loop_restart edges. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub restart_failure_signatures: HashMap, + /// Per-node visit counts persisted for accurate resume. + #[serde(default, skip_serializing_if = "HashMap::is_empty")] + pub node_visits: HashMap, } impl Checkpoint { @@ -47,6 +50,7 @@ impl Checkpoint { next_node_id: Option, loop_failure_signatures: HashMap, restart_failure_signatures: HashMap, + node_visits: HashMap, ) -> Self { Self { timestamp: Utc::now(), @@ -60,6 +64,7 @@ impl Checkpoint { git_commit_sha: None, loop_failure_signatures, restart_failure_signatures, + node_visits, } } @@ -103,6 +108,7 @@ mod tests { None, HashMap::new(), HashMap::new(), + HashMap::new(), ); assert_eq!(cp.current_node, "node_a"); @@ -142,6 +148,7 @@ mod tests { Some("next_step".to_string()), HashMap::new(), HashMap::new(), + HashMap::new(), ); cp.save(&path).unwrap(); @@ -190,6 +197,7 @@ mod tests { None, HashMap::new(), HashMap::new(), + HashMap::new(), ); let json = serde_json::to_string(&cp).unwrap(); @@ -230,6 +238,7 @@ mod tests { None, loop_sigs, restart_sigs, + HashMap::new(), ); cp.save(&path).unwrap(); diff --git a/crates/arc-workflows/src/engine.rs b/crates/arc-workflows/src/engine.rs index 26ec6250a..6115ecbd5 100644 --- a/crates/arc-workflows/src/engine.rs +++ b/crates/arc-workflows/src/engine.rs @@ -303,6 +303,7 @@ fn write_manifest( run_branch: config.run_branch.clone(), base_sha: config.base_sha.clone(), labels: config.labels.clone(), + base_branch: config.base_branch.clone(), }; let _ = std::fs::create_dir_all(logs_root); let _ = manifest.save(&logs_root.join("manifest.json")); @@ -1328,9 +1329,13 @@ impl WorkflowRunEngine { context.append_log(log_entry.clone()); } completed_nodes = cp.completed_nodes.clone(); - // Rebuild visit counts from completed_nodes (which records every visit) - for id in &completed_nodes { - *loop_state.node_visits.entry(id.clone()).or_insert(0) += 1; + // Use persisted node_visits; fall back to reconstruction for old checkpoints + if cp.node_visits.is_empty() { + for id in &completed_nodes { + *loop_state.node_visits.entry(id.clone()).or_insert(0) += 1; + } + } else { + loop_state.node_visits = cp.node_visits.clone(); } // Gap #5: Restore retry counters from checkpoint node_retries = cp.node_retries.clone(); @@ -1781,6 +1786,7 @@ impl WorkflowRunEngine { } // Step 3: Record completion + outcome.duration_ms = Some(stage_duration_ms); completed_nodes.push(node.id.clone()); node_outcomes.insert(node.id.clone(), outcome.clone()); previous_node_id = Some(node.id.clone()); @@ -1881,6 +1887,7 @@ impl WorkflowRunEngine { next_node_id_for_checkpoint, loop_state.loop_failure_signatures.clone(), loop_state.restart_failure_signatures.clone(), + loop_state.node_visits.clone(), ); let checkpoint_path = config.logs_root.join("checkpoint.json"); if let Err(e) = checkpoint.save(&checkpoint_path) { diff --git a/crates/arc-workflows/src/git.rs b/crates/arc-workflows/src/git.rs index f80891965..3f70981e7 100644 --- a/crates/arc-workflows/src/git.rs +++ b/crates/arc-workflows/src/git.rs @@ -850,6 +850,7 @@ mod tests { Some("node_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp_json = serde_json::to_vec_pretty(&cp).unwrap(); store.write_checkpoint("RUN2", &cp_json, &[]).unwrap(); @@ -884,6 +885,7 @@ mod tests { None, std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp1_json = serde_json::to_vec_pretty(&cp1).unwrap(); store.write_checkpoint("RUN3", &cp1_json, &[]).unwrap(); @@ -897,6 +899,7 @@ mod tests { Some("node_c".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let cp2_json = serde_json::to_vec_pretty(&cp2).unwrap(); store.write_checkpoint("RUN3", &cp2_json, &[]).unwrap(); diff --git a/crates/arc-workflows/src/manifest.rs b/crates/arc-workflows/src/manifest.rs index 8ad14f40e..f4e508b1a 100644 --- a/crates/arc-workflows/src/manifest.rs +++ b/crates/arc-workflows/src/manifest.rs @@ -20,6 +20,8 @@ pub struct Manifest { pub base_sha: Option, #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub labels: HashMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_branch: Option, } impl Manifest { @@ -47,6 +49,7 @@ mod tests { run_branch: Some("feature/test".to_string()), base_sha: Some("abc123".to_string()), labels: HashMap::from([("env".into(), "test".into())]), + base_branch: None, } } diff --git a/crates/arc-workflows/src/outcome.rs b/crates/arc-workflows/src/outcome.rs index 158892ada..3756c2721 100644 --- a/crates/arc-workflows/src/outcome.rs +++ b/crates/arc-workflows/src/outcome.rs @@ -102,6 +102,9 @@ pub struct Outcome { /// Used by the parallel handler to skip re-executing branch nodes. #[serde(default, skip_serializing_if = "Option::is_none")] pub jump_to_node: Option, + /// Wall-clock duration of the stage execution in milliseconds. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub duration_ms: Option, } impl Outcome { @@ -117,6 +120,7 @@ impl Outcome { usage: None, files_touched: Vec::new(), jump_to_node: None, + duration_ms: None, } } @@ -132,6 +136,7 @@ impl Outcome { usage: None, files_touched: Vec::new(), jump_to_node: None, + duration_ms: None, } } @@ -149,6 +154,7 @@ impl Outcome { usage: None, files_touched: Vec::new(), jump_to_node: None, + duration_ms: None, } } @@ -166,6 +172,7 @@ impl Outcome { usage: None, files_touched: Vec::new(), jump_to_node: None, + duration_ms: None, } } @@ -190,6 +197,7 @@ impl Outcome { usage: None, files_touched: Vec::new(), jump_to_node: None, + duration_ms: None, } } diff --git a/crates/arc-workflows/src/retro.rs b/crates/arc-workflows/src/retro.rs index b8f283e9e..906a8437c 100644 --- a/crates/arc-workflows/src/retro.rs +++ b/crates/arc-workflows/src/retro.rs @@ -350,6 +350,7 @@ mod tests { git_commit_sha: None, loop_failure_signatures: HashMap::new(), restart_failure_signatures: HashMap::new(), + node_visits: HashMap::new(), } } @@ -410,6 +411,7 @@ mod tests { git_commit_sha: None, loop_failure_signatures: HashMap::new(), restart_failure_signatures: HashMap::new(), + node_visits: HashMap::new(), }; let retro = derive_retro( diff --git a/crates/arc-workflows/tests/integration.rs b/crates/arc-workflows/tests/integration.rs index 1236f87e2..5e815f8a6 100644 --- a/crates/arc-workflows/tests/integration.rs +++ b/crates/arc-workflows/tests/integration.rs @@ -1141,6 +1141,7 @@ fn checkpoint_save_and_resume_roundtrip() { None, std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); checkpoint.save(&path).expect("save should succeed"); @@ -1620,6 +1621,7 @@ async fn resume_from_checkpoint_completes_pipeline() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -1721,6 +1723,7 @@ async fn resume_from_checkpoint_preserves_goal_gate_outcomes() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -2672,6 +2675,7 @@ async fn scenario_crash_recovery() { Some("b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let dir = tempfile::tempdir().unwrap(); @@ -4678,6 +4682,7 @@ async fn fidelity_resume_degrades_full_to_summary_high() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -4776,6 +4781,7 @@ async fn fidelity_resume_degrade_only_affects_first_hop() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -4861,6 +4867,7 @@ async fn fidelity_resume_no_degrade_when_not_full() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -5739,6 +5746,7 @@ async fn fidelity_resume_preserves_context_values_across_checkpoint() { Some("step_b".to_string()), std::collections::HashMap::new(), std::collections::HashMap::new(), + std::collections::HashMap::new(), ); let captures = FidelityCaptures::new(); @@ -11903,6 +11911,7 @@ fn e2e_checkpoint_signatures_roundtrip() { None, loop_sigs, restart_sigs, + std::collections::HashMap::new(), ); cp.save(&path).unwrap();