apl_core/

evaluator.rs

// Location: ./crates/apl-core/src/evaluator.rs
// Copyright 2026
// SPDX-License-Identifier: Apache-2.0
// Authors: Teryl Taylor
//
// APL evaluator — walks the IR against an AttributeBag and returns a Decision.
//
// The evaluator is sync and infallible by design. Missing attributes resolve
// to `false` (DSL spec §2.6); operator type mismatches resolve to `false`.
// The host drives the four phases separately by calling `evaluate_rules` once
// per declared phase — phase orchestration lives in `apl-cpex`.
//
// Semantics anchored in:
//   - DSL spec apl-dsl-spec.md §2 (operators), §3 (actions), §8.1 (require)
//   - apl-design.md §7 (native fast-path, sync inside async outer)

use std::sync::Arc;

use crate::attributes::{AttributeBag, AttributeValue};
use crate::pipeline::{Pipeline, ScanKind, Stage, TaintEvent, TaintScope, TypeCheck};
use crate::rules::{CompareOp, Condition, Effect, Expression, Literal, Rule};
use crate::step::{PdpResolver, PluginInvocation, PluginInvoker};

/// Outcome of evaluating a phase's rule list.
#[derive(Debug, Clone, PartialEq)]
pub enum Decision {
    /// No `deny` rule fired. Pipeline proceeds.
    Allow,
    /// A `deny` rule fired. Pipeline halts.
    Deny {
        reason: Option<String>,
        /// `Rule.source` of the rule that produced the deny — for audit logs.
        rule_source: String,
    },
}

/// Evaluate a phase's rules against the bag.
///
/// Spec §3 semantics:
/// - First `deny` halts; subsequent rules / effects don't run.
/// - `allow` effects *do not* short-circuit — evaluation continues to
///   the next effect (then to the next rule).
/// - If no rule denies, the phase resolves to `Decision::Allow`.
///
/// Sync fast path — only handles control effects (`Allow` / `Deny`).
/// Rules containing `Plugin` / `Delegate` / `Taint` effects must go
/// through [`evaluate_steps`] instead, which has the async invoker
/// traits wired up. This function silently skips non-control effects
/// so a rule list mixed with `Plugin` still terminates cleanly on a
/// later `Deny` — but the side effects don't fire. Caller's job to
/// pick the right entry point for the effects in the rules.
pub fn evaluate_rules(rules: &[Rule], bag: &AttributeBag) -> Decision {
    for rule in rules {
        if !eval_expression(&rule.condition, bag) {
            continue;
        }
        for effect in &rule.effects {
            match effect {
                Effect::Allow => continue,
                Effect::Deny { reason, code } => {
                    // `code` override on the effect takes precedence
                    // over the auto-generated rule source position,
                    // so author-stable categories survive YAML edits.
                    let rule_source = code.clone().unwrap_or_else(|| rule.source.clone());
                    return Decision::Deny {
                        reason: reason.clone(),
                        rule_source,
                    };
                },
                // Plugin / Delegate / Taint require the async step
                // path; ignore here. See doc comment above.
                _ => continue,
            }
        }
    }
    Decision::Allow
}

fn eval_expression(expr: &Expression, bag: &AttributeBag) -> bool {
    match expr {
        Expression::Condition(c) => eval_condition(c, bag),
        Expression::And(parts) => parts.iter().all(|e| eval_expression(e, bag)),
        Expression::Or(parts) => parts.iter().any(|e| eval_expression(e, bag)),
        Expression::Not(inner) => !eval_expression(inner, bag),
        Expression::Always => true,
    }
}

fn eval_condition(cond: &Condition, bag: &AttributeBag) -> bool {
    match cond {
        Condition::IsTrue { key } => bag.get_bool(key).unwrap_or(false),
        Condition::IsFalse { key } => !bag.get_bool(key).unwrap_or(false),
        Condition::Exists { key } => bag.contains(key),
        Condition::Comparison { key, op, value } => eval_comparison(key, *op, value, bag),
        Condition::InSet {
            value_key,
            set_key,
            negate,
        } => {
            let in_set = match (bag.get_string(value_key), bag.get_string_set(set_key)) {
                (Some(s), Some(set)) => set.contains(s),
                _ => false, // missing key or wrong type → not in set
            };
            if *negate {
                !in_set
            } else {
                in_set
            }
        },
    }
}

fn eval_comparison(key: &str, op: CompareOp, lit: &Literal, bag: &AttributeBag) -> bool {
    let attr = match bag.get(key) {
        Some(v) => v,
        None => return false, // missing → false (spec §2.6)
    };

    match op {
        CompareOp::Contains => match (attr, lit) {
            (AttributeValue::StringSet(_), Literal::String(s)) => bag.set_contains(key, s),
            _ => false,
        },
        CompareOp::Eq => values_eq(attr, lit),
        CompareOp::NotEq => !values_eq(attr, lit),
        CompareOp::Gt | CompareOp::GtEq | CompareOp::Lt | CompareOp::LtEq => {
            numeric_compare(attr, lit, op)
        },
    }
}

fn values_eq(attr: &AttributeValue, lit: &Literal) -> bool {
    match (attr, lit) {
        (AttributeValue::Bool(a), Literal::Bool(b)) => a == b,
        (AttributeValue::Int(a), Literal::Int(b)) => a == b,
        (AttributeValue::Float(a), Literal::Float(b)) => a == b,
        (AttributeValue::String(a), Literal::String(b)) => a == b,
        // Int↔Float promotion for equality (matches AttributeBag::get_float).
        (AttributeValue::Int(a), Literal::Float(b)) => (*a as f64) == *b,
        (AttributeValue::Float(a), Literal::Int(b)) => *a == (*b as f64),
        _ => false,
    }
}

fn numeric_compare(attr: &AttributeValue, lit: &Literal, op: CompareOp) -> bool {
    // Coerce both operands to f64 for the order comparison. Numeric-looking
    // strings are coerced — LLM tool arguments routinely arrive as strings
    // (e.g. `"amount": "25000"`), and a policy author writing
    // `args.amount > 10000` plainly means a numeric comparison. A string
    // that doesn't parse as a number is genuinely non-numeric → false
    // (order operators don't apply, spec §2.3).
    let a = match coerce_f64_attr(attr) {
        Some(a) => a,
        None => return false,
    };
    let b = match coerce_f64_lit(lit) {
        Some(b) => b,
        None => return false,
    };
    match op {
        CompareOp::Gt => a > b,
        CompareOp::GtEq => a >= b,
        CompareOp::Lt => a < b,
        CompareOp::LtEq => a <= b,
        _ => unreachable!("numeric_compare called with non-numeric op"),
    }
}

/// Coerce a bag attribute to `f64` for an order comparison: numbers pass
/// through; a string is parsed (numeric-looking strings only); anything
/// else is non-numeric.
fn coerce_f64_attr(attr: &AttributeValue) -> Option<f64> {
    match attr {
        AttributeValue::Int(a) => Some(*a as f64),
        AttributeValue::Float(a) => Some(*a),
        AttributeValue::String(s) => s.trim().parse::<f64>().ok(),
        _ => None,
    }
}

/// Coerce a literal to `f64` for an order comparison. Same rules as
/// [`coerce_f64_attr`] so `args.x > "10"` works symmetrically.
fn coerce_f64_lit(lit: &Literal) -> Option<f64> {
    match lit {
        Literal::Int(b) => Some(*b as f64),
        Literal::Float(b) => Some(*b),
        Literal::String(s) => s.trim().parse::<f64>().ok(),
        _ => None,
    }
}

/// Heuristic: does `s` look like a bag attribute reference (e.g.
/// `claim.manager`, `user.sub`) rather than a literal identity (e.g.
/// `alice@corp.com`, a bare username)? Used to decide whether an
/// unresolved elicitation `from` should fail closed (an attribute that
/// didn't resolve) or pass through as a literal.
///
/// A reference is a dotted path of identifier characters only — starts
/// with a letter and contains only `[A-Za-z0-9_.]` with at least one dot.
/// Literals like emails (contain `@`) or display names (contain spaces)
/// are excluded, so they still pass through verbatim.
fn looks_like_attribute_ref(s: &str) -> bool {
    s.contains('.')
        && s.starts_with(|c: char| c.is_ascii_alphabetic())
        && s.chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '.')
}

// =====================================================================
// Async effect evaluator (policy: / post_policy: walks Vec<Effect>)
// =====================================================================

/// Walk an Effect list against the bag, dispatching PDP calls via `pdp`
/// and plugin invocations via `plugins`. Returns the phase's overall
/// decision.
///
/// Semantics (DSL §3, §7.5):
/// - `Effect::When` — evaluate the condition; if true, run the body in
///   order with the same first-deny-wins logic.
/// - `Effect::Pdp` — call resolver; on Allow run `on_allow` reactions and
///   continue; on Deny run `on_deny` reactions and return the deny
///   (reactions can override with their own deny, but cannot turn a deny
///   into an allow).
/// - `Effect::Plugin` — invoke; Allow continues, Deny returns.
/// - `Effect::Delegate` — mint downstream credential; writes
///   `delegation.granted.*` keys back into the bag; deny-on-failure unless
///   the step's `on_error` overrides.
/// - `Effect::Taint` — record the label; never halts.
/// - `Effect::FieldOp` — apply a pipe chain to `args.X` / `result.X`;
///   may set `args_modified` / `result_modified`.
/// - `Effect::Sequential` — run children in order, halt on first Deny.
/// - `Effect::Parallel` — run children concurrently, abort on first Deny.
/// - `Effect::Allow` — explicit no-op; continues the phase.
/// - `Effect::Deny` — halt with the supplied reason/code.
///
/// PDP / plugin errors map to a Deny with the error in the reason, per
/// the design's fail-closed default (DSL §8.9). Pre-E4 `evaluate_steps`
/// is preserved as a deprecated alias that forwards here.
#[allow(clippy::too_many_arguments)]
pub async fn evaluate_effects(
    effects: &[Effect],
    bag: &mut AttributeBag,
    pdp: &Arc<dyn PdpResolver>,
    plugins: &Arc<dyn PluginInvoker>,
    delegations: &Arc<dyn crate::step::DelegationInvoker>,
    elicitations: &Arc<dyn crate::step::ElicitationInvoker>,
    phase: crate::step::DispatchPhase,
    payload: &mut crate::route::RoutePayload,
) -> StepsEvaluation {
    let mut taints: Vec<crate::pipeline::TaintEvent> = Vec::new();
    let mut args_modified = false;
    let mut result_modified = false;
    let mut pending: Option<crate::step::PendingElicitation> = None;
    for effect in effects {
        // Each top-level effect runs against the shared mutable state.
        // `Effect::When` / `Effect::Pdp` handle their own internal
        // walking via dispatch_effect's recursive call.
        let fallback_source = match effect {
            Effect::When { source, .. } => source.as_str(),
            _ => "",
        };
        match Box::pin(dispatch_effect(
            effect,
            fallback_source,
            bag,
            pdp,
            plugins,
            delegations,
            elicitations,
            phase,
            &mut taints,
            &mut args_modified,
            &mut result_modified,
            payload,
        ))
        .await
        {
            EffectOutcome::Continue => {},
            EffectOutcome::Halt(decision) => {
                return StepsEvaluation::deny(decision, taints, args_modified, result_modified);
            },
            EffectOutcome::Pending(bundle) => {
                // Suspend the phase: stop walking (sequential elicitation),
                // carry the bundle out. Decision stays Allow — the host
                // gates on `pending` being set, not on a deny.
                pending = Some(bundle);
                break;
            },
        }
    }
    StepsEvaluation {
        decision: Decision::Allow,
        taints,
        args_modified,
        result_modified,
        pending,
    }
}

/// Outcome of `evaluate_effects`: the phase's decision plus taints emitted
/// by any plugin steps that ran. Taints are accumulated even when the
/// phase ultimately denies — audit needs to see what the plugins
/// reported before the deny landed. Empty `taints` is the common case
/// (most steps are predicates / PDP calls, not label emitters).
///
/// `args_modified` / `result_modified` are set when an `Effect::FieldOp`
/// inside a `do:` body successfully mutated the route payload — the
/// orchestrator uses them to OR-into the route-level "did anything
/// change" signals so the host knows to re-serialize the body.
#[derive(Debug, Clone)]
pub struct StepsEvaluation {
    pub decision: Decision,
    pub taints: Vec<crate::pipeline::TaintEvent>,
    pub args_modified: bool,
    pub result_modified: bool,
    /// Set when a phase suspended on an unresolved elicitation. `Some`
    /// means "do not forward — emit `-32120` with this bundle." `decision`
    /// is `Allow` in that case (nothing denied); the host gates forwarding
    /// on `pending.is_none()`. See [`crate::step::PendingElicitation`].
    pub pending: Option<crate::step::PendingElicitation>,
}

impl StepsEvaluation {
    fn deny(
        d: Decision,
        taints: Vec<crate::pipeline::TaintEvent>,
        args_modified: bool,
        result_modified: bool,
    ) -> Self {
        Self {
            decision: d,
            taints,
            args_modified,
            result_modified,
            pending: None,
        }
    }
}

/// Outcome of dispatching one effect. Internal control-flow signal —
/// never serialized, never exposed in the IR. Sits between the per-
/// effect dispatch (When / Pdp / Plugin / Delegate / Taint / Allow /
/// Deny / FieldOp / Sequential / Parallel) and the caller's "do I keep
/// walking the effects list or halt?" loop.
enum EffectOutcome {
    /// Effect completed without producing a Deny — caller moves on to
    /// the next effect in the surrounding list.
    Continue,
    /// Effect produced a Deny decision — caller halts the rest of the
    /// surrounding list, the rest of the phase, and the route.
    Halt(Decision),
    /// Effect dispatched an elicitation that hasn't resolved yet — the
    /// phase *suspends*. Like `Halt` it short-circuits the surrounding
    /// list, but it is NOT a deny: the carried bundle propagates up to
    /// the host, which emits `-32120` (retry) instead of forwarding. The
    /// phase decision stays `Allow`; pending being non-empty is what
    /// blocks the forward (see [`PendingElicitation`]).
    Pending(crate::step::PendingElicitation),
}

