dataflow-rs 3.8.0

//! `Task::terminal` and task groups: the guard-clause shape.
//!
//! A step in a workflow's `tasks` list is either a task (carries `function`) or
//! a group (carries `tasks`). Both accept `condition` and `terminal`, which
//! gives `if (…) { … }`, an early `return`, and `if (…) { …; return; }`.

use async_trait::async_trait;
use dataflow_rs::engine::functions::AsyncFunctionHandler;
use dataflow_rs::engine::message::Message;
use dataflow_rs::engine::trace::TraceOptions;
use dataflow_rs::{Engine, Result, TaskContext, TaskOutcome, Workflow};
use serde_json::{Value, json};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

mod common;

use common::dv;

/// Writes `data.<key>` so a test can assert exactly which tasks ran, and counts
/// its own calls. Not a sync built-in, so it also forces the async task path
/// and breaks a sync stretch in two.
#[derive(Debug)]
struct MarkAsync {
    calls: Arc<AtomicUsize>,
}

#[async_trait]
impl AsyncFunctionHandler for MarkAsync {
    type Input = Value;

    async fn execute(&self, ctx: &mut TaskContext<'_>, input: &Value) -> Result<TaskOutcome> {
        self.calls.fetch_add(1, Ordering::SeqCst);
        let key = input
            .get("key")
            .and_then(Value::as_str)
            .unwrap_or("async")
            .to_string();
        ctx.set(
            &format!("data.{key}"),
            datavalue::OwnedDataValue::Bool(true),
        );
        Ok(TaskOutcome::Success)
    }
}

/// Returns whatever outcome the test asked for, so `Skip` / `Status(500)` can
/// be paired with `terminal`.
#[derive(Debug)]
struct Outcome(TaskOutcome);

#[async_trait]
impl AsyncFunctionHandler for Outcome {
    type Input = Value;

    async fn execute(&self, _ctx: &mut TaskContext<'_>, _input: &Value) -> Result<TaskOutcome> {
        Ok(self.0)
    }
}

/// Always fails, so `terminal` can be tested against a failed task.
#[derive(Debug)]
struct Boom;

#[async_trait]
impl AsyncFunctionHandler for Boom {
    type Input = Value;

    async fn execute(&self, _ctx: &mut TaskContext<'_>, _input: &Value) -> Result<TaskOutcome> {
        Err(dataflow_rs::DataflowError::Task("boom".to_string()))
    }
}

/// A `map` task writing `data.<key> = true`, as a JSON step.
fn mark(id: &str) -> String {
    format!(
        r#"{{ "id": "{id}", "name": "{id}", "function": {{ "name": "map",
             "input": {{ "mappings": [ {{ "path": "data.{id}", "logic": true }} ] }} }} }}"#
    )
}

/// Wrap `steps` in a workflow whose condition is always true.
fn workflow(steps: &str) -> Workflow {
    Workflow::from_json(&format!(
        r#"{{ "id": "w", "name": "w", "priority": 0, "tasks": [{steps}] }}"#
    ))
    .expect("workflow should parse")
}

/// Run one message through `engine` and hand back the finished message.
async fn run(engine: &Engine, data: Value) -> Message {
    let mut message = Message::builder().data(dv(data)).build();
    engine
        .process_message(&mut message)
        .await
        .expect("processing should succeed");
    message
}

/// Which of `data.<key>` were written, in the order asked about.
fn ran(message: &Message, keys: &[&str]) -> Vec<String> {
    keys.iter()
        .filter(|k| Value::from(&message.context["data"][**k]) == json!(true))
        .map(|k| (*k).to_string())
        .collect()
}

fn engine_for(workflows: Vec<Workflow>) -> Engine {
    Engine::builder()
        .with_workflows(workflows)
        .build()
        .expect("engine should build")
}

// =============================================================================
// `terminal` on a task
// =============================================================================

#[tokio::test]
async fn terminal_task_stops_the_workflow_when_its_condition_holds() {
    let engine = engine_for(vec![workflow(&format!(
        r#"{},
           {{ "id": "exit", "name": "exit", "terminal": true,
              "condition": {{ "var": "data.stop" }},
              "function": {{ "name": "map", "input": {{ "mappings": [
                  {{ "path": "data.exit", "logic": true }} ] }} }} }},
           {}"#,
        mark("first"),
        mark("last")
    ))]);

    let message = run(&engine, json!({ "stop": true })).await;
    assert_eq!(
        ran(&message, &["first", "exit", "last"]),
        vec!["first", "exit"],
        "the terminal task ran and nothing after it did"
    );
}

