car-server-core 0.49.0

//! `evolution.plan` + `evolution.run` — the self-evolution governor's live
//! daemon surface, end-to-end through `run_dispatch` (arXiv 2507.21046, the
//! "remaining daemon step" of `docs/proposals/self-evolution-governor.md`).
//!
//! Verifies: the five-component live signal assembly (a component with no
//! observable source is omitted, not fabricated), the dry-run dispatch
//! skeleton (no side effects, honest "no inference engine" for the
//! inference-backed Skills arm), and the real Memory arm (consolidate runs,
//! its maintenance sizing is recorded, and the cycle is audited as an
//! `EvolutionTriggered` event on the session log).

use car_memgine::{MemgineEngine, SkillOutcome, SkillTrigger};
use car_server_core::{run_dispatch, ServerState, ServerStateConfig};
use chrono::Utc;
use futures::{SinkExt, StreamExt};
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::Arc;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::sync::Mutex;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{accept_async, connect_async, MaybeTlsStream, WebSocketStream};

fn state_with_engine(journal_dir: std::path::PathBuf, engine: MemgineEngine) -> Arc<ServerState> {
    // Per-test approval journal — never the real user's ~/.car/approvals.jsonl.
    let approvals = journal_dir.join("approvals.jsonl");
    let cfg = ServerStateConfig::new(journal_dir)
        .with_shared_memgine(Arc::new(Mutex::new(engine)))
        .with_approval_journal(approvals);
    Arc::new(ServerState::with_config(cfg))
}

/// Seed an engine whose live signals put Memory and Skills under evolvable
/// pressure: ≥20 facts with >20% flagged outdated (Memory pressure ≥ 0.2,
/// evidence ≥ 20) and 1-of-2 skills degraded with 14 recorded outcomes
/// (Skills pressure 0.5, evidence ≥ 10). Conversation turns give Context a
/// live signal too.
fn pressured_engine() -> MemgineEngine {
    let mut e = MemgineEngine::new(None);
    for i in 0..25 {
        e.ingest_fact(
            &format!("f{i}"),
            &format!("k{i}"),
            &format!("value {i}"),
            "test",
            "user",
            Utc::now(),
            "global",
            None,
            vec![],
            false,
        );
    }
    for i in 0..8 {
        e.report_fact_outdated(&format!("f{i}"));
    }
    for name in ["good", "bad"] {
        e.ingest_skill(
            name,
            "",
            "shell",
            SkillTrigger::default(),
            "s",
            None,
            vec![],
            vec![],
        );
    }
    for _ in 0..4 {
        e.report_outcome("bad", SkillOutcome::Fail);
    }
    for _ in 0..10 {
        e.report_outcome("good", SkillOutcome::Success);
    }
    for i in 0..12 {
        e.ingest_conversation("user", &format!("turn {i}: context signal"), Utc::now());
    }
    e
}

async fn connect(
    state: &Arc<ServerState>,
) -> WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>> {
    let listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new(
        Ipv4Addr::new(127, 0, 0, 1),
        0,
    )))
    .await
    .expect("bind loopback");
    let addr = listener.local_addr().expect("local_addr");
    let st = state.clone();
    tokio::spawn(async move {
        let (stream, peer) = listener.accept().await.expect("accept");
        let ws = accept_async(stream).await.expect("ws handshake");
        let (write, read) = ws.split();
        let _ = run_dispatch(read, Box::pin(write), peer.to_string(), st).await;
    });
    let url = format!("ws://{}", addr);
    let (ws, _resp) = connect_async(&url).await.expect("ws client connect");
    ws
}

async fn send_recv(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    request: serde_json::Value,
) -> serde_json::Value {
    let body = serde_json::to_string(&request).expect("request to_string");
    ws.send(Message::Text(body.into())).await.expect("send");
    let resp = ws.next().await.expect("frame").expect("frame ok");
    let text = match resp {
        Message::Text(t) => t.to_string(),
        other => panic!("expected Text frame, got {:?}", other),
    };
    serde_json::from_str(&text).expect("parse response JSON")
}

fn decision_components(plan: &serde_json::Value) -> Vec<String> {
    plan["decisions"]
        .as_array()
        .expect("decisions")
        .iter()
        .map(|d| d["component"].as_str().unwrap().to_string())
        .collect()
}

#[tokio::test]
async fn evolution_plan_populates_live_components_and_omits_absent_sources() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), pressured_engine());
    let mut ws = connect(&state).await;

    let resp = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "p1", "method": "evolution.plan", "params": {}
        }),
    )
    .await;
    let plan = resp.get("result").expect("result");
    let components = decision_components(plan);

    // Memory / Skills / Context come from the live engine.
    for c in ["memory", "skills", "context"] {
        assert!(components.contains(&c.to_string()), "{c} in {components:?}");
    }
    // Harness has no observable source on a fresh session (empty event log) —
    // omitted, not zero-faked. (Tools is machine-dependent: it reads the real
    // `~/.car/connectors.json` like `connectors.list` does, so its presence is
    // not asserted here; the pure fold is covered by the unit tests in
    // `car_server_core::evolution`.)
    assert!(
        !components.contains(&"harness".to_string()),
        "harness must be absent on an empty session log: {components:?}"
    );
    // The seeded pressure elects Memory and Skills.
    let evolve_now: Vec<&str> = plan["evolve_now"]
        .as_array()
        .unwrap()
        .iter()
        .map(|v| v.as_str().unwrap())
        .collect();
    assert!(evolve_now.contains(&"memory"), "{evolve_now:?}");
    assert!(evolve_now.contains(&"skills"), "{evolve_now:?}");
}

#[tokio::test]
async fn evolution_run_dry_run_reports_without_side_effects() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), pressured_engine());
    let mut ws = connect(&state).await;

    let resp = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "r1", "method": "evolution.run",
            "params": { "dry_run": true }
        }),
    )
    .await;
    let result = resp.get("result").expect("result");
    let steps = result["steps"].as_array().expect("steps");

    // Memory: dry_run reports the sizing + would-run, no consolidate.
    let memory = steps
        .iter()
        .find(|s| s["component"] == "memory")
        .expect("memory step");
    assert_eq!(memory["ran"], true);
    let outcome = memory["outcome"].as_str().unwrap();
    assert!(
        outcome.starts_with("dry_run: would consolidate"),
        "{outcome}"
    );
    assert!(outcome.contains("maintenance:"), "{outcome}");

    // Skills: inference-backed — with no model the step is an HONEST error,
    // not a stubbed success, even under dry_run.
    let skills = steps
        .iter()
        .find(|s| s["component"] == "skills")
        .expect("skills step");
    assert_eq!(skills["ran"], false);
    assert_eq!(skills["outcome"], "no inference engine");

    // No side effects: dry_run appends no EvolutionTriggered audit event.
    let events = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "q1", "method": "events.query",
            "params": { "kinds": ["evolution_triggered"] }
        }),
    )
    .await;
    assert_eq!(events["result"]["count"], 0, "{events}");
}

#[tokio::test]
async fn evolution_run_real_memory_consolidates_and_audits() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), pressured_engine());
    let mut ws = connect(&state).await;

    let resp = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "r2", "method": "evolution.run", "params": {}
        }),
    )
    .await;
    let result = resp.get("result").expect("result");
    let steps = result["steps"].as_array().expect("steps");

    let memory = steps
        .iter()
        .find(|s| s["component"] == "memory")
        .expect("memory step");
    assert_eq!(memory["ran"], true);
    let outcome = memory["outcome"].as_str().unwrap();
    assert!(
        outcome.contains("\"mechanism\":\"consolidate\""),
        "{outcome}"
    );
    assert!(outcome.contains("\"maintenance\""), "{outcome}");
    let evolved: Vec<&str> = result["evolved"]
        .as_array()
        .unwrap()
        .iter()
        .map(|v| v.as_str().unwrap())
        .collect();
    assert!(evolved.contains(&"memory"), "{evolved:?}");

    // The real run is audited on the session event log.
    let events = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "q2", "method": "events.query",
            "params": { "kinds": ["evolution_triggered"] }
        }),
    )
    .await;
    assert_eq!(events["result"]["count"], 1, "{events}");
    let ev = &events["result"]["events"][0];
    assert_eq!(ev["data"]["source"], "evolution.run");
}

/// Held-out harness telemetry that (a) elects the Harness component (20
/// attempts, 60% failing) and (b) diagnoses a RetryConfig mutation with a
/// concrete patch (retries/success = 1.5 > 0.5).
///
/// It carries the task-suite fields — `task_pass_rate` over an explicit
/// `task_pass_denominator`, with `tasks_unrunnable` — because the real
/// `BenchHarnessMeasurer` always produces them (`harness_bench.rs` sets the
/// denominator whenever it sets the rate). A fixture without them cannot reach
/// the comparability guard at all, so the daemon tests would never exercise a
/// branch every real measurement can hit.
fn harness_baseline() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4
        },
        "recovery": { "retries": 12 },
        "task_pass_rate": 0.5,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

async fn run_evolution(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    id: &str,
) -> serde_json::Value {
    send_recv(
        ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "evolution.run",
            "params": { "harness_baseline_metrics": harness_baseline() }
        }),
    )
    .await
}

fn harness_step(result: &serde_json::Value) -> serde_json::Value {
    result["steps"]
        .as_array()
        .expect("steps")
        .iter()
        .find(|s| s["component"] == "harness")
        .expect("harness step")
        .clone()
}

