//! Composition runtime: walk a [`super::ResolvedComposition`] in
//! topological order, dispatching each node against a server pool and
//! recording a [`super::CompositionTrace`].
//!
//! Serial execution today. Parallel scheduling for independent nodes
//! is a deliberate follow-up; the trace shape and the substitution
//! contract here do not change when the scheduler upgrades, so suites
//! authored against this runtime keep working unchanged. The runner
//! still walks every node in `topo_order`, but a future scheduler is
//! free to group nodes at the same indegree-zero rank and dispatch
//! them concurrently.
//!
//! Failure semantics are simple: an `err` node short-circuits every
//! downstream consumer. The runner marks every descendant that has
//! the failed node as an ancestor `skipped` and the trace's
//! `composition.ran` flag stays `true` because the run finished in a
//! defined state; a downstream assertion that gates on `composition.ran`
//! can still address `nodes.<id>.status == ok` to fail the test.

use std::collections::{BTreeMap, HashMap, HashSet};
use std::time::Instant;

use serde_json::{json, Value};

use crate::connector::ServerPool;

use super::data_flow::substitute_with;
use super::{
    substitute, substitute_with_item, CaseSpec, CompositionTrace, NodeStatus, NodeTrace,
    OutputSpec, ResolvedComposition, ResolvedNode, SubstitutionError,
};
use crate::connector::LiveServer;