#[tokio::test]
async fn terminal_task_with_a_false_condition_does_not_halt() {
    let engine = engine_for(vec![workflow(&format!(
        r#"{},
           {{ "id": "exit", "name": "exit", "terminal": true,
              "condition": {{ "var": "data.stop" }},
              "function": {{ "name": "map", "input": {{ "mappings": [
                  {{ "path": "data.exit", "logic": true }} ] }} }} }},
           {}"#,
        mark("first"),
        mark("last")
    ))]);

    let message = run(&engine, json!({ "stop": false })).await;
    assert_eq!(
        ran(&message, &["first", "exit", "last"]),
        vec!["first", "last"],
        "a guard that did not fire cannot end the workflow"
    );
}

#[tokio::test]
async fn terminal_task_returning_skip_does_not_halt() {
    // `TaskOutcome::Skip` opts out of the per-task record entirely — the task
    // declined to act, so "stop after this task" has nothing to stop after.
    let engine = Engine::builder()
        .with_workflows(vec![workflow(&format!(
            r#"{{ "id": "skipper", "name": "skipper", "terminal": true,
                  "function": {{ "name": "skipper", "input": {{}} }} }},
               {}"#,
            mark("last")
        ))])
        .register("skipper", Outcome(TaskOutcome::Skip))
        .build()
        .expect("engine should build");

    let message = run(&engine, json!({})).await;
    assert_eq!(
        ran(&message, &["last"]),
        vec!["last"],
        "a skipped task does not halt, even when marked terminal"
    );
}

#[tokio::test]
async fn terminal_task_that_failed_under_continue_on_error_still_halts() {
    // `terminal` is about position, not outcome: the author said "nothing after
    // this runs". The error is still recorded.
    let engine = Engine::builder()
        .with_workflows(vec![workflow(&format!(
            r#"{{ "id": "boom", "name": "boom", "terminal": true,
                  "continue_on_error": true,
                  "function": {{ "name": "boom", "input": {{}} }} }},
               {}"#,
            mark("last")
        ))])
        .register("boom", Boom)
        .build()
        .expect("engine should build");

    let message = run(&engine, json!({})).await;
    assert!(
        ran(&message, &["last"]).is_empty(),
        "a failed terminal task still ends the workflow"
    );
    assert_eq!(
        message.errors().len(),
        1,
        "and its error is still on the message"
    );
    assert_eq!(message.errors()[0].task_id.as_deref(), Some("boom"));
}

#[tokio::test]
async fn terminal_task_returning_5xx_still_records_and_propagates() {
    // Regression guard: folding `terminal` into the halt decision *before* the
    // status classification would skip the `TASK_STATUS_ERROR` push and the
    // `Err` return, silently turning a failure into a clean stop.
    let engine = Engine::builder()
        .with_workflows(vec![workflow(
            r#"{ "id": "five_hundred", "name": "five_hundred", "terminal": true,
                 "function": { "name": "five_hundred", "input": {} } }"#,
        )])
        .register("five_hundred", Outcome(TaskOutcome::Status(500)))
        .build()
        .expect("engine should build");

    let mut message = Message::builder().data(dv(json!({}))).build();
    let result = engine.process_message(&mut message).await;

    assert!(
        result.is_err(),
        "a 5xx with continue_on_error unset still stops the engine"
    );
    assert!(
        message
            .errors()
            .iter()
            .any(|e| e.code == "TASK_STATUS_ERROR"),
        "and the status error is still recorded, got {:?}",
        message.errors()
    );
}

#[tokio::test]
async fn terminal_task_keeps_its_own_audit_status() {
    let engine = Engine::builder()
        .with_workflows(vec![workflow(
            r#"{ "id": "not_found", "name": "not_found", "terminal": true,
                 "function": { "name": "not_found", "input": {} } }"#,
        )])
        .register("not_found", Outcome(TaskOutcome::Status(404)))
        .build()
        .expect("engine should build");

    let message = run(&engine, json!({})).await;
    let statuses: Vec<usize> = message.audit_trail().iter().map(|e| e.status).collect();
    assert_eq!(
        statuses,
        vec![404],
        "the task's own status survives; `terminal` is not `HALT_STATUS_CODE`"
    );
}

