Aura/docs/plans/0109-persist-taps.md

# persist_taps Wiring — Implementation Plan (cycle 0109)

> **Parent spec:** `docs/specs/0109-persist-taps.md`
>
> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run
> this plan. Steps use `- [ ]` checkboxes for tracking.

**Goal:** `aura campaign run` honours `presentation.persist_taps`: closed tap
vocabulary validated intrinsically, nominee-only non-reduce re-run per cell
(METRICS-equality C1 guard), traces via the existing TraceStore under
`{campaign8}-{run}`, one sparse `trace_name` pointer on the record.

**Architecture:** Task 1 lands the vocabulary + UnknownTap validation
(aura-research + CLI twins; the deferral eprintln STAYS until Task 3; the old
deferral test splits into its two 0109 pins). Task 2 widens the record and
lands the claim-sentinel → composed-name contract across registry and
executor. Task 3 ships `persist_campaign_traces` (the consumer wiring; the
deferral line dies; the Task-1(b) seam test tightens). Task 4 extends the
gated real-data e2e (trace files + chart read-back).

**Tech Stack:** Rust workspace — aura-research, aura-registry, aura-campaign,
aura-cli.

**Task-order constraints:** strictly sequential 1→4. Task 3's old-fences in
campaign_run.rs are valid because Tasks 1-2 do not touch that file; Task 3
finds the Task-1(b) test by its fixture (`campaign_doc_json(.., (1,2),
"\"equity\"", "")`), not by name.

---
### Task 1: Tap vocabulary + UnknownTap intrinsic validation (aura-research + aura-cli twins)

