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 import;
pub mod ingest_git;
pub mod install;
pub mod recall;
pub mod serve_bolt;
pub mod structure;

use export::ExportFormat;
#[cfg(test)]
use mushroomdb::restore::holds_a_store;
/// Re-exported where it has always been named: `cli::RestoreOutcome`.
pub use mushroomdb::restore::RestoreOutcome;
use mushroomdb::schema::Schema;
use mushroomdb::{
    default_max_edges, is_write_query, valid_namespace, AlgoDir, BackupReport, DegreeConfig,
    Explanation, GraphDb, IngestOptions, LouvainConfig, OpenOptions, PageRankConfig, Predicate,
    RealFs, ResultSet, RuleDef, RuleSuggestion, SharedDb, SnapshotOptions, Stats, Value, WccConfig,
    WriteGuard, NS_MAX_LEN,
};
use std::collections::BTreeMap;
use std::fmt::Write as _;
use std::net::SocketAddr;
use std::num::NonZeroU64;
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<(), mushroomdb::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<(), mushroomdb::GraphError> {
    snapshot_automatically(&mut db.write_with_wait(SNAPSHOT_LOCK_WAIT)?)
}

/// Set, to anything, in the environment of a `mushroomdb mcp` that must not
/// snapshot the store it opens.
///
/// `mushroomdb doctor` sets it on the server it starts to see whether it
/// answers: a diagnostic does not rewrite what it diagnoses, and doctor kills
/// that server two seconds after closing its stdin, which is no time to be
/// halfway through a snapshot.
pub const NO_AUTO_SNAPSHOT_ENV: &str = "MUSHROOMDB_NO_AUTO_SNAPSHOT";

/// Take the automatic snapshot if the store's WAL has grown past
/// [`SNAPSHOT_WAL_BYTES`](crate::ingest_git::SNAPSHOT_WAL_BYTES).
///
/// For a process that holds a store open without a timer to snapshot on:
/// `mushroomdb mcp` calls this when it has opened the store and again when
/// its session ends. `wal.bin` is exactly what the next open replays, so its
/// length is the measure.
///
/// Never fails its caller, for the reason a tick of `--snapshot-every` does
/// not: a snapshot shortens the next open and nothing else, and what was
/// written is durable either way. `Busy` — another process holds the write
/// lock — is skipped without a word; anything else is reported on stderr.
/// [`NO_AUTO_SNAPSHOT_ENV`] turns it off.
pub fn snapshot_if_wal_grew(db: &SharedDb, db_dir: &Path) {
    if std::env::var_os(NO_AUTO_SNAPSHOT_ENV).is_some_and(|v| !v.is_empty()) {
        return;
    }
    let grown = std::fs::metadata(db_dir.join("wal.bin"))
        .is_ok_and(|m| m.len() > crate::ingest_git::SNAPSHOT_WAL_BYTES);
    if !grown {
        return;
    }
    match snapshot_shared(db) {
        Ok(()) | Err(mushroomdb::GraphError::Busy { .. }) => {}
        Err(e) => eprintln!("snapshot skipped: {e}"),
    }
}

