cyberbrain 0.7.0

//! The composition (SPEC §8.2). This is the only place that knows about all six crates and
//! the only place a seam is tied. Commands, hooks, MCP and `serve` are thin adapters over
//! [`App`]; none of them may reach around it.
//!
//! Startup order, and why (SPEC §8.2):
//!
//! 1. **Config** — everything below is parameterised by it.
//! 2. **`AuditStore`** (`audit.db`) — opens before anything that could produce a record.
//! 3. **`AuditLog`** (policy) over that store, through [`StoreAuditSink`]; the only writer.
//! 4. **`Egress`** from config and log, handed out as `Arc<dyn EgressGate>`.
//! 5. **`Index`** (`cyberbrain.db`), migrating if needed.
//! 6. **Embedder**, lazily: only when a command needs vectors, never on a hot-path hook.
//! 7. **`LlmClient`**, lazily and last; absence is a caveat, not an error.
//!
//! Steps 3 and 4 both happen inside `Policy::new`, which is why it is constructed exactly
//! once here and handed the sink from step 2.

use crate::audit_bridge::{AUDIT_DB_FILE, StoreAuditSink};
use crate::hostload;
use crate::usage;
use crate::writers::{
    FsNoteWriter, IndexWriter, NoopIndexWriter, NoopNoteWriter, NoteWriter, SqliteIndexWriter,
    StoreEraser, lock_index,
};
use cyberbrain_core::Slash;
use cyberbrain_core::blocks::{MAX_BLOCK_TOKENS, OversizedReason, blocks_of};
use cyberbrain_core::config::DEFAULT_STORE_DIR;
use cyberbrain_core::frontmatter;
use cyberbrain_core::links::link_targets;
use cyberbrain_core::store::{DB_FILE, NOTES_DIR, write_atomic};
use cyberbrain_core::{
    Citation, Config, EgressGate, Embedder, Error, Frontmatter, Note, NoteId, NoteKind, PiiState,
    RecallResult, Result, Ring, Store,
};
use cyberbrain_embed::{ArtefactManifest, ModelPaths, StaticEmbedder};
use cyberbrain_index::{
    AuditStore, EmbeddingProfile, Erased, Index, IndexStats, NoteStamp, ProfileChange,
    RecallOptions, content_hash,
};
use cyberbrain_llm::{LlmClient, LlmConfig, Probe};
use cyberbrain_policy::profile::ProfileExt;
use cyberbrain_policy::{
    Actor, AuditAction, AuditFilter, EgressEntry, EraseReason, EraseRequest, ErasureReport,
    ExportFormat, Finding, Identifier, MemoryAuditSink, ModelCard, ModelInventory, ModelRole,
    OperatorChoice, Policy, PolicyConfig, PolicyStatus, RetentionItem, RetentionQueue,
    SubjectAccessReport, SubjectBlock, SubjectSource, WriteVerdict,
};
use serde::Serialize;
use serde_json::json;
use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::Instant;

/// Name of the manifest the embedder is verified against, next to the two model files.
/// Core's config carries no digest fields (see the report), so the manifest lives with the
/// artefact: `<model_dir>/manifest.json` = `{"weights_blake3": ..., "tokenizer_blake3": ...}`.
pub const MANIFEST_FILE: &str = "manifest.json";

// ---------------------------------------------------------------------------------------
// Store discovery

fn is_store(p: &Path) -> bool {
    p.join(NOTES_DIR).is_dir()
}

/// `--store` / `CYBERBRAIN_STORE` (the CLI folds both into `explicit`), else `.cyberbrain`
/// walking up from the working directory. Never creates anything.
pub fn discover_store(explicit: Option<&Path>) -> Result<PathBuf> {
    if let Some(p) = explicit {
        if is_store(p) {
            return Ok(p.to_path_buf());
        }
        return Err(Error::Config(format!(
            "{} is not a cyberbrain store (no notes/ directory inside it); create one with \
             `cyberbrain init --path {}`",
            Slash(p),
            Slash(p)
        )));
    }
    discover_store_from(None)
}

/// [`discover_store`], starting the walk somewhere other than the process's directory.
///
/// A hook is told which project the session is in, in its payload. The harness happens to
/// start the hook in that directory today, so both agree and the distinction is invisible.
/// It is not guaranteed to: a hook that resolves its store from the process's directory
/// would, the day that changes, quietly read and write the wrong project's memory. The
/// session's own statement of where it is wins.
pub fn discover_store_from(start: Option<&Path>) -> Result<PathBuf> {
    let cwd = match start {
        Some(p) => p.to_path_buf(),
        None => std::env::current_dir().map_err(|e| Error::Io {
            path: PathBuf::from("."),
            source: e,
        })?,
    };
    for dir in cwd.ancestors() {
        let candidate = dir.join(DEFAULT_STORE_DIR);
        if is_store(&candidate) {
            return Ok(candidate);
        }
    }
    Err(Error::Config(format!(
        "no {DEFAULT_STORE_DIR} store found in {} or any directory above it; run \
         `cyberbrain init` in the project root, or point at one with --store or \
         CYBERBRAIN_STORE",
        Slash(&cwd)
    )))
}

// ---------------------------------------------------------------------------------------
// Report types. All `Serialize`, so `--json` and the HTTP API print the same numbers.

#[derive(Debug, Clone, Serialize)]
pub struct InitReport {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub store: PathBuf,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub config: PathBuf,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub audit_db: PathBuf,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub index_db: PathBuf,
    pub next_steps: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct SkippedFile {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub reason: String,
}

#[derive(Debug, Clone, Serialize)]
pub struct OversizedBlock {
    pub note: String,
    pub block_idx: u32,
    pub approx_tokens: u32,
    pub reason: &'static str,
}

#[derive(Debug, Clone, Default, Serialize)]
pub struct EmbedderSummary {
    pub loaded: bool,
    pub profile_id: Option<String>,
    pub dim: Option<usize>,
    /// Why no model is loaded. Always present when `loaded` is false.
    pub reason: Option<String>,
}

#[derive(Debug, Clone, Copy, Default)]
pub struct ScanOptions {
    pub full: bool,
    pub dry_run: bool,
}

/// Every count names the side of the boundary it counts (SPEC §14.3).
#[derive(Debug, Clone, Serialize)]
pub struct ScanReport {
    pub dry_run: bool,
    pub full: bool,
    /// Entries the walker listed under `notes/r*/` plus what it declined to list.
    pub files_listed: usize,
    pub indexed_new: usize,
    pub reindexed_changed: usize,
    /// Content unchanged, but blocks had no vectors and a model is now present.
    pub revectorised: usize,
    pub unchanged: usize,
    /// mtime moved, bytes identical: not reindexed.
    pub touched_only: usize,
    /// Notes the index knew whose file is gone; removed from the index and recorded.
    pub dropped_missing_file: Vec<String>,
    pub skipped: Vec<SkippedFile>,
    pub links_written_back: usize,
    pub link_writeback_failed: Vec<String>,
    pub oversized_blocks: Vec<OversizedBlock>,
    pub embedder: EmbedderSummary,
    pub profile_change: Option<ProfileChange>,
    pub cleared: Option<Erased>,
    /// The index after the scan.
    pub index: IndexStats,
    /// Dry run only: the audit actions a real run would have appended.
    pub audit_preview: Vec<String>,
    pub elapsed_ms: u128,
}

#[derive(Debug, Clone, Serialize)]
pub struct NoteView {
    pub front: Frontmatter,
    pub kind: String,
    pub body: String,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    /// Citations of its blocks as the index has them; empty when not indexed.
    pub blocks: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct BlockView {
    pub citation: String,
    pub idx: u32,
    pub text: String,
    pub token_count: u32,
}

#[derive(Debug, Clone, Serialize)]
pub struct Expanded {
    pub citation: String,
    pub block: BlockView,
    pub note: NoteView,
}

#[derive(Debug, Clone, Default)]
pub struct RecallRequest {
    pub n: Option<usize>,
    pub ring: Option<Ring>,
    /// Restrict to notes of this department, team or domain.
    pub bereich: Option<String>,
    /// Judge validity as of this moment instead of now (`recall --stand`).
    pub at: Option<jiff::Timestamp>,
}

#[derive(Debug, Clone)]
pub struct WriteRequest {
    pub ring: Ring,
    pub kind: NoteKind,
    pub name: String,
    pub body: String,
    pub tags: Vec<String>,
    /// Which department, team or domain the note belongs to. Filters recall, never ranks.
    ///
    /// Three states, not two: `None` leaves whatever the note has, `Some(None)` removes it,
    /// `Some(Some(b))` sets it. A plain `Option` cannot tell "not mentioned" from "clear
    /// it", and a caller that means the second gets the first — silently.
    pub bereich: Option<Option<String>>,
    /// How long the note is kept. Three states, like `bereich` and for the same reason:
    /// `None` leaves whatever the note has, `Some(None)` removes it, `Some(Some(p))` sets
    /// it. As a plain `Option` an ordinary edit that did not mention a period dropped the
    /// one the note had — an agreed deletion date, gone because somebody fixed a typo.
    pub retention: Option<Option<String>>,
    /// Names of notes this one replaces. `None` keeps what the note has; `Some` sets it.
    pub supersedes: Option<Vec<String>>,
    /// From when the note holds. Three states, like `bereich`: `None` keeps what the note
    /// has, `Some(None)` removes it, `Some(Some(t))` sets it.
    pub valid_from: Option<Option<jiff::Timestamp>>,
    /// From when the note no longer holds. Three states, like `valid_from`.
    pub invalid_at: Option<Option<jiff::Timestamp>>,
    /// Write despite findings, stamping `flagged`. The CLI's `--force`.
    pub force: bool,
    /// The operator's answer to a hold, when the caller already asked (the UI, §8.1).
    pub choice: Option<OperatorChoice>,
    /// §8.1: refuse to overwrite a note somebody else changed in the meantime.
    pub expected_updated: Option<jiff::Timestamp>,
    /// Set only when the note was written on another machine and is arriving here.
    ///
    /// Everything in it is identity the note already has: its id, when it was created, when
    /// it was last changed, its tags, its retention. Without it a pull invented all five,
    /// and each one cost something — a citation that resolved on one machine and nowhere
    /// else, a retention clock restarted, an agreed deletion date dropped, and an `updated`
    /// of "now" that made the next real change from the other machine look older than a
    /// note this machine had never touched.
    pub arriving: Option<Arriving>,
    pub dry_run: bool,
}

/// The identity a note keeps when it moves between machines.
#[derive(Debug, Clone)]
pub struct Arriving {
    pub id: NoteId,
    pub created: jiff::Timestamp,
    pub updated: jiff::Timestamp,
    pub tags: Vec<String>,
    pub retention: Option<String>,
}

/// What `propose` did, or what it is waiting to be told.
#[derive(Debug, Serialize)]
#[serde(untagged)]
pub enum Proposed {
    Written(ProposeReport),
    /// The PII gate held it. Same shape as a held write, and answered the same way.
    Held {
        rendered: String,
        name: String,
        /// Never serialised: a finding carries the matched text.
        #[serde(skip)]
        findings: Vec<cyberbrain_policy::Finding>,
    },
}

#[derive(Debug, Clone, Serialize)]
pub struct ManifestReport {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub weights_blake3: String,
    pub tokenizer_blake3: String,
}

#[derive(Debug, Clone, Serialize)]
pub struct ProposeReport {
    pub name: String,
    pub ring: Ring,
    pub kind: String,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub proposed_by: String,
    pub bytes: usize,
    pub pii: PiiState,
    pub redacted: usize,
    /// A note of this name already exists, so accepting this would change it rather than
    /// add one. Worth saying at propose time, not first at review time.
    pub changes_existing: bool,
    pub dry_run: bool,
    pub audit_preview: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ProposalSummary {
    pub name: String,
    pub ring: Ring,
    pub kind: String,
    /// `None` when the audit log has no `note.proposed` row for it — a file that appeared
    /// in `proposals/` some other way. It cannot be accepted; `review` says why.
    pub proposed_by: Option<String>,
    pub created: jiff::Timestamp,
    pub changes_existing: bool,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
}

#[derive(Debug, Clone)]
pub struct ReviewRequest {
    pub name: String,
    pub accept: bool,
    /// Required when rejecting.
    pub reason: String,
    /// Who is deciding. Never the proposer.
    pub by: String,
    /// Accept even though the note changed after the proposal was made.
    pub force: bool,
    pub dry_run: bool,
}

#[derive(Debug, Clone, Serialize)]
pub struct ReviewReport {
    pub name: String,
    pub accepted: bool,
    pub by: String,
    pub proposed_by: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    #[serde(
        skip_serializing_if = "Option::is_none",
        serialize_with = "cyberbrain_core::path_serde::slash_opt"
    )]
    pub path: Option<PathBuf>,
    pub blocks: usize,
    pub vectors: usize,
    pub dry_run: bool,
    pub audit_preview: Vec<String>,
}

/// `cyberbrain invalidate`: declare after the fact that a note stopped holding.
#[derive(Debug, Clone)]
pub struct InvalidateRequest {
    pub name: String,
    /// From when it no longer holds. `None` is now.
    pub at: Option<jiff::Timestamp>,
    /// Take a declaration back instead: the note holds again, open-ended.
    pub clear: bool,
    /// The note that replaces it, written into its head as `superseded_by`.
    pub by: Option<String>,
    pub dry_run: bool,
}

#[derive(Debug, Clone, Serialize)]
pub struct InvalidatedNote {
    pub id: NoteId,
    pub name: String,
    pub ring: Ring,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    /// What the head says now; `None` after `--clear`.
    pub invalid_at: Option<jiff::Timestamp>,
    /// What it said before, so a mistaken date can be put back by hand.
    pub previous_invalid_at: Option<jiff::Timestamp>,
    pub superseded_by: Option<String>,
    pub updated: jiff::Timestamp,
    pub blocks: usize,
    pub vectors: usize,
    pub dry_run: bool,
    pub audit_preview: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct WrittenNote {
    pub id: NoteId,
    pub name: String,
    pub ring: Ring,
    pub kind: String,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub bytes: usize,
    /// `true` for a new note, `false` for an update of an existing one.
    pub created: bool,
    pub updated: jiff::Timestamp,
    pub pii: PiiState,
    pub redacted: usize,
    pub blocks: usize,
    pub vectors: usize,
    pub links: usize,
    pub embedder_reason: Option<String>,
    pub dry_run: bool,
    pub audit_preview: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
#[serde(tag = "outcome", rename_all = "kebab-case")]
pub enum WriteOutcome {
    Written(WrittenNote),
    /// SPEC §12.4: the write is held and the operator has four choices. Exit code 3 at
    /// the CLI, 409 over HTTP.
    Held {
        name: String,
        findings: Vec<Finding>,
        rendered: String,
    },
    /// `expected_updated` did not match. 409 over HTTP.
    Conflict {
        name: String,
        current_updated: jiff::Timestamp,
    },
}

#[derive(Debug, Clone, Serialize)]
pub struct DoctorFinding {
    /// `error` | `warning`
    pub severity: &'static str,
    pub check: &'static str,
    pub detail: String,
}

#[derive(Debug, Clone, Serialize)]
pub struct DoctorReport {
    pub clean: bool,
    pub checks_run: Vec<&'static str>,
    pub findings: Vec<DoctorFinding>,
}

#[derive(Debug, Clone, Serialize)]
pub struct AuditSummary {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub rows: usize,
    pub schema_version: u32,
    /// `Ok(rows verified)` or the first break.
    pub chain: std::result::Result<usize, String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct EmbeddingStatus {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub model_dir: PathBuf,
    pub manifest_present: bool,
    pub embedder: EmbedderSummary,
    pub index_profile: Option<EmbeddingProfile>,
    /// `None` when there is nothing to compare (no model loaded or no vectors stored).
    pub matches_index: Option<bool>,
}

#[derive(Debug, Clone, Serialize)]
pub struct InferenceStatus {
    pub endpoint: String,
    pub model: Option<String>,
    pub state: String,
    pub probe: Option<Probe>,
}

#[derive(Debug, Clone, Serialize)]
pub struct StatusReport {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub store: PathBuf,
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub config: PathBuf,
    pub notes_on_disk: usize,
    pub notes_per_ring: [usize; 5],
    pub files_skipped: usize,
    pub resident_tokens: usize,
    pub resident_cap: usize,
    pub index: IndexStats,
    pub index_stale: bool,
    pub audit: AuditSummary,
    pub embedding: EmbeddingStatus,
    pub inference: InferenceStatus,
    pub policy: PolicyStatus,
}

#[derive(Debug, Clone, Serialize)]
pub struct AuditView {
    pub rows: usize,
    pub verified: Option<std::result::Result<usize, String>>,
    pub rendered: String,
}

/// `policy obligations`, one profile's whole catalogue.
#[derive(Debug, Clone, Serialize)]
pub struct ObligationsView {
    pub profile: cyberbrain_core::PolicyProfile,
    /// The law the profile encodes, as the profile names it.
    pub law: String,
    pub obligations: Vec<cyberbrain_policy::profile::Obligation>,
}

#[derive(Debug, Clone, Serialize)]
pub struct RetentionOutcome {
    pub name: String,
    pub item: RetentionItem,
    pub result: std::result::Result<ErasureReport, String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct RetentionReport {
    pub queue: RetentionQueue,
    pub unreadable: Vec<String>,
    pub applied_run: bool,
    pub dry_run: bool,
    pub applied: Vec<RetentionOutcome>,
    pub audit_preview: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ModelCardReport {
    pub cards: Vec<ModelCard>,
    pub absent: Vec<String>,
}

#[derive(Debug, Clone, Serialize)]
pub struct ConsentReport {
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub path: PathBuf,
    pub consent: bool,
    pub model_source: Option<String>,
    pub warnings: Vec<String>,
}

/// One definition found by `find` (SPEC §10). Line numbers are 1-based and
/// `start_line..=end_line` is inclusive; `line` names the symbol and lies inside it.
#[derive(Debug, Clone, Serialize)]
pub struct FindHit {
    /// Relative to `FindReport::root`, forward slashes.
    pub path: String,
    pub start_line: u32,
    pub end_line: u32,
    /// The line that names the symbol; `start_line` may be earlier when doc comments,
    /// attributes or decorators precede it.
    pub line: u32,
    /// `function`, `method`, `class`, `struct`, `enum`, `trait`, `interface`, `type`,
    /// `impl`, `module`, `namespace`, `macro`, `const`, `static`, `variable`, `table`,
    /// `view`, `index`, `trigger`, `schema`, `section`, `key`, `heading`.
    pub kind: &'static str,
    pub language: &'static str,
    /// The symbol as found.
    pub name: String,
    /// The enclosing named thing (impl target, class, TOML table, parent key path).
    pub scope: Option<String>,
    /// `exact`, `case-insensitive`, `contains`.
    pub matched: &'static str,
    /// The defining line, trimmed, at most 160 characters.
    pub snippet: String,
}

/// What `find` declined to read, by reason (SPEC §14.3: every count names the side of
/// the boundary it counts). A directory kept out by an ignore rule counts once as an
/// entry that was not entered; nothing claims to know how many files were inside it.
#[derive(Debug, Clone, Serialize)]
pub struct FindSkipped {
    /// Entries matched by a `.cyberbrainignore` rule, not entered.
    pub ignored_entries: usize,
    /// Entries matched by a `.gitignore` rule, not entered.
    pub gitignored_entries: usize,
    /// Dot-files and dot-directories, not entered.
    pub hidden_entries: usize,
    /// The store directory itself, not entered.
    pub store_entries: usize,
    /// Symbolic links, never followed.
    pub symlinks: usize,
    /// Lockfiles by name, not read.
    pub lockfiles: usize,
    /// Over the size cap, not read.
    pub too_large: usize,
    /// A NUL byte in the first 8 KiB, not parsed.
    pub binary: usize,
    /// No extractor for the extension, not read.
    pub unsupported: usize,
    pub unsupported_by_extension: std::collections::BTreeMap<String, usize>,
    /// Entries the filesystem refused, with the error.
    pub unreadable: Vec<SkippedFile>,
}

#[derive(Debug, Clone, Serialize)]
pub struct FindReport {
    /// The symbol as given.
    pub symbol: String,
    /// The name part after scope splitting (`find` for `App::find`).
    pub name: String,
    /// The scope part, when the symbol carried one and it selected something.
    pub scope: Option<String>,
    /// The tree that was scanned.
    #[serde(serialize_with = "cyberbrain_core::path_serde::slash")]
    pub root: PathBuf,
    /// Best first, at most `limit`.
    pub hits: Vec<FindHit>,
    /// Matches before truncation.
    pub matched_total: usize,
    pub truncated: bool,
    pub limit: usize,
    /// Files whose text reached an extractor.
    pub files_scanned: usize,
    pub bytes_scanned: u64,
    /// Definitions extracted across those files, matched or not.
    pub definitions_indexed: usize,
    pub skipped: FindSkipped,
    /// Root-relative paths of the ignore files honoured, in walk order.
    pub ignore_files: Vec<String>,
    /// What the counts cannot say: no ignore file at the root, a symbol defined in
    /// several files, a scope that matched nothing, a truncated list.
    pub caveats: Vec<String>,
    pub elapsed_ms: u128,
}

impl From<cyberbrain_code::FindResult> for FindReport {
    fn from(r: cyberbrain_code::FindResult) -> Self {
        FindReport {
            symbol: r.symbol,
            name: r.name,
            scope: r.scope,
            root: r.root,
            hits: r
                .hits
                .into_iter()
                .map(|h| FindHit {
                    path: h.def.path,
                    start_line: h.def.start_line,
                    end_line: h.def.end_line,
                    line: h.def.line,
                    kind: h.def.kind.as_str(),
                    language: h.def.language.as_str(),
                    name: h.def.name,
                    scope: h.def.scope,
                    matched: h.matched.as_str(),
                    snippet: h.def.snippet,
                })
                .collect(),
            matched_total: r.matched_total,
            truncated: r.truncated,
            limit: r.limit,
            files_scanned: r.files_scanned,
            bytes_scanned: r.bytes_scanned,
            definitions_indexed: r.definitions_indexed,
            skipped: FindSkipped {
                ignored_entries: r.skipped.ignored_entries,
                gitignored_entries: r.skipped.gitignored_entries,
                hidden_entries: r.skipped.hidden_entries,
                store_entries: r.skipped.excluded_entries,
                symlinks: r.skipped.symlinks,
                lockfiles: r.skipped.lockfiles,
                too_large: r.skipped.too_large,
                binary: r.skipped.binary,
                unsupported: r.skipped.unsupported,
                unsupported_by_extension: r.skipped.unsupported_by_extension,
                unreadable: r
                    .skipped
                    .unreadable
                    .into_iter()
                    .map(|(path, reason)| SkippedFile {
                        path: PathBuf::from(path),
                        reason,
                    })
                    .collect(),
            },
            ignore_files: r.ignore_files,
            caveats: r.caveats,
            elapsed_ms: r.elapsed.as_millis(),
        }
    }
}

// ---------------------------------------------------------------------------------------
// Lazy pieces

#[derive(Clone)]
enum EmbedderState {
    Loaded {
        embedder: Arc<StaticEmbedder>,
        manifest: ArtefactManifest,
        paths: ModelPaths,
    },
    Absent {
        reason: String,
    },
}

/// The model's identity without the model (`App::model_identity`).
enum ModelIdentity {
    Ready(EmbeddingProfile),
    Absent(String),
}

#[derive(Clone)]
enum LlmState {
    Ready(Box<LlmClient>),
    Absent(String),
}

/// The audit sink a set of writers records into. Real runs share the store-backed policy;
/// dry runs get a fresh `Policy` over a `MemoryAuditSink` so the real policy code runs and
/// its rows can be shown without being kept.
enum PolicyRef<'a> {
    Real(&'a Policy),
    // Boxed because the owned `Policy` dwarfs the reference in the other variant, and every
    // value of this enum would otherwise carry that much stack whether or not it is a dry
    // run. It grew past the threshold when the policy config took the hub endpoint.
    Dry(Box<Policy>, Arc<MemoryAuditSink>),
}

impl PolicyRef<'_> {
    fn get(&self) -> &Policy {
        match self {
            PolicyRef::Real(p) => p,
            PolicyRef::Dry(p, _) => p,
        }
    }

    fn preview(&self) -> Vec<String> {
        match self {
            PolicyRef::Real(_) => Vec::new(),
            PolicyRef::Dry(_, sink) => sink.actions(),
        }
    }
}

struct Writers<'a> {
    notes: Box<dyn NoteWriter>,
    index: Box<dyn IndexWriter>,
    policy: PolicyRef<'a>,
}

fn kind_name(k: NoteKind) -> String {
    match serde_json::to_value(k) {
        Ok(serde_json::Value::String(s)) => s,
        _ => format!("{k:?}").to_lowercase(),
    }
}

fn stamp_of(path: &Path) -> Option<NoteStamp> {
    let md = std::fs::metadata(path).ok()?;
    let ns = md
        .modified()
        .ok()?
        .duration_since(std::time::UNIX_EPOCH)
        .ok()?
        .as_nanos();
    Some(NoteStamp {
        mtime_ns: i64::try_from(ns).ok()?,
        size: md.len(),
    })
}

fn oversized_reason(r: OversizedReason) -> &'static str {
    match r {
        OversizedReason::CodeFence => "a fenced code block is never split",
        OversizedReason::UnbreakableRun => "a single run of characters longer than the limit",
    }
}

fn policy_config(cfg: &Config) -> PolicyConfig {
    let mut pc = PolicyConfig::from_core(cfg)
        .with_model_download_consent(cfg.embedding.model_download_consent);
    if let Some(src) = &cfg.embedding.model_source {
        pc = pc.with_model_source(src.clone());
    }
    pc
}

// ---------------------------------------------------------------------------------------
// App

pub struct App {
    root: PathBuf,
    config: Config,
    store: Store,
    audit_sink: Arc<StoreAuditSink>,
    policy: Policy,
    gate: Arc<dyn EgressGate>,
    index: Arc<Mutex<Index>>,
    actor: Actor,
    embedder: OnceLock<EmbedderState>,
    identity: OnceLock<ModelIdentity>,
    llm: Mutex<Option<LlmState>>,
}

/// The loaded model is not freed, it is left to the operating system.
///
/// Measured 2026-09-25: freeing the tokenizer (500,353 vocabulary entries, ~570 MB of small
/// allocations) and the 512 MB matrix took 0.41–0.50 s, a fifth of every `recall` from the
/// CLI, between the last line of output and exit. An `App` is dropped when its process is
/// about to end (CLI) or never (serve, mcp), so the memory goes back either way; the kernel
/// takes it in one step. Everything else in `App` — index, audit sink — still drops
/// normally, so nothing that has to be flushed is skipped.
impl Drop for App {
    fn drop(&mut self) {
        if let Some(state) = self.embedder.take() {
            std::mem::forget(state);
        }
    }
}

/// From this share of its comparable blocks on, `doctor` names a note as a copy of
/// another. Half: below that the notes are two notes that quote each other.
const SHARED_BLOCKS_PERCENT: usize = 50;

/// Ledger task names for the contradiction check. The abandoned one is separate on purpose:
/// a call that was cut off is not a call that cost that much, and the usage page should not
/// average the two together.
const TASK_CONTRADICTION: &str = "contradiction-check";
const TASK_CONTRADICTION_ABANDONED: &str = "contradiction-check-abandoned";

/// How the previous contradiction check on this store ended.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum LastCheck {
    /// No row in the ledger: nothing has been measured here yet.
    Unknown,
    /// It finished, and took this long.
    Completed { ms: u64 },
    /// It was still running when the budget ran out.
    Abandoned,
}

/// What to do about the contradiction check, decided before a token is spent.
#[derive(Debug, PartialEq, Eq)]
enum CheckPlan {
    /// Run it, but hand the hits over if it takes longer than this.
    Run(std::time::Duration),
    /// Do not start it: the last one measured longer than the budget, and the endpoint is
    /// not going to have become fast since.
    SkipMeasuredSlow { last_ms: u64, budget_ms: u64 },
    /// Do not start it: the last one was cut off at the budget, and the same bet on the
    /// same endpoint pays the same nothing.
    SkipAbandoned { budget_ms: u64 },
    /// Budget 0: the operator asked to wait however long the client's own timeout allows.
    RunUnbounded,
}

/// The budget is spent on the *next* call, so the decision rests on how the last one ended.
/// Without that memory a slow endpoint costs the full budget on every recall and returns
/// nothing for it, which is the same stall in smaller instalments — measured: with only
/// completed calls remembered, the second recall paid three seconds again.
fn plan_contradiction_check(budget_ms: u64, last: LastCheck) -> CheckPlan {
    if budget_ms == 0 {
        return CheckPlan::RunUnbounded;
    }
    match last {
        LastCheck::Abandoned => CheckPlan::SkipAbandoned { budget_ms },
        LastCheck::Completed { ms } if ms > budget_ms => CheckPlan::SkipMeasuredSlow {
            last_ms: ms,
            budget_ms,
        },
        _ => CheckPlan::Run(std::time::Duration::from_millis(budget_ms)),
    }
}

/// How long a measurement of the inference endpoint speaks for the next call. A day: long
/// enough that a slow endpoint is not re-probed on every recall, short enough that adding a
/// GPU is noticed by tomorrow without anyone deleting a ledger file.
const CHECK_MEASUREMENT_GOOD_FOR: std::time::Duration = std::time::Duration::from_secs(86_400);

/// Whether a ledger timestamp is old enough to say nothing about now. An unparseable one
/// counts as stale: the fallback is to try, not to stay silent forever on a bad string.
fn stale(at: &str) -> bool {
    let Ok(then) = at.parse::<jiff::Timestamp>() else {
        return true;
    };
    jiff::Timestamp::now().duration_since(then).unsigned_abs() > CHECK_MEASUREMENT_GOOD_FOR
}

/// Milliseconds a reader can hold in their head: `126 s`, `3.0 s`, `450 ms`.
fn secs(ms: u64) -> String {
    if ms < 1000 {
        format!("{ms} ms")
    } else if ms < 10_000 {
        format!("{:.1} s", ms as f64 / 1000.0)
    } else {
        format!("{} s", ms / 1000)
    }
}

impl App {
    /// Steps 1 to 5 of §8.2. Steps 6 and 7 happen on first use.
    pub fn open(store: Option<&Path>, actor: Actor) -> Result<App> {
        Self::open_from(store, None, actor)
    }

