plan: 0081 cost-model constant-cost

Task-by-task plan for spec 0081: the ConstantCost node (aura-std), the summarize_r scalar->cost-stream change (the workspace-wide call-site migration as one compile-gate task), and the run-path wiring (net_r_equity tap + --cost-per-trade) — RED-first per task. Scoped to the run path (non-reduce); sweep/walkforward/mc pass None this cycle. refs #148
2026-06-28 13:21:09 +02:00
parent 2e63ed9e26
commit 725545cbde
1 changed files with 671 additions and 0 deletions
@@ -0,0 +1,671 @@
+# Cost-model graph (in R) — cycle 1: ConstantCost + net-R seam — Implementation Plan
+
+> **Parent spec:** `docs/specs/0081-cost-model-constant-cost.md`
+>
+> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run
+> this plan. Steps use `- [ ]` checkboxes for tracking.
+
+**Goal:** Add one in-graph cost node (`ConstantCost`) and the net-R seam so
+`aura run --harness stage1-r --cost-per-trade C` yields a `net_r_equity` trace and a
+cost-adjusted `net_expectancy_r`, with a no-cost run byte-identical to today.
+
+**Architecture:** `ConstantCost` (aura-std) emits a 3-field cost-in-R record
+`{cost_in_r, cum_cost_in_r, open_cost_in_r}` isomorphic to PM's `{realized_r,
+cum_realized_r, unrealized_r}`. Two consumers read that one stream: an in-graph
+`net_r_equity` sink (`LinComb(4,[1,1,-1,-1])` + `Recorder`, the `r_equity` idiom plus
+two cost terms) and the post-run `summarize_r`, whose scalar `round_trip_cost: f64` is
+replaced by consuming the co-temporal cost stream. The cost layer is opt-in on the
+run path (non-reduce); a no-cost run is unchanged.
+
+**Tech Stack:** aura-std (new node), aura-analysis (`summarize_r`), aura-cli
+(`stage1_r_graph`, `run_stage1_r`, `persist_traces_r`, `RunArgs`/`parse_run_args`/
+`run_dispatch`). Cross-crate seam guarded by the existing `r_col` lockstep test.
+
+**Scope note (cycle-1 minimal cut, recorded on #148 J/K):** the run path only
+(non-reduce / trace mode, where `net_r_equity` and the full cost stream wire cleanly).
+`stage1_r_graph`'s sweep/walkforward/mc callers pass `None` (no cost): cost in the
+reduce-mode sweep path + the OOS-pooling cost in `r_metrics_from_rs` are a deferred
+follow-up cycle. Acceptance evidence is the JSON `net_expectancy_r` field + the
+chartable `net_r_equity` trace — no human-readable pretty-printer is added (the spec's
+console mock is illustrative; `aura run` emits JSON).
+
+**Files this plan creates or modifies:**
+
+- Create: `crates/aura-std/src/constant_cost.rs` — the `ConstantCost` node (4 PM-geometry
+  inputs → 3-field cost-in-R record, one `cost_per_trade` F64 param).
+- Modify: `crates/aura-std/src/lib.rs:18-70` — register `mod constant_cost;` + re-export.
+- Modify: `crates/aura-analysis/src/lib.rs:154-287` — `summarize_r` signature + net fold;
+  rewrite its cost unit test `:744-755`.
+- Modify (migrate call sites, compile gate): all `summarize_r(...)` callers across
+  `crates/aura-analysis/src/lib.rs`, `crates/aura-composites/tests/risk_executor.rs`,
+  `crates/aura-engine/tests/stage1_r_e2e.rs`,
+  `crates/aura-engine/tests/streaming_reduction_equivalence.rs`,
+  `crates/aura-cli/src/main.rs`.
+- Modify: `crates/aura-cli/src/main.rs` — `stage1_r_graph:2614-2698` (+ its 7 callers),
+  `run_stage1_r:2917-2971`, `persist_traces_r:2976-2992`, `RunArgs:3004-3008`,
+  `parse_run_args:3015-3074`, `run_dispatch:3078-3091`, `USAGE:3093`.
+- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` — rewrite the two cost properties
+  (`:194-241`) onto the node; add the in-graph↔post-run agreement assertion.
+- Test: `crates/aura-cli/tests/cli_run.rs` — add a `net_r_equity` persistence + cost run
+  test (sibling of `:1599`); the no-cost C18 golden `:1686` stays byte-identical.
+
+---
+
+## Task 1: `ConstantCost` node in aura-std
+
+**Files:**
+- Create: `crates/aura-std/src/constant_cost.rs`
+- Modify: `crates/aura-std/src/lib.rs:18-70`
+- Test: `crates/aura-std/src/constant_cost.rs` (`#[cfg(test)] mod tests`)
+
+- [ ] **Step 1: Write the node + its failing unit tests (one diff, a new file)**
+
+Create `crates/aura-std/src/constant_cost.rs` with the node AND its tests. (RED here is
+the test module referencing a node that does not yet compile cleanly; Step 2 verifies,
+Step 3 is the node body. For a single new file the test and impl co-exist in the diff —
+the orchestrator accepts RED on the new-file's first failing run, per the project's
+RED-in-single-diff allowance.)
+
+```rust
+//! `ConstantCost` — a flat round-trip cost charged once per closed trade, in R.
+//! The first cost node of the C10 cost-model graph (#148): an ordinary downstream
+//! node (C9) that reads the executor's trade-geometry and emits a cost-in-R record
+//! isomorphic to `PositionManagement`'s R-triple — `cost_in_r` (charged on a close)
+//! / `cum_cost_in_r` (its running sum) / `open_cost_in_r` (the open trade's would-be
+//! cost, for the window-end trade `summarize_r` synthesises). R-pure: with flat-1R
+//! sizing the cost is `cost_per_trade / |entry - stop|`; notional cancels (C10).
+
+use aura_core::{
+    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
+    ScalarKind,
+};
+
+/// A flat per-trade cost in price units (`cost_per_trade`), emitted in R. Inputs are
+/// four executor-exposed fields: `closed` (closed_this_cycle), `open`, `entry_price`,
+/// `stop_price`. Emits `None` until the executor has produced a record this cycle.
+pub struct ConstantCost {
+    cost_per_trade: f64,
+    cum: f64,
+    out: [Cell; 3],
+}
+
+impl ConstantCost {
+    pub fn new(cost_per_trade: f64) -> Self {
+        Self { cost_per_trade, cum: 0.0, out: [Cell::from_f64(0.0); 3] }
+    }
+
+    /// The param-generic recipe: one `cost_per_trade` F64 knob.
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "ConstantCost",
+            NodeSchema {
+                inputs: vec![
+                    PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "closed".into() },
+                    PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "open".into() },
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "entry_price".into() },
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "stop_price".into() },
+                ],
+                output: vec![
+                    FieldSpec { name: "cost_in_r".into(), kind: ScalarKind::F64 },
+                    FieldSpec { name: "cum_cost_in_r".into(), kind: ScalarKind::F64 },
+                    FieldSpec { name: "open_cost_in_r".into(), kind: ScalarKind::F64 },
+                ],
+                params: vec![ParamSpec { name: "cost_per_trade".into(), kind: ScalarKind::F64 }],
+            },
+            |p| Box::new(ConstantCost::new(p[0].f64())),
+        )
+    }
+}
+
+impl Node for ConstantCost {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1, 1, 1, 1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        // Withhold until the executor's geometry is present this cycle (price warmed up).
+        let ew = ctx.f64_in(2);
+        if ew.is_empty() {
+            return None;
+        }
+        let entry = ew[0];
+        let stop = ctx.f64_in(3).get(0).unwrap_or(0.0);
+        let closed = ctx.bool_in(0).get(0).unwrap_or(false);
+        let open = ctx.bool_in(1).get(0).unwrap_or(false);
+        let latched = (entry - stop).abs();
+        let per = if latched > 0.0 { self.cost_per_trade / latched } else { 0.0 };
+        let cost_in_r = if closed { per } else { 0.0 };
+        let open_cost_in_r = if open { per } else { 0.0 };
+        self.cum += cost_in_r;
+        self.out = [
+            Cell::from_f64(cost_in_r),
+            Cell::from_f64(self.cum),
+            Cell::from_f64(open_cost_in_r),
+        ];
+        Some(&self.out)
+    }
+
+    fn label(&self) -> String {
+        "ConstantCost".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, Timestamp};
+
+    fn cols() -> Vec<AnyColumn> {
+        vec![
+            AnyColumn::with_capacity(ScalarKind::Bool, 1), // closed
+            AnyColumn::with_capacity(ScalarKind::Bool, 1), // open
+            AnyColumn::with_capacity(ScalarKind::F64, 1),  // entry
+            AnyColumn::with_capacity(ScalarKind::F64, 1),  // stop
+        ]
+    }
+
+    #[test]
+    fn no_geometry_yet_withholds() {
+        let mut c = ConstantCost::new(2.0);
+        let inputs = cols(); // entry column empty
+        assert_eq!(c.eval(Ctx::new(&inputs, Timestamp(0))), None);
+    }
+
+    #[test]
+    fn closed_charges_cost_per_trade_over_latched() {
+        let mut c = ConstantCost::new(2.0);
+        let mut inputs = cols();
+        inputs[0].push(Scalar::bool(true)).unwrap(); // closed
+        inputs[1].push(Scalar::bool(false)).unwrap(); // not open
+        inputs[2].push(Scalar::f64(100.0)).unwrap(); // entry
+        inputs[3].push(Scalar::f64(96.0)).unwrap(); // stop -> latched 4.0
+        // cost_in_r = 2.0/4.0 = 0.5; cum = 0.5; open_cost_in_r = 0.0
+        assert_eq!(
+            c.eval(Ctx::new(&inputs, Timestamp(0))),
+            Some([Cell::from_f64(0.5), Cell::from_f64(0.5), Cell::from_f64(0.0)].as_slice())
+        );
+    }
+
+    #[test]
+    fn open_emits_would_be_cost_not_charged_to_cum() {
+        let mut c = ConstantCost::new(2.0);
+        let mut inputs = cols();
+        inputs[0].push(Scalar::bool(false)).unwrap(); // not closed
+        inputs[1].push(Scalar::bool(true)).unwrap(); // open
+        inputs[2].push(Scalar::f64(100.0)).unwrap();
+        inputs[3].push(Scalar::f64(96.0)).unwrap(); // latched 4.0
+        // cost_in_r = 0 (no close); cum stays 0; open_cost_in_r = 0.5
+        assert_eq!(
+            c.eval(Ctx::new(&inputs, Timestamp(0))),
+            Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.5)].as_slice())
+        );
+    }
+
+    #[test]
+    fn zero_latched_contributes_no_cost() {
+        let mut c = ConstantCost::new(2.0);
+        let mut inputs = cols();
+        inputs[0].push(Scalar::bool(true)).unwrap();
+        inputs[1].push(Scalar::bool(false)).unwrap();
+        inputs[2].push(Scalar::f64(100.0)).unwrap();
+        inputs[3].push(Scalar::f64(100.0)).unwrap(); // latched 0 -> no divide
+        assert_eq!(
+            c.eval(Ctx::new(&inputs, Timestamp(0))),
+            Some([Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)].as_slice())
+        );
+    }
+
+    #[test]
+    fn cum_accumulates_across_closes() {
+        let mut c = ConstantCost::new(2.0);
+        let mut a = cols();
+        a[0].push(Scalar::bool(true)).unwrap();
+        a[1].push(Scalar::bool(false)).unwrap();
+        a[2].push(Scalar::f64(100.0)).unwrap();
+        a[3].push(Scalar::f64(96.0)).unwrap(); // 0.5
+        let _ = c.eval(Ctx::new(&a, Timestamp(0)));
+        let mut b = cols();
+        b[0].push(Scalar::bool(true)).unwrap();
+        b[1].push(Scalar::bool(false)).unwrap();
+        b[2].push(Scalar::f64(100.0)).unwrap();
+        b[3].push(Scalar::f64(98.0)).unwrap(); // latched 2.0 -> 1.0; cum 1.5
+        assert_eq!(
+            c.eval(Ctx::new(&b, Timestamp(1))),
+            Some([Cell::from_f64(1.0), Cell::from_f64(1.5), Cell::from_f64(0.0)].as_slice())
+        );
+    }
+}
+```
+
+- [ ] **Step 2: Register the module in aura-std**
+
+In `crates/aura-std/src/lib.rs`, add the `mod` line in the alphabetical block (between
+`mod bias;` at `:20` and `mod delay;` at `:21`):
+
+```rust
+mod constant_cost;
+```
+
+and the re-export in the `pub use` block (between `pub use bias::Bias;` at `:45` and
+`pub use delay::Delay;` at `:46`):
+
+```rust
+pub use constant_cost::ConstantCost;
+```
+
+- [ ] **Step 3: Run the node tests to verify they pass**
+
+Run: `cargo test -p aura-std constant_cost`
+Expected: PASS — `closed_charges_cost_per_trade_over_latched`,
+`open_emits_would_be_cost_not_charged_to_cum`, `zero_latched_contributes_no_cost`,
+`cum_accumulates_across_closes`, `no_geometry_yet_withholds` all green.
+
+- [ ] **Step 4: Workspace builds clean with the new node**
+
+Run: `cargo build -p aura-std`
+Expected: builds, 0 warnings (the node has no dead code; `Cell::from_f64(0.0); 3`
+relies on `Cell: Copy` — if `Cell` is not `Copy`, use
+`[Cell::from_f64(0.0), Cell::from_f64(0.0), Cell::from_f64(0.0)]` instead).
+
+---
+
+## Task 2: `summarize_r` folds the cost stream (signature change + all call sites)
+
+This is a single compile unit: changing `summarize_r`'s signature breaks every caller
+in the workspace, so **all** call sites are migrated in this task (a `cargo build
+--workspace` 0-errors gate is unsatisfiable until they are). The fold's net token form
+is preserved so a no-cost call is byte-identical.
+
+**Files:**
+- Modify: `crates/aura-analysis/src/lib.rs:154-287` (signature, `Trade`, collection loop,
+  net fold) and `:744-755` (its cost unit test).
+- Modify (migrate callers): `crates/aura-analysis/src/lib.rs` (19 test sites),
+  `crates/aura-composites/tests/risk_executor.rs` (4), `crates/aura-engine/tests/stage1_r_e2e.rs` (9),
+  `crates/aura-engine/tests/streaming_reduction_equivalence.rs` (2),
+  `crates/aura-cli/src/main.rs` (9 production).
+- Test: the new exact-subsumption test in `crates/aura-analysis/src/lib.rs`.
+
+- [ ] **Step 1: Write the failing exact-subsumption test**
+
+Add to `crates/aura-analysis/src/lib.rs`'s `#[cfg(test)] mod tests`. It builds a tiny PM
+record (one closed trade + one window-end open trade) and the co-temporal cost stream a
+`ConstantCost(C)` node would emit, then asserts `net_expectancy_r` equals the
+hand-computed `mean(rᵢ − C/latchedᵢ)` over BOTH trades — proving exact subsumption of
+the old scalar (not just at C=0). (Helper `row14(...)` building a 14-wide PM row and the
+cost row builder follow the existing test fixtures in this module; mirror the closest
+existing `summarize_r` test's row construction at `:609-755`.)
+
+```rust
+#[test]
+fn summarize_r_folds_cost_stream_exactly_like_the_scalar_over_all_trades() {
+    // Trade 1: closed, entry 100 stop 96 (latched 4), realized R = 1.0.
+    // Trade 2: still open at window end, entry 100 stop 98 (latched 2), unrealized R = 0.5.
+    let c = 2.0_f64;
+    // PM record: (col0 closed, col1 realized_r, col4 dir, col6 entry, col7 stop,
+    //  col9 conviction, col11 open, col12 unrealized_r). Use the module's row helper.
+    let record = vec![
+        (Timestamp(0), pm_row(/*closed*/ true, /*r*/ 1.0, /*entry*/ 100.0, /*stop*/ 96.0,
+                              /*conv*/ 1.0, /*open*/ false, /*unreal*/ 0.0)),
+        (Timestamp(1), pm_row(/*closed*/ false, /*r*/ 0.0, /*entry*/ 100.0, /*stop*/ 98.0,
+                              /*conv*/ 1.0, /*open*/ true, /*unreal*/ 0.5)),
+    ];
+    // Cost stream the ConstantCost(2.0) node emits, co-temporal: [cost_in_r, open_cost_in_r].
+    let cost = vec![
+        (Timestamp(0), vec![Scalar::f64(c / 4.0), Scalar::f64(0.0)]),       // closed: 0.5 charged
+        (Timestamp(1), vec![Scalar::f64(0.0), Scalar::f64(c / 2.0)]),       // open: would-be 1.0
+    ];
+    let m = summarize_r(&record, &cost);
+    // gross E[R] = mean(1.0, 0.5) = 0.75; net = mean(1.0-0.5, 0.5-1.0) = mean(0.5,-0.5) = 0.0
+    assert!((m.expectancy_r - 0.75).abs() < 1e-12);
+    assert!((m.net_expectancy_r - 0.0).abs() < 1e-12);
+}
+```
+
+(If no `pm_row` helper exists, add one to the test module:
+`fn pm_row(closed: bool, r: f64, entry: f64, stop: f64, conv: f64, open: bool, unreal: f64)
+-> Vec<Scalar>` returning a 14-element vector with the documented `r_col` positions set
+and the rest `Scalar::f64(0.0)`/`Scalar::bool(false)`/`Scalar::i64(0)`/`Scalar::ts(Timestamp(0))`
+per `PM_RECORD_KINDS`.)
+
+- [ ] **Step 2: Run the new test to verify it fails to compile**
+
+Run: `cargo test -p aura-analysis summarize_r_folds_cost_stream`
+Expected: FAIL — compile error `this function takes ... arguments` / `expected f64, found
+&[...]` at the call (the signature still takes `round_trip_cost: f64`).
+
+- [ ] **Step 3: Change `summarize_r`'s signature, `Trade`, the collection loop, and the net fold**
+
+In `crates/aura-analysis/src/lib.rs`:
+
+Signature `:154`:
+```rust
+pub fn summarize_r(
+    record: &[(Timestamp, Vec<Scalar>)],
+    cost: &[(Timestamp, Vec<Scalar>)],
+) -> RMetrics {
+```
+
+`Trade` struct `:159-163` — add a per-trade `cost` term:
+```rust
+    struct Trade {
+        r: f64,
+        bias_abs: f64,
+        latched: f64,
+        cost: f64,
+    }
+```
+
+Collection loop `:165-184` — **positional** join: the cost stream is co-temporal 1:1
+with the record (both recorded per cycle off the same executor cadence — `ConstantCost`
+emits exactly when PM emits, so `record[i]` and `cost[i]` are the same cycle). A closed
+row `i` reads `cost[i]`'s `cost_in_r` (col 0); the window-end open last row reads its
+`open_cost_in_r` (col 1). An empty `cost` slice ⇒ every `.get(i)` is `None` ⇒ cost `0.0`
+(the gross-R baseline). Positional avoids the shared-timestamp collision a `ts`-keyed map
+would have (C4: same `ts` can be two cycles):
+```rust
+    let mut trades: Vec<Trade> = Vec::new();
+    for (i, (_, row)) in record.iter().enumerate() {
+        if row[r_col::CLOSED].as_bool() {
+            let c = cost.get(i).map(|(_, cr)| cr[0].as_f64()).unwrap_or(0.0);
+            trades.push(Trade {
+                r: row[r_col::REALIZED_R].as_f64(),
+                bias_abs: row[r_col::CONVICTION_AT_ENTRY].as_f64(),
+                latched: (row[r_col::ENTRY_PRICE].as_f64() - row[r_col::STOP_PRICE].as_f64()).abs(),
+                cost: c,
+            });
+        }
+    }
+    let mut n_open_at_end = 0u64;
+    if let Some((_, last)) = record.last()
+        && last[r_col::OPEN].as_bool()
+    {
+        let c = cost.get(record.len() - 1).map(|(_, cr)| cr[1].as_f64()).unwrap_or(0.0);
+        trades.push(Trade {
+            r: last[r_col::UNREALIZED_R].as_f64(),
+            bias_abs: last[r_col::CONVICTION_AT_ENTRY].as_f64(),
+            latched: (last[r_col::ENTRY_PRICE].as_f64() - last[r_col::STOP_PRICE].as_f64()).abs(),
+            cost: c,
+        });
+        n_open_at_end = 1;
+    }
+```
+(The empty-input early return at `:186-202` and `n == 0` are unchanged.)
+
+Net fold `:244-250` — subtract the per-trade cost the node supplied (empty stream ⇒
+every `t.cost == 0.0` ⇒ `net_sum == Σ t.r`, byte-identical to the old cost=0 result):
+```rust
+    // net-of-cost: subtract the cost-model's per-trade cost-in-R (from the cost stream;
+    // an empty stream is the gross-R baseline — every trade's cost is 0.0).
+    let net_sum: f64 = trades.iter().map(|t| t.r - t.cost).sum();
+    let net_expectancy_r = net_sum / n as f64;
+```
+`r_metrics_from_rs` (`:301-353`) is **unchanged** — it carries no cost (the pooled-OOS
+copy stays `net_expectancy_r = expectancy_r` under the cost=0 invariant `:349`).
+
+- [ ] **Step 4: Migrate the cost unit test `summarize_r_net_of_cost_subtracts_round_trip_per_trade` (`:744-755`)**
+
+Rewrite it to drive cost via the stream (the node's emission), preserving its assertion
+that a per-trade round-trip cost is charged in R. Replace the `summarize_r(&record, 1.0)`
+call with a constructed cost stream that emits `1.0/latched` on each closed row (and the
+window-end `open_cost_in_r` if the fixture leaves a position open), and assert the same
+net result the old scalar produced. Keep the fixture's trades; only the cost-injection
+mechanism changes.
+
+- [ ] **Step 5: Migrate every other `summarize_r` call site to the new signature**
+
+The cost-0 sites take the empty stream; verify with grep first (Step 6). Mechanical
+transform `summarize_r(<record>, 0.0)` → `summarize_r(<record>, &[])` at:
+- `crates/aura-analysis/src/lib.rs`: `:609, :627, :638, :647, :660, :667, :670, :681, :702, :719, :734, :741, :749` (the `,0.0` sites; the cross-reducer test `:948` too).
+- `crates/aura-composites/tests/risk_executor.rs`: `:97, :141, :178, :210`.
+- `crates/aura-engine/tests/stage1_r_e2e.rs`: `:80, :121, :207, :403` and the first call at `:232` and `:235`.
+- `crates/aura-engine/tests/streaming_reduction_equivalence.rs`: `:63, :78`.
+- `crates/aura-cli/src/main.rs` (production): `:1266, :1278, :1340, :1381, :1437, :1448, :1509, :1520, :2969` → `summarize_r(<record>, &[])`.
+
+The **non-zero** cost test sites — `crates/aura-analysis/src/lib.rs:750` (`,1.0`, handled
+in Step 4) and `crates/aura-engine/tests/stage1_r_e2e.rs:208` (`,2.0`) and the second
+calls at `:232`/`:235` (`,2.0`) — are rewritten in Task-2 Step 4 / Task-3 Step 7 to
+construct a `2.0`-cost stream aligned to their trade record (build
+`[(ts, [2.0/latched_at_ts, open_cost_at_ts]), …]`) and assert the same net.
+
+- [ ] **Step 6: Verify no `summarize_r(... , <f64>)` call survives**
+
+Run: `git grep -nE 'summarize_r\([^)]*,\s*[0-9]' -- 'crates/**/*.rs'`
+Expected: no output (every call now passes a `&[...]` cost stream; the `, 0.0`/`, 1.0`/
+`, 2.0` second arguments are gone). A hit is an un-migrated site — fix it.
+
+- [ ] **Step 7: Build the workspace — the compile gate**
+
+Run: `cargo build --workspace --all-targets`
+Expected: `0 errors` (every `summarize_r` caller migrated).
+
+- [ ] **Step 8: Run the analysis + cross-crate suites**
+
+Run: `cargo test -p aura-analysis`
+Expected: PASS incl. `summarize_r_folds_cost_stream_exactly_like_the_scalar_over_all_trades`,
+the migrated cost test, and `summarize_r_includes_open_trade_and_matches_r_metrics_from_rs`
+(cross-reducer equality holds at cost=0).
+
+---
+
+## Task 3: Run-path wiring — `net_r_equity` tap, cost stream, `--cost-per-trade`
+
+**Files:**
+- Modify: `crates/aura-cli/src/main.rs` — `stage1_r_graph:2614-2698` (+ its 7 call sites),
+  `run_stage1_r:2917-2971`, `persist_traces_r:2976-2992`, `RunArgs:3004-3008`,
+  `parse_run_args:3015-3074`, `run_dispatch:3078-3091`, `USAGE:3093`.
+- Test: `crates/aura-engine/tests/stage1_r_e2e.rs` (in-graph↔post-run agreement),
+  `crates/aura-cli/tests/cli_run.rs` (net_r_equity persistence; no-cost golden held).
+
+- [ ] **Step 1: Write the failing CLI cost-run test**
+
+Add to `crates/aura-cli/tests/cli_run.rs` (sibling of
+`stage1_r_trace_persists_r_equity_and_charts_it:1599`). Run `aura run --harness stage1-r
+--cost-per-trade 2 --trace <n>`, assert (a) `runs/traces/<n>/net_r_equity.json` exists,
+(b) the run JSON has `net_expectancy_r` strictly less than `expectancy_r`. (Mirror the
+existing test's harness invocation + `runs/traces/.../r_equity.json` assertion.)
+
+```rust
+#[test]
+fn stage1_r_cost_run_persists_net_r_equity_and_charges_cost() {
+    let tmp = tempdir().unwrap();
+    let out = run_cli(&tmp, &["run", "--harness", "stage1-r", "--cost-per-trade", "2", "--trace", "c1"]);
+    assert!(tmp.path().join("runs/traces/c1/net_r_equity.json").exists(),
+        "net_r_equity trace persisted");
+    let v: serde_json::Value = serde_json::from_str(&out).unwrap();
+    let gross = v["metrics"]["r"]["expectancy_r"].as_f64().unwrap();
+    let net = v["metrics"]["r"]["net_expectancy_r"].as_f64().unwrap();
+    assert!(net < gross, "cost charged: net {net} < gross {gross}");
+}
+```
+
+(Use the test module's existing CLI-invocation helper and temp-dir pattern; match the
+`run_cli` / output-capture shape already used at `:1599-1621`.)
+
+- [ ] **Step 2: Run it — fails (no flag, no tap yet)**
+
+Run: `cargo test -p aura-cli stage1_r_cost_run_persists_net_r_equity`
+Expected: FAIL — `--cost-per-trade` is an unknown flag (`parse_run_args` returns the
+usage error) / the run exits non-zero.
+
+- [ ] **Step 3: Add the cost taps to `stage1_r_graph`**
+
+Add a `cost` parameter (the bundle) to `stage1_r_graph` and wire the node + net tap when
+present and not in reduce mode. New signature (append after `reduce: bool`):
+```rust
+    cost: Option<(f64, mpsc::Sender<(Timestamp, Vec<Scalar>)>, mpsc::Sender<(Timestamp, Vec<Scalar>)>)>,
+```
+Inside the existing `if !reduce { … }` block (after the `r_equity` wiring at `:2695`),
+add:
+```rust
+        if let Some((cost_per_trade, tx_net, tx_cost)) = cost {
+            let cost_node = g.add(
+                ConstantCost::builder().bind("cost_per_trade", Scalar::f64(cost_per_trade)),
+            );
+            g.connect(exec.output("closed_this_cycle"), cost_node.input("closed"));
+            g.connect(exec.output("open"), cost_node.input("open"));
+            g.connect(exec.output("entry_price"), cost_node.input("entry_price"));
+            g.connect(exec.output("stop_price"), cost_node.input("stop_price"));
+            // net_r_equity = cum_realized_r + unrealized_r - cum_cost_in_r - open_cost_in_r
+            let net_eq = g.add(
+                LinComb::builder(4)
+                    .bind("weights[0]", Scalar::f64(1.0))
+                    .bind("weights[1]", Scalar::f64(1.0))
+                    .bind("weights[2]", Scalar::f64(-1.0))
+                    .bind("weights[3]", Scalar::f64(-1.0)),
+            );
+            g.connect(exec.output("cum_realized_r"), net_eq.input("term[0]"));
+            g.connect(exec.output("unrealized_r"), net_eq.input("term[1]"));
+            g.connect(cost_node.output("cum_cost_in_r"), net_eq.input("term[2]"));
+            g.connect(cost_node.output("open_cost_in_r"), net_eq.input("term[3]"));
+            let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_net));
+            g.connect(net_eq.output("value"), net_rec.input("col[0]"));
+            // the cost stream the post-run fold reads: [cost_in_r, open_cost_in_r]
+            let cost_rec = g.add(
+                Recorder::builder(vec![ScalarKind::F64, ScalarKind::F64], Firing::Any, tx_cost),
+            );
+            g.connect(cost_node.output("cost_in_r"), cost_rec.input("col[0]"));
+            g.connect(cost_node.output("open_cost_in_r"), cost_rec.input("col[1]"));
+        }
+```
+Add `use aura_std::ConstantCost;` to the imports if not already pulled via the existing
+`aura_std::{…}` use (LinComb/Recorder are already imported).
+
+- [ ] **Step 4: Thread `None` through every non-run `stage1_r_graph` caller**
+
+`stage1_r_graph` is called from the sweep/walkforward/mc grid builders. Enumerate them
+(line numbers may have drifted — grep is the authoritative channel):
+
+Run: `git grep -n 'stage1_r_graph(' -- crates/aura-cli/src/main.rs`
+Expected (recon estimate): `:1202, :1236, :1294, :1308, :1324, :1359` (grid builders) plus
+`:2928` (the run caller, updated in Step 5). Append `, None` to each grid-builder call
+(cycle-1 scope: no cost on the reduce-mode sweep path). A call this step misses is a hard
+compile error at the Step-9/10 gate.
+
+- [ ] **Step 5: Thread cost through `run_stage1_r`**
+
+Change `run_stage1_r`'s signature to `fn run_stage1_r(data: RunData, trace: Option<&str>,
+cost: Option<f64>) -> RunReport`. Create the two extra channels, pass the bundle when
+cost is set, drain the streams, fold cost into `summarize_r`, and persist `net_r_equity`:
+```rust
+    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_net, rx_net) = mpsc::channel();
+    let (tx_cost, rx_cost) = mpsc::channel();
+    let cost_bundle = cost.map(|c| (c, tx_net, tx_cost));
+    let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false, cost_bundle)
+        .compile_with_params(&[])
+        .expect("valid stage1-r blueprint");
+```
+After `h.run(sources)` and the existing drains, add:
+```rust
+    let net_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_net.try_iter().collect();
+    let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
+```
+Persist + fold (replace `:2966-2969`):
+```rust
+    if let Some(name) = trace {
+        persist_traces_r(name, &manifest, &eq_rows, &ex_rows, &req_rows, &net_rows);
+    }
+    let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
+    metrics.r = Some(summarize_r(&r_rows, &cost_rows));
+```
+(When `cost` is `None`, `cost_bundle` is `None` ⇒ no cost node / no `net_rows` / `cost_rows`
+empty ⇒ `summarize_r(&r_rows, &[])` ⇒ net == gross, and `persist_traces_r` gets an empty
+`net_rows`.)
+
+- [ ] **Step 6: Add the `net_r_equity` trace to `persist_traces_r`**
+
+Add a `net_rows: &[(Timestamp, Vec<Scalar>)]` parameter and emit the tap only when
+non-empty (so a no-cost run's on-disk trace set is byte-unchanged):
+```rust
+fn persist_traces_r(
+    name: &str,
+    manifest: &RunManifest,
+    eq_rows: &[(Timestamp, Vec<Scalar>)],
+    ex_rows: &[(Timestamp, Vec<Scalar>)],
+    req_rows: &[(Timestamp, Vec<Scalar>)],
+    net_rows: &[(Timestamp, Vec<Scalar>)],
+) {
+    let mut taps = vec![
+        ColumnarTrace::from_rows("equity", &[ScalarKind::F64], eq_rows),
+        ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], ex_rows),
+        ColumnarTrace::from_rows("r_equity", &[ScalarKind::F64], req_rows),
+    ];
+    if !net_rows.is_empty() {
+        taps.push(ColumnarTrace::from_rows("net_r_equity", &[ScalarKind::F64], net_rows));
+    }
+    if let Err(e) = TraceStore::open("runs").write(name, manifest, &taps) {
+        eprintln!("aura: trace persist failed: {e}");
+        std::process::exit(2);
+    }
+}
+```
+
+- [ ] **Step 7: Rewrite the e2e cost properties + add the in-graph↔post-run agreement assertion**
+
+In `crates/aura-engine/tests/stage1_r_e2e.rs:194-241`, rewrite the two scalar-cost
+properties (`net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_dist`
+and `net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it`) to construct a cost stream
+(the `ConstantCost` emission for the test's trades) and pass it to `summarize_r` — same
+assertions, cost now sourced from the node/stream not the scalar. Add an assertion that
+the final `net_r_equity` sample equals `cum_realized_r + unrealized_r − total_cost` (the
+post-run net total) — the in-graph↔post-run agreement. (If building the full graph is
+heavy here, assert the algebraic identity on the constructed records directly.)
+
+- [ ] **Step 8: Add `--cost-per-trade` to `RunArgs` / `parse_run_args` / `run_dispatch` / `USAGE`**
+
+`RunArgs` `:3004-3008`:
+```rust
+struct RunArgs {
+    harness: HarnessKind,
+    data: RunData,
+    trace: Option<String>,
+    cost: Option<f64>,
+}
+```
+`parse_run_args` — declare `let mut cost: Option<f64> = None;` near `:3024`, add a flag
+arm (alongside `--trace` at `:3056`):
+```rust
+            "--cost-per-trade" if cost.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                cost = Some(value.parse().map_err(|_| usage())?);
+                tail = t;
+            }
+```
+and add `cost` to the `Ok(RunArgs { … })` at `:3073`. Update the `usage` string `:3017`
+to include `[--cost-per-trade <f64>]`.
+
+`run_dispatch` `:3083` — thread it into the stage1-r arm (other harness arms ignore cost
+this cycle):
+```rust
+        (HarnessKind::Stage1R, data) => run_stage1_r(data, trace, args.cost),
+```
+Update the top-level `USAGE` const `:3093` to add `[--cost-per-trade <f64>]` to the
+`aura run` synopsis.
+
+Thread the **other** `run_stage1_r` caller too — the test at `crates/aura-cli/src/main.rs:4311`
+calls `run_stage1_r(...)` and must pass the new arg or the build breaks. Find both callers
+and update them: `git grep -n 'run_stage1_r(' -- crates/aura-cli/src/main.rs` (expect the
+dispatch `:3083` and the test `:4311`); the test caller passes `None` (no cost) unless it
+specifically asserts a cost run.
+
+- [ ] **Step 9: Run the run-path tests**
+
+Run: `cargo test -p aura-cli stage1_r_cost_run_persists_net_r_equity`
+Expected: PASS — `net_r_equity.json` persisted; `net < gross`.
+
+Run: `cargo test -p aura-cli stage1_r_single_run_output_golden`
+Expected: PASS, byte-identical — the no-cost golden at `cli_run.rs:1686`
+(`"net_expectancy_r":1.2710005136982836`) is unchanged (no-cost run ⇒ `summarize_r(&r_rows,
+&[])` ⇒ net == gross, no `net_r_equity` tap).
+
+- [ ] **Step 10: Full regression gate**
+
+Run: `cargo test --workspace`
+Expected: PASS, the full suite green (the 665-test baseline plus the new tests), no
+golden moved.
+
+Run: `cargo clippy --workspace --all-targets -- -D warnings`
+Expected: clean.