feat(cli): campaign cost models reach the member run and both re-run sides (#234 tasks 4-5)

cost_nodes_for (beside stop_rule_for_regime) is the one CostSpec ->
bound-builder binding — every component fully bound, so the wrapped
param space stays cost-invariant. run_blueprint_member joins the cost
rows into summarize_r (member metrics genuinely net) and stamps the
cost components into the manifest params; CliMemberRunner threads the
doc's cost. The re-run sides re-derive the model from the manifest:
reproduce via cost_specs_from_params (the #233 stop pattern — a costed
family reproduces bit-identically) and persist_campaign_traces binds
the same model so the C1 drift alarm compares like with like.
Hand-computed pin: a constant cost model yields
net_expectancy_r == expectancy_r - cost_per_trade/|entry-stop| on a
synthetic member.

Verified: full workspace suite green, clippy -D warnings clean; in-loop
spec + quality gates passed per task.

refs #234

crates/aura-cli/src/campaign_run.rs

@@ -19,7 +19,7 @@ use std::path::{Path, PathBuf};
 use std::sync::{mpsc, Arc};
 
 use aura_campaign::{CampaignOutcome, CellOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
-use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp};
+use aura_core::{Cell, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
     blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection,
     RunReport,
@@ -28,8 +28,9 @@ use aura_ingest::{instrument_geometry, open_columns_window, unix_ms_to_epoch_ns}
 use aura_registry::{CampaignRunRecord, WriteKind};
 use aura_research::{
     campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign,
-    validate_process, CampaignDoc, DocRef,
+    validate_process, CampaignDoc, CostSpec, DocRef,
 };
+use aura_std::{CarryCost, ConstantCost, VolSlippageCost};
 
 use crate::project::Env;
 use crate::research_docs::{
@@ -61,6 +62,40 @@ fn stop_rule_for_regime(regime: Option<aura_research::RiskRegime>) -> aura_compo
     }
 }
 
+/// The one CostSpec -> cost-node binding (#234), beside its risk sibling
+/// `stop_rule_for_regime` and shared the same way: `CliMemberRunner::run_member`
+/// (via `run_blueprint_member`) and `persist_campaign_traces`'s drift-alarm
+/// re-run both bind through here, so the persist-side re-run structurally
+/// cannot diverge from the run-side cost model (the #219 divergence class,
+/// cost edition). The bound knob names are the builders' own `ParamSpec` names
+/// — the `CostSpec` serde vocabulary conforms to them.
+pub(crate) fn cost_nodes_for(specs: &[CostSpec]) -> Vec<PrimitiveBuilder> {
+    specs
+        .iter()
+        .map(|s| {
+            let (knob, v) = cost_knob(s);
+            match s {
+                CostSpec::Constant { .. } => ConstantCost::builder().bind(knob, Scalar::f64(v)),
+                CostSpec::VolSlippage { .. } => VolSlippageCost::builder().bind(knob, Scalar::f64(v)),
+                CostSpec::Carry { .. } => CarryCost::builder().bind(knob, Scalar::f64(v)),
+            }
+        })
+        .collect()
+}
+
+/// The one CostSpec-variant -> (knob name, value) mapping, shared by
+/// `cost_nodes_for`'s bind above and `run_blueprint_member`'s manifest stamp
+/// (main.rs): the stamp key must equal the bind key for reproduce to
+/// re-derive a `CostSpec` from a stored manifest, so both sites read the
+/// name off this single function rather than each carrying its own literal.
+pub(crate) fn cost_knob(spec: &CostSpec) -> (&'static str, f64) {
+    match spec {
+        CostSpec::Constant { cost_per_trade } => ("cost_per_trade", *cost_per_trade),
+        CostSpec::VolSlippage { slip_vol_mult } => ("slip_vol_mult", *slip_vol_mult),
+        CostSpec::Carry { carry_per_cycle } => ("carry_per_cycle", *carry_per_cycle),
+    }
+}
+
 /// Phrase an [`ExecFault`] for stderr — Debug-leak-free, path-addressed
 /// (stage index + block id), the `doc_fault_prose`/`ref_fault_prose` register.
 pub(crate) fn exec_fault_prose(f: &ExecFault) -> String {
@@ -232,6 +267,9 @@ struct CliMemberRunner<'a> {
     /// into per-member binding resolution (#231: rebind per campaign without
     /// touching the blueprint's content id).
     bindings: BTreeMap<String, String>,
+    /// The campaign's cost model (#234), threaded into every member run via
+    /// `cost_nodes_for` (empty = zero costs, today's graph).
+    cost: Vec<CostSpec>,
 }
 
 impl MemberRunner for CliMemberRunner<'_> {
@@ -313,6 +351,7 @@ impl MemberRunner for CliMemberRunner<'_> {
             self.env,
             stop,
             &binding,
+            &self.cost,
         );
         report.manifest.instrument = Some(cell.instrument.clone());
         Ok(report)