    /// [`open`](Self::open), with the store discovery walk starting at `start` when no
    /// explicit store was given. Used by the hooks, which are told where the session is.
    pub fn open_from(store: Option<&Path>, start: Option<&Path>, actor: Actor) -> Result<App> {
        let root = match store {
            Some(_) => discover_store(store)?,
            None => discover_store_from(start)?,
        };
        // 1. Config.
        let config = Config::load(&root)?;
        let store = Store::with_config(&root, &config)?;
        // 2. The record, before anything that could produce one.
        let audit_sink = Arc::new(StoreAuditSink::new(AuditStore::open(
            &root.join(AUDIT_DB_FILE),
        )?));
        // 3 + 4. The only audit writer, and the gate built from it.
        let policy = Policy::new(policy_config(&config), audit_sink.clone(), actor.clone());
        let gate = policy.gate();
        // 5. The cache.
        let index = Index::open(&root.join(DB_FILE))?;
        Ok(App {
            root,
            config,
            store,
            audit_sink,
            policy,
            gate,
            index: Arc::new(Mutex::new(index)),
            actor,
            embedder: OnceLock::new(),
            identity: OnceLock::new(),
            llm: Mutex::new(None),
        })
    }

    /// Create a store. Refuses to touch a directory that already is one.
    pub fn init(path: &Path, actor: &Actor) -> Result<InitReport> {
        if Config::path_in(path).exists() || is_store(path) {
            return Err(Error::Config(format!(
                "{} is already a cyberbrain store; nothing was changed",
                Slash(path)
            )));
        }
        let cap = Config::default().rings.resident_cap_tokens;
        Store::create(path, cap)?;
        let config = Config::write_default(path)?;
        let audit_db = path.join(AUDIT_DB_FILE);
        let sink = Arc::new(StoreAuditSink::new(AuditStore::open(&audit_db)?));
        let index_db = path.join(DB_FILE);
        let _ = Index::open(&index_db)?;
        let cfg = Config::load(path)?;
        let policy = Policy::new(policy_config(&cfg), sink, actor.clone());
        policy.audit().record_raw(
            &actor.to_string(),
            "store.init",
            format!("store:{}", Slash(path)),
            json!({ "profile": cfg.policy.profile }),
        )?;
        Ok(InitReport {
            store: path.to_path_buf(),
            config,
            audit_db,
            index_db,
            next_steps: vec![
                "write a note:   cyberbrain write --ring 2 --kind knowledge --name first-note --body 'what you learned'"
                    .into(),
                "index the tree: cyberbrain scan".into(),
                "search it:      cyberbrain recall 'what you learned'".into(),
                format!(
                    "for semantic search, place model.safetensors, tokenizer.json and {MANIFEST_FILE} under {}",
                    Slash(&cfg.model_dir())
                ),
                "read the compliance profile: cyberbrain policy egress".into(),
            ],
        })
    }

    // The accessors below are the surface hooks, MCP and `serve` build on next; nothing
    // in the CLI dispatch needs them yet, and a bin crate reports that.
    #[allow(dead_code)]
    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Loads the model now rather than on the first embed (`daemon`: the first request
    /// should not be the one that waits).
    #[cfg_attr(not(unix), allow(dead_code))] // the daemon is Unix only
    pub fn preload_model(&self) {
        let _ = self.embedder();
    }

    /// Takes over another `App`'s loaded model instead of loading its own. The daemon keeps
    /// one `App` per actor, so the audit log names who asked, and one model for all of them.
    #[cfg_attr(not(unix), allow(dead_code))] // the daemon is Unix only
    pub fn share_model_with(&self, other: &App) {
        if let Some(state) = other.embedder.get() {
            let _ = self.embedder.set(state.clone());
        }
    }

    /// Files whose change makes a long-running process's view of this store stale: the
    /// configuration and the model manifest (the binary is the daemon's own business).
    #[cfg_attr(not(unix), allow(dead_code))] // the daemon is Unix only
    pub fn watched_files(&self) -> Vec<PathBuf> {
        vec![
            cyberbrain_core::config::Config::path_in(&self.root),
            self.config.model_dir().join(MANIFEST_FILE),
        ]
    }

    #[allow(dead_code)]
    pub fn config(&self) -> &Config {
        &self.config
    }

    #[allow(dead_code)]
    pub fn store(&self) -> &Store {
        &self.store
    }

    #[allow(dead_code)]
    pub fn policy(&self) -> &Policy {
        &self.policy
    }

    #[allow(dead_code)]
    pub fn gate(&self) -> Arc<dyn EgressGate> {
        self.gate.clone()
    }

    #[allow(dead_code)]
    pub fn actor(&self) -> &Actor {
        &self.actor
    }

    fn writers(&self, dry_run: bool) -> Writers<'_> {
        if dry_run {
            let sink = Arc::new(MemoryAuditSink::new());
            let policy = Policy::new(
                policy_config(&self.config),
                sink.clone(),
                self.actor.clone(),
            );
            Writers {
                notes: Box::new(NoopNoteWriter {
                    store: self.store.clone(),
                }),
                index: Box::new(NoopIndexWriter {
                    index: self.index.clone(),
                }),
                policy: PolicyRef::Dry(Box::new(policy), sink),
            }
        } else {
            Writers {
                notes: Box::new(FsNoteWriter {
                    store: self.store.clone(),
                }),
                index: Box::new(SqliteIndexWriter {
                    index: self.index.clone(),
                }),
                policy: PolicyRef::Real(&self.policy),
            }
        }
    }

    // ----- step 6: the embedder, lazily --------------------------------------------------

    fn embedder(&self) -> &EmbedderState {
        self.embedder.get_or_init(|| self.load_embedder())
    }

    /// The model's files and manifest, or why there is no model to speak of.
    fn model_artefact(&self) -> std::result::Result<(ModelPaths, ArtefactManifest), String> {
        let dir = self.config.model_dir();
        let paths = ModelPaths::in_dir(&dir);
        let manifest_path = dir.join(MANIFEST_FILE);
        if !paths.weights.is_file() || !paths.tokenizer.is_file() {
            return Err(format!(
                "no model artefact at {} (expected model.safetensors and tokenizer.json); \
                 search is lexical only",
                Slash(&dir)
            ));
        }
        match std::fs::read_to_string(&manifest_path) {
            Ok(text) => serde_json::from_str(&text)
                .map(|m| (paths, m))
                .map_err(|e| format!("{} is not a manifest: {e}", Slash(&manifest_path))),
            Err(_) => Err(format!(
                "model files are present but {} is missing; refusing to load unverified \
                 weights (SPEC §6.1)",
                Slash(&manifest_path)
            )),
        }
    }

    fn load_embedder(&self) -> EmbedderState {
        let (paths, manifest) = match self.model_artefact() {
            Ok(a) => a,
            Err(reason) => return EmbedderState::Absent { reason },
        };
        match StaticEmbedder::load(&paths, &manifest) {
            Ok(e) => EmbedderState::Loaded {
                embedder: Arc::new(e),
                manifest,
                paths,
            },
            Err(e) => EmbedderState::Absent {
                reason: e.to_string(),
            },
        }
    }

    /// Which model this store would embed with — profile, dimension, weights digest — without
    /// loading it, unless something already did.
    ///
    /// 2026-09-25: `status`, `doctor`, `write` and a `scan` with nothing to embed all asked
    /// `embedder()` for these three values and paid the full load for them: tokenizer parse,
    /// matrix read, 2.6 s and 1.6 GB, 26 times the SPEC budget for an unchanged `scan`.
    /// `StaticEmbedder::describe` verifies both files the same way a load does and reads only
    /// the safetensors header. The model itself is still loaded by whoever embeds.
    fn model_identity(&self) -> &ModelIdentity {
        self.identity.get_or_init(|| {
            if let Some(EmbedderState::Loaded {
                embedder, manifest, ..
            }) = self.embedder.get()
            {
                return ModelIdentity::Ready(EmbeddingProfile {
                    id: embedder.profile_id().to_string(),
                    dim: embedder.dim(),
                    model_hash: manifest.weights_blake3.clone(),
                });
            }
            if let Some(EmbedderState::Absent { reason }) = self.embedder.get() {
                return ModelIdentity::Absent(reason.clone());
            }
            match self.model_artefact() {
                Err(reason) => ModelIdentity::Absent(reason),
                Ok((paths, manifest)) => match StaticEmbedder::describe(&paths, &manifest) {
                    Ok(d) => ModelIdentity::Ready(EmbeddingProfile {
                        id: d.profile_id,
                        dim: d.dim,
                        model_hash: manifest.weights_blake3.clone(),
                    }),
                    Err(e) => ModelIdentity::Absent(e.to_string()),
                },
            }
        })
    }

    /// Does the index's recorded profile belong to this model? `None` when either side is
    /// missing. The same comparison `Index::check_embedder` makes, on the identity alone.
    fn profile_mismatch(&self, stored: &EmbeddingProfile) -> Option<Error> {
        match self.model_identity() {
            ModelIdentity::Ready(p) if p.id != stored.id || p.dim != stored.dim => {
                Some(Error::EmbeddingProfileMismatch {
                    stored: stored.describe(),
                    configured: format!("{} (dim {})", p.id, p.dim),
                })
            }
            _ => None,
        }
    }

    fn embedder_summary(&self) -> EmbedderSummary {
        match self.model_identity() {
            ModelIdentity::Ready(p) => EmbedderSummary {
                loaded: true,
                profile_id: Some(p.id.clone()),
                dim: Some(p.dim),
                reason: None,
            },
            ModelIdentity::Absent(reason) => EmbedderSummary {
                loaded: false,
                profile_id: None,
                dim: None,
                reason: Some(reason.clone()),
            },
        }
    }

    fn embedding_profile(&self) -> Option<EmbeddingProfile> {
        match self.model_identity() {
            ModelIdentity::Ready(p) => Some(p.clone()),
            ModelIdentity::Absent(_) => None,
        }
    }

    /// Record the loaded model's profile in the index, logging a change. The index wipes
    /// vectors from a previous profile itself; the row saying so is written here, because
    /// the index writes no audit rows (SPEC §4, one writer).
    fn declare_profile(&self, w: &Writers<'_>) -> Result<Option<ProfileChange>> {
        let Some(profile) = self.embedding_profile() else {
            return Ok(None);
        };
        let change = w.index.set_embedding_profile(&profile)?;
        if change.changed {
            w.policy.get().audit().record_raw(
                &Actor::Cli.to_string(),
                "index.embedding-profile-changed",
                format!("embedding:{}", change.current.id),
                json!({
                    "previous": change.previous,
                    "current": change.current,
                    "vectors_wiped": change.vectors_wiped,
                }),
            )?;
        }
        Ok(Some(change))
    }

