1use std::collections::BTreeMap;
2
3use serde::{Deserialize, Serialize};
4use sha2::{Digest, Sha256};
5use thiserror::Error;
6
7use crate::{
8 BranchResult, BranchSpec, CondSpec, ControlNodeKind, ControlNodeResult, ExpandSpec, LeafResult,
9 LeafSpec, LoopUntilSpec, SequenceSpec, WorkflowExecution, WorkflowExecutionError,
10 WorkflowMemoUsage, WorkflowNode, WorkflowRunStatus, WorkflowSpec, WorkflowUsage,
11 validate_workflow_node_shapes, validate_workflow_nodes,
12};
13
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)]
15pub struct ReplayOptions {
16 #[serde(default)]
17 pub allow_live_replay: bool,
18}
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct WorkflowReplayTrace {
22 pub trace_id: String,
23 #[serde(default)]
24 pub leaf_records: Vec<ReplayLeafRecord>,
25 #[serde(default)]
26 pub control_records: Vec<ReplayControlRecord>,
27}
28
29#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
30pub struct ReplayLeafRecord {
31 pub trace_id: String,
32 pub leaf_id: String,
33 pub input_hash: String,
34 pub result: LeafResult,
35}
36
37#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
38pub struct ReplayControlRecord {
39 pub trace_id: String,
40 pub node_id: String,
41 pub kind: ControlNodeKind,
42 pub result: ControlNodeResult,
43 #[serde(default)]
44 pub generated_nodes: Vec<WorkflowNode>,
45}
46
47#[derive(Debug, Clone)]
48pub struct WorkflowReplayExecutor {
49 trace_id: String,
50 options: ReplayOptions,
51 leaf_records: BTreeMap<ReplayLeafKey, LeafResult>,
52 control_records: BTreeMap<ReplayControlKey, ReplayControlRecord>,
53 resolved_outputs: BTreeMap<String, Option<String>>,
54}
55
56impl WorkflowReplayExecutor {
57 pub fn new(trace: WorkflowReplayTrace) -> Self {
58 Self::with_options(trace, ReplayOptions::default())
59 }
60
61 pub fn with_options(trace: WorkflowReplayTrace, options: ReplayOptions) -> Self {
62 let trace_id = trace.trace_id;
63 let leaf_records = trace
64 .leaf_records
65 .into_iter()
66 .map(|record| {
67 (
68 ReplayLeafKey {
69 trace_id: record.trace_id,
70 leaf_id: record.leaf_id,
71 input_hash: record.input_hash,
72 },
73 record.result,
74 )
75 })
76 .collect();
77 let control_records = trace
78 .control_records
79 .into_iter()
80 .map(|record| {
81 (
82 ReplayControlKey {
83 trace_id: record.trace_id.clone(),
84 node_id: record.node_id.clone(),
85 kind: record.kind,
86 },
87 record,
88 )
89 })
90 .collect();
91
92 Self {
93 trace_id,
94 options,
95 leaf_records,
96 control_records,
97 resolved_outputs: BTreeMap::new(),
98 }
99 }
100
101 pub fn run(&mut self, spec: &WorkflowSpec) -> Result<WorkflowExecution, WorkflowReplayError> {
102 validate_workflow_nodes(&spec.nodes)?;
103 let mut execution = WorkflowExecution::default();
104 self.execute_nodes(spec, &spec.nodes, &mut execution)?;
105 Ok(execution)
106 }
107
108 fn execute_nodes(
109 &mut self,
110 spec: &WorkflowSpec,
111 nodes: &[WorkflowNode],
112 execution: &mut WorkflowExecution,
113 ) -> Result<(), WorkflowReplayError> {
114 for node in nodes {
115 self.execute_node(spec, node, execution)?;
116 }
117 Ok(())
118 }
119
120 fn execute_node(
121 &mut self,
122 spec: &WorkflowSpec,
123 node: &WorkflowNode,
124 execution: &mut WorkflowExecution,
125 ) -> Result<(), WorkflowReplayError> {
126 match node {
127 WorkflowNode::BranchSet(branch) => self.execute_branch_set(spec, branch, execution),
128 WorkflowNode::Leaf(leaf) => self.execute_leaf(spec, leaf, execution),
129 WorkflowNode::Sequence(sequence) => self.execute_sequence(spec, sequence, execution),
130 WorkflowNode::Reduce(reduce) => self.replay_recorded_control(
131 reduce.id.as_str(),
132 ControlNodeKind::Reduce,
133 execution,
134 Some(reduce.inputs.clone()),
135 Some(reduce.prompt.clone()),
136 ),
137 WorkflowNode::TeacherReview(review) => self.replay_recorded_control(
138 review.id.as_str(),
139 ControlNodeKind::TeacherReview,
140 execution,
141 Some(review.candidates.clone()),
142 Some("teacher review replayed from recorded candidates".to_string()),
143 ),
144 WorkflowNode::LoopUntil(loop_until) => {
145 self.execute_loop_until(spec, loop_until, execution)
146 }
147 WorkflowNode::Cond(cond) => self.execute_cond(spec, cond, execution),
148 WorkflowNode::Expand(expand) => self.execute_expand(spec, expand, execution),
149 }
150 }
151
152 fn execute_branch_set(
153 &mut self,
154 spec: &WorkflowSpec,
155 branch: &BranchSpec,
156 execution: &mut WorkflowExecution,
157 ) -> Result<(), WorkflowReplayError> {
158 let before = execution.leaf_results.len();
159 self.execute_nodes(spec, &branch.children, execution)?;
160 let status = branch_status(&execution.leaf_results[before..]);
161 let mut usage = WorkflowUsage::default();
162 let mut memo_usage = WorkflowMemoUsage::default();
163 for result in &execution.leaf_results[before..] {
164 usage.add_assign(result.usage);
165 memo_usage.add_assign(result.memo_usage);
166 }
167 if status == WorkflowRunStatus::ReplayDiverged {
168 execution.mark_replay_diverged();
169 } else if status == WorkflowRunStatus::Failed {
170 execution.mark_failed();
171 }
172 execution.branch_results.push(BranchResult {
173 branch_id: branch.id.clone(),
174 task_id: branch.id.clone(),
175 status,
176 usage,
177 memo_usage,
178 artifacts: Vec::new(),
179 notes: Some("replay branch set evaluated from recorded leaf results".to_string()),
180 });
181 self.replay_recorded_control(
182 branch.id.as_str(),
183 ControlNodeKind::BranchSet,
184 execution,
185 Some(branch.children.iter().map(workflow_node_id).collect()),
186 Some("branch set replayed declared children".to_string()),
187 )
188 }
189
190 fn execute_leaf(
191 &mut self,
192 spec: &WorkflowSpec,
193 leaf: &LeafSpec,
194 execution: &mut WorkflowExecution,
195 ) -> Result<(), WorkflowReplayError> {
196 let inputs = resolved_inputs_for_leaf(leaf, &self.resolved_outputs);
197 let input_hash = compute_leaf_input_hash(spec, leaf, &inputs)?;
198 let key = ReplayLeafKey {
199 trace_id: self.trace_id.clone(),
200 leaf_id: leaf.id.clone(),
201 input_hash,
202 };
203
204 let Some(result) = self.leaf_records.get(&key).cloned() else {
205 if self.options.allow_live_replay {
206 return Err(WorkflowReplayError::LiveReplayUnavailable {
207 leaf: leaf.id.clone(),
208 });
209 }
210 execution.mark_replay_diverged();
211 let result = LeafResult {
212 leaf_id: leaf.id.clone(),
213 task_id: leaf.id.clone(),
214 role: leaf.role.clone(),
215 profile: leaf.profile.clone(),
216 status: WorkflowRunStatus::ReplayDiverged,
217 usage: WorkflowUsage::default(),
218 memo_usage: WorkflowMemoUsage::default(),
219 output: None,
220 artifacts: Vec::new(),
221 schema_error: None,
222 };
223 self.resolved_outputs.insert(leaf.id.clone(), None);
224 execution.leaf_results.push(result);
225 return Ok(());
226 };
227
228 if result.status == WorkflowRunStatus::ReplayDiverged {
229 execution.mark_replay_diverged();
230 } else if result.status == WorkflowRunStatus::Failed {
231 execution.mark_failed();
232 }
233 execution.usage.add_assign(result.usage);
234 execution.memo_usage.add_assign(result.memo_usage);
235 self.resolved_outputs
236 .insert(leaf.id.clone(), result.output.clone());
237 execution.leaf_results.push(result);
238 Ok(())
239 }
240
241 fn execute_sequence(
242 &mut self,
243 spec: &WorkflowSpec,
244 sequence: &SequenceSpec,
245 execution: &mut WorkflowExecution,
246 ) -> Result<(), WorkflowReplayError> {
247 self.execute_nodes(spec, &sequence.children, execution)?;
248 self.replay_recorded_control(
249 sequence.id.as_str(),
250 ControlNodeKind::Sequence,
251 execution,
252 Some(sequence.children.iter().map(workflow_node_id).collect()),
253 Some("sequence replayed in declaration order".to_string()),
254 )
255 }
256
257 fn execute_loop_until(
258 &mut self,
259 spec: &WorkflowSpec,
260 loop_until: &LoopUntilSpec,
261 execution: &mut WorkflowExecution,
262 ) -> Result<(), WorkflowReplayError> {
263 let record = self.control_record(loop_until.id.as_str(), ControlNodeKind::LoopUntil);
264 let selected = record
265 .as_ref()
266 .map(|record| record.result.selected_children.clone())
267 .unwrap_or_else(|| loop_until.children.iter().map(workflow_node_id).collect());
268 let children = select_nodes(&loop_until.children, &selected);
269 self.execute_nodes(spec, &children, execution)?;
270 self.push_control_or_diverge(
271 loop_until.id.as_str(),
272 ControlNodeKind::LoopUntil,
273 execution,
274 record,
275 Some(selected),
276 Some("loop_until replayed recorded child selection".to_string()),
277 );
278 Ok(())
279 }
280
281 fn execute_cond(
282 &mut self,
283 spec: &WorkflowSpec,
284 cond: &CondSpec,
285 execution: &mut WorkflowExecution,
286 ) -> Result<(), WorkflowReplayError> {
287 let record = self.control_record(cond.id.as_str(), ControlNodeKind::Cond);
288 let selected = record
289 .as_ref()
290 .map(|record| record.result.selected_children.clone())
291 .unwrap_or_default();
292 let available = cond
293 .then_nodes
294 .iter()
295 .chain(cond.else_nodes.iter())
296 .cloned()
297 .collect::<Vec<_>>();
298 let nodes = select_nodes(&available, &selected);
299 self.execute_nodes(spec, &nodes, execution)?;
300 self.push_control_or_diverge(
301 cond.id.as_str(),
302 ControlNodeKind::Cond,
303 execution,
304 record,
305 Some(selected),
306 Some("cond replayed recorded branch selection".to_string()),
307 );
308 Ok(())
309 }
310
311 fn execute_expand(
312 &mut self,
313 spec: &WorkflowSpec,
314 expand: &ExpandSpec,
315 execution: &mut WorkflowExecution,
316 ) -> Result<(), WorkflowReplayError> {
317 let record = self.control_record(expand.id.as_str(), ControlNodeKind::Expand);
318 let generated_nodes = record
319 .as_ref()
320 .map(|record| record.generated_nodes.clone())
321 .unwrap_or_default();
322 validate_workflow_node_shapes(&generated_nodes)?;
323 self.execute_nodes(spec, &generated_nodes, execution)?;
324 let selected = record
325 .as_ref()
326 .map(|record| record.result.selected_children.clone())
327 .unwrap_or_else(|| generated_nodes.iter().map(workflow_node_id).collect());
328 self.push_control_or_diverge(
329 expand.id.as_str(),
330 ControlNodeKind::Expand,
331 execution,
332 record,
333 Some(selected),
334 Some(format!(
335 "expand replayed recorded nodes from {}",
336 expand.source
337 )),
338 );
339 Ok(())
340 }
341
342 fn replay_recorded_control(
343 &self,
344 node_id: &str,
345 kind: ControlNodeKind,
346 execution: &mut WorkflowExecution,
347 fallback_children: Option<Vec<String>>,
348 fallback_summary: Option<String>,
349 ) -> Result<(), WorkflowReplayError> {
350 let record = self.control_record(node_id, kind);
351 self.push_control_or_diverge(
352 node_id,
353 kind,
354 execution,
355 record,
356 fallback_children,
357 fallback_summary,
358 );
359 Ok(())
360 }
361
362 fn control_record(&self, node_id: &str, kind: ControlNodeKind) -> Option<ReplayControlRecord> {
363 self.control_records
364 .get(&ReplayControlKey {
365 trace_id: self.trace_id.clone(),
366 node_id: node_id.to_string(),
367 kind,
368 })
369 .cloned()
370 }
371
372 fn push_control_or_diverge(
373 &self,
374 node_id: &str,
375 kind: ControlNodeKind,
376 execution: &mut WorkflowExecution,
377 record: Option<ReplayControlRecord>,
378 fallback_children: Option<Vec<String>>,
379 fallback_summary: Option<String>,
380 ) {
381 let Some(record) = record else {
382 execution.mark_replay_diverged();
383 execution.control_node_results.push(ControlNodeResult {
384 node_id: node_id.to_string(),
385 kind,
386 status: WorkflowRunStatus::ReplayDiverged,
387 selected_children: fallback_children.unwrap_or_default(),
388 summary: fallback_summary
389 .or_else(|| Some("missing replay control record".to_string())),
390 });
391 return;
392 };
393 if record.result.status == WorkflowRunStatus::ReplayDiverged {
394 execution.mark_replay_diverged();
395 } else if record.result.status == WorkflowRunStatus::Failed {
396 execution.mark_failed();
397 }
398 execution.control_node_results.push(record.result);
399 }
400}
401
402#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
403struct ReplayLeafKey {
404 trace_id: String,
405 leaf_id: String,
406 input_hash: String,
407}
408
409#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
410struct ReplayControlKey {
411 trace_id: String,
412 node_id: String,
413 kind: ControlNodeKind,
414}
415
416#[derive(Debug, Clone, PartialEq, Eq, Error)]
417pub enum WorkflowReplayError {
418 #[error(transparent)]
419 Validation(#[from] WorkflowExecutionError),
420 #[error("live replay requested for leaf `{leaf}`, but no live replay provider is configured")]
421 LiveReplayUnavailable { leaf: String },
422 #[error("failed to compute replay input hash: {reason}")]
423 InputHash { reason: String },
424}
425
426pub fn compute_leaf_input_hash(
427 spec: &WorkflowSpec,
428 leaf: &LeafSpec,
429 resolved_inputs: &BTreeMap<String, Option<String>>,
430) -> Result<String, WorkflowReplayError> {
431 let input = ReplayLeafInput {
432 workflow_id: spec.id.as_deref(),
433 workflow_goal: spec.goal.as_str(),
434 leaf,
435 resolved_inputs,
436 };
437 let bytes = serde_json::to_vec(&input).map_err(|error| WorkflowReplayError::InputHash {
438 reason: error.to_string(),
439 })?;
440 let digest = Sha256::digest(bytes);
441 Ok(hex_bytes(digest))
442}
443
444fn hex_bytes(bytes: impl AsRef<[u8]>) -> String {
445 let bytes = bytes.as_ref();
446 let mut out = String::with_capacity(bytes.len() * 2);
447 for byte in bytes {
448 use std::fmt::Write as _;
449 let _ = write!(&mut out, "{byte:02x}");
450 }
451 out
452}
453
454#[derive(Serialize)]
455struct ReplayLeafInput<'a> {
456 workflow_id: Option<&'a str>,
457 workflow_goal: &'a str,
458 leaf: &'a LeafSpec,
459 resolved_inputs: &'a BTreeMap<String, Option<String>>,
460}
461
462fn resolved_inputs_for_leaf(
463 leaf: &LeafSpec,
464 resolved_outputs: &BTreeMap<String, Option<String>>,
465) -> BTreeMap<String, Option<String>> {
466 leaf.depends_on_results
467 .iter()
468 .map(|dependency| {
469 (
470 dependency.clone(),
471 resolved_outputs.get(dependency).cloned().unwrap_or(None),
472 )
473 })
474 .collect()
475}
476
477fn branch_status(results: &[LeafResult]) -> WorkflowRunStatus {
478 if results
479 .iter()
480 .any(|result| result.status == WorkflowRunStatus::ReplayDiverged)
481 {
482 WorkflowRunStatus::ReplayDiverged
483 } else if results
484 .iter()
485 .any(|result| result.status != WorkflowRunStatus::Succeeded)
486 {
487 WorkflowRunStatus::Failed
488 } else {
489 WorkflowRunStatus::Succeeded
490 }
491}
492
493fn select_nodes(nodes: &[WorkflowNode], selected: &[String]) -> Vec<WorkflowNode> {
494 let by_id: BTreeMap<_, _> = nodes
495 .iter()
496 .map(|node| (workflow_node_id(node), node.clone()))
497 .collect();
498 selected
499 .iter()
500 .filter_map(|id| by_id.get(id).cloned())
501 .collect()
502}
503
504fn workflow_node_id(node: &WorkflowNode) -> String {
505 match node {
506 WorkflowNode::BranchSet(spec) => spec.id.clone(),
507 WorkflowNode::Leaf(spec) => spec.id.clone(),
508 WorkflowNode::Sequence(spec) => spec.id.clone(),
509 WorkflowNode::Reduce(spec) => spec.id.clone(),
510 WorkflowNode::TeacherReview(spec) => spec.id.clone(),
511 WorkflowNode::LoopUntil(spec) => spec.id.clone(),
512 WorkflowNode::Cond(spec) => spec.id.clone(),
513 WorkflowNode::Expand(spec) => spec.id.clone(),
514 }
515}
516
517#[cfg(test)]
518mod tests {
519 use super::*;
520 use crate::{
521 AgentType, BudgetSpec, CondSpec, ControlNodeKind, ControlNodeResult, ExpandSpec, LeafSpec,
522 ModelPolicy, PermissionSpec, TaskMode,
523 };
524
525 fn leaf(id: &str) -> LeafSpec {
526 LeafSpec {
527 id: id.to_string(),
528 prompt: format!("run {id}"),
529 agent_type: AgentType::General,
530 role: None,
531 profile: None,
532 mode: TaskMode::ReadOnly,
533 isolation: crate::IsolationMode::Shared,
534 file_scope: Vec::new(),
535 depends_on_results: Vec::new(),
536 budget: BudgetSpec::default(),
537 permissions: PermissionSpec::default(),
538 model_policy: ModelPolicy::default(),
539 }
540 }
541
542 fn leaf_node(id: &str) -> WorkflowNode {
543 WorkflowNode::Leaf(leaf(id))
544 }
545
546 fn workflow(nodes: Vec<WorkflowNode>) -> WorkflowSpec {
547 WorkflowSpec {
548 id: Some("wf".to_string()),
549 goal: "replay safely".to_string(),
550 description: None,
551 budget: BudgetSpec::default(),
552 permissions: PermissionSpec::default(),
553 model_policy: ModelPolicy::default(),
554 promotion_policy: crate::PromotionPolicy::default(),
555 gates: Vec::new(),
556 nodes,
557 }
558 }
559
560 fn leaf_result(id: &str, output: &str) -> LeafResult {
561 LeafResult {
562 leaf_id: id.to_string(),
563 task_id: id.to_string(),
564 role: None,
565 profile: None,
566 status: WorkflowRunStatus::Succeeded,
567 usage: WorkflowUsage {
568 input_tokens: 10,
569 output_tokens: 5,
570 cost_microusd: 2,
571 },
572 memo_usage: WorkflowMemoUsage::default(),
573 output: Some(output.to_string()),
574 artifacts: Vec::new(),
575 schema_error: None,
576 }
577 }
578
579 fn leaf_record(spec: &WorkflowSpec, leaf: &LeafSpec, result: LeafResult) -> ReplayLeafRecord {
580 ReplayLeafRecord {
581 trace_id: "trace-1".to_string(),
582 leaf_id: leaf.id.clone(),
583 input_hash: compute_leaf_input_hash(spec, leaf, &BTreeMap::new()).unwrap(),
584 result,
585 }
586 }
587
588 fn control_record(
589 id: &str,
590 kind: ControlNodeKind,
591 status: WorkflowRunStatus,
592 selected_children: Vec<&str>,
593 ) -> ReplayControlRecord {
594 ReplayControlRecord {
595 trace_id: "trace-1".to_string(),
596 node_id: id.to_string(),
597 kind,
598 result: ControlNodeResult {
599 node_id: id.to_string(),
600 kind,
601 status,
602 selected_children: selected_children.into_iter().map(str::to_string).collect(),
603 summary: Some("recorded".to_string()),
604 },
605 generated_nodes: Vec::new(),
606 }
607 }
608
609 #[test]
610 fn replay_uses_recorded_leaf_outputs_not_live_calls() {
611 let scan = leaf("scan");
612 let spec = workflow(vec![WorkflowNode::Leaf(scan.clone())]);
613 let trace = WorkflowReplayTrace {
614 trace_id: "trace-1".to_string(),
615 leaf_records: vec![leaf_record(
616 &spec,
617 &scan,
618 leaf_result("scan", "recorded output"),
619 )],
620 control_records: Vec::new(),
621 };
622
623 let execution = WorkflowReplayExecutor::new(trace)
624 .run(&spec)
625 .expect("replay should run");
626
627 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
628 assert_eq!(
629 execution.leaf_results[0].output.as_deref(),
630 Some("recorded output")
631 );
632 assert_eq!(execution.usage.cost_microusd, 2);
633 }
634
635 #[test]
636 fn workflow_trace_can_replay_from_records() {
637 let scan = leaf("scan");
638 let summarize = leaf("summarize");
639 let spec = workflow(vec![WorkflowNode::BranchSet(BranchSpec {
640 id: "discover".to_string(),
641 description: None,
642 parallel: true,
643 budget: BudgetSpec::default(),
644 permissions: PermissionSpec::default(),
645 model_policy: ModelPolicy::default(),
646 children: vec![
647 WorkflowNode::Leaf(scan.clone()),
648 WorkflowNode::Leaf(summarize.clone()),
649 ],
650 })]);
651 let trace = WorkflowReplayTrace {
652 trace_id: "trace-1".to_string(),
653 leaf_records: vec![
654 leaf_record(&spec, &scan, leaf_result("scan", "scan ok")),
655 leaf_record(&spec, &summarize, leaf_result("summarize", "summary ok")),
656 ],
657 control_records: vec![control_record(
658 "discover",
659 ControlNodeKind::BranchSet,
660 WorkflowRunStatus::Succeeded,
661 vec!["scan", "summarize"],
662 )],
663 };
664
665 let execution = WorkflowReplayExecutor::new(trace)
666 .run(&spec)
667 .expect("replay should run");
668
669 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
670 assert_eq!(execution.leaf_results.len(), 2);
671 assert_eq!(
672 execution.branch_results[0].status,
673 WorkflowRunStatus::Succeeded
674 );
675 assert_eq!(execution.branch_results[0].usage.cost_microusd, 4);
676 assert_eq!(execution.usage.cost_microusd, 4);
677 }
678
679 #[test]
680 fn workflow_replay_diverges_on_missing_leaf_record() {
681 let spec = workflow(vec![leaf_node("scan")]);
682 let trace = WorkflowReplayTrace {
683 trace_id: "trace-1".to_string(),
684 leaf_records: Vec::new(),
685 control_records: Vec::new(),
686 };
687
688 let execution = WorkflowReplayExecutor::new(trace)
689 .run(&spec)
690 .expect("missing records should be reported as divergence");
691
692 assert_eq!(execution.status, WorkflowRunStatus::ReplayDiverged);
693 assert_eq!(
694 execution.leaf_results[0].status,
695 WorkflowRunStatus::ReplayDiverged
696 );
697 assert_eq!(execution.leaf_results[0].output, None);
698 }
699
700 #[test]
701 fn live_replay_requires_explicit_opt_in() {
702 let spec = workflow(vec![leaf_node("scan")]);
703 let trace = WorkflowReplayTrace {
704 trace_id: "trace-1".to_string(),
705 leaf_records: Vec::new(),
706 control_records: Vec::new(),
707 };
708 let err = WorkflowReplayExecutor::with_options(
709 trace,
710 ReplayOptions {
711 allow_live_replay: true,
712 },
713 )
714 .run(&spec)
715 .expect_err("live replay cannot run without a configured provider");
716
717 assert!(matches!(
718 err,
719 WorkflowReplayError::LiveReplayUnavailable { .. }
720 ));
721 assert!(!ReplayOptions::default().allow_live_replay);
722 }
723
724 #[test]
725 fn leaf_input_hash_is_stable_across_object_key_order() {
726 let mut downstream = leaf("summarize");
727 downstream.depends_on_results = vec!["b".to_string(), "a".to_string()];
728 let spec = workflow(vec![WorkflowNode::Leaf(downstream.clone())]);
729 let mut left = BTreeMap::new();
730 left.insert("a".to_string(), Some("one".to_string()));
731 left.insert("b".to_string(), Some("two".to_string()));
732 let mut right = BTreeMap::new();
733 right.insert("b".to_string(), Some("two".to_string()));
734 right.insert("a".to_string(), Some("one".to_string()));
735
736 let left_hash = compute_leaf_input_hash(&spec, &downstream, &left).unwrap();
737 let right_hash = compute_leaf_input_hash(&spec, &downstream, &right).unwrap();
738
739 assert_eq!(left_hash, right_hash);
740 }
741
742 #[test]
743 fn leaf_input_hash_diverges_on_profile_change() {
744 let base = leaf("review");
745 let mut profiled = base.clone();
746 profiled.profile = Some("reviewer".to_string());
747 let spec = workflow(vec![WorkflowNode::Leaf(base.clone())]);
748
749 let base_hash = compute_leaf_input_hash(&spec, &base, &BTreeMap::new()).unwrap();
750 let profiled_hash = compute_leaf_input_hash(&spec, &profiled, &BTreeMap::new()).unwrap();
751
752 assert_ne!(base_hash, profiled_hash);
753 }
754
755 #[test]
756 fn replay_control_records_drive_cond_expand_loop_until() {
757 let patch = leaf("patch");
758 let generated = leaf("generated-check");
759 let spec = workflow(vec![
760 WorkflowNode::Cond(CondSpec {
761 id: "choose".to_string(),
762 condition: "patch?".to_string(),
763 then_nodes: vec![WorkflowNode::Leaf(patch.clone())],
764 else_nodes: vec![leaf_node("report")],
765 }),
766 WorkflowNode::Expand(ExpandSpec {
767 id: "split".to_string(),
768 source: "choose".to_string(),
769 max_children: None,
770 template: None,
771 }),
772 WorkflowNode::LoopUntil(crate::LoopUntilSpec {
773 id: "verify".to_string(),
774 condition: "done".to_string(),
775 max_iterations: Some(3),
776 children: vec![leaf_node("unused-live-child")],
777 }),
778 ]);
779 let mut expand_record = control_record(
780 "split",
781 ControlNodeKind::Expand,
782 WorkflowRunStatus::Succeeded,
783 vec!["generated-check"],
784 );
785 expand_record.generated_nodes = vec![WorkflowNode::Leaf(generated.clone())];
786 let trace = WorkflowReplayTrace {
787 trace_id: "trace-1".to_string(),
788 leaf_records: vec![
789 leaf_record(&spec, &patch, leaf_result("patch", "patched")),
790 leaf_record(&spec, &generated, leaf_result("generated-check", "checked")),
791 ],
792 control_records: vec![
793 control_record(
794 "choose",
795 ControlNodeKind::Cond,
796 WorkflowRunStatus::Succeeded,
797 vec!["patch"],
798 ),
799 expand_record,
800 control_record(
801 "verify",
802 ControlNodeKind::LoopUntil,
803 WorkflowRunStatus::Succeeded,
804 Vec::new(),
805 ),
806 ],
807 };
808
809 let execution = WorkflowReplayExecutor::new(trace)
810 .run(&spec)
811 .expect("replay should run");
812
813 assert_eq!(
814 execution
815 .leaf_results
816 .iter()
817 .map(|result| result.leaf_id.as_str())
818 .collect::<Vec<_>>(),
819 vec!["patch", "generated-check"]
820 );
821 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
822 }
823}