/// Dispatch every node in topological order against `pool` and
/// return the assembled [`CompositionTrace`].
///
/// The default `server:` name is the only entry of the pool when the
/// composition's nodes do not name one explicitly. This mirrors the
/// loader's "single-default" rule for tool tests and keeps the YAML
/// short when a suite declares one server.
#[allow(clippy::too_many_lines)]
pub async fn run(
    composition: &ResolvedComposition,
    pool: &ServerPool,
    default_server: Option<&str>,
) -> CompositionTrace {
    let mut traces: BTreeMap<String, NodeTrace> = BTreeMap::new();
    let mut order: Vec<String> = Vec::with_capacity(composition.topo_order.len());
    let mut failed_ancestors: HashSet<String> = HashSet::new();

    for id in &composition.topo_order {
        let node = composition
            .nodes
            .get(id)
            .expect("topo_order contains an id not declared in nodes");

        // Skip when any parent already failed: the substitution that
        // would consume the parent's output cannot complete, and
        // running a half-fed node is worse than recording skip.
        let any_parent_failed = node.needs.iter().any(|p| failed_ancestors.contains(p));
        if any_parent_failed {
            traces.insert(
                id.clone(),
                NodeTrace {
                    id: id.clone(),
                    status: NodeStatus::Skipped,
                    inputs: Value::Null,
                    output: Value::Null,
                    duration_ms: None,
                    ..Default::default()
                },
            );
            failed_ancestors.insert(id.clone());
            continue;
        }

        // Assemble the args at dispatch time. Parent outputs are
        // pulled from the trace map we have built so far; the
        // substitution layer rejects an unknown id (which the load
        // validator already prevented) or an unknown field.
        let parent_outputs: HashMap<String, Value> = node
            .needs
            .iter()
            .filter_map(|p| traces.get(p).map(|t| (p.clone(), t.output.clone())))
            .collect();

        // Evaluate `when:` predicate. A falsy result skips the node
        // (and propagates the skip to descendants); a jq runtime
        // error fails the node so the assertion pipeline can see the
        // failure without the node ever dispatching.
        match evaluate_when(node.when.as_deref(), &parent_outputs) {
            WhenOutcome::Run => {}
            WhenOutcome::Skip => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Skipped,
                        inputs: Value::Null,
                        output: Value::Null,
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
            WhenOutcome::Error(message) => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs: Value::Null,
                        output: Value::String(message),
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
        }
        let raw_args = node.args.clone().unwrap_or(Value::Null);
        let inputs = match substitute(&raw_args, &parent_outputs) {
            Ok(v) => v,
            Err(err) => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs: Value::String(substitution_error_message(&err)),
                        output: Value::Null,
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
        };

        // Inline mock short-circuit: record the fixture as the node's
        // output without a server lookup or a dispatch. Apply the
        // per-node response transform exactly as the live path does
        // so a frozen and a live node are indistinguishable from
        // downstream's point of view.
        if let Some(mock) = &node.mock {
            let raw = mock.result.clone();
            let (status, output) = match apply_transform(node.transform.as_ref(), raw) {
                Ok(reshaped) => (NodeStatus::Ok, reshaped),
                Err(message) => {
                    failed_ancestors.insert(id.clone());
                    (NodeStatus::Err, Value::String(message))
                }
            };
            traces.insert(
                id.clone(),
                NodeTrace {
                    id: id.clone(),
                    status,
                    inputs,
                    output,
                    duration_ms: Some(0),
                    ..Default::default()
                },
            );
            order.push(id.clone());
            continue;
        }

        // Pick the server: explicit `server:` on the node wins; the
        // caller-provided default fills in when the node omits one.
        let server_name = node.server.as_deref().or(default_server).unwrap_or("");
        let server = match pool.get(server_name).await {
            Some(s) => s,
            None => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs,
                        output: Value::String(format!(
                            "composition node `{id}` references unknown server `{server_name}`"
                        )),
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
        };

        let started = Instant::now();
        let (status, output) = if node.for_each.is_some() {
            dispatch_fan_out(
                node,
                &parent_outputs,
                server.as_ref(),
                &mut failed_ancestors,
            )
            .await
        } else {
            dispatch_single(node, &inputs, server.as_ref()).await
        };
        let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);

        traces.insert(
            id.clone(),
            NodeTrace {
                id: id.clone(),
                status,
                inputs,
                output,
                duration_ms: Some(duration_ms),
                ..Default::default()
            },
        );
        order.push(id.clone());
    }

    let output = assemble_output(&composition.output, &composition.topo_order, &traces);
    let (cost_total, tokens_total) = totals(&traces);
    let budget_tripped = check_budget(composition.budget.as_ref(), cost_total, tokens_total);

    CompositionTrace {
        name: composition.name.clone(),
        order,
        ran: budget_tripped.is_none()
            && !traces
                .values()
                .any(|t| t.status == NodeStatus::Err && t.duration_ms.is_none()),
        nodes: traces,
        output,
        cost_total,
        tokens_total,
        budget_tripped,
    }
}

/// Sum the cost_usd and tokens fields across every node so the
/// trace surfaces `cost.total` and `tokens.total` without the
/// assertion pipeline re-walking the map.
fn totals(traces: &BTreeMap<String, NodeTrace>) -> (f64, u64) {
    let cost = traces.values().map(|t| t.cost_usd).sum();
    let tokens = traces.values().map(|t| t.tokens).sum();
    (cost, tokens)
}

/// Decide whether any composition-level budget cap tripped and
/// return the name of the cap that fired. Returns `None` when every
/// cap held. Ordering: cost, tokens, duration; the first to fire
/// wins so the reporter shows the user a single clear pointer.
fn check_budget(
    budget: Option<&super::BudgetSpec>,
    cost_total: f64,
    tokens_total: u64,
) -> Option<String> {
    let spec = budget?;
    if let Some(cap) = spec.max_cost_usd {
        if cost_total > cap {
            return Some("max_cost_usd".to_string());
        }
    }
    if let Some(cap) = spec.max_tokens {
        if tokens_total > cap {
            return Some("max_tokens".to_string());
        }
    }
    None
}

fn substitution_error_message(err: &SubstitutionError) -> String {
    err.to_string()
}

