diff --git a/Cargo.lock b/Cargo.lock index c5693e0..ba74e9f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -93,6 +93,7 @@ version = "0.1.0" dependencies = [ "aura-core", "aura-engine", + "serde", "serde_json", ] diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 8ebf1da..d158547 100644 --- a/crates/aura-engine/src/harness.rs +++ b/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]) -> 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 = Box::new(VecSource::new(f64_stream(&[(10, 1.0), (40, 2.0)]))); + let b: Box = 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 = Box::new(VecSource::new(vec![])); + let e2: Box = 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 { diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index d45ed26..b048c74 100644 --- a/crates/aura-engine/src/lib.rs +++ b/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}; diff --git a/crates/aura-ingest/src/lib.rs b/crates/aura-ingest/src/lib.rs index a411226..196356a 100644 --- a/crates/aura-ingest/src/lib.rs +++ b/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, + to_ms: Option, } impl M1FieldSource { @@ -183,7 +187,7 @@ impl M1FieldSource { field: M1Field, ) -> Option { 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)] diff --git a/crates/aura-ingest/tests/streaming_seam.rs b/crates/aura-ingest/tests/streaming_seam.rs index 849f1f0..58a0595 100644 --- a/crates/aura-ingest/tests/streaming_seam.rs +++ b/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); +} diff --git a/crates/aura-registry/Cargo.toml b/crates/aura-registry/Cargo.toml index 530f75c..1e7fdfa 100644 --- a/crates/aura-registry/Cargo.toml +++ b/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] diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index fd9308d..fc9ac9e 100644 --- a/crates/aura-registry/src/lib.rs +++ b/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![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![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![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::::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 = 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 + } } diff --git a/crates/aura-registry/src/lineage.rs b/crates/aura-registry/src/lineage.rs new file mode 100644 index 0000000..491727b --- /dev/null +++ b/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, +} + +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 { + let existing: HashSet = + 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, 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 { + (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) -> Vec { + let mut order: Vec = Vec::new(); + let mut by_id: HashMap = 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 = + 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 { + 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 { + 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 { + result.windows.iter().map(|w| w.run.oos_report.clone()).collect() +}