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) run holistically in
//! `finish`. Both cadences call the SAME §A predicates — no second validator
//! (C24). Root-role boundness is a *runnability* gate, not a construction gate:
//! `finish()` does not run it (#317, open patterns), so an op-script may
//! finish with an open root role; only `compile`/bootstrap enforce it. 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::{ArgOpError, BindOpError, NodeSchema, PrimitiveBuilder, Scalar, ScalarKind};

use crate::blueprint::{
    check_param_namespace_injective, edge_kind_check, name_gate, validate_wiring, BlueprintNode,
    Composite, Gang, GangMember, NameGateFault, OutField, Role, Tap, TapWire,
};
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 construction
    /// `args` (the typed, closed channel — #271) applied BEFORE `bind` params.
    Add {
        type_id: String,
        as_name: Option<String>,
        args: Vec<(String, 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 },
    /// Declare a measurement tap on `"node.field"` under `as_name` (#284) — the
    /// output-side twin of `Expose` (a `Tap`, not an `OutField`).
    Tap { from: String, as_name: String },
    /// Fuse two or more sibling params into one public knob (#61).
    Gang { as_name: String, into: Vec<String> },
    /// Declare the composite's one-line meaning (C29, #316) — the op-script
    /// twin of the builder's `.doc(...)`. At most one per script; a second
    /// is a fault (a declarative script carries no last-wins ambiguity).
    Doc { text: String },
    /// Splice a registered blueprint into the building graph as a nested
    /// composite under an instance identifier (#317). `ref_id` is the full
    /// content id the injected `subgraph` resolver is keyed by — the engine
    /// never sees a label, a content-id prefix, or the registry (CLI-side
    /// resolution happens at DTO conversion, before replay). `name` is the
    /// instance identifier (`None` -> the fetched composite's own `name()`);
    /// `bind` is applied path-qualified (e.g. `"sma.length"`), after the
    /// splice.
    Use { ref_id: String, name: Option<String>, bind: Vec<(String, Scalar)> },
    /// Set the composite's render name (#331). At most once per script.
    Name { name: 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),
    /// A tap-name collision (the separate `tap_names` namespace) — two taps named
    /// alike would collide as recorded series. A distinct fault class from an
    /// interior id (`DuplicateIdentifier`) or a boundary-output clash
    /// (`DuplicateOutput`).
    DuplicateTap(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 },
    /// A construction-arg fault (#271): an arg-bearing type's `args` failed
    /// `try_args` (unknown/duplicate/missing arg, or a malformed value) — or a
    /// `Plain` type was given args it does not declare. By-identifier, naming
    /// the offending node.
    BadArg { node: String, err: ArgOpError },
    /// 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 },
    /// 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 gang member path resolves onto a spliced instance (`Op::Use`'s
    /// `Composite` node), not a primitive (#339 item 1 harvest): a gang
    /// fuses a PRIMITIVE's raw `(name, pos)` param slot, and an instance's
    /// params are nested/path-qualified, so ganging a used instance's
    /// member params is unsupported today. Distinct from
    /// `BadParam::UnknownParam` so the refusal names the rule instead of
    /// reading like a typo hint on the leaf case.
    GangOfSplicedInstance { node: String, member: String },
    /// A holistic finalize fault (totality / injectivity / unbound root role),
    /// wrapping the unchanged engine gate's `CompileError`.
    Incomplete(CompileError),
    /// A second `doc` op — the meaning line is declared at most once.
    DuplicateDoc,
    /// `Op::Use`'s injected `subgraph` resolver found no composite for
    /// `ref_id` (#317) — by-identifier. Unreachable from the CLI (which
    /// resolves and fetches before replay, refusing there on a miss), but
    /// total: a bare `replay`/`GraphSession` caller can still hit it.
    UnknownSubgraph { ref_id: String },
    /// A second `name` op — the render name is declared at most once (mirrors
    /// `DuplicateDoc`).
    DuplicateName,
    /// A `name` op's `name` (or, at the CLI's blueprint-envelope intake, an
    /// envelope's root name) failed the deterministic `name_gate` shape check
    /// (#331). Carries the offending name and the shape fault detail.
    BadName { name: String, fault: NameGateFault },
}

/// 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>,
    /// Second injected resolver (#317), mirroring `vocab`'s closed-lookup
    /// posture: `Op::Use` fetches a registered blueprint by full content id.
    /// The engine stays store-free — CLI-side resolution (label/prefix,
    /// C29 doc gate, the store fetch itself) happens at DTO conversion,
    /// before replay; this closure is a pure lookup into an already-fetched
    /// id->`Composite` cache. The engine places no obligation on a caller
    /// either way: a build-free CLI path MAY still resolve `use` refs
    /// through the store first and pass a real cache-backed closure here
    /// (#339 item 4 harvest: `introspect --unwired` now does); a truly
    /// resolver-less caller (a bare `replay`, most tests) passes `&|_| None`.
    subgraph: &'v dyn Fn(&str) -> Option<Composite>,
    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>,
    taps: Vec<Tap>,
    tap_names: HashSet<String>,
    coverage: HashMap<(usize, usize), usize>,
    gangs: Vec<Gang>,
    doc: Option<String>,
    /// At-most-once guard for `Op::Name` (#331), mirroring `doc`'s own
    /// uniqueness posture but tracked separately (`name` itself is always
    /// populated, seeded from `::new`'s `name` argument, so it cannot double
    /// as its own "was it authored" flag).
    name_authored: bool,
}

impl<'v> GraphSession<'v> {
    /// Start a session for a composite named `name`, resolving node types through
    /// the injected closed vocabulary `vocab` and registered-blueprint splices
    /// (`Op::Use`, #317) through the injected `subgraph` lookup.
    pub fn new(
        name: impl Into<String>,
        vocab: &'v dyn Fn(&str) -> Option<PrimitiveBuilder>,
        subgraph: &'v dyn Fn(&str) -> Option<Composite>,
    ) -> Self {
        Self {
            name: name.into(),
            vocab,
            subgraph,
            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(),
            taps: Vec::new(),
            tap_names: HashSet::new(),
            coverage: HashMap::new(),
            gangs: Vec::new(),
            doc: None,
            name_authored: false,
        }
    }

