feat(0088): §B construction op-script surface — GraphSession/Op/replay + introspection

Iteration-1 §B of the construction service (#157), on the §A shared predicates
(ea1ca32): a declarative, replayable by-identifier op-script that builds a
runnable blueprint through validated ops, reusing the engine's existing gates at
two cadences — no second validator (C24).

- construction.rs: `Op` (Source/Input/Add/Feed/Connect/Expose, 1:1 with the
  document, dotted by-identifier ports), `OpError` (by-identifier causes),
  `GraphSession` (per-op-fallible accumulator) and the free `replay` driver
  (stop-at-first-failing-op, naming it by `(op_index, OpError)`).
  - Eager (per-op): name resolution + edge_kind_check + the >1-cover
    DoubleWiredPort arm + try_bind — all calling the §A shared predicates.
  - Holistic (finish): the 0-cover totality arm, param-namespace injectivity, and
    root-role boundness — the SAME unchanged engine gates
    (check_param_namespace_injective / validate_wiring), now at end-of-document.
  - Build-free introspection: `unwired()` (a partial document's still-open slots).
- aura-std std_vocabulary_types(): the enumerable companion to the closed
  std_vocabulary match (a `fn(&str)->Option<…>` resolver cannot be listed),
  lockstep-tested in the list->resolver direction; the reverse fails safe (#160).
- aura-engine exports replay/GraphSession/Op/OpError + re-exports BindOpError.

Acceptance (engine level): a sink-free price->fast/slow SMA->Sub->Bias signal
root built through the ops compiles to a FlatGraph byte-identical (edges +
sources) to its GraphBuilder twin (C1); an invalid op is rejected at the op,
named; vocabulary + unwired slots answer without a build. Pinned by
construction.rs unit tests + a construction_e2e.rs integration test over the
public seam the iteration-2 CLI will consume.

Notable judgement calls folded in from the implement loop: feed is all-or-nothing
(two-phase resolve-then-commit, no partial wiring on a mid-fan-out fault); the
node identifier is stamped onto the builder (`named(&id)`) so two same-type nodes
get distinct param paths and do not trip the injectivity gate prematurely; a
module-level `#![allow(dead_code)]` covers the public surface until the
iteration-2 CLI consumes it. Full suite green; clippy --all-targets -D warnings
clean.

Iteration 2 (the JSON op-list encoding + `aura graph build`/`introspect` CLI)
remains; #157 stays open. refs #157

crates/aura-engine/src/construction.rs

@@ -0,0 +1,608 @@
+//! Per-op-fallible blueprint construction — the eager half of the §A gate split
+//! (#157). A `GraphSession` applies `Op`s one at a time, running the
+//! eagerly-decidable checks (name resolution, edge kind-match, the >1-cover
+//! double-wire arm, param bind) at the offending op; the only-at-end-decidable
+//! checks (0-cover totality, param-namespace injectivity, root-role boundness)
+//! run holistically in `finish`. Both cadences call the SAME §A predicates — no
+//! second validator (C24). The node vocabulary is an injected closed resolver
+//! (`Fn(&str)->Option<PrimitiveBuilder>`), so the engine stays domain-free
+//! (invariant 9).
+
+// `construction` is a private module whose public surface — `Op` / `OpError` /
+// `GraphSession` (with `add` / `source` / `input` / `connect` / `feed` /
+// `expose` / `unwired` / `finish`) and the free `replay` fn, plus the session
+// accumulator fields (`nodes` / `ids` / `edges` / `out` / `name`) — has no
+// non-test consumer yet: the iteration-2 CLI (`aura graph build`/`introspect`)
+// is what wires it in. Until then the compiler sees the whole surface as dead,
+// so this allow is still load-bearing; it is removed when that consumer lands.
+#![allow(dead_code)]
+
+use std::collections::{HashMap, HashSet};
+
+use aura_core::{BindOpError, NodeSchema, PrimitiveBuilder, Scalar, ScalarKind};
+
+use crate::blueprint::{
+    check_param_namespace_injective, edge_kind_check, validate_wiring, BlueprintNode, Composite,
+    OutField, Role,
+};
+use crate::builder::{resolve_input_slot, resolve_output_field, PortResolveError};
+use crate::harness::{Edge, Target};
+use crate::CompileError;
+
+/// One construction act — the engine-side op vocabulary, 1:1 with the document.
+/// Port references are dotted by-identifier (`"node.field"` / `"node.slot"`),
+/// split at the first `.`. (serde `Deserialize` is iteration 2.)
+#[derive(Debug, PartialEq)]
+pub enum Op {
+    /// Reserve a bound root source role of `kind`.
+    Source { role: String, kind: ScalarKind },
+    /// Reserve an open input role (wired by an enclosing graph).
+    Input { role: String },
+    /// Add a node of `type_id`, optionally named `as_name`, with `bind` params.
+    Add { type_id: String, as_name: Option<String>, bind: Vec<(String, Scalar)> },
+    /// Fan a role into one or more `"node.slot"` consumers.
+    Feed { role: String, into: Vec<String> },
+    /// Wire `"node.field"` -> `"node.slot"`.
+    Connect { from: String, to: String },
+    /// Re-export `"node.field"` at the boundary under `as_name`.
+    Expose { from: String, as_name: String },
+}
+
+/// A per-op construction fault, by-identifier so the cause names the op.
+#[derive(Debug, PartialEq)]
+pub enum OpError {
+    UnknownNodeType(String),
+    /// An interior node-identifier collision (the `names` namespace).
+    DuplicateIdentifier(String),
+    /// A boundary output-name collision (the separate `out_names` namespace) — a
+    /// distinct fault class from an interior node-id clash, so callers can tell the
+    /// two apart from the error value alone.
+    DuplicateOutput(String),
+    DuplicateRole(String),
+    UnknownIdentifier(String),
+    UnknownRole(String),
+    MalformedPort(String),
+    UnknownInPort { node: String, name: String },
+    AmbiguousInPort { node: String, name: String },
+    UnknownOutPort { node: String, name: String },
+    AmbiguousOutPort { node: String, name: String },
+    KindMismatch { from: String, to: String, producer: ScalarKind, consumer: ScalarKind },
+    SlotAlreadyWired { node: String, slot: String },
+    BadParam { node: String, err: BindOpError },
+    /// A holistic finalize fault (totality / injectivity / unbound root role),
+    /// wrapping the unchanged engine gate's `CompileError`.
+    Incomplete(CompileError),
+}
+
+/// A per-op-fallible blueprint accumulator. Holds the same interior data a
+/// `Composite` is built from, plus the incremental coverage map (eager
+/// double-wire front-run) and the identifier→index maps the document references.
+pub struct GraphSession<'v> {
+    name: String,
+    vocab: &'v dyn Fn(&str) -> Option<PrimitiveBuilder>,
+    nodes: Vec<BlueprintNode>,
+    schemas: Vec<NodeSchema>,
+    ids: Vec<String>,
+    names: HashMap<String, usize>,
+    edges: Vec<Edge>,
+    roles: Vec<(String, Option<ScalarKind>, Vec<Target>)>,
+    role_names: HashMap<String, usize>,
+    out: Vec<OutField>,
+    out_names: HashSet<String>,
+    coverage: HashMap<(usize, usize), usize>,
+}
+
+impl<'v> GraphSession<'v> {
+    /// Start a session for a composite named `name`, resolving node types through
+    /// the injected closed vocabulary `vocab`.
+    pub fn new(name: impl Into<String>, vocab: &'v dyn Fn(&str) -> Option<PrimitiveBuilder>) -> Self {
+        Self {
+            name: name.into(),
+            vocab,
+            nodes: Vec::new(),
+            schemas: Vec::new(),
+            ids: Vec::new(),
+            names: HashMap::new(),
+            edges: Vec::new(),
+            roles: Vec::new(),
+            role_names: HashMap::new(),
+            out: Vec::new(),
+            out_names: HashSet::new(),
+            coverage: HashMap::new(),
+        }
+    }
+
+    /// Apply one op, running the eager checks. The first `Err` is the op-script's
+    /// stop point (`replay` attributes it to the op index).
+    pub fn apply(&mut self, op: Op) -> Result<(), OpError> {
+        match op {
+            Op::Source { role, kind } => self.add_role(role, Some(kind)),
+            Op::Input { role } => self.add_role(role, None),
+            Op::Add { type_id, as_name, bind } => self.add_node(type_id, as_name, bind),
+            Op::Feed { role, into } => self.feed(role, into),
+            Op::Connect { from, to } => self.connect(from, to),
+            Op::Expose { from, as_name } => self.expose(from, as_name),
+        }
+    }
+
+    fn add_role(&mut self, role: String, kind: Option<ScalarKind>) -> Result<(), OpError> {
+        if self.role_names.contains_key(&role) {
+            return Err(OpError::DuplicateRole(role));
+        }
+        let idx = self.roles.len();
+        self.role_names.insert(role.clone(), idx);
+        self.roles.push((role, kind, Vec::new()));
+        Ok(())
+    }
+
+    fn add_node(
+        &mut self,
+        type_id: String,
+        as_name: Option<String>,
+        bind: Vec<(String, Scalar)>,
+    ) -> Result<(), OpError> {
+        let mut builder =
+            (self.vocab)(&type_id).ok_or_else(|| OpError::UnknownNodeType(type_id.clone()))?;
+        let id = as_name.unwrap_or_else(|| builder.node_name());
+        for (slot, value) in bind {
+            builder = builder
+                .try_bind(&slot, value)
+                .map_err(|err| OpError::BadParam { node: id.clone(), err })?;
+        }
+        if self.names.contains_key(&id) {
+            return Err(OpError::DuplicateIdentifier(id));
+        }
+        // Stamp the chosen identifier onto the builder so the node's own name (and
+        // hence its `param_space()` path prefix) matches the session id — without
+        // this, two same-type nodes keep the default type-label name and collide on
+        // the holistic param-namespace injectivity gate in `finish`.
+        let builder = builder.named(&id);
+        let node = BlueprintNode::from(builder);
+        let schema = node.signature();
+        let idx = self.nodes.len();
+        self.names.insert(id.clone(), idx);
+        self.ids.push(id);
+        self.nodes.push(node);
+        self.schemas.push(schema);
+        Ok(())
+    }
+
+    /// Split a dotted `"node.port"` reference at the first `.`.
+    fn split_port(dotted: &str) -> Result<(String, String), OpError> {
+        match dotted.split_once('.') {
+            Some((node, port)) if !node.is_empty() && !port.is_empty() => {
+                Ok((node.to_string(), port.to_string()))
+            }
+            _ => Err(OpError::MalformedPort(dotted.to_string())),
+        }
+    }
+
+    fn node_index(&self, id: &str) -> Result<usize, OpError> {
+        self.names.get(id).copied().ok_or_else(|| OpError::UnknownIdentifier(id.to_string()))
+    }
+
+    /// 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"),
+        })?;
+        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(())
+    }
+
+    /// The still-unwired interior input slots (build-free introspection): every
+    /// added node's declared input slots minus the covered ones, by-identifier.
+    pub fn unwired(&self) -> Vec<(String, ScalarKind)> {
+        let mut open = Vec::new();
+        for (n, schema) in self.schemas.iter().enumerate() {
+            for (slot, port) in schema.inputs.iter().enumerate() {
+                if self.coverage.get(&(n, slot)).copied().unwrap_or(0) == 0 {
+                    open.push((format!("{}.{}", self.ids[n], port.name), port.kind));
+                }
+            }
+        }
+        open
+    }
+
+    /// Assemble the `Composite` and run the holistic gates (totality,
+    /// param-namespace injectivity, root-role boundness) — the SAME engine
+    /// predicates the eager path shares, now at end-of-document cadence. Any
+    /// fault wraps as `OpError::Incomplete`.
+    pub fn finish(self) -> Result<Composite, OpError> {
+        let roles: Vec<Role> = self
+            .roles
+            .into_iter()
+            .map(|(name, source, targets)| Role { name, targets, source })
+            .collect();
+        let c = Composite::new(self.name, self.nodes, self.edges, roles, self.out);
+        check_param_namespace_injective(&c.param_space()).map_err(OpError::Incomplete)?;
+        validate_wiring(c.nodes(), c.edges(), c.input_roles(), c.output()).map_err(OpError::Incomplete)?;
+        for (r, role) in c.input_roles().iter().enumerate() {
+            if role.source.is_none() {
+                return Err(OpError::Incomplete(CompileError::UnboundRootRole { role: r }));
+            }
+        }
+        Ok(c)
+    }
+}
+
+/// Replay a document of ops from scratch, stopping at the first failing op and
+/// naming it by `(op_index, OpError)`; a holistic finalize fault is attributed to
+/// index `ops.len()` (the implicit finalize step).
+pub fn replay(
+    name: impl Into<String>,
+    ops: Vec<Op>,
+    vocab: &dyn Fn(&str) -> Option<PrimitiveBuilder>,
+) -> Result<Composite, (usize, OpError)> {
+    let mut s = GraphSession::new(name, vocab);
+    let n = ops.len();
+    for (i, op) in ops.into_iter().enumerate() {
+        s.apply(op).map_err(|e| (i, e))?;
+    }
+    s.finish().map_err(|e| (n, e))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{CompileError, GraphSession, Op, OpError};
+    use aura_core::{BindOpError, Scalar, ScalarKind};
+    use aura_std::std_vocabulary;
+
+    fn session(name: &str) -> GraphSession<'static> {
+        GraphSession::new(name, &std_vocabulary)
+    }
+
+    #[test]
+    fn add_unknown_type_rejected_at_op() {
+        let mut s = session("g");
+        assert_eq!(
+            s.apply(Op::Add { type_id: "Nope".into(), as_name: None, bind: vec![] }),
+            Err(OpError::UnknownNodeType("Nope".into()))
+        );
+    }
+
+    #[test]
+    fn add_duplicate_identifier_rejected() {
+        let mut s = session("g");
+        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("x".into()), bind: vec![] }).unwrap();
+        assert_eq!(
+            s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("x".into()), bind: vec![] }),
+            Err(OpError::DuplicateIdentifier("x".into()))
+        );
+    }
+
+    #[test]
+    fn add_bad_bind_param_rejected_at_op() {
+        let mut s = session("g");
+        assert_eq!(
+            s.apply(Op::Add {
+                type_id: "SMA".into(),
+                as_name: Some("fast".into()),
+                bind: vec![("length".into(), Scalar::f64(2.0))], // wrong kind (i64 param)
+            }),
+            Err(OpError::BadParam {
+                node: "fast".into(),
+                err: BindOpError::KindMismatch {
+                    param: "length".into(),
+                    expected: ScalarKind::I64,
+                    got: ScalarKind::F64,
+                },
+            })
+        );
+    }
+
+    #[test]
+    fn duplicate_role_rejected() {
+        let mut s = session("g");
+        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
+        assert_eq!(
+            s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }),
+            Err(OpError::DuplicateRole("price".into()))
+        );
+    }
+
+    // helper: a fast/slow/sub scaffold sharing a price source
+    fn scaffold() -> GraphSession<'static> {
+        let mut s = session("g");
+        s.apply(Op::Source { role: "price".into(), kind: ScalarKind::F64 }).unwrap();
+        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] }).unwrap();
+        s.apply(Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] }).unwrap();
+        s.apply(Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] }).unwrap();
+        s
+    }
+
+    #[test]
+    fn connect_unknown_out_port_rejected_at_op() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Connect { from: "fast.nope".into(), to: "sub.lhs".into() }),
+            Err(OpError::UnknownOutPort { node: "fast".into(), name: "nope".into() })
+        );
+    }
+
+    #[test]
+    fn connect_double_wire_rejected_at_op() {
+        let mut s = scaffold();
+        s.apply(Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() }).unwrap();
+        assert_eq!(
+            s.apply(Op::Connect { from: "slow.value".into(), to: "sub.lhs".into() }),
+            Err(OpError::SlotAlreadyWired { node: "sub".into(), slot: "lhs".into() })
+        );
+    }
+
+    #[test]
+    fn connect_kind_mismatch_rejected_at_op() {
+        // Gt produces bool; Bias.signal consumes f64 -> eager kind mismatch
+        let mut s = session("g");
+        s.apply(Op::Add { type_id: "Gt".into(), as_name: None, bind: vec![] }).unwrap();
+        s.apply(Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] }).unwrap();
+        let r = s.apply(Op::Connect { from: "gt.value".into(), to: "bias.signal".into() });
+        assert!(matches!(
+            r,
+            Err(OpError::KindMismatch { producer: ScalarKind::Bool, consumer: ScalarKind::F64, .. })
+        ));
+    }
+
+    #[test]
+    fn feed_into_unknown_role_rejected() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Feed { role: "ghost".into(), into: vec!["fast.series".into()] }),
+            Err(OpError::UnknownRole("ghost".into()))
+        );
+    }
+
+    /// `feed` covers each named slot: a successful fan-out registers coverage, so
+    /// a second `feed` into an already-covered slot is rejected at the op.
+    #[test]
+    fn feed_success_covers_each_slot() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Feed {
+                role: "price".into(),
+                into: vec!["fast.series".into(), "slow.series".into()],
+            }),
+            Ok(())
+        );
+        assert_eq!(
+            s.apply(Op::Feed { role: "price".into(), into: vec!["fast.series".into()] }),
+            Err(OpError::SlotAlreadyWired { node: "fast".into(), slot: "series".into() })
+        );
+    }
+
+    /// `feed` is all-or-nothing: when a later target in the same fan-out fails, the
+    /// earlier (valid) targets are left unwired — the failed op leaks no partial
+    /// coverage into the session (so `unwired`/a subsequent op see consistent state).
+    #[test]
+    fn feed_partial_failure_is_atomic() {
+        let mut s = scaffold();
+        // fast.series is valid; ghost.series names an unknown node, so the feed fails
+        // on its second target.
+        assert_eq!(
+            s.apply(Op::Feed {
+                role: "price".into(),
+                into: vec!["fast.series".into(), "ghost.series".into()],
+            }),
+            Err(OpError::UnknownIdentifier("ghost".into()))
+        );
+        // fast.series must still be open: the failed feed wired nothing.
+        assert!(s.unwired().iter().any(|(p, _)| p == "fast.series"));
+    }
+
+    /// An open `Input` role (kind = None) is reserved like a `Source`, sharing the
+    /// duplicate-role guard.
+    #[test]
+    fn input_role_reserved_and_duplicate_rejected() {
+        let mut s = session("g");
+        assert_eq!(s.apply(Op::Input { role: "ext".into() }), Ok(()));
+        assert_eq!(
+            s.apply(Op::Input { role: "ext".into() }),
+            Err(OpError::DuplicateRole("ext".into()))
+        );
+    }
+
+    /// A port reference without a `.` separator is rejected as malformed at the op.
+    #[test]
+    fn connect_malformed_port_rejected() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Connect { from: "fastvalue".into(), to: "sub.lhs".into() }),
+            Err(OpError::MalformedPort("fastvalue".into()))
+        );
+    }
+
+    /// A port naming a node that was never added is rejected by-identifier.
+    #[test]
+    fn connect_unknown_node_identifier_rejected() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Connect { from: "ghost.value".into(), to: "sub.lhs".into() }),
+            Err(OpError::UnknownIdentifier("ghost".into()))
+        );
+    }
+
+    /// `expose` re-exports a resolved output field; a second export under the same
+    /// boundary name is rejected as a duplicate identifier.
+    #[test]
+    fn expose_success_and_duplicate_output_rejected() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Expose { from: "fast.value".into(), as_name: "out".into() }),
+            Ok(())
+        );
+        assert_eq!(
+            s.apply(Op::Expose { from: "slow.value".into(), as_name: "out".into() }),
+            Err(OpError::DuplicateOutput("out".into()))
+        );
+    }
+
+    /// `expose` maps an unresolved output field onto the by-identifier port fault.
+    #[test]
+    fn expose_unknown_out_port_rejected() {
+        let mut s = scaffold();
+        assert_eq!(
+            s.apply(Op::Expose { from: "fast.nope".into(), as_name: "out".into() }),
+            Err(OpError::UnknownOutPort { node: "fast".into(), name: "nope".into() })
+        );
+    }
+
+    #[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::Incomplete(CompileError::UnconnectedPort { .. })));
+    }
+
+    #[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"));
+    }
+
+    #[test]
+    fn replay_stops_at_first_failing_op_naming_index() {
+        let ops = vec![
+            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] },
+            Op::Add { type_id: "Nope".into(), as_name: None, bind: vec![] }, // op 1 fails
+            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
+        ];
+        // `.err().unwrap()` rather than `.unwrap_err()`: the Ok arm `Composite` is
+        // not `Debug` (it holds build closures), which `unwrap_err` would require.
+        let err = super::replay("g", ops, &aura_std::std_vocabulary).err().unwrap();
+        assert_eq!(err, (1, OpError::UnknownNodeType("Nope".into())));
+    }
+
+    /// Acceptance-1 (C24 replay-equals-builder): an op-script replayed through the
+    /// shared §A gates compiles to the byte-identical `FlatGraph` topology
+    /// (`edges` + `sources`) as the same signal root hand-wired on the
+    /// `GraphBuilder` surface — the two construction paths are one construction
+    /// semantics. Models `builder_harness_compiles_identically_to_hand_wired`
+    /// (builder.rs:269), sink-free over the zero-arg vocabulary.
+    #[test]
+    fn replay_built_signal_compiles_identically_to_graphbuilder() {
+        use crate::GraphBuilder;
+        use aura_core::Scalar;
+        use aura_std::{Bias, Sma, Sub};
+
+        // param space of the flat root: fast.length(i64), slow.length(i64),
+        // bias.scale(f64) — same vector applied to both sides, so it cancels.
+        let params = [Scalar::i64(2), Scalar::i64(4), Scalar::f64(0.5)];
+
+        // (a) op-script replay
+        let ops = vec![
+            Op::Source { role: "price".into(), kind: ScalarKind::F64 },
+            Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] },
+            Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] },
+            Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
+            Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] },
+            Op::Feed { role: "price".into(), into: vec!["fast.series".into(), "slow.series".into()] },
+            Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() },
+            Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() },
+            Op::Connect { from: "sub.value".into(), to: "bias.signal".into() },
+            Op::Expose { from: "bias.bias".into(), as_name: "bias".into() },
+        ];
+        let replay_flat = super::replay("root", ops, &aura_std::std_vocabulary)
+            .expect("replay resolves")
+            .compile_with_params(&params)
+            .expect("replay compiles");
+
+        // (b) the GraphBuilder twin — identical topology
+        let mut g = GraphBuilder::new("root");
+        let fast = g.add(Sma::builder().named("fast"));
+        let slow = g.add(Sma::builder().named("slow"));
+        let sub = g.add(Sub::builder());
+        let bias = g.add(Bias::builder());
+        let price = g.source_role("price", ScalarKind::F64);
+        g.feed(price, [fast.input("series"), slow.input("series")]);
+        g.connect(fast.output("value"), sub.input("lhs"));
+        g.connect(slow.output("value"), sub.input("rhs"));
+        g.connect(sub.output("value"), bias.input("signal"));
+        g.expose(bias.output("bias"), "bias");
+        let builder_flat = g.build().expect("root resolves").compile_with_params(&params).expect("compiles");
+
+        assert_eq!(replay_flat.edges, builder_flat.edges);
+        assert_eq!(replay_flat.sources, builder_flat.sources);
+    }
+}