#[tokio::test]
async fn terminal_halt_stops_only_its_own_workflow() {
    let engine = engine_for(vec![
        Workflow::from_json(&format!(
            r#"{{ "id": "first", "name": "first", "priority": 0, "tasks": [
                 {{ "id": "exit", "name": "exit", "terminal": true,
                    "function": {{ "name": "map", "input": {{ "mappings": [
                        {{ "path": "data.exit", "logic": true }} ] }} }} }},
                 {}
               ] }}"#,
            mark("unreachable")
        ))
        .unwrap(),
        Workflow::from_json(&format!(
            r#"{{ "id": "second", "name": "second", "priority": 1, "tasks": [{}] }}"#,
            mark("later_workflow")
        ))
        .unwrap(),
    ]);

    let message = run(&engine, json!({})).await;
    assert_eq!(
        ran(&message, &["exit", "unreachable", "later_workflow"]),
        vec!["exit", "later_workflow"],
        "halt scopes to the workflow, exactly like TaskOutcome::Halt"
    );
}

#[tokio::test]
async fn terminal_task_breaks_the_whole_loop() {
    let engine = engine_for(vec![
        Workflow::from_json(
            r#"{ "id": "sweeps", "name": "sweeps", "priority": 1,
                 "loop": { "counter": "i", "max": 5 },
                 "tasks": [
                   { "id": "tick", "name": "tick", "function": { "name": "map",
                     "input": { "mappings": [ { "path": "data.ticks",
                       "logic": {"+": [{"var": "data.ticks"}, 1]} } ] } } },
                   { "id": "exit", "name": "exit", "terminal": true,
                     "condition": {">=": [{"var": "temp_data.i"}, 2]},
                     "function": { "name": "map", "input": { "mappings": [
                       { "path": "data.exit", "logic": true } ] } } }
                 ] }"#,
        )
        .unwrap(),
    ]);

    let message = run(&engine, json!({ "ticks": 0 })).await;
    assert_eq!(
        Value::from(&message.context["data"]["ticks"]),
        json!(3),
        "sweeps 0, 1 and 2 ran; the terminal task on sweep 2 broke the loop"
    );
}

// =============================================================================
// Task groups
// =============================================================================

#[tokio::test]
async fn group_with_a_false_condition_skips_every_member() {
    let engine = engine_for(vec![workflow(&format!(
        r#"{{ "id": "guarded", "condition": {{ "var": "data.go" }}, "tasks": [{}, {}] }},
           {}"#,
        mark("a"),
        mark("b"),
        mark("after")
    ))]);

    let message = run(&engine, json!({ "go": false })).await;
    assert_eq!(
        ran(&message, &["a", "b", "after"]),
        vec!["after"],
        "the whole span is skipped, and the list resumes past it"
    );

    let message = run(&engine, json!({ "go": true })).await;
    assert_eq!(
        ran(&message, &["a", "b", "after"]),
        vec!["a", "b", "after"],
        "and runs in full when the condition holds"
    );
}

#[tokio::test]
async fn a_skipped_group_records_one_skipped_step_per_member() {
    let engine = engine_for(vec![workflow(&format!(
        r#"{{ "id": "guarded", "condition": {{ "var": "data.go" }}, "tasks": [{}, {}] }},
           {}"#,
        mark("a"),
        mark("b"),
        mark("after")
    ))]);

    let mut message = Message::builder().data(dv(json!({ "go": false }))).build();
    let mut trace =
        dataflow_rs::engine::trace::ExecutionTrace::with_options(TraceOptions::default());
    engine
        .process_message_tracing(&mut message, &mut trace)
        .await
        .expect("processing should succeed");

    let skipped: Vec<&str> = trace
        .steps
        .iter()
        .filter(|s| s.result == dataflow_rs::engine::trace::StepResult::Skipped)
        .filter_map(|s| s.task_id.as_deref())
        .collect();
    assert_eq!(
        skipped,
        vec!["a", "b"],
        "the trace stays task-granular — no group-level step variant"
    );
}

#[tokio::test]
async fn a_group_condition_is_evaluated_once_on_entry() {
    // The behaviour that rules out AND-folding the group condition into each
    // child: a task inside the block mutates what the block's condition reads,
    // and its siblings must still run. This is an `if` block, not N guards.
    let engine = engine_for(vec![workflow(&format!(
        r#"{{ "id": "guarded", "condition": {{ "var": "data.go" }}, "tasks": [
             {{ "id": "clear", "name": "clear", "function": {{ "name": "map",
                "input": {{ "mappings": [ {{ "path": "data.go", "logic": false }} ] }} }} }},
             {}
           ] }}"#,
        mark("sibling")
    ))]);

    let message = run(&engine, json!({ "go": true })).await;
    assert_eq!(
        ran(&message, &["sibling"]),
        vec!["sibling"],
        "the group was entered once; a member turning the condition off mid-block \
         must not switch off its own siblings"
    );
}

