car-server-core 0.49.0

//! Self-evolution governor — live daemon wiring (arXiv 2507.21046, the
//! remaining daemon steps from `docs/proposals/self-evolution-governor.md` /
//! `docs/proposals/remaining-integration-work.md` §3).
//!
//! Three pieces live here:
//!
//! 1. **Signal populaters for the components memgine can't see.** The engine
//!    folds Memory / Skills / Context off its own graph
//!    (`MemgineEngine::evolution_component_states`); the daemon appends:
//!    - **Harness** ← [`harness_component_from_events`] over the session event
//!      log via `car_eventlog::harness_adapt::diagnose`. Pressure =
//!      `min(1, implicated / total_events)` where `implicated` sums each
//!      recurring intervention's `evidence_count` — the fraction of logged
//!      events implicated in a *recurring* failure pattern (one-offs are noise
//!      by the diagnosis's own rule). Evidence = the number of events
//!      diagnosed over.
//!    - **Tools** ← [`tools_component_from_connectors`] over the live
//!      connector registry. Pressure = disconnected / total connectors.
//!      Evidence = the connector count — `ConnectorStatus` carries no
//!      per-connector call counters, so the population size is the honest
//!      evidence figure (min_evidence 1: even one broken connector is real).
//! 2. **The real executor** behind `evolution.run` and the cadence timer:
//!    Memory → `consolidate()` sized by `maintenance::decide_maintenance`,
//!    Skills → `evolve_skills(failed_events, domain)` over failure traces
//!    folded from the event log ([`failed_trace_events`]), Harness → the
//!    `harness_evolution` diagnose→gate→apply loop (HITL-gated; handler.rs
//!    owns that arm because it needs the session `ApprovalLedger`), Context →
//!    [`run_context_evolution`], the `context_evolution` loop over the engine's
//!    own conversation-layer saturation. That arm now has TWO authorization
//!    paths: an opt-in **pre-activation grade** (`context_measure` → two bench
//!    replays over the same split, one under the live `MemgineConfig` and one
//!    under it plus the patch, graded on task outcomes by
//!    `EvolutionAgent::evaluate_context`, promoting or rejecting with no human
//!    in the loop) and, for everything the grade did not run on or did not
//!    decide, the original diagnose→approve→apply→measure→revert human path
//!    with its post-apply margin check retained as defence in depth. Tools has
//!    no mechanism *by decision* and
//!    says so ([`TOOLS_OUT_OF_SCOPE_REASON`]) rather than erroring: a scope
//!    boundary reported as a failed step is how a working system reads as a
//!    broken one.
//! 3. **The autonomous cadence timer** ([`spawn_evolution_cadence`]): one
//!    background task over the daemon's *shared* engine, opt-in via
//!    `.car/config.toml` `evolution_interval_secs` (absent/0 = off). Guarded
//!    by [`CycleGuard`] so a slow cycle is never overlapped by the next tick;
//!    each cycle's outcome is appended as an `EvolutionTriggered` event to a
//!    dedicated journal (`<journal_dir>/evolution.jsonl`). The task dies with
//!    the daemon's tokio runtime, like every other boot timer.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;

use car_connectors::ConnectorStatus;
use car_eventlog::{Event, EventKind, EventLog, RetentionPolicy};
use car_memgine::maintenance::{decide_maintenance, MaintenanceDecision, MaintenanceInput};
use car_memgine::self_evolution::{
    run_evolution_cycle, ComponentState, EvolutionCycleReport, EvolutionOutcome, EvolutionPolicy,
    EvolutionSignals, EvolvableComponent,
};
use car_memgine::{MemgineEngine, TraceEvent};
use serde::{Deserialize, Serialize};
use serde_json::Value;

use crate::session::ServerState;

// ---------------------------------------------------------------------------
// The in-daemon harness evaluator seam
// ---------------------------------------------------------------------------

/// What the daemon needs in order to grade a harness candidate ITSELF.
///
/// Implemented above this crate (`car-bench` owns the task suite and the
/// assistant-loop replay) and installed on
/// [`ServerState`](crate::session::ServerState) by the daemon binary, because
/// `car-bench` depends on `car-server-core` and the dependency cannot run the
/// other way.
///
/// Injected rather than called directly for the same reason the rest of this
/// module injects execution: the `evolution.run` orchestration — mutual
/// exclusion, the dry-run rule, which mutations are measured at all, and the
/// gate wiring — stays unit-testable against a stub that spends no model calls.
#[async_trait::async_trait]
pub trait HarnessMeasurer: Send + Sync {
    /// Replay the requested split in process under `harness_config` and
    /// `memgine_config`, and fold the runs' own event logs into one
    /// `HarnessMetrics`.
    ///
    /// `harness_config` is the operating config the replay must run UNDER —
    /// `None` means the runtime default. Measuring a candidate without
    /// installing its config produces a run byte-identical to the baseline, so
    /// an implementation that ignores this argument reports a comparison of a
    /// config with itself.
    ///
    /// `memgine_config` is the **context-assembly** config the replay's memory
    /// fixtures are seeded under — `None` means the memgine default. It is the
    /// second pillar's twin of the argument above, and the identical warning
    /// applies: a "candidate" context measurement taken without installing the
    /// candidate config is a second measurement of the default, and the gate
    /// would be comparing a config with itself. It matters because a bench task
    /// that declares a `memory:` fixture is replayed with a real memgine
    /// attached and the shipped `recall` tool advertised, so the assembled
    /// context — and therefore the answer the task is graded on — genuinely
    /// moves with `conversation_keep_recent`.
    ///
    /// ONE trait, not one per pillar: the replay is the same replay over the
    /// same split with the same seed, and only which config is varied differs.
    /// Two traits would let the two arms drift into measuring different task
    /// sets, at which point a context grade and a harness grade stop being
    /// comparable to each other or to a `car-bench-harness` CLI run.
    ///
    /// Every call must be a REAL measurement or an `Err`. Returning a
    /// default/zero document is the one thing this trait must never do: an
    /// all-zero `HarnessMetrics` is structurally indistinguishable from a
    /// measurement of a harness that spends nothing, and the regression gate
    /// would read it as one.
    async fn measure(
        &self,
        request: &HarnessMeasureRequest,
        harness_config: Option<&car_memgine::HarnessConfig>,
        memgine_config: Option<&car_memgine::MemgineConfig>,
    ) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String>;
}

fn default_split() -> String {
    "held-out".to_string()
}

fn default_held_in_fraction() -> f64 {
    0.5
}

fn default_max_turns() -> u32 {
    20
}

/// What to replay. Serde-derived: it is the `harness_measure` param of
/// `evolution.run` verbatim.
///
/// The defaults mirror `car_bench::harness_bench::HarnessBenchConfig::default()`
/// exactly — held-out, a 0.5 held-in fraction, seed 0, 20 turns — so an
/// in-daemon measurement and a `car-bench-harness` CLI run are the *same*
/// measurement over the *same* task split. A default that drifted from the
/// CLI's would silently make an operator's baseline file incomparable with a
/// daemon-measured candidate.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct HarnessMeasureRequest {
    /// Model id under test. REQUIRED — metrics not attributable to a model are
    /// not metrics, and a router substitution would attribute a token count to
    /// the wrong model.
    pub model: String,
    /// Which half of the deterministic split. Default `"held-out"` — the
    /// regression-gate half, the tasks a mutation was not tuned on.
    #[serde(default = "default_split")]
    pub split: String,
    /// Share of tasks assigned to held-in.
    #[serde(default = "default_held_in_fraction")]
    pub held_in_fraction: f64,
    /// Seed for the deterministic split shuffle. Fix it for a lineage: two
    /// runs at different seeds are over different task sets.
    #[serde(default)]
    pub split_seed: u64,
    /// Assistant-loop turn cap per task.
    #[serde(default = "default_max_turns")]
    pub max_turns: u32,
    /// Override the task suite directory. `None` = the suite built into the
    /// binary.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub tasks_dir: Option<PathBuf>,
}

/// Read the `harness_measure` param of `evolution.run`. `Ok(None)` = not
/// requested (measuring is strictly opt-in); a malformed object is an error,
/// never a silently defaulted request — a typo'd model id would otherwise
/// spend a real benchmark replay on the wrong model.
pub fn parse_harness_measure_request(
    params: &Value,
) -> Result<Option<HarnessMeasureRequest>, String> {
    match params.get("harness_measure") {
        Some(v) if !v.is_null() => Ok(Some(
            serde_json::from_value(v.clone())
                .map_err(|e| format!("invalid harness_measure: {e}"))?,
        )),
        _ => Ok(None),
    }
}

/// Measure the harness AS IT STANDS — the baseline half of the comparison.
///
/// Measured under the session runtime's live `HarnessConfig` (`None` = the
/// runtime default), because the candidate is that same config plus one patch:
/// baseline and candidate must differ by the mutation and nothing else.
pub async fn measure_baseline(
    measurer: &dyn HarnessMeasurer,
    request: &HarnessMeasureRequest,
    live_config: Option<&car_memgine::HarnessConfig>,
) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
    measurer
        // `None` context config: this is the HARNESS arm, and its baseline and
        // candidate must differ by the harness patch and nothing else. Passing
        // a context config here would vary two things at once and the verdict
        // would not attribute to either.
        .measure(request, live_config, None)
        .await
        .map_err(|e| format!("baseline harness measurement failed: {e}"))
}

/// Measure a candidate: `base` with `patch` applied, replayed on the same
/// split under the mutated config.
///
/// The projection is `HarnessConfig::with_patch_for_measurement` —
/// deliberately ungoverned, because this config exists only to be graded and
/// the grade is what authorizes the real apply. It returns a copy, so the live
/// config is untouched until (and unless) the gate promotes.
pub async fn measure_candidate(
    measurer: &dyn HarnessMeasurer,
    request: &HarnessMeasureRequest,
    base: &car_memgine::HarnessConfig,
    patch: &car_memgine::harness_evolution::HarnessConfigPatch,
) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
    let candidate = base.with_patch_for_measurement(patch);
    // `None` context config, for the reason spelled out on `measure_baseline`:
    // one variable per comparison.
    measurer.measure(request, Some(&candidate), None).await
}

// ---------------------------------------------------------------------------
// The CONTEXT arm's measure pair (the pre-activation grader)
// ---------------------------------------------------------------------------

