Skip to main content

oximedia_workflow/
workflow_simulation.rs

1//! Workflow simulation (dry-run) engine.
2//!
3//! Traces the execution path of a workflow DAG without actually running any
4//! tasks. Useful for pre-flight validation, capacity planning, cost estimation,
5//! and debugging complex conditional workflows.
6//!
7//! The simulator walks the DAG in topological order, evaluates edge conditions
8//! against simulated outputs, and produces a detailed execution trace showing
9//! which nodes would run, be skipped, or potentially fail.
10
11use crate::dag::{DagError, NodeId, WorkflowDag};
12use serde::{Deserialize, Serialize};
13use std::collections::{HashMap, HashSet};
14use std::time::Duration;
15
16/// Configuration for a simulation run.
17#[derive(Debug, Clone)]
18pub struct SimulationConfig {
19    /// Simulated outputs per node task type. When a node of this task type
20    /// runs, these outputs are written to its output map.
21    pub simulated_outputs: HashMap<String, HashMap<String, serde_json::Value>>,
22    /// Task types that should be marked as failed during simulation.
23    pub failing_task_types: HashSet<String>,
24    /// Estimated duration per task type (for cost/time estimation).
25    pub estimated_durations: HashMap<String, Duration>,
26    /// Whether to continue tracing after a simulated failure.
27    pub continue_on_failure: bool,
28    /// Maximum number of nodes to trace (safety limit).
29    pub max_nodes: usize,
30    /// Condition evaluator: maps condition string to boolean result.
31    /// If not provided, all conditions default to true.
32    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    /// Create a new empty simulation config.
50    #[must_use]
51    pub fn new() -> Self {
52        Self::default()
53    }
54
55    /// Add simulated outputs for a task type.
56    #[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    /// Mark a task type as failing.
67    #[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    /// Set estimated duration for a task type.
74    #[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    /// Set whether to continue after failure.
81    #[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    /// Override a condition expression to return a specific boolean.
88    #[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/// A single entry in the simulation trace.
96#[derive(Debug, Clone, Serialize, Deserialize)]
97pub struct SimulationTraceEntry {
98    /// Node that was evaluated.
99    pub node_id: NodeId,
100    /// Task type of the node.
101    pub task_type: String,
102    /// Outcome of the simulated execution.
103    pub outcome: SimulatedOutcome,
104    /// Estimated duration (if configured).
105    pub estimated_duration: Option<Duration>,
106    /// Dependencies that were checked.
107    pub dependencies: Vec<NodeId>,
108    /// Edge conditions that were evaluated.
109    pub evaluated_conditions: Vec<EvaluatedCondition>,
110    /// Order in which this node was visited (0-based).
111    pub visit_order: usize,
112}
113
114/// Outcome of a simulated node execution.
115#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
116pub enum SimulatedOutcome {
117    /// Node would execute successfully.
118    WouldSucceed,
119    /// Node would fail.
120    WouldFail {
121        /// Reason for failure.
122        reason: String,
123    },
124    /// Node would be skipped due to unsatisfied conditions.
125    WouldSkip {
126        /// Reason for skipping.
127        reason: String,
128    },
129    /// Node not reachable from root nodes.
130    Unreachable,
131    /// Node blocked because a dependency failed and continue_on_failure is false.
132    Blocked {
133        /// The failed dependency.
134        failed_dependency: NodeId,
135    },
136}
137
138/// Result of evaluating an edge condition during simulation.
139#[derive(Debug, Clone, Serialize, Deserialize)]
140pub struct EvaluatedCondition {
141    /// The condition expression.
142    pub condition: String,
143    /// The result of evaluation.
144    pub result: bool,
145    /// Source of the result (override or default).
146    pub source: ConditionSource,
147}
148
149/// How a condition was resolved.
150#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
151pub enum ConditionSource {
152    /// Resolved from a config override.
153    Override,
154    /// Defaulted to true (no override provided).
155    DefaultTrue,
156}
157
158/// Complete result of a simulation run.
159#[derive(Debug, Clone, Serialize, Deserialize)]
160pub struct SimulationResult {
161    /// Ordered trace of all visited nodes.
162    pub trace: Vec<SimulationTraceEntry>,
163    /// Summary statistics.
164    pub summary: SimulationSummary,
165    /// Final node statuses.
166    pub node_statuses: HashMap<NodeId, SimulatedOutcome>,
167}
168
169/// Summary statistics from a simulation.
170#[derive(Debug, Clone, Serialize, Deserialize)]
171pub struct SimulationSummary {
172    /// Total nodes in the DAG.
173    pub total_nodes: usize,
174    /// Nodes that would succeed.
175    pub would_succeed: usize,
176    /// Nodes that would fail.
177    pub would_fail: usize,
178    /// Nodes that would be skipped.
179    pub would_skip: usize,
180    /// Nodes that are unreachable.
181    pub unreachable: usize,
182    /// Nodes blocked by failures.
183    pub blocked: usize,
184    /// Estimated total duration (sum of all would-succeed node durations).
185    pub estimated_total_duration: Duration,
186    /// Estimated critical path duration (max path sum).
187    pub estimated_critical_path_duration: Duration,
188    /// Whether the workflow would complete successfully.
189    pub would_complete: bool,
190}
191
192/// The workflow simulation engine.
193#[derive(Debug)]
194pub struct WorkflowSimulator {
195    config: SimulationConfig,
196}
197
198impl WorkflowSimulator {
199    /// Create a new simulator with the given config.
200    #[must_use]
201    pub fn new(config: SimulationConfig) -> Self {
202        Self { config }
203    }
204
205    /// Create a simulator with default config (all nodes succeed).
206    #[must_use]
207    pub fn default_simulator() -> Self {
208        Self::new(SimulationConfig::default())
209    }
210
211    /// Run the simulation on a DAG without modifying it.
212    ///
213    /// # Errors
214    ///
215    /// Returns `DagError` if the DAG contains a cycle.
216    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            // Check if any dependency failed
232            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            // Evaluate incoming edge conditions
240            let evaluated_conditions = self.evaluate_incoming_conditions(dag, node_id);
241
242            // Check if any required condition is false
243            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                // Would succeed: store simulated outputs
265                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        // Identify unreachable nodes
293        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    /// Evaluate incoming edge conditions for a node.
310    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    /// Compute summary statistics from the simulation results.
341    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    /// Compute the critical path duration using longest-path in DAG.
389    fn compute_critical_path_duration(
390        &self,
391        dag: &WorkflowDag,
392        trace: &[SimulationTraceEntry],
393    ) -> Duration {
394        // Build duration map from trace
395        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        // Longest path via topological order
406        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
433/// Helper to create a simple simulation for a DAG.
434///
435/// Returns a result indicating whether the workflow would complete successfully.
436///
437/// # Errors
438///
439/// Returns `DagError` if the DAG contains a cycle.
440pub 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    // --- Basic simulation ---
468
469    #[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); // ingest + deliver still succeed
491    }
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); // deliver is blocked
504        assert_eq!(result.summary.would_fail, 1);
505        assert_eq!(result.summary.would_succeed, 1); // only ingest
506    }
507
508    // --- Condition evaluation ---
509
510    #[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        // Simulate with 4k resolution
533        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); // check + high_res
540        assert_eq!(result.summary.would_skip, 1); // low_res
541    }
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        // No override -> defaults to true
555        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    // --- Duration estimation ---
563
564    #[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        // Critical path = sequential chain = 75s
579        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        // Total = 5 + 10 + 30 + 5 = 50
611        assert_eq!(
612            result.summary.estimated_total_duration,
613            Duration::from_secs(50)
614        );
615        // Critical path = start(5) -> slow(30) -> finish(5) = 40
616        assert_eq!(
617            result.summary.estimated_critical_path_duration,
618            Duration::from_secs(40)
619        );
620    }
621
622    // --- Trace details ---
623
624    #[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        // Second node (transcode) should have ingest as dependency
642        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    // --- Edge cases ---
652
653    #[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        // root -> A, root -> B, A -> join, B -> join
702        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// =============================================================================
747// WorkflowDryRun — high-level dry-run simulator for the Workflow type
748// =============================================================================
749
750/// A simple simulation report produced by dry-running a [`crate::workflow::Workflow`].
751#[derive(Debug, Clone)]
752pub struct SimulationReport {
753    /// Total number of tasks (steps) in the workflow.
754    pub step_count: usize,
755    /// Estimated total duration in milliseconds (sum of per-task estimates).
756    pub estimated_duration_ms: u64,
757    /// Descriptions of dependency relationships found
758    /// (e.g. `"task_b [<id>] depends on task_a [<id>]"`).
759    pub dependencies: Vec<String>,
760    /// Warnings detected during simulation (e.g. empty workflow, no timeout set).
761    pub warnings: Vec<String>,
762}
763
764impl SimulationReport {
765    /// Return `true` if no warnings were generated.
766    #[must_use]
767    pub fn is_clean(&self) -> bool {
768        self.warnings.is_empty()
769    }
770}
771
772/// Dry-run simulator for [`crate::workflow::Workflow`].
773///
774/// Analyses a workflow statically — no tasks are executed.
775pub struct WorkflowDryRun;
776
777impl WorkflowDryRun {
778    /// Simulate a workflow and return a [`SimulationReport`].
779    ///
780    /// Analyses:
781    /// - `step_count`: number of tasks in the workflow.
782    /// - `estimated_duration_ms`: sum of per-task duration estimates.
783    /// - `dependencies`: textual descriptions of every `Task::dependencies` relationship.
784    /// - `warnings`: issues detected (empty workflow, missing task refs, etc.).
785    #[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        // Compute estimated duration per task and build dep descriptions.
804        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            // Record dependency descriptions from task.dependencies
811            for dep_id in &task.dependencies {
812                // Look up the dependency task name if available.
813                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        // Also record edge-based dependencies from workflow.edges.
825        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            // Avoid duplicate entries (task.dependencies vs edges may overlap)
846            if !dependencies.contains(&dep_str) {
847                dependencies.push(dep_str);
848            }
849        }
850
851        // Cycle detection warnings.
852        let cycle_warnings = Self::check_cycles(workflow);
853        warnings.extend(cycle_warnings);
854
855        // Warn about tasks with no timeout set (uses the default 1-hour timeout).
856        for task in &tasks {
857            if task.timeout == std::time::Duration::from_secs(3600) {
858                // Default timeout — not necessarily a problem but worth noting.
859            }
860            // Warn about tasks with conditions that reference unknown variables.
861            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    /// Check for circular dependencies in task.dependencies lists.
880    fn check_cycles(workflow: &crate::workflow::Workflow) -> Vec<String> {
881        // Build adjacency: task_id -> [dep_task_ids]
882        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        // Also add edges from workflow.edges.
890        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(); // 0=white,1=gray,2=black
897
898        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); // gray
917        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; // back edge → cycle
924                }
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); // black
933        false
934    }
935
936    /// Estimate the execution duration of a single task in milliseconds.
937    ///
938    /// Uses task-type-specific heuristics:
939    /// - `Wait { duration }` → uses the actual wait duration.
940    /// - `Transcode` → 60 000 ms (1 minute default estimate).
941    /// - `QualityControl` → 30 000 ms.
942    /// - `Transfer` → 20 000 ms.
943    /// - `Analysis` → 15 000 ms.
944    /// - `Notification` / `HttpRequest` → 1 000 ms.
945    /// - `CustomScript` / `Conditional` → 5 000 ms.
946    #[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                // Convert Duration to ms, capping at u64::MAX.
951                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// =============================================================================
966// WorkflowDryRun tests
967// =============================================================================
968
969#[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    // ── empty workflow ────────────────────────────────────────────────────────
1012
1013    #[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    // ── single task ───────────────────────────────────────────────────────────
1028
1029    #[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    // ── multiple tasks ────────────────────────────────────────────────────────
1046
1047    #[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    // ── Wait task duration ────────────────────────────────────────────────────
1058
1059    #[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)); // 10 s = 10_000 ms
1063        let report = WorkflowDryRun::simulate(&wf);
1064        assert_eq!(report.estimated_duration_ms, 10_000);
1065    }
1066
1067    // ── Duration sum ──────────────────────────────────────────────────────────
1068
1069    #[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")); // 60_000 ms
1073        wf.add_task(make_notify_task("notify")); // 1_000 ms
1074        let report = WorkflowDryRun::simulate(&wf);
1075        assert_eq!(report.estimated_duration_ms, 61_000);
1076    }
1077
1078    // ── Dependencies via task.dependencies ───────────────────────────────────
1079
1080    #[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    // ── Dependencies via workflow.edges ───────────────────────────────────────
1101
1102    #[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    // ── Warnings ─────────────────────────────────────────────────────────────
1121
1122    #[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        // Simple valid workflow should have no warnings.
1135        assert!(
1136            report.warnings.is_empty(),
1137            "unexpected warnings: {:?}",
1138            report.warnings
1139        );
1140    }
1141
1142    // ── SimulationReport fields ───────────────────────────────────────────────
1143
1144    #[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        // Just verify all fields are accessible.
1150        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    // ── Multiple dependencies ─────────────────────────────────────────────────
1158
1159    #[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        // Should have at least 2 dependency entries (one per edge).
1171        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}