//! The `RiskExecutor` composite (Stage-1, #128): a per-symbol risk-based executor over a
//! bias stream — `stop-rule -> Sizer -> position-management`, exposing the dense R-record.
//! Proves the composite bootstraps, runs end-to-end, and folds to the SAME R-outcomes as
//! the hand-wired iter-1 chain, and that R is invariant under the Sizer's `risk_budget`
//! (Stage-1 feed-forward). The Veto is a DOCUMENTED SEAM, not a runtime node (a
//! pass-through identity is exactly what C19/C23 DCE deletes), so it appears nowhere here.
use aura_core::{
    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar,
    ScalarKind, Timestamp,
};
use aura_engine::{summarize_r, Composite, GraphBuilder, RunMetrics, VecSource};
use aura_std::{FixedStop, PositionManagement, Recorder, Sizer, PM_FIELD_NAMES, PM_RECORD_KINDS};
use std::sync::mpsc::channel;

// The dense-record columns this fixture reads, named in lockstep with the sibling
// `stage1_r_e2e.rs` (which names `REALIZED_R = 1` the same way) so the cross-crate
// `PM_FIELD_NAMES` layout is never referenced by a bare literal.
const REALIZED_R: usize = 1;
const SIZE: usize = 10;

/// The per-symbol RiskExecutor: open input roles `bias` + `price`, internal
/// `FixedStop(stop_distance) -> Sizer(risk_budget) -> PositionManagement`, exposing every
/// field of PM's dense R-record. Price fans to BOTH the stop-rule and PM; bias fans to the
/// Sizer and PM; the Sizer's `size` feeds PM's size slot. (Stage-1 ships `FixedStop` here;
/// the volatility stop is a drop-in composite, see `vol_stop_composite.rs`.)
fn risk_executor(stop_distance: f64, risk_budget: f64) -> Composite {
    let mut g = GraphBuilder::new("risk_executor");
    let bias = g.input_role("bias");
    let price = g.input_role("price");
    let stop = g.add(FixedStop::builder().bind("distance", Scalar::f64(stop_distance)));
    let sizer = g.add(Sizer::builder().bind("risk_budget", Scalar::f64(risk_budget)));
    let pm = g.add(PositionManagement::builder());
    g.feed(price, [stop.input("price"), pm.input("price")]); // price fans to stop + PM
    g.feed(bias, [sizer.input("bias"), pm.input("bias")]); // bias fans to sizer + PM
    g.connect(stop.output("stop_distance"), sizer.input("stop_distance"));
    g.connect(stop.output("stop_distance"), pm.input("stop_distance"));
    g.connect(sizer.output("size"), pm.input("size")); // the flat-1R size into PM
    for field in PM_FIELD_NAMES {
        g.expose(pm.output(field), field);
    }
    g.build().expect("risk_executor wires")
}

/// An always-long strategy stand-in: emits a constant `+1` bias once price is present. The
/// strategy is upstream of the RiskExecutor; this is the minimal in-graph producer so the
/// whole chain runs off the single price source (a second bias *source* would k-way-merge
/// into separate cycles and mark stale prices — see harness.rs C4 tie-breaking).
struct ConstLongBias {
    out: [Cell; 1],
}
impl ConstLongBias {
    fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "ConstLongBias",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "price".into() }],
                output: vec![FieldSpec { name: "bias".into(), kind: ScalarKind::F64 }],
                params: vec![],
            },
            |_| Box::new(ConstLongBias { out: [Cell::from_f64(0.0)] }),
        )
    }
}
impl Node for ConstLongBias {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }
    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        if ctx.f64_in(0).is_empty() {
            return None;
        }
        self.out[0] = Cell::from_f64(1.0);
        Some(&self.out)
    }
    fn label(&self) -> String {
        "ConstLongBias".into()
    }
}

/// Bootstrap a harness: one price source -> ConstLongBias (the strategy) + RiskExecutor;
/// the RiskExecutor's dense R-record into a Recorder. Returns the drained ledger.
fn run_executor(prices: &[f64], stop_distance: f64, risk_budget: f64) -> Vec<(Timestamp, Vec<Scalar>)> {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("risk_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(stop_distance, risk_budget));
    let rec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx));
    g.feed(price, [strat.input("price"), exec.input("price")]);
    g.connect(strat.output("bias"), exec.input("bias"));
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        // `input` takes a `&'static str` (names resolve at the authoring boundary, C23);
        // leak the per-column port name so the runtime-built `col[i]` satisfies that bound.
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("risk_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let stream: Vec<(Timestamp, Scalar)> = prices
        .iter()
        .enumerate()
        .map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p)))
        .collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    rx.try_iter().collect()
}

