areev_loop/

engine.rs

//! The engine: the analyze → DISCOVER → ENRICH → validate/dedup → store
//! pipeline, the run-outcome contract, and the review/apply/rollback lifecycle
//! with the governance gates. The DETERMINISTIC output is a pure function of
//! (store state, params, now); the optional LLM stages (§9) only *add* cited
//! drafts (origin=llm, never auto-apply) and whitelisted guidance — with no
//! backend they are the identity, so the deterministic path is unchanged.
//! Auto-apply execution is gated behind a conservative shape check and stays
//! off by default.

use crate::analyzer::{AnalyzeCtx, Analyzer, OutcomeInput};
use crate::cal;
use crate::config::{AppliedRecord, LoopPersisted};
use crate::error::{Error, Result};
use crate::manifest::{AnalyzerManifest, Capability};
use crate::model::{normalize_ident, ActionKind, GrainRecord, Origin, Severity, TargetRef};
use crate::recommendation::{
    dedup_key, AuditRecord, ObserverType, Proposal, RecStatus, Recommendation, Summary,
    MAX_BECAUSE, MAX_EVIDENCE, Checkpoint};
use crate::substrate::{Capabilities, OmsSubstrate, ReadOpts, SubstrateRead};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use std::collections::{BTreeMap, BTreeSet};

/// The namespace the loop's own grains (recommendations, audit) live in.
pub const LOOP_NS: &str = "areev-loop";

/// Host-granted authority, per connection. `admin` implies all.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Scope {
    Read,
    Write,
    Review,
    Apply,
    Admin,
}

/// A set of granted scopes.
#[derive(Debug, Clone, Default)]
pub struct ScopeSet(Vec<Scope>);

impl ScopeSet {
    pub fn of(scopes: &[Scope]) -> Self {
        ScopeSet(scopes.to_vec())
    }
    /// The local root of trust: whoever can run against the file holds all
    /// scopes (the CLI/embedded posture).
    pub fn all() -> Self {
        ScopeSet(vec![Scope::Admin])
    }
    pub fn has(&self, s: Scope) -> bool {
        self.0.contains(&Scope::Admin) || self.0.contains(&s)
    }
}

/// A review decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Decision {
    Approve,
    Reject,
}

/// Gating and scoping options for a run.
#[derive(Debug, Clone, Default)]
pub struct RunOptions {
    pub min_new: Option<u64>,
    pub min_new_errors: Option<u64>,
    pub if_stale_ms: Option<i64>,
    /// Optional global namespace filter (empty = all).
    pub namespaces: Vec<String>,
    /// Re-analyze the WHOLE memory this pass, not just grains since the last-run
    /// watermark (`areev loop reflect`). Dedup/cooldowns still suppress anything
    /// already queued, and the watermark still advances at the end — so a sweep
    /// is safe to run any time and later runs stay incremental. Mainly widens the
    /// watermark-sensitive inputs (tool-failure window, the non-parasitic LLM
    /// evidence bundle) to the full history.
    pub full_sweep: bool,
    /// The principal that invoked this run. Recorded as co-creator on every
    /// non-`Builtin` (LLM / external-command) recommendation the run stores,
    /// so the review gate's self-approval block also fires for whoever
    /// triggered the model that authored the finding — an LLM draft is
    /// authored *via* its trigger, unlike a deterministic finding, which is
    /// computed. `None` (a headless/scheduled run) records no co-creator.
    pub triggering_actor: Option<String>,
}

/// Whether a run executed or was skipped.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum RunOutcome {
    Ran,
    Skipped,
}

/// Why a run was a no-op. `LockHeld` is produced by the host adapter (a
/// concurrent writer), surfaced here for a single contract.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SkipReason {
    MinNewNotMet,
    NotStale,
    LockHeld,
    /// The host policy's `cadence` block set a threshold and none was met.
    CadenceNotDue,
}

/// One analyzer that did not contribute drafts, with why.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AnalyzerSkip {
    pub id: String,
    pub reason: String,
}

/// The run-outcome contract (proposal §13): one shape across CLI/API/MCP/bindings.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct RunResult {
    pub outcome: RunOutcome,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub skip_reason: Option<SkipReason>,
    pub new_grains: u64,
    pub new_error_events: u64,
    pub proposed: u64,
    pub deduped: u64,
    pub stored: u64,
    /// Of the stored recommendations, how many were auto-applied by policy.
    #[serde(default)]
    pub auto_applied: u64,
    #[serde(default)]
    pub analyzers_run: Vec<String>,
    #[serde(default)]
    pub analyzers_skipped: Vec<AnalyzerSkip>,
    /// Where the LLM's contribution went, stage by stage. `None` when no
    /// backend is attached.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub llm_funnel: Option<LlmFunnel>,
}

/// The DISCOVER pipeline's attrition, counted.
///
/// "The model contributed nothing" has at least five distinct causes, and
/// they call for opposite responses: an empty bundle is a capture problem, a
/// model that abstained may need better evidence or a better prompt, drafts
/// dying at GROUND suggest fabrication, drafts dying at VERIFY suggest they
/// were vague, and drafts dying at the floor were merely unconfident. Without
/// this they are indistinguishable from the outside — every one of them
/// renders as an empty ledger and reads like a clean null. That ambiguity
/// cost a six-cell measurement run before it was noticed.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct LlmFunnel {
    /// Grains offered to the model as evidence. Zero means nothing to reflect on.
    pub evidence: u64,
    /// Drafts the model returned.
    pub proposed: u64,
    /// Survived the cite-check and target-class filter.
    pub cited: u64,
    /// Dropped for citing a hash the bundle does not hold. Its own counter
    /// because the fix is specific: models copy long hex badly, and that is
    /// a different problem from one aiming at a surface it may not touch.
    pub dropped_uncited: u64,
    /// Dropped for targeting a class the vocabulary may not reach — a prompt,
    /// host config, or its own grader.
    pub dropped_target: u64,
    /// Survived GROUND — their premises were found in the cited evidence.
    pub grounded: u64,
    /// Verdicts GROUND actually returned. `grounded = 0` with verdicts > 0 is
    /// the gate refusing every draft; `grounded = 0` with verdicts = 0 is a
    /// grader that answered with nothing usable, which is a backend problem
    /// wearing a gate's clothes.
    pub ground_verdicts: u64,
    /// The GROUND call itself failed — no response at all. The engine
    /// fail-softs here by design, so without this the run looks like a model
    /// that had nothing to say.
    pub ground_call_failed: bool,
    /// Survived VERIFY's adversarial pass.
    pub kept: u64,
    /// Cleared the confidence floor and reached the queue.
    pub stored: u64,
    /// Kept, but demoted to advisory for citing fewer distinct evidence
    /// grains than `Policy::min_evidence`. Omitted when zero, so a run under
    /// the default policy reads exactly as before.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub advisory_thin_evidence: u64,
    /// Kept, but dropped before the queue for restating a live lesson on the
    /// same entity in other words (`Policy::near_duplicate = suppress`). Its
    /// own counter beside `dropped_uncited` and `dropped_target`, so "the
    /// model contributed nothing" keeps its distinct causes. Omitted when
    /// zero.
    #[serde(default, skip_serializing_if = "is_zero")]
    pub dropped_near_duplicate: u64,
}

fn is_zero(n: &u64) -> bool {
    *n == 0
}

impl RunResult {
    fn skipped(reason: SkipReason, new_grains: u64, new_error_events: u64) -> Self {
        RunResult {
            outcome: RunOutcome::Skipped,
            skip_reason: Some(reason),
            new_grains,
            new_error_events,
            proposed: 0,
            deduped: 0,
            stored: 0,
            auto_applied: 0,
            llm_funnel: None,
            analyzers_run: vec![],
            analyzers_skipped: vec![],
        }
    }

    pub fn ran(&self) -> bool {
        self.outcome == RunOutcome::Ran
    }
}

/// The engine holds the registered analyzers, the host policy, and an optional
/// LLM enrichment backend (§9).
pub struct Engine {
    analyzers: Vec<Box<dyn Analyzer>>,
    policy: crate::policy::Policy,
    /// Optional LLM backend. `None` → the DISCOVER/ENRICH stages are the
    /// identity, so the pipeline is byte-for-byte the deterministic path.
    llm: Option<Box<dyn crate::llm::LlmBackend>>,
    /// Optional separate backend for the GROUND stage (§5.2, §11). `None` →
    /// grounding rides `llm`. Lets a team point entailment at a cheaper or
    /// specialized model (or take the generative model out of grounding
    /// entirely) without changing the proposer/verifier.
    ground_llm: Option<Box<dyn crate::llm::LlmBackend>>,
}

pub(crate) struct AnalysisPass {
    pub(crate) survivors: Vec<Recommendation>,
    proposed: u64,
    deduped: u64,
    analyzers_run: Vec<String>,
    pub(crate) analyzers_skipped: Vec<AnalyzerSkip>,
    llm_funnel: Option<LlmFunnel>,
}

impl Engine {
    /// An engine with the default built-ins and a default (fully closed)
    /// policy — nothing auto-applies, no LLM.
    pub fn with_builtins() -> Self {
        Engine {
            analyzers: crate::analyzer::builtin_analyzers(),
            policy: crate::policy::Policy::default(),
            llm: None,
            ground_llm: None,
        }
    }

    /// An engine with no analyzers (register your own).
    pub fn empty() -> Self {
        Engine {
            analyzers: vec![],
            policy: crate::policy::Policy::default(),
            llm: None,
            ground_llm: None,
        }
    }

    /// Install a host policy (the only place auto-apply is granted).
    pub fn with_policy(mut self, policy: crate::policy::Policy) -> Self {
        self.policy = policy;
        self
    }

    /// Attach an optional LLM enrichment backend (§9). Only ever *adds* cited
    /// draft recommendations (stamped `origin = llm`, never auto-applied) and
    /// whitelisted guidance notes — it can never gate or rewrite deterministic
    /// output.
    pub fn with_llm(mut self, backend: Box<dyn crate::llm::LlmBackend>) -> Self {
        self.llm = Some(backend);
        self
    }

    /// Attach a separate backend for the GROUND stage (§5.2). Without this,
    /// grounding uses the `with_llm` backend. Independent of the proposer so an
    /// operator can run entailment on a cheaper/specialized model.
    pub fn with_ground_llm(mut self, backend: Box<dyn crate::llm::LlmBackend>) -> Self {
        self.ground_llm = Some(backend);
        self
    }

    pub fn policy(&self) -> &crate::policy::Policy {
        &self.policy
    }

    /// Whether an LLM backend is attached (replay reports it as not replayed).
    pub fn has_llm(&self) -> bool {
        self.llm.is_some()
    }

    /// Register an additional analyzer (the linked-Rust seam).
    pub fn register(&mut self, analyzer: Box<dyn Analyzer>) {
        self.analyzers.push(analyzer);
    }

    pub fn analyzers(&self) -> &[Box<dyn Analyzer>] {
        &self.analyzers
    }

    /// Run the exact production analysis/validation path without Phase 0
    /// measurement or Phase 3 persistence. The immutable substrate borrow is
    /// the replay safety boundary: recommendations, audit grains, state,
    /// cooldowns, outcomes, and the op-log cannot be changed here.
    ///
    /// `overrides` is keyed by full analyzer id and overlays the file's stored
    /// parameter map. Unknown keys fail closed through `resolve_params` and
    /// surface as an analyzer skip, exactly as in a production run.
    pub fn analyze_only<S: OmsSubstrate>(
        &self,
        sub: &S,
        opts: &RunOptions,
        overrides: &BTreeMap<String, Map<String, Value>>,
        now_ms: i64,
    ) -> Result<Vec<Recommendation>> {
        let persisted = LoopPersisted::from_value(sub.load_state()?)?;
        let analysis_watermark = if opts.full_sweep {
            None
        } else {
            persisted.state.watermark_ms
        };
        Ok(self
            .analysis_pass(
                sub,
                &persisted,
                opts,
                overrides,
                analysis_watermark,
                now_ms,
                &[],
            )?
            .survivors)
    }