    fn embed_blocks(&self, texts: &[&str]) -> Result<Option<Vec<Vec<f32>>>> {
        match self.embedder() {
            EmbedderState::Loaded { embedder, .. } => embedder.embed(texts).map(Some),
            EmbedderState::Absent { .. } => Ok(None),
        }
    }

    // ----- step 7: the inference client, lazily and last ---------------------------------

    async fn llm(&self) -> LlmState {
        let cached = self.llm.lock().ok().and_then(|g| g.clone());
        if let Some(s) = cached {
            return s;
        }
        let state = self.connect_llm().await;
        if let Ok(mut g) = self.llm.lock() {
            *g = Some(state.clone());
        }
        state
    }

    /// Why no client can be opened, when the configuration alone says so. Cheap and
    /// silent: no endpoint is validated and nothing is audited.
    fn llm_unconfigured(&self) -> Option<String> {
        let configured = self
            .config
            .inference
            .model
            .as_deref()
            .is_some_and(|m| !m.trim().is_empty());
        (!configured).then(|| {
            "no inference model is configured (inference.model in cyberbrain.toml); the \
             endpoint is never contacted without one"
                .into()
        })
    }

    async fn connect_llm(&self) -> LlmState {
        if let Some(reason) = self.llm_unconfigured() {
            return LlmState::Absent(reason);
        }
        let inf = &self.config.inference;
        let model = inf.model.as_deref().map(str::trim).unwrap_or_default();
        let cfg = LlmConfig {
            base_url: inf.base_url.clone(),
            model: model.to_string(),
            timeout: std::time::Duration::from_millis(inf.timeout_ms),
            allow_public_endpoint: inf.allow_public_endpoint,
            allow_overlay_network: inf.allow_overlay_network,
            ..LlmConfig::default()
        };
        let audit: Arc<dyn cyberbrain_llm::AuditSink> = Arc::new(self.policy.audit().clone());
        match LlmClient::connect(cfg, audit, self.gate.clone()).await {
            Ok(c) => LlmState::Ready(Box::new(c)),
            Err(e) => LlmState::Absent(e.to_string()),
        }
    }

    // ----- scan --------------------------------------------------------------------------

    pub fn scan(&self, opts: ScanOptions) -> Result<ScanReport> {
        let started = Instant::now();
        let w = self.writers(opts.dry_run);
        let listing = self.store.list()?;
        let mut report = ScanReport {
            dry_run: opts.dry_run,
            full: opts.full,
            files_listed: listing.entries.len() + listing.skipped.len(),
            indexed_new: 0,
            reindexed_changed: 0,
            revectorised: 0,
            unchanged: 0,
            touched_only: 0,
            dropped_missing_file: Vec::new(),
            skipped: listing
                .skipped
                .iter()
                .map(|s| SkippedFile {
                    path: s.path.clone(),
                    reason: s.reason.clone(),
                })
                .collect(),
            links_written_back: 0,
            link_writeback_failed: Vec::new(),
            oversized_blocks: Vec::new(),
            embedder: EmbedderSummary::default(),
            profile_change: None,
            cleared: None,
            index: lock_index(&self.index)?.stats()?,
            audit_preview: Vec::new(),
            elapsed_ms: 0,
        };

        if opts.full {
            let cleared = w.index.clear()?;
            w.policy.get().audit().record_raw(
                &Actor::Cli.to_string(),
                "index.cleared",
                "index",
                json!({ "erased": cleared, "dry_run": opts.dry_run }),
            )?;
            report.cleared = Some(cleared);
        }

        // The embedder is loaded here and nowhere earlier: scan needs vectors.
        report.embedder = self.embedder_summary();
        report.profile_change = self.declare_profile(&w)?;
        let model_present = report.embedder.loaded;

        let mut seen: HashSet<NoteId> = HashSet::new();
        for entry in &listing.entries {
            let mut note = match self.store.read_path(&entry.path) {
                Ok(n) => n,
                Err(e) => {
                    report.skipped.push(SkippedFile {
                        path: entry.path.clone(),
                        reason: match &e {
                            Error::Frontmatter { reason, .. } => {
                                format!("unreadable frontmatter: {reason}")
                            }
                            other => other.to_string(),
                        },
                    });
                    continue;
                }
            };
            if !seen.insert(note.front.id) {
                report.skipped.push(SkippedFile {
                    path: entry.path.clone(),
                    reason: format!(
                        "id {} is already used by another note in this tree; ids are unique",
                        note.front.id
                    ),
                });
                continue;
            }

            // Links are derived from the body and written back (SPEC §3.1).
            let targets = link_targets(&note.body);
            if targets != note.front.links {
                note.front.links = targets;
                match w.notes.write(&note) {
                    Ok(_) => report.links_written_back += 1,
                    Err(e) => report
                        .link_writeback_failed
                        .push(format!("{}: {e}", note.front.name)),
                }
            }

            // One hash decides. `content_hash` is the function the index stores, so it is
            // the one the comparison must use.
            let hash = content_hash(&note);
            let existing = lock_index(&self.index)?.note(&note.front.id)?;
            enum Decision {
                New,
                Changed,
                NeedsVectors,
                TouchedOnly,
                Unchanged,
            }
            let decision = match &existing {
                None => Decision::New,
                Some(rec) if rec.hash != hash => Decision::Changed,
                Some(rec) if model_present && rec.vector_count < rec.block_count => {
                    Decision::NeedsVectors
                }
                Some(rec) if rec.stamp != stamp_of(&note.path) => Decision::TouchedOnly,
                Some(_) => Decision::Unchanged,
            };
            match decision {
                Decision::Unchanged => {
                    report.unchanged += 1;
                    continue;
                }
                Decision::TouchedOnly => {
                    report.touched_only += 1;
                    continue;
                }
                _ => {}
            }

            let (blocks, oversized) = blocks_of(&note, MAX_BLOCK_TOKENS);
            for o in oversized {
                report.oversized_blocks.push(OversizedBlock {
                    note: note.front.name.clone(),
                    block_idx: o.block_idx,
                    approx_tokens: o.approx_tokens,
                    reason: oversized_reason(o.reason),
                });
            }
            let texts: Vec<&str> = blocks.iter().map(|b| b.text.as_str()).collect();
            let vectors = match self.embed_blocks(&texts) {
                Ok(v) => v,
                Err(e) => {
                    report.skipped.push(SkippedFile {
                        path: entry.path.clone(),
                        reason: format!("embedding failed: {e}"),
                    });
                    continue;
                }
            };
            match w.index.upsert_note(&note, &blocks, vectors.as_deref()) {
                Ok(_) => match decision {
                    Decision::New => report.indexed_new += 1,
                    Decision::Changed => report.reindexed_changed += 1,
                    Decision::NeedsVectors => report.revectorised += 1,
                    _ => unreachable!(),
                },
                Err(e) => report.skipped.push(SkippedFile {
                    path: entry.path.clone(),
                    reason: format!("the index refused it: {e}"),
                }),
            }
        }

        // Rows whose file is gone. The index is a cache of the files; a note deleted by
        // hand is dropped here and the drop is recorded, because a row vanishing without a
        // record is indistinguishable from a bug.
        let known = lock_index(&self.index)?.notes()?;
        for rec in known {
            if seen.contains(&rec.front.id) {
                continue;
            }
            let erased = w.index.delete_note(&rec.front.id)?;
            w.policy.get().audit().record_raw(
                &Actor::Cli.to_string(),
                "index.note-dropped",
                format!("note:{}", erased.id),
                json!({
                    "name": erased.name,
                    "ring": erased.ring,
                    "path": erased.path,
                    "reason": "file missing at scan",
                    "counts": erased.counts,
                    "dry_run": opts.dry_run,
                }),
            )?;
            report.dropped_missing_file.push(erased.name);
        }

        report.index = lock_index(&self.index)?.stats()?;
        report.audit_preview = w.policy.preview();
        report.elapsed_ms = started.elapsed().as_millis();
        Ok(report)
    }

    /// Whether `scan` would change the index, decided without loading the model: a full
    /// rebuild, a file that differs from the index or is gone, or vectors that are missing
    /// or from another model. The CLI sends such a scan to the daemon, which has the
    /// model loaded, and runs a scan with nothing to do itself.
    pub fn scan_has_work(&self, full: bool) -> Result<bool> {
        if full {
            return Ok(true);
        }
        let (not_indexed, changed, gone, _) = self.staleness()?;
        if not_indexed + changed + gone > 0 {
            return Ok(true);
        }
        let Some(current) = self.embedding_profile() else {
            return Ok(false);
        };
        let stats = lock_index(&self.index)?.stats()?;
        Ok(stats.vectors < stats.blocks || stats.embedding.as_ref() != Some(&current))
    }

    /// The comparison half of `scan`, for `doctor` and `status`: how many files differ
    /// from the index. Reads everything, writes nothing.
    fn staleness(&self) -> Result<(usize, usize, usize, Vec<SkippedFile>)> {
        let listing = self.store.list()?;
        let ix = lock_index(&self.index)?;
        let mut not_indexed = 0;
        let mut changed = 0;
        let mut unreadable = Vec::new();
        let mut seen = HashSet::new();
        for e in &listing.entries {
            match self.store.read_path(&e.path) {
                Ok(n) => {
                    seen.insert(n.front.id);
                    let mut probe = n.clone();
                    probe.front.links = link_targets(&n.body);
                    match ix.note(&n.front.id)? {
                        None => not_indexed += 1,
                        Some(rec) if rec.hash != content_hash(&probe) => changed += 1,
                        Some(_) => {}
                    }
                }
                Err(err) => unreadable.push(SkippedFile {
                    path: e.path.clone(),
                    reason: err.to_string(),
                }),
            }
        }
        let gone = ix
            .notes()?
            .into_iter()
            .filter(|r| !seen.contains(&r.front.id))
            .count();
        Ok((not_indexed, changed, gone, unreadable))
    }

    // ----- recall ------------------------------------------------------------------------

    pub async fn recall(&self, query: &str, req: &RecallRequest) -> Result<RecallResult> {
        let r = &self.config.retrieval;
        let opts = RecallOptions {
            n: req.n.unwrap_or(r.n),
            k_lex: r.k_lex,
            k_sem: r.k_sem,
            ring: req.ring,
            bereich: req.bereich.clone(),
            min_cosine: 0.0,
            // Rings 0 and 1 are injected whole when an agent's session starts
            // (hook/events.rs); for an agent a hit from them is a second copy of its own
            // context. A person — at the terminal, in the UI, whose simple view answers with
            // the ring 0 note — has no such context, and gets them. `--ring 0|1` always
            // searches them.
            skip_resident: matches!(self.actor, Actor::Agent(_)),
            at: req.at,
        };
        let embedder_state = self.embedder();
        let mut result = {
            let ix = lock_index(&self.index)?;
            let embedder: Option<&dyn Embedder> = match embedder_state {
                EmbedderState::Loaded { embedder, .. } => Some(embedder.as_ref()),
                EmbedderState::Absent { .. } => None,
            };
            ix.recall(query, embedder, &opts)?
        };
        if let EmbedderState::Absent { reason } = embedder_state {
            result
                .caveats
                .push(format!("embedder not loaded: {reason}"));
        }
        // Contradiction check (SPEC §7): only with a configured local model, and its
        // absence is said out loud. It runs under a budget, because the hits are ready in
        // milliseconds and this call is the only reason a recall ever feels slow.
        //
        // Everything that can decide "not this time" is decided before the client is
        // opened. Opening it validates the endpoint and asks the egress register, and both
        // leave a row in the audit log; measured 2026-09-25 on the orderflow store, that
        // was two rows for every read and 27 % of the whole log, for checks that were then
        // skipped anyway.
        match self.plan_check(&result.hits) {
            Err(caveat) => result.caveats.push(caveat),
            Ok(plan) => match self.llm().await {
                LlmState::Absent(reason) => result
                    .caveats
                    .push(format!("contradiction check skipped: {reason}")),
                LlmState::Ready(client) => {
                    let budget_ms = self.config.inference.contradiction_budget_ms;
                    let probe = self.load_probe();
                    let started = std::time::Instant::now();
                    let checked = match plan {
                        CheckPlan::Run(budget) => tokio::time::timeout(
                            budget,
                            cyberbrain_llm::tasks::find_conflicts(&client, &result.hits),
                        )
                        .await
                        .ok(),
                        _ => {
                            Some(cyberbrain_llm::tasks::find_conflicts(&client, &result.hits).await)
                        }
                    };
                    match checked {
                        Some((conflicts, caveats)) => {
                            self.record_load(TASK_CONTRADICTION, probe, started.elapsed());
                            result.conflicts = conflicts;
                            result.caveats.extend(caveats);
                        }
                        None => {
                            // Abandoned, not completed: recorded under its own task so the
                            // ledger keeps saying what happened, and so the next call in
                            // another process knows without paying again.
                            self.record_load(
                                TASK_CONTRADICTION_ABANDONED,
                                probe,
                                started.elapsed(),
                            );
                            result.caveats.push(format!(
                                "contradiction check gave up after {} \
                                 (inference.contradiction_budget_ms); the hits are not \
                                 checked against each other",
                                secs(budget_ms)
                            ));
                        }
                    }
                }
            },
        }
        self.record_recall_usage(&result);
        Ok(result)
    }

    /// Whether the contradiction check runs for these hits, decided from what is on disk
    /// alone: the hits, the configuration and the load ledger. `Err` is the caveat saying
    /// why not. Nothing here opens the inference client, so nothing here is audited.
    fn plan_check(&self, hits: &[cyberbrain_core::Hit]) -> std::result::Result<CheckPlan, String> {
        if hits.len() < 2 {
            return Err(
                "contradiction check skipped: fewer than two hits, nothing to compare".into(),
            );
        }
        // Conflicts are defined across rings (SPEC §7). Before, this was found out inside
        // the check, after the client was opened, and the instant return was booked in the
        // ledger as a completed check of 0 ms.
        if !cyberbrain_llm::tasks::spans_rings(hits) {
            return Err(cyberbrain_llm::tasks::one_ring_caveat(hits.len()));
        }
        if let Some(reason) = self.llm_unconfigured() {
            return Err(format!("contradiction check skipped: {reason}"));
        }
        let budget_ms = self.config.inference.contradiction_budget_ms;
        let last = match hostload::LoadLog::new(&self.root)
            .last_of(&[TASK_CONTRADICTION, TASK_CONTRADICTION_ABANDONED])
        {
            None => LastCheck::Unknown,
            // A measurement from another day says nothing about this one: a faster model,
            // a GPU, a machine that was busy last time. Without this, one slow afternoon
            // would switch the check off for good and nothing would ever try again.
            Some(r) if stale(&r.at) => LastCheck::Unknown,
            Some(r) if r.task == TASK_CONTRADICTION_ABANDONED => LastCheck::Abandoned,
            Some(r) => LastCheck::Completed { ms: r.wall_ms },
        };
        match plan_contradiction_check(budget_ms, last) {
            CheckPlan::SkipMeasuredSlow { last_ms, budget_ms } => Err(format!(
                "contradiction check skipped: the last one took {}, over the {} budget \
                 (inference.contradiction_budget_ms); the hits are not checked against each \
                 other",
                secs(last_ms),
                secs(budget_ms)
            )),
            CheckPlan::SkipAbandoned { budget_ms } => Err(format!(
                "contradiction check skipped: the last one was still running when its {} \
                 budget ran out (inference.contradiction_budget_ms); the hits are not checked \
                 against each other",
                secs(budget_ms)
            )),
            plan => Ok(plan),
        }
    }

    /// Ledger row for one recall: the tokens handed over against the tokens the notes those
    /// hits came from hold in full. Both sides come from the index, which counts tokens per
    /// block, so neither is an estimate. Failure to measure is failure to record, never a
    /// zero: a zero here would read as "saved nothing".
    fn record_recall_usage(&self, result: &RecallResult) {
        if result.hits.is_empty() {
            return;
        }
        let cited: std::collections::HashSet<&str> =
            result.hits.iter().map(|h| h.citation.as_str()).collect();
        let notes: std::collections::BTreeSet<&NoteId> =
            result.hits.iter().map(|h| &h.note_id).collect();
        let Ok(ix) = lock_index(&self.index) else {
            return;
        };
        let (mut returned, mut full) = (0u64, 0u64);
        for id in &notes {
            let Ok(blocks) = ix.blocks_of(id) else {
                return;
            };
            for b in blocks {
                full += u64::from(b.token_count);
                if cited.contains(b.citation.to_string().as_str()) {
                    returned += u64::from(b.token_count);
                }
            }
        }
        drop(ix);
        self.usage().append(&usage::UsageRow {
            at: usage::now(),
            op: "recall".into(),
            unit: "tokens".into(),
            returned,
            full,
            hits: result.hits.len() as u64,
            sources: notes.len() as u64,
        });
    }

    /// The before-half of a load measurement. Cheap enough (two small reads, three when a
    /// cgroup is configured) to take around every model call.
    fn load_probe(&self) -> (Option<hostload::HostSample>, Option<hostload::CgroupSample>) {
        (hostload::read_host(), self.cgroup_sample())
    }

    fn cgroup_sample(&self) -> Option<hostload::CgroupSample> {
        let dir = self.config.inference.load_cgroup.as_ref()?;
        hostload::read_cgroup(Path::new(dir))
    }

    fn record_load(
        &self,
        task: &str,
        before: (Option<hostload::HostSample>, Option<hostload::CgroupSample>),
        wall: std::time::Duration,
    ) {
        let after = (hostload::read_host(), self.cgroup_sample());
        let row = hostload::row(task, wall, (before.0, after.0), (before.1, after.1));
        hostload::LoadLog::new(&self.root).append(&row);
    }

    /// The last `days` calendar days of everything the two ledgers and the audit log know,
    /// oldest first, with empty days present and zero. Three sources are merged here rather
    /// than in the page, because the page must not be the place where "no data" and "zero"
    /// get to look alike.
    pub fn usage_by_day(&self, days: usize) -> Vec<usage::DayBucket> {
        let axis = usage::day_axis(days);
        let mut by_date: std::collections::BTreeMap<String, usage::DayBucket> = axis
            .iter()
            .map(|d| {
                (
                    d.clone(),
                    usage::DayBucket {
                        date: d.clone(),
                        ..Default::default()
                    },
                )
            })
            .collect();

        // Retrieval ledger.
        if let Ok(text) = std::fs::read_to_string(self.root.join("usage.jsonl")) {
            for line in text.lines() {
                let Ok(r) = serde_json::from_str::<usage::UsageRow>(line) else {
                    continue;
                };
                let Some(day) = usage::day_of(&r.at).and_then(|d| by_date.get_mut(d)) else {
                    continue;
                };
                let t = if r.op == "find" {
                    &mut day.find
                } else {
                    &mut day.recall
                };
                t.ops += 1;
                t.returned += r.returned;
                t.full += r.full;
                t.hits += r.hits;
            }
        }

        // Model calls, from the audit log.
        let filter = AuditFilter {
            action: Some("inference.call".into()),
            ..AuditFilter::default()
        };
        if let Ok(rows) = self.policy.audit().read(&filter) {
            for r in rows {
                let d = &r.detail;
                let call = d.get("call").and_then(|v| v.as_str()).unwrap_or("");
                if call != "chat-completion" && call != "chat-completion-stream" {
                    continue;
                }
                let ts = r.ts.to_string();
                let Some(day) = usage::day_of(&ts).and_then(|d| by_date.get_mut(d)) else {
                    continue;
                };
                let num = |k: &str| d.get(k).and_then(|v| v.as_u64()).unwrap_or(0);
                day.calls += 1;
                day.prompt_tokens += num("prompt_tokens");
                day.cached_prompt_tokens += num("cached_prompt_tokens");
                day.completion_tokens += num("completion_tokens");
            }
        }

        // Load ledger: averaged, so the day carries a rate and not a sum of rates.
        let mut cores: std::collections::BTreeMap<String, (f64, u64, f64, u64)> =
            std::collections::BTreeMap::new();
        if let Ok(text) = std::fs::read_to_string(self.root.join("load.jsonl")) {
            for line in text.lines() {
                let Ok(r) = serde_json::from_str::<hostload::LoadRow>(line) else {
                    continue;
                };
                let Some(date) = usage::day_of(&r.at).map(str::to_owned) else {
                    continue;
                };
                if let Some(day) = by_date.get_mut(&date) {
                    day.wall_ms += r.wall_ms;
                }
                let e = cores.entry(date).or_insert((0.0, 0, 0.0, 0));
                if let Some(c) = r.endpoint_cores {
                    e.0 += c;
                    e.1 += 1;
                }
                if let Some(c) = r.machine_cores {
                    e.2 += c;
                    e.3 += 1;
                }
            }
        }
        for (date, (ep, epn, ma, man)) in cores {
            let Some(day) = by_date.get_mut(&date) else {
                continue;
            };
            if epn > 0 {
                day.endpoint_cores = Some(ep / epn as f64);
            }
            if man > 0 {
                day.machine_cores = Some(ma / man as f64);
            }
        }
        by_date.into_values().collect()
    }