/// Run a single effect against the evaluator's state. Called by both
/// `evaluate_effects` (top-level walk of `policy:` / `post_policy:`)
/// and by recursive arms (Sequential, Parallel, When body, Pdp
/// reactions), so there's exactly one place that knows how each
/// effect kind dispatches.
///
/// `fallback_source` is the rule-source-position string used as the
/// `rule_source` field on a `Decision::Deny` when the effect itself
/// doesn't carry an explicit code (i.e. `Effect::Deny { code: None }`,
/// or a deny coming back from a plugin / delegator without overriding
/// the default).
#[allow(clippy::too_many_arguments)]
async fn dispatch_effect(
    effect: &Effect,
    fallback_source: &str,
    bag: &mut AttributeBag,
    pdp: &Arc<dyn PdpResolver>,
    plugins: &Arc<dyn PluginInvoker>,
    delegations: &Arc<dyn crate::step::DelegationInvoker>,
    elicitations: &Arc<dyn crate::step::ElicitationInvoker>,
    phase: crate::step::DispatchPhase,
    taints: &mut Vec<crate::pipeline::TaintEvent>,
    args_modified: &mut bool,
    result_modified: &mut bool,
    payload: &mut crate::route::RoutePayload,
) -> EffectOutcome {
    match effect {
        Effect::Allow => EffectOutcome::Continue,

        Effect::Deny { reason, code } => {
            // Author-supplied code overrides the auto-generated source
            // position. Lets MCP clients dispatch on stable categories
            // (`quota.exceeded`) rather than positional codes that
            // shift with YAML edits.
            let rule_source = code.clone().unwrap_or_else(|| fallback_source.to_string());
            EffectOutcome::Halt(Decision::Deny {
                reason: reason.clone(),
                rule_source,
            })
        },

        Effect::Plugin { name } => {
            match plugins
                .invoke(name, bag, PluginInvocation::Step { phase })
                .await
            {
                Ok(outcome) => {
                    // Plugins can emit taints regardless of decision —
                    // collect first, then act on the decision.
                    taints.extend(outcome.taints);
                    match outcome.decision {
                        Decision::Allow => EffectOutcome::Continue,
                        deny @ Decision::Deny { .. } => EffectOutcome::Halt(deny),
                    }
                },
                Err(e) => EffectOutcome::Halt(Decision::Deny {
                    reason: Some(format!("plugin `{}` error: {}", name, e)),
                    rule_source: format!("plugin:{}", name),
                }),
            }
        },

        Effect::Delegate(delegate_step) => {
            match delegations.delegate(delegate_step).await {
                Ok(outcome) => match &outcome.decision {
                    Decision::Allow => {
                        // Surface granted_* keys into the bag so
                        // downstream rules in this same step list can
                        // read them (`require(delegation.granted.permissions
                        // contains "X")`, etc.).
                        use crate::attributes::AttributeValue;
                        use crate::step::delegation_bag_keys as bk;

                        bag.set(bk::GRANTED, AttributeValue::Bool(true));
                        if !outcome.granted_permissions.is_empty() {
                            let set: std::collections::HashSet<String> =
                                outcome.granted_permissions.iter().cloned().collect();
                            bag.set(bk::GRANTED_PERMISSIONS, AttributeValue::StringSet(set));
                        }
                        if let Some(aud) = &outcome.granted_audience {
                            bag.set(bk::GRANTED_AUDIENCE, aud.clone());
                        }
                        if let Some(exp) = &outcome.granted_expires_at {
                            bag.set(bk::GRANTED_EXPIRES_AT, exp.clone());
                        }
                        EffectOutcome::Continue
                    },
                    Decision::Deny { .. } => {
                        // Apply the step's on_error policy. Default
                        // ("deny") halts; "continue" lets the pipeline
                        // keep going so subsequent rules can branch on
                        // the absent `delegation.granted` flag.
                        let on_error = delegate_step
                            .on_error
                            .as_deref()
                            .unwrap_or("deny")
                            .to_ascii_lowercase();
                        if on_error == "continue" {
                            EffectOutcome::Continue
                        } else {
                            EffectOutcome::Halt(outcome.decision)
                        }
                    },
                },
                Err(e) => {
                    // Transport / lookup failure. on_error treats this
                    // the same way as a plugin-side deny.
                    let on_error = delegate_step
                        .on_error
                        .as_deref()
                        .unwrap_or("deny")
                        .to_ascii_lowercase();
                    if on_error == "continue" {
                        EffectOutcome::Continue
                    } else {
                        EffectOutcome::Halt(Decision::Deny {
                            reason: Some(format!(
                                "delegate `{}` error: {}",
                                delegate_step.plugin_name, e
                            )),
                            rule_source: delegate_step.source.clone(),
                        })
                    }
                },
            }
        },

        Effect::Elicit(elicit_step) => dispatch_elicitation(elicit_step, bag, elicitations).await,

        Effect::Taint { label, scopes } => {
            // Emit the taint into the phase's accumulator so it flows
            // into `RouteDecision.taints`. Apl-cpex's invoker handles
            // the session-store persistence side at request end — here
            // we only record the event. Scopes come straight from the
            // parser (`taint(label, session, message)` syntax).
            taints.push(crate::pipeline::TaintEvent {
                label: label.clone(),
                scopes: scopes.clone(),
            });
            EffectOutcome::Continue
        },

        Effect::FieldOp { path, stages } => {
            dispatch_field_op(
                path,
                stages,
                fallback_source,
                bag,
                plugins,
                phase,
                taints,
                args_modified,
                result_modified,
                payload,
            )
            .await
        },

        Effect::Sequential(effects) => {
            // Semantically the same as inlining the list into the
            // enclosing scope — walk in order, stop on first Halt.
            // The variant exists for explicit grouping and to pair
            // with `Parallel` in the IR.
            for inner in effects {
                match Box::pin(dispatch_effect(
                    inner,
                    fallback_source,
                    bag,
                    pdp,
                    plugins,
                    delegations,
                    elicitations,
                    phase,
                    taints,
                    args_modified,
                    result_modified,
                    payload,
                ))
                .await
                {
                    EffectOutcome::Continue => continue,
                    // Halt (deny) and Pending (suspend) both propagate up.
                    other => return other,
                }
            }
            EffectOutcome::Continue
        },

        Effect::Parallel(effects) => {
            // `dispatch_parallel` returns an explicit `BoxFuture<'_, _>`
            // (Send by construction) so the recursive
            // dispatch_effect → dispatch_parallel → dispatch_effect
            // chain doesn't trip the compiler's Send-inference cycle.
            dispatch_parallel(
                effects,
                fallback_source,
                bag,
                pdp,
                plugins,
                delegations,
                elicitations,
                phase,
                taints,
                payload,
            )
            .await
        },

        Effect::When {
            condition,
            body,
            source,
        } => {
            // Predicate-gated body — replaces the historical
            // `Step::Rule`. Skip silently when the condition is false;
            // otherwise walk the body in order and halt on first Deny.
            if !eval_expression(condition, bag) {
                return EffectOutcome::Continue;
            }
            for inner in body {
                match Box::pin(dispatch_effect(
                    inner,
                    source,
                    bag,
                    pdp,
                    plugins,
                    delegations,
                    elicitations,
                    phase,
                    taints,
                    args_modified,
                    result_modified,
                    payload,
                ))
                .await
                {
                    EffectOutcome::Continue => continue,
                    // Halt (deny) and Pending (suspend) both propagate up.
                    other => return other,
                }
            }
            EffectOutcome::Continue
        },

        Effect::Pdp {
            call,
            on_allow,
            on_deny,
        } => {
            // External PDP call — replaces `Step::Pdp`. Reactions run
            // through the same dispatch_effect path (recursively).
            match pdp.evaluate(call, bag).await {
                Ok(pdp_result) => match pdp_result.decision {
                    Decision::Allow => {
                        // Walk on_allow; if it ends without a Halt the
                        // PDP allow stands and we continue.
                        for inner in on_allow {
                            match Box::pin(dispatch_effect(
                                inner,
                                fallback_source,
                                bag,
                                pdp,
                                plugins,
                                delegations,
                                elicitations,
                                phase,
                                taints,
                                args_modified,
                                result_modified,
                                payload,
                            ))
                            .await
                            {
                                EffectOutcome::Continue => continue,
                                // Halt (deny) and Pending (suspend) propagate.
                                other => return other,
                            }
                        }
                        EffectOutcome::Continue
                    },
                    deny @ Decision::Deny { .. } => {
                        // Reactions can override the PDP's deny reason
                        // (e.g. `on_deny: [deny "..."]`) but cannot
                        // upgrade the deny to allow — if reactions
                        // walked clean, the PDP's original deny stands.
                        for inner in on_deny {
                            match Box::pin(dispatch_effect(
                                inner,
                                fallback_source,
                                bag,
                                pdp,
                                plugins,
                                delegations,
                                elicitations,
                                phase,
                                taints,
                                args_modified,
                                result_modified,
                                payload,
                            ))
                            .await
                            {
                                EffectOutcome::Continue => {},
                                // A reaction can override the deny reason;
                                // a suspended reaction (Pending) surfaces.
                                other => return other,
                            }
                        }
                        EffectOutcome::Halt(deny)
                    },
                },
                Err(e) => EffectOutcome::Halt(Decision::Deny {
                    reason: Some(format!("PDP error: {}", e)),
                    rule_source: format!("pdp:{:?}", call.dialect),
                }),
            }
        },
    }
}

/// Drive one [`Effect::Elicit`] step: dispatch on first arrival, check
/// status on every pass, and on an approved-and-genuine response apply
/// the runtime's `scope`-over-args sufficiency check before allowing the
/// phase to continue. See `docs/apl-manager-approval-ciba-design.md`.
///
/// Failure handling follows the step's `on_error` (default `deny`,
/// fail-closed). An explicit human *denial* always halts regardless of
/// `on_error` — `on_error` governs channel/validation *failures*, not a
/// valid "no" from the approver.
///
/// While the human hasn't answered, the phase *suspends*: the step yields
/// an [`EffectOutcome::Pending`] carrying the bundle the host turns into a
/// JSON-RPC `-32120` (retry) instead of forwarding. The bag also records
/// `elicitation.status = pending` so the host/audit can see the in-flight
/// state.
async fn dispatch_elicitation(
    step: &crate::step::ElicitStep,
    bag: &mut AttributeBag,
    elicitations: &Arc<dyn crate::step::ElicitationInvoker>,
) -> EffectOutcome {
    use crate::step::{elicitation_bag_keys as bk, ElicitationOutcome, ElicitationStatus};

    // `on_error` applies to *failures* (channel error, invalid response,
    // expiry, still-pending) — not to a genuine denial.
    let on_error_continue = step
        .on_error
        .as_deref()
        .unwrap_or("deny")
        .eq_ignore_ascii_case("continue");
    let fail = |reason: String| -> EffectOutcome {
        if on_error_continue {
            EffectOutcome::Continue
        } else {
            EffectOutcome::Halt(Decision::Deny {
                reason: Some(reason),
                rule_source: step.source.clone(),
            })
        }
    };

    // First arrival vs. retry: the agent echoes `elicitation.id` on
    // retry. Absent → first arrival: dispatch and record the pending
    // bundle in the bag.
    //
    // Resolve `from` against the bag (e.g. `claim.manager` → the
    // manager's identity). A `from` written as a literal identity (e.g.
    // `alice@corp.com`) that isn't a bag key falls through to the literal.
    // But a `from` that *looks like* an unresolved attribute reference
    // (e.g. `claim.manager` when the claim is absent) must NOT be sent to
    // the channel verbatim — that would dispatch an approval to a bogus
    // `login_hint`. This is an auth boundary: fail closed instead. The
    // attribute vocabulary lives here in the runtime, so the invoker
    // receives the resolved identity rather than re-deriving it.
    let resolved_from = match bag.get_string(&step.from) {
        Some(v) => v.to_string(),
        None if looks_like_attribute_ref(&step.from) => {
            return fail(format!(
                "elicitation `from` attribute `{}` did not resolve to an identity",
                step.from
            ));
        },
        None => step.from.clone(),
    };
    let id = match bag.get_string(bk::ID) {
        Some(existing) => existing.to_string(),
        None => match elicitations.dispatch(step, &resolved_from).await {
            Ok(d) => {
                bag.set(bk::ID, d.id.clone());
                bag.set(bk::STATUS, "pending".to_string());
                // Optional audit label — not a routing key.
                if let Some(channel) = &step.channel {
                    bag.set(bk::CHANNEL, channel.clone());
                }
                if let Some(approver) = &d.approver {
                    bag.set(bk::APPROVER, approver.clone());
                }
                if let Some(intent) = &d.intent_id {
                    bag.set(bk::INTENT_ID, intent.clone());
                }
                if let Some(exp) = &d.expires_at {
                    bag.set(bk::EXPIRES_AT, exp.clone());
                }
                d.id
            },
            Err(e) => return fail(format!("elicitation dispatch failed: {e}")),
        },
    };

    // Status check — non-blocking read of the channel's current state.
    let status = match elicitations.check(step, &id).await {
        Ok(s) => s,
        Err(e) => return fail(format!("elicitation check failed: {e}")),
    };

    match status {
        ElicitationStatus::Pending => {
            // Suspend, don't deny: hand the host a pending bundle so it
            // emits `-32120` (retry) instead of forwarding. Built from the
            // bag, which dispatch populated — works the same on first
            // arrival and on a later still-pending retry.
            bag.set(bk::STATUS, "pending".to_string());
            let approver = bag.get_string(bk::APPROVER).map(str::to_string);
            let intent_id = bag.get_string(bk::INTENT_ID).map(str::to_string);
            let expires_at = bag.get_string(bk::EXPIRES_AT).map(str::to_string);
            EffectOutcome::Pending(crate::step::PendingElicitation {
                id,
                plugin_name: step.plugin_name.clone(),
                approver,
                intent_id,
                channel: step.channel.clone(),
                expires_at,
                source: step.source.clone(),
            })
        },
        ElicitationStatus::Expired => {
            bag.set(bk::STATUS, "expired".to_string());
            fail("elicitation expired before a response".to_string())
        },
        ElicitationStatus::Resolved {
            outcome: ElicitationOutcome::Denied,
        } => {
            bag.set(bk::STATUS, "resolved".to_string());
            bag.set(bk::OUTCOME, "denied".to_string());
            // A genuine "no" halts unconditionally — not subject to
            // `on_error`.
            EffectOutcome::Halt(Decision::Deny {
                reason: Some("elicitation denied by approver".to_string()),
                rule_source: step.source.clone(),
            })
        },
        ElicitationStatus::Resolved {
            outcome: ElicitationOutcome::Approved,
        } => {
            // Genuineness (plugin): signature / intent binding /
            // responder identity.
            let validation = match elicitations.validate(step, &id).await {
                Ok(v) => v,
                Err(e) => return fail(format!("elicitation validation failed: {e}")),
            };
            if !validation.valid {
                let why = validation
                    .reason
                    .unwrap_or_else(|| "response failed validation".to_string());
                return fail(format!("elicitation invalid: {why}"));
            }

            // Sufficiency (runtime): evaluate `scope` against the live
            // args. Kept in APL because the channel (Keycloak CIBA) can't
            // bind args — no RFC 9396 RAR.
            if let Some(scope_src) = &step.scope {
                match crate::parser::parse_predicate(scope_src) {
                    Ok(expr) => {
                        if !eval_expression(&expr, bag) {
                            return fail(format!("elicitation scope not satisfied: `{scope_src}`"));
                        }
                    },
                    Err(e) => {
                        return fail(format!(
                            "elicitation scope `{scope_src}` failed to parse: {e}"
                        ));
                    },
                }
            }

            // Approved + genuine + sufficient. Record resolved facts for
            // downstream rules / audit, then continue.
            bag.set(bk::STATUS, "resolved".to_string());
            bag.set(bk::OUTCOME, "approved".to_string());
            if let Some(approver) = validation.approver {
                bag.set(bk::APPROVER, approver);
            }
            if let Some(intent) = validation.intent_id {
                bag.set(bk::INTENT_ID, intent);
            }
            EffectOutcome::Continue
        },
    }
}