    /// Run one analysis pass. Idempotent under `dedup_key`; the watermark is
    /// advanced at the end, so a crashed run simply re-runs.
    pub fn run<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        opts: &RunOptions,
        now_ms: i64,
    ) -> Result<RunResult> {
        let mut persisted = LoopPersisted::from_value(sub.load_state()?)?;
        let watermark = persisted.state.watermark_ms;
        // A full sweep analyzes the whole memory (watermark ignored for the
        // analysis inputs), while gating, `new` counts, and the end-of-run
        // watermark advance still use the real watermark. Dedup/cooldowns keep
        // it from re-proposing what is already queued.
        let analysis_watermark = if opts.full_sweep { None } else { watermark };

        let new = count_new(sub, watermark)?;
        let (new_grains, new_error_events) = (new.grains, new.error_events);
        if let Some(reason) = gate(opts, &persisted, new_grains, new_error_events, now_ms) {
            return Ok(RunResult::skipped(reason, new_grains, new_error_events));
        }
        // The policy's cadence applies when the caller set no gate of its own
        // (host CLI flags > policy file) and is not asking for a sweep — a
        // sweep is a command, not a tick.
        let flags_set = opts.min_new.is_some() || opts.min_new_errors.is_some() || opts.if_stale_ms.is_some();
        if !flags_set && !opts.full_sweep {
            if let Some(reason) = cadence_gate(&self.policy.cadence, &persisted, new, now_ms) {
                return Ok(RunResult::skipped(reason, new_grains, new_error_events));
            }
        }

        // Phase 0: re-measure applied recommendations due for review (the
        // Verify gate). Records a measured outcome per due recommendation —
        // and, under policy, asks the gate's second question: does each
        // applied recommendation's PREMISE still stand?
        let mut outcome_inputs = measure_outcomes(sub, &mut persisted, &self.policy, now_ms)?;
        if self.policy.premise_drift {
            outcome_inputs.extend(detect_premise_drift(sub, &mut persisted, now_ms)?);
        }

        let AnalysisPass {
            survivors,
            proposed,
            deduped,
            analyzers_run,
            analyzers_skipped,
            llm_funnel,
        } = self.analysis_pass(
            &*sub,
            &persisted,
            opts,
            &BTreeMap::new(),
            analysis_watermark,
            now_ms,
            &outcome_inputs,
        )?;

        // Phase 3 (needs &mut): store survivors + propose audit, then
        // auto-apply the ones the host policy grants (all gates in §6.3).
        let mut stored = 0u64;
        let mut auto_applied = 0u64;
        for mut rec in survivors {
            let spec = rec.to_grain_spec(LOOP_NS)?;
            let hash = sub.put_grain(&spec)?;
            rec.hash = hash.clone();
            let actor = format!("engine:{}", rec.analyzer);
            let audit = AuditRecord {
                rec_hash: hash.clone(),
                from: None,
                to: RecStatus::Pending,
                actor: actor.clone(),
                observer_type: ObserverType::System,
                because: "analyzer proposed".into(),
                previous_audit_hash: None,
                gating: None,
                at_ms: now_ms,
            };
            let audit_hash = sub.put_grain(&audit.to_grain_spec(LOOP_NS))?;
            persisted
                .status_index
                .insert(hash.clone(), RecStatus::Pending);
            persisted.creators.insert(hash.clone(), actor);
            // An LLM or external-command finding exists because someone ran
            // it — record that principal too, so review can refuse the
            // trigger approving their own model's output. Builtin analyzers
            // stay engine-only: deterministic output has no human author.
            if !matches!(rec.origin, Origin::Builtin) {
                if let Some(trigger) = &opts.triggering_actor {
                    persisted.co_creators.insert(hash.clone(), trigger.clone());
                }
            }
            persisted.audit_heads.insert(hash.clone(), audit_hash);
            stored += 1;

            if self.can_auto_apply(&*sub, &rec) {
                self.auto_apply(sub, &mut persisted, &rec, now_ms)?;
                auto_applied += 1;
            }
        }

        persisted.state.last_run_ms = Some(now_ms);
        persisted.state.watermark_ms = Some(now_ms);
        sub.store_state(&persisted.to_value()?)?;

        Ok(RunResult {
            outcome: RunOutcome::Ran,
            skip_reason: None,
            new_grains,
            new_error_events,
            proposed,
            deduped,
            stored,
            auto_applied,
            analyzers_run,
            analyzers_skipped,
            llm_funnel,
        })
    }

    /// Shared Phase 1–2 implementation for production and replay. Cooldowns
    /// and the live recommendation queue are honored in both modes; only the
    /// production caller proceeds into the mutating Phase 3 below.
    #[allow(clippy::too_many_arguments)]
    #[allow(clippy::too_many_arguments)]
    fn analysis_pass<S: OmsSubstrate>(
        &self,
        sub: &S,
        persisted: &LoopPersisted,
        opts: &RunOptions,
        external_overrides: &BTreeMap<String, Map<String, Value>>,
        analysis_watermark: Option<i64>,
        now_ms: i64,
        outcome_inputs: &[OutcomeInput],
    ) -> Result<AnalysisPass> {
        let existing = existing_dedup_keys(sub, persisted)?;
        self.analysis_pass_inner(
            sub,
            persisted,
            &self.policy,
            opts,
            external_overrides,
            analysis_watermark,
            now_ms,
            outcome_inputs,
            &existing,
            None,
        )
    }

    /// The pass itself. `existing` is the set of dedup keys already open
    /// (production computes it from the stored queue; replay carries its own
    /// simulated queue). `replay` is `Some(reason)` when the pass is a
    /// rehearsal: the LLM stage and every analyzer that is not a pure
    /// function of the grains (external commands, telemetry rollups) are
    /// skipped with that reason rather than run against the present.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn analysis_pass_inner<S: OmsSubstrate>(
        &self,
        sub: &S,
        persisted: &LoopPersisted,
        policy: &crate::policy::Policy,
        opts: &RunOptions,
        external_overrides: &BTreeMap<String, Map<String, Value>>,
        analysis_watermark: Option<i64>,
        now_ms: i64,
        outcome_inputs: &[OutcomeInput],
        existing: &BTreeSet<String>,
        replay: Option<&str>,
    ) -> Result<AnalysisPass> {
        let mut analyzers_run = Vec::new();
        let mut analyzers_skipped = Vec::new();
        let mut candidates: Vec<Recommendation> = Vec::new();
        let caps = sub.capabilities();
        let verdicts = latest_verdicts(persisted);

        for analyzer in &self.analyzers {
            let m = analyzer.manifest();
            if let Some(why) = replay {
                let out_of_process = m.trust_class == crate::manifest::TrustClass::Command;
                let telemetry_fed = m.requires.contains(&crate::manifest::Capability::Telemetry);
                if out_of_process || telemetry_fed {
                    analyzers_skipped.push(AnalyzerSkip {
                        id: m.id.clone(),
                        reason: format!(
                            "not replayed: {}",
                            if out_of_process { why } else { "telemetry rollups are not time-indexed" }
                        ),
                    });
                    continue;
                }
            }
            let cfg = persisted.config.get(&m.id);
            let enabled = cfg.and_then(|c| c.enabled).unwrap_or(m.default_on);
            if !enabled {
                analyzers_skipped.push(AnalyzerSkip {
                    id: m.id.clone(),
                    reason: "disabled".into(),
                });
                continue;
            }
            if policy.denies(m.family()) {
                analyzers_skipped.push(AnalyzerSkip {
                    id: m.id.clone(),
                    reason: "denied by host policy".into(),
                });
                continue;
            }
            if let Some(missing) = missing_capability(m, caps) {
                analyzers_skipped.push(AnalyzerSkip {
                    id: m.id.clone(),
                    reason: format!("missing capability: {missing}"),
                });
                continue;
            }
            let mut param_overrides = cfg.map(|c| c.params.clone()).unwrap_or_default();
            if let Some(extra) = external_overrides.get(&m.id) {
                for (key, value) in extra {
                    param_overrides.insert(key.clone(), value.clone());
                }
            }
            let params = match m.resolve_params(&param_overrides) {
                Ok(p) => p,
                Err(e) => {
                    analyzers_skipped.push(AnalyzerSkip {
                        id: m.id.clone(),
                        reason: e.to_string(),
                    });
                    continue;
                }
            };
            let ns_owned = cfg.map(|c| c.namespaces.clone()).unwrap_or_default();
            let ns_slice: &[String] = if ns_owned.is_empty() {
                &opts.namespaces
            } else {
                &ns_owned
            };
            let reader: &dyn SubstrateRead = sub;
            let ctx = AnalyzeCtx::new(
                reader,
                &params,
                ns_slice,
                analysis_watermark,
                now_ms,
                outcome_inputs,
                &verdicts,
            );
            match analyzer.analyze(&ctx) {
                Ok(drafts) => {
                    analyzers_run.push(m.id.clone());
                    for draft in drafts {
                        match stamp(m, &params, draft, now_ms) {
                            Ok(rec) => candidates.push(rec),
                            Err(e) => analyzers_skipped.push(AnalyzerSkip {
                                id: m.id.clone(),
                                reason: e.to_string(),
                            }),
                        }
                    }
                }
                Err(e) => analyzers_skipped.push(AnalyzerSkip {
                    id: m.id.clone(),
                    reason: e.to_string(),
                }),
            }
        }

        let mut funnel = LlmFunnel::default();
        if self.llm.is_some() && replay.is_none() {
            candidates.extend(self.discover(
                sub,
                &candidates,
                analysis_watermark,
                &opts.namespaces,
                now_ms,
                &mut funnel,
            ));
        }

        let proposed = candidates.len() as u64;
        let mut seen = BTreeSet::new();
        let mut survivors = Vec::new();
        for candidate in candidates {
            let family = crate::manifest::analyzer_family(&candidate.analyzer);
            let floor = [
                severity_floor_for(persisted, &candidate.analyzer),
                policy.severity_floor(family),
            ]
            .into_iter()
            .flatten()
            .max();
            if floor.is_some_and(|floor| candidate.severity < floor) {
                continue;
            }
            if !seen.insert(candidate.dedup_key.clone()) {
                continue;
            }
            if existing.contains(&candidate.dedup_key) {
                continue;
            }
            if persisted
                .cooldowns
                .get(&candidate.dedup_key)
                .is_some_and(|until| now_ms < *until)
            {
                continue;
            }
            survivors.push(candidate);
        }
        let deduped = proposed - survivors.len() as u64;
        if self.llm.is_some() && replay.is_none() {
            self.enrich(&mut survivors);
        }
        Ok(AnalysisPass {
            survivors,
            proposed,
            deduped,
            analyzers_run,
            analyzers_skipped,
            llm_funnel: self.llm.is_some().then_some(funnel),
        })
    }

    /// DISCOVER (§9): ask the LLM for additional draft recommendations, given
    /// the deterministic findings as *context* and a bounded, provenance-tagged
    /// evidence bundle. Every returned draft must cite evidence present in the
    /// bundle and target a memory/query surface; it is stamped `origin = llm`
    /// (so it can never auto-apply) and enters the ordinary dedup/store path. A
    /// failed or garbled response yields no drafts — never a failed run.
    fn discover<S: OmsSubstrate>(
        &self,
        sub: &S,
        candidates: &[Recommendation],
        watermark: Option<i64>,
        namespaces: &[String],
        now_ms: i64,
        funnel: &mut LlmFunnel,
    ) -> Vec<Recommendation> {
        let Some(llm) = &self.llm else {
            return Vec::new();
        };
        let findings: Vec<crate::llm::FindingBrief> = candidates
            .iter()
            .take(32)
            .map(|c| crate::llm::FindingBrief {
                analyzer: c.analyzer.clone(),
                summary: c.summary.render(),
                target: c.target_ref.clone(),
                severity: c.severity.as_str().to_string(),
            })
            .collect();
        // Evidence bundle. Seeded first from the grains the deterministic
        // findings cite, THEN — the non-parasitic step (§11) — topped up with
        // RECENT grains (created since the last run) so the LLM gets its own
        // lens and can find issues in grains no analyzer flagged. Without this
        // the LLM could only elaborate near what determinism already caught.
        let attribution = self.policy.evidence_attribution;
        let mut evidence: Vec<crate::llm::EvidenceItem> = Vec::new();
        let mut bundle: BTreeSet<String> = BTreeSet::new();
        let mut ns_by_hash: std::collections::BTreeMap<String, String> = Default::default();
        'cited: for c in candidates {
            for h in &c.evidence {
                // Every source gets a RESERVED share of the bundle, because a
                // cap that only one source respects is not a budget. A single
                // tool_failure finding may cite up to MAX_EVIDENCE (64)
                // grains — the whole bundle — so without this the "give the
                // LLM its own lens" seeding below could be starved to nothing
                // by the very determinism it is supposed to look past. Found
                // live: with the deterministic findings citing enough, the
                // model never saw a single non-cited grain.
                if evidence.len() >= CITED_SEED_CAP {
                    break 'cited;
                }
                if !bundle.contains(h) {
                    if let Ok(Some(g)) = sub.grain(h) {
                        push_evidence(&mut evidence, &mut bundle, &mut ns_by_hash, &g, attribution);
                    }
                }
            }
        }
        let scan_ns: Vec<Option<&str>> = if namespaces.is_empty() {
            vec![None]
        } else {
            namespaces.iter().map(|n| Some(n.as_str())).collect()
        };
        let opts = ReadOpts { live_only: true, since_ms: watermark };
        // Tool grains carry the raw experience of a tool-using agent, and an
        // ERROR is the part a reflection pass can act on. Seeded after the
        // cited grains and inside its own reserved share, so a busy desk
        // cannot crowd out the facts and observations below.
        //
        // Without this the LLM saw tool failures only through the
        // deterministic findings that happened to cite them: it could
        // elaborate on what clustering already caught, but could never find
        // a failure clustering missed — the one thing it is here for. The
        // top-up below called itself "non-parasitic" while omitting the very
        // grain type the flagship analyzer reads.
        //
        // Two passes over the same reserved share: failures first, then — only
        // when the host lets the proposer author skills — the calls that
        // SUCCEEDED. A skill's evidence is a trajectory that worked, and a
        // model shown only what broke can never propose one (measured: with a
        // successful procedure in the memory and no failure, the bundle was
        // empty and DISCOVER was never called). Errors keep first claim on the
        // share, so a busy desk's successes cannot bury the failure signal the
        // lesson path exists for.
        let mut tool_seeded = 0usize;
        let seed_successes = self.policy.skills.enabled || self.policy.plans.enabled;
        'tools: for want_error in [true, false] {
            if !want_error && !seed_successes {
                break;
            }
            for ns in &scan_ns {
                if let Ok(recent) = sub.grains_of_type(crate::model::grain_type::TOOL, *ns, opts) {
                    for g in recent {
                        if tool_seeded >= TOOL_SEED_CAP
                            || evidence.len() >= EVIDENCE_CAP - LENS_RESERVE
                        {
                            break 'tools;
                        }
                        if g.is_error() != want_error {
                            continue;
                        }
                        let before = evidence.len();
                        push_evidence(&mut evidence, &mut bundle, &mut ns_by_hash, &g, attribution);
                        if evidence.len() > before {
                            tool_seeded += 1;
                        }
                    }
                }
            }
        }
        // Harness evidence, ahead of the general passes and read by EXPLICIT
        // namespace — the one path that sees past the `agent:` exclusion, and
        // only for the kinds named above. Early and reserved for the same
        // reason human notes are: a fold summary is written once per long run
        // and would lose every time to routine volume.
        //
        // A missing namespace or no read grant is nothing to reflect on, not an
        // error — the same posture `run_outcome` takes over the same rows.
        if let Ok(rows) =
            sub.grains_of_type(crate::model::grain_type::OBSERVATION, Some(crate::eval::HARNESS_NS), opts)
        {
            let mut seeded = 0usize;
            for g in rows {
                if seeded >= HARNESS_SEED_CAP || evidence.len() >= EVIDENCE_CAP - LENS_RESERVE {
                    break;
                }
                let kind = g.str_field("observation_kind").unwrap_or_default();
                if !HARNESS_EVIDENCE_KINDS.contains(&kind) {
                    continue;
                }
                let before = evidence.len();
                push_evidence(&mut evidence, &mut bundle, &mut ns_by_hash, &g, attribution);
                if evidence.len() > before {
                    seeded += 1;
                }
            }
        }
        // Observations BEFORE facts, with their own small reserve.
        //
        // An Observation is where a human's own words land — a supervisor's
        // note, a reviewer's correction, an instruction for next time. That
        // signal is stated ONCE by nature, so recency and frequency seeding
        // structurally bury it: one note loses to three hundred routine
        // records every time, and the rarest evidence is usually the most
        // valuable. Exhausting facts first (as this did) meant a desk with
        // any volume showed the model no human input at all.
        'notes: for ns in &scan_ns {
            if let Ok(recent) =
                sub.grains_of_type(crate::model::grain_type::OBSERVATION, *ns, opts)
            {
                for g in recent {
                    if evidence.len() >= CITED_SEED_CAP + TOOL_SEED_CAP + NOTE_SEED_CAP {
                        break 'notes;
                    }
                    push_evidence(&mut evidence, &mut bundle, &mut ns_by_hash, &g, attribution);
                }
            }
        }
        'seed: for gt in [
            crate::model::grain_type::FACT,
            crate::model::grain_type::OBSERVATION,
        ] {
            for ns in &scan_ns {
                if let Ok(recent) = sub.grains_of_type(gt, *ns, opts) {
                    for g in recent {
                        if evidence.len() >= EVIDENCE_CAP {
                            break 'seed;
                        }
                        push_evidence(&mut evidence, &mut bundle, &mut ns_by_hash, &g, attribution);
                    }
                }
            }
        }
        funnel.evidence = evidence.len() as u64;
        if evidence.is_empty() {
            return Vec::new(); // nothing to reflect on
        }
        // PROPOSE (§5.1): the abstention-legitimate objective — "nothing to
        // report" is a first-class, zero-penalty answer. The operator-taste
        // history (recent approve/reject decisions on llm findings) is passed so
        // the model learns what this reviewer accepts.
        let (approved, rejected) = self.llm_history(sub);
        let base = match self.policy.discover_objective {
            crate::policy::DiscoverObjective::ReviewQueue => DISCOVER_INSTRUCTIONS,
            crate::policy::DiscoverObjective::Learner => DISCOVER_LEARNER_INSTRUCTIONS,
        };
        // The skill kind is offered only when the host allows it, so the
        // vocabulary the model sees is exactly the vocabulary that can apply.
        let mut instructions = base.to_string();
        if self.policy.skills.enabled {
            instructions.push_str(&skill_instructions(self.policy.skills.min_steps));
        }
        if self.policy.plans.enabled {
            instructions.push_str(&plan_instructions(self.policy.plans.min_nodes));
        }
        // Offered only when a pile is on the table: the paragraph would
        // otherwise invite the model to invent one.
        if findings.iter().any(|f| f.analyzer.starts_with("loop.lesson_pile/")) {
            instructions.push_str(CONSOLIDATION_INSTRUCTIONS);
        }
        let request = crate::llm::LlmRequest {
            loop_proto: 1,
            op: "discover",
            instructions: &instructions,
            findings: findings.clone(),
            evidence: evidence.clone(),
            rejected,
            approved,
        };
        let Ok(body) = serde_json::to_string(&request) else {
            return Vec::new();
        };
        let raw = match llm.complete(&body) {
            Ok(r) => r,
            Err(_) => return Vec::new(), // fail-soft
        };
        // Cheap structural validation (cite-check + target class); collect the
        // survivors for the verifier. Storing the normalized target string
        // avoids a TargetRef clone through the pipeline.
        let caps = sub.capabilities();
        let mut validated: Vec<ValidatedDraft> = Vec::new();
        let drafts: Vec<_> = crate::llm::parse_discover(&raw)
            .recommendations
            .into_iter()
            .take(crate::llm::MAX_LLM_DRAFTS)
            .collect();
        funnel.proposed = drafts.len() as u64;
        let id_to_hash: std::collections::BTreeMap<&str, &str> = evidence
            .iter()
            .map(|e| (e.id.as_str(), e.hash.as_str()))
            .collect();
        for d in drafts {
            let mut cited: Vec<String> = Vec::new();
            for c in &d.evidence {
                if let Some(h) = resolve_citation(c, &bundle, &id_to_hash) {
                    if !cited.contains(&h) {
                        cited.push(h);
                    }
                }
            }
            if cited.is_empty() {
                funnel.dropped_uncited += 1;
                continue; // uncited → drop (no fabrication)
            }
            let Ok(target) = TargetRef::parse(&d.target) else {
                funnel.dropped_target += 1;
                continue;
            };
            let tc = target.target_class();
            // The classes the proposal vocabulary can reach: memory
            // (entity/grain), query (query/template) and code (tool). The
            // prompt (`doc:`), `host:`, `evalset:` and `model:` classes stay
            // closed to the model — it may not rewrite the agent's prompt, its
            // host config, or the gate that grades its own code.
            if !matches!(tc, "memory" | "query" | "code") {
                funnel.dropped_target += 1;
                continue;
            }
            // Resolve BEFORE the gates: what GROUND entails and VERIFY
            // stress-tests is exactly what an apply would do. A draft citing
            // fewer distinct grains than the host's evidence floor is kept
            // as a finding but offered as nothing a reviewer could apply — a
            // single instance may be worth a person's attention; it is not,
            // under that policy, a rule.
            let thin = cited.len() < self.policy.min_evidence as usize;
            if thin {
                funnel.advisory_thin_evidence += 1;
            }
            let resolved = if thin {
                None
            } else {
                resolve_proposal(sub, &d, &target, &cited, &ns_by_hash, caps, &self.policy)
            };
            // A `tool:` target has exactly one legal shape (Rule E1: a code
            // target REQUIRES action_kind code_revision), so an unresolved one
            // could not even be stamped advisory — drop it here rather than
            // spend two model calls on something that fails validation after.
            if tc == "code" && resolved.is_none() {
                funnel.dropped_target += 1;
                continue;
            }
            validated.push(ValidatedDraft {
                draft: d,
                target_ref: target.as_string(),
                cited,
                resolved,
            });
        }
        funnel.cited = validated.len() as u64;
        if validated.is_empty() {
            return Vec::new();
        }
        // GROUND → VERIFY → ROUTE (§5.2–5.4): only drafts that survive an
        // independent grounding entailment check *and* an adversarial
        // verification pass (each a separate call — proposer ≠ scorer) reach the
        // queue, stamped with the verifier's calibrated confidence.
        // GROUND may run on a separate backend (§11); VERIFY always uses the
        // main llm (the proposer≠scorer independence is on VERIFY, not GROUND).
        let ground = self.ground_llm.as_deref().unwrap_or(&**llm);
        let outcome_metric = self.outcome_metric_template(sub);
        self.verify_drafts(sub, &**llm, ground, validated, &evidence, outcome_metric, now_ms, funnel)
    }

    /// The metric an applicable LLM-authored proposal will be re-measured by,
    /// when the host policy names an evalset (`Policy::outcome_evalset`).
    /// The baseline is the newest run journaled so far — the state of the
    /// world BEFORE the proposal, which is what "did applying it help" has
    /// to be read against. No run journaled yet → no metric: the lesson is
    /// honestly unmeasured rather than scored against a number nobody
    /// recorded.
    fn outcome_metric_template<S: OmsSubstrate>(
        &self,
        sub: &S,
    ) -> Option<crate::recommendation::MetricSnapshot> {
        let e = self.policy.outcome_evalset.as_ref()?;
        let run = crate::eval::newest_eval_run(sub, &e.hash, None).ok().flatten()?;
        let baseline = crate::eval::run_value(&run, &e.field)?;
        // The schedule in the host's unit. `review_after_ms`/`horizons_ms` keep
        // the time view for anything that only understands time; when the
        // host counts runs or grains, `checkpoints` is the schedule.
        let schedule = e.schedule();
        let ms_only: Vec<i64> = schedule.iter().filter_map(Checkpoint::as_ms).collect();
        let all_ms = ms_only.len() == schedule.len();
        let horizons = if ms_only.is_empty() { vec![86_400_000] } else { ms_only };
        Some(crate::recommendation::MetricSnapshot {
            metric: format!("evalset:{}:{}", e.hash, e.field),
            baseline,
            unit: e.field.clone(),
            n: run.total(),
            window: "per-run".into(),
            subject: None,
            namespace: None,
            relation: None,
            query: format!(
                "RECALL facts WHERE subject = \"evalset:{}\" AND relation = \"mg:eval_run\"",
                e.hash
            ),
            review_after_ms: horizons[0],
            horizons_ms: if all_ms { horizons } else { Vec::new() },
            checkpoints: if all_ms { Vec::new() } else { schedule },
            higher_is_better: e.higher_is_better,
        })
    }

    /// GROUND → VERIFY → ROUTE (§5.2–5.4). Two independent model calls, batched
    /// over the drafts: a grounding-entailment gate ("does the cited evidence
    /// support the claim?"), then an adversarial keep/kill with a calibrated
    /// confidence. A draft reaches the queue only if it is grounded **and** kept
    /// **and** clears the confidence floor. Any failed call drops the whole LLM
    /// contribution for the run (safe default), never the run.
    #[allow(clippy::too_many_arguments)]
    #[allow(clippy::too_many_arguments)]
    fn verify_drafts<S: SubstrateRead>(
        &self,
        sub: &S,
        llm: &dyn crate::llm::LlmBackend,
        ground: &dyn crate::llm::LlmBackend,
        validated: Vec<ValidatedDraft>,
        evidence: &[crate::llm::EvidenceItem],
        outcome_metric: Option<crate::recommendation::MetricSnapshot>,
        now_ms: i64,
        funnel: &mut LlmFunnel,
    ) -> Vec<Recommendation> {
        use crate::llm::*;
        let ev_by_hash: std::collections::BTreeMap<&str, &EvidenceItem> =
            evidence.iter().map(|e| (e.hash.as_str(), e)).collect();
        let ev_for = |cited: &[String]| -> Vec<EvidenceItem> {
            cited
                .iter()
                .filter_map(|h| ev_by_hash.get(h.as_str()).map(|e| (*e).clone()))
                .collect()
        };

        // GROUND (§5.2): decompose-then-entail per draft, batched into one call.
        // The claim includes any authored lesson — the gate must judge exactly
        // what an apply would record, not only the finding's summary.
        let claims: Vec<GroundItem> = validated
            .iter()
            .enumerate()
            .map(|(i, v)| GroundItem {
                id: i,
                claim: claim_text(&v.draft, v.resolved.as_ref()),
                evidence: ev_for(&v.cited),
            })
            .collect();
        let ground_req = GroundRequest {
            loop_proto: 1,
            op: "ground",
            instructions: GROUND_INSTRUCTIONS,
            claims,
        };
        // A grounding pass that REFUSED every draft and one that never
        // answered are the same number of survivors and opposite problems:
        // the first is the gate doing its job, the second is a backend having
        // a bad minute while the engine fail-softs. Count the verdicts
        // actually returned so the two are distinguishable afterwards.
        let grounded: std::collections::BTreeSet<usize> = match serde_json::to_string(&ground_req)
            .ok()
            .and_then(|b| ground.complete(&b).ok())
        {
            Some(raw) => {
                let parsed = parse_ground(&raw);
                funnel.ground_verdicts = parsed.results.len() as u64;
                parsed
                    .results
                    .into_iter()
                    .filter(|r| r.supported)
                    .map(|r| r.id)
                    .collect()
            }
            None => {
                funnel.ground_call_failed = true;
                return Vec::new();
            }
        };
        funnel.grounded = grounded.len() as u64;
        if grounded.is_empty() {
            return Vec::new();
        }

        // VERIFY (§5.3): adversarial keep/kill over the grounded drafts, a
        // separate call from the proposer. Soundness + abstention only — NOT
        // novelty. Novelty is steered at DISCOVER and settled by human review;
        // asking a weak verifier to judge it just makes it hallucinate "already
        // known" and kill genuine findings (§11).
        let items: Vec<VerifyItem> = validated
            .iter()
            .enumerate()
            .filter(|(i, _)| grounded.contains(i))
            .map(|(i, v)| VerifyItem {
                id: i,
                // Same rule as GROUND: the adversarial pass sees the change.
                summary: claim_text(&v.draft, v.resolved.as_ref()),
                target: v.target_ref.clone(),
                evidence: ev_for(&v.cited),
            })
            .collect();
        let verify_req = VerifyRequest {
            loop_proto: 1,
            op: "verify",
            instructions: VERIFY_INSTRUCTIONS,
            findings: items,
        };
        let verdicts: std::collections::BTreeMap<usize, f64> =
            match serde_json::to_string(&verify_req).ok().and_then(|b| llm.complete(&b).ok()) {
                Some(raw) => parse_verify(&raw)
                    .results
                    .into_iter()
                    .filter(|r| r.keep)
                    .map(|r| (r.id, r.confidence.clamp(0.0, 1.0)))
                    .collect(),
                None => return Vec::new(),
            };

        funnel.kept = verdicts.len() as u64;
        // ROUTE (§5.4): grounded ∧ kept ∧ verifier-confidence ≥ floor. The
        // verifier's confidence (the independent signal) is what we trust and
        // stamp — not the proposer's self-report.
        let mut out = Vec::new();
        for (i, v) in validated.into_iter().enumerate() {
            if let Some(&conf) = verdicts.get(&i) {
                if conf >= MIN_LLM_CONFIDENCE {
                    // A lesson that restates, in other words, a live lesson
                    // on the same entity. `authored_dedup_key` collapses the
                    // same TEXT; meaning is measured here — cosine over the
                    // substrate's embedder when it has one, token-set Jaccard
                    // as the T0 floor — and the policy says whether the
                    // reviewer sees it marked or never sees it. A
                    // consolidation is exempt: superseding the pile is its
                    // whole point.
                    let near = match v.resolved.as_ref() {
                        Some(r) if r.action != ActionKind::Consolidate => r
                            .fact_fields
                            .as_ref()
                            .filter(|f| f.get("relation").and_then(Value::as_str) == Some("lesson"))
                            .map(|f| {
                                near_duplicates_of(
                                    sub,
                                    f.get("subject").and_then(Value::as_str).unwrap_or(""),
                                    f.get("namespace").and_then(Value::as_str),
                                    f.get("object").and_then(Value::as_str).unwrap_or(""),
                                )
                            })
                            .unwrap_or_default(),
                        _ => Vec::new(),
                    };
                    if !near.is_empty()
                        && self.policy.near_duplicate == crate::policy::NearDuplicateMode::Suppress
                    {
                        funnel.dropped_near_duplicate += 1;
                        continue;
                    }
                    let mut rec = stamp_llm(
                        llm.model(),
                        &v.draft,
                        v.target_ref,
                        v.cited,
                        v.resolved,
                        conf,
                        now_ms,
                    );
                    if let Some(best) = near.first() {
                        // The summary says so, and names the closest rule.
                        rec.summary.args.insert("near_count".into(), Value::from(near.len() as u64));
                        rec.summary.args.insert("near_score".into(), Value::from(best.score));
                        rec.summary.args.insert("near_method".into(), Value::from(best.method.clone()));
                        rec.summary.args.insert(
                            "near_hash".into(),
                            Value::from(best.hash.chars().take(12).collect::<String>()),
                        );
                        rec.summary.template_id = "llm.lesson_near_duplicate".into();
                        rec.near_duplicate_of = near;
                    }
                    // Only a proposal an apply can execute (and roll back)
                    // is measured: an advisory flag changes nothing, so
                    // there is nothing to hold or regress.
                    if rec.rollbackable {
                        rec.metric = outcome_metric.clone();
                    }
                    out.push(rec);
                }
            }
        }
        funnel.stored = out.len() as u64;
        out
    }

    /// Recent operator decisions on `origin = llm` findings — approved (incl.
    /// applied) and rejected summaries, most-recent first and bounded — so
    /// DISCOVER can learn what this reviewer accepts (§9). Best-effort: a read
    /// failure yields empty history, never an error.
    fn llm_history<S: OmsSubstrate>(&self, sub: &S) -> (Vec<String>, Vec<String>) {
        const MAX: usize = 20;
        let Ok(mut recs) = self.recommendations(sub, None) else {
            return (Vec::new(), Vec::new());
        };
        recs.retain(|r| matches!(r.origin, Origin::Llm { .. }));
        recs.sort_by_key(|r| std::cmp::Reverse(r.created_at_ms));
        let mut approved = Vec::new();
        let mut rejected = Vec::new();
        for r in &recs {
            match r.status {
                RecStatus::Approved | RecStatus::Applied | RecStatus::RolledBack
                    if approved.len() < MAX =>
                {
                    approved.push(r.summary.render());
                }
                RecStatus::Rejected if rejected.len() < MAX => rejected.push(r.summary.render()),
                _ => {}
            }
        }
        (approved, rejected)
    }

    /// ENRICH (§9): ask the LLM to add a short guidance note to the surviving
    /// deterministic recommendations. Whitelist-only — only `guidance` is
    /// merged (capped), and only onto recs that don't already have one; the
    /// engine-templated summary is never touched. Fail-soft.
    fn enrich(&self, survivors: &mut [Recommendation]) {
        let Some(llm) = &self.llm else {
            return;
        };
        if survivors.is_empty() {
            return;
        }
        let findings: Vec<crate::llm::FindingBrief> = survivors
            .iter()
            .map(|r| crate::llm::FindingBrief {
                analyzer: r.analyzer.clone(),
                summary: r.summary.render(),
                target: r.target_ref.clone(),
                severity: r.severity.as_str().to_string(),
            })
            .collect();
        let request = crate::llm::LlmRequest {
            loop_proto: 1,
            op: "enrich",
            instructions: ENRICH_INSTRUCTIONS,
            findings,
            evidence: Vec::new(),
            rejected: Vec::new(),
            approved: Vec::new(),
        };
        let Ok(body) = serde_json::to_string(&request) else {
            return;
        };
        let raw = match llm.complete(&body) {
            Ok(r) => r,
            Err(_) => return,
        };
        for note in crate::llm::parse_enrich(&raw).notes {
            if note.guidance.trim().is_empty() {
                continue;
            }
            if let Some(r) = survivors
                .iter_mut()
                .find(|r| r.target_ref == note.target && r.guidance.is_none())
            {
                r.guidance = Some(crate::llm::cap(&note.guidance, crate::llm::MAX_GUIDANCE_LEN));
            }
        }
    }

    /// Evaluate the auto-apply gate (§6.3) — ALL preconditions must hold:
    /// host opt-in + policy grant, builtin origin, memory/query target,
    /// non-destructive, and engine-side shape verification: SUPERSEDE-only
    /// structural curation (never an ADD that introduces evidence-derived
    /// text) whose every replacement is **value-identical** to the grain it
    /// supersedes (the exact-equality check — a near-duplicate consolidation
    /// stays pending). A default (closed) policy never grants, so nothing
    /// auto-applies.
    fn can_auto_apply<S: OmsSubstrate>(&self, sub: &S, rec: &Recommendation) -> bool {
        if !rec.origin.auto_apply_eligible() || rec.destructive {
            return false;
        }
        // The analyzer must declare its curation auto-appliable. An analyzer
        // whose manifest is `Never` (e.g. fork surfacing — a lossy merge) is
        // never auto-applied even if the payload passes the shape check.
        let manifest_ok = self
            .analyzers
            .iter()
            .map(|a| a.manifest())
            .find(|m| m.id == rec.analyzer)
            .is_some_and(|m| m.auto_apply == crate::manifest::AutoApplyClass::StructuralCuration);
        if !manifest_ok {
            return false;
        }
        let Ok(target) = TargetRef::parse(&rec.target_ref) else {
            return false;
        };
        let family = crate::manifest::analyzer_family(&rec.analyzer);
        if !self.policy.grants_auto_apply(family, target.target_class(), rec.severity) {
            return false;
        }
        // Shape verification: only a CAL batch of pure SUPERSEDE statements
        // whose replacements change no value is structural curation. An ADD
        // (introducing content), a FORGET (destructive), or a supersession
        // that alters any field disqualifies.
        match &rec.proposal {
            Proposal::Cal { cal } => cal
                .lines()
                .map(str::trim)
                .filter(|l| !l.is_empty())
                .all(|l| supersede_is_value_identical(sub, l)),
            _ => false,
        }
    }

    /// Apply a recommendation as `policy:auto` (the only `pending → applied`
    /// path). Records the applied inverse + a hash-chained audit grain.
    fn auto_apply<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        p: &mut LoopPersisted,
        rec: &Recommendation,
        now_ms: i64,
    ) -> Result<()> {
        let mut created = Vec::new();
        if let Proposal::Cal { cal } = &rec.proposal {
            // Belt and braces over the policy: `grants_auto_apply` already
            // excludes the `query` class, so a definition rewrite cannot reach
            // this path. If one ever did, it would apply with no recorded
            // inverse and no human BECAUSE — refuse instead.
            if cal.lines().map(str::trim).any(is_definition_statement) {
                return Err(Error::InvalidProposal(
                    "a definition rewrite (DEFINE QUERY / DEFINE TEMPLATE) is never \
                     auto-applied: it changes what every future context contains, so it \
                     requires a human APPROVE + APPLY with BECAUSE"
                        .into(),
                ));
            }
            for r in sub.execute_cal(cal)? {
                if let Some(h) = r.get("hash").and_then(Value::as_str) {
                    created.push(h.to_string());
                }
            }
        }
        let applied = AppliedRecord {
            applied_at_ms: now_ms,
            target_ref: rec.target_ref.clone(),
            rollbackable: rec.rollbackable,
            created_hashes: created,
            inverse_cal: None,
            metric: rec.metric.clone(),
        };
        let prev = p.audit_heads.get(&rec.hash).cloned();
        let audit = AuditRecord {
            rec_hash: rec.hash.clone(),
            from: Some(RecStatus::Pending),
            to: RecStatus::Applied,
            actor: "policy:auto".into(),
            observer_type: ObserverType::Policy,
            because: "auto-applied per host policy".into(),
            previous_audit_hash: prev,
            gating: None,
            at_ms: now_ms,
        };
        let audit_hash = sub.put_grain(&audit.to_grain_spec(LOOP_NS))?;
        p.audit_heads.insert(rec.hash.clone(), audit_hash);
        p.status_index.insert(rec.hash.clone(), RecStatus::Applied);
        p.applied.insert(rec.hash.clone(), applied);
        Ok(())
    }

    /// Approve or reject a pending recommendation. Requires the `review` scope,
    /// a mandatory BECAUSE, and blocks self-approval against the creating actor.
    #[allow(clippy::too_many_arguments)]
    pub fn review<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        rec_hash: &str,
        decision: Decision,
        actor: &str,
        observer: ObserverType,
        scopes: &ScopeSet,
        because: &str,
        now_ms: i64,
    ) -> Result<()> {
        if !scopes.has(Scope::Review) {
            return Err(Error::ScopeDenied("review".into()));
        }
        let because = validate_because(because)?;
        let mut p = LoopPersisted::from_value(sub.load_state()?)?;
        let status = *p
            .status_index
            .get(rec_hash)
            .ok_or_else(|| Error::NotFound(rec_hash.into()))?;
        let to = match decision {
            Decision::Approve => RecStatus::Approved,
            Decision::Reject => RecStatus::Rejected,
        };
        if !status.can_transition_to(to, false) {
            return Err(Error::LifecycleViolation(format!(
                "{} -> {}",
                status.as_str(),
                to.as_str()
            )));
        }
        if to == RecStatus::Approved {
            if let Some(creator) = p.creators.get(rec_hash) {
                if creator == actor {
                    return Err(Error::SelfApproval(format!(
                        "{actor} created this recommendation"
                    )));
                }
            }
            if let Some(trigger) = p.co_creators.get(rec_hash) {
                if trigger == actor {
                    return Err(Error::SelfApproval(format!(
                        "{actor} triggered the run that authored this recommendation"
                    )));
                }
            }
        }
        let prev = p.audit_heads.get(rec_hash).cloned();
        let audit = AuditRecord {
            rec_hash: rec_hash.into(),
            from: Some(status),
            to,
            actor: actor.into(),
            observer_type: observer,
            because,
            previous_audit_hash: prev,
            gating: None,
            at_ms: now_ms,
        };
        let audit_hash = sub.put_grain(&audit.to_grain_spec(LOOP_NS))?;
        p.audit_heads.insert(rec_hash.into(), audit_hash);
        p.status_index.insert(rec_hash.into(), to);
        if to == RecStatus::Rejected {
            if let Ok(rec) = load_rec(sub, rec_hash) {
                strike_cooldown(&mut p, rec.dedup_key, now_ms);
            }
        }
        sub.store_state(&p.to_value()?)?;
        Ok(())
    }

    /// Check everything [`apply`](Self::apply) would refuse on, without writing
    /// anything.
    ///
    /// This exists for the fused approve-and-apply callers (the bindings'
    /// `apply_recommendation`). Recording the approval first and *then* hitting
    /// the destructive gate strands the recommendation in `approved`, which has
    /// no exit but `applied` or `expired` — `approved → rejected` is not a
    /// legal transition — so a refused apply left the reviewer unable to
    /// dismiss it. Ask first, then approve.
    ///
    /// Deliberately does not check the lifecycle transition: the caller is
    /// about to make it legal by approving.
    /// `has_gating` is whether the caller will supply a gating run at apply:
    /// a gated revision (code or adapter) without one is refused HERE, before
    /// a fused approve-and-apply records the approval — `approved` has no
    /// exit but `applied` or `expired`, so asking after would strand it.
    pub fn preflight_apply<S: OmsSubstrate>(
        &self,
        sub: &S,
        rec_hash: &str,
        scopes: &ScopeSet,
        allow_destructive: bool,
        has_gating: bool,
    ) -> Result<()> {
        if !scopes.has(Scope::Apply) {
            return Err(Error::ScopeDenied("apply".into()));
        }
        let rec = load_rec(sub, rec_hash)?;
        if rec.destructive && (!scopes.has(Scope::Admin) || !allow_destructive) {
            return Err(Error::DestructiveGated(
                "destructive apply requires admin scope + allow_destructive".into(),
            ));
        }
        ensure_executable(rec.action_kind, &rec.proposal)?;
        if requires_gating(rec.action_kind) && !has_gating {
            return Err(Error::InvalidProposal(GATING_REQUIRED.into()));
        }
        Ok(())
    }

    /// Apply an approved recommendation. Requires `apply`; destructive payloads
    /// additionally require `admin` + `allow_destructive`. Records the applied
    /// info (inverse plan) for rollback.
    #[allow(clippy::too_many_arguments)]
    pub fn apply<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        rec_hash: &str,
        actor: &str,
        observer: ObserverType,
        scopes: &ScopeSet,
        because: &str,
        allow_destructive: bool,
        now_ms: i64,
    ) -> Result<AppliedRecord> {
        self.apply_inner(
            sub, rec_hash, actor, observer, scopes, because, allow_destructive, None, now_ms,
        )
    }

    /// Load the gating evidence for `rec_hash` from the RECORDED
    /// `mg:eval_run` summary named by `run_id` — the one loader every
    /// surface (CLI, bindings, MCP, HTTP) shares, so the stats that admit a
    /// gated revision can never come from a caller: they are read back from
    /// the Fact `areev eval run` journaled, within the recommendation's own
    /// pinned evalset (a run id from a different evalset simply isn't found).
    pub fn gating_evidence<S: OmsSubstrate>(
        &self,
        sub: &S,
        rec_hash: &str,
        run_id: &str,
    ) -> Result<crate::recommendation::GatingEvidence> {
        let rec = self
            .recommendations(sub, None)?
            .into_iter()
            .find(|r| r.hash == rec_hash)
            .ok_or_else(|| {
                Error::InvalidProposal(format!("recommendation {rec_hash} not found"))
            })?;
        let pin = rec.evalset_hash.ok_or_else(|| {
            Error::InvalidProposal(
                "this recommendation pins no evalset — a gating run applies only \
                 to code and adapter revisions"
                    .into(),
            )
        })?;
        // Read through the shared evalset reader, so the run that GATES an
        // apply and the runs that later JUDGE it are parsed by exactly one
        // piece of code — two parsers drifting apart would let a rule be
        // admitted on one reading of a summary and measured on another.
        match crate::eval::eval_run_by_id(sub, &pin, run_id)? {
            Some(run) => Ok(crate::recommendation::GatingEvidence {
                evalset_hash: pin,
                run_id: run.run_id,
                passed: run.passed,
                failed: run.failed,
            }),
            None => Err(Error::InvalidProposal(format!(
                "no recorded gate run '{run_id}' for evalset {pin} — run \
                 `areev eval run --evalset {pin} ...` first"
            ))),
        }
    }

    /// Apply WITH the §7.4 evalset-run edge — the only path that can apply a
    /// gated (code or adapter) revision. The evidence is validated against
    /// the recommendation's pin and recorded on the audit Observation.
    #[allow(clippy::too_many_arguments)]
    pub fn apply_gated<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        rec_hash: &str,
        actor: &str,
        observer: ObserverType,
        scopes: &ScopeSet,
        because: &str,
        allow_destructive: bool,
        gating: &crate::recommendation::GatingEvidence,
        now_ms: i64,
    ) -> Result<AppliedRecord> {
        self.apply_inner(
            sub,
            rec_hash,
            actor,
            observer,
            scopes,
            because,
            allow_destructive,
            Some(gating),
            now_ms,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn apply_inner<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        rec_hash: &str,
        actor: &str,
        observer: ObserverType,
        scopes: &ScopeSet,
        because: &str,
        allow_destructive: bool,
        gating: Option<&crate::recommendation::GatingEvidence>,
        now_ms: i64,
    ) -> Result<AppliedRecord> {
        if !scopes.has(Scope::Apply) {
            return Err(Error::ScopeDenied("apply".into()));
        }
        let because = validate_because(because)?;
        let mut p = LoopPersisted::from_value(sub.load_state()?)?;
        let status = *p
            .status_index
            .get(rec_hash)
            .ok_or_else(|| Error::NotFound(rec_hash.into()))?;
        if !status.can_transition_to(RecStatus::Applied, false) {
            return Err(Error::LifecycleViolation(format!(
                "{} -> applied (approve first)",
                status.as_str()
            )));
        }
        let rec = load_rec(sub, rec_hash)?;
        if rec.destructive && (!scopes.has(Scope::Admin) || !allow_destructive) {
            return Err(Error::DestructiveGated(
                "destructive apply requires admin scope + allow_destructive".into(),
            ));
        }
        // §7.4: a code or adapter revision applies ONLY through the
        // evalset-run edge — the pin must match, the pinned evalset must
        // still be LIVE (a superseded evalset invalidates in-flight
        // recommendations: they re-gate), and a failing gate admits nothing.
        if requires_gating(rec.action_kind) {
            let g = gating
                .ok_or_else(|| Error::InvalidProposal(GATING_REQUIRED.into()))?;
            let pin = rec.evalset_hash.as_deref().unwrap_or_default();
            if g.evalset_hash != pin {
                return Err(Error::InvalidProposal(format!(
                    "gating ran evalset {} but the recommendation is pinned \
                     to {pin} (Rule E1)",
                    g.evalset_hash
                )));
            }
            match sub.grain(pin)? {
                Some(evalset) if evalset.is_live() => {}
                Some(_) => {
                    return Err(Error::InvalidProposal(
                        "the pinned evalset was superseded after gating — \
                         the recommendation must re-gate (Rule E1)"
                            .into(),
                    ))
                }
                None => {
                    return Err(Error::InvalidProposal(format!(
                        "pinned evalset {pin} not found in the substrate"
                    )))
                }
            }
            if g.failed > 0 {
                return Err(Error::InvalidProposal(format!(
                    "the gating run failed {}/{} cases — a failing gate \
                     admits nothing",
                    g.failed,
                    g.passed + g.failed
                )));
            }
        }

        // Execute the proposal.
        let mut created = Vec::new();
        // The inverse of a change that creates no grain — see
        // `AppliedRecord::inverse_cal`. Captured BEFORE execution, because
        // afterwards the previous definition is gone.
        let mut inverse_cal: Option<String> = None;
        match &rec.proposal {
            Proposal::Cal { cal } => {
                for line in cal.lines().map(str::trim).filter(|l| !l.is_empty()) {
                    if !is_definition_statement(line) {
                        continue;
                    }
                    match sub.definition_inverse(line)? {
                        Some(inv) => inverse_cal = Some(inv),
                        None => {
                            return Err(Error::InvalidProposal(format!(
                                "this substrate cannot record a rollback inverse for {line:?}; \
                                 a definition rewrite that ROLLBACK could not undo is refused \
                                 rather than applied"
                            )))
                        }
                    }
                }
                let rows = sub.execute_cal(cal)?;
                for r in rows {
                    if let Some(h) = r.get("hash").and_then(Value::as_str) {
                        created.push(h.to_string());
                    }
                }
            }
            // The engine has no executable Edit primitive. Marking this
            // Applied used to be a lie (and rollback had no inverse).
            Proposal::Edit { .. } => {
                return Err(Error::InvalidProposal(ADVISORY_EDIT.into()));
            }
            // A gated code or adapter revision EXECUTES by writing the
            // promotion grain: an immutable record that this target now
            // resolves to the proposed code (a §7.4 blob address) or adapter
            // (the tuning seam's registry tuple). Hosts read the promotion
            // to re-resolve — `(tool:X, mg:code_promotion)` or
            // `(model:X, mg:adapter_promotion)`; retracting it is the
            // rollback inverse, so the apply is rollbackable end-to-end.
            Proposal::Data { data } if requires_gating(rec.action_kind) => {
                let relation = if rec.action_kind == ActionKind::AdapterRevision {
                    "mg:adapter_promotion"
                } else {
                    "mg:code_promotion"
                };
                // A code revision authored by DISCOVER carries its SOURCE
                // inline, because the discovery pass reads the substrate and
                // cannot write to it. Move it into the CAS here so the
                // promotion names a content ADDRESS: §7.4's rule that code
                // enters the substrate only through the blob seam holds
                // however the revision was authored, and the promotion grain
                // stays a pointer rather than swelling to hold a program.
                let mut promoted = data.clone();
                if let Some(Value::String(src)) = promoted.remove("source") {
                    let address = sub.put_blob(src.as_bytes())?;
                    promoted.insert("code_address".into(), Value::from(address));
                }
                let mut spec = crate::substrate::GrainSpec::new(
                    crate::model::grain_type::FACT,
                    LOOP_NS,
                )
                .with_field("subject", rec.target_ref.clone())
                .with_field("relation", relation)
                .with_field(
                    "object",
                    serde_json::to_string(&promoted)
                        .map_err(|e| Error::Internal(format!("encode promotion: {e}")))?,
                )
                .with_field("rec_hash", rec_hash.to_string());
                if let Some(g) = gating {
                    spec = spec
                        .with_field("gating_evalset", g.evalset_hash.clone())
                        .with_field("gating_run_id", g.run_id.clone());
                }
                created.push(sub.put_grain(&spec)?);
            }
            Proposal::Data { data } => {
                // OutcomeReview is the one executable Data shape: its
                // `revert_of` points at an earlier applied recommendation.
                // Reuse the ordinary rollback path so the created hashes are
                // really retracted and the original lifecycle/audit advances.
                let revert_of = data
                    .get("revert_of")
                    .and_then(Value::as_str)
                    .ok_or_else(|| Error::InvalidProposal(ADVISORY_DATA.into()))?;
                self.rollback(
                    sub,
                    revert_of,
                    actor,
                    observer,
                    scopes,
                    &because,
                    now_ms,
                )?;
                // rollback stored a newer lifecycle state; merge this apply
                // into that state rather than overwriting the rollback.
                p = LoopPersisted::from_value(sub.load_state()?)?;
                // A measured revert is a verdict on the finding, not only on
                // this apply: the lesson was tried and it hurt. Rolled-back
                // findings normally re-propose ("the situation returned"),
                // which is right for an operator's rollback — but here the
                // situation never left, so the next pass would re-propose the
                // same lesson at once and the reviewer would be asked to
                // re-approve what the Verify gate just retracted. Put the
                // reverted finding on the same doubling cooldown a rejection
                // earns; the operator can still re-propose it by hand.
                if let Ok(reverted) = load_rec(sub, revert_of) {
                    strike_cooldown(&mut p, reverted.dedup_key, now_ms);
                }
            }
        }

        let applied = AppliedRecord {
            applied_at_ms: now_ms,
            target_ref: rec.target_ref.clone(),
            rollbackable: rec.rollbackable,
            created_hashes: created,
            inverse_cal,
            metric: rec.metric.clone(),
        };
        let prev = p.audit_heads.get(rec_hash).cloned();
        let audit = AuditRecord {
            rec_hash: rec_hash.into(),
            from: Some(status),
            to: RecStatus::Applied,
            actor: actor.into(),
            observer_type: observer,
            because,
            previous_audit_hash: prev,
            gating: gating.cloned(),
            at_ms: now_ms,
        };
        let audit_hash = sub.put_grain(&audit.to_grain_spec(LOOP_NS))?;
        p.audit_heads.insert(rec_hash.into(), audit_hash);
        p.status_index.insert(rec_hash.into(), RecStatus::Applied);
        p.applied.insert(rec_hash.into(), applied.clone());
        sub.store_state(&p.to_value()?)?;
        Ok(applied)
    }

    /// Roll back an applied recommendation by retracting the grains it created.
    /// Fails for non-rollbackable applies (e.g. FORGET).
    #[allow(clippy::too_many_arguments)]
    pub fn rollback<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        rec_hash: &str,
        actor: &str,
        observer: ObserverType,
        scopes: &ScopeSet,
        because: &str,
        now_ms: i64,
    ) -> Result<()> {
        if !scopes.has(Scope::Apply) {
            return Err(Error::ScopeDenied("apply".into()));
        }
        let because = validate_because(because)?;
        let mut p = LoopPersisted::from_value(sub.load_state()?)?;
        let status = *p
            .status_index
            .get(rec_hash)
            .ok_or_else(|| Error::NotFound(rec_hash.into()))?;
        if !status.can_transition_to(RecStatus::RolledBack, false) {
            return Err(Error::LifecycleViolation(format!(
                "{} -> rolled_back",
                status.as_str()
            )));
        }
        let applied = p
            .applied
            .get(rec_hash)
            .cloned()
            .ok_or_else(|| Error::NotFound(format!("no applied record for {rec_hash}")))?;
        if !applied.rollbackable {
            return Err(Error::LifecycleViolation(
                "recommendation is non-rollbackable (FORGET has no inverse)".into(),
            ));
        }
        for h in &applied.created_hashes {
            sub.retract(h, &format!("rollback of {rec_hash}"))?;
        }
        // A definition rewrite creates no grain, so retracting `created_hashes`
        // undoes nothing. Restoring it means re-running the statement captured
        // at apply time — the previous definition, or a DROP when there was
        // none. Runs BEFORE the audit is written, so a failed restore leaves
        // the recommendation `applied` (still true) rather than recording a
        // rollback that did not happen.
        if let Some(inverse) = &applied.inverse_cal {
            sub.execute_cal(inverse)?;
        }
        let prev = p.audit_heads.get(rec_hash).cloned();
        let audit = AuditRecord {
            rec_hash: rec_hash.into(),
            from: Some(status),
            to: RecStatus::RolledBack,
            actor: actor.into(),
            observer_type: observer,
            because,
            previous_audit_hash: prev,
            gating: None,
            at_ms: now_ms,
        };
        let audit_hash = sub.put_grain(&audit.to_grain_spec(LOOP_NS))?;
        p.audit_heads.insert(rec_hash.into(), audit_hash);
        p.status_index
            .insert(rec_hash.into(), RecStatus::RolledBack);
        sub.store_state(&p.to_value()?)?;
        Ok(())
    }

    /// List stored recommendations, optionally filtered by status. Status comes
    /// from the rebuildable index, not the immutable grain body. Ordered for
    /// review triage — highest severity first, then oldest first — and stable
    /// across runs for identical input.
    pub fn recommendations<S: OmsSubstrate>(
        &self,
        sub: &S,
        status_filter: Option<RecStatus>,
    ) -> Result<Vec<Recommendation>> {
        let p = LoopPersisted::from_value(sub.load_state()?)?;
        let grains = sub.grains_of_type(
            crate::model::grain_type::RECOMMENDATION,
            Some(LOOP_NS),
            ReadOpts {
                live_only: false,
                since_ms: None,
            },
        )?;
        let mut out = Vec::new();
        for g in grains {
            let mut rec = Recommendation::from_fields(&g.hash, &g.fields)?;
            rec.status = p
                .status_index
                .get(&g.hash)
                .copied()
                .unwrap_or(RecStatus::Pending);
            if let Some(f) = status_filter {
                if rec.status != f {
                    continue;
                }
            }
            out.push(rec);
        }
        // Review-queue order: worst first, then oldest first. Hash is only the
        // final tiebreak — sorting by it alone is deterministic per run but
        // meaningless across runs, because a grain's hash covers its timestamp,
        // so an identical queue comes back in a different order every time.
        // `dedup_key` is the last tiebreak that actually decides anything: it is
        // content-derived and stable across runs, whereas findings proposed in
        // the same sweep routinely share a `created_at_ms`. Hash trails it only
        // to make the ordering total.
        out.sort_by(|a, b| {
            b.severity
                .cmp(&a.severity)
                .then(a.created_at_ms.cmp(&b.created_at_ms))
                .then(a.dedup_key.cmp(&b.dedup_key))
                .then(a.hash.cmp(&b.hash))
        });
        Ok(out)
    }

    /// Per-analyzer effective settings for the Setup view: the manifest facts
    /// merged with the file-config (override or manifest default). Read-only.
    pub fn analyzer_settings<S: OmsSubstrate>(
        &self,
        sub: &S,
    ) -> Result<Vec<crate::config::AnalyzerSetting>> {
        let p = LoopPersisted::from_value(sub.load_state()?)?;
        Ok(self
            .analyzers
            .iter()
            .map(|a| {
                let m = a.manifest();
                let cfg = p.config.get(&m.id);
                crate::config::AnalyzerSetting {
                    id: m.id.clone(),
                    title: m.title.clone(),
                    description: m.description.clone(),
                    tier: format!("{:?}", m.tier),
                    trust_class: format!("{:?}", m.trust_class).to_lowercase(),
                    default_on: m.default_on,
                    enabled: cfg.and_then(|c| c.enabled).unwrap_or(m.default_on),
                    severity_floor: cfg
                        .and_then(|c| c.severity_floor)
                        .map(|s| s.as_str().to_string()),
                }
            })
            .collect())
    }

    /// Update one analyzer's file-config (enable/disable, severity floor, param
    /// overrides, namespace scoping). Requires `Admin`. Params are validated
    /// against the analyzer's manifest first (unknown keys rejected, fail-closed),
    /// and the analyzer must exist. Returns the merged config as stored. This is
    /// the only write into `persisted.config` — the config layer, never a grain.
    pub fn set_analyzer_config<S: OmsSubstrate>(
        &self,
        sub: &mut S,
        analyzer_id: &str,
        update: crate::config::AnalyzerConfigUpdate,
        scopes: &ScopeSet,
    ) -> Result<crate::config::AnalyzerConfig> {
        if !scopes.has(Scope::Admin) {
            return Err(Error::ScopeDenied("admin".into()));
        }
        let manifest = self
            .analyzers
            .iter()
            .map(|a| a.manifest())
            .find(|m| m.id == analyzer_id)
            .ok_or_else(|| Error::NotFound(format!("unknown analyzer {analyzer_id:?}")))?;
        // Validate params against the manifest BEFORE touching state.
        if let Some(params) = &update.params {
            manifest.resolve_params(params)?;
        }
        let mut p = LoopPersisted::from_value(sub.load_state()?)?;
        let cfg = p.config.entry(analyzer_id.to_string()).or_default();
        if let Some(enabled) = update.enabled {
            cfg.enabled = Some(enabled);
        }
        if update.clear_floor {
            cfg.severity_floor = None;
        } else if let Some(floor) = update.severity_floor {
            cfg.severity_floor = Some(floor);
        }
        if let Some(params) = update.params {
            cfg.params = params;
        }
        if let Some(ns) = update.namespaces {
            cfg.namespaces = ns;
        }
        let stored = cfg.clone();
        sub.store_state(&p.to_value()?)?;
        Ok(stored)
    }

    /// The measured outcome time series (the Verify gate's history) across all
    /// recommendations, ordered by when each checkpoint was measured.
    pub fn outcomes<S: OmsSubstrate>(&self, sub: &S) -> Result<Vec<crate::recommendation::OutcomeResult>> {
        let p = LoopPersisted::from_value(sub.load_state()?)?;
        let mut out: Vec<_> = p.outcomes.into_values().flatten().collect();
        // `metric` and `rec_hash` break the tie: checkpoints measured in the
        // same sweep share a `measured_at_ms`, and without a tiebreak the order
        // falls through to the map's rec_hash ordering, which shifts every run.
        out.sort_by(|a, b| {
            a.measured_at_ms
                .cmp(&b.measured_at_ms)
                .then(a.horizon_ms.cmp(&b.horizon_ms))
                .then(a.metric.cmp(&b.metric))
                .then(a.rec_hash.cmp(&b.rec_hash))
        });
        Ok(out)
    }

    /// A health snapshot — when the loop last ran, how much is un-analyzed
    /// since, and the queue counts. Lets a host surface "the loop may be stale"
    /// so a forgotten SessionEnd hook / cron doesn't silently kill it.
    pub fn health<S: OmsSubstrate>(&self, sub: &S, now_ms: i64) -> Result<Health> {
        let p = LoopPersisted::from_value(sub.load_state()?)?;
        let new = count_new(sub, p.state.watermark_ms)?;
        let (grains_since_run, error_events_since_run) = (new.grains, new.error_events);
        let recs = self.recommendations(sub, None)?;
        let mut pending = 0;
        let mut applied = 0;
        for r in &recs {
            match r.status {
                RecStatus::Pending => pending += 1,
                RecStatus::Applied => applied += 1,
                _ => {}
            }
        }
        // Stale if it has never run, or it's been a while / a lot has piled up.
        let stale = match p.state.last_run_ms {
            None => true,
            Some(last) => now_ms - last >= 7 * 86_400_000 || grains_since_run >= 100,
        };
        Ok(Health {
            last_run_ms: p.state.last_run_ms,
            grains_since_run,
            error_events_since_run,
            pending,
            applied,
            total: recs.len() as u64,
            stale,
        })
    }

    /// Approval-rate metric for `origin = llm` recommendations (reflection
    /// design §6b) — the live field-quality signal that accrues off the audit
    /// chain: what fraction of the model's *surfaced* proposals a reviewer
    /// accepts. Complements the offline Effective-Reliability eval.
    pub fn llm_metrics<S: OmsSubstrate>(&self, sub: &S) -> Result<LlmMetrics> {
        let recs = self.recommendations(sub, None)?;
        let mut m = LlmMetrics::default();
        for r in &recs {
            if !matches!(r.origin, Origin::Llm { .. }) {
                continue;
            }
            m.proposed += 1;
            match r.status {
                RecStatus::Pending => m.pending += 1,
                RecStatus::Approved | RecStatus::Applied | RecStatus::RolledBack => m.approved += 1,
                RecStatus::Rejected => m.rejected += 1,
                RecStatus::Expired => {}
            }
        }
        let decided = m.approved + m.rejected;
        m.approval_rate = (decided > 0).then(|| m.approved as f64 / decided as f64);
        Ok(m)
    }
}