The tap namespace becomes a **closed vocabulary** (#201 decision 1, spec `docs/specs/0109-persist-taps.md`): `aura-research` gains `tap_vocabulary()` + `DocFault::UnknownTap` in `validate_campaign`, a new `SlotKind::TapKinds` for the `persist_taps` slot (mirroring `EmitKinds`), and the open-slot hint names the vocabulary. `aura-cli` gains the `doc_fault_prose` twin arm. The 0107 deferral seam test `campaign_run_persist_taps_deferred_loudly` splits into two pins.

**Sequencing decision — state loudly:** this task **keeps** the loud-deferral `eprintln!` in `crates/aura-cli/src/campaign_run.rs` (lines 360–368) untouched. It still prints on any run with non-empty valid `persist_taps`. Therefore the new test (b) below asserts **nothing about the deferral line, in either direction** — it pins only validate-pass + the member-data-seam refusal + the absence of any trace output. Task 3 of this plan removes the `eprintln!` and tightens test (b) to pin its absence. Do not remove the `eprintln!` in this task; do not assert on `"not yet honored"` in test (b).

**Files:**
- `crates/aura-research/src/lib.rs`
- `crates/aura-cli/src/research_docs.rs`
- `crates/aura-cli/tests/research_docs.rs`

- [ ] **Step 1: RED — aura-research unit tests (vocabulary, fault, hint label, open-slot hint)**

  In `crates/aura-research/src/lib.rs`, inside the `#[cfg(test)] mod tests` module, insert four new tests immediately after the closing brace of `validate_campaign_accepts_the_fixture_and_reports_each_fault`. Anchor (byte-exact, the test's last lines):

  old:
  ```rust
        let mut emit = ok.clone();
        emit.presentation.emit.push("pie_chart".into());
        assert!(validate_campaign(&emit).contains(&DocFault::UnknownEmitKind("pie_chart".into())));
    }
  ```
  new:
  ```rust
        let mut emit = ok.clone();
        emit.presentation.emit.push("pie_chart".into());
        assert!(validate_campaign(&emit).contains(&DocFault::UnknownEmitKind("pie_chart".into())));
    }

    /// #201 decision 1 (cycle 0109): the tap namespace is a CLOSED vocabulary —
    /// the wrap convention's four persisted sink names. A genuinely new
    /// observable is a new vocabulary entry (or an authored blueprint sink),
    /// never an open node-path namespace in the document.
    #[test]
    fn tap_vocabulary_is_the_wrap_conventions_four_sink_names() {
        assert_eq!(tap_vocabulary(), ["equity", "exposure", "r_equity", "net_r_equity"]);
    }

    /// Each `presentation.persist_taps` entry outside `tap_vocabulary()` is an
    /// intrinsic fault carrying its index (for path-addressed prose downstream)
    /// and the offending name — mirroring the UnknownEmitKind loop. The
    /// fixture's own "net_r_equity" tap is in-vocabulary, so the clean-fixture
    /// assert in `validate_campaign_accepts_the_fixture_and_reports_each_fault`
    /// stays green unchanged.
    #[test]
    fn validate_campaign_refuses_unknown_taps_by_index() {
        let ok = parse_campaign(CAMPAIGN_FIXTURE).unwrap();
        let mut tap = ok.clone();
        tap.presentation.persist_taps.push("bias".into());
        assert_eq!(
            validate_campaign(&tap),
            vec![DocFault::UnknownTap { index: 1, tap: "bias".into() }]
        );
        // every in-vocabulary name passes
        let mut all = ok.clone();
        all.presentation.persist_taps = tap_vocabulary().iter().map(|t| t.to_string()).collect();
        assert_eq!(validate_campaign(&all), Vec::new());
    }

    /// The `std::presentation` persist_taps slot advertises the closed tap
    /// vocabulary through the describe path (`describe_block` +
    /// `slot_kind_label`); the label is cross-pinned against `tap_vocabulary()`
    /// so the static label and the validator cannot drift apart.
    #[test]
    fn persist_taps_slot_advertises_the_tap_vocabulary() {
        let block = describe_block("std::presentation").expect("presentation describable");
        let slot = block
            .slots
            .iter()
            .find(|s| s.name == "persist_taps")
            .expect("presentation has a persist_taps slot");
        assert_eq!(slot.kind, SlotKind::TapKinds);
        assert_eq!(
            slot_kind_label(slot.kind),
            format!("list of: {}", tap_vocabulary().join(" | "))
        );
        assert_eq!(
            slot_kind_label(slot.kind),
            "list of: equity | exposure | r_equity | net_r_equity"
        );
    }

    /// A draft missing its `presentation` section is pointed at the section
    /// AND the closed tap vocabulary (the guide never advertises an open
    /// namespace the validator would refuse).
    #[test]
    fn presentation_open_slot_hint_names_the_tap_vocabulary() {
        let draft = r#"{ "format_version": 1, "kind": "campaign", "name": "draft",
            "data": { "instruments": ["GER40"], "windows": [ { "from_ms": 1, "to_ms": 2 } ] },
            "strategies": [ { "ref": { "content_id": "9f3a" },
                              "axes": { "slow": { "kind": "I64", "values": [10] } } } ],
            "process": { "ref": { "content_id": "4e2d" } },
            "seed": 1 }"#;
        let slots = open_slots_campaign(draft).unwrap();
        assert_eq!(
            slots,
            vec![open(
                "presentation",
                "required section: persist_taps (equity | exposure | r_equity | net_r_equity) + emit",
            )]
        );
    }
  ```

  Run: `cargo test -p aura-research`
  Expected: **compilation failure of the test target** — errors naming `tap_vocabulary` (E0425, cannot find function), `UnknownTap` (E0599, no variant on `DocFault`), and `TapKinds` (E0599, no variant on `SlotKind`). This compile-fail IS the RED evidence: tests of a not-yet-existing API cannot fail at runtime. Do not "fix" this here; Step 3 turns it green.

- [ ] **Step 2: RED — split the 0107 deferral seam test into the two 0109 pins (aura-cli)**

  In `crates/aura-cli/tests/research_docs.rs`, replace the entire `campaign_run_persist_taps_deferred_loudly` test (doc comment included) with the two new tests. The old deferral-prose assert dies here.

  old:
  ```rust
/// Non-empty persist_taps defers LOUDLY, and the note's ORDER is pinned: it
/// prints before member execution, so it is asserted here on a run
/// that refuses at the member-data seam. The [1, 2] epoch-ms window (1970)
/// makes that refusal deterministic on every machine: a data-less host
/// refuses on missing geometry, a data-ful host on a window no archive file
/// overlaps — exit 1 either way, tap note already on stderr.
#[test]
fn campaign_run_persist_taps_deferred_loudly() {
    let _fixture = project_lock();
    let dir = built_project();
    let runs_dir = dir.join("runs");
    std::fs::remove_dir_all(&runs_dir).ok();
    let _cleanup = ScratchGuard(vec![
        ScratchPath::Dir(runs_dir.clone()),
        ScratchPath::File(dir.join("taps.process.json")),
        ScratchPath::File(dir.join("taps.campaign.json")),
    ]);
    let bp_id = seed_blueprint(dir, "campaign-run-taps-seed");
    let proc_id = register_process_doc(dir, "taps.process.json", SWEEP_ONLY_PROCESS_DOC);
    write_doc(
        dir,
        "taps.campaign.json",
        &campaign_doc_json(&bp_id, &proc_id, (1, 2), "\"r_record\"", ""),
    );
    let (out, code) = run_code_in(dir, &["campaign", "run", "taps.campaign.json"]);
    assert_eq!(code, Some(1), "stdout/stderr: {out}");
    assert!(
        out.contains("aura: persist_taps not yet honored (1 tap(s) ignored)"),
        "the tap note must precede the member-data refusal: {out}"
    );
    assert!(
        out.contains("no recorded geometry") || out.contains("no data for instrument"),
        "the refusal names the data condition: {out}"
    );
}
  ```
  new:
  ```rust
/// 0109 split (a) — the tap namespace is CLOSED: the 0107 deferral fixture's
/// made-up "r_record" tap now refuses INTRINSICALLY, exit 1, path-addressed
/// and vocabulary-enumerating. The old test drove `campaign run` on a file
/// target, so this pin drives the same path: the refusal fires at the run's
/// parse-valid gate (`parse_valid_campaign` — the same `validate_campaign` +
/// `doc_fault_prose` composition `campaign validate` shares), before
/// registration, before the referential tier, before any member runs — and
/// therefore before the (Task-3-doomed) loud-deferral note, whose absence is
/// pinned here.
#[test]
fn campaign_run_refuses_unknown_tap_at_validate() {
    let _fixture = project_lock();
    let dir = built_project();
    let runs_dir = dir.join("runs");
    std::fs::remove_dir_all(&runs_dir).ok();
    let _cleanup = ScratchGuard(vec![
        ScratchPath::Dir(runs_dir.clone()),
        ScratchPath::File(dir.join("badtap.process.json")),
        ScratchPath::File(dir.join("badtap.campaign.json")),
    ]);
    let bp_id = seed_blueprint(dir, "campaign-run-badtap-seed");
    let proc_id = register_process_doc(dir, "badtap.process.json", SWEEP_ONLY_PROCESS_DOC);
    write_doc(
        dir,
        "badtap.campaign.json",
        &campaign_doc_json(&bp_id, &proc_id, (1, 2), "\"r_record\"", ""),
    );
    let (out, code) = run_code_in(dir, &["campaign", "run", "badtap.campaign.json"]);
    assert_eq!(code, Some(1), "stdout/stderr: {out}");
    assert!(out.contains("campaign document invalid:"), "stdout/stderr: {out}");
    assert!(
        out.contains(
            "presentation.persist_taps[0]: unknown tap \"r_record\" \
             (taps: equity | exposure | r_equity | net_r_equity)"
        ),
        "the refusal is path-addressed and enumerates the vocabulary: {out}"
    );
    assert!(
        !out.contains("not yet honored"),
        "an invalid document must never reach the deferral note: {out}"
    );
    assert!(!out.contains("UnknownTap"), "Debug leak: {out}");
}

/// 0109 split (b) — the control: an in-vocabulary tap ("equity") sails
/// through the new intrinsic gate (no UnknownTap prose) and the run proceeds
/// to refuse at the member-data seam, exactly as a tap-less run over the
/// [1, 2] 1970 epoch-ms window does (data-less host: missing geometry;
/// data-ful host: no archive file overlaps) — exit 1 either way, and no
/// trace output (tracing never starts on a run that dies before members;
/// there is no "traces persisted" summary line). NOTE: the 0107
/// loud-deferral eprintln still fires in this task's tree and is
/// deliberately NOT asserted either way here — Task 3 deletes it and
/// tightens this test to pin its absence.
#[test]
fn campaign_run_valid_tap_reaches_the_member_data_seam() {
    let _fixture = project_lock();
    let dir = built_project();
    let runs_dir = dir.join("runs");
    std::fs::remove_dir_all(&runs_dir).ok();
    let _cleanup = ScratchGuard(vec![
        ScratchPath::Dir(runs_dir.clone()),
        ScratchPath::File(dir.join("taps.process.json")),
        ScratchPath::File(dir.join("taps.campaign.json")),
    ]);
    let bp_id = seed_blueprint(dir, "campaign-run-taps-seed");
    let proc_id = register_process_doc(dir, "taps.process.json", SWEEP_ONLY_PROCESS_DOC);
    write_doc(
        dir,
        "taps.campaign.json",
        &campaign_doc_json(&bp_id, &proc_id, (1, 2), "\"equity\"", ""),
    );
    let (out, code) = run_code_in(dir, &["campaign", "run", "taps.campaign.json"]);
    assert_eq!(code, Some(1), "stdout/stderr: {out}");
    assert!(
        !out.contains("unknown tap"),
        "\"equity\" is in the closed vocabulary and must pass validate: {out}"
    );
    assert!(
        out.contains("no recorded geometry") || out.contains("no data for instrument"),
        "the refusal names the data condition: {out}"
    );
    assert!(
        !out.contains("traces persisted"),
        "no trace summary on a run that never reached a member: {out}"
    );
}
  ```

  Run: `cargo test -p aura-cli --test research_docs campaign_run_refuses_unknown_tap_at_validate`
  Expected: compiles, **1 test FAILS** (RED). Today "r_record" passes validation, so the run refuses at the member-data seam instead — the `"campaign document invalid:"` and unknown-tap-prose asserts fail, and the deferral line is present (failing the `!contains("not yet honored")` assert).

  Run: `cargo test -p aura-cli --test research_docs campaign_run_valid_tap_reaches_the_member_data_seam`
  Expected: **1 test PASSES already**. This is deliberate and not a defect: test (b) is the control half of the split — it guards the gate Step 3 introduces against over-rejection ("equity" must keep reaching the data seam once unknown taps refuse). Its RED counterpart is test (a).

- [ ] **Step 3: GREEN — aura-research: vocabulary, fault, validation loop, TapKinds, hint**

  Seven edits in `crates/aura-research/src/lib.rs`.

  (1) Add `tap_vocabulary` directly below `emit_vocabulary`:

  old:
  ```rust
/// The closed v1 emit vocabulary (data-level presentation outputs).
pub fn emit_vocabulary() -> &'static [&'static str] {
    &["family_table", "selection_report"]
}
  ```
  new:
  ```rust
/// The closed v1 emit vocabulary (data-level presentation outputs).
pub fn emit_vocabulary() -> &'static [&'static str] {
    &["family_table", "selection_report"]
}

/// The wrap convention's persisted sink names — the closed tap vocabulary
/// (#201 decision 1). A genuinely new observable becomes a new entry or an
/// authored sink in the strategy blueprint (C22: the choice of sinks is part
/// of the experiment) — never an open node-path namespace in the document.
pub fn tap_vocabulary() -> &'static [&'static str] {
    &["equity", "exposure", "r_equity", "net_r_equity"]
}
  ```

  (2) Add the `DocFault` variant (campaign side, after `UnknownEmitKind`):

  old:
  ```rust
    EmptyAxes { strategy: usize },
    EmptyAxis { strategy: usize, axis: String },
    UnknownEmitKind(String),
    ProcessRefMustBeContentId,
}
  ```
  new:
  ```rust
    EmptyAxes { strategy: usize },
    EmptyAxis { strategy: usize, axis: String },
    UnknownEmitKind(String),
    UnknownTap { index: usize, tap: String },
    ProcessRefMustBeContentId,
}
  ```

  (3) Add the taps loop in `validate_campaign`, after the emit loop:

  old:
  ```rust
    for e in &doc.presentation.emit {
        if !emit_vocabulary().contains(&e.as_str()) {
            faults.push(DocFault::UnknownEmitKind(e.clone()));
        }
    }
    faults
}
  ```
  new:
  ```rust
    for e in &doc.presentation.emit {
        if !emit_vocabulary().contains(&e.as_str()) {
            faults.push(DocFault::UnknownEmitKind(e.clone()));
        }
    }
    for (i, t) in doc.presentation.persist_taps.iter().enumerate() {
        if !tap_vocabulary().contains(&t.as_str()) {
            faults.push(DocFault::UnknownTap { index: i, tap: t.clone() });
        }
    }
    faults
}
  ```

  (4) Add the `SlotKind` variant (mirror `EmitKinds`):

  old:
  ```rust
    Axes,
    EmitKinds,
}
  ```
  new:
  ```rust
    Axes,
    EmitKinds,
    /// The closed tap vocabulary (`tap_vocabulary()` entries).
    TapKinds,
}
  ```

  (5) Retype the `persist_taps` slot in `CAMPAIGN_SECTIONS` (`std::presentation`):

  old:
  ```rust
            SlotInfo { name: "persist_taps", kind: SlotKind::Strings, required: true },
  ```
  new:
  ```rust
            SlotInfo { name: "persist_taps", kind: SlotKind::TapKinds, required: true },
  ```

  (6) Add the `slot_kind_label` arm:

  old:
  ```rust
        SlotKind::EmitKinds => "list of: family_table | selection_report",
    }
}
  ```
  new:
  ```rust
        SlotKind::EmitKinds => "list of: family_table | selection_report",
        SlotKind::TapKinds => "list of: equity | exposure | r_equity | net_r_equity",
    }
}
  ```

  (7) Name the vocabulary in the `open_slots_campaign` presentation hint:

  old:
  ```rust
    if v.get("presentation").is_none() {
        slots.push(open("presentation", "required section: persist_taps + emit"));
    }
  ```
  new:
  ```rust
    if v.get("presentation").is_none() {
        slots.push(open(
            "presentation",
            "required section: persist_taps (equity | exposure | r_equity | net_r_equity) + emit",
        ));
    }
  ```

  Run: `cargo test -p aura-research`
  Expected: compiles, **all tests pass** — including the four Step-1 tests (`tap_vocabulary_is_the_wrap_conventions_four_sink_names`, `validate_campaign_refuses_unknown_taps_by_index`, `persist_taps_slot_advertises_the_tap_vocabulary`, `presentation_open_slot_hint_names_the_tap_vocabulary`) and the untouched `validate_campaign_accepts_the_fixture_and_reports_each_fault` (the fixture's "net_r_equity" tap is in-vocabulary).
  NOTE: at this point `cargo build --workspace` FAILS with E0004 (non-exhaustive match): `doc_fault_prose` in `crates/aura-cli/src/research_docs.rs` does not yet cover `UnknownTap`. That is expected mid-task; Step 4 restores the workspace build. (`doc_fault_prose` is the only exhaustive `DocFault` consumer in the workspace — aura-campaign mentions `DocFault` only in a comment.)

- [ ] **Step 4: GREEN — aura-cli prose twin (`doc_fault_prose` arm + unit test)**

  Three edits in `crates/aura-cli/src/research_docs.rs`.

  (1) Import `tap_vocabulary`:

  old:
  ```rust
use aura_research::{
    campaign_to_json, campaign_vocabulary, describe_block, open_slots_campaign,
    open_slots_process, parse_campaign, parse_process, process_to_json, process_vocabulary,
    slot_kind_label, validate_campaign, validate_process, CampaignDoc, DocError, DocFault,
    DocKind, ProcessDoc, StageBlock,
};
  ```
  new:
  ```rust
use aura_research::{
    campaign_to_json, campaign_vocabulary, describe_block, open_slots_campaign,
    open_slots_process, parse_campaign, parse_process, process_to_json, process_vocabulary,
    slot_kind_label, tap_vocabulary, validate_campaign, validate_process, CampaignDoc, DocError,
    DocFault, DocKind, ProcessDoc, StageBlock,
};
  ```

  (2) Add the prose arm (vocabulary composed from `tap_vocabulary()` — one source, no third copy of the four names):

  old:
  ```rust
        DocFault::UnknownEmitKind(kind) => format!("presentation.emit: unknown kind \"{kind}\""),
        DocFault::ProcessRefMustBeContentId => {
            "process.ref: a process is referenced by content id in this version".into()
        }
  ```
  new:
  ```rust
        DocFault::UnknownEmitKind(kind) => format!("presentation.emit: unknown kind \"{kind}\""),
        DocFault::UnknownTap { index, tap } => format!(
            "presentation.persist_taps[{index}]: unknown tap \"{tap}\" (taps: {})",
            tap_vocabulary().join(" | ")
        ),
        DocFault::ProcessRefMustBeContentId => {
            "process.ref: a process is referenced by content id in this version".into()
        }
  ```

  (3) Pin the prose in the file's `mod tests`, after the last existing test:

  old:
  ```rust
    #[test]
    fn zero_walk_forward_length_prose_is_path_addressed_and_debug_free() {
        let prose = doc_fault_prose(&DocFault::ZeroWalkForwardLength { stage: 2, field: "step_ms" });
        assert_eq!(prose, "pipeline[2]: walk_forward step_ms must be > 0");
        assert!(!prose.contains("ZeroWalkForwardLength"), "Debug leak: {prose}");
    }
}
  ```
  new:
  ```rust
    #[test]
    fn zero_walk_forward_length_prose_is_path_addressed_and_debug_free() {
        let prose = doc_fault_prose(&DocFault::ZeroWalkForwardLength { stage: 2, field: "step_ms" });
        assert_eq!(prose, "pipeline[2]: walk_forward step_ms must be > 0");
        assert!(!prose.contains("ZeroWalkForwardLength"), "Debug leak: {prose}");
    }

    #[test]
    /// #201 decision 1 (spec 0109): the unknown-tap refusal is path-addressed
    /// (`presentation.persist_taps[i]`) and enumerates the closed vocabulary
    /// inline — four names are small enough to print, so no introspection
    /// pointer is needed (contrast the 17-name metric roster, which points at
    /// `aura process introspect --metrics` instead).
    fn unknown_tap_prose_is_path_addressed_and_enumerates_the_vocabulary() {
        let prose = doc_fault_prose(&DocFault::UnknownTap { index: 0, tap: "bias".into() });
        assert_eq!(
            prose,
            "presentation.persist_taps[0]: unknown tap \"bias\" \
             (taps: equity | exposure | r_equity | net_r_equity)"
        );
        assert!(!prose.contains("UnknownTap"), "Debug leak: {prose}");
    }
}
  ```

  Run: `cargo test -p aura-cli unknown_tap_prose`
  Expected: compiles, `unknown_tap_prose_is_path_addressed_and_enumerates_the_vocabulary` passes (1 passed; the filter matches only this test).

- [ ] **Step 5: Gates — both crates green, workspace builds**

  Run: `cargo test -p aura-research`
  Expected: green, 0 failures.

  Run: `cargo test -p aura-cli --test research_docs campaign_run`
  Expected: green, 0 failures — `campaign_run_refuses_unknown_tap_at_validate` now passes (the run refuses at the parse-valid gate with the enumerating prose, no deferral note, no Debug leak), `campaign_run_valid_tap_reaches_the_member_data_seam` still passes (the control), and every sibling `campaign_run_*` test is unchanged (`campaign_run_real_e2e_sweep_gate_walkforward` may self-skip on a data-less machine — that skip is green, not a failure).

  Run: `cargo build --workspace`
  Expected: 0 errors (the Step-3 E0004 window is closed by the Step-4 arm).
### Task 2: trace_name widening + name composition + eligibility stamp (aura-registry + aura-campaign)

**Context (cycle 0109, spec `docs/specs/0109-persist-taps.md`, #201 decision 5 + the spec's seam note):** `CampaignRunRecord` gains a sparse `trace_name: Option<String>` — the TraceStore family name a campaign realization claims when its document requests `persist_taps`. The contract: the STORED `campaign_runs.jsonl` line AND the record `execute` RETURNS both carry the same `Some("{campaign8}-{run}")` iff `campaign.presentation.persist_taps` is non-empty, else `None`. Because the per-campaign `run` counter is assigned inside `Registry::append_campaign_run`, the seam is split: `aura-campaign::execute` stamps an eligibility **claim sentinel** (`Some(String::new())`) on the record it hands to the store; `append_campaign_run` replaces any `Some` with the derived name on the stored line (`None` stays `None`); `execute` then mirrors the same derivation onto the returned copy after `record.run = run`. The field is serde-default + skip-when-`None` (the C14/C23 widening convention), so pre-0109 stored lines keep parsing and re-serialized dumps of `None`-records stay byte-identical.

**Files:**
- `crates/aura-registry/src/lib.rs` (tests: RED composition pin, pre-widening extension, literal threading)
- `crates/aura-registry/src/lineage.rs` (the field + the append-side composition)
- `crates/aura-registry/tests/campaign_run_store_e2e.rs` (literal threading)
- `crates/aura-campaign/tests/execute.rs` (RED stamp pin, `None` extension)
- `crates/aura-campaign/src/exec.rs` (the claim sentinel + the returned-copy mirror)

- [ ] **Step 1: RED — registry pins for the name composition and the pre-0109 parse.**

  In `crates/aura-registry/src/lib.rs`, at the very end of the test module, extend the existing pre-widening test and append the new composition test. Replace this exact existing code (the file's final lines — the pre-widening test plus the module-closing brace):

  ```rust
      /// A pre-0108 stored line — no `bootstrap`, no `generalizations` — still
      /// parses (the C14/C23 serde-default widening convention): existing
      /// `campaign_runs.jsonl` stores survive the record widening unchanged.
      #[test]
      fn campaign_run_line_without_new_fields_still_parses() {
          let path = temp_family_dir("campaign_run_pre_widening");
          let store = path.with_file_name("campaign_runs.jsonl");
          let line = r#"{"campaign":"cafe","process":"beef","run":0,"seed":7,"cells":[{"strategy":"3f9c","instrument":"EURUSD","window_ms":[0,1000],"stages":[{"block":"std::gate","survivor_ordinals":[0,2]}]}]}"#;
          fs::write(&store, format!("{line}\n")).expect("write pre-widening line");
          let reg = Registry::open(&path);
          let loaded = reg.load_campaign_runs().expect("pre-widening line parses");
          assert_eq!(loaded.len(), 1);
          assert_eq!(loaded[0].cells[0].stages[0].bootstrap, None);
          assert!(loaded[0].generalizations.is_empty());
      }
  }
  ```

  with:

  ```rust
      /// A pre-0109 stored line — no `bootstrap`, no `generalizations`, no
      /// `trace_name` — still parses (the C14/C23 serde-default widening
      /// convention): existing `campaign_runs.jsonl` stores survive the record
      /// widening unchanged.
      #[test]
      fn campaign_run_line_without_new_fields_still_parses() {
          let path = temp_family_dir("campaign_run_pre_widening");
          let store = path.with_file_name("campaign_runs.jsonl");
          let line = r#"{"campaign":"cafe","process":"beef","run":0,"seed":7,"cells":[{"strategy":"3f9c","instrument":"EURUSD","window_ms":[0,1000],"stages":[{"block":"std::gate","survivor_ordinals":[0,2]}]}]}"#;
          fs::write(&store, format!("{line}\n")).expect("write pre-widening line");
          let reg = Registry::open(&path);
          let loaded = reg.load_campaign_runs().expect("pre-widening line parses");
          assert_eq!(loaded.len(), 1);
          assert_eq!(loaded[0].cells[0].stages[0].bootstrap, None);
          assert!(loaded[0].generalizations.is_empty());
          assert_eq!(
              loaded[0].trace_name, None,
              "pre-0109 line: an absent trace_name parses to None"
          );
      }

      /// The 0109 name composition (#201 d5 / the spec's seam note): a `Some`
      /// trace_name on the input record — the executor's claim sentinel,
      /// content ignored — is replaced on the STORED line with the derived
      /// `"{campaign8}-{run}"` (prefix from the record's own `campaign`, run
      /// from the store's per-campaign counter); `None` stays `None`. The
      /// executor guarantees a 64-hex campaign id, but the registry seam must
      /// not panic on a shorter one — the safe-prefix case is pinned here.
      #[test]
      fn append_campaign_run_composes_trace_name_from_a_claim() {
          let path = temp_family_dir("campaign_run_trace_name");
          let reg = Registry::open(&path);
          let long = "aaaabbbbccccddddeeeeffff0000111122223333444455556666777788889999";

          // claim sentinel (empty string) -> derived name, per-campaign counter
          let mut claimed = campaign_run_record(long);
          claimed.trace_name = Some(String::new());
          assert_eq!(reg.append_campaign_run(&claimed).expect("append claim 0"), 0);
          assert_eq!(reg.append_campaign_run(&claimed).expect("append claim 1"), 1);

          // no claim -> None stays None
          assert_eq!(reg.append_campaign_run(&campaign_run_record(long)).expect("append plain"), 2);

          // shorter-than-8-chars campaign id: composes from the whole id, no panic
          let mut short = campaign_run_record("abc");
          short.trace_name = Some("sentinel content is ignored".to_string());
          assert_eq!(reg.append_campaign_run(&short).expect("append short-id claim"), 0);

          let stored = reg.load_campaign_runs().expect("load");
          assert_eq!(
              stored.iter().map(|r| r.trace_name.as_deref()).collect::<Vec<_>>(),
              vec![Some("aaaabbbb-0"), Some("aaaabbbb-1"), None, Some("abc-0")],
              "a claim composes {{campaign8}}-{{run}} on the stored line; None stays None",
          );
      }
  }
  ```

  (`campaign_run_record` is the existing helper at the top of the campaign-run test block; `temp_family_dir` and `fs` are already in scope in this module.)

  Run: `cargo test -p aura-registry append_campaign_run_composes_trace_name`
  Expected outcome: **compilation fails** with `error[E0609]: no field \`trace_name\` on type ... CampaignRunRecord` (several occurrences); no tests run. This is the RED evidence — the widening does not exist yet.

- [ ] **Step 2: GREEN — the field, the append-side composition, and literal threading.**

  Four files. All edits are old->new byte-exact replacements.

  **(a) `crates/aura-registry/src/lineage.rs` — the field.** The new field goes after `generalizations` (declaration order = serde order for the `Some` case). Replace:

  ```rust
      /// Campaign-scope generalize annotations (cycle 0108), one per
      /// (strategy, window) across instruments. Absent pre-0108 — the
      /// serde-default widening keeps existing stored lines parseable.
      #[serde(default, skip_serializing_if = "Vec::is_empty")]
      pub generalizations: Vec<CampaignGeneralization>,
  }
  ```

  with:

  ```rust
      /// Campaign-scope generalize annotations (cycle 0108), one per
      /// (strategy, window) across instruments. Absent pre-0108 — the
      /// serde-default widening keeps existing stored lines parseable.
      #[serde(default, skip_serializing_if = "Vec::is_empty")]
      pub generalizations: Vec<CampaignGeneralization>,
      /// The TraceStore family name this realization claims when the document
      /// requests persist_taps — a pure derivation ("{campaign8}-{run}"): the
      /// executor hands a claim sentinel (`Some`, any content) and
      /// [`Registry::append_campaign_run`] composes the derived name onto the
      /// stored line; the consumer persists the bytes (#201 d5). `None` when
      /// the document requests no taps. Absent pre-0109 — the serde-default
      /// widening keeps existing stored lines parseable.
      #[serde(default, skip_serializing_if = "Option::is_none")]
      pub trace_name: Option<String>,
  }
  ```

  **(b) `crates/aura-registry/src/lineage.rs` — the composition in `append_campaign_run`.** Replace:

  ```rust
          let mut file = fs::OpenOptions::new().create(true).append(true).open(&path)?;
          let stored = CampaignRunRecord { run, ..record.clone() };
          let line = serde_json::to_string(&stored).expect("a finite CampaignRunRecord serializes");
  ```

  with:

  ```rust
          let mut file = fs::OpenOptions::new().create(true).append(true).open(&path)?;
          // A `Some` trace_name (any content — the executor's claim sentinel,
          // #201 d5) is replaced with the derived name "{campaign8}-{run}";
          // `None` stays `None`. The executor guarantees a 64-hex campaign id,
          // but the registry seam must not panic on a shorter one (in-crate
          // tests use short ids), so the prefix take is the safe `get(..8)`.
          let trace_name = record.trace_name.as_ref().map(|_| {
              let prefix = record.campaign.get(..8).unwrap_or(record.campaign.as_str());
              format!("{prefix}-{run}")
          });
          let stored = CampaignRunRecord { run, trace_name, ..record.clone() };
          let line = serde_json::to_string(&stored).expect("a finite CampaignRunRecord serializes");
  ```

  And extend the method's doc comment. Replace:

  ```rust
      /// run. The input record's own `run` field is ignored. Reads the store once
      /// to pick the run index (a read-before-write; single-process CLI
      /// invocations do not race).
  ```

  with:

  ```rust
      /// run. The input record's own `run` field is ignored; a `Some`
      /// `trace_name` (the executor's claim sentinel, cycle 0109) is replaced
      /// with the derived `"{campaign8}-{run}"` on the stored line, `None`
      /// stays `None`. Reads the store once to pick the run index (a
      /// read-before-write; single-process CLI invocations do not race).
  ```

  **(c) `crates/aura-registry/src/lib.rs` — thread `trace_name: None` through the three test literals.**

  Site 1 (the `campaign_run_record` helper). Replace:

  ```rust
              run: 99,
              seed: 7,
              cells: vec![],
              generalizations: vec![],
          }
  ```

  with:

  ```rust
              run: 99,
              seed: 7,
              cells: vec![],
              generalizations: vec![],
              trace_name: None,
          }
  ```

  Site 2 (`campaign_run_record_roundtrips`). Replace:

  ```rust
                  ],
              }],
              generalizations: vec![],
          };
  ```

  with:

  ```rust
                  ],
              }],
              generalizations: vec![],
              trace_name: None,
          };
  ```

  Site 3 (`campaign_run_record_roundtrips_with_annotator_fields`, where `generalizations` is a non-empty vec). Replace:

  ```rust
                  missing: vec!["US500".to_string()],
              }],
          };
  ```

  with:

  ```rust
                  missing: vec!["US500".to_string()],
              }],
              trace_name: None,
          };
  ```

  **(d) `crates/aura-registry/tests/campaign_run_store_e2e.rs` — the `campaign_run` helper literal.** Replace:

  ```rust
          }],
          generalizations: vec![],
      }
  }
  ```

  with:

  ```rust
          }],
          generalizations: vec![],
          trace_name: None,
      }
  }
  ```

  Run: `cargo test -p aura-registry`
  Expected outcome: green — every target compiles and all tests pass (0 failed), including the new `append_campaign_run_composes_trace_name_from_a_claim` and the extended `campaign_run_line_without_new_fields_still_parses`. The existing whole-record `PartialEq` round-trip asserts stay green: their literals now carry `trace_name: None`, no claim is made, and the stored line keeps `None`.

- [ ] **Step 3: RED — campaign pin for the eligibility stamp + returned/stored agreement.**

  In `crates/aura-campaign/tests/execute.rs`, two edits.

  **(a)** Pin the `None` side on the existing sweep-only test (this assert passes already — the `campaign()` helper builds `persist_taps: vec![]`; it pins that empty taps NEVER grow a name). In `execute_sweep_only_records_family_and_selection`, replace:

  ```rust
      assert_eq!(runs[0].run, 0);
      assert_eq!(runs[0].seed, 7);
      assert_eq!(runs[0].cells.len(), 1);
  ```

  with:

  ```rust
      assert_eq!(runs[0].run, 0);
      assert_eq!(runs[0].seed, 7);
      assert_eq!(
          runs[0].trace_name, None,
          "empty persist_taps claims no trace name (0109 contract)"
      );
      assert_eq!(runs[0].cells.len(), 1);
  ```

  **(b)** Append the new test at the end of the file. Replace the file's final lines:

  ```rust
      // the record round-trips through the store with BOTH annotations intact
      let runs = reg.load_campaign_runs().expect("load campaign runs");
      assert_eq!(runs[0], out.record);
  }
  ```

  with:

  ```rust
      // the record round-trips through the store with BOTH annotations intact
      let runs = reg.load_campaign_runs().expect("load campaign runs");
      assert_eq!(runs[0], out.record);
  }

  /// The 0109 name contract (spec seam note / #201 d5): `execute` claims a
  /// TraceStore family name iff the document requests persist_taps — the
  /// RETURNED record and the STORED line both carry the same derived
  /// `Some("{campaign8}-{run}")` (`append_campaign_run` composes it onto the
  /// stored line from the claim sentinel; `execute` mirrors the same
  /// derivation onto the returned copy after the counter is assigned). The
  /// empty-taps `None` side is pinned on
  /// `execute_sweep_only_records_family_and_selection`.
  #[test]
  fn execute_persist_taps_stamps_trace_name() {
      let reg = temp_registry("persist_taps_trace_name");
      let mut doc = campaign(&["EURUSD"]);
      doc.presentation.persist_taps = vec!["equity".to_string()];
      let proc_doc = process(vec![sweep_stage(false)]);
      let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg)
          .expect("persist_taps campaign executes");

      // returned record: the derived name over the first run counter
      let expected = format!("{}-0", &CAMPAIGN_ID[..8]);
      assert_eq!(
          out.record.trace_name.as_deref(),
          Some(expected.as_str()),
          "the returned record carries the derived trace name",
      );

      // the stored line agrees — in the field and as a whole record
      let runs = reg.load_campaign_runs().expect("load campaign runs");
      assert_eq!(runs.len(), 1);
      assert_eq!(runs[0].trace_name.as_deref(), Some(expected.as_str()));
      assert_eq!(runs[0], out.record, "stored line and returned record are the same record");

      // a second run of the same campaign derives the next counter's name
      let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg)
          .expect("second persist_taps run");
      let expected_1 = format!("{}-1", &CAMPAIGN_ID[..8]);
      assert_eq!(second.record.trace_name.as_deref(), Some(expected_1.as_str()));
  }
  ```

  (`CAMPAIGN_ID` is the file's existing 64-hex constant; its first 8 chars are `aaaabbbb`. `Presentation`'s fields are `pub`, so the `doc.presentation.persist_taps` mutation compiles.)

  Run: `cargo test -p aura-campaign execute_persist_taps_stamps_trace_name`
  Expected outcome: compiles, and the one matched test **FAILS** (`0 passed; 1 failed`) on the first assert — left `None`, right `Some("aaaabbbb-0")` — because `execute` does not yet set the claim sentinel. This is the RED evidence.

- [ ] **Step 4: GREEN — the claim sentinel + the returned-copy mirror in `execute`.**

  In `crates/aura-campaign/src/exec.rs`, replace the record finalization at the end of `execute` (note: `campaign_prefix` is already in scope — `let campaign_prefix = &campaign_id[..8];` earlier in the same function):

  ```rust
      let mut record = CampaignRunRecord {
          campaign: campaign_id.to_string(),
          process: process_id,
          run: 0,
          seed: campaign.seed,
          cells: cells_rec,
          generalizations,
      };
      let run = registry.append_campaign_run(&record).map_err(ExecFault::Registry)?;
      record.run = run;
      Ok(CampaignOutcome { record, run, cells: cells_out })
  ```

  with:

  ```rust
      let mut record = CampaignRunRecord {
          campaign: campaign_id.to_string(),
          process: process_id,
          run: 0,
          seed: campaign.seed,
          cells: cells_rec,
          generalizations,
          // The claim sentinel (#201 d5): a `Some` (any content) tells
          // `append_campaign_run` to compose the derived trace name onto the
          // stored line; `None` means the document requests no taps.
          trace_name: (!campaign.presentation.persist_taps.is_empty()).then(String::new),
      };
      let run = registry.append_campaign_run(&record).map_err(ExecFault::Registry)?;
      record.run = run;
      // Mirror the store-side derivation onto the returned copy (the 0109
      // contract: stored line and returned record carry the same
      // `Some("{campaign8}-{run}")` iff persist_taps is non-empty, else None).
      record.trace_name = (!campaign.presentation.persist_taps.is_empty())
          .then(|| format!("{campaign_prefix}-{run}"));
      Ok(CampaignOutcome { record, run, cells: cells_out })
  ```

  Run: `cargo test -p aura-campaign`
  Expected outcome: green, 0 failed across all targets — `execute_persist_taps_stamps_trace_name` now passes, and every existing whole-record `PartialEq` assert (`execute_sweep_only_records_family_and_selection`, `execute_is_deterministic_twice`, `execute_mc_after_gate_bootstraps_each_survivor`, `execute_generalize_across_two_instruments`, `execute_mc_and_generalize_compose_in_one_pipeline`) stays green: those campaigns have empty `persist_taps`, so both the stored and returned records carry `trace_name: None`.

- [ ] **Step 5: Workspace gates — build + the aura-cli byte-identity pin.**

  No code changes. Run, in order:

  1. `cargo build --workspace`
     Expected outcome: builds cleanly (no other crate constructs a `CampaignRunRecord` literal — the only sites are the ones threaded in Steps 2 and 4).

  2. `cargo test -p aura-cli --test research_docs campaign_runs_lists`
     Expected outcome: `1 passed; 0 failed` — the matched test is the hand-seeded byte-identity pin `campaign_runs_lists_and_dumps_the_bare_stored_record`. It stays green without modification: its seeded store line omits `trace_name`, which parses to `None` via the serde default, and the dump path RE-serializes the parsed record (`serde_json::to_string` in `aura-cli/src/research_docs.rs`), where `skip_serializing_if = "Option::is_none"` omits the field again — so the dump remains byte-identical to the seeded line.
### Task 3: persist_campaign_traces — the consumer wiring (aura-cli)

**Files:** `crates/aura-cli/src/campaign_run.rs`, `crates/aura-cli/tests/research_docs.rs`

- [ ] **Step 1: RED — pin the deferral line's retirement on the Task-1(b) seam test.**
  In `crates/aura-cli/tests/research_docs.rs`, locate the seam test Task 1 created as split (b) of the old `campaign_run_persist_taps_deferred_loudly`: the `#[test]` fn whose fixture writes a campaign via `campaign_doc_json(&bp_id, &proc_id, (1, 2), "\"equity\"", "")` (a VALID tap over the deterministic-refusal 1970 window) and asserts exit 1 with the member-data refusal (`no recorded geometry` / `no data for instrument`).
  1. If any assertion in that test mentions `persist_taps not yet honored` (tolerating or requiring the 0107 deferral line), delete that whole `assert!` statement.
  2. Append, as the LAST two statements of the function body (immediately before its closing `}`):

  ```rust
      assert!(
          !out.contains("persist_taps not yet honored"),
          "the 0107 deferral line is retired; persist_taps is honored from 0109 on: {out}"
      );
      assert!(
          !out.contains("traces persisted:"),
          "a member-data refusal aborts inside execute, before any tracing: {out}"
      );
  ```

  Run: `cargo test -p aura-cli --test research_docs campaign_run`
  Expected: exactly ONE failure — the tightened seam test fails on the new `!out.contains("persist_taps not yet honored")` assertion (the deferral `eprintln!` still fires in `campaign_run.rs`); every other `campaign_run*` test passes or prints its data-less skip note. This is the RED pin for this task.

- [ ] **Step 2: GREEN — wire trace persistence into `crates/aura-cli/src/campaign_run.rs`.**
  Four edits, all in this file.

  **(a) Imports.** Replace this exact block (top of file):

  ```rust
  use std::path::{Path, PathBuf};
  use std::sync::Arc;

  use aura_campaign::{CellSpec, ExecFault, MemberFault, MemberRunner};
  use aura_core::{Cell, ParamSpec, Scalar};
  use aura_engine::{blueprint_from_json, FamilySelection, RunReport};
  use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource};
  use aura_registry::CampaignRunRecord;
  use aura_research::{
      campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign,
      validate_process, DocRef,
  };
  ```

  with:

  ```rust
  use std::path::{Path, PathBuf};
  use std::sync::{mpsc, Arc};

  use aura_campaign::{CampaignOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
  use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp};
  use aura_engine::{
      blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection,
      RunReport,
  };
  use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource};
  use aura_registry::{CampaignRunRecord, WriteKind};
  use aura_research::{
      campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign,
      validate_process, CampaignDoc, DocRef,
  };
  ```

  (The test module's own `use aura_core::ScalarKind;` stays as is — an explicit import shadows the `use super::*` glob without conflict.)

  **(b) Remove the deferral block.** Replace this exact block (inside `run_campaign`, just before the runner construction):

  ```rust
      // Loud deferral (#198 decision 6), once per run — pinned ORDER: this
      // prints after the document gates and BEFORE any member executes, so a
      // data refusal inside execute cannot swallow it.
      if !campaign.presentation.persist_taps.is_empty() {
          eprintln!(
              "aura: persist_taps not yet honored ({} tap(s) ignored)",
              campaign.presentation.persist_taps.len()
          );
      }

      let runner = CliMemberRunner {
  ```

  with:

  ```rust
      let runner = CliMemberRunner {
  ```

  **(c) Call site + the new items.** The persist call lands AFTER the always-on final `campaign_run` stdout line — decided: stdout stays data-pure (the record line is the wire contract), the trace surface is a stderr concern, and the realization record is already stored by `execute`, so a trace failure exits 1 without un-recording it. Replace this exact block (the tail of `run_campaign`, including its closing brace):

  ```rust
      println!(
          "{}",
          serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
              .expect("campaign run record serializes")
      );
      Ok(())
  }
  ```

  with (the `#[cfg(test)]` module that follows stays untouched below the new items):

  ```rust
      println!(
          "{}",
          serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
              .expect("campaign run record serializes")
      );

      // Trace persistence runs LAST, after the always-on record line (0109):
      // stdout stays data-pure (the record line is the wire contract) and the
      // trace surface is a stderr concern; the realization record is already
      // stored by `execute`, so a trace failure exits 1 without un-recording it.
      if let Some(trace_name) = &outcome.record.trace_name {
          persist_campaign_traces(
              trace_name,
              &campaign.presentation.persist_taps,
              &outcome,
              &campaign,
              &strategies,
              &runner.server,
              env,
          )?;
      }
      Ok(())
  }

  /// Which drained wrap-convention channel a requested tap persists from on a
  /// campaign trace re-run. The routing mirrors `persist_traces_r` (main.rs):
  /// equity <- the SimBroker equity recorder (tx_eq), exposure <- the Bias
  /// recorder (tx_ex), r_equity <- the cum_realized_r + unrealized_r recorder
  /// (tx_req). `net_r_equity` has no channel here: the campaign runner wires
  /// no cost leg (`wrap_r(.., cost: None)`), so a net curve is not produced
  /// by this run's configuration.
  #[derive(Clone, Copy, Debug, PartialEq, Eq)]
  enum TapChannel {
      Equity,
      Exposure,
      REquity,
  }

  /// Route one requested tap of the closed vocabulary to its drained channel;
  /// `None` marks a tap this run cannot produce (the caller skips it loudly).
  /// An unknown name cannot reach here (`validate_campaign` refuses it) and
  /// maps to `None` all the same rather than guessing.
  fn tap_channel(tap: &str) -> Option<TapChannel> {
      match tap {
          "equity" => Some(TapChannel::Equity),
          "exposure" => Some(TapChannel::Exposure),
          "r_equity" => Some(TapChannel::REquity),
          _ => None,
      }
  }

  /// The content-derived on-disk key of one campaign cell under
  /// `traces/<trace_name>/` — the `member_key` discipline (content, never a
  /// runtime ordinal): strategy content-id prefix + instrument +
  /// doc-positional window ordinal, sanitized to one portable path component.
  fn campaign_cell_key(strategy: &str, instrument: &str, window_ordinal: usize) -> String {
      let strategy8 = strategy.get(..8).unwrap_or(strategy);
      crate::sanitize_component(&format!("{strategy8}-{instrument}-w{window_ordinal}"))
  }

  /// Persist the requested `persist_taps` for every nominee cell under
  /// `traces/<trace_name>/<cell_key>/` (0109, #201 d5): re-run each nominee
  /// once in non-reduce trace mode over its own recorded `manifest.window`,
  /// assert the re-run METRICS equal the recorded nominee metrics (the C1
  /// drift alarm — manifest fields are fresh-context and not compared), and
  /// write the requested-AND-producible taps through the sweep verbs'
  /// member-dir mechanism: a slash-joined `"{name}/{key}"` handed to
  /// `TraceStore::write` (the `persist_traces_r(&format!("{name}/{key}"), ..)`
  /// layout — the fn itself is not reused: it writes a fixed tap set and
  /// process-exits on error). The written index manifest is the RECORDED
  /// nominee's — the trace documents that run's provenance, not a re-derived
  /// fresh-context one. Loud stderr per no-nominee cell and ONCE per
  /// unproducible requested tap (producibility is run-configuration-level,
  /// not per-cell); one summary line at the end. Every `Err` is a refusal
  /// `campaign_cmd` renders as `aura: {msg}` + exit 1.
  fn persist_campaign_traces(
      trace_name: &str,
      taps: &[String],
      outcome: &CampaignOutcome,
      campaign: &CampaignDoc,
      strategies: &[(String, String)],
      server: &Arc<data_server::DataServer>,
      env: &Env,
  ) -> Result<(), String> {
      let store = env.trace_store();
      store
          .ensure_name_free(trace_name, WriteKind::Family)
          .map_err(|e| e.to_string())?;

      // Requested ∩ producible, in request order. When nothing producible was
      // requested (net_r_equity only), no member dir is written and the
      // summary honestly reports 0 tap(s).
      let mut routed: Vec<(&str, TapChannel)> = Vec::new();
      for tap in taps {
          match tap_channel(tap) {
              Some(ch) => routed.push((tap.as_str(), ch)),
              None => eprintln!(
                  "aura: tap \"{tap}\" is not produced by this run (needs a cost run); skipped"
              ),
          }
      }

      // `outcome.cells` is index-aligned with `outcome.record.cells` by
      // construction: `execute` pushes both in the same cell-loop iteration.
      let mut persisted_cells = 0usize;
      for (cell_rec, cell_out) in outcome.record.cells.iter().zip(&outcome.cells) {
          let Some((_, nominee_report)) = &cell_out.nominee else {
              eprintln!(
                  "aura: cell {}/{}/[{}, {}]: no nominee; no traces persisted",
                  cell_rec.strategy, cell_rec.instrument, cell_rec.window_ms.0, cell_rec.window_ms.1
              );
              continue;
          };
          if routed.is_empty() {
              continue;
          }

          // The cell key is doc-derived: the window ordinal is the POSITION
          // of the cell's window in campaign.data.windows (mirroring the
          // family-name convention), never a loop counter re-derived here.
          let window_ordinal = campaign
              .data
              .windows
              .iter()
              .position(|w| (w.from_ms, w.to_ms) == cell_rec.window_ms)
              .ok_or_else(|| {
                  format!(
                      "cell window [{}, {}] is not one of campaign.data.windows",
                      cell_rec.window_ms.0, cell_rec.window_ms.1
                  )
              })?;
          let cell_key =
              campaign_cell_key(&cell_rec.strategy, &cell_rec.instrument, window_ordinal);

          // The cell's blueprint: the run's own index-aligned resolution, by
          // content id (exactly the bytes the executor's members ran).
          let (_, blueprint_json) = strategies
              .iter()
              .find(|(id, _)| *id == cell_rec.strategy)
              .ok_or_else(|| {
                  format!(
                      "cell strategy {} is not among the run's resolved strategies",
                      cell_rec.strategy
                  )
              })?;

          // Re-run the nominee, non-reduce: the SAME member the executor ran
          // (same wrapped space, same params, same window, seed-free real
          // data), mirroring `CliMemberRunner::run_member` with the reduce
          // fold off so the per-cycle tap streams exist. The window is the
          // nominee report's own `manifest.window` — already epoch-ns
          // (`run_blueprint_member` stamped the post-seam bounds), so no
          // second ms->ns crossing here.
          let space = crate::blueprint_axis_probe(blueprint_json, env).param_space();
          let point = crate::point_from_params(&space, &nominee_report.manifest.params);
          if instrument_geometry(server, &cell_rec.instrument).is_none() {
              return Err(format!(
                  "no recorded geometry for symbol '{}' at {} — refusing to run a \
                   real instrument with a guessed pip",
                  cell_rec.instrument,
                  env.data_path()
              ));
          }
          let (from, to) = nominee_report.manifest.window;
          let no_data = || {
              format!(
                  "no data for instrument {} in the nominee window [{}, {}] (epoch-ns)",
                  cell_rec.instrument, from.0, to.0
              )
          };
          let source = M1FieldSource::open_window(
              server,
              &cell_rec.instrument,
              Some(from),
              Some(to),
              M1Field::Close,
          )
          .ok_or_else(no_data)?;
          if aura_engine::Source::peek(&source).is_none() {
              return Err(no_data());
          }
          let signal = blueprint_from_json(blueprint_json, &|t| env.resolve(t))
              .expect("stored blueprint passed the referential gate; reload is infallible");
          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 = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, false, false, None)
              .bootstrap_with_cells(&point)
              .expect("the nominee's point re-bootstraps (it already ran this realization)");
          let sources: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(source)];
          h.run(sources);
          // Drain ALL FOUR channels (`run_signal_r` leaves req undrained; the
          // trace path must not): eq/ex/req feed the taps, r feeds the metrics.
          let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
          let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
          let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
          let req_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_req.try_iter().collect();

          // The C1 drift alarm: metrics equality against the recorded
          // nominee. The reduce-mode fold shares its arithmetic with this
          // non-reduce reduction (SeriesFold via `summarize`; GatedRecorder
          // emits exactly the rows `summarize_r`'s ledger reads), so equality
          // is bit-exact.
          let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
          rerun_metrics.r = Some(summarize_r(&r_rows, &[]));
          if rerun_metrics != nominee_report.metrics {
              return Err(format!(
                  "trace re-run diverged from the recorded nominee (C1 violation): cell \
                   {cell_key} of trace {trace_name} does not reproduce its recorded \
                   metrics; refusing to persist a silently-wrong trace"
              ));
          }

          let traces: Vec<ColumnarTrace> = routed
              .iter()
              .map(|&(tap, ch)| {
                  let rows: &[(Timestamp, Vec<Scalar>)] = match ch {
                      TapChannel::Equity => &eq_rows,
                      TapChannel::Exposure => &ex_rows,
                      TapChannel::REquity => &req_rows,
                  };
                  ColumnarTrace::from_rows(tap, &[ScalarKind::F64], rows)
              })
              .collect();
          store
              .write(&format!("{trace_name}/{cell_key}"), &nominee_report.manifest, &traces)
              .map_err(|e| e.to_string())?;
          persisted_cells += 1;
      }

      eprintln!(
          "aura: traces persisted: {trace_name} ({} tap(s) x {persisted_cells} cell(s))",
          routed.len()
      );
      Ok(())
  }
  ```

  **(d) Hermetic unit tests.** In the same file's `#[cfg(test)] mod tests`, replace this exact block (the last test fn plus the module's closing brace):

  ```rust
          assert!(
              m.contains("open param \"threshold\"") && m.contains("(wrapped path: sma.threshold)"),
              "raw-namespace prose expected: {m}"
          );
      }
  }
  ```

  with:

  ```rust
          assert!(
              m.contains("open param \"threshold\"") && m.contains("(wrapped path: sma.threshold)"),
              "raw-namespace prose expected: {m}"
          );
      }

      #[test]
      /// 0109: the campaign cell key is content-derived (the `member_key`
      /// discipline — never a runtime ordinal): strategy content-id prefix +
      /// instrument + doc-positional window ordinal, one portable component.
      fn campaign_cell_key_is_content_derived() {
          let strategy = "bb34aa55".repeat(8); // a 64-hex content id
          assert_eq!(campaign_cell_key(&strategy, "GER40", 0), "bb34aa55-GER40-w0");
      }

      #[test]
      /// A non-portable instrument byte maps to `_` (the shared
      /// `sanitize_component` charset) — one path component, never a nested
      /// path or an invalid directory name.
      fn campaign_cell_key_sanitizes_non_portable_bytes() {
          let strategy = "0123abcd".repeat(8);
          assert_eq!(campaign_cell_key(&strategy, "GER/40", 3), "0123abcd-GER_40-w3");
      }

      #[test]
      /// The tap->channel routing mirrors `persist_traces_r` (equity<-eq,
      /// exposure<-ex, r_equity<-req); `net_r_equity` is unproducible on the
      /// campaign runner's cost-free wrap and routes to no channel.
      fn tap_channel_routes_the_producible_vocabulary_only() {
          assert_eq!(tap_channel("equity"), Some(TapChannel::Equity));
          assert_eq!(tap_channel("exposure"), Some(TapChannel::Exposure));
          assert_eq!(tap_channel("r_equity"), Some(TapChannel::REquity));
          assert_eq!(tap_channel("net_r_equity"), None);
      }
  }
  ```

  Run: `cargo build --workspace`
  Expected: clean build, no warnings from the touched crate.

- [ ] **Step 3: gates.**
  Run: `cargo test -p aura-cli --bin aura campaign_cell_key`
  Expected: 2 passed (`campaign_cell_key_is_content_derived`, `campaign_cell_key_sanitizes_non_portable_bytes`), 0 failed.
  Run: `cargo test -p aura-cli --bin aura tap_channel`
  Expected: 1 passed (`tap_channel_routes_the_producible_vocabulary_only`), 0 failed.
  Run: `cargo test -p aura-cli --test research_docs campaign_run`
  Expected: ALL `campaign_run*` tests pass (the Step-1 seam test is now GREEN: the deferral line is gone and the member-data refusal aborts inside `execute` before any tracing; on a data-less host the two real-data e2e tests print their skip note and pass).

### Task 4: gated e2e extension

**Files:** `crates/aura-cli/tests/research_docs.rs`

- [ ] **Step 1: extend `campaign_run_real_e2e_sweep_gate_walkforward` to pin the full trace path.**
  Three edits, all in `crates/aura-cli/tests/research_docs.rs`.

  **(a) Doc comment.** Replace this exact block:

  ```rust
  /// Gated real-data e2e (the `cli_run.rs` skip idiom): a full
  /// sweep -> gate -> walk_forward -> monte_carlo campaign over GER40
  /// Sept-2024 where the local archive is present — exit 0, emit-gated
  /// family/selection lines, a final line parseable as JSON with top-level
  /// `campaign_run` linking a sweep family id, a walk-forward family id, and a
  /// pooled-OOS bootstrap on the terminal mc stage, and the
  /// `campaign_runs.jsonl` sibling store written. Skips with a note elsewhere
  /// so `cargo test --workspace` stays green on a data-less machine.
  ```

  with:

  ```rust
  /// Gated real-data e2e (the `cli_run.rs` skip idiom): a full
  /// sweep -> gate -> walk_forward -> monte_carlo campaign over GER40
  /// Sept-2024 where the local archive is present — exit 0, emit-gated
  /// family/selection lines, a final line parseable as JSON with top-level
  /// `campaign_run` linking a sweep family id, a walk-forward family id, and a
  /// pooled-OOS bootstrap on the terminal mc stage, and the
  /// `campaign_runs.jsonl` sibling store written. Since 0109 the campaign
  /// also requests `persist_taps: ["equity", "r_equity"]`: the record carries
  /// `trace_name`, the nominee's taps land under `traces/<name>/<cell_key>/`
  /// (the persist re-run equality-asserted against the recorded nominee —
  /// C1), and `aura chart <name>/<cell_key>` reads them back through the
  /// shipped viewer. Skips with a note elsewhere so `cargo test --workspace`
  /// stays green on a data-less machine.
  ```

  **(b) The persist_taps splice.** Replace this exact block (inside the test):

  ```rust
      write_doc(
          dir,
          "wf.campaign.json",
          &campaign_doc_json(
              &bp_id,
              &proc_id,
              (1725148800000, 1727740799999),
              "",
              "\"family_table\", \"selection_report\"",
          ),
      );
  ```

  with:

  ```rust
      write_doc(
          dir,
          "wf.campaign.json",
          &campaign_doc_json(
              &bp_id,
              &proc_id,
              (1725148800000, 1727740799999),
              "\"equity\", \"r_equity\"",
              "\"family_table\", \"selection_report\"",
          ),
      );
  ```

  **(c) The trace assertions.** Replace this exact block (the tail of the test, including its closing brace):

  ```rust
      // The registry's new sibling store carries the realization record.
      assert!(
          runs_dir.join("campaign_runs.jsonl").is_file(),
          "campaign_runs.jsonl written beside runs.jsonl"
      );
  }
  ```

  with:

  ```rust
      // The registry's new sibling store carries the realization record.
      assert!(
          runs_dir.join("campaign_runs.jsonl").is_file(),
          "campaign_runs.jsonl written beside runs.jsonl"
      );
      // 0109: the record claims the trace family it persisted under —
      // "{campaign8}-{run}" (fresh store => run 0).
      let trace_name = v["campaign_run"]["trace_name"]
          .as_str()
          .expect("persist_taps non-empty => the record carries trace_name");
      assert!(
          trace_name.len() == 10 && trace_name.ends_with("-0"),
          "trace_name is {{campaign8}}-{{run}}: {trace_name}"
      );
      assert!(
          trace_name[..8].chars().all(|c| c.is_ascii_hexdigit()),
          "trace_name prefix is the campaign id's first 8 hex chars: {trace_name}"
      );
      // The summary stderr line names the tap x cell arity (2 requested taps,
      // one (strategy, instrument, window) cell).
      assert!(
          out.contains(&format!("aura: traces persisted: {trace_name} (2 tap(s) x 1 cell(s))")),
          "the persist summary line: {out}"
      );
      // The taps land under traces/<name>/<cell_key>/ — the cell key is
      // content-derived: strategy8-instrument-w<window_ordinal>.
      let cell_key = format!("{}-GER40-w0", &bp_id[..8]);
      let cell_dir = runs_dir.join("traces").join(trace_name).join(&cell_key);
      assert!(
          cell_dir.join("equity.json").is_file(),
          "equity tap persisted at {}",
          cell_dir.display()
      );
      assert!(
          cell_dir.join("r_equity.json").is_file(),
          "r_equity tap persisted at {}",
          cell_dir.display()
      );
      assert!(
          !cell_dir.join("exposure.json").exists(),
          "unrequested taps are not persisted"
      );
      // The shipped viewer reads the member back (the cli_run.rs chart
      // idiom): exit 0 and an HTML page carrying both persisted tap series.
      let (chart_out, chart_code) =
          run_code_in(dir, &["chart", &format!("{trace_name}/{cell_key}")]);
      assert_eq!(chart_code, Some(0), "aura chart reads the persisted member: {chart_out}");
      assert!(
          chart_out.contains("\"equity\"") && chart_out.contains("\"r_equity\""),
          "the chart page carries the two persisted taps: {chart_out}"
      );
  }
  ```

  Run: `cargo test -p aura-cli --test research_docs campaign_run_real_e2e`
  Expected: 2 passed, 0 failed. On a host with the local GER40 archive, `campaign_run_real_e2e_sweep_gate_walkforward` exercises the full trace path (record `trace_name`, persisted `equity.json` + `r_equity.json`, the C1 equality assert holding, chart read-back); on a data-less host both tests print `skip: no local GER40 data for the campaign e2e` and pass.

- [ ] **Step 2: workspace gates.**
  Run: `cargo test --workspace`
  Expected: green — 0 failed across all crates (data-gated tests pass or skip per host).
  Run: `cargo clippy --workspace --all-targets -- -D warnings`
  Expected: clean, zero warnings.