/// 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,
    /// prints nothing and exits 0. `mushroomdb doctor` lists any that remain
    /// and `install` removes them. Never in [`usage`].
    RetiredHook,
    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>,
        /// `--shutdown-timeout-secs`: the whole graceful shutdown (drain,
        /// snapshot, backup flush, lease release) must finish within this.
        /// Default [`DEFAULT_SHUTDOWN_TIMEOUT_SECS`].
        shutdown_timeout: 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>,
        /// The `--bolt-*` flags; resolved against the environment by
        /// [`serve_bolt::resolve_bolt_settings`].
        bolt: serve_bolt::BoltFlags,
        /// `--tx-max`, `--tx-timeout-secs`, `--tx-max-lifetime-secs`: the HTTP
        /// transaction limits; resolved against the environment by
        /// [`resolve_tx_limits`].
        tx: TxFlags,
    },
    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 the Cypher statement from this file instead of `--query`.
        file: Option<PathBuf>,
        /// Read the Cypher statement from standard input instead of `--query`.
        stdin: bool,
        /// 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,
        /// Read the statement from this file (`--file`), instead of `cypher`.
        file: Option<PathBuf>,
        /// Read the statement from standard input (`--stdin`), instead of `cypher`.
        stdin: bool,
        /// 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,
    },
    /// List the declared indexes (`schema indexes <db-dir>`) or unique
    /// constraints (`schema constraints <db-dir>`) as a table.
    SchemaList {
        db_dir: PathBuf,
        what: SchemaListing,
    },
    /// Migrate an old-format snapshot to the current version and keep `.bak`.
    Migrate {
        db_dir: PathBuf,
    },
    /// Issue an access key: print a new random token once, store only its hash.
    KeysAdd {
        db_dir: PathBuf,
        name: String,
        tier: String,
        role: Option<String>,
    },
    /// Print the `keys.json` hash of the token read from standard input.
    KeysHash,
    /// 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,
    },
    /// Restore a store from a continuous backup in S3-compatible storage
    /// (`restore --from s3://bucket/prefix`). See [`run_restore_s3`].
    RestoreS3 {
        /// `s3://bucket/prefix`, `r2://…` or `minio://…`.
        from: String,
        /// `--at`: a commit number or an RFC 3339 timestamp; `None` restores
        /// the newest restore point.
        at: Option<String>,
        /// `--into`; `None` is [`DEFAULT_RESTORE_INTO`].
        into: Option<PathBuf>,
    },
    /// Export all nodes, edges, and rules to a destination directory.
    Export {
        db_dir: PathBuf,
        dest: PathBuf,
        format: ExportFormat,
    },
    /// Bulk-load CSV and JSONL files.
    Import {
        db_dir: PathBuf,
        inputs: import::ImportInputs,
        opts: import::ImportOptions,
    },
    /// 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<mushroomdb::GraphError> for CliError {
    fn from(e: mushroomdb::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 query <db-dir> [--query \"MATCH ...\"|--file <path>|--stdin] [--role <name>] [--namespace <ns>] <cypher…>
                     --file reads the statement from a file and --stdin from standard input,
                     so a quote or a backslash needs no shell escaping; a trailing newline
                     and one trailing `;` are dropped, and the statement comes from only one of the three places
                     --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 asof <db-dir> --commit N|--at <date> [--query \"MATCH ...\"|--file <path>|--stdin] [--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 stats <db-dir>
  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 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 verify <db-dir>       validate CRC32 integrity of every snapshot section
  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 keys add <db-dir> <name> --tier admin|write|read [--role <name>]
                     issues an access key: prints a new random token once and stores only its
                     blake3 hash in <db-dir>/keys.json (created private, mode 0600); --tier read
                     needs --role, and reads that role's view without ever writing
  mushroomdb keys hash
                     reads a token (at least 32 bytes) from stdin and prints the hash keys.json holds
  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 schema indexes <db-dir>
  mushroomdb schema constraints <db-dir>
  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 builds the index at each open until the next snapshot carries it
  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 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 restore --from s3://bucket/prefix [--at <commit | timestamp>] [--into <dir>]
                     restores a continuous backup (MUSHROOMDB_BACKUP_S3_URL) into <dir>, default ./restored,
                     which must be new or empty; every object is checked against its manifest first
                     --at N restores the store as it was right after commit N (as asof --commit counts);
                     --at 2026-10-01T12:00:00Z the last commit at or before that instant, from the
                     backup's commit times; an instant past the newest recorded time is refused
                     when later commits have none
                     the restored store starts a new backup lineage: back it up to a new prefix
                     r2:// and minio:// need AWS_ENDPOINT_URL; credentials come from the AWS_* variables
  mushroomdb export <db-dir> <dest> --format jsonl|csv|cypher|parquet|graphml [--single-file]   export all data
                                      --single-file (jsonl only) writes one graph.jsonl of typed node, edge and rule lines
                                      csv writes nodes.csv and relationships.csv (plus rules.jsonl when the store has rules);
                                      cypher writes graph.cypher, one statement per line; both load back with `mushroomdb import` / `mushroomdb query`
                                      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 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>] [--shutdown-timeout-secs <secs>] [--restore-from <dir>] [--tls-cert <pem> --tls-key <pem>]
                   [--bolt-port <port>] [--bolt-tls | --bolt-plaintext] [--bolt-advertised-host <host>] [--bolt-advertised-port <port>]
                   [--bolt-database <name>] [--bolt-max-connections <n>] [--bolt-tx-timeout-secs <secs>] [--bookmark-timeout-secs <secs>]
                   [--tx-max <n>] [--tx-timeout-secs <secs>] [--tx-max-lifetime-secs <secs>]
                     --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
                     MUSHROOMDB_RESTORE_FROM_S3=s3://bucket/prefix restores the newest backup there into an
                     empty <db-dir> before the port is bound, as a new lineage that keeps backing up beside
                     it; a prefix with no backup starts an empty store, any other failure exits 1
                     --tls-cert and --tls-key go together and need a build with the `tls` feature
                     --bolt-port (or MUSHROOMDB_BOLT_PORT) also serves Bolt on --addr's IP; off when unset
                     Bolt is TLS when --tls-cert is given, unless --bolt-plaintext (MUSHROOMDB_BOLT_TLS=on|off)
                     --bolt-advertised-host/-port (MUSHROOMDB_BOLT_ADVERTISED_HOST/_PORT) are the address ROUTE
                     hands drivers; a wildcard --addr needs the host. --bolt-max-connections (default 256,
                     MUSHROOMDB_BOLT_MAX_CONNECTIONS); --bolt-database names the store (default mushroomdb);
                     --bolt-tx-timeout-secs (default 30) ends an idle transaction; --bookmark-timeout-secs
                     (default 5) bounds a bookmark wait
                     --tx-max (default 100) open HTTP transactions at once; --tx-timeout-secs (default 30)
                     ends an idle one and --tx-max-lifetime-secs (default 300) any one; each flag wins over
                     MUSHROOMDB_TX_MAX, MUSHROOMDB_TX_TIMEOUT_SECS and MUSHROOMDB_TX_MAX_LIFETIME_SECS
  mushroomdb import [--nodes <file>] [--relationships <file>] [--db <dir>] [--delimiter <c>]
                    [--nodes-jsonl <file|->] [--edges-jsonl <file|->] [--rules-jsonl <file|->] [--jsonl <file|->]
                    [--array-delimiter <c>] [--ignore-errors] [--batch-size N]
                                   bulk-load CSV or JSONL files into <dir> (default ./mushroomdb-data), then write one snapshot
                                   the JSONL files are the ones `export --format jsonl` writes; --jsonl reads the one file
                                   `export --format jsonl --single-file` writes (typed node, edge and rule lines, nodes before the edges that use them);
                                   `-` reads one of them from standard input
                                   nodes need an :ID and a :LABEL column, relationships :START_ID, :END_ID and :TYPE;
                                   other columns are properties, typed by a suffix: age:int weight:float active:boolean tags:string[]
                                   one label per node: Person;Employee is refused; relationship properties are not supported
                                   rules the store already has fire per batch; import first, then declare rules, to backfill once
                                   an error stops the import at its file and line, batches already committed stay; --ignore-errors skips bad rows
  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 demo <db-dir>
  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 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 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 --version
  mushroomdb --help

Default serve address is 127.0.0.1:8080. Non-loopback --addr requires --token, MUSHROOMDB_TOKEN, or an `admin` key in `<db-dir>/keys.json` (read and write keys alone do not count).
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 }),
        "keys" => parse_keys(&args[1..]),
        "backup" => parse_backup(&args[1..]),
        "restore" => parse_restore(&args[1..]),
        "export" => parse_export(&args[1..]),
        "import" => parse_import(&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" | "intercept" | "impact-hook" | "enrich" | "sync" => Ok(Command::RetiredHook),
        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 shutdown_timeout = Duration::from_secs(DEFAULT_SHUTDOWN_TIMEOUT_SECS);
    let mut restore_from: Option<PathBuf> = None;
    let mut tls_cert: Option<PathBuf> = None;
    let mut tls_key: Option<PathBuf> = None;
    let mut bolt = serve_bolt::BoltFlags::default();
    let mut tx = TxFlags::default();
    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 == "--shutdown-timeout-secs" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --shutdown-timeout-secs".to_string())?;
            shutdown_timeout = parse_shutdown_timeout(val)?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--shutdown-timeout-secs=") {
            shutdown_timeout = parse_shutdown_timeout(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 let Some(used) = parse_bolt_flag(args, i, &mut bolt)? {
            i += used;
        } else if let Some(used) = parse_tx_flag(args, i, &mut tx)? {
            i += used;
        } 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,
        shutdown_timeout,
        restore_from,
        tls_cert,
        tls_key,
        bolt,
        tx,
    })
}

/// The HTTP transaction limits as given on `serve`'s command line; `None` is
/// "not given", which leaves the environment variable (or the default) in
/// force. Non-zero integers, not `Duration`s, keep [`Command::Serve`] small.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct TxFlags {
    /// `--tx-max <n>` (`MUSHROOMDB_TX_MAX`).
    pub max_open: Option<NonZeroU64>,
    /// `--tx-timeout-secs <secs>` (`MUSHROOMDB_TX_TIMEOUT_SECS`).
    pub idle_timeout_secs: Option<NonZeroU64>,
    /// `--tx-max-lifetime-secs <secs>` (`MUSHROOMDB_TX_MAX_LIFETIME_SECS`).
    pub max_lifetime_secs: Option<NonZeroU64>,
}

/// The HTTP transaction limits `serve` runs with: each flag given wins over
/// its environment variable, read through `env`; a limit given by neither
/// keeps its default.
pub fn resolve_tx_limits(
    flags: &TxFlags,
    env: &dyn Fn(&str) -> Option<String>,
) -> server::TxLimits {
    let mut limits = server::TxLimits::from_lookup(|name| match name {
        // A flag given: its variable is not read, so an invalid value there
        // does not warn about a limit that does not apply.
        server::TX_MAX_ENV if flags.max_open.is_some() => None,
        server::TX_TIMEOUT_ENV if flags.idle_timeout_secs.is_some() => None,
        server::TX_MAX_LIFETIME_ENV if flags.max_lifetime_secs.is_some() => None,
        _ => env(name),
    });
    if let Some(n) = flags.max_open {
        limits.max_open = usize::try_from(n.get()).unwrap_or(usize::MAX);
    }
    if let Some(secs) = flags.idle_timeout_secs {
        limits.idle_timeout = Duration::from_secs(secs.get());
    }
    if let Some(secs) = flags.max_lifetime_secs {
        limits.max_lifetime = Duration::from_secs(secs.get());
    }
    limits
}

/// Reads one `--tx-*` flag at `args[i]` into `tx`, as `--flag value` or
/// `--flag=value`. Returns how many arguments it used, or `None` when
/// `args[i]` is not one of them. Every value must be a positive integer.
fn parse_tx_flag(args: &[&str], i: usize, tx: &mut TxFlags) -> Result<Option<usize>, String> {
    const VALUED: [&str; 3] = ["--tx-max", "--tx-timeout-secs", "--tx-max-lifetime-secs"];
    let arg = args[i];
    let (flag, value, used) = match arg.split_once('=') {
        Some((flag, value)) if VALUED.contains(&flag) => (flag, value, 1),
        _ if VALUED.contains(&arg) => {
            let value = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| format!("missing value for {arg}"))?;
            (arg, value, 2)
        }
        _ => return Ok(None),
    };
    let n = value
        .parse::<NonZeroU64>()
        .map_err(|_| format!("{flag} must be a positive integer, got: {value}"))?;
    match flag {
        "--tx-max" => tx.max_open = Some(n),
        "--tx-timeout-secs" => tx.idle_timeout_secs = Some(n),
        _ => tx.max_lifetime_secs = Some(n),
    }
    Ok(Some(used))
}

