cli/

lib.rs

//! `mushroomdb` CLI library: hand-rolled arg parsing and the demo dataset builder.
//!
//! The binary in `main.rs` stays thin — it dispatches on [`parse_args`] and
//! prints what the lib functions return.

pub mod doctor;
pub mod export;
pub(crate) mod hook;
pub mod ingest_git;
pub mod install;
pub mod recall;
pub mod structure;

#[cfg(test)]
use core_api::restore::holds_a_store;
/// Re-exported where it has always been named: `cli::RestoreOutcome`.
pub use core_api::restore::RestoreOutcome;
use core_api::schema::Schema;
use core_api::{
    default_max_edges, is_write_query, valid_namespace, AlgoDir, BackupReport, DegreeConfig,
    Explanation, GraphDb, IngestOptions, LouvainConfig, PageRankConfig, Predicate, RealFs,
    ResultSet, RuleDef, RuleSuggestion, SharedDb, SnapshotOptions, Stats, Value, WccConfig,
    WriteGuard, NS_MAX_LEN,
};
use export::ExportFormat;
use std::collections::BTreeMap;
use std::fmt::Write as _;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;

/// What every snapshot mushroomdb takes *on its own* does with the WAL.
///
/// An ingest that ends past the WAL threshold, a `serve --snapshot-every`
/// tick, a graceful shutdown, and a plain `mushroomdb snapshot` all archive the
/// WAL to `wal.<N>.archive` rather than dropping it. `node_history`,
/// `edge_history`, `was_linked` and `open_at` all read WAL frames and all
/// consult archives, so the store opens fast and still remembers how it got
/// here. A truncating snapshot ends that reach — it deletes the genesis marker
/// as well as the tail — so it is never something mushroomdb decides for the
/// user: only `mushroomdb snapshot --truncate` does it.
pub const AUTOMATIC_SNAPSHOT: SnapshotOptions = SnapshotOptions {
    keep_wal: false,
    archive_wal: true,
};

/// How many WAL archives a snapshot mushroomdb takes *on its own* keeps.
/// `None` = all of them. Retention is configured, never defaulted: deleting
/// history nobody asked to delete is not a default worth having.
///
/// Archiving moves the WAL aside rather than deleting it, so without a bound
/// every automatic snapshot leaves one more file behind and nothing ever
/// reclaims them. On a dogfooded repository that is a new archive per
/// [`SNAPSHOT_WAL_BYTES`] of churn, for as long as the store exists — and that
/// growth is the argument for the default, not against it: the reach archives
/// exist to preserve — `node_history`, `edge_history`, `was_linked` — is
/// exactly what pruning costs, so an automatic snapshot never volunteers to
/// pay it. `mushroomdb stats` and `mushroomdb doctor` both print where history
/// currently starts, and `mushroomdb snapshot <db> --retention N` is how a
/// caller bounds the directory once they have decided the trade is worth it.
///
/// One consequence is worth stating plainly for whoever does set a bound,
/// because it is not proportional. The first prune breaks the genesis chain,
/// and `open_at` refuses any commit it cannot reconstruct from a complete
/// prefix — so from that point it answers for commits past the last snapshot
/// and no further, even though the retained archives still answer
/// `node_history` and `was_linked` over their own window. Time travel to a
/// point-in-time state is therefore bounded by the last snapshot once a store
/// has churned this far; the history reads are bounded by the retention.
///
/// Only the automatic path takes this default. `mushroomdb snapshot` is a
/// thing the user asked for, and `--retention N` is theirs to set: an
/// explicit snapshot with no `--retention` still keeps every archive.
///
/// [`SNAPSHOT_WAL_BYTES`]: crate::ingest_git::SNAPSHOT_WAL_BYTES
pub const AUTO_SNAPSHOT_RETENTION: Option<u32> = None;

/// Take [`AUTOMATIC_SNAPSHOT`] under a held write lock, keeping
/// [`AUTO_SNAPSHOT_RETENTION`] archives.
///
/// The single place the automatic disposition and the automatic bound are
/// applied together, so no caller can pick up one without the other.
///
/// # Errors
///
/// Whatever writing the snapshot returned.
pub fn snapshot_automatically(db: &mut WriteGuard<'_>) -> Result<(), core_api::GraphError> {
    db.set_wal_archive_retention(AUTO_SNAPSHOT_RETENTION);
    db.snapshot_with(AUTOMATIC_SNAPSHOT)
}

/// How long a server-initiated snapshot waits for the store's cross-process
/// write lock before giving up.
///
/// Short on purpose. A snapshot is an optimisation — it shortens the next
/// open's replay — so skipping one costs nothing but a longer replay, whereas
/// blocking the shutdown path or piling up timer ticks behind a busy peer
/// costs the operator.
pub const SNAPSHOT_LOCK_WAIT: Duration = Duration::from_millis(500);

/// Take the snapshot `serve` takes — on a `--snapshot-every` tick, and once
/// more on a graceful shutdown.
///
/// Lives here rather than in `main.rs` so the behaviour a running server has is
/// the behaviour a test can call. `Busy` is the caller's to interpret: a tick
/// skips it, since the next one is only a period away.
///
/// # Errors
///
/// Whatever taking the write lock or writing the snapshot returned.
pub fn snapshot_shared(db: &SharedDb) -> Result<(), core_api::GraphError> {
    snapshot_automatically(&mut db.write_with_wait(SNAPSHOT_LOCK_WAIT)?)
}

/// Deterministic demo: 10 Orgs, 20 Projects, 30 People.
pub const N_ORGS: usize = 10;
pub const N_PROJECTS: usize = 20;
pub const N_PEOPLE: usize = 30;

/// Sample query printed by `mushroomdb demo` and executed against the fresh store.
///
/// Scoped to one person so `ORDER BY score DESC` is visibly ranked (a global
/// `LIMIT 5` would be five 1.0 home-project hits).
pub const SAMPLE_QUERY: &str = "\
MATCH (p:Person {id: 'person-01'})-[r:FIT]->(proj:Project)
RETURN p, proj, r.score AS score
ORDER BY score DESC, proj";

const SAMPLE_EXPLAIN_A: &str = "person-01";
const SAMPLE_EXPLAIN_B: &str = "proj-01";

/// Build version, printed by `mushroomdb --version`.
pub const VERSION: &str = env!("CARGO_PKG_VERSION");

/// The one line `--version` and the `version` subcommand print.
#[must_use]
pub fn version_string() -> String {
    format!("mushroomdb {VERSION}")
}

/// Where `--auto` looks for a database, in order.
///
/// 1. `$CLAUDE_PROJECT_DIR/mushroom-memory` — the assistant tells a hook which
///    project it is working in, and that is the most specific answer there is.
/// 2. `<working-tree root>/mushroom-memory`, but only when the working
///    directory is inside a git checkout. Without that guard a command run
///    from a home directory would quietly create a store there.
/// 3. `<home>/.mushroomdb/memory`, the user-scope default `install` writes.
///
/// The two project-scoped answers match [`install::default_db`],
/// so a hook with `--auto` finds the store `install --project` created.
#[must_use]
pub fn resolve_auto_db(
    env_project_dir: Option<&std::ffi::OsStr>,
    cwd: &Path,
    home: &Path,
) -> PathBuf {
    if let Some(dir) = env_project_dir.filter(|d| !d.is_empty()) {
        return Path::new(dir).join("mushroom-memory");
    }
    if let Some(root) = worktree_root(cwd) {
        return root.join("mushroom-memory");
    }
    home.join(".mushroomdb").join("memory")
}

/// The root of the working tree `dir` sits in: the nearest ancestor holding a
/// `.git` entry, or `None` outside a checkout.
///
/// The answer is a *working tree* root, never the `.git` directory several
/// worktrees share. A linked worktree keeps a `.git` **file** at its root
/// (`gitdir: …/worktrees/<name>`) rather than a directory, and both spellings
/// count here, so `git worktree add` produces a checkout that resolves to its
/// own store. Two worktrees are two different sets of files, and a graph built
/// from one answers questions about the other wrongly.
///
/// Walking up matters as much as the file/directory distinction: a hook fires
/// with whatever working directory the tool call had, which is often a
/// subdirectory, and only the root has the `.git` entry.
#[must_use]
pub fn worktree_root(dir: &Path) -> Option<&Path> {
    dir.ancestors().find(|d| d.join(".git").exists())
}

/// How `serve` should mount a UI. Precedence: `--ui dir` > embedded > `--no-ui`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ServeUi {
    Filesystem(PathBuf),
    Embedded,
    None,
}

/// Algorithm subcommand for `mushroomdb algo`.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AlgoSubcmd {
    Pagerank,
    Wcc,
    Degree,
    Communities,
}

/// Parsed `mushroomdb` invocation.
///
/// No `Eq` derive: `Algo { min_weight: Option<f64>, .. }` carries a float.
#[derive(Debug, Clone, PartialEq)]
pub enum Command {
    /// A hook body 0.6 installed and 0.7 retired: `touch`, `intercept`,
    /// `impact-hook` and `enrich` in the assistant's settings, `sync` in the
    /// git hooks. A machine that upgraded without re-running `install` still
    /// calls them, so each is accepted with any arguments, opens nothing, and
    /// prints one stderr line saying how to remove it. Never in [`usage`].
    RetiredHook {
        sub: &'static str,
    },
    Serve {
        db_dir: PathBuf,
        addr: SocketAddr,
        ui: ServeUi,
        /// If the db dir is missing or empty, run [`run_demo`] before serving.
        /// Docker's default CMD uses this so a fresh volume is ready on first boot.
        demo_if_empty: bool,
        /// Bearer token for non-loopback binds. Loopback may omit it.
        token: Option<String>,
        /// Role-bound tokens from `--role-token TOKEN:ROLE` flags.
        /// Merged with `MUSHROOMDB_ROLE_TOKENS` env var in main before serving.
        role_tokens: Vec<(String, String)>,
        /// Periodic snapshot cadence. `None` = off (default).
        snapshot_every: Option<Duration>,
        /// Seed an empty `db_dir` from the newest backup under this directory
        /// before opening it. See [`restore_if_empty`]. `None` = off (default).
        restore_from: Option<PathBuf>,
        /// Path to PEM certificate for native TLS (`--tls-cert`). Requires `--tls-key`.
        tls_cert: Option<PathBuf>,
        /// Path to PEM private key for native TLS (`--tls-key`). Requires `--tls-cert`.
        tls_key: Option<PathBuf>,
    },
    Mcp {
        /// `None` with `auto` set: resolved by [`resolve_auto_db`] at run time.
        db_dir: Option<PathBuf>,
        auto: bool,
        /// `--all-tools`: advertise every served tool in `tools/list`
        /// rather than the association tools every store lists by default. The rest are
        /// callable either way; the flag decides what is listed, and what every
        /// session pays for before its first turn.
        all_tools: bool,
    },
    Stats {
        db_dir: PathBuf,
    },
    Demo {
        db_dir: PathBuf,
    },
    /// Read-only view of the database at a past commit.
    AsOf {
        db_dir: PathBuf,
        /// 0-based WAL commit index to replay up to (inclusive).
        ///
        /// Exactly one of `commit` and `at` is set — the parser refuses both and
        /// refuses neither, because a precedence rule between them is how a
        /// caller ends up believing it asked for a date while the engine
        /// answered a commit.
        commit: Option<u64>,
        /// RFC 3339 date to replay up to, resolved against the store's recorded
        /// commit times once it is open.
        at: Option<String>,
        /// Optional Cypher read query to execute against the as-of view.
        query: Option<String>,
        /// Read only this namespace, as it was at that commit.
        namespace: Option<String>,
    },
    /// Profile the database and suggest linking rules with estimated edge counts.
    Suggest {
        db_dir: PathBuf,
    },
    /// Run a graph algorithm (pagerank / wcc / degree / communities).
    Algo {
        db_dir: PathBuf,
        subcmd: AlgoSubcmd,
        /// Print only the top N results (0 = all).
        top: usize,
        /// Edge direction for degree/pagerank (`out` / `in` / `both`).
        /// Ignored by `wcc` and `communities`, which are always undirected.
        dir: AlgoDir,
        /// `communities` only: restrict to the union of these edge types
        /// (`--edge-type T`, repeatable). Empty means all edge types.
        edge_types: Vec<String>,
        /// `communities` only: edge property to read as the edge weight.
        weight_prop: Option<String>,
        /// `communities` only: drop edges below this resolved weight.
        min_weight: Option<f64>,
    },
    /// Run a Cypher query (read or write).
    Query {
        db_dir: PathBuf,
        /// Positional after dir (remaining args joined), or `--query`.
        cypher: String,
        /// Answer as one of the store's roles: only the nodes it may see.
        role: Option<String>,
        /// Answer from one namespace only. Intersects `role` — never widens it.
        namespace: Option<String>,
    },
    /// Write `snapshot.bin`. The WAL is archived unless told otherwise.
    Snapshot {
        db_dir: PathBuf,
        wal: WalDisposition,
        /// Keep the newest N archives; prune oldest at snapshot time.
        /// None = unlimited. Applies only when the WAL is archived.
        retention: Option<u32>,
    },
    /// Drive any outstanding vector-index build to completion.
    BuildIndex {
        db_dir: PathBuf,
        /// Build only this rule; `None` builds every pending one.
        rule: Option<String>,
    },
    /// Apply a JSON schema file idempotently (`schema apply <db-dir> <schema.json>`),
    /// the built-in memory schema (`schema apply <db-dir> --memory-defaults`), or
    /// the identity preset (`schema apply <db-dir> --memory-identity`).
    SchemaApply {
        db_dir: PathBuf,
        /// `None` when a built-in was named instead.
        schema_file: Option<PathBuf>,
        memory_defaults: bool,
        memory_identity: bool,
    },
    /// Migrate an old-format snapshot to the current version and keep `.bak`.
    Migrate {
        db_dir: PathBuf,
    },
    /// Validate CRC32 integrity of every section in the V8 snapshot.
    Verify {
        db_dir: PathBuf,
    },
    /// Create a consistent, verified copy of the database directory.
    Backup {
        db_dir: PathBuf,
        dest: PathBuf,
    },
    /// Export all nodes, edges, and rules to a destination directory.
    Export {
        db_dir: PathBuf,
        dest: PathBuf,
        format: ExportFormat,
    },
    /// Build (or incrementally sync) a graph of a git repository.
    IngestGit {
        db_dir: PathBuf,
        opts: ingest_git::IngestGitOpts,
    },
    /// Wire the /mushroom skill and MCP server into Claude Code / Cursor.
    Install(install::InstallOpts),
    /// Undo what `install` wrote (manifest-driven).
    Uninstall(install::InstallOpts),
    /// Turn an install off without removing it: strips the hooks, the MCP
    /// entry and the git hook blocks; the skill, the store and the
    /// `.gitignore` line stay.
    Disable(install::ToggleOpts),
    /// Turn a disabled install back on, re-deriving the dynamic parts (the
    /// resolved command, the hooks) rather than replaying stale ones.
    Enable(install::ToggleOpts),
    /// Verify an install end to end: config, store, hooks, and a real MCP handshake.
    Doctor(doctor::DoctorOpts),
    /// Body of the Claude Code UserPromptSubmit hook: reads a prompt payload on
    /// stdin, prints related graph facts on stdout.
    Recall {
        db_dir: Option<PathBuf>,
        auto: bool,
    },
    /// Body of the Claude Code SessionStart hook: prints the repository in one
    /// byte-stable block, which the host caches for the whole session.
    Brief {
        db_dir: Option<PathBuf>,
        auto: bool,
    },
    /// What links two nodes, with the evidence behind each link.
    Why {
        db_dir: PathBuf,
        a: String,
        b: String,
    },
    Version,
    Help,
}

/// Outcome of [`run_demo`]. Counts are deterministic.
#[derive(Debug)]
pub struct DemoOutcome {
    pub auto_fk_rules: Vec<String>,
    pub sample_query: String,
    pub sample_result: ResultSet,
    pub explanations: Vec<Explanation>,
    pub stats: Stats,
    /// First suggestion from the rule suggester (teaser only — not auto-applied).
    pub suggestion: Option<RuleSuggestion>,
}

/// CLI-facing error. [`Display`] is the message printed to stderr.
#[derive(Debug)]
pub struct CliError(pub String);

impl std::fmt::Display for CliError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl std::error::Error for CliError {}

impl From<core_api::GraphError> for CliError {
    fn from(e: core_api::GraphError) -> Self {
        CliError(e.to_string())
    }
}

impl From<std::io::Error> for CliError {
    fn from(e: std::io::Error) -> Self {
        CliError(e.to_string())
    }
}

/// Usage text for no-args / `--help` / `-h`.
pub fn usage() -> &'static str {
    "\
mushroomdb — embedded graph database

Usage:
  mushroomdb install [--platform claude-code|cursor|codex|all] [--project|--user] [--db <path>]
                     [--command <path>] [--no-prewarm] [--delivery cli|mcp|both]
                     [--always-load|--no-always-load]
                     --delivery cli writes the skill and the hooks and registers no MCP
                     server: the skill teaches `mushroomdb <command>` instead (claude-code
                     only; cursor and codex are always registered as MCP servers)
                     --always-load marks the registered MCP server alwaysLoad, so the host
                     keeps its tools in context instead of deferring them; already the
                     default when --db names a store and a server is registered
                     (--delivery mcp|both) — --no-always-load opts out
  mushroomdb uninstall [--platform claude-code|cursor|codex|all] [--project|--user] [--db <path>]
  mushroomdb disable [--platform claude-code|cursor|codex|all] [--project|--user]
                     turn an install off without removing it: hooks and MCP entry are
                     removed; the skill, the store and .gitignore stay
  mushroomdb enable [--platform claude-code|cursor|codex|all] [--project|--user]
                     turn a disabled install back on
  mushroomdb doctor [--project|--user] [--platform claude-code|cursor|codex|all]
                     verify an install: config entry, store, hooks, and a real stdio
                     handshake with the configured MCP command; exits 1 on any `fail`
  mushroomdb serve <db-dir> [--addr 127.0.0.1:8080] [--token <secret>] [--role-token TOKEN:ROLE] [--ui <dist-dir>] [--no-ui] [--demo-if-empty] [--snapshot-every <secs>] [--restore-from <dir>] [--tls-cert <pem> --tls-key <pem>]
                     --restore-from seeds an empty <db-dir> from the newest backup under <dir>
                     (or from <dir> itself if it is one); a no-op when <db-dir> already holds a store
                     --tls-cert and --tls-key go together and need a build with the `tls` feature
  mushroomdb mcp <db-dir>|--auto [--all-tools]
                     --all-tools lists every served tool; the default lists the association tools on every store,
                     a store `ingest-git` built included (the rest stay callable, just unlisted)
  mushroomdb stats <db-dir>
  mushroomdb demo <db-dir>
  mushroomdb recall <db-dir>|--auto   hook body: reads a prompt payload on stdin, prints related graph facts
  mushroomdb brief <db-dir>|--auto    hook body: the store's schema in one block — labels,
                                      edge types, history depth, roles and one worked call per
                                      question kind; byte-stable, so a session host caches it once
  mushroomdb why <db-dir> <a> <b>
                                   every rule edge between two keys with the evidence that
                                   derived it, or a note that there is none
  mushroomdb suggest <db-dir>
  mushroomdb asof <db-dir> --commit N|--at <date> [--query \"MATCH ...\"] [--namespace <ns>]
                     --at takes an RFC 3339 date (2026-06-19, or 2026-06-19T12:00:00Z)
                     and resolves to the last commit at or before it; exactly one of
                     --commit and --at
  mushroomdb query <db-dir> [--query \"MATCH ...\"] [--role <name>] [--namespace <ns>] <cypher…>
                     --role answers as one of the store's roles and --namespace from one
                     namespace; together they intersect, so neither ever widens the other
                     and either one makes the query a read
  mushroomdb snapshot <db-dir> [--keep-wal|--truncate] [--retention N]
                     folds the WAL into snapshot.bin and archives it as wal.<N>.archive,
                     so node_history, edge_history, was_linked and asof keep reaching it;
                     --truncate discards it instead, --keep-wal leaves wal.bin whole
  mushroomdb build-index <db-dir> [--rule <name>]
                     drives a rule's vector index to completion a slice at a time, for an
                     operator who wants the build finished before traffic arrives; a rule
                     created over a large corpus derives no edges until its index is whole
  mushroomdb migrate <db-dir>
  mushroomdb verify <db-dir>       validate CRC32 integrity of every snapshot section
  mushroomdb backup <db-dir> <dest>   process-local consistent copy of the database to <dest>
                                      WARNING: unsafe against a concurrently running serve process;
                                      use POST /backup on the HTTP server for live-serve backups
  mushroomdb export <db-dir> <dest> --format jsonl|parquet|graphml   export all data
                                      graphml writes one file: <dest>/graph.graphml if <dest> is an
                                      existing directory, otherwise <dest> is the file path itself
                                      (nodes + edges only; rules have no GraphML analogue)
  mushroomdb ingest-git <db-dir> <repo-dir> [--exclude <pattern>]... [--max-commits-per-file N]
                        [--recurse-submodules] [--prs] [--no-structure] [--no-docs] [--ensure-gitignore]
                                   graph a git repo (authors, commits, files, symbols, imports, calls, mentions); re-run to sync
                                   --recurse-submodules also walks each initialised submodule
                                   --prs links merged pull requests via gh (skipped when gh is unavailable)
                                   --no-structure skips the working-tree pass (no hashes, symbols, imports or calls)
                                   --no-docs skips Markdown bodies, headings and mentions
                                   --ensure-gitignore adds the database directory to the repo's .gitignore
                                   with no --exclude the defaults apply: target/ node_modules/ dist/ .git/ *.lock *.min.js
  mushroomdb schema apply <db-dir> <schema.json>
  mushroomdb schema apply <db-dir> --memory-defaults
                                   applies the built-in memory schema to an existing store;
                                   full-text on a populated store rebuilds the index at every open from here on
  mushroomdb schema apply <db-dir> --memory-identity
                                   adds the identity preset: sixteen SAME_AS rules over each entity's aliases;
                                   says what it will backfill, then backfills
  mushroomdb algo pagerank <db-dir> [--top N] [--dir out|in|both]
  mushroomdb algo wcc <db-dir> [--top N]
  mushroomdb algo degree <db-dir> [--top N] [--dir out|in|both]
  mushroomdb algo communities <db-dir> [--edge-type T]... [--weight-prop P] [--min-weight X] [--top N]
  mushroomdb --version
  mushroomdb --help

Default serve address is 127.0.0.1:8080. Non-loopback --addr requires --token or MUSHROOMDB_TOKEN.
install defaults: --platform auto-detect; scope auto (project inside a git checkout, else user);
the MCP entry runs `npx -y mushroomdb@<version>` unless a `mushroomdb` on PATH is this binary, or
--command names one (a relative --command or --db is anchored to the current directory).
--no-prewarm means no network and no resolution: neither the one-off package fetch nor locating the
package's binary, so every hook keeps the slower `npx` form.
install and uninstall take out the code-graph hooks and git hook blocks a 0.6 install wrote.
uninstall resolves the same scope and falls back to the other one when the inferred scope has no
manifest; undoing a Codex install needs --platform codex.
A project install inside a git checkout writes --auto rather than a store path, so each `git
worktree` gets its own store; outside a checkout, and with --db, the store is pinned to an absolute
path instead.
--auto resolves the database as $CLAUDE_PROJECT_DIR/mushroom-memory, else mushroom-memory at the
root of the working tree the current directory is in, else ~/.mushroomdb/memory.
"
}

fn parse_install_cmd(args: &[&str]) -> Result<install::InstallOpts, String> {
    let mut platform: Option<install::Platform> = None;
    let mut scope: Option<install::Scope> = None;
    let mut db: Option<PathBuf> = None;
    let mut command: Option<PathBuf> = None;
    let mut prewarm = true;
    let mut delivery = install::Delivery::default();
    // Tri-state on purpose: `None` is "the user said nothing", which is the
    // only case the default below is allowed to decide.
    let mut always_load: Option<bool> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--delivery" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --delivery".to_string())?;
            delivery = install::Delivery::parse(val)?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--delivery=") {
            delivery = install::Delivery::parse(val)?;
            i += 1;
        } else if a == "--platform" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --platform".to_string())?;
            platform = Some(install::Platform::parse(val)?);
            i += 2;
        } else if let Some(val) = a.strip_prefix("--platform=") {
            platform = Some(install::Platform::parse(val)?);
            i += 1;
        } else if a == "--project" || a == "--user" {
            let want = if a == "--project" {
                install::Scope::Project
            } else {
                install::Scope::User
            };
            // Two scopes name two different installs; picking one silently
            // would put files somewhere the user did not ask for.
            if scope.is_some_and(|s| s != want) {
                return Err("--project and --user are mutually exclusive".to_string());
            }
            scope = Some(want);
            i += 1;
        } else if a == "--no-git-hooks" {
            // Accepted and ignored: 0.7 writes no git hooks, so "none" is
            // already what every install gets. Refusing it would break every
            // script that passed it to a 0.6 install for the same result.
            i += 1;
        } else if a == "--intercept-grep" || a == "--impact-before-edit" || a == "--enrich-grep" {
            return Err(format!(
                "{a} was removed in 0.7 along with the code graph it fed. \
                 Nothing replaces it. Pin `mushroomdb@0.6.x` to keep it."
            ));
        } else if a == "--always-load" {
            always_load = Some(true);
            i += 1;
        } else if a == "--no-always-load" {
            always_load = Some(false);
            i += 1;
        } else if a == "--no-prewarm" {
            prewarm = false;
            i += 1;
        } else if a == "--command" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --command".to_string())?;
            command = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--command=") {
            command = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--db" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --db".to_string())?;
            db = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--db=") {
            db = Some(PathBuf::from(val));
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else {
            return Err(format!("unexpected argument: {a}"));
        }
    }
    // An install that named its store with `--db` and registers a server is
    // an entity-store install: the session is being pointed at a graph it
    // could not otherwise find, and the first association run showed what a
    // deferred tool list costs it — turns spent searching for the tools
    // before the first question. So `alwaysLoad` is the default there, and
    // `--no-always-load` is the way out.
    //
    // An install with no `--db` takes whatever store the working directory
    // resolves to, which is usually the code graph: three tools, a skill that
    // teaches them, and no discovery problem worth pinning context for. That
    // one stays opt-in through `--always-load`.
    let always_load =
        always_load.unwrap_or(db.is_some() && !matches!(delivery, install::Delivery::Cli));
    Ok(install::InstallOpts {
        platform,
        scope,
        db,
        command,
        prewarm,
        delivery,
        always_load,
    })
}

