From 559903a14ef817f009f90b5250e1181e48de526a Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 4 Jun 2026 20:45:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-cli):=20aura=20run=20=E2=80=94=20end-?= =?UTF-8?q?to-end=20sample-harness=20CLI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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)> 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 --- Cargo.lock | 2 + crates/aura-cli/Cargo.toml | 2 + crates/aura-cli/src/main.rs | 156 ++++++++++++++++++++++++++++++- crates/aura-cli/tests/cli_run.rs | 36 +++++++ crates/aura-std/src/lib.rs | 2 + crates/aura-std/src/recorder.rs | 119 +++++++++++++++++++++++ 6 files changed, 315 insertions(+), 2 deletions(-) create mode 100644 crates/aura-cli/tests/cli_run.rs create mode 100644 crates/aura-std/src/recorder.rs diff --git a/Cargo.lock b/Cargo.lock index 2817d73..be6ae5c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6,7 +6,9 @@ version = 4 name = "aura-cli" version = "0.1.0" dependencies = [ + "aura-core", "aura-engine", + "aura-std", ] [[package]] diff --git a/crates/aura-cli/Cargo.toml b/crates/aura-cli/Cargo.toml index 344574b..a5e4aae 100644 --- a/crates/aura-cli/Cargo.toml +++ b/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" } diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index f834da7..1a78f8f 100644 --- a/crates/aura-cli/src/main.rs +++ b/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)>, + Receiver<(Timestamp, Vec)>, +) { + 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)> = rx_eq.try_iter().collect(); + let ex_rows: Vec<(Timestamp, Vec)> = 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"); + } } diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs new file mode 100644 index 0000000..7441788 --- /dev/null +++ b/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:?}"); +} diff --git a/crates/aura-std/src/lib.rs b/crates/aura-std/src/lib.rs index d832bc0..8d1fdf1 100644 --- a/crates/aura-std/src/lib.rs +++ b/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; diff --git a/crates/aura-std/src/recorder.rs b/crates/aura-std/src/recorder.rs new file mode 100644 index 0000000..7610a64 --- /dev/null +++ b/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, + firing: Firing, + tx: Sender<(Timestamp, Vec)>, +} + +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)>) -> 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)> = 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)> = rx.try_iter().collect(); + assert_eq!(rows, vec![(Timestamp(2), vec![Scalar::F64(1.0), Scalar::F64(2.0)])]); + } +}