feat(aura-core): typed construction-arg seam on PrimitiveBuilder

A second, closed construction channel beside the scalar bind seam
(refs #271): ArgKind (Tz / TimeOfDay / Count — deliberately NOT
ScalarKind, which stays invariant 4's streamed set), ArgSpec, ArgValue,
ConstructionArg (verbatim strict-form strings, the id-bearing twin of
BoundParam), ArgOpError, and an ArgsState on PrimitiveBuilder:
pending(name, doc, specs, make) declares an arg-bearing recipe that is
introspectable but not constructible; try_args validates strictly
(exact IANA name, zero-padded HH:MM, plain positive decimal — the
accepted form IS the canonical form, no normalization layer), then
runs make and returns the configured builder. build() on a pending
builder panics — the bind panic-contract twin; the data path always
goes through try_args.

chrono-tz enters aura-core for the typed Tz value (already a workspace
dependency of four crates; typed values at the seam beat a
validated-string re-parse split across two coupled sites).

One test per ArgOpError variant plus strict-form refusal pins.

crates/aura-core/Cargo.toml

@@ -7,6 +7,10 @@ publish.workspace = true
 
 [dependencies]
 serde = { workspace = true }
+# ArgValue::Tz carries chrono_tz::Tz (the closed IANA table, exact names only) —
+# already a workspace dependency of four other crates (see docs/specs/
+# construction-args.md § Dependency note).
+chrono-tz = { version = "0.10", default-features = false }
 
 [dev-dependencies]
 serde_json = { workspace = true }

crates/aura-core/src/lib.rs

@@ -44,8 +44,9 @@ pub use column::{Column, Window};
 pub use ctx::Ctx;
 pub use error::KindMismatch;
 pub use node::{
-    doc_gate, zip_params, BindOpError, BoundParam, DocGateFault, FieldSpec, Firing, Node,
-    NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
+    doc_gate, zip_params, ArgKind, ArgOpError, ArgSpec, ArgValue, BindOpError, BoundParam,
+    ConstructionArg, DocGateFault, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec,
+    PrimitiveBuilder,
 };
 pub use scalar::{Scalar, ScalarKind, Timestamp};
 pub use series_fold::SeriesFold;

crates/aura-core/src/node.rs

@@ -112,6 +112,7 @@ pub struct PrimitiveBuilder {
     instance_name: Option<String>,
     schema: NodeSchema,
     bound: Vec<BoundParam>,
+    args: ArgsState, // NEW: `Plain` for every existing (non-arg-bearing) node
     // The build closure's type is exactly the recipe contract (a param slice in, a
     // boxed node out); a type alias would not clarify it.
     #[allow(clippy::type_complexity)]
@@ -127,7 +128,108 @@ impl PrimitiveBuilder {
         schema: NodeSchema,
         build: impl Fn(&[Cell]) -> Box<dyn Node> + 'static,
     ) -> Self {
-        Self { name, instance_name: None, schema, bound: Vec::new(), build: Box::new(build) }
+        Self { name, instance_name: None, schema, bound: Vec::new(), args: ArgsState::Plain, build: Box::new(build) }
+    }
+
+    /// An arg-bearing recipe: introspectable (doc + declared `ArgSpec`s) but
+    /// not yet constructible. `schema` carries only the doc line (empty
+    /// inputs/output/params — the real signature forms inside `make`, once
+    /// `try_args` has parsed values to hand it); `build` on a pending builder
+    /// panics (the `bind` panic-contract twin) — the data path always goes
+    /// through `try_args` first, so no pending builder ever reaches `build`.
+    pub fn pending(
+        name: &'static str,
+        doc: &'static str,
+        specs: &'static [ArgSpec],
+        make: fn(&[(String, ArgValue)]) -> PrimitiveBuilder,
+    ) -> Self {
+        Self {
+            name,
+            instance_name: None,
+            schema: NodeSchema { doc, ..Default::default() },
+            bound: Vec::new(),
+            args: ArgsState::Pending { specs, make },
+            build: Box::new(move |_| {
+                panic!(
+                    "PrimitiveBuilder::build: `{name}` is an unconfigured arg-bearing type — \
+                     call try_args first"
+                )
+            }),
+        }
+    }
+
+    /// Validate raw `(name, value)` pairs against the declared `ArgSpec`s
+    /// (strict form — see `ArgKind::parse`), then run `make`, returning the
+    /// configured builder (`args: Configured{..}`) with this builder's
+    /// `instance_name` carried over. `Plain` + empty raw is `Ok(self)`
+    /// (uniform op application, #157's try_bind precedent); `Plain` +
+    /// non-empty raw is `NotArgBearing`. Validation order (deterministic):
+    /// first unknown arg name, then a duplicate, then the first spec-order
+    /// missing arg, then per-kind parse failures (`BadValue`).
+    pub fn try_args(self, raw: &[(String, String)]) -> Result<Self, ArgOpError> {
+        let (specs, make) = match &self.args {
+            ArgsState::Plain | ArgsState::Configured { .. } => {
+                return if raw.is_empty() { Ok(self) } else { Err(ArgOpError::NotArgBearing) };
+            }
+            ArgsState::Pending { specs, make } => (*specs, *make),
+        };
+
+        for (name, _) in raw {
+            if !specs.iter().any(|s| s.name == name.as_str()) {
+                return Err(ArgOpError::UnknownArg(name.clone()));
+            }
+        }
+        for i in 0..raw.len() {
+            if raw[i + 1..].iter().any(|(n, _)| n == &raw[i].0) {
+                return Err(ArgOpError::DuplicateArg(raw[i].0.clone()));
+            }
+        }
+        for spec in specs {
+            if !raw.iter().any(|(n, _)| n.as_str() == spec.name) {
+                return Err(ArgOpError::MissingArg(spec.name.to_string()));
+            }
+        }
+
+        let mut values = Vec::with_capacity(specs.len());
+        let mut accepted = Vec::with_capacity(specs.len());
+        for spec in specs {
+            let raw_val = &raw.iter().find(|(n, _)| n.as_str() == spec.name).expect("presence checked above").1;
+            let value = spec.kind.parse(raw_val).map_err(|()| ArgOpError::BadValue {
+                arg: spec.name.to_string(),
+                kind: spec.kind,
+                got: raw_val.clone(),
+            })?;
+            values.push((spec.name.to_string(), value));
+            accepted.push(ConstructionArg { name: spec.name.to_string(), value: raw_val.clone() });
+        }
+
+        let mut built = make(&values);
+        built.instance_name = self.instance_name;
+        built.args = ArgsState::Configured { values: accepted };
+        Ok(built)
+    }
+
+    /// The declared construction-arg specs (a view into the pending recipe);
+    /// empty unless this builder `is_pending()`.
+    pub fn arg_specs(&self) -> &[ArgSpec] {
+        match &self.args {
+            ArgsState::Pending { specs, .. } => specs,
+            _ => &[],
+        }
+    }
+
+    /// The accepted, verbatim construction-arg pairs (the serialize surface);
+    /// empty unless this builder was configured via `try_args`.
+    pub fn construction_args(&self) -> &[ConstructionArg] {
+        match &self.args {
+            ArgsState::Configured { values } => values,
+            _ => &[],
+        }
+    }
+
+    /// True for an arg-bearing recipe not yet configured via `try_args`.
+    pub fn is_pending(&self) -> bool {
+        matches!(self.args, ArgsState::Pending { .. })
     }
     /// Set this node instance's explicit name. Must be non-empty: the name forms
     /// a knob-address segment (`<composite>.<name>.<param>`, via `node_name()` in
@@ -344,6 +446,135 @@ pub enum BindOpError {
     AlreadyGanged { param: String, gang: String },
 }
 
+/// Closed construction-arg kinds (spec §Closedness) — deliberately NOT
+/// `ScalarKind`: the four scalar kinds are the streamed set (invariant 4); args
+/// are bootstrap metadata, never streamed, so the two closedness axes stay
+/// separate types.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ArgKind {
+    /// An IANA timezone name (`chrono_tz::Tz`'s own parser — exact, case-sensitive).
+    Tz,
+    /// A local wall-clock time, strict zero-padded `HH:MM` (00-23 / 00-59).
+    TimeOfDay,
+    /// A plain positive decimal count (>= 1, no leading zeros).
+    Count,
+}
+
+impl ArgKind {
+    /// Parse `raw` in this kind's strict canonical form. The accepted form IS
+    /// the canonical form (spec §Closedness) — there is no normalization
+    /// layer, so a near-miss string refuses rather than being silently
+    /// rewritten (content ids stay input-variance-free by refusal).
+    // The spec's pinned shape (`Result<ArgValue, ()>`, mirroring `ArgOpError`'s
+    // caller-side `BadValue` wrapping the plain refusal): the unit error carries
+    // no information of its own — `try_args` is the only caller and always maps
+    // it to `ArgOpError::BadValue`, which is where the real detail lives.
+    #[allow(clippy::result_unit_err)]
+    pub fn parse(&self, raw: &str) -> Result<ArgValue, ()> {
+        match self {
+            ArgKind::Tz => raw.parse::<chrono_tz::Tz>().map(ArgValue::Tz).map_err(|_| ()),
+            ArgKind::TimeOfDay => {
+                let bytes = raw.as_bytes();
+                if bytes.len() != 5 || bytes[2] != b':' {
+                    return Err(());
+                }
+                let (h, m) = (&raw[0..2], &raw[3..5]);
+                if !h.bytes().all(|b| b.is_ascii_digit()) || !m.bytes().all(|b| b.is_ascii_digit()) {
+                    return Err(());
+                }
+                let hour: u32 = h.parse().map_err(|_| ())?;
+                let minute: u32 = m.parse().map_err(|_| ())?;
+                if hour > 23 || minute > 59 {
+                    return Err(());
+                }
+                Ok(ArgValue::TimeOfDay { hour, minute })
+            }
+            ArgKind::Count => {
+                if raw.is_empty() || !raw.bytes().all(|b| b.is_ascii_digit()) {
+                    return Err(());
+                }
+                if raw.len() > 1 && raw.starts_with('0') {
+                    return Err(()); // leading zero: not the canonical form
+                }
+                let n: usize = raw.parse().map_err(|_| ())?;
+                if n == 0 {
+                    return Err(()); // Count is >= 1
+                }
+                Ok(ArgValue::Count(n))
+            }
+        }
+    }
+
+    /// The static, per-kind hint line (introspection surface, C29) — a closed
+    /// table keyed only by `ArgKind`, never per-entry freetext.
+    pub fn hint(&self) -> &'static str {
+        match self {
+            ArgKind::Tz => "IANA timezone name, e.g. Europe/Berlin",
+            ArgKind::TimeOfDay => "local wall-clock HH:MM",
+            ArgKind::Count => "positive integer count",
+        }
+    }
+}
+
+/// One declared construction arg of an arg-bearing `PrimitiveBuilder` (the
+/// pending recipe's twin of `ParamSpec`): its render name and closed `ArgKind`.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct ArgSpec {
+    pub name: &'static str,
+    pub kind: ArgKind,
+}
+
+/// A parsed, validated construction-arg value (in-memory only; documents carry
+/// the canonical string form in [`ConstructionArg`]).
+#[derive(Clone, Debug, PartialEq)]
+pub enum ArgValue {
+    Tz(chrono_tz::Tz),
+    TimeOfDay { hour: u32, minute: u32 },
+    Count(usize),
+}
+
+/// A consumed construction arg — the id-bearing, serialized twin of
+/// `BoundParam`. `value` is the accepted strict-form string, stored verbatim
+/// (never re-normalized — see `ArgKind::parse`).
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct ConstructionArg {
+    pub name: String,
+    pub value: String,
+}
+
+/// A fallible-`try_args` fault — the construction-arg twin of `BindOpError`,
+/// for the data-level construction surface.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum ArgOpError {
+    /// Args were given, but this type declares none (a `Plain` builder).
+    NotArgBearing,
+    /// A given arg name is not among the declared `ArgSpec`s.
+    UnknownArg(String),
+    /// The same arg name was given more than once.
+    DuplicateArg(String),
+    /// A declared arg has no value in the given set.
+    MissingArg(String),
+    /// A given value failed its declared kind's strict-form parse.
+    BadValue { arg: String, kind: ArgKind, got: String },
+}
+
+/// The args-channel state of a [`PrimitiveBuilder`] (private: callers only
+/// ever observe it through `arg_specs`/`construction_args`/`is_pending`).
+enum ArgsState {
+    /// No construction args declared (every existing node, unaffected).
+    Plain,
+    /// An arg-bearing recipe: declared `ArgSpec`s and the `make` that turns a
+    /// parsed value set into the real, configured builder.
+    Pending {
+        specs: &'static [ArgSpec],
+        #[allow(clippy::type_complexity)]
+        make: fn(&[(String, ArgValue)]) -> PrimitiveBuilder,
+    },
+    /// A configured arg-bearing builder: the accepted, verbatim (name, value)
+    /// pairs — the render/serialize surface (`construction_args`).
+    Configured { values: Vec<ConstructionArg> },
+}
+
 /// A node's declared interface: its inputs (in order) and its output record — an
 /// ordered list of named base columns; length 1 is a scalar (the degenerate
 /// case), and an **empty** `output` (`vec![]`) declares a **pure consumer**