fn parse_doctor_cmd(args: &[&str]) -> Result<doctor::DoctorOpts, String> {
    let mut platform: Option<install::Platform> = None;
    let mut scope: Option<install::Scope> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--platform" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --platform".to_string())?;
            platform = Some(install::Platform::parse(val)?);
            i += 2;
        } else if let Some(val) = a.strip_prefix("--platform=") {
            platform = Some(install::Platform::parse(val)?);
            i += 1;
        } else if a == "--project" || a == "--user" {
            let want = if a == "--project" {
                install::Scope::Project
            } else {
                install::Scope::User
            };
            if scope.is_some_and(|s| s != want) {
                return Err("--project and --user are mutually exclusive".to_string());
            }
            scope = Some(want);
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else {
            return Err(format!("unexpected argument: {a}"));
        }
    }
    Ok(doctor::DoctorOpts { platform, scope })
}

/// Shared by `mushroomdb enable` and `mushroomdb disable`: the same
/// `--platform` / `--project` / `--user` flags `doctor` takes, and nothing
/// else — neither command chooses a store or a binary, so there is no `--db`
/// or `--command` to parse.
fn parse_toggle_cmd(args: &[&str]) -> Result<install::ToggleOpts, String> {
    let mut platform: Option<install::Platform> = None;
    let mut scope: Option<install::Scope> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--platform" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --platform".to_string())?;
            platform = Some(install::Platform::parse(val)?);
            i += 2;
        } else if let Some(val) = a.strip_prefix("--platform=") {
            platform = Some(install::Platform::parse(val)?);
            i += 1;
        } else if a == "--project" || a == "--user" {
            let want = if a == "--project" {
                install::Scope::Project
            } else {
                install::Scope::User
            };
            if scope.is_some_and(|s| s != want) {
                return Err("--project and --user are mutually exclusive".to_string());
            }
            scope = Some(want);
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else {
            return Err(format!("unexpected argument: {a}"));
        }
    }
    Ok(install::ToggleOpts { platform, scope })
}

fn parse_ingest_git(args: &[&str]) -> Result<Command, String> {
    let mut positional = Vec::new();
    let mut exclude = Vec::new();
    let mut max_commits_per_file = ingest_git::DEFAULT_MAX_COMMITS_PER_FILE;
    let mut recurse_submodules = false;
    let mut prs = false;
    let mut structure = true;
    let mut docs = true;
    let mut ensure_gitignore = false;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--recurse-submodules" {
            recurse_submodules = true;
            i += 1;
        } else if a == "--prs" {
            prs = true;
            i += 1;
        } else if a == "--no-structure" {
            structure = false;
            i += 1;
        } else if a == "--no-docs" {
            docs = false;
            i += 1;
        } else if a == "--ensure-gitignore" {
            ensure_gitignore = true;
            i += 1;
        } else if a == "--exclude" {
            exclude.push(
                args.get(i + 1)
                    .copied()
                    .ok_or_else(|| "missing value for --exclude".to_string())?
                    .to_string(),
            );
            i += 2;
        } else if let Some(val) = a.strip_prefix("--exclude=") {
            exclude.push(val.to_string());
            i += 1;
        } else if a == "--max-commits-per-file" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --max-commits-per-file".to_string())?;
            max_commits_per_file = val
                .parse()
                .map_err(|e| format!("bad --max-commits-per-file: {e}"))?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--max-commits-per-file=") {
            max_commits_per_file = val
                .parse()
                .map_err(|e| format!("bad --max-commits-per-file: {e}"))?;
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else {
            positional.push(a);
            i += 1;
        }
    }
    let [db_dir, repo] = positional.as_slice() else {
        return Err("ingest-git requires <db-dir> <repo-dir>".into());
    };
    // Structure ingest reads the working tree, where a repository carries
    // build output and vendored dependencies that its history does not. A user
    // who states any pattern of their own is taken to mean exactly that set.
    if exclude.is_empty() {
        exclude = ingest_git::DEFAULT_EXCLUDES
            .iter()
            .map(|p| (*p).to_string())
            .collect();
    }
    Ok(Command::IngestGit {
        db_dir: PathBuf::from(db_dir),
        opts: ingest_git::IngestGitOpts {
            repo: PathBuf::from(repo),
            exclude,
            max_commits_per_file,
            recurse_submodules,
            prs,
            structure,
            docs,
            ensure_gitignore,
        },
    })
}

/// Parse argv after the binary name. Hand-rolled — no clap.
pub fn parse_args<S: AsRef<str>>(args: &[S]) -> Result<Command, String> {
    let args: Vec<&str> = args.iter().map(AsRef::as_ref).collect();
    if args.is_empty() {
        return Ok(Command::Help);
    }
    match args[0] {
        "--help" | "-h" | "help" => Ok(Command::Help),
        "--version" | "-V" | "version" => Ok(Command::Version),
        "serve" => parse_serve(&args[1..]),
        "mcp" => parse_mcp(&args[1..]),
        "stats" => parse_one_dir("stats", &args[1..]).map(|db_dir| Command::Stats { db_dir }),
        "demo" => parse_one_dir("demo", &args[1..]).map(|db_dir| Command::Demo { db_dir }),
        "suggest" => parse_one_dir("suggest", &args[1..]).map(|db_dir| Command::Suggest { db_dir }),
        "asof" => parse_asof(&args[1..]),
        "algo" => parse_algo(&args[1..]),
        "query" => parse_query(&args[1..]),
        "snapshot" => parse_snapshot(&args[1..]),
        "build-index" => parse_build_index(&args[1..]),
        "schema" => parse_schema(&args[1..]),
        "migrate" => parse_one_dir("migrate", &args[1..]).map(|db_dir| Command::Migrate { db_dir }),
        "verify" => parse_one_dir("verify", &args[1..]).map(|db_dir| Command::Verify { db_dir }),
        "backup" => parse_backup(&args[1..]),
        "export" => parse_export(&args[1..]),
        "recall" => parse_dir_or_auto("recall", &args[1..])
            .map(|(db_dir, auto)| Command::Recall { db_dir, auto }),
        "brief" => parse_dir_or_auto("brief", &args[1..])
            .map(|(db_dir, auto)| Command::Brief { db_dir, auto }),
        "why" => parse_positional("why", &args[1..], 2, 2).map(|(db_dir, rest)| Command::Why {
            db_dir,
            a: rest[0].clone(),
            b: rest[1].clone(),
        }),
        "ingest-git" => parse_ingest_git(&args[1..]),
        "install" => parse_install_cmd(&args[1..]).map(Command::Install),
        "uninstall" => parse_install_cmd(&args[1..]).map(Command::Uninstall),
        "disable" => parse_toggle_cmd(&args[1..]).map(Command::Disable),
        "enable" => parse_toggle_cmd(&args[1..]).map(Command::Enable),
        "doctor" => parse_doctor_cmd(&args[1..]).map(Command::Doctor),
        // Arguments are not parsed: whatever a 0.6 install wrote after the
        // subcommand, the answer is the same.
        "touch" => Ok(Command::RetiredHook { sub: "touch" }),
        "intercept" => Ok(Command::RetiredHook { sub: "intercept" }),
        "impact-hook" => Ok(Command::RetiredHook { sub: "impact-hook" }),
        "enrich" => Ok(Command::RetiredHook { sub: "enrich" }),
        "sync" => Ok(Command::RetiredHook { sub: "sync" }),
        other => Err(format!("unknown command: {other}")),
    }
}

fn default_addr() -> SocketAddr {
    SocketAddr::from(([127, 0, 0, 1], 8080))
}

fn parse_serve(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut addr = default_addr();
    let mut ui = ServeUi::Embedded;
    let mut saw_ui = false;
    let mut saw_no_ui = false;
    let mut demo_if_empty = false;
    let mut token = None;
    let mut role_tokens: Vec<(String, String)> = Vec::new();
    let mut snapshot_every = None;
    let mut restore_from: Option<PathBuf> = None;
    let mut tls_cert: Option<PathBuf> = None;
    let mut tls_key: Option<PathBuf> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--addr" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --addr".to_string())?;
            addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--addr=") {
            addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
            i += 1;
        } else if a == "--ui" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --ui".to_string())?;
            ui = ServeUi::Filesystem(PathBuf::from(val));
            saw_ui = true;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--ui=") {
            ui = ServeUi::Filesystem(PathBuf::from(val));
            saw_ui = true;
            i += 1;
        } else if a == "--no-ui" {
            ui = ServeUi::None;
            saw_no_ui = true;
            i += 1;
        } else if a == "--demo-if-empty" {
            demo_if_empty = true;
            i += 1;
        } else if a == "--token" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --token".to_string())?;
            token = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--token=") {
            token = Some(val.to_string());
            i += 1;
        } else if a == "--role-token" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --role-token".to_string())?;
            let (tok, role) = parse_role_token(val)?;
            role_tokens.push((tok, role));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--role-token=") {
            let (tok, role) = parse_role_token(val)?;
            role_tokens.push((tok, role));
            i += 1;
        } else if a == "--snapshot-every" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --snapshot-every".to_string())?;
            snapshot_every = Some(parse_snapshot_every(val)?);
            i += 2;
        } else if let Some(val) = a.strip_prefix("--snapshot-every=") {
            snapshot_every = Some(parse_snapshot_every(val)?);
            i += 1;
        } else if a == "--restore-from" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --restore-from".to_string())?;
            restore_from = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--restore-from=") {
            restore_from = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--tls-cert" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --tls-cert".to_string())?;
            tls_cert = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--tls-cert=") {
            tls_cert = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--tls-key" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --tls-key".to_string())?;
            tls_key = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--tls-key=") {
            tls_key = Some(PathBuf::from(val));
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    if saw_ui && saw_no_ui {
        return Err("cannot combine --ui and --no-ui".to_string());
    }
    match (&tls_cert, &tls_key) {
        (Some(_), None) => return Err("--tls-cert requires --tls-key".to_string()),
        (None, Some(_)) => return Err("--tls-key requires --tls-cert".to_string()),
        _ => {}
    }
    let db_dir = db_dir.ok_or_else(|| "serve requires <db-dir>".to_string())?;
    Ok(Command::Serve {
        db_dir,
        addr,
        ui,
        demo_if_empty,
        token,
        role_tokens,
        snapshot_every,
        restore_from,
        tls_cert,
        tls_key,
    })
}

fn parse_role_token(val: &str) -> Result<(String, String), String> {
    let (tok, role) = val
        .split_once(':')
        .ok_or_else(|| format!("--role-token requires TOKEN:ROLE format, got: {val}"))?;
    if tok.is_empty() {
        return Err("--role-token: TOKEN must not be empty".to_string());
    }
    if role.is_empty() {
        return Err("--role-token: ROLE must not be empty".to_string());
    }
    Ok((tok.to_string(), role.to_string()))
}

fn parse_snapshot_every(val: &str) -> Result<Duration, String> {
    let secs: u64 = val
        .parse()
        .map_err(|_| format!("invalid --snapshot-every: {val}"))?;
    if secs == 0 {
        return Err("--snapshot-every must be a positive number of seconds".into());
    }
    Ok(Duration::from_secs(secs))
}

/// `--ui <dir>` must be a directory that contains `index.html`.
pub fn validate_ui_dir(dir: &Path) -> Result<PathBuf, String> {
    if !dir.is_dir() {
        return Err(format!("--ui directory does not exist: {}", dir.display()));
    }
    let index = dir.join("index.html");
    if !index.is_file() {
        return Err(format!(
            "--ui directory is missing index.html: {}",
            dir.display()
        ));
    }
    Ok(dir.to_path_buf())
}

fn parse_asof(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut commit: Option<u64> = None;
    let mut at: Option<String> = None;
    let mut query: Option<String> = None;
    let mut namespace: Option<String> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--at" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --at".to_string())?;
            at = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--at=") {
            at = Some(val.to_string());
            i += 1;
        } else if a == "--commit" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --commit".to_string())?;
            commit = Some(
                val.parse()
                    .map_err(|_| format!("invalid commit index: {val}"))?,
            );
            i += 2;
        } else if let Some(val) = a.strip_prefix("--commit=") {
            commit = Some(
                val.parse()
                    .map_err(|_| format!("invalid commit index: {val}"))?,
            );
            i += 1;
        } else if a == "--query" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --query".to_string())?;
            query = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--query=") {
            query = Some(val.to_string());
            i += 1;
        } else if a == "--namespace" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --namespace".to_string())?;
            namespace = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--namespace=") {
            namespace = Some(val.to_string());
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "asof requires <db-dir>".to_string())?;
    match (commit.is_some(), at.is_some()) {
        (false, false) => {
            return Err("asof requires --commit N or --at <date>".to_string());
        }
        (true, true) => {
            return Err(
                "asof takes --commit or --at, not both: a precedence rule between a \
                 commit index and a date is how a caller ends up believing it asked \
                 for one and got the other"
                    .to_string(),
            );
        }
        _ => {}
    }
    Ok(Command::AsOf {
        db_dir,
        commit,
        at,
        query,
        namespace,
    })
}

/// Execute an as-of query at the given commit and print results.
///
/// `namespace` restricts the read to one namespace as it was at that commit —
/// a namespace cannot change, so that is simply the nodes which existed then and
/// are in it. A name no node uses answers with nothing, never with everything.
pub fn run_asof(
    db_dir: &Path,
    commit: Option<u64>,
    at: Option<&str>,
    query: Option<&str>,
    namespace: Option<&str>,
) -> Result<String, CliError> {
    let namespace = check_namespace(namespace)?;
    // A date has to be resolved against the store's recorded commit times, and
    // `open_at` needs the commit before it opens — so a dated call costs one
    // ordinary open first. The parser guarantees exactly one of the two is set.
    let commit = match (commit, at) {
        (Some(c), _) => c,
        (None, Some(date)) => GraphDb::open(db_dir)?.resolve_date(date)?,
        (None, None) => {
            return Err(CliError(
                "asof requires --commit N or --at <date>".to_string(),
            ))
        }
    };
    // Counts come off the opened handle: it knows the archives the live WAL no
    // longer holds, and where history now starts.
    let db = GraphDb::open_at(db_dir, commit)?;
    let total = db.wal_total_commits()?;
    let floor = db.wal_horizon_floor();
    let mut out = String::new();
    if floor == 0 {
        let _ = writeln!(out, "as-of commit {} of {}", commit, total);
    } else {
        let _ = writeln!(
            out,
            "as-of commit {} of {} (history reaches back to commit {})",
            commit, total, floor
        );
    }
    if let Some(cypher) = query {
        let params = BTreeMap::new();
        let rs = match namespace {
            Some(ns) => db.query_masked(cypher, &params, &db.mask_for_namespace(ns))?,
            None => db.query(cypher, &params)?,
        };
        out.push_str(&format_result_set(&rs));
    }
    Ok(out)
}

/// A `--namespace` value, refused here rather than resolved to an empty mask: a
/// typo that silently answers "nothing" reads like an empty store.
fn check_namespace(namespace: Option<&str>) -> Result<Option<&str>, CliError> {
    match namespace {
        Some(ns) if !valid_namespace(ns) => Err(CliError(format!(
            "namespace {ns:?} is not a valid namespace name — 1 to {NS_MAX_LEN} characters of \
             [A-Za-z0-9_.-]"
        ))),
        other => Ok(other),
    }
}

fn parse_query(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut query_flag: Option<String> = None;
    let mut role: Option<String> = None;
    let mut namespace: Option<String> = None;
    let mut cypher_parts: Vec<&str> = Vec::new();
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--query" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --query".to_string())?;
            query_flag = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--query=") {
            query_flag = Some(val.to_string());
            i += 1;
        } else if a == "--role" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --role".to_string())?;
            role = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--role=") {
            role = Some(val.to_string());
            i += 1;
        } else if a == "--namespace" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --namespace".to_string())?;
            namespace = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--namespace=") {
            namespace = Some(val.to_string());
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            cypher_parts.push(a);
            i += 1;
        }
    }
    let db_dir = db_dir.ok_or_else(|| "query requires <db-dir>".to_string())?;
    let cypher = if let Some(q) = query_flag {
        if !cypher_parts.is_empty() {
            return Err(
                "query: pass Cypher as remaining arguments or --query, not both".to_string(),
            );
        }
        q
    } else {
        if cypher_parts.is_empty() {
            return Err("query requires a Cypher string".to_string());
        }
        cypher_parts.join(" ")
    };
    Ok(Command::Query {
        db_dir,
        cypher,
        role,
        namespace,
    })
}

/// Run a Cypher read or write and print columns/rows like [`run_asof`].
///
/// `role` answers as one of the store's roles and `namespace` from one
/// namespace; together they **intersect**, so a role bound to one namespace
/// never sees another and naming a namespace outside its binding answers with
/// nothing. Either one makes the query a read: a restricted write is refused.
pub fn run_query(
    db_dir: &Path,
    cypher: &str,
    role: Option<&str>,
    namespace: Option<&str>,
) -> Result<String, CliError> {
    let namespace = check_namespace(namespace)?;
    let params = BTreeMap::new();
    if role.is_some() || namespace.is_some() {
        let db = GraphDb::open(db_dir)?;
        // The same never-widen composition every other surface uses: one mask
        // per leg, intersected.
        let mask = match (role, namespace) {
            (Some(role), Some(ns)) => db
                .mask_for_role(role)?
                .intersect(&db.mask_for_namespace(ns)),
            (Some(role), None) => db.mask_for_role(role)?,
            (None, Some(ns)) => db.mask_for_namespace(ns),
            (None, None) => unreachable!("one of the two is Some in this branch"),
        };
        return Ok(format_result_set(&db.query_masked(cypher, &params, &mask)?));
    }
    let is_write = is_write_query(cypher).map_err(CliError)?;
    let rs = if is_write {
        let mut db = GraphDb::open(db_dir)?;
        db.query_write(cypher, &params)?
    } else {
        let db = GraphDb::open(db_dir)?;
        db.query(cypher, &params)?
    };
    Ok(format_result_set(&rs))
}

fn parse_snapshot(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    // Archiving is the default: a snapshot the user did not ask to be
    // destructive should not cost them their history.
    let mut wal = WalDisposition::Archive;
    let mut retention: Option<u32> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--keep-wal" {
            wal = WalDisposition::Keep;
            i += 1;
        } else if a == "--truncate" {
            wal = WalDisposition::Truncate;
            i += 1;
        } else if a == "--archive-wal" {
            // Kept for the callers that spelled the default out.
            wal = WalDisposition::Archive;
            i += 1;
        } else if a.starts_with("--retention=") {
            let v = a.trim_start_matches("--retention=");
            retention = Some(
                v.parse::<u32>()
                    .map_err(|_| format!("--retention= expects a u32, got: {v}"))?,
            );
            i += 1;
        } else if a == "--retention" {
            i += 1;
            let v = args
                .get(i)
                .ok_or_else(|| "--retention requires a value".to_string())?;
            retention = Some(
                v.parse::<u32>()
                    .map_err(|e| format!("--retention value error: {e}"))?,
            );
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "snapshot requires <db-dir>".to_string())?;
    Ok(Command::Snapshot {
        db_dir,
        wal,
        retention,
    })
}