/// A health snapshot for the backend's self-improvement loop.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Health {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_run_ms: Option<i64>,
    pub grains_since_run: u64,
    pub error_events_since_run: u64,
    pub pending: u64,
    pub applied: u64,
    pub total: u64,
    /// True when the loop looks stalled (never run, or ≥7d / ≥100 new grains
    /// since the last run) — a nudge that a trigger may be unwired.
    pub stale: bool,
}

/// Approval-rate metric for `origin = llm` recommendations (reflection §6b).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct LlmMetrics {
    /// Total llm-origin recommendations ever stored (those that survived the
    /// verifier and reached the queue).
    pub proposed: u64,
    pub pending: u64,
    /// Approved + Applied + RolledBack (a reviewer said yes at least once).
    pub approved: u64,
    pub rejected: u64,
    /// approved / (approved + rejected); `None` until at least one is decided.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub approval_rate: Option<f64>,
}

/// Re-measure applied recommendations at each **checkpoint** past due, via the
/// engine's typed reads — no CAL-scalar round-trip. A recommendation
/// accumulates one `OutcomeResult` per horizon (measured once each), forming a
/// time series, so a late regression (held at 1d, regressed at 30d) is caught.
/// Only *regressed* checkpoints feed the outcome analyzer (→ a revert).
/// Unknown metric kinds are skipped, never faked.
fn measure_outcomes<S: OmsSubstrate>(
    sub: &S,
    p: &mut LoopPersisted,
    policy: &crate::policy::Policy,
    now_ms: i64,
) -> Result<Vec<OutcomeInput>> {
    // Collect all due (recommendation, checkpoint) pairs first. A checkpoint
    // is due in its own unit: elapsed time, evalset runs journaled since the
    // apply, or grains written since it (`Checkpoint`).
    let mut due: Vec<(String, crate::config::AppliedRecord, Checkpoint)> = Vec::new();
    for (h, a) in &p.applied {
        if p.status_index.get(h) != Some(&RecStatus::Applied) {
            continue;
        }
        let Some(metric) = &a.metric else { continue };
        let done = p.measured.get(h).cloned().unwrap_or_default();
        for cp in metric.schedule() {
            if !done.contains(&cp) && checkpoint_due(sub, metric, a.applied_at_ms, cp, now_ms)? {
                due.push((h.clone(), a.clone(), cp));
            }
        }
    }

    let mut out = Vec::new();
    for (rec_hash, applied, checkpoint) in due {
        let metric = applied.metric.as_ref().unwrap();
        let Some(current) = measure_metric(sub, metric, applied.applied_at_ms)? else {
            continue; // metric kind not yet re-measurable
        };
        let bound = cost_bound_for(policy, metric);
        let base = baseline_at_apply(
            sub,
            metric,
            applied.applied_at_ms,
            baseline_kind_for(policy, metric),
            bound.map(|b| b.field.as_str()),
        )?;
        let baseline = base.value;
        // The floor is resolved ONCE here and travels with the input, so the
        // recorded verdict and the revert draft cannot disagree on it.
        let tolerance = tolerance_for(policy, metric, &base);
        let regressed = crate::recommendation::is_regression(
            baseline,
            current,
            metric.higher_is_better,
            tolerance,
        );
        // The run `current` came from, and the cost bound's reading on it.
        // The quality verdict never depends on the cost: `regressed`
        // dominates, a breached bound under a held score is `held_costlier`,
        // and a cost that is not measurable leaves the verdict alone.
        let (current_run_id, cost) = current_run_and_cost(sub, metric, applied.applied_at_ms, bound, &base)?;
        let costlier = cost.as_ref().is_some_and(|c| c.breached());
        let verdict = if regressed {
            "regressed"
        } else if costlier {
            "held_costlier"
        } else {
            "held"
        };
        p.outcomes.entry(rec_hash.clone()).or_default().push(
            crate::recommendation::OutcomeResult {
                rec_hash: rec_hash.clone(),
                metric: metric.metric.clone(),
                baseline,
                current,
                verdict: verdict.into(),
                baseline_kind: base.kind.into(),
                baseline_run_id: base.run_id.clone(),
                best_before: base.best_before,
                tolerance,
                current_run_id: current_run_id.clone(),
                cost: cost.clone(),
                // A time checkpoint speaks through `horizon_ms`, as it always
                // did; the other units carry themselves.
                horizon_ms: checkpoint.as_ms().unwrap_or(0),
                checkpoint: checkpoint.as_ms().is_none().then_some(checkpoint),
                measured_at_ms: now_ms,
            },
        );
        p.measured.entry(rec_hash.clone()).or_default().push(checkpoint);
        if regressed || costlier {
            out.push(OutcomeInput {
                rec_hash,
                target_ref: applied.target_ref.clone(),
                metric: metric.metric.clone(),
                baseline,
                current,
                unit: metric.unit.clone(),
                higher_is_better: metric.higher_is_better,
                baseline_kind: base.kind.into(),
                baseline_run_id: base.run_id,
                best_before: base.best_before,
                tolerance,
                current_run_id,
                cost,
            });
        }
    }
    Ok(out)
}