/// Reads one `--bolt-*` flag (or `--bookmark-timeout-secs`) at `args[i]` into
/// `bolt`, as `--flag value` or `--flag=value`. Returns how many arguments it
/// used, or `None` when `args[i]` is not one of them.
fn parse_bolt_flag(
    args: &[&str],
    i: usize,
    bolt: &mut serve_bolt::BoltFlags,
) -> Result<Option<usize>, String> {
    let arg = args[i];
    match arg {
        "--bolt-tls" | "--bolt-plaintext" => {
            let tls = arg == "--bolt-tls";
            if bolt.tls == Some(!tls) {
                return Err("cannot combine --bolt-tls and --bolt-plaintext".into());
            }
            bolt.tls = Some(tls);
            return Ok(Some(1));
        }
        _ => {}
    }
    const VALUED: [&str; 7] = [
        "--bolt-port",
        "--bolt-advertised-host",
        "--bolt-advertised-port",
        "--bolt-database",
        "--bolt-max-connections",
        "--bolt-tx-timeout-secs",
        "--bookmark-timeout-secs",
    ];
    let (flag, value, used) = match arg.split_once('=') {
        Some((flag, value)) if VALUED.contains(&flag) => (flag, value, 1),
        _ if VALUED.contains(&arg) => {
            let value = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| format!("missing value for {arg}"))?;
            (arg, value, 2)
        }
        _ => return Ok(None),
    };
    let invalid = || format!("invalid {flag}: {value}");
    let port = || value.parse::<u16>().map_err(|_| invalid());
    let seconds = || match value.parse::<u64>() {
        Ok(0) | Err(_) => Err(format!("{flag} must be a positive number of seconds")),
        Ok(secs) => Ok(Duration::from_secs(secs)),
    };
    match flag {
        "--bolt-port" => bolt.port = Some(port()?),
        "--bolt-advertised-host" => bolt.advertised_host = Some(value.to_string()),
        "--bolt-advertised-port" => bolt.advertised_port = Some(port()?),
        "--bolt-database" => bolt.database = Some(value.to_string()),
        "--bolt-max-connections" => {
            bolt.max_connections = Some(value.parse().map_err(|_| invalid())?)
        }
        "--bolt-tx-timeout-secs" => bolt.tx_timeout = Some(seconds()?),
        _ => bolt.bookmark_timeout = Some(seconds()?),
    }
    Ok(Some(used))
}

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()))
}

/// Default for `serve --shutdown-timeout-secs`: drain up to 25 s, snapshot in
/// what is left of the budget (8 s with the default backup bounds), then flush
/// the backup and release the lease within 22 s.
pub const DEFAULT_SHUTDOWN_TIMEOUT_SECS: u64 = 55;
/// The drain phase's share at the default timeout, and its cap at any timeout.
pub const SHUTDOWN_DRAIN_SECS: u64 = 25;
/// The snapshot phase's nominal share, used to scale the phases down when the
/// timeout is shorter than drain + snapshot + backup.
const SHUTDOWN_SNAPSHOT_NOMINAL: Duration = Duration::from_secs(8);
/// A backup flush keeps at least this much of any timeout, so the lease
/// release is still attempted under a tight one.
const SHUTDOWN_BACKUP_FLOOR: Duration = Duration::from_secs(2);

/// How a shutdown's time is divided: the three phases, in order, and the
/// most each may take. They sum to at most the timeout.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ShutdownPlan {
    /// Longest the drain of in-flight requests may take.
    pub drain: Duration,
    /// Snapshot's share; unused drain time also goes to it at run time.
    pub snapshot: Duration,
    /// Backup flush and lease release. Zero without backup.
    pub backup: Duration,
}

/// Divide `total` between the phases. `backup_bound` is the backup's own
/// shutdown bound (`BackupConfig::shutdown_budget`), zero without backup.
///
/// With room (`total` at least drain cap + 8 s + `backup_bound`) the drain gets
/// its cap, the backup its bound, and the snapshot the rest. With less, all
/// three scale down in proportion, except that the backup keeps a floor
/// ([`SHUTDOWN_BACKUP_FLOOR`], never more than its bound) and the drain gives
/// way first: it is the phase a stuck client can stretch.
pub fn shutdown_plan(total: Duration, backup_bound: Duration) -> ShutdownPlan {
    let cap = Duration::from_secs(SHUTDOWN_DRAIN_SECS);
    let nominal = cap + SHUTDOWN_SNAPSHOT_NOMINAL + backup_bound;
    if total >= nominal {
        return ShutdownPlan {
            drain: cap,
            snapshot: total - cap - backup_bound,
            backup: backup_bound,
        };
    }
    let scaled = |d: Duration| d.mul_f64(total.as_secs_f64() / nominal.as_secs_f64());
    let backup = if backup_bound.is_zero() {
        Duration::ZERO
    } else {
        scaled(backup_bound)
            .max(SHUTDOWN_BACKUP_FLOOR.min(backup_bound))
            .min(total)
    };
    let drain = scaled(cap).min(total - backup);
    ShutdownPlan {
        drain,
        snapshot: total - backup - drain,
        backup,
    }
}

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

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 file: Option<PathBuf> = None;
    let mut stdin = false;
    let mut namespace: Option<String> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--file" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --file".to_string())?;
            file = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--file=") {
            file = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--stdin" {
            stdin = true;
            i += 1;
        } else 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(),
            );
        }
        _ => {}
    }
    check_statement_sources("asof", query.is_some(), file.is_some(), stdin)?;
    Ok(Command::AsOf {
        db_dir,
        commit,
        at,
        query,
        file,
        stdin,
        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),
    }
}

/// A statement comes from one place: the command line, `--file` or `--stdin`.
/// Two sources is a refusal, never a precedence rule.
fn check_statement_sources(
    command: &str,
    inline: bool,
    file: bool,
    stdin: bool,
) -> Result<(), String> {
    if [inline, file, stdin].iter().filter(|given| **given).count() > 1 {
        return Err(format!(
            "{command}: give the Cypher statement one way — on the command line, \
             with --file <path>, or with --stdin — not several"
        ));
    }
    Ok(())
}

/// The statement a `query` or `asof` was given: `inline` as it came, or the
/// text of `--file` or of standard input. A file or pipe ends in a newline the
/// editor or `echo` put there, which is not part of the statement, so trailing
/// whitespace is dropped.
pub fn resolve_statement(
    inline: Option<&str>,
    file: Option<&Path>,
    stdin: bool,
    input: &mut dyn std::io::Read,
) -> Result<Option<String>, CliError> {
    let text = match (file, stdin) {
        (Some(path), _) => std::fs::read_to_string(path)
            .map_err(|e| CliError(format!("cannot read --file {}: {e}", path.display())))?,
        (None, true) => {
            let mut text = String::new();
            input
                .read_to_string(&mut text)
                .map_err(|e| CliError(format!("cannot read the statement from stdin: {e}")))?;
            text
        }
        (None, false) => match inline {
            Some(statement) => return Ok(Some(tidy_statement(statement).to_string())),
            None => return Ok(None),
        },
    };
    let statement = tidy_statement(&text);
    if statement.trim().is_empty() {
        return Err(CliError("the Cypher statement is empty".to_string()));
    }
    Ok(Some(statement.to_string()))
}