/// Migrate the snapshot at `db_dir` to the current format version.
///
/// - If the snapshot is already at the current version, prints
///   `already current (V<N>)`.
/// - If the snapshot is an older version, writes `snapshot.bin.bak` (atomic +
///   fsynced) then performs a truncating snapshot at the current version, and
///   prints `migrated V<from> -> V<current>`.
/// - WAL-only stores (no snapshot) are treated as needing a fresh snapshot.
pub fn run_migrate(db_dir: &Path) -> Result<String, CliError> {
    let current = core_api::SNAPSHOT_VERSION;
    let from_ver = core_api::snapshot_version_at(db_dir)?;

    // `>=`, not `==`: a store that has opted in to multiplicity writes V10, a
    // version above the default this binary writes. It is already current —
    // rewriting it would produce another V10 snapshot and report "V10 -> V9".
    if let Some(ver) = from_ver {
        if ver >= current {
            return Ok(format!("already current (V{ver})\n"));
        }
    }

    // Copy the original snapshot to .bak at OS level — no in-memory buffer
    // required for a 2+ GiB file.  The original snapshot.bin is authoritative
    // until snapshot_with's write_atomic (tmp+rename) succeeds, so a torn .bak
    // on crash is acceptable.
    if from_ver.is_some() {
        std::fs::copy(db_dir.join("snapshot.bin"), db_dir.join("snapshot.bin.bak"))?;
    }

    // Open with auto_migrate=false to avoid double-migration, then write
    // the truncating snapshot (CLI migrate always truncates the WAL).
    let mut db = GraphDb::open_with_options(
        db_dir,
        core_api::OpenOptions {
            auto_migrate: false,
            ..Default::default()
        },
    )?;
    db.snapshot()?;

    let msg = match from_ver {
        Some(ver) => format!("migrated V{ver} -> V{current}\n"),
        None => format!("migrated WAL-only -> V{current}\n"),
    };
    Ok(msg)
}

/// Validate the CRC32 integrity of every section in a V8 snapshot.
///
/// Exits with a non-zero code if any section is corrupt.  This is the
/// explicit integrity audit path; mushroomdb does NOT CRC-check large
/// sections on the hot query path (see format-stability.md).
pub fn run_verify(db_dir: &Path) -> Result<String, CliError> {
    // A store that has only ever been written via the WAL has no snapshot yet;
    // give an actionable message instead of a raw "No such file" io error.
    if !db_dir.join("snapshot.bin").exists() {
        return Err(CliError(format!(
            "verify: no snapshot found in {} — take one first with `mushroomdb snapshot {}`",
            db_dir.display(),
            db_dir.display()
        )));
    }
    let results = core_api::verify_snapshot(db_dir)
        .map_err(|e| CliError(format!("verify: cannot open snapshot: {e}")))?;
    let mut any_fail = false;
    let mut out = String::new();
    for (id, section_name, byte_len, result) in &results {
        match result {
            Ok(()) => {
                let _ = writeln!(
                    out,
                    "  section {:2} ({:<12}) {:>10} bytes  OK",
                    id, section_name, byte_len
                );
            }
            Err(msg) => {
                let _ = writeln!(
                    out,
                    "  section {:2} ({:<12}) {:>10} bytes  CORRUPT: {msg}",
                    id, section_name, byte_len
                );
                any_fail = true;
            }
        }
    }
    if any_fail {
        Err(CliError(format!("integrity check FAILED:\n{out}")))
    } else {
        Ok(format!(
            "integrity check OK ({} sections):\n{out}",
            results.len()
        ))
    }
}

/// What a snapshot does with the WAL it folds in.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum WalDisposition {
    /// Move it to `wal.<N>.archive`, where the history reads still find it.
    /// The default, and what every automatic snapshot does: a store should not
    /// forget how it got here in exchange for opening faster.
    #[default]
    Archive,
    /// Leave `wal.bin` whole. Every pre-snapshot commit stays in the live WAL,
    /// and every open replays all of it.
    Keep,
    /// Drop it. The smallest directory and the fastest open, at the price of
    /// every commit before this point: `node_history`, `edge_history`,
    /// `was_linked` and `open_at` stop reaching them, and any archives an
    /// earlier snapshot left become unreachable too.
    Truncate,
}

impl WalDisposition {
    fn options(self) -> SnapshotOptions {
        match self {
            WalDisposition::Archive => AUTOMATIC_SNAPSHOT,
            WalDisposition::Keep => SnapshotOptions {
                keep_wal: true,
                archive_wal: false,
            },
            WalDisposition::Truncate => SnapshotOptions {
                keep_wal: false,
                archive_wal: false,
            },
        }
    }
}

/// Open `dir` and write `snapshot.bin`, archiving the WAL by default.
pub fn run_snapshot(
    db_dir: &Path,
    wal: WalDisposition,
    retention: Option<u32>,
) -> Result<String, CliError> {
    let mut db = GraphDb::open(db_dir)?;
    if wal == WalDisposition::Archive {
        db.set_wal_archive_retention(retention);
    }
    db.snapshot_with(wal.options())?;
    Ok(format!(
        "snapshot written: {}\n",
        db_dir.join("snapshot.bin").display()
    ))
}

fn parse_build_index(args: &[&str]) -> Result<Command, String> {
    let mut db_dir: Option<PathBuf> = None;
    let mut rule: Option<String> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if let Some(v) = a.strip_prefix("--rule=") {
            if v.is_empty() {
                return Err("--rule requires a value".to_string());
            }
            rule = Some(v.to_string());
            i += 1;
        } else if a == "--rule" {
            i += 1;
            let v = args
                .get(i)
                .ok_or_else(|| "--rule requires a value".to_string())?;
            rule = Some((*v).to_string());
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "build-index requires <db-dir>".to_string())?;
    Ok(Command::BuildIndex { db_dir, rule })
}

/// Pump every outstanding vector-index build to completion, one slice per
/// call, printing a line per slice.
///
/// `rule` narrows the report to one rule; the pump itself always advances every
/// pending build, because they share one write lock and splitting them would
/// only mean holding it more often.
pub fn run_build_index(db_dir: &Path, rule: Option<&str>) -> Result<String, CliError> {
    let mut db = GraphDb::open(db_dir)?;
    build_index_on(&mut db, rule)
}

/// [`run_build_index`] against an already-open handle.
///
/// Terminates: every pump advances each pending build's cursor by at least one
/// node (the slice size is never zero), so the outstanding list empties.
///
/// Exposed for tests that need a build this handle itself deferred. Whether a
/// build is outstanding is decided when the rule is created, so a test using a
/// reduced slice size has to keep the handle that created the rule; reopening
/// replays `CreateRule` at the production slice size. A build a *snapshot* cut
/// short needs no such care — the reopen recognises it, which is what
/// [`run_build_index`] relies on.
pub fn build_index_on(db: &mut GraphDb<RealFs>, rule: Option<&str>) -> Result<String, CliError> {
    let mut out = String::new();
    let interesting = |name: &str| rule.is_none_or(|r| r == name);
    // Read before pumping, so that "no such rule" and "that rule is already
    // built" can be told apart in the empty case below.
    let known_rules: Vec<String> = db.stats().rules.iter().map(|r| r.name.clone()).collect();

    // Pump before looking. Open registers a build a snapshot cut short, but a
    // resumed build can still be registered and finished inside a single call,
    // so the outstanding list alone cannot show that anything happened.
    loop {
        let (finished, outstanding) = db.pump_index_build_reporting()?;
        for b in &outstanding {
            if interesting(&b.rule) {
                out.push_str(&format!("building {}: {}/{}\n", b.rule, b.indexed, b.total));
            }
        }
        // Reported from the pump rather than inferred from a shrinking
        // outstanding list: a resumed build is registered and finished inside a
        // single call, so it never appears in that list at all.
        for b in &finished {
            if interesting(&b.rule) {
                out.push_str(&format!("built {}: {} vectors\n", b.rule, b.total));
            }
        }
        if outstanding.is_empty() {
            break;
        }
    }
    if out.is_empty() {
        match rule {
            // Distinguish the two ways this can be empty, because they need
            // different things from the operator: a finished build is fine, a
            // name that is not a rule is a typo.
            Some(r) if !known_rules.iter().any(|k| k == r) => out.push_str(&format!(
                "no rule named {r:?} in this store; nothing to build\n"
            )),
            Some(r) => out.push_str(&format!("nothing to build for rule {r:?}\n")),
            None => out.push_str("nothing to build\n"),
        }
    }
    Ok(out)
}

fn parse_schema(args: &[&str]) -> Result<Command, String> {
    if args.is_empty() {
        return Err("schema requires a subcommand: apply".to_string());
    }
    match args[0] {
        "apply" => parse_schema_apply(&args[1..]),
        other => Err(format!(
            "unknown schema subcommand: {other}; expected apply"
        )),
    }
}

fn parse_schema_apply(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut schema_file = None;
    let mut memory_defaults = false;
    let mut memory_identity = false;
    for a in args {
        if *a == "--memory-defaults" {
            memory_defaults = true;
        } else if *a == "--memory-identity" {
            memory_identity = true;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(*a));
        } else if schema_file.is_none() {
            schema_file = Some(PathBuf::from(*a));
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "schema apply requires <db-dir>".to_string())?;
    if memory_identity {
        if memory_defaults || schema_file.is_some() {
            return Err(
                "schema apply --memory-identity takes no schema file and no other preset"
                    .to_string(),
            );
        }
        return Ok(Command::SchemaApply {
            db_dir,
            schema_file: None,
            memory_defaults: false,
            memory_identity: true,
        });
    }
    if memory_defaults {
        if schema_file.is_some() {
            return Err("schema apply --memory-defaults takes no schema file argument".to_string());
        }
        return Ok(Command::SchemaApply {
            db_dir,
            schema_file: None,
            memory_defaults: true,
            memory_identity: false,
        });
    }
    let schema_file =
        schema_file.ok_or_else(|| "schema apply requires <schema.json>".to_string())?;
    Ok(Command::SchemaApply {
        db_dir,
        schema_file: Some(schema_file),
        memory_defaults: false,
        memory_identity: false,
    })
}

/// Read `schema_file`, open `db_dir`, apply the schema, and return the diff
/// as one line per entry: `"created rule:x"`, `"updated view:y"`, etc.
pub fn run_schema_apply(db_dir: &Path, schema_file: &Path) -> Result<String, CliError> {
    let json = std::fs::read_to_string(schema_file)
        .map_err(|e| CliError(format!("cannot read {}: {e}", schema_file.display())))?;
    let schema: Schema = serde_json::from_str(&json).map_err(|e| {
        CliError(format!(
            "invalid schema JSON in {}: {e}",
            schema_file.display()
        ))
    })?;
    let mut db = GraphDb::open(db_dir)?;
    let diff = db.apply_schema(&schema)?;
    let mut out = String::new();
    for entry in &diff.created {
        let _ = writeln!(out, "created {entry}");
    }
    for entry in &diff.updated {
        let _ = writeln!(out, "updated {entry}");
    }
    for entry in &diff.unchanged {
        let _ = writeln!(out, "unchanged {entry}");
    }
    if diff.created.is_empty() && diff.updated.is_empty() && diff.unchanged.is_empty() {
        let _ = writeln!(out, "schema applied: nothing to do (empty schema)");
    }
    Ok(out)
}

/// `mushroomdb schema apply <db> --memory-defaults`
///
/// Applies `core_api::memory_schema::memory_defaults()`. Kept separate from
/// `run_schema_apply` because it takes no schema file, and separate from
/// `open` because declaring full-text on a populated store rebuilds the index
/// at open — the caller is told what it will cost before it happens.
pub fn run_schema_apply_memory_defaults(db_dir: &Path) -> Result<String, CliError> {
    let mut db = GraphDb::open(db_dir)?;
    let before = db.stats();
    let diff = db.apply_schema(&core_api::memory_schema::memory_defaults())?;
    let mut out = String::new();
    if diff.created.is_empty() && diff.updated.is_empty() {
        let _ = writeln!(out, "schema apply: already current, nothing to do");
        return Ok(out);
    }
    let _ = writeln!(
        out,
        "schema apply: {} created, {} updated, {} unchanged",
        diff.created.len(),
        diff.updated.len(),
        diff.unchanged.len()
    );
    for name in diff.created.iter().chain(diff.updated.iter()) {
        let _ = writeln!(out, "  {name}");
    }
    let _ = writeln!(
        out,
        "note: {} live nodes are now full-text indexed; the index rebuilds at \
         every open from here on.",
        before.nodes_live
    );
    Ok(out)
}

/// `mushroomdb schema apply <db> --memory-identity`
///
/// Applies `core_api::memory_schema::memory_identity()` — opt-in, because a
/// store never gains a rule silently. What it will do is measured and stated
/// first, in the same output, before anything is written: how many entity
/// nodes the rules will compare, and how many need an `aliases` list written
/// before they can. Then the lists are written in one commit and the rules
/// created, each with its own backfill.
///
/// `aliases` is what a node's key and name imply. A node that already carries
/// items those do not imply — a user's own `aliases` property from before
/// 0.7 — keeps them: the same commit moves them to `alias_keys`, the list the
/// five claim rules read, and the output says how many nodes that touched.
/// Nothing is dropped. On a store that already carries the preset's
/// earlier eleven rules, this adds the five claim rules and leaves the rest
/// unchanged.
pub fn run_schema_apply_memory_identity(db_dir: &Path) -> Result<String, CliError> {
    use core_api::memory::identity::{
        aliases_to_backfill, entity_node_count, write_aliases, ALIASES_FIELD, ALIAS_KEYS_FIELD,
        SAME_AS_EDGE,
    };
    let mut db = GraphDb::open(db_dir)?;
    let preset = core_api::memory_schema::memory_identity();
    let entity_nodes = entity_node_count(&db);
    let backfill = aliases_to_backfill(&db)?;
    let mut out = String::new();
    let _ = writeln!(
        out,
        "identity preset: {} SAME_AS rules over each entity's aliases, global — they link \
         across namespaces",
        preset.rules.len()
    );
    let writes = |field: &str| backfill.iter().filter(|node| node.sets(field)).count();
    let _ = writeln!(
        out,
        "before applying: {entity_nodes} entity node(s); {} need an aliases list written \
         first, from the key and the name; each rule then compares aliases across its labels \
         once, and every later write to an entity is re-checked",
        writes(ALIASES_FIELD)
    );
    let moved = writes(ALIAS_KEYS_FIELD);
    if moved == 0 {
        let _ = writeln!(
            out,
            "alias_keys: nothing to move — it holds only declared aliases; an entity that \
             declares one equal to a provisional stub's key links that stub"
        );
    } else {
        let _ = writeln!(
            out,
            "alias_keys: {moved} node(s) carry aliases their key and name do not imply; those \
             are kept, moved to alias_keys as declared aliases — one equal to a provisional \
             stub's key links that stub"
        );
    }
    // Two commits: the aliases, then the rules. A failure between them leaves
    // the lists written and no rule; a re-run recovers, since the backfill is
    // idempotent and finds nothing left to write.
    write_aliases(&mut db, &backfill)?;
    let diff = db.apply_schema(&preset)?;
    let _ = writeln!(
        out,
        "schema apply: {} created, {} updated, {} unchanged",
        diff.created.len(),
        diff.updated.len(),
        diff.unchanged.len()
    );
    // Edges are directed and a same-label rule derives both directions, so
    // the pair count is the number of identities claimed.
    let edges = db.weighted_edges(SAME_AS_EDGE, None);
    let pairs: std::collections::BTreeSet<(&str, &str)> = edges
        .iter()
        .map(|(a, b, _)| {
            if a <= b {
                (a.as_str(), b.as_str())
            } else {
                (b.as_str(), a.as_str())
            }
        })
        .collect();
    let _ = writeln!(
        out,
        "SAME_AS edges now: {} ({} pair(s))",
        edges.len(),
        pairs.len()
    );
    Ok(out)
}

fn parse_backup(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut dest = None;
    for a in args {
        if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        }
        if db_dir.is_none() {
            db_dir = Some(PathBuf::from(*a));
        } else if dest.is_none() {
            dest = Some(PathBuf::from(*a));
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "backup requires <db-dir>".to_string())?;
    let dest = dest.ok_or_else(|| "backup requires <dest>".to_string())?;
    Ok(Command::Backup { db_dir, dest })
}

fn parse_export(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut dest = None;
    let mut format = ExportFormat::Jsonl;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--format" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --format".to_string())?;
            format = ExportFormat::parse(val).ok_or_else(|| {
                format!("unknown format '{val}'; expected jsonl, parquet, or graphml")
            })?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--format=") {
            format = ExportFormat::parse(val).ok_or_else(|| {
                format!("unknown format '{val}'; expected jsonl, parquet, or graphml")
            })?;
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else if dest.is_none() {
            dest = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| "export requires <db-dir>".to_string())?;
    let dest = dest.ok_or_else(|| "export requires <dest>".to_string())?;
    Ok(Command::Export {
        db_dir,
        dest,
        format,
    })
}

/// Create a consistent, verified backup of `db_dir` to `dest`.
pub fn run_backup(db_dir: &Path, dest: &Path) -> Result<BackupReport, CliError> {
    let db = GraphDb::open(db_dir)?;
    Ok(db.backup_to(dest)?)
}

/// [`core_api::restore::restore_if_empty`], in the CLI's error type.
///
/// The implementation moved down into `core-api` in v0.6.10 so the Python
/// binding could reach it without taking a dependency on this crate; the
/// semantics, the messages and [`RestoreOutcome`] are unchanged. The engine
/// carries the message inside [`core_api::GraphError::Io`], whose own `Display`
/// adds a prefix — unwrapped here, so `serve` prints what it always printed.
pub fn restore_if_empty(db_dir: &Path, from: &Path) -> Result<RestoreOutcome, CliError> {
    core_api::restore::restore_if_empty(db_dir, from).map_err(|e| match e {
        core_api::GraphError::Io(io) => CliError(io.to_string()),
        other => CliError(other.to_string()),
    })
}

/// Format a [`BackupReport`] for display.
pub fn format_backup(dest: &Path, report: &BackupReport) -> String {
    let mut out = String::new();
    writeln!(out, "backup to: {}", dest.display()).unwrap();
    writeln!(out, "  files: {}", report.files.join(", ")).unwrap();
    writeln!(out, "  bytes: {}", report.bytes).unwrap();
    writeln!(out, "  verified: {}", report.verified).unwrap();
    out
}

/// Export all data from `db_dir` to `dest` in `format`.
pub fn run_export(db_dir: &Path, dest: &Path, format: &ExportFormat) -> Result<String, CliError> {
    let db = GraphDb::open(db_dir)?;
    let nodes = db.all_nodes_for_export();
    let edges = db.all_edges_for_export();
    let mut rules = db.rules();
    rules.sort_by(|a, b| a.name.cmp(&b.name));
    let node_count = nodes.len();
    let edge_count = edges.len();
    let rule_count = rules.len();
    match format {
        ExportFormat::Jsonl => {
            export::write_jsonl(&nodes, &edges, &rules, dest)?;
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges, {} rules\n",
                dest.display(),
                format.name(),
                node_count,
                edge_count,
                rule_count
            ))
        }
        ExportFormat::Parquet => {
            export::write_parquet(&nodes, &edges, &rules, dest)?;
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges, {} rules\n",
                dest.display(),
                format.name(),
                node_count,
                edge_count,
                rule_count
            ))
        }
        // GraphML has no rule analogue: only nodes and edges are written.
        ExportFormat::Graphml => {
            let file_path = export::write_graphml(&nodes, &edges, dest)?;
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges\n",
                file_path.display(),
                format.name(),
                node_count,
                edge_count,
            ))
        }
    }
}

fn format_result_set(rs: &ResultSet) -> String {
    let mut out = String::new();
    let _ = writeln!(out, "columns: {}", rs.columns().join(", "));
    for i in 0..rs.len() {
        let cells: Vec<String> = rs
            .columns()
            .iter()
            .map(|c| format!("{c}={}", fmt_cell(rs.get(i, c))))
            .collect();
        let _ = writeln!(out, "  {}", cells.join("  "));
    }
    out
}

