behavior_contracts/

behavior.rs

//! behavior — component-graph IR + `run_behavior` unified execution IF.
//!
//! Ports the TS reference `ts/src/behavior.ts` (scp-ir-architecture.md §5-§7).
//!
//! A portable component-graph IR (`components[]{name, inputPorts, body[], output, plan}`)
//! is executed on top of the existing COMMON primitives: `run_plan` (stage exec /
//! Skip propagation / Policy Kind) + `evaluate` (Expression IR). The concrete
//! per-component (leaf) implementation is resolved by name through a [`ComponentExec`]
//! seam — the boundary-injected handler registry (IR + {effects,config,hooks}).
//!
//! Body node kinds:
//!   - componentRef: `{id, component, ports, parent?, bindField?, relationKind?, policy?}`
//!   - map:          `{id, map:{over, as, component, ports, when?, into?, batched?, parent?,
//!                    relationKind?, policy?}}` (`when`/`into`/`batched` are behaviorVersion 2)
//!   - cond:         `{id, cond:{if, then, else, parent?}}` (pure Expression; no handler)
//!
//! behaviorVersion 2 (bc#22):
//!   - `map.when`    — per-element guard, lowered to `{cond:[when,true,false]}` (strict bool;
//!     non-bool fails closed with TYPE_MISMATCH). Falsy elements are skipped (handler not
//!     called, excluded from the result list; order preserved).
//!   - `map.into`    — zip-attach: the node result becomes the `over` list with each
//!     guard-passing element shallow-copied and augmented with `into: <handler result>`
//!     (same length/order as `over`; skipped elements pass through unchanged). Parent
//!     results are never mutated. Non-object kept elements fail closed
//!     (MAP_INTO_ELEMENT_NOT_OBJECT).
//!   - `map.batched` — evaluates the ports of all guard-passing elements first, then calls
//!     the handler ONCE with `{items:[<ports>...]}`; the handler must return a list aligned
//!     to items (violations fail closed with MAP_BATCH_RESULT_MISMATCH). Zero kept elements
//!     means the handler is not called and the result is the empty list.
//!   - handler ctx   — every handler call carries the node identity (nodeId + component)
//!     via the [`ComponentExec::exec_ctx`] seam (additive; default forwards to `exec`).

#[cfg(feature = "ir")]
use crate::expr::evaluate as evaluate_expression;
use crate::expr::ExprFailure;
#[cfg(feature = "ir")]
use crate::plan::{run_plan, ExecutionPlanSpec, OpSpec, RelationKind};
use crate::plan::{ExecOutcome, PlanFailure};
use crate::value::Value;
#[cfg(feature = "ir")]
use serde_json::Value as J;

/// Stable behavior-execution failure codes (in addition to plan/expr codes).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BehaviorFailureCode {
    UnknownComponent,
    UnknownNodeKind,
    MapOverNotArray,
    MapIntoElementNotObject,
    MapBatchResultMismatch,
    FanoutOverNotArray,
    FanoutBatchResultMismatch,
    UnknownEntry,
}

impl BehaviorFailureCode {
    pub fn as_str(self) -> &'static str {
        match self {
            BehaviorFailureCode::UnknownComponent => "UNKNOWN_COMPONENT",
            BehaviorFailureCode::UnknownNodeKind => "UNKNOWN_NODE_KIND",
            BehaviorFailureCode::MapOverNotArray => "MAP_OVER_NOT_ARRAY",
            BehaviorFailureCode::MapIntoElementNotObject => "MAP_INTO_ELEMENT_NOT_OBJECT",
            BehaviorFailureCode::MapBatchResultMismatch => "MAP_BATCH_RESULT_MISMATCH",
            BehaviorFailureCode::FanoutOverNotArray => "FANOUT_OVER_NOT_ARRAY",
            BehaviorFailureCode::FanoutBatchResultMismatch => "FANOUT_BATCH_RESULT_MISMATCH",
            BehaviorFailureCode::UnknownEntry => "UNKNOWN_ENTRY",
        }
    }
}

/// A unified failure for `run_behavior`, carrying a stable string code so the
/// conformance runner can compare against expression / plan / behavior codes alike.
#[derive(Debug, Clone)]
pub struct BehaviorError {
    code: String,
    pub message: String,
}