/// Read the `context_measure` param of `evolution.run`. `Ok(None)` = not
/// requested; a malformed object is an error, never a silently defaulted
/// request.
///
/// Deliberately parsed into the SAME [`HarnessMeasureRequest`] type
/// `harness_measure` uses, rather than a context-specific twin. The two arms
/// replay the same task suite on the same deterministic split under the same
/// seed and the same turn cap — only the config being varied differs — so a
/// separate request type would buy nothing but the freedom for the two splits
/// to drift apart, and a context grade taken over a different task set than a
/// harness grade is not comparable with it or with a `car-bench-harness` CLI
/// run.
///
/// Opt-in for the same reason `harness_measure` is: a benchmark replay is a
/// paid side effect (real model calls, real money), and a daemon that starts
/// spending them because a cycle happened to diagnose something is a daemon
/// nobody can leave running.
pub fn parse_context_measure_request(
    params: &Value,
) -> Result<Option<HarnessMeasureRequest>, String> {
    match params.get("context_measure") {
        Some(v) if !v.is_null() => Ok(Some(
            serde_json::from_value(v.clone())
                .map_err(|e| format!("invalid context_measure: {e}"))?,
        )),
        _ => Ok(None),
    }
}

/// Measure the context config AS IT STANDS — the baseline half of the
/// pre-activation comparison.
///
/// `harness_config` is deliberately `None` on both halves of this pair (see
/// [`measure_context_candidate`]): the replay must differ by the context patch
/// and nothing else, so both arms run under the runtime's default harness
/// config. Varying the harness config here as well would produce a verdict that
/// attributes to neither change.
pub async fn measure_context_baseline(
    measurer: &dyn HarnessMeasurer,
    request: &HarnessMeasureRequest,
    live: &car_memgine::MemgineConfig,
) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
    measurer
        .measure(request, None, Some(live))
        .await
        .map_err(|e| format!("baseline context measurement failed: {e}"))
}

/// Measure a context candidate: `base` with `patch` applied, replayed on the
/// same split under the mutated context config.
///
/// The projection is
/// [`car_memgine::MemgineConfig::with_context_patch_for_measurement`] —
/// deliberately ungoverned, because this config exists only to be graded and
/// the grade is what authorizes the real apply. It returns a copy, so the live
/// engine config is untouched until (and unless) the gate promotes and
/// `apply_context_patch` installs it for real.
///
/// `harness_config: None` here matches [`measure_context_baseline`], and that
/// pairing is the load-bearing part: baseline and candidate then differ by the
/// context patch alone.
pub async fn measure_context_candidate(
    measurer: &dyn HarnessMeasurer,
    request: &HarnessMeasureRequest,
    base: &car_memgine::MemgineConfig,
    patch: &car_memgine::ContextConfigPatch,
) -> Result<car_eventlog::harness_metrics::HarnessMetrics, String> {
    let candidate = base.with_context_patch_for_measurement(patch);
    measurer.measure(request, None, Some(&candidate)).await
}

// ---------------------------------------------------------------------------
// Signal populaters (item 1)
// ---------------------------------------------------------------------------

/// Fold the **Harness** component's evolution signals from an event-log tail.
///
/// Pressure = `min(1, implicated / total)` where `implicated` is the summed
/// `evidence_count` of the recurring interventions
/// `car_eventlog::harness_adapt::diagnose` proposes (min 2 occurrences — its
/// own "one-offs are noise" rule): the fraction of logged events implicated in
/// a recurring interaction-failure pattern. Evidence = the number of events
/// diagnosed over. Returns `None` on an empty log — no telemetry is *absence
/// of signal*, not zero pressure, matching how the engine omits Memory/Skills
/// for an empty store.
pub fn harness_component_from_events(events: &[Event]) -> Option<ComponentState> {
    if events.is_empty() {
        return None;
    }
    let report = car_eventlog::harness_adapt::diagnose(events, 2);
    let implicated: usize = report.interventions.iter().map(|i| i.evidence_count).sum();
    let pressure = (implicated as f64 / events.len() as f64).min(1.0);
    Some(ComponentState {
        component: EvolvableComponent::Harness,
        signals: EvolutionSignals {
            pressure,
            evidence: events.len() as u64,
            // Harness evolution regression-gates against telemetry; a thin
            // tail can't support that.
            min_evidence: 20,
            // Costlier than a memory/skill pass: proposals need gating and
            // possibly a human.
            cost: 2.0,
        },
    })
}

/// Fold the **Harness** component's evolution signals from a caller-supplied
/// [`car_eventlog::harness_metrics::HarnessMetrics`] snapshot — the planning
/// twin of [`harness_component_from_events`] for `evolution.run` callers that
/// pass `harness_baseline_metrics` (their own held-out telemetry). Pressure =
/// failed attempts over total attempts; evidence = total attempts. `None`
/// when the metrics record no attempts (nothing observed).
pub fn harness_component_from_metrics(
    m: &car_eventlog::harness_metrics::HarnessMetrics,
) -> Option<ComponentState> {
    let eff = &m.trajectory_efficiency;
    let attempts = eff.actions_succeeded + eff.failed_attempts;
    if attempts == 0 {
        return None;
    }
    Some(ComponentState {
        component: EvolvableComponent::Harness,
        signals: EvolutionSignals {
            pressure: (eff.failed_attempts as f64 / attempts as f64).clamp(0.0, 1.0),
            evidence: attempts as u64,
            min_evidence: 20,
            cost: 2.0,
        },
    })
}

/// Fold the **Tools** component's evolution signals from connector health.
///
/// Pressure = disconnected connectors / total. Evidence = the connector count:
/// `ConnectorStatus` carries no per-connector call counters (there is no
/// call-volume signal to report honestly), so the population size is the
/// evidence, with `min_evidence` 1 — a single broken connector is a real
/// signal. Returns `None` when no connectors are configured (nothing to
/// evolve).
pub fn tools_component_from_connectors(connectors: &[ConnectorStatus]) -> Option<ComponentState> {
    if connectors.is_empty() {
        return None;
    }
    let unhealthy = connectors.iter().filter(|c| !c.connected).count();
    Some(ComponentState {
        component: EvolvableComponent::Tools,
        signals: EvolutionSignals {
            pressure: (unhealthy as f64 / connectors.len() as f64).clamp(0.0, 1.0),
            evidence: connectors.len() as u64,
            min_evidence: 1,
            cost: 1.0,
        },
    })
}

// ---------------------------------------------------------------------------
// Executor mechanics (item 2) — shared by evolution.run and the cadence timer
// ---------------------------------------------------------------------------

/// Fold the failure events `evolve_skills` consumes from an event-log tail:
/// `ActionFailed` / `ActionRejected` / `PolicyViolation` / `ReplanExhausted`
/// become `TraceEvent`s (kind = the event kind's snake_case name, tool lifted
/// from `data.tool` when the executor recorded one, reward 0.0). This is what
/// the session event log genuinely carries — per-action failure records, not
/// full state-before/after trajectories; those fields stay `None` rather than
/// being fabricated.
pub fn failed_trace_events(events: &[Event]) -> Vec<TraceEvent> {
    events
        .iter()
        .filter(|ev| match ev.kind {
            EventKind::ActionFailed
            | EventKind::ActionRejected
            | EventKind::PolicyViolation
            | EventKind::ReplanExhausted => true,
            // A goal evaluation is a failure signal for skill evolution only
            // when the agent claimed completion (met) yet the runtime could not
            // ground it — the false-success case. Grounded or not-yet-met
            // verdicts are normal and must not be folded as failures.
            EventKind::GoalEvaluated => {
                let met = ev
                    .data
                    .get("met")
                    .and_then(|v| v.as_bool())
                    .unwrap_or(false);
                let grounded = ev
                    .data
                    .get("grounded")
                    .and_then(|v| v.as_bool())
                    .unwrap_or(true);
                met && !grounded
            }
            EventKind::TurnCompleted => {
                // A truncated completion (false-success) or a max_turns/stalled
                // terminal is a failure exemplar for skill evolution; a clean
                // empty_tool_calls finish is not.
                let decision = ev
                    .data
                    .get("decision")
                    .and_then(|v| v.as_str())
                    .unwrap_or("");
                let truncated = ev
                    .data
                    .get("was_truncated")
                    .and_then(|v| v.as_bool())
                    .unwrap_or(false);
                truncated || decision == "max_turns" || decision == "stalled"
            }
            _ => false,
        })
        .map(|ev| TraceEvent {
            kind: serde_json::to_value(&ev.kind)
                .ok()
                .and_then(|v| v.as_str().map(str::to_string))
                .unwrap_or_default(),
            action_id: ev.action_id.clone(),
            tool: ev
                .data
                .get("tool")
                .and_then(|v| v.as_str())
                .map(str::to_string),
            data: Value::Object(ev.data.clone().into_iter().collect()),
            duration_ms: None,
            state_before: None,
            state_after: None,
            reward: Some(0.0),
        })
        .collect()
}

/// Price the localized-vs-global maintenance decision off the live
/// [`car_memgine::memsys::MemoryStats`]: dirty regions = the reconciliation
/// backlog (`outstanding_outdated + facts_superseded` — the same backlog the
/// Memory pressure signal counts), total regions = `total_facts`, unit cost
/// per region on both paths, global structural gain = the supersede-churn
/// share (a store reorganization's upside is proportional to how much of the
/// store has churned), valued at one region per full unit of gain times the
/// store size. Deterministic, no fabricated constants beyond the unit costs.
pub fn maintenance_input_from_stats(stats: &car_memgine::memsys::MemoryStats) -> MaintenanceInput {
    let total = stats.total_facts;
    MaintenanceInput {
        dirty_regions: stats.outstanding_outdated + stats.facts_superseded,
        total_regions: total,
        localized_cost_per_region: 1.0,
        global_cost_per_region: 1.0,
        global_structural_gain: if total > 0 {
            (stats.facts_superseded as f64 / total as f64).clamp(0.0, 1.0)
        } else {
            0.0
        },
        gain_value: total as f64,
    }
}

/// The Memory arm of the evolution executor: size the pass with
/// [`decide_maintenance`] (localized vs global — recorded, since
/// `consolidate()` is the single live mechanism for both today) and run
/// `engine.consolidate()`. `dry_run` skips the consolidate and reports what
/// would run (`applied == false`).
pub async fn run_memory_evolution(
    engine: &Arc<tokio::sync::Mutex<MemgineEngine>>,
    dry_run: bool,
) -> Result<EvolutionOutcome, String> {
    let mut eng = engine.lock().await;
    let decision: MaintenanceDecision =
        decide_maintenance(&maintenance_input_from_stats(&eng.memory_stats()));
    if dry_run {
        return Ok(EvolutionOutcome::no_op(format!(
            "dry_run: would consolidate (maintenance: {:?} — {})",
            decision.strategy, decision.rationale
        )));
    }
    let report = eng.consolidate().await;
    let summary = serde_json::to_string(&serde_json::json!({
        "mechanism": "consolidate",
        "maintenance": decision,
        "expired_pruned": report.expired_pruned,
        "superseded_gc": report.superseded_gc,
        "turns_compacted": report.turns_compacted,
        "domains_evolved": report.domains_evolved,
        "total_nodes": report.total_nodes,
    }))
    .map_err(|e| e.to_string())?;
    // A real consolidate is a real pass over the store (GC, embedding flush,
    // promotion gate) — it applied, even when nothing needed pruning.
    Ok(EvolutionOutcome::applied(summary))
}