@@ -482,6 +521,7 @@ pub(crate) fn run_campaign_returning(
         env,
         server: Arc::new(data_server::DataServer::new(env.data_path())),
         bindings: campaign.data.bindings.clone(),
+        cost: campaign.cost.clone(),
     };
     let outcome = aura_campaign::execute(
         campaign_id,
@@ -862,16 +902,32 @@ pub(crate) fn persist_campaign_traces(
             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, stop, false, geo.pip_size, &binding, None)
+            // The campaign's OWN cost model (#234), bound the same way
-                .bootstrap_with_cells(&point)
+            // `CliMemberRunner::run_member` binds it (via `cost_nodes_for`):
-                .expect("the member's point re-bootstraps (it already ran this realization)");
+            // empty = no leg, so a cost-less campaign's re-run is unchanged.
+            // A non-empty model MUST be threaded here too — the recorded
+            // member ran net (Task 4's `run_blueprint_member`); leaving this
+            // 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 cost_leg = (!campaign.cost.is_empty()).then(|| crate::CostLeg {
+                nodes: cost_nodes_for(&campaign.cost),
+                tx_cost,
+                tx_net,
+            });
+            let mut h =
+                crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding, cost_leg)
+                    .bootstrap_with_cells(&point)
+                    .expect("the member's point re-bootstraps (it already ran this realization)");
             h.run(sources);
-            // Drain ALL FOUR channels (`run_signal_r` leaves req undrained; the
+            // Drain ALL FIVE channels (`run_signal_r` leaves req undrained; the
-            // trace path must not): eq/ex/req feed the taps, r feeds the metrics.
+            // trace path must not): eq/ex/req 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();
 
             // The C1 drift alarm: metrics equality against the recorded
             // member. The reduce-mode fold shares its arithmetic with this
