feat(cli): TapChannel::Net — the net_r_equity tap routes and persists (#234 task 6)

TapChannel gains Net; tap_channel routes the full closed vocabulary
(producibility moved to the caller); the trace path drains the sixth
channel and persists the cost-adjusted net curve. A cost-less document
requesting net_r_equity keeps a skip notice, reworded to name the
remedy (add a cost block) — not a refusal. Two deliberate pin moves:
the absence pin re-pins to routed, the skip-notice e2e re-pins to the
reworded form. Post-review: positive write-path e2e added (a costed
GER40 campaign lands net_r_equity.json with the curve differing from
r_equity by the accrued costs).

Verified: routing pin + positive e2e green (real archive), full
workspace suite green, clippy -D warnings clean.

refs #234

crates/aura-cli/src/campaign_run.rs

@@ -636,36 +636,35 @@ fn present_campaign(
 /// 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: `wrap_r` no longer builds a
+/// (tx_req), net_r_equity <- the cost leg's LinComb(4) net-curve recorder
-/// cost leg at all (#221), so a net curve is not produced by this run's
+/// (tx_net, #234) — produced only when the campaign document carries a cost
-/// configuration.
+/// block (the caller's producibility check).
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum TapChannel {
     Equity,
     Exposure,
     REquity,
+    Net,
 }
 
 /// 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).
+/// `None` marks a name outside the vocabulary (unreachable — `validate_campaign`
-/// An unknown name cannot reach here (`validate_campaign` refuses it) and
+/// refuses it — and mapped defensively rather than guessed at).
-/// maps to `None` all the same rather than guessing.
 fn tap_channel(tap: &str) -> Option<TapChannel> {
     // The emit_vocabulary debug_assert's twin: every vocabulary tap must be
-    // either routable here or the one known cost-only tap — a future fifth
+    // routable here — a future fifth vocabulary entry fails loudly instead of
-    // vocabulary entry fails loudly instead of silently skipping with the
+    // silently skipping.
-    // hardcoded needs-a-cost-run message.
     debug_assert!(
         aura_research::tap_vocabulary()
             .iter()
-            .all(|t| *t == "net_r_equity"
+            .all(|t| matches!(*t, "equity" | "exposure" | "r_equity" | "net_r_equity")),
-                || matches!(*t, "equity" | "exposure" | "r_equity")),
         "tap_vocabulary drifted from the channels campaign_run routes"
     );
     match tap {
         "equity" => Some(TapChannel::Equity),
         "exposure" => Some(TapChannel::Exposure),
         "r_equity" => Some(TapChannel::REquity),
+        "net_r_equity" => Some(TapChannel::Net),
         _ => None,
     }
 }
@@ -771,16 +770,19 @@ pub(crate) fn persist_campaign_traces(
         .ensure_name_free(trace_name, WriteKind::Family)
         .map_err(|e| e.to_string())?;
 
-    // Requested ∩ producible, in request order. When nothing producible was
+    // Requested ∩ producible, in request order. `net_r_equity` is producible
-    // requested (net_r_equity only), no member dir is written and the
+    // exactly when the document carries a cost block (#234); the skip notice
+    // names the remedy — the tap is optional presentation, not a refusal.
+    // When nothing producible was requested, 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)),
+            Some(TapChannel::Net) if campaign.cost.is_empty() => eprintln!(
-            None => eprintln!(
+                "aura: tap \"{tap}\" needs a cost model; add a cost block to the campaign document; skipped"
-                "aura: tap \"{tap}\" is not produced by this run (needs a cost run); skipped"
             ),
+            Some(ch) => routed.push((tap.as_str(), ch)),
+            None => eprintln!("aura: tap \"{tap}\" is not produced by this run; skipped"),
         }
     }
 
@@ -910,7 +912,7 @@ pub(crate) fn persist_campaign_traces(
             // re-run gross would trip the C1 drift alarm below on every
             // legitimate costed campaign, not just on a real divergence.
             let (tx_cost, rx_cost) = mpsc::channel();
-            let (tx_net, _rx_net) = mpsc::channel();
+            let (tx_net, rx_net) = mpsc::channel();
             let cost_leg = (!campaign.cost.is_empty()).then(|| crate::CostLeg {
                 nodes: cost_nodes_for(&campaign.cost),
                 tx_cost,
@@ -921,13 +923,15 @@ pub(crate) fn persist_campaign_traces(
                     .bootstrap_with_cells(&point)
                     .expect("the member's point re-bootstraps (it already ran this realization)");
             h.run(sources);
-            // Drain ALL FIVE channels (`run_signal_r` leaves req undrained; the
+            // Drain ALL SIX channels (`run_signal_r` leaves req undrained; the
-            // trace path must not): eq/ex/req feed the taps, r+cost feed the metrics.
+            // trace path must not): eq/ex/req/net feed the taps, r+cost feed
+            // 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();
             let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
+            let net_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_net.try_iter().collect();
 
             // The C1 drift alarm: metrics equality against the recorded
             // member. The reduce-mode fold shares its arithmetic with this
@@ -951,6 +955,7 @@ pub(crate) fn persist_campaign_traces(
                         TapChannel::Equity => &eq_rows,
                         TapChannel::Exposure => &ex_rows,
                         TapChannel::REquity => &req_rows,
+                        TapChannel::Net => &net_rows,
                     };
                     ColumnarTrace::from_rows(tap, &[ScalarKind::F64], rows)
                 })
@@ -1272,14 +1277,17 @@ mod tests {
     }
 
     #[test]
-    /// The tap->channel routing mirrors `persist_traces_r` (equity<-eq,
+    /// #234 — a DELIBERATE pin move (was: `net_r_equity` routes to `None` on
-    /// exposure<-ex, r_equity<-req); `net_r_equity` is unproducible on the
+    /// the cost-free wrap): the full closed tap vocabulary now routes,
-    /// campaign runner's cost-free wrap and routes to no channel.
+    /// `net_r_equity` to the cost leg's net-curve channel. Producibility
-    fn tap_channel_routes_the_producible_vocabulary_only() {
+    /// (does THIS run carry a cost model?) is the caller's check, not the
+    /// routing's.
+    fn tap_channel_routes_the_full_vocabulary() {
         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);
+        assert_eq!(tap_channel("net_r_equity"), Some(TapChannel::Net));
+        assert_eq!(tap_channel("bogus"), None);
     }
 
     #[test]

crates/aura-cli/tests/research_docs.rs

@@ -2451,6 +2451,123 @@ fn campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm() {
     );
 }
 
+/// Property (#234): a costed campaign requesting `net_r_equity` must actually
+/// land a `net_r_equity.json` trace on disk carrying the cost-adjusted curve
+/// — the write path (`TapChannel::Net` -> `rx_net` -> `net_rows` ->
+/// `ColumnarTrace`) had no positive e2e coverage: the costed siblings above
+/// request only equity/r_equity, and the skip e2e only proves `net_r_equity`
+/// is ABSENT on a cost-FREE document. Exercised through the same real
+/// production path as
+/// `campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm`,
+/// requesting `net_r_equity` alongside `r_equity` on an identically-shaped
+/// costed document: the net file must exist, carry the same row count as its
+/// `r_equity` sibling (both drained off the identical per-cycle firing off
+/// `exec.output("cum_realized_r"/"unrealized_r")` — `wrap_r`, main.rs), and
+/// its values must diverge from the gross r_equity curve at least once
+/// (non-vacuity — a net curve byte-identical to gross would pass even if the
+/// LinComb(4) cost subtraction were never wired). Gated on the local GER40
+/// archive; skips on a data-less host.
+#[test]
+fn campaign_run_real_e2e_net_r_equity_tap_persists_the_cost_adjusted_curve() {
+    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("net-persist.process.json")),
+        ScratchPath::File(dir.join("net-persist.campaign.json")),
+    ]);
+    let bp_id = seed_blueprint(dir, "campaign-persist-net-seed");
+    let proc_id = register_process_doc(dir, "net-persist.process.json", SWEEP_ONLY_PROCESS_DOC);
+    let base = campaign_doc_json(
+        &bp_id,
+        &proc_id,
+        (1725148800000, 1727740799999),
+        "\"net_r_equity\", \"r_equity\"",
+        "\"family_table\"",
+    );
+    let with_cost = base.replacen(
+        "\"seed\": 7,",
+        "\"seed\": 7,\n  \"cost\": [ { \"constant\": { \"cost_per_trade\": 0.0005 } } ],",
+        1,
+    );
+    assert_ne!(with_cost, base, "replacen must actually match the fixture's seed field");
+    write_doc(dir, "net-persist.campaign.json", &with_cost);
+    let (out, code) = run_code_in(dir, &["campaign", "run", "net-persist.campaign.json"]);
+
+    // Skip on a data-less machine: the member-data refusal, never a panic.
+    if code == Some(1)
+        && (out.contains("no recorded geometry") || out.contains("no data for instrument"))
+    {
+        eprintln!("skip: no local GER40 data for the campaign net-persist e2e");
+        return;
+    }
+
+    assert!(
+        !out.contains("C1 violation"),
+        "the costed persist-side re-run must bind the SAME cost model the nominee ran \
+         under, not false-fail C1: {out}"
+    );
+    assert_eq!(code, Some(0), "a deterministic costed campaign persists cleanly: {out}");
+
+    let record_line = out
+        .lines()
+        .find(|l| l.starts_with("{\"campaign_run\":"))
+        .expect("the always-on final campaign_run line");
+    let v: serde_json::Value =
+        serde_json::from_str(record_line).expect("campaign_run line parses as JSON");
+    let trace_name = v["campaign_run"]["trace_name"]
+        .as_str()
+        .expect("persist_taps non-empty => the record carries trace_name");
+    let cell_key = format!("{}-GER40-w0", &bp_id[..8]);
+    let cell_dir = runs_dir.join("traces").join(trace_name).join(&cell_key);
+
+    // The headline finding: net_r_equity.json actually lands on disk.
+    let net_path = cell_dir.join("net_r_equity.json");
+    assert!(
+        net_path.is_file(),
+        "the costed nominee's net_r_equity tap persists at {}",
+        cell_dir.display()
+    );
+    let req_path = cell_dir.join("r_equity.json");
+    assert!(
+        req_path.is_file(),
+        "the sibling gross r_equity tap persists at {}",
+        cell_dir.display()
+    );
+
+    let net_trace: serde_json::Value = serde_json::from_str(
+        &std::fs::read_to_string(&net_path).expect("net_r_equity.json readable"),
+    )
+    .expect("net_r_equity.json parses as JSON");
+    let req_trace: serde_json::Value = serde_json::from_str(
+        &std::fs::read_to_string(&req_path).expect("r_equity.json readable"),
+    )
+    .expect("r_equity.json parses as JSON");
+    let net_col = net_trace["columns"][0].as_array().expect("net_r_equity has a value column");
+    let req_col = req_trace["columns"][0].as_array().expect("r_equity has a value column");
+    assert!(!net_col.is_empty(), "the net curve carries at least one recorded row: {net_trace}");
+    assert_eq!(
+        net_col.len(),
+        req_col.len(),
+        "the net and gross curves are drained off the identical per-cycle firing, same row count"
+    );
+
+    // Non-vacuity: at least one row's net value differs from its gross
+    // sibling — else this test would pass even if the LinComb cost
+    // subtraction (net = gross - cum_cost - open_cost) were never wired and
+    // net silently mirrored gross.
+    let any_row_differs = net_col.iter().zip(req_col.iter()).any(|(n, r)| {
+        n.as_f64().expect("net cell is f64") != r.as_f64().expect("gross cell is f64")
+    });
+    assert!(
+        any_row_differs,
+        "the costed net curve must diverge from the gross r_equity curve at least once \
+         (else the cost subtraction never engaged): net={net_col:?} gross={req_col:?}"
+    );
+}
+
 /// Property (#234): a `vol_slippage` cost component — the structurally
 /// distinct leg from `constant` (it reactivates the vol-proxy RollingMax/
 /// RollingMin/Sub chain feeding a `VolSlippageCost` node, not a flat per-close