/// The Skills arm of the evolution executor: `evolve_skills(failed_events,
/// domain)` for every domain `domains_needing_evolution` flags (success rate
/// below 0.6 with ≥3 recorded outcomes — the engine's own threshold). Errors
/// with `"no inference engine"` when the session engine has no model —
/// evolution is inference-backed and silently returning nothing would be a
/// stub. `failed_events` is whatever failure trace the caller's event source
/// genuinely holds (possibly empty — the domain outcome stats, not the traces,
/// are what elect a domain for evolution).
pub async fn run_skills_evolution(
    engine: &Arc<tokio::sync::Mutex<MemgineEngine>>,
    failed_events: &[TraceEvent],
    dry_run: bool,
) -> Result<EvolutionOutcome, String> {
    let mut eng = engine.lock().await;
    if !eng.has_inference() {
        return Err("no inference engine".to_string());
    }
    let domains = eng.domains_needing_evolution(0.6);
    if domains.is_empty() {
        return Ok(EvolutionOutcome::no_op(
            "no domain below the evolution threshold (success < 0.6 over ≥3 outcomes) — nothing to evolve",
        ));
    }
    if dry_run {
        return Ok(EvolutionOutcome::no_op(format!(
            "dry_run: would evolve domain(s) {:?} over {} failure trace(s)",
            domains,
            failed_events.len()
        )));
    }
    let mut evolved = 0usize;
    for domain in &domains {
        evolved += eng.evolve_skills(failed_events, domain).await.len();
    }
    let summary = format!(
        "evolved {} skill(s) across domain(s) {:?} over {} failure trace(s)",
        evolved,
        domains,
        failed_events.len()
    );
    // Applied only when skills were actually produced (kernel review S2).
    Ok(if evolved > 0 {
        EvolutionOutcome::applied(summary)
    } else {
        EvolutionOutcome::no_op(summary)
    })
}

/// The reason the Tools pillar records when it is planned. Verbatim in both
/// call sites (`evolution.run` and the cadence) so an operator reading either
/// report gets the same explanation, and so changing the boundary is one edit.
pub const TOOLS_OUT_OF_SCOPE_REASON: &str =
    "connector remediation means re-running a connector's OAuth or credential exchange. That is \
     an access change, and this loop deliberately holds no authority to grant, refresh or move \
     credentials — reconnect and re-auth stay operator actions through the `connectors.*` \
     surface. Recorded as a deliberate scope decision, not a failure.";

/// The standing half of every context pending reason: that a pre-activation
/// grade EXISTS and why it is opt-in, plus what approving this fingerprint
/// instead buys and how far that authorization reaches.
///
/// This paragraph used to open by saying no pre-activation measurement was
/// available, because `car-bench-harness` replayed a task runtime with no
/// memgine attached and never offered the model a `recall` tool. That is no
/// longer true: a bench task may declare a `memory:` fixture, and a task that
/// does is replayed with a real memgine seeded from it and the shipped `recall`
/// tool advertised, so the assembled context — and the answer the task is
/// graded on — moves with `conversation_keep_recent`. What remains true is that
/// the grade costs a benchmark replay, so it never runs unless a caller asks
/// for it.
///
/// Prefixed per-mutation with the SPECIFIC precondition that was missing (see
/// [`context_pending_reason`]) — "a grade was available and you did not ask for
/// one" and "you asked and the mutation had nothing to grade" are different
/// facts and lead an operator to different next actions.
const CONTEXT_PENDING_REASON: &str =
    "A pre-activation grade IS available for a patched context mutation: the daemon can replay \
     the deterministic bench split twice — once under the live context config, once under it \
     plus this patch — and promote or reject on the resulting TASK pass rates, because bench \
     tasks that declare a memory fixture answer out of assembled context and are therefore \
     sensitive to this knob. It is opt-in via the `context_measure` param because a benchmark \
     replay is a paid side effect (real model calls, real money), and it is deliberately never \
     supplied by the unattended cadence. Activation therefore falls back to the human gate here. \
     Approving this fingerprint once makes every later cycle that proposes the same change apply \
     it, measure the conversation tokens it actually saved over compacting without it, and roll \
     it back if it saved none — that post-apply margin check is retained on the approved path as \
     defence in depth, and is a weaker claim than the task-outcome grade above because it can \
     only falsify the predicted token saving, never confirm the assembled context still answers. \
     The approval ledger is DAEMON-WIDE and the fingerprint names the change, not the engine: \
     approving it authorizes this same conversation_keep_recent change on any engine this daemon \
     evolves — the shared one the unattended cadence runs over, and every per-agent engine an \
     `evolution.run` names — not only the one that proposed it.";

/// No `context_measure` in the request (which includes every unattended cadence
/// tick — see [`spawn_evolution_cadence`]).
const CONTEXT_NOT_REQUESTED: &str =
    "no pre-activation grade was requested for this cycle: the `context_measure` param was \
     absent, and the unattended cadence never supplies it (a timer must not start spending \
     benchmark replays because someone enabled `evolution_interval_secs`).";

/// `context_measure` supplied together with `dry_run`.
const CONTEXT_DRY_RUN_REASON: &str =
    "`context_measure` was requested with `dry_run` — a benchmark replay is a paid side effect \
     and a dry run performs none, so nothing was measured and nothing could be graded.";

/// Did anything a [`car_memgine::ContextConfigPatch`] can reach move between
/// the config the baseline was measured under and the config live now?
///
/// Compares the patch-REACHABLE fields — today exactly
/// `conversation_keep_recent` — rather than the whole config or a digest of
/// it, and that scope is precisely what the rollback-correctness argument
/// turns on. `apply_context_patch` builds its inverse patch by reading each
/// patched field's CURRENT value, so the inverse describes the measured base
/// if and only if every field the patch touches still holds the value the
/// baseline replay ran under. A field the patch cannot reach moving (a token
/// budget, a layer threshold) does not change what the patch will overwrite or
/// what the inverse will restore, and refusing a graded promotion over it
/// would be a false alarm.
///
/// Whenever `ContextConfigPatch` gains a field this comparison gains a term —
/// the same standing warning the patch struct and `apply_context_patch` both
/// carry, and for the same reason: miss one and the inverse patch silently
/// describes a value nobody measured.
fn context_patch_base_moved(
    measured_under: &car_memgine::MemgineConfig,
    current: &car_memgine::MemgineConfig,
) -> bool {
    measured_under.conversation_keep_recent != current.conversation_keep_recent
}

/// The terminal status a graded promotion reports when the live config moved
/// under it between the baseline replay and the apply.
fn context_config_moved_reason(measured_under: usize, current: usize) -> String {
    format!(
        "the live context config moved while this mutation was being measured: \
         conversation_keep_recent was {measured_under} when the baseline replay ran and is \
         {current} now. Something else moved it — another session's `evolution.run` over the \
         same engine, the human-approved path, or an unattended cadence tick — so the grade \
         was computed against a base that no longer exists, and the inverse patch this apply \
         would hand back for rollback would name the CURRENT value rather than the measured \
         one. Nothing is applied. The mutation is NOT falsified and no backoff is recorded: \
         the measurement was invalidated, not the change, and a later cycle re-diagnoses \
         against the new base and re-measures against it."
    )
}

/// A diagnosed mutation with no concrete patch — nothing to project into a
/// candidate config, so nothing to replay.
const CONTEXT_NO_PATCH_REASON: &str =
    "this mutation carries no concrete config patch, so there is nothing to project into a \
     candidate config and nothing to install on a replay — a human designs this change.";

/// Compose the full reason an operator reads on a pending context mutation:
/// the specific precondition that was missing, then the standing explanation of
/// what a grade would have been and what approving instead authorizes.
///
/// Two parts rather than one blob because only the first half varies, and an
/// operator triaging a queue needs to see *which* precondition failed without
/// re-reading the same three sentences on every entry.
fn context_pending_reason(missing: &str) -> String {
    format!("{missing} {CONTEXT_PENDING_REASON}")
}

/// Why an approved-but-falsified context mutation is skipped this tick.
const CONTEXT_BACKOFF_REASON: &str =
    "this mutation's post-apply measurement falsified it on an earlier unattended tick, so it is \
     in exponential backoff. Re-applying it every tick would re-run a full compaction pass under \
     the engine lock to reach the same verdict — the Skills arm backs off for the same reason. \
     The standing approval is untouched: the next attempt happens automatically once the window \
     elapses, and a re-diagnosis that PAYS clears the backoff.";

/// Per-fingerprint exponential backoff for context mutations whose post-apply
/// measurement falsified them (kernel review S5, applied to Context).
///
/// The unattended cadence re-diagnoses from live signals every tick. A
/// falsified mutation restores the knob it moved, so the *next* tick sees the
/// same signals, mints the same fingerprint, matches the same standing
/// approval, and applies-measures-reverts again — forever, each round paying
/// for a full compaction pass under the engine lock. This is that loop's brake.
/// It is deliberately keyed on the fingerprint (the change), not the component:
/// a *different* proposal for the same pillar is not the thing that failed.
///
/// Only the unattended path uses it. `evolution.run` on a session is a person
/// asking for the check to run now, and there is no reason to answer that with
/// "in backoff" (see [`run_context_evolution`]'s `backoff` argument).
#[derive(Debug, Default)]
pub struct ContextBackoff {
    map: HashMap<String, DomainAttempts>,
}

impl ContextBackoff {
    /// True when `fingerprint` has never been falsified, or its backoff window
    /// has elapsed.
    pub fn is_due(&self, fingerprint: &str, tick: u64) -> bool {
        self.map
            .get(fingerprint)
            .map(|a| tick >= a.next_tick)
            .unwrap_or(true)
    }

    /// Record a falsified apply at `tick`: the next attempt is allowed
    /// `2^attempts` ticks later, exponent capped at [`BACKOFF_MAX_EXPONENT`].
    pub fn note_falsified(&mut self, fingerprint: &str, tick: u64) {
        let entry = self
            .map
            .entry(fingerprint.to_string())
            .or_insert(DomainAttempts {
                attempts: 0,
                next_tick: tick,
            });
        entry.attempts = (entry.attempts + 1).min(BACKOFF_MAX_EXPONENT);
        entry.next_tick = tick + (1u64 << entry.attempts);
    }