    pub fn load_summary(&self) -> hostload::LoadSummary {
        hostload::LoadLog::new(&self.root).summary()
    }

    /// What the endpoint says it currently holds in memory. `None` when no model is
    /// configured, the endpoint is unreachable, or it does not answer the vendor route.
    pub async fn loaded_models(&self) -> Option<Vec<cyberbrain_llm::LoadedModel>> {
        match self.llm().await {
            LlmState::Ready(client) => client.loaded_models().await,
            LlmState::Absent(_) => None,
        }
    }

    fn usage(&self) -> usage::UsageLog {
        usage::UsageLog::new(&self.root)
    }

    pub fn usage_summary(&self) -> usage::UsageSummary {
        self.usage().summary()
    }

    /// What the local model actually cost, read back out of the audit log. Grouped by task,
    /// because "contradiction-check" and "session-summary" are paid for separately. Rows the
    /// endpoint did not report counts for are counted as calls and not as tokens, and the
    /// number of those is carried so the totals cannot be mistaken for complete.
    pub fn inference_usage(&self) -> usage::InferenceUsage {
        let filter = AuditFilter {
            action: Some("inference.call".into()),
            ..AuditFilter::default()
        };
        let Ok(rows) = self.policy.audit().read(&filter) else {
            return usage::InferenceUsage::default();
        };
        let mut out = usage::InferenceUsage::default();
        for r in rows {
            let d = &r.detail;
            let call = d.get("call").and_then(|v| v.as_str()).unwrap_or("");
            if call != "chat-completion" && call != "chat-completion-stream" {
                continue;
            }
            let task = d
                .get("task")
                .and_then(|v| v.as_str())
                .unwrap_or("unnamed")
                .to_string();
            let e = out.tasks.entry(task).or_default();
            let num = |k: &str| d.get(k).and_then(|v| v.as_u64());
            e.calls += 1;
            e.elapsed_ms += num("elapsed_ms").unwrap_or(0);
            if d.get("outcome").and_then(|v| v.as_str()) != Some("ok") {
                e.failed += 1;
            }
            match (num("prompt_tokens"), num("completion_tokens")) {
                (Some(p), c) => {
                    e.prompt_tokens += p;
                    e.completion_tokens += c.unwrap_or(0);
                    match num("cached_prompt_tokens") {
                        Some(c) => e.cached_prompt_tokens += c,
                        None => e.calls_without_cache_report += 1,
                    }
                }
                _ => e.calls_without_counts += 1,
            }
            if out.first.is_none() {
                out.first = Some(r.ts.to_string());
            }
            out.last = Some(r.ts.to_string());
        }
        out
    }

    /// `recall --id`: a citation back to its block and the whole note it came from.
    pub fn recall_id(&self, citation: &str) -> Result<Expanded> {
        let cit: Citation = citation.parse()?;
        let (block, rec) = lock_index(&self.index)?.resolve(&cit)?.ok_or_else(|| {
            Error::NoSuchNote(format!(
                "citation {cit} (not in the index; run `cyberbrain scan` if the note exists)"
            ))
        })?;
        let note = self.note_view_at(&rec.path)?;
        Ok(Expanded {
            citation: cit.to_string(),
            block: BlockView {
                citation: block.citation.to_string(),
                idx: block.idx,
                text: block.text,
                token_count: block.token_count,
            },
            note,
        })
    }

    fn note_view_at(&self, path: &Path) -> Result<NoteView> {
        let n = self.store.read_path(path)?;
        let blocks = lock_index(&self.index)?
            .blocks_of(&n.front.id)?
            .into_iter()
            .map(|b| b.citation.to_string())
            .collect();
        Ok(NoteView {
            kind: kind_name(n.front.kind),
            front: n.front,
            body: n.body,
            path: n.path,
            blocks,
        })
    }

    /// A name, or an id. Names are the common case; ids are tried when the name does not
    /// parse or does not exist.
    fn resolve_target(&self, target: &str) -> Result<Note> {
        if let Ok(n) = self.store.read(target) {
            return Ok(n);
        }
        if let Ok(id) = NoteId::from_string(target) {
            if let Some(rec) = lock_index(&self.index)?.note(&id)? {
                return self.store.read_path(&rec.path);
            }
            return self.store.read_by_id(id);
        }
        Err(Error::NoSuchNote(target.to_string()))
    }

    pub fn export(&self, target: &str) -> Result<NoteView> {
        let n = self.resolve_target(target)?;
        self.note_view_at(&n.path)
    }

    // ----- write -------------------------------------------------------------------------

    /// Put the notes tree back after an audit row that could not be written.
    ///
    /// Removed if the note is new, rewritten from the copy read at the start if it is an
    /// edit. Failures here are printed rather than returned: the caller is already on its
    /// way out with the reason it got here, and replacing that reason with "and the
    /// rollback failed too" would lose the one the operator needs. Staying silent is the
    /// one thing it must not do — then a file really is left behind with nothing saying so.
    fn undo_note_write(w: &Writers<'_>, note: &Note, existing: Option<&Note>, why: &Error) {
        let name = &note.front.name;
        let put_back = match existing {
            Some(prev) => w.notes.write(prev).map(|_| ()),
            None => w.notes.remove(&note.path).map(|_| ()),
        };
        if let Err(e) = put_back {
            eprintln!(
                "cyberbrain: {name} could not be recorded ({why}) and the file could not be \
                 put back either ({e}). The note on disk is not in the audit chain; remove \
                 or rewrite it by hand, then run `cyberbrain scan`."
            );
        }
    }

    /// Rings 0 and 1 belong to the operator (SPEC §3.2): session start injects them as
    /// invariants, and they outrank everything below them.
    ///
    /// 2026-09-22: until now only the hub path (`apply_pulled`) kept that rule; `App::write`
    /// did not look at the actor at all. Over MCP, `write {ring: 0}` therefore landed in
    /// `notes/r0/`, and an agent could plant a standing instruction in every future session
    /// start. The check sits here, where the actor is known, so that every path through
    /// `write` meets it rather than each caller having to remember it. The operator is
    /// `Operator` (CLI, web UI) and `Cli`; anybody else is pointed at `propose`. Nobody else
    /// may overwrite an existing ring 0/1 note either, whatever ring they ask for.
    fn refuse_resident_unless_operator(
        &self,
        name: &str,
        ring: Ring,
        existing: Option<&Note>,
        dry_run: bool,
    ) -> Result<()> {
        if matches!(self.actor, Actor::Operator | Actor::Cli) {
            return Ok(());
        }
        let held = existing.map(|n| n.front.ring).filter(|r| r.is_resident());
        let Some(r) = held.or(Some(ring).filter(|r| r.is_resident())) else {
            return Ok(());
        };
        let reason = format!(
            "ring {} belongs to the operator; {} may not write it (note {name}). Propose the \
             text instead — `cyberbrain propose --ring {} --kind … --name {name}` — and the \
             operator accepts it with `cyberbrain review {name} --accept`",
            r.as_u8(),
            self.actor,
            r.as_u8(),
        );
        // A refusal is the most interesting audit row there is (§12.1). A dry run writes
        // nothing, this row included.
        if !dry_run {
            self.policy.audit().record(
                &self.actor,
                AuditAction::PolicyRefusal,
                format!("name:{name}"),
                serde_json::json!({ "ring": r, "requested_ring": ring, "reason": reason }),
            )?;
        }
        Err(Error::PolicyRefusal {
            profile: "ring-owner".to_string(),
            reason,
        })
    }

    pub fn write(&self, req: WriteRequest) -> Result<WriteOutcome> {
        let name = frontmatter::normalize_name(req.name.trim()).into_owned();
        frontmatter::validate_name(&name).map_err(|why| Error::Frontmatter {
            path: PathBuf::from(format!("{name}.md")),
            reason: format!("name `{name}`: {why}"),
        })?;
        if let Some(Some(r)) = &req.retention {
            frontmatter::validate_retention(r).map_err(|why| Error::Frontmatter {
                path: PathBuf::from(format!("{name}.md")),
                reason: format!("retention `{r}`: {why}"),
            })?;
        }
        for s in req.supersedes.iter().flatten() {
            let why = if *s == name {
                Some("is the note itself")
            } else {
                frontmatter::validate_name(s).err()
            };
            if let Some(why) = why {
                return Err(Error::Frontmatter {
                    path: PathBuf::from(format!("{name}.md")),
                    reason: format!("supersedes `{s}`: {why}"),
                });
            }
        }
        let w = self.writers(req.dry_run);
        let policy = w.policy.get();

        let existing = match self.store.read(&name) {
            Ok(n) => Some(n),
            Err(Error::NoSuchNote(_)) => None,
            Err(e) => return Err(e),
        };
        self.refuse_resident_unless_operator(&name, req.ring, existing.as_ref(), req.dry_run)?;
        if let (Some(cur), Some(expected)) = (&existing, req.expected_updated)
            && cur.front.updated != expected
        {
            return Ok(WriteOutcome::Conflict {
                name,
                current_updated: cur.front.updated,
            });
        }

        // SPEC §12.4: the scan runs before any byte is written.
        let (body, pii, redacted) = match policy.check_write(&name, &req.body)? {
            WriteVerdict::Proceed { pii, .. } => (req.body.clone(), pii, 0),
            WriteVerdict::Held { findings } => {
                let choice = req.choice.or(if req.force {
                    Some(OperatorChoice::ProceedFlagged)
                } else {
                    None
                });
                match choice {
                    None => {
                        return Ok(WriteOutcome::Held {
                            rendered: cyberbrain_policy::write_gate::render_hold(
                                &req.body, &findings,
                            ),
                            name,
                            findings,
                        });
                    }
                    Some(c) => {
                        let r = policy.resolve_hold(&name, &req.body, &findings, c)?;
                        (r.body, r.pii, r.redacted)
                    }
                }
            }
        };

        let now = jiff::Timestamp::now();
        let mut tags: Vec<String> = Vec::new();
        for t in req.tags {
            let t = t.trim().to_string();
            if !t.is_empty() && !tags.contains(&t) {
                tags.push(t);
            }
        }
        let arriving = req.arriving.clone();
        // An arriving note keeps what it already is. The local note's id still wins when
        // there is one — two ids for one name is a store that contradicts itself, and the
        // note here was written first as far as this machine can tell.
        let front = Frontmatter {
            id: existing
                .as_ref()
                .map(|n| n.front.id)
                .or(arriving.as_ref().map(|a| a.id))
                .unwrap_or_else(NoteId::generate),
            name: name.clone(),
            ring: req.ring,
            kind: req.kind,
            created: existing
                .as_ref()
                .map(|n| n.front.created)
                .or(arriving.as_ref().map(|a| a.created))
                .unwrap_or(now),
            // The sender's stamp, not ours. Ours would be newer than every version the hub
            // still has to offer, so the next real change from the other machine would be
            // kept back as "this machine changed it since" — which this machine never did.
            updated: arriving.as_ref().map(|a| a.updated).unwrap_or(now),
            tags: match &arriving {
                Some(a) if tags.is_empty() => a.tags.clone(),
                _ => tags,
            },
            links: link_targets(&body),
            // Absent keeps what the note has; an explicit `Some(None)` removes it.
            bereich: match req.bereich.clone() {
                None => existing.as_ref().and_then(|n| n.front.bereich.clone()),
                Some(v) => v,
            },
            // Absent keeps what the note has, then what arrived with it; an explicit
            // `Some(None)` removes it.
            retention: match req.retention.clone() {
                None => existing
                    .as_ref()
                    .and_then(|n| n.front.retention.clone())
                    .or_else(|| arriving.as_ref().and_then(|a| a.retention.clone())),
                Some(v) => v,
            },
            // Absent keeps what the note has (a hand-set `supersedes` survives an edit).
            supersedes: match req.supersedes.clone() {
                None => existing
                    .as_ref()
                    .map(|n| n.front.supersedes.clone())
                    .unwrap_or_default(),
                Some(v) => v,
            },
            // Only ever set by hand in the head; a write keeps it.
            superseded_by: existing
                .as_ref()
                .and_then(|n| n.front.superseded_by.clone()),
            // Absent keeps what the note has; an explicit `Some(None)` removes it.
            valid_from: match req.valid_from {
                None => existing.as_ref().and_then(|n| n.front.valid_from),
                Some(v) => v,
            },
            invalid_at: match req.invalid_at {
                None => existing.as_ref().and_then(|n| n.front.invalid_at),
                Some(v) => v,
            },
            pii,
        };
        let note = Note {
            front,
            body,
            path: PathBuf::new(),
        };
        let bytes = frontmatter::render(&note.front, &note.body)?.len();
        let path = w.notes.write(&note)?;
        let note = Note { path, ..note };
        // The file is on disk and the row is not, and that is the wrong order to be
        // interrupted in: a note the chain does not mention is the one state the chain
        // exists to rule out. `record_write` already retries a moved chain head eight
        // times, so arriving here means something worse than contention — and then the
        // file goes back to what it was, rather than standing as an unrecorded note.
        if let Err(e) = policy.record_write(&note.front, bytes) {
            Self::undo_note_write(&w, &note, existing.as_ref(), &e);
            return Err(e);
        }

        // A write reindexes the note in the same request (§8.1).
        let (blocks, _) = blocks_of(&note, MAX_BLOCK_TOKENS);
        let texts: Vec<&str> = blocks.iter().map(|b| b.text.as_str()).collect();
        self.declare_profile(&w)?;
        let vectors = self.embed_blocks(&texts)?;
        let outcome = w.index.upsert_note(&note, &blocks, vectors.as_deref())?;
        Ok(WriteOutcome::Written(WrittenNote {
            id: note.front.id,
            name,
            ring: note.front.ring,
            kind: kind_name(note.front.kind),
            path: note.path,
            bytes,
            created: existing.is_none(),
            updated: now,
            pii,
            redacted,
            blocks: outcome.blocks,
            vectors: outcome.vectors,
            links: outcome.links,
            embedder_reason: self.embedder_summary().reason,
            dry_run: req.dry_run,
            audit_preview: w.policy.preview(),
        }))
    }

    /// Declare that a note stopped holding at a moment (or, with `clear`, that it holds
    /// again). Only the head changes: the body is not scanned again, because it is not
    /// rewritten. Everything else is the write path's — the ring owner's guard, one
    /// `note.write` row naming the operation and the bounds, and a reindex in the same
    /// request, so the next recall already ranks the note down and says since when.
    pub fn invalidate(&self, req: InvalidateRequest) -> Result<InvalidatedNote> {
        let name = frontmatter::normalize_name(req.name.trim()).into_owned();
        if let Some(by) = &req.by {
            let by = frontmatter::normalize_name(by.trim());
            let why = if by == name {
                Some("is the note itself")
            } else {
                frontmatter::validate_name(&by).err()
            };
            if let Some(why) = why {
                return Err(Error::Frontmatter {
                    path: PathBuf::from(format!("{name}.md")),
                    reason: format!("replaced by `{by}`: {why}"),
                });
            }
        }
        let w = self.writers(req.dry_run);
        let policy = w.policy.get();
        let existing = self.store.read(&name)?;
        self.refuse_resident_unless_operator(
            &name,
            existing.front.ring,
            Some(&existing),
            req.dry_run,
        )?;

        let now = jiff::Timestamp::now();
        let mut front = existing.front.clone();
        front.invalid_at = if req.clear {
            None
        } else {
            Some(req.at.unwrap_or(now))
        };
        if let Some(by) = &req.by {
            front.superseded_by = Some(frontmatter::normalize_name(by.trim()).into_owned());
        }
        front.updated = now;
        let note = Note {
            front,
            body: existing.body.clone(),
            path: PathBuf::new(),
        };
        // Refuses an `invalid_at` that is not after the note's `valid_from`.
        let bytes = frontmatter::render(&note.front, &note.body)?.len();
        let path = w.notes.write(&note)?;
        let note = Note { path, ..note };
        let op = if req.clear {
            "revalidate"
        } else {
            "invalidate"
        };
        if let Err(e) = policy.record_write_as(&note.front, bytes, Some(op)) {
            Self::undo_note_write(&w, &note, Some(&existing), &e);
            return Err(e);
        }
        let (blocks, _) = blocks_of(&note, MAX_BLOCK_TOKENS);
        let texts: Vec<&str> = blocks.iter().map(|b| b.text.as_str()).collect();
        self.declare_profile(&w)?;
        let vectors = self.embed_blocks(&texts)?;
        let outcome = w.index.upsert_note(&note, &blocks, vectors.as_deref())?;
        Ok(InvalidatedNote {
            id: note.front.id,
            name,
            ring: note.front.ring,
            path: note.path,
            invalid_at: note.front.invalid_at,
            previous_invalid_at: existing.front.invalid_at,
            superseded_by: note.front.superseded_by,
            updated: now,
            blocks: outcome.blocks,
            vectors: outcome.vectors,
            dry_run: req.dry_run,
            audit_preview: w.policy.preview(),
        })
    }

    /// Write the `manifest.json` a model artefact needs, from the files themselves.
    ///
    /// The manifest is what makes a swapped model artefact a refusal rather than a silent
    /// change of meaning (SPEC §6): the loader checks both digests and will not load an
    /// artefact that does not match. Producing one was, until now, an undocumented exercise —
    /// the field names appeared in no Markdown in the repository and the hashing existed only
    /// inside tests, so the shape had to be guessed from a deserialisation error. This is the
    /// same hashing, reachable.
    ///
    /// It does not fetch anything and does not decide whether the artefact is the right one:
    /// it records what is in the folder, so that a later change to it is caught.
    pub fn write_manifest(&self, dir: &Path) -> Result<ManifestReport> {
        let paths = cyberbrain_embed::ModelPaths::in_dir(dir);
        for (what, path) in [
            ("model.safetensors", &paths.weights),
            ("tokenizer.json", &paths.tokenizer),
        ] {
            if !path.is_file() {
                return Err(Error::Config(format!(
                    "no {what} in {}; put the model2vec artefact there first",
                    Slash(dir)
                )));
            }
        }
        let manifest = cyberbrain_embed::ArtefactManifest {
            weights_blake3: cyberbrain_embed::hash_file(&paths.weights)?,
            tokenizer_blake3: cyberbrain_embed::hash_file(&paths.tokenizer)?,
        };
        let path = dir.join("manifest.json");
        let text = serde_json::to_string_pretty(&manifest)
            .map_err(|e| Error::Index(format!("the manifest does not serialise: {e}")))?
            + "\n";
        cyberbrain_core::store::write_atomic(&path, text.as_bytes())?;
        Ok(ManifestReport {
            path,
            weights_blake3: manifest.weights_blake3,
            tokenizer_blake3: manifest.tokenizer_blake3,
        })
    }

    // ----- review: propose, list, accept, reject -----------------------------------------
    //
    // A proposal is a note that is not a note yet. It lives in `proposals/`, outside the
    // notes tree, and that placement is the whole safety of this: `Frontmatter` does not
    // deny unknown fields, so a state written into a note's own header would be read and
    // ignored by every older binary, and an unapproved ring 0 note would be injected as a
    // live invariant. A directory an older `scan` never walks cannot be ignored.
    //
    // It follows that a proposal is not in the index, so `recall` and `find` cannot return
    // one. That is the intent, not a side effect: an agent that finds an unapproved
    // invariant treats it as one.

