orion-server 1.11.1

//! Walking a workflow's steps: its `tasks` array, which since dataflow-rs 3.6
//! holds **steps** rather than tasks, and since 3.14 its loop's `setup` list,
//! which holds steps too and runs first.
//!
//! An element carrying a `tasks` key is a *task group* — `{id, condition,
//! terminal, tasks}` — stating one condition for a contiguous run of tasks
//! instead of repeating it on each. Groups nest. dataflow-rs flattens the tree
//! at parse time and records each group's span on the task that opens it, so
//! the executor still walks a flat `&[Task]` and nothing downstream of parsing
//! has to know.
//!
//! Orion's own analysis does not go through that parse. Every check that asks
//! "what does this workflow reference" reads the **authored JSON**: the
//! connector walk, the `channel_call` walk, the task-shape validator, the
//! unresolvable-JSONLogic advisory, the offline call-log correlation. Each of
//! those iterated the array and looked for `function`, so on a grouped
//! workflow each of them would have skipped the group — and every task inside
//! it — without saying so. A connector referenced only from inside a guard
//! clause would have passed closure checking; a task inside one would have
//! gone unvalidated.
//!
//! So there is one flattener here and every walk uses it. The alternative —
//! each walk learning about groups on its own — is how they come to disagree
//! about what a workflow contains, and the disagreement would be silent in the
//! direction that matters: fewer things checked, still reported green.
//!
//! # Since dataflow-rs 3.7 this is an adapter, not an implementation
//!
//! The traversal, the group test and the depth cap now come from
//! [`walk_authored_steps`], which the engine ships precisely so a host stops
//! mirroring them. What is left here is the *shape* Orion's callers want —
//! leaves and groups pre-split into vectors — over the engine's lazy iterator.
//!
//! Mirroring cost Orion one real bug, and it is the reason this module no
//! longer holds a depth constant of its own. Orion counted depth from 1 while
//! the parser counts enclosing groups from 0, so the mirror refused the eighth
//! nesting level that dataflow-rs accepts: a legal workflow was rejected at
//! create with an "engine refuses to build this" message that was not true.
//! Reading [`MAX_GROUP_DEPTH`] means the limit tracks the parser by
//! construction.

use dataflow_rs::{MAX_GROUP_DEPTH, StepKind, walk_authored_steps, walk_authored_steps_at};
use serde_json::Value;

pub use dataflow_rs::is_group;

/// dataflow-rs refuses a step tree nested deeper than this.
///
/// Re-exported rather than redefined so the number an author is told about is
/// the number the parser enforces.
pub const MAX_STEP_DEPTH: usize = MAX_GROUP_DEPTH;

/// What a `tasks` array holds, flattened.
#[derive(Debug, Default)]
pub struct Steps<'a> {
    /// Every leaf task in document order, with the JSON path that addresses
    /// it — `tasks[0]`, `tasks[1].tasks[0]`. The path is what makes a
    /// field error point at the thing the author wrote.
    pub tasks: Vec<(String, &'a Value)>,
    /// Every group, likewise. Groups are validated in their own right: they
    /// carry an id that shares the task namespace, and a condition.
    pub groups: Vec<(String, &'a Value)>,
    /// Groups nested at or past [`MAX_STEP_DEPTH`]. The walk reports them and
    /// stops descending, so a pathological document is bounded and the author
    /// still gets a node to point at.
    ///
    /// A group listed here is *not* also in [`Self::groups`]: it is the one
    /// finding that matters about it, and the whole workflow is refused over
    /// it either way.
    pub too_deep: Vec<String>,
}

/// Every step of a workflow, in the order the engine runs them: its loop's
/// `setup` list, at `loop.setup[0]`…, then `tasks`.
///
/// Both halves, always. A loop's setup steps are tasks the engine resolves,
/// validates and runs exactly as it does the body's (dataflow-rs 3.14), and a
/// walk handed `tasks` alone passed an unknown function, a missing connector
/// or a duplicate id placed there without a word (#351). Taking the loop in
/// the signature is what makes that omission something a caller has to write
/// out rather than fall into.
fn authored_steps<'a>(
    tasks: &'a Value,
    loop_config: Option<&'a Value>,
) -> impl Iterator<Item = dataflow_rs::AuthoredStep<'a>> {
    let setup = loop_config.and_then(|l| l.get("setup"));
    setup
        .into_iter()
        .flat_map(|setup| walk_authored_steps_at(setup, "loop.setup"))
        .chain(walk_authored_steps(tasks))
}