/// Apply the per-node response transform when one is declared.
/// Returns the reshaped value on success or a one-line error message
/// the trace records and downstream nodes propagate as a skip.
fn apply_transform(
    transform: Option<&super::TransformSpec>,
    value: Value,
) -> Result<Value, String> {
    let Some(spec) = transform else {
        return Ok(value);
    };
    crate::transform_jq::run(&spec.jq, &value).map_err(|e| format!("jq transform failed: {e}"))
}

/// Dispatch a single (non-fan-out) node against the named server.
/// Returns the final (status, output) pair the trace records.
async fn dispatch_single(
    node: &ResolvedNode,
    inputs: &Value,
    server: &LiveServer,
) -> (NodeStatus, Value) {
    match server
        .client
        .request(
            "tools/call".to_string(),
            json!({"name": node.tool, "arguments": inputs}),
        )
        .await
    {
        Ok(v) => match apply_transform(node.transform.as_ref(), v) {
            Ok(reshaped) => (NodeStatus::Ok, reshaped),
            Err(message) => (NodeStatus::Err, Value::String(message)),
        },
        Err(e) => (
            NodeStatus::Err,
            Value::String(format!("tools/call failed: {e}")),
        ),
    }
}

/// Hard ceiling on `for_each` fan-out width (WOR-1313).
///
/// A parent that unexpectedly returns a huge array (a paginated tool
/// gone wrong, a server echoing its whole dataset) would otherwise fan
/// into one tool call per element and hammer the user's server for
/// hours. 10000 is far beyond any sane test suite; exceeding it is
/// treated as a bug in the composition, not a workload to run.
/// A node overrides it with `max_iterations:` (WOR-1322), which the
/// loader bounds to [`FOR_EACH_ITERATIONS_CEILING`].
const MAX_FOR_EACH_ITERATIONS: usize = 10_000;

/// Hard ceiling on the per-node `max_iterations:` override (WOR-1322).
/// Exists so a typo in the override cannot authorize a six-figure
/// fan-out; enforced by the composition validator at load time.
pub(super) const FOR_EACH_ITERATIONS_CEILING: usize = 100_000;

/// Render the over-limit diagnostic. Split out so the guard's message
/// (which must name the offending count) is pinned by a unit test
/// without standing up a live server.
fn fan_out_over_limit_message(parent_id: &str, count: usize, limit: usize) -> String {
    format!(
        "for_each over `{parent_id}` would dispatch {count} iterations, \
         exceeding the limit of {limit}"
    )
}

