Aura/crates/aura-engine/src/construction.rs

//! Per-op-fallible blueprint construction — the eager half of the §A gate split
//! (#157). A `GraphSession` applies `Op`s one at a time, running the
//! eagerly-decidable checks (name resolution, edge kind-match, the >1-cover
//! double-wire arm, param bind) at the offending op; the only-at-end-decidable
//! checks (0-cover totality, param-namespace injectivity, root-role boundness)
//! run holistically in `finish`. Both cadences call the SAME §A predicates — no
//! second validator (C24). The node vocabulary is an injected closed resolver
//! (`Fn(&str)->Option<PrimitiveBuilder>`), so the engine stays domain-free
//! (invariant 9).

use std::collections::{HashMap, HashSet};

use aura_core::{BindOpError, NodeSchema, PrimitiveBuilder, Scalar, ScalarKind};

use crate::blueprint::{
    check_param_namespace_injective, check_root_roles_bound, edge_kind_check, validate_wiring,
    BlueprintNode, Composite, Gang, GangMember, OutField, Role,
};
use crate::builder::{resolve_input_slot, resolve_output_field, PortResolveError};
use crate::harness::{Edge, Target};
use crate::CompileError;

/// One construction act — the engine-side op vocabulary, 1:1 with the document.
/// Port references are dotted by-identifier (`"node.field"` / `"node.slot"`),
/// split at the first `.`. (serde `Deserialize` is iteration 2.)
#[derive(Debug, PartialEq)]
pub enum Op {
    /// Reserve a bound root source role of `kind`.
    Source { role: String, kind: ScalarKind },
    /// Reserve an open input role (wired by an enclosing graph).
    Input { role: String },
    /// Add a node of `type_id`, optionally named `as_name`, with `bind` params.
    Add { type_id: String, as_name: Option<String>, bind: Vec<(String, Scalar)> },
    /// Fan a role into one or more `"node.slot"` consumers.
    Feed { role: String, into: Vec<String> },
    /// Wire `"node.field"` -> `"node.slot"`.
    Connect { from: String, to: String },
    /// Re-export `"node.field"` at the boundary under `as_name`.
    Expose { from: String, as_name: String },
    /// Fuse two or more sibling params into one public knob (#61).
    Gang { as_name: String, into: Vec<String> },
}

/// A per-op construction fault, by-identifier so the cause names the op.
#[derive(Debug, PartialEq)]
pub enum OpError {
    UnknownNodeType(String),
    /// An interior node-identifier collision (the `names` namespace).
    DuplicateIdentifier(String),
    /// A boundary output-name collision (the separate `out_names` namespace) — a
    /// distinct fault class from an interior node-id clash, so callers can tell the
    /// two apart from the error value alone.
    DuplicateOutput(String),
    DuplicateRole(String),
    UnknownIdentifier(String),
    UnknownRole(String),
    MalformedPort(String),
    UnknownInPort { node: String, name: String },
    AmbiguousInPort { node: String, name: String },
    UnknownOutPort { node: String, name: String },
    AmbiguousOutPort { node: String, name: String },
    KindMismatch { from: String, to: String, producer: ScalarKind, consumer: ScalarKind },
    SlotAlreadyWired { node: String, slot: String },
    /// Wiring `from -> to` would close a dataflow cycle among the interior edges:
    /// `to`'s node already reaches `from`'s node (or it is a self-edge). The only
    /// legal feedback path is an explicit delay/state node, so a cycle makes the
    /// blueprint non-acyclic (domain invariant 5 / ledger C9).
    WouldCycle { from: String, to: String },
    BadParam { node: String, err: BindOpError },
    /// Holistic totality fault, by-identifier: an interior input slot covered by
    /// no edge or role target (the finalize 0-cover arm).
    UnconnectedPort { node: String, slot: String },
    /// Holistic role-kind fault, by-identifier: a root input role fans into slots
    /// of differing scalar kinds.
    RoleKindMismatch { role: String },
    /// Holistic root fault, by-identifier: a root input role has no bound source.
    UnboundRootRole { role: String },
    /// Gang members must share one scalar kind.
    GangKindMismatch { member: String, expected: ScalarKind, got: ScalarKind },
    /// The param is already claimed by an earlier gang.
    AlreadyGanged { node: String, param: String },
    /// A gang needs at least two members.
    GangArity { gang: String },
    /// A holistic finalize fault (totality / injectivity / unbound root role),
    /// wrapping the unchanged engine gate's `CompileError`.
    Incomplete(CompileError),
}

/// A per-op-fallible blueprint accumulator. Holds the same interior data a
/// `Composite` is built from, plus the incremental coverage map (eager
/// double-wire front-run) and the identifier→index maps the document references.
pub struct GraphSession<'v> {
    name: String,
    vocab: &'v dyn Fn(&str) -> Option<PrimitiveBuilder>,
    nodes: Vec<BlueprintNode>,
    schemas: Vec<NodeSchema>,
    ids: Vec<String>,
    names: HashMap<String, usize>,
    edges: Vec<Edge>,
    roles: Vec<(String, Option<ScalarKind>, Vec<Target>)>,
    role_names: HashMap<String, usize>,
    out: Vec<OutField>,
    out_names: HashSet<String>,
    coverage: HashMap<(usize, usize), usize>,
    gangs: Vec<Gang>,
}