@@ -2629,8 +2746,8 @@ fn campaign_persist_two_regimes_land_in_distinct_trace_dirs() {
 /// Property: an UNPRODUCIBLE tap requested alongside a producible one degrades
 /// gracefully, never a crash or a silently-dropped whole request. `net_r_equity`
 /// is in the closed tap vocabulary (`validate_campaign` accepts it) but has no
-/// drained channel on the campaign runner's cost-free wrap (`tap_channel`
+/// drained channel on the campaign runner's cost-free wrap (the doc carries
-/// returns `None`); the one-time stderr note names exactly that tap, the
+/// no cost block, #234); the one-time stderr note names exactly that tap, the
 /// summary line's arity counts only what was actually written (1 tap, not 2),
 /// and no `net_r_equity.json` file is ever created — only the producible
 /// `equity` tap lands on disk. Uses `SWEEP_ONLY_PROCESS_DOC` (cheapest
@@ -2684,9 +2801,9 @@ fn campaign_run_real_e2e_skips_unproducible_tap_gracefully() {
 
     assert!(
         out.contains(
-            "aura: tap \"net_r_equity\" is not produced by this run (needs a cost run); skipped"
+            "aura: tap \"net_r_equity\" needs a cost model; add a cost block to the campaign document; skipped"
         ),
-        "the unproducible tap gets a named, one-time note: {out}"
+        "the unproducible tap gets a named, one-time note with the remedy: {out}"
     );
     assert!(
         out.contains(&format!("aura: traces persisted: {trace_name} (1 tap(s) x 1 cell(s))")),