/// Property: the composite bootstraps, runs, and folds to the documented hand value. A
/// constant long over a monotonically rising price never stops or flips, so the position
/// is open at window end: entry @100 latched on FixedStop(10), last mark @105 -> window-end
/// R = (105-100)/10 = +0.5, the only trade -> expectancy 0.5 (the bootstrapped-composite
/// twin of the iter-1 hand-wired `open_at_window_end_is_folded_into_expectancy_not_dropped`).
#[test]
fn risk_executor_bootstraps_and_folds_to_expected_rmetric() {
    let ledger = run_executor(&[100.0, 102.0, 105.0], 10.0, 1.0);
    let m = summarize_r(&ledger, 0.0);
    assert_eq!(m.n_open_at_end, 1, "the open position must be counted");
    assert_eq!(m.n_trades, 1);
    assert!((m.expectancy_r - 0.5).abs() < 1e-9, "window-end R = (105-100)/10; got {}", m.expectancy_r);
}

/// Property: R is invariant under the Sizer's `risk_budget` (Stage-1 feed-forward). The
/// same price path at two budgets yields a bit-identical realised-R ledger, while the
/// `size` column scales with the budget — proving size flows through the Sizer into PM yet
/// never touches R.
#[test]
fn risk_executor_r_invariant_under_risk_budget() {
    let path = [100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0];
    let a = run_executor(&path, 5.0, 1.0);
    let b = run_executor(&path, 5.0, 8.0);
    assert_eq!(a.len(), b.len());
    assert!(!a.is_empty());
    // index realized_r and size by the producer's dense-record layout (named consts above).
    let realized =
        |rows: &[(Timestamp, Vec<Scalar>)]| rows.iter().map(|(_, r)| r[REALIZED_R].as_f64()).collect::<Vec<_>>();
    let size =
        |rows: &[(Timestamp, Vec<Scalar>)]| rows.iter().map(|(_, r)| r[SIZE].as_f64()).collect::<Vec<_>>();
    assert_eq!(realized(&a), realized(&b), "realized_r must be invariant under risk_budget");
    // size scaled 8x: at least one cycle has a nonzero size that is exactly 8x a's (same
    // stop distance per cycle, budget 1 -> 8).
    let (sa, sb) = (size(&a), size(&b));
    assert!(
        sa.iter().zip(&sb).any(|(x, y)| *x > 0.0 && (*y - 8.0 * *x).abs() < 1e-9),
        "risk_budget must scale size 8x: a={sa:?} b={sb:?}"
    );
}

/// Property: **a real folded `RMetrics` survives the `RunMetrics.r` serde round-trip
/// byte-for-byte (the Stage-1 on-disk back-compat contract), driven from an actual run —
/// not a hand-built literal.** A bootstrapped RiskExecutor run is folded by `summarize_r`
/// and the result is attached as `RunMetrics.r = Some(..)`; serializing then
/// deserializing must reproduce an equal value, and the `r` key must be present. The
/// `report.rs` unit test asserts this on a hand-written `RMetrics`; here the value is
/// whatever the live fold produced, so a future `RMetrics` field that the fold sets but
/// serde forgets to thread would round-trip-diverge here (the literal test cannot see it).
/// The sibling pip-only-`None` path (omitted from JSON) is the inverse, covered in report.rs.
#[test]
fn folded_rmetrics_survives_runmetrics_serde_round_trip() {
    let ledger = run_executor(&[100.0, 104.0, 108.0, 110.0, 102.0, 96.0, 94.0], 5.0, 1.0);
    let folded = summarize_r(&ledger, 0.0);
    assert!(folded.n_trades >= 1, "the run must produce at least one trade to fold");
    let m = RunMetrics { total_pips: 0.0, max_drawdown: 0.0, exposure_sign_flips: 0, r: Some(folded) };
    let json = serde_json::to_string(&m).expect("serialize a run's RunMetrics with an r block");
    assert!(json.contains("\"r\":{"), "the folded r block must be present in the JSON: {json}");
    let back: RunMetrics = serde_json::from_str(&json).expect("deserialize round-trips");
    assert_eq!(back, m, "a live-folded RMetrics must round-trip byte-for-byte");
}