feat(aura-cli): aura run — end-to-end sample-harness CLI

The Walking-skeleton milestone's closing seam: a real `aura run` binary that
bootstraps a built-in sample harness, runs it deterministically, and prints the
cycle-0009 metrics+manifest report as canonical JSON to stdout — the headline
C14 "run a sim, emit structured metrics" move, end-to-end, from a binary for the
first time.

Two deliverables:

- aura-std::Recorder — a reusable recording sink node (the glossary *sink* role,
  C8/C22): a pure consumer (output: vec![]) over kinds.len() input columns,
  holding an mpsc::Sender<(Timestamp, Vec<Scalar>)> as its out-of-graph
  destination. mpsc keeps the engine's purity invariant (C7): no Rc/RefCell.
  Supports all four base scalar kinds; returns None until every column is warm.
  Promotes the shape the #[cfg(test)] fixtures already proved into a shipped,
  reusable block (the fixtures stay as historical snapshots; de-dup is a later
  tidy).

- aura-cli `run` subcommand — synthetic_prices / sample_harness / run_sample /
  main. The sample harness (synthetic source → SMA(2)/SMA(4) → Sub → Exposure →
  SimBroker → two Recorder sinks) is authored in plain Rust over the raw
  Harness::bootstrap(nodes, sources, edges) API (C17/C20) — the open
  experiment-builder DSL thread is deliberately not committed this cycle. main
  hand-parses one subcommand: `run` prints run_sample().to_json() (exit 0),
  anything else prints a one-line usage to stderr (exit 2). aura-cli gains
  aura-std + aura-core path deps; the workspace stays zero-(external-)dependency.

The built-in synthetic stream (7 ticks, rises then reverses) yields a non-trivial
demo trace: equity [0,0,0,0,-0.08,-0.17,-0.13] → total_pips -0.13, max_drawdown
0.17, exposure_sign_flips 1. The run_sample unit test pins the integer flip count
exactly and the two f64 metrics within 1e-9 (the real run's float dust is ~7e-15,
confirming the tolerance); determinism is pinned exactly (two runs → identical
JSON). tests/cli_run.rs drives the built binary for the observable exit/stdout
contract.

manifest.commit is filled from option_env!("AURA_COMMIT") (defaults to
"unknown"); real HEAD capture via build.rs is deferred. Non-goals untouched:
experiment-builder DSL, aura new, Aura.toml schema, the data-server source (#7).

One deviation from the plan's verbatim code: a scoped
#[allow(clippy::type_complexity)] on sample_harness (its (Harness, Receiver,
Receiver) return tuple trips the lint under -D warnings) — consistent with the
identical allow on the build_two_sink_harness fixture in aura-engine. Verified
myself: cargo test --workspace (61 green, 0 red), clippy --all-targets
-D warnings (clean), cargo doc -D warnings (clean), and both smoke runs.

closes #8

crates/aura-cli/Cargo.toml

@@ -10,4 +10,6 @@ name = "aura"
 path = "src/main.rs"
 
 [dependencies]
+aura-core = { path = "../aura-core" }
 aura-engine = { path = "../aura-engine" }
+aura-std = { path = "../aura-std" }

crates/aura-cli/src/main.rs

@@ -1,8 +1,160 @@
 //! `aura` — the programmatic / CLI face of the engine (the surface the LLM and
 //! automation drive: author a node, run a sim/sweep, emit structured metrics).
 //!