/// Dispatch a `for_each:` node: resolve the parent array, iterate
/// over each element substituting `${item}`, and collect the
/// per-iteration outputs into an array. Returns the
/// (status, output-array) pair the trace records.
///
/// Empty arrays are not an error: the node ends `ok` with an empty
/// output array. A non-array parent output ends the node `err` so a
/// suite that mistakes the parent shape fails loudly. Concurrency is
/// bounded by the node's `max_concurrency` (defaults to 1, sequential)
/// and the total fan-out width by [`MAX_FOR_EACH_ITERATIONS`].
async fn dispatch_fan_out(
    node: &ResolvedNode,
    parent_outputs: &HashMap<String, Value>,
    server: &LiveServer,
    failed_ancestors: &mut HashSet<String>,
) -> (NodeStatus, Value) {
    let Some(parent_id) = node.for_each.clone() else {
        // Should be unreachable: the caller only routes here when
        // for_each is Some. Bail cleanly so a regression surfaces.
        return (
            NodeStatus::Err,
            Value::String("internal: dispatch_fan_out without for_each".into()),
        );
    };
    let Some(source) = parent_outputs.get(&parent_id) else {
        failed_ancestors.insert(node.id.clone());
        return (
            NodeStatus::Err,
            Value::String(format!(
                "for_each parent `{parent_id}` not found in node `{}`'s parent map",
                node.id
            )),
        );
    };
    let Some(items) = source.as_array() else {
        failed_ancestors.insert(node.id.clone());
        return (
            NodeStatus::Err,
            Value::String(format!(
                "for_each parent `{parent_id}` is not an array; got {}",
                short_type(source)
            )),
        );
    };
    let limit = node.max_iterations.unwrap_or(MAX_FOR_EACH_ITERATIONS);
    if items.len() > limit {
        failed_ancestors.insert(node.id.clone());
        return (
            NodeStatus::Err,
            Value::String(fan_out_over_limit_message(&parent_id, items.len(), limit)),
        );
    }
    let mut outputs: Vec<Value> = Vec::with_capacity(items.len());
    let raw_args = node.args.clone().unwrap_or(Value::Null);
    let cap = node.max_concurrency.unwrap_or(1).max(1);

    // Sequential v1: bound concurrency at 1 unconditionally so an
    // exhaustive parallel fan-out cannot quietly multiply load on a
    // user server. cap is wired through for the upgrade path; the
    // parallel scheduler lands as a follow-up so the trace shape
    // and the concurrency story can be tested in isolation.
    let _ = cap;
    for item in items {
        let inputs = match substitute_with_item(&raw_args, parent_outputs, Some(item)) {
            Ok(v) => v,
            Err(err) => {
                failed_ancestors.insert(node.id.clone());
                return (
                    NodeStatus::Err,
                    Value::String(format!("for_each substitution failed: {err}")),
                );
            }
        };
        let (status, output) = dispatch_single(node, &inputs, server).await;
        if status != NodeStatus::Ok {
            failed_ancestors.insert(node.id.clone());
            return (
                NodeStatus::Err,
                Value::String(format!(
                    "for_each iteration failed on item {}: {}",
                    outputs.len(),
                    stringify_short(&output)
                )),
            );
        }
        outputs.push(output);
    }
    (NodeStatus::Ok, Value::Array(outputs))
}

/// Per-case execution trace. The runner produces one of these per
/// `cases:` row when the composition declares the dataset.
#[derive(Debug, Clone)]
pub struct CaseTrace {
    /// Case name from `cases[].name`.
    pub name: String,
    /// The trace the executor produced for this case.
    pub trace: CompositionTrace,
    /// Per-case golden value (verbatim from the case row), forwarded
    /// for the reporter. The runner records it; comparison against
    /// the combined output is the reporter's job.
    pub golden: Option<Value>,
    /// Per-case assertions appended to the composition's base
    /// `expect:` list when one is present. Empty when the case row
    /// omits its own.
    pub expect: Vec<Value>,
}

/// Run the composition once per `cases:` row, binding each row's
/// `vars` for `${var.X}` substitution. Returns one [`CaseTrace`] per
/// row. When the composition declares no cases, the caller should
/// fall back to [`run`].
pub async fn run_cases(
    composition: &ResolvedComposition,
    pool: &ServerPool,
    default_server: Option<&str>,
) -> Vec<CaseTrace> {
    let mut out = Vec::with_capacity(composition.cases.len());
    for case in &composition.cases {
        let trace = run_one_case(composition, pool, default_server, case).await;
        out.push(CaseTrace {
            name: case.name.clone(),
            trace,
            golden: case.golden.clone(),
            expect: case.expect.clone(),
        });
    }
    out
}