fn parse_algo(args: &[&str]) -> Result<Command, String> {
    if args.is_empty() {
        return Err(
            "algo requires a subcommand: pagerank | wcc | degree | communities".to_string(),
        );
    }
    let subcmd = match args[0] {
        "pagerank" => AlgoSubcmd::Pagerank,
        "wcc" => AlgoSubcmd::Wcc,
        "degree" => AlgoSubcmd::Degree,
        "communities" => AlgoSubcmd::Communities,
        other => {
            return Err(format!(
                "unknown algo subcommand: {other}; expected pagerank | wcc | degree | communities"
            ))
        }
    };
    let rest = &args[1..];
    let mut db_dir = None;
    let mut top: usize = 20;
    let mut dir = AlgoDir::Both;
    let mut edge_types: Vec<String> = Vec::new();
    let mut weight_prop: Option<String> = None;
    let mut min_weight: Option<f64> = None;
    let mut i = 0;
    while i < rest.len() {
        let a = rest[i];
        if a == "--top" {
            let val = rest
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --top".to_string())?;
            top = val
                .parse()
                .map_err(|_| format!("--top must be a non-negative integer, got {val}"))?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--top=") {
            top = val
                .parse()
                .map_err(|_| format!("--top must be a non-negative integer, got {val}"))?;
            i += 1;
        } else if a == "--dir" {
            let val = rest
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --dir".to_string())?;
            dir = parse_algo_dir(val)?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--dir=") {
            dir = parse_algo_dir(val)?;
            i += 1;
        } else if a == "--edge-type" {
            let val = rest
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --edge-type".to_string())?;
            edge_types.push(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--edge-type=") {
            edge_types.push(val.to_string());
            i += 1;
        } else if a == "--weight-prop" {
            let val = rest
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --weight-prop".to_string())?;
            weight_prop = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--weight-prop=") {
            weight_prop = Some(val.to_string());
            i += 1;
        } else if a == "--min-weight" {
            let val = rest
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --min-weight".to_string())?;
            min_weight = Some(
                val.parse()
                    .map_err(|_| format!("--min-weight must be a number, got {val}"))?,
            );
            i += 2;
        } else if let Some(val) = a.strip_prefix("--min-weight=") {
            min_weight = Some(
                val.parse()
                    .map_err(|_| format!("--min-weight must be a number, got {val}"))?,
            );
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    let db_dir = db_dir.ok_or_else(|| format!("algo {} requires <db-dir>", args[0]))?;
    Ok(Command::Algo {
        db_dir,
        subcmd,
        top,
        dir,
        edge_types,
        weight_prop,
        min_weight,
    })
}

/// Parse the `--dir` value for `algo` into an [`AlgoDir`].
fn parse_algo_dir(val: &str) -> Result<AlgoDir, String> {
    match val.to_ascii_lowercase().as_str() {
        "out" => Ok(AlgoDir::Out),
        "in" => Ok(AlgoDir::In),
        "both" => Ok(AlgoDir::Both),
        other => Err(format!("--dir must be one of out | in | both, got {other}")),
    }
}

/// Body of `mushroomdb brief <db-dir>|--auto`, the `SessionStart` hook.
///
/// Byte-stable for a given store: the host caches this output for the whole
/// session, so two prompts of the same session must not disagree about what
/// the repository is. Opened read-only like every other question.
pub fn run_brief(db_dir: &Path) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    let reach = reach_line(db_dir);
    let b = core_api::memory::brief::brief(&db, &core_api::memory::brief::BriefOptions::default());
    let mut text = core_api::memory::brief::render(&b, &reach);
    // An empty store's line names how to fill it, and `render` cannot know
    // which doors this install opened. A cli-only install gets the shell form.
    if text == core_api::memory::brief::EMPTY_BRIEF
        && matches!(install::delivery_for_store(db_dir), install::Delivery::Cli)
    {
        text = empty_brief_cli(db_dir);
    }
    // Said here because a brief is computed once per session and cached,
    // where the prompt hook fires every turn. A 0.6.x store upgraded to 0.7
    // has no text index until someone asks for one, and a user who never
    // calls the `recall` *tool* would otherwise never learn why the hook went
    // quiet.
    //
    // Placed above the reach line rather than after it: a brief ends with how
    // to reach the graph, and the tests holding that line read it as the last.
    // An empty store's brief has no reach line, so there it simply ends.
    //
    // The path is quoted and sanitized as the reach line's is: quoted so a
    // space or a quote in it stays one argument, sanitized because this line
    // is inserted after `render` sanitized everything else.
    if db.fulltext_pairs().is_empty() {
        let notice = format!(
            "no text index: recall cannot match anything here. \
             Run: mushroomdb schema apply {} --memory-defaults\n",
            core_api::digest::sanitize(&install::sh_quote(&db_dir.to_string_lossy()))
        );
        let at = text
            .rfind("\nreach the graph: ")
            .map_or(text.len(), |i| i + 1);
        text.insert_str(at, &notice);
    }
    Ok(text)
}

/// How a Cypher statement is typed at a shell, in the words every place that
/// prints a shell `query` uses: this brief's reach line, the CLI-only
/// empty-store line, and the skill's CLI table.
///
/// The statement is one double-quoted argument because the engine's Cypher
/// takes single-quoted strings only. A shell expands `$` and a backtick inside
/// double quotes, so a fact holding either would run a command on its way in;
/// the rule is in words rather than the characters themselves so it reads the
/// same in a Markdown table as in a plain line.
const SHELL_QUOTING_RULE: &str = "single-quote Cypher strings; inside the double quotes \
                                  backslash every dollar sign, double quote and backtick";

/// The brief's last line: how to reach the graph from this session.
///
/// Two doors, because a session may have either one open — the MCP tool, and
/// the same question typed at a shell. The command names the binary the way
/// `install` would resolve it right now, which is the same resolution the
/// hooks themselves were written with, and it names the store, because both
/// tools take one.
///
/// **Every MCP name here has to be one `tools/list` actually lists.** A
/// session can only call what its client was shown, so naming a tool it
/// cannot see would be worse than naming none — which is also why a `cli`
/// install, registering no server at all, gets the shell form alone. Every
/// store is served [`server::ASSOCIATION_TOOLS`], one written by `ingest-git`
/// included, so every store is reached through `explain_association` and
/// `query`, which is where its entities are.
/// `the_reach_line_names_only_tools_its_surface_lists` holds this line and
/// that list in step.
///
/// The shell half names `query` alone. `mushroomdb why <db> <a> <b>` is
/// `explain_association`'s shell door, but it answers one pair, and `query`
/// answers that question and every other the brief's recipes ask, a Cypher
/// read away.
fn reach_line(db_dir: &Path) -> String {
    let bin = install::detect_mcp_command(None).shell();
    let db = install::sh_quote(&db_dir.to_string_lossy());
    let tools = format!(
        "explain_association <a> <b>{sep}query '<cypher>' (MCP tools; add role: <name> \
         or namespace: <ns> to narrow what it sees){sep}or:",
        sep = core_api::digest::SEP
    );
    let shell = format!("{bin} query {db} \"<cypher>\" ({SHELL_QUOTING_RULE})");
    match install::delivery_for_store(db_dir) {
        install::Delivery::Cli => shell,
        _ => format!("{tools} {shell}"),
    }
}

/// An empty store's brief for an install that registered no MCP server.
///
/// [`core_api::memory::brief::EMPTY_BRIEF`] offers `remember`,
/// `upsert_entity` and `ingest_json`, which are MCP tools; a `--delivery cli`
/// session has a shell and nothing else. This is the same offer in the one
/// form that session can act on — a `query` with a `CREATE`, the write the
/// skill's CLI table names, with the `id:` property a `CREATE` needs.
///
/// The statement is in double quotes for the shell because the engine's
/// Cypher takes single-quoted strings only: the line has to run as a shell
/// would run it, and
/// `an_empty_store_brief_on_a_cli_delivery_install_names_no_mcp_tool` runs
/// it, on a store path with a space and a quote in it. The fact a session
/// fills in is arbitrary text inside those double quotes, so the line ends
/// with [`SHELL_QUOTING_RULE`].
///
/// The id and the text are placeholders, as the skill's `note:…` is. A
/// literal id would be the same key on every paste, and the key is the node.
fn empty_brief_cli(db_dir: &Path) -> String {
    // Sanitized as the reach line is: `render` returned a constant for this
    // store, so nothing downstream sanitizes what is built here.
    let bin = core_api::digest::sanitize(&install::detect_mcp_command(None).shell());
    let db = core_api::digest::sanitize(&install::sh_quote(&db_dir.to_string_lossy()));
    format!(
        "mushroomdb brief — empty store; offer to fill it: {bin} query {db} \
         \"CREATE (n:Note {{id: 'note:<fresh-id>', text: '<the fact>'}})\" writes a fact; \
         fill in both, with an id no note has yet; {SHELL_QUOTING_RULE}\n"
    )
}

/// Open a store the way every question about it is asked: read-only, with both
/// write paths off, so asking never migrates a snapshot, rewrites a torn WAL
/// tail, or makes a writer wait on the cross-process lock.
///
/// A directory that is not there is an error rather than an empty store:
/// `RealFs::new` runs `create_dir_all`, so without this guard a `brief` hook
/// left behind by an uninstall — or any read of a mistyped path — would create
/// the very store it then reports as empty. The same guard `run_recall` opens
/// behind.
fn open_for_reading(db_dir: &Path) -> Result<structure::Db, CliError> {
    if !db_dir.exists() {
        return Err(CliError(format!("no store at {}", db_dir.display())));
    }
    Ok(GraphDb::open_with_options(
        db_dir,
        core_api::OpenOptions {
            auto_migrate: false,
            repair_wal: false,
            read_only: true,
        },
    )?)
}

/// Body of `mushroomdb why <db> <a> <b>`: every rule edge between two keys
/// with the evidence that derived it, or an honest note that there is none.
/// A key the store does not hold is an error (`node key not found: <key>`),
/// not an empty answer.
///
/// The reply is what the `explain_association` tool sends, framing line
/// included: the rule names and matched values are graph content.
pub fn run_why(db_dir: &Path, a: &str, b: &str) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    let explained = core_api::explain_digest::explain_with_evidence(&db, a, b)?;
    Ok(format!(
        "{}{}",
        core_api::digest::UNTRUSTED_FRAMING,
        core_api::explain_digest::render_explain(a, b, &explained)
    ))
}

/// Run a graph algorithm and return a formatted string.
///
/// `dir` selects the edge direction for `degree` and `pagerank`; `wcc` and
/// `communities` are always undirected and ignore it. `edge_types` /
/// `weight_prop` / `min_weight` are used by `communities` only.
#[allow(clippy::too_many_arguments)]
pub fn run_algo(
    db_dir: &Path,
    subcmd: &AlgoSubcmd,
    top: usize,
    dir: AlgoDir,
    edge_types: Vec<String>,
    weight_prop: Option<String>,
    min_weight: Option<f64>,
) -> Result<String, CliError> {
    let db = GraphDb::open(db_dir)?;
    match subcmd {
        AlgoSubcmd::Pagerank => {
            let config = PageRankConfig {
                direction: dir,
                ..PageRankConfig::default()
            };
            let report = db.pagerank(&config);
            Ok(format_pagerank(&report, top))
        }
        AlgoSubcmd::Wcc => {
            let config = WccConfig::default();
            let report = db.connected_components(&config);
            Ok(format_wcc(&report, top))
        }
        AlgoSubcmd::Degree => {
            let config = DegreeConfig {
                direction: dir,
                ..DegreeConfig::default()
            };
            let report = db.degree_centrality(&config);
            Ok(format_degree(&report, top))
        }
        AlgoSubcmd::Communities => {
            let config = LouvainConfig {
                edge_types,
                weight_prop,
                min_weight,
                ..LouvainConfig::default()
            };
            let report = db.communities(&config);
            Ok(format_communities(&report, top))
        }
    }
}

fn format_pagerank(report: &core_api::PageRankReport, top: usize) -> String {
    let mut buf = String::new();
    let _ = writeln!(buf, "== pagerank (converged={}) ==", report.converged);
    let rows = if top == 0 {
        report.scores.as_slice()
    } else {
        &report.scores[..top.min(report.scores.len())]
    };
    for (i, (key, score)) in rows.iter().enumerate() {
        let _ = writeln!(buf, "  {:>4}  {:<40}  {:.6}", i + 1, key, score);
    }
    buf
}

fn format_wcc(report: &core_api::WccReport, top: usize) -> String {
    let mut buf = String::new();
    let _ = writeln!(buf, "== wcc (truncated={}) ==", report.truncated);
    let rows = if top == 0 {
        report.components.as_slice()
    } else {
        &report.components[..top.min(report.components.len())]
    };
    for (key, comp_id) in rows {
        let _ = writeln!(buf, "  {:<40}  component={}", key, comp_id);
    }
    buf
}

fn format_degree(report: &core_api::DegreeReport, top: usize) -> String {
    let mut buf = String::new();
    let _ = writeln!(
        buf,
        "== degree centrality (truncated={}) ==",
        report.truncated
    );
    let rows = if top == 0 {
        report.scores.as_slice()
    } else {
        &report.scores[..top.min(report.scores.len())]
    };
    for (i, (key, deg)) in rows.iter().enumerate() {
        let _ = writeln!(buf, "  {:>4}  {:<40}  degree={}", i + 1, key, deg);
    }
    buf
}

/// One line per community: id, size, cohesion, first 3 members.
/// Prints `(truncated)` in the header when the time budget fired.
fn format_communities(report: &core_api::CommunityReport, top: usize) -> String {
    let mut buf = String::new();
    let trunc = if report.truncated { " (truncated)" } else { "" };
    let _ = writeln!(
        buf,
        "== communities (modularity={:.2}){trunc} ==",
        report.modularity
    );
    let rows = if top == 0 {
        report.communities.as_slice()
    } else {
        &report.communities[..top.min(report.communities.len())]
    };
    for c in rows {
        let preview: Vec<&str> = c.members.iter().take(3).map(String::as_str).collect();
        let _ = writeln!(
            buf,
            "  {:>4}  size={:<6} cohesion={:<6.2} members=[{}]",
            c.id,
            c.members.len(),
            c.cohesion,
            preview.join(", ")
        );
    }
    buf
}

/// `<db-dir>` or `--auto`, for the commands a hook line invokes.
///
/// Exactly one of the two: `--auto` says "work it out from the environment",
/// which a stated path contradicts rather than refines.
fn parse_dir_or_auto(cmd: &str, args: &[&str]) -> Result<(Option<PathBuf>, bool), String> {
    let mut db_dir = None;
    let mut auto = false;
    for a in args {
        if *a == "--auto" {
            auto = true;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_some() {
            return Err(format!("unexpected extra argument: {a}"));
        } else {
            db_dir = Some(PathBuf::from(*a));
        }
    }
    match (&db_dir, auto) {
        (Some(_), true) => Err(format!("{cmd}: --auto takes no <db-dir>")),
        (None, false) => Err(format!("{cmd} requires <db-dir> or --auto")),
        _ => Ok((db_dir, auto)),
    }
}

/// `mcp [<db-dir>|--auto] [--all-tools]`. Every other flag is
/// [`parse_dir_or_auto`]'s to reject, so `--all-tools` is stripped here and
/// the rest of the line parses exactly as `recall`'s does.
fn parse_mcp(args: &[&str]) -> Result<Command, String> {
    let all_tools = args.contains(&"--all-tools");
    let rest: Vec<&str> = args
        .iter()
        .copied()
        .filter(|a| *a != "--all-tools")
        .collect();
    parse_dir_or_auto("mcp", &rest).map(|(db_dir, auto)| Command::Mcp {
        db_dir,
        auto,
        all_tools,
    })
}

/// `<db-dir>` followed by between `min` and `max` further arguments, none of
/// which may look like a flag.
///
/// The graph tools take keys — paths and symbol names — and a key beginning
/// with `-` is far more likely to be a typo'd flag than a file called `-x`, so
/// it is refused rather than looked up and reported missing.
fn parse_positional(
    cmd: &str,
    args: &[&str],
    min: usize,
    max: usize,
) -> Result<(PathBuf, Vec<String>), String> {
    let mut rest: Vec<String> = Vec::new();
    let mut db_dir: Option<PathBuf> = None;
    for a in args {
        if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        }
        match db_dir {
            None => db_dir = Some(PathBuf::from(*a)),
            Some(_) => rest.push((*a).to_string()),
        }
    }
    let db_dir = db_dir.ok_or_else(|| format!("{cmd} requires <db-dir>"))?;
    if rest.len() < min {
        return Err(format!(
            "{cmd} requires <db-dir> and {min} more argument{}",
            if min == 1 { "" } else { "s" }
        ));
    }
    if rest.len() > max {
        return Err(format!("unexpected extra argument: {}", rest[max]));
    }
    Ok((db_dir, rest))
}

fn parse_one_dir(cmd: &str, args: &[&str]) -> Result<PathBuf, String> {
    let mut db_dir = None;
    for a in args {
        if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        }
        if db_dir.is_some() {
            return Err(format!("unexpected extra argument: {a}"));
        }
        db_dir = Some(PathBuf::from(*a));
    }
    db_dir.ok_or_else(|| format!("{cmd} requires <db-dir>"))
}

/// Pretty-print [`Stats`] for `mushroomdb stats` and the demo smoke test.
pub fn format_stats(stats: &Stats) -> String {
    let mut out = String::new();
    let _ = writeln!(
        out,
        "nodes: {} live, {} tombstoned",
        stats.nodes_live, stats.nodes_tombstoned
    );
    let _ = writeln!(out, "edges: {}", stats.edges);
    if stats.history_floor == 0 {
        let _ = writeln!(out, "history: complete (nothing pruned)");
    } else {
        let _ = writeln!(
            out,
            "history: reaches back to commit {}",
            stats.history_floor
        );
    }
    // A store that names no namespace is one implicit `default` namespace, and
    // saying so would be noise on every single-tenant store — so the line
    // appears only once there is more than one, which keeps existing output
    // byte-identical.
    if stats.namespaces.len() > 1 {
        let names: Vec<String> = stats
            .namespaces
            .iter()
            .map(|n| format!("{} ({})", n.name, n.nodes_live))
            .collect();
        let _ = writeln!(out, "namespaces: {}", names.join(", "));
    }
    let _ = writeln!(out, "rules: {}", stats.rules.len());
    for r in &stats.rules {
        let _ = writeln!(
            out,
            "  {:<28} edges={}  tripped={}",
            r.name, r.edges, r.tripped
        );
    }
    out
}

/// Open `dir` and return live stats.
pub fn read_stats(dir: &Path) -> Result<Stats, CliError> {
    let db = SharedDb::open(dir)?;
    let stats = db.read().stats();
    Ok(stats)
}

/// Build the deterministic demo dataset in an empty `dir`.
///
/// Refuses if `dir` already exists and is not empty. Ingests 10 Orgs, 20
/// Projects, 30 People via [`SharedDb`] / `ingest_json` (auto-FK on `*_id`)
/// then declares `skill_fit` plus the three Predicates II rules.
pub fn run_demo(dir: &Path) -> Result<DemoOutcome, CliError> {
    refuse_non_empty(dir)?;

    let db = SharedDb::open(dir)?;
    let opts = IngestOptions::default();
    let mut auto_fk_rules = Vec::new();

    {
        let mut w = db.write();
        // A demo store is the first store most readers open. Declare the
        // memory schema before the first write so `recall` can answer on it.
        w.apply_schema(&core_api::memory_schema::memory_defaults())?;
        for (label, json) in [
            ("Org", org_json()),
            ("Project", project_json()),
            ("Person", person_json()),
        ] {
            let report = w.ingest_json(label, &json, &opts)?;
            if !report.row_errors.is_empty() {
                return Err(CliError(format!(
                    "demo ingest of {label} had row errors: {:?}",
                    report.row_errors
                )));
            }
            auto_fk_rules.extend(report.rules_created);
        }
        let skill_fit = Predicate::Overlap {
            field: "skills".into(),
            min: 0.5,
        };
        let skill_fit_k = Some(default_max_edges(&skill_fit));
        w.create_rule(RuleDef {
            name: "skill_fit".into(),
            src_label: "Person".into(),
            dst_label: "Project".into(),
            predicate: skill_fit,
            edge_type: "FIT".into(),
            weight_prop: Some("score".into()),
            max_edges: skill_fit_k,
            approximate: false,
            via_label: None,
            via_edge: None,
            via_dir: None,
            namespace: None,
        })?;
        let founded_within = Predicate::NumericWithin {
            field: "founded_year".into(),
            tolerance: 2.0,
        };
        let founded_within_k = Some(default_max_edges(&founded_within));
        w.create_rule(RuleDef {
            name: "founded_within".into(),
            src_label: "Org".into(),
            dst_label: "Org".into(),
            predicate: founded_within,
            edge_type: "FOUNDED_WITHIN".into(),
            weight_prop: Some("score".into()),
            max_edges: founded_within_k,
            approximate: false,
            via_label: None,
            via_edge: None,
            via_dir: None,
            namespace: None,
        })?;
        let nearby_office = Predicate::GeoRadius {
            field: "office".into(),
            km: 50.0,
        };
        let nearby_office_k = Some(default_max_edges(&nearby_office));
        w.create_rule(RuleDef {
            name: "nearby_office".into(),
            src_label: "Org".into(),
            dst_label: "Org".into(),
            predicate: nearby_office,
            edge_type: "NEARBY_OFFICE".into(),
            weight_prop: Some("score".into()),
            max_edges: nearby_office_k,
            approximate: false,
            via_label: None,
            via_edge: None,
            via_dir: None,
            namespace: None,
        })?;
        let similar_interests = Predicate::VectorSimilar {
            field: "embedding".into(),
            min: 0.8,
        };
        let similar_interests_k = Some(default_max_edges(&similar_interests));
        w.create_rule(RuleDef {
            name: "similar_interests".into(),
            src_label: "Person".into(),
            dst_label: "Person".into(),
            predicate: similar_interests,
            edge_type: "SIMILAR".into(),
            weight_prop: Some("score".into()),
            max_edges: similar_interests_k,
            approximate: false,
            via_label: None,
            via_edge: None,
            via_dir: None,
            namespace: None,
        })?;
    }

    let r = db.read();
    let sample_result = r.query(SAMPLE_QUERY, &BTreeMap::new())?;
    let explanations = r.explain(SAMPLE_EXPLAIN_A, SAMPLE_EXPLAIN_B)?;
    let stats = r.stats();
    // Rule suggestion teaser: first suggestion sorted by est_edges desc.
    let suggestion = r.suggest_rules().into_iter().next();

    Ok(DemoOutcome {
        auto_fk_rules,
        sample_query: SAMPLE_QUERY.to_string(),
        sample_result,
        explanations,
        stats,
        suggestion,
    })
}

fn dir_is_empty_or_absent(dir: &Path) -> Result<bool, CliError> {
    if dir.is_file() {
        return Err(CliError(format!(
            "demo refuses a non-empty directory: {} is a file",
            dir.display()
        )));
    }
    if !dir.exists() {
        return Ok(true);
    }
    Ok(std::fs::read_dir(dir)?.next().is_none())
}

fn refuse_non_empty(dir: &Path) -> Result<(), CliError> {
    if dir_is_empty_or_absent(dir)? {
        Ok(())
    } else {
        Err(CliError(format!(
            "demo refuses a non-empty directory: {} \
             (directory must be empty — including hidden files)",
            dir.display()
        )))
    }
}

/// Run [`run_demo`] when `dir` is missing or empty; otherwise leave it alone.
pub fn maybe_run_demo_if_empty(dir: &Path) -> Result<Option<DemoOutcome>, CliError> {
    if dir_is_empty_or_absent(dir)? {
        Ok(Some(run_demo(dir)?))
    } else {
        Ok(None)
    }
}

fn json_array(rows: impl IntoIterator<Item = String>) -> String {
    let mut out = String::from("[");
    let mut first = true;
    for row in rows {
        if !first {
            out.push(',');
        }
        first = false;
        out.push_str(&row);
    }
    out.push(']');
    out
}

/// Wrap a 1-based project index into `1..=N_PROJECTS`.
fn wrap_proj(i: usize) -> usize {
    (i - 1) % N_PROJECTS + 1
}