#[tokio::test]
async fn terminal_group_halts_after_its_last_task() {
    let steps = format!(
        r#"{{ "id": "exit_branch", "terminal": true, "condition": {{ "var": "data.stop" }},
              "tasks": [{}, {}] }},
           {}"#,
        mark("body_one"),
        mark("body_two"),
        mark("after")
    );
    let engine = engine_for(vec![workflow(&steps)]);

    let message = run(&engine, json!({ "stop": true })).await;
    assert_eq!(
        ran(&message, &["body_one", "body_two", "after"]),
        vec!["body_one", "body_two"],
        "the whole branch ran, then the workflow ended"
    );

    let message = run(&engine, json!({ "stop": false })).await;
    assert_eq!(
        ran(&message, &["body_one", "body_two", "after"]),
        vec!["after"],
        "a branch that never fired cannot end the workflow"
    );
}

#[tokio::test]
async fn nested_groups_skip_only_their_own_span() {
    let steps = format!(
        r#"{{ "id": "outer", "condition": {{ "var": "data.outer" }}, "tasks": [
             {},
             {{ "id": "inner", "condition": {{ "var": "data.inner" }}, "tasks": [{}] }},
             {}
           ] }},
           {}"#,
        mark("before_inner"),
        mark("in_inner"),
        mark("after_inner"),
        mark("after_outer")
    );
    let engine = engine_for(vec![workflow(&steps)]);
    let keys = ["before_inner", "in_inner", "after_inner", "after_outer"];

    let message = run(&engine, json!({ "outer": true, "inner": false })).await;
    assert_eq!(
        ran(&message, &keys),
        vec!["before_inner", "after_inner", "after_outer"],
        "a false inner condition skips only the inner span"
    );

    let message = run(&engine, json!({ "outer": false, "inner": true })).await;
    assert_eq!(
        ran(&message, &keys),
        vec!["after_outer"],
        "a false outer condition skips the whole subtree"
    );

    let message = run(&engine, json!({ "outer": true, "inner": true })).await;
    assert_eq!(ran(&message, &keys), keys, "both true runs everything");
}

#[tokio::test]
async fn a_terminal_outer_group_halts_even_when_its_only_child_was_skipped() {
    // The case a per-task close counter gets wrong: nothing inside `outer` ever
    // executes, so the task that would carry "outer closes here" is jumped
    // straight over — yet `outer` was entered and is terminal.
    let steps = format!(
        r#"{{ "id": "outer", "terminal": true, "condition": {{ "var": "data.outer" }},
              "tasks": [
                {{ "id": "inner", "condition": {{ "var": "data.inner" }}, "tasks": [{}] }}
              ] }},
           {}"#,
        mark("in_inner"),
        mark("after_outer")
    );
    let engine = engine_for(vec![workflow(&steps)]);

    let message = run(&engine, json!({ "outer": true, "inner": false })).await;
    assert_eq!(
        ran(&message, &["in_inner", "after_outer"]),
        Vec::<String>::new(),
        "the entered terminal group still ends the workflow"
    );
}

#[tokio::test]
async fn a_group_spanning_an_async_boundary_enters_and_exits_correctly() {
    let calls = Arc::new(AtomicUsize::new(0));
    let steps = format!(
        r#"{{ "id": "guarded", "condition": {{ "var": "data.go" }}, "tasks": [
             {},
             {{ "id": "mid", "name": "mid",
                "function": {{ "name": "mark_async", "input": {{ "key": "mid" }} }} }},
             {}
           ] }},
           {}"#,
        mark("pre"),
        mark("post"),
        mark("after")
    );
    let engine = Engine::builder()
        .with_workflows(vec![workflow(&steps)])
        .register(
            "mark_async",
            MarkAsync {
                calls: Arc::clone(&calls),
            },
        )
        .build()
        .expect("engine should build");
    let keys = ["pre", "mid", "post", "after"];

    let message = run(&engine, json!({ "go": true })).await;
    assert_eq!(ran(&message, &keys), keys, "sync, async, sync all ran");
    assert_eq!(calls.load(Ordering::SeqCst), 1);

    let message = run(&engine, json!({ "go": false })).await;
    assert_eq!(
        ran(&message, &keys),
        vec!["after"],
        "a false condition jumps past the async task too"
    );
    assert_eq!(
        calls.load(Ordering::SeqCst),
        1,
        "the async handler was not called for the skipped span"
    );
}

