From 0bf15cb06990d537c64761bbc6d575c4189e15b4 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 4 Jun 2026 15:36:50 +0200 Subject: [PATCH] =?UTF-8?q?fieldtest:=20cycle-0006=20sink=20recording=20?= =?UTF-8?q?=E2=80=94=205=20examples,=206=20findings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit First fieldtest of the project. A standalone downstream-consumer crate (fieldtests/cycle-0006-substrate/) path-depends on the engine crates and exercises the post-0006 substrate from the public interface only (rustdoc + specs + ledger, never crates/ source): 1 custom recording node via the Node contract + Ctx::now() 2 fan-out/fan-in DAG, 3-arg bootstrap, run() -> () 3 two interior streams recorded from one run (the cycle headline) 4 byte-identical recorded output across two runs (C1 determinism) 5 mis-wired recording edge rejected (KindMismatch) at bootstrap Findings: 3 working (carry-on), 2 friction, 1 spec_gap. - friction: a nested standalone consumer crate fights the workspace resolver (needs an empty [workspace] table — Cargo's own hint). - friction: recorder boilerplate is hand-rewritten per author (C9 deliberately ships no library sink). - spec_gap: the public surface never states output: vec![] is THE sink declaration, nor defines a Some return paired with empty output. Spec at docs/specs/fieldtest-0006-substrate.md feeds the 0007 plan. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/specs/fieldtest-0006-substrate.md | 155 ++++++++++++++++++ fieldtests/cycle-0006-substrate/.gitignore | 2 + fieldtests/cycle-0006-substrate/Cargo.toml | 43 +++++ .../c0006_1_custom_node_now.rs | 100 +++++++++++ .../c0006_2_fanout_dag.rs | 108 ++++++++++++ .../c0006_3_multi_sink_record.rs | 152 +++++++++++++++++ .../c0006_4_determinism.rs | 105 ++++++++++++ .../c0006_5_reject_mismatch.rs | 92 +++++++++++ 8 files changed, 757 insertions(+) create mode 100644 docs/specs/fieldtest-0006-substrate.md create mode 100644 fieldtests/cycle-0006-substrate/.gitignore create mode 100644 fieldtests/cycle-0006-substrate/Cargo.toml create mode 100644 fieldtests/cycle-0006-substrate/c0006_1_custom_node_now.rs create mode 100644 fieldtests/cycle-0006-substrate/c0006_2_fanout_dag.rs create mode 100644 fieldtests/cycle-0006-substrate/c0006_3_multi_sink_record.rs create mode 100644 fieldtests/cycle-0006-substrate/c0006_4_determinism.rs create mode 100644 fieldtests/cycle-0006-substrate/c0006_5_reject_mismatch.rs diff --git a/docs/specs/fieldtest-0006-substrate.md b/docs/specs/fieldtest-0006-substrate.md new file mode 100644 index 0000000..f5f914d --- /dev/null +++ b/docs/specs/fieldtest-0006-substrate.md @@ -0,0 +1,155 @@ +# Fieldtest — cycle-0006 (sink recording) — 2026-06-04 + +**Status:** Draft — awaiting orchestrator triage +**Author:** fieldtester (dispatched by fieldtest skill) + +## Scope + +Cycle 0006 settled "a sink is a role, not a type": recording is done by an +ordinary node in its `eval`, pushing a record to a destination it holds as a +field (channel / buffer / chart handle) — an out-of-graph side effect. Three +substrate changes shipped: (1) `Ctx::now() -> Timestamp` (a `Copy` accessor so a +recording node can causally stamp each record with the cycle timestamp); (2) the +old single `observe: usize` recording affordance was removed — `Harness::bootstrap` +dropped its 4th parameter and `Harness::run` returns `()` (retention is owned by +whoever holds the recording node's destination's read end); (3) the run loop is +otherwise an unchanged topological router. No `Sink` type, no `SinkHandler` trait, +no engine sink registry. The fieldtest exercised this from a standalone downstream +consumer crate (`fieldtests/cycle-0006-substrate/`) that `path`-depends on the +engine crates and uses only the public surface (rustdoc + specs + ledger). + +## Examples + +### fieldtests/cycle-0006-substrate/c0006_1_custom_node_now.rs — custom recording node via the Node contract +- Authors a `Recorder` node from scratch: `schema()` declares one f64 input + + `output: vec![]` (pure consumer); `eval` reads the typed window, calls + `ctx.now()`, sends `(now, value)` out of graph, returns `None`. Wired as + `source -> SMA(2) -> Recorder`. +- Fits the "author a custom node incl. `Ctx::now()`" axis. +- Outcome: built, ran, matched expected `[(20,11.0),(30,13.0),(40,15.0)]` on first try. + +### fieldtests/cycle-0006-substrate/c0006_2_fanout_dag.rs — multi-stage fan-out / fan-in DAG, bootstrapped + run +- `source -> {SMA(2), SMA(4)} -> Sub(fast-slow) -> Recorder`. Uses the post-0006 + 3-arg `bootstrap` (no `observe`) and `run() -> ()`; retention via the channel. +- Fits the "fan-out/fan-in DAG" and "bootstrap arity / run returns ()" axes. +- Outcome: built, ran, matched `[(4,2.0),(5,2.0),(6,2.0)]` on first try. + +### fieldtests/cycle-0006-substrate/c0006_3_multi_sink_record.rs — many interior streams, one run (headline) +- Two independent `Recorder` nodes tap SMA(2) and SMA(4) in one harness, draining + to two channels into `Vec<(Timestamp, Vec)>` (the spec's stated drain + shape). Asserts both streams correct AND of different length (sparse, timestamped, + no shared cycle index). +- Fits the "record multiple distinct interior streams of one run" axis (the cycle + headline; impossible before 0006). +- Outcome: built, ran, both streams matched (4 vs 2 records) on first try. + +### fieldtests/cycle-0006-substrate/c0006_4_determinism.rs — byte-identical recorded output across two runs +- Two fresh harnesses, identical input through `{SMA(2),SMA(4)} -> Sub -> Recorder`; + compares recorded `f64::to_bits()` and timestamps for true byte-identity. +- Fits the "verify determinism (C1)" axis. +- Outcome: built, ran, byte-identical on first try. + +### fieldtests/cycle-0006-substrate/c0006_5_reject_mismatch.rs — mis-wired recording edge rejected +- An i64 producer field bound into a recorder's f64 slot; asserts bootstrap returns + `BootstrapError::KindMismatch { producer: I64, consumer: F64 }` before any data + flows. Correct behaviour here is rejection. +- Fits the recording axis from the failure side (recording adds no new wiring hole). +- Outcome: built, ran, rejected with the documented error on first try. + +## Findings + +### [working] Custom recording node + `Ctx::now()` is natural and correct +- Examples: c0006_1, c0006_3, c0006_5. +- What happened: implementing `Node` for a user type, declaring `output: vec![]`, + reading `ctx.f64_in(0)`, calling `ctx.now()`, performing an mpsc side effect, and + returning `None` all worked verbatim from the rustdoc signatures and the 0006 + spec's worked example. `Ctx::now()` returned the present cycle's timestamp on every + fired cycle; recorded streams were sparse + timestamped exactly as documented. +- Why working: the new surface was reached for, used as designed, and correct on the + first run with no surprises. +- Recommended action: carry-on. + +### [working] The cycle headline — many interior streams from one run +- Example: c0006_3. +- What happened: two recorders tapped two different-rate indicators in one harness; + drained streams were individually correct and independently lengthed (4 vs 2). The + thing the cycle exists to enable (more than one recorded stream per run) works. +- Why working: this is the acceptance evidence the spec promised, reproduced + empirically by a downstream consumer. +- Recommended action: carry-on. + +### [working] Post-0006 `Harness` API (3-arg bootstrap, `run() -> ()`) + determinism + rejection +- Examples: c0006_2 (3-arg bootstrap, `run` returns `()`), c0006_4 (byte-identical + re-run), c0006_5 (`KindMismatch` rejection). +- What happened: the shrunk surface is exactly as the rustdoc/ledger state; a + consumer holding the channel's read end owns retention with no engine help. + Determinism held bit-exactly; a mis-wired recorder edge was rejected at bootstrap. +- Why working: the substrate shrink introduced no regression and no forbidden + failure class (C1/C7 hold from the consumer's vantage). +- Recommended action: carry-on. + +### [friction] Standalone consumer crate fights the workspace resolver +- Example: all (the fixture crate's `Cargo.toml`). +- What happened, verbatim: + `error: current package believes it's in a workspace when it's not: ... this may + be fixable by adding ... Alternatively, to keep it out of the workspace, add ... + an empty [workspace] table to the package's manifest.` +- Resolution taken: added an empty `[workspace]` table to the fixture manifest (the + keep-it-out option Cargo itself lists; no edit to the engine root manifest). After + that, all five binaries built and ran. +- Why friction: a real downstream research project (C16: "a project is always a Rust + crate" depending on aura) created *inside* or beside this repo hits the same wall. + The task completed, but only after a non-obvious one-liner. This is the + resolver-fight the carrier explicitly asked be recorded. +- Recommended action: plan — note in onboarding/docs (or the future `aura new` + scaffolder, an open ledger thread) that a project crate needs its own `[workspace]` + root when nested under the engine repo. + +### [friction] Recorder boilerplate is hand-rewritten per author +- Examples: all five (each redeclares a near-identical `Recorder`: a `tx` field, a + `schema` with `output: vec![]`, an `eval` that reads inputs by kind, sends, returns + `None`). +- What happened: the cycle deliberately ships no `Recorder`/sink library type (C9 — + no speculative surface; the spec calls `Recorder` "this cycle's test-local + fixture, not a shipped type"). So every consumer writes the channel-sink dance, + including the per-kind `match` to pull the newest value of each input. +- Why friction: it is redundant across authors and the 0006 spec implies a real + author writes "their ChartSink, their RegistrySink" each from scratch. The cycle's + scope intentionally excludes a library sink, so this is expected friction, not a + bug — but it is the natural candidate for the next tidy. +- Recommended action: plan — consider an `aura-std` channel/buffer recording block + (or a small `Recorder` helper) once a second consumer confirms the shape, so the + boilerplate is written once. Not urgent; C9 correctly kept it out of this cycle. + +### [spec_gap] No public statement of the canonical sink declaration / `Some`-with-empty-output +- Examples: c0006_1, c0006_3, c0006_5 (all declare `output: vec![]`). +- What happened: I needed the canonical way to declare a no-output (pure consumer) + node. The public surface (rustdoc for `NodeSchema`/`Node`; the 0006 spec) shows + `output: vec![]` only implicitly inside the worked example's prose ("a node with + `output: vec![]`"); neither the rustdoc nor the ledger C8 entry states outright + that an empty `output` vec is *the* sink declaration, nor what the engine does if a + node with `output: vec![]` nonetheless returns `Some(row)` (ignored? debug-asserts + on a width mismatch? UB?). I picked `output: vec![]` + return `None` — it matched + the worked example and ran correctly — but the "what if a sink returns `Some`" + corner is unconstrained by the public surface, and I did not probe it (probing the + engine's reaction would require reading implementation source, which ends the + test). +- Why spec_gap: the design surface is silent on a corner a downstream author can hit + by mistake; the natural reading worked but another (a sink that records *and* + forwards via `output: vec![]` + `Some`) is left undefined publicly. +- Recommended action: tighten the design ledger / rustdoc — state explicitly that + `output: vec![]` is the pure-consumer declaration and define the contract for a + `Some` return paired with an empty `output` (reject at bootstrap, debug-assert, or + documented-ignore). The C8 "both" case is documented for a node with a real output + port; the empty-output-plus-`Some` mistake is not. + +## Recommendation summary + +| Finding | Action | +|---|---| +| [working] custom node + `Ctx::now()` | carry-on | +| [working] many interior streams, one run (headline) | carry-on | +| [working] 3-arg bootstrap / `run()->()` / determinism / rejection | carry-on | +| [friction] consumer crate fights workspace resolver | plan | +| [friction] recorder boilerplate rewritten per author | plan | +| [spec_gap] canonical sink declaration / `Some`-with-empty-output undefined | tighten the design ledger | diff --git a/fieldtests/cycle-0006-substrate/.gitignore b/fieldtests/cycle-0006-substrate/.gitignore new file mode 100644 index 0000000..96ef6c0 --- /dev/null +++ b/fieldtests/cycle-0006-substrate/.gitignore @@ -0,0 +1,2 @@ +/target +Cargo.lock diff --git a/fieldtests/cycle-0006-substrate/Cargo.toml b/fieldtests/cycle-0006-substrate/Cargo.toml new file mode 100644 index 0000000..54a59fc --- /dev/null +++ b/fieldtests/cycle-0006-substrate/Cargo.toml @@ -0,0 +1,43 @@ +# Standalone downstream-consumer crate for the cycle-0006 fieldtest. +# +# It is NOT a member of the aura workspace — it path-depends on the engine crates +# exactly as a real research project (C16) would, and is built via +# `cargo run --manifest-path fieldtests/cycle-0006-substrate/Cargo.toml --bin ` +# so HEAD source is always what runs. +# Empty [workspace] table: marks this fixture crate as its OWN workspace root, so +# it stays out of the engine workspace without editing the engine's root manifest. +# (Cargo's own error message lists this as the keep-it-out-of-the-workspace fix.) +# Recorded as a fieldtest finding: a standalone path-dep consumer crate nested +# under a workspace repo needs this one line. +[workspace] + +[package] +name = "c0006-fieldtest" +version = "0.0.0" +edition = "2024" +publish = false + +[dependencies] +aura-core = { path = "../../crates/aura-core" } +aura-engine = { path = "../../crates/aura-engine" } +aura-std = { path = "../../crates/aura-std" } + +[[bin]] +name = "c0006_1_custom_node_now" +path = "c0006_1_custom_node_now.rs" + +[[bin]] +name = "c0006_2_fanout_dag" +path = "c0006_2_fanout_dag.rs" + +[[bin]] +name = "c0006_3_multi_sink_record" +path = "c0006_3_multi_sink_record.rs" + +[[bin]] +name = "c0006_4_determinism" +path = "c0006_4_determinism.rs" + +[[bin]] +name = "c0006_5_reject_mismatch" +path = "c0006_5_reject_mismatch.rs" diff --git a/fieldtests/cycle-0006-substrate/c0006_1_custom_node_now.rs b/fieldtests/cycle-0006-substrate/c0006_1_custom_node_now.rs new file mode 100644 index 0000000..45675ee --- /dev/null +++ b/fieldtests/cycle-0006-substrate/c0006_1_custom_node_now.rs @@ -0,0 +1,100 @@ +//! Fieldtest c0006 #1 — author a custom recording node from scratch. +//! +//! Axis: "Author a custom node via the public Node contract (schema / eval / +//! Ctx, including Ctx::now())." +//! +//! A downstream author writes their own sink node — here a `Recorder` that holds +//! the destination (an mpsc::Sender) as a field and, in eval, stamps each fired +//! cycle's value with `ctx.now()` and pushes `(now, value)` out of the graph. +//! It returns `None` (pure consumer, C8). This is the canonical recording-node +//! shape the 0006 spec's "Concrete code shapes" section describes (`Recorder` is +//! the spec's test-local fixture; a real author writes their ChartSink the same +//! way). +//! +//! What this fixture proves a downstream consumer can do with ONLY the public +//! surface: implement `Node` for their own type, read a typed input window, call +//! `ctx.now()`, perform an out-of-graph side effect, and return `None`. + +use std::sync::mpsc::{self, Sender}; + +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Harness, SourceSpec, Target}; +use aura_std::Sma; + +/// A pure-consumer recording node: one f64 input, no output, pushes +/// `(ctx.now(), value)` to a channel the caller drains. +struct Recorder { + tx: Sender<(Timestamp, f64)>, +} + +impl Recorder { + fn new(tx: Sender<(Timestamp, f64)>) -> Self { + Self { tx } + } +} + +impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { + kind: ScalarKind::F64, + lookback: 1, + firing: Firing::Any, + }], + // Pure consumer (sink): no output port. + output: vec![], + } + } + + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.f64_in(0); + if w.is_empty() { + return None; // not yet warmed + } + // The C2-causal stamp: the present cycle's timestamp, never the future. + let _ = self.tx.send((ctx.now(), w[0])); + None // pure sink — nothing forwarded into the graph + } +} + +fn f64_stream(pairs: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { + pairs + .iter() + .map(|&(t, v)| (Timestamp(t), Scalar::F64(v))) + .collect() +} + +fn main() { + let (tx, rx) = mpsc::channel(); + + // source -> SMA(2) -> Recorder. Recorder stamps each fired SMA value with now(). + let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), // node 0 + Box::new(Recorder::new(tx)), // node 1 + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }], + }], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .expect("valid DAG"); + + h.run(vec![f64_stream(&[(10, 10.0), (20, 12.0), (30, 14.0), (40, 16.0)])]); + + let recorded: Vec<(Timestamp, f64)> = rx.try_iter().collect(); + + // SMA(2) warms up after 2 samples, so it fires at t=20,30,40 with means + // (10+12)/2=11, (12+14)/2=13, (14+16)/2=15. Each carries the cycle's now(). + let expected = vec![ + (Timestamp(20), 11.0), + (Timestamp(30), 13.0), + (Timestamp(40), 15.0), + ]; + + println!("recorded = {recorded:?}"); + println!("expected = {expected:?}"); + assert_eq!(recorded, expected, "Recorder stream (timestamp via Ctx::now)"); + println!("c0006_1 OK: custom node + Ctx::now() recorded a now-stamped stream"); +} diff --git a/fieldtests/cycle-0006-substrate/c0006_2_fanout_dag.rs b/fieldtests/cycle-0006-substrate/c0006_2_fanout_dag.rs new file mode 100644 index 0000000..58af630 --- /dev/null +++ b/fieldtests/cycle-0006-substrate/c0006_2_fanout_dag.rs @@ -0,0 +1,108 @@ +//! Fieldtest c0006 #2 — multi-stage fan-out / fan-in DAG, bootstrapped + run. +//! +//! Axes: "Build a multi-stage fan-out / fan-in DAG (one source fanning into +//! SMA(2) and SMA(4), then a combiner)" and "Bootstrap and run a Harness (note +//! the new bootstrap arity and that run returns ())." +//! +//! source --> SMA(2) --\ +//! \ Sub(fast - slow) --> Recorder +//! \-> SMA(4) --/ +//! +//! A downstream author wires the classic spread DAG and records the combiner's +//! output. This fixture leans on the post-0006 surface: bootstrap takes exactly +//! THREE positional args (no observe), and run returns () — retention is the +//! recorder's channel, owned by the caller. + +use std::sync::mpsc::{self, Sender}; + +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Harness, SourceSpec, Target}; +use aura_std::{Sma, Sub}; + +struct Recorder { + tx: Sender<(Timestamp, f64)>, +} +impl Recorder { + fn new(tx: Sender<(Timestamp, f64)>) -> Self { + Self { tx } + } +} +impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { + kind: ScalarKind::F64, + lookback: 1, + firing: Firing::Any, + }], + output: vec![], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.f64_in(0); + if w.is_empty() { + return None; + } + let _ = self.tx.send((ctx.now(), w[0])); + None + } +} + +fn f64_stream(pairs: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { + pairs + .iter() + .map(|&(t, v)| (Timestamp(t), Scalar::F64(v))) + .collect() +} + +fn main() { + let (tx, rx) = mpsc::channel(); + + // nodes: 0 = SMA(2), 1 = SMA(4), 2 = Sub(0 - 1), 3 = Recorder taps Sub. + let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(Sub::new()), + Box::new(Recorder::new(tx)), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + // source fans out into both SMAs + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0 + Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // SMA(4) -> Sub.in1 + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // Sub -> Recorder + ], + ) + .expect("valid DAG"); + + // ascending-in-price stream: SMA(2) leads SMA(4), spread stays constant once warm. + h.run(vec![f64_stream(&[ + (1, 10.0), + (2, 12.0), + (3, 14.0), + (4, 16.0), + (5, 18.0), + (6, 20.0), + ])]); + + let recorded: Vec<(Timestamp, f64)> = rx.try_iter().collect(); + + // Sub fires once both SMAs are warm. SMA(4) warms at the 4th sample (t=4). + // t=4: SMA2=mean(16,14)=15, SMA4=mean(16,14,12,10)=13 -> 2 + // t=5: SMA2=mean(18,16)=17, SMA4=mean(18,16,14,12)=15 -> 2 + // t=6: SMA2=mean(20,18)=19, SMA4=mean(20,18,16,14)=17 -> 2 + let expected = vec![ + (Timestamp(4), 2.0), + (Timestamp(5), 2.0), + (Timestamp(6), 2.0), + ]; + + println!("recorded = {recorded:?}"); + println!("expected = {expected:?}"); + assert_eq!(recorded, expected, "fan-out/fan-in spread DAG recorded output"); + println!("c0006_2 OK: fan-out/fan-in DAG bootstrapped (3-arg) + run () + recorded"); +} diff --git a/fieldtests/cycle-0006-substrate/c0006_3_multi_sink_record.rs b/fieldtests/cycle-0006-substrate/c0006_3_multi_sink_record.rs new file mode 100644 index 0000000..91b7692 --- /dev/null +++ b/fieldtests/cycle-0006-substrate/c0006_3_multi_sink_record.rs @@ -0,0 +1,152 @@ +//! Fieldtest c0006 #3 — record MANY distinct interior streams of ONE run. +//! +//! Axis: "Record multiple distinct interior streams of one run via +//! recording-nodes (sink-as-role)." +//! +//! This is the cycle headline: impossible before 0006 (one `observe` index = one +//! recorded row). Two independent Recorder nodes tap two interior indicators — +//! SMA(2) and SMA(4) — in the SAME harness, each draining to its own channel. +//! Mirrors the 0006 spec's worked "User-facing program" example. +//! +//! It also records `(Timestamp, Vec)` rows (the exact type the spec's +//! worked example drains into: `Vec<(Timestamp, Vec)>`) to fieldtest +//! that the spec's stated drain shape is reachable. + +use std::sync::mpsc::{self, Sender}; + +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Harness, SourceSpec, Target}; +use aura_std::Sma; + +/// Recorder declared over arbitrary input kinds; records the full input row +/// (one value per declared slot) tagged with now(). Matches the spec's +/// `Recorder::new(&[ScalarKind], tx)` shape and `(Timestamp, Vec)` drain. +struct Recorder { + kinds: Vec, + tx: Sender<(Timestamp, Vec)>, +} +impl Recorder { + fn new(kinds: &[ScalarKind], tx: Sender<(Timestamp, Vec)>) -> Self { + Self { + kinds: kinds.to_vec(), + tx, + } + } +} +impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: self + .kinds + .iter() + .map(|&kind| InputSpec { + kind, + lookback: 1, + firing: Firing::Any, + }) + .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 input, by declared kind + match kind { + ScalarKind::F64 => { + let w = ctx.f64_in(i); + if w.is_empty() { + return None; + } + row.push(Scalar::F64(w[0])); + } + ScalarKind::I64 => { + let w = ctx.i64_in(i); + if w.is_empty() { + return None; + } + row.push(Scalar::I64(w[0])); + } + ScalarKind::Bool => { + let w = ctx.bool_in(i); + if w.is_empty() { + return None; + } + row.push(Scalar::Bool(w[0])); + } + ScalarKind::Timestamp => { + let w = ctx.ts_in(i); + if w.is_empty() { + return None; + } + row.push(Scalar::Ts(w[0])); + } + } + } + let _ = self.tx.send((ctx.now(), row)); + None + } +} + +fn f64_stream(pairs: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { + pairs + .iter() + .map(|&(t, v)| (Timestamp(t), Scalar::F64(v))) + .collect() +} + +fn main() { + let (tx_fast, rx_fast) = mpsc::channel(); + let (tx_slow, rx_slow) = mpsc::channel(); + + let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), // 0 + Box::new(Sma::new(4)), // 1 + Box::new(Recorder::new(&[ScalarKind::F64], tx_fast)), // 2: taps node 0 + Box::new(Recorder::new(&[ScalarKind::F64], tx_slow)), // 3: taps node 1 + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> recorder fast + Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow + ], + ) + .expect("valid DAG"); + + h.run(vec![f64_stream(&[ + (1, 10.0), + (2, 12.0), + (3, 14.0), + (4, 16.0), + (5, 18.0), + ])]); + + let fast: Vec<(Timestamp, Vec)> = rx_fast.try_iter().collect(); + let slow: Vec<(Timestamp, Vec)> = rx_slow.try_iter().collect(); + + // SMA(2): warm at t=2 -> means 11,13,15,17 at t=2,3,4,5. + let fast_expected = vec![ + (Timestamp(2), vec![Scalar::F64(11.0)]), + (Timestamp(3), vec![Scalar::F64(13.0)]), + (Timestamp(4), vec![Scalar::F64(15.0)]), + (Timestamp(5), vec![Scalar::F64(17.0)]), + ]; + // SMA(4): warm at t=4 -> means 13,15 at t=4,5. + let slow_expected = vec![ + (Timestamp(4), vec![Scalar::F64(13.0)]), + (Timestamp(5), vec![Scalar::F64(15.0)]), + ]; + + println!("fast = {fast:?}"); + println!("slow = {slow:?}"); + assert_eq!(fast, fast_expected, "SMA(2) interior stream"); + assert_eq!(slow, slow_expected, "SMA(4) interior stream"); + // The headline: two DIFFERENT-RATE streams recorded from ONE run, no shared + // cycle index — each sparse and timestamped (4 vs 2 records). + assert_ne!(fast.len(), slow.len(), "different-rate recorders are independent"); + println!("c0006_3 OK: two interior streams recorded from one run (sink-as-role)"); +} diff --git a/fieldtests/cycle-0006-substrate/c0006_4_determinism.rs b/fieldtests/cycle-0006-substrate/c0006_4_determinism.rs new file mode 100644 index 0000000..1f7b288 --- /dev/null +++ b/fieldtests/cycle-0006-substrate/c0006_4_determinism.rs @@ -0,0 +1,105 @@ +//! Fieldtest c0006 #4 — recorded output is byte-identical across two runs (C1). +//! +//! Axis: "Then verify determinism: same input => byte-identical recorded output +//! across two runs." +//! +//! Two fresh harnesses (each its own channel), identical input. Drain both +//! recorded streams and assert bit-identical. A recorded side effect cannot feed +//! back into the graph, so determinism must hold (the cycle's own acceptance +//! claim). To make "byte-identical" concrete we compare the bit patterns of the +//! recorded f64s, not just `==` (which treats NaN specially), and the timestamps. + +use std::sync::mpsc::{self, Sender}; + +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Harness, SourceSpec, Target}; +use aura_std::{Sma, Sub}; + +struct Recorder { + tx: Sender<(Timestamp, f64)>, +} +impl Recorder { + fn new(tx: Sender<(Timestamp, f64)>) -> Self { + Self { tx } + } +} +impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { + kind: ScalarKind::F64, + lookback: 1, + firing: Firing::Any, + }], + output: vec![], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.f64_in(0); + if w.is_empty() { + return None; + } + let _ = self.tx.send((ctx.now(), w[0])); + None + } +} + +fn f64_stream(pairs: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { + pairs + .iter() + .map(|&(t, v)| (Timestamp(t), Scalar::F64(v))) + .collect() +} + +fn build_and_run(tx: Sender<(Timestamp, f64)>) { + let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(Sub::new()), + Box::new(Recorder::new(tx)), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + 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 }, + ], + ) + .expect("valid DAG"); + + h.run(vec![f64_stream(&[ + (1, 10.5), + (2, 12.25), + (3, 9.75), + (4, 16.0), + (5, 18.125), + (6, 7.0), + (7, 21.5), + ])]); +} + +fn main() { + let (tx1, rx1) = mpsc::channel(); + build_and_run(tx1); + let run1: Vec<(Timestamp, f64)> = rx1.try_iter().collect(); + + let (tx2, rx2) = mpsc::channel(); + build_and_run(tx2); + let run2: Vec<(Timestamp, f64)> = rx2.try_iter().collect(); + + println!("run1 = {run1:?}"); + println!("run2 = {run2:?}"); + + // bit-exact comparison (f64::to_bits) — true byte-identity, not float ==. + assert_eq!(run1.len(), run2.len(), "same number of recorded events"); + for (a, b) in run1.iter().zip(run2.iter()) { + assert_eq!(a.0, b.0, "timestamps byte-identical"); + assert_eq!(a.1.to_bits(), b.1.to_bits(), "values byte-identical"); + } + assert!(!run1.is_empty(), "the run actually recorded something"); + println!("c0006_4 OK: recorded output byte-identical across two runs (C1)"); +} diff --git a/fieldtests/cycle-0006-substrate/c0006_5_reject_mismatch.rs b/fieldtests/cycle-0006-substrate/c0006_5_reject_mismatch.rs new file mode 100644 index 0000000..122eaa2 --- /dev/null +++ b/fieldtests/cycle-0006-substrate/c0006_5_reject_mismatch.rs @@ -0,0 +1,92 @@ +//! Fieldtest c0006 #5 — a mis-wired recording edge is rejected at bootstrap. +//! +//! Supports the "record interior streams" axis from the failure side: a recording +//! node declares typed input slots like any consumer, so an edge whose producer +//! field kind mismatches the slot kind is rejected at bootstrap (the existing +//! per-field 0005 check; recording adds no new hole). Correct behaviour here is +//! REJECTION — this fixture asserts the rejection happens with the documented +//! error, before any data flows. Mirrors the 0006 spec's must-fail fixture. + +use std::sync::mpsc; + +use aura_core::{Ctx, FieldSpec, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_engine::{BootstrapError, Edge, Harness, SourceSpec, Target}; + +/// A producer that emits one i64 field. +struct I64Source { + out: [Scalar; 1], +} +impl Node for I64Source { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { + kind: ScalarKind::I64, + lookback: 1, + firing: Firing::Any, + }], + output: vec![FieldSpec { name: "v", kind: ScalarKind::I64 }], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.i64_in(0); + if w.is_empty() { + return None; + } + self.out[0] = Scalar::I64(w[0]); + Some(&self.out) + } +} + +/// A recorder declaring an f64 input slot. +struct Recorder { + tx: mpsc::Sender<(Timestamp, f64)>, +} +impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { + kind: ScalarKind::F64, + lookback: 1, + firing: Firing::Any, + }], + output: vec![], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.f64_in(0); + if w.is_empty() { + return None; + } + let _ = self.tx.send((ctx.now(), w[0])); + None + } +} + +fn main() { + let (tx, _rx) = mpsc::channel(); + + // Bind the i64 producer field into the f64 recorder slot -> must be rejected. + let err = Harness::bootstrap( + vec![ + Box::new(I64Source { out: [Scalar::I64(0)] }), + Box::new(Recorder { tx }), + ], + vec![SourceSpec { + kind: ScalarKind::I64, + targets: vec![Target { node: 0, slot: 0 }], + }], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .unwrap_err(); + + println!("err = {err:?}"); + assert_eq!( + err, + BootstrapError::KindMismatch { + producer: ScalarKind::I64, + consumer: ScalarKind::F64, + }, + "kind-mismatched recorder edge must be rejected at bootstrap" + ); + println!("c0006_5 OK: mis-wired recording edge rejected with KindMismatch"); +}