impl<'v> GraphSession<'v> {
    /// Start a session for a composite named `name`, resolving node types through
    /// the injected closed vocabulary `vocab`.
    pub fn new(name: impl Into<String>, vocab: &'v dyn Fn(&str) -> Option<PrimitiveBuilder>) -> Self {
        Self {
            name: name.into(),
            vocab,
            nodes: Vec::new(),
            schemas: Vec::new(),
            ids: Vec::new(),
            names: HashMap::new(),
            edges: Vec::new(),
            roles: Vec::new(),
            role_names: HashMap::new(),
            out: Vec::new(),
            out_names: HashSet::new(),
            coverage: HashMap::new(),
            gangs: Vec::new(),
        }
    }

    /// Apply one op, running the eager checks. The first `Err` is the op-script's
    /// stop point (`replay` attributes it to the op index).
    pub fn apply(&mut self, op: Op) -> Result<(), OpError> {
        match op {
            Op::Source { role, kind } => self.add_role(role, Some(kind)),
            Op::Input { role } => self.add_role(role, None),
            Op::Add { type_id, as_name, bind } => self.add_node(type_id, as_name, bind),
            Op::Feed { role, into } => self.feed(role, into),
            Op::Connect { from, to } => self.connect(from, to),
            Op::Expose { from, as_name } => self.expose(from, as_name),
            Op::Gang { as_name, into } => self.gang(as_name, into),
        }
    }

    fn add_role(&mut self, role: String, kind: Option<ScalarKind>) -> Result<(), OpError> {
        if self.role_names.contains_key(&role) {
            return Err(OpError::DuplicateRole(role));
        }
        let idx = self.roles.len();
        self.role_names.insert(role.clone(), idx);
        self.roles.push((role, kind, Vec::new()));
        Ok(())
    }

    fn add_node(
        &mut self,
        type_id: String,
        as_name: Option<String>,
        bind: Vec<(String, Scalar)>,
    ) -> Result<(), OpError> {
        let mut builder =
            (self.vocab)(&type_id).ok_or_else(|| OpError::UnknownNodeType(type_id.clone()))?;
        let id = as_name.unwrap_or_else(|| builder.node_name());
        for (slot, value) in bind {
            builder = builder
                .try_bind(&slot, value)
                .map_err(|err| OpError::BadParam { node: id.clone(), err })?;
        }
        if self.names.contains_key(&id) {
            return Err(OpError::DuplicateIdentifier(id));
        }
        // Stamp the chosen identifier onto the builder so the node's own name (and
        // hence its `param_space()` path prefix) matches the session id — without
        // this, two same-type nodes keep the default type-label name and collide on
        // the holistic param-namespace injectivity gate in `finish`.
        let builder = builder.named(&id);
        let node = BlueprintNode::from(builder);
        let schema = node.signature();
        let idx = self.nodes.len();
        self.names.insert(id.clone(), idx);
        self.ids.push(id);
        self.nodes.push(node);
        self.schemas.push(schema);
        Ok(())
    }

    /// Split a dotted `"node.port"` reference at the first `.`.
    fn split_port(dotted: &str) -> Result<(String, String), OpError> {
        match dotted.split_once('.') {
            Some((node, port)) if !node.is_empty() && !port.is_empty() => {
                Ok((node.to_string(), port.to_string()))
            }
            _ => Err(OpError::MalformedPort(dotted.to_string())),
        }
    }

    fn node_index(&self, id: &str) -> Result<usize, OpError> {
        self.names.get(id).copied().ok_or_else(|| OpError::UnknownIdentifier(id.to_string()))
    }

    /// Would adding the interior edge `from_idx -> to_idx` close a dataflow cycle?
    /// It does iff `to_idx` already reaches `from_idx` through the interior edges
    /// added so far (or it is a self-edge, the degenerate 1-node loop). Source/input
    /// role feeds are roots, not interior edges, so they cannot participate. The
    /// only legal feedback is an explicit delay/state node (domain invariant 5 /
    /// ledger C9), so this is the acyclicity gate the eager `connect` op enforces.
    fn closes_cycle(&self, from_idx: usize, to_idx: usize) -> bool {
        let mut stack = vec![to_idx];
        let mut seen = HashSet::new();
        while let Some(n) = stack.pop() {
            if n == from_idx {
                return true;
            }
            if !seen.insert(n) {
                continue;
            }
            for e in &self.edges {
                if e.from == n {
                    stack.push(e.to);
                }
            }
        }
        false
    }

    /// Mark `(node, slot)` covered, rejecting a second cover eagerly (the >1 arm
    /// of the exactly-once invariant — the holistic 0 arm is deferred to finish).
    fn cover(&mut self, node: usize, node_id: &str, slot: usize, slot_name: &str) -> Result<(), OpError> {
        let c = self.coverage.entry((node, slot)).or_insert(0);
        if *c >= 1 {
            return Err(OpError::SlotAlreadyWired { node: node_id.to_string(), slot: slot_name.to_string() });
        }
        *c += 1;
        Ok(())
    }