    /// Drop backoff state for a mutation that has now measurably paid — a
    /// later relapse starts fresh rather than inheriting an old window.
    pub fn clear(&mut self, fingerprint: &str) {
        self.map.remove(fingerprint);
    }
}

fn backoff_due(backoff: Option<(&std::sync::Mutex<ContextBackoff>, u64)>, fp: &str) -> bool {
    match backoff {
        Some((b, tick)) => b.lock().unwrap().is_due(fp, tick),
        None => true,
    }
}

fn note_falsified(backoff: Option<(&std::sync::Mutex<ContextBackoff>, u64)>, fp: &str) {
    if let Some((b, tick)) = backoff {
        b.lock().unwrap().note_falsified(fp, tick);
    }
}

fn clear_backoff(backoff: Option<(&std::sync::Mutex<ContextBackoff>, u64)>, fp: &str) {
    if let Some((b, _)) = backoff {
        b.lock().unwrap().clear(fp);
    }
}

/// The Context arm of the evolution executor — the pillar's real mechanism
/// (`car_memgine::context_evolution`), replacing the `not_executable` error
/// that used to make a documented boundary read as a failing subsystem.
///
/// The shape mirrors the Harness arm in `handler.rs`: diagnose from live
/// signals, fingerprint each mutation, resolve it against the daemon's SHARED
/// durable approval ledger (approve on one connection, apply on another,
/// survives a restart), and report a per-mutation detail object plus a summary.
///
/// **Authorization resolves most-binding-first, over two paths.** A durable
/// operator decision always wins; only a mutation nobody has decided on reaches
/// the pre-activation gate; only a mutation the gate could not run on (or could
/// not decide) reaches the human gate:
///
/// 1. `Rejected` in the ledger → `rejected_by_operator`. An operator's "no" is
///    the most binding thing in the system and is never re-litigated by a
///    measurement.
/// 2. `Approved` in the ledger → the human-approved path below, unchanged:
///    backoff check, dry-run, then the one-lock baseline-compact / apply /
///    re-compact / revert-unless-positive margin measurement.
/// 3. No prior decision **and** a grade is runnable — a measurer and a
///    `context_measure` request were handed in, this is not a `dry_run`, and
///    the mutation carries a patch that
///    [`car_memgine::context_evolution::requires_human_approval`] says is
///    gradeable → **the pre-activation gate**. Read the live
///    [`car_memgine::MemgineConfig`] off the engine, replay the split twice
///    (baseline under the live config, candidate under the live config plus the
///    patch — see [`measure_context_baseline`] /
///    [`measure_context_candidate`]), and hand both documents to
///    [`car_memgine::harness_evolution::EvolutionAgent::evaluate_context`],
///    which is literally the same gate, with the same guards, that grades a
///    harness mutation. `Promote` applies for real (`applied`, `governance:
///    "promoted"`); `Reject` applies nothing and reports `rejected_by_gate`;
///    `NeedsApproval` / `Incomparable` fall through to `pending_approval`
///    carrying the gate's own reason; a replay error reports
///    `measurement_failed` and applies nothing; and a `Promote` whose measured
///    base moved before the apply reports `config_moved_during_measurement` and
///    also applies nothing.
/// 4. Anything else → `pending_approval`, with a reason naming the precondition
///    that was missing ([`context_pending_reason`]).
///
/// Three properties of that resolution are load-bearing:
///
/// - **`rejected_by_gate` does NOT fall through to the human gate.** It is a
///   verdict, not an absence of one: the daemon measured this exact change on
///   task outcomes and it came back a regression. Listing it for approval would
///   invite an operator to approve a change the gate had just measured as worse
///   — and because the ledger is daemon-wide and keyed on the change, that
///   approval would then stand on every engine, permanently, over the top of a
///   real measurement. An operator who disagrees can still approve the
///   fingerprint directly through `permission.approve`; what must not happen is
///   the daemon *soliciting* it.
/// - **A `measurement_failed` mutation is not falsified.** The measurement was.
///   [`ContextBackoff::note_falsified`] is deliberately NOT called on that path:
///   backing a mutation off because the bench errored would punish the change
///   for an infrastructure failure and delay the retry that would have graded
///   it honestly.
/// - **A promotion is re-checked against the live config before it applies.**
///   The engine lock is DROPPED across the two replays, so the config the grade
///   was measured under can move before the apply — another session's
///   `evolution.run` over the same engine, the human-approved path, a cadence
///   tick. Under the same lock hold that would apply the patch, the fields a
///   [`car_memgine::ContextConfigPatch`] can reach are compared against the
///   config the baseline ran under (`context_patch_base_moved`). If they moved,
///   the step reports `config_moved_during_measurement` carrying both values
///   and applies NOTHING: the verdict was computed against a base that no
///   longer exists, and the inverse patch `apply_context_patch` hands back
///   would describe the CURRENT value rather than the measured one, so even the
///   rollback the contract promises would restore the wrong config. Like
///   `measurement_failed`, this records NO backoff — the measurement was
///   invalidated, not the change — and the correct recovery is a later cycle
///   re-diagnosing and re-measuring against the new base.
///
/// The post-apply margin measurement is retained, unchanged, on the
/// human-approved path (2) — defence in depth, and the only automatic check on
/// a change an operator authorized without asking for a grade:
///
/// - **On that path the measurement happens AFTER the apply, and it measures
///   the MARGIN.**
///   Under ONE lock acquisition on the engine — otherwise another task's ingest
///   would be credited or blamed — the arm compacts under the *unchanged*
///   `conversation_keep_recent` first (`conversation_tokens_baseline`), then
///   applies the patch, compacts again, and re-reads
///   (`conversation_tokens_after`). Comparing against that baseline rather than
///   against the uncompacted layer is the load-bearing part: the uncompacted
///   comparison would credit the mutation with every token compaction was going
///   to save anyway, and on a change an operator authorized without asking for
///   a grade this is the ONLY automatic check on it. If the margin is not
///   positive, the contract predicted something that did not happen, so the
///   inverse patch goes back on inside the same lock hold and the step reports
///   `rolled_back` (or `rollback_failed`, its own status, when even that does
///   not take) and counts nothing as applied. Note what a rollback does and does
///   not restore: the config knob goes back, the summarization performed while
///   measuring does not — compaction replaces turns with summaries and keeps
///   them in the layer, which is what the engine's own heuristic does at this
///   saturation anyway.
/// - **`backoff`** is `Some` only on the unattended cadence. A falsified
///   mutation restores the knob, so the next tick re-diagnoses it, re-matches
///   the same standing approval and repeats the whole apply-measure-revert
///   round under the engine lock — forever. [`ContextBackoff`] is that brake,
///   keyed per fingerprint, mirroring [`SkillsBackoff`] (kernel review S5). A
///   session-driven `evolution.run` passes `None`: a person asking for the
///   check now should get it now.
/// - **`measure`** is `Some` only when the caller supplied `context_measure`
///   AND this build has an in-process measurer installed. The unattended
///   cadence passes `None` on purpose: a timer that started spending benchmark
///   replays because someone set `evolution_interval_secs` would turn an opt-in
///   cost into a background one. `dry_run` is honoured here rather than by the
///   caller so the pending reason can say *which* precondition was missing — a
///   dry run that reports "you did not ask for a grade" would be lying.
pub async fn run_context_evolution(
    engine: &Arc<tokio::sync::Mutex<MemgineEngine>>,
    state: &Arc<ServerState>,
    dry_run: bool,
    pending: &std::sync::Mutex<Vec<Value>>,
    backoff: Option<(&std::sync::Mutex<ContextBackoff>, u64)>,
    measure: Option<(&dyn HarnessMeasurer, &HarnessMeasureRequest)>,
) -> Result<EvolutionOutcome, String> {
    use car_memgine::context_evolution::{
        context_mutation_fingerprint, diagnose_context, requires_human_approval,
    };
    use car_memgine::harness_evolution::{EvolutionAgent, PromotionDecision};

    let signals = { engine.lock().await.context_evolution_signals() };
    let Some(signals) = signals else {
        return Ok(EvolutionOutcome::no_op("no observable context signal"));
    };
    let mutations = diagnose_context(&signals);
    if mutations.is_empty() {
        return Ok(EvolutionOutcome::no_op(
            "no context mutations diagnosed from live context signals",
        ));
    }

    let mut details: Vec<Value> = Vec::new();
    let mut applied = 0usize;
    let mut pending_count = 0usize;
    // How many mutations ATTEMPTED the pre-activation gate. Incremented on
    // ENTERING the grading path, before either replay runs, so a mutation whose
    // measurement then errors still counts here — which is why it is named for
    // attempts and not for grades. What it lets an operator tell apart is "a
    // grade was requested and every mutation was already decided in the ledger"
    // (0) from "a grade was requested and at least one mutation reached the
    // replays" (>0); which of those attempts actually produced a verdict is in
    // the per-mutation statuses, not in this number.
    let mut grade_attempts = 0usize;

    for m in &mutations {
        let fingerprint = context_mutation_fingerprint(m);
        let prior = {
            let ledger = state.approval_ledger.read().await;
            ledger.lookup(&fingerprint).map(|r| r.decision)
        };
        let status: Value = match prior {
            Some(car_policy::ApprovalDecision::Rejected) => {
                serde_json::json!({ "status": "rejected_by_operator" })
            }
            Some(car_policy::ApprovalDecision::Approved) => match m.patch.as_ref() {
                None => serde_json::json!({
                    "status": "approved_no_patch",
                    "note": "approved but carries no concrete config patch — a human designs this change",
                }),
                // Backoff is read BEFORE `dry_run`: a dry run exists to report
                // what the next real cycle would do, and what it would do with a
                // backed-off fingerprint is wait.
                Some(_) if !backoff_due(backoff, &fingerprint) => serde_json::json!({
                    "status": "in_backoff",
                    "governance": "human_approved",
                    "reason": CONTEXT_BACKOFF_REASON,
                }),
                Some(_) if dry_run => {
                    serde_json::json!({ "status": "would_apply", "governance": "human_approved" })
                }
                Some(patch) => {
                    // ONE lock acquisition: baseline-compact, measure, apply,
                    // compact again, re-measure, and (if it did not pay)
                    // revert — with no window for another task's ingest to land
                    // in the middle and be credited or blamed.
                    let mut eng = engine.lock().await;
                    // The baseline is a compaction under the CURRENT, unchanged
                    // `conversation_keep_recent` — NOT the uncompacted layer.
                    // Measuring from the uncompacted layer would hand this
                    // mutation every token that compaction was about to save
                    // anyway, and it would be promoted on the strength of a
                    // saving it did not cause. What its contract predicts is a
                    // MARGINAL saving: more turns summarized *because* the knob
                    // fell. That is the only thing this compares.
                    //
                    // Two limits worth knowing, because the report says which
                    // one it hit. (1) `compact_conversation_heuristic` gates on
                    // the tokens held in VERBATIM turns, and the baseline pass
                    // has just cut that number — so when the surviving turns
                    // land under the hard threshold the second pass is refused
                    // outright and the margin reads zero. That is the engine's
                    // own "not full enough to compact" policy, not a verdict on
                    // the knob, and it self-corrects: the baseline pass leaves
                    // the layer in the steady-state shape (verbatim turns only,
                    // everything older summarized) where the next attempt's
                    // baseline is a no-op and the margin is the whole effect.
                    // (2) Saturation counts summaries, compaction's gate does
                    // not, so a layer whose pressure is carried by summaries
                    // diagnoses forever and this knob can never relieve it —
                    // that one measures as zero every time, which is exactly
                    // what the backoff bounds.
                    let baseline = eng.compact_conversation_heuristic();
                    let before = eng
                        .context_evolution_signals()
                        .map(|s| s.conversation_tokens)
                        .unwrap_or(0);
                    match eng.apply_context_patch(patch) {
                        Err(e) => {
                            // Backed off like a falsified apply, and for the
                            // stronger reason: an apply that refuses will refuse
                            // again next tick, and reaching it costs a full
                            // baseline compaction pass under the engine lock.
                            note_falsified(backoff, &fingerprint);
                            serde_json::json!({
                                "status": "apply_failed",
                                "error": e,
                                "baseline_turns_summarized": baseline.turns_summarized,
                            })
                        }
                        Ok(inverse) => {
                            let report = eng.compact_conversation_heuristic();
                            let after = eng
                                .context_evolution_signals()
                                .map(|s| s.conversation_tokens)
                                .unwrap_or(before);
                            if after >= before {
                                let why = if report.turns_summarized == 0 {
                                    "compaction under the new conversation_keep_recent did no \
                                     work at all: after the baseline pass the turns still held \
                                     verbatim are below the engine's own hard threshold, so it \
                                     refuses to compact them whatever this knob says. That is \
                                     \"no saving available right now\", not \"this knob cannot \
                                     help\" — the baseline pass has left the layer in the shape \
                                     where a later cycle's attempt can pay, and the backoff \
                                     window is when it retries."
                                } else {
                                    "compaction under the new conversation_keep_recent ran and \
                                     still saved nothing over compaction under the old one, so \
                                     the change's predicted improvement is falsified."
                                };
                                let falsified = format!(
                                    "{why} ({before} → {after} tokens.) The config is reverted; \
                                     the summarization the measurement itself performed is not \
                                     undone — compaction replaces turns with summaries and keeps \
                                     them in the layer, which is what the engine's own heuristic \
                                     does at this same saturation.",
                                );
                                note_falsified(backoff, &fingerprint);
                                match eng.apply_context_patch(&inverse) {
                                    Ok(_) => serde_json::json!({
                                        "status": "rolled_back",
                                        "governance": "human_approved",
                                        "reason": falsified,
                                        "conversation_tokens_baseline": before,
                                        "conversation_tokens_after": after,
                                        "baseline_turns_summarized": baseline.turns_summarized,
                                        "turns_summarized": report.turns_summarized,
                                    }),
                                    // The config is MUTATED and could not be put
                                    // back. Reporting this as `rolled_back` would
                                    // tell an operator nothing changed while the
                                    // engine runs at the new value, so it gets its
                                    // own terminal status and a log line.
                                    Err(rollback_error) => {
                                        tracing::error!(
                                            fingerprint = %fingerprint,
                                            error = %rollback_error,
                                            "context mutation was falsified but its rollback \
                                             failed — the engine is running at the mutated \
                                             conversation_keep_recent"
                                        );
                                        serde_json::json!({
                                            "status": "rollback_failed",
                                            "governance": "human_approved",
                                            "reason": falsified,
                                            "rollback_error": rollback_error,
                                            "rollback_patch": inverse,
                                            "conversation_tokens_baseline": before,
                                            "conversation_tokens_after": after,
                                            "baseline_turns_summarized": baseline.turns_summarized,
                                            "turns_summarized": report.turns_summarized,
                                        })
                                    }
                                }
                            } else {
                                applied += 1;
                                clear_backoff(backoff, &fingerprint);
                                serde_json::json!({
                                    "status": "applied",
                                    "governance": "human_approved",
                                    "rollback_patch": inverse,
                                    "conversation_tokens_baseline": before,
                                    "conversation_tokens_after": after,
                                    "baseline_turns_summarized": baseline.turns_summarized,
                                    "turns_summarized": report.turns_summarized,
                                })
                            }
                        }
                    }
                }
            },
            // Nobody has decided on this fingerprint. The pre-activation gate
            // gets its turn before the human does — that is the whole point of
            // having one — and the human gate is the fallback for everything
            // the gate could not run on or could not decide.
            None => {
                // A mutation is gradeable iff it carries a patch to project
                // into a candidate config. `requires_human_approval` is the
                // authority on that question (today it is exactly
                // `patch.is_none()`), and it is consulted rather than
                // re-derived here so a future rule that makes some patched
                // mutation human-only lands in one place.
                let gradeable = m.patch.as_ref().filter(|_| !requires_human_approval(m));
                match (measure, gradeable) {
                    (Some((measurer, request)), Some(patch)) if !dry_run => {
                        // Read the LIVE config and drop the lock before the
                        // replays. Each replay is a full benchmark run — many
                        // seconds of model calls — and holding the engine lock
                        // across it would stall every ingest, recall and
                        // consolidate on this daemon for the duration. The
                        // config is a cheap clone and the value only has to be
                        // consistent at the moment the comparison is anchored.
                        let live = { engine.lock().await.config().clone() };
                        grade_attempts += 1;
                        let replays = match measure_context_baseline(measurer, request, &live).await
                        {
                            Err(e) => Err(e),
                            Ok(baseline) => {
                                match measure_context_candidate(measurer, request, &live, patch)
                                    .await
                                {
                                    Err(e) => Err(e),
                                    Ok(candidate) => Ok((baseline, candidate)),
                                }
                            }
                        };
                        match replays {
                            // The measurement failed, not the mutation. Nothing
                            // is applied, nothing is synthesized, and
                            // `note_falsified` is deliberately NOT called: the
                            // change was never graded, so backing it off would
                            // punish it for a bench failure and delay the retry
                            // that would have graded it honestly.
                            Err(error) => serde_json::json!({
                                "status": "measurement_failed",
                                "error": error,
                            }),
                            Ok((baseline, candidate)) => {
                                let decision = EvolutionAgent::new()
                                    .evaluate_context(m, &baseline, &candidate);
                                let mut status = match decision {
                                    PromotionDecision::Promote { reason } => {
                                        // Graded and promoted — apply for real,
                                        // through the one mutation door, which
                                        // enforces the floor and hands back the
                                        // inverse patch the contract's rollback
                                        // promises. No ledger entry was needed
                                        // and none is written: the authorization
                                        // here is the measurement.
                                        let mut eng = engine.lock().await;
                                        // TOCTOU: the lock was DROPPED across
                                        // the two replays, which are minutes of
                                        // model calls, so the config the grade
                                        // was measured under may not be the
                                        // config about to be patched. Anything
                                        // else with a handle on this engine can
                                        // have moved it in that window — another
                                        // session's `evolution.run`, the
                                        // human-approved path, an unattended
                                        // cadence tick. Re-read it under the
                                        // SAME lock hold that would apply the
                                        // patch and compare against what the
                                        // baseline ran under; if it moved,
                                        // refuse rather than degrade, which is
                                        // how the rest of this module handles a
                                        // precondition it cannot honour.
                                        //
                                        // Optimistic re-check rather than
                                        // holding the lock across the replays:
                                        // the lock is the engine's ONLY lock, so
                                        // holding it for the duration of a
                                        // benchmark run would stall every
                                        // ingest, recall and consolidate on this
                                        // daemon for minutes to protect a window
                                        // that is almost never contended. The
                                        // re-check costs one config clone and
                                        // turns the rare collision into a
                                        // refusal instead of a promotion
                                        // justified by a comparison that no
                                        // longer applies.
                                        let current = eng.config().clone();
                                        if context_patch_base_moved(&live, &current) {
                                            serde_json::json!({
                                                "status": "config_moved_during_measurement",
                                                "governance": "promoted",
                                                "reason": context_config_moved_reason(
                                                    live.conversation_keep_recent,
                                                    current.conversation_keep_recent,
                                                ),
                                                "gate_reason": reason,
                                                "measured_under_conversation_keep_recent":
                                                    live.conversation_keep_recent,
                                                "current_conversation_keep_recent":
                                                    current.conversation_keep_recent,
                                            })
                                        } else {
                                            match eng.apply_context_patch(patch) {
                                                Ok(inverse) => {
                                                    applied += 1;
                                                    serde_json::json!({
                                                        "status": "applied",
                                                        "governance": "promoted",
                                                        "reason": reason,
                                                        "rollback_patch": inverse,
                                                    })
                                                }
                                                Err(e) => serde_json::json!({
                                                    "status": "apply_failed",
                                                    "governance": "promoted",
                                                    "error": e,
                                                }),
                                            }
                                        }
                                    }
                                    // A VERDICT, and it deliberately stops here
                                    // rather than falling through to
                                    // `pending_approval`. Soliciting an
                                    // operator's approval for a change the
                                    // daemon just measured as a regression —
                                    // onto a daemon-wide ledger keyed on the
                                    // change, where it would stand for every
                                    // engine forever — is how a measured system
                                    // gets talked out of its own measurement.
                                    PromotionDecision::Reject { reason } => serde_json::json!({
                                        "status": "rejected_by_gate",
                                        "reason": reason,
                                    }),
                                    // No verdict. `NeedsApproval` means the gate
                                    // passed but the mutation is human-only
                                    // anyway; `Incomparable` means the two
                                    // documents cannot be compared (task pass
                                    // rates over different task sets). Both are
                                    // "the measurement did not decide", which is
                                    // exactly what the human gate is the
                                    // fallback for — carrying the gate's own
                                    // reason so an operator reads WHY it did not.
                                    PromotionDecision::NeedsApproval { reason }
                                    | PromotionDecision::Incomparable { reason } => {
                                        let reason = context_pending_reason(&reason);
                                        pending_count += 1;
                                        pending.lock().unwrap().push(serde_json::json!({
                                            "fingerprint": fingerprint,
                                            "mutation": m.id,
                                            "component": m.contract.component,
                                            "safety_affecting": m.contract.component.is_safety_affecting(),
                                            "rationale": m.rationale,
                                            "reason": reason,
                                        }));
                                        serde_json::json!({
                                            "status": "pending_approval",
                                            "reason": reason,
                                        })
                                    }
                                };
                                // Audit what the verdict was computed FROM, on
                                // every graded outcome including the ones that
                                // applied nothing. A promotion (or a rejection)
                                // nobody can re-derive from the response is not
                                // an audited one, and these six numbers are
                                // exactly the inputs `evaluate_context` reads
                                // for a `ContextBudget` mutation.
                                if let Some(obj) = status.as_object_mut() {
                                    obj.insert(
                                        "baseline_task_pass_rate".into(),
                                        serde_json::to_value(baseline.task_pass_rate)
                                            .unwrap_or(Value::Null),
                                    );
                                    obj.insert(
                                        "baseline_task_pass_denominator".into(),
                                        serde_json::to_value(baseline.task_pass_denominator)
                                            .unwrap_or(Value::Null),
                                    );
                                    obj.insert(
                                        "baseline_total_tokens".into(),
                                        Value::from(baseline.trajectory_efficiency.total_tokens),
                                    );
                                    obj.insert(
                                        "candidate_task_pass_rate".into(),
                                        serde_json::to_value(candidate.task_pass_rate)
                                            .unwrap_or(Value::Null),
                                    );
                                    obj.insert(
                                        "candidate_task_pass_denominator".into(),
                                        serde_json::to_value(candidate.task_pass_denominator)
                                            .unwrap_or(Value::Null),
                                    );
                                    obj.insert(
                                        "candidate_total_tokens".into(),
                                        Value::from(candidate.trajectory_efficiency.total_tokens),
                                    );
                                }
                                status
                            }
                        }
                    }
                    // No grade was runnable. Say which precondition was
                    // missing — "you did not ask for one", "you asked on a dry
                    // run", and "this mutation has nothing to grade" lead an
                    // operator to three different next actions, and collapsing
                    // them into one sentence is how an opt-in feature reads as
                    // broken.
                    _ => {
                        let missing = if gradeable.is_none() {
                            CONTEXT_NO_PATCH_REASON
                        } else if measure.is_none() {
                            CONTEXT_NOT_REQUESTED
                        } else {
                            CONTEXT_DRY_RUN_REASON
                        };
                        let reason = context_pending_reason(missing);
                        pending_count += 1;
                        pending.lock().unwrap().push(serde_json::json!({
                            "fingerprint": fingerprint,
                            "mutation": m.id,
                            "component": m.contract.component,
                            // The component's own safety classification, not
                            // `requires_human_approval` — that function now
                            // answers "is this mutation gradeable", which is a
                            // different question. Reading it as a safety
                            // classification would report every patchless
                            // proposal as safety-affecting, telling an operator
                            // that a `conversation_keep_recent` change touches
                            // a safety boundary. It does not.
                            "safety_affecting": m.contract.component.is_safety_affecting(),
                            "rationale": m.rationale,
                            "reason": reason,
                        }));
                        serde_json::json!({
                            "status": "pending_approval",
                            "reason": reason,
                        })
                    }
                }
            }
        };
        let mut d = serde_json::json!({
            "mutation": m.id,
            "component": m.contract.component,
            "fingerprint": fingerprint,
            "rationale": m.rationale,
        });
        if let (Some(obj), Some(s)) = (d.as_object_mut(), status.as_object()) {
            for (k, v) in s {
                obj.insert(k.clone(), v.clone());
            }
        }
        details.push(d);
    }

    let mut summary_obj = serde_json::json!({
        "mechanism": "context_evolution",
        "mutations": mutations.len(),
        "applied": applied,
        "pending": pending_count,
        "details": details,
    });
    // Mirrors the Harness arm's `measurement` key, and exists for the same
    // reason: a benchmark replay is a paid side effect, and a caller who asked
    // for one needs to see in the response whether it happened. Present ONLY
    // when `context_measure` was supplied, so its absence means "no measurement
    // was needed" while `grade_attempts: 0` means "one was requested and
    // nothing reached the gate".
    if let (Some(obj), Some((_, request))) = (summary_obj.as_object_mut(), measure) {
        obj.insert(
            "context_measured".into(),
            serde_json::json!({
                "status": if dry_run { "skipped_dry_run" } else { "measured" },
                "grade_attempts": grade_attempts,
                "model": request.model,
                "split": request.split,
                "split_seed": request.split_seed,
            }),
        );
    }
    let summary = serde_json::to_string(&summary_obj).map_err(|e| e.to_string())?;
    // "Evolved" means a patch landed AND survived its post-apply measurement —
    // a rolled-back mutation changed nothing by the time this returns (S2).
    Ok(if applied > 0 {
        EvolutionOutcome::applied(summary)
    } else {
        EvolutionOutcome::no_op(summary)
    })
}