@@ -879,7 +935,7 @@ pub(crate) fn persist_campaign_traces(
             // 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, &[]));
+            rerun_metrics.r = Some(summarize_r(&r_rows, &cost_rows));
             if rerun_metrics != member_report.metrics {
                 return Err(format!(
                     "trace re-run diverged from the recorded member (C1 violation): cell \
@@ -1225,4 +1281,24 @@ mod tests {
         assert_eq!(tap_channel("r_equity"), Some(TapChannel::REquity));
         assert_eq!(tap_channel("net_r_equity"), None);
     }
+
+    #[test]
+    /// #234: the one CostSpec -> builder binding maps each component to its
+    /// shipped cost node with the knob BOUND — a bound component adds no open
+    /// param, so the wrapped param_space stays cost-invariant (the probe/member
+    /// space equivalence the campaign path relies on).
+    fn cost_nodes_for_maps_each_component_to_its_bound_builder() {
+        let nodes = cost_nodes_for(&[
+            aura_research::CostSpec::Constant { cost_per_trade: 2.0 },
+            aura_research::CostSpec::VolSlippage { slip_vol_mult: 0.5 },
+            aura_research::CostSpec::Carry { carry_per_cycle: 0.1 },
+        ]);
+        let labels: Vec<String> = nodes.iter().map(|n| n.label()).collect();
+        assert_eq!(labels, ["ConstantCost", "VolSlippageCost", "CarryCost"]);
+        assert!(
+            nodes.iter().all(|n| n.params().is_empty()),
+            "every component must be fully bound (no open param leaks into param_space)"
+        );
+        assert!(cost_nodes_for(&[]).is_empty(), "an empty model binds no nodes");
+    }
 }

crates/aura-cli/src/main.rs

@@ -1053,6 +1053,39 @@ fn stop_rule_from_params(params: &[(String, Scalar)]) -> StopRule {
     }
 }
 
+/// Re-derive the cost model a member was minted under from its manifest params
+/// (`cost[k].<knob>`, stamped by `run_blueprint_member`) — the #233 stop-regime
+/// pattern: like the stop, the cost model rides OUTSIDE the wrapped
+/// param_space, so `point_from_params` cannot recover it. Components are read
+/// in index order; the builder knob name discriminates the variant (each
+/// shipped cost node has exactly one, distinctly named knob). No `cost[0].*`
+/// param = the empty (gross) model — every pre-cost member widens to it,
+/// mirroring `stop_rule_from_params`'s default arm.
+fn cost_specs_from_params(params: &[(String, Scalar)]) -> Vec<aura_research::CostSpec> {
+    let mut specs = Vec::new();
+    for k in 0.. {
+        let prefix = format!("cost[{k}].");
+        let Some((knob, v)) = params
+            .iter()
+            .find_map(|(n, s)| n.strip_prefix(&prefix).map(|rest| (rest, *s)))
+        else {
+            break;
+        };
+        specs.push(match knob {
+            "cost_per_trade" => aura_research::CostSpec::Constant { cost_per_trade: v.as_f64() },
+            "slip_vol_mult" => aura_research::CostSpec::VolSlippage { slip_vol_mult: v.as_f64() },
+            "carry_per_cycle" => aura_research::CostSpec::Carry { carry_per_cycle: v.as_f64() },
+            other => {
+                // Corrupted-on-disk manifest data — the reproduce path's clean
+                // exit register (`aura:` + exit 1), like point_from_params.
+                eprintln!("aura: manifest cost param cost[{k}].{other} names no cost component");
+                std::process::exit(1);
+            }
+        });
+    }
+    specs
+}
+
 /// Look up a persisted family by id, or exit 1 (unknown id / registry load failure) —
 /// the single place `reproduce_family` and `reproduce_family_in` resolve a family, so
 /// the two exit-1 error phrasings can't drift out of sync between the call sites.
@@ -1115,6 +1148,10 @@ fn reproduce_family_in(
         let (tx_r, _) = mpsc::channel();
         let (tx_req, _) = mpsc::channel();
         let stop = stop_rule_from_params(&stored.manifest.params);
+        // The member's cost model, re-derived from its manifest (the #233
+        // stop pattern): a costed family re-runs under the exact components it
+        // was minted with, so `net_expectancy_r` reproduces bit-identically.
+        let cost = cost_specs_from_params(&stored.manifest.params);
         // The member's binding, re-derived from the stored blueprint's own
         // input roles (name defaults — family manifests carry no overrides).
         let binding = binding::resolve_binding(&hash, reload().input_roles(), &BTreeMap::new())
@@ -1188,6 +1225,7 @@ fn reproduce_family_in(
             env,
             stop,
             &binding,
+            &cost,
         );
         outcomes.push((label, rerun.metrics == stored.metrics));
     }
@@ -1588,12 +1626,23 @@ fn run_blueprint_member(
     env: &project::Env,
     stop: StopRule,
     binding: &binding::ResolvedBinding,
+    cost: &[aura_research::CostSpec],
 ) -> 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_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, None)
+    // The doc's cost model as an optional wrap leg (#234): empty = no leg,
+    // exactly the pre-cost graph. The net curve is a !reduce trace concern;
+    // its sender is wired but unread here (the r_equity tap precedent above).
+    let (tx_cost, rx_cost) = mpsc::channel();
+    let (tx_net, _rx_net) = mpsc::channel();
+    let cost_leg = (!cost.is_empty()).then(|| CostLeg {
+        nodes: campaign_run::cost_nodes_for(cost),
+        tx_cost,
+        tx_net,
+    });
+    let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, cost_leg)
         .bootstrap_with_cells(point)
         .expect("member bootstraps (point kind-checked against param_space)");
     h.run(sources);
@@ -1605,11 +1654,27 @@ fn run_blueprint_member(
         named.push(("stop_length".to_string(), Scalar::i64(length)));
         named.push(("stop_k".to_string(), Scalar::f64(k)));
     }