impl BehaviorError {
    /// Construct a coded behavior error. Public so the straight-line codegen surface
    /// (bc#39) can raise the same fail-closed codes/messages as `run_behavior`
    /// (`UNKNOWN_COMPONENT` / `UNKNOWN_ENTRY` / `UNKNOWN_NODE_KIND`) without depending
    /// on the private `BehaviorFailureCode` enum.
    pub fn new(code: impl Into<String>, message: impl Into<String>) -> Self {
        BehaviorError {
            code: code.into(),
            message: message.into(),
        }
    }

    /// The stable string code used for conformance matching.
    pub fn code(&self) -> &str {
        &self.code
    }
}

impl std::fmt::Display for BehaviorError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code, self.message)
    }
}
impl std::error::Error for BehaviorError {}

impl From<ExprFailure> for BehaviorError {
    fn from(e: ExprFailure) -> Self {
        BehaviorError {
            code: e.code.as_str().to_string(),
            message: e.message,
        }
    }
}
impl From<PlanFailure> for BehaviorError {
    fn from(e: PlanFailure) -> Self {
        BehaviorError {
            code: e.code.as_str().to_string(),
            message: e.message,
        }
    }
}

#[cfg(feature = "ir")]
fn bfail<T>(code: BehaviorFailureCode, message: impl Into<String>) -> Result<T, BehaviorError> {
    Err(BehaviorError {
        code: code.as_str().to_string(),
        message: message.into(),
    })
}

/// The boundary-injected handler registry seam (§7.1). Resolves a catalog name to
/// a leaf implementation and executes it with evaluated ports (+ the map element
/// `bound` value, if any). Returns `None` when the name is unknown so `run_behavior`
/// can fail closed with `UNKNOWN_COMPONENT`.
pub trait ComponentExec {
    fn exec(
        &mut self,
        component: &str,
        ports: &[(String, Value)],
        bound: Option<&Value>,
    ) -> Option<ExecOutcome>;

    /// behaviorVersion 2: like [`ComponentExec::exec`] but carrying the handler ctx
    /// (`node_id` = body node id, `component` = catalog name). `run_behavior` calls
    /// this seam for every handler invocation. The default forwards to `exec`, so
    /// existing implementations keep working unchanged; override it to observe the
    /// node identity (error context / tracing).
    fn exec_ctx(
        &mut self,
        node_id: &str,
        component: &str,
        ports: &[(String, Value)],
        bound: Option<&Value>,
    ) -> Option<ExecOutcome> {
        let _ = node_id;
        self.exec(component, ports, bound)
    }
}

/// Blanket forwarding impl so a **trait object** handler registry
/// (`&mut dyn ComponentExec`) itself satisfies [`ComponentExec`] (bc#68).
///
/// The generic straight-line / typed codegen surface exposes generic-by-value
/// entries (`bind<H: ComponentExec>(handlers: H)` / `run_*<H: ComponentExec>(
/// handlers: &mut H, …)`). A consumer that holds its handler registry behind a
/// trait object (`handlers: &mut dyn ComponentExec` — exactly how [`run_behavior`]
/// takes it) could not name a concrete `H` to hand those entries. This impl makes
/// `H = &mut dyn ComponentExec` a valid instantiation, so the trait-object handler
/// drives the generated straight-line/typed module through the *unchanged* generic
/// path (no `bind_dyn` variant needed, no emitter change). It is purely additive:
/// every existing concrete `H` keeps working exactly as before.
///
/// Both methods forward to the underlying `dyn` (a plain reborrow), so calling a
/// generated module through `&mut dyn` runs the SAME straight-line/typed code as a
/// concrete handler — it never falls back to `run_behavior`.
impl ComponentExec for &mut dyn ComponentExec {
    fn exec(
        &mut self,
        component: &str,
        ports: &[(String, Value)],
        bound: Option<&Value>,
    ) -> Option<ExecOutcome> {
        (**self).exec(component, ports, bound)
    }