/// The policy's cost bound, for the evalset the policy names.
fn cost_bound_for<'p>(
    policy: &'p crate::policy::Policy,
    metric: &crate::recommendation::MetricSnapshot,
) -> Option<&'p crate::policy::CostBound> {
    match (policy.outcome_evalset.as_ref(), crate::eval::parse_evalset_metric(&metric.metric)) {
        (Some(e), Some((hash, _))) if e.hash == hash => e.cost.as_ref(),
        _ => None,
    }
}

/// For an evalset metric: the run the current value was read from (the
/// same lookup `measure_metric` made — one reader, one run), and the cost
/// bound's reading of it against the baseline run. `not_measurable` when
/// either side lacks the field or carries it malformed; the figures that
/// did measure still ride on the record.
fn current_run_and_cost<S: SubstrateRead>(
    sub: &S,
    metric: &crate::recommendation::MetricSnapshot,
    applied_at_ms: i64,
    bound: Option<&crate::policy::CostBound>,
    base: &BaselineRead,
) -> Result<(Option<String>, Option<crate::recommendation::CostRead>)> {
    let Some((evalset, _)) = crate::eval::parse_evalset_metric(&metric.metric) else {
        return Ok((None, None));
    };
    let Some(run) = crate::eval::newest_eval_run(sub, evalset, Some(applied_at_ms))? else {
        return Ok((None, None));
    };
    let cost = bound.map(|b| {
        let current = crate::eval::run_value(&run, &b.field);
        let status = match (base.cost, current) {
            (Some(bl), Some(cur)) if cur > bl * b.max_increase_ratio + 1e-9 => "breached",
            (Some(_), Some(_)) => "within",
            _ => "not_measurable",
        };
        crate::recommendation::CostRead {
            field: b.field.clone(),
            max_increase_ratio: b.max_increase_ratio,
            baseline: base.cost,
            current,
            status: status.into(),
        }
    });
    Ok((Some(run.run_id), cost))
}

/// The policy's minimum effect size in the metric's unit, for the evalset
/// the policy names; zero for everything else. The `points` form scales a
/// count field by the baseline run's total — or, with no run before the
/// apply, by the total the proposal's snapshot recorded.
fn tolerance_for(
    policy: &crate::policy::Policy,
    metric: &crate::recommendation::MetricSnapshot,
    base: &BaselineRead,
) -> f64 {
    match (policy.outcome_evalset.as_ref(), crate::eval::parse_evalset_metric(&metric.metric)) {
        (Some(e), Some((hash, field))) if e.hash == hash => e
            .min_effect
            .map(|m| m.resolve(field, base.total.unwrap_or(metric.n)))
            .unwrap_or(0.0),
        _ => 0.0,
    }
}

/// Has this checkpoint come due for a recommendation applied at `applied_at_ms`?
///
/// Time is a subtraction. Runs are counted from the evalset the metric names
/// — journaled strictly after the apply, the same set the measurement itself
/// reads, so "due" and "measurable" can never disagree; on a metric that is
/// not evalset-backed a run checkpoint never fires, honestly, rather than
/// guessing what a run would be. Grains are the loop's own activity count
/// since the apply.
fn checkpoint_due<S: SubstrateRead>(
    sub: &S,
    metric: &crate::recommendation::MetricSnapshot,
    applied_at_ms: i64,
    cp: Checkpoint,
    now_ms: i64,
) -> Result<bool> {
    Ok(match cp {
        Checkpoint::AfterMs(ms) => now_ms - applied_at_ms >= ms,
        Checkpoint::AfterRuns(n) => match crate::eval::parse_evalset_metric(&metric.metric) {
            Some((evalset, _)) => {
                crate::eval::eval_runs(sub, evalset, Some(applied_at_ms + 1))?.len() >= n as usize
            }
            None => false,
        },
        Checkpoint::AfterGrains(n) => count_new(sub, Some(applied_at_ms))?.grains >= n as u64,
    })
}

/// The number a verdict compares against, and where it came from.
pub(crate) struct BaselineRead {
    pub value: f64,
    /// `newest_before_apply`, `high_water` or `snapshot` — see
    /// `OutcomeResult::baseline_kind`.
    pub kind: &'static str,
    pub run_id: Option<String>,
    pub best_before: Option<f64>,
    /// Cases the baseline run graded, when it was a run.
    pub total: Option<u64>,
    /// The policy's cost field on the baseline run, when both exist and it
    /// is measurable there.
    pub cost: Option<f64>,
}

/// The host's baseline choice applies to the evalset it names; any other
/// evalset-backed metric keeps the marginal comparison.
fn baseline_kind_for(
    policy: &crate::policy::Policy,
    metric: &crate::recommendation::MetricSnapshot,
) -> crate::policy::BaselineKind {
    match (policy.outcome_evalset.as_ref(), crate::eval::parse_evalset_metric(&metric.metric)) {
        (Some(e), Some((hash, _))) if e.hash == hash => e.baseline,
        _ => crate::policy::BaselineKind::default(),
    }
}

/// The number a verdict compares against: for an evalset metric, a run
/// journaled BEFORE the apply when there is one — the newest by default, the
/// best under `high_water` — else the snapshot the proposal froze. "Did
/// applying it help" is a question about the state of the world at the
/// apply, not at the proposal — and a deployment that measures once, at day
/// one, and then approves its twentieth rule would otherwise read a rule
/// that cost twenty points as `held` against a baseline the first nineteen
/// had already left far behind. Found on a real corpus
/// (`crates/areev-bench/CURVE.md`: a rule that contradicted an earlier one
/// took the agent from 86% to 66% and measured as held against 26%). With
/// nothing journaled between proposal and apply the two are the same run,
/// so no verdict recorded before this changes.
///
/// `best_before` is read regardless of the choice: the marginal verdict is
/// blind to a fall from the peak by construction (`ADBUY.md`, seed 3: 35 →
/// 238 → 128 → 133 reads `held` against 35), so the receipt carries the
/// peak beside the baseline even when it is not the baseline.
pub(crate) fn baseline_at_apply<S: SubstrateRead>(
    sub: &S,
    metric: &crate::recommendation::MetricSnapshot,
    applied_at_ms: i64,
    kind: crate::policy::BaselineKind,
    cost_field: Option<&str>,
) -> Result<BaselineRead> {
    use crate::policy::BaselineKind;
    if let Some((evalset, field)) = crate::eval::parse_evalset_metric(&metric.metric) {
        // Oldest first, the field read through the one reader every consumer
        // uses; a run whose summary lacks the field is not a candidate.
        let before: Vec<(crate::eval::EvalRun, f64)> = crate::eval::eval_runs(sub, evalset, None)?
            .into_iter()
            .filter(|r| r.recorded_ms < applied_at_ms)
            .filter_map(|r| crate::eval::run_value(&r, field).map(|v| (r, v)))
            .collect();
        if let Some(newest) = before.last() {
            // The first run to attain the best value is the high-water mark:
            // a later tie did not raise it.
            let best = before
                .iter()
                .fold(None::<&(crate::eval::EvalRun, f64)>, |acc, r| match acc {
                    None => Some(r),
                    Some(b) => {
                        let better = if metric.higher_is_better { r.1 > b.1 } else { r.1 < b.1 };
                        Some(if better { r } else { b })
                    }
                })
                .expect("non-empty");
            let pick = match kind {
                BaselineKind::NewestBeforeApply => newest,
                BaselineKind::HighWater => best,
            };
            return Ok(BaselineRead {
                value: pick.1,
                kind: kind.as_str(),
                run_id: Some(pick.0.run_id.clone()),
                best_before: Some(best.1),
                total: Some(pick.0.total()),
                cost: cost_field.and_then(|f| crate::eval::run_value(&pick.0, f)),
            });
        }
    }
    Ok(BaselineRead { value: metric.baseline, kind: "snapshot", run_id: None, best_before: None, total: None, cost: None })
}

/// Typed re-measurement for the fixed set of metric kinds the engine knows.
pub(crate) fn measure_metric<S: SubstrateRead>(
    sub: &S,
    metric: &crate::recommendation::MetricSnapshot,
    since_ms: i64,
) -> Result<Option<f64>> {
    match metric.metric.as_str() {
        // How many times did this tool fail again *with the same signature*
        // after the lesson was applied? Scoped to the signature (metric.relation)
        // so an unrelated later failure of the same tool is not read as a
        // regression of this specific lesson.
        "tool_error_recurrence" => {
            let Some(tool) = &metric.subject else { return Ok(None) };
            let tools = sub.grains_of_type(
                crate::model::grain_type::TOOL,
                None,
                ReadOpts { live_only: true, since_ms: Some(since_ms) },
            )?;
            let n = tools
                .iter()
                .filter(|t| t.tool_name() == Some(tool.as_str()) && t.is_error())
                .filter(|t| {
                    // No stored signature (legacy metric) → fall back to the
                    // whole-tool count so old recommendations still measure.
                    metric.relation.as_deref().is_none_or(|sig| {
                        crate::analyzers::tool_failure::normalize_signature(
                            t.tool_content().unwrap_or(""),
                        ) == sig
                    })
                })
                .count();
            Ok(Some(n as f64))
        }
        // After a resolve-to-latest, does the subject again hold more than one
        // live value under the functional relation? Live-state read (no since
        // filter): the excess beyond one distinct object is the regression.
        "contradiction_recurrence" => {
            let (Some(subject), Some(relation)) = (&metric.subject, &metric.relation) else {
                return Ok(None);
            };
            let facts = scoped_live_facts(sub, metric.namespace.as_deref(), subject)?;
            let distinct: BTreeSet<String> = facts
                .iter()
                .filter(|f| {
                    f.fact_relation()
                        .is_some_and(|r| normalize_ident(r) == *relation)
                })
                .filter_map(|f| f.fact_object().map(normalize_ident))
                .collect();
            Ok(Some(distinct.len().saturating_sub(1) as f64))
        }
        // `evalset:<hash>:<field>` — external correctness, measured the only
        // way that keeps the honesty rule ("Areev Loop improves the agent's
        // memory, not its outputs") intact: an evalset run is an INTERNAL,
        // BOUNDED, ATTRIBUTABLE measurement. The engine never runs the evalset;
        // it only reads summaries a host journaled with `areev eval run`.
        //
        // `since_ms` is the apply time, and it is load-bearing rather than an
        // optimization: a run journaled BEFORE the apply cannot be evidence of
        // what applying did. Comparing the baseline run against itself would
        // report "held" forever — a fabricated receipt, which is worse than no
        // receipt at all. No run since the apply → `None` → not yet
        // measurable, and the checkpoint stays due.
        m if m.starts_with("evalset:") => {
            let Some((evalset, field)) = crate::eval::parse_evalset_metric(m) else {
                return Ok(None);
            };
            let Some(run) = crate::eval::newest_eval_run(sub, evalset, Some(since_ms))? else {
                return Ok(None);
            };
            Ok(crate::eval::run_value(&run, field))
        }
        _ => Ok(None),
    }
}

/// Live facts for one normalized (namespace?, subject) — the shared scope of
/// the fact-shaped recurrence metrics. `namespace: None` spans all namespaces.
fn scoped_live_facts<S: SubstrateRead>(
    sub: &S,
    namespace: Option<&str>,
    subject: &str,
) -> Result<Vec<GrainRecord>> {
    let facts = sub.grains_of_type(
        crate::model::grain_type::FACT,
        None,
        ReadOpts { live_only: true, since_ms: None },
    )?;
    Ok(facts
        .into_iter()
        .filter(|f| namespace.is_none_or(|ns| normalize_ident(&f.namespace) == ns))
        .filter(|f| {
            f.fact_subject()
                .is_some_and(|s| normalize_ident(s) == subject)
        })
        .collect())
}