/// One leading byte-order mark, trailing whitespace and one trailing `;` are
/// not part of the statement: an editor, a pipe and a pasted line put them
/// there, and the lexer has no use for any of them.
fn tidy_statement(text: &str) -> &str {
    let text = text.strip_prefix('\u{feff}').unwrap_or(text).trim_end();
    text.strip_suffix(';').unwrap_or(text).trim_end()
}

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 file: Option<PathBuf> = None;
    let mut stdin = false;
    let mut cypher_parts: Vec<&str> = Vec::new();
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--file" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --file".to_string())?;
            file = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--file=") {
            file = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--stdin" {
            stdin = true;
            i += 1;
        } else 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())?;
    check_statement_sources(
        "query",
        query_flag.is_some() || !cypher_parts.is_empty(),
        file.is_some(),
        stdin,
    )?;
    let cypher = if file.is_some() || stdin {
        String::new()
    } else 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,
        file,
        stdin,
        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 = mushroomdb::SNAPSHOT_VERSION;
    let from_ver = mushroomdb::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,
        mushroomdb::OpenOptions {
            auto_migrate: false,
            ..Default::default()
        },
    )?;
    db.snapshot()?;
    // The version the snapshot was written at: a WAL-only store whose WAL
    // holds a 0.10 record is written at V11, not at the default.
    let to_ver = mushroomdb::snapshot_version_at(db_dir)?.unwrap_or(current);

    let msg = match from_ver {
        Some(ver) => format!("migrated V{ver} -> V{to_ver}\n"),
        None => format!("migrated WAL-only -> V{to_ver}\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 = mushroomdb::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 | indexes | constraints".to_string());
    }
    match args[0] {
        "apply" => parse_schema_apply(&args[1..]),
        "indexes" => {
            parse_one_dir("schema indexes", &args[1..]).map(|db_dir| Command::SchemaList {
                db_dir,
                what: SchemaListing::Indexes,
            })
        }
        "constraints" => {
            parse_one_dir("schema constraints", &args[1..]).map(|db_dir| Command::SchemaList {
                db_dir,
                what: SchemaListing::Constraints,
            })
        }
        other => Err(format!(
            "unknown schema subcommand: {other}; expected apply | indexes | constraints"
        )),
    }
}

/// What `mushroomdb schema indexes|constraints` lists.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum SchemaListing {
    Indexes,
    Constraints,
}

/// `mushroomdb schema indexes|constraints`: the rows `SHOW INDEXES` /
/// `SHOW CONSTRAINTS` give, as a table. The store opens read-only, so this
/// writes nothing and never waits on a writer, a server included.
pub fn run_schema_list(db_dir: &Path, what: SchemaListing) -> Result<String, CliError> {
    let db = GraphDb::open_with_options(
        db_dir,
        OpenOptions {
            auto_migrate: false,
            repair_wal: false,
            read_only: true,
        },
    )?;
    let statement = match what {
        SchemaListing::Indexes => "SHOW INDEXES",
        SchemaListing::Constraints => "SHOW CONSTRAINTS",
    };
    let rs = db.query(statement, &BTreeMap::new())?;
    let columns = rs.columns();
    let rows: Vec<Vec<String>> = (0..rs.len())
        .map(|i| {
            columns
                .iter()
                .map(|c| fmt_cell_plain(rs.get(i, c)))
                .collect()
        })
        .collect();
    let widths: Vec<usize> = columns
        .iter()
        .enumerate()
        .map(|(n, c)| {
            rows.iter()
                .map(|r| r[n].chars().count())
                .chain([c.chars().count()])
                .max()
                .unwrap_or(0)
        })
        .collect();
    let line = |cells: &[String]| {
        let padded: Vec<String> = cells
            .iter()
            .zip(&widths)
            .map(|(c, w)| format!("{c:<w$}"))
            .collect();
        format!("{}\n", padded.join(" | ").trim_end())
    };
    let mut out = line(columns);
    for row in &rows {
        out.push_str(&line(row));
    }
    Ok(out)
}

/// A cell as text, without the quoting [`fmt_cell`] puts around a string.
fn fmt_cell_plain(cell: Option<&Value>) -> String {
    match cell {
        Some(Value::Str(s)) => s.clone(),
        other => fmt_cell(other),
    }
}

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 `mushroomdb::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(&mushroomdb::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 `mushroomdb::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 mushroomdb::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 = mushroomdb::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 })
}

/// `restore --from <url> [--at <commit | timestamp>] [--into <dir>]`.
fn parse_restore(args: &[&str]) -> Result<Command, String> {
    let (mut from, mut at, mut into) = (None, None, None);
    let mut i = 0;
    while i < args.len() {
        if let Some(v) = flag_value(args, &mut i, "--from")? {
            from = Some(v.to_string());
        } else if let Some(v) = flag_value(args, &mut i, "--at")? {
            // Checked here so a typo fails before anything is downloaded.
            backup::RestoreTarget::parse(v)?;
            at = Some(v.to_string());
        } else if let Some(v) = flag_value(args, &mut i, "--into")? {
            into = Some(PathBuf::from(v));
        } else if args[i].starts_with('-') {
            return Err(format!("unexpected flag: {}", args[i]));
        } else {
            return Err(format!(
                "unexpected argument: {} (restore takes --from, --at and --into)",
                args[i]
            ));
        }
    }
    let from = from
        .filter(|f| !f.trim().is_empty())
        .ok_or_else(|| "restore requires --from s3://bucket/prefix".to_string())?;
    Ok(Command::RestoreS3 { from, at, into })
}

/// Take the value of `--flag value` or `--flag=value` at `args[*i]`.
fn flag_value<'a>(args: &[&'a str], i: &mut usize, flag: &str) -> Result<Option<&'a str>, String> {
    let a = args[*i];
    if a == flag {
        let val = args
            .get(*i + 1)
            .copied()
            .ok_or_else(|| format!("missing value for {flag}"))?;
        *i += 2;
        Ok(Some(val))
    } else if let Some(val) = a.strip_prefix(flag).and_then(|r| r.strip_prefix('=')) {
        *i += 1;
        Ok(Some(val))
    } else {
        Ok(None)
    }
}

/// A one-character option value; `tab` and `\t` name the tab.
fn single_char(flag: &str, val: &str) -> Result<char, String> {
    let val = if val == "tab" || val == "\\t" {
        "\t"
    } else {
        val
    };
    let mut chars = val.chars();
    match (chars.next(), chars.next()) {
        (Some(c), None) => Ok(c),
        _ => Err(format!("{flag} takes one character, got '{val}'")),
    }
}

fn parse_import(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = PathBuf::from("mushroomdb-data");
    let mut inputs = import::ImportInputs::default();
    let mut opts = import::ImportOptions::default();
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--ignore-errors" {
            opts.ignore_errors = true;
            i += 1;
        } else if let Some(v) = flag_value(args, &mut i, "--nodes")? {
            inputs.nodes_csv = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--relationships")? {
            inputs.rels_csv = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--nodes-jsonl")? {
            inputs.nodes_jsonl = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--edges-jsonl")? {
            inputs.edges_jsonl = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--jsonl")? {
            inputs.stream_jsonl = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--rules-jsonl")? {
            inputs.rules_jsonl = Some(PathBuf::from(v));
        } else if let Some(v) = flag_value(args, &mut i, "--db")? {
            db_dir = PathBuf::from(v);
        } else if let Some(v) = flag_value(args, &mut i, "--delimiter")? {
            let c = single_char("--delimiter", v)?;
            opts.delimiter = u8::try_from(u32::from(c))
                .ok()
                .filter(u8::is_ascii)
                .ok_or_else(|| format!("--delimiter must be an ASCII character, got '{v}'"))?;
        } else if let Some(v) = flag_value(args, &mut i, "--array-delimiter")? {
            opts.array_delimiter = single_char("--array-delimiter", v)?;
        } else if let Some(v) = flag_value(args, &mut i, "--batch-size")? {
            opts.batch_size = v
                .parse::<usize>()
                .ok()
                .filter(|n| *n > 0)
                .ok_or_else(|| format!("--batch-size must be a positive integer, got '{v}'"))?;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else {
            return Err(format!("unexpected argument: {a}"));
        }
    }
    if inputs == import::ImportInputs::default() {
        return Err(
            "import needs --nodes, --relationships, --nodes-jsonl, --edges-jsonl, --rules-jsonl or --jsonl"
                .to_string(),
        );
    }
    Ok(Command::Import {
        db_dir,
        inputs,
        opts,
    })
}