    fn connect(&mut self, from: String, to: String) -> Result<(), OpError> {
        let (from_node, from_field_name) = Self::split_port(&from)?;
        let (to_node, to_slot_name) = Self::split_port(&to)?;
        let fi = self.node_index(&from_node)?;
        let ti = self.node_index(&to_node)?;
        let from_field = resolve_output_field(&self.schemas[fi], &from_field_name).map_err(|e| match e {
            PortResolveError::Unknown => OpError::UnknownOutPort { node: from_node.clone(), name: from_field_name.clone() },
            PortResolveError::Ambiguous => OpError::AmbiguousOutPort { node: from_node.clone(), name: from_field_name.clone() },
        })?;
        let slot = resolve_input_slot(&self.schemas[ti], &to_slot_name).map_err(|e| match e {
            PortResolveError::Unknown => OpError::UnknownInPort { node: to_node.clone(), name: to_slot_name.clone() },
            PortResolveError::Ambiguous => OpError::AmbiguousInPort { node: to_node.clone(), name: to_slot_name.clone() },
        })?;
        edge_kind_check(&self.schemas[fi], from_field, &self.schemas[ti], slot).map_err(|e| match e {
            CompileError::Bootstrap(crate::BootstrapError::KindMismatch { producer, consumer }) => {
                OpError::KindMismatch { from: from.clone(), to: to.clone(), producer, consumer }
            }
            // `edge_kind_check` over the just-resolved (in-range) field + slot indices
            // can only yield `Ok` or a `KindMismatch`; the index-fault arm is dead.
            _ => unreachable!("edge_kind_check on resolved indices yields only KindMismatch"),
        })?;
        if self.closes_cycle(fi, ti) {
            return Err(OpError::WouldCycle { from, to });
        }
        self.cover(ti, &to_node, slot, &to_slot_name)?;
        self.edges.push(Edge { from: fi, to: ti, slot, from_field });
        Ok(())
    }

    fn feed(&mut self, role: String, into: Vec<String>) -> Result<(), OpError> {
        let ri = *self.role_names.get(&role).ok_or_else(|| OpError::UnknownRole(role.clone()))?;
        // Phase 1: resolve + coverage-check every target without mutating the session,
        // so a later failing target leaves no partial wiring behind (feed is
        // all-or-nothing). Coverage is checked against both prior ops (`self.coverage`)
        // and the slots already claimed earlier in this same fan-out batch.
        let mut resolved: Vec<(usize, usize)> = Vec::with_capacity(into.len());
        for target in &into {
            let (to_node, to_slot_name) = Self::split_port(target)?;
            let ti = self.node_index(&to_node)?;
            let slot = resolve_input_slot(&self.schemas[ti], &to_slot_name).map_err(|e| match e {
                PortResolveError::Unknown => OpError::UnknownInPort { node: to_node.clone(), name: to_slot_name.clone() },
                PortResolveError::Ambiguous => OpError::AmbiguousInPort { node: to_node.clone(), name: to_slot_name.clone() },
            })?;
            let already = self.coverage.get(&(ti, slot)).copied().unwrap_or(0) > 0
                || resolved.contains(&(ti, slot));
            if already {
                return Err(OpError::SlotAlreadyWired { node: to_node, slot: to_slot_name });
            }
            resolved.push((ti, slot));
        }
        // Phase 2: commit — every target validated, so this cannot fail.
        for (ti, slot) in resolved {
            *self.coverage.entry((ti, slot)).or_insert(0) += 1;
            self.roles[ri].2.push(Target { node: ti, slot });
        }
        Ok(())
    }

    fn expose(&mut self, from: String, as_name: String) -> Result<(), OpError> {
        let (from_node, from_field_name) = Self::split_port(&from)?;
        let fi = self.node_index(&from_node)?;
        let field = resolve_output_field(&self.schemas[fi], &from_field_name).map_err(|e| match e {
            PortResolveError::Unknown => OpError::UnknownOutPort { node: from_node.clone(), name: from_field_name.clone() },
            PortResolveError::Ambiguous => OpError::AmbiguousOutPort { node: from_node.clone(), name: from_field_name.clone() },
        })?;
        if !self.out_names.insert(as_name.clone()) {
            return Err(OpError::DuplicateOutput(as_name));
        }
        self.out.push(OutField { node: fi, field, name: as_name });
        Ok(())
    }