    /// Write a note into `proposals/` for somebody else to accept.
    ///
    /// The PII gate runs here rather than at acceptance, so the person who wrote the text is
    /// the one who answers for it. It runs again at acceptance over the same body, because
    /// the profile may have changed in between.
    pub fn propose(&self, req: WriteRequest, who: &str) -> Result<Proposed> {
        let name = frontmatter::normalize_name(req.name.trim()).into_owned();
        frontmatter::validate_name(&name).map_err(|why| Error::Frontmatter {
            path: PathBuf::from(format!("{name}.md")),
            reason: format!("name `{name}`: {why}"),
        })?;
        if let Some(Some(r)) = &req.retention {
            frontmatter::validate_retention(r).map_err(|why| Error::Frontmatter {
                path: PathBuf::from(format!("{name}.md")),
                reason: format!("retention `{r}`: {why}"),
            })?;
        }
        for s in req.supersedes.iter().flatten() {
            let why = if *s == name {
                Some("is the note itself")
            } else {
                frontmatter::validate_name(s).err()
            };
            if let Some(why) = why {
                return Err(Error::Frontmatter {
                    path: PathBuf::from(format!("{name}.md")),
                    reason: format!("supersedes `{s}`: {why}"),
                });
            }
        }
        if self.store.read_proposal(&name).is_ok() {
            return Err(Error::Config(format!(
                "a proposal named {name} is already waiting; `cyberbrain review {name} --reject` \
                 it first, or propose under another name"
            )));
        }

        let w = self.writers(req.dry_run);
        let policy = w.policy.get();

        // SPEC §12.4: the scan runs before any byte is written, here as anywhere.
        let (body, pii, redacted) = match policy.check_write(&name, &req.body)? {
            WriteVerdict::Proceed { pii, .. } => (req.body.clone(), pii, 0),
            WriteVerdict::Held { findings } => {
                let choice = req.choice.or(if req.force {
                    Some(OperatorChoice::ProceedFlagged)
                } else {
                    None
                });
                match choice {
                    None => {
                        return Ok(Proposed::Held {
                            rendered: cyberbrain_policy::write_gate::render_hold(
                                &req.body, &findings,
                            ),
                            name,
                            findings,
                        });
                    }
                    Some(c) => {
                        let r = policy.resolve_hold(&name, &req.body, &findings, c)?;
                        (r.body, r.pii, r.redacted)
                    }
                }
            }
        };

        let now = jiff::Timestamp::now();
        let mut tags: Vec<String> = Vec::new();
        for t in req.tags {
            let t = t.trim().to_string();
            if !t.is_empty() && !tags.contains(&t) {
                tags.push(t);
            }
        }
        let replaces = match self.store.read(&name) {
            Ok(n) => Some(n.front.updated),
            Err(Error::NoSuchNote(_)) => None,
            Err(e) => return Err(e),
        };
        let note = Note {
            front: Frontmatter {
                id: NoteId::generate(),
                name: name.clone(),
                ring: req.ring,
                kind: req.kind,
                created: now,
                updated: now,
                tags,
                links: link_targets(&body),
                bereich: req.bereich.flatten(),
                retention: req.retention.flatten(),
                supersedes: req.supersedes.clone().unwrap_or_default(),
                superseded_by: None,
                valid_from: req.valid_from.flatten(),
                invalid_at: req.invalid_at.flatten(),
                pii,
            },
            body,
            path: PathBuf::new(),
        };
        let bytes = frontmatter::render(&note.front, &note.body)?.len();

        // The real path with the file write no-op'd, rather than a simulation beside it
        // (SPEC §8): everything above ran, including the gate and the rendering.
        let path = if req.dry_run {
            self.store.proposal_path(&name)?
        } else {
            self.store.write_proposal(&note)?
        };

        // Who proposed it lives here and only here. The chain is hashed, which makes it a
        // worse thing to forge than a line of YAML in a file anyone can edit — and it means
        // no note format changed for this feature.
        policy.audit().record(
            &self.actor,
            AuditAction::NoteProposed,
            format!("note:{name}"),
            serde_json::json!({
                "by": who,
                "ring": req.ring.as_u8(),
                "kind": kind_name(req.kind),
                "bytes": bytes,
                "pii": pii,
                "changes_existing": replaces.is_some(),
                "dry_run": req.dry_run,
            }),
        )?;

        Ok(Proposed::Written(ProposeReport {
            name,
            ring: note.front.ring,
            kind: kind_name(note.front.kind),
            path,
            proposed_by: who.to_string(),
            bytes,
            pii,
            redacted,
            changes_existing: replaces.is_some(),
            dry_run: req.dry_run,
            audit_preview: w.policy.preview(),
        }))
    }

    /// Everything waiting, oldest first, with who proposed it.
    pub fn proposals(&self) -> Result<Vec<ProposalSummary>> {
        let mut out = Vec::new();
        for note in self.store.list_proposals()? {
            let name = note.front.name.clone();
            out.push(ProposalSummary {
                proposed_by: self.proposer_of(&name)?,
                changes_existing: self.store.read(&name).is_ok(),
                name,
                ring: note.front.ring,
                kind: kind_name(note.front.kind),
                created: note.front.created,
                path: note.path,
            });
        }
        Ok(out)
    }

    /// Who the audit log says proposed the proposal that is **open right now**, if any.
    ///
    /// `None` means there is no open proposal of that name — either nothing was ever
    /// proposed under it, or what was has already been accepted or rejected. Either way the
    /// two-person rule cannot be applied, and `review` says so rather than waving it
    /// through.
    ///
    /// The lifecycle is what makes this sound, and asking only whether a `note.proposed`
    /// row exists was not enough. Those rows stay in the log forever — they have to, it is
    /// hash-chained — so a name that was once proposed and then rejected kept answering
    /// this question for good. Anyone could put a file of that name back into `proposals/`
    /// by hand, with any content and any ring, and `review --accept` would find the old row,
    /// be satisfied, compare the two-person rule against a person who had nothing to do with
    /// it, and write an unapproved ring 0 note whose audit trail then named that person as
    /// its proposer. So the question is not "was this ever proposed" but "is the newest
    /// thing that happened to this name a proposal".
    fn proposer_of(&self, name: &str) -> Result<Option<String>> {
        // Every row about this name in one read: they come back in sequence order, so the
        // last of the three that concern a proposal is the current state. Comparing
        // timestamps across three separate reads would be the same question asked worse —
        // a row carries no sequence number, and two rows can share a timestamp.
        let filter = AuditFilter {
            subject: Some(format!("note:{name}")),
            ..AuditFilter::default()
        };
        let rows = self.policy.audit().read(&filter)?;
        let last = rows.iter().rev().find(|e| {
            e.action == AuditAction::NoteProposed.as_str()
                || e.action == AuditAction::NoteProposalAccepted.as_str()
                || e.action == AuditAction::NoteProposalRejected.as_str()
        });
        Ok(match last {
            Some(e) if e.action == AuditAction::NoteProposed.as_str() => e
                .detail
                .get("by")
                .and_then(|v| v.as_str())
                .map(str::to_string),
            // Accepted, rejected, or never proposed: whatever is in `proposals/` under this
            // name now did not get there through `propose`.
            _ => None,
        })
    }

    /// Accept or reject a proposal.
    pub fn review(&self, req: ReviewRequest) -> Result<ReviewReport> {
        let note = self.store.read_proposal(&req.name)?;
        let name = note.front.name.clone();

        let Some(proposer) = self.proposer_of(&name)? else {
            return Err(Error::Config(format!(
                "the audit log has no record of {name} being proposed, so there is nobody to \
                 check this against. A file that appeared in proposals/ without going through \
                 `cyberbrain propose` is not a proposal; delete it or propose it properly"
            )));
        };
        // The hub's rule, in the same words, for the same reason (§14.1 of docs/HUB.md).
        if proposer == req.by {
            return Err(Error::PolicyRefusal {
                profile: "review".to_string(),
                reason: format!(
                    "a proposal cannot be reviewed by the person who made it. {name} was \
                     proposed by {proposer}"
                ),
            });
        }
        // 2026-09-25: `by` is a name anybody can type, so an agent that proposed a ring 0 note
        // could accept it under a second name and the note moved into notes/r0 — propose was
        // no gate at all. Deciding on a ring 0/1 proposal is the operator's, whichever way.
        self.refuse_resident_unless_operator(&name, note.front.ring, None, req.dry_run)?;

        let w = self.writers(req.dry_run);
        let policy = w.policy.get();

        if !req.accept {
            let reason = req.reason.trim();
            if reason.is_empty() {
                return Err(Error::Config(
                    "rejecting needs a reason: it is the only place the proposer will look".into(),
                ));
            }
            if !req.dry_run {
                self.store.remove_proposal(&name)?;
            }
            policy.audit().record(
                &self.actor,
                AuditAction::NoteProposalRejected,
                format!("note:{name}"),
                serde_json::json!({
                    "by": req.by, "proposed_by": proposer, "reason": reason,
                    "dry_run": req.dry_run,
                }),
            )?;
            return Ok(ReviewReport {
                name,
                accepted: false,
                by: req.by,
                proposed_by: proposer,
                reason: Some(reason.to_string()),
                path: None,
                blocks: 0,
                vectors: 0,
                dry_run: req.dry_run,
                audit_preview: w.policy.preview(),
            });
        }

        // A proposal written against a note that has moved on since would overwrite the
        // newer text. Same failure `expected_updated` exists to prevent, over a longer
        // interval: the review may be days after the proposal.
        if let Ok(existing) = self.store.read(&name)
            && existing.front.updated > note.front.created
            && !req.force
        {
            return Err(Error::Config(format!(
                "{name} changed after this was proposed ({} against {}); accepting would \
                 overwrite the newer text. Read both, then `--force` if the proposal is \
                 still right",
                existing.front.updated, note.front.created
            )));
        }

        // Scanned again over the same body: the proposal may have sat for a week and the
        // profile may have changed under it.
        let pii = match policy.check_write(&name, &note.body)? {
            WriteVerdict::Proceed { pii, .. } => pii,
            WriteVerdict::Held { findings } => {
                return Err(Error::PolicyRefusal {
                    profile: "pii".to_string(),
                    reason: format!(
                        "{name} holds {} possible personal data item(s) and cannot be accepted \
                         as it stands. Reject it with a reason; the proposer resolves it and \
                         proposes again",
                        findings.len()
                    ),
                });
            }
        };

        // The id is minted here, not at propose time: a citation must point at something
        // that exists, and until this moment nothing did.
        let now = jiff::Timestamp::now();
        let existing = self.store.read(&name).ok();
        let accepted = Note {
            front: Frontmatter {
                id: existing
                    .as_ref()
                    .map(|n| n.front.id)
                    .unwrap_or_else(NoteId::generate),
                created: existing.as_ref().map(|n| n.front.created).unwrap_or(now),
                updated: now,
                pii,
                ..note.front.clone()
            },
            body: note.body.clone(),
            path: PathBuf::new(),
        };
        let bytes = frontmatter::render(&accepted.front, &accepted.body)?.len();
        let ring = accepted.front.ring;

        let (path, blocks, vectors) = if req.dry_run {
            (self.store.note_path(ring, &name)?, 0, 0)
        } else {
            let path = w.notes.write(&accepted)?;
            let accepted = Note {
                path: path.clone(),
                ..accepted
            };
            policy.record_write(&accepted.front, bytes)?;
            let (blocks, _) = blocks_of(&accepted, MAX_BLOCK_TOKENS);
            let texts: Vec<&str> = blocks.iter().map(|b| b.text.as_str()).collect();
            self.declare_profile(&w)?;
            let vectors = self.embed_blocks(&texts)?;
            let outcome = w
                .index
                .upsert_note(&accepted, &blocks, vectors.as_deref())?;
            self.store.remove_proposal(&name)?;
            (path, outcome.blocks, outcome.vectors)
        };

        policy.audit().record(
            &self.actor,
            AuditAction::NoteProposalAccepted,
            format!("note:{name}"),
            serde_json::json!({
                "by": req.by, "proposed_by": proposer,
                "ring": ring.as_u8(), "bytes": bytes,
                "dry_run": req.dry_run,
            }),
        )?;

        Ok(ReviewReport {
            name,
            accepted: true,
            by: req.by,
            proposed_by: proposer,
            reason: None,
            path: Some(path),
            blocks,
            vectors,
            dry_run: req.dry_run,
            audit_preview: w.policy.preview(),
        })
    }

    // ----- erasure: one path -------------------------------------------------------------

    /// `forget`. The retention sweep and the API's `DELETE` reach the same eraser through
    /// `Policy`; this method is the operator's entry to it.
    pub fn forget(&self, target: &str, dry_run: bool) -> Result<ErasureReport> {
        let req = self.erase_request(target, EraseReason::OperatorForget, dry_run)?;
        let w = self.writers(dry_run);
        let mut eraser = StoreEraser {
            notes: w.notes.as_ref(),
            index: w.index.as_ref(),
        };
        let mut report = w.policy.get().forget(&mut eraser, &req)?;
        if dry_run {
            for a in w.policy.preview() {
                report
                    .notes
                    .push(format!("audit row a real run would append: {a}"));
            }
        }
        Ok(report)
    }

    fn erase_request(
        &self,
        target: &str,
        reason: EraseReason,
        dry_run: bool,
    ) -> Result<EraseRequest> {
        // Prefer the file (authoritative); fall back to the index for a note whose file is
        // already gone, so `forget` can still clean the rows up.
        let (id, name, ring, path) = match self.resolve_target(target) {
            Ok(n) => (n.front.id, n.front.name, n.front.ring, n.path),
            Err(Error::NoSuchNote(_)) => {
                let ix = lock_index(&self.index)?;
                let rec = match ix.note_by_name(target)? {
                    Some(r) => Some(r),
                    None => match NoteId::from_string(target) {
                        Ok(id) => ix.note(&id)?,
                        Err(_) => None,
                    },
                };
                let rec = rec.ok_or_else(|| Error::NoSuchNote(target.to_string()))?;
                (rec.front.id, rec.front.name, rec.front.ring, rec.path)
            }
            Err(e) => return Err(e),
        };
        Ok(EraseRequest {
            note_id: id,
            name,
            ring,
            path,
            reason,
            dry_run,
        })
    }

    // ----- doctor ------------------------------------------------------------------------

    pub fn doctor(&self) -> Result<DoctorReport> {
        let mut findings = Vec::new();
        let mut checks = Vec::new();
        let mut push = |severity: &'static str, check: &'static str, detail: String| {
            findings.push(DoctorFinding {
                severity,
                check,
                detail,
            })
        };

        checks.push("notes tree");
        let listing = self.store.list()?;
        for s in &listing.skipped {
            push(
                "warning",
                "notes tree",
                format!("{}: {}", Slash(&s.path), s.reason),
            );
        }

        checks.push("stale index");
        let (not_indexed, changed, gone, unreadable) = self.staleness()?;
        for u in unreadable {
            push(
                "error",
                "unreadable note",
                format!("{}: {}", Slash(&u.path), u.reason),
            );
        }
        if not_indexed + changed + gone > 0 {
            push(
                "warning",
                "stale index",
                format!(
                    "{not_indexed} notes not indexed, {changed} changed on disk since the last \
                     scan, {gone} indexed notes whose file is gone; run `cyberbrain scan`"
                ),
            );
        }

        // Two different facts, and lumping them together hides the actionable one. A link
        // to a name that could exist is intent: somebody will write that note. A link to a
        // name that can never be a note — underscores, capitals, a path — is a typo or a
        // leftover from another tool's naming, and no amount of writing notes will ever
        // resolve it. Reporting both as "does not exist (valid: it names intent)" tells the
        // operator to wait for something that is never coming.
        checks.push("dangling links");
        checks.push("unresolvable links");
        {
            let ix = lock_index(&self.index)?;
            for l in ix.dangling_links()? {
                let from = ix
                    .note(&l.from_note)?
                    .map(|r| r.front.name)
                    .unwrap_or_else(|| l.from_note.to_string());
                match cyberbrain_core::validate_name(&l.to_name) {
                    Ok(()) => push(
                        "warning",
                        "dangling links",
                        format!(
                            "{from} links to [[{}]] which does not exist yet (valid name: it names intent)",
                            l.to_name
                        ),
                    ),
                    Err(reason) => {
                        // Offer the name form when a note actually sits under it.
                        // Most of these come from another tool's file names, and the
                        // note the author meant is often already in the store.
                        let normalised = normalise_link_target(&l.to_name);
                        let hint = match ix.note_by_name(&normalised)? {
                            Some(_) => format!("; did you mean [[{normalised}]]?"),
                            None => String::new(),
                        };
                        push(
                            "warning",
                            "unresolvable links",
                            format!(
                                "{from} links to [[{}]], which can never resolve: a note name {reason}{hint}",
                                l.to_name
                            ),
                        )
                    }
                }
            }
        }

        // Notes that are mostly copies of each other. Recall already shows each text once;
        // this says where the copies are, so somebody can merge them. Measured on the
        // orderflow store: 19 % of all blocks had a textual twin in another note.
        checks.push("shared blocks");
        for sb in lock_index(&self.index)?.shared_blocks(SHARED_BLOCKS_PERCENT)? {
            push(
                "warning",
                "shared blocks",
                format!(
                    "{} shares {} of {} block(s) ({}%) with {}; consider merging them or linking \
                     one from the other",
                    sb.note,
                    sb.shared,
                    sb.of,
                    sb.shared * 100 / sb.of.max(1),
                    sb.other
                ),
            );
        }

        checks.push("ring cap");
        let resident = self.store.resident_tokens()?;
        let cap = self.store.resident_cap();
        if resident > cap {
            push(
                "error",
                "ring cap",
                format!(
                    "rings 0+1 hold ~{resident} tokens, over the cap of {cap}; a hand edit crossed it"
                ),
            );
        } else if resident * 10 >= cap * 8 {
            push(
                "warning",
                "ring cap",
                format!(
                    "rings 0+1 hold ~{resident} of {cap} tokens ({}%)",
                    resident * 100 / cap
                ),
            );
        }

        checks.push("index integrity");
        for p in lock_index(&self.index)?.integrity()? {
            push("error", "index integrity", p);
        }

        checks.push("embedding profile");
        let stored = lock_index(&self.index)?.embedding_profile()?;
        match (stored, self.model_identity()) {
            (Some(p), ModelIdentity::Ready(_)) => {
                if let Some(e) = self.profile_mismatch(&p) {
                    push("error", "embedding profile", e.to_string());
                }
            }
            (Some(p), ModelIdentity::Absent(reason)) => push(
                "warning",
                "embedding profile",
                format!(
                    "index vectors come from {} but no model is loaded ({reason}); semantic search is off",
                    p.id
                ),
            ),
            (None, ModelIdentity::Ready(_)) => {
                if lock_index(&self.index)?.stats()?.blocks > 0 {
                    push(
                        "warning",
                        "embedding profile",
                        "a model is present but the index holds no vectors; run `cyberbrain scan`"
                            .into(),
                    );
                }
            }
            (None, ModelIdentity::Absent(_)) => {}
        }

        checks.push("audit chain");
        if let Err(e) = self.policy.verify_audit() {
            push("error", "audit chain", e.to_string());
        }

        checks.push("retention");
        let (queue, _) = self.retention_queue()?;
        for i in &queue.items {
            if let cyberbrain_policy::RetentionStatus::Invalid { reason } = &i.status {
                push("warning", "retention", format!("{}: {reason}", i.name));
            }
        }
        if queue.due > 0 {
            push(
                "warning",
                "retention",
                format!(
                    "{} note(s) past their retention; nothing expires by itself, run `cyberbrain policy retention`",
                    queue.due
                ),
            );
        }

        Ok(DoctorReport {
            clean: findings.is_empty(),
            checks_run: checks,
            findings,
        })
    }

    // ----- status ------------------------------------------------------------------------

