diff --git a/docs/plans/0094-content-addressed-reproduction.md b/docs/plans/0094-content-addressed-reproduction.md new file mode 100644 index 0000000..7ee82da --- /dev/null +++ b/docs/plans/0094-content-addressed-reproduction.md @@ -0,0 +1,541 @@ +# Content-addressed reproduction — iteration 1 (the heart) — Implementation Plan + +> **Parent spec:** `docs/specs/0094-content-addressed-reproduction.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Content-address a blueprint-sweep's topology and add `aura reproduce +`, which re-derives every persisted member bit-identically from the +stored blueprint (acceptance 2). + +**Architecture:** A dumb content-addressed bytes store in `aura-registry` +(`runs/blueprints/.json`, keyed by the `topology_hash` the manifest already +carries) is written once per sweep at the existing persist seam. The per-member +reduce-mode run is extracted from the sweep closure into a shared +`run_blueprint_member`, so `reproduce_family` re-runs the *identical* path and a +match is bit-identical by construction (C1). Reproduction reconstructs each +member's bootstrap point from its recorded params (the inverse of `zip_params`). + +**Tech Stack:** `aura-registry` (store), `aura-cli` (persist write, reproduce +path, dispatch arm), reusing `aura-engine::blueprint_serde` (untouched, C9). + +**Acceptance 1 (`graph introspect --content-id`) and 3 (Tier-1 id stability) are +DEFERRED to a follow-on plan 0094b — not in this iteration.** + +--- + +## Files this plan creates or modifies + +- Modify: `crates/aura-registry/src/lib.rs` — add `put_blueprint` / `get_blueprint` + / `blueprints_dir` to `impl Registry` (~42-97); a round-trip unit test in the + test module. +- Modify: `crates/aura-cli/src/main.rs` — extract `run_blueprint_member` from + `blueprint_sweep_family`'s member closure (~3017-3045); add the store write in + `run_blueprint_sweep` (~3063-3082); add `reproduce_family` / `reproduce_family_in` + / `point_from_params` / `ReproduceReport`; add the `["reproduce", id]` dispatch + arm (~3853); a reproduce unit test in the `#[cfg(test)] mod tests`. +- Test: `crates/aura-cli/tests/cli_run.rs` — an E2E driving the built binary: + `aura sweep … --name X` then `aura reproduce X`; plus a missing-family rejection. + +--- + +## Task 1: Content-addressed blueprint store (aura-registry) + +**Files:** +- Modify: `crates/aura-registry/src/lib.rs:42-97` (add three methods to `impl Registry`) +- Test: `crates/aura-registry/src/lib.rs` (test module, ~639+) + +- [ ] **Step 1: Write the failing test** + +Add to the `#[cfg(test)] mod tests` (use the existing `temp_path(name)` helper at ~639): + +```rust +#[test] +fn blueprint_store_round_trips_by_content_id() { + let reg = Registry::open(temp_path("blueprint_store_round_trip")); + let canonical = r#"{"format_version":1,"input_roles":[],"nodes":[],"edges":[]}"#; + reg.put_blueprint("deadbeef", canonical).expect("put"); + // exact bytes come back, keyed by content id + assert_eq!(reg.get_blueprint("deadbeef").expect("get"), Some(canonical.to_string())); + // an absent id is Ok(None), not an error (treat-as-empty discipline) + assert_eq!(reg.get_blueprint("never-written").expect("get"), None); +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cargo test -p aura-registry blueprint_store_round_trips_by_content_id` +Expected: FAIL — compile error (`no method named put_blueprint`). + +- [ ] **Step 3: Write the store methods** + +Add to `impl Registry` (the block at ~42-97). `RegistryError::Io` already exists +(491-504) with `impl From` (531-535), so `?` auto-maps: + +```rust + /// The content-addressed blueprint store dir — a sibling of the runs store, + /// `.with_file_name("blueprints")`. A topology is stored once, + /// keyed by its content id (`topology_hash`), so a whole sweep family's + /// members share one stored blueprint (C18 tiny manifest; C11/C12 dedup). + fn blueprints_dir(&self) -> std::path::PathBuf { + self.path.with_file_name("blueprints") + } + + /// Write-once content-addressed put: `blueprints/.json` = `canonical_json`. + /// Idempotent — the same content id always addresses identical canonical bytes, + /// so a repeated write re-writes identical content. The registry does NOT + /// verify `sha256(bytes) == hash` (no `sha2` dep here): the caller owns the + /// hash, and reproduction's bit-identical metric compare is the integrity check. + pub fn put_blueprint(&self, hash: &str, canonical_json: &str) -> Result<(), RegistryError> { + let dir = self.blueprints_dir(); + std::fs::create_dir_all(&dir)?; + std::fs::write(dir.join(format!("{hash}.json")), canonical_json)?; + Ok(()) + } + + /// Read a stored blueprint by content id; `Ok(None)` if absent — the same + /// treat-as-empty discipline `load` applies to a missing runs store. + pub fn get_blueprint(&self, hash: &str) -> Result, RegistryError> { + match std::fs::read_to_string(self.blueprints_dir().join(format!("{hash}.json"))) { + Ok(s) => Ok(Some(s)), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None), + Err(e) => Err(RegistryError::Io(e)), + } + } +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cargo test -p aura-registry blueprint_store_round_trips_by_content_id` +Expected: PASS. + +--- + +## Task 2: Extract `run_blueprint_member` (behaviour-preserving) + +**Files:** +- Modify: `crates/aura-cli/src/main.rs:2983-3046` (`blueprint_sweep_family` — extract its member closure) + +The member closure at 3017-3045 is lifted verbatim into a named fn so reproduction +re-runs the identical reduce-mode path. NO behaviour change: the existing keystone +`blueprint_sweep_member_equals_single_run_and_shares_topology_hash` must stay green. + +- [ ] **Step 1: Add the extracted function** + +Insert immediately above `blueprint_sweep_family` (~2982). `ParamSpec`, `Cell`, +`Source` are already in scope in this file (used by the closure today): + +```rust +/// Run one bootstrapped member of a loaded-signal sweep: the reduce-mode path +/// (`SeriesReducer` folds eq/ex, `GatedRecorder` retains the gated R rows — the +/// O(cycles)→O(trades) fold), shared by the live sweep AND reproduction so a +/// reproduced member re-derives bit-identically (C1). `signal` is a freshly +/// reloaded blueprint (`Composite` is `!Clone`); `point` is the member's bound +/// cells; `space` gives the by-name manifest params; `topo` the shared signal hash. +#[allow(clippy::too_many_arguments)] +fn run_blueprint_member( + signal: Composite, + point: &[Cell], + space: &[ParamSpec], + sources: Vec>, + window: (Timestamp, Timestamp), + seed: u64, + pip: f64, + topo: &str, +) -> RunReport { + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_ex, rx_ex) = mpsc::channel(); + let (tx_r, rx_r) = mpsc::channel(); + let (tx_req, _rx_req) = mpsc::channel(); + let mut h = wrap_stage1r(signal, tx_eq, tx_ex, tx_r, tx_req, false, true, None) + .bootstrap_with_cells(point) + .expect("member bootstraps (point kind-checked against param_space)"); + h.run(sources); + let named = zip_params(space, point); // by-name params for the manifest record + let mut manifest = sim_optimal_manifest(named, window, seed, pip); + manifest.broker = stage1_r_broker_label(pip); + manifest.topology_hash = Some(topo.to_string()); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let (total_pips, max_drawdown) = rx_eq + .try_iter() + .next() + .map(|(_, row)| (row[0].as_f64(), row[1].as_f64())) + .unwrap_or((0.0, 0.0)); + let bias_sign_flips = + rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); + let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; + m.r = Some(summarize_r(&r_rows, &[])); + RunReport { manifest, metrics: m } +} +``` + +- [ ] **Step 2: Rewrite the sweep closure to call it** + +Replace the body of `binder.sweep(|point| { … })` at 3017-3045 with: + +```rust + binder.sweep(|point| { + // fresh per-member graph (Composite is !Clone, reload per member) run through + // the shared reduce-mode member path — the same fn reproduction re-runs. + run_blueprint_member(reload(doc), point, &space, data.run_sources(), window, 0, pip, &topo) + }) +``` + +- [ ] **Step 3: Build** + +Run: `cargo build -p aura-cli` +Expected: 0 errors. + +- [ ] **Step 4: Verify the extraction preserved behaviour** + +Run: `cargo test -p aura-cli --bin aura blueprint_sweep_member_equals_single_run_and_shares_topology_hash` +Expected: PASS (the reduce member still equals the full single run at fast=2/slow=4 — the extraction changed no behaviour). + +--- + +## Task 3: Store the canonical blueprint at the persist seam + +**Files:** +- Modify: `crates/aura-cli/src/main.rs:3063-3082` (`run_blueprint_sweep`) + +- [ ] **Step 1: Add the store write before `append_family`** + +Replace `run_blueprint_sweep`'s body (3064-3081) with (the `let _ = persist;` line and +the append/print tail are unchanged; only the `reg` binding + the put are new): + +```rust + let _ = persist; // reserved for the deferred per-member trace path; the family record below is unconditional + let family = blueprint_sweep_family(doc, axes, &data).unwrap_or_else(|e| { + eprintln!("aura: {e:?}"); + std::process::exit(2); + }); + let reg = default_registry(); + // Store the canonical blueprint ONCE, keyed by the family's shared topology_hash — + // exactly the bytes whose SHA256 the members carry (#164 byte-canonical, round-trip + // idempotent). One stored topology per family (C18/C11/C12). + let topo = family.points[0] + .report + .manifest + .topology_hash + .clone() + .expect("a blueprint sweep stamps every member's topology_hash"); + let canonical = blueprint_to_json( + &blueprint_from_json(doc, &|t| std_vocabulary(t)) + .expect("doc parse-validated at the dispatch boundary"), + ) + .expect("a loaded blueprint re-serializes"); + reg.put_blueprint(&topo, &canonical).unwrap_or_else(|e| { + eprintln!("aura: {e}"); + std::process::exit(2); + }); + // Record the family unconditionally (C18/C21 lineage), exactly like `run_sweep`. + let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) { + Ok(id) => id, + Err(e) => { + eprintln!("aura: {e}"); + std::process::exit(2); + } + }; + for pt in &family.points { + println!("{}", family_member_line(&id, &pt.report)); + } +``` + +- [ ] **Step 2: Build + verify existing sweep behaviour unperturbed** + +Run: `cargo build -p aura-cli` +Expected: 0 errors. + +- [ ] **Step 3: Verify the existing blueprint-sweep E2E still passes** + +Run: `cargo test -p aura-cli --test cli_run aura_sweep_loads_a_blueprint_and_persists_a_sweep_family` +Expected: PASS (the store write is additive + stdout-silent; the persisted family + printed lines are unchanged). + +--- + +## Task 4: Reproduce path + dispatch arm + +**Files:** +- Modify: `crates/aura-cli/src/main.rs` (add `ReproduceReport`, `point_from_params`, + `reproduce_family_in`, `reproduce_family`, the `["reproduce", id]` arm ~3853) +- Test: `crates/aura-cli/src/main.rs` (`#[cfg(test)] mod tests`) + +- [ ] **Step 1: Write the failing test** + +Add to `#[cfg(test)] mod tests` (mirrors the keystone setup at 4986-4997, adds a temp +registry + the persist + reproduce round-trip). The grid includes fast=2/slow=4, the +open-at-end param (`n_open_at_end=1`): + +```rust +#[test] +fn reproduce_family_re_derives_every_member_bit_identically() { + // a unique temp runs store so the on-disk family + blueprint store do not collide. + let dir = std::env::temp_dir().join(format!("aura-repro-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("temp dir"); + let reg = Registry::open(dir.join("runs.jsonl")); + + let open = stage1_signal(None, None); + let doc = blueprint_to_json(&open).expect("serializes"); + let data = DataSource::Synthetic; + // 2x grid over slow.length {4,6} at fast=2 — slow=4 is the open-at-end member. + let axes = vec![ + ("stage1_signal.fast.length".to_string(), vec![Scalar::i64(2)]), + ("stage1_signal.slow.length".to_string(), vec![Scalar::i64(4), Scalar::i64(6)]), + ]; + let family = blueprint_sweep_family(&doc, &axes, &data).expect("axes resolve"); + + // persist exactly as run_blueprint_sweep does: store the blueprint, append the family. + let topo = family.points[0].report.manifest.topology_hash.clone().expect("topo"); + let canonical = + blueprint_to_json(&blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap()).unwrap(); + reg.put_blueprint(&topo, &canonical).expect("store blueprint"); + let id = reg + .append_family("repro", FamilyKind::Sweep, &sweep_member_reports(&family)) + .expect("append"); + + // reproduce: every member re-derives bit-identically (incl the open-at-end member). + let rep = reproduce_family_in(®, &id, &data); + assert_eq!(rep.outcomes.len(), 2, "two members reproduced"); + assert!( + rep.outcomes.iter().all(|(_, ok)| *ok), + "every member re-derives bit-identically: {:?}", + rep.outcomes + ); + + let _ = std::fs::remove_dir_all(&dir); +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cargo test -p aura-cli --bin aura reproduce_family_re_derives_every_member_bit_identically` +Expected: FAIL — compile error (`cannot find function reproduce_family_in` / `ReproduceReport`). + +- [ ] **Step 3: Write the reproduce path** + +Add near `runs_family` (~2470). `group_families`, `load_family_members`, `Cell`, +`ParamSpec`, `Scalar`, `topology_hash`, `blueprint_from_json`, `std_vocabulary` are all +in scope: + +```rust +/// The outcome of reproducing one persisted family: per member, whether its re-run +/// metrics are bit-identical to the stored metrics (C1). +struct ReproduceReport { + outcomes: Vec<(String, bool)>, +} + +/// Reconstruct a member's bootstrap point from its recorded named params — the inverse +/// of `zip_params(space, point)`. Walks the reloaded signal's `param_space` in order +/// (deterministic for the same blueprint) and reads each knob's value from the manifest. +fn point_from_params(space: &[ParamSpec], params: &[(String, Scalar)]) -> Vec { + space + .iter() + .map(|ps| { + params + .iter() + .find(|(n, _)| n == &ps.name) + .map(|(_, s)| s.cell()) + .unwrap_or_else(|| panic!("manifest is missing param {}", ps.name)) + }) + .collect() +} + +/// Re-derive every member of a persisted sweep family from the content-addressed store +/// and compare to the stored result, against an explicit registry (testable seam). +fn reproduce_family_in(reg: &Registry, id: &str, data: &DataSource) -> ReproduceReport { + let members = reg.load_family_members().unwrap_or_else(|e| { + eprintln!("aura: {e}"); + std::process::exit(2); + }); + let Some(family) = group_families(members).into_iter().find(|f| f.id == id) else { + // reproduce is an action, not a lookup: an unknown id is a hard error (distinct + // from `runs family `'s treat-as-empty exit 0). + eprintln!("aura: no such family '{id}'"); + std::process::exit(2); + }; + let pip = data.pip_size(); + let window = data.full_window(); + let mut outcomes = Vec::new(); + for member in &family.members { + let stored = &member.report; + let hash = stored.manifest.topology_hash.clone().unwrap_or_else(|| { + eprintln!("aura: family member has no topology_hash; not a generated run"); + std::process::exit(2); + }); + let doc = reg + .get_blueprint(&hash) + .unwrap_or_else(|e| { + eprintln!("aura: {e}"); + std::process::exit(2); + }) + .unwrap_or_else(|| { + eprintln!("aura: blueprint {hash} missing from store"); + std::process::exit(2); + }); + let signal = blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap_or_else(|e| { + eprintln!("aura: stored blueprint {hash} does not parse: {e:?}"); + std::process::exit(2); + }); + let space = signal.param_space(); + let point = point_from_params(&space, &stored.manifest.params); + let label = stored + .manifest + .params + .iter() + .map(|(n, v)| format!("{n}={v:?}")) + .collect::>() + .join(", "); + let rerun = run_blueprint_member( + signal, + &point, + &space, + data.run_sources(), + window, + stored.manifest.seed, + pip, + &hash, + ); + outcomes.push((label, rerun.metrics == stored.metrics)); + } + ReproduceReport { outcomes } +} + +/// `aura reproduce `: re-derive a persisted sweep family from the +/// content-addressed blueprint store and verify each member reproduces bit-identically +/// (C18 "re-derives full results on demand"). Synthetic data this cycle (deterministic); +/// recorded-dataset reproduction rides the DataServer seam (#124). +fn reproduce_family(id: &str) { + let rep = reproduce_family_in(&default_registry(), id, &DataSource::Synthetic); + let total = rep.outcomes.len(); + let ok = rep.outcomes.iter().filter(|(_, b)| *b).count(); + for (label, identical) in &rep.outcomes { + let verdict = if *identical { "bit-identical" } else { "DIVERGED" }; + println!("{id} member {label} reproduced: {verdict}"); + } + println!("reproduced {ok}/{total} members bit-identically"); + if ok != total { + std::process::exit(1); + } +} +``` + +- [ ] **Step 4: Add the dispatch arm** + +In the top-level `match` (~3736), beside `["runs", "family", id]` (~3853), add: + +```rust + ["reproduce", id] => reproduce_family(id), +``` + +- [ ] **Step 5: Run test to verify it passes** + +Run: `cargo test -p aura-cli --bin aura reproduce_family_re_derives_every_member_bit_identically` +Expected: PASS (every member, incl the open-at-end one, re-derives bit-identically). + +--- + +## Task 5: End-to-end CLI coverage + +**Files:** +- Test: `crates/aura-cli/tests/cli_run.rs` (append two tests, mirroring + `aura_sweep_loads_a_blueprint_and_persists_a_sweep_family` at 3269-3288) + +- [ ] **Step 1: Write the E2E tests** + +Append to `crates/aura-cli/tests/cli_run.rs` (`BIN` at :8, `temp_cwd` at :13): + +```rust +/// E2E (acc 2): persist a blueprint sweep, then `aura reproduce ` re-derives every +/// member bit-identically from the content-addressed store and exits 0. +#[test] +fn aura_reproduce_re_derives_a_persisted_sweep_bit_identically() { + let cwd = temp_cwd("reproduce_roundtrip"); + let fixture = format!("{}/tests/fixtures/stage1_signal_open.json", env!("CARGO_MANIFEST_DIR")); + + let sweep = std::process::Command::new(BIN) + .args([ + "sweep", + &fixture, + "--axis", + "stage1_signal.fast.length=2", + "--axis", + "stage1_signal.slow.length=4,6", + "--name", + "smacross", + ]) + .current_dir(&cwd) + .output() + .expect("run aura sweep"); + assert!(sweep.status.success(), "sweep ok: {}", String::from_utf8_lossy(&sweep.stderr)); + + let repro = std::process::Command::new(BIN) + .args(["reproduce", "smacross-1"]) + .current_dir(&cwd) + .output() + .expect("run aura reproduce"); + let stdout = String::from_utf8_lossy(&repro.stdout); + assert!(repro.status.success(), "reproduce exits 0: {}", String::from_utf8_lossy(&repro.stderr)); + assert!( + stdout.contains("reproduced 2/2 members bit-identically"), + "every member re-derives: {stdout}" + ); + + let _ = std::fs::remove_dir_all(&cwd); +} + +/// E2E (negative): `aura reproduce ` is a hard error (exit non-zero + named +/// cause on stderr), distinct from `runs family `'s treat-as-empty exit 0. +#[test] +fn aura_reproduce_rejects_an_unknown_family() { + let cwd = temp_cwd("reproduce_unknown"); + let out = std::process::Command::new(BIN) + .args(["reproduce", "no-such-family-1"]) + .current_dir(&cwd) + .output() + .expect("run aura reproduce"); + assert!(!out.status.success(), "unknown family exits non-zero"); + assert!( + String::from_utf8_lossy(&out.stderr).contains("no such family 'no-such-family-1'"), + "names the cause: {}", + String::from_utf8_lossy(&out.stderr) + ); + let _ = std::fs::remove_dir_all(&cwd); +} +``` + +- [ ] **Step 2: Build the binary + run the E2E** + +Run: `cargo test -p aura-cli --test cli_run aura_reproduce_re_derives_a_persisted_sweep_bit_identically` +Expected: PASS. + +- [ ] **Step 3: Run the negative E2E** + +Run: `cargo test -p aura-cli --test cli_run aura_reproduce_rejects_an_unknown_family` +Expected: PASS. + +- [ ] **Step 4: Full workspace gate** + +Run: `cargo test --workspace` +Expected: PASS (no regressions). + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: clean. + +--- + +## Notes for the implementer + +- **`member.report` hop:** a grouped family's members are `FamilyRunRecord`; the + payload is `member.report` (a `RunReport`), so fields are `member.report.manifest.*` + / `member.report.metrics` (the spec pseudo wrote `member.manifest.*` for brevity). +- **`RMetrics` equality excludes `trade_rs`** (hand-written `PartialEq`, + aura-analysis/src/lib.rs:111-124; `trade_rs` is `#[serde(skip)]`). A stored member's + empty `trade_rs` vs. a re-run's populated one therefore does NOT break + `rerun.metrics == stored.metrics` — this is what makes the compare well-defined. +- **`Scalar::cell()`** (aura-core/src/scalar.rs:89) is the `Scalar → Cell` conversion for + `point_from_params`; the forward `zip_params` lives at aura-core/src/node.rs:83. +- **Seed:** the sweep records seed 0 (`run_blueprint_member(..., 0, ...)` from the sweep + call site); reproduction threads `stored.manifest.seed` (= 0) through the same fn. +- Do NOT add `graph introspect --content-id` or the Tier-1 pin here — they are plan 0094b.