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