/// Sliding window of `len` skill tokens starting at project `start`.
fn skill_window_json(start: usize, len: usize) -> String {
    let parts: Vec<String> = (0..len)
        .map(|k| format!(r#""s{:02}""#, wrap_proj(start + k)))
        .collect();
    format!("[{}]", parts.join(","))
}

/// Real city [lat, lon] for org `i` (1-based). Four clusters sit inside 50 km:
/// NYC / Jersey City / Newark, SF / Oakland / Berkeley, London / Greenwich,
/// Paris / Versailles.
fn org_office(i: usize) -> (f64, f64) {
    match i {
        1 => (40.7128, -74.0060),  // New York
        2 => (48.8566, 2.3522),    // Paris
        3 => (51.5074, -0.1278),   // London
        4 => (37.7749, -122.4194), // San Francisco
        5 => (37.8044, -122.2711), // Oakland
        6 => (37.8715, -122.2730), // Berkeley
        7 => (40.7178, -74.0431),  // Jersey City
        8 => (51.4769, 0.0005),    // Greenwich
        9 => (48.8014, 2.1301),    // Versailles
        10 => (40.7357, -74.1724), // Newark
        _ => unreachable!("demo orgs are 1..=10"),
    }
}

/// Dim-8 embedding for person `i`. Groups of three share a unit axis (cos = 1);
/// two extra groups are (0.8, 0.6, …) and (0.6, 0.8, …) so cos = 0.8 / 0.96
/// against the first two axes is hand-checkable.
fn person_embedding_json(i: usize) -> String {
    let mut v = [0.0_f64; 8];
    match i {
        9 | 19 | 29 => {
            v[0] = 0.8;
            v[1] = 0.6;
        }
        10 | 20 | 30 => {
            v[0] = 0.6;
            v[1] = 0.8;
        }
        _ => {
            let axis = (i - 1) % 10;
            debug_assert!(axis < 8);
            v[axis] = 1.0;
        }
    }
    let parts: Vec<String> = v.iter().map(|x| format!("{x}")).collect();
    format!("[{}]", parts.join(","))
}

fn org_json() -> String {
    json_array((1..=N_ORGS).map(|i| {
        let year = 2010 + (i as i64 - 1);
        let (lat, lon) = org_office(i);
        format!(
            r#"{{"id":"org-{i:02}","name":"Org {i}","founded_year":{year},"office":[{lat},{lon}],"skills":{}}}"#,
            skill_window_json(i, 3)
        )
    }))
}

fn project_json() -> String {
    json_array((1..=N_PROJECTS).map(|i| {
        let org = (i - 1) % N_ORGS + 1;
        format!(
            r#"{{"id":"proj-{i:02}","name":"Project {i}","org_id":"org-{org:02}","skills":{}}}"#,
            skill_window_json(i, 3)
        )
    }))
}

fn person_json() -> String {
    json_array((1..=N_PEOPLE).map(|i| {
        let org = (i - 1) % N_ORGS + 1;
        let proj = (i - 1) % N_PROJECTS + 1;
        format!(
            r#"{{"id":"person-{i:02}","name":"Person {i}","org_id":"org-{org:02}","project_id":"proj-{proj:02}","embedding":{},"skills":{}}}"#,
            person_embedding_json(i),
            skill_window_json(proj, 3)
        )
    }))
}

/// Render a [`DemoOutcome`] the way `mushroomdb demo` prints it.
pub fn format_demo(dir: &Path, out: &DemoOutcome) -> String {
    let mut buf = String::new();
    let _ = writeln!(buf, "== demo ==");
    let _ = writeln!(
        buf,
        "ingested {N_ORGS} Orgs, {N_PROJECTS} Projects, {N_PEOPLE} People"
    );
    let _ = writeln!(
        buf,
        "overlap rule: skill_fit (Person.skills ∩ Project.skills, min 0.5)"
    );
    let _ = writeln!(
        buf,
        "numeric rule: founded_within (Org.founded_year, tolerance 2)"
    );
    let _ = writeln!(buf, "geo rule: nearby_office (Org.office [lat,lon], 50 km)");
    let _ = writeln!(
        buf,
        "vector rule: similar_interests (Person.embedding dim 8, min 0.8)"
    );
    let _ = writeln!(buf);
    let _ = writeln!(buf, "== auto-FK rules ==");
    let mut names = out.auto_fk_rules.clone();
    names.sort();
    for name in names {
        let _ = writeln!(buf, "  {name}");
    }
    let _ = writeln!(buf);
    let _ = writeln!(buf, "== query ==");
    let _ = writeln!(buf, "{}", out.sample_query);
    let _ = writeln!(buf);
    let _ = writeln!(buf, "columns: {}", out.sample_result.columns().join(", "));
    for i in 0..out.sample_result.len() {
        let cells: Vec<String> = out
            .sample_result
            .columns()
            .iter()
            .map(|c| format!("{c}={}", fmt_cell(out.sample_result.get(i, c))))
            .collect();
        let _ = writeln!(buf, "  {}", cells.join("  "));
    }
    let _ = writeln!(buf);
    let _ = writeln!(
        buf,
        "== explain ({SAMPLE_EXPLAIN_A}, {SAMPLE_EXPLAIN_B}) =="
    );
    for e in &out.explanations {
        let weight = e
            .weight
            .map(|w| fmt_value(&Value::Float(w)))
            .unwrap_or_else(|| "none".into());
        let _ = writeln!(
            buf,
            "  rule={}  type={}  {}→{}  weight={}",
            e.rule, e.edge_type, e.src_key, e.dst_key, weight
        );
    }
    let _ = writeln!(buf);
    let _ = writeln!(buf, "== serve ==");
    let _ = writeln!(buf, "  mushroomdb serve {}", dir.display());

    // Teaser: one suggestion from the rule suggester (not auto-applied).
    if let Some(s) = &out.suggestion {
        let _ = writeln!(buf);
        let _ = writeln!(buf, "== suggested rule (teaser) ==");
        let _ = writeln!(buf, "  {}", s.def.name);
        let _ = writeln!(
            buf,
            "  {} → {} via {:?}",
            s.def.src_label, s.def.dst_label, s.def.predicate
        );
        let _ = writeln!(buf, "  est_edges: ~{}", s.est_edges);
        let _ = writeln!(buf, "  {}", s.rationale);
        let _ = writeln!(
            buf,
            "  (run `mushroomdb suggest {}` for full analysis)",
            dir.display()
        );
    }

    buf
}

/// Profile the database at `dir` and return all rule suggestions.
pub fn run_suggest(dir: &Path) -> Result<Vec<RuleSuggestion>, CliError> {
    let db = GraphDb::open(dir)?;
    Ok(db.suggest_rules())
}

/// Pretty-print a list of [`RuleSuggestion`]s for `mushroomdb suggest`.
pub fn format_suggest(suggestions: &[RuleSuggestion]) -> String {
    let mut buf = String::new();
    if suggestions.is_empty() {
        let _ = writeln!(
            buf,
            "no rule suggestions (database may be empty or rules already cover all patterns)"
        );
        return buf;
    }
    let _ = writeln!(buf, "== rule suggestions ({}) ==", suggestions.len());
    for (i, s) in suggestions.iter().enumerate() {
        let _ = writeln!(buf);
        let _ = writeln!(buf, "[{}] {}", i + 1, s.def.name);
        let _ = writeln!(
            buf,
            "    {} → {}  via {:?}",
            s.def.src_label, s.def.dst_label, s.def.predicate
        );
        let _ = writeln!(buf, "    est_edges : ~{}", s.est_edges);
        let _ = writeln!(buf, "    rationale : {}", s.rationale);
        if !s.examples.is_empty() {
            let _ = writeln!(buf, "    examples  :");
            for (src, dst, score) in &s.examples {
                let _ = writeln!(buf, "      {src} → {dst}  score={score:.4}");
            }
        }
        let _ = writeln!(buf, "    predicate : {:?}", s.def.predicate);
        let _ = writeln!(
            buf,
            "    to apply  : POST /rules  or  db.create_rule(suggestion.def)"
        );
    }
    buf
}

fn fmt_value(v: &Value) -> String {
    match v {
        Value::Int(i) => i.to_string(),
        Value::Float(f) => {
            let s = format!("{f}");
            if s.contains('.') || s.contains('e') || s.contains('E') {
                s
            } else {
                format!("{s}.0")
            }
        }
        Value::Str(s) => s.clone(),
        Value::Bool(b) => b.to_string(),
        Value::List(xs) => {
            let inner: Vec<String> = xs.iter().map(fmt_value).collect();
            format!("[{}]", inner.join(", "))
        }
        Value::Map(m) => {
            let inner: Vec<String> = m
                .iter()
                .map(|(k, v)| format!("{k}: {}", fmt_value(v)))
                .collect();
            format!("{{{}}}", inner.join(", "))
        }
    }
}

fn fmt_cell(cell: Option<&Value>) -> String {
    match cell {
        None => "null".into(),
        Some(v) => fmt_value(v),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::BTreeSet;
    use std::net::SocketAddr;
    use std::path::PathBuf;

    fn tmp(name: &str) -> PathBuf {
        let nanos = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let d = std::env::temp_dir().join(format!(
            "graphdb-cli-{}-{}-{}",
            name,
            std::process::id(),
            nanos
        ));
        let _ = std::fs::remove_dir_all(&d);
        d
    }

    fn directed_pairs(db: &SharedDb, etype: &str) -> BTreeSet<(String, String)> {
        let g = db.read();
        let mut out = BTreeSet::new();
        for i in 1..=N_ORGS {
            let src = format!("org-{i:02}");
            if let Ok(nbrs) = g.neighbors(&src, etype, core_api::Direction::Out) {
                for dst in nbrs {
                    out.insert((src.clone(), dst));
                }
            }
        }
        for i in 1..=N_PEOPLE {
            let src = format!("person-{i:02}");
            if let Ok(nbrs) = g.neighbors(&src, etype, core_api::Direction::Out) {
                for dst in nbrs {
                    out.insert((src.clone(), dst));
                }
            }
        }
        out
    }

    fn assert_weight(db: &SharedDb, a: &str, b: &str, rule: &str, want: f64) {
        let hits: Vec<_> = db
            .read()
            .explain(a, b)
            .expect("explain")
            .into_iter()
            .filter(|e| e.rule == rule && e.src_key == a && e.dst_key == b)
            .collect();
        assert_eq!(hits.len(), 1, "explain {a}/{b} rule={rule}: {hits:?}");
        let got = hits[0].weight.expect("weighted");
        assert!(
            (got - want).abs() < 1e-12,
            "{rule} {a}→{b}: got {got} want {want}"
        );
    }

    fn haversine_km(lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
        const R: f64 = 6371.0088;
        let phi1 = lat1.to_radians();
        let phi2 = lat2.to_radians();
        let dphi = (lat2 - lat1).to_radians();
        let dlam = (lon2 - lon1).to_radians();
        let a = ((dphi / 2.0).sin().powi(2) + phi1.cos() * phi2.cos() * (dlam / 2.0).sin().powi(2))
            .clamp(0.0, 1.0);
        let c = 2.0 * a.sqrt().atan2((1.0 - a).sqrt());
        R * c
    }

    fn default_bind() -> SocketAddr {
        SocketAddr::from(([127, 0, 0, 1], 8080))
    }

    #[test]
    fn parse_args_table() {
        struct Case {
            args: &'static [&'static str],
            check: fn(Result<Command, String>),
        }

        let cases = [
            Case {
                args: &[],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("no-args → Help, got {other:?}"),
                },
            },
            Case {
                args: &["--help"],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("--help → Help, got {other:?}"),
                },
            },
            Case {
                args: &["-h"],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("-h → Help, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        restore_from,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(addr, default_bind());
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        assert!(role_tokens.is_empty());
                        assert_eq!(snapshot_every, None);
                        assert_eq!(restore_from, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve <dir> → Serve default addr, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr", "127.0.0.1:8080"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        restore_from,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(
                            addr,
                            "127.0.0.1:8080".parse::<std::net::SocketAddr>().unwrap()
                        );
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        assert!(role_tokens.is_empty());
                        assert_eq!(snapshot_every, None);
                        assert_eq!(restore_from, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve --addr after dir, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr=127.0.0.1:9090"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        restore_from,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(
                            addr,
                            "127.0.0.1:9090".parse::<std::net::SocketAddr>().unwrap()
                        );
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        let _ = role_tokens; // empty, not asserted
                        assert_eq!(snapshot_every, None);
                        assert_eq!(restore_from, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve --addr=VALUE, got {other:?}"),
                },
            },
            Case {
                args: &["mcp", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Mcp {
                        db_dir,
                        auto,
                        all_tools,
                    }) => {
                        assert_eq!(db_dir, Some(PathBuf::from("/tmp/demo-db")));
                        assert!(!auto);
                        assert!(!all_tools, "the short list is the default");
                    }
                    other => panic!("mcp <dir>, got {other:?}"),
                },
            },
            Case {
                args: &["stats", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Stats { db_dir }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                    }
                    other => panic!("stats <dir>, got {other:?}"),
                },
            },
            Case {
                args: &["demo", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Demo { db_dir }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                    }
                    other => panic!("demo <dir>, got {other:?}"),
                },
            },
            Case {
                args: &["serve"],
                check: |r| {
                    let e = r.expect_err("serve without dir");
                    assert!(
                        e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
                        "missing-dir error should mention dir, got {e}"
                    );
                },
            },
            Case {
                args: &["mcp"],
                check: |r| {
                    let e = r.expect_err("mcp without dir");
                    assert!(
                        e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
                        "missing-dir error should mention dir, got {e}"
                    );
                },
            },
            Case {
                args: &["stats"],
                check: |r| {
                    let e = r.expect_err("stats without dir");
                    assert!(
                        e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
                        "missing-dir error should mention dir, got {e}"
                    );
                },
            },
            Case {
                args: &["demo"],
                check: |r| {
                    let e = r.expect_err("demo without dir");
                    assert!(
                        e.to_lowercase().contains("db-dir") || e.to_lowercase().contains("dir"),
                        "missing-dir error should mention dir, got {e}"
                    );
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr"],
                check: |r| {
                    let e = r.expect_err("--addr missing value");
                    assert!(
                        e.to_lowercase().contains("addr"),
                        "--addr missing value should mention addr, got {e}"
                    );
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr", "not-an-addr"],
                check: |r| {
                    let e = r.expect_err("invalid addr");
                    assert!(
                        e.to_lowercase().contains("addr") || e.to_lowercase().contains("address"),
                        "invalid addr should mention address, got {e}"
                    );
                },
            },
            Case {
                args: &["frobnicate", "/tmp/demo-db"],
                check: |r| {
                    let e = r.expect_err("unknown command");
                    assert!(
                        e.to_lowercase().contains("unknown")
                            || e.to_lowercase().contains("frobnicate"),
                        "unknown command should name it, got {e}"
                    );
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--ui", "/tmp/ui-dist"],
                check: |r| match r {
                    Ok(Command::Serve { ui, .. }) => {
                        assert_eq!(
                            ui,
                            super::ServeUi::Filesystem(PathBuf::from("/tmp/ui-dist"))
                        );
                    }
                    other => panic!("serve --ui <dir>, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--ui=/tmp/ui-eq"],
                check: |r| match r {
                    Ok(Command::Serve { ui, .. }) => {
                        assert_eq!(ui, super::ServeUi::Filesystem(PathBuf::from("/tmp/ui-eq")));
                    }
                    other => panic!("serve --ui=VALUE, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--ui"],
                check: |r| {
                    let e = r.expect_err("--ui missing value");
                    assert!(
                        e.to_lowercase().contains("ui"),
                        "--ui missing value should mention ui, got {e}"
                    );
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--no-ui"],
                check: |r| match r {
                    Ok(Command::Serve { ui, .. }) => {
                        assert_eq!(ui, super::ServeUi::None);
                    }
                    other => panic!("serve --no-ui, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--ui", "/tmp/x", "--no-ui"],
                check: |r| {
                    let e = r.expect_err("combine --ui and --no-ui");
                    assert!(
                        e.contains("--ui") && e.contains("--no-ui"),
                        "conflict should name both flags, got {e}"
                    );
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "extra"],
                check: |r| {
                    let e = r.expect_err("extra positional");
                    assert!(
                        e.to_lowercase().contains("unexpected")
                            || e.to_lowercase().contains("extra"),
                        "extra arg should be rejected, got {e}"
                    );
                },
            },
            Case {
                args: &[
                    "serve",
                    "/data",
                    "--addr",
                    "0.0.0.0:8080",
                    "--demo-if-empty",
                ],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        demo_if_empty,
                        ui,
                        token,
                        snapshot_every,
                        ..
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/data"));
                        assert_eq!(
                            addr,
                            "0.0.0.0:8080".parse::<std::net::SocketAddr>().unwrap()
                        );
                        assert!(demo_if_empty);
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert_eq!(token, None);
                        assert_eq!(snapshot_every, None);
                    }
                    other => panic!("serve --demo-if-empty docker default, got {other:?}"),
                },
            },
            Case {
                args: &["install", "--project", "--delivery", "cli"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => {
                        assert_eq!(opts.scope, Some(install::Scope::Project));
                        assert_eq!(opts.delivery, install::Delivery::Cli);
                    }
                    other => panic!("install --delivery cli, got {other:?}"),
                },
            },
            Case {
                args: &["install", "--delivery=mcp"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => {
                        assert_eq!(opts.delivery, install::Delivery::Mcp)
                    }
                    other => panic!("install --delivery=mcp, got {other:?}"),
                },
            },
            Case {
                // No flag is the default, and it is the one that opens both
                // doors: an upgrade must not quietly drop a user's server.
                args: &["install"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => {
                        assert_eq!(opts.delivery, install::Delivery::Both)
                    }
                    other => panic!("install, got {other:?}"),
                },
            },
            Case {
                args: &["install", "--delivery", "sideways"],
                check: |r| match r {
                    Err(e) => assert!(e.contains("--delivery must be cli | mcp | both"), "{e}"),
                    other => panic!("a bad --delivery must be refused, got {other:?}"),
                },
            },
            Case {
                args: &["install", "--always-load"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(opts.always_load),
                    other => panic!("install --always-load, got {other:?}"),
                },
            },
            Case {
                // 0.7 writes no git hooks, so `--no-git-hooks` asks for what
                // every install already gets: accepted, not refused.
                args: &["install", "--no-git-hooks"],
                check: |r| match r {
                    Ok(Command::Install(_)) => {}
                    other => panic!("install --no-git-hooks, got {other:?}"),
                },
            },
            Case {
                // An install that names its store and registers a server is
                // an entity-store install: `alwaysLoad` without asking, so a
                // session meets the tools rather than searching for them.
                args: &["install", "--delivery", "mcp", "--db", "./mem"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(opts.always_load),
                    other => panic!("install --delivery mcp --db, got {other:?}"),
                },
            },
            Case {
                // `both` registers a server too, so it defaults the same way.
                args: &["install", "--delivery", "both", "--db=./mem"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(opts.always_load),
                    other => panic!("install --delivery both --db, got {other:?}"),
                },
            },
            Case {
                // …and `--no-always-load` is the way out of it.
                args: &[
                    "install",
                    "--delivery",
                    "mcp",
                    "--db",
                    "./mem",
                    "--no-always-load",
                ],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(!opts.always_load),
                    other => panic!("install --no-always-load, got {other:?}"),
                },
            },
            Case {
                // A `--delivery cli` install registers no server, so there is
                // no entry for the key to go on: naming a store cannot turn
                // it on.
                args: &["install", "--delivery", "cli", "--db", "./mem"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(!opts.always_load),
                    other => panic!("install --delivery cli --db, got {other:?}"),
                },
            },
            Case {
                // An install with no `--db` resolves whatever store the
                // directory has — usually the code graph — and stays opt-in.
                // `--always-load` still forces it.
                args: &["install", "--always-load"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(opts.always_load),
                    other => panic!("install --always-load, got {other:?}"),
                },
            },
            Case {
                // The one experiment left is off unless it is asked for.
                args: &["install"],
                check: |r| match r {
                    Ok(Command::Install(opts)) => assert!(!opts.always_load),
                    other => panic!("install, got {other:?}"),
                },
            },
        ];

        for case in &cases {
            (case.check)(parse_args(case.args));
        }
    }

    /// The five code-graph subcommands a person typed, left in 0.7. Each is an
    /// unknown command now — not a stub that prints a deprecation — and the
    /// help text names none of them. The other five were hook bodies; see
    /// [`the_retired_hook_bodies_parse_as_inert`].
    #[test]
    fn the_retired_code_graph_subcommands_are_unknown() {
        for sub in ["map", "explore", "context", "impact", "owners"] {
            match parse_args(&[sub, "/tmp/db"]) {
                Err(e) => assert_eq!(e, format!("unknown command: {sub}")),
                other => panic!("{sub} still parses: {other:?}"),
            }
            assert!(
                !usage().contains(&format!("mushroomdb {sub} ")),
                "the help still names {sub}"
            );
        }
    }

    /// The five hook bodies 0.6 installed parse, with any arguments, to the
    /// inert [`Command::RetiredHook`], so an upgraded machine that has not
    /// re-run `install` is not shown a hook error on every tool call. The help
    /// still names none of them.
    #[test]
    fn the_retired_hook_bodies_parse_as_inert() {
        for sub in ["touch", "intercept", "impact-hook", "enrich", "sync"] {
            for args in [
                vec![sub],
                vec![sub, "--auto"],
                vec![sub, "/tmp/db", "--whatever", "x"],
            ] {
                assert_eq!(
                    parse_args(&args),
                    Ok(Command::RetiredHook { sub }),
                    "{args:?}"
                );
            }
            assert!(
                !usage().contains(&format!("mushroomdb {sub}")),
                "the help still names {sub}"
            );
        }
    }

    #[test]
    fn serve_default_addr_is_loopback_8080() {
        match parse_args(&["serve", "/tmp/db"]).unwrap() {
            Command::Serve { addr, .. } => {
                assert_eq!(
                    addr,
                    "127.0.0.1:8080".parse::<std::net::SocketAddr>().unwrap()
                );
            }
            other => panic!("{other:?}"),
        }
    }

    #[test]
    fn serve_snapshot_every_parses_seconds() {
        match parse_args(&["serve", "/tmp/db", "--snapshot-every", "30"]).unwrap() {
            Command::Serve { snapshot_every, .. } => {
                assert_eq!(snapshot_every, Some(Duration::from_secs(30)));
            }
            other => panic!("{other:?}"),
        }
        match parse_args(&["serve", "/tmp/db", "--snapshot-every=5"]).unwrap() {
            Command::Serve { snapshot_every, .. } => {
                assert_eq!(snapshot_every, Some(Duration::from_secs(5)));
            }
            other => panic!("{other:?}"),
        }
        match parse_args(&["serve", "/tmp/db"]).unwrap() {
            Command::Serve { snapshot_every, .. } => {
                assert_eq!(snapshot_every, None);
            }
            other => panic!("{other:?}"),
        }
        let err = parse_args(&["serve", "/tmp/db", "--snapshot-every"]).unwrap_err();
        assert!(
            err.contains("snapshot-every"),
            "missing value should name the flag, got {err}"
        );
        let err = parse_args(&["serve", "/tmp/db", "--snapshot-every", "0"]).unwrap_err();
        assert!(
            err.contains("snapshot-every"),
            "zero should be rejected, got {err}"
        );
        let err = parse_args(&["serve", "/tmp/db", "--snapshot-every", "nope"]).unwrap_err();
        assert!(
            err.contains("snapshot-every"),
            "invalid value should name the flag, got {err}"
        );
    }

    #[test]
    fn serve_token_flag_and_non_loopback_without_token_is_parsed() {
        // parse succeeds; main() enforces the bind rule. Token is stored.
        match parse_args(&[
            "serve",
            "/tmp/db",
            "--addr",
            "0.0.0.0:8080",
            "--token",
            "s3cret",
        ])
        .unwrap()
        {
            Command::Serve { token, addr, .. } => {
                assert_eq!(token.as_deref(), Some("s3cret"));
                assert_eq!(addr.ip().to_string(), "0.0.0.0");
            }
            other => panic!("{other:?}"),
        }
    }

    #[test]
    fn parse_build_index_both_forms_and_a_missing_value() {
        match parse_args(&["build-index", "/tmp/db"]).unwrap() {
            Command::BuildIndex { db_dir, rule } => {
                assert_eq!(db_dir, PathBuf::from("/tmp/db"));
                assert_eq!(rule, None);
            }
            other => panic!("{other:?}"),
        }
        for args in [
            vec!["build-index", "/tmp/db", "--rule", "sim"],
            vec!["build-index", "/tmp/db", "--rule=sim"],
        ] {
            match parse_args(&args).unwrap() {
                Command::BuildIndex { db_dir, rule } => {
                    assert_eq!(db_dir, PathBuf::from("/tmp/db"));
                    assert_eq!(rule.as_deref(), Some("sim"), "{args:?}");
                }
                other => panic!("{other:?}"),
            }
        }
        assert_eq!(
            parse_args(&["build-index", "/tmp/db", "--rule"]).unwrap_err(),
            "--rule requires a value"
        );
        assert_eq!(
            parse_args(&["build-index", "/tmp/db", "--rule="]).unwrap_err(),
            "--rule requires a value"
        );
        assert_eq!(
            parse_args(&["build-index"]).unwrap_err(),
            "build-index requires <db-dir>"
        );
        assert!(usage().contains("mushroomdb build-index <db-dir> [--rule <name>]"));
    }

    #[test]
    fn parse_snapshot_and_query() {
        // Archiving is what a snapshot does unless the user says otherwise.
        for (args, want) in [
            (vec!["snapshot", "/tmp/db"], WalDisposition::Archive),
            (
                vec!["snapshot", "/tmp/db", "--archive-wal"],
                WalDisposition::Archive,
            ),
            (
                vec!["snapshot", "/tmp/db", "--keep-wal"],
                WalDisposition::Keep,
            ),
            (
                vec!["snapshot", "/tmp/db", "--truncate"],
                WalDisposition::Truncate,
            ),
        ] {
            match parse_args(&args).unwrap() {
                Command::Snapshot { wal, .. } => assert_eq!(wal, want, "{args:?}"),
                other => panic!("{other:?}"),
            }
        }
        match parse_args(&["query", "/tmp/db", "MATCH (n) RETURN n LIMIT 1"]).unwrap() {
            Command::Query { cypher, .. } => assert!(cypher.contains("MATCH")),
            other => panic!("{other:?}"),
        }
        match parse_args(&["query", "/tmp/db", "MATCH", "(n)", "RETURN", "n"]).unwrap() {
            Command::Query { cypher, .. } => assert_eq!(cypher, "MATCH (n) RETURN n"),
            other => panic!("{other:?}"),
        }
        match parse_args(&["query", "/tmp/db", "--query", "MATCH (n) RETURN n"]).unwrap() {
            Command::Query { cypher, .. } => assert_eq!(cypher, "MATCH (n) RETURN n"),
            other => panic!("{other:?}"),
        }
        let text = usage();
        assert!(
            text.contains("query"),
            "usage should mention query, got:\n{text}"
        );
        assert!(
            text.contains("snapshot"),
            "usage should mention snapshot, got:\n{text}"
        );
    }

    /// Binding: the help describes the tree 0.7 ships. One tool surface, the
    /// association tools listed on every store, and one brief — the memory
    /// schema's. The code-graph listing and the repository brief are gone.
    #[test]
    fn usage_describes_one_surface_and_the_memory_brief() {
        let text = usage();
        assert!(
            text.contains(
                "--all-tools lists every served tool; the default lists the association tools on every store,\n\
                 \x20                    a store `ingest-git` built included"
            ),
            "mcp help line, got:\n{text}"
        );
        assert!(
            text.contains(
                "mushroomdb brief <db-dir>|--auto    hook body: the store's schema in one block — labels,\n\
                 \x20                                     edge types, history depth, roles and one worked call per\n\
                 \x20                                     question kind; byte-stable, so a session host caches it once"
            ),
            "brief help line, got:\n{text}"
        );
        for stale in [
            "28 tools",
            "explore, query, stats",
            "most called symbols",
            "synced sha",
        ] {
            assert!(!text.contains(stale), "usage still says {stale:?}");
        }
    }

    #[test]
    fn usage_lists_every_subcommand() {
        let text = usage();
        for word in [
            "serve",
            "mcp",
            "stats",
            "demo",
            "query",
            "snapshot",
            "--keep-wal",
            "mushroomdb",
            "--ui",
            "--no-ui",
            "--demo-if-empty",
            "--token",
            "--role-token TOKEN:ROLE",
            "--tls-cert <pem> --tls-key <pem>",
            "a build with the `tls` feature",
            "--snapshot-every",
        ] {
            assert!(
                text.contains(word),
                "usage should mention {word}, got:\n{text}"
            );
        }
    }

    #[test]
    fn validate_ui_dir_requires_index_html() {
        let missing = tmp("ui-missing");
        let err = super::validate_ui_dir(&missing).expect_err("missing dir");
        assert!(
            err.contains("does not exist"),
            "missing dir error, got {err}"
        );

        let empty = tmp("ui-empty");
        std::fs::create_dir_all(&empty).unwrap();
        let err = super::validate_ui_dir(&empty).expect_err("no index");
        assert!(
            err.contains("index.html"),
            "missing index.html error, got {err}"
        );

        let ok = tmp("ui-ok");
        std::fs::create_dir_all(&ok).unwrap();
        std::fs::write(ok.join("index.html"), "<!doctype html>").unwrap();
        let got = super::validate_ui_dir(&ok).expect("valid ui dir");
        assert_eq!(got, ok);
    }

    #[test]
    fn maybe_run_demo_if_empty_seeds_then_skips() {
        let dir = tmp("boot-empty");
        let first = super::maybe_run_demo_if_empty(&dir)
            .expect("empty dir demos")
            .expect("Some(DemoOutcome)");
        assert_eq!(first.stats.nodes_live, 60);
        let db = SharedDb::open(&dir).expect("reopen");
        assert!(db.read().has_node("person-01"));
        let second = super::maybe_run_demo_if_empty(&dir).expect("non-empty is ok");
        assert!(
            second.is_none(),
            "second boot must not re-demo a populated volume"
        );

        let occupied = tmp("boot-occupied");
        std::fs::create_dir_all(&occupied).unwrap();
        std::fs::write(occupied.join("keep-me"), b"x").unwrap();
        let skipped = super::maybe_run_demo_if_empty(&occupied).expect("occupied skip");
        assert!(skipped.is_none());
        assert_eq!(
            std::fs::read(occupied.join("keep-me")).unwrap(),
            b"x",
            "existing volume contents must be untouched"
        );
    }

    #[test]
    fn demo_builder_is_deterministic_and_refuses_second_run() {
        let dir = tmp("demo");
        let out = run_demo(&dir).expect("first demo run");

        assert_eq!(
            out.stats.nodes_live, 60,
            "10 orgs + 20 projects + 30 people"
        );
        assert_eq!(out.stats.nodes_tombstoned, 0);
        // Auto-FK: 20 project→org + 30 person→org + 30 person→project = 80.
        // FIT: each of 30 people matches home (Jaccard 1.0) and two adjacent
        // projects (3-skill window shifted ±1 → Jaccard 2/4 = 0.5) = 30*3 = 90.
        // founded_within: |year_i − year_j| ≤ 2 on 2010+(i-1) → 17 pairs × 2 = 34.
        // nearby_office: 4 city clusters (NYC/SF/London/Paris) → 8 pairs × 2 = 16.
        // similar_interests: dim-8 groups → 57 pairs × 2 = 114.
        // Total: 80 + 90 + 34 + 16 + 114 = 334.
        assert_eq!(out.stats.edges, 334);
        assert_eq!(
            out.stats.rules.len(),
            7,
            "3 auto-FK + overlap + numeric + geo + vector"
        );
        let fit = out
            .stats
            .rules
            .iter()
            .find(|r| r.name == "skill_fit")
            .expect("skill_fit");
        assert_eq!(fit.edges, 90, "30 people × 3 FIT edges");
        let founded = out
            .stats
            .rules
            .iter()
            .find(|r| r.name == "founded_within")
            .expect("founded_within");
        assert_eq!(founded.edges, 34);
        let nearby = out
            .stats
            .rules
            .iter()
            .find(|r| r.name == "nearby_office")
            .expect("nearby_office");
        assert_eq!(nearby.edges, 16);
        let similar = out
            .stats
            .rules
            .iter()
            .find(|r| r.name == "similar_interests")
            .expect("similar_interests");
        assert_eq!(similar.edges, 114);

        let mut names: Vec<&str> = out.stats.rules.iter().map(|r| r.name.as_str()).collect();
        names.sort_unstable();
        assert_eq!(
            names,
            vec![
                "auto_fk_person_org_id",
                "auto_fk_person_project_id",
                "auto_fk_project_org_id",
                "founded_within",
                "nearby_office",
                "similar_interests",
                "skill_fit",
            ]
        );

        // `recall` needs a name index; the memory schema applied at store
        // creation adds it, plus every other memory-entity label — no nodes,
        // edges or rules.
        let db = SharedDb::open(&dir).expect("reopen demo");
        assert_eq!(
            db.read().fulltext_pairs(),
            vec![
                ("Concept".to_string(), "name".to_string()),
                ("Concept".to_string(), "summary".to_string()),
                ("Entity".to_string(), "name".to_string()),
                ("Event".to_string(), "name".to_string()),
                ("Note".to_string(), "text".to_string()),
                ("Org".to_string(), "name".to_string()),
                ("Person".to_string(), "name".to_string()),
                ("Project".to_string(), "name".to_string()),
            ]
        );

        let mut auto = out.auto_fk_rules.clone();
        auto.sort();
        assert_eq!(
            auto,
            vec![
                "auto_fk_person_org_id".to_string(),
                "auto_fk_person_project_id".to_string(),
                "auto_fk_project_org_id".to_string(),
            ]
        );

        assert!(
            !out.sample_result.is_empty(),
            "sample Cypher query must return rows"
        );
        assert!(
            out.sample_query.contains("ORDER BY score DESC"),
            "sample query must rank by score, got {}",
            out.sample_query
        );
        let scores: Vec<f64> = (0..out.sample_result.len())
            .map(|i| match out.sample_result.get(i, "score") {
                Some(Value::Float(f)) => *f,
                other => panic!("score col should be Float, got {other:?}"),
            })
            .collect();
        let distinct: std::collections::BTreeSet<u64> =
            scores.iter().map(|s| s.to_bits()).collect();
        assert!(
            distinct.len() >= 2,
            "sample results must be visibly ranked, got {scores:?}"
        );
        for w in scores.windows(2) {
            assert!(
                w[0] >= w[1],
                "scores must be non-increasing, got {scores:?}"
            );
        }
        assert!(
            !out.explanations.is_empty(),
            "explain(person-01, proj-01) must find the derived edges"
        );

        let db = SharedDb::open(&dir).expect("reopen demo");
        assert_eq!(
            directed_pairs(&db, "FOUNDED_WITHIN"),
            [
                ("org-01", "org-02"),
                ("org-01", "org-03"),
                ("org-02", "org-01"),
                ("org-02", "org-03"),
                ("org-02", "org-04"),
                ("org-03", "org-01"),
                ("org-03", "org-02"),
                ("org-03", "org-04"),
                ("org-03", "org-05"),
                ("org-04", "org-02"),
                ("org-04", "org-03"),
                ("org-04", "org-05"),
                ("org-04", "org-06"),
                ("org-05", "org-03"),
                ("org-05", "org-04"),
                ("org-05", "org-06"),
                ("org-05", "org-07"),
                ("org-06", "org-04"),
                ("org-06", "org-05"),
                ("org-06", "org-07"),
                ("org-06", "org-08"),
                ("org-07", "org-05"),
                ("org-07", "org-06"),
                ("org-07", "org-08"),
                ("org-07", "org-09"),
                ("org-08", "org-06"),
                ("org-08", "org-07"),
                ("org-08", "org-09"),
                ("org-08", "org-10"),
                ("org-09", "org-07"),
                ("org-09", "org-08"),
                ("org-09", "org-10"),
                ("org-10", "org-08"),
                ("org-10", "org-09"),
            ]
            .into_iter()
            .map(|(a, b)| (a.to_string(), b.to_string()))
            .collect::<BTreeSet<_>>()
        );
        assert_eq!(
            directed_pairs(&db, "NEARBY_OFFICE"),
            [
                ("org-01", "org-07"),
                ("org-01", "org-10"),
                ("org-02", "org-09"),
                ("org-03", "org-08"),
                ("org-04", "org-05"),
                ("org-04", "org-06"),
                ("org-05", "org-04"),
                ("org-05", "org-06"),
                ("org-06", "org-04"),
                ("org-06", "org-05"),
                ("org-07", "org-01"),
                ("org-07", "org-10"),
                ("org-08", "org-03"),
                ("org-09", "org-02"),
                ("org-10", "org-01"),
                ("org-10", "org-07"),
            ]
            .into_iter()
            .map(|(a, b)| (a.to_string(), b.to_string()))
            .collect::<BTreeSet<_>>()
        );
        assert_weight(&db, "org-01", "org-02", "founded_within", 0.5);
        let nyc_jc = 1.0 - haversine_km(40.7128, -74.0060, 40.7178, -74.0431) / 50.0;
        assert_weight(&db, "org-01", "org-07", "nearby_office", nyc_jc);
        assert_weight(&db, "person-01", "person-11", "similar_interests", 1.0);
        assert_weight(&db, "person-01", "person-09", "similar_interests", 0.8);

        let err = run_demo(&dir).expect_err("second run into the same dir");
        let msg = err.to_string().to_lowercase();
        assert!(
            msg.contains("not empty") || msg.contains("non-empty") || msg.contains("non empty"),
            "refuse message must mention non-empty dir, got {err}"
        );
        assert!(
            msg.contains("hidden"),
            "refuse message must mention hidden files, got {err}"
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn run_snapshot_writes_snapshot_bin() {
        let dir = tmp("snapshot-cli");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node("Person", "alice", vec![]).expect("insert");
        }
        assert!(
            !dir.join("snapshot.bin").exists(),
            "GraphDb Drop must not snapshot"
        );
        let out = run_snapshot(&dir, WalDisposition::Archive, None).expect("snapshot");
        assert!(
            dir.join("snapshot.bin").is_file(),
            "run_snapshot must write snapshot.bin"
        );
        assert!(
            out.contains("snapshot.bin"),
            "snapshot output should mention snapshot.bin, got {out}"
        );
        let db = GraphDb::open(&dir).expect("reopen");
        assert!(db.has_node("alice"), "reopen after snapshot must recover");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Every snapshot mushroomdb takes on its own — the ingest's, a
    /// `serve --snapshot-every` tick, the graceful-shutdown one — archives the
    /// WAL, so a store never loses its past to a write nobody asked for. Only
    /// an explicit `--truncate` ends that reach.
    #[test]
    fn an_automatic_snapshot_keeps_history_reachable_and_truncate_ends_it() {
        let dir = tmp("snapshot-archive");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node("Person", "alice", vec![]).expect("insert");
        }
        let before = core_api::wal_commit_count_at(&dir).expect("count");
        assert!(before > 0, "the insert is a commit");

        // Exactly the call `serve` makes on a tick and on shutdown.
        {
            let shared = SharedDb::open(&dir).expect("open");
            snapshot_shared(&shared).expect("snapshot");
        }

        let archives = || {
            std::fs::read_dir(&dir)
                .expect("read dir")
                .filter_map(Result::ok)
                .filter(|e| e.file_name().to_string_lossy().ends_with(".archive"))
                .count()
        };
        assert_eq!(archives(), 1, "the WAL was archived, not dropped");
        assert!(
            dir.join("wal.genesis").is_file(),
            "the genesis marker is what lets asof reach an archived commit"
        );
        {
            let db = GraphDb::open(&dir).expect("reopen");
            assert!(db.has_node("alice"));
            assert!(
                !db.node_history("alice").expect("history").items.is_empty(),
                "the insert is still explainable"
            );
        }
        assert!(
            GraphDb::open_at(&dir, before - 1).is_ok(),
            "asof still reaches a commit the snapshot folded in"
        );

        // A truncating snapshot is the destructive one, and only the user asks
        // for it.
        run_snapshot(&dir, WalDisposition::Truncate, None).expect("truncate");
        assert!(
            !dir.join("wal.genesis").exists(),
            "truncating ends asof's reach into the archives"
        );
        let db = GraphDb::open(&dir).expect("reopen");
        assert!(
            db.has_node("alice"),
            "the data survives; only the past goes"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Automatic snapshots keep every archive: history is the thing archives
    /// exist for, and nothing deletes it unless a caller asks with
    /// `--retention`.
    #[test]
    fn automatic_snapshots_keep_every_archive() {
        let dir = tmp("snapshot-retention");
        let archives = |d: &Path| {
            std::fs::read_dir(d)
                .expect("read dir")
                .filter_map(Result::ok)
                .filter(|e| e.file_name().to_string_lossy().ends_with(".archive"))
                .count()
        };

        // Ten rounds of "commit something, then take the snapshot the ingest,
        // the sync hook and the server tick all take".
        let rounds = 10;
        for i in 0..rounds {
            {
                let mut db = GraphDb::open(&dir).expect("open");
                db.insert_node("Person", &format!("p{i}"), vec![])
                    .expect("insert");
            }
            let shared = SharedDb::open(&dir).expect("open shared");
            snapshot_shared(&shared).expect("snapshot");
        }

        assert_eq!(
            archives(&dir),
            rounds,
            "an automatic snapshot must not delete an archive"
        );

        let db = GraphDb::open(&dir).expect("reopen");
        assert_eq!(
            db.wal_horizon_floor(),
            0,
            "nothing was pruned, so the floor stays at 0"
        );
        for i in 0..rounds {
            assert!(db.has_node(&format!("p{i}")), "p{i} survived");
        }
        assert!(
            !db.node_history("p0").expect("history").items.is_empty(),
            "the oldest history is still there: that is the point of the default"
        );
        assert_eq!(db.node_history("p0").expect("history").horizon, 0);
        drop(db);

        // The explicit command is the user's, and keeps everything unless the
        // user says otherwise.
        let manual = tmp("snapshot-retention-manual");
        for i in 0..3 {
            {
                let mut db = GraphDb::open(&manual).expect("open");
                db.insert_node("Person", &format!("p{i}"), vec![])
                    .expect("insert");
            }
            run_snapshot(&manual, WalDisposition::Archive, None).expect("snapshot");
        }
        assert_eq!(
            archives(&manual),
            3,
            "`mushroomdb snapshot` with no --retention keeps every archive"
        );

        let _ = std::fs::remove_dir_all(&dir);
        let _ = std::fs::remove_dir_all(&manual);
    }

    /// `--retention N` still bounds the archives when a caller asks for it —
    /// the automatic default changed, not the escape hatch.
    #[test]
    fn retention_is_still_available_when_configured() {
        let dir = tmp("retention-configured");
        for i in 0..5 {
            {
                let mut db = GraphDb::open(&dir).unwrap();
                db.insert_node("Person", &format!("p{i}"), vec![]).unwrap();
            }
            run_snapshot(&dir, WalDisposition::Archive, Some(2)).unwrap();
        }
        let archives = std::fs::read_dir(&dir)
            .expect("read dir")
            .filter_map(Result::ok)
            .filter(|e| e.file_name().to_string_lossy().ends_with(".archive"))
            .count();
        assert_eq!(archives, 2, "--retention 2 still prunes to two");
        assert!(GraphDb::open(&dir).unwrap().wal_horizon_floor() > 0);
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn run_query_formats_like_asof() {
        let dir = tmp("query-cli");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node(
                "Person",
                "alice",
                vec![("id".into(), Value::Str("alice".into()))],
            )
            .expect("insert");
        }
        let out = run_query(&dir, "MATCH (n:Person) RETURN n.id AS id", None, None).expect("query");
        assert!(out.contains("columns:"), "got {out}");
        assert!(out.contains("id=alice"), "got {out}");
        let _ = run_query(&dir, "CREATE (n:Person {id: 'bob'})", None, None).expect("write");
        let db = GraphDb::open(&dir).expect("reopen");
        assert!(db.has_node("bob"), "query_write must persist CREATE");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// The as-of header says how far back history reaches, and counts every
    /// commit the store still holds — including the archived ones the live
    /// WAL no longer carries.
    #[test]
    fn asof_header_names_the_horizon() {
        let dir = tmp("asof-horizon");
        // Ten rounds of "write, then snapshot with an explicit retention": the
        // bound prunes the oldest archives and the floor advances past 0. The
        // automatic path no longer prunes on its own, so the horizon here is
        // built with `--retention` rather than the automatic default.
        for i in 0..10 {
            {
                let mut db = GraphDb::open(&dir).expect("open");
                db.insert_node("Person", &format!("p{i}"), vec![])
                    .expect("insert");
            }
            run_snapshot(&dir, WalDisposition::Archive, Some(2)).expect("snapshot");
        }
        // One more write after the last snapshot: that commit lives in the live
        // WAL, which is what `asof` can still reconstruct on a pruned store.
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node("Person", "after", vec![]).expect("insert");
        }
        let (floor, total) = {
            let db = GraphDb::open(&dir).expect("reopen");
            (
                db.wal_horizon_floor(),
                db.wal_total_commits().expect("total"),
            )
        };
        assert!(floor > 0, "the retention must have pruned something");

        let out = run_asof(&dir, Some(total - 1), None, None, None).expect("asof");
        assert_eq!(
            out.trim(),
            format!(
                "as-of commit {} of {total} (history reaches back to commit {floor})",
                total - 1
            ),
            "the header must name the horizon it can reach"
        );
        let _ = std::fs::remove_dir_all(&dir);

        // A store that has pruned nothing reads exactly as it always did.
        let clean = tmp("asof-clean");
        {
            let mut db = GraphDb::open(&clean).expect("open");
            db.insert_node("Person", "a", vec![]).expect("insert");
        }
        assert_eq!(
            run_asof(&clean, Some(0), None, None, None)
                .expect("asof")
                .trim(),
            "as-of commit 0 of 1"
        );
        let _ = std::fs::remove_dir_all(&clean);
    }

    /// `stats` says whether history is complete, and where it starts when it
    /// is not.
    #[test]
    fn format_stats_says_how_far_back_history_reaches() {
        let dir = tmp("stats-horizon");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node("Person", "a", vec![]).expect("insert");
        }
        let text = format_stats(&read_stats(&dir).expect("stats"));
        assert!(
            text.contains("history: complete (nothing pruned)"),
            "an unpruned store says so, got:\n{text}"
        );
        let _ = std::fs::remove_dir_all(&dir);

        let pruned = tmp("stats-horizon-pruned");
        for i in 0..10 {
            {
                let mut db = GraphDb::open(&pruned).expect("open");
                db.insert_node("Person", &format!("p{i}"), vec![])
                    .expect("insert");
            }
            run_snapshot(&pruned, WalDisposition::Archive, Some(2)).expect("snapshot");
        }
        let stats = read_stats(&pruned).expect("stats");
        assert!(stats.history_floor > 0, "the retention must have pruned");
        assert!(
            format_stats(&stats).contains(&format!(
                "history: reaches back to commit {}",
                stats.history_floor
            )),
            "a pruned store names its floor"
        );
        let _ = std::fs::remove_dir_all(&pruned);
    }

    #[test]
    fn format_stats_contains_counts() {
        let dir = tmp("stats-smoke");
        let out = run_demo(&dir).expect("demo for stats smoke");
        let text = format_stats(&out.stats);
        assert!(
            text.contains("60"),
            "stats output should include live node count, got:\n{text}"
        );
        assert!(
            text.contains("334"),
            "stats output should include edge count, got:\n{text}"
        );
        assert!(
            text.to_lowercase().contains("node"),
            "stats output should mention nodes, got:\n{text}"
        );
        assert!(
            text.to_lowercase().contains("edge"),
            "stats output should mention edges, got:\n{text}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    // ── backup CLI tests ──────────────────────────────────────────────────────

    #[test]
    fn parse_backup_round_trip() {
        let r = parse_args(&["backup", "/db/dir", "/backup/dest"]);
        match r {
            Ok(Command::Backup { db_dir, dest }) => {
                assert_eq!(db_dir, PathBuf::from("/db/dir"));
                assert_eq!(dest, PathBuf::from("/backup/dest"));
            }
            other => panic!("backup parse, got {other:?}"),
        }
    }

    #[test]
    fn parse_backup_missing_dest_errors() {
        let r = parse_args(&["backup", "/db/dir"]);
        assert!(r.is_err(), "backup without <dest> should error");
        let e = r.unwrap_err();
        assert!(
            e.to_lowercase().contains("dest"),
            "error should mention dest, got: {e}"
        );
    }

    #[test]
    fn parse_export_defaults_to_jsonl() {
        let r = parse_args(&["export", "/db/dir", "/export/dest"]);
        match r {
            Ok(Command::Export { format, .. }) => {
                assert_eq!(format, ExportFormat::Jsonl);
            }
            other => panic!("export parse, got {other:?}"),
        }
    }

    #[test]
    fn parse_export_parquet_flag() {
        let r = parse_args(&["export", "/db/dir", "/export/dest", "--format", "parquet"]);
        match r {
            Ok(Command::Export { format, .. }) => {
                assert_eq!(format, ExportFormat::Parquet);
            }
            other => panic!("export --format parquet parse, got {other:?}"),
        }
    }

    #[test]
    fn parse_export_parquet_flag_eq() {
        let r = parse_args(&["export", "/db/dir", "/dest", "--format=parquet"]);
        match r {
            Ok(Command::Export { format, .. }) => {
                assert_eq!(format, ExportFormat::Parquet);
            }
            other => panic!("export --format=parquet parse, got {other:?}"),
        }
    }

    #[test]
    fn run_backup_cli_produces_verified_report() {
        let src = tmp("cli-backup-src");
        let dst = tmp("cli-backup-dst");
        let _ = run_demo(&src).expect("demo");
        let report = run_backup(&src, &dst).expect("run_backup");
        assert!(report.verified, "backup must be verified");
        assert!(!report.files.is_empty());
        assert!(report.bytes > 0);
        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst);
    }

    /// Build a one-node store at `dir` and snapshot it, so `backup_to` has a
    /// `snapshot.bin` to copy.
    fn seed_store(dir: &Path, key: &str) {
        let mut db = GraphDb::open(dir).expect("open seed store");
        db.insert_node("N", key, vec![]).expect("insert seed node");
        db.snapshot().expect("snapshot seed store");
    }

    #[test]
    fn restore_from_seeds_an_empty_dir() {
        let src = tmp("restore-src");
        seed_store(&src, "a");
        let vault = tmp("restore-vault");
        run_backup(&src, &vault.join("2026-09-10T00-00Z")).expect("backup");

        let fresh = tmp("restore-fresh");
        match restore_if_empty(&fresh, &vault).expect("restore_if_empty") {
            RestoreOutcome::Restored { files, bytes, .. } => {
                assert!(
                    files.contains(&"snapshot.bin".to_string()),
                    "expected snapshot.bin among {files:?}"
                );
                assert!(bytes > 0, "expected a non-zero byte count");
            }
            other => panic!("expected Restored, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh).expect("open restored").has_node("a"));

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&vault);
        let _ = std::fs::remove_dir_all(&fresh);
    }

    #[test]
    fn restore_from_a_backup_dir_itself() {
        let src = tmp("restore-direct-src");
        seed_store(&src, "a");
        let backup = tmp("restore-direct-backup");
        run_backup(&src, &backup).expect("backup");

        let fresh = tmp("restore-direct-fresh");
        match restore_if_empty(&fresh, &backup).expect("restore_if_empty") {
            RestoreOutcome::Restored { from, .. } => assert_eq!(from, backup),
            other => panic!("expected Restored, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh).expect("open restored").has_node("a"));

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&backup);
        let _ = std::fs::remove_dir_all(&fresh);
    }

    /// A store that has never snapshotted backs up as `wal.bin` and nothing
    /// else — `mushroomdb demo` then `mushroomdb backup` is exactly that — and
    /// it carries every commit. Ranking on `snapshot.bin` alone would skip it
    /// and start the server empty, which is the failure the flag exists to
    /// prevent.
    #[test]
    fn restore_from_a_wal_only_backup() {
        let src = tmp("restore-walonly-src");
        {
            let mut db = GraphDb::open(&src).expect("open src");
            db.insert_node("N", "a", vec![]).expect("insert");
        }
        assert!(
            !src.join("snapshot.bin").exists(),
            "test setup: src must not have snapshotted"
        );
        let vault = tmp("restore-walonly-vault");
        run_backup(&src, &vault.join("2026-09-10T00-00Z")).expect("backup");

        let fresh = tmp("restore-walonly-fresh");
        match restore_if_empty(&fresh, &vault).expect("restore_if_empty") {
            RestoreOutcome::Restored { files, .. } => assert!(
                files.contains(&"wal.bin".to_string()),
                "expected wal.bin among {files:?}"
            ),
            other => panic!("expected Restored, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh).expect("open restored").has_node("a"));

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&vault);
        let _ = std::fs::remove_dir_all(&fresh);
    }

    #[test]
    fn restore_from_picks_latest_then_newest() {
        let older_src = tmp("restore-rank-old-src");
        seed_store(&older_src, "old");
        let newer_src = tmp("restore-rank-new-src");
        seed_store(&newer_src, "new");
        let latest_src = tmp("restore-rank-latest-src");
        seed_store(&latest_src, "named-latest");

        // Newest by mtime, with no `latest/` present.
        let vault = tmp("restore-rank-vault");
        run_backup(&older_src, &vault.join("2026-09-01")).expect("backup old");
        std::thread::sleep(std::time::Duration::from_millis(1100));
        run_backup(&newer_src, &vault.join("2026-09-02")).expect("backup new");

        let fresh = tmp("restore-rank-fresh");
        match restore_if_empty(&fresh, &vault).expect("restore by mtime") {
            RestoreOutcome::Restored { from, .. } => assert_eq!(from, vault.join("2026-09-02")),
            other => panic!("expected Restored, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh)
            .expect("open restored")
            .has_node("new"));

        // `latest/` wins outright, even though it is older than 2026-09-02.
        run_backup(&latest_src, &vault.join("latest")).expect("backup latest");
        let older_than_latest = std::fs::metadata(vault.join("2026-09-02").join("snapshot.bin"))
            .expect("stat newest")
            .modified()
            .expect("mtime");
        let latest_mtime = std::fs::metadata(vault.join("latest").join("snapshot.bin"))
            .expect("stat latest")
            .modified()
            .expect("mtime");
        assert!(
            latest_mtime >= older_than_latest,
            "test setup: latest/ should not be older here"
        );

        let fresh2 = tmp("restore-rank-fresh2");
        match restore_if_empty(&fresh2, &vault).expect("restore by name") {
            RestoreOutcome::Restored { from, .. } => assert_eq!(from, vault.join("latest")),
            other => panic!("expected Restored, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh2)
            .expect("open restored")
            .has_node("named-latest"));

        for d in [&older_src, &newer_src, &latest_src, &vault, &fresh, &fresh2] {
            let _ = std::fs::remove_dir_all(d);
        }
    }

    #[test]
    fn restore_from_is_a_no_op_when_a_store_exists() {
        let src = tmp("restore-noop-src");
        seed_store(&src, "a");
        let vault = tmp("restore-noop-vault");
        run_backup(&src, &vault.join("2026-09-10T00-00Z")).expect("backup");

        let existing = tmp("restore-noop-existing");
        seed_store(&existing, "b");

        assert_eq!(
            restore_if_empty(&existing, &vault).expect("restore_if_empty"),
            RestoreOutcome::AlreadyPresent
        );
        let db = GraphDb::open(&existing).expect("open existing");
        assert!(db.has_node("b"), "the existing store must survive");
        assert!(!db.has_node("a"), "the backup must not have been copied in");

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&vault);
        let _ = std::fs::remove_dir_all(&existing);
    }

    #[test]
    fn restore_from_an_empty_vault_is_not_an_error() {
        let vault = tmp("restore-empty-vault");
        std::fs::create_dir_all(&vault).expect("mkdir vault");
        let fresh = tmp("restore-empty-fresh");
        assert_eq!(
            restore_if_empty(&fresh, &vault).expect("restore_if_empty"),
            RestoreOutcome::Empty
        );
        // A missing `from` is the same warning: a first boot with no volume yet.
        assert_eq!(
            restore_if_empty(&fresh, &vault.join("nope")).expect("restore_if_empty"),
            RestoreOutcome::Empty
        );

        let _ = std::fs::remove_dir_all(&vault);
        let _ = std::fs::remove_dir_all(&fresh);
    }

    #[test]
    fn restore_from_a_corrupt_backup_fails_loudly() {
        let src = tmp("restore-corrupt-src");
        seed_store(&src, "a");
        let vault = tmp("restore-corrupt-vault");
        let backup = vault.join("2026-09-10T00-00Z");
        run_backup(&src, &backup).expect("backup");

        // Truncate the copied snapshot: the CRC check on open must reject it.
        let snap = backup.join("snapshot.bin");
        let bytes = std::fs::read(&snap).expect("read snapshot");
        std::fs::write(&snap, &bytes[..bytes.len() / 2]).expect("truncate snapshot");

        let fresh = tmp("restore-corrupt-fresh");
        let err = restore_if_empty(&fresh, &vault).expect_err("expected a hard failure");
        let msg = err.to_string();
        assert!(
            msg.contains(&fresh.display().to_string()),
            "error must name the restored dir, got: {msg}"
        );
        assert!(
            msg.contains(&backup.display().to_string()),
            "error must name the backup, got: {msg}"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&vault);
        let _ = std::fs::remove_dir_all(&fresh);
    }

    /// A failed restore is all-or-nothing: nothing of the bad copy reaches
    /// `db_dir`, so the next boot restores rather than reporting
    /// `AlreadyPresent` over a half-written store.
    #[test]
    fn a_failed_restore_leaves_the_db_dir_untouched() {
        let src = tmp("restore-atomic-src");
        seed_store(&src, "a");

        let bad_vault = tmp("restore-atomic-bad-vault");
        let bad = bad_vault.join("2026-09-10T00-00Z");
        run_backup(&src, &bad).expect("backup the bad one");
        let snap = bad.join("snapshot.bin");
        let bytes = std::fs::read(&snap).expect("read snapshot");
        std::fs::write(&snap, &bytes[..bytes.len() / 2]).expect("truncate snapshot");

        let good_vault = tmp("restore-atomic-good-vault");
        run_backup(&src, &good_vault.join("2026-09-11T00-00Z")).expect("backup the good one");

        let fresh = tmp("restore-atomic-fresh");
        let err = restore_if_empty(&fresh, &bad_vault).expect_err("expected a hard failure");
        let msg = err.to_string();
        assert!(
            msg.contains(&fresh.display().to_string()) && msg.contains(&bad.display().to_string()),
            "error must name both paths, got: {msg}"
        );

        // Nothing was left behind — not the copied files, not the staging dir.
        let leftovers: Vec<String> = std::fs::read_dir(&fresh)
            .expect("read fresh")
            .flatten()
            .filter_map(|e| e.file_name().into_string().ok())
            .collect();
        assert!(
            leftovers.is_empty(),
            "a failed restore must leave nothing behind, found: {leftovers:?}"
        );
        assert!(!holds_a_store(&fresh), "the dir must not hold a store");

        // So the retry against a good backup restores, rather than deciding a
        // store is already there.
        match restore_if_empty(&fresh, &good_vault).expect("retry must restore") {
            RestoreOutcome::Restored { from, .. } => {
                assert_eq!(from, good_vault.join("2026-09-11T00-00Z"))
            }
            other => panic!("expected Restored on retry, got {other:?}"),
        }
        assert!(GraphDb::open(&fresh).expect("open restored").has_node("a"));

        for d in [&src, &bad_vault, &good_vault, &fresh] {
            let _ = std::fs::remove_dir_all(d);
        }
    }

    /// The staging directory never outlives a successful restore either — a
    /// served store directory holds the store and nothing else.
    #[test]
    fn a_successful_restore_leaves_no_staging_dir() {
        let src = tmp("restore-staging-src");
        seed_store(&src, "a");
        let vault = tmp("restore-staging-vault");
        run_backup(&src, &vault.join("2026-09-11T00-00Z")).expect("backup");

        let fresh = tmp("restore-staging-fresh");
        restore_if_empty(&fresh, &vault).expect("restore");

        let stray: Vec<String> = std::fs::read_dir(&fresh)
            .expect("read fresh")
            .flatten()
            .filter_map(|e| e.file_name().into_string().ok())
            .filter(|n| n.starts_with(".restore-"))
            .collect();
        assert!(
            stray.is_empty(),
            "staging dir must be gone, found: {stray:?}"
        );

        for d in [&src, &vault, &fresh] {
            let _ = std::fs::remove_dir_all(d);
        }
    }

    #[test]
    fn serve_parses_restore_from() {
        match parse_args(&["serve", "/tmp/db", "--restore-from", "/vol/backups"]) {
            Ok(Command::Serve { restore_from, .. }) => {
                assert_eq!(restore_from, Some(PathBuf::from("/vol/backups")))
            }
            other => panic!("--restore-from parse, got {other:?}"),
        }
        match parse_args(&["serve", "/tmp/db", "--restore-from=/vol/backups"]) {
            Ok(Command::Serve { restore_from, .. }) => {
                assert_eq!(restore_from, Some(PathBuf::from("/vol/backups")))
            }
            other => panic!("--restore-from= parse, got {other:?}"),
        }
        match parse_args(&["serve", "/tmp/db"]) {
            Ok(Command::Serve { restore_from, .. }) => assert_eq!(restore_from, None),
            other => panic!("default restore_from, got {other:?}"),
        }
        let err = parse_args(&["serve", "/tmp/db", "--restore-from"])
            .expect_err("missing value must be an error");
        assert!(
            err.contains("--restore-from"),
            "error must name the flag, got: {err}"
        );
    }

    #[test]
    fn run_export_jsonl_two_runs_byte_identical() {
        let src = tmp("cli-export-src");
        let dst1 = tmp("cli-export-dst1");
        let dst2 = tmp("cli-export-dst2");
        let _ = run_demo(&src).expect("demo");

        run_export(&src, &dst1, &ExportFormat::Jsonl).expect("first export");
        run_export(&src, &dst2, &ExportFormat::Jsonl).expect("second export");

        for filename in &["nodes.jsonl", "edges.jsonl", "rules.jsonl"] {
            let f1 = std::fs::read(dst1.join(filename)).expect("read first");
            let f2 = std::fs::read(dst2.join(filename)).expect("read second");
            assert_eq!(
                f1, f2,
                "{filename} must be byte-identical across two export runs"
            );
        }
        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst1);
        let _ = std::fs::remove_dir_all(&dst2);
    }

    #[test]
    fn run_export_jsonl_nodes_are_sorted() {
        let src = tmp("cli-export-sorted");
        let dst = tmp("cli-export-sorted-dst");
        let _ = run_demo(&src).expect("demo");
        run_export(&src, &dst, &ExportFormat::Jsonl).expect("export");

        let content = std::fs::read_to_string(dst.join("nodes.jsonl")).expect("read nodes");
        let keys: Vec<String> = content
            .lines()
            .filter(|l| !l.is_empty())
            .map(|l| {
                let v: serde_json::Value = serde_json::from_str(l).expect("parse line");
                v["key"].as_str().unwrap_or("").to_string()
            })
            .collect();
        let mut sorted = keys.clone();
        sorted.sort();
        assert_eq!(keys, sorted, "nodes.jsonl must be sorted by key");
        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst);
    }

    #[test]
    fn run_export_jsonl_derived_edges_have_rule() {
        let src = tmp("cli-export-derived");
        let dst = tmp("cli-export-derived-dst");
        let _ = run_demo(&src).expect("demo");
        run_export(&src, &dst, &ExportFormat::Jsonl).expect("export");

        let content = std::fs::read_to_string(dst.join("edges.jsonl")).expect("read edges");
        let derived_lines: Vec<serde_json::Value> = content
            .lines()
            .filter(|l| !l.is_empty())
            .map(|l| serde_json::from_str(l).expect("parse line"))
            .filter(|v: &serde_json::Value| v["derived"].as_bool().unwrap_or(false))
            .collect();
        assert!(
            !derived_lines.is_empty(),
            "demo store should have derived edges"
        );
        for edge in &derived_lines {
            assert!(
                !edge["rule"].is_null(),
                "derived edge must have non-null rule: {edge}"
            );
        }
        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst);
    }

    #[test]
    fn run_export_parquet_produces_files() {
        let src = tmp("cli-export-parq-src");
        let dst = tmp("cli-export-parq-dst");
        let _ = run_demo(&src).expect("demo");
        run_export(&src, &dst, &ExportFormat::Parquet).expect("parquet export");

        assert!(
            dst.join("nodes.parquet").exists(),
            "nodes.parquet must exist"
        );
        assert!(
            dst.join("edges.parquet").exists(),
            "edges.parquet must exist"
        );
        assert!(
            dst.join("rules.parquet").exists(),
            "rules.parquet must exist"
        );
        // All files must be non-empty.
        for f in &["nodes.parquet", "edges.parquet", "rules.parquet"] {
            let meta = std::fs::metadata(dst.join(f)).expect("metadata");
            assert!(meta.len() > 0, "{f} must be non-empty");
        }
        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst);
    }

    #[test]
    fn parse_export_graphml_flag() {
        let r = parse_args(&["export", "/db/dir", "/dest", "--format", "graphml"]);
        match r {
            Ok(Command::Export { format, .. }) => {
                assert_eq!(format, ExportFormat::Graphml);
            }
            other => panic!("export --format graphml parse, got {other:?}"),
        }
    }

    #[test]
    fn run_export_graphml_structure() {
        let src = tmp("cli-export-gml-src");
        let dst_dir = tmp("cli-export-gml-dst");
        let dst = dst_dir.join("graph.graphml");
        let _ = run_demo(&src).expect("demo");
        run_export(&src, &dst, &ExportFormat::Graphml).expect("graphml export");

        let content = std::fs::read_to_string(&dst).expect("read graphml");

        assert!(
            content.starts_with("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"),
            "must start with an XML declaration"
        );
        assert!(
            content.contains("<graphml xmlns=\"http://graphml.graphdrawing.org/xmlns\">"),
            "must use the standard GraphML namespace"
        );
        assert!(
            content.contains(
                "<key id=\"n_label\" for=\"node\" attr.name=\"label\" attr.type=\"string\"/>"
            ),
            "must declare the node label key"
        );
        assert!(
            content.contains(
                "<key id=\"e_type\" for=\"edge\" attr.name=\"type\" attr.type=\"string\"/>"
            ),
            "must declare the edge type key"
        );
        assert!(
            content.contains(
                "<key id=\"e_derived\" for=\"edge\" attr.name=\"derived\" attr.type=\"boolean\"/>"
            ),
            "must declare the edge derived key"
        );
        assert!(
            content.contains(
                "<key id=\"e_rule\" for=\"edge\" attr.name=\"rule\" attr.type=\"string\"/>"
            ),
            "must declare the edge rule key"
        );
        assert!(
            content.contains(
                "<key id=\"e_weight\" for=\"edge\" attr.name=\"weight\" attr.type=\"double\"/>"
            ),
            "must declare the edge weight key"
        );
        // The demo store's Org nodes carry `founded_year` as a JSON integer
        // (`Value::Int`, a 64-bit i64). GraphML's informal convention treats
        // `attr.type="int"` as 32-bit, so this must declare `"long"`.
        assert!(
            content.contains(
                "<key id=\"n_founded_year\" for=\"node\" attr.name=\"founded_year\" attr.type=\"long\"/>"
            ),
            "an int-valued prop must declare attr.type=\"long\", not \"int\", got: {content}"
        );
        assert!(
            content.contains("<graph id=\"G\" edgedefault=\"directed\">"),
            "must declare a single directed graph element"
        );
        assert!(content.contains("<node id="), "must contain node elements");
        assert!(
            content.contains("<edge id=\"e0\" source=\""),
            "must contain a sequentially-numbered edge starting at e0"
        );
        assert!(
            content.trim_end().ends_with("</graphml>"),
            "must close the root element"
        );

        // The demo store's `skill_fit` rule declares weight_prop "score"; its
        // derived FIT edges must carry both a rule and a weight in GraphML.
        assert!(
            content.contains("<data key=\"e_rule\">skill_fit</data>")
                || content.contains("<data key=\"e_rule\">founded_within</data>"),
            "at least one derived edge must carry its rule name"
        );
        assert!(
            content.contains(&format!(
                "<data key=\"{}\">",
                "e_weight" /* declared above */
            )),
            "at least one derived edge must carry a weight value"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir);
    }

    #[test]
    fn run_export_graphml_dest_dir_writes_graph_dot_graphml() {
        let src = tmp("cli-export-gml-dir-src");
        let dst_dir = tmp("cli-export-gml-dir-dst");
        std::fs::create_dir_all(&dst_dir).expect("mkdir dest");
        let _ = run_demo(&src).expect("demo");

        let msg = run_export(&src, &dst_dir, &ExportFormat::Graphml).expect("graphml export");

        assert!(
            dst_dir.join("graph.graphml").exists(),
            "an existing directory dest must produce dest/graph.graphml"
        );
        assert!(
            msg.contains("graph.graphml"),
            "report must name the file actually written, got: {msg}"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir);
    }

    /// python3-gated: parses the exported file with the stdlib XML parser to
    /// confirm it is well-formed. Skipped (not failed) when python3 is absent.
    #[test]
    fn run_export_graphml_is_well_formed_xml() {
        let has_python3 = std::process::Command::new("python3")
            .arg("--version")
            .output()
            .map(|o| o.status.success())
            .unwrap_or(false);
        if !has_python3 {
            eprintln!("skipping run_export_graphml_is_well_formed_xml: python3 not found");
            return;
        }

        let src = tmp("cli-export-gml-wf-src");
        let dst_dir = tmp("cli-export-gml-wf-dst");
        let dst = dst_dir.join("graph.graphml");
        let _ = run_demo(&src).expect("demo");
        run_export(&src, &dst, &ExportFormat::Graphml).expect("graphml export");

        let status = std::process::Command::new("python3")
            .arg("-c")
            .arg("import sys, xml.etree.ElementTree as E; E.parse(sys.argv[1])")
            .arg(&dst)
            .status()
            .expect("run python3");
        assert!(
            status.success(),
            "python3's XML parser must accept the exported GraphML file"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir);
    }

    #[test]
    fn run_export_graphml_two_runs_byte_identical() {
        let src = tmp("cli-export-gml-bi-src");
        let dst_dir1 = tmp("cli-export-gml-bi-dst1");
        let dst_dir2 = tmp("cli-export-gml-bi-dst2");
        let dst1 = dst_dir1.join("graph.graphml");
        let dst2 = dst_dir2.join("graph.graphml");
        let _ = run_demo(&src).expect("demo");

        run_export(&src, &dst1, &ExportFormat::Graphml).expect("first export");
        run_export(&src, &dst2, &ExportFormat::Graphml).expect("second export");

        let f1 = std::fs::read(&dst1).expect("read first");
        let f2 = std::fs::read(&dst2).expect("read second");
        assert_eq!(
            f1, f2,
            "graph.graphml must be byte-identical across two export runs"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir1);
        let _ = std::fs::remove_dir_all(&dst_dir2);
    }

    #[test]
    fn run_export_graphml_escapes_and_lists() {
        use core_api::{GraphDb, Value};
        let src = tmp("cli-export-gml-esc-src");
        let dst_dir = tmp("cli-export-gml-esc-dst");
        let dst = dst_dir.join("graph.graphml");

        {
            let mut db = GraphDb::open(&src).unwrap();
            db.insert_node(
                "Widget",
                "w1",
                vec![
                    (
                        "title".into(),
                        Value::Str("Tom & Jerry <says> \"hi\" 'bye'".into()),
                    ),
                    (
                        "tags".into(),
                        Value::List(vec![Value::Str("a".into()), Value::Str("b".into())]),
                    ),
                ],
            )
            .unwrap();
        }

        run_export(&src, &dst, &ExportFormat::Graphml).expect("graphml export");
        let content = std::fs::read_to_string(&dst).expect("read graphml");

        assert!(
            content.contains("Tom &amp; Jerry &lt;says&gt; &quot;hi&quot; &apos;bye&apos;"),
            "special XML characters in string props must be escaped, got: {content}"
        );
        assert!(
            !content.contains("Tom & Jerry <says>"),
            "unescaped special characters must not appear verbatim"
        );
        assert!(
            content.contains(
                "<key id=\"n_tags\" for=\"node\" attr.name=\"tags\" attr.type=\"string\"/>"
            ),
            "list-valued props must declare attr.type=\"string\""
        );
        assert!(
            content.contains("<data key=\"n_tags\">[&quot;a&quot;,&quot;b&quot;]</data>"),
            "list-valued props must render as XML-escaped JSON text, got: {content}"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir);
    }

    /// I2: when nodes disagree on the `Value` variant for a prop name (one
    /// int, one string), the key must declare `attr.type="string"` — a value
    /// of either type fits text — rather than picking one node's type and
    /// risking a value that doesn't fit it.
    #[test]
    fn run_export_graphml_mixed_type_prop_declares_string() {
        use core_api::{GraphDb, Value};
        let src = tmp("cli-export-gml-mixed-src");
        let dst_dir = tmp("cli-export-gml-mixed-dst");
        let dst = dst_dir.join("graph.graphml");

        {
            let mut db = GraphDb::open(&src).unwrap();
            db.insert_node("Metric", "m1", vec![("score".into(), Value::Int(5))])
                .unwrap();
            db.insert_node(
                "Metric",
                "m2",
                vec![("score".into(), Value::Str("high".into()))],
            )
            .unwrap();
        }

        run_export(&src, &dst, &ExportFormat::Graphml).expect("graphml export");
        let content = std::fs::read_to_string(&dst).expect("read graphml");

        assert!(
            content.contains(
                "<key id=\"n_score\" for=\"node\" attr.name=\"score\" attr.type=\"string\"/>"
            ),
            "a prop name with conflicting value types across nodes must declare \
             attr.type=\"string\", got: {content}"
        );
        assert!(
            !content.contains("attr.name=\"score\" attr.type=\"long\""),
            "must not declare a narrower type once a conflict is seen, got: {content}"
        );
        // Each node's own value still renders in its own literal text form,
        // regardless of the declared (fallback) attr.type.
        assert!(
            content.contains("<data key=\"n_score\">5</data>"),
            "the int-valued node must still render its literal int text, got: {content}"
        );
        assert!(
            content.contains("<data key=\"n_score\">high</data>"),
            "the string-valued node must still render its literal string text, got: {content}"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst_dir);
    }

    #[test]
    fn parse_algo_degree_defaults_dir_both() {
        let cmd = parse_args(&["algo", "degree", "/db"]).unwrap();
        match cmd {
            Command::Algo { dir, .. } => assert_eq!(dir, AlgoDir::Both),
            other => panic!("expected Algo, got {other:?}"),
        }
    }

    #[test]
    fn parse_algo_degree_with_dir_flag() {
        for (arg, want) in [
            ("out", AlgoDir::Out),
            ("in", AlgoDir::In),
            ("both", AlgoDir::Both),
        ] {
            let cmd = parse_args(&["algo", "degree", "/db", "--dir", arg]).unwrap();
            match cmd {
                Command::Algo { dir, .. } => assert_eq!(dir, want, "--dir {arg}"),
                other => panic!("expected Algo, got {other:?}"),
            }
        }
        // `--dir=out` form too.
        let cmd = parse_args(&["algo", "degree", "/db", "--dir=in"]).unwrap();
        match cmd {
            Command::Algo { dir, .. } => assert_eq!(dir, AlgoDir::In),
            other => panic!("expected Algo, got {other:?}"),
        }
    }

    #[test]
    fn parse_algo_rejects_unknown_dir() {
        assert!(parse_args(&["algo", "degree", "/db", "--dir", "sideways"]).is_err());
    }

    #[test]
    fn parse_algo_communities_parses_edge_type_weight_prop_min_weight() {
        let cmd = parse_args(&[
            "algo",
            "communities",
            "/db",
            "--edge-type",
            "IMPORTS",
            "--edge-type=CO_CHANGED",
            "--weight-prop",
            "score",
            "--min-weight",
            "0.3",
            "--top",
            "5",
        ])
        .unwrap();
        match cmd {
            Command::Algo {
                subcmd,
                top,
                edge_types,
                weight_prop,
                min_weight,
                ..
            } => {
                assert_eq!(subcmd, AlgoSubcmd::Communities);
                assert_eq!(top, 5);
                assert_eq!(
                    edge_types,
                    vec!["IMPORTS".to_string(), "CO_CHANGED".to_string()]
                );
                assert_eq!(weight_prop, Some("score".to_string()));
                assert_eq!(min_weight, Some(0.3));
            }
            other => panic!("expected Algo, got {other:?}"),
        }
    }

    #[test]
    fn parse_algo_communities_defaults_have_no_edge_type_or_weight_filter() {
        let cmd = parse_args(&["algo", "communities", "/db"]).unwrap();
        match cmd {
            Command::Algo {
                subcmd,
                edge_types,
                weight_prop,
                min_weight,
                ..
            } => {
                assert_eq!(subcmd, AlgoSubcmd::Communities);
                assert!(edge_types.is_empty());
                assert_eq!(weight_prop, None);
                assert_eq!(min_weight, None);
            }
            other => panic!("expected Algo, got {other:?}"),
        }
    }

    /// I1: exporting a store containing NaN/Inf floats must succeed, not panic.
    /// The NaN field must be serialised as JSON null (lossy but safe).
    #[test]
    fn run_export_jsonl_nan_float_becomes_null() {
        use core_api::{GraphDb, Value};
        let src = tmp("cli-export-nan-src");
        let dst = tmp("cli-export-nan-dst");

        // Insert a node with NaN, +Inf, and -Inf properties via the public API.
        {
            let mut db = GraphDb::open(&src).unwrap();
            db.insert_node(
                "Sensor",
                "s1",
                vec![
                    ("nan_val".into(), Value::Float(f64::NAN)),
                    ("pos_inf".into(), Value::Float(f64::INFINITY)),
                    ("neg_inf".into(), Value::Float(f64::NEG_INFINITY)),
                    ("normal".into(), Value::Float(1.5)),
                ],
            )
            .unwrap();
        }

        // Export must succeed.
        run_export(&src, &dst, &ExportFormat::Jsonl).expect("export with NaN must succeed");

        // nodes.jsonl must exist and the NaN fields must be null.
        let content =
            std::fs::read_to_string(dst.join("nodes.jsonl")).expect("nodes.jsonl missing");
        let row: serde_json::Value =
            serde_json::from_str(content.lines().next().unwrap()).expect("valid json line");
        assert_eq!(
            row["nan_val"],
            serde_json::Value::Null,
            "NaN must export as null"
        );
        assert_eq!(
            row["pos_inf"],
            serde_json::Value::Null,
            "+Inf must export as null"
        );
        assert_eq!(
            row["neg_inf"],
            serde_json::Value::Null,
            "-Inf must export as null"
        );
        // Normal float must survive.
        assert_eq!(
            row["normal"],
            serde_json::json!(1.5),
            "normal float roundtrips"
        );

        let _ = std::fs::remove_dir_all(&src);
        let _ = std::fs::remove_dir_all(&dst);
    }

    #[test]
    fn serve_tls_flags_parse_both_forms() {
        // --tls-cert VALUE --tls-key VALUE (space form)
        match parse_args(&[
            "serve",
            "/tmp/db",
            "--tls-cert",
            "/a/cert.pem",
            "--tls-key",
            "/a/key.pem",
        ])
        .unwrap()
        {
            Command::Serve {
                tls_cert, tls_key, ..
            } => {
                assert_eq!(tls_cert, Some(PathBuf::from("/a/cert.pem")));
                assert_eq!(tls_key, Some(PathBuf::from("/a/key.pem")));
            }
            other => panic!("{other:?}"),
        }
        // --tls-cert=VALUE --tls-key=VALUE (equals form)
        match parse_args(&[
            "serve",
            "/tmp/db",
            "--tls-cert=/b/cert.pem",
            "--tls-key=/b/key.pem",
        ])
        .unwrap()
        {
            Command::Serve {
                tls_cert, tls_key, ..
            } => {
                assert_eq!(tls_cert, Some(PathBuf::from("/b/cert.pem")));
                assert_eq!(tls_key, Some(PathBuf::from("/b/key.pem")));
            }
            other => panic!("{other:?}"),
        }
        // Neither → both None.
        match parse_args(&["serve", "/tmp/db"]).unwrap() {
            Command::Serve {
                tls_cert, tls_key, ..
            } => {
                assert_eq!(tls_cert, None);
                assert_eq!(tls_key, None);
            }
            other => panic!("{other:?}"),
        }
    }

    #[test]
    fn serve_tls_flags_require_both() {
        // --tls-cert alone → error
        let err = parse_args(&["serve", "/tmp/db", "--tls-cert", "/a/cert.pem"]).unwrap_err();
        assert!(
            err.contains("tls-key"),
            "--tls-cert alone must mention --tls-key in error, got {err}"
        );
        // --tls-key alone → error
        let err = parse_args(&["serve", "/tmp/db", "--tls-key", "/a/key.pem"]).unwrap_err();
        assert!(
            err.contains("tls-cert"),
            "--tls-key alone must mention --tls-cert in error, got {err}"
        );
    }

    #[test]
    fn version_flag_parses() {
        assert_eq!(parse_args(&["--version"]).unwrap(), Command::Version);
        assert_eq!(parse_args(&["-V"]).unwrap(), Command::Version);
        assert_eq!(parse_args(&["version"]).unwrap(), Command::Version);
    }

    #[test]
    fn recall_parses_one_dir_and_is_listed_in_usage() {
        assert_eq!(
            parse_args(&["recall", "/tmp/db"]).unwrap(),
            Command::Recall {
                db_dir: Some(PathBuf::from("/tmp/db")),
                auto: false,
            }
        );
        assert!(
            parse_args(&["recall"]).is_err(),
            "one of <db-dir> or --auto is required"
        );
        assert!(usage().contains("mushroomdb recall <db-dir>"));
    }

    #[test]
    fn why_takes_a_dir_and_two_keys() {
        assert_eq!(
            parse_args(&["why", "/tmp/db", "a.rs", "b.rs"]).unwrap(),
            Command::Why {
                db_dir: PathBuf::from("/tmp/db"),
                a: "a.rs".to_string(),
                b: "b.rs".to_string(),
            }
        );

        // Too few arguments, too many, and a key that looks like a flag.
        for args in [
            vec!["why", "/tmp/db", "a"],
            vec!["why", "/tmp/db", "a", "b", "c"],
            vec!["why", "/tmp/db", "-a", "b"],
        ] {
            assert!(parse_args(&args).is_err(), "{args:?} must not parse");
        }
        assert!(usage().contains("mushroomdb why <db-dir> <a> <b>"));
    }

    /// Every hook-driven command takes either a path or `--auto`, never both
    /// and never neither.
    #[test]
    fn hook_commands_take_a_dir_or_auto() {
        assert_eq!(
            parse_args(&["mcp", "--auto"]).unwrap(),
            Command::Mcp {
                db_dir: None,
                auto: true,
                all_tools: false
            }
        );
        assert_eq!(
            parse_args(&["recall", "--auto"]).unwrap(),
            Command::Recall {
                db_dir: None,
                auto: true
            }
        );
        assert_eq!(
            parse_args(&["brief", "--auto"]).unwrap(),
            Command::Brief {
                db_dir: None,
                auto: true
            }
        );
        assert_eq!(
            parse_args(&["brief", "/tmp/db"]).unwrap(),
            Command::Brief {
                db_dir: Some(PathBuf::from("/tmp/db")),
                auto: false
            }
        );
        for cmd in ["mcp", "recall", "brief"] {
            assert!(parse_args(&[cmd]).is_err(), "{cmd} with no target");
            assert!(
                parse_args(&[cmd, "/tmp/db", "--auto"]).is_err(),
                "{cmd} with both"
            );
        }
        assert!(usage().contains("--auto"));
    }

    /// Binding: `--all-tools` is `mcp`'s alone, sits either side of the store
    /// path, and every other flag is still rejected.
    #[test]
    fn mcp_takes_all_tools() {
        for args in [
            &["mcp", "/tmp/db", "--all-tools"][..],
            &["mcp", "--all-tools", "/tmp/db"][..],
        ] {
            assert_eq!(
                parse_args(args).unwrap(),
                Command::Mcp {
                    db_dir: Some(PathBuf::from("/tmp/db")),
                    auto: false,
                    all_tools: true
                },
                "{args:?}"
            );
        }
        assert_eq!(
            parse_args(&["mcp", "--auto", "--all-tools"]).unwrap(),
            Command::Mcp {
                db_dir: None,
                auto: true,
                all_tools: true
            }
        );
        assert!(parse_args(&["mcp", "--all-tools"]).is_err(), "no target");
        assert!(parse_args(&["mcp", "/tmp/db", "--nope"]).is_err());
        assert!(parse_args(&["recall", "/tmp/db", "--all-tools"]).is_err());
        assert!(usage().contains("--all-tools"));
    }

    #[test]
    fn ingest_git_parses_excludes() {
        let cmd = parse_args(&[
            "ingest-git",
            "/tmp/db",
            "/tmp/repo",
            "--exclude",
            "target/",
            "--exclude=*.lock",
            "--max-commits-per-file",
            "50",
            "--recurse-submodules",
            "--prs",
            "--ensure-gitignore",
        ])
        .unwrap();
        assert_eq!(
            cmd,
            Command::IngestGit {
                db_dir: PathBuf::from("/tmp/db"),
                opts: ingest_git::IngestGitOpts {
                    repo: PathBuf::from("/tmp/repo"),
                    exclude: vec!["target/".into(), "*.lock".into()],
                    max_commits_per_file: 50,
                    recurse_submodules: true,
                    prs: true,
                    structure: true,
                    docs: true,
                    ensure_gitignore: true,
                },
            }
        );
        // Defaults and arity.
        let Command::IngestGit { opts, .. } =
            parse_args(&["ingest-git", "/tmp/db", "/tmp/repo"]).unwrap()
        else {
            panic!("expected IngestGit");
        };
        assert_eq!(
            opts.exclude,
            ingest_git::DEFAULT_EXCLUDES
                .iter()
                .map(|p| (*p).to_string())
                .collect::<Vec<_>>(),
            "with no --exclude the defaults apply"
        );
        assert_eq!(
            opts.max_commits_per_file,
            ingest_git::DEFAULT_MAX_COMMITS_PER_FILE
        );
        assert!(!opts.recurse_submodules && !opts.prs && !opts.ensure_gitignore);
        assert!(
            opts.structure && opts.docs,
            "structure and docs default on and are recorded on the marker"
        );
        let Command::IngestGit { opts, .. } = parse_args(&[
            "ingest-git",
            "/tmp/db",
            "/tmp/repo",
            "--no-structure",
            "--no-docs",
        ])
        .unwrap() else {
            panic!("expected IngestGit");
        };
        assert!(!opts.structure && !opts.docs);
        assert!(parse_args(&["ingest-git", "/tmp/db"]).is_err());
        assert!(parse_args(&["ingest-git", "/tmp/db", "/tmp/repo", "--nope"]).is_err());
        assert!(parse_args(&["ingest-git", "/tmp/db", "/tmp/repo", "--exclude"]).is_err());
        assert!(usage().contains("mushroomdb ingest-git <db-dir> <repo-dir>"));
    }

    #[test]
    fn version_constant_matches_cargo() {
        assert_eq!(VERSION, env!("CARGO_PKG_VERSION"));
        assert!(usage().contains("--version"));
    }
    // ── Namespaces on the CLI ────────────────────────────────────────────────

    /// A store with two namespaces and one role bound to the first.
    fn ns_store(name: &str) -> PathBuf {
        let dir = tmp(name);
        let mut db = GraphDb::open(&dir).expect("open");
        for (key, ns) in [
            ("d1", None),
            ("a1", Some("tenant-a")),
            ("a2", Some("tenant-a")),
            ("b1", Some("tenant-b")),
        ] {
            let mut props = vec![("id".into(), Value::Str(key.into()))];
            if let Some(ns) = ns {
                props.push(("ns".into(), Value::Str(ns.into())));
            }
            db.insert_node("Doc", key, props).expect("insert");
        }
        db.apply_schema(&Schema {
            roles: vec![core_api::RoleDef {
                name: "a-reader".into(),
                keys: vec![],
                labels: vec!["Doc".into()],
                visible_where: None,
                namespaces: Some(vec!["tenant-a".into()]),
                write: None,
            }],
            ..Default::default()
        })
        .expect("schema");
        dir
    }

    /// `stats` names every namespace and its live count once a store has more
    /// than the default one.
    #[test]
    fn format_stats_lists_namespaces() {
        let dir = ns_store("stats-namespaces");
        let text = format_stats(&read_stats(&dir).expect("stats"));
        assert!(
            text.contains("namespaces: default (1), tenant-a (2), tenant-b (1)"),
            "got:\n{text}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// A single-tenant store's `stats` output is byte-identical to what it was
    /// before namespaces existed: no line at all.
    #[test]
    fn format_stats_omits_the_namespaces_line_on_one_namespace() {
        let dir = tmp("stats-one-namespace");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node("Person", "a", vec![]).expect("insert");
        }
        let text = format_stats(&read_stats(&dir).expect("stats"));
        assert!(
            !text.contains("namespaces"),
            "a store with only `default` says nothing about namespaces, got:\n{text}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `query --namespace` and `query --role` parse, and they compose.
    #[test]
    fn parse_query_takes_a_role_and_a_namespace() {
        let Ok(Command::Query {
            role, namespace, ..
        }) = parse_args(&[
            "query",
            "/tmp/db",
            "--role",
            "a-reader",
            "--namespace",
            "tenant-a",
            "MATCH (n) RETURN n",
        ])
        else {
            panic!("expected Query");
        };
        assert_eq!(role.as_deref(), Some("a-reader"));
        assert_eq!(namespace.as_deref(), Some("tenant-a"));

        let Ok(Command::Query {
            role, namespace, ..
        }) = parse_args(&[
            "query",
            "/tmp/db",
            "--namespace=tenant-b",
            "MATCH (n) RETURN n",
        ])
        else {
            panic!("expected Query");
        };
        assert_eq!(role, None);
        assert_eq!(namespace.as_deref(), Some("tenant-b"));

        assert!(parse_args(&["query", "/tmp/db", "--namespace"]).is_err());
        assert!(parse_args(&["query", "/tmp/db", "--role"]).is_err());
        assert!(usage().contains("--namespace <ns>"));
    }

    /// `query --role` answers as that role, `--namespace` narrows, and the two
    /// together intersect — a role bound to one namespace never sees another.
    #[test]
    fn run_query_with_a_role_and_a_namespace_never_widens() {
        let dir = ns_store("query-namespace");
        let q = "MATCH (n) RETURN n.id AS id ORDER BY n.id";

        let all = run_query(&dir, q, None, None).expect("query");
        assert!(all.contains("id=a1") && all.contains("id=b1") && all.contains("id=d1"));

        let ns = run_query(&dir, q, None, Some("tenant-a")).expect("query");
        assert!(ns.contains("id=a1") && ns.contains("id=a2"), "got {ns}");
        assert!(!ns.contains("id=b1") && !ns.contains("id=d1"), "got {ns}");

        let role = run_query(&dir, q, Some("a-reader"), None).expect("query");
        assert!(
            role.contains("id=a1") && !role.contains("id=b1"),
            "got {role}"
        );

        let both = run_query(&dir, q, Some("a-reader"), Some("tenant-b")).expect("query");
        assert!(
            !both.contains("id=a1") && !both.contains("id=b1"),
            "role ∩ namespace, never role ∪ namespace: {both}"
        );

        // Either argument makes the query a read: a restricted write is refused
        // and nothing lands.
        for (role, namespace) in [
            (None, Some("tenant-a")),
            (Some("a-reader"), None),
            (Some("a-reader"), Some("tenant-a")),
        ] {
            let write = run_query(&dir, "CREATE (n:Doc {id: 'z1'})", role, namespace);
            assert!(
                write
                    .as_ref()
                    .err()
                    .is_some_and(|e| e.0.contains("read-only")),
                "a restricted write must be refused, got {write:?}"
            );
            assert!(
                !GraphDb::open(&dir).expect("reopen").has_node("z1"),
                "the write must not have landed"
            );
        }

        let unknown = run_query(&dir, q, Some("nobody"), None);
        assert!(unknown.is_err(), "an unknown role is an error");
        let invalid = run_query(&dir, q, None, Some("no spaces"));
        assert!(
            invalid
                .as_ref()
                .err()
                .is_some_and(|e| e.0.contains("valid namespace name")),
            "{invalid:?}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `asof --namespace` reads one namespace as it was at a commit.
    #[test]
    fn run_asof_in_a_namespace() {
        let dir = ns_store("asof-namespace");
        let at = {
            let db = GraphDb::open(&dir).expect("open");
            db.wal_total_commits().expect("commits") - 1
        };
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node(
                "Doc",
                "a3",
                vec![
                    ("id".into(), Value::Str("a3".into())),
                    ("ns".into(), Value::Str("tenant-a".into())),
                ],
            )
            .expect("insert");
        }
        let q = "MATCH (n) RETURN n.id AS id ORDER BY n.id";
        let then = run_asof(&dir, Some(at), None, Some(q), Some("tenant-a")).expect("asof");
        assert!(
            then.contains("id=a1") && then.contains("id=a2"),
            "got {then}"
        );
        assert!(
            !then.contains("id=a3") && !then.contains("id=b1"),
            "a3 did not exist then and b1 is another namespace: {then}"
        );
        let now = run_asof(&dir, Some(at + 1), None, Some(q), Some("tenant-a")).expect("asof");
        assert!(now.contains("id=a3"), "got {now}");
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// `schema apply` takes a v4 roles file — a role bound to namespaces — and
    /// the sidecar it writes says version 4.
    #[test]
    fn schema_apply_accepts_v4_roles_with_namespaces() {
        let dir = tmp("schema-v4");
        {
            let mut db = GraphDb::open(&dir).expect("open");
            db.insert_node(
                "Doc",
                "a1",
                vec![("ns".into(), Value::Str("tenant-a".into()))],
            )
            .expect("insert");
        }
        let file = dir.join("schema.json");
        std::fs::write(
            &file,
            r#"{"roles": [{"name": "a-reader", "labels": ["Doc"], "keys": [],
                 "namespaces": ["tenant-a"]}]}"#,
        )
        .expect("write schema");
        let out = run_schema_apply(&dir, &file).expect("apply");
        assert!(out.contains("a-reader"), "got {out}");
        let sidecar = std::fs::read_to_string(dir.join("roles.json")).expect("roles.json");
        assert!(
            sidecar.contains("\"version\": 4") || sidecar.contains("\"version\":4"),
            "a role with namespaces writes version 4: {sidecar}"
        );
        assert!(sidecar.contains("tenant-a"), "{sidecar}");
        let _ = std::fs::remove_dir_all(&dir);
    }
}