    pub async fn status(&self) -> Result<StatusReport> {
        let listing = self.store.list()?;
        let mut per_ring = [0usize; 5];
        for e in &listing.entries {
            per_ring[e.ring.as_u8() as usize] += 1;
        }
        let index = lock_index(&self.index)?.stats()?;
        let (not_indexed, changed, gone, _) = self.staleness()?;
        let embedder = self.embedder_summary();
        let matches_index = match (&index.embedding, self.model_identity()) {
            (Some(stored), ModelIdentity::Ready(_)) => {
                Some(self.profile_mismatch(stored).is_none())
            }
            _ => None,
        };
        let model_dir = self.config.model_dir();
        let inf = &self.config.inference;
        let inference = match self.llm().await {
            LlmState::Ready(c) => {
                let probe = c.probe().await;
                InferenceStatus {
                    endpoint: inf.base_url.clone(),
                    model: inf.model.clone(),
                    state: if probe.reachable {
                        "reachable".into()
                    } else {
                        "configured but unreachable".into()
                    },
                    probe: Some(probe),
                }
            }
            LlmState::Absent(reason) => InferenceStatus {
                endpoint: inf.base_url.clone(),
                model: inf.model.clone(),
                state: format!("not in use: {reason}"),
                probe: None,
            },
        };
        Ok(StatusReport {
            store: self.root.clone(),
            config: self.store.config_path(),
            notes_on_disk: listing.entries.len(),
            notes_per_ring: per_ring,
            files_skipped: listing.skipped.len(),
            resident_tokens: self.store.resident_tokens()?,
            resident_cap: self.store.resident_cap(),
            index_stale: not_indexed + changed + gone > 0,
            index,
            audit: AuditSummary {
                path: self.root.join(AUDIT_DB_FILE),
                rows: self.audit_sink.count()?,
                schema_version: self.audit_sink.schema_version()?,
                chain: self.policy.verify_audit().map_err(|e| e.to_string()),
            },
            embedding: EmbeddingStatus {
                manifest_present: model_dir.join(MANIFEST_FILE).is_file(),
                model_dir,
                embedder,
                index_profile: lock_index(&self.index)?.embedding_profile()?,
                matches_index,
            },
            inference,
            policy: self.policy.status(),
        })
    }

    // ----- find --------------------------------------------------------------------------

    /// The tree `find` scans. The store lives at `<project>/.cyberbrain` by default
    /// (SPEC §4), so the project is the store's parent. A store placed elsewhere with
    /// `--store` says nothing about where the code is; the working directory is the
    /// only other candidate and it is what an agent's hooks run in. The report names
    /// the root it used either way, so the caller can see which tree answered.
    fn code_root(&self) -> PathBuf {
        let is_default_store = self
            .root
            .file_name()
            .is_some_and(|n| n == DEFAULT_STORE_DIR);
        match (is_default_store, self.root.parent()) {
            (true, Some(parent)) if !parent.as_os_str().is_empty() => parent.to_path_buf(),
            (true, Some(_)) => PathBuf::from("."),
            _ => std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")),
        }
    }

    /// `find <symbol>` (SPEC §10): line ranges of the definitions of `symbol` in the
    /// project tree, so the agent reads a slice instead of a file.
    ///
    /// Synchronous and index-free: hooks and MCP call it and neither has a runtime to
    /// spare, and a scan of the tree as it is now cannot hand back a stale range. The
    /// store itself is excluded from the walk; its notes belong to `recall`. Nothing in
    /// `cyberbrain.toml` configures the code index yet (see the report), so the crate's
    /// defaults apply: `.cyberbrainignore` and `.gitignore` honoured, hidden entries
    /// skipped, files over 1 MiB not read.
    pub fn find(&self, symbol: &str, limit: usize) -> Result<FindReport> {
        let opts = cyberbrain_code::FindOptions {
            exclude: vec![self.root.clone()],
            ..cyberbrain_code::FindOptions::default()
        };
        let result = cyberbrain_code::find(&self.code_root(), symbol, limit, &opts)?;
        let report = FindReport::from(result);
        self.record_find_usage(&report);
        Ok(report)
    }

    /// Ledger row for one find, counted in lines: the spans the caller is told to read
    /// against the length of the files they sit in. The files were just walked, so reading
    /// their line count back costs a warm read of at most `limit` files; a file that cannot
    /// be read is left out of both sides rather than counted as free.
    fn record_find_usage(&self, report: &FindReport) {
        if report.hits.is_empty() {
            return;
        }
        let mut returned = 0u64;
        let mut full = 0u64;
        let mut counted: std::collections::BTreeSet<&str> = std::collections::BTreeSet::new();
        for h in &report.hits {
            returned += u64::from(h.end_line.saturating_sub(h.start_line)) + 1;
            if counted.insert(h.path.as_str()) {
                match std::fs::read_to_string(report.root.join(&h.path)) {
                    Ok(text) => full += text.lines().count() as u64,
                    Err(_) => {
                        counted.remove(h.path.as_str());
                    }
                }
            }
        }
        if full == 0 {
            return;
        }
        self.usage().append(&usage::UsageRow {
            at: usage::now(),
            op: "find".into(),
            unit: "lines".into(),
            returned,
            full,
            hits: report.hits.len() as u64,
            sources: counted.len() as u64,
        });
    }

    // ----- policy ------------------------------------------------------------------------

    pub fn policy_egress(&self) -> Vec<EgressEntry> {
        self.policy.egress().register()
    }

    /// What the active profile claims about the law, with the basis and how sure the author
    /// is of each line.
    ///
    /// The catalogue has existed since the profiles did and was read by nothing but a test:
    /// a compliance claim nobody can print is a comment. Printing it is also the honest
    /// move, because roughly a third of the lines carry a confidence below `high`, and a
    /// reader who cannot see that gap will assume there is none.
    pub fn policy_obligations(&self) -> ObligationsView {
        let profile = self.config.policy.profile;
        ObligationsView {
            profile,
            law: profile.law().to_string(),
            obligations: profile.obligations(),
        }
    }

    /// The audit log, rendered. Reading through `export` records the export itself
    /// (SPEC §12.6), after the rows were rendered.
    /// Point this store at a hub, from an invitation. Returns the inference endpoint if the
    /// invitation carried one.
    pub fn enrol_with_hub(
        &self,
        hub_url: &str,
        device: &str,
        inference_url: Option<&str>,
    ) -> Result<Option<String>> {
        let path = self.store.config_path();
        let text = std::fs::read_to_string(&path).map_err(|e| Error::Io {
            path: path.clone(),
            source: e,
        })?;
        let mut updated = crate::hub::client::set_hub_in_config(&text, hub_url, device);
        if let Some(url) = inference_url {
            updated = crate::hub::client::set_inference_url(&updated, url);
        }
        // Parsed before it is written: a config file this command broke would leave the
        // store unusable, and the person would have no idea what changed.
        cyberbrain_core::config::Config::parse(&updated, &self.root)
            .map_err(|e| Error::Config(format!("enrolment would break the config file: {e}")))?;
        std::fs::write(&path, updated).map_err(|e| Error::Io {
            path: path.clone(),
            source: e,
        })?;
        Ok(inference_url.map(str::to_owned))
    }

    /// Enrol with a fleet invitation: ask the hub it names for a device of this store's own.
    ///
    /// The gate for this one call gets the invitation's address as its hub, because the store
    /// has none yet, and nothing is written here. The caller stores the token and the config
    /// only once the hub has answered, so a refused or unreachable enrolment leaves nothing but
    /// the audit rows of the attempt.
    pub async fn enrol_with_fleet_invitation(
        &self,
        inv: &crate::hub::client::FleetInvitation,
        machine: &str,
    ) -> Result<crate::hub::client::Enrolled> {
        let mut cfg = self.policy.config().clone();
        cfg.hub_endpoint = Some(inv.hub_url.clone());
        let gate = cyberbrain_policy::Egress::new(
            cfg,
            self.policy.audit().clone(),
            self.policy.actor().clone(),
        );
        crate::hub::client::enrol_at_hub(
            &gate,
            self.policy.actor(),
            inv,
            machine,
            &self.project_label(),
        )
        .await
    }

    /// This store's project as a device name on a hub shows it: the folder the store sits in.
    pub fn project_label(&self) -> String {
        self.root
            .parent()
            .and_then(|p| p.file_name())
            .or_else(|| self.root.file_name())
            .and_then(|f| crate::hub::store::project_label(&f.to_string_lossy()))
            .unwrap_or_else(|| "store".to_string())
    }

    /// What this machine is, as far as a company hub is concerned.
    ///
    /// The dashboard's own question. Somebody looking at their notes cannot tell from that
    /// screen whether this machine reports to anybody, and "check with `cyberbrain hub
    /// push`" is not an answer for the person the delivery is *about*.
    ///
    /// The last delivery is read out of this store's own audit log, not from a note kept on
    /// the side: the audit log is what was actually sent, so the two cannot disagree.
    pub fn hub_status(&self) -> Result<serde_json::Value> {
        let Some(url) = self.config.hub.url.clone() else {
            return Ok(serde_json::json!({ "enrolled": false }));
        };
        let filter = cyberbrain_policy::AuditFilter {
            action: Some("egress.completed".into()),
            contains: Some("audit-sync".into()),
            ..Default::default()
        };
        let last = self
            .policy
            .audit()
            .read(&filter)
            .ok()
            .and_then(|rows| rows.last().map(|e| e.ts.to_string()));
        Ok(serde_json::json!({
            "enrolled": true,
            "hub": url,
            "device": self.config.hub.device,
            "last_delivery": last,
        }))
    }

    /// Deliver audit rows to the hub this store was enrolled with.
    ///
    /// Returns the report and the exit code. A hub that is not collecting, and a gap it can
    /// still close, are **not** failures of this command: they are states a timer should see
    /// and carry on from. Only something the operator has to fix exits non-zero.
    /// Offer this store's notes to the hub.
    ///
    /// The selection lives here and only here: rings 2 to 4, and only notes that carry a
    /// bereich. Rings 0 and 1 are not filtered out as a courtesy — they are not eligible,
    /// and the hub refuses them again on arrival. A reviewer who wants to know what this
    /// machine offers reads this function and is done.
    /// If this note carries a bereich and this store delivers to a hub, the pieces needed
    /// to say so. Returns nothing when there is nothing to warn about, so the caller has no
    /// condition of its own to get wrong.
    pub fn shared_copy_hint(&self, target: &str) -> Option<(String, String, String)> {
        let hub = self.config.hub.url.clone()?;
        let note = self.store.read(target).ok()?;
        let bereich = note.front.bereich.clone()?;
        Some((bereich, note.front.name.clone(), hub))
    }

    /// The sentence `forget` owes somebody when the note it just erased was shared.
    ///
    /// Worded once and used by every surface that erases. It was only ever printed by the
    /// CLI, so the same erasure through the web page or the API said "removed: file,
    /// blocks, vectors, fts_rows" and nothing about the copy on the hub — a report that is
    /// accurate about this disk and reads as if it were about the note.
    ///
    /// Has to be called before the erasure: afterwards there is nothing left to ask.
    pub fn shared_copy_sentence(&self, target: &str) -> Option<String> {
        let (bereich, name, hub) = self.shared_copy_hint(target)?;
        Some(format!(
            concat!(
                "This note was in bereich {b} and this store is enrolled with {h}. The hub ",
                "may hold a copy, and erasing it there is a separate step: ",
                "cyberbrain hub erase {n} --bereich {b}"
            ),
            b = bereich,
            h = hub,
            n = name
        ))
    }

    /// What a pull does with the hub's answer: everything except fetching it and moving
    /// the cursor. Separate so it can be tested without a hub at the other end of a socket.
    pub(crate) fn apply_pulled(
        &self,
        notes: &[serde_json::Value],
        erased: &[serde_json::Value],
        apply_erasures: bool,
        dry_run: bool,
    ) -> Result<Pulled> {
        let mut written = Vec::new();
        let mut kept_local = Vec::new();
        let mut refused = Vec::new();
        let mut flagged: Vec<String> = Vec::new();
        let mut retry = false;
        for n in notes {
            let name = n["name"].as_str().unwrap_or_default().to_string();
            let ring_u8 = n["ring"].as_u64().unwrap_or(9) as u8;
            // The hub never holds rings 0 or 1, and this checks anyway. A store that trusts
            // its hub about which ring something is has handed over its invariants.
            if !(2..=4).contains(&ring_u8) {
                refused.push(serde_json::json!({
                    "name": name,
                    "why": format!("ring {ring_u8} may not arrive over the network"),
                }));
                continue;
            }
            let incoming_updated = n["updated"].as_str().unwrap_or_default();
            if let Ok(local) = self.store.read(&name) {
                let local_updated = local.front.updated.to_string();
                if local_updated.as_str() > incoming_updated {
                    kept_local.push(serde_json::json!({
                        "name": name,
                        "local": local_updated,
                        "offered": incoming_updated,
                    }));
                    continue;
                }
                if local_updated == incoming_updated {
                    continue;
                }
            }
            // The hub carries the sender's frontmatter verbatim, and until now the pull
            // read five fields out of the wire note and rebuilt the rest. Everything it
            // rebuilt was wrong in the same direction: a new id (so a citation from the
            // other machine resolved nowhere), `created` of now (so a retention clock
            // started over), no tags, no retention (so an agreed deletion date quietly
            // stopped applying) and `updated` of now.
            let sent_text = n["frontmatter"].as_str().unwrap_or_default();
            let sent = frontmatter::parse(std::path::Path::new(&name), sent_text)
                .ok()
                .map(|p| p.front);
            let incoming_id = sent
                .as_ref()
                .map(|f| f.id)
                .unwrap_or_else(cyberbrain_core::NoteId::generate);
            let incoming_stamp = match incoming_updated.parse::<jiff::Timestamp>() {
                Ok(t) => t,
                // A note whose stamp we cannot read is not a note we can order against the
                // local one, and taking it with a stamp of our own is what caused the
                // trouble. Refused and named, rather than taken on a guess.
                Err(e) => {
                    refused.push(serde_json::json!({
                        "name": name,
                        "why": format!("its `updated` is not a timestamp: {e}"),
                    }));
                    continue;
                }
            };
            let incoming_created = sent.as_ref().map(|f| f.created).unwrap_or(incoming_stamp);
            let incoming_tags = sent.as_ref().map(|f| f.tags.clone()).unwrap_or_default();
            let incoming_retention = sent.as_ref().and_then(|f| f.retention.clone());
            // Validity travels with the note: an expiry declared on one machine that
            // arrived here without it would make this machine's recall say the opposite.
            let incoming_validity = sent.as_ref().map(|f| (f.valid_from, f.invalid_at));

            let taken = serde_json::json!({
                "name": name,
                "updated": incoming_updated,
                "retention": incoming_retention,
                "tags": incoming_tags,
            });
            if !dry_run {
                let req = WriteRequest {
                    ring: Ring::try_from(ring_u8)?,
                    kind: serde_json::from_value(n["kind"].clone())
                        .unwrap_or(cyberbrain_core::NoteKind::Knowledge),
                    name: name.clone(),
                    body: n["body"].as_str().unwrap_or_default().to_string(),
                    tags: Vec::new(),
                    bereich: Some(n["bereich"].as_str().map(str::to_string)),
                    retention: None,
                    force: true,
                    choice: None,
                    expected_updated: None,
                    supersedes: None,
                    valid_from: incoming_validity.map(|v| v.0),
                    invalid_at: incoming_validity.map(|v| v.1),
                    arriving: Some(Arriving {
                        id: incoming_id,
                        created: incoming_created,
                        updated: incoming_stamp,
                        tags: incoming_tags,
                        retention: incoming_retention,
                    }),
                    dry_run: false,
                };
                // `force` is what makes an unattended pull possible at all: a hold waits for
                // an operator, and there is none at the other end of a timer. What it must
                // not do is pass in silence — a note can arrive here carrying personal data
                // that this machine's own gate would have stopped, and the person running
                // the pull is the one who has to know it landed.
                match self.write(req) {
                    Ok(WriteOutcome::Written(w)) => {
                        if w.pii == cyberbrain_core::PiiState::Flagged {
                            flagged.push(name.clone());
                        }
                    }
                    Ok(_) => {}
                    // A note this machine cannot write is named and left out, and the rest of
                    // the delivery still arrives. Returning here, as it did, stopped every
                    // note after it, and the next pull stopped at the same place, for good:
                    // what an older machine meets the first time a newer one shares a name
                    // it cannot read. The cursor is held back, so the note comes again and
                    // lands once this machine can take it.
                    Err(e) => {
                        refused.push(serde_json::json!({
                            "name": name,
                            "why": e.to_string(),
                        }));
                        retry = true;
                        continue;
                    }
                }
            }
            written.push(taken);
        }

        let mut erasures = Vec::new();
        for e in erased {
            let name = e["name"].as_str().unwrap_or_default().to_string();
            let exists = self.store.read(&name).is_ok();
            erasures.push(serde_json::json!({
                "name": name,
                "erased_at": e["erased_at"],
                "held_here": exists,
                "removed": apply_erasures && exists && !dry_run,
            }));
            if apply_erasures && exists && !dry_run {
                // The same path `cyberbrain forget` takes, so a distributed erasure is not a
                // second, weaker way of deleting things.
                self.forget(&name, false)?;
            }
        }

        Ok(Pulled {
            written,
            kept_local,
            refused,
            flagged,
            erasures,
            retry,
        })
    }

    /// Take what the hub has for this machine.
    ///
    /// The hub decided what this device may see; this decides what to keep, and it is
    /// deliberately timid. A note that changed here since the last pull is never overwritten
    /// — it is reported and left alone. Erasures are reported too and only acted on when
    /// asked, because deleting a local file on the strength of a network message is not
    /// something to do quietly.
    pub async fn pull_notes_from_hub(
        &self,
        apply_erasures: bool,
        dry_run: bool,
    ) -> Result<(serde_json::Value, i32)> {
        use crate::hub::client;
        let hub_url = self
            .config
            .hub
            .url
            .clone()
            .ok_or_else(|| Error::Config("this store is not enrolled with a hub".into()))?;
        let token = client::token_for(&hub_url, self.config.hub.device.as_deref())?;
        let pin = client::pin_for(&hub_url);
        let since = client::read_cursor(&hub_url, self.config.hub.device.as_deref());
        let answer = client::fetch_from_hub(
            self.policy.egress(),
            &cyberbrain_policy::Actor::Operator,
            &hub_url,
            &token,
            pin.as_deref(),
            since.as_deref(),
        )
        .await?;

        let notes = answer
            .get("notes")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();
        let erased = answer
            .get("erased")
            .and_then(|v| v.as_array())
            .cloned()
            .unwrap_or_default();

        let Pulled {
            written,
            kept_local,
            refused,
            flagged,
            erasures,
            retry,
        } = self.apply_pulled(&notes, &erased, apply_erasures, dry_run)?;

        // An erasure this machine was told about but did not act on must come round again.
        // Advancing the cursor past it would mean the only warning was the one nobody
        // acted on, and the copy stays here for good.
        let unfinished = erasures.iter().any(|e| {
            e["held_here"].as_bool().unwrap_or(false) && !e["removed"].as_bool().unwrap_or(false)
        });
        // The same for a note this machine could not write: it has to be offered again.
        let unfinished = unfinished || retry;
        if !dry_run {
            // Everything the hub just showed us is, by definition, what it holds.
            let mut known = client::read_known(&hub_url, self.config.hub.device.as_deref());
            for n in &notes {
                if let (Some(name), Some(u)) = (n["name"].as_str(), n["updated"].as_str()) {
                    known.insert(name.to_string(), u.to_string());
                }
            }
            let _ = client::write_known(&hub_url, self.config.hub.device.as_deref(), &known);
            if let Some(c) = answer.get("cursor").and_then(|v| v.as_str())
                && !unfinished
            {
                client::write_cursor(&hub_url, self.config.hub.device.as_deref(), c)?;
            }
        }

        let v = serde_json::json!({
            "hub": hub_url,
            "since": since,
            "written": written,
            "kept_local": kept_local,
            "refused": refused,
            "flagged": flagged,
            "erasures": erasures,
            "cursor_held_back": unfinished,
            "message": format!(
                "{} note(s) taken, {} kept because this machine changed them since, {} \
                 refused, {} erasure(s) reported{}{}",
                written.len(),
                kept_local.len(),
                refused.len(),
                erasures.len(),
                if flagged.is_empty() {
                    String::new()
                } else {
                    // Named, not counted: somebody has to be able to go and look at them.
                    format!(
                        ".\n{} note(s) arrived carrying personal data and were written \
                         flagged, because a hold has nobody to ask at this end: {}",
                        flagged.len(),
                        flagged.join(", ")
                    )
                },
                if unfinished {
                    ". The position was not advanced: an erasure is still waiting here, and \
                     it has to come round again"
                } else {
                    ""
                }
            ),
        });
        Ok((v, 0))
    }