// --- free helpers ---

/// The §6.3 exact-equality check: a SUPERSEDE is *value-identical* when every
/// replacement field equals the superseded grain's value — strings after
/// case-fold/trim (upstream NFC is an OMS invariant), `namespace` against the
/// grain's own namespace, everything else exactly. This is what makes an
/// auto-applied consolidation provably information-preserving; a
/// near-duplicate (an observation body off by one token) fails it and stays
/// pending for human review. Fails closed: an unrecognized line shape, an
/// empty replacement, a missing grain, or a field the original never had all
/// disqualify.
fn supersede_is_value_identical<S: SubstrateRead>(sub: &S, line: &str) -> bool {
    let Some((target, _gtype, fields)) = crate::cal::parse_own_supersede(line) else {
        return false;
    };
    if fields.is_empty() {
        return false;
    }
    let Ok(Some(grain)) = sub.grain(&target) else {
        return false;
    };
    // An expiry lives OUTSIDE `fields`, so the replacement (built from a fixed
    // field set) can never carry it — consolidating away a grain that has a
    // valid_to would silently drop the expiry, invisibly to the field-by-field
    // check below. Fail closed. (A dup that additionally carries an extra
    // *content* field the replacement omits is a real but subtler info-loss;
    // catching it soundly needs a full canonical-vs-extra comparison rather
    // than this replacement-scoped check, since a real fact's fields also carry
    // OMS metadata like `confidence` that consolidation legitimately keeps —
    // left as a follow-up so this narrow fix can't block valid consolidations.)
    if grain.valid_to_ms.is_some() {
        return false;
    }
    // Forward check: every replacement field equals the grain's value.
    fields.iter().all(|(k, v)| {
        if k == "namespace" {
            return v
                .as_str()
                .is_some_and(|s| normalize_ident(s) == normalize_ident(&grain.namespace));
        }
        match (v, grain.fields.get(k)) {
            (Value::String(a), Some(Value::String(b))) => {
                normalize_ident(a) == normalize_ident(b)
            }
            (a, Some(b)) => a == b,
            (_, None) => false,
        }
    })
}

/// The DISCOVER evidence bundle's total size, and the reserved share each
/// source gets inside it.
///
/// Three sources feed the bundle and they answer different questions, so each
/// is budgeted rather than served first-come: the grains the deterministic
/// findings CITED (what clustering already caught), recent tool ERRORS (what
/// clustering could have caught but did not), and recent facts/observations —
/// the LLM's own lens, and the ONLY source that can carry a problem with no
/// error shape at all. `LENS_RESERVE` is what the last of those is guaranteed.
const EVIDENCE_CAP: usize = 64;
const CITED_SEED_CAP: usize = 24;
const TOOL_SEED_CAP: usize = 16;
/// Human-authored Observations get a small guaranteed share, taken before the
/// general top-up. Learning does not only come from what went wrong: a person
/// saying "from now on, do X" is a complete rule stated once, and no amount of
/// counting recovers it from a corpus that never surfaced it.
const NOTE_SEED_CAP: usize = 8;
/// Harness Observations the LLM lens is allowed to see, and how many.
///
/// An all-namespace scan deliberately hides every `agent:` namespace
/// (`areev-loop-adapter`): those hold the file's own grants and Tier-2 audit
/// records, and an analyzer that swept them as ordinary memory once proposed
/// tombstoning the grants — which locks every non-owner out of the file. That
/// exclusion stays exactly as it is.
///
/// But not everything the harness records is governance. A fold summary is the
/// agent's OWN account of what a long run had worked out, written when the
/// transcript outgrew the model's window — experience, in the harness
/// namespace only because it is evidence about a run rather than memory the
/// agent asserts. Invisible to the lens, it may as well not have been written.
///
/// So: an explicit, named list read from an EXPLICIT namespace (which is what
/// distinguishes it from a sweep), with a reserve of its own. Adding a kind
/// here is a one-line decision someone reviews — never a blanket un-hiding of
/// `agent:*`.
const HARNESS_EVIDENCE_KINDS: &[&str] = &["fold_summary"];
const HARNESS_SEED_CAP: usize = 6;
const LENS_RESERVE: usize = 24;

/// The confidence floor (§5.4): a verified draft below this is dropped. The
/// verifier's calibrated confidence is the gate, not the proposer's self-report.
const MIN_LLM_CONFIDENCE: f64 = 0.75;

/// The fixed DISCOVER instruction (§5.1), in two objectives that differ in
/// exactly one paragraph — the scoring rule — so the vocabulary, the cite
/// rule and the JSON contract cannot drift between them. Kept in its own
/// request field so it never interleaves with (attacker-influenced) evidence
/// text. The review-queue rule makes "nothing to report" a first-class,
/// zero-penalty answer — the structural antidote to over-generation; the
/// learner rule makes abstaining over evidence that plainly holds a lesson
/// cost the same as a wrong one. Which applies is host policy
/// (`Policy::discover_objective`), never the model's or the file's choice.
macro_rules! discover_instructions {
    ($scoring:literal) => {
        concat!(
            "You review an agent's memory for quality. \
Given deterministic findings and the evidence they cite, propose ADDITIONAL \
findings the deterministic checks would miss (e.g. a semantic contradiction, a \
stale assumption, a duplicated meaning, a recurring preventable mistake, a \
recurring cost or hand-off the agent's own setup could remove). \
The deterministic findings already cover what the ERROR TEXT says; restating \
one of them earns nothing. The evidence may also contain OUTCOME records — a \
run's observable shape together with whether it was accepted or rejected. A \
problem that raised no error at all is exactly the kind the deterministic \
checks cannot see, so compare the rejected outcomes against the accepted \
ones: a feature they share and the accepted ones lack is a candidate rule. \
Require at least two rejected outcomes before proposing one — a single \
rejection is an anecdote, not a pattern. ",
            $scoring,
            " The 'approved' and 'rejected' lists, when \
present, show findings this reviewer recently accepted or rejected — prefer the \
kind they accept and avoid the kind they reject. Every proposal MUST cite one \
or more evidence items from the bundle by their 'id' (or 'hash'), name a \
'target', and include your confidence 0.0-1.0. Return JSON: \
{\"recommendations\":[{\"summary\":\"...\",\
\"target\":\"...\",\"guidance\":\"...\",\"evidence\":[\"<id>\"],\
\"confidence\":0.0,\"proposal\":{...}}]}. \
OMIT 'proposal' for an advisory finding — one worth a human's attention that \
you are not asking to change anything. Include it ONLY when the evidence \
supports a specific change, choosing exactly one kind: \
(1) {\"kind\":\"lesson\",\"lesson\":\"...\"} with target \
\"entity:<ns>/<subject>\" — ONE short imperative rule (max 240 chars) naming \
an action the agent itself takes on the next occasion. Either ADD an action \
it is failing to take ('Record the vendor name and the amount on every \
invoice, not just the date') or ORDER one that goes wrong ('Refund a \
subscription before cancelling it; refunds on cancelled subscriptions are \
refused'). Name the action, not a check on it: 'validate', 'verify' and \
'ensure ... is correct' describe a review step the agent has no way to \
perform, and such a rule changes nothing even once applied. \
(2) {\"kind\":\"fact\",\"relation\":\"...\",\"object\":\"...\"} with the same \
entity target — a durable fact the agent keeps having to be told (an alias, a \
settled default, a preference). 'relation' is a short identifier (letters, \
digits, _ - . :), not a sentence. \
(3) {\"kind\":\"query_revision\",\"body\":\"<CAL>\"} with target \
\"query:<name>\" or \"template:<name>\" — a rewrite of the saved query that \
assembles the agent's context, when the evidence shows it retrieves the wrong \
things. Give the FULL new body; it replaces the old one. \
(4) {\"kind\":\"plan_revision\",\"edits\":[{\"path\":\"...\",\"from\":X,\"to\":Y}]} \
with target \"grain:<workflow hash>\" — at most 8 field-level edits to the \
workflow. Only these paths are editable: 'edges.<i>.cond', \
'edges.<i>.max_cycles', 'retries.<node>'. 'from' MUST equal what the plan \
holds now, or the edit is refused. You cannot add, remove or rewire nodes. \
(5) {\"kind\":\"code_revision\",\"source\":\"...\"} with target \"tool:<name>\" \
— full replacement source for that tool. It is applied only after a recorded \
evaluation run passes, so propose one only when the evidence shows the current \
code is the defect. \
The subject of a fact, the name of a query, the plan hash and the tool name \
all come from 'target' — do not repeat them inside 'proposal'. A proposal \
becomes a change a human reviewer may apply, so it must be fully supported by \
the cited evidence. Propose nothing you cannot ground in the evidence."
        )
    };
}

/// The review-queue objective (the default).
const DISCOVER_INSTRUCTIONS: &str = discover_instructions!(
    "SCORING: propose a finding ONLY if you \
are more than 0.75 confident it is BOTH correct AND materially useful. A correct, \
useful finding earns 1; a wrong or trivial one is penalized 2; returning nothing \
earns 0. When in doubt, propose nothing — an empty list is the correct answer \
when there is nothing worth flagging."
);

/// The learner objective (`Policy::discover_objective = learner`).
const DISCOVER_LEARNER_INSTRUCTIONS: &str = discover_instructions!(
    "SCORING: you are the learning stage of a deployed agent, and what you \
propose now is what it will do differently next time — a lesson you withhold \
is a mistake it repeats. A correct, actionable proposal earns 1; a wrong or \
trivial one is penalized 1; returning nothing while the evidence holds a \
recurring failure, two or more rejected outcomes, an instruction from a \
person, or a multi-step procedure the agent completed successfully that no \
saved skill or plan covers, is ALSO penalized 1. Abstain only when the \
evidence shows none of those. Prefer the one proposal that addresses the most \
frequent or most costly failure — or, when nothing failed, the procedure that \
worked — over several speculative ones, and report your confidence honestly — \
an independent verifier, not you, decides what survives."
);

/// The fixed GROUND instruction (§5.2): verify the finding's factual PREMISES
/// are real (anti-fabrication), while allowing an inference. A self-improvement
/// finding may reason BEYOND the evidence (e.g. 'HQ=SF and country=Germany are
/// inconsistent'); grounding checks the premises (HQ=SF, country=Germany) are in
/// the evidence — the soundness of the inference is VERIFY's job, not this one.
const GROUND_INSTRUCTIONS: &str = "You are a grounding checker guarding against \
fabrication. A finding may draw an INFERENCE from facts — your job is to confirm \
the facts it relies on are actually present in the cited evidence, NOT that its \
conclusion is stated verbatim. Decompose the finding into the factual claims it \
depends on. Mark supported=true when those facts are present in the evidence \
(even if the finding reasons beyond them). Mark supported=false ONLY if it relies \
on a fact that is NOT in the evidence (a fabrication) or cites evidence about a \
different subject. Return JSON: {\"results\":[{\"id\":0,\"supported\":true,\"reason\":\"...\"}]}.";

/// The fixed VERIFY instruction (§5.3): adversarial, abstention-biased.
const VERIFY_INSTRUCTIONS: &str = "You are an adversarial reviewer stress-testing \
each finding for SOUNDNESS — reject unsound findings, but only for real reasons, \
never invented ones. For each finding, ask: (1) Reality — is there a genuine, \
SPECIFIC problem, or is it vague/speculative? Reject hedged 'potential' or \
'possible' findings with no concrete defect, and reject any claimed \
inconsistency or contradiction that is not backed by at least two actually \
conflicting facts in the cited evidence. (2) Context — does the finding \
correctly read its cited evidence, or misinterpret what the grains say? Keep a \
finding when it names a genuine, specific problem grounded in its evidence and \
materially useful to a human reviewer; otherwise reject it, and default to \
keep=false when uncertain. Do NOT reject a finding for being 'already known', \
redundant, or a 'common type of error' — duplication is handled elsewhere, and a \
grounded cross-fact inconsistency with two conflicting facts is exactly what to \
KEEP. Give a calibrated confidence 0.0-1.0. Return JSON: \
{\"results\":[{\"id\":0,\"keep\":true,\"confidence\":0.0,\"reason\":\"...\"}]}.";

/// The fixed ENRICH instruction.
const ENRICH_INSTRUCTIONS: &str = "For each finding, optionally add a one-sentence \
guidance note to help a human reviewer decide. Do not restate the finding. Return \
JSON: {\"notes\":[{\"target\":\"<target_ref>\",\"guidance\":\"...\"}]}.";

/// Add a grain to the evidence bundle — deduplicated, bounded at 64, text
/// capped. Shared by the deterministic-citation and recent-grain seeding.
/// `ns_by_hash` records each bundled grain's namespace so an authored lesson
/// can later land in the namespace its evidence lives in (never one the
/// model names).
fn push_evidence(
    evidence: &mut Vec<crate::llm::EvidenceItem>,
    bundle: &mut BTreeSet<String>,
    ns_by_hash: &mut std::collections::BTreeMap<String, String>,
    g: &GrainRecord,
    attribution: crate::policy::EvidenceAttribution,
) {
    if evidence.len() < EVIDENCE_CAP && bundle.insert(g.hash.clone()) {
        ns_by_hash.insert(g.hash.clone(), g.namespace.clone());
        evidence.push(crate::llm::EvidenceItem {
            id: format!("e{}", evidence.len() + 1),
            hash: g.hash.clone(),
            grain_type: g.grain_type.clone(),
            text: crate::llm::cap(&grain_brief_with(g, attribution), 400),
        });
    }
}

/// Resolve one citation to a bundled grain's hash: the full hash, the
/// bundle-local `id` the evidence item carried, or an unambiguous hash prefix
/// of at least 12 hex chars. Anything else is a fabrication and resolves to
/// nothing. Small models copy 64-hex hashes badly — measured live, the
/// cite-check was where most of a cheap model's drafts died ("proposed 3 →
/// cited 1") — and a citation that plainly names one bundled grain is not
/// the thing that check exists to catch.
pub(crate) fn resolve_citation(
    cite: &str,
    bundle: &BTreeSet<String>,
    id_to_hash: &std::collections::BTreeMap<&str, &str>,
) -> Option<String> {
    let cite = cite.trim();
    if bundle.contains(cite) {
        return Some(cite.to_string());
    }
    if let Some(h) = id_to_hash.get(cite) {
        return Some((*h).to_string());
    }
    const MIN_PREFIX: usize = 12;
    if cite.len() >= MIN_PREFIX && cite.chars().all(|c| c.is_ascii_hexdigit()) {
        let lower = cite.to_ascii_lowercase();
        let mut it = bundle.iter().filter(|h| h.starts_with(&lower));
        if let (Some(h), None) = (it.next(), it.next()) {
            return Some(h.clone());
        }
    }
    None
}

/// A short human-readable projection of a grain for the evidence bundle,
/// under the host's attribution policy.
fn grain_brief_with(g: &GrainRecord, attribution: crate::policy::EvidenceAttribution) -> String {
    if let (Some(s), Some(r), Some(o)) = (g.fact_subject(), g.fact_relation(), g.fact_object()) {
        return format!("{s} {r} {o}");
    }
    // Tool grains — the evidence most lesson drafts cite. Rendering them
    // empty starved GROUND of the very facts it exists to check: a correct
    // lesson would be refused as unverifiable (found live — the gate
    // rightly rejected a claim over evidence it could not see).
    if let Some(t) = g.tool_name() {
        let status = if g.is_error() { "error" } else { "ok" };
        let out = g.tool_content().unwrap_or("");
        // The call's input, when recorded: without it a successful trajectory
        // reads as a list of tool names and outputs, and a procedure — WHICH
        // ticket was fetched, WHAT tag was set — cannot be reconstructed from
        // it. A skill proposal needs the arguments; a lesson usually does not,
        // and the cap on the brief bounds the cost either way.
        let input = match g.fields.get("input") {
            Some(Value::String(v)) if !v.is_empty() => format!(" input={v}"),
            Some(v @ Value::Object(_)) | Some(v @ Value::Array(_)) => format!(" input={v}"),
            _ => String::new(),
        };
        return format!("tool {t}{input} {status}: {out}");
    }
    // `object` is last but it is not optional: an Observation stores its text
    // there (subject + object, no relation), so it misses the fact-triple
    // branch above and used to fall through this list to an empty string —
    // every human note in a memory reached the model as a blank line. That is
    // the single highest-value evidence a memory holds, and it was the one
    // shape that rendered to nothing. Callers had started duplicating the text
    // into `body` to work around it; nothing should have to.
    for key in ["content", "body", "text", "summary", "object"] {
        if let Some(v) = g.fields.get(key).and_then(|v| v.as_str()) {
            if v.is_empty() {
                continue;
            }
            // Who said it, when the grain records it. An Observation reaches
            // the model as a bare sentence otherwise, and a bare sentence is
            // ambiguous about direction in exactly the way that matters: a
            // person's correction ("Vendor Name is ACME") reads identically
            // to the agent having been told something it asked for. Measured
            // live on the receipts corpus, a model given 31 unattributed
            // corrections concluded the agent was repeatedly *requesting*
            // data it already had, and proposed rules to stop it asking.
            // The observer is already on the grain; only the projection
            // dropped it.
            if attribution == crate::policy::EvidenceAttribution::Anonymous {
                return v.to_string();
            }
            if let Some(who) = g.fields.get("observer_id").and_then(|v| v.as_str()) {
                if !who.is_empty() {
                    let kind = g
                        .fields
                        .get("observer_type")
                        .and_then(|v| v.as_str())
                        .unwrap_or("");
                    let about = g.fields.get("subject").and_then(|v| v.as_str()).unwrap_or("");
                    let mut prefix = if kind == "human" {
                        format!("{who} (a person) said")
                    } else {
                        format!("{who} observed")
                    };
                    if !about.is_empty() {
                        prefix.push_str(&format!(" of {about}"));
                    }
                    return format!("{prefix}: {v}");
                }
            }
            return v.to_string();
        }
    }
    String::new()
}

/// One imperative line, no control characters, capped: the only shape an
/// authored lesson may take. A literal newline could otherwise smuggle a
/// second CAL statement past review (belt: serde_json escapes it anyway) or
/// break the one-line prompt rendering hosts assume.
fn sanitize_lesson(s: &str) -> String {
    sanitize_line(s, crate::llm::MAX_LESSON_LEN)
}

/// One line, no control characters, capped. Every free-text field the model
/// can put into an executable proposal goes through this: a literal newline
/// could otherwise smuggle a second CAL statement past review (belt:
/// serde_json escapes it anyway), split a one-line DEFINE across the batch the
/// apply path iterates, or break the one-line prompt rendering hosts assume.
fn sanitize_line(s: &str, max: usize) -> String {
    let cleaned: String =
        s.chars().map(|c| if c.is_control() { ' ' } else { c }).collect();
    crate::llm::cap(cleaned.trim(), max)
}

/// A relation is an identifier, not prose — it becomes a queryable predicate,
/// and whitespace or quotes in one would make the Fact unfindable by the very
/// recall that should surface it. `None` rejects the draft's `fact` proposal.
fn sanitize_relation(s: &str) -> Option<String> {
    let r = sanitize_line(s, crate::llm::MAX_RELATION_LEN);
    if r.is_empty()
        || !r
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.' | ':'))
    {
        return None;
    }
    Some(r)
}

/// A model-supplied query body is placed INSIDE a `DEFINE … AS { … }` block,
/// so it is the one place in the vocabulary where model text becomes part of a
/// statement's structure rather than its content. Two belts, because the
/// substrate's parser strength is not something this engine gets to assume:
///
/// - **No braces.** Closing the block early is the injection shape; a saved
///   RECALL/ASSEMBLE body needs no braces of its own, so refusing them costs
///   nothing and fails closed.
/// - **No destructive keyword, anywhere in the body.** `cal::contains_destructive`
///   scans each LINE's leading keyword, which a single-line injection slips
///   past by construction — so scan every token here instead.
///
/// The substrate's own `validate_cal` and the saved-query read-only
/// verification pass still run after this; this is the layer that does not
/// depend on either of them being strict.
fn safe_definition_body(body: &str) -> bool {
    if body.contains('{') || body.contains('}') {
        return false;
    }
    !body
        .split(|c: char| !c.is_ascii_alphanumeric() && c != '_')
        .any(|tok| {
            ["FORGET", "PURGE", "DROP", "DEFINE"]
                .iter()
                .any(|kw| tok.eq_ignore_ascii_case(kw))
        })
}

/// The claim GROUND entails and VERIFY stress-tests. When the draft carries a
/// resolvable proposal the claim names exactly what an apply would do, so what
/// survives the gates is what gets written — never a summary standing in for
/// a change the gates never saw.
fn claim_text(d: &crate::llm::LlmDraft, resolved: Option<&ResolvedProposal>) -> String {
    let summary = crate::llm::cap(&d.summary, crate::llm::MAX_SUMMARY_LEN);
    match resolved {
        Some(r) => format!("{summary} {}", r.rendered),
        None => summary,
    }
}

/// A DISCOVER draft that survived structural validation, carrying the
/// executable form of its proposal. Resolution happens BEFORE GROUND/VERIFY,
/// so both gates judge exactly what an apply would do — the rule the authored
/// lesson already followed, generalized to the whole vocabulary. It also means
/// a malformed proposal costs no model call: it dies here, not at apply.
struct ValidatedDraft {
    draft: crate::llm::LlmDraft,
    target_ref: String,
    cited: Vec<String>,
    resolved: Option<ResolvedProposal>,
}

/// The executable shape of a validated draft. `None` on a [`ValidatedDraft`]
/// means advisory — the model said something a human may want to see, but
/// nothing the engine will ever execute.
struct ResolvedProposal {
    action: ActionKind,
    proposal: Proposal,
    /// One line naming exactly what an apply would do; folded into the claim
    /// both gates judge and shown in the review summary.
    rendered: String,
    summary_key: &'static str,
    summary_args: serde_json::Map<String, Value>,
    rollbackable: bool,
    evalset_hash: Option<String>,
    importance: f64,
    /// Set for the two Fact-writing shapes. Held rather than pre-rendered so
    /// the grain can carry the VERIFIER's confidence, which is not known until
    /// after the gates have run.
    fact_fields: Option<serde_json::Map<String, Value>>,
    /// Statements appended after the `fact_fields` ADD when the proposal is
    /// rendered — a consolidation's supersessions of the pile it replaces.
    extra_statements: Vec<String>,
    /// A plan revision's rehearsal report (`SubstrateRead::plan_replay`),
    /// carried onto the recommendation so the reviewer sees it.
    replay: Option<Value>,
}

