feat(0081): cost-model graph cycle 1 — ConstantCost node + net-R seam

Lands the first cycle of the "Cost-model graph (in R)" milestone: cost is
now a composable in-graph node, not a post-run scalar (C10).

- ConstantCost (aura-std): a flat round-trip cost per trade, emitted in R
  as a 3-field record {cost_in_r, cum_cost_in_r, open_cost_in_r} isomorphic
  to PositionManagement's R-triple. R-pure: cost_in_r = cost_per_trade /
  |entry-stop|; notional cancels, no equity held.
- summarize_r folds the cost stream into net_expectancy_r, replacing its
  scalar round_trip_cost (one home for cost, no double-count). Positional
  co-temporal join; an empty stream is the gross-R baseline.
- net_r_equity tap (stage1_r_graph): LinComb(4) over [cum_realized_r,
  unrealized_r, -cum_cost_in_r, -open_cost_in_r] -> Recorder, a sibling of
  r_equity; --cost-per-trade on the run path wires the node + tap + fold.
- A no-cost run is byte-identical: no net_r_equity tap, net == gross, the
  C18 golden unchanged.

Tests: exact-subsumption (the node-stream net == mean(r_i - cost_i) over
ALL trades incl. the window-end open one), the in-graph net_r_equity tap +
net < gross E2E, the no-cost golden held. Full suite green; clippy clean.

Implement-time decisions ratified (recorded on #148): the cost stream is
recorded 3-wide (the node's full output; summarize_r reads col 0/2, col 1
feeds the in-graph net curve) — resolving a 2-wide/3-wide inconsistency in
the plan's Step 3; Trade drops its `latched` field (cost now arrives in R),
so r_col ENTRY_PRICE/STOP_PRICE become #[allow(dead_code)] layout contract.

Scoped to the run path (non-reduce); sweep/walkforward/mc pass None — cost
on the reduce-mode sweep path and OOS-pooling cost are a deferred follow-up.

Known pre-existing concern (not this cycle): aura-std lib.rs module doc
still names "realistic broker profiles", retired by the C10 rework.

refs #148

crates/aura-engine/tests/stage1_r_e2e.rs

@@ -27,7 +27,7 @@
 
 use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
 use aura_engine::{PositionAction, RMetrics, derive_position_events, summarize_r};
-use aura_std::{FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement};
+use aura_std::{ConstantCost, FixedStop, PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement};
 
 // The indices `report.rs::summarize_r` reads the dense record by. Re-declared here
 // (the consumer's `r_col` module is private to `aura-engine`) so the guard test can
@@ -41,6 +41,11 @@ const STOP_PRICE: usize = 7;
 const CONVICTION_AT_ENTRY: usize = 9;
 const SIZE: usize = 10;
 const DIRECTION: usize = 4;
+const CUM_REALIZED_R: usize = 13;
+// `ConstantCost`'s 3-wide output schema [cost_in_r, cum_cost_in_r, open_cost_in_r] — the
+// run-path's in-graph `net_r_equity` tap reads cum (1) + open (2); summarize_r reads 0/2.
+const CUM_COST_IN_R: usize = 1;
+const OPEN_COST_IN_R: usize = 2;
 
 /// Property: the real producer->consumer seam composes. Driving `FixedStop` ->
 /// `PositionManagement` directly (node-by-node, not a bootstrapped graph) over a
@@ -77,7 +82,7 @@ fn synthetic_long_then_stop_produces_a_sane_rmetric() {
             ));
         }
     }
-    let m = summarize_r(&ledger, 0.0);
+    let m = summarize_r(&ledger, &[]);
     assert!(m.n_trades >= 1, "expected at least one trade (entry then stop), got {m:?}");
     assert!(m.expectancy_r.is_finite());
     // entered ~100, stopped at the 95 level (price gapped to 94 through it) -> a loss.