// ---------------------------------------------------------------------------
// Cadence Skills backoff (kernel review S5): the UNATTENDED loop must not
// re-spend inference on the same failing domain every tick forever. Each
// attempted domain gets an exponentially growing tick-skip; the counter
// resets only when the domain is observed recovered (no longer flagged by
// `domains_needing_evolution`).
// ---------------------------------------------------------------------------

/// Per-domain exponential backoff state for the cadence Skills arm.
#[derive(Debug, Default)]
pub struct SkillsBackoff {
    map: HashMap<String, DomainAttempts>,
}

#[derive(Debug)]
struct DomainAttempts {
    attempts: u32,
    next_tick: u64,
}

/// Cap on the backoff exponent: 2^6 = 64 ticks max between attempts.
const BACKOFF_MAX_EXPONENT: u32 = 6;

impl SkillsBackoff {
    /// The flagged domains that are due an attempt at `tick` (never attempted,
    /// or past their backoff window).
    pub fn due(&self, flagged: &[String], tick: u64) -> Vec<String> {
        flagged
            .iter()
            .filter(|d| {
                self.map
                    .get(*d)
                    .map(|a| tick >= a.next_tick)
                    .unwrap_or(true)
            })
            .cloned()
            .collect()
    }

    /// Record that `domain` was attempted at `tick`: the next attempt is
    /// allowed `2^attempts` ticks later (exponent capped at
    /// [`BACKOFF_MAX_EXPONENT`]). Every attempt widens the window — a domain
    /// only stops backing off by *recovering* (see
    /// [`Self::reset_recovered`]), so unattended inference spend on a domain
    /// that stays broken decays geometrically instead of repeating each tick.
    pub fn note_attempt(&mut self, domain: &str, tick: u64) {
        let entry = self
            .map
            .entry(domain.to_string())
            .or_insert(DomainAttempts {
                attempts: 0,
                next_tick: tick,
            });
        entry.attempts = (entry.attempts + 1).min(BACKOFF_MAX_EXPONENT);
        entry.next_tick = tick + (1u64 << entry.attempts);
    }