/// Execute the composition for a single case row. The case's `vars`
/// are threaded through every node's args via [`substitute_with`] so
/// `${var.X}` references resolve and `${id.field}` / `${item}` keep
/// their existing semantics.
#[allow(clippy::too_many_lines)]
async fn run_one_case(
    composition: &ResolvedComposition,
    pool: &ServerPool,
    default_server: Option<&str>,
    case: &CaseSpec,
) -> CompositionTrace {
    let case_vars: HashMap<String, Value> = case
        .vars
        .iter()
        .map(|(k, v)| (k.clone(), v.clone()))
        .collect();
    let mut traces: BTreeMap<String, NodeTrace> = BTreeMap::new();
    let mut order: Vec<String> = Vec::with_capacity(composition.topo_order.len());
    let mut failed_ancestors: HashSet<String> = HashSet::new();
    for id in &composition.topo_order {
        let node = composition
            .nodes
            .get(id)
            .expect("topo_order contains an id not declared in nodes");
        if node.needs.iter().any(|p| failed_ancestors.contains(p)) {
            traces.insert(
                id.clone(),
                NodeTrace {
                    id: id.clone(),
                    status: NodeStatus::Skipped,
                    inputs: Value::Null,
                    output: Value::Null,
                    duration_ms: None,
                    ..Default::default()
                },
            );
            failed_ancestors.insert(id.clone());
            continue;
        }
        let parent_outputs: HashMap<String, Value> = node
            .needs
            .iter()
            .filter_map(|p| traces.get(p).map(|t| (p.clone(), t.output.clone())))
            .collect();
        let raw_args = node.args.clone().unwrap_or(Value::Null);
        let inputs = match substitute_with(&raw_args, &parent_outputs, None, Some(&case_vars)) {
            Ok(v) => v,
            Err(err) => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs: Value::String(err.to_string()),
                        output: Value::Null,
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
        };
        if let Some(mock) = &node.mock {
            let raw = mock.result.clone();
            let (status, output) = match apply_transform(node.transform.as_ref(), raw) {
                Ok(reshaped) => (NodeStatus::Ok, reshaped),
                Err(message) => {
                    failed_ancestors.insert(id.clone());
                    (NodeStatus::Err, Value::String(message))
                }
            };
            traces.insert(
                id.clone(),
                NodeTrace {
                    id: id.clone(),
                    status,
                    inputs,
                    output,
                    duration_ms: Some(0),
                    ..Default::default()
                },
            );
            order.push(id.clone());
            continue;
        }
        let server_name = node.server.as_deref().or(default_server).unwrap_or("");
        let server = match pool.get(server_name).await {
            Some(s) => s,
            None => {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs,
                        output: Value::String(format!(
                            "composition node `{id}` references unknown server `{server_name}`"
                        )),
                        duration_ms: None,
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
                order.push(id.clone());
                continue;
            }
        };
        let started = Instant::now();
        let (status, output) = dispatch_single(node, &inputs, server.as_ref()).await;
        let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
        traces.insert(
            id.clone(),
            NodeTrace {
                id: id.clone(),
                status,
                inputs,
                output,
                duration_ms: Some(duration_ms),
                ..Default::default()
            },
        );
        order.push(id.clone());
    }
    let output = assemble_output(&composition.output, &composition.topo_order, &traces);
    let (cost_total, tokens_total) = totals(&traces);
    let budget_tripped = check_budget(composition.budget.as_ref(), cost_total, tokens_total);
    CompositionTrace {
        name: composition.name.clone(),
        order,
        ran: budget_tripped.is_none()
            && !traces
                .values()
                .any(|t| t.status == NodeStatus::Err && t.duration_ms.is_none()),
        nodes: traces,
        output,
        cost_total,
        tokens_total,
        budget_tripped,
    }
}

/// Compute the composition's combined output.
///
/// `None`: fall back to the last topologically-ordered node's
/// output. `OutputSpec::Single`: select that node's output verbatim.
/// `OutputSpec::Assemble`: run the jq program against `{nodes: {<id>:
/// <output>, ...}}` and use the result.
fn assemble_output(
    output: &Option<OutputSpec>,
    topo_order: &[String],
    traces: &BTreeMap<String, NodeTrace>,
) -> Value {
    match output {
        None => topo_order
            .last()
            .and_then(|id| traces.get(id))
            .map(|t| t.output.clone())
            .unwrap_or(Value::Null),
        Some(OutputSpec::Single(id)) => traces
            .get(id)
            .map(|t| t.output.clone())
            .unwrap_or(Value::Null),
        Some(OutputSpec::Assemble(spec)) => {
            let context = json!({
                "nodes": traces
                    .iter()
                    .map(|(k, v)| (k.clone(), v.output.clone()))
                    .collect::<serde_json::Map<_, _>>()
            });
            match crate::transform_jq::run(&spec.assemble, &context) {
                Ok(v) => v,
                Err(e) => Value::String(format!("output.assemble failed: {e}")),
            }
        }
    }
}

