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 profile: leaf.profile.clone(),
215 status: WorkflowRunStatus::ReplayDiverged,
216 usage: WorkflowUsage::default(),
217 memo_usage: WorkflowMemoUsage::default(),
218 output: None,
219 artifacts: Vec::new(),
220 schema_error: None,
221 };
222 self.resolved_outputs.insert(leaf.id.clone(), None);
223 execution.leaf_results.push(result);
224 return Ok(());
225 };
226
227 if result.status == WorkflowRunStatus::ReplayDiverged {
228 execution.mark_replay_diverged();
229 } else if result.status == WorkflowRunStatus::Failed {
230 execution.mark_failed();
231 }
232 execution.usage.add_assign(result.usage);
233 execution.memo_usage.add_assign(result.memo_usage);
234 self.resolved_outputs
235 .insert(leaf.id.clone(), result.output.clone());
236 execution.leaf_results.push(result);
237 Ok(())
238 }
239
240 fn execute_sequence(
241 &mut self,
242 spec: &WorkflowSpec,
243 sequence: &SequenceSpec,
244 execution: &mut WorkflowExecution,
245 ) -> Result<(), WorkflowReplayError> {
246 self.execute_nodes(spec, &sequence.children, execution)?;
247 self.replay_recorded_control(
248 sequence.id.as_str(),
249 ControlNodeKind::Sequence,
250 execution,
251 Some(sequence.children.iter().map(workflow_node_id).collect()),
252 Some("sequence replayed in declaration order".to_string()),
253 )
254 }
255
256 fn execute_loop_until(
257 &mut self,
258 spec: &WorkflowSpec,
259 loop_until: &LoopUntilSpec,
260 execution: &mut WorkflowExecution,
261 ) -> Result<(), WorkflowReplayError> {
262 let record = self.control_record(loop_until.id.as_str(), ControlNodeKind::LoopUntil);
263 let selected = record
264 .as_ref()
265 .map(|record| record.result.selected_children.clone())
266 .unwrap_or_else(|| loop_until.children.iter().map(workflow_node_id).collect());
267 let children = select_nodes(&loop_until.children, &selected);
268 self.execute_nodes(spec, &children, execution)?;
269 self.push_control_or_diverge(
270 loop_until.id.as_str(),
271 ControlNodeKind::LoopUntil,
272 execution,
273 record,
274 Some(selected),
275 Some("loop_until replayed recorded child selection".to_string()),
276 );
277 Ok(())
278 }
279
280 fn execute_cond(
281 &mut self,
282 spec: &WorkflowSpec,
283 cond: &CondSpec,
284 execution: &mut WorkflowExecution,
285 ) -> Result<(), WorkflowReplayError> {
286 let record = self.control_record(cond.id.as_str(), ControlNodeKind::Cond);
287 let selected = record
288 .as_ref()
289 .map(|record| record.result.selected_children.clone())
290 .unwrap_or_default();
291 let available = cond
292 .then_nodes
293 .iter()
294 .chain(cond.else_nodes.iter())
295 .cloned()
296 .collect::<Vec<_>>();
297 let nodes = select_nodes(&available, &selected);
298 self.execute_nodes(spec, &nodes, execution)?;
299 self.push_control_or_diverge(
300 cond.id.as_str(),
301 ControlNodeKind::Cond,
302 execution,
303 record,
304 Some(selected),
305 Some("cond replayed recorded branch selection".to_string()),
306 );
307 Ok(())
308 }
309
310 fn execute_expand(
311 &mut self,
312 spec: &WorkflowSpec,
313 expand: &ExpandSpec,
314 execution: &mut WorkflowExecution,
315 ) -> Result<(), WorkflowReplayError> {
316 let record = self.control_record(expand.id.as_str(), ControlNodeKind::Expand);
317 let generated_nodes = record
318 .as_ref()
319 .map(|record| record.generated_nodes.clone())
320 .unwrap_or_default();
321 validate_workflow_node_shapes(&generated_nodes)?;
322 self.execute_nodes(spec, &generated_nodes, execution)?;
323 let selected = record
324 .as_ref()
325 .map(|record| record.result.selected_children.clone())
326 .unwrap_or_else(|| generated_nodes.iter().map(workflow_node_id).collect());
327 self.push_control_or_diverge(
328 expand.id.as_str(),
329 ControlNodeKind::Expand,
330 execution,
331 record,
332 Some(selected),
333 Some(format!(
334 "expand replayed recorded nodes from {}",
335 expand.source
336 )),
337 );
338 Ok(())
339 }
340
341 fn replay_recorded_control(
342 &self,
343 node_id: &str,
344 kind: ControlNodeKind,
345 execution: &mut WorkflowExecution,
346 fallback_children: Option<Vec<String>>,
347 fallback_summary: Option<String>,
348 ) -> Result<(), WorkflowReplayError> {
349 let record = self.control_record(node_id, kind);
350 self.push_control_or_diverge(
351 node_id,
352 kind,
353 execution,
354 record,
355 fallback_children,
356 fallback_summary,
357 );
358 Ok(())
359 }
360
361 fn control_record(&self, node_id: &str, kind: ControlNodeKind) -> Option<ReplayControlRecord> {
362 self.control_records
363 .get(&ReplayControlKey {
364 trace_id: self.trace_id.clone(),
365 node_id: node_id.to_string(),
366 kind,
367 })
368 .cloned()
369 }
370
371 fn push_control_or_diverge(
372 &self,
373 node_id: &str,
374 kind: ControlNodeKind,
375 execution: &mut WorkflowExecution,
376 record: Option<ReplayControlRecord>,
377 fallback_children: Option<Vec<String>>,
378 fallback_summary: Option<String>,
379 ) {
380 let Some(record) = record else {
381 execution.mark_replay_diverged();
382 execution.control_node_results.push(ControlNodeResult {
383 node_id: node_id.to_string(),
384 kind,
385 status: WorkflowRunStatus::ReplayDiverged,
386 selected_children: fallback_children.unwrap_or_default(),
387 summary: fallback_summary
388 .or_else(|| Some("missing replay control record".to_string())),
389 });
390 return;
391 };
392 if record.result.status == WorkflowRunStatus::ReplayDiverged {
393 execution.mark_replay_diverged();
394 } else if record.result.status == WorkflowRunStatus::Failed {
395 execution.mark_failed();
396 }
397 execution.control_node_results.push(record.result);
398 }
399}
400
401#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
402struct ReplayLeafKey {
403 trace_id: String,
404 leaf_id: String,
405 input_hash: String,
406}
407
408#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
409struct ReplayControlKey {
410 trace_id: String,
411 node_id: String,
412 kind: ControlNodeKind,
413}
414
415#[derive(Debug, Clone, PartialEq, Eq, Error)]
416pub enum WorkflowReplayError {
417 #[error(transparent)]
418 Validation(#[from] WorkflowExecutionError),
419 #[error("live replay requested for leaf `{leaf}`, but no live replay provider is configured")]
420 LiveReplayUnavailable { leaf: String },
421 #[error("failed to compute replay input hash: {reason}")]
422 InputHash { reason: String },
423}
424
425pub fn compute_leaf_input_hash(
426 spec: &WorkflowSpec,
427 leaf: &LeafSpec,
428 resolved_inputs: &BTreeMap<String, Option<String>>,
429) -> Result<String, WorkflowReplayError> {
430 let input = ReplayLeafInput {
431 workflow_id: spec.id.as_deref(),
432 workflow_goal: spec.goal.as_str(),
433 leaf,
434 resolved_inputs,
435 };
436 let bytes = serde_json::to_vec(&input).map_err(|error| WorkflowReplayError::InputHash {
437 reason: error.to_string(),
438 })?;
439 let digest = Sha256::digest(bytes);
440 Ok(hex_bytes(digest))
441}
442
443fn hex_bytes(bytes: impl AsRef<[u8]>) -> String {
444 let bytes = bytes.as_ref();
445 let mut out = String::with_capacity(bytes.len() * 2);
446 for byte in bytes {
447 use std::fmt::Write as _;
448 let _ = write!(&mut out, "{byte:02x}");
449 }
450 out
451}
452
453#[derive(Serialize)]
454struct ReplayLeafInput<'a> {
455 workflow_id: Option<&'a str>,
456 workflow_goal: &'a str,
457 leaf: &'a LeafSpec,
458 resolved_inputs: &'a BTreeMap<String, Option<String>>,
459}
460
461fn resolved_inputs_for_leaf(
462 leaf: &LeafSpec,
463 resolved_outputs: &BTreeMap<String, Option<String>>,
464) -> BTreeMap<String, Option<String>> {
465 leaf.depends_on_results
466 .iter()
467 .map(|dependency| {
468 (
469 dependency.clone(),
470 resolved_outputs.get(dependency).cloned().unwrap_or(None),
471 )
472 })
473 .collect()
474}
475
476fn branch_status(results: &[LeafResult]) -> WorkflowRunStatus {
477 if results
478 .iter()
479 .any(|result| result.status == WorkflowRunStatus::ReplayDiverged)
480 {
481 WorkflowRunStatus::ReplayDiverged
482 } else if results
483 .iter()
484 .any(|result| result.status != WorkflowRunStatus::Succeeded)
485 {
486 WorkflowRunStatus::Failed
487 } else {
488 WorkflowRunStatus::Succeeded
489 }
490}
491
492fn select_nodes(nodes: &[WorkflowNode], selected: &[String]) -> Vec<WorkflowNode> {
493 let by_id: BTreeMap<_, _> = nodes
494 .iter()
495 .map(|node| (workflow_node_id(node), node.clone()))
496 .collect();
497 selected
498 .iter()
499 .filter_map(|id| by_id.get(id).cloned())
500 .collect()
501}
502
503fn workflow_node_id(node: &WorkflowNode) -> String {
504 match node {
505 WorkflowNode::BranchSet(spec) => spec.id.clone(),
506 WorkflowNode::Leaf(spec) => spec.id.clone(),
507 WorkflowNode::Sequence(spec) => spec.id.clone(),
508 WorkflowNode::Reduce(spec) => spec.id.clone(),
509 WorkflowNode::TeacherReview(spec) => spec.id.clone(),
510 WorkflowNode::LoopUntil(spec) => spec.id.clone(),
511 WorkflowNode::Cond(spec) => spec.id.clone(),
512 WorkflowNode::Expand(spec) => spec.id.clone(),
513 }
514}
515
516#[cfg(test)]
517mod tests {
518 use super::*;
519 use crate::{
520 AgentType, BudgetSpec, CondSpec, ControlNodeKind, ControlNodeResult, ExpandSpec, LeafSpec,
521 ModelPolicy, PermissionSpec, TaskMode,
522 };
523
524 fn leaf(id: &str) -> LeafSpec {
525 LeafSpec {
526 id: id.to_string(),
527 prompt: format!("run {id}"),
528 agent_type: AgentType::General,
529 profile: None,
530 mode: TaskMode::ReadOnly,
531 isolation: crate::IsolationMode::Shared,
532 file_scope: Vec::new(),
533 depends_on_results: Vec::new(),
534 budget: BudgetSpec::default(),
535 permissions: PermissionSpec::default(),
536 model_policy: ModelPolicy::default(),
537 }
538 }
539
540 fn leaf_node(id: &str) -> WorkflowNode {
541 WorkflowNode::Leaf(leaf(id))
542 }
543
544 fn workflow(nodes: Vec<WorkflowNode>) -> WorkflowSpec {
545 WorkflowSpec {
546 id: Some("wf".to_string()),
547 goal: "replay safely".to_string(),
548 description: None,
549 budget: BudgetSpec::default(),
550 permissions: PermissionSpec::default(),
551 model_policy: ModelPolicy::default(),
552 promotion_policy: crate::PromotionPolicy::default(),
553 nodes,
554 }
555 }
556
557 fn leaf_result(id: &str, output: &str) -> LeafResult {
558 LeafResult {
559 leaf_id: id.to_string(),
560 task_id: id.to_string(),
561 profile: None,
562 status: WorkflowRunStatus::Succeeded,
563 usage: WorkflowUsage {
564 input_tokens: 10,
565 output_tokens: 5,
566 cost_microusd: 2,
567 },
568 memo_usage: WorkflowMemoUsage::default(),
569 output: Some(output.to_string()),
570 artifacts: Vec::new(),
571 schema_error: None,
572 }
573 }
574
575 fn leaf_record(spec: &WorkflowSpec, leaf: &LeafSpec, result: LeafResult) -> ReplayLeafRecord {
576 ReplayLeafRecord {
577 trace_id: "trace-1".to_string(),
578 leaf_id: leaf.id.clone(),
579 input_hash: compute_leaf_input_hash(spec, leaf, &BTreeMap::new()).unwrap(),
580 result,
581 }
582 }
583
584 fn control_record(
585 id: &str,
586 kind: ControlNodeKind,
587 status: WorkflowRunStatus,
588 selected_children: Vec<&str>,
589 ) -> ReplayControlRecord {
590 ReplayControlRecord {
591 trace_id: "trace-1".to_string(),
592 node_id: id.to_string(),
593 kind,
594 result: ControlNodeResult {
595 node_id: id.to_string(),
596 kind,
597 status,
598 selected_children: selected_children.into_iter().map(str::to_string).collect(),
599 summary: Some("recorded".to_string()),
600 },
601 generated_nodes: Vec::new(),
602 }
603 }
604
605 #[test]
606 fn replay_uses_recorded_leaf_outputs_not_live_calls() {
607 let scan = leaf("scan");
608 let spec = workflow(vec![WorkflowNode::Leaf(scan.clone())]);
609 let trace = WorkflowReplayTrace {
610 trace_id: "trace-1".to_string(),
611 leaf_records: vec![leaf_record(
612 &spec,
613 &scan,
614 leaf_result("scan", "recorded output"),
615 )],
616 control_records: Vec::new(),
617 };
618
619 let execution = WorkflowReplayExecutor::new(trace)
620 .run(&spec)
621 .expect("replay should run");
622
623 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
624 assert_eq!(
625 execution.leaf_results[0].output.as_deref(),
626 Some("recorded output")
627 );
628 assert_eq!(execution.usage.cost_microusd, 2);
629 }
630
631 #[test]
632 fn workflow_trace_can_replay_from_records() {
633 let scan = leaf("scan");
634 let summarize = leaf("summarize");
635 let spec = workflow(vec![WorkflowNode::BranchSet(BranchSpec {
636 id: "discover".to_string(),
637 description: None,
638 parallel: true,
639 budget: BudgetSpec::default(),
640 permissions: PermissionSpec::default(),
641 model_policy: ModelPolicy::default(),
642 children: vec![
643 WorkflowNode::Leaf(scan.clone()),
644 WorkflowNode::Leaf(summarize.clone()),
645 ],
646 })]);
647 let trace = WorkflowReplayTrace {
648 trace_id: "trace-1".to_string(),
649 leaf_records: vec![
650 leaf_record(&spec, &scan, leaf_result("scan", "scan ok")),
651 leaf_record(&spec, &summarize, leaf_result("summarize", "summary ok")),
652 ],
653 control_records: vec![control_record(
654 "discover",
655 ControlNodeKind::BranchSet,
656 WorkflowRunStatus::Succeeded,
657 vec!["scan", "summarize"],
658 )],
659 };
660
661 let execution = WorkflowReplayExecutor::new(trace)
662 .run(&spec)
663 .expect("replay should run");
664
665 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
666 assert_eq!(execution.leaf_results.len(), 2);
667 assert_eq!(
668 execution.branch_results[0].status,
669 WorkflowRunStatus::Succeeded
670 );
671 assert_eq!(execution.branch_results[0].usage.cost_microusd, 4);
672 assert_eq!(execution.usage.cost_microusd, 4);
673 }
674
675 #[test]
676 fn workflow_replay_diverges_on_missing_leaf_record() {
677 let spec = workflow(vec![leaf_node("scan")]);
678 let trace = WorkflowReplayTrace {
679 trace_id: "trace-1".to_string(),
680 leaf_records: Vec::new(),
681 control_records: Vec::new(),
682 };
683
684 let execution = WorkflowReplayExecutor::new(trace)
685 .run(&spec)
686 .expect("missing records should be reported as divergence");
687
688 assert_eq!(execution.status, WorkflowRunStatus::ReplayDiverged);
689 assert_eq!(
690 execution.leaf_results[0].status,
691 WorkflowRunStatus::ReplayDiverged
692 );
693 assert_eq!(execution.leaf_results[0].output, None);
694 }
695
696 #[test]
697 fn live_replay_requires_explicit_opt_in() {
698 let spec = workflow(vec![leaf_node("scan")]);
699 let trace = WorkflowReplayTrace {
700 trace_id: "trace-1".to_string(),
701 leaf_records: Vec::new(),
702 control_records: Vec::new(),
703 };
704 let err = WorkflowReplayExecutor::with_options(
705 trace,
706 ReplayOptions {
707 allow_live_replay: true,
708 },
709 )
710 .run(&spec)
711 .expect_err("live replay cannot run without a configured provider");
712
713 assert!(matches!(
714 err,
715 WorkflowReplayError::LiveReplayUnavailable { .. }
716 ));
717 assert!(!ReplayOptions::default().allow_live_replay);
718 }
719
720 #[test]
721 fn leaf_input_hash_is_stable_across_object_key_order() {
722 let mut downstream = leaf("summarize");
723 downstream.depends_on_results = vec!["b".to_string(), "a".to_string()];
724 let spec = workflow(vec![WorkflowNode::Leaf(downstream.clone())]);
725 let mut left = BTreeMap::new();
726 left.insert("a".to_string(), Some("one".to_string()));
727 left.insert("b".to_string(), Some("two".to_string()));
728 let mut right = BTreeMap::new();
729 right.insert("b".to_string(), Some("two".to_string()));
730 right.insert("a".to_string(), Some("one".to_string()));
731
732 let left_hash = compute_leaf_input_hash(&spec, &downstream, &left).unwrap();
733 let right_hash = compute_leaf_input_hash(&spec, &downstream, &right).unwrap();
734
735 assert_eq!(left_hash, right_hash);
736 }
737
738 #[test]
739 fn leaf_input_hash_diverges_on_profile_change() {
740 let base = leaf("review");
741 let mut profiled = base.clone();
742 profiled.profile = Some("reviewer".to_string());
743 let spec = workflow(vec![WorkflowNode::Leaf(base.clone())]);
744
745 let base_hash = compute_leaf_input_hash(&spec, &base, &BTreeMap::new()).unwrap();
746 let profiled_hash = compute_leaf_input_hash(&spec, &profiled, &BTreeMap::new()).unwrap();
747
748 assert_ne!(base_hash, profiled_hash);
749 }
750
751 #[test]
752 fn replay_control_records_drive_cond_expand_loop_until() {
753 let patch = leaf("patch");
754 let generated = leaf("generated-check");
755 let spec = workflow(vec![
756 WorkflowNode::Cond(CondSpec {
757 id: "choose".to_string(),
758 condition: "patch?".to_string(),
759 then_nodes: vec![WorkflowNode::Leaf(patch.clone())],
760 else_nodes: vec![leaf_node("report")],
761 }),
762 WorkflowNode::Expand(ExpandSpec {
763 id: "split".to_string(),
764 source: "choose".to_string(),
765 max_children: None,
766 template: None,
767 }),
768 WorkflowNode::LoopUntil(crate::LoopUntilSpec {
769 id: "verify".to_string(),
770 condition: "done".to_string(),
771 max_iterations: Some(3),
772 children: vec![leaf_node("unused-live-child")],
773 }),
774 ]);
775 let mut expand_record = control_record(
776 "split",
777 ControlNodeKind::Expand,
778 WorkflowRunStatus::Succeeded,
779 vec!["generated-check"],
780 );
781 expand_record.generated_nodes = vec![WorkflowNode::Leaf(generated.clone())];
782 let trace = WorkflowReplayTrace {
783 trace_id: "trace-1".to_string(),
784 leaf_records: vec![
785 leaf_record(&spec, &patch, leaf_result("patch", "patched")),
786 leaf_record(&spec, &generated, leaf_result("generated-check", "checked")),
787 ],
788 control_records: vec![
789 control_record(
790 "choose",
791 ControlNodeKind::Cond,
792 WorkflowRunStatus::Succeeded,
793 vec!["patch"],
794 ),
795 expand_record,
796 control_record(
797 "verify",
798 ControlNodeKind::LoopUntil,
799 WorkflowRunStatus::Succeeded,
800 Vec::new(),
801 ),
802 ],
803 };
804
805 let execution = WorkflowReplayExecutor::new(trace)
806 .run(&spec)
807 .expect("replay should run");
808
809 assert_eq!(
810 execution
811 .leaf_results
812 .iter()
813 .map(|result| result.leaf_id.as_str())
814 .collect::<Vec<_>>(),
815 vec!["patch", "generated-check"]
816 );
817 assert_eq!(execution.status, WorkflowRunStatus::Succeeded);
818 }
819}