/// The C1 kernel-review scenario end-to-end: the runner connection surfaces a
/// pending harness mutation; a DIFFERENT connection approves the fingerprint
/// via `permission.approve`; a THIRD connection's `evolution.run` then applies
/// the patch — possible only because the approval ledger is shared
/// daemon-wide, not per-connection. Also exercises C2: the fingerprint minted
/// on run 1 must match run 2's re-diagnosis (content-bound, not
/// rationale-bound).
#[tokio::test]
async fn approval_on_one_connection_applies_on_another() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));

    // Connection 1 (the runner): the mutation is diagnosed but unapproved →
    // pending, NOT applied, NOT in `evolved` (S2).
    let mut runner = connect(&state).await;
    let r1 = run_evolution(&mut runner, "e1").await;
    let result1 = r1.get("result").expect("result: {r1}");
    let step1 = harness_step(result1);
    assert_eq!(step1["ran"], true);
    assert_eq!(step1["applied"], false, "{step1}");
    assert!(
        !result1["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "harness"),
        "pending-only run must not claim harness evolved (S2): {result1}"
    );
    let pending = result1["pending_approvals"]
        .as_array()
        .expect("pending_approvals surfaced");
    let fingerprint = pending[0]["fingerprint"].as_str().unwrap().to_string();
    // C2 shape: bound to component + patch content.
    assert!(fingerprint.starts_with("harness:retry:"), "{fingerprint}");

    // Connection 2 (the approver — e.g. a host UI): approve by fingerprint.
    let mut approver = connect(&state).await;
    let a = send_recv(
        &mut approver,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "reviewed retry tuning" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    // Connection 3 (a fresh runner session): the standing approval is visible
    // through the SHARED ledger → the patch applies.
    let mut runner2 = connect(&state).await;
    let r2 = run_evolution(&mut runner2, "e2").await;
    let result2 = r2.get("result").expect("result");
    let step2 = harness_step(result2);
    assert_eq!(step2["ran"], true);
    assert_eq!(step2["applied"], true, "{step2}");
    assert!(
        result2["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "harness"),
        "{result2}"
    );
    let outcome2 = step2["outcome"].as_str().unwrap();
    assert!(outcome2.contains("\"applied\":1"), "{outcome2}");
    assert!(outcome2.contains("human_approved"), "{outcome2}");
}

/// `dry_run` still LISTS the approvals a real run would need — reporting is
/// response data, not a side effect — while appending no audit event.
#[tokio::test]
async fn dry_run_lists_pending_approvals_without_side_effects() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let mut ws = connect(&state).await;

    let r = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "d1", "method": "evolution.run",
            "params": { "dry_run": true, "harness_baseline_metrics": harness_baseline() }
        }),
    )
    .await;
    let result = r.get("result").expect("result");
    let pending = result["pending_approvals"]
        .as_array()
        .expect("dry_run must still list what needs approval");
    assert!(!pending.is_empty());
    assert!(
        pending[0]["reason"]
            .as_str()
            .unwrap_or("")
            .contains("harness_candidate_metrics"),
        "the no-candidate-metrics reason is stated: {pending:?}"
    );

    let events = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "q", "method": "events.query",
            "params": { "kinds": ["evolution_triggered"] }
        }),
    )
    .await;
    assert_eq!(
        events["result"]["count"], 0,
        "dry_run appends no audit event"
    );
}

/// Approvals survive a daemon restart: a second `ServerState` over the same
/// approval journal honours the standing decision without re-asking.
#[tokio::test]
async fn approval_survives_daemon_restart() {
    let tmp = TempDir::new().unwrap();

    // Daemon #1: surface + approve.
    let fingerprint = {
        let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
        let mut ws = connect(&state).await;
        let r = run_evolution(&mut ws, "e1").await;
        let fp = r["result"]["pending_approvals"][0]["fingerprint"]
            .as_str()
            .unwrap()
            .to_string();
        let a = send_recv(
            &mut ws,
            serde_json::json!({
                "jsonrpc": "2.0", "id": "ap", "method": "permission.approve",
                "params": { "fingerprint": fp, "reason": "ok" }
            }),
        )
        .await;
        assert!(a.get("result").is_some(), "{a}");
        fp
    };

    // Daemon #2 (same approval journal — a restart): the approval is loaded
    // from disk and the patch applies without re-asking.
    let state2 = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let mut ws2 = connect(&state2).await;
    let r2 = run_evolution(&mut ws2, "e2").await;
    let result2 = r2.get("result").expect("result");
    let step2 = harness_step(result2);
    assert_eq!(step2["applied"], true, "{step2}");
    assert!(
        result2.get("pending_approvals").is_none()
            || !result2["pending_approvals"]
                .as_array()
                .unwrap()
                .iter()
                .any(|p| p["fingerprint"] == fingerprint.as_str()),
        "an approved fingerprint must not re-surface as pending: {result2}"
    );
}

// ---------------------------------------------------------------------------
// In-daemon candidate measurement (`harness_measure`)
// ---------------------------------------------------------------------------
//
// The daemon grades a harness candidate ITSELF: it replays the held-out split
// under the live config, then once per measurable mutation under that config
// plus the mutation's patch, and feeds the regression gate — no operator
// handing files in. These tests drive that orchestration with a STUB
// `HarnessMeasurer`, so they spend no model calls and no money while still
// exercising the real handler, the real gate and the real apply path.

/// A `HarnessMeasurer` that returns scripted documents and records every call.
///
/// Call 1 is the baseline (measured under the session's live config); every
/// later call is a candidate (measured under that config plus one patch). The
/// recorded `configs` are what proves the loop is not simply grading the
/// baseline twice.
struct StubMeasurer {
    baseline: Result<car_eventlog::harness_metrics::HarnessMetrics, String>,
    candidate: Result<car_eventlog::harness_metrics::HarnessMetrics, String>,
    configs: std::sync::Mutex<Vec<Option<car_memgine::HarnessConfig>>>,
}

impl StubMeasurer {
    fn new(
        baseline: serde_json::Value,
        candidate: Result<serde_json::Value, String>,
    ) -> Arc<StubMeasurer> {
        Arc::new(StubMeasurer {
            baseline: Ok(serde_json::from_value(baseline).expect("baseline HarnessMetrics")),
            candidate: candidate.map(|v| serde_json::from_value(v).expect("candidate metrics")),
            configs: std::sync::Mutex::new(Vec::new()),
        })
    }

    /// A measurer whose very first replay — the baseline — fails. Nothing can
    /// be graded after that, so no candidate document is ever needed.
    fn failing_baseline(error: &str) -> Arc<StubMeasurer> {
        Arc::new(StubMeasurer {
            baseline: Err(error.to_string()),
            candidate: Err(error.to_string()),
            configs: std::sync::Mutex::new(Vec::new()),
        })
    }

    fn calls(&self) -> Vec<Option<car_memgine::HarnessConfig>> {
        self.configs.lock().unwrap().clone()
    }
}

#[async_trait::async_trait]
impl car_server_core::evolution::HarnessMeasurer for StubMeasurer {
    async fn measure(
        &self,
        _request: &car_server_core::evolution::HarnessMeasureRequest,
        harness_config: Option<&car_memgine::HarnessConfig>,
        memgine_config: Option<&car_memgine::MemgineConfig>,
    ) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
        // This stub serves the HARNESS arm only. The context arm passes a
        // `MemgineConfig` and its own stub answers those calls; if one ever
        // landed here the harness fixtures would grade a context mutation and
        // the test would pass for a reason nobody intended.
        assert!(
            memgine_config.is_none(),
            "the harness arm must never vary the context config: a replay that \
             varies two configs at once produces a verdict that attributes to neither"
        );
        let n = {
            let mut calls = self.configs.lock().unwrap();
            calls.push(harness_config.cloned());
            calls.len()
        };
        if n == 1 {
            self.baseline.clone()
        } else {
            self.candidate.clone()
        }
    }
}

/// A candidate that improves the retry mutation's target (retries 12 -> 3, a
/// 75% cut) while holding attempt-level success and still doing work — and
/// solving MORE of the same 12 graded tasks (0.5 -> 0.75), so the task-level
/// guard is armed and satisfied rather than skipped for want of a rate.
fn improved_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4
        },
        "recovery": { "retries": 3 },
        "task_pass_rate": 0.75,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// A candidate whose attempt-level success collapses — the guard the gate
/// exists for. Its task pass rate is HELD at the baseline's 0.5 over the same
/// 12 tasks, so the rejection still has to come from the attempt-level guard;
/// a lower rate here would let the task-level guard reject it first and this
/// test would stop covering what it says it covers.
fn regressed_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.1
        },
        "recovery": { "retries": 3 },
        "task_pass_rate": 0.5,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// A candidate that looks perfect and is not comparable: every metric the
/// attempt-level guard reads is held or improved and its task pass rate is a
/// flawless 1.0 — but over 8 tasks, not the baseline's 12, because 4 became
/// unmeasurable under the candidate config. That is the shape a harness which
/// LOST a capability produces: the tasks it would have failed leave the
/// denominator and the surviving rate rises.
fn incomparable_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4
        },
        "recovery": { "retries": 3 },
        "task_pass_rate": 1.0,
        "task_pass_denominator": 8,
        "tasks_unrunnable": 4
    })
}

async fn run_with_measure(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    id: &str,
    extra: serde_json::Value,
) -> serde_json::Value {
    let mut params = serde_json::json!({
        "harness_measure": { "model": "stub-model", "split": "held-out", "split_seed": 0 }
    });
    let obj = params.as_object_mut().unwrap();
    for (k, v) in extra.as_object().expect("extra params object") {
        obj.insert(k.clone(), v.clone());
    }
    send_recv(
        ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "evolution.run", "params": params
        }),
    )
    .await
}

/// The harness step's summary, parsed back from its JSON string.
fn harness_outcome(result: &serde_json::Value) -> serde_json::Value {
    let step = harness_step(result);
    let outcome = step["outcome"].as_str().expect("harness outcome string");
    serde_json::from_str(outcome).expect("harness outcome is JSON")
}

fn detail_for(outcome: &serde_json::Value, component: &str) -> serde_json::Value {
    outcome["details"]
        .as_array()
        .expect("details")
        .iter()
        .find(|d| d["component"] == component)
        .unwrap_or_else(|| panic!("no detail for {component} in {outcome}"))
        .clone()
}

/// **The headline assertion of the whole feature**: with `harness_measure` and
/// no `harness_candidate_metrics` anywhere in the request, the daemon measures
/// the candidate itself and the regression gate promotes it — an unattended
/// cycle that reaches an applied change.
#[tokio::test]
async fn daemon_measured_candidate_promotes_without_supplied_metrics() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let stub = StubMeasurer::new(harness_baseline(), Ok(improved_candidate()));
    state.set_harness_measurer(stub.clone());
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m1", serde_json::json!({})).await;
    let result = r.get("result").unwrap_or_else(|| panic!("result: {r}"));
    let step = harness_step(result);
    assert_eq!(step["applied"], true, "{step}");
    assert!(
        result["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "harness"),
        "{result}"
    );

    let outcome = harness_outcome(result);
    let retry = detail_for(&outcome, "retry_config");
    assert_eq!(retry["status"], "applied", "{retry}");
    assert_eq!(retry["governance"], "promoted", "{retry}");

    // Auditability: the verdict must be re-derivable from the response alone.
    assert_eq!(outcome["measurement"]["status"], "measured", "{outcome}");
    assert_eq!(outcome["measurement"]["split"], "held-out");
    assert_eq!(outcome["measurement"]["model"], "stub-model");
    assert_eq!(
        outcome["measurement"]["baseline_metrics"]["recovery"]["retries"],
        12
    );
    assert_eq!(retry["candidate_metrics"]["recovery"]["retries"], 3);
}