+    // Stamp the cost model the member ran under, beside the stop knobs (#234,
+    // the #233 pattern): one `cost[k].<knob>` param per component, in
+    // component order — the knob name discriminates the variant (each shipped
+    // cost node has exactly one, distinctly named knob). The name is read off
+    // `campaign_run::cost_knob`, the same function `cost_nodes_for` binds
+    // through, so the stamp key cannot drift from the bind key: the manifest
+    // carries enough to re-derive the exact model later. `reproduce_family_in`
+    // reads this stamp back via `cost_specs_from_params` (the #233 stop-regime
+    // pattern), so a costed family reproduces net, not gross.
+    for (k, spec) in cost.iter().enumerate() {
+        let (knob, v) = campaign_run::cost_knob(spec);
+        named.push((format!("cost[{k}].{knob}"), Scalar::f64(v)));
+    }
     let mut manifest = sim_optimal_manifest(named, window, seed, pip);
     manifest.broker = r_sma_broker_label(pip);
     manifest.topology_hash = Some(topo.to_string());
     manifest.project = env.provenance();
     let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
+    // The member's cost rows (empty when no cost model) join the R reduction:
+    // net_expectancy_r diverges from gross by exactly the modelled costs.
+    let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
     let (total_pips, max_drawdown) = rx_eq
         .try_iter()
         .next()
@@ -1618,7 +1683,7 @@ fn run_blueprint_member(
     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, &[]));
+    m.r = Some(summarize_r(&r_rows, &cost_rows));
     RunReport { manifest, metrics: m }
 }
 
@@ -1725,7 +1790,7 @@ fn blueprint_sweep_family(
         .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(env, &binding.columns()), window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding)
+            run_blueprint_member(reload(doc), point, &space, data.run_sources(env, &binding.columns()), window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[])
         })
         // render the sweep terminal's BindError to a message (the fn's String error contract),
         // so `UnknownKnob("nope")` still surfaces verbatim at the IO wrapper.
@@ -1758,7 +1823,7 @@ fn blueprint_sweep_over(
     binder.sweep_with_lattice(|point| {
         let sources = data.windowed_sources(from, to, env, &binding.columns());
         let window = window_of(&sources).expect("non-empty in-sample window");
-        run_blueprint_member(reload(doc), point, &space, sources, window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding)
+        run_blueprint_member(reload(doc), point, &space, sources, window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[])
     })
 }
 
@@ -1776,7 +1841,7 @@ fn run_oos_blueprint(
     let pip = data.pip_size();
     let sources = data.windowed_sources(from, to, env, &binding.columns());
     let window = window_of(&sources).expect("non-empty out-of-sample window");
-    let report = run_blueprint_member(reload, params, space, sources, window, 0, pip, topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding);
+    let report = run_blueprint_member(reload, params, space, sources, window, 0, pip, topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[]);
     (Vec::new(), report)
 }
 
@@ -1903,7 +1968,7 @@ fn blueprint_mc_family(
     let family = monte_carlo(&base_point, &seeds, |seed, _base| {
         let sources = synthetic_walk_sources(seed);
         let window = window_of(&sources).expect("non-empty synthetic walk");
-        run_blueprint_member(reload(doc), &[], &space, sources, window, seed, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding)
+        run_blueprint_member(reload(doc), &[], &space, sources, window, seed, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[])
     });
     // Silent-vacuous MC guard (refuse-don't-guess, C10): with >= 2 seeds, if every draw's
     // metrics are bit-identical to the first, no seed reached a distinguishable realization —
@@ -3964,6 +4029,184 @@ mod tests {
         );
     }
 