fn short_type(value: &Value) -> &'static str {
    match value {
        Value::Null => "null",
        Value::Bool(_) => "boolean",
        Value::Number(_) => "number",
        Value::String(_) => "string",
        Value::Array(_) => "array",
        Value::Object(_) => "object",
    }
}

fn stringify_short(value: &Value) -> String {
    match value {
        Value::String(s) => s.clone(),
        other => other.to_string(),
    }
}

/// Outcome of evaluating a node's `when:` predicate.
enum WhenOutcome {
    /// The predicate returned a truthy value (or no predicate was
    /// declared); dispatch the node normally.
    Run,
    /// The predicate returned a falsy value; mark the node skipped
    /// and propagate the skip to descendants.
    Skip,
    /// The jq program failed to run against the parents context;
    /// surface as a node error so the suite can fail loudly.
    Error(String),
}

/// Evaluate the optional `when:` predicate against a context object
/// built from the node's parents. The jq program runs against
/// `{parents: {<id>: <output>, ...}}` so a predicate can reference
/// `.parents.search.count > 0` without re-implementing reference
/// resolution.
fn evaluate_when(when: Option<&str>, parent_outputs: &HashMap<String, Value>) -> WhenOutcome {
    let Some(predicate) = when else {
        return WhenOutcome::Run;
    };
    let context = json!({
        "parents": parent_outputs.iter().map(|(k, v)| (k.clone(), v.clone())).collect::<serde_json::Map<_, _>>()
    });
    match crate::transform_jq::run(predicate, &context) {
        Ok(value) => {
            if is_truthy(&value) {
                WhenOutcome::Run
            } else {
                WhenOutcome::Skip
            }
        }
        Err(e) => WhenOutcome::Error(format!("when predicate failed: {e}")),
    }
}