/// Run a list of effects concurrently. Each branch gets its own
/// cloned bag and payload — mutations inside a branch don't
/// propagate back to the shared outer state. Taints from every
/// branch are merged into the outer `taints` vec (taints are
/// append-only event logs, safe to concatenate). First Halt by
/// branch index wins; the remaining branches are aborted via
/// `cpex_orchestration::run_branches`'s `short_circuit_on_deny`.
///
/// Config-load already rejected `FieldOp` / `Delegate` here via
/// [`Effect::validate_parallel_purity`], so at runtime we trust the
/// IR not to contain mutation effects.
///
/// # Concurrency model (E3.2)
///
/// Built on [`cpex_orchestration::run_branches`] — the same JoinSet
/// + abort-on-deny primitive `cpex-core`'s executor uses for its
/// concurrent phase. Each branch is `tokio::spawn`ed onto the
/// runtime, so branches get true OS-thread parallelism (vs. the v1
/// implementation's `join_all`, which only interleaved on one
/// task). To meet the `'static + Send` bounds for spawning, the
/// invoker references are `&Arc<dyn ...>` — we `Arc::clone` an
/// owned reference into each branch closure.
///
/// Note: no per-branch timeout. The DSL doesn't expose one, and
/// plugin-level timeouts upstream of this call (in cpex-core's
/// executor) bound individual plugin invocations. If a route ever
/// needs a per-branch budget the orchestration crate already
/// supports `BranchConfig::timeout_per_branch` — wire it through a
/// `Effect::Parallel` extension if/when needed.
// Returns an explicit `BoxFuture` rather than `impl Future` so the
// caller (`dispatch_effect`'s `Effect::Parallel` arm, which is itself
// `async fn`) can break the Send-inference cycle this would otherwise
// introduce: dispatch_effect's opaque return type would depend on
// dispatch_parallel's, and dispatch_parallel spawns futures that
// recursively re-enter dispatch_effect. A concrete `BoxFuture` is
// `Pin<Box<dyn Future + Send + 'a>>` — already Send by construction,
// no inference required.
fn dispatch_parallel<'a>(
    effects: &'a [Effect],
    fallback_source: &'a str,
    bag: &'a AttributeBag,
    pdp: &'a Arc<dyn PdpResolver>,
    plugins: &'a Arc<dyn PluginInvoker>,
    delegations: &'a Arc<dyn crate::step::DelegationInvoker>,
    elicitations: &'a Arc<dyn crate::step::ElicitationInvoker>,
    phase: crate::step::DispatchPhase,
    taints: &'a mut Vec<crate::pipeline::TaintEvent>,
    payload: &'a crate::route::RoutePayload,
) -> futures::future::BoxFuture<'a, EffectOutcome> {
    Box::pin(async move {
        use cpex_orchestration::{run_branches, BranchConfig, BranchOutcome, ErasedBranch};

        if effects.is_empty() {
            return EffectOutcome::Continue;
        }

        // Build one spawn-ready branch future per effect. Each branch
        // owns:
        //   * a cloned bag and payload — branch mutations stay local;
        //   * cloned Arcs to the invokers — `'static + Send`, ready for
        //     `tokio::spawn`;
        //   * an owned copy of the effect to evaluate (clone is cheap
        //     for the variants `Parallel` can hold: Allow, Deny, Plugin,
        //     Taint, Sequential, Parallel, When, Pdp).
        let mut branches: Vec<ErasedBranch<(EffectOutcome, Vec<crate::pipeline::TaintEvent>)>> =
            Vec::with_capacity(effects.len());
        for effect in effects.iter() {
            let effect = effect.clone();
            let fallback = fallback_source.to_string();
            let mut branch_bag = bag.clone();
            let mut branch_payload = payload.clone();
            let pdp = Arc::clone(pdp);
            let plugins = Arc::clone(plugins);
            let delegations = Arc::clone(delegations);
            let elicitations = Arc::clone(elicitations);
            branches.push(Box::pin(async move {
                let mut branch_taints: Vec<crate::pipeline::TaintEvent> = Vec::new();
                let mut branch_args_modified = false;
                let mut branch_result_modified = false;
                let outcome = Box::pin(dispatch_effect(
                    &effect,
                    &fallback,
                    &mut branch_bag,
                    &pdp,
                    &plugins,
                    &delegations,
                    &elicitations,
                    phase,
                    &mut branch_taints,
                    &mut branch_args_modified,
                    &mut branch_result_modified,
                    &mut branch_payload,
                ))
                .await;
                (outcome, branch_taints)
            }));
        }

        // `is_deny` short-circuits the moment any branch returns
        // `EffectOutcome::Halt(_)`. The remaining branches get
        // `BranchOutcome::Aborted` and we drop their (already-cancelled)
        // futures. Taints from already-completed branches still land.
        let cfg = BranchConfig {
            timeout_per_branch: None,
            short_circuit_on_deny: true,
        };
        let outcomes = run_branches(
            branches,
            cfg,
            |v: &(EffectOutcome, Vec<crate::pipeline::TaintEvent>)| {
                matches!(v.0, EffectOutcome::Halt(_))
            },
        )
        .await;

        // Aggregate in input order: append every branch's taints; pick
        // the first Halt (by branch index, not wall-clock order) as the
        // overall result. Aborted / panicked branches contribute no
        // taints — they didn't run to completion. A panicked branch is
        // *not* converted into a Halt; we log via `tracing::warn!` and
        // continue. (A misbehaving plugin shouldn't take down the
        // parallel block any more than it would the host process.)
        let mut first_halt: Option<Decision> = None;
        for (idx, outcome) in outcomes.into_iter().enumerate() {
            match outcome {
                BranchOutcome::Completed((effect_outcome, branch_taints)) => {
                    taints.extend(branch_taints);
                    if first_halt.is_none() {
                        if let EffectOutcome::Halt(d) = effect_outcome {
                            first_halt = Some(d);
                        }
                    }
                },
                BranchOutcome::Aborted => {
                    // Short-circuit cancelled this branch — intentional,
                    // no diagnostic needed.
                },
                BranchOutcome::TimedOut => {
                    // Unreachable today (no per-branch timeout
                    // configured). Treat as a no-op if it ever fires
                    // post-config-extension.
                },
                BranchOutcome::Panicked(msg) => {
                    // A panicking branch is a misbehaving plugin/effect;
                    // dropping its output (no Halt, no taints) keeps the
                    // parallel block's other branches intact rather than
                    // taking the whole block down. apl-core has no
                    // tracing dep — host integrations that care can
                    // surface the panic via cpex-core's plugin error
                    // path. `idx`/`msg` are eaten here.
                    let _ = (idx, msg);
                },
            }
        }

        match first_halt {
            Some(d) => EffectOutcome::Halt(d),
            None => EffectOutcome::Continue,
        }
    })
}

/// Apply a `FieldOp` effect — resolve the path in args/result, run
/// the pipeline stages, write the outcome back into the payload.
///
/// Out-of-phase ops are silent no-ops: a Pre-phase rule with
/// `result.X | redact` skips because the result hasn't been produced
/// yet; a Post-phase rule with `args.X | redact` skips because the
/// args were already sent on the wire. This is intentional so the
/// same `when:`/`do:` rule body can be reused across phases without
/// the author needing to branch on phase.
///
/// Missing fields skip silently too (same as the args:/result: phase
/// pipelines) — a pipeline can't transform what isn't there. If the
/// author needs presence semantics, that's a `require(exists(args.X))`
/// upstream of the `do:` body.
#[allow(clippy::too_many_arguments)]
async fn dispatch_field_op(
    path: &str,
    stages: &[crate::pipeline::Stage],
    fallback_source: &str,
    bag: &mut AttributeBag,
    plugins: &Arc<dyn PluginInvoker>,
    phase: crate::step::DispatchPhase,
    taints: &mut Vec<crate::pipeline::TaintEvent>,
    args_modified: &mut bool,
    result_modified: &mut bool,
    payload: &mut crate::route::RoutePayload,
) -> EffectOutcome {
    use crate::route::{get_dotted, remove_dotted, set_dotted};
    use crate::step::DispatchPhase;

    // Pick the right side of the payload based on the path prefix.
    // Out-of-phase ops drop silently (see the doc comment).
    enum Side {
        Args,
        Result,
    }
    let (root, subpath, side) = if let Some(rest) = path.strip_prefix("args.") {
        if !matches!(phase, DispatchPhase::Pre) {
            return EffectOutcome::Continue;
        }
        (&mut payload.args, rest, Side::Args)
    } else if let Some(rest) = path.strip_prefix("result.") {
        if !matches!(phase, DispatchPhase::Post) {
            return EffectOutcome::Continue;
        }
        let Some(result) = payload.result.as_mut() else {
            return EffectOutcome::Continue;
        };
        (result, rest, Side::Result)
    } else {
        return EffectOutcome::Halt(Decision::Deny {
            reason: Some(format!(
                "FieldOp path `{}` must start with `args.` or `result.`",
                path
            )),
            rule_source: fallback_source.to_string(),
        });
    };

    let Some(current) = get_dotted(root, subpath).cloned() else {
        return EffectOutcome::Continue; // missing field → silent no-op
    };

    let pipeline = crate::pipeline::Pipeline {
        stages: stages.to_vec(),
    };
    let eval = evaluate_pipeline(&pipeline, &current, bag, plugins, path, phase).await;
    taints.extend(eval.taints);
    let mark_modified = |side: Side, args: &mut bool, result: &mut bool| match side {
        Side::Args => *args = true,
        Side::Result => *result = true,
    };
    match eval.outcome {
        FieldOutcome::Pass => EffectOutcome::Continue,
        FieldOutcome::Replace(new_val) => {
            if set_dotted(root, subpath, new_val) {
                mark_modified(side, args_modified, result_modified);
            }
            EffectOutcome::Continue
        },
        FieldOutcome::Omit => {
            if remove_dotted(root, subpath) {
                mark_modified(side, args_modified, result_modified);
            }
            EffectOutcome::Continue
        },
        FieldOutcome::Deny {
            reason,
            stage_index: _,
        } => EffectOutcome::Halt(Decision::Deny {
            reason: Some(reason),
            rule_source: fallback_source.to_string(),
        }),
    }
}

// =====================================================================
// Pipe-chain evaluator (args: / result: field pipelines)
// =====================================================================

/// Result of running a pipeline against one field's value.
///
/// `Pass`: every stage succeeded; the original value should be kept.
/// `Replace`: a transform produced a new value (also covers conditional
/// `redact` firing).
/// `Omit`: an `omit` stage fired; the field should be dropped from output.
/// `Deny`: a validator failed; pipeline halted; the route should deny.
#[derive(Debug, Clone, PartialEq)]
pub enum FieldOutcome {
    Pass,
    Replace(serde_json::Value),
    Omit,
    Deny { reason: String, stage_index: usize },
}