/// A measured candidate that regresses is rejected by the gate, and nothing is
/// applied — the same verdict a supplied-metrics caller would have got.
#[tokio::test]
async fn daemon_measured_regression_is_rejected_by_the_gate() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state.set_harness_measurer(StubMeasurer::new(
        harness_baseline(),
        Ok(regressed_candidate()),
    ));
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m2", serde_json::json!({})).await;
    let result = r.get("result").expect("result");
    let step = harness_step(result);
    assert_eq!(step["applied"], false, "{step}");
    assert!(
        !result["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "harness"),
        "a rejected candidate must not report the harness as evolved: {result}"
    );
    let retry = detail_for(&harness_outcome(result), "retry_config");
    assert_eq!(retry["status"], "rejected_by_gate", "{retry}");
}

/// A measurement that fails is reported as a measurement failure. Nothing is
/// applied and — the point — no document is synthesized to stand in for the
/// one that was never measured.
#[tokio::test]
async fn a_failed_candidate_measurement_fabricates_nothing() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state.set_harness_measurer(StubMeasurer::new(
        harness_baseline(),
        Err("model backend unreachable".into()),
    ));
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m3", serde_json::json!({})).await;
    let result = r.get("result").expect("result");
    assert_eq!(harness_step(result)["applied"], false);
    let outcome = harness_outcome(result);
    let retry = detail_for(&outcome, "retry_config");
    assert_eq!(retry["status"], "measurement_failed", "{retry}");
    assert!(
        retry["error"]
            .as_str()
            .unwrap_or_default()
            .contains("model backend unreachable"),
        "{retry}"
    );
    assert!(
        retry.get("candidate_metrics").is_none(),
        "a failed measurement must carry NO metrics: {retry}"
    );
    // The cycle still completed — a bench failure is not a cycle failure.
    assert_eq!(harness_step(result)["ran"], true);
}

/// Give the session's own event log real harness telemetry: the SAME action id
/// failing twice, which is what `harness_adapt::diagnose` counts as a
/// *recurring* failure pattern (its min-2-occurrences rule) and therefore what
/// elects the Harness component.
///
/// Needed because a `dry_run` + `harness_measure` cycle measures nothing, so
/// there is no measured baseline to elect or diagnose the component with — the
/// live log is the only honest source left, and a fresh session's is empty.
/// An `assertion` action is the seed because the runtime evaluates it itself:
/// it needs no registered tool and no client-side tool callback (a `tool_call`
/// on this session is rejected by the static-verification gate as unregistered,
/// and a rejection feeds `verification_strength` — the opposite signal). An
/// assertion against an unset state key fails, twice, under one action id.
async fn seed_recurring_action_failures(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
) {
    // Ten, not two: `harness_component_from_events` sets `min_evidence: 20`
    // (a thin telemetry tail cannot support a regression gate) and the default
    // policy needs pressure >= 0.2, so the log needs enough events AND enough of
    // them implicated in the recurring pattern for the component to be elected.
    for i in 0..10 {
        let r = send_recv(
            ws,
            serde_json::json!({
                "jsonrpc": "2.0", "id": format!("seed{i}"), "method": "proposal.submit",
                "params": { "proposal": {
                    "source": "test",
                    "actions": [{
                        "id": "a1",
                        "type": "assertion",
                        "parameters": { "key": "never_set", "expected": "something" }
                    }]
                }}
            }),
        )
        .await;
        let results = r["result"]["results"]
            .as_array()
            .unwrap_or_else(|| panic!("submit {i} returned no results: {r}"));
        assert_eq!(
            results[0]["status"], "failed",
            "the seed action must FAIL (not be rejected) — a rejection would \
             feed verification_strength instead of the failure signal: {r}"
        );
    }
}

/// `dry_run` performs NO measurement: a benchmark replay is a paid side effect
/// and dry_run performs none. Every mutation routes to the existing pending
/// path, with a reason that names dry_run.
///
/// Note what this test had to do to exist, because it is the semantics, not a
/// test detail: with nothing measured there is no measured baseline, so the
/// Harness component is elected and diagnosed from the session's LIVE event log
/// exactly as it was before in-daemon measurement existed. A fresh session has
/// no such telemetry and correctly produces no harness step at all — the daemon
/// does not invent a diagnosis for a measurement it deliberately did not take.
#[tokio::test]
async fn dry_run_measures_nothing_and_says_so() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let stub = StubMeasurer::new(harness_baseline(), Ok(improved_candidate()));
    state.set_harness_measurer(stub.clone());
    let mut ws = connect(&state).await;
    seed_recurring_action_failures(&mut ws).await;

    // An explicit `pressure_threshold`, because ten seeded failures put the
    // Harness pressure at ~0.11 (implicated events over total logged events) —
    // real signal, below the 0.20 default. The policy is a documented param;
    // lowering it here elects the component without touching any assertion
    // about what the dry run then does.
    let r = run_with_measure(
        &mut ws,
        "m4",
        serde_json::json!({ "dry_run": true, "policy": { "pressure_threshold": 0.05 } }),
    )
    .await;
    let result = r.get("result").expect("result");
    assert!(
        stub.calls().is_empty(),
        "dry_run must not spend a single measurement: {:?}",
        stub.calls()
    );
    let outcome = harness_outcome(result);
    assert_eq!(
        outcome["measurement"]["status"], "skipped_dry_run",
        "{outcome}"
    );
    let pending = result["pending_approvals"]
        .as_array()
        .expect("dry_run still lists what needs approval");
    assert!(pending
        .iter()
        .all(|p| p["reason"].as_str().unwrap_or_default().contains("dry_run")));
    assert_eq!(outcome["applied"], 0, "{outcome}");
}

/// Measuring and handing metrics in are two answers to the same question.
/// Combining them is an error naming both params — never a silent winner.
#[tokio::test]
async fn harness_measure_and_supplied_metrics_are_mutually_exclusive() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state.set_harness_measurer(StubMeasurer::new(
        harness_baseline(),
        Ok(improved_candidate()),
    ));
    let mut ws = connect(&state).await;

    let r = run_with_measure(
        &mut ws,
        "m5",
        serde_json::json!({ "harness_candidate_metrics": improved_candidate() }),
    )
    .await;
    let err = r
        .get("error")
        .unwrap_or_else(|| panic!("expected error: {r}"));
    let msg = err["message"].as_str().unwrap_or_default();
    assert!(msg.contains("harness_measure"), "{msg}");
    assert!(msg.contains("harness_candidate_metrics"), "{msg}");
}

/// A build with no evaluator installed ERRS on `harness_measure` rather than
/// quietly degrading to HITL: an opt-in that silently does nothing would report
/// an unattended cycle that never measured anything.
#[tokio::test]
async fn harness_measure_without_an_installed_measurer_errs() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m6", serde_json::json!({})).await;
    let err = r
        .get("error")
        .unwrap_or_else(|| panic!("expected error: {r}"));
    let msg = err["message"].as_str().unwrap_or_default();
    assert!(
        msg.contains("no in-process harness evaluator"),
        "the error must say the build has no evaluator: {msg}"
    );
}

/// Without this the loop could be grading the baseline twice: the baseline call
/// must carry the session's LIVE config and the candidate call must carry that
/// config with the mutation's patch applied.
#[tokio::test]
async fn the_measurer_receives_the_candidate_config_not_the_baseline_twice() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let stub = StubMeasurer::new(harness_baseline(), Ok(improved_candidate()));
    state.set_harness_measurer(stub.clone());
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m7", serde_json::json!({})).await;
    assert!(r.get("result").is_some(), "{r}");

    let calls = stub.calls();
    assert_eq!(
        calls.len(),
        2,
        "one baseline replay plus one candidate replay — the patchless \
         safety-affecting mutation is never measured: {calls:?}"
    );
    assert_eq!(
        calls[0], None,
        "the baseline runs under the session's live config, which is the \
         runtime default on a fresh session: {calls:?}"
    );
    // The diagnosed retry patch is `max_retries: 2, retry_backoff_ms: 100`,
    // projected onto the live (default) config.
    assert_eq!(
        calls[1],
        Some(car_memgine::HarnessConfig {
            max_retries: 2,
            retry_backoff_ms: 100,
            ..Default::default()
        }),
        "the candidate must be measured under the PATCHED config: {calls:?}"
    );

    // The patchless, safety-affecting Validator mutation is surfaced for
    // approval rather than measured.
    let outcome = harness_outcome(r.get("result").unwrap());
    let validator = detail_for(&outcome, "validator");
    assert_eq!(validator["status"], "pending_approval", "{validator}");
    assert!(
        validator.get("candidate_metrics").is_none(),
        "an unmeasured mutation must carry no metrics: {validator}"
    );
}

/// Baseline telemetry from a HEALTHY harness: 20 attempts, none failed. It
/// still elects the Harness component (`harness_component_from_metrics` needs
/// only `attempts > 0`) but at pressure 0.0, which the default policy `skip`s
/// — so the cycle never reaches the Harness arm.
fn healthy_baseline() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 20,
            "failed_attempts": 0,
            "success_rate": 1.0
        },
        "recovery": { "retries": 0 },
        "task_pass_rate": 1.0,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// The measurement is reported whether or not the Harness component is elected.