-//! Skeleton only: real subcommands arrive with the walking-skeleton milestone.
+//! The walking skeleton's closing seam: `aura run` bootstraps a built-in sample
+//! signal-quality harness (synthetic source → SMA-cross → Exposure → SimBroker →
+//! recording sinks), runs it deterministically (C1), and prints the run's
+//! metrics + manifest (#6) as canonical JSON to stdout (the headline C14 move).
+
+use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_engine::{
+    f64_field, summarize, Edge, Harness, RunManifest, RunReport, SourceSpec, Target,
+};
+use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
+use std::sync::mpsc::{self, Receiver};
+
+/// The built-in synthetic price stream: rises through t=4 then reverses, so the
+/// demo trace carries one exposure sign flip and a real drawdown (C22 populated
+/// trace). Deterministic and fixed (C1).
+fn synthetic_prices() -> Vec<(Timestamp, Scalar)> {
+    [
+        (1_i64, 1.0000_f64),
+        (2, 1.0010),
+        (3, 1.0030),
+        (4, 1.0060),
+        (5, 1.0040),
+        (6, 1.0010),
+        (7, 0.9990),
+    ]
+    .iter()
+    .map(|&(t, p)| (Timestamp(t), Scalar::F64(p)))
+    .collect()
+}
+
+/// Bootstrap the sample signal-quality harness with two recording sinks (equity
+/// tapped on the SimBroker, exposure tapped on the Exposure node). Rust-authored
+/// wiring (C17/C20) over the raw bootstrap API — no builder DSL this cycle. The
+/// price taps both SMAs and the broker's price slot (slot 1); exposure feeds the
+/// broker's slot 0 (slot order is load-bearing — both are f64).
+// The harness-plus-two-drained-sink-receivers tuple has exactly one call site
+// (`run_sample`); a named type would be speculative abstraction this cycle.
+#[allow(clippy::type_complexity)]
+fn sample_harness() -> (
+    Harness,
+    Receiver<(Timestamp, Vec<Scalar>)>,
+    Receiver<(Timestamp, Vec<Scalar>)>,
+) {
+    let (tx_eq, rx_eq) = mpsc::channel();
+    let (tx_ex, rx_ex) = mpsc::channel();
+    let h = Harness::bootstrap(
+        vec![
+            Box::new(Sma::new(2)),                                          // 0 fast SMA
+            Box::new(Sma::new(4)),                                          // 1 slow SMA
+            Box::new(Sub::new()),                                           // 2 spread
+            Box::new(Exposure::new(0.5)),                                   // 3 exposure
+            Box::new(SimBroker::new(0.0001)),                               // 4 sim-optimal broker
+            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
+            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
+        ],
+        vec![SourceSpec {
+            kind: ScalarKind::F64,
+            targets: vec![
+                Target { node: 0, slot: 0 },
+                Target { node: 1, slot: 0 },
+                Target { node: 4, slot: 1 }, // price into the broker's price slot
+            ],
+        }],
+        vec![
+            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
+            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
+            Edge { from: 2, to: 3, slot: 0, from_field: 0 },
+            Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure into broker slot 0
+            Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
+            Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
+        ],
+    )
+    .expect("valid sample signal-quality DAG");
+    (h, rx_eq, rx_ex)
+}
+
+/// Run the sample harness and fold it into a `RunReport` (drain both sinks →
+/// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic
+/// (C1): the same build yields the same report.
+fn run_sample() -> RunReport {
+    let (mut h, rx_eq, rx_ex) = sample_harness();
+    let prices = synthetic_prices();
+    let window = (
+        prices.first().expect("non-empty stream").0,
+        prices.last().expect("non-empty stream").0,
+    );
+    h.run(vec![prices]);
+
+    let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
+    let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
+    let equity = f64_field(&eq_rows, 0);
+    let exposure = f64_field(&ex_rows, 0);
+    let metrics = summarize(&equity, &exposure);
+
+    RunReport {
+        manifest: RunManifest {
+            commit: option_env!("AURA_COMMIT").unwrap_or("unknown").to_string(),
+            params: vec![
+                ("sma_fast".to_string(), 2.0),
+                ("sma_slow".to_string(), 4.0),
+                ("exposure_scale".to_string(), 0.5),
+            ],
+            window,
+            seed: 0,
+            broker: "sim-optimal(pip_size=0.0001)".to_string(),
+        },
+        metrics,
+    }
+}
 
 fn main() {
-    println!("aura: walking-skeleton scaffold — no subcommands yet.");
+    let mut args = std::env::args().skip(1);
+    match args.next().as_deref() {
+        Some("run") => println!("{}", run_sample().to_json()),
+        _ => {
+            eprintln!("aura: usage: aura run");
+            std::process::exit(2);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn run_sample_is_deterministic_and_non_trivial() {
+        let r1 = run_sample();
+        let r2 = run_sample();
+        // C1 determinism: two runs are bit-identical (metrics + rendered JSON).
+        assert_eq!(r1.metrics, r2.metrics);
+        assert_eq!(r1.to_json(), r2.to_json());
+
+        let m = &r1.metrics;
+        // exactly one exposure sign flip in the demo trace (rises then reverses).
+        assert_eq!(m.exposure_sign_flips, 1);
+        // a non-trivial, populated trace: a real drawdown.
+        assert!(m.max_drawdown > 0.0);
+        // hand-computed magnitudes for the chosen stream (float tolerance; the
+        // computation's dust is ~1e-15).
+        assert!(
+            (m.max_drawdown - 0.17).abs() < 1e-9,
+            "max_drawdown = {}",
+            m.max_drawdown
+        );
+        assert!(
+            (m.total_pips - (-0.13)).abs() < 1e-9,
+            "total_pips = {}",
+            m.total_pips
+        );
+
+        // manifest carries the sample's known configuration.
+        let (from, to) = r1.manifest.window;
+        assert_eq!((from.0, to.0), (1, 7));
+        assert_eq!(r1.manifest.commit, "unknown");
+    }
 }

crates/aura-cli/tests/cli_run.rs

@@ -0,0 +1,36 @@
+//! Integration test: drive the built `aura` binary as a downstream user would,
+//! asserting the `run` subcommand's stdout/exit contract and the bad-args path.
+
+use std::process::Command;
+
+/// Path to the freshly-built `aura` binary (Cargo sets this env var for the test
+/// crate; the binary is named `aura` in `Cargo.toml`).
+const BIN: &str = env!("CARGO_BIN_EXE_aura");
+
+#[test]
+fn run_prints_json_and_exits_zero() {
+    let out = Command::new(BIN).arg("run").output().expect("spawn aura run");
+    assert!(out.status.success(), "exit status: {:?}", out.status);
+
+    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
+    // exactly one line (the JSON object + a trailing newline from println!).
+    assert_eq!(stdout.lines().count(), 1, "stdout was: {stdout:?}");
+    let line = stdout.trim_end();
+
+    // canonical cycle-0009 JSON shape: nested manifest + metrics, stable keys.
+    assert!(line.starts_with("{\"manifest\":{\"commit\":\"unknown\","), "got: {line}");
+    assert!(line.contains("\"broker\":\"sim-optimal(pip_size=0.0001)\""), "got: {line}");
+    assert!(line.contains("\"window\":[1,7]"), "got: {line}");
+    assert!(line.contains("\"metrics\":{\"total_pips\":"), "got: {line}");
+    // the integer sign-flip count is stable across float renderings.
+    assert!(line.ends_with("\"exposure_sign_flips\":1}}"), "got: {line}");
+}
+
+#[test]
+fn no_args_prints_usage_and_exits_two() {
+    let out = Command::new(BIN).output().expect("spawn aura");
+    assert_eq!(out.status.code(), Some(2), "exit status: {:?}", out.status);
+    assert!(out.stdout.is_empty(), "stdout should be empty on the usage path");
+    let stderr = String::from_utf8(out.stderr).expect("utf-8 stderr");
+    assert!(stderr.contains("usage"), "stderr was: {stderr:?}");
+}

crates/aura-std/src/lib.rs

@@ -18,12 +18,14 @@
 mod add;
 mod exposure;
 mod lincomb;
+mod recorder;
 mod sim_broker;
 mod sma;
 mod sub;
 pub use add::Add;
 pub use exposure::Exposure;
 pub use lincomb::LinComb;
+pub use recorder::Recorder;
 pub use sim_broker::SimBroker;
 pub use sma::Sma;
 pub use sub::Sub;

crates/aura-std/src/recorder.rs

@@ -0,0 +1,119 @@
+//! `Recorder` — a reusable recording sink (the glossary *sink* role, C8/C22):
+//! a pure consumer that, each fired cycle, sends `(ctx.now(), row)` — the newest
+//! value of each declared input column — to an out-of-graph `mpsc` destination it
+//! holds. It produces nothing (`output: vec![]`), so it is a leaf in the DAG. The
+//! `mpsc::Sender` keeps the engine's purity invariant (C7): the node carries no
+//! `Rc`/`RefCell` interior mutability, only an owned channel handle. Supports all
+//! four base scalar kinds so any column can be persisted; returns `None` (filters)
+//! until every input column is warm.
+
+use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
+use std::sync::mpsc::Sender;
+
+/// A recording sink over `kinds.len()` input columns. Each fired cycle it reads
+/// the newest value of every column and sends the row to `tx`; it returns `None`
+/// (records, forwards nothing) and `None` during warm-up until all columns have a
+/// value.
+pub struct Recorder {
+    kinds: Vec<ScalarKind>,
+    firing: Firing,
+    tx: Sender<(Timestamp, Vec<Scalar>)>,
+}
+
+impl Recorder {
+    /// A recorder over one input column per entry in `kinds`, each with the given
+    /// `firing` policy, sending recorded `(timestamp, row)` pairs to `tx`.
+    pub fn new(kinds: &[ScalarKind], firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        Self { kinds: kinds.to_vec(), firing, tx }
+    }
+}
+
+impl Node for Recorder {
+    fn schema(&self) -> NodeSchema {
+        NodeSchema {
+            inputs: self
+                .kinds
+                .iter()
+                .map(|&kind| InputSpec { kind, lookback: 1, firing: self.firing })
+                .collect(),
+            output: vec![],
+        }
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> {
+        let mut row = Vec::with_capacity(self.kinds.len());
+        for (i, &kind) in self.kinds.iter().enumerate() {
+            // newest of each column by kind; `?` returns None (warm-up) if cold.
+            let scalar = match kind {
+                ScalarKind::F64 => Scalar::F64(ctx.f64_in(i).get(0)?),
+                ScalarKind::I64 => Scalar::I64(ctx.i64_in(i).get(0)?),
+                ScalarKind::Bool => Scalar::Bool(ctx.bool_in(i).get(0)?),
+                ScalarKind::Timestamp => Scalar::Ts(ctx.ts_in(i).get(0)?),
+            };
+            row.push(scalar);
+        }
+        let _ = self.tx.send((ctx.now(), row));
+        None
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Timestamp};
+    use std::sync::mpsc;
+
+    #[test]
+    fn recorder_captures_f64_stream_after_warmup() {
+        let (tx, rx) = mpsc::channel();
+        let mut rec = Recorder::new(&[ScalarKind::F64], Firing::Any, tx);
+
+        // size the one f64 input column from the schema, as the engine would.
+        let schema = rec.schema();
+        assert!(schema.output.is_empty(), "a sink declares no output (C8)");
+        let mut inputs = vec![AnyColumn::with_capacity(
+            schema.inputs[0].kind,
+            schema.inputs[0].lookback,
+        )];
+
+        // cold: returns None and records nothing.
+        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert!(rx.try_recv().is_err());
+
+        // warm: returns None (pure consumer) but records (now, [F64(newest)]).
+        for (t, v) in [(2_i64, 10.0_f64), (3, 20.0), (4, 30.0)] {
+            inputs[0].push(Scalar::F64(v)).unwrap();
+            assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+        assert_eq!(
+            rows,
+            vec![
+                (Timestamp(2), vec![Scalar::F64(10.0)]),
+                (Timestamp(3), vec![Scalar::F64(20.0)]),
+                (Timestamp(4), vec![Scalar::F64(30.0)]),
+            ]
+        );
+    }
+
+    #[test]
+    fn recorder_is_none_until_all_columns_warm() {
+        let (tx, rx) = mpsc::channel();
+        let mut rec = Recorder::new(&[ScalarKind::F64, ScalarKind::F64], Firing::Any, tx);
+        let mut inputs = vec![
+            AnyColumn::with_capacity(ScalarKind::F64, 1),
+            AnyColumn::with_capacity(ScalarKind::F64, 1),
+        ];
+
+        // only column 0 present -> None, nothing recorded.
+        inputs[0].push(Scalar::F64(1.0)).unwrap();
+        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert!(rx.try_recv().is_err());
+
+        // both present -> records the full row (still returns None).
+        inputs[1].push(Scalar::F64(2.0)).unwrap();
+        assert_eq!(rec.eval(Ctx::new(&inputs, Timestamp(2))), None);
+        let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+        assert_eq!(rows, vec![(Timestamp(2), vec![Scalar::F64(1.0), Scalar::F64(2.0)])]);
+    }
+}