/// Full result of a pipeline run: value-level outcome plus accumulated
/// taint side effects.
///
/// `taint(...)` stages, plugin invocations, and `scan(...)` stages can all
/// emit taints; the evaluator collects them here and hands them to the host
/// (apl-cpex) for SessionStore writes. Taints accumulate even on `Replace`
/// and `Omit` outcomes; they do not accumulate past a `Deny` (the pipeline
/// halts at the failing stage).
#[derive(Debug, Clone, PartialEq)]
pub struct PipelineEvaluation {
    pub outcome: FieldOutcome,
    pub taints: Vec<TaintEvent>,
}

/// Walk a pipeline against `value` and the bag, applying stages left-to-right.
///
/// Async because pipe-chain `plugin(name)` stages dispatch through
/// `PluginInvoker`, which is async.
///
/// `field_name` is the field this pipeline is attached to (from the wrapping
/// `FieldRule`). It's threaded into `PluginInvocation::Field` when a
/// `Stage::Plugin` fires so the invoker knows which field is in focus.
/// Pass `""` for standalone pipeline runs that aren't part of a field rule.
///
/// `Stage::Validate { name }` is currently a no-op with a TODO — the named
/// validator registry lands in a later step.
pub async fn evaluate_pipeline(
    pipeline: &Pipeline,
    value: &serde_json::Value,
    bag: &AttributeBag,
    plugins: &Arc<dyn PluginInvoker>,
    field_name: &str,
    phase: crate::step::DispatchPhase,
) -> PipelineEvaluation {
    let mut current = value.clone();
    let mut replaced = false;
    let mut taints: Vec<TaintEvent> = Vec::new();

    for (idx, stage) in pipeline.stages.iter().enumerate() {
        match stage {
            // ----- Validators -----
            Stage::Type(tc) => {
                if !type_check(tc, &current) {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!("expected {:?}, got {}", tc, value_kind(&current)),
                            stage_index: idx,
                        },
                        taints,
                    };
                }
            },
            Stage::Length { min, max } => {
                let Some(s) = current.as_str() else {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!(
                                "len(...) requires string value, got {}",
                                value_kind(&current)
                            ),
                            stage_index: idx,
                        },
                        taints,
                    };
                };
                let len = s.chars().count();
                if min.map_or(false, |m| len < m) || max.map_or(false, |m| len > m) {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!("length {} outside [{:?}, {:?}]", len, min, max),
                            stage_index: idx,
                        },
                        taints,
                    };
                }
            },
            Stage::Range { min, max } => {
                let Some(n) = current.as_i64() else {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!(
                                "range requires integer value, got {}",
                                value_kind(&current)
                            ),
                            stage_index: idx,
                        },
                        taints,
                    };
                };
                if min.map_or(false, |m| n < m) || max.map_or(false, |m| n > m) {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!("value {} outside [{:?}, {:?}]", n, min, max),
                            stage_index: idx,
                        },
                        taints,
                    };
                }
            },
            Stage::Enum { values } => {
                let Some(s) = current.as_str() else {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!(
                                "enum(...) requires string value, got {}",
                                value_kind(&current)
                            ),
                            stage_index: idx,
                        },
                        taints,
                    };
                };
                if !values.iter().any(|v| v == s) {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!("value `{}` not in enum {:?}", s, values),
                            stage_index: idx,
                        },
                        taints,
                    };
                }
            },
            Stage::Regex { pattern } => {
                // Compile-at-eval for now. A future step can swap to a
                // route-level pre-compile cache keyed by pattern.
                let re = match regex::Regex::new(pattern) {
                    Ok(r) => r,
                    Err(e) => {
                        return PipelineEvaluation {
                            outcome: FieldOutcome::Deny {
                                reason: format!("invalid regex `{}`: {}", pattern, e),
                                stage_index: idx,
                            },
                            taints,
                        };
                    },
                };
                let Some(s) = current.as_str() else {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!(
                                "regex requires string value, got {}",
                                value_kind(&current)
                            ),
                            stage_index: idx,
                        },
                        taints,
                    };
                };
                if !re.is_match(s) {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!("value did not match regex `{}`", pattern),
                            stage_index: idx,
                        },
                        taints,
                    };
                }
            },
            Stage::Validate { name } => {
                // Named-validator dispatch is not implemented in this
                // build. The parser rejects `validate(...)` at compile
                // time (parser.rs); this branch covers IR built
                // programmatically bypassing the parser. Same shape
                // as the parser's diagnostic — operators reach for
                // `regex(...)` or `plugin(...)` instead.
                return PipelineEvaluation {
                    outcome: FieldOutcome::Deny {
                        reason: format!(
                            "`validate({})` is not implemented; use `regex(...)` \
                             or `plugin({})` instead",
                            name, name,
                        ),
                        stage_index: idx,
                    },
                    taints,
                };
            },

            // ----- Transforms -----
            Stage::Mask { keep_last } => {
                let Some(s) = current.as_str() else {
                    return PipelineEvaluation {
                        outcome: FieldOutcome::Deny {
                            reason: format!(
                                "mask(...) requires string value, got {}",
                                value_kind(&current)
                            ),
                            stage_index: idx,
                        },
                        taints,
                    };
                };
                let chars: Vec<char> = s.chars().collect();
                let keep = (*keep_last).min(chars.len());
                let mask_count = chars.len() - keep;
                let masked: String = std::iter::repeat('*')
                    .take(mask_count)
                    .chain(chars.into_iter().skip(mask_count))
                    .collect();
                current = serde_json::Value::String(masked);
                replaced = true;
            },
            Stage::Redact { condition } => {
                let should_redact = match condition {
                    None => true,
                    Some(expr) => eval_expression(expr, bag),
                };
                if should_redact {
                    current = serde_json::Value::String("[REDACTED]".into());
                    replaced = true;
                }
            },
            Stage::Omit => {
                return PipelineEvaluation {
                    outcome: FieldOutcome::Omit,
                    taints,
                };
            },
            Stage::Hash => {
                // Simple deterministic digest — DefaultHasher is fine for
                // de-identification (not for cryptographic use).
                use std::hash::{Hash, Hasher};
                let mut h = std::collections::hash_map::DefaultHasher::new();
                value_for_hash(&current).hash(&mut h);
                current = serde_json::Value::String(format!("hash:{:016x}", h.finish()));
                replaced = true;
            },

            // ----- Effects -----
            Stage::Taint { label, scopes } => {
                taints.push(TaintEvent {
                    label: label.clone(),
                    scopes: scopes.clone(),
                });
            },
            Stage::Plugin { name } => {
                let invocation = PluginInvocation::Field {
                    name: field_name,
                    value: &current,
                    phase,
                };
                match plugins.invoke(name, bag, invocation).await {
                    Ok(outcome) => {
                        // Plugins can emit taints regardless of decision.
                        taints.extend(outcome.taints);
                        match outcome.decision {
                            Decision::Allow => {
                                if let Some(new_value) = outcome.modified_value {
                                    current = new_value;
                                    replaced = true;
                                }
                            },
                            Decision::Deny {
                                reason,
                                rule_source: _,
                            } => {
                                return PipelineEvaluation {
                                    outcome: FieldOutcome::Deny {
                                        reason: reason
                                            .unwrap_or_else(|| format!("plugin `{}` denied", name)),
                                        stage_index: idx,
                                    },
                                    taints,
                                };
                            },
                        }
                    },
                    Err(e) => {
                        // Fail-closed: plugin dispatch failure halts the pipeline.
                        return PipelineEvaluation {
                            outcome: FieldOutcome::Deny {
                                reason: format!("plugin `{}` error: {}", name, e),
                                stage_index: idx,
                            },
                            taints,
                        };
                    },
                }
            },
            Stage::Scan { kind } => {
                // Spec mapping (apl-dsl-spec §4): scan stages are taint
                // emitters. The actual PII detection / injection signal
                // lives in plugin(...) variants of the same scanners; this
                // stage just records the label so downstream policies can
                // gate on it. `pii.redact` additionally rewrites the value.
                let (label, redact): (&str, bool) = match kind {
                    ScanKind::PiiDetect => ("PII", false),
                    ScanKind::PiiRedact => ("PII", true),
                    ScanKind::InjectionScan => ("injection", false),
                };
                taints.push(TaintEvent {
                    label: label.to_string(),
                    scopes: vec![TaintScope::Session],
                });
                if redact {
                    current = serde_json::Value::String("[REDACTED]".into());
                    replaced = true;
                }
            },
        }
    }

    let outcome = if replaced {
        FieldOutcome::Replace(current)
    } else {
        FieldOutcome::Pass
    };
    PipelineEvaluation { outcome, taints }
}

fn type_check(tc: &TypeCheck, v: &serde_json::Value) -> bool {
    match tc {
        TypeCheck::Str => v.is_string(),
        TypeCheck::Int => v.is_i64(),
        TypeCheck::Bool => v.is_boolean(),
        TypeCheck::Float => v.is_f64() || v.is_i64(),
        TypeCheck::Email => v
            .as_str()
            .map_or(false, |s| s.contains('@') && s.contains('.')),
        TypeCheck::Url => v.as_str().map_or(false, |s| {
            s.starts_with("http://") || s.starts_with("https://")
        }),
        TypeCheck::Uuid => v.as_str().map_or(false, is_uuid_shape),
    }
}

fn is_uuid_shape(s: &str) -> bool {
    // 8-4-4-4-12 hex with `-` separators.
    let bytes = s.as_bytes();
    if bytes.len() != 36 {
        return false;
    }
    for (i, &b) in bytes.iter().enumerate() {
        match i {
            8 | 13 | 18 | 23 => {
                if b != b'-' {
                    return false;
                }
            },
            _ => {
                if !b.is_ascii_hexdigit() {
                    return false;
                }
            },
        }
    }
    true
}

fn value_kind(v: &serde_json::Value) -> &'static str {
    match v {
        serde_json::Value::Null => "null",
        serde_json::Value::Bool(_) => "bool",
        serde_json::Value::Number(n) if n.is_i64() => "int",
        serde_json::Value::Number(_) => "float",
        serde_json::Value::String(_) => "string",
        serde_json::Value::Array(_) => "array",
        serde_json::Value::Object(_) => "object",
    }
}

