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();