    /// 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, args, bind } => self.add_node(type_id, as_name, args, 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::Tap { from, as_name } => self.tap(from, as_name),
            Op::Gang { as_name, into } => self.gang(as_name, into),
            Op::Doc { text } => {
                if self.doc.is_some() {
                    return Err(OpError::DuplicateDoc);
                }
                self.doc = Some(text);
                Ok(())
            }
            Op::Use { ref_id, name, bind } => self.use_subgraph(ref_id, name, bind),
            Op::Name { name } => self.set_name(name),
        }
    }

    /// `Op::Name` (#331): set the composite's render name — the op-script's
    /// data-borne twin of `replay`'s seeded default name. Guarded by (a)
    /// at-most-once (mirrors the `doc` op's uniqueness guard just above) and
    /// (b) the shared deterministic shape gate (`name_gate`) — the SAME gate
    /// the CLI's blueprint-envelope intake applies to a hand-edited
    /// envelope's root name (the other data-borne birth route, skeptic
    /// finding minuted on #331).
    fn set_name(&mut self, name: String) -> Result<(), OpError> {
        if self.name_authored {
            return Err(OpError::DuplicateName);
        }
        if let Err(fault) = name_gate(&name) {
            return Err(OpError::BadName { name, fault });
        }
        self.name = name;
        self.name_authored = true;
        Ok(())
    }

    /// `Op::Use` (#317): fetch a registered blueprint through the injected
    /// `subgraph` resolver, rename it to the instance identifier, apply its
    /// path-qualified `bind` entries, and push it as an ordinary
    /// `BlueprintNode::Composite` — sharing the node-identifier namespace
    /// with every `Add`ed primitive (so an instance and a primitive collide
    /// the same way two primitives do). A miss on `ref_id` is a by-identifier
    /// `UnknownSubgraph`; a `bind` path with no matching open param anywhere
    /// in the instance reuses the existing `BadParam` bind-fault shape,
    /// naming the instance (`node`) and the qualified within-instance path
    /// (`err`'s carried name) — the pair spells out the full qualified path.
    fn use_subgraph(
        &mut self,
        ref_id: String,
        name: Option<String>,
        bind: Vec<(String, Scalar)>,
    ) -> Result<(), OpError> {
        let composite = (self.subgraph)(&ref_id).ok_or_else(|| OpError::UnknownSubgraph { ref_id: ref_id.clone() })?;
        let id = name.unwrap_or_else(|| composite.name().to_string());
        if self.names.contains_key(&id) {
            return Err(OpError::DuplicateIdentifier(id));
        }
        let mut composite = composite.renamed(&id);
        for (path, value) in bind {
            composite = composite
                .bind_path(&path, value)
                .map_err(|err| OpError::BadParam { node: id.clone(), err })?;
        }
        let node = BlueprintNode::Composite(composite);
        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(())
    }

    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>,
        args: Vec<(String, 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());
        // Construction args apply BEFORE the bind loop (spec §aura-engine): a
        // pending arg-bearing builder with no args therefore refuses here as
        // `MissingArg` — no pending builder ever reaches `try_bind`/enters the
        // graph.
        builder = builder.try_args(&args).map_err(|err| OpError::BadArg { node: id.clone(), err })?;
        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(())
    }

    /// Declare a measurement tap on a resolved interior output wire — the
    /// output-side twin of `expose`. Node resolution (`node_index` →
    /// `UnknownIdentifier`), the output-field resolver, and the name-dedup are the
    /// same as `expose`'s; only the appended item differs: a `Tap` on the separate
    /// `taps`/`tap_names` namespace (dedup → `DuplicateTap`), not an `OutField`.
    fn tap(&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.tap_names.insert(as_name.clone()) {
            return Err(OpError::DuplicateTap(as_name));
        }
        self.taps.push(Tap { name: as_name, from: TapWire { node: fi, field } });
        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 {
                // #317: `Op::Use` makes a session node legitimately a
                // Composite, not just a Primitive — reachable now, not a
                // defect. A gang fuses a PRIMITIVE's raw (name, pos) param
                // slot; an instance has no such flat slot (its params are
                // nested/path-qualified). #339 item 1 harvest: name the
                // rule here instead of falling into the ordinary
                // no-such-open-param shape, which reads as a typo hint on
                // an otherwise correct member path.
                return Err(OpError::GangOfSplicedInstance { node: node_name, member: param_name });
            };
            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 construction gates (totality,
    /// param-namespace injectivity) — the SAME engine predicates the eager path
    /// shares, now at end-of-document cadence. Root-role boundness is
    /// deliberately NOT checked here (#317, open patterns): a `finish()`ed
    /// composite may still carry an open root role, declaring a reusable
    /// pattern's formal parameters; only `compile`/bootstrap (the runnability
    /// gate) refuse an unbound root role. 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();
        // The doc op (C29, #316) threads through `Composite::with_doc`.
        let mut c = Composite::new(self.name, self.nodes, self.edges, roles, self.out)
            .with_taps(self.taps)
            .with_gangs(self.gangs)
            .map_err(OpError::Incomplete)?;
        if let Some(t) = self.doc {
            c = c.with_doc(t);
        }
        check_param_namespace_injective(&c.param_space()).map_err(OpError::Incomplete)?;
        validate_wiring(c.nodes(), c.edges(), c.input_roles(), c.output(), c.taps()).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),
        })?;
        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). `subgraph` is the injected