    /// Ask the hub to erase its copy.
    pub async fn erase_at_hub(
        &self,
        bereich: &str,
        name: &str,
    ) -> Result<(serde_json::Value, i32)> {
        use crate::hub::client::{self, Reply};
        let hub_url = self
            .config
            .hub
            .url
            .clone()
            .ok_or_else(|| Error::Config("this store is not enrolled with a hub".into()))?;
        let token = client::token_for(&hub_url, self.config.hub.device.as_deref())?;
        let pin = client::pin_for(&hub_url);
        let reply = client::erase_at_hub(
            self.policy.egress(),
            &cyberbrain_policy::Actor::Operator,
            &hub_url,
            &token,
            pin.as_deref(),
            bereich,
            name,
        )
        .await?;
        let mut v = serde_json::json!({ "bereich": bereich, "name": name, "hub": hub_url });
        let code = match reply {
            Reply::Ok(d) => {
                v["notes_removed"] = serde_json::json!(d.accepted);
                v["conflict_rows_removed"] = serde_json::json!(d.total_rows);
                v["message"] = serde_json::json!(format!(
                    "{name} erased at {hub_url}: {} note row(s) and {} conflict row(s) removed",
                    d.accepted, d.total_rows
                ));
                0
            }
            Reply::NotCollecting(m) => {
                v["message"] = serde_json::json!(m);
                1
            }
            Reply::Gap { expected } => {
                v["message"] = serde_json::json!(format!("unexpected: {expected}"));
                1
            }
            Reply::Refused { status, message } => {
                v["message"] = serde_json::json!(format!("refused ({status}): {message}"));
                1
            }
        };
        Ok((v, code))
    }

    pub async fn push_notes_to_hub(
        &self,
        bereich: Option<&str>,
        dry_run: bool,
    ) -> Result<(serde_json::Value, i32)> {
        use crate::hub::client::{self, Reply};

        // Not required for a dry run. "What would I be giving away?" is a question to answer
        // before enrolling, not after — refusing to answer it until a hub is configured gets
        // the decision made in the wrong order.
        let hub_url = self.config.hub.url.clone();

        let known = match &hub_url {
            Some(u) => client::read_known(u, self.config.hub.device.as_deref()),
            None => Default::default(),
        };
        let mut offered = Vec::new();
        let mut skipped_no_bereich = 0usize;
        let mut skipped_resident = 0usize;
        for entry in self.store.list()?.entries {
            let note = self.store.read(&entry.name)?;
            let f = &note.front;
            if matches!(f.ring, Ring::Invariant | Ring::Protocol) {
                skipped_resident += 1;
                continue;
            }
            let Some(b) = f.bereich.as_deref() else {
                skipped_no_bereich += 1;
                continue;
            };
            if let Some(want) = bereich
                && want != b
            {
                continue;
            }
            offered.push(serde_json::json!({
                "id": f.id.to_string(),
                "name": f.name,
                "ring": f.ring.as_u8(),
                "kind": kind_name(f.kind),
                "bereich": b,
                "updated": f.updated.to_string(),
                "frontmatter": cyberbrain_core::frontmatter::render(f, "")?,
                "body": note.body,
                // What this store last knew the hub to hold for this note. Without it every
                // delivery from a second machine looks like a collision.
                "based_on": known.get(&f.name),
            }));
        }

        let summary = serde_json::json!({
            "hub": hub_url.clone(),
            "offered": offered.len(),
            "skipped_without_bereich": skipped_no_bereich,
            "skipped_rings_0_1": skipped_resident,
        });
        if dry_run {
            let mut v = summary;
            v["message"] = serde_json::json!(format!(
                "{} note(s) would be offered to {}. {} carry no bereich and {} are ring 0 or 1, \
                 which never leave this machine.",
                v["offered"],
                hub_url
                    .as_deref()
                    .unwrap_or("a hub, once this store is enrolled with one"),
                skipped_no_bereich,
                skipped_resident
            ));
            return Ok((v, 0));
        }

        let hub_url = hub_url.ok_or_else(|| {
            Error::Config(
                "this store is not enrolled with a hub; run `cyberbrain hub enrol <invitation>`"
                    .into(),
            )
        })?;
        let token = client::token_for(&hub_url, self.config.hub.device.as_deref())?;
        let version = env!("CARGO_PKG_VERSION");
        let egress = self.policy.egress();
        let actor = cyberbrain_policy::Actor::Operator;
        let pin = client::pin_for(&hub_url);
        let batch = serde_json::json!({ "notes": offered }).to_string();
        let reply = client::deliver_notes(
            egress,
            &actor,
            &hub_url,
            &token,
            pin.as_deref(),
            version,
            batch,
        )
        .await?;

        let mut v = summary;
        let code = match reply {
            Reply::Ok(d) => {
                v["accepted"] = serde_json::json!(d.accepted);
                v["stored"] = serde_json::json!(d.total_rows);
                v["conflicts"] = serde_json::json!(d.conflicts);
                // Only what the hub actually took counts as known. A note it turned away is
                // one this store still has no agreed version of.
                let mut known = client::read_known(&hub_url, self.config.hub.device.as_deref());
                for n in &offered {
                    let name = n["name"].as_str().unwrap_or_default();
                    let conflicted = d.conflicts.iter().any(|c| c["name"].as_str() == Some(name));
                    if !conflicted && let Some(u) = n["updated"].as_str() {
                        known.insert(name.to_string(), u.to_string());
                    }
                }
                let _ = client::write_known(&hub_url, self.config.hub.device.as_deref(), &known);
                v["message"] = serde_json::json!(if d.conflicts.is_empty() {
                    format!(
                        "{} of {} note(s) accepted by {}, {} newer than what it held",
                        d.accepted, v["offered"], hub_url, d.total_rows
                    )
                } else {
                    format!(
                        "{} of {} note(s) accepted by {}, {} stored, {} in conflict: \
                         another machine changed them. Nothing was overwritten — \
                         `cyberbrain hub pull` to see the other version.",
                        d.accepted,
                        v["offered"],
                        hub_url,
                        d.total_rows,
                        d.conflicts.len()
                    )
                });
                if d.conflicts.is_empty() { 0 } else { 1 }
            }
            Reply::NotCollecting(m) => {
                v["message"] = serde_json::json!(format!("the hub is not collecting: {m}"));
                0
            }
            Reply::Gap { expected } => {
                v["message"] = serde_json::json!(format!("the hub expected {expected}"));
                1
            }
            Reply::Refused { status, message } => {
                v["message"] = serde_json::json!(format!("refused ({status}): {message}"));
                1
            }
        };
        Ok((v, code))
    }

    pub async fn push_to_hub(
        &self,
        since: Option<jiff::Timestamp>,
    ) -> Result<(serde_json::Value, i32)> {
        use crate::hub::client::{self, Reply};

        let hub_url = self.config.hub.url.clone().ok_or_else(|| {
            Error::Config(
                "this store is not enrolled with a hub; run `cyberbrain hub enrol <invitation>`"
                    .into(),
            )
        })?;
        let token = client::token_for(&hub_url, self.config.hub.device.as_deref())?;
        let version = env!("CARGO_PKG_VERSION");

        let filter = cyberbrain_policy::AuditFilter {
            since,
            ..Default::default()
        };
        let bundle = self.export_audit_bundle(&filter)?;

        let egress = self.policy.egress();
        let actor = cyberbrain_policy::Actor::Operator;
        // The pin, like the token, is what this machine was enrolled with: both are read
        // here, and the delivery itself decides nothing about who it trusts.
        let pin = client::pin_for(&hub_url);
        let reply = client::deliver(
            egress,
            &actor,
            &hub_url,
            &token,
            pin.as_deref(),
            version,
            bundle,
        )
        .await?;

        Ok(match reply {
            Reply::Ok(d) => (
                serde_json::json!({
                    "state": "delivered",
                    "accepted": d.accepted,
                    "total_rows": d.total_rows,
                    "hub": d.hub,
                    "message": format!(
                        "delivered {} new row(s) to {}; the hub now holds {}",
                        d.accepted, d.hub, d.total_rows
                    ),
                }),
                0,
            ),
            Reply::NotCollecting(m) => (
                serde_json::json!({
                    "state": "not-collecting",
                    "message": format!("{m}\nNothing was lost; this store keeps its rows."),
                }),
                0,
            ),
            Reply::Gap { expected } => (
                serde_json::json!({
                    "state": "gap",
                    "expected_anchor": expected,
                    "message": format!(
                        "the hub is at {} and this delivery did not reach back that far. \
                         Send a wider period: `cyberbrain hub push` without --since covers \
                         everything.",
                        &expected[..expected.len().min(12)]
                    ),
                }),
                0,
            ),
            Reply::Refused { status, message } => (
                serde_json::json!({
                    "state": "refused",
                    "status": status,
                    "message": format!("the hub refused the delivery ({status}): {message}"),
                }),
                1,
            ),
        })
    }

    /// A period of the log as a self-checking bundle, for somebody outside to verify.
    ///
    /// The tool string goes in the header for the reader's benefit; nothing in the check
    /// depends on it, which is the point — a file that only this version can verify would
    /// not survive the retention period it exists for.
    pub fn export_audit_bundle(&self, filter: &AuditFilter) -> Result<String> {
        let tool = concat!("cyberbrain ", env!("CARGO_PKG_VERSION"));
        self.policy.export_audit_bundle(filter, tool)
    }

    pub fn policy_audit(
        &self,
        filter: &AuditFilter,
        verify: bool,
        format: ExportFormat,
    ) -> Result<AuditView> {
        let verified = verify.then(|| self.policy.verify_audit().map_err(|e| e.to_string()));
        let rows = self.policy.audit().read(filter)?.len();

        // An `--action` that matches nothing is refused, not answered with an empty table.
        // In an audit tool the two readings are opposite: "no such action name" and
        // "nothing of that kind ever happened". Someone checking whether erasures occurred
        // types the obvious name, gets zero rows and concludes the wrong thing. So when a
        // filter selects nothing out of a non-empty log, say which actions are actually in
        // it — measured from the log rather than a hard-coded list, because the binary and
        // the index write names the policy vocabulary does not contain.
        if rows == 0 && filter.action.is_some() {
            let all = self.policy.audit().read(&AuditFilter::default())?;
            if !all.is_empty() {
                let wanted = filter.action.as_deref().unwrap_or_default();
                let mut present: Vec<String> = all
                    .iter()
                    .map(|e| e.action.clone())
                    .collect::<std::collections::BTreeSet<_>>()
                    .into_iter()
                    .collect();
                present.sort();
                if !present
                    .iter()
                    .any(|a| a == wanted || a.starts_with(&format!("{wanted}.")))
                {
                    return Err(Error::Config(format!(
                        "no audit action named {wanted:?}; the log contains: {}. Refused rather than answered with an empty table: a missing name and a thing that never happened are different answers",
                        present.join(", ")
                    )));
                }
            }
        }

        let rendered = self.policy.export_audit(filter, format)?;
        Ok(AuditView {
            rows,
            verified,
            rendered,
        })
    }

    pub fn policy_subject(&self, identifier: &str) -> Result<SubjectAccessReport> {
        let source = IndexSubjectSource { index: &self.index };
        self.policy.subject_access(&source, identifier)
    }

    /// Every note on disk with its frontmatter, for retention. Files are authoritative,
    /// so this reads the tree rather than the index.
    fn retention_queue(&self) -> Result<(RetentionQueue, Vec<String>)> {
        let listing = self.store.list()?;
        let mut notes = Vec::new();
        let mut unreadable = Vec::new();
        for e in &listing.entries {
            match self.store.read_path(&e.path) {
                Ok(n) => notes.push((n.front, n.path)),
                Err(err) => unreadable.push(format!("{}: {err}", Slash(&e.path))),
            }
        }
        let queue = self
            .policy
            .retention_queue(notes.iter().map(|(f, p)| (f, p.as_path())));
        Ok((queue, unreadable))
    }

    pub fn policy_retention(&self, apply: bool, dry_run: bool) -> Result<RetentionReport> {
        let (queue, unreadable) = self.retention_queue()?;
        let mut report = RetentionReport {
            queue,
            unreadable,
            applied_run: apply,
            dry_run,
            applied: Vec::new(),
            audit_preview: Vec::new(),
        };
        if !apply {
            return Ok(report);
        }
        let w = self.writers(dry_run);
        let mut eraser = StoreEraser {
            notes: w.notes.as_ref(),
            index: w.index.as_ref(),
        };
        for (item, result) in w
            .policy
            .get()
            .apply_retention(&mut eraser, &report.queue, dry_run)
        {
            report.applied.push(RetentionOutcome {
                name: item.name.clone(),
                item,
                result: result.map_err(|e| e.to_string()),
            });
        }
        report.audit_preview = w.policy.preview();
        Ok(report)
    }

    pub fn policy_model_card(&self) -> ModelCardReport {
        let cards = self.policy.model_cards(&[self]);
        let mut absent = Vec::new();
        if let EmbedderState::Absent { reason } = self.embedder() {
            absent.push(format!("embedding model: {reason}"));
        }
        if self.config.inference.model.is_none() {
            absent.push(
                "inference model: none configured (inference.model); the endpoint is not contacted"
                    .into(),
            );
        }
        ModelCardReport { cards, absent }
    }

    /// Persist consent in `cyberbrain.toml` (SPEC §12.1: consent lives in the file, not in
    /// memory). Edits the one key in place so the operator's comments survive.
    pub fn policy_consent(&self, grant: bool) -> Result<ConsentReport> {
        let path = self.store.config_path();
        let text = match std::fs::read_to_string(&path) {
            Ok(t) => t,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                cyberbrain_core::config::DEFAULT_TOML.to_string()
            }
            Err(e) => return Err(Error::Io { path, source: e }),
        };
        let new_line = format!("model_download_consent = {grant}");
        let mut lines: Vec<String> = text.lines().map(str::to_owned).collect();
        let mut in_embedding = false;
        let mut replaced = false;
        let mut embedding_header: Option<usize> = None;
        for (i, line) in lines.iter_mut().enumerate() {
            let t = line.trim();
            if t.starts_with('[') {
                in_embedding = t == "[embedding]";
                if in_embedding {
                    embedding_header = Some(i);
                }
                continue;
            }
            if in_embedding
                && !t.starts_with('#')
                && t.split('=').next().map(str::trim) == Some("model_download_consent")
            {
                *line = new_line.clone();
                replaced = true;
            }
        }
        if !replaced {
            match embedding_header {
                Some(i) => lines.insert(i + 1, new_line),
                None => {
                    lines.push(String::new());
                    lines.push("[embedding]".into());
                    lines.push(new_line);
                }
            }
        }
        let mut out = lines.join("\n");
        out.push('\n');
        // Validate before writing: a config that no longer parses is worse than no consent.
        let parsed = Config::parse(&out, &path)?;
        write_atomic(&path, out.as_bytes())?;
        let mut warnings = Vec::new();
        if parsed.embedding.model_source.is_none() {
            warnings.push(
                "embedding.model_source is unset, so nothing can be downloaded regardless of \
                 consent; the artefact has to be placed by hand"
                    .into(),
            );
        }
        warnings.push(
            "takes effect on the next command; this process keeps the register it started with"
                .into(),
        );
        self.policy.audit().record_raw(
            &self.actor.to_string(),
            "consent.model-download",
            "model-download",
            json!({ "consent": grant, "model_source": parsed.embedding.model_source }),
        )?;
        Ok(ConsentReport {
            path,
            consent: grant,
            model_source: parsed.embedding.model_source,
            warnings,
        })
    }
}

/// SPEC §12.7: what this binary can vouch for about the models it uses. Fields it did not
/// read from the artefact stay `None` and print as *not stated*.
impl ModelInventory for App {
    fn model_cards(&self) -> Vec<ModelCard> {
        let mut cards = Vec::new();
        if let EmbedderState::Loaded {
            embedder,
            manifest,
            paths,
        } = self.embedder()
        {
            let info = embedder.info();
            let dir = self.config.model_dir();
            let mut c = ModelCard::new(
                ModelRole::Embedding,
                dir.file_name()
                    .map(|n| n.to_string_lossy().into_owned())
                    .unwrap_or_else(|| "model2vec".into()),
                self.config
                    .embedding
                    .model_source
                    .clone()
                    .unwrap_or_else(|| "placed by hand (no model_source configured)".into()),
                "static token embeddings for hybrid recall over the notes in this store",
            );
            c.blake3 = Some(manifest.weights_blake3.clone());
            c.hash_verified = Some(true);
            c.dimension = Some(info.dim);
            c.pooling = Some(info.pooling.to_string());
            c.format = Some(format!(
                "model2vec: safetensors [{} x {}] {} + tokenizer.json (blake3 {})",
                info.vocab_rows, info.dim, info.weights_dtype, manifest.tokenizer_blake3
            ));
            c.size_bytes = std::fs::metadata(&paths.weights)
                .ok()
                .map(|m| m.len())
                .and_then(|w| {
                    std::fs::metadata(&paths.tokenizer)
                        .ok()
                        .map(|t| w + t.len())
                });
            c.artefact_path = Some(dir);
            c.limitations.push(
                "static embeddings: no context, no word order; retrieval quality below a transformer".into(),
            );
            cards.push(c);
        }
        if let Some(model) = &self.config.inference.model {
            let mut c = ModelCard::new(
                ModelRole::Inference,
                model.clone(),
                self.config.inference.base_url.clone(),
                "session summaries, contradiction checks between retrieved blocks, ring and tag suggestions, note supersession (SPEC §11)",
            );
            c.format = Some("OpenAI-compatible HTTP, operated by the deployer".into());
            c.notes.push(
                "licence, version and weights are the endpoint operator's; not read by this tool"
                    .into(),
            );
            cards.push(c);
        }
        cards
    }
}

struct IndexSubjectSource<'a> {
    index: &'a Mutex<Index>,
}

impl SubjectSource for IndexSubjectSource<'_> {
    fn blocks_mentioning(&self, identifier: &Identifier) -> Result<Vec<SubjectBlock>> {
        Ok(lock_index(self.index)?
            .blocks_containing(identifier.raw())?
            .into_iter()
            .map(|(cit, name, text)| SubjectBlock {
                citation: cit.to_string(),
                note_id: None,
                note_name: name,
                ring: Some(cit.ring),
                text,
            })
            .collect())
    }
}

/// The part of a hub pull that touches this store.
pub(crate) struct Pulled {
    pub written: Vec<serde_json::Value>,
    pub kept_local: Vec<serde_json::Value>,
    pub refused: Vec<serde_json::Value>,
    pub flagged: Vec<String>,
    pub erasures: Vec<serde_json::Value>,
    /// A note could not be written here, so the cursor must not move past it.
    pub retry: bool,
}