#[tokio::test]
async fn groups_behave_the_same_in_the_shared_arena_cross_workflow_run() {
    // Two consecutive fully-sync workflows run inside one shared `ArenaContext`
    // via `execute_sync_workflow_run`; the group gate is per-workflow.
    let engine = engine_for(vec![
        Workflow::from_json(&format!(
            r#"{{ "id": "first", "name": "first", "priority": 0, "tasks": [
                 {{ "id": "g1", "condition": {{ "var": "data.go" }}, "tasks": [{}] }},
                 {}
               ] }}"#,
            mark("a"),
            mark("b")
        ))
        .unwrap(),
        Workflow::from_json(&format!(
            r#"{{ "id": "second", "name": "second", "priority": 1, "tasks": [
                 {{ "id": "g2", "condition": {{ "var": "data.go" }}, "tasks": [{}] }},
                 {}
               ] }}"#,
            mark("c"),
            mark("d")
        ))
        .unwrap(),
    ]);

    let message = run(&engine, json!({ "go": false })).await;
    assert_eq!(
        ran(&message, &["a", "b", "c", "d"]),
        vec!["b", "d"],
        "each workflow's group is gated independently"
    );

    let message = run(&engine, json!({ "go": true })).await;
    assert_eq!(ran(&message, &["a", "b", "c", "d"]), ["a", "b", "c", "d"]);
}

// =============================================================================
// Parsing and validation
// =============================================================================

#[tokio::test]
async fn a_workflow_without_groups_records_no_spans() {
    let wf = workflow(&format!("{}, {}", mark("a"), mark("b")));
    assert_eq!(wf.tasks.len(), 2);
    assert!(
        wf.tasks.iter().all(|t| t.group_starts.is_empty()),
        "no groups authored, no spans recorded"
    );
    assert!(
        wf.tasks.iter().all(|t| !t.terminal),
        "`terminal` defaults to false"
    );
}

#[test]
fn a_group_flattens_into_the_task_list_in_document_order() {
    let wf = workflow(&format!(
        r#"{}, {{ "id": "g", "condition": true, "tasks": [{}, {}] }}, {}"#,
        mark("a"),
        mark("b"),
        mark("c"),
        mark("d")
    ));

    let ids: Vec<&str> = wf.tasks.iter().map(|t| t.id.as_str()).collect();
    assert_eq!(ids, vec!["a", "b", "c", "d"]);
    assert_eq!(wf.tasks[1].group_starts.len(), 1, "the group opens at `b`");
    assert_eq!(
        wf.tasks[1].group_starts[0].end, 3,
        "and its half-open span ends after `c`"
    );
}

#[test]
fn nested_groups_are_recorded_outermost_first() {
    let wf = workflow(&format!(
        r#"{{ "id": "outer", "condition": true, "tasks": [
             {{ "id": "inner", "condition": true, "tasks": [{}] }}
           ] }}"#,
        mark("only")
    ));

    let ids: Vec<&str> = wf.tasks[0]
        .group_starts
        .iter()
        .map(|g| g.id.as_str())
        .collect();
    assert_eq!(
        ids,
        vec!["outer", "inner"],
        "the gate evaluates the outer condition first"
    );
}

#[test]
fn an_empty_group_is_rejected_at_parse_time() {
    let err = Workflow::from_json(
        r#"{ "id": "w", "name": "w", "tasks": [ { "id": "g", "tasks": [] } ] }"#,
    )
    .expect_err("an empty group should not parse");
    assert!(format!("{err}").contains("contains no tasks"), "got: {err}");
}

#[test]
fn groups_nested_too_deeply_are_rejected_at_parse_time() {
    let mut json = mark("leaf");
    for i in 0..9 {
        json = format!(r#"{{ "id": "g{i}", "condition": true, "tasks": [{json}] }}"#);
    }
    let err = Workflow::from_json(&format!(
        r#"{{ "id": "w", "name": "w", "tasks": [{json}] }}"#
    ))
    .expect_err("nine levels should not parse");
    assert!(format!("{err}").contains("nested deeper"), "got: {err}");
}

#[test]
fn a_group_id_colliding_with_a_task_id_is_rejected() {
    let wf = workflow(&format!(
        r#"{{ "id": "a", "condition": true, "tasks": [{}] }}"#,
        mark("a")
    ));
    let err = wf.validate().expect_err("the collision should be rejected");
    assert!(format!("{err}").contains("Duplicate"), "got: {err}");
}

#[test]
fn a_malformed_task_still_reports_its_own_missing_field() {
    // The reason this is not `#[serde(untagged)]`: an untagged enum would
    // report "data did not match any variant" instead.
    let err = Workflow::from_json(
        r#"{ "id": "w", "name": "w", "tasks": [ { "id": "t", "name": "t" } ] }"#,
    )
    .expect_err("a task with no function should not parse");
    assert!(
        format!("{err}").contains("missing field") && format!("{err}").contains("function"),
        "got: {err}"
    );
}