    /// Eager arm of the gang gate: resolve every member against the CURRENT
    /// (post-bind) open param lists — a member bound at `Add` has left the open
    /// schema and correctly fails as `UnknownParam`, mirroring `try_bind` — enforce
    /// kind uniformity and single-gang membership, then commit. Structural validity
    /// is re-run holistically at `finish` through `Composite::with_gangs` (one
    /// predicate, two cadences, like `connect`'s eager `edge_kind_check` + the
    /// holistic `validate_wiring`).
    fn gang(&mut self, as_name: String, into: Vec<String>) -> Result<(), OpError> {
        if into.len() < 2 {
            return Err(OpError::GangArity { gang: as_name });
        }
        let mut members: Vec<GangMember> = Vec::with_capacity(into.len());
        let mut kind: Option<ScalarKind> = None;
        for port in &into {
            let (node_name, param_name) = Self::split_port(port)?;
            let ti = self.node_index(&node_name)?;
            let BlueprintNode::Primitive(b) = &self.nodes[ti] else {
                unreachable!("sessions only ever add primitives")
            };
            let hits: Vec<usize> = b
                .params()
                .iter()
                .enumerate()
                .filter(|(_, p)| p.name == param_name)
                .map(|(i, _)| i)
                .collect();
            let idx = match hits.as_slice() {
                [i] => *i,
                [] => {
                    return Err(OpError::BadParam {
                        node: node_name,
                        err: BindOpError::UnknownParam(param_name),
                    })
                }
                _ => {
                    return Err(OpError::BadParam {
                        node: node_name,
                        err: BindOpError::AmbiguousParam(param_name),
                    })
                }
            };
            let member_kind = b.params()[idx].kind;
            match kind {
                None => kind = Some(member_kind),
                Some(expected) if expected != member_kind => {
                    return Err(OpError::GangKindMismatch { member: port.clone(), expected, got: member_kind })
                }
                _ => {}
            }
            let pos = b.original_pos(idx);
            let already_claimed = self.gangs.iter().flat_map(|g| g.members.iter()).any(|m| m.node == ti && m.pos == pos)
                || members.iter().any(|m| m.node == ti && m.pos == pos);
            if already_claimed {
                return Err(OpError::AlreadyGanged { node: node_name, param: param_name });
            }
            members.push(GangMember { node: ti, pos, name: param_name });
        }
        self.gangs.push(Gang { name: as_name, kind: kind.expect(">=2 members checked above"), members });
        Ok(())
    }

    /// The still-unwired interior input slots (build-free introspection): every
    /// added node's declared input slots minus the covered ones, by-identifier.
    pub fn unwired(&self) -> Vec<(String, ScalarKind)> {
        let mut open = Vec::new();
        for (n, schema) in self.schemas.iter().enumerate() {
            for (slot, port) in schema.inputs.iter().enumerate() {
                if self.coverage.get(&(n, slot)).copied().unwrap_or(0) == 0 {
                    open.push((format!("{}.{}", self.ids[n], port.name), port.kind));
                }
            }
        }
        open
    }

    /// Assemble the `Composite` and run the holistic gates (totality,
    /// param-namespace injectivity, root-role boundness) — the SAME engine
    /// predicates the eager path shares, now at end-of-document cadence. The
    /// index-carrying `CompileError`s the gates return are translated back to the
    /// surface's by-identifier `OpError` variants (slot/role names via the session's
    /// retained `ids`/`schemas` and the composite's roles); a fault with no
    /// by-identifier mapping falls through to `OpError::Incomplete`.
    pub fn finish(self) -> Result<Composite, OpError> {
        let roles: Vec<Role> = self
            .roles
            .into_iter()
            .map(|(name, source, targets)| Role { name, targets, source })
            .collect();
        let c = Composite::new(self.name, self.nodes, self.edges, roles, self.out)
            .with_gangs(self.gangs)
            .map_err(OpError::Incomplete)?;
        check_param_namespace_injective(&c.param_space()).map_err(OpError::Incomplete)?;
        validate_wiring(c.nodes(), c.edges(), c.input_roles(), c.output()).map_err(|e| match e {
            CompileError::UnconnectedPort { node, slot } => OpError::UnconnectedPort {
                node: self.ids[node].clone(),
                slot: self.schemas[node].inputs[slot].name.clone(),
            },
            CompileError::RoleKindMismatch { role } => OpError::RoleKindMismatch {
                role: c.input_roles()[role].name.clone(),
            },
            other => OpError::Incomplete(other),
        })?;
        check_root_roles_bound(c.input_roles()).map_err(|e| match e {
            CompileError::UnboundRootRole { role } => OpError::UnboundRootRole {
                role: c.input_roles()[role].name.clone(),
            },
            other => OpError::Incomplete(other),
        })?;
        Ok(c)
    }
}

/// Replay a document of ops from scratch, stopping at the first failing op and
/// naming it by `(op_index, OpError)`; a holistic finalize fault is attributed to
/// index `ops.len()` (the implicit finalize step).
pub fn replay(
    name: impl Into<String>,
    ops: Vec<Op>,
    vocab: &dyn Fn(&str) -> Option<PrimitiveBuilder>,
) -> Result<Composite, (usize, OpError)> {
    let mut s = GraphSession::new(name, vocab);
    let n = ops.len();
    for (i, op) in ops.into_iter().enumerate() {
        s.apply(op).map_err(|e| (i, e))?;
    }
    s.finish().map_err(|e| (n, e))
}

#[cfg(test)]
mod tests {
    use super::{GraphSession, Op, OpError};
    use aura_core::{BindOpError, Scalar, ScalarKind};
    use aura_std::std_vocabulary;

