spec: 0041 source ingestion seam (boss-signed)

Re-types the engine's ingestion input from a materialized Vec<Vec<(Timestamp, Scalar)>> into a `Source` producer trait the k-way merge drives by peek/next, and proves it against a real, lazily-streamed data-server window in the same cycle — closing the cycle-0011 deliberate eager-materialization gap (C12 Arc<[T]> cross-sim sharing). A VecSource adapter keeps every existing run call site byte-identical (behaviour-preserving); an M1FieldSource streams data-server chunks with O(one-chunk) residency, independent of window length. Foundational cycle of the World-II milestone; #66/#68/#69 inherit the seam. Signed under the boss auto-sign gate: precondition clean, grounding PASS on the final bytes, unanimous five-lens spec-skeptic panel (two editorial ambiguity rounds on the residency predicate repaired to a measured per-pull CHUNK_SIZE ceiling). refs #71
2026-06-15 10:00:44 +02:00
parent 2b56e090a9
commit ae240140c4
1 changed files with 459 additions and 0 deletions
@@ -0,0 +1,459 @@
+# Source ingestion seam — Design Spec
+
+**Date:** 2026-06-15
+**Status:** Draft — awaiting user spec review
+**Authors:** orchestrator + Claude
+
+## Goal
+
+Replace the engine's single materialized-`Vec` ingestion input with a **producer
+seam** — a `Source` trait the k-way merge drives by `peek`/`next` — and prove it
+against a real, lazily-streamed `data-server` window in the same cycle, so the
+seam is designed against the real data source from the first line.
+
+`Harness::run` today takes `Vec<Vec<(Timestamp, Scalar)>>`: one fully
+materialized stream per source. That is the **only** engine type encoding an
+eager-dataflow assumption. At realistic research scale (20 years, 3 tick streams,
+~7.9e9 ticks) the eager layout is ~190 GB resident (24 B/tick) — non-residable.
+The merge loop itself never needs the whole stream: it only ever peeks the head
+timestamp of each live source and pops one record (`harness.rs:237`–`257`). So
+the eager `Vec` is gratuitous; a `peek`/`next` producer is the faithful seam, and
+a `data-server`-backed producer streams `Arc<[T]>` chunks lazily — resident
+footprint O(Σ node lookbacks + one chunk), flat across the horizon.
+
+This is the foundational cycle of the World-II milestone: the orchestration
+families (#66 / #68 / #69) all bootstrap harnesses over data windows, so they
+inherit this seam. Ledger cycle 0011 (`INDEX.md:530`) already names today's eager
+materialization "a known, deliberate gap" whose zero-copy `Arc<[T]>` cross-sim
+sharing is "the target for the orchestration cycle that introduces [the axes]" —
+this milestone. This cycle closes that gap by making a streaming path exist; it
+does not delete the eager convenience path (still valid for bounded loads).
+
+**Contracts touched:** C2 (bounded read-only windows), C3 (one merge at
+ingestion, ms→epoch-ns normalized there), C4 (cycle granularity / tie-by-source-
+index), C7 (SoA base columns), C12 (`Arc<[T]>` cross-sim sharing).
+
+## Architecture
+
+Three layers, one new boundary type:
+
+1. **`aura-engine`** gains a public `Source` trait and a `VecSource` adapter.
+   `Harness::run` is re-typed from `Vec<Vec<(Timestamp, Scalar)>>` to
+   `Vec<Box<dyn Source>>`. The merge loop's index arithmetic
+   (`streams[s][cursor[s]]`) becomes `sources[s].peek()` / `sources[s].next()`.
+   Every existing call site wraps its `Vec<(Timestamp, Scalar)>` in
+   `VecSource::new(..)` — behaviour-preserving, byte-identical run output, no
+   residency change. `VecSource` *is* the old behaviour, named.
+
+2. **`aura-ingest`** gains `M1FieldSource`: a streaming `Source` that holds a
+   `data-server` `SymbolChunkIter<M1Parsed>`, the current `Arc<[M1Parsed]>` chunk
+   (an `Arc` clone of the cache's chunk — zero-copy), a cursor into it, a field
+   selector, and a **pre-decoded head** record. It constructs each `Scalar`
+   per-pull from the borrowed chunk (no owned-`Vec` materialization of the
+   window) and refills via `next_chunk()` when the current chunk drains.
+
+3. **End-to-end**: a gated integration test bootstraps the signal-quality sample
+   harness, opens an `M1FieldSource` over a real `[from, to]` window, runs it
+   through the seam to a `RunReport`, and asserts determinism (C1).
+
+The `peek(&self)`/`next(&mut self)` split is what forces the pre-decoded head in
+the streaming source: answering `peek` without `&mut` is impossible over an
+underlying iterator whose refill (`next_chunk`) is `&mut`, so the head is decoded
+ahead of demand at construction and after every `next`. This is the standard
+peekable shape, determined by the trait signature (settled in the issue), not an
+open design choice.
+
+`SourceSpec` (`harness.rs:49`) is unchanged: it stays the *wiring* metadata
+(scalar `kind` + target slots), declared on the `FlatGraph`. A `Box<dyn Source>`
+is the *runtime* producer bound to source index `s` at `run` time. The two were
+already distinct (the old `Vec` was the runtime half); this cycle only re-types
+the runtime half from an owned `Vec` to a producer.
+
+### Scope boundary
+
+In scope: the trait, `VecSource`, the `run` re-type with every call site adapted,
+`M1FieldSource` (the streaming source), and one streaming end-to-end test.
+
+Out of scope (deliberately, to keep the seam focused):
+- Deleting the eager `load_m1_window` / `close_stream` path — it stays a valid
+  convenience for bounded loads; the CLI keeps using it (wrapped in `VecSource`).
+  The deliberate-eager gap is closed by a streaming path *existing*, not by
+  removing the eager one.
+- Migrating the CLI's real-data commands to streaming — follow-on.
+- Tick sources (`stream_tick_windowed`) — the same shape applies (`M1FieldSource`
+  generalizes), but the e2e and the SMA-cross sample consume M1 close; tick is
+  not built here.
+- Seed-as-input (#66) — the seed lives at the recording edge, outside this seam.
+
+## Concrete code shapes
+
+### User-facing program — a streaming real-window backtest
+
+What a project author (or a World program) writes to backtest a strategy over a
+real data window. This is the acceptance criterion's empirical evidence: the
+audience's natural reach, shown.
+
+```rust
+// Bootstrap the harness (unchanged — topology is FlatGraph as today).
+let mut h = Harness::bootstrap(sma_cross_signal_quality_graph())?;
+
+// Open a streaming source over a real window. None = no archived file overlaps
+// the window (data-server's file-level Option contract, propagated).
+let close = M1FieldSource::open(&server, "AAPL.US", Some(from_ms), Some(to_ms), M1Field::Close)
+    .expect("window overlaps real data");
+
+// Drive the seam. The source streams Arc<[M1Parsed]> chunks lazily: resident
+// footprint is one chunk + the engine's per-node lookback windows, NOT the
+// whole window. A 20-year window streams in the same memory as a one-day one.
+h.run(vec![Box::new(close)]);
+
+let report = fold_recorded_into_report(/* sinks */);
+```
+
+The eager path still works unchanged for bounded loads (every existing call
+site, now wrapped):
+
+```rust
+let prices: Vec<(Timestamp, Scalar)> =
+    load_m1_window(&server, "AAPL.US", Some(from_ms), Some(to_ms))?.close_stream();
+h.run(vec![Box::new(VecSource::new(prices))]); // was: h.run(vec![prices]);
+```
+
+### Delivered code — the `Source` trait (aura-engine)
+
+```rust
+/// The ingestion-boundary producer the k-way merge drives (C3). A source yields
+/// timestamped scalars in ascending timestamp order (the C3 ingestion
+/// precondition). Object-safe: the merge holds `Vec<Box<dyn Source>>`.
+pub trait Source {
+    /// The timestamp of the head record, without consuming it. `None` = the
+    /// source is exhausted. Cheap and side-effect-free: it reads a pre-decoded
+    /// head, never advances the underlying stream.
+    fn peek(&self) -> Option<Timestamp>;
+
+    /// 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)>;
+}
+```
+
+### Delivered code — `VecSource` (aura-engine), behaviour-preserving
+
+```rust
+/// A `Source` over a fully materialized stream — the cycle-0011 eager shape,
+/// named. Every pre-seam call site wraps its `Vec<(Timestamp, Scalar)>` in this,
+/// so run output is byte-identical (C1) and residency is unchanged.
+pub struct VecSource {
+    stream: Vec<(Timestamp, Scalar)>,
+    cursor: usize,
+}
+
+impl VecSource {
+    pub fn new(stream: Vec<(Timestamp, Scalar)>) -> Self {
+        Self { stream, cursor: 0 }
+    }
+}
+
+impl Source for VecSource {
+    fn peek(&self) -> Option<Timestamp> {
+        self.stream.get(self.cursor).map(|&(t, _)| t)
+    }
+    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
+        let item = self.stream.get(self.cursor).copied()?; // (Timestamp, Scalar): Copy
+        self.cursor += 1;
+        Some(item)
+    }
+}
+```
+
+### `Harness::run` — the merge loop re-type (before → after)
+
+```rust
+// BEFORE (harness.rs:220) — eager Vec input, index arithmetic.
+pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
+    assert_eq!(streams.len(), self.sources.len(), /* ... */);
+    let mut cursor: Vec<usize> = vec![0; streams.len()];
+    loop {
+        let mut pick: Option<usize> = None;
+        for (s, stream) in streams.iter().enumerate() {
+            if cursor[s] < stream.len() {
+                match pick {
+                    None => pick = Some(s),
+                    Some(p) => if stream[cursor[s]].0 < streams[p][cursor[p]].0 { pick = Some(s); }
+                }
+            }
+        }
+        let s = match pick { Some(s) => s, None => break };
+        let (ts, value) = streams[s][cursor[s]];
+        cursor[s] += 1;
+        /* ... forward into target slots, eval in topo order ... unchanged ... */
+    }
+}
+
+// AFTER — producer seam, peek/next. The pick scan, tie-by-source-index, and the
+// entire forward+eval body are preserved; only the head access changes.
+pub fn run(&mut self, mut sources: Vec<Box<dyn Source>>) {
+    assert_eq!(sources.len(), self.sources.len(), /* ... */);
+    loop {
+        // pick the live source head with the smallest (timestamp, source index).
+        // strictly-`<` replace + source-order scan preserves C4 tie-breaking.
+        let mut pick: Option<(usize, Timestamp)> = None;
+        for (s, src) in sources.iter().enumerate() {
+            if let Some(ts) = src.peek() {
+                match pick {
+                    None => pick = Some((s, ts)),
+                    Some((_, best)) if ts < best => pick = Some((s, ts)),
+                    _ => {}
+                }
+            }
+        }
+        let s = match pick { Some((s, _)) => s, None => break };
+        let (ts, value) = sources[s].next().expect("peeked Some ⇒ next is Some");
+        /* ... forward into target slots, eval in topo order ... byte-identical ... */
+    }
+}
+```
+
+Note the borrow shape: `self.sources` (the `SourceSpec` wiring) is read inside the
+forward step while `sources` (the producers) is the separate `run` argument, so
+the existing destructuring of `self` into disjoint field borrows is unaffected.
+
+### Delivered code — `M1FieldSource` (aura-ingest), the streaming source
+
+```rust
+/// Which base column of an M1 bar a source streams (C7: a composite window is a
+/// bundle of base columns; one `Source` per consumed field).
+pub enum M1Field { Open, High, Low, Close, Spread, Volume }
+
+/// A streaming `Source` over a data-server M1 window. Holds one Arc chunk
+/// (zero-copy clone of the cache's chunk) and a cursor; constructs each `Scalar`
+/// per-pull; refills via `next_chunk()` when the chunk drains. Resident footprint
+/// is O(one chunk), independent of window length.
+pub struct M1FieldSource {
+    iter: SymbolChunkIter<M1Parsed>,    // data-server's lazy chunk producer
+    chunk: Option<Arc<[M1Parsed]>>,     // current chunk, borrowed (Arc refcount, not a data copy)
+    pos: usize,                         // cursor within `chunk`
+    field: M1Field,                     // which column this source projects
+    head: Option<(Timestamp, Scalar)>,  // pre-decoded head ⇒ peek(&self) is cheap
+}
+
+impl M1FieldSource {
+    /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract).
+    /// `None` when no archived file overlaps the window (an unknown symbol or a
+    /// far-future window) — propagating `stream_m1_windowed`'s file-level Option.
+    /// A window that overlaps a file but holds zero matching bars yields a source
+    /// whose first `peek` is `None` (immediately exhausted), not `None` here.
+    pub fn open(
+        server: &Arc<DataServer>, symbol: &str,
+        from_ms: Option<i64>, to_ms: Option<i64>, 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 };
+        s.advance(); // pre-decode the first head
+        Some(s)
+    }
+
+    /// Project the bar at the cursor into (normalized ts, scalar) for `field`.
+    fn decode(&self, bar: &M1Parsed) -> (Timestamp, Scalar) {
+        let value = match self.field {
+            M1Field::Open => Scalar::F64(bar.open),
+            M1Field::High => Scalar::F64(bar.high),
+            M1Field::Low => Scalar::F64(bar.low),
+            M1Field::Close => Scalar::F64(bar.close),
+            M1Field::Spread => Scalar::F64(bar.spread),
+            M1Field::Volume => Scalar::I64(bar.volume),
+        };
+        (unix_ms_to_epoch_ns(bar.time_ms), value) // ms → epoch-ns at the seam (C3)
+    }
+
+    /// Decode the head at the cursor, refilling chunks as needed. Sets
+    /// `self.head = None` at exhaustion.
+    fn advance(&mut self) {
+        loop {
+            if self.chunk.is_none() {
+                self.chunk = self.iter.next_chunk(); // &mut refill; None = window exhausted
+                self.pos = 0;
+                if self.chunk.is_none() { self.head = None; return; }
+            }
+            let chunk = self.chunk.as_ref().unwrap();
+            if self.pos < chunk.len() {
+                self.head = Some(self.decode(&chunk[self.pos]));
+                return;
+            }
+            self.chunk = None; // current chunk drained — loop to refill
+        }
+    }
+
+    /// Records resident in this source right now: the current chunk's length (or
+    /// 0 at exhaustion). The residency probe — bounded by one chunk length by
+    /// construction (the struct holds at most one `Arc` chunk and no accumulating
+    /// field), so it can never grow with window length. Sampled by the residency
+    /// test to make the O(one chunk) claim a measured quantity, not prose.
+    pub fn resident_records(&self) -> usize {
+        self.chunk.as_ref().map_or(0, |c| c.len())
+    }
+}
+
+impl Source for M1FieldSource {
+    fn peek(&self) -> Option<Timestamp> {
+        self.head.map(|(t, _)| t)
+    }
+    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
+        let item = self.head?;
+        self.pos += 1;
+        self.advance(); // pre-decode the next head (may refill the chunk)
+        Some(item)
+    }
+}
+```
+
+### N-field generalization (shape the trait permits; not all-tested here)
+
+Two `M1FieldSource`s over the same `(symbol, window)` get the **same** `Arc`
+chunks from the data-server cache (the `SymbolGuard` keeps the symbol loaded; the
+cache shares `Arc<[T]>`), so they stream different fields sharing the chunk's ts
+axis zero-copy — C12's cross-source `Arc<[T]>` sharing:
+
+```rust
+let close  = M1FieldSource::open(&server, sym, f, t, M1Field::Close)?;
+let volume = M1FieldSource::open(&server, sym, f, t, M1Field::Volume)?;
+h.run(vec![Box::new(close), Box::new(volume)]); // co-scheduled by timestamp (C4)
+```
+
+The e2e headline uses the close source alone; a focused test asserts two sources
+over one window both stream (the sharing property), but the full OHLCV-as-N-
+sources harness is not built in this cycle.
+
+## Components
+
+| Component | Crate | Change |
+|-----------|-------|--------|
+| `Source` trait | aura-engine | **new** — `peek(&self)`, `next(&mut)`; object-safe; re-exported from the crate root |
+| `VecSource` | aura-engine | **new** — behaviour-preserving adapter over `Vec<(Timestamp, Scalar)>` |
+| `Harness::run` | aura-engine | **re-typed** — `Vec<Box<dyn Source>>`; merge loop drives `peek`/`next`; forward+eval body byte-identical |
+| ~53 `run` call sites | aura-engine, aura-cli, aura-ingest | **adapted** — wrap streams in `VecSource::new(..)`; mostly tests |
+| `M1Field`, `M1FieldSource` | aura-ingest | **new** — streaming M1 source, per-pull decode, lazy chunk refill, ms→ns normalization |
+| streaming e2e test | aura-ingest | **new** — real-window backtest through the seam to a `RunReport` |
+| `load_m1_window`, `close_stream`, `transpose_m1` | aura-ingest | **kept** — eager convenience for bounded loads; unchanged |
+
+## Data flow
+
+```
+data-server files ──stream_m1_windowed(sym, from_ms, to_ms)──▶ SymbolChunkIter<M1Parsed>
+                                                                    │  next_chunk() : Arc<[M1Parsed]>  (lazy, N+2 prefetch)
+                                                                    ▼
+M1FieldSource { chunk: Arc<[M1Parsed]>, pos, field, head } ──decode(bar)──▶ (epoch-ns ts, Scalar)
+                                                                    │  peek() / next()
+                                                                    ▼
+Harness::run(Vec<Box<dyn Source>>) ──k-way merge by (peek ts, source idx)──▶ one chronological cycle stream
+                                                                    │  (forward into target slots → eval in topo order — unchanged)
+                                                                    ▼
+                                                             recorded sinks → RunReport
+```
+
+The only resident data at any instant: each source's current `Arc` chunk
+(≤1 chunk) + one pre-decoded head, plus the engine's per-node lookback ring
+buffers (already O(Σ lookbacks), unchanged). Nothing scales with window length.
+
+## Error handling
+
+- **Source open failure** is an `Option`, not a `Result`: `M1FieldSource::open`
+  returns `None` exactly when `stream_m1_windowed` does — no archived file
+  overlaps the window (unknown symbol / far-future window). This mirrors the
+  existing `load_m1_window` file-level Option contract (`lib.rs:101`–`109`).
+- **Empty-but-overlapping window** (a file overlaps but holds zero matching
+  bars): `open` returns `Some`, and the source's first `peek` is `None` —
+  immediately exhausted. `run` over it drives zero cycles. This preserves the
+  existing distinction "data source present" vs "nothing to read from".
+- **Mid-stream**: `next`/`peek` never error — exhaustion is `None`. `next_chunk`
+  yields `None` at window end; `advance` maps that to `head = None`.
+- **`run`'s length assertion** (one source per `SourceSpec`) is unchanged.
+- The ascending-timestamp ingestion precondition (C3) is unchanged: data-server
+  files are chronological, and `decode` preserves order within and across chunks.
+
+## Testing strategy
+
+1. **Behaviour-preserving (the load-bearing guarantee).** The full
+   `cargo test --workspace` suite stays green and byte-identical after the
+   re-type — every call site wrapped in `VecSource`. The existing C1-determinism
+   assertions (e.g. `real_bars.rs:118`–`120`, `blueprint.rs` sweep equality) are
+   the evidence; no test output changes.
+2. **`VecSource` unit tests** (aura-engine): `peek` is non-consuming and idempotent;
+   `peek` then `next` agree; exhaustion yields `None`; an empty stream peeks `None`.
+3. **`decode` projection unit tests** (aura-ingest, hermetic — pure `decode` on
+   hand-made `M1Parsed`, no data-server files): each `M1Field` projects to the
+   right `Scalar` kind/value with ts normalized ms→ns. `decode` is a pure
+   function of `(bar, field)`, so it is fully testable without an iterator or
+   files. (The `peek`/`next`/refill walk over a real iterator is exercised by the
+   gated e2e below, which is where a real `SymbolChunkIter` exists.)
+4. **Streaming e2e + residency proof** (aura-ingest, gated — skips where the
+   local Pepperstone directory is absent, like `real_bars.rs`): bootstrap the
+   signal-quality sample harness, `M1FieldSource::open` over a real window
+   spanning **many** data-server chunks, `h.run` through the seam, fold to a
+   `RunReport`; assert `total_pips.is_finite()` and C1 (two runs bit-identical
+   JSON). **Residency predicate — the measurable O(one chunk) claim:** sampling
+   `resident_records()` at every pull across the whole run, its maximum is
+   **≤ one data-server chunk** — the `CHUNK_SIZE` bound (1024 records;
+   `data_server`'s documented chunk size, asserted against the constant, not a
+   magic literal). This per-pull bound is **independent of window length**: it
+   holds at every pull whether the window spans a handful of chunks or millions,
+   which is the O(one chunk)-not-O(window) content. The bound is correct on real
+   chunks regardless of boundary-chunk lengths (`filter_chunk` only ever yields a
+   chunk *≤* `CHUNK_SIZE`). A source that accumulated chunks would blow past the
+   one-chunk bound on a long window; a source that accumulated decoded state is
+   excluded structurally by §5. (Deliberately *not* an "equal peak across two
+   windows" assertion — boundary/file-final chunks are short, so per-window
+   maxima legitimately differ; the load-bearing invariant is the per-pull
+   one-chunk *ceiling*, which is what bounds residency.)
+5. **Structural no-accumulation guarantee** (always-on, not gated): `M1FieldSource`
+   holds exactly `{ iter, one Option<Arc<[M1Parsed]>>, cursor, field, one head }`
+   — no `Vec<Arc<..>>` or other field that grows with the stream — so
+   `resident_records()` is bounded by one chunk length *by construction*. This is
+   the always-on half of the residency claim (verifiable from the struct
+   definition + clippy); the gated item 4 is the dynamic-invariance half.
+6. **Sharing property** (gated): two `M1FieldSource`s (close + volume) over one
+   window both stream to exhaustion in a single `run` — the N-sources-share-the-
+   window shape.
+
+## Acceptance criteria
+
+- [ ] `Source` (`peek` + `next`, object-safe) and `VecSource` exist in
+      aura-engine; `Harness::run` takes `Vec<Box<dyn Source>>`.
+- [ ] Every `run` call site is adapted; `cargo test --workspace` is green and
+      run output is byte-identical to pre-seam (behaviour-preserving).
+- [ ] `M1FieldSource` streams a data-server M1 window through the seam: per-pull
+      `Scalar` decode from a borrowed `Arc` chunk, lazy `next_chunk` refill,
+      ms→epoch-ns normalized at the seam (C3).
+- [ ] A real short-window backtest runs end-to-end through the seam to a
+      `RunReport`, deterministically (C1).
+- [ ] **Residency (measured predicate):** across a full run over a window
+      spanning many chunks, peak `M1FieldSource::resident_records()` is
+      **≤ one data-server chunk** (`CHUNK_SIZE` = 1024) — a per-pull ceiling
+      **independent of window length** — so the source holds O(one chunk), NOT
+      O(window length), measured on real multi-chunk data (Testing strategy §4).
+      The struct carries no accumulating field, so this bound holds by
+      construction (§5). The engine's per-node lookback ring buffers are
+      unchanged — already O(Σ lookbacks), pinned by the existing suite — so total
+      resident memory is flat across the horizon.
+- [ ] `cargo clippy --workspace --all-targets -- -D warnings` clean;
+      `cargo doc --workspace --no-deps` clean.
+
+## Surfaced boundary frictions (recorded, per the issue's friction-first intent)
+
+- **Transpose granularity.** The streaming source does **per-pull** field decode
+  from the borrowed chunk, not `transpose_m1`'s per-window owned-SoA transpose.
+  The eager `M1Columns`/`transpose_m1` path is a different shape, kept for bounded
+  loads; the two now coexist (a follow-on may unify them).
+- **`Arc` borrow lifetime.** Each source holds an `Arc<[M1Parsed]>` clone; across
+  disjoint sims (C1/C12) the data-server cache shares the same chunks zero-copy.
+  The `SymbolGuard` inside `SymbolChunkIter` keeps the symbol loaded for the
+  source's lifetime.
+- **Data-server cache file retention.** The aura-side source is O(one chunk); the
+  data-server *cache* retains loaded files (with N+2 prefetch). Whether files
+  behind the cursor are evicted is the data-server's contract, **owned by that
+  lib, not this cycle** — the residency claim here is scoped to aura's source
+  footprint. Flagged for the milestone, not resolved here.
+- **Error channel asymmetry.** `Option` at open (file-level), `None`-on-
+  exhaustion mid-stream — no `Result` anywhere, matching `stream_m1_windowed`.
+- **`peek(&self)` vs `&mut` refill.** Resolved by the pre-decoded head; the head
+  decode happens in `advance`, called from `open` and `next`.