1use crate::dag::{DagError, NodeId, WorkflowDag};
12use serde::{Deserialize, Serialize};
13use std::collections::{HashMap, HashSet};
14use std::time::Duration;
15
16#[derive(Debug, Clone)]
18pub struct SimulationConfig {
19 pub simulated_outputs: HashMap<String, HashMap<String, serde_json::Value>>,
22 pub failing_task_types: HashSet<String>,
24 pub estimated_durations: HashMap<String, Duration>,
26 pub continue_on_failure: bool,
28 pub max_nodes: usize,
30 pub condition_overrides: HashMap<String, bool>,
33}
34
35impl Default for SimulationConfig {
36 fn default() -> Self {
37 Self {
38 simulated_outputs: HashMap::new(),
39 failing_task_types: HashSet::new(),
40 estimated_durations: HashMap::new(),
41 continue_on_failure: true,
42 max_nodes: 10_000,
43 condition_overrides: HashMap::new(),
44 }
45 }
46}
47
48impl SimulationConfig {
49 #[must_use]
51 pub fn new() -> Self {
52 Self::default()
53 }
54
55 #[must_use]
57 pub fn with_outputs(
58 mut self,
59 task_type: impl Into<String>,
60 outputs: HashMap<String, serde_json::Value>,
61 ) -> Self {
62 self.simulated_outputs.insert(task_type.into(), outputs);
63 self
64 }
65
66 #[must_use]
68 pub fn with_failing_type(mut self, task_type: impl Into<String>) -> Self {
69 self.failing_task_types.insert(task_type.into());
70 self
71 }
72
73 #[must_use]
75 pub fn with_duration(mut self, task_type: impl Into<String>, duration: Duration) -> Self {
76 self.estimated_durations.insert(task_type.into(), duration);
77 self
78 }
79
80 #[must_use]
82 pub fn with_continue_on_failure(mut self, cont: bool) -> Self {
83 self.continue_on_failure = cont;
84 self
85 }
86
87 #[must_use]
89 pub fn with_condition_override(mut self, condition: impl Into<String>, value: bool) -> Self {
90 self.condition_overrides.insert(condition.into(), value);
91 self
92 }
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize)]
97pub struct SimulationTraceEntry {
98 pub node_id: NodeId,
100 pub task_type: String,
102 pub outcome: SimulatedOutcome,
104 pub estimated_duration: Option<Duration>,
106 pub dependencies: Vec<NodeId>,
108 pub evaluated_conditions: Vec<EvaluatedCondition>,
110 pub visit_order: usize,
112}
113
114#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116pub enum SimulatedOutcome {
117 WouldSucceed,
119 WouldFail {
121 reason: String,
123 },
124 WouldSkip {
126 reason: String,
128 },
129 Unreachable,
131 Blocked {
133 failed_dependency: NodeId,
135 },
136}
137
138#[derive(Debug, Clone, Serialize, Deserialize)]
140pub struct EvaluatedCondition {
141 pub condition: String,
143 pub result: bool,
145 pub source: ConditionSource,
147}
148
149#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
151pub enum ConditionSource {
152 Override,
154 DefaultTrue,
156}
157
158#[derive(Debug, Clone, Serialize, Deserialize)]
160pub struct SimulationResult {
161 pub trace: Vec<SimulationTraceEntry>,
163 pub summary: SimulationSummary,
165 pub node_statuses: HashMap<NodeId, SimulatedOutcome>,
167}
168
169#[derive(Debug, Clone, Serialize, Deserialize)]
171pub struct SimulationSummary {
172 pub total_nodes: usize,
174 pub would_succeed: usize,
176 pub would_fail: usize,
178 pub would_skip: usize,
180 pub unreachable: usize,
182 pub blocked: usize,
184 pub estimated_total_duration: Duration,
186 pub estimated_critical_path_duration: Duration,
188 pub would_complete: bool,
190}
191
192#[derive(Debug)]
194pub struct WorkflowSimulator {
195 config: SimulationConfig,
196}
197
198impl WorkflowSimulator {
199 #[must_use]
201 pub fn new(config: SimulationConfig) -> Self {
202 Self { config }
203 }
204
205 #[must_use]
207 pub fn default_simulator() -> Self {
208 Self::new(SimulationConfig::default())
209 }
210
211 pub fn simulate(&self, dag: &WorkflowDag) -> Result<SimulationResult, DagError> {
217 let order = dag.topological_sort()?;
218
219 let mut trace = Vec::new();
220 let mut node_statuses: HashMap<NodeId, SimulatedOutcome> = HashMap::new();
221 let mut failed_nodes: HashSet<NodeId> = HashSet::new();
222 let mut node_outputs: HashMap<NodeId, HashMap<String, serde_json::Value>> = HashMap::new();
223 let mut visit_order = 0usize;
224
225 for &node_id in &order {
226 let node = match dag.nodes.get(&node_id) {
227 Some(n) => n,
228 None => continue,
229 };
230
231 let deps = dag.predecessors(node_id);
233 let blocked_by = if !self.config.continue_on_failure {
234 deps.iter().find(|d| failed_nodes.contains(d)).copied()
235 } else {
236 None
237 };
238
239 let evaluated_conditions = self.evaluate_incoming_conditions(dag, node_id);
241
242 let conditions_met = evaluated_conditions.iter().all(|ec| ec.result);
244
245 let outcome = if let Some(failed_dep) = blocked_by {
246 SimulatedOutcome::Blocked {
247 failed_dependency: failed_dep,
248 }
249 } else if !conditions_met {
250 let failed_conds: Vec<&str> = evaluated_conditions
251 .iter()
252 .filter(|ec| !ec.result)
253 .map(|ec| ec.condition.as_str())
254 .collect();
255 SimulatedOutcome::WouldSkip {
256 reason: format!("conditions not met: {}", failed_conds.join(", ")),
257 }
258 } else if self.config.failing_task_types.contains(&node.task_type) {
259 failed_nodes.insert(node_id);
260 SimulatedOutcome::WouldFail {
261 reason: format!("task type '{}' configured to fail", node.task_type),
262 }
263 } else {
264 if let Some(outputs) = self.config.simulated_outputs.get(&node.task_type) {
266 node_outputs.insert(node_id, outputs.clone());
267 }
268 SimulatedOutcome::WouldSucceed
269 };
270
271 let estimated_duration = self
272 .config
273 .estimated_durations
274 .get(&node.task_type)
275 .copied();
276
277 let entry = SimulationTraceEntry {
278 node_id,
279 task_type: node.task_type.clone(),
280 outcome: outcome.clone(),
281 estimated_duration,
282 dependencies: deps,
283 evaluated_conditions,
284 visit_order,
285 };
286
287 node_statuses.insert(node_id, outcome);
288 trace.push(entry);
289 visit_order += 1;
290 }
291
292 let visited: HashSet<NodeId> = node_statuses.keys().copied().collect();
294 for &node_id in dag.nodes.keys() {
295 if !visited.contains(&node_id) {
296 node_statuses.insert(node_id, SimulatedOutcome::Unreachable);
297 }
298 }
299
300 let summary = self.compute_summary(dag, &node_statuses, &trace);
301
302 Ok(SimulationResult {
303 trace,
304 summary,
305 node_statuses,
306 })
307 }
308
309 fn evaluate_incoming_conditions(
311 &self,
312 dag: &WorkflowDag,
313 node_id: NodeId,
314 ) -> Vec<EvaluatedCondition> {
315 let mut results = Vec::new();
316
317 for edge in &dag.edges {
318 if edge.to_node != node_id {
319 continue;
320 }
321 if let Some(ref condition) = edge.condition {
322 let (result, source) =
323 if let Some(&override_val) = self.config.condition_overrides.get(condition) {
324 (override_val, ConditionSource::Override)
325 } else {
326 (true, ConditionSource::DefaultTrue)
327 };
328
329 results.push(EvaluatedCondition {
330 condition: condition.clone(),
331 result,
332 source,
333 });
334 }
335 }
336
337 results
338 }
339
340 fn compute_summary(
342 &self,
343 dag: &WorkflowDag,
344 statuses: &HashMap<NodeId, SimulatedOutcome>,
345 trace: &[SimulationTraceEntry],
346 ) -> SimulationSummary {
347 let total_nodes = dag.nodes.len();
348 let mut would_succeed = 0usize;
349 let mut would_fail = 0usize;
350 let mut would_skip = 0usize;
351 let mut unreachable = 0usize;
352 let mut blocked = 0usize;
353 let mut total_duration = Duration::ZERO;
354
355 for outcome in statuses.values() {
356 match outcome {
357 SimulatedOutcome::WouldSucceed => would_succeed += 1,
358 SimulatedOutcome::WouldFail { .. } => would_fail += 1,
359 SimulatedOutcome::WouldSkip { .. } => would_skip += 1,
360 SimulatedOutcome::Unreachable => unreachable += 1,
361 SimulatedOutcome::Blocked { .. } => blocked += 1,
362 }
363 }
364
365 for entry in trace {
366 if entry.outcome == SimulatedOutcome::WouldSucceed {
367 if let Some(d) = entry.estimated_duration {
368 total_duration += d;
369 }
370 }
371 }
372
373 let critical_path = self.compute_critical_path_duration(dag, trace);
374
375 SimulationSummary {
376 total_nodes,
377 would_succeed,
378 would_fail,
379 would_skip,
380 unreachable,
381 blocked,
382 estimated_total_duration: total_duration,
383 estimated_critical_path_duration: critical_path,
384 would_complete: would_fail == 0 && blocked == 0,
385 }
386 }
387
388 fn compute_critical_path_duration(
390 &self,
391 dag: &WorkflowDag,
392 trace: &[SimulationTraceEntry],
393 ) -> Duration {
394 let duration_map: HashMap<NodeId, Duration> = trace
396 .iter()
397 .filter(|e| e.outcome == SimulatedOutcome::WouldSucceed)
398 .filter_map(|e| e.estimated_duration.map(|d| (e.node_id, d)))
399 .collect();
400
401 if duration_map.is_empty() {
402 return Duration::ZERO;
403 }
404
405 let order = match dag.topological_sort() {
407 Ok(o) => o,
408 Err(_) => return Duration::ZERO,
409 };
410
411 let mut dist: HashMap<NodeId, Duration> = HashMap::new();
412
413 for &node_id in &order {
414 let self_dur = duration_map
415 .get(&node_id)
416 .copied()
417 .unwrap_or(Duration::ZERO);
418 let max_pred = dag
419 .predecessors(node_id)
420 .iter()
421 .filter_map(|p| dist.get(p))
422 .max()
423 .copied()
424 .unwrap_or(Duration::ZERO);
425
426 dist.insert(node_id, max_pred + self_dur);
427 }
428
429 dist.values().max().copied().unwrap_or(Duration::ZERO)
430 }
431}
432
433pub fn quick_simulate(dag: &WorkflowDag) -> Result<bool, DagError> {
441 let sim = WorkflowSimulator::default_simulator();
442 let result = sim.simulate(dag)?;
443 Ok(result.summary.would_complete)
444}
445
446#[cfg(test)]
447mod tests {
448 use super::*;
449 use crate::dag::{WorkflowEdge, WorkflowNode};
450
451 fn make_node(task_type: &str) -> WorkflowNode {
452 WorkflowNode::new(task_type)
453 }
454
455 fn make_linear_dag() -> (WorkflowDag, NodeId, NodeId, NodeId) {
456 let mut dag = WorkflowDag::new();
457 let a = dag.add_node(make_node("ingest")).expect("add node");
458 let b = dag.add_node(make_node("transcode")).expect("add node");
459 let c = dag.add_node(make_node("deliver")).expect("add node");
460 dag.add_edge(WorkflowEdge::new(a, b, "raw"))
461 .expect("add edge");
462 dag.add_edge(WorkflowEdge::new(b, c, "encoded"))
463 .expect("add edge");
464 (dag, a, b, c)
465 }
466
467 #[test]
470 fn test_simulate_all_succeed() {
471 let (dag, _, _, _) = make_linear_dag();
472 let sim = WorkflowSimulator::default_simulator();
473 let result = sim.simulate(&dag).expect("simulate");
474
475 assert!(result.summary.would_complete);
476 assert_eq!(result.summary.would_succeed, 3);
477 assert_eq!(result.summary.would_fail, 0);
478 assert_eq!(result.trace.len(), 3);
479 }
480
481 #[test]
482 fn test_simulate_with_failing_task() {
483 let (dag, _, _, _) = make_linear_dag();
484 let config = SimulationConfig::new().with_failing_type("transcode");
485 let sim = WorkflowSimulator::new(config);
486 let result = sim.simulate(&dag).expect("simulate");
487
488 assert!(!result.summary.would_complete);
489 assert_eq!(result.summary.would_fail, 1);
490 assert_eq!(result.summary.would_succeed, 2); }
492
493 #[test]
494 fn test_simulate_blocked_on_failure() {
495 let (dag, _, _, _) = make_linear_dag();
496 let config = SimulationConfig::new()
497 .with_failing_type("transcode")
498 .with_continue_on_failure(false);
499 let sim = WorkflowSimulator::new(config);
500 let result = sim.simulate(&dag).expect("simulate");
501
502 assert!(!result.summary.would_complete);
503 assert_eq!(result.summary.blocked, 1); assert_eq!(result.summary.would_fail, 1);
505 assert_eq!(result.summary.would_succeed, 1); }
507
508 #[test]
511 fn test_simulate_with_conditions() {
512 let mut dag = WorkflowDag::new();
513 let a = dag.add_node(make_node("check")).expect("add node");
514 let b = dag.add_node(make_node("high_res")).expect("add node");
515 let c = dag.add_node(make_node("low_res")).expect("add node");
516
517 dag.add_edge(WorkflowEdge::with_condition(
518 a,
519 b,
520 "video",
521 "resolution == 4k",
522 ))
523 .expect("add edge");
524 dag.add_edge(WorkflowEdge::with_condition(
525 a,
526 c,
527 "video",
528 "resolution != 4k",
529 ))
530 .expect("add edge");
531
532 let config = SimulationConfig::new()
534 .with_condition_override("resolution == 4k", true)
535 .with_condition_override("resolution != 4k", false);
536 let sim = WorkflowSimulator::new(config);
537 let result = sim.simulate(&dag).expect("simulate");
538
539 assert_eq!(result.summary.would_succeed, 2); assert_eq!(result.summary.would_skip, 1); }
542
543 #[test]
544 fn test_conditions_default_true() {
545 let mut dag = WorkflowDag::new();
546 let a = dag.add_node(make_node("source")).expect("add node");
547 let b = dag.add_node(make_node("sink")).expect("add node");
548 dag.add_edge(WorkflowEdge::with_condition(a, b, "data", "some_condition"))
549 .expect("add edge");
550
551 let sim = WorkflowSimulator::default_simulator();
552 let result = sim.simulate(&dag).expect("simulate");
553
554 assert_eq!(result.summary.would_succeed, 2);
556 assert!(result.trace[1]
557 .evaluated_conditions
558 .iter()
559 .all(|ec| ec.source == ConditionSource::DefaultTrue));
560 }
561
562 #[test]
565 fn test_estimated_durations() {
566 let (dag, _, _, _) = make_linear_dag();
567 let config = SimulationConfig::new()
568 .with_duration("ingest", Duration::from_secs(10))
569 .with_duration("transcode", Duration::from_secs(60))
570 .with_duration("deliver", Duration::from_secs(5));
571 let sim = WorkflowSimulator::new(config);
572 let result = sim.simulate(&dag).expect("simulate");
573
574 assert_eq!(
575 result.summary.estimated_total_duration,
576 Duration::from_secs(75)
577 );
578 assert_eq!(
580 result.summary.estimated_critical_path_duration,
581 Duration::from_secs(75)
582 );
583 }
584
585 #[test]
586 fn test_critical_path_parallel() {
587 let mut dag = WorkflowDag::new();
588 let root = dag.add_node(make_node("start")).expect("add node");
589 let branch_a = dag.add_node(make_node("fast")).expect("add node");
590 let branch_b = dag.add_node(make_node("slow")).expect("add node");
591 let join = dag.add_node(make_node("finish")).expect("add node");
592
593 dag.add_edge(WorkflowEdge::new(root, branch_a, "x"))
594 .expect("add edge");
595 dag.add_edge(WorkflowEdge::new(root, branch_b, "x"))
596 .expect("add edge");
597 dag.add_edge(WorkflowEdge::new(branch_a, join, "x"))
598 .expect("add edge");
599 dag.add_edge(WorkflowEdge::new(branch_b, join, "x"))
600 .expect("add edge");
601
602 let config = SimulationConfig::new()
603 .with_duration("start", Duration::from_secs(5))
604 .with_duration("fast", Duration::from_secs(10))
605 .with_duration("slow", Duration::from_secs(30))
606 .with_duration("finish", Duration::from_secs(5));
607 let sim = WorkflowSimulator::new(config);
608 let result = sim.simulate(&dag).expect("simulate");
609
610 assert_eq!(
612 result.summary.estimated_total_duration,
613 Duration::from_secs(50)
614 );
615 assert_eq!(
617 result.summary.estimated_critical_path_duration,
618 Duration::from_secs(40)
619 );
620 }
621
622 #[test]
625 fn test_trace_visit_order() {
626 let (dag, _, _, _) = make_linear_dag();
627 let sim = WorkflowSimulator::default_simulator();
628 let result = sim.simulate(&dag).expect("simulate");
629
630 for (i, entry) in result.trace.iter().enumerate() {
631 assert_eq!(entry.visit_order, i);
632 }
633 }
634
635 #[test]
636 fn test_trace_dependencies_recorded() {
637 let (dag, a, _, _) = make_linear_dag();
638 let sim = WorkflowSimulator::default_simulator();
639 let result = sim.simulate(&dag).expect("simulate");
640
641 let transcode_entry = result
643 .trace
644 .iter()
645 .find(|e| e.task_type == "transcode")
646 .expect("find transcode");
647 assert_eq!(transcode_entry.dependencies.len(), 1);
648 assert_eq!(transcode_entry.dependencies[0], a);
649 }
650
651 #[test]
654 fn test_simulate_empty_dag() {
655 let dag = WorkflowDag::new();
656 let sim = WorkflowSimulator::default_simulator();
657 let result = sim.simulate(&dag).expect("simulate");
658
659 assert!(result.summary.would_complete);
660 assert_eq!(result.summary.total_nodes, 0);
661 assert!(result.trace.is_empty());
662 }
663
664 #[test]
665 fn test_simulate_single_node() {
666 let mut dag = WorkflowDag::new();
667 dag.add_node(make_node("solo")).expect("add node");
668
669 let sim = WorkflowSimulator::default_simulator();
670 let result = sim.simulate(&dag).expect("simulate");
671
672 assert!(result.summary.would_complete);
673 assert_eq!(result.summary.would_succeed, 1);
674 }
675
676 #[test]
677 fn test_quick_simulate_success() {
678 let (dag, _, _, _) = make_linear_dag();
679 assert!(quick_simulate(&dag).expect("simulate"));
680 }
681
682 #[test]
683 fn test_simulate_with_outputs() {
684 let (dag, _, _, _) = make_linear_dag();
685 let mut outputs = HashMap::new();
686 outputs.insert(
687 "path".to_string(),
688 serde_json::json!(std::env::temp_dir()
689 .join("oximedia-workflow-sim-out.mp4")
690 .to_string_lossy()),
691 );
692 let config = SimulationConfig::new().with_outputs("ingest", outputs);
693 let sim = WorkflowSimulator::new(config);
694 let result = sim.simulate(&dag).expect("simulate");
695
696 assert!(result.summary.would_complete);
697 }
698
699 #[test]
700 fn test_simulate_diamond_dag() {
701 let mut dag = WorkflowDag::new();
703 let root = dag.add_node(make_node("root")).expect("add node");
704 let node_a = dag.add_node(make_node("branch_a")).expect("add node");
705 let node_b = dag.add_node(make_node("branch_b")).expect("add node");
706 let join = dag.add_node(make_node("join")).expect("add node");
707
708 dag.add_edge(WorkflowEdge::new(root, node_a, "x"))
709 .expect("add edge");
710 dag.add_edge(WorkflowEdge::new(root, node_b, "x"))
711 .expect("add edge");
712 dag.add_edge(WorkflowEdge::new(node_a, join, "x"))
713 .expect("add edge");
714 dag.add_edge(WorkflowEdge::new(node_b, join, "x"))
715 .expect("add edge");
716
717 let sim = WorkflowSimulator::default_simulator();
718 let result = sim.simulate(&dag).expect("simulate");
719
720 assert!(result.summary.would_complete);
721 assert_eq!(result.summary.would_succeed, 4);
722 }
723
724 #[test]
725 fn test_node_statuses_map() {
726 let (dag, a, b, c) = make_linear_dag();
727 let sim = WorkflowSimulator::default_simulator();
728 let result = sim.simulate(&dag).expect("simulate");
729
730 assert_eq!(result.node_statuses.len(), 3);
731 assert_eq!(
732 result.node_statuses.get(&a),
733 Some(&SimulatedOutcome::WouldSucceed)
734 );
735 assert_eq!(
736 result.node_statuses.get(&b),
737 Some(&SimulatedOutcome::WouldSucceed)
738 );
739 assert_eq!(
740 result.node_statuses.get(&c),
741 Some(&SimulatedOutcome::WouldSucceed)
742 );
743 }
744}
745
746#[derive(Debug, Clone)]
752pub struct SimulationReport {
753 pub step_count: usize,
755 pub estimated_duration_ms: u64,
757 pub dependencies: Vec<String>,
760 pub warnings: Vec<String>,
762}
763
764impl SimulationReport {
765 #[must_use]
767 pub fn is_clean(&self) -> bool {
768 self.warnings.is_empty()
769 }
770}
771
772pub struct WorkflowDryRun;
776
777impl WorkflowDryRun {
778 #[must_use]
786 pub fn simulate(workflow: &crate::workflow::Workflow) -> SimulationReport {
787 let mut warnings = Vec::new();
788 let mut dependencies = Vec::new();
789
790 let tasks: Vec<&crate::task::Task> = workflow.tasks().collect();
791 let step_count = tasks.len();
792
793 if step_count == 0 {
794 warnings.push("Workflow has no tasks".to_string());
795 return SimulationReport {
796 step_count: 0,
797 estimated_duration_ms: 0,
798 dependencies,
799 warnings,
800 };
801 }
802
803 let mut estimated_duration_ms: u64 = 0;
805
806 for task in &tasks {
807 let dur = Self::estimate_task_duration_ms(task);
808 estimated_duration_ms = estimated_duration_ms.saturating_add(dur);
809
810 for dep_id in &task.dependencies {
812 let dep_name = workflow
814 .get_task(dep_id)
815 .map(|t| t.name.as_str())
816 .unwrap_or("<unknown>");
817 dependencies.push(format!(
818 "'{}' [{}] depends on '{}' [{}]",
819 task.name, task.id, dep_name, dep_id
820 ));
821 }
822 }
823
824 for edge in &workflow.edges {
826 let from_name = workflow
827 .get_task(&edge.from)
828 .map(|t| t.name.as_str())
829 .unwrap_or("<unknown>");
830 let to_name = workflow
831 .get_task(&edge.to)
832 .map(|t| t.name.as_str())
833 .unwrap_or("<unknown>");
834 let dep_str = if let Some(ref cond) = edge.condition {
835 format!(
836 "edge: '{}' [{}] -> '{}' [{}] (condition: {cond})",
837 from_name, edge.from, to_name, edge.to
838 )
839 } else {
840 format!(
841 "edge: '{}' [{}] -> '{}' [{}]",
842 from_name, edge.from, to_name, edge.to
843 )
844 };
845 if !dependencies.contains(&dep_str) {
847 dependencies.push(dep_str);
848 }
849 }
850
851 let cycle_warnings = Self::check_cycles(workflow);
853 warnings.extend(cycle_warnings);
854
855 for task in &tasks {
857 if task.timeout == std::time::Duration::from_secs(3600) {
858 }
860 for cond in &task.conditions {
862 if cond.trim().is_empty() {
863 warnings.push(format!(
864 "Task '{}' has an empty condition string",
865 task.name
866 ));
867 }
868 }
869 }
870
871 SimulationReport {
872 step_count,
873 estimated_duration_ms,
874 dependencies,
875 warnings,
876 }
877 }
878
879 fn check_cycles(workflow: &crate::workflow::Workflow) -> Vec<String> {
881 let mut adj: std::collections::HashMap<crate::task::TaskId, Vec<crate::task::TaskId>> =
883 std::collections::HashMap::new();
884
885 for task in workflow.tasks() {
886 adj.insert(task.id, task.dependencies.clone());
887 }
888
889 for edge in &workflow.edges {
891 adj.entry(edge.to).or_default().push(edge.from);
892 }
893
894 let mut warnings = Vec::new();
895 let mut color: std::collections::HashMap<crate::task::TaskId, u8> =
896 std::collections::HashMap::new(); for &start in adj.keys() {
899 if color.get(&start).copied().unwrap_or(0) == 0 {
900 let mut path = Vec::new();
901 if Self::dfs_cycle(start, &adj, &mut color, &mut path) {
902 warnings.push(format!("Potential cycle detected near task [{}]", start));
903 }
904 }
905 }
906
907 warnings
908 }
909
910 fn dfs_cycle(
911 node: crate::task::TaskId,
912 adj: &std::collections::HashMap<crate::task::TaskId, Vec<crate::task::TaskId>>,
913 color: &mut std::collections::HashMap<crate::task::TaskId, u8>,
914 path: &mut Vec<crate::task::TaskId>,
915 ) -> bool {
916 color.insert(node, 1); path.push(node);
918
919 if let Some(neighbors) = adj.get(&node) {
920 for &next in neighbors {
921 let c = color.get(&next).copied().unwrap_or(0);
922 if c == 1 {
923 return true; }
925 if c == 0 && Self::dfs_cycle(next, adj, color, path) {
926 return true;
927 }
928 }
929 }
930
931 path.pop();
932 color.insert(node, 2); false
934 }
935
936 #[must_use]
947 fn estimate_task_duration_ms(task: &crate::task::Task) -> u64 {
948 match &task.task_type {
949 crate::task::TaskType::Wait { duration } => {
950 duration.as_millis().try_into().unwrap_or(u64::MAX)
952 }
953 crate::task::TaskType::Transcode { .. } => 60_000,
954 crate::task::TaskType::QualityControl { .. } => 30_000,
955 crate::task::TaskType::Transfer { .. } => 20_000,
956 crate::task::TaskType::Analysis { .. } => 15_000,
957 crate::task::TaskType::Notification { .. } => 1_000,
958 crate::task::TaskType::HttpRequest { .. } => 1_000,
959 crate::task::TaskType::CustomScript { .. } => 5_000,
960 crate::task::TaskType::Conditional { .. } => 5_000,
961 }
962 }
963}
964
965#[cfg(test)]
970mod dry_run_tests {
971 use super::*;
972 use crate::task::{Task, TaskType};
973 use crate::workflow::Workflow;
974 use std::path::PathBuf;
975 use std::time::Duration;
976
977 fn make_transcode_task(name: &str) -> Task {
978 Task::new(
979 name,
980 TaskType::Transcode {
981 input: PathBuf::from("/in/video.mp4"),
982 output: PathBuf::from("/out/video.mp4"),
983 preset: "h264".to_string(),
984 params: std::collections::HashMap::new(),
985 },
986 )
987 }
988
989 fn make_wait_task(name: &str, secs: u64) -> Task {
990 Task::new(
991 name,
992 TaskType::Wait {
993 duration: Duration::from_secs(secs),
994 },
995 )
996 }
997
998 fn make_notify_task(name: &str) -> Task {
999 Task::new(
1000 name,
1001 TaskType::Notification {
1002 channel: crate::task::NotificationChannel::Webhook {
1003 url: "https://example.com/hook".to_string(),
1004 },
1005 message: "done".to_string(),
1006 metadata: std::collections::HashMap::new(),
1007 },
1008 )
1009 }
1010
1011 #[test]
1014 fn test_dry_run_empty_workflow() {
1015 let wf = Workflow::new("empty");
1016 let report = WorkflowDryRun::simulate(&wf);
1017 assert_eq!(report.step_count, 0);
1018 assert_eq!(report.estimated_duration_ms, 0);
1019 assert!(!report.warnings.is_empty(), "should warn about no tasks");
1020 assert!(
1021 report.warnings.iter().any(|w| w.contains("no tasks")),
1022 "warning: {:?}",
1023 report.warnings
1024 );
1025 }
1026
1027 #[test]
1030 fn test_dry_run_single_task_step_count() {
1031 let mut wf = Workflow::new("single");
1032 wf.add_task(make_transcode_task("encode"));
1033 let report = WorkflowDryRun::simulate(&wf);
1034 assert_eq!(report.step_count, 1);
1035 }
1036
1037 #[test]
1038 fn test_dry_run_single_task_no_deps() {
1039 let mut wf = Workflow::new("single");
1040 wf.add_task(make_transcode_task("encode"));
1041 let report = WorkflowDryRun::simulate(&wf);
1042 assert!(report.dependencies.is_empty());
1043 }
1044
1045 #[test]
1048 fn test_dry_run_multiple_tasks_step_count() {
1049 let mut wf = Workflow::new("multi");
1050 wf.add_task(make_transcode_task("encode"));
1051 wf.add_task(make_notify_task("notify"));
1052 wf.add_task(make_wait_task("pause", 5));
1053 let report = WorkflowDryRun::simulate(&wf);
1054 assert_eq!(report.step_count, 3);
1055 }
1056
1057 #[test]
1060 fn test_dry_run_wait_task_uses_actual_duration() {
1061 let mut wf = Workflow::new("wait-wf");
1062 wf.add_task(make_wait_task("sleep", 10)); let report = WorkflowDryRun::simulate(&wf);
1064 assert_eq!(report.estimated_duration_ms, 10_000);
1065 }
1066
1067 #[test]
1070 fn test_dry_run_estimated_duration_sum() {
1071 let mut wf = Workflow::new("sum-wf");
1072 wf.add_task(make_transcode_task("encode")); wf.add_task(make_notify_task("notify")); let report = WorkflowDryRun::simulate(&wf);
1075 assert_eq!(report.estimated_duration_ms, 61_000);
1076 }
1077
1078 #[test]
1081 fn test_dry_run_dependencies_listed_via_task_deps() {
1082 let mut wf = Workflow::new("dep-wf");
1083 let encode_id = wf.add_task(make_transcode_task("encode"));
1084 let mut notify = make_notify_task("notify");
1085 notify.add_dependency(encode_id);
1086 wf.add_task(notify);
1087
1088 let report = WorkflowDryRun::simulate(&wf);
1089 assert!(!report.dependencies.is_empty(), "should record dependency");
1090 assert!(
1091 report
1092 .dependencies
1093 .iter()
1094 .any(|d| d.contains("notify") && d.contains("encode")),
1095 "deps: {:?}",
1096 report.dependencies
1097 );
1098 }
1099
1100 #[test]
1103 fn test_dry_run_dependencies_listed_via_edges() {
1104 let mut wf = Workflow::new("edge-wf");
1105 let encode_id = wf.add_task(make_transcode_task("encode"));
1106 let notify_id = wf.add_task(make_notify_task("notify"));
1107 wf.add_edge(encode_id, notify_id).expect("add edge");
1108
1109 let report = WorkflowDryRun::simulate(&wf);
1110 assert!(
1111 report
1112 .dependencies
1113 .iter()
1114 .any(|d| d.contains("encode") && d.contains("notify")),
1115 "deps: {:?}",
1116 report.dependencies
1117 );
1118 }
1119
1120 #[test]
1123 fn test_dry_run_warnings_for_empty_workflow() {
1124 let wf = Workflow::new("empty");
1125 let report = WorkflowDryRun::simulate(&wf);
1126 assert!(!report.warnings.is_empty());
1127 }
1128
1129 #[test]
1130 fn test_dry_run_no_warnings_for_simple_valid_workflow() {
1131 let mut wf = Workflow::new("clean");
1132 wf.add_task(make_transcode_task("encode"));
1133 let report = WorkflowDryRun::simulate(&wf);
1134 assert!(
1136 report.warnings.is_empty(),
1137 "unexpected warnings: {:?}",
1138 report.warnings
1139 );
1140 }
1141
1142 #[test]
1145 fn test_dry_run_simulation_report_all_fields_accessible() {
1146 let mut wf = Workflow::new("fields-test");
1147 wf.add_task(make_transcode_task("encode"));
1148 let report = WorkflowDryRun::simulate(&wf);
1149 let _ = report.step_count;
1151 let _ = report.estimated_duration_ms;
1152 let _ = report.dependencies;
1153 let _ = report.warnings;
1154 assert!(report.is_clean());
1155 }
1156
1157 #[test]
1160 fn test_dry_run_multiple_dependencies() {
1161 let mut wf = Workflow::new("multi-dep");
1162 let a = wf.add_task(make_transcode_task("encode-a"));
1163 let b = wf.add_task(make_transcode_task("encode-b"));
1164 let notify_id = wf.add_task(make_notify_task("notify"));
1165
1166 wf.add_edge(a, notify_id).expect("edge a->notify");
1167 wf.add_edge(b, notify_id).expect("edge b->notify");
1168
1169 let report = WorkflowDryRun::simulate(&wf);
1170 let edge_deps: Vec<_> = report
1172 .dependencies
1173 .iter()
1174 .filter(|d| d.contains("notify"))
1175 .collect();
1176 assert!(
1177 edge_deps.len() >= 2,
1178 "expected 2 deps to notify, got: {:?}",
1179 report.dependencies
1180 );
1181 }
1182}