fn parse_export(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut dest = None;
    let mut format = ExportFormat::Jsonl;
    let mut single_file = false;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--single-file" {
            single_file = true;
            i += 1;
        } else 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, csv, cypher, 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, csv, cypher, 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())?;
    if single_file {
        if format != ExportFormat::Jsonl {
            return Err("--single-file goes with --format jsonl".to_string());
        }
        format = ExportFormat::JsonlSingle;
    }
    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)?)
}

/// Where `restore` writes when `--into` is not given.
pub const DEFAULT_RESTORE_INTO: &str = "restored";

/// `mushroomdb restore --from s3://…`: download, verify and install a restore
/// point into `into` (default [`DEFAULT_RESTORE_INTO`]) with
/// [`backup::restore_from_s3`], and say what was restored.
pub fn run_restore_s3(
    from: &str,
    at: Option<&str>,
    into: Option<&Path>,
) -> Result<String, CliError> {
    let target = at
        .map(backup::RestoreTarget::parse)
        .transpose()
        .map_err(CliError)?;
    let into = into.map_or_else(|| PathBuf::from(DEFAULT_RESTORE_INTO), Path::to_path_buf);
    let rt = tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()?;
    let out = rt
        .block_on(backup::restore_from_s3(from, target, &into))
        .map_err(|e| CliError(format!("restore from {from} failed: {e}")))?;
    Ok(format_restore_s3(from, target, &out))
}

/// Format a [`backup::RestoreS3Outcome`] for display.
pub fn format_restore_s3(
    from: &str,
    at: Option<backup::RestoreTarget>,
    out: &backup::RestoreS3Outcome,
) -> String {
    let mut s = String::new();
    let point = at.map_or_else(|| "the newest restore point".to_string(), |t| t.to_string());
    writeln!(s, "restored {from} at {point} into {}", out.into.display()).unwrap();
    writeln!(s, "  manifest: {}", out.manifest).unwrap();
    writeln!(s, "  commits: {}", out.commits).unwrap();
    writeln!(s, "  files: {}", out.files.join(", ")).unwrap();
    writeln!(s, "  segments: {}", out.segments).unwrap();
    writeln!(s, "  bytes: {}", out.bytes).unwrap();
    for path in &out.swept {
        writeln!(
            s,
            "  removed a killed restore's leftover: {}",
            path.display()
        )
        .unwrap();
    }
    if let Some(note) = &out.install_note {
        writeln!(s, "  warning: {note}").unwrap();
    }
    if let Some(note) = &out.tail_note {
        writeln!(s, "  warning: {note}").unwrap();
    }
    writeln!(
        s,
        "  lineage: {} (new; restored from {})",
        out.lineage, out.source_lineage
    )
    .unwrap();
    writeln!(
        s,
        "back this store up to a new prefix: the one it came from belongs to the source"
    )
    .unwrap();
    s
}

/// [`mushroomdb::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 [`mushroomdb::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> {
    mushroomdb::restore::restore_if_empty(db_dir, from).map_err(|e| match e {
        mushroomdb::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
}

/// Bulk-load CSV and JSONL files into `db_dir`; `mushroomdb import`.
pub fn run_import(
    db_dir: &Path,
    inputs: &import::ImportInputs,
    opts: &import::ImportOptions,
) -> Result<import::ImportReport, CliError> {
    import::run_import_inputs(db_dir, inputs, opts, &mut |_| {})
}

/// The warning line for NaN and infinite floats, which JSON and CSV cannot
/// carry; empty when there are none.
fn non_finite_note(stats: &export::ExportStats) -> String {
    match stats.non_finite_floats {
        0 => String::new(),
        n => format!(
            "warning: {n} NaN or infinite float values cannot be written in this format and were left out\n"
        ),
    }
}

fn export_summary(
    dest: &Path,
    format: &ExportFormat,
    stats: &export::ExportStats,
    rule_count: usize,
    extra: &str,
) -> String {
    let manual = stats.edges - stats.derived_edges;
    let mut out = format!(
        "exported to {} (format={}): {} nodes, {} edges, {} rules\n",
        dest.display(),
        format.name(),
        stats.nodes,
        manual,
        rule_count
    );
    if stats.derived_edges > 0 {
        let _ = writeln!(
            out,
            "{} derived edges are not written; they belong to their rules",
            stats.derived_edges
        );
    }
    out.push_str(extra);
    out
}

/// Export all data from `db_dir` to `dest` in `format`.
///
/// JSONL, CSV and Cypher stream: the store is read node by node and the files
/// are written as it goes. Parquet and GraphML build their whole file in
/// memory, so they read the store whole.
pub fn run_export(db_dir: &Path, dest: &Path, format: &ExportFormat) -> Result<String, CliError> {
    let db = GraphDb::open(db_dir)?;
    let mut rules = db.rules();
    rules.sort_by(|a, b| a.name.cmp(&b.name));
    let rule_count = rules.len();
    match format {
        ExportFormat::Jsonl => {
            let stats = export::write_jsonl(&db, &rules, dest)?;
            // JSONL has always counted every edge, derived included.
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges, {} rules\n{}",
                dest.display(),
                format.name(),
                stats.nodes,
                stats.edges,
                rule_count,
                non_finite_note(&stats)
            ))
        }
        ExportFormat::JsonlSingle => {
            let stats = export::write_jsonl_single(&db, &rules, dest)?;
            Ok(format!(
                "exported to {} (format={}, single file): {} nodes, {} edges, {} rules\n{}",
                dest.display(),
                format.name(),
                stats.nodes,
                stats.edges,
                rule_count,
                non_finite_note(&stats)
            ))
        }
        ExportFormat::Csv => {
            let stats = export::write_csv(&db, &rules, dest)?;
            let note = non_finite_note(&stats);
            Ok(export_summary(dest, format, &stats, rule_count, &note))
        }
        ExportFormat::Cypher => {
            let (stats, loss) = export::write_cypher(&db, dest)?;
            let mut extra = String::new();
            if rule_count > 0 {
                extra.push_str(&format!(
                    "{rule_count} rules are not written: Cypher has no statement for one\n"
                ));
            }
            if loss.ids_added > 0 {
                extra.push_str(&format!(
                    "warning: {} nodes had no id property and gain one, equal to their key, when the file is run\n",
                    loss.ids_added
                ));
            }
            if loss.typed_properties + loss.unwritable_numbers + loss.differing_ids > 0 {
                extra.push_str(&format!(
                    "warning: not written to the Cypher file: {} properties of the date, time, duration or binary types, \
                     {} NaN, infinite or -2^63 numbers or names with a line break, {} nodes whose id property is not their key\n",
                    loss.typed_properties, loss.unwritable_numbers, loss.differing_ids
                ));
            }
            Ok(export_summary(dest, format, &stats, 0, &extra))
        }
        ExportFormat::Parquet => {
            let nodes = db.all_nodes_for_export();
            let edges = db.all_edges_for_export();
            export::write_parquet(&nodes, &edges, &rules, dest)?;
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges, {} rules\n",
                dest.display(),
                format.name(),
                nodes.len(),
                edges.len(),
                rule_count
            ))
        }
        // GraphML has no rule analogue: only nodes and edges are written.
        ExportFormat::Graphml => {
            let nodes = db.all_nodes_for_export();
            let edges = db.all_edges_for_export();
            let file_path = export::write_graphml(&nodes, &edges, dest)?;
            Ok(format!(
                "exported to {} (format={}): {} nodes, {} edges\n",
                file_path.display(),
                format.name(),
                nodes.len(),
                edges.len(),
            ))
        }
    }
}

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 =
        mushroomdb::memory::brief::brief(&db, &mushroomdb::memory::brief::BriefOptions::default());
    let mut text = mushroomdb::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 == mushroomdb::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.
    //
    // A store with one text field can still be missing the memory defaults:
    // `remember` declares `Note.text` as it writes, so a store that never took
    // them stops having "no text index" at its first note, while the entities
    // already in it stay unsearchable by name. That store gets the second
    // notice, naming the fields. A store with no node under a memory label
    // gets neither: the defaults would index nothing there.
    let store = mushroomdb::digest::sanitize(&install::sh_quote(&db_dir.to_string_lossy()));
    let notice = if db.fulltext_pairs().is_empty() {
        Some(format!(
            "no text index: recall cannot match anything here. \
             Run: mushroomdb schema apply {store} --memory-defaults\n"
        ))
    } else {
        let unindexed = mushroomdb::memory::recall::unindexed_memory_fields(&db);
        (!unindexed.is_empty()).then(|| {
            format!(
                "text index incomplete: recall does not search {}. \
                 Run: mushroomdb schema apply {store} --memory-defaults\n",
                mushroomdb::memory::recall::unindexed_fields_list(&unindexed)
            )
        })
    };
    if let Some(notice) = notice {
        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 = mushroomdb::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.
///
/// [`mushroomdb::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 = mushroomdb::digest::sanitize(&install::detect_mcp_command(None).shell());
    let db = mushroomdb::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,
        mushroomdb::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 = mushroomdb::explain_digest::explain_with_evidence(&db, a, b)?;
    Ok(format!(
        "{}{}",
        mushroomdb::digest::UNTRUSTED_FRAMING,
        mushroomdb::explain_digest::render_explain(a, b, &explained)
    ))
}