crates/aura-engine/src/lib.rs

@@ -45,6 +45,7 @@
 mod blueprint;
 mod blueprint_serde;
 mod builder;
+mod construction;
 mod graph_model;
 mod harness;
 mod mc;
@@ -61,6 +62,7 @@ pub use blueprint_serde::{
     PrimitiveData, SerializeError, BLUEPRINT_FORMAT_VERSION,
 };
 pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHandle};
+pub use construction::{replay, GraphSession, Op, OpError};
 pub use graph_model::model_to_json;
 pub use harness::{
     window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SplitMix64,
@@ -85,7 +87,9 @@ pub use walkforward::{
 // #29: re-export the core scalar vocabulary a Blueprint builder needs
 // (SourceSpec.kind is a ScalarKind; sources/Recorder columns are Scalar /
 // Firing / Timestamp) so a graph builder has one import surface, not two.
-pub use aura_core::{Firing, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp};
+pub use aura_core::{
+    BindOpError, Firing, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp,
+};
 
 #[cfg(test)]
 mod test_fixtures;

crates/aura-engine/tests/construction_e2e.rs

@@ -0,0 +1,134 @@
+//! End-to-end coverage for the construction op-script (#157, cycle 0088): a
+//! declarative, replayable by-identifier op-script (`replay` over a `Vec<Op>`)
+//! builds a runnable blueprint through validated ops, reusing the engine's
+//! existing gates at two cadences — eager (per-op) and holistic (finalize).
+//!
+//! The in-module tests in `construction.rs` reach into crate internals
+//! (`crate::GraphBuilder`, `crate::CompileError`, `super::replay`). These tests
+//! use only the **exported** surface — `aura_engine::{replay, Op, OpError,
+//! GraphBuilder, Composite, FlatGraph, Scalar, ScalarKind}` plus the injected
+//! `aura_std::std_vocabulary` — the exact seam the iteration-2 CLI (`aura graph
+//! build`/`introspect`) and the World consume. A refactor that made `Op`,
+//! `OpError`, `replay`, or a needed re-export (`ScalarKind`, `Scalar`)
+//! unreachable across the crate boundary would break that consumer while leaving
+//! every in-crate test green; that is the regression class this file pins, in
+//! addition to the per-property invariants below.
+
+use aura_engine::{replay, CompileError, GraphBuilder, Op, OpError, Scalar, ScalarKind};
+use aura_std::{std_vocabulary, Bias, Sma, Sub};
+
+/// The flat root's open param point: fast.length(i64), slow.length(i64),
+/// bias.scale(f64), in param-space traversal order. The same vector is applied to
+/// both construction paths, so it cancels and the only thing under test is
+/// construction identity. Deterministic — no time, no randomness.
+fn params() -> [Scalar; 3] {
+    [Scalar::i64(2), Scalar::i64(4), Scalar::f64(0.5)]
+}
+
+/// The SMA-cross signal as a construction op-script: a bound `price` source, a
+/// fast/slow SMA pair fed from it, their spread through `Sub`, a `Bias`, and the
+/// `bias` field exposed at the boundary. The smallest script that exercises every
+/// op kind (`Source`/`Add`/`Feed`/`Connect`/`Expose`) at once.
+fn signal_ops() -> Vec<Op> {
+    vec![
+        Op::Source { role: "price".into(), kind: ScalarKind::F64 },
+        Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] },
+        Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] },
+        Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },
+        Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] },
+        Op::Feed { role: "price".into(), into: vec!["fast.series".into(), "slow.series".into()] },
+        Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() },
+        Op::Connect { from: "slow.value".into(), to: "sub.rhs".into() },
+        Op::Connect { from: "sub.value".into(), to: "bias.signal".into() },
+        Op::Expose { from: "bias.bias".into(), as_name: "bias".into() },
+    ]
+}
+
+/// Property (C24 acceptance-1, through the public seam): an op-script replayed via
+/// the exported `replay` + the injected `aura_std::std_vocabulary` compiles to a
+/// `FlatGraph` whose index-wired topology (`edges`) and lowered bound sources
+/// (`sources`) are **identical** to the same signal hand-wired on the exported
+/// `GraphBuilder` surface and compiled with the same param point. The two
+/// authoring surfaces are one construction semantics (C24) — proven entirely
+/// through `aura_engine`/`aura_std` public exports (the CLI/World seam). The
+/// non-empty `edges` guard keeps the equality from being the vacuous match of two
+/// empty graphs.
+#[test]
+fn replayed_construction_compiles_identically_to_graphbuilder() {
+    // (a) op-script replay through the public construction surface.
+    let replay_flat = replay("root", signal_ops(), &std_vocabulary)
+        .expect("op-script resolves")
+        .compile_with_params(&params())
+        .expect("replay compiles");
+
+    // (b) the GraphBuilder twin — identical topology, identical param point.
+    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("twin compiles");
+
+    assert!(!replay_flat.edges.is_empty(), "the signal must wire a non-trivial topology");
+    assert_eq!(replay_flat.edges, builder_flat.edges, "construction topology diverged from the builder twin");
+    assert_eq!(replay_flat.sources, builder_flat.sources, "lowered bound sources diverged from the builder twin");
+}
+
+/// Property (eager fail-fast cadence, at the public boundary): an eagerly-decidable
+/// fault — here an `Add` of a `type_id` outside the closed vocabulary — stops
+/// `replay` AT the offending op and is reported by `(op_index, OpError)`, the
+/// op_index being the position of the bad op (2 here), not the end of the script.
+/// The earlier valid ops do not mask it and the later ops never run. This is the
+/// by-index error attribution the CLI/World relies on to point an author at the
+/// exact failing op, asserted through the exported `replay`/`Op`/`OpError`.
+#[test]
+fn replay_attributes_eager_fault_to_its_op_index() {
+    let ops = vec![
+        Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] }, // 0 ok
+        Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },                // 1 ok
+        Op::Add { type_id: "Nope".into(), as_name: None, bind: vec![] },               // 2 fails
+        Op::Add { type_id: "Bias".into(), as_name: None, bind: vec![] },               // 3 never runs
+    ];
+    // `.err().unwrap()` (not `unwrap_err`): the Ok arm `Composite` holds build
+    // closures and is not `Debug`, which `unwrap_err`'s bound would require.
+    let err = replay("g", ops, &std_vocabulary).err().unwrap();
+    assert_eq!(err, (2, OpError::UnknownNodeType("Nope".into())));
+}
+
+/// Property (holistic finalize cadence, at the public boundary): a fault that is
+/// only decidable at end-of-document — input-slot totality (the 0-cover arm of
+/// exactly-once wiring) — passes every per-op check and is caught by the implicit
+/// finalize step, attributed to index `ops.len()` (6 here), strictly past any real
+/// op index. Every op below applies cleanly; `sub.rhs` is left unwired, so only the
+/// holistic `validate_wiring` gate at finalize can reject it, as
+/// `OpError::Incomplete(CompileError::UnconnectedPort { .. })`. Together with the
+/// eager-cadence test, this pins the iteration's eager/holistic gate split as
+/// observable error attribution across the crate boundary.
+#[test]
+fn replay_attributes_holistic_fault_to_finalize_step() {
+    let ops = vec![
+        Op::Source { role: "price".into(), kind: ScalarKind::F64 },                    // 0
+        Op::Add { type_id: "SMA".into(), as_name: Some("fast".into()), bind: vec![] }, // 1
+        Op::Add { type_id: "SMA".into(), as_name: Some("slow".into()), bind: vec![] }, // 2
+        Op::Add { type_id: "Sub".into(), as_name: None, bind: vec![] },                // 3
+        Op::Feed { role: "price".into(), into: vec!["fast.series".into(), "slow.series".into()] }, // 4
+        Op::Connect { from: "fast.value".into(), to: "sub.lhs".into() },               // 5
+        // sub.rhs deliberately left unwired -> only finalize can decide totality.
+    ];
+    let finalize_index = ops.len(); // 6 — the implicit finalize step
+    let err = replay("g", ops, &std_vocabulary).err().unwrap();
+    assert!(
+        matches!(err, (i, OpError::Incomplete(CompileError::UnconnectedPort { .. })) if i == finalize_index),
+        "an only-at-end-decidable fault must be attributed to the finalize step, got {err:?}",
+    );
+}