    /// Drop backoff state for domains no longer flagged — an observed
    /// recovery resets the counter, so a relapse starts fresh.
    pub fn reset_recovered(&mut self, flagged: &[String]) {
        self.map.retain(|d, _| flagged.iter().any(|f| f == d));
    }
}

/// The cadence-timer Skills arm: like [`run_skills_evolution`] but
/// backoff-gated per domain (kernel review S5) and — being unattended, with
/// no session — running over an empty failure-trace set (the engine's own
/// per-domain outcome stats are what elect a domain; see the cadence scope
/// notes on [`run_evolution_cadence_cycle`]).
pub async fn run_skills_evolution_backoff(
    engine: &Arc<tokio::sync::Mutex<MemgineEngine>>,
    backoff: &std::sync::Mutex<SkillsBackoff>,
    tick: u64,
) -> Result<EvolutionOutcome, String> {
    let mut eng = engine.lock().await;
    if !eng.has_inference() {
        return Err("no inference engine".to_string());
    }
    let flagged = eng.domains_needing_evolution(0.6);
    let due = {
        let mut b = backoff.lock().unwrap();
        b.reset_recovered(&flagged);
        b.due(&flagged, tick)
    };
    if flagged.is_empty() {
        return Ok(EvolutionOutcome::no_op(
            "no domain below the evolution threshold — nothing to evolve",
        ));
    }
    if due.is_empty() {
        return Ok(EvolutionOutcome::no_op(format!(
            "all {} flagged domain(s) in backoff — no attempt this tick",
            flagged.len()
        )));
    }
    let mut evolved = 0usize;
    for domain in &due {
        evolved += eng.evolve_skills(&[], domain).await.len();
        backoff.lock().unwrap().note_attempt(domain, tick);
    }
    let summary = format!(
        "evolved {} skill(s) across due domain(s) {:?} ({} flagged total)",
        evolved,
        due,
        flagged.len()
    );
    Ok(if evolved > 0 {
        EvolutionOutcome::applied(summary)
    } else {
        EvolutionOutcome::no_op(summary)
    })
}

// ---------------------------------------------------------------------------
// Cadence timer (item 3)
// ---------------------------------------------------------------------------

/// Non-overlap guard for the cadence timer: a tick that arrives while the
/// previous cycle is still running is skipped, never queued. RAII — dropping
/// the token releases the guard even if the cycle errors.
#[derive(Debug, Default)]
pub struct CycleGuard {
    running: std::sync::atomic::AtomicBool,
}

/// Held while a cycle runs; releases the guard on drop.
pub struct CycleToken<'a> {
    guard: &'a CycleGuard,
}

impl CycleGuard {
    /// Claim the guard. `None` when a cycle is already in flight.
    pub fn try_begin(&self) -> Option<CycleToken<'_>> {
        self.running
            .compare_exchange(
                false,
                true,
                std::sync::atomic::Ordering::SeqCst,
                std::sync::atomic::Ordering::SeqCst,
            )
            .ok()
            .map(|_| CycleToken { guard: self })
    }
}

impl Drop for CycleToken<'_> {
    fn drop(&mut self) {
        self.guard
            .running
            .store(false, std::sync::atomic::Ordering::SeqCst);
    }
}

/// Read the opt-in cadence interval from the `.car/` project's `config.toml`
/// (`evolution_interval_secs`), discovered from the same anchor as
/// [`crate::seed_memgine_config`]: `$CAR_PROJECT_DIR` when set, else the
/// process cwd. Absent, `0`, or no project → `None` (off — the no-surprise
/// default).
pub fn seed_evolution_interval() -> Option<u64> {
    let anchor = std::env::var_os("CAR_PROJECT_DIR")
        .map(std::path::PathBuf::from)
        .or_else(|| std::env::current_dir().ok())?;
    let car_dir = car_memgine::project::discover_project(&anchor)?;
    car_memgine::project::load_config_overrides(&car_dir)?
        .evolution_interval_secs
        .filter(|s| *s > 0)
}