/// The name form of a link target, for suggesting what the author probably meant.
/// Underscores and spaces become hyphens, capitals fold down (`Für` to `für`, which is a name
/// now), everything else is dropped, and runs of hyphens collapse. Only ever used to look up
/// an existing note, never to rewrite a link: a `[[target]]` in a note is the author's text
/// and stays theirs.
fn normalise_link_target(name: &str) -> String {
    use cyberbrain_core::frontmatter::{is_name_letter, normalize_name};
    let mut out = String::with_capacity(name.len());
    for ch in normalize_name(name).chars() {
        match ch {
            '0'..='9' => out.push(ch),
            c if is_name_letter(c) => out.push(c),
            c if c.is_uppercase() && c.to_lowercase().all(is_name_letter) => {
                out.extend(c.to_lowercase())
            }
            '_' | ' ' | '-' | '.' | '/' if !out.ends_with('-') => out.push('-'),
            _ => {}
        }
    }
    out.trim_matches('-').to_string()
}

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

    /// Calibrated against the state that made this exist: a store whose ledger says the
    /// last check took 126 s is not asked to try again, because it would stall the recall
    /// for two minutes and then be cancelled anyway.
    #[test]
    fn a_measured_slow_endpoint_is_not_asked_again() {
        assert_eq!(
            plan_contradiction_check(3_000, LastCheck::Completed { ms: 126_184 }),
            CheckPlan::SkipMeasuredSlow {
                last_ms: 126_184,
                budget_ms: 3_000
            }
        );
    }

    /// The case the first implementation got wrong, found by timing two calls in a row: a
    /// check that was cut off leaves no completed measurement, so remembering only
    /// completed calls made every recall pay the budget again and get nothing for it.
    #[test]
    fn an_abandoned_check_is_not_retried_either() {
        assert_eq!(
            plan_contradiction_check(3_000, LastCheck::Abandoned),
            CheckPlan::SkipAbandoned { budget_ms: 3_000 }
        );
    }

    #[test]
    fn a_fast_one_runs_under_the_budget() {
        assert_eq!(
            plan_contradiction_check(3_000, LastCheck::Completed { ms: 800 }),
            CheckPlan::Run(std::time::Duration::from_millis(3_000))
        );
        // Nothing measured yet: try once. That is what makes the first call the one that
        // learns, rather than every call paying for the ignorance of the last.
        assert_eq!(
            plan_contradiction_check(3_000, LastCheck::Unknown),
            CheckPlan::Run(std::time::Duration::from_millis(3_000))
        );
    }

    /// Exactly at the budget is not over it.
    #[test]
    fn the_boundary_is_not_slow() {
        assert_eq!(
            plan_contradiction_check(3_000, LastCheck::Completed { ms: 3_000 }),
            CheckPlan::Run(std::time::Duration::from_millis(3_000))
        );
    }

    /// Yesterday's measurement is not evidence about today's endpoint.
    #[test]
    fn a_measurement_expires_after_a_day() {
        let now = jiff::Timestamp::now();
        let fresh = (now - jiff::SignedDuration::from_hours(2)).to_string();
        let old = (now - jiff::SignedDuration::from_hours(30)).to_string();
        assert!(!stale(&fresh), "two hours old still counts");
        assert!(stale(&old), "thirty hours old does not");
        assert!(
            stale("not a timestamp"),
            "an unreadable stamp means try again"
        );
    }

    #[test]
    fn zero_means_wait_however_long_it_takes() {
        assert_eq!(
            plan_contradiction_check(0, LastCheck::Completed { ms: 126_184 }),
            CheckPlan::RunUnbounded
        );
        assert_eq!(
            plan_contradiction_check(0, LastCheck::Abandoned),
            CheckPlan::RunUnbounded
        );
    }

    /// The caveat is read by a person deciding whether to care, so the number has to be
    /// one they can hold: seconds, not 126184.
    #[test]
    fn durations_read_like_durations() {
        assert_eq!(secs(450), "450 ms");
        assert_eq!(secs(3_000), "3.0 s");
        assert_eq!(secs(126_184), "126 s");
    }
}

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

    /// The seam `find` ties: the tree scanned is the store's parent, the store itself is
    /// kept out of it, and the report serialises with the field names the CLI, HTTP and
    /// MCP surfaces all print.
    #[test]
    fn find_scans_the_project_and_not_the_store() {
        let dir = tempfile::tempdir().unwrap();
        let project = dir.path().join("proj");
        std::fs::create_dir_all(project.join("src")).unwrap();
        std::fs::write(
            project.join("src/lib.rs"),
            "/// Greets.\npub fn greet() -> &'static str {\n    \"hi\"\n}\n\nfn main() {\n    greet();\n}\n",
        )
        .unwrap();
        let store = project.join(DEFAULT_STORE_DIR);
        App::init(&store, &Actor::Cli).unwrap();
        // A note whose heading is the same word must not come back from `find`.
        std::fs::write(store.join("notes/r2/greet.md"), "# greet\n\nnot code\n").unwrap();

        let app = App::open(Some(&store), Actor::Cli).unwrap();
        let r = app.find("greet", 10).unwrap();
        assert_eq!(r.root, std::path::absolute(&project).unwrap());
        assert_eq!(r.hits.len(), 1, "{:?}", r.hits);
        let h = &r.hits[0];
        assert_eq!(
            (
                h.path.as_str(),
                h.kind,
                h.language,
                h.start_line,
                h.line,
                h.end_line,
                h.matched
            ),
            ("src/lib.rs", "function", "rust", 1, 2, 4, "exact")
        );
        assert_eq!(
            r.skipped.store_entries, 1,
            "the store is excluded, not merely hidden"
        );
        assert_eq!(r.files_scanned, 1);
        assert!(!r.truncated);
        assert!(
            r.caveats.iter().any(|c| c.contains(".cyberbrainignore")),
            "no ignore file in the project: the report must say so: {:?}",
            r.caveats
        );

        let v = serde_json::to_value(&r).unwrap();
        for key in [
            "symbol",
            "name",
            "scope",
            "root",
            "hits",
            "matched_total",
            "truncated",
            "limit",
            "files_scanned",
            "bytes_scanned",
            "definitions_indexed",
            "skipped",
            "ignore_files",
            "caveats",
            "elapsed_ms",
        ] {
            assert!(v.get(key).is_some(), "FindReport lacks `{key}`");
        }
        for key in [
            "path",
            "start_line",
            "end_line",
            "line",
            "kind",
            "language",
            "name",
            "scope",
            "matched",
            "snippet",
        ] {
            assert!(v["hits"][0].get(key).is_some(), "FindHit lacks `{key}`");
        }

        let e = app.find("", 10).unwrap_err();
        assert_eq!(e.exit_code(), 1);
    }
}

#[cfg(test)]
mod path_rendering_tests {
    use super::*;
    use cyberbrain_core::slash;

    /// A doctor finding names the file it is about, and the name is spelt with forward
    /// slashes on every platform: the report is pasted and diffed, not opened.
    #[test]
    fn doctor_findings_spell_paths_with_forward_slashes() {
        let dir = tempfile::tempdir().unwrap();
        let store = dir.path().join("store");
        App::init(&store, &Actor::Operator).unwrap();
        let stray = store.join("notes").join("r2").join("stray.txt");
        std::fs::write(&stray, "not a note").unwrap();
        let app = App::open(Some(&store), Actor::Operator).unwrap();
        let r = app.doctor().unwrap();
        let f = r
            .findings
            .iter()
            .find(|f| f.check == "notes tree")
            .unwrap_or_else(|| panic!("{r:?}"));
        assert!(!f.detail.contains('\\'), "{}", f.detail);
        assert!(f.detail.contains(&slash(&stray)), "{}", f.detail);

        // The JSON form says the same thing as the human form.
        let v = serde_json::to_value(&r).unwrap();
        for finding in v["findings"].as_array().unwrap() {
            let detail = finding["detail"].as_str().unwrap();
            assert!(!detail.contains('\\'), "{detail}");
        }
    }
}

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

    fn wire_note(name: &str, body: &str) -> serde_json::Value {
        serde_json::json!({
            "name": name,
            "ring": 2,
            "kind": "knowledge",
            "updated": "2026-09-11T08:00:00Z",
            "bereich": "disposition",
            "frontmatter": "",
            "body": body,
        })
    }

    /// One note this machine cannot take must not cost it the rest of the delivery.
    ///
    /// Before, `self.write(req)?` ended the pull at the first such note: the notes after it
    /// never arrived, the cursor did not move, and the next pull stopped at the same place,
    /// for good. An older machine meets exactly that the first time a newer one shares a
    /// name it does not know how to read.
    #[test]
    fn a_note_this_machine_cannot_take_is_refused_and_the_rest_still_arrive() {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("store");
        App::init(&root, &cyberbrain_policy::Actor::Operator).unwrap();
        let app = App::open(Some(&root), cyberbrain_policy::Actor::Operator).unwrap();

        let notes = [
            wire_note("Not A Name", "arrives first and cannot be written"),
            wire_note("arrives-anyway", "arrives second"),
        ];
        let pulled = app.apply_pulled(&notes, &[], false, false).unwrap();

        assert_eq!(pulled.refused.len(), 1, "{:?}", pulled.refused);
        assert_eq!(pulled.refused[0]["name"], "Not A Name");
        assert_eq!(pulled.written.len(), 1, "{:?}", pulled.written);
        assert_eq!(pulled.written[0]["name"], "arrives-anyway");
        assert!(app.store.read("arrives-anyway").is_ok());
        assert!(
            pulled.retry,
            "the cursor must wait for the note that did not land"
        );
    }
}

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

    fn req(ring: Ring, name: &str) -> WriteRequest {
        WriteRequest {
            ring,
            kind: cyberbrain_core::NoteKind::Decision,
            name: name.to_string(),
            body: "*Für: probe*\n\nText.".to_string(),
            tags: Vec::new(),
            bereich: None,
            retention: None,
            force: false,
            choice: None,
            expected_updated: None,
            supersedes: None,
            valid_from: None,
            invalid_at: None,
            arriving: None,
            dry_run: false,
        }
    }

    fn review(name: &str, by: &str) -> ReviewRequest {
        ReviewRequest {
            name: name.to_string(),
            accept: true,
            reason: String::new(),
            by: by.to_string(),
            force: false,
            dry_run: false,
        }
    }

    /// Calibrated against the broken state first (2026-09-25): an agent proposed a ring 0
    /// note, accepted it under a second name, and it landed in notes/r0.
    #[test]
    fn an_agent_cannot_accept_a_ring_0_proposal_under_another_name() {
        let dir = tempfile::tempdir().unwrap();
        let store = dir.path().join("store");
        App::init(&store, &Actor::Operator).unwrap();
        let agent = App::open(Some(&store), Actor::Agent("claude-code:test".into())).unwrap();
        agent
            .propose(req(Ring::Invariant, "vorschlag"), "agent-a")
            .unwrap();
        let err = agent.review(review("vorschlag", "agent-b")).unwrap_err();
        assert!(matches!(err, Error::PolicyRefusal { .. }), "{err}");
        assert!(!store.join("notes").join("r0").join("vorschlag.md").exists());

        // The operator still decides it, and a ring 2 proposal stays the agent's to take.
        let op = App::open(Some(&store), Actor::Operator).unwrap();
        op.review(review("vorschlag", "christoph")).unwrap();
        assert!(store.join("notes").join("r0").join("vorschlag.md").exists());
        agent
            .propose(req(Ring::Knowledge, "r2-vorschlag"), "agent-a")
            .unwrap();
        agent.review(review("r2-vorschlag", "agent-b")).unwrap();
    }

    #[test]
    fn an_agent_cannot_write_ring_0_or_1_directly() {
        let dir = tempfile::tempdir().unwrap();
        let store = dir.path().join("store");
        App::init(&store, &Actor::Operator).unwrap();
        let agent = App::open(Some(&store), Actor::Agent("claude-code:test".into())).unwrap();
        for ring in [Ring::Invariant, Ring::Protocol] {
            let err = agent.write(req(ring, "direkt")).unwrap_err();
            assert!(matches!(err, Error::PolicyRefusal { .. }), "{err}");
        }
        agent.write(req(Ring::Knowledge, "direkt")).unwrap();
    }
}

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

    fn store_with_model(dir: &Path) -> PathBuf {
        let store = dir.join("store");
        App::init(&store, &Actor::Operator).unwrap();
        let toml = store.join("cyberbrain.toml");
        let text = std::fs::read_to_string(&toml)
            .unwrap()
            .replace("# model = \"qwen3:8b\"", "model = \"probe\"")
            // A loopback port nobody listens on: validation passes, a call fails at once.
            .replace("http://127.0.0.1:11434/v1", "http://127.0.0.1:9/v1");
        std::fs::write(&toml, text).unwrap();
        store
    }

    fn note(app: &App, ring: Ring, name: &str, body: &str) {
        app.write(WriteRequest {
            ring,
            kind: cyberbrain_core::NoteKind::Knowledge,
            name: name.to_string(),
            body: body.to_string(),
            tags: Vec::new(),
            bereich: None,
            retention: None,
            force: false,
            choice: None,
            expected_updated: None,
            supersedes: None,
            valid_from: None,
            invalid_at: None,
            arriving: None,
            dry_run: false,
        })
        .unwrap();
    }

    /// Audit rows the inference layer wrote: opening the client, asking the register.
    fn inference_rows(store: &Path) -> Vec<String> {
        let app = App::open(Some(store), Actor::Operator).unwrap();
        app.policy
            .audit()
            .read(&AuditFilter::default())
            .unwrap()
            .into_iter()
            .map(|r| r.action.to_string())
            .filter(|a| a.starts_with("inference.") || a.starts_with("egress."))
            .collect()
    }

    fn recall(app: &App, q: &str) -> RecallResult {
        runtime_for_tests()
            .block_on(app.recall(q, &RecallRequest::default()))
            .unwrap()
    }

    fn runtime_for_tests() -> tokio::runtime::Runtime {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
    }

    /// Calibrated against the state it fixes (2026-09-25): the orderflow store skipped the
    /// check on every recall because the last one was abandoned, and still opened the
    /// client first, which put `egress.permitted` and `inference.call` into the audit log
    /// for each read — 27 % of all rows.
    #[test]
    fn a_skipped_check_opens_no_client_and_writes_no_audit_row() {
        let dir = tempfile::tempdir().unwrap();
        let store = store_with_model(dir.path());
        let app = App::open(Some(&store), Actor::Operator).unwrap();
        note(&app, Ring::Knowledge, "wolf-an", "wolfpack läuft im Juli");
        note(
            &app,
            Ring::Session,
            "wolf-aus",
            "wolfpack abgeschaltet im September",
        );
        hostload::LoadLog::new(&store).append(&hostload::row(
            TASK_CONTRADICTION_ABANDONED,
            std::time::Duration::from_millis(3_001),
            (None, None),
            (None, None),
        ));
        let before = inference_rows(&store).len();
        let r = recall(&app, "wolfpack");
        assert_eq!(r.hits.len(), 2);
        assert!(
            r.caveats.iter().any(|c| c.contains("still running")),
            "{:?}",
            r.caveats
        );
        assert_eq!(
            inference_rows(&store).len(),
            before,
            "a check that does not run must not open the client"
        );
    }

    /// All hits in one ring: there is nothing the check could report (conflicts are
    /// defined across rings), so no client, no audit row and no ledger row — before, the
    /// early return was booked as a completed 0 ms check.
    #[test]
    fn hits_from_one_ring_are_not_a_check_and_not_a_measurement() {
        let dir = tempfile::tempdir().unwrap();
        let store = store_with_model(dir.path());
        let app = App::open(Some(&store), Actor::Operator).unwrap();
        note(&app, Ring::Knowledge, "wolf-an", "wolfpack läuft im Juli");
        note(
            &app,
            Ring::Knowledge,
            "wolf-aus",
            "wolfpack abgeschaltet im September",
        );
        let before = inference_rows(&store).len();
        let r = recall(&app, "wolfpack");
        assert_eq!(r.hits.len(), 2);
        assert!(
            r.caveats
                .iter()
                .any(|c| c.contains("all 2 hits are in one ring")),
            "{:?}",
            r.caveats
        );
        assert_eq!(inference_rows(&store).len(), before);
        assert!(
            hostload::LoadLog::new(&store)
                .last_of(&[TASK_CONTRADICTION, TASK_CONTRADICTION_ABANDONED])
                .is_none(),
            "nothing was measured, so nothing is in the ledger"
        );
    }

    /// The other side: a check that is due still opens the client and says so in the log.
    #[test]
    fn a_due_check_still_opens_the_client() {
        let dir = tempfile::tempdir().unwrap();
        let store = store_with_model(dir.path());
        let app = App::open(Some(&store), Actor::Operator).unwrap();
        note(&app, Ring::Knowledge, "wolf-an", "wolfpack läuft im Juli");
        note(
            &app,
            Ring::Session,
            "wolf-aus",
            "wolfpack abgeschaltet im September",
        );
        let before = inference_rows(&store).len();
        let _ = recall(&app, "wolfpack");
        let rows = inference_rows(&store);
        assert!(rows.len() > before, "{rows:?}");
    }
}

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

    fn write(app: &App, ring: Ring, name: &str, body: &str) {
        app.write(WriteRequest {
            ring,
            kind: cyberbrain_core::NoteKind::Knowledge,
            name: name.to_string(),
            body: body.to_string(),
            tags: Vec::new(),
            bereich: None,
            retention: None,
            force: false,
            choice: None,
            expected_updated: None,
            supersedes: None,
            valid_from: None,
            invalid_at: None,
            arriving: None,
            dry_run: false,
        })
        .unwrap();
    }

    fn recall(app: &App, q: &str, ring: Option<Ring>) -> RecallResult {
        let req = RecallRequest {
            ring,
            ..RecallRequest::default()
        };
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(app.recall(q, &req))
            .unwrap()
    }

    const RULE: &str = "Never copy config.py from server one to server two: server two keeps \
                        its own copy, and overwriting it once broke live trading there.";

    fn store() -> (tempfile::TempDir, App) {
        let dir = tempfile::tempdir().unwrap();
        let root = dir.path().join("store");
        App::init(&root, &Actor::Operator).unwrap();
        let app = App::open(Some(&root), Actor::Operator).unwrap();
        (dir, app)
    }

    /// The same store, as an agent whose session start injected rings 0 and 1.
    fn agent(op: &App) -> App {
        App::open(Some(op.root()), Actor::Agent("claude-code:test".into())).unwrap()
    }

    /// Rings 0 and 1 are injected whole at every session start (hook/events.rs), and the
    /// usage text there promises that recall searches rings 2 to 4. Calibrated against the
    /// state before: the ring 0 note came back as the first hit, a second copy of what the
    /// agent already had in its context.
    #[test]
    fn resident_rings_are_not_recalled_unless_asked_for() {
        let (_d, app) = store();
        write(
            &app,
            Ring::Invariant,
            "wolfpack-regel",
            "wolfpack bleibt abgeschaltet.",
        );
        write(
            &app,
            Ring::Knowledge,
            "wolfpack-stand",
            "wolfpack wurde am 10.09. abgeschaltet.",
        );
        let r = recall(&agent(&app), "wolfpack abgeschaltet", None);
        let names: Vec<&str> = r.hits.iter().map(|h| h.note_name.as_str()).collect();
        assert_eq!(names, ["wolfpack-stand"], "{r:?}");
        assert!(
            r.caveats.iter().any(|c| c.contains("rings 0/1")),
            "{:?}",
            r.caveats
        );
        // Asked for by ring, they are there.
        let r0 = recall(&agent(&app), "wolfpack abgeschaltet", Some(Ring::Invariant));
        assert_eq!(r0.hits.len(), 1);
        assert_eq!(r0.hits[0].note_name, "wolfpack-regel");
    }

    /// A person has no session context: for them the ring 0 note is the answer (the UI's
    /// simple view is built on that). Found by the browser tests: with rings 0/1 left out
    /// for everybody, the simple view had no answer to "can we release on a friday".
    #[test]
    fn a_person_still_gets_the_resident_rings() {
        let (_d, app) = store();
        write(
            &app,
            Ring::Invariant,
            "wolfpack-regel",
            "wolfpack bleibt abgeschaltet.",
        );
        write(
            &app,
            Ring::Knowledge,
            "wolfpack-stand",
            "wolfpack wurde am 10.09. abgeschaltet.",
        );
        let r = recall(&app, "wolfpack abgeschaltet", None);
        let names: Vec<&str> = r.hits.iter().map(|h| h.note_name.as_str()).collect();
        assert_eq!(names, ["wolfpack-regel", "wolfpack-stand"], "{r:?}");
        assert!(!r.caveats.iter().any(|c| c.contains("rings 0/1")));
    }

    /// A ring 2 copy of a ring 1 paragraph is the ring 1 paragraph: already in context.
    #[test]
    fn a_copy_of_resident_text_is_not_recalled_either() {
        let (_d, app) = store();
        write(&app, Ring::Protocol, "protokoll", RULE);
        write(&app, Ring::Knowledge, "abschrift", RULE);
        write(
            &app,
            Ring::Knowledge,
            "anderes",
            "config.py on server two is its own.",
        );
        let r = recall(&agent(&app), "config.py server two", None);
        let names: Vec<&str> = r.hits.iter().map(|h| h.note_name.as_str()).collect();
        assert_eq!(names, ["anderes"], "{r:?}");
    }

    /// Calibrated against the state before: doctor had no such check and said nothing.
    #[test]
    fn doctor_names_notes_that_share_most_of_their_blocks() {
        let (_d, app) = store();
        let other = "A second paragraph long enough to count as a block of its own, about \
                     something else entirely, written once and copied along with the rule.";
        write(
            &app,
            Ring::Knowledge,
            "original",
            &format!("{RULE}\n\n{other}"),
        );
        write(
            &app,
            Ring::Knowledge,
            "kopie",
            &format!("{RULE}\n\n{other}"),
        );
        write(
            &app,
            Ring::Knowledge,
            "eigen",
            "Nothing here is shared with anybody.",
        );
        let r = app.doctor().unwrap();
        let shared: Vec<&DoctorFinding> = r
            .findings
            .iter()
            .filter(|f| f.check == "shared blocks")
            .collect();
        assert_eq!(shared.len(), 1, "{:?}", r.findings);
        assert!(shared[0].detail.contains("kopie"), "{}", shared[0].detail);
        assert!(
            shared[0].detail.contains("original"),
            "{}",
            shared[0].detail
        );
        assert!(shared[0].detail.contains("(100%)"), "{}", shared[0].detail);
        assert!(r.checks_run.contains(&"shared blocks"));
    }
}