crates/aura-std/src/lib.rs

@@ -83,5 +83,5 @@ pub use sma::Sma;
 pub use sqrt::Sqrt;
 pub use stop_rule::FixedStop;
 pub use sub::Sub;
-pub use vocabulary::std_vocabulary;
+pub use vocabulary::{std_vocabulary, std_vocabulary_types};
 pub use vol_slippage_cost::VolSlippageCost;

crates/aura-std/src/vocabulary.rs

@@ -55,6 +55,22 @@ pub fn std_vocabulary(type_id: &str) -> Option<PrimitiveBuilder> {
     })
 }
 
+/// The enumerable companion to [`std_vocabulary`]: the closed set of zero-arg
+/// type identities the resolver answers. A `fn(&str)->Option<…>` resolver cannot
+/// be listed, so build-free vocabulary introspection (#157) reads this. Kept in
+/// lockstep with the `std_vocabulary` match arms by maintainer discipline — a
+/// type added there must be added here. The reverse (an arm with no entry here)
+/// is not test-enforced and cannot be, from a `fn` resolver; it fails safe (the
+/// type stays load-resolvable, only unlisted; #160).
+pub fn std_vocabulary_types() -> &'static [&'static str] {
+    &[
+        "Add", "And", "Bias", "CarryCost", "ConstantCost", "Delay", "EMA",
+        "EqConst", "FixedStop", "Gt", "Latch", "LongOnly", "Mul",
+        "PositionManagement", "Resample", "RollingMax", "RollingMin", "Sizer",
+        "SMA", "Sqrt", "Sub", "VolSlippageCost",
+    ]
+}
+
 #[cfg(test)]
 mod tests {
     use super::std_vocabulary;
@@ -107,4 +123,29 @@ mod tests {
         assert!(std_vocabulary("LinComb").is_none());
         assert!(std_vocabulary("Recorder").is_none());
     }
+
+    #[test]
+    fn std_vocabulary_types_lists_exactly_the_resolvable_keys() {
+        use super::std_vocabulary_types;
+        // every listed type resolves to a builder carrying that exact label
+        for &type_id in std_vocabulary_types() {
+            let builder = std_vocabulary(type_id)
+                .unwrap_or_else(|| panic!("{type_id} listed but unresolvable"));
+            assert_eq!(
+                builder.label(),
+                type_id,
+                "listed type must round-trip to its label"
+            );
+        }
+        // the list is the WHOLE resolvable set: a couple of known non-members stay out
+        assert!(!std_vocabulary_types().contains(&"LinComb")); // construction-arg node
+        assert!(!std_vocabulary_types().contains(&"Recorder")); // sink
+        assert!(!std_vocabulary_types().contains(&"nope"));
+        // count guard: pins the list at exactly 22 entries, so a dropped or added
+        // list entry trips this; with the round-trip loop above, the list is
+        // exactly the 22 listed resolvable keys. The reverse drift — a resolver
+        // arm with no list entry — is not catchable from a `fn(&str)` resolver
+        // (not enumerable); it fails safe (load-resolvable, only unlisted; #160).
+        assert_eq!(std_vocabulary_types().len(), 22);
+    }
 }