/// Run one unattended evolution cycle over the daemon's **shared** engine:
/// fold its live Memory / Skills / Context signals plus Tools from connector
/// health, plan under the default policy, and dispatch `EvolveNow` components
/// to the same mechanics as `evolution.run` (`dry_run = false`).
///
/// Scope boundaries, stated rather than papered over:
/// - **Harness is not populated** here — harness telemetry (action events)
///   lives in per-session event logs; the daemon holds no cross-session action
///   log, and the harness apply path is HITL-gated on a *session's* approval
///   flow. Harness evolution runs through `evolution.run` on a session.
/// - **Skills** run with an empty failure-trace set for the same reason; the
///   engine's own per-domain outcome stats (what `domains_needing_evolution`
///   folds) are the evidence that elects a domain. Attempts are per-domain
///   exponentially backed off across ticks ([`SkillsBackoff`], kernel review
///   S5) so unattended inference spend never repeats the same failing domain
///   every tick.
/// - **Context** runs the real mechanism ([`run_context_evolution`]) — the
///   cadence has both the shared engine and the shared approval ledger, which
///   is everything that arm needs — but with **no pre-activation grader**: it
///   passes `None` for `measure`, because a grade costs a benchmark replay per
///   mutation and an unattended timer must not start spending model calls
///   because someone enabled `evolution_interval_secs`. The cadence therefore
///   still applies only what a human approved once. It runs under a
///   per-fingerprint [`ContextBackoff`],
///   so an approved mutation the measurement falsifies is not re-applied and
///   re-reverted on every tick forever (the Skills arm's S5 rule, applied to
///   Context). Its pending approvals are collected locally and folded into the
///   step summary as a count: an unattended cycle has no response to attach a
///   `pending_approvals` array to, and the durable ledger is where an operator
///   actually resolves them.
/// - **Tools** is recorded as `out_of_scope` (not a failure): connector
///   remediation is a credential operation this loop holds no authority to
///   perform ([`TOOLS_OUT_OF_SCOPE_REASON`]).
///
/// Returns the typed cycle report (`None` when the daemon has no shared
/// engine); the cadence loop decides logging.
pub async fn run_evolution_cadence_cycle(
    state: &Arc<ServerState>,
    backoff: &std::sync::Mutex<SkillsBackoff>,
    context_backoff: &std::sync::Mutex<ContextBackoff>,
    tick: u64,
) -> Option<EvolutionCycleReport> {
    let engine = state.shared_memgine.as_ref()?.clone();

    let mut components = { engine.lock().await.evolution_component_states() };
    state.ensure_connectors_loaded().await;
    let connector_list = state.connectors().list().await;
    if let Some(t) = tools_component_from_connectors(&connector_list) {
        components.push(t);
    }

    // The cadence has no response to hang a `pending_approvals` array off, so
    // the Context arm's pending entries land here and are reported as a count
    // in its summary; the durable ledger is where they are actually resolved.
    let context_pending: std::sync::Mutex<Vec<Value>> = std::sync::Mutex::new(Vec::new());
    let context_pending_ref = &context_pending;

    let policy = EvolutionPolicy::default();
    let report = run_evolution_cycle(&components, &policy, |c| {
        let engine = engine.clone();
        async move {
            match c {
                EvolvableComponent::Memory => run_memory_evolution(&engine, false).await,
                EvolvableComponent::Skills => {
                    run_skills_evolution_backoff(&engine, backoff, tick).await
                }
                // A boundary, not a breakage. This used to be an
                // `Err("not_executable: …")`, which the cycle records as
                // `ran: false` — the same shape a crashed mechanism produces.
                // The cadence never appends Harness to `components` today, so
                // the arm is unreachable in practice; it is corrected here so
                // that stays true if a future cadence does plan it.
                EvolvableComponent::Harness => Ok(EvolutionOutcome::out_of_scope(
                    "harness telemetry and the HITL apply path are per-session — the cadence has \
                     no session, so harness evolution is driven via evolution.run on one. \
                     Recorded as a deliberate scope decision, not a failure.",
                )),
                EvolvableComponent::Context => {
                    // `None` measure, deliberately. The pre-activation grade
                    // costs a full benchmark replay per mutation — real model
                    // calls, real money — and an unattended timer must not
                    // start spending them because someone set
                    // `evolution_interval_secs`. That is a boundary, not an
                    // omission: an operator who wants an unattended cycle
                    // graded runs `evolution.run` with `context_measure`, where
                    // the spend is something they asked for. The cadence still
                    // applies changes whose fingerprint a human approved once,
                    // under the post-apply margin check and the backoff.
                    run_context_evolution(
                        &engine,
                        state,
                        false,
                        context_pending_ref,
                        Some((context_backoff, tick)),
                        None,
                    )
                    .await
                }
                EvolvableComponent::Tools => {
                    Ok(EvolutionOutcome::out_of_scope(TOOLS_OUT_OF_SCOPE_REASON))
                }
            }
        }
    })
    .await;

    let pending = context_pending.into_inner().unwrap();
    if !pending.is_empty() {
        tracing::info!(
            count = pending.len(),
            "evolution cadence surfaced context mutation(s) awaiting operator approval; \
             approve by fingerprint via permission.approve"
        );
    }

    Some(report)
}

/// Retention cap for the cadence's dedicated event log (kernel review S4): a
/// long-running daemon's twin log must not grow unboundedly in memory.
const EVOLUTION_LOG_MAX_EVENTS: usize = 1000;

/// Spawn the autonomous cadence timer (opt-in via `.car/config.toml`
/// `evolution_interval_secs`): ONE background task over the daemon's shared
/// engine that every `interval_secs` runs [`run_evolution_cadence_cycle`] and
/// appends the outcome as an `EvolutionTriggered` event (`data.source =
/// "cadence"`) to `<journal_dir>/evolution.jsonl` — capped at
/// [`EVOLUTION_LOG_MAX_EVENTS`] in memory, and **no-op cycles (nothing
/// planned, nothing run) are not appended** (kernel review S4), so an idle
/// daemon doesn't mint an audit line per tick. A tick that lands while the
/// previous cycle is still running is **skipped** ([`CycleGuard`]); each cycle
/// runs in its own task so a panic is isolated to that tick. The task ends
/// with the daemon's tokio runtime. Returns `None` (and warns) when the daemon
/// has no shared engine to evolve.
pub fn spawn_evolution_cadence(
    state: Arc<ServerState>,
    interval_secs: u64,
) -> Option<tokio::task::JoinHandle<()>> {
    if state.shared_memgine.is_none() {
        tracing::warn!(
            "evolution_interval_secs set but the daemon has no shared engine; cadence not started"
        );
        return None;
    }
    let journal = state.journal_dir.join("evolution.jsonl");
    Some(tokio::spawn(async move {
        let mut log = EventLog::with_journal(journal);
        log.set_retention(Some(RetentionPolicy {
            max_events: Some(EVOLUTION_LOG_MAX_EVENTS),
            max_age_secs: None,
        }));
        let guard = Arc::new(CycleGuard::default());
        // Per-domain Skills backoff persists across ticks for the daemon's
        // lifetime (kernel review S5).
        let backoff = Arc::new(std::sync::Mutex::new(SkillsBackoff::default()));
        // Per-fingerprint Context backoff, same lifetime and same reason: a
        // mutation the measurement falsified must not be re-applied every tick.
        let context_backoff = Arc::new(std::sync::Mutex::new(ContextBackoff::default()));
        let mut tick_no: u64 = 0;
        let mut tick = tokio::time::interval(std::time::Duration::from_secs(interval_secs.max(1)));
        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        // Skip the immediate first tick — nothing has accrued at boot.
        tick.tick().await;
        loop {
            tick.tick().await;
            tick_no += 1;
            let Some(_token) = guard.try_begin() else {
                tracing::warn!("evolution cadence tick skipped: previous cycle still running");
                continue;
            };
            // Isolate a panicking cycle to its tick; the token is held across
            // the await so an overlapping tick still skips.
            let cycle_state = state.clone();
            let cycle_backoff = backoff.clone();
            let cycle_context_backoff = context_backoff.clone();
            let outcome = tokio::spawn(async move {
                run_evolution_cadence_cycle(
                    &cycle_state,
                    &cycle_backoff,
                    &cycle_context_backoff,
                    tick_no,
                )
                .await
            })
            .await;
            let mut data: HashMap<String, Value> = HashMap::new();
            data.insert("source".into(), Value::from("cadence"));
            match outcome {
                Ok(Some(report)) => {
                    // No-op cycle: nothing planned, nothing run — skip the
                    // append entirely (S4).
                    if report.plan.evolve_now.is_empty() && report.steps.is_empty() {
                        continue;
                    }
                    data.insert(
                        "report".into(),
                        serde_json::to_value(&report).unwrap_or(Value::Null),
                    );
                }
                Ok(None) => {
                    data.insert("error".into(), Value::from("no shared engine"));
                }
                Err(e) => {
                    data.insert("error".into(), Value::from(format!("cycle panicked: {e}")));
                }
            }
            log.append(EventKind::EvolutionTriggered, None, None, data);
        }
    }))
}

#[cfg(test)]
mod tests {
    use super::*;

    fn ev(kind: EventKind, action: Option<&str>) -> Event {
        Event {
            kind,
            action_id: action.map(str::to_string),
            proposal_id: None,
            data: HashMap::new(),
            timestamp: chrono::Utc::now(),
            prev_hash: None,
            hash: None,
        }
    }

    fn connector(slug: &str, connected: bool) -> ConnectorStatus {
        ConnectorStatus {
            slug: slug.into(),
            name: slug.into(),
            url: format!("https://example.com/{slug}"),
            connected,
            tool_count: 1,
            enabled_count: 1,
            last_error: if connected {
                None
            } else {
                Some("dial failed".into())
            },
        }
    }

    // --- populaters ---

    #[test]
    fn harness_pressure_counts_recurring_failure_share() {
        // 4 recurring rejections of the same action + 4 unrelated successes:
        // implicated 4 of 8 events → pressure 0.5, evidence 8.
        let mut events = vec![
            ev(EventKind::ActionRejected, Some("a1")),
            ev(EventKind::ActionRejected, Some("a1")),
            ev(EventKind::ActionRejected, Some("a1")),
            ev(EventKind::ActionRejected, Some("a1")),
        ];
        for _ in 0..4 {
            events.push(ev(EventKind::ActionSucceeded, Some("ok")));
        }
        let c = harness_component_from_events(&events).expect("component");
        assert_eq!(c.component, EvolvableComponent::Harness);
        assert!((c.signals.pressure - 0.5).abs() < 1e-9, "{c:?}");
        assert_eq!(c.signals.evidence, 8);
    }