/// registered-blueprint lookup `Op::Use` (#317) resolves splices through — pass
/// `&|_| None` when no store is in play.
pub fn replay(
    name: impl Into<String>,
    ops: Vec<Op>,
    vocab: &dyn Fn(&str) -> Option<PrimitiveBuilder>,
    subgraph: &dyn Fn(&str) -> Option<Composite>,
) -> Result<Composite, (usize, OpError)> {
    let mut s = GraphSession::new(name, vocab, subgraph);
    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 crate::blueprint::{BlueprintNode, Composite, Role};
    use aura_core::{ArgKind, ArgOpError, BindOpError, Scalar, ScalarKind};
    use aura_vocabulary::std_vocabulary;

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

    #[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, args: vec![], 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()), args: vec![], bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("x".into()), args: vec![], 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()),
                args: vec![], 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()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Sub".into(), as_name: None, args: vec![], 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, args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Bias".into(), as_name: None, args: vec![], 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() })
        );
    }

    /// `tap` names an interior output wire by `"node.field"` (the output-side twin
    /// of `expose`), appending a resolved `Tap` to the composite — the
    /// name-addressed wire resolves to the same interior `{node, field}` a
    /// raw-index tap would carry (`fast` is session node 0, its `value` field 0).
    #[test]
    fn tap_op_resolves_and_appends() {
        use crate::blueprint::{Tap, TapWire};
        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();
        s.apply(Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() }).unwrap();
        s.apply(Op::Expose { from: "sub.value".into(), as_name: "bias".into() }).unwrap();
        assert_eq!(s.apply(Op::Tap { from: "fast.value".into(), as_name: "fast_ma".into() }), Ok(()));
        let c = s.finish().expect("finishes");
        assert_eq!(c.taps(), &[Tap { name: "fast_ma".into(), from: TapWire { node: 0, field: 0 } }]);
    }

    /// `tap` reuses `expose`'s resolvers: a bad output field maps to the
    /// by-identifier `UnknownOutPort`; an unknown source node to `UnknownIdentifier`
    /// (via `node_index`), never the nonexistent `UnknownNode`.
    #[test]
    fn tap_op_bad_out_port_and_ghost_node_rejected() {
        let mut s = scaffold();
        assert_eq!(
            s.apply(Op::Tap { from: "fast.nope".into(), as_name: "t".into() }),
            Err(OpError::UnknownOutPort { node: "fast".into(), name: "nope".into() })
        );
        assert_eq!(
            s.apply(Op::Tap { from: "ghost.value".into(), as_name: "t".into() }),
            Err(OpError::UnknownIdentifier("ghost".into()))
        );
    }

    /// A second tap under the same boundary name is a `DuplicateTap` — the twin of
    /// `expose`'s `DuplicateOutput`, over the separate `tap_names` namespace.
    #[test]
    fn tap_op_duplicate_name_rejected() {
        let mut s = scaffold();
        assert_eq!(s.apply(Op::Tap { from: "fast.value".into(), as_name: "t".into() }), Ok(()));
        assert_eq!(
            s.apply(Op::Tap { from: "slow.value".into(), as_name: "t".into() }),
            Err(OpError::DuplicateTap("t".into()))
        );
    }

    /// `finish()` threads op-declared taps into `Composite.taps`, and the #282
    /// compile path hoists them into `FlatGraph.taps` — proving the `.with_taps`
    /// thread this op adds reaches the flat graph (guards the latent drop `finish`
    /// had: it built `.with_gangs` but never `.with_taps`). `fast` lowers to flat
    /// node 0, its `value` field 0.
    #[test]
    fn tap_op_threads_into_flat_graph() {
        use crate::harness::FlatTap;
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: None, args: vec![], 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::Tap { from: "fast.value".into(), as_name: "fast_ma".into() },
            Op::Expose { from: "sub.value".into(), as_name: "bias".into() },
        ];
        let flat = super::replay("g", ops, &aura_vocabulary::std_vocabulary, &|_: &str| None)
            .expect("replays")
            .compile_with_params(&[Scalar::i64(2), Scalar::i64(4)])
            .expect("compiles");
        assert_eq!(flat.taps, vec![FlatTap { name: "fast_ma".into(), node: 0, field: 0 }]);
    }

    /// 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()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), args: vec![], 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, args: vec![], 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, &|_: &str| None).expect("replays");
        let names: Vec<String> = c.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, ["length"]);
    }

    /// #261 — the Frankfurt `Session` preset is reachable from a blueprint. The
    /// property this pins: a data-only op-script (a blueprint) that names the
    /// preset's closed-roster type-id resolves it through `std_vocabulary`,
    /// carries its one scalar `period_minutes` knob and the `bars_since_open:
    /// i64` session-index output, wires its `trigger` input through the ordinary
    /// edge convention, and builds to a finished `Composite`. Reachability is the
    /// whole ask — the node's tz-aware/DST session SEMANTICS are already pinned
    /// in `aura-market/src/session.rs` and are not re-pinned here. Today the preset
    /// id is absent from the roster (the base `Session` builder takes structural
    /// construction args, so the zero-arg macro cannot host it), so the lookup
    /// returns `None` and the op-script's `Add` would fail `UnknownNodeType`.
    #[test]
    fn session_frankfurt_preset_is_reachable_from_a_blueprint() {
        // (a) the preset resolves through the closed roster as the Session preset
        //     — its one scalar knob and its i64 session-index output, not some
        //     other node parked under the id.
        let preset = std_vocabulary("SessionFrankfurt")
            .expect("the Frankfurt Session preset is in the std vocabulary roster");
        assert_eq!(preset.label(), "SessionFrankfurt");
        let schema = preset.schema();
        assert_eq!(schema.params.len(), 1, "exactly one scalar knob");
        assert_eq!(schema.params[0].name, "period_minutes");
        assert_eq!(schema.params[0].kind, ScalarKind::I64);
        assert_eq!(schema.inputs.len(), 1);
        assert_eq!(schema.inputs[0].name, "trigger");
        assert_eq!(schema.output[0].name, "bars_since_open");
        assert_eq!(schema.output[0].kind, ScalarKind::I64);

        // (b) an op-script naming that id builds: a bound trigger source feeds the
        //     preset's `trigger`, its `period_minutes` knob binds, and the i64 bar
        //     index exposes — a data-only blueprint reaching the session stream.
        let ops = vec![
            Op::Source { role: "trigger".into(), kind: ScalarKind::F64 },
            Op::Add {
                type_id: "SessionFrankfurt".into(),
                as_name: Some("session".into()),
                args: vec![], bind: vec![("period_minutes".into(), Scalar::i64(15))],
            },
            Op::Feed { role: "trigger".into(), into: vec!["session.trigger".into()] },
            Op::Expose { from: "session.bars_since_open".into(), as_name: "bars".into() },
        ];
        super::replay("session_blueprint", ops, &std_vocabulary, &|_: &str| None)
            .expect("the preset resolves, wires and builds through an op-script");
    }

    /// 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()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add {
            type_id: "SMA".into(),
            as_name: Some("b".into()),
            args: vec![], 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()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "Bias".into(), as_name: Some("b".into()), args: vec![], 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()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("c".into()), args: vec![], 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()), args: vec![], 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()), args: vec![], 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()))
        );
    }

    /// C29 (#316): the doc op sets the composite's meaning line — the
    /// op-script twin of the builder's .doc(...).
    #[test]
    fn doc_op_sets_the_composite_doc() {
        let mut s = scaffold();
        s.apply(Op::Doc { text: "a described op-built composite".into() }).unwrap();
        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();
        s.apply(Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() }).unwrap();
        s.apply(Op::Expose { from: "sub.value".into(), as_name: "out".into() }).unwrap();
        let c = s.finish().expect("finishes");
        assert_eq!(c.doc(), Some("a described op-built composite"));
    }

    /// A second doc op is a fault, not a silent overwrite.
    #[test]
    fn duplicate_doc_op_is_refused() {
        let mut s = scaffold();
        s.apply(Op::Doc { text: "first".into() }).unwrap();
        assert_eq!(
            s.apply(Op::Doc { text: "second".into() }),
            Err(OpError::DuplicateDoc)
        );
    }

    // ---- Op::Name (#331) -----------------------------------------------

    /// A `name` op sets the finished composite's render name — the property
    /// the whole op exists for.
    #[test]
    fn name_op_sets_the_finished_composite_name() {
        let mut s = scaffold();
        s.apply(Op::Name { name: "x".into() }).unwrap();
        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();
        s.apply(Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() }).unwrap();
        s.apply(Op::Expose { from: "sub.value".into(), as_name: "out".into() }).unwrap();
        let c = s.finish().expect("finishes");
        assert_eq!(c.name(), "x");
    }

    /// Byte-compat (#331 AC1): a script that never authors a `name` op keeps
    /// the seed name `replay`/`GraphSession::new` was given — the CLI's own
    /// seed stays `"graph"`, unchanged by this op's addition.
    #[test]
    fn no_name_op_keeps_the_seeded_replay_name() {
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), args: vec![], bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["a.series".into()] },
            Op::Expose { from: "a.value".into(), as_name: "out".into() },
        ];
        let c = super::replay("graph", ops, &std_vocabulary, &|_: &str| None).expect("replays");
        assert_eq!(c.name(), "graph");
    }

    /// A second `name` op is a fault, not a silent last-wins overwrite —
    /// mirrors the `doc` op's own at-most-once guard.
    #[test]
    fn duplicate_name_op_is_refused() {
        let mut s = session("g");
        s.apply(Op::Name { name: "x".into() }).unwrap();
        assert_eq!(
            s.apply(Op::Name { name: "y".into() }),
            Err(OpError::DuplicateName)
        );
    }

    /// Shape violations (#331): empty, containing a path separator, and the
    /// dot segments all fault at the op via the shared `name_gate`.
    #[test]
    fn name_op_shape_violations_are_refused() {
        use crate::blueprint::NameGateFault;
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Name { name: "".into() }),
            Err(OpError::BadName { name: "".into(), fault: NameGateFault::Empty })
        );
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Name { name: "a/b".into() }),
            Err(OpError::BadName { name: "a/b".into(), fault: NameGateFault::ContainsSeparator })
        );
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Name { name: "..".into() }),
            Err(OpError::BadName { name: "..".into(), fault: NameGateFault::DotSegment })
        );
    }

    #[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:?}");
    }

    /// #317 (open patterns): `finish()` no longer runs the root-role gate — a
    /// reserved `Input` root role that is fed but never source-bound now
    /// finishes cleanly (the composite declares an open formal parameter, the
    /// pattern's reusable interior). The runnability gate moves entirely to
    /// `compile`, which refuses the SAME composite with the unchanged
    /// `CompileError::UnboundRootRole` — the split this cycle ratified: finish
    /// = construction-complete, compile = runnable.
    #[test]
    fn finish_accepts_an_open_input_role_and_compile_refuses_it() {
        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()), args: vec![], 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 c = s.finish().expect("finish() no longer gates root-role boundness");
        assert_eq!(
            c.compile().err(),
            Some(crate::CompileError::UnboundRootRole { role: 0 }),
            "compile() is the runnability gate that refuses the open root role"
        );
    }

    /// 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()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "And".into(), as_name: Some("conj".into()), args: vec![], 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()), args: vec![], bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: Some("b".into()), args: vec![], 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_vocabulary::std_vocabulary, &|_: &str| None)
            .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()), args: vec![], 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_vocabulary::std_vocabulary, &|_: &str| None).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()), args: vec![], bind: vec![] },
            Op::Add { type_id: "Nope".into(), as_name: None, args: vec![], bind: vec![] }, // op 1 fails
            Op::Add { type_id: "Sub".into(), as_name: None, args: vec![], 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_vocabulary::std_vocabulary, &|_: &str| None).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::{Sma, Sub};
        use aura_strategy::Bias;

        // 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()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: None, args: vec![], bind: vec![] },
            Op::Add { type_id: "Bias".into(), as_name: None, args: vec![], 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_vocabulary::std_vocabulary, &|_: &str| None)
            .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()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), args: vec![], 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, args: vec![], 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, &|_: &str| None).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);
    }

    /// C24 lockstep extended to taps (#284): the op-script's `Op::Tap` replay and
    /// the `GraphBuilder::tap` twin serialize to byte-identical blueprint JSON and
    /// compile to byte-identical `FlatGraph.taps` — the tap op is the by-identifier
    /// face of the same construction semantics `GraphBuilder::build` runs, not a
    /// second one. Guards the builder/op-script tap symmetry (both surfaces can
    /// declare a tap, and they resolve it identically).
    #[test]
    fn tap_replay_matches_builder() {
        use crate::blueprint_serde::blueprint_to_json;
        use crate::GraphBuilder;
        use aura_core::Scalar;
        use aura_std::{Sma, Sub};

        // (a) op-script: an SMA crossover with a tap on the raw spread (sub.value).
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "Sub".into(), as_name: Some("sub".into()), args: vec![], 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::Tap { from: "sub.value".into(), as_name: "spread".into() },
            Op::Expose { from: "sub.value".into(), as_name: "bias".into() },
        ];
        let replayed = super::replay("g", ops, &std_vocabulary, &|_: &str| None).expect("replays");

        // (b) the GraphBuilder twin — same adds/feeds/connects + g.tap(...) / g.expose(...).
        let mut g = GraphBuilder::new("g");
        let fast = g.add(Sma::builder().named("fast"));
        let slow = g.add(Sma::builder().named("slow"));
        let sub = g.add(Sub::builder().named("sub"));
        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.tap(sub.output("value"), "spread");
        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), Scalar::i64(4)];
        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.taps, built_flat.taps);
        assert_eq!(replay_flat.edges, built_flat.edges);
    }

    // ---- construction args (#271) -------------------------------------

    /// `add_node` applies `try_args` AFTER resolve and BEFORE the bind loop
    /// (spec §aura-engine): an arg-bearing type's `add` op, given a valid
    /// `args` set, configures the real signature and the subsequent `bind`
    /// sees the now-real `period_minutes` param — a data-only op-script
    /// reaches a `Session` for any IANA timezone/open, not just the
    /// `SessionFrankfurt` preset.
    #[test]
    fn add_op_with_args_builds_a_session() {
        let mut s = session("g");
        s.apply(Op::Source { role: "trigger".into(), kind: ScalarKind::F64 }).unwrap();
        assert_eq!(
            s.apply(Op::Add {
                type_id: "Session".into(),
                as_name: Some("ny".into()),
                args: vec![("tz".into(), "America/New_York".into()), ("open".into(), "09:30".into())],
                bind: vec![("period_minutes".into(), Scalar::i64(15))],
            }),
            Ok(()),
        );
        s.apply(Op::Feed { role: "trigger".into(), into: vec!["ny.trigger".into()] }).unwrap();
        s.apply(Op::Expose { from: "ny.bars_since_open".into(), as_name: "bars".into() }).unwrap();
        s.finish().expect("an args-configured Session finishes and builds");
    }

    /// `Session` is arg-bearing (declares `tz`/`open`): an `add` op naming it
    /// with NO `args` refuses at the op as `BadArg(MissingArg(..))` — the
    /// pending builder never reaches the bind loop or the graph (spec
    /// §aura-engine: "a pending resolve with no args therefore fails as
    /// MissingArg").
    #[test]
    fn add_op_arg_bearing_type_without_args_refuses() {
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Add {
                type_id: "Session".into(),
                as_name: Some("ny".into()),
                args: vec![],
                bind: vec![],
            }),
            Err(OpError::BadArg { node: "ny".into(), err: ArgOpError::MissingArg("tz".into()) }),
        );
    }

    /// A malformed arg value (strict-form refusal, spec §Closedness) surfaces
    /// as `BadArg(BadValue{..})`, naming the arg, its declared `ArgKind`, and
    /// the rejected string.
    #[test]
    fn add_op_bad_tz_refuses() {
        let mut s = session("g");
        assert_eq!(
            s.apply(Op::Add {
                type_id: "Session".into(),
                as_name: Some("ny".into()),
                args: vec![("tz".into(), "not_a_zone".into()), ("open".into(), "09:30".into())],
                bind: vec![],
            }),
            Err(OpError::BadArg {
                node: "ny".into(),
                err: ArgOpError::BadValue { arg: "tz".into(), kind: ArgKind::Tz, got: "not_a_zone".into() },
            }),
        );
    }

    // ---- Op::Use (#317) -----------------------------------------------

    /// A small two-node fixture composite for the `Op::Use` tests (#317): an
    /// open `price` input role feeds an `Sma`, whose `value` feeds a `Bias`;
    /// `Bias.bias` re-exports as `out`. Built fresh via `GraphBuilder` on every
    /// call — mirroring how a CLI-side registry lookup hands back a freshly
    /// loaded `Composite` each fetch — never shared or cloned (`Composite`
    /// holds build closures and is not `Clone`).
    fn use_fixture() -> Composite {
        use crate::GraphBuilder;
        use aura_std::Sma;
        use aura_strategy::Bias;
        let mut g = GraphBuilder::new("fixture");
        let sma = g.add(Sma::builder().named("sma"));
        let bias = g.add(Bias::builder().named("bias"));
        let price = g.input_role("price");
        g.feed(price, [sma.input("series")]);
        g.connect(sma.output("value"), bias.input("signal"));
        g.expose(bias.output("bias"), "out");
        g.build().expect("fixture resolves")
    }

    /// Wrap a sink-free signal (one open input role, one exposed field) in a
    /// runnable root that records the exposed field via a channel-backed
    /// `Recorder`, feeds a fixed synthetic price stream through it, and
    /// returns the recorded trace. Mirrors `blueprint_serde.rs`'s
    /// `run_recording` test helper (same shape); duplicated here rather than
    /// shared, since that one is private to its own file.
    fn run_recording(signal: Composite) -> Vec<(aura_core::Timestamp, Vec<Scalar>)> {
        use crate::harness::{Edge, Target};
        use aura_core::Firing;
        use aura_std::Recorder;
        use std::sync::mpsc;
        let (tx, rx) = mpsc::channel();
        let root = Composite::new(
            "h",
            vec![
                BlueprintNode::Composite(signal),
                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx).into(),
            ],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        );
        // only `bias.scale` is open (every fixture in this module's tests binds
        // `sma.length` at the use seam), so one F64 param point.
        let mut h = root.bootstrap_with_params(vec![Scalar::f64(0.5)]).expect("bootstraps");
        h.run(vec![Box::new(crate::VecSource::new(crate::test_fixtures::synthetic_prices()))]);
        rx.try_iter().collect()
    }

    /// C1 twin (#317 headline): an op-script using `Op::Use` splices a
    /// registered-style subgraph into the building graph byte-identically to
    /// the same splice authored directly on `GraphBuilder`
    /// (`g.add(BlueprintNode::Composite(x))` plus the same rename/bind) — the
    /// correctness oracle for the whole splice path. Mirrors
    /// `replay_built_signal_compiles_identically_to_graphbuilder` (topology
    /// identity) and `blueprint_serde::serialized_blueprint_runs_bit_identical_
    /// to_rust_built` (bit-identical run).
    #[test]
    fn use_op_splices_byte_identical_to_the_rust_built_twin() {
        use crate::blueprint_serde::blueprint_to_json;
        use crate::GraphBuilder;

        let subgraph = |ref_id: &str| (ref_id == "fixture-id").then(use_fixture);

        // (a) op-script: a bound `price` source feeds the spliced instance's
        // own "price" port (`Op::Source`, not `Op::Input` — this test
        // bootstraps and runs the result, and only compile/bootstrap gate
        // root-role boundness now, #317; finish() itself no longer cares);
        // the instance's "out" field re-exports as "bias".
        let ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Use {
                ref_id: "fixture-id".into(),
                name: Some("gate".into()),
                bind: vec![("sma.length".into(), Scalar::i64(5))],
            },
            Op::Feed { role: "price".into(), into: vec!["gate.price".into()] },
            Op::Expose { from: "gate.out".into(), as_name: "bias".into() },
        ];
        let replayed = super::replay("root", ops, &std_vocabulary, &subgraph).expect("replays");

        // (b) the GraphBuilder twin — the identical splice: the SAME fixture,
        // renamed to "gate", the SAME bind, added the way a Rust author would.
        let mut g = GraphBuilder::new("root");
        let price = g.source_role("price", ScalarKind::F64);
        let spliced = use_fixture()
            .renamed("gate")
            .bind_path("sma.length", Scalar::i64(5))
            .expect("binds");
        let gate = g.add(spliced);
        g.feed(price, [gate.input("price")]);
        g.expose(gate.output("out"), "bias");
        let built = g.build().expect("root resolves");

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

        // bootstrap + run bit-identical (only bias.scale is open: sma.length
        // was bound at the use seam).
        let trace_replayed = run_recording(replayed);
        let trace_built = run_recording(built);
        assert_eq!(
            trace_replayed, trace_built,
            "the op-script splice diverged from the Rust-built twin at run time"
        );
        assert!(!trace_replayed.is_empty(), "trace must be populated (non-degenerate)");
    }

    /// A miss on the injected `subgraph` resolver is a by-identifier fault
    /// naming the `ref_id` (#317) — reachable directly against `GraphSession`/
    /// `replay` even though the CLI (Task 3) always pre-resolves.
    #[test]
    fn use_op_unknown_subgraph_is_a_by_identifier_fault() {
        let mut s = GraphSession::new("g", &std_vocabulary, &|_: &str| None);
        assert_eq!(
            s.apply(Op::Use { ref_id: "ghost".into(), name: None, bind: vec![] }),
            Err(OpError::UnknownSubgraph { ref_id: "ghost".into() })
        );
    }

    /// An instance identifier colliding with an already-added node is the
    /// existing `DuplicateIdentifier` refusal — instance names share the
    /// interior node-identifier namespace with primitives (#317).
    #[test]
    fn use_op_duplicate_instance_identifier_rejected() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("gate".into()), args: vec![], bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Use { ref_id: "fixture".into(), name: Some("gate".into()), bind: vec![] }),
            Err(OpError::DuplicateIdentifier("gate".into()))
        );
    }

    /// #331 ratifying pin (spec test 5): `use_subgraph`'s instance-identifier
    /// default (`name: None` -> the fetched composite's own `name()`,
    /// `construction.rs:221-230` — code-inspected, previously untested since
    /// every other test in this section passes an explicit instance name)
    /// actually reaches the session's identifier namespace: `use_fixture()`'s
    /// own name is "fixture", so with no instance name the spliced ports
    /// resolve under "fixture.*", not some other placeholder.
    #[test]
    fn use_op_with_no_name_defaults_instance_identifier_to_the_fetched_composites_name() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
        assert_eq!(
            s.apply(Op::Use { ref_id: "fixture-id".into(), name: None, bind: vec![] }),
            Ok(())
        );
        s.apply(Op::Feed { role: "price".into(), into: vec!["fixture.price".into()] }).unwrap();
        assert_eq!(
            s.apply(Op::Expose { from: "fixture.out".into(), as_name: "bias".into() }),
            Ok(()),
            "the spliced instance resolves under \"fixture\" — the fetched composite's own name"
        );
    }

    /// A post-splice `bind` path with no matching open param anywhere in the
    /// instance faults through the existing bind-fault shape (`BadParam`),
    /// naming the instance and the qualified within-instance path (#317).
    #[test]
    fn use_op_bind_of_a_closed_path_faults_with_the_qualified_path() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        assert_eq!(
            s.apply(Op::Use {
                ref_id: "fixture".into(),
                name: Some("gate".into()),
                bind: vec![("ghost.length".into(), Scalar::i64(2))],
            }),
            Err(OpError::BadParam {
                node: "gate".into(),
                err: BindOpError::UnknownParam("ghost.length".into()),
            })
        );
    }

    /// A gang-bearing fixture for the ganged-member-bind refusal test below
    /// (review finding, #317 follow-up): two `SMA`s ganged on `length` under
    /// one public `channel_length` knob, feeding a `Sub` whose difference
    /// re-exports as `out`. Mirrors `gang_op_projects_the_param_space`'s
    /// op-script shape; built fresh on every call like `use_fixture` (never
    /// shared/cloned — `Composite` holds build closures and is not `Clone`).
    fn use_gang_fixture() -> Composite {
        let ops = vec![
            Op::Input { role: "price".into() },
            Op::Add { type_id: "SMA".into(), as_name: Some("channel_hi".into()), args: vec![], bind: vec![] },
            Op::Add { type_id: "SMA".into(), as_name: Some("channel_lo".into()), args: vec![], bind: vec![] },
            Op::Gang {
                as_name: "channel_length".into(),
                into: vec!["channel_hi.length".into(), "channel_lo.length".into()],
            },
            Op::Add { type_id: "Sub".into(), as_name: None, args: vec![], bind: vec![] },
            Op::Feed {
                role: "price".into(),
                into: vec!["channel_hi.series".into(), "channel_lo.series".into()],
            },
            Op::Connect { from: "channel_hi.value".into(), to: "sub.lhs".into() },
            Op::Connect { from: "channel_lo.value".into(), to: "sub.rhs".into() },
            Op::Expose { from: "sub.value".into(), as_name: "out".into() },
        ];
        super::replay("gang_fixture", ops, &std_vocabulary, &|_: &str| None).expect("gang fixture replays")
    }

    /// Silent gang de-fuse guard (review finding, #317 follow-up): binding a
    /// GANGED member's raw path through `Op::Use`'s post-splice `bind` must
    /// not quietly freeze that one member while the gang's public knob keeps
    /// driving its sibling — refused by identifier, naming the full
    /// qualified path and the gang's own public name. The gang's own public
    /// knob (`gate.channel_length`) staying unbindable at the use seam is
    /// accepted residue, not covered here.
    #[test]
    fn use_op_bind_of_a_ganged_member_path_refuses_by_identifier() {
        let subgraph = |_: &str| Some(use_gang_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        assert_eq!(
            s.apply(Op::Use {
                ref_id: "fixture".into(),
                name: Some("gate".into()),
                bind: vec![("channel_hi.length".into(), Scalar::i64(20))],
            }),
            Err(OpError::BadParam {
                node: "gate".into(),
                err: BindOpError::AlreadyGanged {
                    param: "channel_hi.length".into(),
                    gang: "channel_length".into(),
                },
            })
        );
    }

    /// Documented residue pin (#317 audit, prose updated #339 item 1
    /// harvest): a `gang` op cannot fuse a spliced instance's member param —
    /// a gang fuses a PRIMITIVE's raw `(name, pos)` slot, and an instance's
    /// params are nested/path-qualified. The refusal now names the rule
    /// (`OpError::GangOfSplicedInstance`) instead of the bare
    /// no-such-open-param shape, so it reads as "instance-member ganging is
    /// unsupported", not as a typo hint — matching the authoring-guide/C24
    /// claim "the gang op refuses a composite instance's member path".
    #[test]
    fn gang_of_a_spliced_instance_member_path_refuses() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        s.apply(Op::Use { ref_id: "fixture".into(), name: Some("gate".into()), bind: vec![] })
            .unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("solo".into()), args: vec![], bind: vec![] })
            .unwrap();
        assert_eq!(
            s.apply(Op::Gang {
                as_name: "fused".into(),
                into: vec!["gate.sma.length".into(), "solo.length".into()],
            }),
            Err(OpError::GangOfSplicedInstance { node: "gate".into(), member: "sma.length".into() })
        );
    }

    /// Sibling pin to `gang_of_a_spliced_instance_member_path_refuses`: the
    /// plain-typo case — a `gang` member naming a param that never existed
    /// on a PRIMITIVE (leaf) node — stays the ordinary no-such-open-param
    /// shape (`BindOpError::UnknownParam`), unaffected by the instance-path
    /// rule above (#339 item 1 harvest).
    #[test]
    fn gang_of_a_leaf_node_with_a_wrong_param_name_refuses_as_unknown_param() {
        let mut s = session("g");
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("a".into()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("b".into()), args: vec![], bind: vec![] }).unwrap();
        assert_eq!(
            s.apply(Op::Gang { as_name: "typo".into(), into: vec!["a.lenght".into(), "b.length".into()] }),
            Err(OpError::BadParam { node: "a".into(), err: BindOpError::UnknownParam("lenght".into()) })
        );
    }

    /// DAG invariant (domain invariant 5 / C9), through an instance: a
    /// `connect` closing a cycle through a spliced instance is rejected
    /// eagerly, the instance treated as one opaque node (#317) — the same
    /// `closes_cycle` gate `replay_rejects_dataflow_cycle` pins for primitives.
    #[test]
    fn use_op_connect_closing_a_cycle_through_an_instance_is_rejected_eagerly() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        s.apply(Op::Add { type_id: "Sub".into(), as_name: Some("s".into()), args: vec![], bind: vec![] }).unwrap();
        s.apply(Op::Use { ref_id: "fixture".into(), name: Some("gate".into()), bind: vec![] }).unwrap();
        s.apply(Op::Connect { from: "s.value".into(), to: "gate.price".into() }).unwrap();
        assert_eq!(
            s.apply(Op::Connect { from: "gate.out".into(), to: "s.lhs".into() }),
            Err(OpError::WouldCycle { from: "gate.out".into(), to: "s.lhs".into() })
        );
    }

    /// An unfed instance in-port finishes with the by-identifier
    /// `UnconnectedPort`, naming the instance and the instance's own role name
    /// (rendered `<instance>.<role>` by the CLI's `format_op_error`) — the
    /// instance's open input roles ARE its in-ports (#317).
    #[test]
    fn use_op_unfed_instance_role_reports_the_qualified_identifier() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        s.apply(Op::Use { ref_id: "fixture".into(), name: Some("gate".into()), bind: vec![] }).unwrap();
        s.apply(Op::Expose { from: "gate.out".into(), as_name: "bias".into() }).unwrap();
        // gate.price deliberately left unfed.
        let err = s.finish().err().unwrap();
        assert!(
            matches!(&err, OpError::UnconnectedPort { node, slot } if node == "gate" && slot == "price"),
            "an unfed instance role finishes by-identifier as <instance>.<role>, got {err:?}"
        );
    }

    /// An instance's still-open interior params surface path-qualified under
    /// the instance identifier (`<instance>.<node>.<param>`, #317) — the
    /// existing `collect_params` composite-arm prefix rule, now exercised
    /// through `Op::Use`'s rename-on-splice (extends the
    /// `param_space_is_flat_path_qualified_and_slot_disambiguated` family).
    #[test]
    fn use_op_instance_params_surface_path_qualified() {
        let subgraph = |_: &str| Some(use_fixture());
        let mut s = GraphSession::new("g", &std_vocabulary, &subgraph);
        // a bound root role (mirroring the headline twin test's shape;
        // `finish()` itself no longer cares either way, #317).
        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
        s.apply(Op::Use { ref_id: "fixture".into(), name: Some("gate".into()), bind: vec![] }).unwrap();
        s.apply(Op::Feed { role: "price".into(), into: vec!["gate.price".into()] }).unwrap();
        s.apply(Op::Expose { from: "gate.out".into(), as_name: "bias".into() }).unwrap();
        let c = s.finish().expect("finishes");
        let names: Vec<String> = c.param_space().into_iter().map(|p| p.name).collect();
        assert_eq!(names, ["gate.sma.length", "gate.bias.scale"]);
    }

    /// Open patterns end-to-end (#317, spec §"Open patterns"): a pattern
    /// declaring its formal parameter as an `input` role `finish()`es even
    /// though nothing ever binds it (the root-role gate no longer runs in
    /// `finish`), survives a real serialize/deserialize round trip through
    /// the stored JSON form (the same bytes a registry fetch would hand
    /// back), and a source-rooted consumer script splices the RELOADED
    /// pattern by ref id through the injected `subgraph` lookup — the outer
    /// script compiles and runs, proving the whole author's round trip
    /// (build open pattern -> store -> reload -> `use` -> run), not just an
    /// in-memory splice.
    #[test]
    fn open_input_pattern_finishes_registers_shaped_and_splices() {
        use crate::blueprint_serde::{blueprint_from_json, blueprint_to_json};

        // (a) the reusable pattern: an open `x` input role feeds an Sma, whose
        // value re-exports as `out`.
        let pattern_ops = vec![
            Op::Input { role: "x".into() },
            Op::Add {
                type_id: "SMA".into(),
                as_name: Some("sma".into()),
                args: vec![], bind: vec![("length".into(), Scalar::i64(3))],
            },
            Op::Feed { role: "x".into(), into: vec!["sma.series".into()] },
            Op::Expose { from: "sma.value".into(), as_name: "out".into() },
        ];
        let pattern = super::replay("pattern", pattern_ops, &std_vocabulary, &|_: &str| None)
            .expect("an input-role pattern finishes without root-role boundness");

        // (b) round-trip through the serialized store form.
        let json = blueprint_to_json(&pattern).expect("an open pattern serializes");

        // (c) a source-rooted outer script splices the RELOADED pattern by ref
        // id (the injected `subgraph` lookup re-parses the stored JSON on
        // every call, mirroring a fresh registry fetch — `Composite` is not
        // `Clone`).
        let subgraph = |ref_id: &str| {
            (ref_id == "pattern-id")
                .then(|| blueprint_from_json(&json, &std_vocabulary).expect("an open pattern deserializes"))
        };
        let outer_ops = vec![
            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
            Op::Use { ref_id: "pattern-id".into(), name: Some("p".into()), bind: vec![] },
            Op::Feed { role: "price".into(), into: vec!["p.x".into()] },
            Op::Expose { from: "p.out".into(), as_name: "bias".into() },
        ];
        let outer = super::replay("root", outer_ops, &std_vocabulary, &subgraph)
            .expect("the outer script splices the reloaded open pattern");

        // (d) compiles and runs — the same recording-harness shape
        // `run_recording` uses, but with zero open params (this pattern binds
        // `sma.length` at the pattern seam, unlike the shared fixture, so
        // `run_recording`'s hardcoded one-F64-param point does not apply).
        use crate::harness::{Edge, Target};
        use aura_core::Firing;
        use aura_std::Recorder;
        use std::sync::mpsc;
        let (tx, rx) = mpsc::channel();
        let root = Composite::new(
            "h",
            vec![
                BlueprintNode::Composite(outer),
                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx).into(),
            ],
            vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
            vec![Role {
                name: "src".into(),
                targets: vec![Target { node: 0, slot: 0 }],
                source: Some(ScalarKind::F64),
            }],
            vec![],
        );
        let mut h = root.bootstrap().expect("bootstraps with no open params");
        h.run(vec![Box::new(crate::VecSource::new(crate::test_fixtures::synthetic_prices()))]);
        let trace: Vec<_> = rx.try_iter().collect();
        assert!(!trace.is_empty(), "the spliced open pattern must actually produce output");
    }
}