/// jq's truthiness rule: every value is truthy except `false` and
/// `null`. Numbers (including zero), empty strings, empty arrays, and
/// empty objects are all truthy. Documented in the composition
/// reference so suite authors do not have to reverse-engineer it.
fn is_truthy(value: &Value) -> bool {
    !matches!(value, Value::Null | Value::Bool(false))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::composition::{CompositionSpec, ResolvedComposition};

    #[test]
    fn when_predicate_with_no_predicate_runs_the_node() {
        let outcome = evaluate_when(None, &HashMap::new());
        assert!(matches!(outcome, WhenOutcome::Run));
    }

    #[test]
    fn when_predicate_returning_true_runs_the_node() {
        let parents = HashMap::from([("a".to_string(), json!({"count": 3}))]);
        let outcome = evaluate_when(Some(".parents.a.count > 0"), &parents);
        assert!(matches!(outcome, WhenOutcome::Run));
    }

    #[test]
    fn when_predicate_returning_false_skips_the_node() {
        let parents = HashMap::from([("a".to_string(), json!({"count": 0}))]);
        let outcome = evaluate_when(Some(".parents.a.count > 0"), &parents);
        assert!(matches!(outcome, WhenOutcome::Skip));
    }

    #[test]
    fn when_predicate_returning_null_skips_the_node() {
        let parents = HashMap::from([("a".to_string(), json!({}))]);
        let outcome = evaluate_when(Some(".parents.a.missing"), &parents);
        assert!(matches!(outcome, WhenOutcome::Skip));
    }

    #[test]
    fn when_predicate_returning_a_number_is_truthy() {
        let outcome = evaluate_when(Some("0"), &HashMap::new());
        // jq truthiness: 0 is truthy. The composition reference
        // documents this so suite authors can rely on it.
        assert!(matches!(outcome, WhenOutcome::Run));
    }

    #[test]
    fn when_predicate_referencing_a_missing_parent_field_returns_null_and_skips() {
        let parents = HashMap::from([("a".to_string(), json!({"v": 1}))]);
        let outcome = evaluate_when(Some(".parents.a.absent"), &parents);
        assert!(matches!(outcome, WhenOutcome::Skip));
    }

    #[test]
    fn is_truthy_table() {
        // Reference table the executor follows. Every case here is
        // documented in compositions.md so a suite author can predict
        // the runtime decision from reading the doc.
        assert!(!is_truthy(&Value::Null));
        assert!(!is_truthy(&Value::Bool(false)));
        assert!(is_truthy(&Value::Bool(true)));
        assert!(is_truthy(&json!(0)));
        assert!(is_truthy(&json!("")));
        assert!(is_truthy(&json!([])));
        assert!(is_truthy(&json!({})));
    }

    #[test]
    fn for_each_beyond_iteration_cap_reports_count_and_limit() {
        // WOR-1313: the guard's diagnostic must carry both the actual
        // fan-out width and the cap so a suite author sees immediately
        // which parent array exploded.
        let message = fan_out_over_limit_message(
            "search",
            MAX_FOR_EACH_ITERATIONS + 1,
            MAX_FOR_EACH_ITERATIONS,
        );
        assert!(message.contains("search"), "message: {message}");
        assert!(message.contains("10001"), "message: {message}");
        assert!(message.contains("10000"), "message: {message}");
        // WOR-1322: a per-node override lands in the same diagnostic.
        let message = fan_out_over_limit_message("search", 3, 2);
        assert!(message.contains("3 iterations"), "message: {message}");
        assert!(message.contains("limit of 2"), "message: {message}");
    }

    #[test]
    fn substitution_error_message_round_trips_the_error_text() {
        // Lightweight contract test on the error rendering: a future
        // change that swaps SubstitutionError to a richer payload must
        // still produce a non-empty diagnostic on the trace.
        let err = SubstitutionError::UnknownParent {
            template: "${x.y}".into(),
            referent: "x".into(),
        };
        assert!(!substitution_error_message(&err).is_empty());
    }

    #[test]
    fn skip_propagation_marks_descendants_of_a_failed_node() {
        // No server pool needed: we drive this through the
        // `failed_ancestors` accumulator alone. Build a composition
        // where b consumes a and a fails; the propagation rule should
        // mark b skipped without dispatching it.
        let spec: CompositionSpec = serde_yaml::from_str(
            r#"
name: chain
nodes:
  - id: a
    tool: t
  - id: b
    needs: [a]
    tool: u
    args: { x: "${a.id}" }
"#,
        )
        .unwrap();
        let (resolved, _) = ResolvedComposition::resolve(&spec).unwrap();
        // Walk the topo order manually with a synthetic failure on a:
        // a is err, b should observe failed_ancestors and skip.
        let mut traces: BTreeMap<String, NodeTrace> = BTreeMap::new();
        let mut failed_ancestors: HashSet<String> = HashSet::new();
        for id in &resolved.topo_order {
            if id == "a" {
                traces.insert(
                    id.clone(),
                    NodeTrace {
                        id: id.clone(),
                        status: NodeStatus::Err,
                        inputs: Value::Null,
                        output: Value::String("synth".into()),
                        duration_ms: Some(0),
                        ..Default::default()
                    },
                );
                failed_ancestors.insert(id.clone());
            } else {
                let node = resolved.nodes.get(id).unwrap();
                let any_parent_failed = node.needs.iter().any(|p| failed_ancestors.contains(p));
                assert!(any_parent_failed, "b should observe a's failure");
            }
        }
    }
}