/// The Workflow fields a `plan_revision` may touch. Thresholds and limits —
/// never who calls what.
///
/// The exclusion is structural, not advisory: `nodes`, `edges[].src`,
/// `edges[].dst` and `bindings` are simply not matchable here, so a topology
/// change cannot be expressed by any proposal the model can write. That keeps
/// a plan revision reviewable as a short list of scalar deltas rather than a
/// re-drawn graph, which is the difference between a reviewer checking a
/// number and a reviewer re-deriving a plan.
fn plan_edit_allowed(path: &str) -> bool {
    let seg: Vec<&str> = path.split('.').collect();
    match seg.as_slice() {
        ["edges", i, "cond"] | ["edges", i, "max_cycles"] => i.parse::<usize>().is_ok(),
        ["retries", node] => !node.is_empty(),
        _ => false,
    }
}

/// Read the value at an allowlisted path (absent → `Value::Null`, which is
/// what an edit adding a `retries` entry must declare as its `from`).
fn plan_get(body: &Value, path: &str) -> Value {
    let mut cur = body;
    for seg in path.split('.') {
        cur = match cur {
            Value::Array(a) => match seg.parse::<usize>().ok().and_then(|i| a.get(i)) {
                Some(v) => v,
                None => return Value::Null,
            },
            Value::Object(o) => match o.get(seg) {
                Some(v) => v,
                None => return Value::Null,
            },
            _ => return Value::Null,
        };
    }
    cur.clone()
}

/// Write the value at an allowlisted path. Only creates a missing key in an
/// object (the `retries.<node>` case — including the `retries` map itself,
/// which a stored plan with no retries omits entirely under omit-defaults
/// serialization, so a first retry edit on such a plan used to fail to
/// resolve and stay advisory); never grows an array.
fn plan_set(body: &mut Value, path: &str, to: Value) -> bool {
    let segs: Vec<&str> = path.split('.').collect();
    let Some((last, parents)) = segs.split_last() else {
        return false;
    };
    let mut cur = body;
    for (depth, seg) in parents.iter().enumerate() {
        cur = match cur {
            Value::Array(a) => match seg.parse::<usize>().ok().and_then(move |i| a.get_mut(i)) {
                Some(v) => v,
                None => return false,
            },
            Value::Object(o) => {
                if depth == 0 && *seg == "retries" && !o.contains_key("retries") {
                    o.insert("retries".into(), Value::Object(serde_json::Map::new()));
                }
                match o.get_mut(*seg) {
                    Some(v) => v,
                    None => return false,
                }
            }
            _ => return false,
        };
    }
    match cur {
        Value::Array(a) => match last.parse::<usize>().ok().and_then(move |i| a.get_mut(i)) {
            Some(slot) => {
                *slot = to;
                true
            }
            None => false,
        },
        Value::Object(o) => {
            o.insert((*last).to_string(), to);
            true
        }
        _ => false,
    }
}

/// Type-check one plan edit's new value against the field it targets. Without
/// this a string in `max_cycles` would be dropped by the grain deserializer
/// and the "applied" revision would silently mean *unlimited* — a proposal
/// that reads as a tightening and lands as a removal.
fn plan_value_ok(path: &str, to: &Value) -> bool {
    let seg: Vec<&str> = path.split('.').collect();
    match seg.as_slice() {
        ["edges", _, "cond"] => to.as_str().is_some_and(|c| {
            !c.trim().is_empty() && c.len() <= 200 && !c.chars().any(char::is_control)
        }),
        ["edges", _, "max_cycles"] | ["retries", _] => {
            to.as_u64().is_some_and(|n| n <= 1_000)
        }
        _ => false,
    }
}

/// Resolve a DISCOVER draft's proposal into the executable form an apply would
/// run, or `None` for advisory. Every variant takes its SCOPE from the draft's
/// target and its supporting facts from the substrate — the model names the
/// change, never the subject it lands on, the namespace it lands in, or (for
/// code) the evalset that grades it.
fn resolve_proposal<S: OmsSubstrate>(
    sub: &S,
    d: &crate::llm::LlmDraft,
    target: &TargetRef,
    cited: &[String],
    ns_by_hash: &std::collections::BTreeMap<String, String>,
    caps: Capabilities,
    policy: &crate::policy::Policy,
) -> Option<ResolvedProposal> {
    use crate::llm::DraftProposal as P;
    let (skills, plans) = (&policy.skills, &policy.plans);
    let mut args = serde_json::Map::new();
    match d.parsed_proposal()? {
        // ---- plan: a procedure as a validated Workflow + its Skill prose ----
        P::Plan { description, when_to_use, nodes, edges } => {
            if !plans.enabled || !caps.plans {
                return None;
            }
            let PlanFields { skill, workflow, name, n_nodes, n_edges, existing_skill, existing_plan } =
                derived_plan_fields(sub, target, &description, &when_to_use, &nodes, &edges, cited, ns_by_hash, plans)?;
            args.insert("name".into(), Value::from(name.clone()));
            args.insert("nodes".into(), Value::from(n_nodes as u64));
            let mut stmts = vec![match &existing_skill {
                Some(h) => cal::supersede(h, "skill", &skill),
                None => cal::add("skill", &skill),
            }];
            // A graph the runtime would refuse is not minted as a plan — but
            // the procedure it describes is still the thing worth keeping, so
            // it is recorded as a skill. Measured need: a 30B proposer writes
            // conditions like "tickets.length > 0", outside the frozen v1
            // grammar, and discarding the draft for that threw away the
            // captured procedure entirely (PERSIST.md §11 #27).
            let (summary_key, kind) = match &workflow {
                Some(wf) => {
                    stmts.push(match &existing_plan {
                        Some(h) => cal::supersede(h, "workflow", wf),
                        None => cal::add("workflow", wf),
                    });
                    args.insert("edges".into(), Value::from(n_edges as u64));
                    ("llm.plan", "plan")
                }
                None => {
                    args.insert("steps".into(), Value::from(n_nodes as u64));
                    ("llm.skill", "skill")
                }
            };
            let patched = existing_skill.is_some() || (workflow.is_some() && existing_plan.is_some());
            let (action, verb) = if patched {
                (ActionKind::Revise, "revise")
            } else {
                (ActionKind::Record, "record")
            };
            Some(ResolvedProposal {
                action,
                proposal: Proposal::Cal { cal: cal::batch(&stmts) },
                rendered: format!(
                    "Proposed {kind} to {verb}: \"{name}\" — {n_nodes} steps; when: {}",
                    skill.get("when_to_use").and_then(Value::as_str).unwrap_or("")
                ),
                summary_key,
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.65,
                fact_fields: None,
                extra_statements: Vec::new(),
                replay: None,
            })
        }
        // ---- skill: a reusable procedure from a trajectory that succeeded ----
        P::Skill { description, when_to_use, steps } => {
            if !skills.enabled {
                return None;
            }
            let SkillFields { fields, name, n_steps, existing } =
                derived_skill_fields(sub, target, &description, &when_to_use, &steps, cited, ns_by_hash, skills)?;
            args.insert("name".into(), Value::from(name.clone()));
            args.insert("steps".into(), Value::from(n_steps as u64));
            // A live skill of the same name in the same namespace is PATCHED
            // (superseded), never duplicated beside itself.
            let (action, cal, verb) = match existing {
                Some(hash) => (ActionKind::Revise, cal::supersede(&hash, "skill", &fields), "revise"),
                None => (ActionKind::Record, cal::add("skill", &fields), "record"),
            };
            Some(ResolvedProposal {
                action,
                proposal: Proposal::Cal { cal },
                rendered: format!(
                    "Proposed skill to {verb}: \"{name}\" — {n_steps} steps; when: {}",
                    fields.get("when_to_use").and_then(Value::as_str).unwrap_or("")
                ),
                summary_key: "llm.skill",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.6,
                fact_fields: None,
                extra_statements: Vec::new(),
                replay: None,
            })
        }
        // ---- consolidation: one lesson replacing a pile ----
        P::Consolidation { lesson, supersedes } => {
            let lesson = sanitize_lesson(&lesson);
            if lesson.is_empty() {
                return None;
            }
            let fields = derived_fact_fields(target, "lesson", &lesson, cited, ns_by_hash)?;
            let subject = fields.get("subject").and_then(Value::as_str).unwrap_or("").to_string();
            // Every member must be a LIVE lesson on this very entity, all in
            // one namespace, and there must be a pile — a "consolidation" of
            // one grain, or of grains the model picked from elsewhere, is
            // not a consolidation and stays advisory.
            let mut hashes: Vec<String> = supersedes.into_iter().collect();
            hashes.sort();
            hashes.dedup();
            let mut members = Vec::new();
            for h in &hashes {
                let g = sub.grain(h).ok().flatten()?;
                if !g.is_live()
                    || g.fact_relation() != Some("lesson")
                    || g.fact_subject().is_none_or(|s| normalize_ident(s) != normalize_ident(&subject))
                {
                    return None;
                }
                members.push(g);
            }
            if members.len() < 2 || members.iter().any(|m| m.namespace != members[0].namespace) {
                return None;
            }
            let ns = members[0].namespace.clone();
            let mut fields = fields;
            if !ns.is_empty() {
                fields.insert("namespace".into(), Value::from(ns.clone()));
            }
            fields.insert("consolidates".into(), Value::from(hashes.clone()));
            // Each member is superseded by a marker naming the line that
            // replaced it — not by a copy of the lesson, which would leave N
            // live copies in the prompt. Rollback retracts the markers and
            // the added line; the members come back as heads.
            let extra_statements: Vec<String> = members
                .iter()
                .map(|m| {
                    let mut marker = serde_json::Map::new();
                    marker.insert("subject".into(), Value::from(subject.clone()));
                    marker.insert("relation".into(), Value::from("mg:lesson_consolidated"));
                    marker.insert("object".into(), Value::from(lesson.clone()));
                    if !ns.is_empty() {
                        marker.insert("namespace".into(), Value::from(ns.clone()));
                    }
                    cal::supersede(&m.hash, "fact", &marker)
                })
                .collect();
            args.insert("lesson".into(), Value::from(lesson.clone()));
            args.insert("count".into(), Value::from(members.len() as u64));
            Some(ResolvedProposal {
                action: ActionKind::Consolidate,
                proposal: Proposal::Cal { cal: cal::batch(&extra_statements) },
                rendered: format!(
                    "Proposed consolidation of {} lessons on \"{subject}\" into one: \"{lesson}\"",
                    members.len()
                ),
                summary_key: "llm.consolidation",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.6,
                fact_fields: Some(fields),
                extra_statements,
                replay: None,
            })
        }
        // ---- lesson: the pre-vocabulary shape, unchanged ----
        P::Lesson { lesson } => {
            let lesson = sanitize_lesson(&lesson);
            if lesson.is_empty() {
                return None;
            }
            let fields = derived_fact_fields(target, "lesson", &lesson, cited, ns_by_hash)?;
            args.insert("lesson".into(), Value::from(lesson.clone()));
            Some(ResolvedProposal {
                // Same action as the deterministic lesson path — "record a
                // failure-derived lesson" — so dedup groups authored lessons
                // per target and review UIs need no new vocabulary.
                action: ActionKind::ClusterFailure,
                proposal: Proposal::Cal { cal: cal::add("fact", &fields) },
                rendered: format!("Proposed lesson to record: \"{lesson}\""),
                summary_key: "llm.lesson",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.5,
                fact_fields: Some(fields),
                extra_statements: Vec::new(),
                replay: None,
            })
        }
        // ---- fact: a durable fact under a model-chosen relation ----
        P::Fact { relation, object } => {
            let relation = sanitize_relation(&relation)?;
            let object = sanitize_line(&object, crate::llm::MAX_OBJECT_LEN);
            if object.is_empty() {
                return None;
            }
            let fields = derived_fact_fields(target, &relation, &object, cited, ns_by_hash)?;
            let subject = fields.get("subject").and_then(Value::as_str).unwrap_or("");
            args.insert("relation".into(), Value::from(relation.clone()));
            args.insert("object".into(), Value::from(object.clone()));
            Some(ResolvedProposal {
                action: ActionKind::Record,
                proposal: Proposal::Cal { cal: cal::add("fact", &fields) },
                rendered: format!("Proposed fact to record: {subject} {relation} \"{object}\""),
                summary_key: "llm.fact",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.5,
                fact_fields: Some(fields),
                extra_statements: Vec::new(),
                replay: None,
            })
        }
        // ---- query_revision: how the agent assembles its own context ----
        P::QueryRevision { body } => {
            let name = target.opaque();
            // The name comes from the target, but it still ends up inside a
            // statement — a quote or control character in one would change the
            // statement's shape rather than its content.
            if name.is_empty()
                || name.chars().any(|c| c.is_control() || c == '"' || c == '\\')
            {
                return None;
            }
            let body = sanitize_line(&body, crate::llm::MAX_QUERY_BODY_LEN);
            if body.is_empty() || !safe_definition_body(&body) {
                return None;
            }
            let stmt = match target.scheme() {
                "query" => format!("DEFINE QUERY \"{name}\" AS {{ {body} }}"),
                "template" => format!("DEFINE TEMPLATE {name} AS {{ {body} }}"),
                _ => return None,
            };
            // The substrate owns the grammar: if it will not parse, or will
            // not hand back an inverse, this is not something a reviewer
            // should be offered as applicable. A definition change ROLLBACK
            // could not undo must not be applied at all.
            sub.validate_cal(&stmt).ok()?;
            sub.definition_inverse(&stmt).ok().flatten()?;
            args.insert("name".into(), Value::from(name));
            args.insert("body".into(), Value::from(body.clone()));
            Some(ResolvedProposal {
                action: ActionKind::Revise,
                proposal: Proposal::Cal { cal: stmt },
                rendered: format!("Proposed rewrite of saved {} \"{name}\" to: {body}", target.scheme()),
                summary_key: "llm.query_revision",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.6,
                fact_fields: None,
                extra_statements: Vec::new(),
                replay: None,
            })
        }
        // ---- plan_revision: field-level edits to a Workflow grain ----
        P::PlanRevision { edits } => {
            if !caps.plans
                || target.scheme() != "grain"
                || edits.is_empty()
                || edits.len() > crate::llm::MAX_PLAN_EDITS
            {
                return None;
            }
            let hash = target.opaque();
            let g = sub.grain(hash).ok().flatten()?;
            if g.grain_type != "workflow" || !g.is_live() {
                return None;
            }
            let mut body = Value::Object(g.fields.clone());
            let mut deltas = Vec::new();
            let nodes: std::collections::BTreeSet<String> = body
                .get("nodes")
                .and_then(Value::as_array)
                .map(|a| a.iter().filter_map(Value::as_str).map(str::to_string).collect())
                .unwrap_or_default();
            for e in &edits {
                if !plan_edit_allowed(&e.path) || !plan_value_ok(&e.path, &e.to) {
                    return None;
                }
                // A retry count for a node that does not exist is inert, but
                // applying it still mints a new plan hash — and every trigger
                // pointing at the old one must then be walked forward. A
                // no-op is not worth that.
                if let Some(node) = e.path.strip_prefix("retries.") {
                    if !nodes.contains(node) {
                        return None;
                    }
                }
                // `from` is the staleness check: a proposal authored against
                // an older plan does not silently apply to a newer one.
                if plan_get(&body, &e.path) != e.from {
                    return None;
                }
                // A no-op edit is not a revision; it would apply, mint a new
                // plan hash, and orphan every trigger pointing at the old one
                // for nothing.
                if e.from == e.to {
                    return None;
                }
                if !plan_set(&mut body, &e.path, e.to.clone()) {
                    return None;
                }
                deltas.push(format!("{}: {} -> {}", e.path, e.from, e.to));
            }
            // The substrate owns the plan grammar (unique + reachable nodes,
            // conditions parse, every cycle bounded). An edit that would make
            // the plan unrunnable never reaches a reviewer as applicable.
            sub.validate_plan(&body).ok()?;
            // The rehearsal: the candidate re-driven through the runtime's
            // scheduler over the live plan's journaled runs, every effect
            // answered from the journal (`areev run shadow --plan-file`).
            // The report rides on the card either way; under a
            // `plan_replay` policy it is also the gate — a candidate worse
            // than the incumbent on the same runs, or rehearsable on too
            // few of them, is stored as advisory with the reason naming the
            // runs, never offered to apply.
            let replay = sub.plan_replay(hash, &body).ok().flatten();
            let refused = match (&policy.plan_replay, &replay) {
                (Some(gate), Some(report)) => gate.refusal(report),
                _ => None,
            };
            let Value::Object(fields) = body else {
                return None;
            };
            args.insert("plan".into(), Value::from(hash));
            args.insert("edits".into(), Value::from(deltas.join("; ")));
            if let Some(reason) = refused {
                let mut data = serde_json::Map::new();
                data.insert("plan".into(), Value::from(hash));
                data.insert("edits".into(), Value::from(deltas.clone()));
                data.insert("refused".into(), Value::from(reason.clone()));
                args.insert("reason".into(), Value::from(reason.clone()));
                return Some(ResolvedProposal {
                    action: ActionKind::Flag,
                    proposal: Proposal::Data { data },
                    rendered: format!(
                        "Plan revision ({}) refused by the rehearsal: {reason}",
                        deltas.join("; ")
                    ),
                    summary_key: "llm.plan_revision_refused",
                    summary_args: args,
                    rollbackable: false,
                    evalset_hash: None,
                    importance: 0.4,
                    fact_fields: None,
                    extra_statements: Vec::new(),
                    replay,
                });
            }
            let stmt = cal::supersede(hash, "workflow", &fields);
            // Same rule as the definition rewrite: a statement the substrate
            // will not accept is not something to offer a reviewer as
            // applicable. `validate_plan` checked the GRAPH; this checks the
            // statement that carries it.
            sub.validate_cal(&stmt).ok()?;
            Some(ResolvedProposal {
                action: ActionKind::Revise,
                proposal: Proposal::Cal { cal: stmt },
                rendered: format!("Proposed plan revision ({})", deltas.join("; ")),
                summary_key: "llm.plan_revision",
                summary_args: args,
                rollbackable: true,
                evalset_hash: None,
                importance: 0.7,
                fact_fields: None,
                extra_statements: Vec::new(),
                replay,
            })
        }
        // ---- code_revision: §7.4, gated by the tool's own evalset ----
        P::CodeRevision { source } => {
            if !caps.code || target.scheme() != "tool" || source.trim().is_empty() {
                return None;
            }
            if source.chars().count() > crate::llm::MAX_CODE_LEN {
                return None;
            }
            // Rule E1's pin, resolved from the substrate. A proposer that
            // could name its own grader is not gated, and a tool that
            // declares no evalset has no gate to pass — advisory either way.
            let evalset = sub.tool_evalset(target.opaque()).ok().flatten()?;
            let mut data = serde_json::Map::new();
            data.insert("tool".into(), Value::from(target.opaque()));
            data.insert("source".into(), Value::from(source.clone()));
            args.insert("tool".into(), Value::from(target.opaque()));
            args.insert("bytes".into(), Value::from(source.len() as u64));
            Some(ResolvedProposal {
                action: ActionKind::CodeRevision,
                proposal: Proposal::Data { data },
                rendered: format!(
                    "Proposed new source for tool {} ({} bytes), gated by evalset {}",
                    target.opaque(),
                    source.len(),
                    evalset
                ),
                summary_key: "llm.code_revision",
                summary_args: args,
                rollbackable: true,
                evalset_hash: Some(evalset),
                importance: 0.8,
                fact_fields: None,
                extra_statements: Vec::new(),
                replay: None,
            })
        }
    }
}

/// Stamp a validated DISCOVER draft as an `origin = llm` recommendation.
/// Default shape: an advisory `Flag` carrying `Proposal::Data` (no executable
/// mutation). A draft whose proposal RESOLVED (see [`resolve_proposal`])
/// instead stamps as that executable change, reviewable with the exact line an
/// apply would run. Either way `Origin::Llm` plus the no-manifest analyzer id
/// leave it structurally ineligible for auto-apply — and independently, no
/// class this vocabulary can reach except `memory` is auto-appliable at all
/// (`Policy::grants_auto_apply`). The only path into the agent is a human
/// review with a BECAUSE followed by an explicit apply.
fn stamp_llm(
    model: &str,
    d: &crate::llm::LlmDraft,
    target_ref: String,
    cited: Vec<String>,
    resolved: Option<ResolvedProposal>,
    confidence: f64,
    now_ms: i64,
) -> Recommendation {
    let summary_text = crate::llm::cap(&d.summary, crate::llm::MAX_SUMMARY_LEN);
    let guidance = if d.guidance.trim().is_empty() {
        None
    } else {
        Some(crate::llm::cap(&d.guidance, crate::llm::MAX_GUIDANCE_LEN))
    };
    let replay = resolved.as_ref().and_then(|r| r.replay.clone());
    let (action, proposal, summary, rollbackable, importance, evalset_hash, content) = match resolved {
        Some(mut r) => {
            // What the proposal would DO, before the verifier's confidence
            // is folded into the fact: the dedup key fingerprints this, so
            // the same lesson at a different confidence is one finding.
            let content = match &r.fact_fields {
                Some(fields) => format!(
                    "{} {}",
                    fields.get("relation").and_then(Value::as_str).unwrap_or(""),
                    fields.get("object").and_then(Value::as_str).unwrap_or("")
                ),
                None => match &r.proposal {
                    Proposal::Cal { cal } => cal.clone(),
                    Proposal::Data { data } => Value::Object(data.clone()).to_string(),
                    Proposal::Edit { diff, .. } => diff.clone(),
                },
            };
            // The grain records the VERIFIER's calibrated confidence — the
            // independent signal — never the proposer's self-report.
            if let Some(mut fields) = r.fact_fields.take() {
                fields.insert("confidence".into(), Value::from(confidence.clamp(0.0, 1.0)));
                let mut statements = vec![cal::add("fact", &fields)];
                statements.extend(r.extra_statements.iter().cloned());
                r.proposal = Proposal::Cal { cal: cal::batch(&statements) };
            }
            let mut args = r.summary_args;
            args.insert("text".into(), Value::from(summary_text));
            (
                r.action,
                r.proposal,
                Summary::new(r.summary_key, args),
                r.rollbackable,
                r.importance,
                r.evalset_hash,
                Some(content),
            )
        }
        None => {
            let mut args = serde_json::Map::new();
            args.insert("text".into(), Value::from(summary_text));
            let mut data = serde_json::Map::new();
            data.insert("source".into(), Value::from("llm"));
            (
                ActionKind::Flag,
                Proposal::Data { data },
                Summary::new("llm.discover", args),
                false,
                0.3,
                None,
                None,
            )
        }
    };
    // An advisory flag keeps the analyzer-style key (one open flag per
    // target); an executable proposal keys on its content too, because
    // there the content is the finding.
    let dedup = match &content {
        Some(c) => crate::recommendation::authored_dedup_key("llm", &target_ref, action, c),
        None => dedup_key("llm", &target_ref, action),
    };
    Recommendation {
        hash: String::new(),
        analyzer: "loop.llm/1".to_string(),
        params_snapshot: serde_json::Map::new(),
        origin: Origin::Llm { model: model.to_string() },
        target_ref: target_ref.clone(),
        action_kind: action,
        dedup_key: dedup,
        summary,
        severity: Severity::Low,
        proposal,
        destructive: false,
        rollbackable,
        evidence: cited,
        evidence_query: None,
        metric: None,
        // The verifier's calibrated confidence — not a hardcoded default.
        confidence: confidence.clamp(0.0, 1.0),
        importance,
        created_at_ms: now_ms,
        guidance,
        evalset_hash,
        near_duplicate_of: Vec::new(),
        replay,
        status: RecStatus::Pending,
    }
}