///
/// A baseline replay is spent BEFORE the plan is assembled, and the plan can
/// legitimately never dispatch Harness — here because a healthy harness has
/// zero pressure and the policy skips it. Reported only from the Harness step,
/// this caller would be billed for a full held-out replay and get a response
/// that never mentions it. A paid side effect nobody can see in the response is
/// one nobody can audit.
#[tokio::test]
async fn the_measurement_is_reported_even_when_harness_is_not_elected() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    let stub = StubMeasurer::new(healthy_baseline(), Ok(improved_candidate()));
    state.set_harness_measurer(stub.clone());
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m8", serde_json::json!({})).await;
    let result = r.get("result").unwrap_or_else(|| panic!("result: {r}"));

    // The premise: no harness step ran at all.
    assert!(
        result["steps"]
            .as_array()
            .expect("steps")
            .iter()
            .all(|s| s["component"] != "harness"),
        "a zero-pressure harness must be skipped, not dispatched: {result}"
    );
    // The money was nevertheless spent — one baseline replay, no candidates.
    assert_eq!(
        stub.calls().len(),
        1,
        "the baseline replay runs before the plan: {:?}",
        stub.calls()
    );
    // …and the response says so, at the top level.
    let measurement = &result["measurement"];
    assert_eq!(measurement["status"], "measured", "{result}");
    assert_eq!(measurement["split"], "held-out", "{measurement}");
    assert_eq!(measurement["model"], "stub-model", "{measurement}");
    assert_eq!(
        measurement["baseline_metrics"]["trajectory_efficiency"]["attempts_total"], 20,
        "the full baseline document is what makes the replay re-derivable: {measurement}"
    );
}

/// A FAILED baseline replay is surfaced at the top level too.
///
/// This is the worst version of the same hole: with the Harness component never
/// elected (a fresh session's log carries no recurring failure pattern, and a
/// failed measurement supplies no baseline to elect one from) the failure had
/// nowhere to be reported. The operator got a normal-looking successful cycle
/// with no hint that the measurement they explicitly asked for never happened.
#[tokio::test]
async fn a_failed_baseline_measurement_is_never_silently_swallowed() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state.set_harness_measurer(StubMeasurer::failing_baseline("bench backend unreachable"));
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m9", serde_json::json!({})).await;
    let result = r.get("result").unwrap_or_else(|| panic!("result: {r}"));

    assert!(
        result["steps"]
            .as_array()
            .expect("steps")
            .iter()
            .all(|s| s["component"] != "harness"),
        "premise: a failed baseline elects no harness component on a fresh \
         session, so the step cannot carry the error: {result}"
    );
    let measurement = &result["measurement"];
    assert_eq!(measurement["status"], "measurement_failed", "{result}");
    assert!(
        measurement["error"]
            .as_str()
            .unwrap_or_default()
            .contains("bench backend unreachable"),
        "the failure must name what went wrong: {measurement}"
    );
    assert!(
        measurement.get("baseline_metrics").is_none(),
        "a failed measurement must carry NO metrics: {measurement}"
    );
    // Nothing was promoted on a measurement that never happened.
    assert!(
        result["evolved"]
            .as_array()
            .expect("evolved")
            .iter()
            .all(|v| v != "harness"),
        "{result}"
    );
}

/// A candidate graded over a SMALLER task set than the baseline reaches the
/// fourth decision — `incomparable` — and applies nothing.
///
/// The real `BenchHarnessMeasurer` always reports a `task_pass_denominator`, so
/// this is a shape the daemon can genuinely produce: a candidate config that
/// loses a capability also loses the ability to MEASURE the tasks needing it,
/// those tasks drop out of the denominator, and the surviving rate rises to a
/// flawless 1.0. Read as an improvement it would promote a strictly worse
/// harness. The verdict has to survive the trip out through the daemon surface,
/// not just live inside `EvolutionAgent::evaluate`.
#[tokio::test]
async fn a_shrunken_task_denominator_is_incomparable_and_applies_nothing() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state.set_harness_measurer(StubMeasurer::new(
        harness_baseline(),
        Ok(incomparable_candidate()),
    ));
    let mut ws = connect(&state).await;

    let r = run_with_measure(&mut ws, "m10", serde_json::json!({})).await;
    let result = r.get("result").unwrap_or_else(|| panic!("result: {r}"));
    let step = harness_step(result);
    assert_eq!(step["applied"], false, "{step}");
    assert!(
        result["evolved"]
            .as_array()
            .expect("evolved")
            .iter()
            .all(|v| v != "harness"),
        "an undecidable comparison must not report the harness as evolved: {result}"
    );

    let outcome = harness_outcome(result);
    let retry = detail_for(&outcome, "retry_config");
    assert_eq!(
        retry["status"], "incomparable",
        "not `rejected_by_gate` — nothing says the candidate is bad, only that \
         this evidence cannot decide it: {retry}"
    );
    let reason = retry["reason"].as_str().unwrap_or_default();
    assert!(
        reason.contains("12") && reason.contains("8"),
        "the reason must name both denominators: {retry}"
    );
    assert!(
        retry.get("rollback_patch").is_none(),
        "nothing was applied, so there is nothing to roll back: {retry}"
    );
    assert_eq!(outcome["applied"], 0, "{outcome}");
    // The document the non-verdict was computed from is still published.
    assert_eq!(retry["candidate_metrics"]["task_pass_denominator"], 8);
    assert_eq!(outcome["measurement"]["status"], "measured", "{outcome}");
}

// ---------------------------------------------------------------------------
// Context and Tools: a real mechanism, and a boundary reported as a boundary.
//
// Both pillars used to return `Err("not_executable: …")`, which the cycle
// records as `ran: false` — the same shape a crashed consolidate produces. So a
// documented scope decision (Tools) and a genuinely-tunable knob nobody had
// wired (Context) both read as failing subsystems in every report.
// ---------------------------------------------------------------------------

/// An engine whose conversation layer is genuinely over its layer-3 assembly
/// budget and STAYS there — the live pressure the Context pillar diagnoses.
///
/// The saturation is real, not staged: with a 400-token budget the layer-3
/// allowance is 120 tokens and twelve ~58-token turns spend nearly six times
/// it. It persists because the engine's own eager compaction (which runs inside
/// `ingest_conversation` at the hard threshold) can only touch the turns beyond
/// `conversation_keep_recent`, and with turns arriving one at a time that is a
/// single turn per ingest — a batch of one summarizes verbatim, so the token
/// total never moves. That is exactly the situation the mechanism exists for:
/// compaction has already run and the layer is still full.
///
/// `batch_size` is the lever the two outcomes hang off. At 8 the three turns
/// handed to compaction after the patch collapse into one summary and the token
/// total falls (the change pays). At 1 each is summarized verbatim on its own,
/// the total is unchanged, and the post-apply measurement must revert it.
/// Speculative compaction is off so the summary text is a pure function of the
/// turns.
fn saturated_engine_with(batch_size: usize) -> MemgineEngine {
    let cfg = car_memgine::MemgineConfig {
        token_budget: 400, // layer-3 budget = 400 * 0.30 = 120 tokens
        compaction_batch_size: batch_size,
        speculative_compaction_interval: 0,
        ..Default::default()
    };
    let mut e = MemgineEngine::new(Some(cfg));
    let base = Utc::now();
    for i in 0..12i64 {
        e.ingest_conversation(
            "user",
            &format!("turn {i}: {}", "x".repeat(220)),
            base + chrono::Duration::seconds(i),
        );
    }
    e
}

fn saturated_engine() -> MemgineEngine {
    saturated_engine_with(8)
}

fn step_for(result: &serde_json::Value, component: &str) -> serde_json::Value {
    result["steps"]
        .as_array()
        .expect("steps")
        .iter()
        .find(|s| s["component"] == component)
        .unwrap_or_else(|| panic!("no {component} step in {result}"))
        .clone()
}

/// The context step's summary, parsed back from its JSON string.
fn context_outcome(result: &serde_json::Value) -> serde_json::Value {
    let step = step_for(result, "context");
    let outcome = step["outcome"].as_str().expect("context outcome string");
    serde_json::from_str(outcome).expect("context outcome is JSON")
}

/// The one context mutation's detail object. Its `component` is the
/// `HarnessComponent` the change contract is typed on — `ContextBudget`, which
/// existed and was unreachable until context diagnoses carried a patch.
fn context_detail(result: &serde_json::Value) -> serde_json::Value {
    detail_for(&context_outcome(result), "context_budget")
}

async fn run_context_cycle(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    id: &str,
    dry_run: bool,
) -> serde_json::Value {
    let resp = send_recv(
        ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "evolution.run",
            "params": { "dry_run": dry_run }
        }),
    )
    .await;
    resp.get("result")
        .unwrap_or_else(|| panic!("evolution.run failed: {resp}"))
        .clone()
}

async fn keep_recent(state: &Arc<ServerState>) -> usize {
    state
        .shared_memgine
        .as_ref()
        .expect("shared engine")
        .lock()
        .await
        .context_evolution_signals()
        .expect("live context signal")
        .conversation_keep_recent
}

async fn conversation_tokens(state: &Arc<ServerState>) -> usize {
    state
        .shared_memgine
        .as_ref()
        .expect("shared engine")
        .lock()
        .await
        .context_evolution_signals()
        .expect("live context signal")
        .conversation_tokens
}

/// Context now has a mechanism: it diagnoses from the live conversation-layer
/// saturation and reports a `context_evolution` summary, instead of erroring
/// with `not_executable` for a knob that was tunable all along.
#[tokio::test]
async fn context_is_no_longer_reported_as_not_executable() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let mut ws = connect(&state).await;

    let result = run_context_cycle(&mut ws, "c1", false).await;
    let step = step_for(&result, "context");
    assert_eq!(step["ran"], true, "{step}");
    assert_eq!(
        step["out_of_scope"], false,
        "context is executed, not excused: {step}"
    );

    let outcome = context_outcome(&result);
    assert_eq!(outcome["mechanism"], "context_evolution", "{outcome}");
    assert_eq!(
        outcome["mutations"], 1,
        "the saturated layer must diagnose exactly one mutation: {outcome}"
    );

    // Nothing anywhere in the cycle claims a pillar is not executable.
    let steps = serde_json::to_string(&result["steps"]).unwrap();
    assert!(
        !steps.contains("not_executable"),
        "no step may report a scope decision as an execution failure: {steps}"
    );
}