/// Flatten a workflow's steps into its leaf tasks and its groups, setup
/// first.
pub fn walk_steps<'a>(tasks: &'a Value, loop_config: Option<&'a Value>) -> Steps<'a> {
    let mut out = Steps::default();
    for step in authored_steps(tasks, loop_config) {
        match step.kind {
            StepKind::Leaf => out.tasks.push((step.path, step.node)),
            StepKind::Group => out.groups.push((step.path, step.node)),
            StepKind::TooDeep => out.too_deep.push(step.path),
        }
    }
    out
}

/// Just the leaf tasks, setup first, for the walks that do not need paths.
///
/// The engine's walker formats a path for every node, so this no longer saves
/// the allocation it once did — it saves the caller a `filter`/`map` and keeps
/// "which tasks are in this workflow" a single spelling. The traversal is the
/// same one [`walk_steps`] uses, which is the property that matters: the
/// premise of this module is that every walk sees the same tasks.
pub fn leaf_tasks<'a>(tasks: &'a Value, loop_config: Option<&'a Value>) -> Vec<&'a Value> {
    authored_steps(tasks, loop_config)
        .filter(|step| step.kind == StepKind::Leaf)
        .map(|step| step.node)
        .collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn task(id: &str) -> Value {
        json!({"id": id, "name": id, "function": {"name": "map", "input": {"mappings": []}}})
    }

    /// #351: a loop's `setup` steps are steps the engine runs, first, and
    /// every walk must see them, at the coordinate the author typed.
    #[test]
    fn a_loops_setup_steps_come_first_at_their_own_paths() {
        let tasks = json!([task("body")]);
        let loop_config = json!({"max": 2, "setup": [
            task("prime"),
            {"id": "g", "tasks": [task("inner")]}
        ]});
        let steps = walk_steps(&tasks, Some(&loop_config));
        assert_eq!(
            steps
                .tasks
                .iter()
                .map(|(p, _)| p.as_str())
                .collect::<Vec<_>>(),
            ["loop.setup[0]", "loop.setup[1].tasks[0]", "tasks[0]"]
        );
        assert_eq!(steps.groups.len(), 1);
        assert_eq!(leaf_tasks(&tasks, Some(&loop_config)).len(), 3);
        // A loop without `setup` adds nothing.
        assert_eq!(leaf_tasks(&tasks, Some(&json!({"max": 2}))).len(), 1);
    }

    /// The flat case must be untouched: every workflow written before 3.6 is
    /// one, and this walk replaced the loops that handled them.
    #[test]
    fn a_flat_array_yields_its_tasks_in_order() {
        let tasks = json!([task("a"), task("b")]);
        let steps = walk_steps(&tasks, None);
        assert!(steps.groups.is_empty());
        assert_eq!(
            steps
                .tasks
                .iter()
                .map(|(p, _)| p.as_str())
                .collect::<Vec<_>>(),
            ["tasks[0]", "tasks[1]"]
        );
    }

    /// The case that motivated this: a task inside a guard clause is a task,
    /// and every check that walks the array must see it.
    #[test]
    fn a_group_yields_its_members_with_addressable_paths() {
        let tasks = json!([
            task("first"),
            {"id": "guard", "condition": true, "terminal": true, "tasks": [
                task("inner_a"),
                task("inner_b")
            ]},
            task("last")
        ]);
        let steps = walk_steps(&tasks, None);

        assert_eq!(
            steps
                .tasks
                .iter()
                .map(|(p, _)| p.as_str())
                .collect::<Vec<_>>(),
            [
                "tasks[0]",
                "tasks[1].tasks[0]",
                "tasks[1].tasks[1]",
                "tasks[2]"
            ],
            "document order, with the path that addresses each task"
        );
        assert_eq!(steps.groups.len(), 1);
        assert_eq!(steps.groups[0].0, "tasks[1]");
    }

    #[test]
    fn groups_nest() {
        let tasks = json!([
            {"id": "outer", "tasks": [
                {"id": "inner", "tasks": [task("deep")]}
            ]}
        ]);
        let steps = walk_steps(&tasks, None);
        assert_eq!(steps.tasks.len(), 1);
        assert_eq!(steps.tasks[0].0, "tasks[0].tasks[0].tasks[0]");
        assert_eq!(steps.groups.len(), 2, "both levels are groups");
        assert!(steps.too_deep.is_empty());
    }

    /// Deeper than the engine accepts: reported, and the walk stops rather
    /// than recursing without bound on a pathological document.
    #[test]
    fn nesting_past_the_engine_limit_is_reported() {
        let mut inner = json!([task("leaf")]);
        for i in 0..MAX_STEP_DEPTH + 2 {
            inner = json!([{"id": format!("g{i}"), "tasks": inner}]);
        }
        let steps = walk_steps(&inner, None);
        assert!(
            !steps.too_deep.is_empty(),
            "a tree past the limit must be reported, not silently truncated"
        );
    }

    /// The bug the mirror had: the engine accepts [`MAX_STEP_DEPTH`] nesting
    /// levels, and Orion must accept exactly those too. Counting depth from
    /// the wrong base made this workflow a 400 describing an engine refusal
    /// that would never have happened.
    #[test]
    fn the_deepest_nesting_the_parser_accepts_is_not_reported() {
        let mut inner = json!([task("leaf")]);
        for i in 0..MAX_STEP_DEPTH {
            inner = json!([{"id": format!("g{i}"), "tasks": inner}]);
        }
        assert!(
            walk_steps(&inner, None).too_deep.is_empty(),
            "{MAX_STEP_DEPTH} levels of nesting is what the parser accepts"
        );
        assert!(
            serde_json::from_value::<dataflow_rs::Workflow>(json!({
                "id": "w", "name": "w", "condition": true, "tasks": inner,
            }))
            .is_ok(),
            "and the parser agrees, which is the whole claim"
        );
    }

    /// A malformed task with neither `function` nor `tasks` is a broken task,
    /// not an empty group — reporting it as a group would send the author to
    /// the wrong place.
    #[test]
    fn a_task_missing_its_function_is_still_a_task() {
        let tasks = json!([{"id": "broken", "name": "broken"}]);
        let steps = walk_steps(&tasks, None);
        assert_eq!(steps.tasks.len(), 1);
        assert!(steps.groups.is_empty());
    }

    /// [`leaf_tasks`] and [`walk_steps`] must not drift apart. Upstream pins
    /// its walker against the engine's own flattening; this pins Orion's two
    /// spellings of it against each other.
    #[test]
    fn leaf_tasks_sees_exactly_what_walk_steps_sees() {
        let mut deep = json!([task("leaf")]);
        for i in 0..MAX_STEP_DEPTH + 2 {
            deep = json!([{"id": format!("g{i}"), "tasks": deep}]);
        }
        let cases = [
            json!([task("a"), task("b")]),
            json!([task("first"), {"id": "guard", "tasks": [task("x"), task("y")]}, task("last")]),
            json!([{"id": "outer", "tasks": [{"id": "inner", "tasks": [task("deep")]}]}]),
            json!([{"id": "broken", "name": "broken"}]),
            json!([{"id": "empty", "tasks": []}]),
            json!({"not": "an array"}),
            deep,
        ];
        for tasks in cases {
            let expected: Vec<&Value> = walk_steps(&tasks, None)
                .tasks
                .into_iter()
                .map(|(_, t)| t)
                .collect();
            assert_eq!(
                leaf_tasks(&tasks, None),
                expected,
                "disagreement on {tasks}"
            );
        }
    }
}