/// The Fact an approved `lesson` or `fact` proposal writes: subject from the
/// entity target, the model's relation (`"lesson"` for the lesson shape —
/// prescriptive prose, distinct from the deterministic `fails_with` signature
/// facts), the sanitized object, and the DOMINANT namespace of the cited
/// evidence (max count, ties to the lexicographically smallest — the
/// tool_failure rule), never a namespace the model names. `None` when the
/// target gives no subject.
/// The `skill` paragraph appended to the DISCOVER instructions when the host
/// allows skill authoring. Kept beside the fixed instruction text it extends.
fn skill_instructions(min_steps: u32) -> String {
    format!(
        " (6) {{\"kind\":\"skill\",\"description\":\"...\",\"when_to_use\":\"...\",\
\"steps\":[\"...\",\"...\"]}} with target \"entity:<ns>/<skill-name>\" — a REUSABLE \
PROCEDURE the agent carried out successfully in the evidence: a sequence of tool \
calls that reached its goal, which a later session facing the same situation \
should not have to rediscover. Give {min_steps} to {} ordered steps, each naming \
the tool called and the values that mattered (the field checked, the tag set, the \
exact format produced), a one-line description, and 'when_to_use' — the situation \
that should trigger it. The skill-name is a short identifier (letters, digits, \
_ -). If a saved skill already covers this procedure, use ITS name so it is \
patched rather than duplicated. Do not propose a skill for a procedure that \
failed, or for one already saved and unchanged. A finding that itself describes \
two or more steps the agent should carry out in order ('after listing the \
tickets, fetch each, then …') IS a procedure: propose it as a skill or a plan, \
never as a lesson — a lesson is one rule, and a procedure written as one is a \
procedure nobody can open.",
        crate::llm::MAX_SKILL_STEPS
    )
}

/// The `plan` paragraph appended to the DISCOVER instructions when the host
/// allows plan authoring. The condition grammar is the runtime's frozen v1
/// grammar, stated so the model writes conditions the plan validator accepts.
fn plan_instructions(min_nodes: u32) -> String {
    format!(
        " (7) {{\"kind\":\"plan\",\"description\":\"...\",\"when_to_use\":\"...\",\
\"nodes\":[{{\"id\":\"list_open\",\"tool\":\"<tool name>\",\"step\":\"...\"}},...],\
\"edges\":[{{\"src\":\"list_open\",\"dst\":\"tag\",\"cond\":\"shared_incident == true\"}},...]}} \
with target \"entity:<ns>/<plan-name>\" — the same reusable procedure as a skill, \
but as a PLAN the runtime can validate and run: {min_nodes} to {} steps, each an \
'id' (letters, digits, _ -), the 'tool' it calls — which MUST be a tool named in \
the cited evidence — and what the step does with it; and 'edges' from step to \
step. An edge 'cond' is optional and uses exactly this grammar: 'path == literal', \
'path != literal', 'path exists' or '!path', where path is dotted names and the \
literal is a JSON string, number, true, false or null — no other operators; state \
a threshold as a flag the step sets ('reporters_ge_3 == true'). A loop back to an \
earlier step needs 'max_cycles'. Prefer a plan over a skill when the procedure \
has branches or a loop; prefer a skill when it is a straight list. If a saved \
plan already covers this procedure, use ITS name so it is patched.",
        crate::llm::MAX_PLAN_NODES
    )
}

/// What a `plan` proposal resolves to: the Skill grain (prose), the Workflow
/// grain (structure), the name both take from the target, the counts, and the
/// live pair of that name (to supersede) if there is one.
struct PlanFields {
    skill: serde_json::Map<String, Value>,
    /// `None` when the graph the model wrote would not run — an edge
    /// condition outside the runtime's frozen grammar, an edge naming no
    /// step. The procedure is still captured, as a skill.
    workflow: Option<serde_json::Map<String, Value>>,
    name: String,
    n_nodes: usize,
    n_edges: usize,
    existing_skill: Option<String>,
    existing_plan: Option<String>,
}

/// The two grains a `plan` proposal would write.
///
/// Grounding is structural: every step's tool must be one the cited evidence
/// shows was called, so the model cannot plan around a tool it invented. The
/// workflow body is handed to the substrate's own plan validator before the
/// draft can be stamped applicable — a plan a reviewer could approve is one
/// the runtime would accept. The pair shares a `name`; the Workflow carries it
/// as a host field (the type is a container by design), and that is how the
/// live plan of a name is found to be patched rather than duplicated.
#[allow(clippy::too_many_arguments)]
fn derived_plan_fields<S: SubstrateRead>(
    sub: &S,
    target: &TargetRef,
    description: &str,
    when_to_use: &str,
    nodes: &[crate::llm::PlanNodeDraft],
    edges: &[crate::llm::PlanEdgeDraft],
    cited: &[String],
    ns_by_hash: &std::collections::BTreeMap<String, String>,
    plans: &crate::policy::PlanAuthoring,
) -> Option<PlanFields> {
    if target.scheme() != "entity" {
        return None;
    }
    let name = sanitize_skill_name(
        target.opaque().rsplit_once('/').map(|(_, n)| n).unwrap_or(target.opaque()),
    )?;
    let description = sanitize_line(description, crate::llm::MAX_OBJECT_LEN);
    let when_to_use = sanitize_line(when_to_use, crate::llm::MAX_OBJECT_LEN);
    if description.is_empty() || when_to_use.is_empty() {
        return None;
    }
    if nodes.len() < plans.min_nodes.max(1) as usize || nodes.len() > crate::llm::MAX_PLAN_NODES {
        return None;
    }
    // The tools the evidence shows were actually called.
    let known_tools: BTreeSet<String> = cited
        .iter()
        .filter_map(|h| sub.grain(h).ok().flatten())
        .filter_map(|g| g.tool_name().map(normalize_ident))
        .collect();
    let mut ids: Vec<String> = Vec::new();
    let mut steps: Vec<String> = Vec::new();
    let mut seen: BTreeSet<String> = BTreeSet::new();
    let mut grounded = 0usize;
    for n in nodes {
        let id = sanitize_skill_name(&n.id)?;
        if !seen.insert(id.clone()) {
            return None; // duplicate step id
        }
        let step = sanitize_line(&n.step, crate::llm::MAX_SKILL_STEP_LEN);
        if step.is_empty() {
            return None;
        }
        // A step whose tool the evidence never shows keeps its instruction and
        // loses the attribution — it is not recorded as calling anything. A
        // real procedure has steps that call nothing (deciding, grouping,
        // comparing), and a model writes them with a placeholder tool;
        // rejecting the whole draft for one of those threw away procedures
        // that were three-quarters grounded (PERSIST.md §11 #27). Nodes carry
        // no bindings here, so an unattributed step executes nothing and
        // claims nothing — but the prose must not tell a later session to
        // call a tool that does not exist.
        let tool = sanitize_line(&n.tool, crate::llm::MAX_SKILL_NAME_LEN);
        if !tool.is_empty() && known_tools.contains(&normalize_ident(&tool)) {
            grounded += 1;
            steps.push(format!("{}. {id} [{tool}]: {step}", steps.len() + 1));
        } else {
            steps.push(format!("{}. {id}: {step}", steps.len() + 1));
        }
        ids.push(id);
    }
    // Anchored in the trajectory: at least one step calls a tool the evidence
    // actually shows, or this is not a procedure the agent carried out.
    if grounded == 0 {
        return None;
    }
    // Edges are built leniently: one the runtime could not run costs the
    // plan, never the procedure. `runnable` goes false and the flow line is
    // still written into the skill's prose, where it is description rather
    // than a promise.
    let mut edge_vals: Vec<Value> = Vec::new();
    let mut flow_lines: Vec<String> = Vec::new();
    let mut runnable = true;
    for e in edges {
        let (Some(src), Some(dst)) = (sanitize_skill_name(&e.src), sanitize_skill_name(&e.dst)) else {
            runnable = false;
            continue;
        };
        if !seen.contains(&src) || !seen.contains(&dst) {
            // An edge naming no step — a model's "end" node, typically.
            flow_lines.push(format!("{} → {}", e.src.trim(), e.dst.trim()));
            runnable = false;
            continue;
        }
        let mut ev = serde_json::Map::new();
        ev.insert("src".into(), Value::from(src.clone()));
        ev.insert("dst".into(), Value::from(dst.clone()));
        let mut label = format!("{src} → {dst}");
        if let Some(c) = e
            .cond
            .as_deref()
            .map(|c| sanitize_line(c, crate::llm::MAX_COND_LEN))
            .filter(|c| !c.is_empty())
        {
            label.push_str(&format!(" if {c}"));
            ev.insert("cond".into(), Value::from(c));
        }
        if let Some(m) = e.max_cycles {
            if m == 0 || m > 100 {
                runnable = false;
            } else {
                label.push_str(&format!(" (at most {m} times)"));
                ev.insert("max_cycles".into(), Value::from(m));
            }
        }
        flow_lines.push(label);
        edge_vals.push(Value::Object(ev));
    }
    if edge_vals.len() > 4 * ids.len() {
        runnable = false;
    }
    // Namespace: where the evidence lives, by majority — a lesson's rule.
    let mut ns_counts: std::collections::BTreeMap<&str, usize> = Default::default();
    for h in cited {
        if let Some(ns) = ns_by_hash.get(h) {
            if !ns.is_empty() {
                *ns_counts.entry(ns.as_str()).or_default() += 1;
            }
        }
    }
    let ns = ns_counts
        .iter()
        .max_by(|a, b| a.1.cmp(b.1).then_with(|| b.0.cmp(a.0)))
        .map(|(ns, _)| ns.to_string());

    // The Workflow: what the runtime validates. Unbound steps are abstract
    // nodes — legal, and what a plan over a host's own tools is.
    let mut workflow = serde_json::Map::new();
    workflow.insert("nodes".into(), Value::from(ids.clone()));
    workflow.insert("edges".into(), Value::Array(edge_vals));
    workflow.insert("name".into(), Value::from(name.clone()));
    if let Some(ns) = &ns {
        workflow.insert("namespace".into(), Value::from(ns.clone()));
    }
    // The substrate owns the grammar: the runtime's own validator decides
    // whether this is a plan (unique and reachable steps, conditions that
    // parse, every cycle bounded). The engine carries no second opinion.
    let workflow = (runnable && sub.validate_plan(&Value::Object(workflow.clone())).is_ok())
        .then_some(workflow);

    // The Skill: the same procedure as prose, with the graph's edges spelled
    // out under the steps so a reader sees the branches the plan encodes.
    let mut instructions = steps.join("\n");
    if !flow_lines.is_empty() {
        instructions.push_str("\n\nFlow:\n");
        instructions.push_str(&flow_lines.iter().map(|l| format!("- {l}")).collect::<Vec<_>>().join("\n"));
    }
    let mut skill = serde_json::Map::new();
    skill.insert("name".into(), Value::from(name.clone()));
    skill.insert("description".into(), Value::from(description));
    skill.insert("when_to_use".into(), Value::from(when_to_use));
    skill.insert("instructions".into(), Value::from(instructions));
    if let Some(ns) = &ns {
        skill.insert("namespace".into(), Value::from(ns.clone()));
    }
    let live = |gt: &str, pick: &dyn Fn(&GrainRecord) -> bool| -> Option<String> {
        sub.grains_of_type(gt, ns.as_deref(), ReadOpts { live_only: true, since_ms: None })
            .ok()?
            .into_iter()
            .find(|g| pick(g))
            .map(|g| g.hash)
    };
    let existing_skill = live(crate::model::grain_type::SKILL, &|g| g.skill_name() == Some(name.as_str()));
    let existing_plan = workflow
        .is_some()
        .then(|| live(crate::model::grain_type::WORKFLOW, &|g| g.str_field("name") == Some(name.as_str())))
        .flatten();
    let n_edges = workflow
        .as_ref()
        .and_then(|w| w.get("edges"))
        .and_then(Value::as_array)
        .map_or(0, |a| a.len());
    Some(PlanFields { skill, workflow, name, n_nodes: ids.len(), n_edges, existing_skill, existing_plan })
}

/// The metric name under which the Verify gate records a premise that moved.
pub const PREMISE_DRIFT_METRIC: &str = "premise_drift";

/// The Verify gate's second question. For every applied recommendation, the
/// grains it cited are looked up again: one that has been retracted, or
/// superseded by a grain holding a DIFFERENT value, is a premise that moved.
/// A value-identical supersession — what consolidation does — is not, and
/// neither is a supersession the recommendation's OWN apply performed: a
/// contradiction resolution cites the two conflicting facts and retires one
/// of them; that is the change it was approved to make, not its premise
/// moving out from under it.
///
/// Compared with the wrong rule (counting the apply's own work as drift), this
/// is what keeps the gate quiet on the analyzers that exist to supersede.
/// Records `drifted` in the outcome series once per distinct count (so a
/// pass does not re-record what the last pass already did) and returns an
/// input `outcome_review` turns into the revert proposal. The reviewer
/// decides; nothing here applies.
fn detect_premise_drift<S: OmsSubstrate>(
    sub: &S,
    p: &mut LoopPersisted,
    now_ms: i64,
) -> Result<Vec<OutcomeInput>> {
    let mut out = Vec::new();
    let applied: Vec<(String, String, Vec<String>)> = p
        .applied
        .iter()
        .filter(|(h, _)| p.status_index.get(*h) == Some(&RecStatus::Applied))
        .map(|(h, a)| (h.clone(), a.target_ref.clone(), a.created_hashes.clone()))
        .collect();
    for (rec_hash, target_ref, own) in applied {
        let Ok(rec) = load_rec(sub, &rec_hash) else { continue };
        if rec.evidence.is_empty() {
            continue;
        }
        let mut moved = 0u64;
        for e in &rec.evidence {
            match sub.grain(e)? {
                None => moved += 1, // retracted or gone
                Some(g) => {
                    let Some(newer) = &g.superseded_by else { continue };
                    if own.iter().any(|c| c == newer) {
                        continue; // the apply's own supersession
                    }
                    match sub.grain(newer)? {
                        // Superseded by something unreadable: a retraction
                        // (the reference substrate marks FORGET this way).
                        None => moved += 1,
                        Some(n) => {
                            if !same_value(&g, &n) {
                                moved += 1;
                            }
                        }
                    }
                }
            }
        }
        if moved == 0 {
            continue;
        }
        let already = p
            .outcomes
            .get(&rec_hash)
            .and_then(|v| v.iter().rev().find(|o| o.metric == PREMISE_DRIFT_METRIC))
            .is_some_and(|o| o.current == moved as f64);
        if !already {
            p.outcomes.entry(rec_hash.clone()).or_default().push(
                crate::recommendation::OutcomeResult {
                    rec_hash: rec_hash.clone(),
                    metric: PREMISE_DRIFT_METRIC.into(),
                    baseline: 0.0,
                    current: moved as f64,
                    verdict: "drifted".into(),
                    baseline_kind: "snapshot".into(),
                    baseline_run_id: None,
                    best_before: None,
                    tolerance: 0.0,
                    current_run_id: None,
                    cost: None,
                    horizon_ms: 0,
                    checkpoint: None,
                    measured_at_ms: now_ms,
                },
            );
        }
        out.push(OutcomeInput {
            rec_hash,
            target_ref,
            metric: PREMISE_DRIFT_METRIC.into(),
            baseline: 0.0,
            current: moved as f64,
            unit: "superseded premises".into(),
            higher_is_better: false,
            baseline_kind: "snapshot".into(),
            baseline_run_id: None,
            best_before: None,
            tolerance: 0.0,
            current_run_id: None,
            cost: None,
        });
    }
    Ok(out)
}

/// Does the superseding grain say the same thing as the one it replaced? A
/// fact compares its object; anything else compares its text body. Two grains
/// that cannot be compared are treated as different — the fail-closed
/// reading, since a premise we cannot confirm still holds is one that moved.
fn same_value(old: &GrainRecord, new: &GrainRecord) -> bool {
    if let (Some(a), Some(b)) = (old.fact_object(), new.fact_object()) {
        return normalize_ident(a) == normalize_ident(b);
    }
    for key in ["content", "tool_content", "body", "text", "object"] {
        if let (Some(a), Some(b)) = (old.str_field(key), new.str_field(key)) {
            return normalize_ident(a) == normalize_ident(b);
        }
    }
    false
}

/// A skill name is an identifier: `[A-Za-z0-9_-]`, bounded, case preserved.
fn sanitize_skill_name(s: &str) -> Option<String> {
    let t = s.trim();
    if t.is_empty()
        || t.chars().count() > crate::llm::MAX_SKILL_NAME_LEN
        || !t.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
    {
        return None;
    }
    Some(t.to_string())
}

/// What a `skill` proposal resolves to: the grain's fields, the name it took
/// from the target, how many steps survived sanitizing, and the live skill of
/// that name in the same namespace (to supersede) if there is one.
struct SkillFields {
    fields: serde_json::Map<String, Value>,
    name: String,
    n_steps: usize,
    existing: Option<String>,
}

/// The fields of the Skill grain a `skill` proposal would write.
///
/// The namespace is the one most of the cited evidence lives in — the same
/// rule a lesson follows — never one the model names.
#[allow(clippy::too_many_arguments)]
fn derived_skill_fields<S: SubstrateRead>(
    sub: &S,
    target: &TargetRef,
    description: &str,
    when_to_use: &str,
    steps: &[String],
    cited: &[String],
    ns_by_hash: &std::collections::BTreeMap<String, String>,
    skills: &crate::policy::SkillAuthoring,
) -> Option<SkillFields> {
    if target.scheme() != "entity" {
        return None;
    }
    let name = sanitize_skill_name(
        target.opaque().rsplit_once('/').map(|(_, n)| n).unwrap_or(target.opaque()),
    )?;
    let description = sanitize_line(description, crate::llm::MAX_OBJECT_LEN);
    let when_to_use = sanitize_line(when_to_use, crate::llm::MAX_OBJECT_LEN);
    let steps: Vec<String> = steps
        .iter()
        .map(|st| sanitize_line(st, crate::llm::MAX_SKILL_STEP_LEN))
        .filter(|st| !st.is_empty())
        .take(crate::llm::MAX_SKILL_STEPS)
        .collect();
    if description.is_empty() || when_to_use.is_empty() || steps.len() < skills.min_steps.max(1) as usize {
        return None;
    }
    // Namespace: where the evidence lives, by majority (ties → lexically
    // first), exactly as a lesson's.
    let mut ns_counts: std::collections::BTreeMap<&str, usize> = Default::default();
    for h in cited {
        if let Some(ns) = ns_by_hash.get(h) {
            if !ns.is_empty() {
                *ns_counts.entry(ns.as_str()).or_default() += 1;
            }
        }
    }
    let ns = ns_counts
        .iter()
        .max_by(|a, b| a.1.cmp(b.1).then_with(|| b.0.cmp(a.0)))
        .map(|(ns, _)| ns.to_string());
    let instructions = steps
        .iter()
        .enumerate()
        .map(|(i, st)| format!("{}. {st}", i + 1))
        .collect::<Vec<_>>()
        .join("\n");
    let mut fields = serde_json::Map::new();
    fields.insert("name".into(), Value::from(name.clone()));
    fields.insert("description".into(), Value::from(description));
    fields.insert("when_to_use".into(), Value::from(when_to_use));
    fields.insert("instructions".into(), Value::from(instructions));
    if let Some(ns) = &ns {
        fields.insert("namespace".into(), Value::from(ns.clone()));
    }
    // Patch, don't duplicate: the live skill of this name in this namespace.
    let existing = sub
        .grains_of_type(
            crate::model::grain_type::SKILL,
            ns.as_deref(),
            ReadOpts { live_only: true, since_ms: None },
        )
        .ok()?
        .into_iter()
        .find(|g| g.skill_name() == Some(name.as_str()))
        .map(|g| g.hash);
    Some(SkillFields { fields, name, n_steps: steps.len(), existing })
}

fn derived_fact_fields(
    target: &TargetRef,
    relation: &str,
    object: &str,
    cited: &[String],
    ns_by_hash: &std::collections::BTreeMap<String, String>,
) -> Option<serde_json::Map<String, Value>> {
    if target.scheme() != "entity" {
        return None;
    }
    let subject = target
        .opaque()
        .rsplit_once('/')
        .map(|(_, s)| s)
        .unwrap_or(target.opaque());
    if subject.is_empty() {
        return None;
    }
    let mut ns_counts: std::collections::BTreeMap<&str, usize> = Default::default();
    for h in cited {
        if let Some(ns) = ns_by_hash.get(h) {
            if !ns.is_empty() {
                *ns_counts.entry(ns.as_str()).or_default() += 1;
            }
        }
    }
    let lesson_ns = ns_counts
        .iter()
        .max_by(|a, b| a.1.cmp(b.1).then_with(|| b.0.cmp(a.0)))
        .map(|(ns, _)| ns.to_string());
    let mut fields = serde_json::Map::new();
    fields.insert("subject".into(), Value::from(subject));
    fields.insert("relation".into(), Value::from(relation));
    fields.insert("object".into(), Value::from(object));
    // `confidence` is stamped by the caller from the VERIFIER's calibrated
    // score, not the proposer's self-report — so it is deliberately absent
    // here, where only the proposer has spoken.
    if let Some(ns) = lesson_ns {
        fields.insert("namespace".into(), Value::from(ns));
    }
    Some(fields)
}

/// The latest Verify-gate verdict for every grain an applied recommendation
/// created — keyed by the CREATED hash, so an analyzer looking at a live
/// lesson can say how it measured (`held`, `regressed`, `drifted`,
/// `held_costlier`) without reaching the engine's state itself.
fn latest_verdicts(p: &LoopPersisted) -> BTreeMap<String, String> {
    let mut out = BTreeMap::new();
    for (rec_hash, applied) in &p.applied {
        let latest = p
            .outcomes
            .get(rec_hash)
            .and_then(|v| v.iter().max_by_key(|o| o.measured_at_ms))
            .map(|o| o.verdict.clone());
        for h in &applied.created_hashes {
            out.insert(h.clone(), latest.clone().unwrap_or_else(|| "unmeasured".into()));
        }
    }
    out
}

/// Cosine similarity at or above which two lesson embeddings are one
/// instruction (the T1 leg).
pub const NEAR_DUPLICATE_COSINE: f64 = 0.90;
/// Token-set Jaccard at or above which two lesson texts are one instruction
/// (the T0 floor — weak, and honest about it).
pub const NEAR_DUPLICATE_JACCARD: f64 = 0.60;
/// How many near-duplicates a recommendation names, best first.
const NEAR_DUPLICATE_CAP: usize = 8;

