feat(aura-engine,aura-registry): producer-supplied window + registry lineage (0045 iter 1)

Iteration 1 of spec 0045 — the engine + registry core for storing
orchestration families as linked records.

aura-engine: add Source::bounds() — the inclusive (from, to) data extent a
producer will stream, known without materialization (#71 firewall) — and a
free window_of() that folds the union extent across a run's sources. This is
the producer-supplied replacement for the prices.first()/.last() Vec scans
at the CLI call sites (those migrate in iteration 2); a lazy producer now
reports its window so stored lineage stays byte-identical whether a source
is eager or streamed.

aura-ingest: M1FieldSource stores its requested [from_ms, to_ms] window at
open and reports it via bounds() (normalized to epoch-ns; None when
open-ended — the archive-extent query is a deferred non-goal).

aura-registry: new lineage module — FamilyKind / FamilyRunRecord (a
RunReport stamped with family_id + kind + ordinal) / Family, plus
Registry::append_family (assigns family_id = "{name}-{counter}" via a
per-name generate-and-check counter, writes members to a sibling
families.jsonl) and load_family_members; group_families re-derives the
families (the round-trip), and the three *_member_reports extractors pull
the per-kind member reports. The flat runs.jsonl store and its
append/load/rank_by/optimize API are byte-for-byte unchanged (a test pins
the two stores' disjointness). C9 preserved: aura-engine gains no registry
dependency. New dep: serde derive on aura-registry (per-case policy, same
basis as serde_json already is).

Verification: cargo build --workspace; cargo test --workspace (228 green,
incl. 3 new engine bounds/window_of, 1 data-gated ingest bounds run against
real archive data, 5 new registry lineage round-trip/counter/disjointness
tests); cargo clippy --workspace --all-targets -D warnings clean. The CLI
surface (aura mc, family-aware aura runs, the window_of migration) is
iteration 2.

refs #70

Cargo.lock

@@ -93,6 +93,7 @@ version = "0.1.0"
 dependencies = [
  "aura-core",
  "aura-engine",
+ "serde",
  "serde_json",
 ]
 

crates/aura-engine/src/harness.rs

@@ -63,6 +63,13 @@ pub trait Source {
     /// Pop the head `(timestamp, scalar)` and advance. `None` = exhausted.
     /// After `next` returns `Some`, `peek` reflects the new head.
     fn next(&mut self) -> Option<(Timestamp, Scalar)>;
+
+    /// The inclusive `(from, to)` data extent this producer will stream, known
+    /// without materializing it (#71/C18): a lazy producer reports its window
+    /// bounds; an eager one reports its buffer's first/last. `None` = empty /
+    /// open-ended. Cursor-independent: a property of the data extent, not the
+    /// read position.
+    fn bounds(&self) -> Option<(Timestamp, Timestamp)>;
 }
 
 /// A `Source` over a fully materialized stream — the cycle-0011 eager shape,
@@ -88,6 +95,33 @@ impl Source for VecSource {
         self.cursor += 1;
         Some(item)
     }
+    fn bounds(&self) -> Option<(Timestamp, Timestamp)> {
+        // the eager source reads its own buffer ends (O(1) on the slice) — the
+        // consumer no longer scans a Vec. Cursor-independent (ignores self.cursor).
+        Some((self.stream.first()?.0, self.stream.last()?.0))
+    }
+}
+
+/// The window of a run: the union extent across its sources (min `from`, max
+/// `to`), each source contributing its [`Source::bounds`]. `None` when every
+/// source is empty / open-ended. The producer-supplied replacement for scanning a
+/// materialized `Vec` at the call site (#71/C18): byte-identical whether a source
+/// is eager (`VecSource`) or lazily streamed. Uses `<`/`>` only, so it needs no
+/// `Ord` on `Timestamp`.
+pub fn window_of(sources: &[Box<dyn Source>]) -> Option<(Timestamp, Timestamp)> {
+    let mut acc: Option<(Timestamp, Timestamp)> = None;
+    for s in sources {
+        if let Some((from, to)) = s.bounds() {
+            acc = Some(match acc {
+                None => (from, to),
+                Some((lo, hi)) => (
+                    if from < lo { from } else { lo },
+                    if to > hi { to } else { hi },
+                ),
+            });
+        }
+    }
+    acc
 }
 
 /// A tiny, fully-deterministic, dependency-free PRNG (SplitMix64). The seed
@@ -517,6 +551,40 @@ mod tests {
         assert_eq!(Source::next(&mut s), None);
     }
 
+    #[test]
+    fn vec_source_bounds_are_buffer_ends_and_cursor_independent() {
+        let mut s = VecSource::new(f64_stream(&[(10, 1.0), (20, 2.0), (30, 3.0)]));
+        // bounds = (first ts, last ts) of the buffer
+        assert_eq!(s.bounds(), Some((Timestamp(10), Timestamp(30))));
+        // cursor-independent: advancing does not move the reported extent
+        let _ = Source::next(&mut s);
+        assert_eq!(s.bounds(), Some((Timestamp(10), Timestamp(30))));
+        // an empty buffer has no extent
+        let empty = VecSource::new(vec![]);
+        assert_eq!(empty.bounds(), None);
+    }
+
+    #[test]
+    fn window_of_folds_union_extent_across_sources() {
+        let a: Box<dyn Source> = Box::new(VecSource::new(f64_stream(&[(10, 1.0), (40, 2.0)])));
+        let b: Box<dyn Source> = Box::new(VecSource::new(f64_stream(&[(5, 1.0), (25, 2.0)])));
+        // union extent: min first, max last
+        assert_eq!(window_of(&[a, b]), Some((Timestamp(5), Timestamp(40))));
+        // all-empty -> None
+        let e1: Box<dyn Source> = Box::new(VecSource::new(vec![]));
+        let e2: Box<dyn Source> = Box::new(VecSource::new(vec![]));
+        assert_eq!(window_of(&[e1, e2]), None);
+    }
+
+    #[test]
+    fn synthetic_source_bounds_match_analytic_span() {
+        let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
+        let s = spec.source(7);
+        // first ts = 1 (i=0); last ts = (len-1)*step + 1
+        let last = (spec.len as i64 - 1) * spec.step + 1;
+        assert_eq!(s.bounds(), Some((Timestamp(1), Timestamp(last))));
+    }
+
     /// One f64 input port with the given firing policy (lookback 1 is the bootstrap
     /// default for these test fixtures, supplied by their `lookbacks()`).
     fn f64_port(firing: Firing) -> PortSpec {

crates/aura-engine/src/lib.rs

@@ -58,7 +58,8 @@ pub use blueprint::{
 pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHandle};
 pub use graph_model::model_to_json;
 pub use harness::{
-    BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target, VecSource,
+    window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target,
+    VecSource,
 };
 pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
 pub use sweep::{sweep, GridSpace, SweepError, SweepFamily, SweepPoint};

crates/aura-ingest/src/lib.rs

@@ -167,6 +167,10 @@ pub struct M1FieldSource {
     pos: usize,
     field: M1Field,
     head: Option<(Timestamp, Scalar)>,
+    /// The requested window (inclusive Unix-ms), kept so `bounds()` can report a
+    /// producer-supplied extent without streaming (#71). `None` = open-ended.
+    from_ms: Option<i64>,
+    to_ms: Option<i64>,
 }
 
 impl M1FieldSource {
@@ -183,7 +187,7 @@ impl M1FieldSource {
         field: M1Field,
     ) -> Option<Self> {
         let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
-        let mut s = Self { iter, chunk: None, pos: 0, field, head: None };
+        let mut s = Self { iter, chunk: None, pos: 0, field, head: None, from_ms, to_ms };
         s.advance();
         Some(s)
     }
@@ -228,6 +232,15 @@ impl aura_engine::Source for M1FieldSource {
         self.advance();
         Some(item)
     }
+    fn bounds(&self) -> Option<(Timestamp, Timestamp)> {
+        // the producer-supplied window: the requested [from_ms, to_ms] normalized
+        // to epoch-ns, known without streaming a bar (#71). Open-ended (either
+        // bound absent) -> no known extent (Non-goals: archive-extent query).
+        match (self.from_ms, self.to_ms) {
+            (Some(from), Some(to)) => Some((unix_ms_to_epoch_ns(from), unix_ms_to_epoch_ns(to))),
+            _ => None,
+        }
+    }
 }
 
 #[cfg(test)]

crates/aura-ingest/tests/streaming_seam.rs

@@ -189,3 +189,27 @@ fn two_field_sources_share_one_window() {
     assert_eq!(n_close, n_volume, "both fields share the same ts axis ⇒ same count");
     assert!(n_close > 0);
 }
+
+#[test]
+fn field_source_bounds_are_the_requested_window_normalized() {
+    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
+    if skip_if_no_data(&server) {
+        return;
+    }
+    let src = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
+        .expect("AAPL.US has data in the 2006-08 window");
+    // bounds report the *requested* window normalized to epoch-ns, without
+    // streaming a single bar (producer-supplied window; #71 firewall).
+    assert_eq!(
+        Source::bounds(&src),
+        Some((
+            aura_ingest::unix_ms_to_epoch_ns(FROM_MS),
+            aura_ingest::unix_ms_to_epoch_ns(TO_MS),
+        ))
+    );
+    // an open-ended source (no window) has no known extent (Non-goals: archive
+    // extent query is deferred).
+    let open_ended = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close)
+        .expect("AAPL.US archive exists");
+    assert_eq!(Source::bounds(&open_ended), None);
+}

crates/aura-registry/Cargo.toml

@@ -10,6 +10,9 @@ publish.workspace = true
 # back (admitted under the amended C16 per-case policy, INDEX.md). RunReport
 # derives serde in aura-engine.
 aura-engine = { path = "../aura-engine" }
+# the lineage record types (FamilyKind / FamilyRunRecord) derive serde; admitted
+# under the same per-case policy as serde_json (INDEX.md).
+serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
 
 [dev-dependencies]

crates/aura-registry/src/lib.rs

@@ -4,6 +4,11 @@
 //! experiments over time", which has no home in git or Gitea). Storage is
 //! serde_json; display is the caller's concern (the CLI prints via
 //! `RunReport::to_json`).
+//!
+//! Orchestration **families** (sweep / Monte-Carlo / walk-forward, C12/C21) are
+//! stored as related records in a sibling family store (`families.jsonl`): each
+//! member is a `RunReport` stamped with a shared `family_id`, re-derived as a unit
+//! by [`group_families`]. The flat runs store and its API are untouched.
 
 use std::cmp::Ordering;
 use std::fmt;
@@ -13,6 +18,12 @@ use std::path::{Path, PathBuf};
 
 use aura_engine::{RunReport, SweepFamily, SweepPoint};
 
+mod lineage;
+pub use lineage::{
+    group_families, mc_member_reports, sweep_member_reports, walkforward_member_reports, Family,
+    FamilyKind, FamilyRunRecord,
+};
+
 /// An append-only run registry over a JSONL file: one serde_json line per
 /// `RunReport`.
 pub struct Registry {
@@ -282,4 +293,91 @@ mod tests {
         // the whole winning point is returned, params AND report together.
         assert_eq!(winner, family.points[1]);
     }
+
+    fn temp_family_dir(name: &str) -> PathBuf {
+        // a unique per-test directory so the families.jsonl sibling never collides
+        // across tests sharing the temp dir; the registry binds to runs.jsonl inside.
+        let dir =
+            std::env::temp_dir().join(format!("aura-registry-fam-{}-{}", std::process::id(), name));
+        let _ = fs::remove_dir_all(&dir);
+        fs::create_dir_all(&dir).expect("create temp family dir");
+        dir.join("runs.jsonl")
+    }
+
+    #[test]
+    fn lineage_round_trips_one_family() {
+        let path = temp_family_dir("roundtrip");
+        let reg = Registry::open(&path);
+        let reports =
+            vec![report_with(1.0, 0.5, 1), report_with(2.0, 0.3, 2), report_with(3.0, 0.1, 0)];
+        let id = reg.append_family("f", FamilyKind::MonteCarlo, &reports).expect("append family");
+        assert_eq!(id, "f-0");
+        let families = group_families(reg.load_family_members().expect("load members"));
+        assert_eq!(families.len(), 1);
+        let fam = &families[0];
+        assert_eq!(fam.id, "f-0");
+        assert_eq!(fam.kind, FamilyKind::MonteCarlo);
+        // members re-derived as a unit, ordinal-ordered, equal to the stamped reports
+        let member_reports: Vec<_> = fam.members.iter().map(|m| m.report.clone()).collect();
+        assert_eq!(member_reports, reports);
+        assert_eq!(fam.members.iter().map(|m| m.ordinal).collect::<Vec<_>>(), vec![0, 1, 2]);
+    }
+
+    #[test]
+    fn per_name_counter_increments_independently() {
+        let path = temp_family_dir("counter");
+        let reg = Registry::open(&path);
+        let reports = vec![report_with(1.0, 0.5, 1)];
+        let x0 = reg.append_family("x", FamilyKind::Sweep, &reports).expect("x-0");
+        let x1 = reg.append_family("x", FamilyKind::Sweep, &reports).expect("x-1");
+        let y0 = reg.append_family("y", FamilyKind::Sweep, &reports).expect("y-0");
+        assert_eq!((x0.as_str(), x1.as_str(), y0.as_str()), ("x-0", "x-1", "y-0"));
+        let families = group_families(reg.load_family_members().expect("load"));
+        assert_eq!(families.len(), 3);
+    }
+
+    #[test]
+    fn distinct_families_regroup_in_first_seen_order() {
+        let path = temp_family_dir("regroup");
+        let reg = Registry::open(&path);
+        let a = vec![report_with(1.0, 0.5, 1), report_with(2.0, 0.5, 1)];
+        let b = vec![report_with(3.0, 0.5, 1)];
+        let ida = reg.append_family("a", FamilyKind::Sweep, &a).expect("a");
+        let idb = reg.append_family("b", FamilyKind::WalkForward, &b).expect("b");
+        let families = group_families(reg.load_family_members().expect("load"));
+        // first-seen file order: a before b
+        assert_eq!(families.iter().map(|f| f.id.clone()).collect::<Vec<_>>(), vec![ida, idb]);
+        assert_eq!(families[0].members.len(), 2);
+        assert_eq!(families[1].kind, FamilyKind::WalkForward);
+        // members ordinal-sorted within each family
+        assert_eq!(families[0].members.iter().map(|m| m.ordinal).collect::<Vec<_>>(), vec![0, 1]);
+    }
+
+    #[test]
+    fn family_store_and_flat_store_are_disjoint() {
+        let path = temp_family_dir("disjoint");
+        let reg = Registry::open(&path);
+        // a family write leaves the flat runs store empty...
+        reg.append_family("f", FamilyKind::Sweep, &[report_with(1.0, 0.5, 1)]).expect("family");
+        assert_eq!(reg.load().expect("flat load"), Vec::<RunReport>::new());
+        // ...and a flat write leaves the family store untouched at one family.
+        reg.append(&report_with(9.0, 0.5, 1)).expect("flat append");
+        assert_eq!(reg.load().expect("flat load 2").len(), 1);
+        assert_eq!(group_families(reg.load_family_members().expect("members")).len(), 1);
+    }
+
+    #[test]
+    fn rank_a_family_as_a_unit() {
+        let path = temp_family_dir("rankfam");
+        let reg = Registry::open(&path);
+        let reports =
+            vec![report_with(1.0, 0.5, 1), report_with(3.0, 0.5, 1), report_with(2.0, 0.5, 1)];
+        reg.append_family("f", FamilyKind::Sweep, &reports).expect("family");
+        let fam = group_families(reg.load_family_members().expect("members"))
+            .pop()
+            .expect("one family");
+        let member_reports: Vec<RunReport> = fam.members.iter().map(|m| m.report.clone()).collect();
+        let ranked = rank_by(member_reports, "total_pips").expect("rank");
+        assert_eq!(ranked[0].metrics.total_pips, 3.0); // best-first within the family
+    }
 }

crates/aura-registry/src/lineage.rs

@@ -0,0 +1,153 @@
+//! Registry lineage for orchestration families (C18/C21): a family — a sweep,
+//! Monte-Carlo, or walk-forward run — is a set of run records sharing one
+//! `family_id` (the parent reference). This module adds the persisted lineage
+//! unit ([`FamilyRunRecord`]), the family-store accessors on [`Registry`]
+//! ([`Registry::append_family`] / [`Registry::load_family_members`]), and the
+//! round-trip that re-derives a [`Family`] from the stored links
+//! ([`group_families`]). The family store is a sibling JSONL of the flat runs
+//! store (`families.jsonl`), so the flat `runs.jsonl` path and API are untouched.
+
+use std::collections::{HashMap, HashSet};
+use std::fs;
+use std::io::Write;
+
+use aura_engine::{McFamily, RunReport, SweepFamily, WalkForwardResult};
+use serde::{Deserialize, Serialize};
+
+use crate::{Registry, RegistryError};
+
+/// Which C12 orchestration axis produced a family.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub enum FamilyKind {
+    Sweep,
+    MonteCarlo,
+    WalkForward,
+}
+
+/// One persisted family member: a run record stamped with its lineage link.
+/// `family_id` is the shared parent reference; `ordinal` is the member's position
+/// in the family's canonical order (odometer / seed-input / roll).
+#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
+pub struct FamilyRunRecord {
+    pub family_id: String,
+    pub kind: FamilyKind,
+    pub ordinal: usize,
+    pub report: RunReport,
+}
+
+/// A re-derived family: the members sharing one `family_id`, ordinal-sorted — the
+/// round-trip result of [`group_families`].
+#[derive(Clone, Debug, PartialEq)]
+pub struct Family {
+    pub id: String,
+    pub kind: FamilyKind,
+    pub members: Vec<FamilyRunRecord>,
+}
+
+impl Registry {
+    /// Assign a fresh `family_id = "{name}-{counter}"` (counter = the smallest `k`
+    /// for which `"{name}-{k}"` is absent in the family store — a per-name run
+    /// index), stamp `reports` as ordinal-ordered members, and append them as N
+    /// JSONL lines to the family store (a sibling of the flat runs store,
+    /// `families.jsonl`, leaving the flat path untouched). Returns the assigned id.
+    /// Reads the store once to pick the counter (a read-before-write; single-process
+    /// CLI invocations do not race).
+    pub fn append_family(
+        &self,
+        name: &str,
+        kind: FamilyKind,
+        reports: &[RunReport],
+    ) -> Result<String, RegistryError> {
+        let existing: HashSet<String> =
+            self.load_family_members()?.into_iter().map(|r| r.family_id).collect();
+        let family_id = next_family_id(name, &existing);
+
+        let path = self.path.with_file_name("families.jsonl");
+        if let Some(parent) = path.parent().filter(|p| !p.as_os_str().is_empty()) {
+            fs::create_dir_all(parent)?;
+        }
+        let mut file = fs::OpenOptions::new().create(true).append(true).open(&path)?;
+        for (ordinal, report) in reports.iter().enumerate() {
+            let record = FamilyRunRecord {
+                family_id: family_id.clone(),
+                kind,
+                ordinal,
+                report: report.clone(),
+            };
+            // a finite RunReport wrapped in a FamilyRunRecord serializes infallibly.
+            let line = serde_json::to_string(&record).expect("a finite FamilyRunRecord serializes");
+            writeln!(file, "{line}")?;
+        }
+        Ok(family_id)
+    }
+
+    /// Parse every stored family member, in file order. A missing file is an empty
+    /// family store (`Ok(vec![])`), exactly as [`Registry::load`] treats a missing
+    /// flat store.
+    pub fn load_family_members(&self) -> Result<Vec<FamilyRunRecord>, RegistryError> {
+        let path = self.path.with_file_name("families.jsonl");
+        let text = match fs::read_to_string(&path) {
+            Ok(t) => t,
+            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
+            Err(e) => return Err(RegistryError::Io(e)),
+        };
+        let mut records = Vec::new();
+        for (i, raw) in text.lines().enumerate() {
+            if raw.trim().is_empty() {
+                continue;
+            }
+            let record = serde_json::from_str(raw)
+                .map_err(|source| RegistryError::Parse { line: i + 1, source })?;
+            records.push(record);
+        }
+        Ok(records)
+    }
+}
+
+/// The next free `"{name}-{counter}"` id: the smallest `k >= 0` for which
+/// `"{name}-{k}"` is not in `existing`. Robust to `-` inside `name` — it tests
+/// membership of generated candidates, never parses the stored strings. Module-
+/// private: the id form is an `append_family` implementation detail.
+fn next_family_id(name: &str, existing: &HashSet<String>) -> String {
+    (0usize..)
+        .map(|k| format!("{name}-{k}"))
+        .find(|id| !existing.contains(id))
+        .expect("the usize counter space is effectively unbounded")
+}
+
+/// Group member records into families by `family_id` (first-seen family order;
+/// each family's members ordinal-sorted). The round-trip: `append_family(...)`
+/// then `group_families(load_family_members())` recovers the families.
+pub fn group_families(records: Vec<FamilyRunRecord>) -> Vec<Family> {
+    let mut order: Vec<String> = Vec::new();
+    let mut by_id: HashMap<String, Family> = HashMap::new();
+    for rec in records {
+        let fam = by_id.entry(rec.family_id.clone()).or_insert_with(|| {
+            order.push(rec.family_id.clone());
+            Family { id: rec.family_id.clone(), kind: rec.kind, members: Vec::new() }
+        });
+        fam.members.push(rec);
+    }
+    let mut families: Vec<Family> =
+        order.into_iter().map(|id| by_id.remove(&id).expect("id was inserted")).collect();
+    for fam in &mut families {
+        fam.members.sort_by_key(|m| m.ordinal);
+    }
+    families
+}
+
+/// The member reports of a sweep family, in enumeration order — the trivial
+/// extractor the CLI hands to [`Registry::append_family`].
+pub fn sweep_member_reports(family: &SweepFamily) -> Vec<RunReport> {
+    family.points.iter().map(|p| p.report.clone()).collect()
+}
+
+/// The member reports of a Monte-Carlo family, in seed-input order.
+pub fn mc_member_reports(family: &McFamily) -> Vec<RunReport> {
+    family.draws.iter().map(|d| d.report.clone()).collect()
+}
+
+/// The per-window OOS reports of a walk-forward result, in roll order.
+pub fn walkforward_member_reports(result: &WalkForwardResult) -> Vec<RunReport> {
+    result.windows.iter().map(|w| w.run.oos_report.clone()).collect()
+}