//! The `RiskExecutor` composite (#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`
//! (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_composites::{risk_executor, StopRule};
use aura_engine::{summarize_r, GraphBuilder, RunMetrics, VecSource};
use aura_std::{Recorder, PM_FIELD_NAMES, PM_RECORD_KINDS};
use std::sync::mpsc::channel;

// The dense-record columns this fixture reads, named in lockstep with the sibling
// `r_sma_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;

/// 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(StopRule::Fixed(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, &[]);
    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` (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 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, &[]);
    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, bias_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");
}

/// Property: the RiskExecutor embeds the `vol_stop` composite (the new `StopRule::Vol`
/// arm) and folds to a finite RMetric — the volatility-defined default the `r-sma`
/// harness uses. A short k·σ stop over a rising-then-falling path opens a trade and
/// (stop or window-end) closes at least one, so the fold is non-empty and sane.
#[test]
fn risk_executor_vol_stop_arm_bootstraps_and_folds() {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("vol_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0));
    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() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("vol_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let path = [100.0, 101.0, 100.0, 101.0, 103.0, 106.0, 104.0, 99.0, 95.0, 96.0];
    let stream: Vec<(Timestamp, Scalar)> =
        path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    let ledger: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    let m = summarize_r(&ledger, &[]);
    assert!(m.n_trades >= 1, "the vol-stop executor must fold at least one trade");
    assert!(m.sqn.is_finite(), "SQN must be finite, got {}", m.sqn);
}

/// Property (#262): the RiskExecutor embeds the `VolTfStop` primitive (the new
/// `StopRule::VolTf` arm) and folds to a finite RMetric, proving the arm's
/// `price -> stop_distance` port/kind/firing wiring actually bootstraps and
/// runs — not merely compiles. Timestamps are real epoch-ns minute ticks
/// (`i * 60s` in ns) so `period_minutes: 1` rolls a bucket every cycle,
/// matching the node's own bucket-rollover contract; a short k·σ stop over a
/// rising-then-falling path opens a trade and closes at least one.
#[test]
fn risk_executor_vol_tf_stop_arm_bootstraps_and_folds() {
    const MINUTE_NS: i64 = 60 * 1_000_000_000;
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("vol_tf_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(
        StopRule::VolTf { period_minutes: 1, length: 1, k: 2.0 },
        1.0,
    ));
    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() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("vol_tf_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let path = [100.0, 101.0, 100.0, 101.0, 103.0, 106.0, 104.0, 99.0, 95.0, 96.0];
    let stream: Vec<(Timestamp, Scalar)> =
        path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64 * MINUTE_NS), Scalar::f64(p))).collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    let ledger: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    let m = summarize_r(&ledger, &[]);
    assert!(m.n_trades >= 1, "the vol_tf-stop executor must fold at least one trade");
    assert!(m.sqn.is_finite(), "SQN must be finite, got {}", m.sqn);
}

/// Drain one `risk_executor`-variant harness over the shared rising-then-falling path,
/// folding the dense R-record. `exec` is the variant under test (bound or open); `params`
/// is the positional point the open variant needs (empty for the bound one). Shared by the
/// open-vs-bound equivalence test below so the two arms run byte-identical inputs.
fn fold_executor(exec: aura_engine::Composite, params: Vec<Scalar>) -> aura_engine::RMetrics {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("vol_open_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(exec);
    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() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("vol_open_harness wires")
        .bootstrap_with_params(params)
        .expect("bootstraps");
    let path = [100.0, 101.0, 100.0, 101.0, 103.0, 106.0, 104.0, 99.0, 95.0, 96.0];
    let stream: Vec<(Timestamp, Scalar)> =
        path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    let ledger: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    summarize_r(&ledger, &[])
}

/// Property (#137): `risk_executor_vol_open` exposes the vol-stop's EWMA length and
/// `k`-multiplier as exactly two FREE `param_space` axes — the I64 `length` under the
/// `vol_stop.stop_length.length` suffix and the F64 `weights[0]` under
/// `vol_stop.stop_k.weights[0]` — so a sweep can grid the stop timescale through the
/// `.axis(..)` mechanism. (As the root composite here its names carry no outer prefix;
/// embedded in a harness they gain the enclosing node's path, hence the suffix match.)
/// The bound `risk_executor(StopRule::Vol { .. }, ..)` exposes none of these (its knobs
/// are constants), which is why the gridding variant exists.
#[test]
fn risk_executor_vol_open_exposes_the_two_stop_knobs_as_free_axes() {
    let open = aura_composites::risk_executor_vol_open(1.0);
    let space = open.param_space();
    let length = space
        .iter()
        .find(|p| p.name.ends_with("vol_stop.stop_length.length"))
        .expect("open vol-stop exposes a length axis");
    assert_eq!(length.kind, ScalarKind::I64, "stop length axis is I64");
    let k = space
        .iter()
        .find(|p| p.name.ends_with("vol_stop.stop_k.weights[0]"))
        .expect("open vol-stop exposes a k axis");
    assert_eq!(k.kind, ScalarKind::F64, "stop k axis is F64");
    // exactly the two stop knobs are open (the rest — Delay lag, Sizer risk_budget — are bound).
    assert_eq!(space.len(), 2, "only the two stop knobs are free: {space:?}");

    // the bound arm has no open knobs at all (its stop is a constant).
    assert!(
        aura_composites::risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0)
            .param_space()
            .is_empty(),
        "the bound vol-stop arm exposes no sweepable knobs"
    );
}

/// Property (#137): the OPEN vol-stop arm, bootstrapped at a given `(length, k)`, folds to
/// the SAME R-outcome as the BOUND arm at the same values — the byte-equivalence the
/// no-flags r-sma sweep relies on to stay golden (the gridding variant only frees the
/// two knobs; topology and every other constant are unchanged). The open arm's positional
/// point is `[length, k]` in `param_space` slot order (length first, then k).
#[test]
fn risk_executor_vol_open_matches_the_bound_arm_at_the_same_point() {
    let bound = fold_executor(risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0), vec![]);
    let open = fold_executor(
        aura_composites::risk_executor_vol_open(1.0),
        vec![Scalar::i64(3), Scalar::f64(2.0)],
    );
    assert_eq!(open.n_trades, bound.n_trades, "trade count must match the bound arm");
    assert_eq!(open.sqn.to_bits(), bound.sqn.to_bits(), "SQN must be bit-identical to the bound arm");
    assert_eq!(
        open.expectancy_r.to_bits(),
        bound.expectancy_r.to_bits(),
        "E[R] must be bit-identical to the bound arm"
    );
}