/// The live lessons on `subject` (in `namespace`, when known) that `text`
/// restates. Cosine over the substrate's embedder when it embeds both sides;
/// otherwise normalized token-set Jaccard. Best first, capped. A read that
/// fails yields nothing — a near-duplicate check must never block a draft.
pub(crate) fn near_duplicates_of<S: SubstrateRead + ?Sized>(
    sub: &S,
    subject: &str,
    namespace: Option<&str>,
    text: &str,
) -> Vec<crate::recommendation::NearDuplicate> {
    use crate::analyzers::duplicate_sweep::{jaccard, tokenize};
    if subject.is_empty() || text.trim().is_empty() {
        return Vec::new();
    }
    let Ok(facts) = sub.grains_of_type(
        crate::model::grain_type::FACT,
        namespace,
        ReadOpts { live_only: true, since_ms: None },
    ) else {
        return Vec::new();
    };
    let mine = sub.embed(text).ok().flatten();
    let my_tokens = tokenize(text);
    let mut out: Vec<crate::recommendation::NearDuplicate> = facts
        .iter()
        .filter(|f| f.fact_relation() == Some("lesson"))
        .filter(|f| f.fact_subject().is_some_and(|s| normalize_ident(s) == normalize_ident(subject)))
        .filter_map(|f| {
            let other = f.fact_object()?;
            let (score, method, floor) = match (&mine, sub.embed(other).ok().flatten()) {
                (Some(a), Some(b)) => (cosine(a, &b), "cosine", NEAR_DUPLICATE_COSINE),
                _ => (jaccard(&my_tokens, &tokenize(other)), "jaccard", NEAR_DUPLICATE_JACCARD),
            };
            (score >= floor).then(|| crate::recommendation::NearDuplicate {
                hash: f.hash.clone(),
                score: (score * 1000.0).round() / 1000.0,
                method: method.into(),
            })
        })
        .collect();
    out.sort_by(|a, b| b.score.partial_cmp(&a.score).unwrap_or(std::cmp::Ordering::Equal).then(a.hash.cmp(&b.hash)));
    out.truncate(NEAR_DUPLICATE_CAP);
    out
}

fn cosine(a: &[f32], b: &[f32]) -> f64 {
    if a.len() != b.len() || a.is_empty() {
        return 0.0;
    }
    let (mut dot, mut na, mut nb) = (0f64, 0f64, 0f64);
    for (x, y) in a.iter().zip(b) {
        dot += *x as f64 * *y as f64;
        na += *x as f64 * *x as f64;
        nb += *y as f64 * *y as f64;
    }
    if na == 0.0 || nb == 0.0 {
        0.0
    } else {
        dot / (na.sqrt() * nb.sqrt())
    }
}

/// The `consolidation` paragraph appended to the DISCOVER instructions. It
/// is answerable only to a `lesson_pile` finding, which lists the hashes the
/// proposal must name — the model cannot pick a pile of its own.
const CONSOLIDATION_INSTRUCTIONS: &str = " (8) {\"kind\":\"consolidation\",\"lesson\":\"...\",\
\"supersedes\":[\"<hash>\",...]} with the same entity target — ONLY in answer to a \
'Lesson pile' finding, which lists the live lessons on one entity that exceed \
its budget. Write ONE short imperative rule (max 240 chars) that says what \
those lessons say together, dropping nothing a lesson that measured 'held' \
required and keeping nothing only a lesson that measured 'regressed' or \
'drifted' added; 'supersedes' MUST be exactly the hashes that finding lists \
(cite them as evidence too). Applying it replaces every listed lesson with \
the one line; the reviewer can restore them all.";

/// The action kinds that apply ONLY through the evalset-run gating edge
/// (§7.4 for tool code; the tuning seam's adapter promotion inherits the
/// same rule). One predicate so the gate, the rollbackable stamp, and the
/// promotion write can never disagree on membership.
fn requires_gating(kind: ActionKind) -> bool {
    matches!(
        kind,
        ActionKind::CodeRevision | ActionKind::AdapterRevision
    )
}

fn stamp(
    m: &AnalyzerManifest,
    params: &crate::manifest::Params,
    d: crate::recommendation::RecDraft,
    now_ms: i64,
) -> Result<Recommendation> {
    let target = TargetRef::parse(&d.target_ref)?;
    // Rule E1 at the door (§7.4): an unpinned code revision, a pinned
    // non-code target, or a code action on a non-tool target never becomes
    // a recommendation at all.
    crate::recommendation::validate_code_rules(
        d.action_kind,
        target.target_class(),
        d.evalset_hash.as_deref(),
    )?;
    // A revert's identity is the recommendation it retracts, not just its
    // target: two regressed lessons on one entity are two reverts.
    let revert_of = match (&d.action_kind, &d.proposal) {
        (ActionKind::Revert, Proposal::Data { data }) => {
            data.get("revert_of").and_then(|v| v.as_str()).map(str::to_string)
        }
        _ => None,
    };
    let dedup = match revert_of.as_deref() {
        Some(h) => crate::recommendation::revert_dedup_key(m.family(), &d.target_ref, h),
        None => dedup_key(m.family(), &d.target_ref, d.action_kind),
    };
    let destructive = match &d.proposal {
        Proposal::Cal { cal } => cal::contains_destructive(cal),
        _ => false,
    };
    let rollbackable = match &d.proposal {
        Proposal::Cal { .. } => !destructive,
        Proposal::Edit { .. } => false,
        // A code or adapter revision applies by WRITING the promotion
        // grain; retracting it is the exact inverse — rollbackable by
        // construction.
        Proposal::Data { .. } => requires_gating(d.action_kind),
    };
    let mut evidence = d.evidence;
    evidence.truncate(MAX_EVIDENCE);
    // Provenance follows the analyzer's trust class: a subprocess
    // (`--analyzer-cmd`) finding is stamped `Command` — structurally
    // auto-apply-ineligible and badged [external] on the recall surface —
    // not `Builtin`.
    let origin = match m.trust_class {
        crate::manifest::TrustClass::Command => Origin::Command { id: m.id.clone() },
        _ => Origin::Builtin,
    };
    Ok(Recommendation {
        hash: String::new(),
        analyzer: m.id.clone(),
        params_snapshot: params.snapshot(),
        origin,
        target_ref: target.as_string(),
        action_kind: d.action_kind,
        dedup_key: dedup,
        summary: d.summary,
        severity: d.severity,
        proposal: d.proposal,
        destructive,
        rollbackable,
        evidence,
        evidence_query: d.evidence_query,
        metric: d.metric,
        confidence: d.confidence,
        importance: d.importance,
        created_at_ms: now_ms,
        guidance: None,
        evalset_hash: d.evalset_hash,
        near_duplicate_of: Vec::new(),
        replay: None,
        status: RecStatus::Pending,
    })
}

fn validate_because(because: &str) -> Result<String> {
    let trimmed = because.trim();
    if trimmed.is_empty() {
        return Err(Error::InvalidProposal(
            "a BECAUSE reason is required".into(),
        ));
    }
    if trimmed.chars().count() > MAX_BECAUSE {
        return Err(Error::InvalidProposal(format!(
            "BECAUSE exceeds {MAX_BECAUSE} chars"
        )));
    }
    Ok(trimmed.to_string())
}

fn missing_capability(m: &AnalyzerManifest, caps: Capabilities) -> Option<&'static str> {
    for req in &m.requires {
        match req {
            Capability::Forks if !caps.forks => return Some("forks"),
            Capability::Telemetry if !caps.telemetry => return Some("telemetry"),
            Capability::Embeddings if !caps.embeddings => return Some("embeddings"),
            _ => {}
        }
    }
    None
}

fn severity_floor_for(p: &LoopPersisted, analyzer_id: &str) -> Option<Severity> {
    p.config.get(analyzer_id).and_then(|c| c.severity_floor)
}

fn gate(
    opts: &RunOptions,
    p: &LoopPersisted,
    new_grains: u64,
    new_errors: u64,
    now_ms: i64,
) -> Option<SkipReason> {
    let any = opts.min_new.is_some() || opts.min_new_errors.is_some() || opts.if_stale_ms.is_some();
    if !any {
        return None;
    }
    let min_new_ok = opts.min_new.is_some_and(|m| new_grains >= m);
    let min_err_ok = opts.min_new_errors.is_some_and(|m| new_errors >= m);
    let stale_ok = opts
        .if_stale_ms
        .is_some_and(|d| p.state.last_run_ms.is_none_or(|last| now_ms - last >= d));
    if min_new_ok || min_err_ok || stale_ok {
        return None;
    }
    // Choose the honest reason: staleness-only gate → not_stale, else min_new.
    if opts.if_stale_ms.is_some() && opts.min_new.is_none() && opts.min_new_errors.is_none() {
        Some(SkipReason::NotStale)
    } else {
        Some(SkipReason::MinNewNotMet)
    }
}

/// What landed since a watermark, in every unit a gate can count.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct NewSince {
    /// Grains of the four evidence-bearing types.
    pub grains: u64,
    /// Tool grains recording a failure (`--min-new-errors`).
    pub error_events: u64,
    /// Event grains — turns, in a chat deployment (`cadence.every_events`).
    pub events: u64,
    /// Distinct `session_id`s among those Events (`cadence.every_sessions`).
    pub sessions: u64,
}

fn count_new<S: SubstrateRead>(sub: &S, watermark: Option<i64>) -> Result<NewSince> {
    let opts = ReadOpts {
        live_only: false,
        since_ms: watermark.map(|w| w + 1),
    };
    let mut n = NewSince::default();
    let mut sessions: BTreeSet<&str> = BTreeSet::new();
    let mut events_held: Vec<GrainRecord> = Vec::new();
    for t in [
        crate::model::grain_type::FACT,
        crate::model::grain_type::EVENT,
        crate::model::grain_type::TOOL,
        crate::model::grain_type::OBSERVATION,
    ] {
        let g = sub.grains_of_type(t, None, opts)?;
        n.grains += g.len() as u64;
        // The error gate (--min-new-errors) watches captured tool failures.
        if t == crate::model::grain_type::TOOL {
            n.error_events += g.iter().filter(|e| e.is_error()).count() as u64;
        }
        if t == crate::model::grain_type::EVENT {
            n.events = g.len() as u64;
            events_held = g;
        }
    }
    for e in &events_held {
        if let Some(sid) = e.str_field("session_id").filter(|s| !s.is_empty()) {
            sessions.insert(sid);
        }
    }
    n.sessions = sessions.len() as u64;
    Ok(n)
}

/// The policy cadence, evaluated: `None` when a pass is due.
///
/// OR over the thresholds the host set — the pass runs when any one is met.
/// The same shape as the per-call gate above, deliberately: the flags and
/// the policy block are one mechanism spelled in two places, and the flags
/// win when both are present.
fn cadence_gate(c: &crate::policy::Cadence, p: &LoopPersisted, new: NewSince, now_ms: i64) -> Option<SkipReason> {
    if !c.is_set() {
        return None;
    }
    let time_ok = c
        .every_ms
        .is_some_and(|d| p.state.last_run_ms.is_none_or(|last| now_ms - last >= d));
    let grains_ok = c.every_grains.is_some_and(|m| new.grains >= m);
    let events_ok = c.every_events.is_some_and(|m| new.events >= m);
    let sessions_ok = c.every_sessions.is_some_and(|m| new.sessions >= m);
    if time_ok || grains_ok || events_ok || sessions_ok {
        None
    } else {
        Some(SkipReason::CadenceNotDue)
    }
}

fn existing_dedup_keys<S: SubstrateRead>(sub: &S, p: &LoopPersisted) -> Result<BTreeSet<String>> {
    let grains = sub.grains_of_type(
        crate::model::grain_type::RECOMMENDATION,
        Some(LOOP_NS),
        ReadOpts {
            live_only: false,
            since_ms: None,
        },
    )?;
    let mut set = BTreeSet::new();
    for g in grains {
        let status = p
            .status_index
            .get(&g.hash)
            .copied()
            .unwrap_or(RecStatus::Pending);
        // Pending/approved (still open) and applied (already handled)
        // recommendations suppress re-proposal of the same finding. Rejected
        // is handled by cooldowns, and so is a rollback the Verify gate
        // caused (`strike_cooldown` at the revert apply); an operator's own
        // rollback and expiry may legitimately re-propose (the situation
        // returned).
        if matches!(
            status,
            RecStatus::Pending | RecStatus::Approved | RecStatus::Applied
        ) {
            if let Some(key) = g.str_field("dedup_key") {
                set.insert(key.to_string());
            }
        }
    }
    Ok(set)
}

/// Put a finding's `dedup_key` on an exponential cooldown: 7d, 14d, 28d, …
/// capped at 90d, so a finding a reviewer keeps rejecting stops re-surfacing
/// on a fixed 7d cadence (it was a flat 7d despite the "doubling" comment).
/// Two events earn a strike: a reviewer's rejection, and a revert the Verify
/// gate proposed on a measured regression — both are a verdict that the
/// finding, as it stands, should not come back on the next pass.
pub(crate) fn strike_cooldown(p: &mut LoopPersisted, dedup_key: String, now_ms: i64) {
    const BASE_MS: i64 = 7 * 86_400_000;
    const CAP_MS: i64 = 90 * 86_400_000;
    let strikes = p.cooldown_strikes.entry(dedup_key.clone()).or_insert(0);
    let interval = BASE_MS.saturating_mul(1_i64 << (*strikes).min(31)).min(CAP_MS);
    *strikes = strikes.saturating_add(1);
    p.cooldowns.insert(dedup_key, now_ms + interval);
}

fn load_rec<S: SubstrateRead>(sub: &S, rec_hash: &str) -> Result<Recommendation> {
    let g = sub
        .grain(rec_hash)?
        .ok_or_else(|| Error::NotFound(rec_hash.into()))?;
    Recommendation::from_fields(rec_hash, &g.fields)
}

/// Is this line a definition rewrite — a statement that changes a saved
/// `qry:`/`tpl:` registry row rather than writing a grain?
///
/// A keyword test, not a parse: the engine deliberately contains a CAL
/// *writer*, never a parser (parsing is the substrate's job). Both spellings
/// are matched case-insensitively, and `DROP` is intentionally absent — the
/// loop may propose defining a query, never removing one.
pub(crate) fn is_definition_statement(line: &str) -> bool {
    let up = line.trim_start().to_ascii_uppercase();
    up.starts_with("DEFINE QUERY") || up.starts_with("DEFINE TEMPLATE")
}

/// The refusal an advisory `Edit` earns. The engine has no executable edit
/// primitive; the change belongs in the host.
const ADVISORY_EDIT: &str = "This recommendation is advisory: the engine cannot execute this edit. \
     Make the change in the host. Dismiss it with REJECT … BECAUSE while it is still pending, or \
     approve it to acknowledge it and let it expire.";

/// The refusal an advisory `Data` finding earns — every `Data` shape except
/// `outcome_review`'s revert, which carries `revert_of`.
/// Shared by [`Engine::preflight_apply`] and the apply gate so a fused
/// approve-and-apply caller is refused BEFORE the approval lands, not after.
const GATING_REQUIRED: &str = "code and adapter revisions apply only with a recorded gating run \
     (evalset hash + run id + stats) — use apply_gated";

const ADVISORY_DATA: &str = "This finding is advisory: the engine cannot execute it. Act on its \
     guidance. Dismiss it with REJECT … BECAUSE while it is still pending, or approve it to \
     acknowledge it and let it expire.";

/// Whether [`Engine::apply`] can execute this proposal at all.
///
/// One source of truth, shared by [`Engine::preflight_apply`] and
/// [`Engine::apply`] so the two can never disagree.
///
/// Deliberately **not** consulted on approve. An advisory finding is a `Flag`:
/// approving it means "yes, this is real", which is the whole workflow for the
/// LLM path and the telemetry analyzers. What must not happen is a caller being
/// walked into an approval and *then* refused — which is exactly what the fused
/// approve-and-apply path in the bindings did, leaving the recommendation in
/// `approved`, whose only exits are `applied` and `expired`. Preflight asks
/// first, so that path now refuses before it commits anything.
pub(crate) fn ensure_executable(action_kind: ActionKind, proposal: &Proposal) -> Result<()> {
    match proposal {
        Proposal::Cal { .. } => Ok(()),
        Proposal::Edit { .. } => Err(Error::InvalidProposal(ADVISORY_EDIT.into())),
        // Two executable Data shapes: `outcome_review`'s `revert_of` (names
        // an earlier applied recommendation to roll back), and a gated
        // revision (code or adapter), which executes by writing its
        // promotion grain — the gating-run requirement itself is checked at
        // apply, not here.
        Proposal::Data { data } => {
            if requires_gating(action_kind)
                || data.get("revert_of").and_then(Value::as_str).is_some()
            {
                Ok(())
            } else {
                Err(Error::InvalidProposal(ADVISORY_DATA.into()))
            }
        }
    }
}

#[cfg(test)]
mod definition_body_tests {
    use super::safe_definition_body;

    #[test]
    fn ordinary_bodies_pass() {
        assert!(safe_definition_body("RECALL facts WHERE relation = \"lesson\" LIMIT 20"));
        assert!(safe_definition_body("ASSEMBLE context FOR \"desk\" BUDGET 2000"));
    }

    #[test]
    fn a_body_cannot_close_its_own_block_or_carry_destruction() {
        // The injection shape: close the DEFINE block, append a statement.
        // Newlines are already collapsed by `sanitize_line`, so the payload
        // arrives as ONE line — which is exactly what a line-leading-keyword
        // destructive scan cannot see.
        assert!(!safe_definition_body("RECALL facts } FORGET abc {"));
        assert!(!safe_definition_body("RECALL facts } PURGE OLDER THAN 1d {"));
        // …and the keyword alone is refused even without the braces.
        assert!(!safe_definition_body("RECALL facts FORGET abc"));
        assert!(!safe_definition_body("recall facts purge older than 1d"));
        assert!(!safe_definition_body("RECALL facts DROP QUERY \"x\""));
        // A nested DEFINE would redefine something the target does not name.
        assert!(!safe_definition_body("RECALL facts DEFINE QUERY \"other\""));
        // Substring matches are not keywords — this must still pass.
        assert!(safe_definition_body("RECALL facts WHERE subject = \"purged_at\""));
    }
}

#[cfg(test)]
mod plan_edit_tests {
    use super::{plan_edit_allowed, plan_get, plan_set, plan_value_ok};
    use serde_json::json;

    fn plan() -> serde_json::Value {
        json!({
            "nodes": ["fetch", "review", "post"],
            "edges": [
                {"src": "fetch", "dst": "review"},
                {"src": "review", "dst": "fetch", "cond": "confidence < 0.9", "max_cycles": 2}
            ],
            "bindings": {"fetch": "sha256:tool1"},
            "retries": {"fetch": 1}
        })
    }

    #[test]
    fn the_allowlist_admits_thresholds_and_refuses_topology() {
        assert!(plan_edit_allowed("edges.1.cond"));
        assert!(plan_edit_allowed("edges.1.max_cycles"));
        assert!(plan_edit_allowed("retries.fetch"));
        // Topology is not expressible — the structural half of the guarantee
        // that a plan revision stays reviewable as scalar deltas.
        for path in [
            "nodes",
            "nodes.0",
            "edges.0.src",
            "edges.0.dst",
            "edges",
            "bindings.fetch",
            "edges.x.cond",
            "",
        ] {
            assert!(!plan_edit_allowed(path), "{path} must not be editable");
        }
    }

    #[test]
    fn values_are_type_checked_against_the_field() {
        // A string in max_cycles would be DROPPED by the grain deserializer,
        // so an "applied" tightening would silently mean unlimited.
        assert!(!plan_value_ok("edges.1.max_cycles", &json!("2")));
        assert!(plan_value_ok("edges.1.max_cycles", &json!(2)));
        assert!(!plan_value_ok("edges.1.max_cycles", &json!(-1)));
        assert!(!plan_value_ok("retries.fetch", &json!(10_000)));
        assert!(plan_value_ok("retries.fetch", &json!(3)));
        assert!(plan_value_ok("edges.1.cond", &json!("confidence < 0.8")));
        assert!(!plan_value_ok("edges.1.cond", &json!("  ")));
        assert!(!plan_value_ok("edges.1.cond", &json!("a\nb")));
        assert!(!plan_value_ok("edges.0.src", &json!("other")));
    }

    #[test]
    fn get_reads_through_arrays_and_objects_and_absence_is_null() {
        let p = plan();
        assert_eq!(plan_get(&p, "edges.1.max_cycles"), json!(2));
        assert_eq!(plan_get(&p, "retries.fetch"), json!(1));
        // An edit that ADDS a retry declares `from: null` — so absence has to
        // read as Null rather than as an error.
        assert_eq!(plan_get(&p, "retries.review"), json!(null));
        assert_eq!(plan_get(&p, "edges.9.cond"), json!(null));
        assert_eq!(plan_get(&p, "edges.0.cond"), json!(null));
    }

    #[test]
    fn set_writes_scalars_and_adds_a_missing_retry_but_never_grows_an_array() {
        let mut p = plan();
        assert!(plan_set(&mut p, "edges.1.max_cycles", json!(5)));
        assert_eq!(plan_get(&p, "edges.1.max_cycles"), json!(5));
        assert!(plan_set(&mut p, "retries.review", json!(2)));
        assert_eq!(plan_get(&p, "retries.review"), json!(2));
        assert!(!plan_set(&mut p, "edges.7.cond", json!("x")));
        assert_eq!(p["edges"].as_array().unwrap().len(), 2, "no array growth");
        // A stored plan with no retries omits the map entirely (omit-defaults
        // serialization); the first retry edit on it must create the map, or
        // the revision can never resolve. Found by the golden E2E on the demo
        // plan, which the reference fixture — it always had a map — hid.
        let mut q = plan();
        q.as_object_mut().unwrap().remove("retries");
        assert!(plan_set(&mut q, "retries.greet", json!(1)));
        assert_eq!(plan_get(&q, "retries.greet"), json!(1));
        // Only `retries` is created; any other missing parent still refuses.
        let mut r = plan();
        r.as_object_mut().unwrap().remove("edges");
        assert!(!plan_set(&mut r, "edges.0.cond", json!("x")));
    }
}

#[cfg(test)]
mod definition_proposal_tests {
    use super::is_definition_statement;

    #[test]
    fn definition_statements_are_recognized_in_both_spellings() {
        assert!(is_definition_statement(r#"DEFINE QUERY "triage" AS { RECALL facts }"#));
        assert!(is_definition_statement("  define template foo AS { x }"));
        assert!(is_definition_statement("DEFINE TEMPLATE bar AS { y }"));
        // Ordinary proposals are untouched.
        assert!(!is_definition_statement("ADD fact {}"));
        assert!(!is_definition_statement("SUPERSEDE abc WITH fact {}"));
        assert!(!is_definition_statement("FORGET abc"));
        // `DROP` is never proposable, so it is deliberately NOT a definition
        // statement here — a proposal containing one still fails validation
        // rather than being handed an inverse.
        assert!(!is_definition_statement(r#"DROP QUERY "triage""#));
    }
}