/// Tools is a decision, not a breakage: `ran: true`, `out_of_scope: true`,
/// absent from `evolved`, present in `out_of_scope`, and the reason names the
/// credential authority this loop does not hold.
#[tokio::test]
async fn tools_is_recorded_as_out_of_scope_not_a_failure() {
    let tmp = TempDir::new().unwrap();
    // A connector that cannot be dialled gives the Tools pillar a real,
    // deterministic pressure signal (1 of 1 disconnected) instead of depending
    // on whatever this machine has in the operator's own connectors.json.
    //
    // Injected into THIS state's manager rather than via `CAR_HOME`. Setting
    // that env var would race every other test in this binary — each one's
    // `evolution.plan`/`evolution.run` planning pass reaches `getenv` through
    // connector discovery, and `set_var` is not thread-safe — and clearing it
    // afterwards would point a later test at the operator's real connectors,
    // secret headers and all.
    let manifest = tmp.path().join("connectors.json");
    std::fs::write(
        &manifest,
        r#"{"connectors":[{"slug":"unreachable","name":"Unreachable",
             "url":"http://127.0.0.1:1/mcp","secret_headers":[],"enabled_tools":[]}]}"#,
    )
    .unwrap();

    let state = state_with_engine(tmp.path().join("journals"), MemgineEngine::new(None));
    state
        .connectors
        .set(Arc::new(car_connectors::ConnectorManager::with_path(
            state.mcp_executor.clone(),
            manifest,
        )))
        .unwrap_or_else(|_| panic!("connector manager was already initialized"));
    let mut ws = connect(&state).await;
    let result = run_context_cycle(&mut ws, "t1", false).await;

    let step = step_for(&result, "tools");
    assert_eq!(step["ran"], true, "a boundary is not a failure: {step}");
    assert_eq!(step["applied"], false, "{step}");
    assert_eq!(step["out_of_scope"], true, "{step}");
    assert!(
        !result["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "tools"),
        "{result}"
    );
    assert!(
        result["out_of_scope"]
            .as_array()
            .expect("out_of_scope list")
            .iter()
            .any(|v| v == "tools"),
        "{result}"
    );
    let reason = step["outcome"].as_str().unwrap();
    assert!(reason.contains("credentials"), "{reason}");
    assert!(reason.contains("connectors.*"), "{reason}");
}

/// The C1 approval path for the new pillar: a context mutation is surfaced as
/// pending on one connection, approved by fingerprint on a SECOND, and applied
/// by a third run — with the before/after conversation token counts that make
/// the apply auditable.
#[tokio::test]
async fn a_context_proposal_waits_for_approval_then_applies_on_another_connection() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;

    // Run 1: diagnosed, unapproved → pending, nothing applied.
    let mut runner = connect(&state).await;
    let r1 = run_context_cycle(&mut runner, "c1", false).await;
    let d1 = context_detail(&r1);
    assert_eq!(d1["status"], "pending_approval", "{d1}");
    assert!(
        !r1["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "a pending proposal has evolved nothing: {r1}"
    );
    assert_eq!(keep_recent(&state).await, original_keep);

    let pending = r1["pending_approvals"]
        .as_array()
        .expect("pending_approvals surfaced");
    let entry = pending
        .iter()
        .find(|p| {
            p["fingerprint"]
                .as_str()
                .is_some_and(|f| f.starts_with("context:"))
        })
        .unwrap_or_else(|| panic!("no context fingerprint in {pending:?}"));
    let fingerprint = entry["fingerprint"].as_str().unwrap().to_string();
    // A context approval must never be confusable with a harness one.
    assert!(!fingerprint.starts_with("harness:"), "{fingerprint}");
    // The reason states which precondition was missing and what approving
    // instead authorizes. Rewritten with the contract: a pre-activation grade
    // now EXISTS for a patched context mutation and is opt-in via
    // `context_measure`; this cycle simply did not ask for one. The text this
    // used to assert on said no grade was possible at all, which is the claim
    // the memory-fixture bench path falsified.
    let reason = entry["reason"].as_str().unwrap();
    assert!(reason.contains("context_measure"), "{reason}");
    assert!(reason.contains("human gate"), "{reason}");
    assert!(
        !reason.contains("no memgine attached"),
        "the old \"the bench cannot see context\" claim must not survive: {reason}"
    );

    // Connection 2 (the approver): approve by fingerprint.
    let mut approver = connect(&state).await;
    let a = send_recv(
        &mut approver,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "reviewed the keep-recent cut" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    // Connection 3: the standing approval is visible through the SHARED ledger
    // → the patch applies, and its measurement is published.
    let tokens_before = conversation_tokens(&state).await;
    let mut runner2 = connect(&state).await;
    let r2 = run_context_cycle(&mut runner2, "c2", false).await;
    let d2 = context_detail(&r2);
    assert_eq!(d2["status"], "applied", "{d2}");
    assert_eq!(d2["governance"], "human_approved", "{d2}");
    assert_eq!(
        d2["conversation_tokens_baseline"], tokens_before as u64,
        "{d2}"
    );
    assert!(
        d2["conversation_tokens_after"].as_u64().unwrap() < tokens_before as u64,
        "an applied context change must have cut conversation tokens: {d2}"
    );
    assert_eq!(
        d2["rollback_patch"]["conversation_keep_recent"], original_keep as u64,
        "{d2}"
    );
    assert!(
        r2["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "{r2}"
    );
    assert_eq!(keep_recent(&state).await, original_keep / 2);
    assert_eq!(
        conversation_tokens(&state).await,
        d2["conversation_tokens_after"].as_u64().unwrap() as usize
    );
}

/// The measured half of the pillar, and the half that makes it honest: when the
/// applied change does NOT cut conversation tokens, its contract's predicted
/// improvement is falsified, so the inverse patch goes straight back on and
/// nothing is reported as evolved.
#[tokio::test]
async fn a_context_change_that_does_not_cut_tokens_is_rolled_back() {
    let tmp = TempDir::new().unwrap();
    // Batch size 1: every turn handed to compaction is summarized on its own,
    // and a one-turn summary is the turn verbatim — so compaction cannot save a
    // single token however many turns it is given.
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine_with(1));
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;

    let mut ws = connect(&state).await;
    let r1 = run_context_cycle(&mut ws, "c1", false).await;
    let fingerprint = r1["pending_approvals"]
        .as_array()
        .expect("pending_approvals")
        .iter()
        .find(|p| {
            p["fingerprint"]
                .as_str()
                .is_some_and(|f| f.starts_with("context:"))
        })
        .expect("context fingerprint")["fingerprint"]
        .as_str()
        .unwrap()
        .to_string();
    let a = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "approved" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    let r2 = run_context_cycle(&mut ws, "c2", false).await;
    let d = context_detail(&r2);
    assert_eq!(d["status"], "rolled_back", "{d}");
    assert_eq!(
        d["conversation_tokens_after"], d["conversation_tokens_baseline"],
        "the case under test is a change that saved nothing: {d}"
    );
    // A rollback that itself fails is a different, louder status.
    assert!(d.get("rollback_error").is_none(), "{d}");
    assert!(
        !r2["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "a rolled-back change evolved nothing: {r2}"
    );
    assert_eq!(context_outcome(&r2)["applied"], 0);
    // The knob is back where it started, in the same lock hold that moved it.
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
}

/// A conversation layer that is over budget with EVERY turn still verbatim.
///
/// Built by ingesting under a `conversation_keep_recent` high enough that the
/// engine's eager compaction has nothing to take, then lowering the knob to the
/// value the diagnosis starts from. That is the only way to reach this state:
/// ingesting at the final value lets eager compaction summarize turns one at a
/// time on the way in, which leaves nothing for a later compaction pass to do.
///
/// The budget is sized so 40 verbatim turns are far over the layer-3 allowance
/// but the 6 that `conversation_keep_recent` protects are comfortably under it.
/// That is what makes this fixture discriminating: compacting under the
/// UNCHANGED knob already cuts most of these tokens, and the lowered knob then
/// adds nothing at all.
fn uncompacted_engine() -> MemgineEngine {
    let cfg = car_memgine::MemgineConfig {
        token_budget: 1600, // layer-3 budget = 1600 * 0.30 = 480 tokens
        compaction_batch_size: 8,
        speculative_compaction_interval: 0,
        conversation_keep_recent: 40,
        ..Default::default()
    };
    let mut e = MemgineEngine::new(Some(cfg));
    let base = Utc::now();
    for i in 0..40i64 {
        e.ingest_conversation(
            "user",
            &format!("turn {i}: {}", "x".repeat(220)),
            base + chrono::Duration::seconds(i),
        );
    }
    e.apply_context_patch(&car_memgine::ContextConfigPatch {
        conversation_keep_recent: Some(6),
    })
    .expect("lower keep_recent to the value under test");
    e
}

/// The measurement is MARGINAL, and this is the case that proves it: compaction
/// under the unchanged `conversation_keep_recent` already cuts the layer, and
/// the change is credited with none of that.
///
/// Measured from the *uncompacted* layer instead, this same run reports a large
/// saving and promotes the mutation — crediting it with every token compaction
/// was going to save anyway. This run takes the human-approved path — the
/// operator approved the fingerprint without requesting a `context_measure`
/// grade — so no pre-activation grade ran, and this post-apply check is the
/// only automatic thing standing between an approved self-modification and a
/// promotion. It has to attribute honestly.
///
/// Be precise about *why* the margin is zero here, because the two reasons are
/// not the same verdict. It is not that the lowered knob had nothing left to
/// give — it is that the baseline pass cut the verbatim turns below the
/// engine's own hard threshold, so `compact_conversation_heuristic` refuses to
/// run again at all (`turns_summarized == 0` on the second pass, asserted
/// below). "Nothing to compact right now" is a transient state that the
/// baseline pass itself creates and that a later cycle clears; the mechanism
/// reverts, says which of the two it hit, and retries after its backoff.
#[tokio::test]
async fn a_context_change_is_measured_against_a_baseline_compaction_not_the_uncompacted_layer() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), uncompacted_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_uncompacted = conversation_tokens(&state).await;

    let mut ws = connect(&state).await;
    let r1 = run_context_cycle(&mut ws, "c1", false).await;
    let fingerprint = r1["pending_approvals"]
        .as_array()
        .expect("pending_approvals")
        .iter()
        .find(|p| {
            p["fingerprint"]
                .as_str()
                .is_some_and(|f| f.starts_with("context:"))
        })
        .expect("context fingerprint")["fingerprint"]
        .as_str()
        .unwrap()
        .to_string();
    let a = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "approved" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    let r2 = run_context_cycle(&mut ws, "c2", false).await;
    let d = context_detail(&r2);
    let baseline = d["conversation_tokens_baseline"].as_u64().unwrap();
    let after = d["conversation_tokens_after"].as_u64().unwrap();

    // The fixture is the discriminating one only if compacting WITHOUT the
    // change already saved most of the tokens. Assert that, or the test below
    // would pass for the wrong reason.
    assert!(
        baseline < tokens_uncompacted as u64,
        "baseline compaction under the unchanged knob must itself cut tokens \
         ({tokens_uncompacted} → {baseline}), else this fixture proves nothing: {d}"
    );
    assert!(d["baseline_turns_summarized"].as_u64().unwrap() > 0, "{d}");
    assert_eq!(
        after, baseline,
        "the change is credited with none of the baseline's saving: {d}"
    );
    // The precise reason, so this test cannot quietly start passing for a
    // different one: the second pass did no work, because the baseline left the
    // verbatim turns under the engine's hard threshold.
    assert_eq!(
        d["turns_summarized"], 0,
        "the engine's own gate refused the second pass; that is the case under test: {d}"
    );
    assert!(
        d["reason"].as_str().unwrap().contains("did no work at all"),
        "the report must distinguish \"nothing to compact right now\" from \"this knob \
         cannot help\": {d}"
    );
    assert_eq!(
        d["status"], "rolled_back",
        "a change that saved nothing over the baseline must be reverted, however \
         much the baseline itself saved: {d}"
    );
    assert!(
        !r2["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "{r2}"
    );
    assert_eq!(keep_recent(&state).await, original_keep);
}

/// The cadence's brake, end to end: a falsified mutation is re-diagnosed and
/// re-matched to the same standing approval on the very next tick, and the
/// backoff — not the approval, not the diagnosis — is what stops it being
/// applied and reverted again. Without it, every tick pays for a full
/// compaction pass under the engine lock to reach the same verdict, forever.
#[tokio::test]
async fn a_falsified_context_mutation_is_backed_off_on_the_next_cadence_tick() {
    use car_server_core::evolution::{run_context_evolution, ContextBackoff};

    let tmp = TempDir::new().unwrap();
    // Batch size 1: compaction can never save a token, so the measurement
    // falsifies whatever it is given.
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine_with(1));
    let engine = state.shared_memgine.as_ref().unwrap().clone();
    let backoff = std::sync::Mutex::new(ContextBackoff::default());
    let pending = std::sync::Mutex::new(Vec::new());

    // Tick 1 surfaces it for approval; approve by fingerprint.
    run_context_evolution(&engine, &state, false, &pending, Some((&backoff, 1)), None)
        .await
        .expect("tick 1");
    let entry = pending.lock().unwrap().first().cloned().expect("pending");
    let fingerprint = entry["fingerprint"].as_str().unwrap().to_string();
    let mut ws = connect(&state).await;
    let a = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "approved" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    // Tick 2 applies it, measures nothing, reverts, and records the backoff.
    let out = run_context_evolution(&engine, &state, false, &pending, Some((&backoff, 2)), None)
        .await
        .expect("tick 2");
    let d2 = detail_for(
        &serde_json::from_str::<serde_json::Value>(&out.summary).unwrap(),
        "context_budget",
    );
    assert_eq!(d2["status"], "rolled_back", "{d2}");
    assert!(
        !out.applied,
        "a rolled-back mutation evolved nothing: {out:?}"
    );

    // Tick 3: same signals, same fingerprint, same standing approval — and the
    // backoff is the only thing that stops a second apply-measure-revert round.
    let out3 = run_context_evolution(&engine, &state, false, &pending, Some((&backoff, 3)), None)
        .await
        .expect("tick 3");
    let d3 = detail_for(
        &serde_json::from_str::<serde_json::Value>(&out3.summary).unwrap(),
        "context_budget",
    );
    assert_eq!(d3["status"], "in_backoff", "{d3}");
    assert_eq!(
        d3["governance"], "human_approved",
        "the approval still stands: {d3}"
    );
    assert!(
        d3.get("conversation_tokens_baseline").is_none(),
        "no pass was run: {d3}"
    );

    // With no backoff — the session path — the same call applies-and-reverts
    // again, which is the behaviour the cadence must NOT have.
    let session = run_context_evolution(&engine, &state, false, &pending, None, None)
        .await
        .expect("session run");
    let ds = detail_for(
        &serde_json::from_str::<serde_json::Value>(&session.summary).unwrap(),
        "context_budget",
    );
    assert_eq!(
        ds["status"], "rolled_back",
        "a person asking now gets the check now: {ds}"
    );
}