    fn exec_ctx(
        &mut self,
        node_id: &str,
        component: &str,
        ports: &[(String, Value)],
        bound: Option<&Value>,
    ) -> Option<ExecOutcome> {
        (**self).exec_ctx(node_id, component, ports, bound)
    }
}

#[cfg(feature = "ir")]
fn scope_get<'a>(scope: &'a [(String, Value)], key: &str) -> Option<&'a Value> {
    scope.iter().find(|(k, _)| k == key).map(|(_, v)| v)
}

/// fanout_dedup_drop — THE ONE dedup/drop definition (behaviorVersion 3) — Rust twin of behavior.ts
/// fanoutDedupDrop. Applies first-seen dedupe (by the body's `dedupe_key` field) + dangling drop
/// (Null / non-object / absent-key body when `drop=="dangling"`) + implicitSource strip to the aligned
/// raw list; returns the connection items (the caller wraps into {items,cursor:Null}). Byte-equal to
/// the TS/Python/PHP/Go definitions. Bodies are `Value::Obj`; a Null / non-Obj body is dangling.
#[cfg(feature = "ir")]
fn fanout_dedup_drop(
    aligned_bodies: Vec<Value>,
    dedupe_key: &str,
    drop: &str,
    implicit_source: Option<&str>,
) -> Vec<Value> {
    use std::collections::HashSet;
    let mut items: Vec<Value> = Vec::with_capacity(aligned_bodies.len());
    let mut seen: HashSet<String> = HashSet::new();
    for body in aligned_bodies.into_iter() {
        let fields = match &body {
            Value::Obj(f) => Some(f),
            _ => None,
        };
        let key_val = fields.and_then(|f| f.iter().find(|(k, _)| k == dedupe_key).map(|(_, v)| v));
        let has_key = matches!(key_val, Some(v) if !matches!(v, Value::Null));
        if !has_key {
            if drop == "dangling" {
                continue;
            }
            items.push(body); // drop:none keeps a dangling body
            continue;
        }
        let seen_key = match key_val.unwrap() {
            Value::Str(s) => format!("s:{s}"),
            other => format!("j:{other:?}"),
        };
        if seen.contains(&seen_key) {
            continue;
        }
        seen.insert(seen_key);
        match (implicit_source, fields) {
            (Some(src), Some(f)) if f.iter().any(|(k, _)| k == src) => {
                let stripped: Vec<(String, Value)> =
                    f.iter().filter(|(k, _)| k != src).cloned().collect();
                items.push(Value::Obj(stripped));
            }
            _ => items.push(body),
        }
    }
    items
}

#[cfg(feature = "ir")]
fn node_kind(n: &J) -> Result<&'static str, BehaviorError> {
    if n.get("fanout").is_some() {
        Ok("fanout")
    } else if n.get("map").is_some() {
        Ok("map")
    } else if n.get("cond").is_some() {
        Ok("cond")
    } else if n.get("component").is_some() {
        Ok("componentRef")
    } else {
        bfail(
            BehaviorFailureCode::UnknownNodeKind,
            format!(
                "body node '{}' is not componentRef/map/cond/fanout",
                n.get("id").and_then(|v| v.as_str()).unwrap_or("?")
            ),
        )
    }
}

#[cfg(feature = "ir")]
fn node_sub(n: &J) -> &J {
    if let Some(f) = n.get("fanout") {
        f
    } else if let Some(m) = n.get("map") {
        m
    } else if let Some(c) = n.get("cond") {
        c
    } else {
        n
    }
}

#[cfg(feature = "ir")]
fn node_parent(n: &J) -> Option<&str> {
    node_sub(n).get("parent").and_then(|v| v.as_str())
}

#[cfg(feature = "ir")]
fn node_bind_field(n: &J) -> Option<String> {
    if n.get("map").is_some() || n.get("cond").is_some() || n.get("fanout").is_some() {
        return None;
    }
    n.get("bindField")
        .and_then(|v| v.as_str())
        .map(str::to_string)
}

#[cfg(feature = "ir")]
fn node_relation_kind(n: &J) -> Option<RelationKind> {
    if n.get("cond").is_some() {
        return None;
    }
    let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
        node_sub(n)
    } else {
        n
    };
    match sub.get("relationKind").and_then(|v| v.as_str()) {
        Some("connection") => Some(RelationKind::Connection),
        Some(_) => Some(RelationKind::Single),
        None => None,
    }
}