@@ -844,6 +1075,132 @@ mod tests {
     fn doc_gate_accepts_a_meaning_line() {
         assert_eq!(doc_gate("EMA", "exponential moving average over the input series"), Ok(()));
     }
+
+    // --- construction args (ArgKind/ArgSpec/ArgValue/ConstructionArg/ArgOpError) ---
+
+    const DEMO_ARG_SPECS: &[ArgSpec] =
+        &[ArgSpec { name: "tz", kind: ArgKind::Tz }, ArgSpec { name: "open", kind: ArgKind::TimeOfDay }];
+
+    /// A minimal `make`: the fixture does not need `values` to shape a real
+    /// schema (that's `Session::make`'s job, Task 2) — it only proves
+    /// `try_args` reaches `make` with parsed values and carries the result
+    /// forward as `Configured`.
+    fn demo_make(values: &[(String, ArgValue)]) -> PrimitiveBuilder {
+        let _ = values;
+        PrimitiveBuilder::new(
+            "Demo",
+            NodeSchema { inputs: vec![], output: vec![], params: vec![], doc: "test-only arg-bearing schema" },
+            |_| Box::new(Bare),
+        )
+    }
+
+    fn demo_pending() -> PrimitiveBuilder {
+        PrimitiveBuilder::pending("Demo", "test-only arg-bearing schema", DEMO_ARG_SPECS, demo_make)
+    }
+
+    #[test]
+    fn try_args_not_arg_bearing_on_a_plain_builder_with_args() {
+        let plain = PrimitiveBuilder::new("Bare", NodeSchema::default(), |_| Box::new(Bare));
+        assert_eq!(
+            plain.try_args(&[("x".into(), "1".into())]).err(),
+            Some(ArgOpError::NotArgBearing),
+        );
+    }
+
+    #[test]
+    fn try_args_plain_with_empty_raw_is_ok_unchanged() {
+        let plain = PrimitiveBuilder::new(
+            "Bare",
+            NodeSchema { inputs: vec![], output: vec![], params: vec![], doc: "plain doc" },
+            |_| Box::new(Bare),
+        );
+        let after = plain.try_args(&[]).expect("empty raw against a Plain builder is Ok(self)");
+        assert_eq!(after.schema().doc, "plain doc");
+        assert!(after.construction_args().is_empty());
+    }
+
+    #[test]
+    fn try_args_unknown_arg_names_the_first_unknown() {
+        assert_eq!(
+            demo_pending()
+                .try_args(&[("nope".into(), "x".into()), ("tz".into(), "Europe/Berlin".into())])
+                .err(),
+            Some(ArgOpError::UnknownArg("nope".into())),
+        );
+    }
+
+    #[test]
+    fn try_args_duplicate_arg_names_the_repeated_name() {
+        assert_eq!(
+            demo_pending()
+                .try_args(&[
+                    ("tz".into(), "Europe/Berlin".into()),
+                    ("tz".into(), "Europe/Berlin".into()),
+                    ("open".into(), "09:30".into()),
+                ])
+                .err(),
+            Some(ArgOpError::DuplicateArg("tz".into())),
+        );
+    }
+
+    #[test]
+    fn try_args_missing_arg_names_the_first_missing_in_spec_order() {
+        assert_eq!(
+            demo_pending().try_args(&[("tz".into(), "Europe/Berlin".into())]).err(),
+            Some(ArgOpError::MissingArg("open".into())),
+        );
+    }
+
+    #[test]
+    fn try_args_bad_value_names_arg_kind_and_the_rejected_string() {
+        assert_eq!(
+            demo_pending()
+                .try_args(&[("tz".into(), "berlin".into()), ("open".into(), "09:30".into())])
+                .err(),
+            Some(ArgOpError::BadValue { arg: "tz".into(), kind: ArgKind::Tz, got: "berlin".into() }),
+        );
+    }
+
+    /// Strict form IS the canonical form (spec §Closedness): no normalization
+    /// layer, so a near-miss string refuses rather than being rewritten.
+    #[test]
+    fn arg_kind_parse_strict_form_refusals_and_accepts() {
+        assert!(ArgKind::TimeOfDay.parse("9:30").is_err(), "not zero-padded");
+        assert!(ArgKind::Tz.parse("berlin").is_err(), "not the exact IANA name");
+        assert!(ArgKind::Count.parse("02").is_err(), "leading zero");
+        assert!(ArgKind::TimeOfDay.parse("09:30").is_ok());
+        assert!(ArgKind::Tz.parse("Europe/Berlin").is_ok());
+        assert!(ArgKind::Count.parse("2").is_ok());
+    }
+
+    #[test]
+    fn try_args_accepts_a_valid_set_and_echoes_verbatim_pairs() {
+        let configured = demo_pending()
+            .try_args(&[("tz".into(), "America/New_York".into()), ("open".into(), "09:30".into())])
+            .expect("a full valid set configures");
+        assert!(!configured.is_pending());
+        assert_eq!(
+            configured.construction_args(),
+            &[
+                ConstructionArg { name: "tz".into(), value: "America/New_York".into() },
+                ConstructionArg { name: "open".into(), value: "09:30".into() },
+            ],
+        );
+    }
+
+    #[test]
+    fn pending_arg_specs_are_visible_before_configuration() {
+        let pending = demo_pending();
+        assert!(pending.is_pending());
+        assert_eq!(pending.arg_specs().iter().map(|s| s.name).collect::<Vec<_>>(), ["tz", "open"]);
+        assert!(pending.construction_args().is_empty());
+    }
+
+    #[test]
+    #[should_panic(expected = "unconfigured")]
+    fn pending_build_panics() {
+        let _ = demo_pending().build(&[]);
+    }
 }
 
 #[cfg(test)]