/// `dry_run` over an already-approved fingerprint reports what it WOULD do and
/// touches nothing — an apply is a side effect, and dry_run performs none.
#[tokio::test]
async fn dry_run_applies_no_context_change() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;

    let mut ws = connect(&state).await;
    // dry_run still LISTS what needs approval — reporting is response data.
    let r1 = run_context_cycle(&mut ws, "c1", true).await;
    let fingerprint = r1["pending_approvals"]
        .as_array()
        .expect("pending_approvals")
        .iter()
        .find(|p| {
            p["fingerprint"]
                .as_str()
                .is_some_and(|f| f.starts_with("context:"))
        })
        .expect("context fingerprint")["fingerprint"]
        .as_str()
        .unwrap()
        .to_string();
    let a = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "ap1", "method": "permission.approve",
            "params": { "fingerprint": fingerprint, "reason": "approved" }
        }),
    )
    .await;
    assert!(a.get("result").is_some(), "approve failed: {a}");

    let r2 = run_context_cycle(&mut ws, "c2", true).await;
    let d = context_detail(&r2);
    assert_eq!(d["status"], "would_apply", "{d}");
    assert_eq!(d["governance"], "human_approved", "{d}");
    assert!(d.get("conversation_tokens_baseline").is_none(), "{d}");
    assert!(
        !r2["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "{r2}"
    );
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
}

// ---------------------------------------------------------------------------
// The CONTEXT pillar's PRE-ACTIVATION grader (`context_measure`).
//
// The pillar's old contract was that no automatic check could authorize a
// context change: the bench replayed a runtime with no memgine attached and
// never offered the model a `recall` tool, so a "measured" gate over that
// replay would have returned the same numbers whatever `conversation_keep_
// recent` was set to. That is no longer true — a bench task may declare a
// `memory:` fixture and is then replayed with a real memgine and the shipped
// `recall` tool, so the assembled context, and the answer the task is graded
// on, move with the knob.
//
// These tests drive that orchestration with a STUB measurer, so they spend no
// model calls and no money while exercising the real handler, the real gate
// (`EvolutionAgent::evaluate_context`) and the real `apply_context_patch`.
// ---------------------------------------------------------------------------

/// A `HarnessMeasurer` for the CONTEXT arm.
///
/// Records the `conversation_keep_recent` of every context config it is handed,
/// which is what proves the candidate replay ran under the PATCHED config
/// rather than grading the live one twice — the single defect that would make
/// every context verdict meaningless while looking perfectly healthy.
struct ContextStubMeasurer {
    baseline: Result<car_eventlog::harness_metrics::HarnessMetrics, String>,
    candidate: Result<car_eventlog::harness_metrics::HarnessMetrics, String>,
    keep_recents: std::sync::Mutex<Vec<usize>>,
    /// The TOCTOU seam. When set, the stub moves the shared engine's
    /// `conversation_keep_recent` to this value at the moment the CANDIDATE
    /// replay is requested — i.e. after the baseline was measured under the
    /// live config and while the gate holds no lock. That is exactly the window
    /// another `evolution.run`, the human-approved path or a cadence tick can
    /// land in, and it is the only way to observe it from outside.
    interloper: Option<(Arc<tokio::sync::Mutex<MemgineEngine>>, usize)>,
}

impl ContextStubMeasurer {
    fn new(
        baseline: serde_json::Value,
        candidate: Result<serde_json::Value, String>,
    ) -> Arc<ContextStubMeasurer> {
        Arc::new(ContextStubMeasurer {
            baseline: Ok(serde_json::from_value(baseline).expect("baseline HarnessMetrics")),
            candidate: candidate.map(|v| serde_json::from_value(v).expect("candidate metrics")),
            keep_recents: std::sync::Mutex::new(Vec::new()),
            interloper: None,
        })
    }

    /// A measurer whose very first replay — the baseline — fails. Nothing can
    /// be graded after that, so no candidate document is ever needed. Mirrors
    /// [`StubMeasurer::failing_baseline`] on the harness arm.
    fn failing_baseline(error: &str) -> Arc<ContextStubMeasurer> {
        Arc::new(ContextStubMeasurer {
            baseline: Err(error.to_string()),
            candidate: Err(error.to_string()),
            keep_recents: std::sync::Mutex::new(Vec::new()),
            interloper: None,
        })
    }

    /// A measurer that moves the live `conversation_keep_recent` to
    /// `new_keep_recent` between the baseline and the candidate replay, so the
    /// grade lands on a base that no longer exists.
    fn mutating_the_engine_between_replays(
        baseline: serde_json::Value,
        candidate: serde_json::Value,
        engine: Arc<tokio::sync::Mutex<MemgineEngine>>,
        new_keep_recent: usize,
    ) -> Arc<ContextStubMeasurer> {
        Arc::new(ContextStubMeasurer {
            baseline: Ok(serde_json::from_value(baseline).expect("baseline HarnessMetrics")),
            candidate: Ok(serde_json::from_value(candidate).expect("candidate metrics")),
            keep_recents: std::sync::Mutex::new(Vec::new()),
            interloper: Some((engine, new_keep_recent)),
        })
    }

    /// The `conversation_keep_recent` each replay actually ran under, in call
    /// order: baseline first, then the candidate.
    fn keep_recents(&self) -> Vec<usize> {
        self.keep_recents.lock().unwrap().clone()
    }
}

#[async_trait::async_trait]
impl car_server_core::evolution::HarnessMeasurer for ContextStubMeasurer {
    async fn measure(
        &self,
        _request: &car_server_core::evolution::HarnessMeasureRequest,
        harness_config: Option<&car_memgine::HarnessConfig>,
        memgine_config: Option<&car_memgine::MemgineConfig>,
    ) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
        // Both halves of a context comparison must run under the runtime's
        // default harness config: a replay that varies two configs at once
        // produces a verdict that attributes to neither.
        assert!(
            harness_config.is_none(),
            "the context arm must not vary the harness config"
        );
        let cfg = memgine_config.expect("the context arm always installs a context config");
        let n = {
            let mut calls = self.keep_recents.lock().unwrap();
            calls.push(cfg.conversation_keep_recent);
            calls.len()
        };
        if n == 1 {
            self.baseline.clone()
        } else {
            // The candidate replay is where the interloper strikes: the gate
            // has measured its baseline, has released the engine lock, and is
            // minutes of model calls away from re-acquiring it.
            if let Some((engine, keep)) = &self.interloper {
                engine
                    .lock()
                    .await
                    .apply_context_patch(&car_memgine::ContextConfigPatch {
                        conversation_keep_recent: Some(*keep),
                    })
                    .expect("the interloping config change must land");
            }
            self.candidate.clone()
        }
    }
}