+    /// #234: a member run under a constant cost model nets the hand-computed
+    /// per-trade cost — `net_expectancy_r ≈ expectancy_r − mean(cost_per_trade
+    /// / |entry − stop|)` over summarize_r's ledger (closed rows + the
+    /// window-end open row; the CostRunner R-normalization contract), while
+    /// every gross field stays byte-identical to the cost-less run; and the
+    /// manifest stamps the component for reproduce to re-derive.
+    #[test]
+    fn run_blueprint_member_joins_a_constant_cost_model_into_net_metrics() {
+        let env = project::Env::std();
+        let data = DataSource::Synthetic;
+        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
+        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
+        let space = blueprint_axis_probe(&doc, &env).param_space();
+        let binding = binding::resolve_binding("costnet", reload().input_roles(), &BTreeMap::new())
+            .expect("the price role resolves");
+        let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K };
+        let window = data.full_window(&env);
+        let pip = data.pip_size();
+        const CPT: f64 = 0.0005; // price units; the synthetic stream trades near 1.0
+        let run = |cost: &[aura_research::CostSpec]| {
+            run_blueprint_member(
+                reload(),
+                &[],
+                &space,
+                data.run_sources(&env, &binding.columns()),
+                window,
+                0,
+                pip,
+                "topo",
+                &env,
+                stop,
+                &binding,
+                cost,
+            )
+        };
+        let gross = run(&[]);
+        let netted = run(&[aura_research::CostSpec::Constant { cost_per_trade: CPT }]);
+
+        // The trade geometry, independently: a non-reduce cost-less wrap over
+        // the same realization, draining the dense R record whose ledger
+        // summarize_r reads.
+        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 flat = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, None)
+            .compile_with_params(&[])
+            .expect("wraps");
+        let mut h = Harness::bootstrap(flat).expect("bootstraps");
+        h.run(data.run_sources(&env, &binding.columns()));
+        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
+
+        // summarize_r's ledger: closed rows, plus the final row if open — each
+        // charged cost_per_trade / |entry − stop| (0 when the latched distance
+        // is 0). PM record cols: closed=0, entry_price=6, stop_price=7,
+        // open=11 (the aura-analysis r_col contract).
+        let mut costs: Vec<f64> = Vec::new();
+        for (i, (_, row)) in r_rows.iter().enumerate() {
+            let is_last = i == r_rows.len() - 1;
+            if row[0].as_bool() || (is_last && row[11].as_bool()) {
+                let latched = (row[6].as_f64() - row[7].as_f64()).abs();
+                costs.push(if latched > 0.0 { CPT / latched } else { 0.0 });
+            }
+        }
+        let g = gross.metrics.r.as_ref().expect("gross member carries R metrics");
+        let n = netted.metrics.r.as_ref().expect("netted member carries R metrics");
+        assert_eq!(
+            costs.len() as u64, g.n_trades,
+            "the independent ledger walk must see the member's trades"
+        );
+        assert!(costs.iter().any(|&c| c > 0.0), "at least one trade must charge (non-vacuous)");
+        let mean_cost = costs.iter().sum::<f64>() / costs.len() as f64;
+
+        assert_eq!(gross.metrics.total_pips, netted.metrics.total_pips, "gross pips unchanged");
+        assert_eq!(g.expectancy_r, n.expectancy_r, "gross R stays byte-identical under cost");
+        assert_eq!(g.n_trades, n.n_trades);
+        assert_eq!(g.net_expectancy_r, g.expectancy_r, "cost-less: net == gross");
+        assert!(
+            (n.net_expectancy_r - (g.expectancy_r - mean_cost)).abs() < 1e-12,
+            "net = gross − mean per-trade cost: net {} gross {} mean_cost {mean_cost}",
+            n.net_expectancy_r,
+            g.expectancy_r
+        );
+        // The manifest stamps the component (the reproduce re-derivation carrier)...
+        assert!(
+            netted
+                .manifest
+                .params
+                .iter()
+                .any(|(k, v)| k == "cost[0].cost_per_trade" && *v == Scalar::f64(CPT)),
+            "member manifest stamps the cost component: {:?}",
+            netted.manifest.params
+        );
+        // ...and a cost-less member stamps nothing (content/label stability).
+        assert!(
+            !gross.manifest.params.iter().any(|(k, _)| k.starts_with("cost[")),
+            "a cost-less member stamps no cost params"
+        );
+    }
+
+    /// #234: `campaign_run::persist_campaign_traces`'s C1 drift alarm re-runs a
+    /// costed member in `!reduce` mode (fresh per-cycle taps + a `CostLeg`
+    /// bound the same way `cost_nodes_for` binds it) and compares the result
+    /// against the recorded reduce-mode member. This pins that cross-mode
+    /// equivalence holds UNDER a non-empty cost model: the `!reduce` +
+    /// `CostLeg` wiring (this test, verbatim) and the reduce-mode
+    /// `run_blueprint_member` path must net to the exact same `RunMetrics`
+    /// over the same synthetic realization — the equality the drift alarm
+    /// structurally depends on to not false-positive on every legitimate
+    /// costed campaign.
+    #[test]
+    fn persist_side_nonreduce_rerun_matches_reduce_mode_net_metrics_under_cost() {
+        let env = project::Env::std();
+        let data = DataSource::Synthetic;
+        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
+        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
+        let space = blueprint_axis_probe(&doc, &env).param_space();
+        let binding =
+            binding::resolve_binding("persistnet", reload().input_roles(), &BTreeMap::new())
+                .expect("the price role resolves");
+        let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K };
+        let window = data.full_window(&env);
+        let pip = data.pip_size();
+        let cost = [aura_research::CostSpec::Constant { cost_per_trade: 0.0005 }];
+
+        // The recorded member: reduce-mode, cost-bound — `run_blueprint_member`,
+        // the exact path `CliMemberRunner::run_member` calls.
+        let recorded = run_blueprint_member(
+            reload(),
+            &[],
+            &space,
+            data.run_sources(&env, &binding.columns()),
+            window,
+            0,
+            pip,
+            "topo",
+            &env,
+            stop,
+            &binding,
+            &cost,
+        );
+
+        // The persist-side re-run, verbatim structure: `!reduce` + a `CostLeg`
+        // built through the SAME `campaign_run::cost_nodes_for`, then
+        // `summarize` + `summarize_r(&r_rows, &cost_rows)` — exactly
+        // `persist_campaign_traces`'s C1 drift-alarm computation.
+        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 (tx_cost, rx_cost) = mpsc::channel();
+        let (tx_net, _rx_net) = mpsc::channel();
+        let cost_leg =
+            Some(CostLeg { nodes: campaign_run::cost_nodes_for(&cost), tx_cost, tx_net });
+        let flat = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, cost_leg)
+            .compile_with_params(&[])
+            .expect("the persist-side wrap builds");
+        let mut h = Harness::bootstrap(flat).expect("the persist-side harness bootstraps");
+        h.run(data.run_sources(&env, &binding.columns()));
+        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 cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
+        let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
+        rerun_metrics.r = Some(summarize_r(&r_rows, &cost_rows));
+
+        assert!(
+            rerun_metrics.r.as_ref().is_some_and(|r| r.n_trades > 0),
+            "the fixture must actually close at least one costed trade"
+        );
+        assert_eq!(
+            rerun_metrics, recorded.metrics,
+            "the persist path's !reduce + CostLeg re-run must net to the exact same \
+             RunMetrics as the reduce-mode member under the same cost model — the \
+             C1 drift alarm `persist_campaign_traces` relies on"
+        );
+    }
+
     /// Property: the pip a run resolves and stamps into its manifest must be the
     /// pip the in-graph `SimBroker` divides by — the resolved (real, per-instrument)
     /// pip has to reach the graph, not just the manifest label. The `SimBroker`