#[cfg(feature = "ir")]
fn node_relation_kind_str(n: &J) -> Option<&str> {
    if n.get("cond").is_some() {
        return None;
    }
    let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
        node_sub(n)
    } else {
        n
    };
    sub.get("relationKind").and_then(|v| v.as_str())
}

#[cfg(feature = "ir")]
fn node_policy(n: &J) -> Option<String> {
    if n.get("cond").is_some() {
        return None;
    }
    let sub = if n.get("map").is_some() || n.get("fanout").is_some() {
        node_sub(n)
    } else {
        n
    };
    sub.get("policy")
        .and_then(|v| v.as_str())
        .map(str::to_string)
}

#[cfg(feature = "ir")]
fn eval_ports(ports: &J, scope: &[(String, Value)]) -> Result<Vec<(String, Value)>, BehaviorError> {
    let obj = ports.as_object().ok_or_else(|| BehaviorError {
        code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
        message: "ports must be an object".into(),
    })?;
    let mut out = Vec::with_capacity(obj.len());
    for (k, v) in obj {
        out.push((k.clone(), evaluate_expression(v, scope)?));
    }
    Ok(out)
}

/// Parse an `ExecutionPlanSpec` from a component's `plan` field (`null`/absent → None).
#[cfg(feature = "ir")]
fn parse_plan(plan: Option<&J>) -> Option<ExecutionPlanSpec> {
    let p = plan?;
    if p.is_null() {
        return None;
    }
    let groups = p
        .get("groups")?
        .as_array()?
        .iter()
        .map(|g| {
            g.as_array()
                .map(|row| {
                    row.iter()
                        .filter_map(|i| i.as_u64().map(|x| x as usize))
                        .collect()
                })
                .unwrap_or_default()
        })
        .collect();
    let concurrency = p.get("concurrency").and_then(|c| c.as_i64()).unwrap_or(1);
    Some(ExecutionPlanSpec {
        groups,
        concurrency,
    })
}

/// Run a component-graph IR (scp-ir-architecture.md §7).
///
/// * `ir`      — the portable component-graph IR (`{components:[...]}`).
/// * `handlers`— the boundary-injected [`ComponentExec`] seam (catalog name → leaf impl).
/// * `input`   — the entry component's inputPorts bindings (param values).
/// * `entry`   — the component name to run (`None` = first component).
///
/// Returns the evaluated `output` (Φ merge) or a coded [`BehaviorError`].
#[cfg(feature = "ir")]
pub fn run_behavior(
    ir: &J,
    handlers: &mut dyn ComponentExec,
    input: &[(String, Value)],
    entry: Option<&str>,
) -> Result<Value, BehaviorError> {
    let components = ir
        .get("components")
        .and_then(|c| c.as_array())
        .ok_or_else(|| BehaviorError {
            code: BehaviorFailureCode::UnknownEntry.as_str().to_string(),
            message: "IR.components must be an array".into(),
        })?;
    let comp = match entry {
        Some(name) => components
            .iter()
            .find(|c| c.get("name").and_then(|n| n.as_str()) == Some(name)),
        None => components.first(),
    };
    let comp = match comp {
        Some(c) => c,
        None => {
            return bfail(
                BehaviorFailureCode::UnknownEntry,
                format!("component '{}' not found in IR", entry.unwrap_or("<first>")),
            )
        }
    };

    let body: Vec<J> = comp
        .get("body")
        .and_then(|b| b.as_array())
        .cloned()
        .unwrap_or_default();

    let index_of = |id: &str| {
        body.iter()
            .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
    };

    // Build OpSpecs for run_plan staging (parent id → index via Wire).
    let mut ops: Vec<OpSpec> = Vec::with_capacity(body.len());
    for n in &body {
        let parent = node_parent(n).and_then(index_of);
        ops.push(OpSpec {
            id: n
                .get("id")
                .and_then(|v| v.as_str())
                .unwrap_or_default()
                .to_string(),
            parent,
            bind_field: node_bind_field(n),
            relation_kind: node_relation_kind(n),
            policy: node_policy(n),
        });
    }

    run_behavior_inner(
        &body,
        &ops,
        parse_plan(comp.get("plan")),
        comp.get("output").cloned().unwrap_or(J::Null),
        &bind_input_scope(comp, input),
        handlers,
    )
}