    fn session(name: &str) -> GraphSession<'static> {
        GraphSession::new(name, &std_vocabulary)
    }

    #[test]
    fn add_unknown_type_rejected_at_op() {
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Add { type_id: "Nope".into(), as_name: None, bind: vec![] }),
            Err(OpError::UnknownNodeType("Nope".into()))
        );
    }

    #[test]
    fn add_duplicate_identifier_rejected() {
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("x".into()), bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("x".into()), bind: vec![] }),
            Err(OpError::DuplicateIdentifier("x".into()))
        );
    }

    #[test]
    fn add_bad_bind_param_rejected_at_op() {
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Add {
                type_id: "SMA".into(),
                as_name: Some("fast".into()),
                bind: vec![("length".into(), Scalar::f64(2.0))], // wrong kind (i64 param)
            }),
            Err(OpError::BadParam {
                node: "fast".into(),
                err: BindOpError::KindMismatch {
                    param: "length".into(),
                    expected: ScalarKind::I64,
                    got: ScalarKind::F64,
                },
            })
        );
    }

    #[test]
    fn duplicate_role_rejected() {
        let mut s = session("g");
        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
        assert_eq!(
            s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }),
            Err(OpError::DuplicateRole("price".into()))
        );
    }

    // helper: a fast/slow/sub scaffold sharing a price source
    fn scaffold() -> GraphSession<'static> {
        let mut s = session("g");
        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] }).unwrap();
        s
    }

    #[test]
    fn connect_unknown_out_port_rejected_at_op() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Connect { from: "fast.nope".into(), to: "sub.lhs".into() }),
            Err(OpError::UnknownOutPort { node: "fast".into(), name: "nope".into() })
        );
    }

    #[test]
    fn connect_double_wire_rejected_at_op() {
        let mut s = scaffold();
        s.apply(Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() }).unwrap();
        assert_eq!(
            s.apply(Op::Connect { from: "slow.value".into(), to: "sub.lhs".into() }),
            Err(OpError::SlotAlreadyWired { node: "sub".into(), slot: "lhs".into() })
        );
    }

    #[test]
    fn connect_kind_mismatch_rejected_at_op() {
        // Gt produces bool; Bias.signal consumes f64 -> eager kind mismatch
        let mut s = session("g");
        s.apply(Op::Add { type_id: "Gt".into(), as_name: None, bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] }).unwrap();
        let r = s.apply(Op::Connect { from: "gt.value".into(), to: "bias.signal".into() });
        assert!(matches!(
            r,
            Err(OpError::KindMismatch { producer: ScalarKind::Bool, consumer: ScalarKind::F64, .. })
        ));
    }

    #[test]
    fn feed_into_unknown_role_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Feed { role: "ghost".into(), into: vec!["fast.series".into()] }),
            Err(OpError::UnknownRole("ghost".into()))
        );
    }

    /// `feed` covers each named slot: a successful fan-out registers coverage, so
    /// a second `feed` into an already-covered slot is rejected at the op.
    #[test]
    fn feed_success_covers_each_slot() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Feed {
                role: "price".into(),
                into: vec!["fast.series".into(), "slow.series".into()],
            }),
            Ok(())
        );
        assert_eq!(
            s.apply(Op::Feed { role: "price".into(), into: vec!["fast.series".into()] }),
            Err(OpError::SlotAlreadyWired { node: "fast".into(), slot: "series".into() })
        );
    }

    /// `feed` is all-or-nothing: when a later target in the same fan-out fails, the
    /// earlier (valid) targets are left unwired — the failed op leaks no partial
    /// coverage into the session (so `unwired`/a subsequent op see consistent state).
    #[test]
    fn feed_partial_failure_is_atomic() {
        let mut s = scaffold();
        // fast.series is valid; ghost.series names an unknown node, so the feed fails
        // on its second target.
        assert_eq!(
            s.apply(Op::Feed {
                role: "price".into(),
                into: vec!["fast.series".into(), "ghost.series".into()],
            }),
            Err(OpError::UnknownIdentifier("ghost".into()))
        );
        // fast.series must still be open: the failed feed wired nothing.
        assert!(s.unwired().iter().any(|(p, _)| p == "fast.series"));
    }

    /// An open `Input` role (kind = None) is reserved like a `Source`, sharing the
    /// duplicate-role guard.
    #[test]
    fn input_role_reserved_and_duplicate_rejected() {
        let mut s = session("g");
        assert_eq!(s.apply(Op::Input { role: "ext".into() }), Ok(()));
        assert_eq!(
            s.apply(Op::Input { role: "ext".into() }),
            Err(OpError::DuplicateRole("ext".into()))
        );
    }

    /// A port reference without a `.` separator is rejected as malformed at the op.
    #[test]
    fn connect_malformed_port_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Connect { from: "fastvalue".into(), to: "sub.lhs".into() }),
            Err(OpError::MalformedPort("fastvalue".into()))
        );
    }

    /// A port naming a node that was never added is rejected by-identifier.
    #[test]
    fn connect_unknown_node_identifier_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Connect { from: "ghost.value".into(), to: "sub.lhs".into() }),
            Err(OpError::UnknownIdentifier("ghost".into()))
        );
    }

    /// `expose` re-exports a resolved output field; a second export under the same
    /// boundary name is rejected as a duplicate identifier.
    #[test]
    fn expose_success_and_duplicate_output_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Expose { from: "fast.value".into(), as_name: "out".into() }),
            Ok(())
        );
        assert_eq!(
            s.apply(Op::Expose { from: "slow.value".into(), as_name: "out".into() }),
            Err(OpError::DuplicateOutput("out".into()))
        );
    }

    /// `expose` maps an unresolved output field onto the by-identifier port fault.
    #[test]
    fn expose_unknown_out_port_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Expose { from: "fast.nope".into(), as_name: "out".into() }),
            Err(OpError::UnknownOutPort { node: "fast".into(), name: "nope".into() })
        );
    }

    /// The gang op fuses two sibling params into one public knob; the replayed
    /// composite's `param_space()` carries exactly the gang address.
    #[test]
    fn gang_op_projects_the_param_space() {
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), bind: vec![] },
            Op::Gang { as_name: "length".into(), into: vec!["a.length".into(), "b.length".into()] },
            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["a.series".into(), "b.series".into()] },
            Op::Connect { from: "a.value".into(), to: "sub.lhs".into() },
            Op::Connect { from: "b.value".into(), to: "sub.rhs".into() },
            Op::Expose { from: "sub.value".into(), as_name: "bias".into() },
        ];
        let c = super::replay("sig", ops, &std_vocabulary).expect("replays");
        let names: Vec<String> = c.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, ["length"]);
    }

    /// Eager refusals: unknown member param (a member bound at `Add` has left the
    /// open schema), kind mismatch across members, a doubly-ganged member, a
    /// single-member gang, and an unknown node identifier.
    #[test]
    fn gang_op_eager_refusals() {
        // bound member: `b.length` was bound at Add, so it left the open param
        // list and is unknown to the gang gate, exactly as `try_bind` would see it.
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add {
            type_id: "SMA".into(),
            as_name: Some("b".into()),
            bind: vec![("length".into(), Scalar::i64(5))],
        })
        .unwrap();
        assert_eq!(
            s.apply(Op::Gang { as_name: "length".into(), into: vec!["a.length".into(), "b.length".into()] }),
            Err(OpError::BadParam { node: "b".into(), err: BindOpError::UnknownParam("length".into()) })
        );

        // kind mismatch: SMA.length is i64, Bias.scale is f64.
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Bias".into(), as_name: Some("b".into()), bind: vec![] }).unwrap();
        let r = s.apply(Op::Gang { as_name: "g".into(), into: vec!["a.length".into(), "b.scale".into()] });
        assert!(
            matches!(r, Err(OpError::GangKindMismatch { expected: ScalarKind::I64, got: ScalarKind::F64, .. })),
            "{r:?}"
        );

        // doubly-ganged member: a.length is already claimed by an earlier gang.
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("c".into()), bind: vec![] }).unwrap();
        s.apply(Op::Gang { as_name: "lo".into(), into: vec!["a.length".into(), "b.length".into()] }).unwrap();
        assert_eq!(
            s.apply(Op::Gang { as_name: "hi".into(), into: vec!["a.length".into(), "c.length".into()] }),
            Err(OpError::AlreadyGanged { node: "a".into(), param: "length".into() })
        );

        // single member: a gang needs at least two.
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Gang { as_name: "solo".into(), into: vec!["a.length".into()] }),
            Err(OpError::GangArity { gang: "solo".into() })
        );

        // unknown node: a member names a node that was never added.
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Gang { as_name: "g".into(), into: vec!["a.length".into(), "ghost.length".into()] }),
            Err(OpError::UnknownIdentifier("ghost".into()))
        );
    }

    #[test]
    fn finish_reports_incomplete_on_unwired_slot() {
        // a fully-named scaffold missing the sub.rhs connection -> holistic 0-arm
        let mut s = scaffold();
        s.apply(Op::Feed { role: "price".into(), into: vec!["fast.series".into(), "slow.series".into()] }).unwrap();
        s.apply(Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() }).unwrap();
        // sub.rhs deliberately left unwired
        // (`.err().unwrap()` rather than `.unwrap_err()`: the Ok arm `Composite`
        // holds build closures and is not `Debug`, which `unwrap_err` would require.)
        let err = s.finish().err().unwrap();
        assert!(matches!(&err, OpError::UnconnectedPort { node, slot } if node == "sub" && slot == "rhs"),
            "finalize names the open slot by-identifier, got {err:?}");
    }

    /// A reserved `Input` root role that is fed but never source-bound finishes with
    /// the by-identifier `UnboundRootRole` (not a raw role index).
    #[test]
    fn finish_reports_unbound_input_root_role_by_identifier() {
        let mut s = session("g");
        s.apply(Op::Input { role: "ext".into() }).unwrap();
        s.apply(Op::Add { type_id: "Sub".into(), as_name: Some("sub".into()), bind: vec![] }).unwrap();
        s.apply(Op::Feed { role: "ext".into(), into: vec!["sub.lhs".into(), "sub.rhs".into()] }).unwrap();
        s.apply(Op::Expose { from: "sub.value".into(), as_name: "out".into() }).unwrap();
        let err = s.finish().err().unwrap();
        assert!(matches!(&err, OpError::UnboundRootRole { role } if role == "ext"),
            "an unbound Input root role finishes by-identifier, got {err:?}");
    }

    /// A root role fed into two slots of differing scalar kinds (an `f64` `Sub.lhs`
    /// and a `bool` `And.a`) finishes with the by-identifier `RoleKindMismatch`
    /// naming the role (not a raw role index). `feed` runs no eager kind check, so
    /// this fault surfaces only at finalize and must be translated by-identifier
    /// there (the role-index -> name mapping `c.input_roles()[role].name`).
    #[test]
    fn finish_reports_role_kind_mismatch_by_identifier() {
        let mut s = session("g");
        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
        s.apply(Op::Add { type_id: "Sub".into(), as_name: Some("sub".into()), bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "And".into(), as_name: Some("conj".into()), bind: vec![] }).unwrap();
        // price fans into sub.lhs (f64) AND conj.a (bool): differing slot kinds.
        s.apply(Op::Feed { role: "price".into(), into: vec!["sub.lhs".into(), "conj.a".into()] }).unwrap();
        let err = s.finish().err().unwrap();
        assert!(matches!(&err, OpError::RoleKindMismatch { role } if role == "price"),
            "a role fanning into differing-kind slots finishes by-identifier, got {err:?}");
    }

    #[test]
    fn unwired_lists_open_slots() {
        let mut s = scaffold();
        s.apply(Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() }).unwrap();
        let open = s.unwired();
        // fast.series, slow.series, sub.rhs are still open; sub.lhs is covered
        assert!(open.iter().any(|(p, _)| p == "sub.rhs"));
        assert!(open.iter().any(|(p, _)| p == "fast.series"));
        assert!(!open.iter().any(|(p, _)| p == "sub.lhs"));
    }

    /// DAG invariant (domain invariant 5 / ledger C9): the construction surface
    /// must reject a dataflow cycle. The only legal feedback path is an explicit
    /// delay/state node; a `connect` that closes a cycle among interior nodes
    /// makes the blueprint non-acyclic and must be a construction fault, never a
    /// silent `Ok`.
    #[test]
    fn replay_rejects_dataflow_cycle() {
        // Mirror fieldtests/cycle-0088-construction-op-script/c0088_3m_two_cycle.json:
        // two `Sub` nodes a,b each fed `rhs` from price, then a.value->b.lhs and
        // b.value->a.lhs closing an a<->b 2-cycle. Every interior slot is covered
        // exactly once and the root role is bound, so every current gate (eager
        // per-op cover/kind + holistic totality/injectivity/root-boundness) passes
        // — only an acyclicity check can catch it.
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "Sub".into(), as_name: Some("a".into()), bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: Some("b".into()), bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["a.rhs".into(), "b.rhs".into()] },
            Op::Connect { from: "a.value".into(), to: "b.lhs".into() }, // op 4: a -> b
            Op::Connect { from: "b.value".into(), to: "a.lhs".into() }, // op 5: closes b -> a
            Op::Expose { from: "a.value".into(), as_name: "out".into() },
        ];
        let closing_connect = 5;
        let finalize = ops.len();
        // `.err()` not `.unwrap_err()`: the Ok arm `Composite` holds build closures
        // and is not `Debug`.
        let (idx, _err) = super::replay("g", ops, &aura_std::std_vocabulary)
            .err()
            .expect("a cyclic op-list must be rejected (DAG invariant 5 / C9)");
        // An eager realisation attributes the fault to the closing connect op; a
        // holistic topological check at finish attributes it to the implicit
        // finalize index. Accept either so the GREEN side is free to choose.
        assert!(
            idx == closing_connect || idx == finalize,
            "cycle fault should name the closing connect (op {closing_connect}) \
             or finalize (op {finalize}), got op {idx}"
        );
    }

    /// A self-loop (a node feeding its own input) is the degenerate 1-node cycle
    /// and must be rejected for the same DAG reason (domain invariant 5 / C9).
    #[test]
    fn replay_rejects_self_loop() {
        // single `Sub` s: rhs fed from price, then s.value -> s.lhs closes a 1-node
        // loop. Both slots covered, role bound — only an acyclicity check catches it.
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "Sub".into(), as_name: Some("s".into()), bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["s.rhs".into()] },
            Op::Connect { from: "s.value".into(), to: "s.lhs".into() }, // self-edge
            Op::Expose { from: "s.value".into(), as_name: "out".into() },
        ];
        assert!(
            super::replay("g", ops, &aura_std::std_vocabulary).is_err(),
            "a self-loop op-list must be rejected (DAG invariant 5 / C9)"
        );
    }

    #[test]
    fn replay_stops_at_first_failing_op_naming_index() {
        let ops = vec![
            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] },
            Op::Add { type_id: "Nope".into(), as_name: None, bind: vec![] }, // op 1 fails
            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
        ];
        // `.err().unwrap()` rather than `.unwrap_err()`: the Ok arm `Composite` is
        // not `Debug` (it holds build closures), which `unwrap_err` would require.
        let err = super::replay("g", ops, &aura_std::std_vocabulary).err().unwrap();
        assert_eq!(err, (1, OpError::UnknownNodeType("Nope".into())));
    }

    /// Acceptance-1 (C24 replay-equals-builder): an op-script replayed through the
    /// shared §A gates compiles to the byte-identical `FlatGraph` topology
    /// (`edges` + `sources`) as the same signal root hand-wired on the
    /// `GraphBuilder` surface — the two construction paths are one construction
    /// semantics. Models `builder_harness_compiles_identically_to_hand_wired`
    /// (builder.rs:269), sink-free over the zero-arg vocabulary.
    #[test]
    fn replay_built_signal_compiles_identically_to_graphbuilder() {
        use crate::GraphBuilder;
        use aura_core::Scalar;
        use aura_std::{Bias, Sma, Sub};

        // param space of the flat root: fast.length(i64), slow.length(i64),
        // bias.scale(f64) — same vector applied to both sides, so it cancels.
        let params = [Scalar::i64(2), Scalar::i64(4), Scalar::f64(0.5)];

        // (a) op-script replay
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
            Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["fast.series".into(), "slow.series".into()] },
            Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() },
            Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() },
            Op::Connect { from: "sub.value".into(), to: "bias.signal".into() },
            Op::Expose { from: "bias.bias".into(), as_name: "bias".into() },
        ];
        let replay_flat = super::replay("root", ops, &aura_std::std_vocabulary)
            .expect("replay resolves")
            .compile_with_params(&params)
            .expect("replay compiles");

        // (b) the GraphBuilder twin — identical topology
        let mut g = GraphBuilder::new("root");
        let fast = g.add(Sma::builder().named("fast"));
        let slow = g.add(Sma::builder().named("slow"));
        let sub = g.add(Sub::builder());
        let bias = g.add(Bias::builder());
        let price = g.source_role("price", ScalarKind::F64);
        g.feed(price, [fast.input("series"), slow.input("series")]);
        g.connect(fast.output("value"), sub.input("lhs"));
        g.connect(slow.output("value"), sub.input("rhs"));
        g.connect(sub.output("value"), bias.input("signal"));
        g.expose(bias.output("bias"), "bias");
        let builder_flat = g.build().expect("root resolves").compile_with_params(&params).expect("compiles");

        assert_eq!(replay_flat.edges, builder_flat.edges);
        assert_eq!(replay_flat.sources, builder_flat.sources);
    }

    /// C24 lockstep extended to gangs: the op-script's `Op::Gang` replay and the
    /// `GraphBuilder::gang` twin serialize to byte-identical blueprint JSON and
    /// compile to the byte-identical `FlatGraph` topology — the gang op is not a
    /// second construction semantics, just the by-identifier face of the same
    /// gate `GraphBuilder::build` runs (builder.rs).
    #[test]
    fn gang_replay_serializes_identically_to_builder() {
        use crate::blueprint_serde::blueprint_to_json;
        use crate::GraphBuilder;
        use aura_core::Scalar;
        use aura_std::{Sma, Sub};

        // (a) op-script: the Task-4 fixture (two SMA + gang + Sub + expose).
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), bind: vec![] },
            Op::Gang { as_name: "length".into(), into: vec!["a.length".into(), "b.length".into()] },
            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["a.series".into(), "b.series".into()] },
            Op::Connect { from: "a.value".into(), to: "sub.lhs".into() },
            Op::Connect { from: "b.value".into(), to: "sub.rhs".into() },
            Op::Expose { from: "sub.value".into(), as_name: "bias".into() },
        ];
        let replayed = super::replay("g", ops, &std_vocabulary).expect("replays");

        // (b) the GraphBuilder twin — same adds/feeds/connects/expose + g.gang(...).
        let mut g = GraphBuilder::new("g");
        let a = g.add(Sma::builder().named("a"));
        let b = g.add(Sma::builder().named("b"));
        g.gang("length", [a.param("length"), b.param("length")]);
        let sub = g.add(Sub::builder().named("sub"));
        let price = g.source_role("price", ScalarKind::F64);
        g.feed(price, [a.input("series"), b.input("series")]);
        g.connect(a.output("value"), sub.input("lhs"));
        g.connect(b.output("value"), sub.input("rhs"));
        g.expose(sub.output("value"), "bias");
        let built = g.build().expect("root resolves");

        assert_eq!(
            blueprint_to_json(&replayed).expect("replay serializes"),
            blueprint_to_json(&built).expect("builder serializes"),
        );

        let params = [Scalar::i64(2)];
        let replay_flat = replayed.compile_with_params(&params).expect("replay compiles");
        let built_flat = built.compile_with_params(&params).expect("builder compiles");
        assert_eq!(replay_flat.edges, built_flat.edges);
        assert_eq!(replay_flat.sources, built_flat.sources);
    }
}