/// Stable byte representation of a value for hashing — serde_json's
/// `to_string` is canonical enough for our use.
fn value_for_hash(v: &serde_json::Value) -> String {
    serde_json::to_string(v).unwrap_or_default()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::rules::{Condition, Expression, Rule};
    use crate::step::{DelegationInvoker, NoopDelegationInvoker};
    use std::collections::HashSet;
    use std::sync::Arc;

    fn rule(condition: Expression, effect: Effect, source: &str) -> Rule {
        Rule::single(condition, effect, source)
    }

    // Wrap stateless test invokers in `Arc<dyn ...>` once per call. The
    // public evaluator API takes `&Arc<dyn PluginInvoker>` so internal
    // dispatch (notably `Effect::Parallel`) can `Arc::clone` an owned,
    // 'static reference into each spawned branch (slice E3.2).
    fn null_pipe_plugins() -> Arc<dyn PluginInvoker> {
        Arc::new(NullPipelinePlugins)
    }
    fn null_plugins() -> Arc<dyn PluginInvoker> {
        Arc::new(NullPlugins)
    }
    fn noop_delegations() -> Arc<dyn DelegationInvoker> {
        Arc::new(NoopDelegationInvoker)
    }
    fn noop_elicitations() -> Arc<dyn crate::step::ElicitationInvoker> {
        Arc::new(crate::step::NoopElicitationInvoker)
    }
    fn auto_elicitations() -> Arc<dyn crate::step::ElicitationInvoker> {
        Arc::new(crate::step::AutoApprovingElicitor)
    }

    fn deny(reason: &str) -> Effect {
        Effect::Deny {
            reason: Some(reason.into()),
            code: None,
        }
    }

    /// Build an `Effect::Elicit` with the given scope / on_error for the
    /// elicitation-arm tests. Channel/kind are fixed — the arm doesn't
    /// branch on them (that's the plugin's job, which the fakes stub).
    fn elicit_effect(scope: Option<&str>, on_error: Option<&str>) -> Effect {
        Effect::Elicit(crate::step::ElicitStep {
            kind: crate::step::ElicitKind::Approval,
            plugin_name: "manager-approver".into(),
            channel: Some("ciba".into()),
            from: "user.manager".into(),
            purpose: Some("approve payroll adjustment".into()),
            scope: scope.map(|s| s.to_string()),
            timeout: None,
            config_override: None,
            on_error: on_error.map(|s| s.to_string()),
            source: "route.test.policy[0]".into(),
        })
    }

    fn cond(c: Condition) -> Expression {
        Expression::Condition(c)
    }

    // ----- Decision-level semantics -----

    #[test]
    fn empty_rules_allow() {
        let mut bag = AttributeBag::new();
        assert_eq!(evaluate_rules(&[], &bag), Decision::Allow);
    }

    #[test]
    fn first_deny_halts() {
        let mut bag = AttributeBag::new();
        bag.set("a", true);
        bag.set("b", true);

        let rules = vec![
            rule(
                cond(Condition::IsTrue { key: "a".into() }),
                deny("first"),
                "r0",
            ),
            rule(
                cond(Condition::IsTrue { key: "b".into() }),
                deny("second"),
                "r1",
            ),
        ];

        match evaluate_rules(&rules, &bag) {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert_eq!(reason.as_deref(), Some("first"));
                assert_eq!(rule_source, "r0");
            },
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    #[test]
    fn allow_does_not_short_circuit() {
        // Spec §3: explicit allow continues evaluation. A later deny still fires.
        let mut bag = AttributeBag::new();
        bag.set("ok", true);
        bag.set("bad", true);

        let rules = vec![
            rule(
                cond(Condition::IsTrue { key: "ok".into() }),
                Effect::Allow,
                "r0_allow",
            ),
            rule(
                cond(Condition::IsTrue { key: "bad".into() }),
                deny("later"),
                "r1_deny",
            ),
        ];

        match evaluate_rules(&rules, &bag) {
            Decision::Deny { rule_source, .. } => assert_eq!(rule_source, "r1_deny"),
            d => panic!("allow short-circuited; expected later deny, got {:?}", d),
        }
    }

    #[test]
    fn unmatched_rules_dont_fire() {
        let mut bag = AttributeBag::new(); // "denied" missing → false
        let rules = vec![rule(
            cond(Condition::IsTrue {
                key: "denied".into(),
            }),
            deny("shouldn't fire"),
            "r0",
        )];
        assert_eq!(evaluate_rules(&rules, &bag), Decision::Allow);
    }

    // ----- Predicate semantics -----

    #[test]
    fn missing_key_is_false() {
        let mut bag = AttributeBag::new();
        assert!(!eval_condition(
            &Condition::IsTrue {
                key: "missing".into()
            },
            &bag
        ));
        assert!(eval_condition(
            &Condition::IsFalse {
                key: "missing".into()
            },
            &bag
        ));
        // Comparison on missing → false (spec §2.6).
        assert!(!eval_condition(
            &Condition::Comparison {
                key: "missing".into(),
                op: CompareOp::Eq,
                value: 1_i64.into(),
            },
            &bag,
        ));
    }

    #[test]
    fn and_or_not_combinators() {
        let mut bag = AttributeBag::new();
        bag.set("a", true);
        bag.set("b", false);

        let a = cond(Condition::IsTrue { key: "a".into() });
        let b = cond(Condition::IsTrue { key: "b".into() });

        assert!(eval_expression(
            &Expression::And(vec![a.clone(), a.clone()]),
            &bag
        ));
        assert!(!eval_expression(
            &Expression::And(vec![a.clone(), b.clone()]),
            &bag
        ));
        assert!(eval_expression(
            &Expression::Or(vec![a.clone(), b.clone()]),
            &bag
        ));
        assert!(!eval_expression(
            &Expression::Or(vec![b.clone(), b.clone()]),
            &bag
        ));
        assert!(eval_expression(&Expression::Not(Box::new(b)), &bag));
    }

    // ----- Comparison operators -----

    #[test]
    fn int_comparisons() {
        let mut bag = AttributeBag::new();
        bag.set("delegation.depth", 3_i64);

        let cmp = |op| Condition::Comparison {
            key: "delegation.depth".into(),
            op,
            value: 2_i64.into(),
        };
        assert!(eval_condition(&cmp(CompareOp::Gt), &bag));
        assert!(eval_condition(&cmp(CompareOp::GtEq), &bag));
        assert!(!eval_condition(&cmp(CompareOp::Lt), &bag));
        assert!(!eval_condition(&cmp(CompareOp::Eq), &bag));
        assert!(eval_condition(&cmp(CompareOp::NotEq), &bag));
    }

    #[test]
    fn int_to_float_promotion_in_comparison() {
        let mut bag = AttributeBag::new();
        bag.set("delegation.depth", 2_i64);
        // `delegation.depth > 2.5` — int promotes to float for the compare.
        assert!(!eval_condition(
            &Condition::Comparison {
                key: "delegation.depth".into(),
                op: CompareOp::Gt,
                value: 2.5_f64.into(),
            },
            &bag,
        ));
        assert!(eval_condition(
            &Condition::Comparison {
                key: "delegation.depth".into(),
                op: CompareOp::Lt,
                value: 2.5_f64.into(),
            },
            &bag,
        ));
    }

    #[test]
    fn numeric_string_args_coerce_for_order_comparison() {
        // LLMs routinely send numeric tool-args as strings, e.g.
        // `args.amount = "25000"`. An order comparison must still fire so a
        // gate like `when: args.amount > 10000` (and the elicitation
        // `scope: args.amount <= 25000`) work. Regression for the demo bug
        // where a string amount slipped past the approval threshold.
        let mut bag = AttributeBag::new();
        bag.set("args.amount", "25000");
        let cmp = |op, v: i64| Condition::Comparison {
            key: "args.amount".into(),
            op,
            value: v.into(),
        };
        assert!(
            eval_condition(&cmp(CompareOp::Gt, 10000), &bag),
            "\"25000\" > 10000"
        );
        assert!(
            eval_condition(&cmp(CompareOp::LtEq, 25000), &bag),
            "\"25000\" <= 25000"
        );
        assert!(
            !eval_condition(&cmp(CompareOp::Gt, 25000), &bag),
            "\"25000\" > 25000 is false"
        );

        // A genuinely non-numeric string still doesn't order-compare.
        bag.set("args.amount", "lots");
        assert!(
            !eval_condition(&cmp(CompareOp::Gt, 10000), &bag),
            "\"lots\" > 10000 is false"
        );
    }

    #[test]
    fn string_equality_no_ordering() {
        let mut bag = AttributeBag::new();
        bag.set("subject.id", "alice");

        assert!(eval_condition(
            &Condition::Comparison {
                key: "subject.id".into(),
                op: CompareOp::Eq,
                value: "alice".into(),
            },
            &bag,
        ));
        // Order operators on strings → false (spec §2.3).
        assert!(!eval_condition(
            &Condition::Comparison {
                key: "subject.id".into(),
                op: CompareOp::Gt,
                value: "alice".into(),
            },
            &bag,
        ));
    }

    #[test]
    fn contains_set_membership() {
        let mut bag = AttributeBag::new();
        bag.set(
            "session.labels",
            HashSet::from(["PII".to_string(), "financial".to_string()]),
        );

        assert!(eval_condition(
            &Condition::Comparison {
                key: "session.labels".into(),
                op: CompareOp::Contains,
                value: "PII".into(),
            },
            &bag,
        ));
        assert!(!eval_condition(
            &Condition::Comparison {
                key: "session.labels".into(),
                op: CompareOp::Contains,
                value: "PHI".into(),
            },
            &bag,
        ));
        // Contains on a non-set attribute → false.
        bag.set("subject.id", "alice");
        assert!(!eval_condition(
            &Condition::Comparison {
                key: "subject.id".into(),
                op: CompareOp::Contains,
                value: "alice".into(),
            },
            &bag,
        ));
    }

    // ----- Realistic end-to-end -----

    #[test]
    fn hr_compensation_scenario() {
        // From the HR demo: alice (hr role + view_ssn perm) requests compensation
        // with delegation.depth = 1. Rules:
        //   1. require(authenticated)
        //   2. require(role.hr | role.finance)
        //   3. delegation.depth > 2 & include_ssn: deny
        //   4. !perm.view_ssn & include_ssn: deny
        let mut bag = AttributeBag::new();
        bag.set("authenticated", true);
        bag.set("role.hr", true);
        bag.set("perm.view_ssn", true);
        bag.set("delegation.depth", 1_i64);
        bag.set("include_ssn", true);

        let rules = vec![
            // require(authenticated) → deny if !authenticated
            rule(
                Expression::Not(Box::new(cond(Condition::IsTrue {
                    key: "authenticated".into(),
                }))),
                deny("not authenticated"),
                "r0",
            ),
            // require(role.hr | role.finance) → deny if neither
            // Desugars to: when !(role.hr | role.finance) do deny
            //             = when (role.hr is false) AND (role.finance is false), deny
            rule(
                Expression::And(vec![
                    cond(Condition::IsFalse {
                        key: "role.hr".into(),
                    }),
                    cond(Condition::IsFalse {
                        key: "role.finance".into(),
                    }),
                ]),
                deny("not in hr/finance"),
                "r1",
            ),
            // delegation.depth > 2 & include_ssn: deny
            rule(
                Expression::And(vec![
                    cond(Condition::Comparison {
                        key: "delegation.depth".into(),
                        op: CompareOp::Gt,
                        value: 2_i64.into(),
                    }),
                    cond(Condition::IsTrue {
                        key: "include_ssn".into(),
                    }),
                ]),
                deny("delegation too deep for SSN"),
                "r2",
            ),
        ];

        assert_eq!(evaluate_rules(&rules, &bag), Decision::Allow);

        // Now make Alice undelegated-but-deep — should still allow at depth=1.
        // Change to depth=3 and the SSN rule fires.
        bag.set("delegation.depth", 3_i64);
        match evaluate_rules(&rules, &bag) {
            Decision::Deny { rule_source, .. } => assert_eq!(rule_source, "r2"),
            d => panic!("expected r2 deny, got {:?}", d),
        }
    }

    // ===================================================================
    // Pipe-chain evaluator tests
    // ===================================================================

    use crate::pipeline::{Stage, TypeCheck};
    use serde_json::json;

    fn make_pipeline(stages: Vec<Stage>) -> crate::pipeline::Pipeline {
        crate::pipeline::Pipeline { stages }
    }

    // Helper: a plugin invoker that's never expected to fire (pipelines
    // without `plugin(...)` stages). Panics if called. Defined alongside
    // the other null fixtures further down in this module.

    async fn run_pipeline(
        p: &crate::pipeline::Pipeline,
        v: &serde_json::Value,
        bag: &AttributeBag,
    ) -> FieldOutcome {
        evaluate_pipeline(
            p,
            v,
            bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await
        .outcome
    }

    /// Pipeline-test null invoker — distinct from the step-test `NullPlugins`
    /// so each test can panic with a clearer "wrong fixture" message if it
    /// ever does dispatch a plugin call by accident.
    struct NullPipelinePlugins;
    #[async_trait]
    impl PluginInvoker for NullPipelinePlugins {
        async fn invoke(
            &self,
            name: &str,
            _bag: &AttributeBag,
            _invocation: PluginInvocation<'_>,
        ) -> Result<PluginOutcome, PluginError> {
            panic!(
                "NullPipelinePlugins should not dispatch; got plugin({})",
                name
            );
        }
    }

    #[tokio::test]
    async fn pipeline_empty_is_pass() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![]);
        assert_eq!(
            run_pipeline(&p, &json!("anything"), &bag).await,
            FieldOutcome::Pass
        );
    }

    #[tokio::test]
    async fn pipeline_type_check_passes_and_denies() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Type(TypeCheck::Str)]);
        assert_eq!(
            run_pipeline(&p, &json!("hello"), &bag).await,
            FieldOutcome::Pass
        );
        match run_pipeline(&p, &json!(42), &bag).await {
            FieldOutcome::Deny {
                reason,
                stage_index,
            } => {
                assert!(reason.contains("expected Str"));
                assert_eq!(stage_index, 0);
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_mask_preserves_last_n() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Mask { keep_last: 4 }]);
        match run_pipeline(&p, &json!("123-45-6789"), &bag).await {
            FieldOutcome::Replace(v) => assert_eq!(v, json!("*******6789")),
            other => panic!("expected Replace, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_mask_handles_short_strings() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Mask { keep_last: 4 }]);
        // keep_last >= length → no mask chars; full string preserved.
        match run_pipeline(&p, &json!("ab"), &bag).await {
            FieldOutcome::Replace(v) => assert_eq!(v, json!("ab")),
            other => panic!("expected Replace, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_unconditional_redact() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Redact { condition: None }]);
        match run_pipeline(&p, &json!("secret"), &bag).await {
            FieldOutcome::Replace(v) => assert_eq!(v, json!("[REDACTED]")),
            other => panic!("expected Replace, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_conditional_redact_fires_when_condition_true() {
        // redact(!perm.view_ssn): condition is `!perm.view_ssn`. Missing key
        // → IsTrue is false → `!IsTrue` is true → redact fires.
        let mut bag = AttributeBag::new();
        let cond = Expression::Not(Box::new(Expression::Condition(Condition::IsTrue {
            key: "perm.view_ssn".into(),
        })));
        let p = make_pipeline(vec![Stage::Redact {
            condition: Some(cond),
        }]);
        match run_pipeline(&p, &json!("123-45-6789"), &bag).await {
            FieldOutcome::Replace(v) => assert_eq!(v, json!("[REDACTED]")),
            other => panic!("expected Replace (redact fired), got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_conditional_redact_skips_when_condition_false() {
        let mut bag = AttributeBag::new();
        bag.set("perm.view_ssn", true);
        let cond = Expression::Not(Box::new(Expression::Condition(Condition::IsTrue {
            key: "perm.view_ssn".into(),
        })));
        let p = make_pipeline(vec![Stage::Redact {
            condition: Some(cond),
        }]);
        // perm.view_ssn=true → !true=false → redact skipped → Pass.
        assert_eq!(
            run_pipeline(&p, &json!("123-45-6789"), &bag).await,
            FieldOutcome::Pass,
        );
    }

    #[tokio::test]
    async fn pipeline_omit_short_circuits() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Omit,
            // This stage should never run.
            Stage::Type(TypeCheck::Int),
        ]);
        assert_eq!(
            run_pipeline(&p, &json!("anything"), &bag).await,
            FieldOutcome::Omit
        );
    }

    #[tokio::test]
    async fn pipeline_range_validator() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Type(TypeCheck::Int),
            Stage::Range {
                min: Some(0),
                max: Some(1_000_000),
            },
        ]);
        assert_eq!(
            run_pipeline(&p, &json!(500_000), &bag).await,
            FieldOutcome::Pass
        );
        // Above max → deny.
        match run_pipeline(&p, &json!(2_000_000), &bag).await {
            FieldOutcome::Deny {
                reason,
                stage_index,
            } => {
                assert!(reason.contains("outside"));
                assert_eq!(stage_index, 1);
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_length_validator() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Length {
            min: None,
            max: Some(5),
        }]);
        assert_eq!(
            run_pipeline(&p, &json!("hi"), &bag).await,
            FieldOutcome::Pass
        );
        assert!(matches!(
            run_pipeline(&p, &json!("too long"), &bag).await,
            FieldOutcome::Deny { .. },
        ));
    }

    #[tokio::test]
    async fn pipeline_enum_validator() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Enum {
            values: vec!["low".into(), "medium".into(), "high".into()],
        }]);
        assert_eq!(
            run_pipeline(&p, &json!("medium"), &bag).await,
            FieldOutcome::Pass
        );
        assert!(matches!(
            run_pipeline(&p, &json!("extreme"), &bag).await,
            FieldOutcome::Deny { .. },
        ));
    }

    #[tokio::test]
    async fn pipeline_uuid_validator() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Type(TypeCheck::Uuid)]);
        assert_eq!(
            run_pipeline(&p, &json!("550e8400-e29b-41d4-a716-446655440000"), &bag).await,
            FieldOutcome::Pass,
        );
        assert!(matches!(
            run_pipeline(&p, &json!("not-a-uuid"), &bag).await,
            FieldOutcome::Deny { .. },
        ));
    }

    #[tokio::test]
    async fn pipeline_hash_replaces_value() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Hash]);
        match run_pipeline(&p, &json!("secret"), &bag).await {
            FieldOutcome::Replace(v) => {
                let s = v.as_str().unwrap();
                assert!(s.starts_with("hash:"));
                assert_eq!(s.len(), "hash:".len() + 16);
            },
            other => panic!("expected Replace, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_validate_named_denies_at_runtime() {
        // `validate(name)` is unimplemented in this build. The parser
        // rejects it at compile time; this test exercises the runtime
        // defense-in-depth path for IR built programmatically. The
        // deny message points operators at the working alternatives
        // (`regex(...)` / `plugin(...)`).
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Type(TypeCheck::Str),
            Stage::Validate {
                name: "ssn_format".into(),
            },
            Stage::Mask { keep_last: 4 },
        ]);
        match run_pipeline(&p, &json!("123-45-6789"), &bag).await {
            FieldOutcome::Deny {
                reason,
                stage_index,
            } => {
                assert_eq!(stage_index, 1, "validate stage is at index 1");
                assert!(
                    reason.contains("not implemented"),
                    "deny reason should explain that validate is unimplemented: {reason}",
                );
                assert!(
                    reason.contains("regex") || reason.contains("plugin"),
                    "deny reason should point at alternatives: {reason}",
                );
            },
            other => panic!("expected Deny on validate(...) stage, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_validator_short_circuits_before_transform() {
        // If the validator fails, the transform never runs.
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Type(TypeCheck::Int), // will fail on a string
            Stage::Mask { keep_last: 4 },
        ]);
        match run_pipeline(&p, &json!("hello"), &bag).await {
            FieldOutcome::Deny { stage_index, .. } => assert_eq!(stage_index, 0),
            other => panic!("expected Deny at stage 0, got {:?}", other),
        }
    }

    // ----- Regex stage -----

    #[tokio::test]
    async fn pipeline_regex_match_passes() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Regex {
            pattern: r"^\d{3}-\d{2}-\d{4}$".into(),
        }]);
        assert_eq!(
            run_pipeline(&p, &json!("123-45-6789"), &bag).await,
            FieldOutcome::Pass
        );
    }

    #[tokio::test]
    async fn pipeline_regex_no_match_denies() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Regex {
            pattern: r"^\d{3}-\d{2}-\d{4}$".into(),
        }]);
        match run_pipeline(&p, &json!("not an ssn"), &bag).await {
            FieldOutcome::Deny {
                reason,
                stage_index,
            } => {
                assert!(reason.contains("did not match"));
                assert_eq!(stage_index, 0);
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_regex_invalid_pattern_denies() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Regex {
            pattern: "(unclosed".into(),
        }]);
        match run_pipeline(&p, &json!("anything"), &bag).await {
            FieldOutcome::Deny { reason, .. } => {
                assert!(reason.contains("invalid regex"));
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_regex_non_string_denies() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Regex {
            pattern: r"^\d+$".into(),
        }]);
        match run_pipeline(&p, &json!(42), &bag).await {
            FieldOutcome::Deny { reason, .. } => {
                assert!(reason.contains("requires string"));
            },
            other => panic!("expected Deny on non-string regex input, got {:?}", other),
        }
    }

    // ----- Taint and Scan stages -----

    #[tokio::test]
    async fn pipeline_taint_records_event() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Type(TypeCheck::Str),
            Stage::Taint {
                label: "PII".into(),
                scopes: vec![TaintScope::Session],
            },
            Stage::Mask { keep_last: 4 },
        ]);
        let result = evaluate_pipeline(
            &p,
            &json!("123-45-6789"),
            &bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert_eq!(result.outcome, FieldOutcome::Replace(json!("*******6789")));
        assert_eq!(
            result.taints,
            vec![TaintEvent {
                label: "PII".into(),
                scopes: vec![TaintScope::Session],
            }]
        );
    }

    #[tokio::test]
    async fn pipeline_scan_pii_detect_emits_taint() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Scan {
            kind: ScanKind::PiiDetect,
        }]);
        let result = evaluate_pipeline(
            &p,
            &json!("some text"),
            &bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        // PII detect: value unchanged, one taint event emitted.
        assert_eq!(result.outcome, FieldOutcome::Pass);
        assert_eq!(
            result.taints,
            vec![TaintEvent {
                label: "PII".into(),
                scopes: vec![TaintScope::Session],
            }]
        );
    }

    #[tokio::test]
    async fn pipeline_scan_pii_redact_replaces_and_taints() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Scan {
            kind: ScanKind::PiiRedact,
        }]);
        let result = evaluate_pipeline(
            &p,
            &json!("123-45-6789"),
            &bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert_eq!(result.outcome, FieldOutcome::Replace(json!("[REDACTED]")));
        assert_eq!(result.taints.len(), 1);
        assert_eq!(result.taints[0].label, "PII");
    }

    #[tokio::test]
    async fn pipeline_scan_injection_emits_injection_taint() {
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![Stage::Scan {
            kind: ScanKind::InjectionScan,
        }]);
        let result = evaluate_pipeline(
            &p,
            &json!("user input"),
            &bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert_eq!(result.outcome, FieldOutcome::Pass);
        assert_eq!(result.taints[0].label, "injection");
    }

    #[tokio::test]
    async fn pipeline_deny_does_not_accumulate_later_taints() {
        // Pipeline halts at the first failing validator; taints emitted
        // before the failure stick, taints after do not.
        let mut bag = AttributeBag::new();
        let p = make_pipeline(vec![
            Stage::Taint {
                label: "before".into(),
                scopes: vec![TaintScope::Session],
            },
            Stage::Type(TypeCheck::Int), // fails on string input
            Stage::Taint {
                label: "after".into(),
                scopes: vec![TaintScope::Session],
            },
        ]);
        let result = evaluate_pipeline(
            &p,
            &json!("hello"),
            &bag,
            &null_pipe_plugins(),
            "test_field",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert!(matches!(result.outcome, FieldOutcome::Deny { .. }));
        assert_eq!(
            result.taints,
            vec![TaintEvent {
                label: "before".into(),
                scopes: vec![TaintScope::Session],
            }]
        );
    }

    // ----- Plugin stage in pipe chain -----

    /// Pipe-context plugin invoker that returns canned outcomes by name.
    struct PipePlugin {
        outcomes: std::collections::HashMap<String, PluginOutcome>,
    }
    #[async_trait]
    impl PluginInvoker for PipePlugin {
        async fn invoke(
            &self,
            name: &str,
            _bag: &AttributeBag,
            _invocation: PluginInvocation<'_>,
        ) -> Result<PluginOutcome, PluginError> {
            self.outcomes
                .get(name)
                .cloned()
                .ok_or_else(|| PluginError::NotFound(name.into()))
        }
    }

    #[tokio::test]
    async fn pipeline_plugin_allow_continues() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(PipePlugin {
            outcomes: std::collections::HashMap::from([(
                "noop".to_string(),
                PluginOutcome::allow(),
            )]),
        });
        let p = make_pipeline(vec![
            Stage::Type(TypeCheck::Str),
            Stage::Plugin {
                name: "noop".into(),
            },
            Stage::Mask { keep_last: 4 },
        ]);
        let result = evaluate_pipeline(
            &p,
            &json!("123-45-6789"),
            &bag,
            &plugins,
            "compensation",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert_eq!(result.outcome, FieldOutcome::Replace(json!("*******6789")));
        assert!(result.taints.is_empty());
    }

    #[tokio::test]
    async fn pipeline_plugin_can_replace_value() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(PipePlugin {
            outcomes: std::collections::HashMap::from([(
                "scrubber".to_string(),
                PluginOutcome {
                    decision: Decision::Allow,
                    taints: vec![TaintEvent {
                        label: "PII".to_string(),
                        scopes: vec![TaintScope::Session],
                    }],
                    modified_value: Some(json!("***scrubbed***")),
                },
            )]),
        });
        let p = make_pipeline(vec![Stage::Plugin {
            name: "scrubber".into(),
        }]);
        let result = evaluate_pipeline(
            &p,
            &json!("sensitive data"),
            &bag,
            &plugins,
            "notes",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        assert_eq!(
            result.outcome,
            FieldOutcome::Replace(json!("***scrubbed***"))
        );
        assert_eq!(
            result.taints,
            vec![TaintEvent {
                label: "PII".into(),
                scopes: vec![TaintScope::Session],
            }]
        );
    }

    #[tokio::test]
    async fn pipeline_plugin_deny_halts() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(PipePlugin {
            outcomes: std::collections::HashMap::from([(
                "guard".to_string(),
                PluginOutcome {
                    decision: Decision::Deny {
                        reason: Some("policy violation".into()),
                        rule_source: "guard".into(),
                    },
                    taints: vec![],
                    modified_value: None,
                },
            )]),
        });
        let p = make_pipeline(vec![
            Stage::Plugin {
                name: "guard".into(),
            },
            // Should never run.
            Stage::Mask { keep_last: 4 },
        ]);
        let result = evaluate_pipeline(
            &p,
            &json!("data"),
            &bag,
            &plugins,
            "payload",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        match result.outcome {
            FieldOutcome::Deny {
                reason,
                stage_index,
            } => {
                assert_eq!(reason, "policy violation");
                assert_eq!(stage_index, 0);
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn pipeline_plugin_missing_fails_closed() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(PipePlugin {
            outcomes: Default::default(),
        });
        let p = make_pipeline(vec![Stage::Plugin {
            name: "missing".into(),
        }]);
        let result = evaluate_pipeline(
            &p,
            &json!("data"),
            &bag,
            &plugins,
            "payload",
            crate::step::DispatchPhase::Pre,
        )
        .await;
        match result.outcome {
            FieldOutcome::Deny { reason, .. } => assert!(reason.contains("missing")),
            other => panic!("expected Deny on missing plugin, got {:?}", other),
        }
    }

    // ===================================================================
    // 5c additions: Exists, InSet, Always
    // ===================================================================

    #[test]
    fn exists_distinguishes_missing_from_falsy() {
        let mut bag = AttributeBag::new();
        bag.set("args.flag", false);
        // Key is present with a falsy value — IsTrue says false, Exists says true.
        assert!(!eval_condition(
            &Condition::IsTrue {
                key: "args.flag".into()
            },
            &bag
        ));
        assert!(eval_condition(
            &Condition::Exists {
                key: "args.flag".into()
            },
            &bag
        ));
        // Missing key — Exists is false.
        assert!(!eval_condition(
            &Condition::Exists {
                key: "args.nonexistent".into()
            },
            &bag
        ));
    }

    #[test]
    fn in_set_member_and_non_member() {
        let mut bag = AttributeBag::new();
        bag.set("subject.type", "user");
        bag.set(
            "allowed_types",
            std::collections::HashSet::from(["user".to_string(), "service".to_string()]),
        );

        assert!(eval_condition(
            &Condition::InSet {
                value_key: "subject.type".into(),
                set_key: "allowed_types".into(),
                negate: false,
            },
            &bag
        ));

        bag.set("subject.type", "agent");
        assert!(!eval_condition(
            &Condition::InSet {
                value_key: "subject.type".into(),
                set_key: "allowed_types".into(),
                negate: false,
            },
            &bag
        ));
    }

    #[test]
    fn in_set_negate() {
        let mut bag = AttributeBag::new();
        bag.set("subject.type", "agent");
        bag.set(
            "blocked_types",
            std::collections::HashSet::from(["service".to_string()]),
        );

        // agent is not in blocked_types → not in → true
        assert!(eval_condition(
            &Condition::InSet {
                value_key: "subject.type".into(),
                set_key: "blocked_types".into(),
                negate: true,
            },
            &bag
        ));
    }

    #[test]
    fn in_set_missing_keys_resolve_to_false() {
        let mut bag = AttributeBag::new();
        // Both missing → in = false → not in = true (spec §2.6 missing→false
        // applies to the underlying `in` lookup; negate flips it).
        assert!(!eval_condition(
            &Condition::InSet {
                value_key: "x".into(),
                set_key: "y".into(),
                negate: false,
            },
            &bag
        ));
        assert!(eval_condition(
            &Condition::InSet {
                value_key: "x".into(),
                set_key: "y".into(),
                negate: true,
            },
            &bag
        ));
    }

    #[test]
    fn always_evaluates_true() {
        let mut bag = AttributeBag::new();
        assert!(eval_expression(&Expression::Always, &bag));
    }

    #[test]
    fn always_rule_unconditional_deny() {
        let mut bag = AttributeBag::new();
        let r = Rule {
            condition: Expression::Always,
            effects: vec![Effect::Deny {
                reason: Some("unconditional".into()),
                code: None,
            }],
            source: "test".into(),
        };
        match evaluate_rules(&[r], &bag) {
            Decision::Deny { reason, .. } => assert_eq!(reason.as_deref(), Some("unconditional")),
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    // ===================================================================
    // 5c-v/vi: async step evaluator with mock resolvers
    // ===================================================================

    use crate::step::{
        PdpCall, PdpDecision, PdpDialect, PdpError, PdpResolver, PluginError, PluginInvocation,
        PluginInvoker, PluginOutcome,
    };
    use async_trait::async_trait;

    /// PDP resolver that returns the decision baked into it. Doesn't
    /// inspect call.args — tests assert on call.dialect / on the decision
    /// flow, not on Cedar/OPA-specific arg parsing.
    struct FakePdp {
        decision: Decision,
    }
    #[async_trait]
    impl PdpResolver for FakePdp {
        fn dialect(&self) -> PdpDialect {
            PdpDialect::Cedar
        }
        async fn evaluate(
            &self,
            _call: &PdpCall,
            _bag: &AttributeBag,
        ) -> Result<PdpDecision, PdpError> {
            Ok(PdpDecision {
                decision: self.decision.clone(),
                diagnostics: vec![],
            })
        }
    }

    /// PDP resolver that returns an error — exercises fail-closed path.
    struct ErroringPdp;
    #[async_trait]
    impl PdpResolver for ErroringPdp {
        fn dialect(&self) -> PdpDialect {
            PdpDialect::Cedar
        }
        async fn evaluate(
            &self,
            _call: &PdpCall,
            _bag: &AttributeBag,
        ) -> Result<PdpDecision, PdpError> {
            Err(PdpError::Dispatch("simulated PDP outage".into()))
        }
    }

    /// Elicitation invoker that dispatches fine but resolves to a human
    /// *denial* — exercises the "genuine no halts unconditionally" path.
    struct DenyingElicitor;
    #[async_trait]
    impl crate::step::ElicitationInvoker for DenyingElicitor {
        async fn dispatch(
            &self,
            _step: &crate::step::ElicitStep,
            resolved_from: &str,
        ) -> Result<crate::step::ElicitationDispatch, crate::step::ElicitationError> {
            Ok(crate::step::ElicitationDispatch {
                id: "deny-1".into(),
                approver: Some(resolved_from.to_string()),
                intent_id: None,
                expires_at: None,
            })
        }
        async fn check(
            &self,
            _step: &crate::step::ElicitStep,
            _id: &str,
        ) -> Result<crate::step::ElicitationStatus, crate::step::ElicitationError> {
            Ok(crate::step::ElicitationStatus::Resolved {
                outcome: crate::step::ElicitationOutcome::Denied,
            })
        }
        async fn validate(
            &self,
            _step: &crate::step::ElicitStep,
            _id: &str,
        ) -> Result<crate::step::ElicitationValidation, crate::step::ElicitationError> {
            // Never reached on a denial — the arm halts before validate.
            unreachable!("validate must not run on a denied elicitation")
        }
    }

    /// Elicitation invoker that dispatches fine but never resolves —
    /// every `check` returns Pending. Exercises the suspend path.
    struct StillPendingElicitor;
    #[async_trait]
    impl crate::step::ElicitationInvoker for StillPendingElicitor {
        async fn dispatch(
            &self,
            _step: &crate::step::ElicitStep,
            resolved_from: &str,
        ) -> Result<crate::step::ElicitationDispatch, crate::step::ElicitationError> {
            Ok(crate::step::ElicitationDispatch {
                id: "pending-1".into(),
                approver: Some(resolved_from.to_string()),
                intent_id: Some("intent-xyz".into()),
                expires_at: Some("2026-12-31T00:00:00Z".into()),
            })
        }
        async fn check(
            &self,
            _step: &crate::step::ElicitStep,
            _id: &str,
        ) -> Result<crate::step::ElicitationStatus, crate::step::ElicitationError> {
            Ok(crate::step::ElicitationStatus::Pending)
        }
        async fn validate(
            &self,
            _step: &crate::step::ElicitStep,
            _id: &str,
        ) -> Result<crate::step::ElicitationValidation, crate::step::ElicitationError> {
            unreachable!("validate must not run while pending")
        }
    }

    /// Plugin invoker keyed by name → outcome.
    struct FakePlugin {
        decisions: std::collections::HashMap<String, Decision>,
    }
    #[async_trait]
    impl PluginInvoker for FakePlugin {
        async fn invoke(
            &self,
            name: &str,
            _bag: &AttributeBag,
            _invocation: PluginInvocation<'_>,
        ) -> Result<PluginOutcome, PluginError> {
            match self.decisions.get(name) {
                Some(d) => Ok(PluginOutcome {
                    decision: d.clone(),
                    taints: vec![],
                    modified_value: None,
                }),
                None => Err(PluginError::NotFound(name.into())),
            }
        }
    }

    /// Null invoker — fails any plugin call (for PDP-only tests).
    struct NullPlugins;
    #[async_trait]
    impl PluginInvoker for NullPlugins {
        async fn invoke(
            &self,
            name: &str,
            _bag: &AttributeBag,
            _invocation: PluginInvocation<'_>,
        ) -> Result<PluginOutcome, PluginError> {
            Err(PluginError::NotFound(name.into()))
        }
    }

    fn pdp_step(decision_diagnostic_label: &str) -> Effect {
        Effect::Pdp {
            call: PdpCall {
                dialect: PdpDialect::Cedar,
                args: serde_yaml::Value::String(decision_diagnostic_label.into()),
            },
            on_deny: vec![],
            on_allow: vec![],
        }
    }

    #[tokio::test]
    async fn steps_rule_only_path() {
        let mut bag = AttributeBag::new();
        let steps = vec![Effect::When {
            condition: Expression::Always,
            body: vec![Effect::Allow],
            source: "test".into(),
        }];
        let r = evaluate_effects(
            &steps,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await;
        assert_eq!(r.decision, Decision::Allow);
    }

    #[tokio::test]
    async fn pdp_allow_continues() {
        let mut bag = AttributeBag::new();
        let steps = vec![pdp_step("dummy")];
        let pdp: Arc<dyn PdpResolver> = Arc::new(FakePdp {
            decision: Decision::Allow,
        });
        assert_eq!(
            evaluate_effects(
                &steps,
                &mut bag,
                &pdp,
                &null_plugins(),
                &noop_delegations(),
                &noop_elicitations(),
                crate::step::DispatchPhase::Pre,
                &mut crate::route::RoutePayload::new(serde_json::Value::Null)
            )
            .await
            .decision,
            Decision::Allow,
        );
    }

    #[tokio::test]
    async fn pdp_deny_returns_deny() {
        let mut bag = AttributeBag::new();
        let steps = vec![pdp_step("dummy")];
        let pdp: Arc<dyn PdpResolver> = Arc::new(FakePdp {
            decision: Decision::Deny {
                reason: Some("forbidden".into()),
                rule_source: "pdp".into(),
            },
        });
        match evaluate_effects(
            &steps,
            &mut bag,
            &pdp,
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny { reason, .. } => assert_eq!(reason.as_deref(), Some("forbidden")),
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn pdp_on_deny_reaction_can_override_reason() {
        // PDP denies, on_deny reaction includes a more specific deny rule that
        // fires before the PDP's deny is returned.
        let mut bag = AttributeBag::new();
        let steps = vec![Effect::Pdp {
            call: PdpCall {
                dialect: PdpDialect::Cedar,
                args: serde_yaml::Value::Null,
            },
            on_deny: vec![Effect::When {
                condition: Expression::Always,
                body: vec![Effect::Deny {
                    reason: Some("reaction took over".into()),
                    code: None,
                }],
                source: "on_deny[0]".into(),
            }],
            on_allow: vec![],
        }];
        let pdp: Arc<dyn PdpResolver> = Arc::new(FakePdp {
            decision: Decision::Deny {
                reason: Some("pdp original".into()),
                rule_source: "p".into(),
            },
        });
        match evaluate_effects(
            &steps,
            &mut bag,
            &pdp,
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert_eq!(reason.as_deref(), Some("reaction took over"));
                assert_eq!(rule_source, "on_deny[0]");
            },
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn pdp_on_allow_can_deny() {
        // PDP allows, but an on_allow reaction can still deny (e.g., a
        // taint check that fails). Outcome: deny.
        let mut bag = AttributeBag::new();
        let steps = vec![Effect::Pdp {
            call: PdpCall {
                dialect: PdpDialect::Cedar,
                args: serde_yaml::Value::Null,
            },
            on_deny: vec![],
            on_allow: vec![Effect::When {
                condition: Expression::Always,
                body: vec![Effect::Deny {
                    reason: Some("reaction veto".into()),
                    code: None,
                }],
                source: "on_allow[0]".into(),
            }],
        }];
        let pdp: Arc<dyn PdpResolver> = Arc::new(FakePdp {
            decision: Decision::Allow,
        });
        match evaluate_effects(
            &steps,
            &mut bag,
            &pdp,
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny { reason, .. } => assert_eq!(reason.as_deref(), Some("reaction veto")),
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn pdp_error_is_fail_closed() {
        let mut bag = AttributeBag::new();
        let steps = vec![pdp_step("dummy")];
        match evaluate_effects(
            &steps,
            &mut bag,
            &(Arc::new(ErroringPdp) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny { reason, .. } => {
                assert!(reason.unwrap().contains("PDP error"));
            },
            d => panic!("expected Deny on PDP error, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn plugin_allow_continues_deny_halts() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(FakePlugin {
            decisions: std::collections::HashMap::from([
                ("ok_plugin".to_string(), Decision::Allow),
                (
                    "blocking_plugin".to_string(),
                    Decision::Deny {
                        reason: Some("rate limit hit".into()),
                        rule_source: "plugin".into(),
                    },
                ),
            ]),
        });

        let allow_only = vec![Effect::Plugin {
            name: "ok_plugin".into(),
        }];
        assert_eq!(
            evaluate_effects(
                &allow_only,
                &mut bag,
                &(Arc::new(FakePdp {
                    decision: Decision::Allow
                }) as Arc<dyn PdpResolver>),
                &plugins,
                &noop_delegations(),
                &noop_elicitations(),
                crate::step::DispatchPhase::Pre,
                &mut crate::route::RoutePayload::new(serde_json::Value::Null)
            )
            .await
            .decision,
            Decision::Allow,
        );

        let with_deny = vec![
            Effect::Plugin {
                name: "ok_plugin".into(),
            },
            Effect::Plugin {
                name: "blocking_plugin".into(),
            },
        ];
        match evaluate_effects(
            &with_deny,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &plugins,
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny { reason, .. } => assert_eq!(reason.as_deref(), Some("rate limit hit")),
            d => panic!("expected Deny from blocking_plugin, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn plugin_error_is_fail_closed() {
        let mut bag = AttributeBag::new();
        let plugins: std::sync::Arc<dyn PluginInvoker> = std::sync::Arc::new(FakePlugin {
            decisions: Default::default(),
        });
        let steps = vec![Effect::Plugin {
            name: "missing".into(),
        }];
        match evaluate_effects(
            &steps,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &plugins,
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
        .decision
        {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert!(reason.unwrap().contains("missing"));
                assert!(rule_source.contains("missing"));
            },
            d => panic!("expected Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn taint_step_always_continues_and_accumulates() {
        let mut bag = AttributeBag::new();
        let steps = vec![
            Effect::Taint {
                label: "PII".into(),
                scopes: vec![crate::pipeline::TaintScope::Session],
            },
            // A later rule should still fire — taint doesn't short-circuit.
            Effect::When {
                condition: Expression::Always,
                body: vec![Effect::Deny {
                    reason: Some("after taint".into()),
                    code: None,
                }],
                source: "p[1]".into(),
            },
        ];
        let eval = evaluate_effects(
            &steps,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await;
        match eval.decision {
            Decision::Deny { reason, .. } => assert_eq!(reason.as_deref(), Some("after taint")),
            d => panic!("expected Deny from rule after Taint, got {:?}", d),
        }
        // Step::Taint should have been accumulated into the phase's taints
        // before the deny landed — audit needs to see what tainted before
        // the policy halted.
        assert_eq!(eval.taints.len(), 1);
        assert_eq!(eval.taints[0].label, "PII");
        assert_eq!(
            eval.taints[0].scopes,
            vec![crate::pipeline::TaintScope::Session]
        );
    }

    // ----- E2: FieldOp end-to-end through evaluate_steps -----

    #[tokio::test]
    async fn field_op_in_do_redacts_args_during_pre_phase() {
        // Sketches the demo case: when condition holds, redact args.ssn
        // — verifies the dispatcher walks effects, lifts the FieldOp
        // out, and rewrites the payload.
        let mut bag = AttributeBag::new();
        // Predicate is the rule's `when:`; here we make it always true.
        let stages = vec![Stage::Redact { condition: None }];
        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::FieldOp {
                path: "args.ssn".into(),
                stages,
            }],
            source: "demo.policy[0]".into(),
        };
        let steps = vec![Effect::from(rule)];
        let mut payload = crate::route::RoutePayload::new(json!({
            "ssn": "123-45-6789",
            "name": "Jane",
        }));

        let eval = evaluate_effects(
            &steps,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;

        assert_eq!(eval.decision, Decision::Allow);
        assert!(eval.args_modified, "FieldOp should flag args_modified");
        // The ssn field should now read `[REDACTED]` (the stock value
        // the Stage::Redact applier writes when no when-clause is set).
        assert_eq!(
            payload.args.get("ssn").and_then(|v| v.as_str()),
            Some("[REDACTED]")
        );
        // Other fields untouched.
        assert_eq!(
            payload.args.get("name").and_then(|v| v.as_str()),
            Some("Jane")
        );
    }

    #[tokio::test]
    async fn field_op_targeting_result_in_pre_phase_is_skipped() {
        // A `result.X | ...` op encountered during the Pre phase is a
        // no-op — the result hasn't been produced yet. Same rule body
        // can be reused across phases without branching.
        let mut bag = AttributeBag::new();
        let stages = vec![Stage::Redact { condition: None }];
        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::FieldOp {
                path: "result.ssn".into(),
                stages,
            }],
            source: "demo.policy[0]".into(),
        };
        let mut payload = crate::route::RoutePayload::new(json!({}));
        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;
        assert_eq!(eval.decision, Decision::Allow);
        assert!(!eval.args_modified);
        assert!(!eval.result_modified);
    }

    #[tokio::test]
    async fn field_op_with_invalid_path_denies() {
        // Path missing the `args.` / `result.` prefix is an author bug
        // — fail closed with a clear violation rather than silently
        // doing nothing.
        let mut bag = AttributeBag::new();
        let stages = vec![Stage::Redact { condition: None }];
        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::FieldOp {
                path: "ssn".into(), // missing prefix
                stages,
            }],
            source: "demo.policy[0]".into(),
        };
        let mut payload = crate::route::RoutePayload::new(json!({"ssn": "x"}));
        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;
        match eval.decision {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert!(reason.unwrap_or_default().contains("must start with"));
                assert_eq!(rule_source, "demo.policy[0]");
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    // ----- E3: Sequential / Parallel orchestration -----

    #[tokio::test]
    async fn sequential_runs_effects_in_order_until_deny() {
        // A Sequential block runs each effect in order. Allow-only
        // effects pass through; the first Deny halts the rest of the
        // sequential body AND the parent step.
        let mut bag = AttributeBag::new();
        let mut payload = crate::route::RoutePayload::new(json!({}));

        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::Sequential(vec![
                Effect::Allow,
                Effect::Deny {
                    reason: Some("blocked by sequential".into()),
                    code: Some("seq.test".into()),
                },
                Effect::Allow, // unreachable
            ])],
            source: "test.policy[0]".into(),
        };

        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;

        match eval.decision {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert_eq!(reason.as_deref(), Some("blocked by sequential"));
                // The `code` override on the effect won — `seq.test`
                // rather than the rule's `test.policy[0]` source.
                assert_eq!(rule_source, "seq.test");
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn parallel_allows_when_no_branch_denies() {
        // Both branches are no-op Allow → overall Continue → route Allow.
        let mut bag = AttributeBag::new();
        let mut payload = crate::route::RoutePayload::new(json!({}));

        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::Parallel(vec![
                Effect::Allow,
                Effect::Taint {
                    label: "audit_branch".into(),
                    scopes: vec![crate::pipeline::TaintScope::Session],
                },
            ])],
            source: "test.policy[0]".into(),
        };

        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;

        assert_eq!(eval.decision, Decision::Allow);
        // Taints from parallel branches accumulate into the outer.
        assert_eq!(eval.taints.len(), 1);
        assert_eq!(eval.taints[0].label, "audit_branch");
    }

    #[tokio::test]
    async fn parallel_denies_when_any_branch_denies() {
        // One Allow, one Deny — overall Deny.
        let mut bag = AttributeBag::new();
        let mut payload = crate::route::RoutePayload::new(json!({}));

        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::Parallel(vec![
                Effect::Allow,
                Effect::Deny {
                    reason: Some("branch 1 denied".into()),
                    code: None,
                },
            ])],
            source: "test.policy[0]".into(),
        };

        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;

        match eval.decision {
            Decision::Deny { reason, .. } => {
                assert_eq!(reason.as_deref(), Some("branch 1 denied"));
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn parallel_picks_first_index_halt_not_first_to_complete() {
        // When two branches both deny, the one with the lower index
        // in the effects list wins — not the one that physically
        // finishes first.
        let mut bag = AttributeBag::new();
        let mut payload = crate::route::RoutePayload::new(json!({}));

        let rule = Rule {
            condition: Expression::Always,
            effects: vec![Effect::Parallel(vec![
                Effect::Deny {
                    reason: Some("idx-0".into()),
                    code: None,
                },
                Effect::Deny {
                    reason: Some("idx-1".into()),
                    code: None,
                },
            ])],
            source: "test.policy[0]".into(),
        };

        let eval = evaluate_effects(
            &vec![Effect::from(rule)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &noop_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut payload,
        )
        .await;

        match eval.decision {
            Decision::Deny { reason, .. } => {
                assert_eq!(reason.as_deref(), Some("idx-0"), "lower-index halt wins");
            },
            other => panic!("expected Deny, got {:?}", other),
        }
    }

    // ----- Effect::Elicit (human-in-the-loop) -----

    /// Run a single `Effect::Elicit` against the given invoker + bag.
    /// Seeds the `user.manager` attribute (the `from` these tests use) so
    /// it resolves — an unresolved attribute `from` now fails closed, which
    /// its own dedicated test covers.
    async fn run_elicit(
        effect: Effect,
        elicitations: &Arc<dyn crate::step::ElicitationInvoker>,
        bag: &mut AttributeBag,
    ) -> StepsEvaluation {
        if bag.get_string("user.manager").is_none() {
            bag.set("user.manager", "manager@corp.com");
        }
        evaluate_effects(
            &[effect],
            bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            elicitations,
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await
    }

    #[tokio::test]
    async fn elicit_auto_approve_allows_and_records_facts() {
        let mut bag = AttributeBag::new();
        let eval = run_elicit(elicit_effect(None, None), &auto_elicitations(), &mut bag).await;
        assert_eq!(eval.decision, Decision::Allow);
        // Resolved facts land in the bag for downstream rules / audit.
        assert_eq!(bag.get_string("elicitation.status"), Some("resolved"));
        assert_eq!(bag.get_string("elicitation.outcome"), Some("approved"));
        assert_eq!(
            bag.get_string("elicitation.approver"),
            Some("manager@corp.com")
        );
        assert_eq!(bag.get_string("elicitation.intent_id"), Some("auto-intent"));
    }

    #[tokio::test]
    async fn elicit_scope_satisfied_allows() {
        // Sufficiency check passes: amount within the approved bound.
        let mut bag = AttributeBag::new();
        bag.set("args.amount", 100_i64);
        let eval = run_elicit(
            elicit_effect(Some("args.amount <= 25000"), None),
            &auto_elicitations(),
            &mut bag,
        )
        .await;
        assert_eq!(eval.decision, Decision::Allow);
    }

    #[tokio::test]
    async fn elicit_scope_unsatisfied_denies_even_when_approved() {
        // The human approved, but the live args exceed scope — the runtime
        // sufficiency check fails closed regardless of the genuine approval.
        let mut bag = AttributeBag::new();
        bag.set("args.amount", 50_000_i64);
        let eval = run_elicit(
            elicit_effect(Some("args.amount <= 25000"), None),
            &auto_elicitations(),
            &mut bag,
        )
        .await;
        match eval.decision {
            Decision::Deny { reason, .. } => {
                assert!(
                    reason
                        .as_deref()
                        .unwrap_or("")
                        .contains("scope not satisfied"),
                    "got: {:?}",
                    reason
                );
            },
            d => panic!("expected scope Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn elicit_noop_dispatch_fails_closed() {
        // No channel wired → dispatch errors → fail closed (default deny).
        let mut bag = AttributeBag::new();
        let eval = run_elicit(elicit_effect(None, None), &noop_elicitations(), &mut bag).await;
        match eval.decision {
            Decision::Deny {
                reason,
                rule_source,
            } => {
                assert!(reason.as_deref().unwrap_or("").contains("dispatch failed"));
                assert_eq!(rule_source, "route.test.policy[0]");
            },
            d => panic!("expected fail-closed Deny, got {:?}", d),
        }
    }

    #[tokio::test]
    async fn elicit_on_error_continue_allows_on_dispatch_failure() {
        // Same noop failure, but on_error=continue lets the pipeline pass.
        let mut bag = AttributeBag::new();
        let eval = run_elicit(
            elicit_effect(None, Some("continue")),
            &noop_elicitations(),
            &mut bag,
        )
        .await;
        assert_eq!(eval.decision, Decision::Allow);
    }

    #[tokio::test]
    async fn elicit_denied_halts_regardless_of_on_error() {
        // A genuine human "no" halts even with on_error=continue — on_error
        // governs failures, not a valid denial.
        let denier: Arc<dyn crate::step::ElicitationInvoker> = Arc::new(DenyingElicitor);
        let mut bag = AttributeBag::new();
        let eval = run_elicit(elicit_effect(None, Some("continue")), &denier, &mut bag).await;
        match eval.decision {
            Decision::Deny { reason, .. } => {
                assert!(reason
                    .as_deref()
                    .unwrap_or("")
                    .contains("denied by approver"));
            },
            d => panic!("expected denial Deny, got {:?}", d),
        }
        assert_eq!(bag.get_string("elicitation.outcome"), Some("denied"));
    }

    #[tokio::test]
    async fn elicit_pending_suspends_not_denies() {
        // Unresolved elicitation → the phase suspends: decision stays Allow
        // and a pending bundle is carried out (the host emits -32120). This
        // is the key difference from the fail-closed placeholder.
        let pending: Arc<dyn crate::step::ElicitationInvoker> = Arc::new(StillPendingElicitor);
        let mut bag = AttributeBag::new();
        let eval = run_elicit(elicit_effect(None, None), &pending, &mut bag).await;

        assert_eq!(eval.decision, Decision::Allow, "pending is not a deny");
        let bundle = eval.pending.expect("pending bundle carried out");
        assert_eq!(bundle.id, "pending-1");
        assert_eq!(bundle.plugin_name, "manager-approver");
        assert_eq!(bundle.approver.as_deref(), Some("manager@corp.com"));
        assert_eq!(bundle.intent_id.as_deref(), Some("intent-xyz"));
        assert_eq!(bundle.channel.as_deref(), Some("ciba"));
        assert_eq!(bundle.expires_at.as_deref(), Some("2026-12-31T00:00:00Z"));
        assert_eq!(bundle.source, "route.test.policy[0]");
        // Bag reflects the in-flight state for audit.
        assert_eq!(bag.get_string("elicitation.status"), Some("pending"));
    }

    #[tokio::test]
    async fn elicit_pending_even_with_on_error_continue() {
        // on_error governs *failures*; a genuine pending is not a failure,
        // so on_error=continue must NOT collapse it into a plain allow —
        // the bundle still surfaces so the host suspends.
        let pending: Arc<dyn crate::step::ElicitationInvoker> = Arc::new(StillPendingElicitor);
        let mut bag = AttributeBag::new();
        let eval = run_elicit(elicit_effect(None, Some("continue")), &pending, &mut bag).await;
        assert_eq!(eval.decision, Decision::Allow);
        assert!(
            eval.pending.is_some(),
            "pending must survive on_error=continue"
        );
    }

    #[tokio::test]
    async fn elicit_pending_short_circuits_later_effects() {
        // A pending elicitation suspends the phase: effects after it in the
        // list do not run this pass. A trailing `deny` proves it — if the
        // phase kept walking, the decision would be Deny, not Allow+pending.
        let pending: Arc<dyn crate::step::ElicitationInvoker> = Arc::new(StillPendingElicitor);
        let mut bag = AttributeBag::new();
        bag.set("user.manager", "manager@corp.com");
        let effects = vec![elicit_effect(None, None), deny("should not run")];
        let eval = evaluate_effects(
            &effects,
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &pending,
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await;
        assert_eq!(eval.decision, Decision::Allow, "trailing deny must not run");
        assert!(eval.pending.is_some());
    }

    #[tokio::test]
    async fn elicit_unresolved_from_attribute_fails_closed() {
        // `from` is an attribute reference (`user.manager`) that isn't in
        // the bag. It must fail closed BEFORE dispatch — never send the
        // literal `"user.manager"` to the channel as a bogus login_hint.
        let mut bag = AttributeBag::new(); // note: no user.manager seeded
        let eval = evaluate_effects(
            &[elicit_effect(None, None)],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &auto_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await;
        match eval.decision {
            Decision::Deny { reason, .. } => {
                let r = reason.as_deref().unwrap_or("");
                assert!(r.contains("did not resolve"), "got: {r}");
                assert!(r.contains("user.manager"), "got: {r}");
            },
            d => panic!(
                "expected fail-closed Deny on unresolved `from`, got {:?}",
                d
            ),
        }
        // Nothing was dispatched, so no elicitation id was recorded.
        assert!(bag.get_string("elicitation.id").is_none());
    }

    #[tokio::test]
    async fn elicit_literal_from_passes_through_unresolved() {
        // A `from` that is a literal identity (an email — not an attribute
        // reference) is used verbatim even when it isn't a bag key.
        let mut bag = AttributeBag::new();
        let effect = Effect::Elicit(crate::step::ElicitStep {
            from: "alice@corp.com".into(),
            ..match elicit_effect(None, None) {
                Effect::Elicit(e) => e,
                _ => unreachable!(),
            }
        });
        let eval = evaluate_effects(
            &[effect],
            &mut bag,
            &(Arc::new(FakePdp {
                decision: Decision::Allow,
            }) as Arc<dyn PdpResolver>),
            &null_plugins(),
            &noop_delegations(),
            &auto_elicitations(),
            crate::step::DispatchPhase::Pre,
            &mut crate::route::RoutePayload::new(serde_json::Value::Null),
        )
        .await;
        assert_eq!(eval.decision, Decision::Allow);
        // The literal identity flowed through as the resolved approver.
        assert_eq!(
            bag.get_string("elicitation.approver"),
            Some("alice@corp.com")
        );
    }

    #[test]
    fn looks_like_attribute_ref_classification() {
        // Attribute references: dotted identifier paths.
        assert!(looks_like_attribute_ref("user.manager"));
        assert!(looks_like_attribute_ref("claim.manager"));
        assert!(looks_like_attribute_ref("user.sub"));
        // Literals: emails, display names, bare tokens.
        assert!(!looks_like_attribute_ref("alice@corp.com"));
        assert!(!looks_like_attribute_ref("Alice Smith"));
        assert!(!looks_like_attribute_ref("alice"));
        assert!(!looks_like_attribute_ref(""));
    }
}