/// bind_input_scope — the entry component's input bindings, read against its declaration
/// (`inputPorts`).
///
/// An input port declared `{opt:T}` (PortSchema `required:false`) is OMITTABLE and its value domain is
/// `T | null`. "Key omitted" and "key present as null" are two spellings of the same "no value", so an
/// omitted key binds to null. This only reads a declaration the portable IR already carries — no new
/// information, no new operator.
///
/// Required ports and undeclared names are NOT bound: referencing one unbound is still
/// `UNKNOWN_BINDING` (the fail-closed for a typo / an unwired param stands — only a port DECLARED
/// omittable is relaxed).
#[cfg(feature = "ir")]
fn bind_input_scope(comp: &J, input: &[(String, Value)]) -> Vec<(String, Value)> {
    let mut scope: Vec<(String, Value)> = input.to_vec();
    if let Some(ports) = comp.get("inputPorts").and_then(|p| p.as_object()) {
        for (name, schema) in ports {
            let optional = schema.get("required").and_then(|r| r.as_bool()) == Some(false);
            if optional && !scope.iter().any(|(k, _)| k == name) {
                scope.push((name.clone(), Value::Null));
            }
        }
    }
    scope
}

/// Single-closure implementation: builds the exec closure that owns `handlers`,
/// `results`, and `pending_err`, delegates staging to `run_plan`, then evaluates
/// the component `output`. Kept separate to sidestep split-borrow layering.
#[cfg(feature = "ir")]
fn run_behavior_inner(
    body: &[J],
    ops: &[OpSpec],
    plan: Option<ExecutionPlanSpec>,
    output: J,
    input: &[(String, Value)],
    handlers: &mut dyn ComponentExec,
) -> Result<Value, BehaviorError> {
    let index_of = |id: &str| {
        body.iter()
            .position(|n| n.get("id").and_then(|v| v.as_str()) == Some(id))
    };

    let mut results: Vec<(String, Value)> = Vec::new();
    let mut pending_err: Option<BehaviorError> = None;

    let run = {
        let results_cell = &mut results;
        let err_cell = &mut pending_err;
        let exec = |op: &OpSpec, _bound: Option<&Value>| -> ExecOutcome {
            if err_cell.is_some() {
                return ExecOutcome::Error("aborted".into());
            }
            let idx = match index_of(&op.id) {
                Some(i) => i,
                None => {
                    *err_cell = Some(BehaviorError {
                        code: BehaviorFailureCode::UnknownNodeKind.as_str().to_string(),
                        message: format!("no body node for op '{}'", op.id),
                    });
                    return ExecOutcome::Error("aborted".into());
                }
            };
            let node = &body[idx];

            let base_scope = |extra: Option<(&str, &Value)>| -> Vec<(String, Value)> {
                let mut s: Vec<(String, Value)> = input.to_vec();
                for (k, v) in results_cell.iter() {
                    s.push((k.clone(), v.clone()));
                }
                if let Some((k, v)) = extra {
                    s.push((k.to_string(), v.clone()));
                }
                s
            };

            let outcome = match node_kind(node) {
                Err(e) => {
                    *err_cell = Some(e);
                    return ExecOutcome::Error("aborted".into());
                }
                Ok("cond") => {
                    let c = node.get("cond").unwrap();
                    let cond_expr = serde_json::json!({
                        "cond": [c.get("if"), c.get("then"), c.get("else")]
                    });
                    match evaluate_expression(&cond_expr, &base_scope(None)) {
                        Ok(v) => ExecOutcome::Ok(v),
                        Err(e) => {
                            *err_cell = Some(e.into());
                            return ExecOutcome::Error("aborted".into());
                        }
                    }
                }
                Ok("map") => {
                    let m = node.get("map").unwrap();
                    let over = match evaluate_expression(
                        m.get("over").unwrap_or(&J::Null),
                        &base_scope(None),
                    ) {
                        Ok(v) => v,
                        Err(e) => {
                            *err_cell = Some(e.into());
                            return ExecOutcome::Error("aborted".into());
                        }
                    };
                    let arr = match &over {
                        Value::Arr(a) => a.clone(),
                        _ => {
                            *err_cell = Some(BehaviorError {
                                code: BehaviorFailureCode::MapOverNotArray.as_str().to_string(),
                                message: format!("map '{}': 'over' is not an array", op.id),
                            });
                            return ExecOutcome::Error("aborted".into());
                        }
                    };
                    let component = m.get("component").and_then(|v| v.as_str()).unwrap_or("");
                    let as_name = m.get("as").and_then(|v| v.as_str()).unwrap_or("$");
                    let ports_j = m
                        .get("ports")
                        .cloned()
                        .unwrap_or(J::Object(Default::default()));
                    let when = m.get("when");
                    let batched = m.get("batched").and_then(|v| v.as_bool()).unwrap_or(false);
                    let into = m.get("into").and_then(|v| v.as_str());

                    // per-element guard (v2): lowered to `{cond:[when,true,false]}`
                    // (strict bool — a non-bool guard fails closed via the cond operator).
                    let keep = |scope: &[(String, Value)]| -> Result<bool, BehaviorError> {
                        match when {
                            None => Ok(true),
                            Some(w) => {
                                let cond_expr = serde_json::json!({ "cond": [w, true, false] });
                                Ok(matches!(
                                    evaluate_expression(&cond_expr, scope)?,
                                    Value::Bool(true)
                                ))
                            }
                        }
                    };

                    let mut kept_idx: Vec<usize> = Vec::new(); // guard-passing over indices (for into)
                    let collected: Vec<Value>;
                    if batched {
                        // batched (v2): evaluate all guard-passing ports first, call handler ONCE.
                        let mut items: Vec<Value> = Vec::new();
                        for (i, el) in arr.iter().enumerate() {
                            let scope = base_scope(Some((as_name, el)));
                            match keep(&scope) {
                                Ok(false) => continue,
                                Ok(true) => {}
                                Err(e) => {
                                    *err_cell = Some(e);
                                    return ExecOutcome::Error("aborted".into());
                                }
                            }
                            match eval_ports(&ports_j, &scope) {
                                Ok(p) => items.push(Value::Obj(p)),
                                Err(e) => {
                                    *err_cell = Some(e);
                                    return ExecOutcome::Error("aborted".into());
                                }
                            }
                            kept_idx.push(i);
                        }
                        if items.is_empty() {
                            collected = Vec::new(); // all filtered: handler is not called
                        } else {
                            let want = items.len();
                            let batch_ports = vec![("items".to_string(), Value::Arr(items))];
                            match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
                                None => {
                                    *err_cell = Some(BehaviorError {
                                        code: BehaviorFailureCode::UnknownComponent
                                            .as_str()
                                            .to_string(),
                                        message: format!(
                                            "component '{component}' has no handler (fail-closed)"
                                        ),
                                    });
                                    return ExecOutcome::Error("aborted".into());
                                }
                                Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
                                Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
                                    collected = r;
                                }
                                Some(ExecOutcome::Ok(_)) => {
                                    *err_cell = Some(BehaviorError {
                                        code: BehaviorFailureCode::MapBatchResultMismatch
                                            .as_str()
                                            .to_string(),
                                        message: format!(
                                            "map '{}': batched handler must return a list aligned to items (want {want})",
                                            op.id
                                        ),
                                    });
                                    return ExecOutcome::Error("aborted".into());
                                }
                            }
                        }
                    } else {
                        let mut out: Vec<Value> = Vec::with_capacity(arr.len());
                        for (i, el) in arr.iter().enumerate() {
                            let scope = base_scope(Some((as_name, el)));
                            match keep(&scope) {
                                Ok(false) => continue,
                                Ok(true) => {}
                                Err(e) => {
                                    *err_cell = Some(e);
                                    return ExecOutcome::Error("aborted".into());
                                }
                            }
                            let ports = match eval_ports(&ports_j, &scope) {
                                Ok(p) => p,
                                Err(e) => {
                                    *err_cell = Some(e);
                                    return ExecOutcome::Error("aborted".into());
                                }
                            };
                            match handlers.exec_ctx(&op.id, component, &ports, Some(el)) {
                                None => {
                                    *err_cell = Some(BehaviorError {
                                        code: BehaviorFailureCode::UnknownComponent
                                            .as_str()
                                            .to_string(),
                                        message: format!(
                                            "component '{component}' has no handler (fail-closed)"
                                        ),
                                    });
                                    return ExecOutcome::Error("aborted".into());
                                }
                                Some(ExecOutcome::Error(e)) => return ExecOutcome::Error(e),
                                Some(ExecOutcome::Ok(v)) => out.push(v),
                            }
                            kept_idx.push(i);
                        }
                        collected = out;
                    }

                    match into {
                        None => ExecOutcome::Ok(Value::Arr(collected)),
                        Some(key) => {
                            // into (v2): result = over list with kept elements shallow-copied
                            // + `into` key attached; skipped elements pass through unchanged.
                            let mut augmented: Vec<Value> = Vec::with_capacity(arr.len());
                            let mut k = 0usize;
                            for (i, el) in arr.iter().enumerate() {
                                if k < kept_idx.len() && kept_idx[k] == i {
                                    let fields = match el {
                                        Value::Obj(f) => f.clone(),
                                        _ => {
                                            *err_cell = Some(BehaviorError {
                                                code: BehaviorFailureCode::MapIntoElementNotObject
                                                    .as_str()
                                                    .to_string(),
                                                message: format!(
                                                    "map '{}': 'into' requires object elements (element {i} is not an object)",
                                                    op.id
                                                ),
                                            });
                                            return ExecOutcome::Error("aborted".into());
                                        }
                                    };
                                    let mut fields = fields;
                                    let val = collected[k].clone();
                                    match fields.iter_mut().find(|(fk, _)| fk == key) {
                                        Some(slot) => slot.1 = val,
                                        None => fields.push((key.to_string(), val)),
                                    }
                                    augmented.push(Value::Obj(fields));
                                    k += 1;
                                } else {
                                    augmented.push(el.clone());
                                }
                            }
                            ExecOutcome::Ok(Value::Arr(augmented))
                        }
                    }
                }
                Ok("fanout") => {
                    // fanout (v3): over (id-list) → deduped ONE batched handler → dedupe/drop/strip →
                    // connection {items, cursor:null}. The MAP_BATCH_RESULT_MISMATCH alignment does NOT
                    // apply (the result is a connection, not an aligned list).
                    let f = node.get("fanout").unwrap();
                    let over = match evaluate_expression(
                        f.get("over").unwrap_or(&J::Null),
                        &base_scope(None),
                    ) {
                        Ok(v) => v,
                        Err(e) => {
                            *err_cell = Some(e.into());
                            return ExecOutcome::Error("aborted".into());
                        }
                    };
                    let arr = match &over {
                        Value::Arr(a) => a.clone(),
                        _ => {
                            *err_cell = Some(BehaviorError {
                                code: BehaviorFailureCode::FanoutOverNotArray.as_str().to_string(),
                                message: format!("fanout '{}': 'over' is not an array", op.id),
                            });
                            return ExecOutcome::Error("aborted".into());
                        }
                    };
                    let component = f.get("component").and_then(|v| v.as_str()).unwrap_or("");
                    let as_name = f.get("as").and_then(|v| v.as_str()).unwrap_or("$");
                    let ports_j = f
                        .get("ports")
                        .cloned()
                        .unwrap_or(J::Object(Default::default()));
                    let dedupe_key = f.get("dedupeKey").and_then(|v| v.as_str()).unwrap_or("");
                    let drop = f.get("drop").and_then(|v| v.as_str()).unwrap_or("dangling");
                    let implicit_source = f.get("implicitSource").and_then(|v| v.as_str());
                    let mut items: Vec<Value> = Vec::with_capacity(arr.len());
                    for el in arr.iter() {
                        let scope = base_scope(Some((as_name, el)));
                        match eval_ports(&ports_j, &scope) {
                            Ok(p) => items.push(Value::Obj(p)),
                            Err(e) => {
                                *err_cell = Some(e);
                                return ExecOutcome::Error("aborted".into());
                            }
                        }
                    }
                    if items.is_empty() {
                        ExecOutcome::Ok(Value::Obj(vec![
                            ("items".into(), Value::Arr(vec![])),
                            ("cursor".into(), Value::Null),
                        ]))
                    } else {
                        let want = items.len();
                        let batch_ports = vec![("items".to_string(), Value::Arr(items))];
                        match handlers.exec_ctx(&op.id, component, &batch_ports, None) {
                            None => {
                                *err_cell = Some(BehaviorError {
                                    code: BehaviorFailureCode::UnknownComponent
                                        .as_str()
                                        .to_string(),
                                    message: format!(
                                        "component '{component}' has no handler (fail-closed)"
                                    ),
                                });
                                ExecOutcome::Error("aborted".into())
                            }
                            Some(ExecOutcome::Error(e)) => ExecOutcome::Error(e),
                            Some(ExecOutcome::Ok(Value::Arr(r))) if r.len() == want => {
                                let deduped =
                                    fanout_dedup_drop(r, dedupe_key, drop, implicit_source);
                                ExecOutcome::Ok(Value::Obj(vec![
                                    ("items".into(), Value::Arr(deduped)),
                                    ("cursor".into(), Value::Null),
                                ]))
                            }
                            Some(ExecOutcome::Ok(_)) => {
                                *err_cell = Some(BehaviorError {
                                    code: BehaviorFailureCode::FanoutBatchResultMismatch
                                        .as_str()
                                        .to_string(),
                                    message: format!(
                                        "fanout '{}': batched handler must return a list aligned to the deduped id list (want {want})",
                                        op.id
                                    ),
                                });
                                ExecOutcome::Error("aborted".into())
                            }
                        }
                    }
                }
                Ok(_) => {
                    let component = node.get("component").and_then(|v| v.as_str()).unwrap_or("");
                    let ports_j = node
                        .get("ports")
                        .cloned()
                        .unwrap_or(J::Object(Default::default()));
                    let ports = match eval_ports(&ports_j, &base_scope(None)) {
                        Ok(p) => p,
                        Err(e) => {
                            *err_cell = Some(e);
                            return ExecOutcome::Error("aborted".into());
                        }
                    };
                    match handlers.exec_ctx(&op.id, component, &ports, None) {
                        None => {
                            *err_cell = Some(BehaviorError {
                                code: BehaviorFailureCode::UnknownComponent.as_str().to_string(),
                                message: format!(
                                    "component '{component}' has no handler (fail-closed)"
                                ),
                            });
                            return ExecOutcome::Error("aborted".into());
                        }
                        Some(o) => o,
                    }
                }
            };

            if let ExecOutcome::Ok(v) = &outcome {
                results_cell.push((op.id.clone(), v.clone()));
            }
            outcome
        };

        run_plan(plan.as_ref(), ops, exec)
    };

    // A BehaviorError raised inside exec takes precedence over the synthetic plan
    // OP_FAILED it produced (we injected an "aborted" error to unwind run_plan).
    if let Some(e) = pending_err {
        return Err(e);
    }
    let run = run?;

    // Skipped nodes contribute their unproduced representation to the scope so the
    // component `output` can `ref`/`coalesce` over them.
    for (i, id_val) in run.skipped.iter().enumerate() {
        let _ = i;
        if let Some(idx) = index_of(id_val) {
            let rk = node_relation_kind_str(&body[idx]);
            let unproduced = if rk == Some("connection") {
                Value::Obj(vec![
                    ("items".into(), Value::Arr(vec![])),
                    ("cursor".into(), Value::Null),
                ])
            } else {
                Value::Null
            };
            if scope_get(&results, id_val).is_none() {
                results.push((id_val.clone(), unproduced));
            }
        }
    }

    let scope: Vec<(String, Value)> = {
        let mut s: Vec<(String, Value)> = input.to_vec();
        s.extend(results.iter().cloned());
        s
    };
    Ok(evaluate_expression(&output, &scope)?)
}