    #[test]
    fn harness_one_off_failures_are_zero_pressure() {
        // Single occurrences never form an intervention (min_occurrences 2).
        let events = vec![
            ev(EventKind::ActionRejected, Some("a1")),
            ev(EventKind::ActionFailed, Some("a2")),
            ev(EventKind::ActionSucceeded, Some("a3")),
        ];
        let c = harness_component_from_events(&events).unwrap();
        assert_eq!(c.signals.pressure, 0.0);
        assert_eq!(c.signals.evidence, 3);
    }

    #[test]
    fn harness_empty_log_is_absent_not_zero() {
        assert!(harness_component_from_events(&[]).is_none());
    }

    #[test]
    fn tools_pressure_is_disconnected_share() {
        let list = vec![
            connector("up", true),
            connector("down1", false),
            connector("down2", false),
            connector("up2", true),
        ];
        let c = tools_component_from_connectors(&list).expect("component");
        assert_eq!(c.component, EvolvableComponent::Tools);
        assert!((c.signals.pressure - 0.5).abs() < 1e-9, "{c:?}");
        assert_eq!(c.signals.evidence, 4);
    }

    #[test]
    fn tools_absent_when_no_connectors_configured() {
        assert!(tools_component_from_connectors(&[]).is_none());
    }

    // --- failure-trace folding ---

    #[test]
    fn failed_trace_events_fold_failure_kinds_only() {
        let mut failed = ev(EventKind::ActionFailed, Some("a1"));
        failed.data.insert("tool".into(), Value::from("http_get"));
        failed.data.insert("error".into(), Value::from("timeout"));
        let events = vec![
            failed,
            ev(EventKind::ActionSucceeded, Some("a2")),
            ev(EventKind::PolicyViolation, Some("a3")),
            ev(EventKind::ReplanExhausted, None),
        ];
        let traces = failed_trace_events(&events);
        assert_eq!(traces.len(), 3);
        assert_eq!(traces[0].kind, "action_failed");
        assert_eq!(traces[0].tool.as_deref(), Some("http_get"));
        assert_eq!(traces[0].action_id.as_deref(), Some("a1"));
        assert_eq!(traces[0].reward, Some(0.0));
        assert_eq!(traces[1].kind, "policy_violation");
        assert_eq!(traces[2].kind, "replan_exhausted");
    }

    #[test]
    fn failed_trace_events_fold_only_ungrounded_completions() {
        // A completion the runtime could not ground (met but !grounded) is a
        // false-success failure exemplar; a grounded or not-yet-met verdict is not.
        let mut ungrounded = ev(EventKind::GoalEvaluated, None);
        ungrounded.data.insert("met".into(), Value::Bool(true));
        ungrounded
            .data
            .insert("grounded".into(), Value::Bool(false));
        let mut grounded = ev(EventKind::GoalEvaluated, None);
        grounded.data.insert("met".into(), Value::Bool(true));
        grounded.data.insert("grounded".into(), Value::Bool(true));
        let mut in_progress = ev(EventKind::GoalEvaluated, None);
        in_progress.data.insert("met".into(), Value::Bool(false));
        in_progress
            .data
            .insert("grounded".into(), Value::Bool(false));

        let traces = failed_trace_events(&[ungrounded, grounded, in_progress]);
        assert_eq!(
            traces.len(),
            1,
            "only the met-but-ungrounded completion is a failure"
        );
        assert_eq!(traces[0].kind, "goal_evaluated");
        assert_eq!(traces[0].reward, Some(0.0));
    }

    #[test]
    fn failed_trace_events_fold_problematic_turn_completions_only() {
        let mut truncated = ev(EventKind::TurnCompleted, None);
        truncated
            .data
            .insert("decision".into(), Value::from("empty_tool_calls"));
        truncated
            .data
            .insert("was_truncated".into(), Value::Bool(true));
        let mut capped = ev(EventKind::TurnCompleted, None);
        capped
            .data
            .insert("decision".into(), Value::from("max_turns"));
        capped
            .data
            .insert("was_truncated".into(), Value::Bool(false));
        let mut clean = ev(EventKind::TurnCompleted, None);
        clean
            .data
            .insert("decision".into(), Value::from("empty_tool_calls"));
        clean
            .data
            .insert("was_truncated".into(), Value::Bool(false));

        let traces = failed_trace_events(&[truncated, capped, clean]);
        assert_eq!(traces.len(), 2, "a clean finish is not a failure");
        assert!(traces.iter().all(|t| t.kind == "turn_completed"));
    }

    // --- maintenance sizing ---

    #[test]
    fn maintenance_input_prices_backlog_off_live_stats() {
        let stats = car_memgine::memsys::MemoryStats {
            total_facts: 100,
            outstanding_outdated: 5,
            facts_superseded: 10,
            ..Default::default()
        };
        let input = maintenance_input_from_stats(&stats);
        assert_eq!(input.dirty_regions, 15);
        assert_eq!(input.total_regions, 100);
        assert!((input.global_structural_gain - 0.10).abs() < 1e-9);
        // Low churn → localized wins under decide_maintenance.
        let d = decide_maintenance(&input);
        assert_eq!(
            d.strategy,
            car_memgine::maintenance::MaintenanceStrategy::Localized,
            "{d:?}"
        );
    }

    #[test]
    fn maintenance_input_clean_store_is_noop() {
        let input = maintenance_input_from_stats(&car_memgine::memsys::MemoryStats::default());
        let d = decide_maintenance(&input);
        assert_eq!(
            d.strategy,
            car_memgine::maintenance::MaintenanceStrategy::NoOp
        );
    }

    // --- executor mechanics ---

    #[tokio::test]
    async fn memory_evolution_dry_run_reports_without_consolidating() {
        let engine = Arc::new(tokio::sync::Mutex::new(MemgineEngine::new(None)));
        let out = run_memory_evolution(&engine, true).await.unwrap();
        assert!(out.summary.starts_with("dry_run"), "{out:?}");
        assert!(!out.applied, "dry run must not count as applied (S2)");
        // Real run against an empty engine still completes (consolidate is
        // side-effect-observable via its report fields) and IS an applied pass.
        let real = run_memory_evolution(&engine, false).await.unwrap();
        assert!(
            real.summary.contains("\"mechanism\":\"consolidate\""),
            "{real:?}"
        );
        assert!(real.applied);
    }

    #[tokio::test]
    async fn skills_evolution_without_inference_is_an_honest_error() {
        let engine = Arc::new(tokio::sync::Mutex::new(MemgineEngine::new(None)));
        let err = run_skills_evolution(&engine, &[], false).await.unwrap_err();
        assert_eq!(err, "no inference engine");
    }

    // --- cadence context backoff (S5, applied to Context) ---

    #[test]
    fn context_backoff_widens_exponentially_and_clears_when_the_change_pays() {
        let fp = "context:context:abcd1234";
        let mut b = ContextBackoff::default();

        // Never falsified → due immediately.
        assert!(b.is_due(fp, 1));
        // Falsified at tick 1 → next attempt allowed at 1 + 2^1 = 3.
        b.note_falsified(fp, 1);
        assert!(!b.is_due(fp, 2), "tick 2 still backing off");
        assert!(b.is_due(fp, 3));
        // Falsified again at 3 → next at 3 + 2^2 = 7.
        b.note_falsified(fp, 3);
        assert!(!b.is_due(fp, 6));
        assert!(b.is_due(fp, 7));
        // A different proposal for the same pillar is not the thing that
        // failed, so it is unaffected.
        assert!(b.is_due("context:context:99999999", 4));
        // A change that measurably pays clears its window; a relapse starts
        // fresh rather than inheriting the old exponent.
        b.clear(fp);
        assert!(b.is_due(fp, 4));
        b.note_falsified(fp, 4);
        assert!(b.is_due(fp, 6), "exponent restarted at 2^1");
    }

    #[test]
    fn context_backoff_helpers_are_transparent_without_a_backoff() {
        // The session path passes `None`: a person asking for the check now
        // gets it now, and the helpers must never gate on absence.
        assert!(backoff_due(None, "context:context:abcd1234"));
        note_falsified(None, "context:context:abcd1234");
        clear_backoff(None, "context:context:abcd1234");
    }

    // --- cadence skills backoff (S5) ---

    #[test]
    fn skills_backoff_widens_exponentially_and_resets_on_recovery() {
        let mut b = SkillsBackoff::default();
        let flagged = vec!["web".to_string()];

        // Never attempted → due immediately.
        assert_eq!(b.due(&flagged, 1), flagged);
        // Attempt at tick 1 → next allowed at 1 + 2^1 = 3.
        b.note_attempt("web", 1);
        assert!(b.due(&flagged, 2).is_empty(), "tick 2 still backing off");
        assert_eq!(b.due(&flagged, 3), flagged);
        // Second attempt at tick 3 → next at 3 + 2^2 = 7.
        b.note_attempt("web", 3);
        assert!(b.due(&flagged, 6).is_empty());
        assert_eq!(b.due(&flagged, 7), flagged);
        // Recovery (no longer flagged) resets the counter.
        b.reset_recovered(&[]);
        b.note_attempt("web", 10);
        // Fresh entry: attempts back to 1 → next at 10 + 2 = 12.
        assert_eq!(b.due(&flagged, 12), flagged);
    }

    #[test]
    fn skills_backoff_exponent_is_capped() {
        let mut b = SkillsBackoff::default();
        for t in 0..20 {
            b.note_attempt("stuck", t);
        }
        // Cap: 2^6 = 64 ticks after the last attempt (at tick 19).
        let flagged = vec!["stuck".to_string()];
        assert!(b.due(&flagged, 19 + 63).is_empty());
        assert_eq!(b.due(&flagged, 19 + 64), flagged);
    }

    #[test]
    fn skills_backoff_only_gates_the_attempted_domain() {
        let mut b = SkillsBackoff::default();
        let flagged = vec!["a".to_string(), "b".to_string()];
        b.note_attempt("a", 1);
        assert_eq!(b.due(&flagged, 2), vec!["b".to_string()]);
    }

    // --- cadence guard ---

    #[test]
    fn cycle_guard_blocks_overlap_and_releases_on_drop() {
        let guard = CycleGuard::default();
        let token = guard.try_begin().expect("first claim");
        assert!(guard.try_begin().is_none(), "in-flight cycle must block");
        drop(token);
        assert!(guard.try_begin().is_some(), "released after drop");
    }

    #[test]
    fn cycle_guard_releases_even_when_cycle_errors() {
        let guard = CycleGuard::default();
        let r: Result<(), ()> = {
            let _token = guard.try_begin().unwrap();
            Err(())
        };
        assert!(r.is_err());
        assert!(guard.try_begin().is_some(), "drop on error path releases");
    }
}