/// The context baseline document: a real replay's shape, with the task-suite
/// fields the gate's comparability and task-level guards both need.
fn context_baseline() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4,
            "total_tokens": 100_000,
            "model_calls": 40
        },
        "task_pass_rate": 0.5,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// The shape a `conversation_keep_recent` cut is supposed to produce: 20% fewer
/// tokens (`ContextBudget` is a cost-reduction target, and the gate wants at
/// least `min_target_improvement` = 5%) with the task pass rate HELD — the
/// summaries still carried what the tasks needed. Attempt-level success and the
/// task denominator are unchanged, so no guard fires.
fn improved_context_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4,
            "total_tokens": 80_000,
            "model_calls": 40
        },
        "task_pass_rate": 0.5,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// The failure this whole pillar exists to catch, and the one the post-apply
/// margin check is blind to: the token cut is REAL (100k → 80k, the target
/// metric improves) but the agent now answers out of a summary that dropped
/// what mattered, so it solves half as many tasks (0.5 → 0.25). Only a
/// task-outcome grade can see that; a token measurement would call it a win.
fn regressed_context_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4,
            "total_tokens": 80_000,
            "model_calls": 40
        },
        "task_pass_rate": 0.25,
        "task_pass_denominator": 12,
        "tasks_unrunnable": 0
    })
}

/// A context candidate that looks perfect and is not comparable: the token cut
/// is real (100k → 80k) and its task pass rate is a flawless 1.0 — but over 8
/// tasks, not the baseline's 12, because 4 became unmeasurable under the
/// candidate config. That is the shape a context config which LOST the ability
/// to answer produces: the tasks it would have failed leave the denominator and
/// the surviving rate rises. The harness arm's `incomparable_candidate` twin.
fn incomparable_context_candidate() -> serde_json::Value {
    serde_json::json!({
        "trajectory_efficiency": {
            "attempts_total": 20,
            "actions_succeeded": 8,
            "failed_attempts": 12,
            "success_rate": 0.4,
            "total_tokens": 80_000,
            "model_calls": 40
        },
        "task_pass_rate": 1.0,
        "task_pass_denominator": 8,
        "tasks_unrunnable": 4
    })
}

/// `evolution.run` with `context_measure` (and nothing else), returning the
/// `result` object.
async fn run_context_measured(
    ws: &mut WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>,
    id: &str,
    dry_run: bool,
) -> serde_json::Value {
    let resp = send_recv(
        ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": id, "method": "evolution.run",
            "params": {
                "dry_run": dry_run,
                "context_measure": { "model": "stub-model", "split": "held-out", "split_seed": 0 }
            }
        }),
    )
    .await;
    resp.get("result")
        .unwrap_or_else(|| panic!("evolution.run failed: {resp}"))
        .clone()
}

/// True when any pending approval in the response is a context fingerprint.
fn has_context_pending(result: &serde_json::Value) -> bool {
    result["pending_approvals"]
        .as_array()
        .map(|a| {
            a.iter().any(|p| {
                p["fingerprint"]
                    .as_str()
                    .is_some_and(|f| f.starts_with("context:"))
            })
        })
        .unwrap_or(false)
}

/// **The headline assertion of the Context grader**: with `context_measure` and
/// no operator anywhere in the loop — no `permission.approve`, nothing in the
/// ledger — the daemon replays the split twice, grades the two documents on
/// TASK outcomes, and applies the change itself.
#[tokio::test]
async fn a_measured_context_mutation_promotes_and_applies_with_no_approval() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let stub = ContextStubMeasurer::new(context_baseline(), Ok(improved_context_candidate()));
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm1", false).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "applied", "{d}");
    assert_eq!(
        d["governance"], "promoted",
        "the authorization here is the measurement, not a human: {d}"
    );
    assert_eq!(
        d["rollback_patch"]["conversation_keep_recent"], original_keep as u64,
        "{d}"
    );
    assert!(
        result["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "{result}"
    );
    // The knob really moved on the live engine.
    assert_eq!(keep_recent(&state).await, original_keep / 2);
    // And nobody was asked to approve anything.
    assert!(
        !has_context_pending(&result),
        "a graded promotion must not also solicit an approval: {result}"
    );

    // The candidate replay ran under the PATCHED config. Without this the two
    // replays would be the same measurement and every verdict would be noise.
    assert_eq!(
        stub.keep_recents(),
        vec![original_keep, original_keep / 2],
        "baseline under the live config, candidate under the halved one"
    );

    // The verdict is re-derivable from the response.
    assert_eq!(d["baseline_task_pass_rate"], 0.5, "{d}");
    assert_eq!(d["candidate_task_pass_rate"], 0.5, "{d}");
    assert_eq!(d["baseline_task_pass_denominator"], 12, "{d}");
    assert_eq!(d["candidate_task_pass_denominator"], 12, "{d}");
    assert_eq!(d["baseline_total_tokens"], 100_000, "{d}");
    assert_eq!(d["candidate_total_tokens"], 80_000, "{d}");

    // A paid replay the response never mentions is one nobody can audit.
    let measured = &context_outcome(&result)["context_measured"];
    assert_eq!(measured["status"], "measured", "{measured}");
    assert_eq!(measured["grade_attempts"], 1, "{measured}");
    assert_eq!(measured["model"], "stub-model", "{measured}");
    assert_eq!(measured["split"], "held-out", "{measured}");
}

/// The gate's teeth: a candidate that cuts tokens while solving fewer TASKS is
/// rejected, nothing is applied — and it does NOT then fall through to the
/// human gate. Soliciting an operator's approval for a change the daemon just
/// measured as a regression, onto a daemon-wide ledger keyed on the change, is
/// how a measured system gets talked out of its own measurement.
#[tokio::test]
async fn a_measured_context_regression_is_rejected_by_the_gate_and_applies_nothing() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;
    let stub = ContextStubMeasurer::new(context_baseline(), Ok(regressed_context_candidate()));
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm2", false).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "rejected_by_gate", "{d}");
    assert!(
        d["reason"].as_str().unwrap().contains("TASK pass rate"),
        "the rejection must name the guard that fired: {d}"
    );
    assert_eq!(context_outcome(&result)["applied"], 0, "{result}");
    assert!(
        !result["evolved"]
            .as_array()
            .unwrap()
            .iter()
            .any(|v| v == "context"),
        "{result}"
    );
    // Nothing touched: not the knob, not the conversation layer (the gate runs
    // BEFORE any apply, so no compaction pass happens either).
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
    // A verdict is not an absence of one.
    assert!(
        !has_context_pending(&result),
        "a measured regression must not be offered to an operator for approval: {result}"
    );
    // Still auditable — the numbers the rejection was computed from.
    assert_eq!(d["baseline_task_pass_rate"], 0.5, "{d}");
    assert_eq!(d["candidate_task_pass_rate"], 0.25, "{d}");
}

/// A replay that errors is reported as exactly that: `measurement_failed`,
/// nothing applied, no document synthesized. The mutation was not falsified —
/// the measurement was — so it must also not be backed off, or an
/// infrastructure failure would delay the retry that would have graded it
/// honestly.
#[tokio::test]
async fn a_failed_context_measurement_fabricates_nothing_and_backs_nothing_off() {
    use car_server_core::evolution::{run_context_evolution, ContextBackoff};

    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;
    let stub = ContextStubMeasurer::new(
        context_baseline(),
        Err("the bench host ran out of disk staging the task suite".to_string()),
    );
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm3", false).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "measurement_failed", "{d}");
    assert!(
        d["error"].as_str().unwrap().contains("ran out of disk"),
        "{d}"
    );
    assert!(
        d.get("baseline_task_pass_rate").is_none(),
        "no verdict was reached, so there is nothing to audit it against: {d}"
    );
    assert_eq!(context_outcome(&result)["applied"], 0, "{result}");
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
    assert!(
        !has_context_pending(&result),
        "a failed measurement is a broken instrument, not a proposal for review: {result}"
    );

    // `note_falsified` must NOT have fired. Observed the only way it can be:
    // drive the arm directly with a backoff installed and check the next tick
    // still measures instead of reporting `in_backoff`. (The live cadence never
    // supplies a measurer; this combination is legal on the API and is the only
    // way to see the backoff table from outside.)
    let engine = state.shared_memgine.as_ref().unwrap().clone();
    let backoff = std::sync::Mutex::new(ContextBackoff::default());
    let pending = std::sync::Mutex::new(Vec::new());
    let request = serde_json::from_value(
        serde_json::json!({ "model": "stub-model", "split": "held-out", "split_seed": 0 }),
    )
    .expect("HarnessMeasureRequest shape");
    for tick in 1..=2u64 {
        let out = run_context_evolution(
            &engine,
            &state,
            false,
            &pending,
            Some((&backoff, tick)),
            Some((stub.as_ref(), &request)),
        )
        .await
        .unwrap_or_else(|e| panic!("tick {tick}: {e}"));
        let detail = detail_for(
            &serde_json::from_str::<serde_json::Value>(&out.summary).unwrap(),
            "context_budget",
        );
        assert_eq!(
            detail["status"], "measurement_failed",
            "tick {tick} must retry the measurement rather than back the mutation off: {detail}"
        );
    }
}