@@ -4001,10 +4244,10 @@ mod tests {
         let binding = binding::resolve_binding(signal_a.name(), signal_a.input_roles(), &BTreeMap::new())
             .expect("the price role resolves");
         let report_a = run_blueprint_member(
-            signal_a, &[], &space, make_source(), window, 0, pip_a, "topo", &env, stop, &binding,
+            signal_a, &[], &space, make_source(), window, 0, pip_a, "topo", &env, stop, &binding, &[],
         );
         let report_b = run_blueprint_member(
-            r_meanrev_signal(Some(3), Some(0.0)), &[], &space, make_source(), window, 0, pip_b, "topo", &env, stop, &binding,
+            r_meanrev_signal(Some(3), Some(0.0)), &[], &space, make_source(), window, 0, pip_b, "topo", &env, stop, &binding, &[],
         );
         // Guard: the run must have actually traded, else the invariant is vacuous.
         assert!(
@@ -4636,6 +4879,7 @@ mod tests {
             &env,
             non_default,
             &binding,
+            &[],
         );
 
         // persist exactly as the sweep/campaign paths do: store the blueprint, append the family.
@@ -4659,6 +4903,68 @@ mod tests {
         let _ = std::fs::remove_dir_all(&dir);
     }
 
+    /// #234: a family minted under a cost model must reproduce bit-identically —
+    /// reproduce re-derives the components from the member manifest (the #233
+    /// stop-regime pattern). Without the re-derivation the re-run joins an
+    /// empty cost slice, its net_expectancy_r reverts to gross, and the member
+    /// reports DIVERGED. All three variants ride along so every knob name
+    /// round-trips (the vol_slippage component also exercises the reduce-mode
+    /// vol proxy).
+    #[test]
+    fn reproduce_family_re_derives_a_costed_member_bit_identically() {
+        let dir = std::env::temp_dir().join(format!("aura-repro-cost-{}", 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 env = project::Env::std();
+        let data = DataSource::Synthetic;
+        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
+        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
+        let topo = topology_hash(&reload());
+        let space = blueprint_axis_probe(&doc, &env).param_space();
+        let pip = data.pip_size();
+        let window = data.full_window(&env);
+        let binding = binding::resolve_binding("costrepro", reload().input_roles(), &BTreeMap::new())
+            .expect("the price role resolves");
+        let cost = vec![
+            aura_research::CostSpec::Constant { cost_per_trade: 0.0005 },
+            aura_research::CostSpec::VolSlippage { slip_vol_mult: 0.5 },
+            aura_research::CostSpec::Carry { carry_per_cycle: 0.0001 },
+        ];
+        let report = run_blueprint_member(
+            reload(),
+            &[],
+            &space,
+            data.run_sources(&env, &binding.columns()),
+            window,
+            0,
+            pip,
+            &topo,
+            &env,
+            StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
+            &binding,
+            &cost,
+        );
+        // Non-vacuity: the cost model must actually bite, else a DIVERGED
+        // verdict could never be observed and this pin proves nothing.
+        let r = report.metrics.r.as_ref().expect("member carries R metrics");
+        assert_ne!(r.net_expectancy_r, r.expectancy_r, "the cost model must move net off gross");
+
+        let canonical = blueprint_to_json(&reload()).unwrap();
+        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
+        let id = reg.append_family("costrepro", FamilyKind::Sweep, &[report]).expect("append");
+
+        let rep = reproduce_family_in(&reg, &id, &data, &env);
+        assert_eq!(rep.outcomes.len(), 1, "one member reproduced");
+        assert!(
+            rep.outcomes.iter().all(|(_, ok)| *ok),
+            "the costed member re-derives bit-identically: {:?}",
+            rep.outcomes
+        );
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
     #[test]
     fn reproduce_family_re_derives_every_mc_member_bit_identically() {
         let dir = std::env::temp_dir().join(format!("aura-repro-mc-{}", std::process::id()));

crates/aura-cli/tests/research_docs.rs

@@ -2359,6 +2359,184 @@ fn campaign_persist_non_default_regime_does_not_false_fail_c1() {
     );
 }
 
+/// Property (#234 Task 5, quality-review Minor finding): a costed campaign's
+/// trace-persist re-run must not false-fail the C1 drift alarm, exercised
+/// through the REAL production path — `campaign run` -> `run_campaign_returning`
+/// -> `CliMemberRunner` (net-mode member run) -> `persist_campaign_traces`
+/// (non-reduce re-run binding a `CostLeg` via `cost_nodes_for`) — not the
+/// hand-mirrored arithmetic pinned by
+/// `persist_side_nonreduce_rerun_matches_reduce_mode_net_metrics_under_cost` in
+/// `main.rs`'s unit tests. Mirrors the sibling non-default-regime e2e above
+/// (same #219-class divergence risk, cost edition): a costed campaign
+/// document, `persist_taps` non-empty so the re-run actually executes, must
+/// exit 0 with its nominee's cost stamp on disk and no C1 violation. Gated on
+/// the local GER40 archive; skips on a data-less host.
+#[test]
+fn campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm() {
+    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("cost-persist.process.json")),
+        ScratchPath::File(dir.join("cost-persist.campaign.json")),
+    ]);
+    let bp_id = seed_blueprint(dir, "campaign-persist-cost-seed");
+    let proc_id = register_process_doc(dir, "cost-persist.process.json", SWEEP_ONLY_PROCESS_DOC);
+    let base = campaign_doc_json(
+        &bp_id,
+        &proc_id,
+        (1725148800000, 1727740799999),
+        "\"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, "cost-persist.campaign.json", &with_cost);
+    let (out, code) = run_code_in(dir, &["campaign", "run", "cost-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 cost-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}");
+
+    // Non-vacuity: every emitted member actually stamped the cost component
+    // it ran under (else the C1 comparison above never engaged a non-empty
+    // CostLeg and this test would pass for the wrong reason).
+    let member_lines: Vec<&str> =
+        out.lines().filter(|l| l.starts_with("{\"family_id\":")).collect();
+    assert!(!member_lines.is_empty(), "family_table member lines emitted: {out}");
+    for line in &member_lines {
+        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
+        let params = v["report"]["manifest"]["params"]
+            .as_array()
+            .expect("manifest.params is an array");
+        assert!(
+            params.iter().any(|p| p[0].as_str() == Some("cost[0].cost_per_trade")),
+            "the costed member stamps its cost component: {line}"
+        );
+    }
+
+    // The nominee's taps actually landed under the costed re-run.
+    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);
+    assert!(
+        cell_dir.join("r_equity.json").is_file(),
+        "the costed nominee's r_equity tap persists at {}",
+        cell_dir.display()
+    );
+}
+
+/// 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
+/// charge) — nets a real campaign member through the SAME production path as
+/// its `constant` sibling
+/// (`campaign_run_real_e2e_costed_campaign_persists_without_drift_alarm`):
+/// `campaign run` -> `CliMemberRunner` (net-mode) -> `persist_campaign_traces`
+/// (non-reduce re-run binding a `CostLeg`). The re-run must bind the identical
+/// vol-slippage model the nominee ran under, so it must not false-fail the C1
+/// drift alarm; the emitted member must stamp `cost[0].slip_vol_mult`; and the
+/// vol-driven charge must actually move net off gross (non-vacuity — a
+/// zero-effect charge would let this pass even if the vol proxy were never
+/// wired). Gated on the local GER40 archive; skips on a data-less host.
+#[test]
+fn campaign_run_real_e2e_vol_slippage_cost_persists_without_drift_alarm() {
+    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("volcost-persist.process.json")),
+        ScratchPath::File(dir.join("volcost-persist.campaign.json")),
+    ]);
+    let bp_id = seed_blueprint(dir, "campaign-persist-volcost-seed");
+    let proc_id =
+        register_process_doc(dir, "volcost-persist.process.json", SWEEP_ONLY_PROCESS_DOC);
+    let base = campaign_doc_json(
+        &bp_id,
+        &proc_id,
+        (1725148800000, 1727740799999),
+        "\"r_equity\"",
+        "\"family_table\"",
+    );
+    let with_cost = base.replacen(
+        "\"seed\": 7,",
+        "\"seed\": 7,\n  \"cost\": [ { \"vol_slippage\": { \"slip_vol_mult\": 0.5 } } ],",
+        1,
+    );
+    assert_ne!(with_cost, base, "replacen must actually match the fixture's seed field");
+    write_doc(dir, "volcost-persist.campaign.json", &with_cost);
+    let (out, code) = run_code_in(dir, &["campaign", "run", "volcost-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 vol-slippage-persist e2e");
+        return;
+    }
+
+    assert!(
+        !out.contains("C1 violation"),
+        "the vol-slippage 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 vol-slippage-costed campaign persists cleanly: {out}");
+
+    let member_lines: Vec<&str> =
+        out.lines().filter(|l| l.starts_with("{\"family_id\":")).collect();
+    assert!(!member_lines.is_empty(), "family_table member lines emitted: {out}");
+    let mut any_net_moved = false;
+    for line in &member_lines {
+        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
+        let params = v["report"]["manifest"]["params"]
+            .as_array()
+            .expect("manifest.params is an array");
+        assert!(
+            params.iter().any(|p| p[0].as_str() == Some("cost[0].slip_vol_mult")),
+            "the costed member stamps its vol_slippage component: {line}"
+        );
+        if let Some(r) = v["report"]["metrics"]["r"].as_object() {
+            let gross = r["expectancy_r"].as_f64().expect("expectancy_r is a number");
+            let net = r["net_expectancy_r"].as_f64().expect("net_expectancy_r is a number");
+            if gross != net {
+                any_net_moved = true;
+            }
+        }
+    }
+    assert!(
+        any_net_moved,
+        "at least one member's vol-slippage charge must move net off gross \
+         (else the vol proxy never engaged): {out}"
+    );
+}
+
 /// RED (#212 — the trace-dir collision facet of the same #219 record gap): two
 /// risk regimes over the SAME (strategy, instrument, window) cell must persist
 /// their traces into DISTINCT directories. `campaign_cell_key` names dirs