/// `why`'s reply as it goes to stdout. A person at a terminal reads the
/// framing line as a warning, so `styled` dims it; the text stays, because an
/// agent may be reading through a terminal too. Piped output is untouched.
pub fn why_for_stdout(out: String, styled: bool) -> String {
    let framing = mushroomdb::digest::UNTRUSTED_FRAMING;
    match out.strip_prefix(framing) {
        Some(rest) if styled => {
            format!("\x1b[2m{}\x1b[0m\n{rest}", framing.trim_end_matches('\n'))
        }
        _ => out,
    }
}

/// 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: &mushroomdb::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: &mushroomdb::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: &mushroomdb::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: &mushroomdb::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_keys(args: &[&str]) -> Result<Command, String> {
    match args.first().copied() {
        Some("hash") if args.len() == 1 => Ok(Command::KeysHash),
        Some("hash") => Err("keys hash takes no arguments; it reads the token from stdin".into()),
        Some("add") => {
            let mut positional: Vec<&str> = Vec::new();
            let mut tier: Option<String> = None;
            let mut role: Option<String> = None;
            let mut i = 1;
            while i < args.len() {
                match args[i] {
                    "--tier" | "--role" => {
                        let flag = args[i];
                        i += 1;
                        let value = args
                            .get(i)
                            .ok_or_else(|| format!("{flag} requires a value"))?;
                        let slot = if flag == "--tier" {
                            &mut tier
                        } else {
                            &mut role
                        };
                        if slot.replace((*value).to_string()).is_some() {
                            return Err(format!("{flag} given twice"));
                        }
                    }
                    flag if flag.starts_with('-') => {
                        return Err(format!("unexpected flag: {flag}"))
                    }
                    other => positional.push(other),
                }
                i += 1;
            }
            let [db_dir, name] = positional.as_slice() else {
                return Err("keys add requires <db-dir> <name>".into());
            };
            let tier = tier.ok_or("keys add requires --tier admin|write|read")?;
            match (tier.as_str(), role.is_some()) {
                ("admin" | "write", false) => {}
                ("admin" | "write", true) => return Err("--role goes only with --tier read".into()),
                ("read", true) => {}
                ("read", false) => return Err("--tier read requires --role <name>".into()),
                _ => return Err("--tier must be admin, write or read".into()),
            }
            Ok(Command::KeysAdd {
                db_dir: PathBuf::from(*db_dir),
                name: (*name).to_string(),
                tier,
                role,
            })
        }
        _ => Err("keys requires a subcommand: add or hash".into()),
    }
}

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(&mushroomdb::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(", "))
        }
        Value::Date(_)
        | Value::LocalDateTime(_)
        | Value::DateTime { .. }
        | Value::Duration { .. }
        | Value::Binary(_) => mushroomdb::v11_display_text(v).unwrap_or_default(),
        // A null list element or map value.
        Value::Null => "null".to_string(),
    }
}

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
    }

    #[test]
    fn why_for_stdout_dims_the_framing_only_when_styled() {
        let framing = mushroomdb::digest::UNTRUSTED_FRAMING;
        let out = format!("{framing}mushroomdb explain — a ↔ b: 0 relationship(s)\n");
        assert_eq!(why_for_stdout(out.clone(), false), out);
        let styled = why_for_stdout(out.clone(), true);
        assert_eq!(
            styled,
            format!(
                "\x1b[2m{}\x1b[0m\nmushroomdb explain — a ↔ b: 0 relationship(s)\n",
                framing.trim_end_matches('\n')
            )
        );
        // The framing text survives styling: strip the escapes and it is back.
        assert_eq!(styled.replace("\x1b[2m", "").replace("\x1b[0m", ""), out);
        // A reply without the framing is never touched.
        assert_eq!(why_for_stdout("plain\n".into(), true), "plain\n");
    }

    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, mushroomdb::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, mushroomdb::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,
                        shutdown_timeout: _,
                        restore_from,
                        tls_cert,
                        tls_key,
                        bolt,
                        tx,
                    }) => {
                        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);
                        assert_eq!(bolt, super::serve_bolt::BoltFlags::default());
                        assert_eq!(tx, super::TxFlags::default());
                    }
                    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,
                        shutdown_timeout: _,
                        restore_from,
                        tls_cert,
                        tls_key,
                        bolt,
                        tx,
                    }) => {
                        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);
                        assert_eq!(bolt, super::serve_bolt::BoltFlags::default());
                        assert_eq!(tx, super::TxFlags::default());
                    }
                    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,
                        shutdown_timeout: _,
                        restore_from,
                        tls_cert,
                        tls_key,
                        bolt,
                        tx,
                    }) => {
                        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);
                        assert_eq!(bolt, super::serve_bolt::BoltFlags::default());
                        assert_eq!(tx, super::TxFlags::default());
                    }
                    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), "{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 the_shutdown_plan_uses_the_nominal_split_with_room_and_scales_without() {
        let s = Duration::from_secs;
        // Default timeout, default backup bound: 25 + 8 + 22.
        let p = shutdown_plan(s(55), s(22));
        assert_eq!((p.drain, p.snapshot, p.backup), (s(25), s(8), s(22)));
        // More room goes to the snapshot.
        let p = shutdown_plan(s(100), s(22));
        assert_eq!((p.drain, p.snapshot, p.backup), (s(25), s(53), s(22)));
        // Less room: every phase shrinks, and the sum never exceeds the timeout.
        for total in [1, 2, 5, 10, 25, 40, 54] {
            for bound in [0, 1, 3, 22, 60] {
                let p = shutdown_plan(s(total), s(bound));
                assert!(
                    p.drain + p.snapshot + p.backup <= s(total),
                    "{total}/{bound}: {p:?}"
                );
                assert!(p.drain <= s(SHUTDOWN_DRAIN_SECS));
                if bound == 0 {
                    assert_eq!(p.backup, Duration::ZERO);
                } else {
                    assert!(p.backup <= s(bound).min(s(total)));
                    assert!(p.backup >= s(2).min(s(bound)).min(s(total)), "{p:?}");
                }
            }
        }
        // 25 s with backup: the backup keeps a real share.
        let p = shutdown_plan(s(25), s(22));
        assert!(p.backup >= s(9) && p.drain >= s(10), "{p:?}");
    }

    #[test]
    fn serve_parses_the_bolt_flags_in_both_spellings() {
        let bolt = |args: &[&str]| match parse_args(args).unwrap() {
            Command::Serve { bolt, .. } => bolt,
            other => panic!("{other:?}"),
        };
        let flags = bolt(&[
            "serve",
            "/tmp/db",
            "--bolt-port",
            "7687",
            "--bolt-tls",
            "--bolt-advertised-host=db.example.test",
            "--bolt-advertised-port",
            "17687",
            "--bolt-database=orders",
            "--bolt-max-connections",
            "8",
            "--bolt-tx-timeout-secs=12",
            "--bookmark-timeout-secs",
            "3",
        ]);
        assert_eq!(
            flags,
            serve_bolt::BoltFlags {
                port: Some(7687),
                tls: Some(true),
                advertised_host: Some("db.example.test".into()),
                advertised_port: Some(17687),
                database: Some("orders".into()),
                max_connections: Some(8),
                tx_timeout: Some(Duration::from_secs(12)),
                bookmark_timeout: Some(Duration::from_secs(3)),
            }
        );
        assert_eq!(
            bolt(&["serve", "/tmp/db", "--bolt-port=0", "--bolt-plaintext"]).tls,
            Some(false)
        );
        for bad in [
            &["serve", "/tmp/db", "--bolt-port", "70000"][..],
            &["serve", "/tmp/db", "--bolt-port"],
            &["serve", "/tmp/db", "--bolt-tls", "--bolt-plaintext"],
            &["serve", "/tmp/db", "--bolt-tx-timeout-secs", "0"],
            &["serve", "/tmp/db", "--bookmark-timeout-secs=x"],
            &["serve", "/tmp/db", "--bolt-max-connections", "-1"],
        ] {
            assert!(parse_args(bad).is_err(), "{bad:?}");
        }
    }

    #[test]
    fn serve_parses_the_tx_flags_in_both_spellings_and_refuses_zero() {
        let tx = |args: &[&str]| match parse_args(args).unwrap() {
            Command::Serve { tx, .. } => tx,
            other => panic!("{other:?}"),
        };
        assert_eq!(tx(&["serve", "/tmp/db"]), TxFlags::default());
        assert_eq!(
            tx(&[
                "serve",
                "/tmp/db",
                "--tx-max",
                "7",
                "--tx-timeout-secs=4",
                "--tx-max-lifetime-secs",
                "60",
            ]),
            TxFlags {
                max_open: NonZeroU64::new(7),
                idle_timeout_secs: NonZeroU64::new(4),
                max_lifetime_secs: NonZeroU64::new(60),
            }
        );
        for bad in [
            &["serve", "/tmp/db", "--tx-max", "0"][..],
            &["serve", "/tmp/db", "--tx-max"],
            &["serve", "/tmp/db", "--tx-timeout-secs=0"],
            &["serve", "/tmp/db", "--tx-max-lifetime-secs", "-5"],
            &["serve", "/tmp/db", "--tx-timeout-secs", "soon"],
        ] {
            assert!(parse_args(bad).is_err(), "{bad:?}");
        }
    }

    #[test]
    fn a_tx_flag_wins_over_its_variable_and_an_unset_flag_leaves_the_variable() {
        let env = |name: &str| match name {
            server::TX_MAX_ENV => Some("5".to_string()),
            server::TX_TIMEOUT_ENV => Some("9".to_string()),
            server::TX_MAX_LIFETIME_ENV => Some("99".to_string()),
            _ => None,
        };
        let s = Duration::from_secs;
        let from_env = resolve_tx_limits(&TxFlags::default(), &env);
        assert_eq!(
            (
                from_env.max_open,
                from_env.idle_timeout,
                from_env.max_lifetime
            ),
            (5, s(9), s(99))
        );
        let flags = TxFlags {
            max_open: NonZeroU64::new(2),
            idle_timeout_secs: None,
            max_lifetime_secs: NonZeroU64::new(40),
        };
        let mixed = resolve_tx_limits(&flags, &env);
        assert_eq!(
            (mixed.max_open, mixed.idle_timeout, mixed.max_lifetime),
            (2, s(9), s(40))
        );
        // Neither: the defaults.
        assert_eq!(
            resolve_tx_limits(&TxFlags::default(), &|_| None),
            server::TxLimits::default()
        );
        // A flag given: an invalid variable beside it is not read.
        let invalid = |name: &str| (name == server::TX_MAX_ENV).then(|| "lots".to_string());
        let given = TxFlags {
            max_open: NonZeroU64::new(3),
            ..TxFlags::default()
        };
        assert_eq!(resolve_tx_limits(&given, &invalid).max_open, 3);
    }

    #[test]
    fn serve_shutdown_timeout_parses_seconds_and_defaults_to_55() {
        match parse_args(&["serve", "/tmp/db"]).unwrap() {
            Command::Serve {
                shutdown_timeout, ..
            } => assert_eq!(shutdown_timeout, Duration::from_secs(55)),
            other => panic!("{other:?}"),
        }
        match parse_args(&["serve", "/tmp/db", "--shutdown-timeout-secs", "90"]).unwrap() {
            Command::Serve {
                shutdown_timeout, ..
            } => assert_eq!(shutdown_timeout, Duration::from_secs(90)),
            other => panic!("{other:?}"),
        }
        match parse_args(&["serve", "/tmp/db", "--shutdown-timeout-secs=7"]).unwrap() {
            Command::Serve {
                shutdown_timeout, ..
            } => assert_eq!(shutdown_timeout, Duration::from_secs(7)),
            other => panic!("{other:?}"),
        }
        for bad in ["0", "nope"] {
            let err =
                parse_args(&["serve", "/tmp/db", "--shutdown-timeout-secs", bad]).unwrap_err();
            assert!(err.contains("shutdown-timeout-secs"), "{err}");
        }
        assert!(parse_args(&["serve", "/tmp/db", "--shutdown-timeout-secs"]).is_err());
    }

    #[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 parse_keys_subcommands() {
        match parse_args(&["keys", "add", "/d", "app", "--tier", "write"]).unwrap() {
            Command::KeysAdd {
                name, tier, role, ..
            } => assert_eq!((name.as_str(), tier.as_str(), role), ("app", "write", None)),
            other => panic!("{other:?}"),
        }
        match parse_args(&["keys", "add", "/d", "t", "--role", "r", "--tier", "read"]).unwrap() {
            Command::KeysAdd { tier, role, .. } => {
                assert_eq!((tier.as_str(), role.as_deref()), ("read", Some("r")))
            }
            other => panic!("{other:?}"),
        }
        assert!(matches!(
            parse_args(&["keys", "hash"]),
            Ok(Command::KeysHash)
        ));
        for bad in [
            vec!["keys"],
            vec!["keys", "hash", "x"],
            vec!["keys", "add", "/d", "app"],
            vec!["keys", "add", "/d", "--tier", "write"],
            vec!["keys", "add", "/d", "app", "--tier", "read"],
            vec!["keys", "add", "/d", "app", "--tier", "write", "--role", "r"],
            vec!["keys", "add", "/d", "app", "--tier", "root"],
            vec![
                "keys", "add", "/d", "app", "--tier", "write", "--tier", "read",
            ],
            vec!["keys", "add", "/d", "app", "--tier", "write", "--bogus"],
        ] {
            assert!(parse_args(&bad).is_err(), "{bad:?}");
        }
    }

    #[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 = mushroomdb::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);

        // A truncating snapshot discards history too, and says so the same way.
        let truncated = tmp("stats-horizon-truncated");
        {
            let mut db = GraphDb::open(&truncated).expect("open");
            db.insert_node("Person", "a", vec![]).expect("insert");
            db.insert_node("Person", "b", vec![]).expect("insert");
        }
        run_snapshot(&truncated, WalDisposition::Truncate, None).expect("truncate");
        let stats = read_stats(&truncated).expect("stats");
        assert_eq!(stats.history_floor, 2, "two commits were discarded");
        assert!(
            format_stats(&stats).contains("history: reaches back to commit 2"),
            "a truncated store names its floor, got:\n{}",
            format_stats(&stats)
        );
        let _ = std::fs::remove_dir_all(&truncated);
    }

    #[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_restore_takes_from_at_and_into() {
        assert_eq!(
            parse_args(&["restore", "--from", "s3://b/p"]),
            Ok(Command::RestoreS3 {
                from: "s3://b/p".into(),
                at: None,
                into: None,
            })
        );
        assert_eq!(
            parse_args(&[
                "restore",
                "--from=minio://b/p",
                "--at",
                "75",
                "--into",
                "/tmp/r",
            ]),
            Ok(Command::RestoreS3 {
                from: "minio://b/p".into(),
                at: Some("75".into()),
                into: Some(PathBuf::from("/tmp/r")),
            })
        );
        match parse_args(&["restore", "--from", "s3://b", "--at=2026-10-01T12:00:00Z"]) {
            Ok(Command::RestoreS3 { at, .. }) => {
                assert_eq!(at.as_deref(), Some("2026-10-01T12:00:00Z"))
            }
            other => panic!("restore parse, got {other:?}"),
        }
    }

    #[test]
    fn parse_restore_refuses_what_it_cannot_use() {
        for (args, needle) in [
            (vec!["restore"], "--from"),
            (vec!["restore", "--into", "x"], "--from"),
            (
                vec!["restore", "--from", "s3://b", "--at", "yesterday"],
                "RFC 3339",
            ),
            (vec!["restore", "--from", "s3://b", "--at"], "missing value"),
            (
                vec!["restore", "--from", "s3://b", "extra"],
                "unexpected argument",
            ),
            (
                vec!["restore", "--from", "s3://b", "--force"],
                "unexpected flag",
            ),
        ] {
            let e = parse_args(&args).unwrap_err();
            assert!(e.contains(needle), "{args:?}: {e}");
        }
    }

    #[test]
    fn restore_with_a_bad_url_fails_before_touching_anything() {
        let parent = tmp("restore-bad-url");
        let into = parent.join("restored");
        let e = run_restore_s3("gs://b/p", None, Some(&into)).unwrap_err();
        assert!(e.0.contains("--from"), "{e}");
        assert!(!into.exists());
    }

    #[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 mushroomdb::{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 mushroomdb::{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 mushroomdb::{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![mushroomdb::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` and `asof` take the statement from `--file` or `--stdin`, and
    /// refuse a second source rather than pick one.
    #[test]
    fn parse_query_and_asof_take_a_file_or_stdin() {
        let Ok(Command::Query {
            cypher,
            file,
            stdin,
            ..
        }) = parse_args(&["query", "/tmp/db", "--file", "/tmp/q.cyp"])
        else {
            panic!("expected Query");
        };
        assert_eq!(cypher, "");
        assert_eq!(file, Some(PathBuf::from("/tmp/q.cyp")));
        assert!(!stdin);

        let Ok(Command::Query { file, stdin, .. }) = parse_args(&["query", "/tmp/db", "--stdin"])
        else {
            panic!("expected Query");
        };
        assert_eq!(file, None);
        assert!(stdin);

        let Ok(Command::AsOf {
            commit,
            query,
            file,
            stdin,
            ..
        }) = parse_args(&["asof", "/tmp/db", "--commit", "1", "--file=/tmp/q.cyp"])
        else {
            panic!("expected AsOf");
        };
        assert_eq!(commit, Some(1));
        assert_eq!(query, None);
        assert_eq!(file, Some(PathBuf::from("/tmp/q.cyp")));
        assert!(!stdin);

        let Ok(Command::AsOf { stdin, .. }) =
            parse_args(&["asof", "/tmp/db", "--commit", "1", "--stdin"])
        else {
            panic!("expected AsOf");
        };
        assert!(stdin);

        for args in [
            &["query", "/tmp/db", "--stdin", "--file", "/tmp/q"][..],
            &["query", "/tmp/db", "--stdin", "MATCH (n) RETURN n"],
            &[
                "query",
                "/tmp/db",
                "--file",
                "/tmp/q",
                "--query",
                "MATCH (n) RETURN n",
            ],
            &[
                "asof",
                "/tmp/db",
                "--commit",
                "1",
                "--stdin",
                "--query",
                "MATCH (n) RETURN n",
            ],
            &[
                "asof", "/tmp/db", "--commit", "1", "--stdin", "--file", "/tmp/q",
            ],
            &["query", "/tmp/db", "--file"],
        ] {
            assert!(parse_args(args).is_err(), "{args:?} should be refused");
        }
        assert!(usage().contains("--file <path>|--stdin"));
    }

    /// The trailing newline a file or pipe ends in is not part of the
    /// statement; an empty one is refused.
    #[test]
    fn resolve_statement_drops_the_trailing_newline() {
        let mut piped: &[u8] = b"MATCH (n) RETURN count(n)\n\n";
        let got = resolve_statement(None, None, true, &mut piped).expect("stdin");
        assert_eq!(got.as_deref(), Some("MATCH (n) RETURN count(n)"));

        let mut unused: &[u8] = b"";
        let inline = resolve_statement(Some("MATCH (n) RETURN n"), None, false, &mut unused)
            .expect("inline");
        assert_eq!(inline.as_deref(), Some("MATCH (n) RETURN n"));
        assert_eq!(
            resolve_statement(None, None, false, &mut unused).unwrap(),
            None
        );

        // A byte-order mark and one trailing semicolon are not the statement,
        // whichever way it arrives.
        let mut bom: &[u8] = b"\xEF\xBB\xBFMATCH (n) RETURN n;  \n";
        let got = resolve_statement(None, None, true, &mut bom).expect("bom");
        assert_eq!(got.as_deref(), Some("MATCH (n) RETURN n"));
        let got = resolve_statement(Some("MATCH (n) RETURN n ; "), None, false, &mut unused)
            .expect("inline semicolon");
        assert_eq!(got.as_deref(), Some("MATCH (n) RETURN n"));
        let got = resolve_statement(Some("RETURN ';;'"), None, false, &mut unused).unwrap();
        assert_eq!(got.as_deref(), Some("RETURN ';;'"));

        let mut blank: &[u8] = b" \n";
        assert!(resolve_statement(None, None, true, &mut blank).is_err());
        assert!(resolve_statement(
            None,
            Some(Path::new("/nonexistent/q.cyp")),
            false,
            &mut unused
        )
        .is_err());
    }

    /// `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);
    }
}