Brummel/Aura
1
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases Wiki Activity

plan: 0094 content-addressed reproduction (iteration 1)

Browse Source 
Iteration 1 (acceptance 2, the heart): the content-addressed blueprint store
(Registry::put_blueprint/get_blueprint under runs/blueprints/), the store write
at the sweep persist seam, the extracted shared run_blueprint_member, and
`aura reproduce <family-id>` re-deriving every persisted member bit-identically
from the stored blueprint. acceptance 1 (graph introspect --content-id) and 3
(Tier-1 id stability) deferred to plan 0094b.

refs #158
This commit is contained in:
Brummel
2026-07-01 01:29:34 +02:00
parent 4b6bc52794
commit a6d3774d04
1 changed files with 541 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/0094-content-addressed-reproduction.md
+541
View File
@@ -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
<family-id>`, 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/<hash>.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<std::io::Error>` (531-535), so `?` auto-maps:
```rust
/// The content-addressed blueprint store dir — a sibling of the runs store,
/// `<runs.jsonl>.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/<hash>.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<Option<String>, 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<Box<dyn aura_engine::Source>>,
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<Scalar>)> = 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(&reg, &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<Cell> {
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 <id>`'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::<Vec<_>>()
.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 <family-id>`: 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 <id>` 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 <unknown>` is a hard error (exit non-zero + named
/// cause on stderr), distinct from `runs family <id>`'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.