@@ -118,7 +123,26 @@ fn run_chain_ledger(stop: &mut dyn Node, path: &[(f64, f64)]) -> Vec<(Timestamp,
 /// over a `(bias, price)` path. This is the actual producer->consumer seam,
 /// node-by-node; the resulting `RMetrics` is what a recorded Stage-1 run would report.
 fn run_chain(stop: &mut dyn Node, path: &[(f64, f64)]) -> RMetrics {
-    summarize_r(&run_chain_ledger(stop, path), 0.0)
+    summarize_r(&run_chain_ledger(stop, path), &[])
+}
+
+/// The co-temporal cost stream a `ConstantCost(c)` node would emit over `ledger`: per row,
+/// the round-trip cost `c` (price units) expressed in R via that row's latched distance
+/// (`|entry_price - stop_price|`), placed in BOTH the closed-cost (col 0) and the open-cost
+/// (col 2) slots of the producer's 3-wide `[cost_in_r, cum_cost_in_r, open_cost_in_r]` row.
+/// A zero-latched (flat) row contributes no cost. `summarize_r` reads col 0 for a closed
+/// trade and col 2 for the window-end open trade, so this reproduces the retired scalar
+/// `round_trip_cost` argument exactly — the cost recovered from the producer's own
+/// `entry_price`/`stop_price` columns, end-to-end. Col 1 (cum) is unread by `summarize_r`.
+fn const_cost_stream(ledger: &[(Timestamp, Vec<Scalar>)], c: f64) -> Vec<(Timestamp, Vec<Scalar>)> {
+    ledger
+        .iter()
+        .map(|(ts, row)| {
+            let latched = (row[ENTRY_PRICE].as_f64() - row[STOP_PRICE].as_f64()).abs();
+            let cir = if latched > 0.0 { c / latched } else { 0.0 };
+            (*ts, vec![Scalar::f64(cir), Scalar::f64(0.0), Scalar::f64(cir)])
+        })
+        .collect()
 }
 
 /// A constant-long-bias path (`bias = +1` every cycle) over `prices`.
@@ -204,8 +228,8 @@ fn clean_stop_folds_to_exactly_minus_one_r() {
 fn net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_dist() {
     let mut stop = FixedStop::new(10.0);
     let ledger = run_chain_ledger(&mut stop, &long_path(&[100.0, 105.0, 110.0]));
-    let gross = summarize_r(&ledger, 0.0);
-    let net = summarize_r(&ledger, 2.0);
+    let gross = summarize_r(&ledger, &[]);
+    let net = summarize_r(&ledger, &const_cost_stream(&ledger, 2.0));
     assert_eq!(gross.n_trades, 1);
     assert!((gross.expectancy_r - 1.0).abs() < 1e-9, "gross window-end R = (110-100)/10; got {}", gross.expectancy_r);
     // gross is untouched by the cost; only net_expectancy_r absorbs it.
@@ -229,10 +253,10 @@ fn net_of_cost_charges_one_round_trip_per_trade_through_the_recovered_latched_di
 fn net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it() {
     let mut tight = FixedStop::new(5.0);
     let tight_l = run_chain_ledger(&mut tight, &long_path(&[100.0, 102.0, 105.0]));
-    let (tg, tn) = (summarize_r(&tight_l, 0.0), summarize_r(&tight_l, 2.0));
+    let (tg, tn) = (summarize_r(&tight_l, &[]), summarize_r(&tight_l, &const_cost_stream(&tight_l, 2.0)));
     let mut wide = FixedStop::new(20.0);
     let wide_l = run_chain_ledger(&mut wide, &long_path(&[100.0, 102.0, 105.0]));
-    let (wg, wn) = (summarize_r(&wide_l, 0.0), summarize_r(&wide_l, 2.0));
+    let (wg, wn) = (summarize_r(&wide_l, &[]), summarize_r(&wide_l, &const_cost_stream(&wide_l, 2.0)));
     let tight_cost = tg.expectancy_r - tn.net_expectancy_r; // 2.0/5 = 0.4
     let wide_cost = wg.expectancy_r - wn.net_expectancy_r; // 2.0/20 = 0.1
     assert!((tight_cost - 0.4).abs() < 1e-9, "tight cost = 2/5; got {tight_cost}");
@@ -240,6 +264,75 @@ fn net_of_cost_is_charged_per_r_so_a_wider_stop_dilutes_it() {
     assert!(tight_cost > wide_cost, "a tighter stop must pay more R per fixed price-unit cost");
 }
 
+/// Drive the REAL `aura_std::ConstantCost(c)` node over a recorded PM `ledger`,
+/// node-by-node off the producer's own `closed`/`open`/`entry_price`/`stop_price`
+/// columns — the actual cost producer the run-path graph taps (not an algebraic stand-in
+/// like `const_cost_stream`). Returns the node's 3-wide `[cost_in_r, cum_cost_in_r,
+/// open_cost_in_r]` rows, co-temporal 1:1 with `ledger`. Fresh single-element columns per
+/// cycle (the node's lookback is 1), matching the node's own unit-test driving pattern.
+fn const_cost_node_stream(
+    ledger: &[(Timestamp, Vec<Scalar>)],
+    c: f64,
+) -> Vec<(Timestamp, Vec<Scalar>)> {
+    let mut node = ConstantCost::new(c);
+    ledger
+        .iter()
+        .map(|(ts, row)| {
+            let mut cols = 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
+            ];
+            cols[0].push(Scalar::bool(row[CLOSED].as_bool())).unwrap();
+            cols[1].push(Scalar::bool(row[OPEN].as_bool())).unwrap();
+            cols[2].push(Scalar::f64(row[ENTRY_PRICE].as_f64())).unwrap();
+            cols[3].push(Scalar::f64(row[STOP_PRICE].as_f64())).unwrap();
+            let out = node.eval(Ctx::new(&cols, *ts)).expect("cost row co-temporal with PM record");
+            (*ts, out.iter().map(|cell| Scalar::f64(cell.f64())).collect())
+        })
+        .collect()
+}
+
+/// Property (Task 3, the run-path net seam): **the in-graph `net_r_equity` tap and the
+/// post-run `summarize_r` net fold agree on the same recorded streams.** The run-path
+/// graph taps a per-cycle `net_r_equity = cum_realized_r + unrealized_r − cum_cost_in_r −
+/// open_cost_in_r` (a `LinComb` over the executor + `ConstantCost` outputs), while
+/// `summarize_r` independently folds the same PM records + cost stream into
+/// `net_expectancy_r`. The FINAL `net_r_equity` sample must equal the post-run net total
+/// (`net_expectancy_r × n_trades` = Σ over trades of `r − cost`) — a divergence would mean
+/// the charted net-R-equity series and the reported net metric disagree. The path carries
+/// one clean stopped loser AND a window-end open trade, so both the cumulative-close-cost
+/// (`cum_cost_in_r`) and the window-end open-cost (`open_cost_in_r`) terms bite. The cost is
+/// produced by the real `ConstantCost` node, the same producer the run-path wires.
+#[test]
+fn net_r_equity_final_sample_agrees_with_summarize_r_net_total() {
+    let mut stop = FixedStop::new(10.0);
+    // open long @100, stop out at the 90 level (bias->0): a clean -1R closed loser; then a
+    // fresh long @100 held up to 105, open at window end (+0.5R). Closed-cost charges
+    // cum_cost_in_r; the open trade charges open_cost_in_r — both net terms exercised.
+    let ledger = run_chain_ledger(
+        &mut stop,
+        &[(1.0, 100.0), (1.0, 100.0), (0.0, 90.0), (1.0, 100.0), (1.0, 102.0), (1.0, 105.0)],
+    );
+    let cost = const_cost_node_stream(&ledger, 2.0);
+    let m = summarize_r(&ledger, &cost);
+    assert_eq!(m.n_trades, 2, "one closed loser + one window-end open trade");
+    assert_eq!(m.n_open_at_end, 1, "the second long is still open at window end");
+    // post-run net total = mean net R × trade count = Σ (r − cost) over all trades.
+    let post_run_net_total = m.net_expectancy_r * m.n_trades as f64;
+    // in-graph final net_r_equity sample = the LinComb(1,1,−1,−1) at the last cycle.
+    let (_, last_pm) = ledger.last().unwrap();
+    let (_, last_cost) = cost.last().unwrap();
+    let net_eq_final = last_pm[CUM_REALIZED_R].as_f64() + last_pm[UNREALIZED_R].as_f64()
+        - last_cost[CUM_COST_IN_R].as_f64()
+        - last_cost[OPEN_COST_IN_R].as_f64();
+    assert!(
+        (net_eq_final - post_run_net_total).abs() < 1e-9,
+        "in-graph net_r_equity {net_eq_final} must equal post-run net total {post_run_net_total}",
+    );
+}
+
 /// Property (#130): **`sqn` and `sqn_normalized` (SQN100) are computed from the
 /// producer's *recorded* dense records, not hand-built rows, and below the
 /// 100-trade cap the two are identical.** Every `summarize_r` test that touches
@@ -400,7 +493,7 @@ fn derived_event_table_agrees_with_r_metrics_over_one_ledger() {
     let mut stop = FixedStop::new(5.0);
     let ledger =
         run_chain_ledger(&mut stop, &long_path(&[100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0]));
-    let m: RMetrics = summarize_r(&ledger, 0.0);
+    let m: RMetrics = summarize_r(&ledger, &[]);
     let events = derive_position_events(&ledger, 7);
 
     let closes = events.iter().filter(|e| e.action == PositionAction::Close).count() as u64;

crates/aura-engine/tests/streaming_reduction_equivalence.rs

@@ -60,7 +60,7 @@ fn gated_recorder_then_summarize_r_equals_raw_summarize_r() {
     .collect();
 
     // RAW: summarize_r over the whole dense record.
-    let raw = summarize_r(&full, 0.0);
+    let raw = summarize_r(&full, &[]);
 
     // FOLDED: drive GatedRecorder (gate = CLOSED), collect its emitted rows, fold.
     let (tx, rx) = mpsc::channel();
@@ -75,7 +75,7 @@ fn gated_recorder_then_summarize_r_equals_raw_summarize_r() {
     }
     g.finalize();
     let folded_rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
-    let folded = summarize_r(&folded_rows, 0.0);
+    let folded = summarize_r(&folded_rows, &[]);
 
     assert_eq!(folded, raw, "GatedRecorder+summarize_r must equal raw summarize_r");
 }