/// `dry_run` with `context_measure`: a benchmark replay is a paid side effect
/// and a dry run performs none, so NO replay happens, nothing is applied, and
/// the mutation falls back to the human gate with a reason naming `dry_run` as
/// the precondition that was missing.
#[tokio::test]
async fn dry_run_with_context_measure_replays_nothing_and_falls_back_to_the_human_gate() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;
    let stub = ContextStubMeasurer::new(context_baseline(), Ok(improved_context_candidate()));
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm4", true).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "pending_approval", "{d}");
    assert!(
        stub.keep_recents().is_empty(),
        "a dry run must spend no replay at all, got {:?}",
        stub.keep_recents()
    );
    let reason = d["reason"].as_str().unwrap();
    assert!(
        reason.contains("dry_run") && reason.contains("paid side effect"),
        "the reason must name the precondition that was missing: {reason}"
    );
    // The human gate is the fallback, and the entry is listed for approval —
    // reporting what needs approval is response data, not a side effect.
    assert!(has_context_pending(&result), "{result}");
    // A requested-but-unspent measurement is still reported, so an operator can
    // tell it from "no measurement was needed".
    let measured = &context_outcome(&result)["context_measured"];
    assert_eq!(measured["status"], "skipped_dry_run", "{measured}");
    assert_eq!(measured["grade_attempts"], 0, "{measured}");
    // Nothing moved.
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
}

/// The other half of the contract: with no `context_measure` in the request,
/// the arm behaves exactly as it did before the grader existed — no replay is
/// spent, the mutation goes to the human gate, and the summary carries no
/// measurement key at all (absent means "none was needed", which is different
/// from `grade_attempts: 0`).
#[tokio::test]
async fn without_context_measure_no_replay_is_spent_and_no_measurement_is_reported() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let stub = ContextStubMeasurer::new(context_baseline(), Ok(improved_context_candidate()));
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_cycle(&mut ws, "cm5", false).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "pending_approval", "{d}");
    assert!(stub.keep_recents().is_empty(), "no replay may be spent");
    assert!(
        context_outcome(&result).get("context_measured").is_none(),
        "an unrequested measurement has nothing to report: {result}"
    );
    assert!(d["reason"].as_str().unwrap().contains("was absent"), "{d}");
}

/// `context_measure` on a build with no evaluator installed is an ERROR, not a
/// quiet degrade to the human gate — an opt-in that silently does nothing would
/// report an unattended cycle that never measured anything. Mirrors the
/// `harness_measure` rule.
#[tokio::test]
async fn context_measure_without_an_installed_measurer_errs() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let mut ws = connect(&state).await;

    let resp = send_recv(
        &mut ws,
        serde_json::json!({
            "jsonrpc": "2.0", "id": "cm6", "method": "evolution.run",
            "params": { "context_measure": { "model": "stub-model" } }
        }),
    )
    .await;
    let err = resp["error"]["message"]
        .as_str()
        .unwrap_or_else(|| panic!("expected an error, got {resp}"));
    assert!(err.contains("context_measure"), "{err}");
    assert!(err.contains("no in-process harness"), "{err}");
}

/// The context arm's twin of `a_failed_baseline_measurement_is_never_silently_
/// swallowed`: when the FIRST replay errors there is nothing to compare, so the
/// candidate replay is never even attempted, the step says `measurement_failed`
/// naming the failure, and nothing is applied. The existing failure test covers
/// a failed *candidate*; a baseline that dies takes a different path through
/// the arm and must not be able to vanish into a normal-looking cycle either.
#[tokio::test]
async fn a_failed_context_baseline_measurement_is_never_silently_swallowed() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;
    let stub = ContextStubMeasurer::failing_baseline("bench backend unreachable");
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm7", false).await;
    let d = context_detail(&result);

    assert_eq!(d["status"], "measurement_failed", "{d}");
    let error = d["error"].as_str().unwrap_or_default();
    assert!(
        error.contains("bench backend unreachable"),
        "the failure must name what went wrong: {d}"
    );
    assert!(
        error.contains("baseline"),
        "and which half of the comparison died — a failed baseline and a failed \
         candidate are different infrastructure problems: {d}"
    );
    // The candidate replay was never attempted: with no baseline there is
    // nothing to compare it against, and it would be a paid replay spent on a
    // comparison that cannot happen.
    assert_eq!(
        stub.keep_recents(),
        vec![original_keep],
        "only the baseline replay may be spent"
    );
    assert!(
        d.get("baseline_task_pass_rate").is_none(),
        "a failed measurement must carry NO metrics: {d}"
    );
    // Nothing was promoted on a measurement that never happened.
    assert_eq!(context_outcome(&result)["applied"], 0, "{result}");
    assert!(
        !result["evolved"]
            .as_array()
            .expect("evolved")
            .iter()
            .any(|v| v == "context"),
        "{result}"
    );
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
    assert!(
        !has_context_pending(&result),
        "a failed measurement is a broken instrument, not a proposal for review: {result}"
    );
}

/// The context arm's twin of `a_shrunken_task_denominator_is_incomparable_and_
/// applies_nothing`: a candidate graded over a SMALLER task set than the
/// baseline reaches the fourth decision — `Incomparable` — and applies nothing.
///
/// The context arm routes that non-verdict to `pending_approval` rather than to
/// its own status, because "the measurement did not decide" is exactly what the
/// human gate is the fallback for; what it must carry is the GATE's reason, not
/// a generic one, so an operator reads why the numbers could not decide it.
#[tokio::test]
async fn a_shrunken_context_task_denominator_is_incomparable_and_applies_nothing() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    let tokens_before = conversation_tokens(&state).await;
    let stub = ContextStubMeasurer::new(context_baseline(), Ok(incomparable_context_candidate()));
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm8", false).await;
    let d = context_detail(&result);

    assert_eq!(
        d["status"], "pending_approval",
        "not `rejected_by_gate` — nothing says the candidate is bad, only that \
         this evidence cannot decide it: {d}"
    );
    let reason = d["reason"].as_str().unwrap_or_default();
    assert!(
        reason.contains("12") && reason.contains("8"),
        "the reason must name both denominators: {d}"
    );
    assert!(
        d.get("rollback_patch").is_none(),
        "nothing was applied, so there is nothing to roll back: {d}"
    );
    assert_eq!(context_outcome(&result)["applied"], 0, "{result}");
    assert!(
        !result["evolved"]
            .as_array()
            .expect("evolved")
            .iter()
            .any(|v| v == "context"),
        "an undecidable comparison must not report context as evolved: {result}"
    );
    assert_eq!(keep_recent(&state).await, original_keep);
    assert_eq!(conversation_tokens(&state).await, tokens_before);
    // The human gate IS the fallback here, and the entry is listed for it.
    assert!(has_context_pending(&result), "{result}");
    // The document the non-verdict was computed from is still published.
    assert_eq!(d["candidate_task_pass_denominator"], 8, "{d}");
    assert_eq!(d["baseline_task_pass_denominator"], 12, "{d}");
    let measured = &context_outcome(&result)["context_measured"];
    assert_eq!(measured["status"], "measured", "{measured}");
    assert_eq!(measured["grade_attempts"], 1, "{measured}");
}

/// **The TOCTOU refusal.** The gate reads the live config, drops the engine
/// lock, and spends minutes of model calls replaying — so by the time it
/// re-acquires the lock to apply, something else may have moved the very knob
/// it graded. Here the stub moves `conversation_keep_recent` out from under it
/// between the two replays, which is what a second `evolution.run` over the
/// same engine, the human-approved path, or a cadence tick would do.
///
/// The promotion must then be REFUSED, not degraded: the verdict was computed
/// against a base that no longer exists, and the inverse patch the apply hands
/// back for rollback would name the interloper's value rather than the measured
/// one. It gets its own terminal status carrying both values, applies nothing,
/// and — like `measurement_failed` — is not treated as a falsified mutation,
/// because the measurement was invalidated and the change never graded against
/// the config it would now be changing.
#[tokio::test]
async fn a_context_base_that_moves_during_measurement_refuses_to_apply() {
    let tmp = TempDir::new().unwrap();
    let state = state_with_engine(tmp.path().join("journals"), saturated_engine());
    let original_keep = keep_recent(&state).await;
    // Distinct from BOTH the measured base (6) and what the graded patch would
    // install (3), so the final assertion can only pass if nothing was applied.
    let interloper_keep = 5;
    assert!(interloper_keep != original_keep && interloper_keep != original_keep / 2);
    let stub = ContextStubMeasurer::mutating_the_engine_between_replays(
        context_baseline(),
        improved_context_candidate(),
        state
            .shared_memgine
            .as_ref()
            .expect("shared engine")
            .clone(),
        interloper_keep,
    );
    state.set_harness_measurer(stub.clone());

    let mut ws = connect(&state).await;
    let result = run_context_measured(&mut ws, "cm9", false).await;
    let d = context_detail(&result);

    // Premise: the gate really did reach a Promote — this test is about what
    // happens AFTER a favourable verdict, not about a verdict that never came.
    assert_eq!(d["status"], "config_moved_during_measurement", "{d}");
    assert_eq!(
        d["measured_under_conversation_keep_recent"], original_keep as u64,
        "{d}"
    );
    assert_eq!(
        d["current_conversation_keep_recent"], interloper_keep as u64,
        "{d}"
    );
    let reason = d["reason"].as_str().unwrap_or_default();
    assert!(
        reason.contains(&original_keep.to_string())
            && reason.contains(&interloper_keep.to_string()),
        "the reason must name the value the grade was measured under and the value live now: {d}"
    );
    assert!(
        d.get("rollback_patch").is_none(),
        "nothing was applied, so there is nothing to roll back: {d}"
    );

    // NOTHING was applied. The knob still holds the interloper's value, not the
    // graded patch's — the one assertion the whole finding turns on.
    assert_eq!(keep_recent(&state).await, interloper_keep);
    assert_eq!(context_outcome(&result)["applied"], 0, "{result}");
    assert!(
        !result["evolved"]
            .as_array()
            .expect("evolved")
            .iter()
            .any(|v| v == "context"),
        "a refused promotion must not report context as evolved: {result}"
    );
    // And it is not quietly re-routed to an operator: a measurement nobody can
    // trust is not a proposal for review.
    assert!(
        !has_context_pending(&result),
        "an invalidated measurement must not solicit an approval: {result}"
    );
    // The replays really did straddle the mutation: baseline under the measured
    // base, candidate under the patched projection of it.
    assert_eq!(stub.keep_recents(), vec![original_keep, original_keep / 2]);
    // The attempt is still reported and still auditable.
    let measured = &context_outcome(&result)["context_measured"];
    assert_eq!(measured["grade_attempts"], 1, "{measured}");
    assert_eq!(d["baseline_task_pass_rate"], 0.5, "{d}");
    assert_eq!(d["candidate_total_tokens"], 80_000, "{d}");
}