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

use core_api::repograph;
use core_api::schema::Schema;
use core_api::{
    default_max_edges, is_write_query, wal_commit_count_at, AlgoDir, BackupReport, DegreeConfig,
    Explanation, GraphDb, IngestOptions, LouvainConfig, PageRankConfig, Predicate, ResultSet,
    RuleDef, RuleSuggestion, SharedDb, SnapshotOptions, Stats, Value, WccConfig,
};
use export::ExportFormat;
use std::collections::{BTreeMap, BTreeSet};
use std::fmt::Write as _;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::time::Duration;

/// 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. `<cwd>/mushroom-memory`, but only when the working directory is 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 cwd.join(".git").exists() {
        return cwd.join("mushroom-memory");
    }
    home.join(".mushroomdb").join("memory")
}

/// 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 {
    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>,
        /// Path to PEM certificate for native TLS (`--tls-cert`). Requires `--tls-key`.
        tls_cert: Option<PathBuf>,
        /// Path to PEM private key for native TLS (`--tls-key`). Requires `--tls-cert`.
        tls_key: Option<PathBuf>,
    },
    Mcp {
        /// `None` with `auto` set: resolved by [`resolve_auto_db`] at run time.
        db_dir: Option<PathBuf>,
        auto: bool,
    },
    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).
        commit: u64,
        /// Optional Cypher read query to execute against the as-of view.
        query: 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,
    },
    /// Write `snapshot.bin` (default truncates WAL unless `--keep-wal`).
    Snapshot {
        db_dir: PathBuf,
        keep_wal: bool,
        /// Rename WAL to wal.<commit_seq>.archive before writing fresh baseline.
        archive_wal: bool,
        /// Keep the newest N archives; prune oldest at snapshot time.
        /// None = unlimited. Applies only when archive_wal is true.
        retention: Option<u32>,
    },
    /// Apply a JSON schema file idempotently (`schema apply <db-dir> <schema.json>`).
    SchemaApply {
        db_dir: PathBuf,
        schema_file: PathBuf,
    },
    /// Migrate an old-format snapshot to the current version and keep `.bak`.
    Migrate {
        db_dir: PathBuf,
    },
    /// Validate CRC32 integrity of every section in the V8 snapshot.
    Verify {
        db_dir: PathBuf,
    },
    /// Create a consistent, verified copy of the database directory.
    Backup {
        db_dir: PathBuf,
        dest: PathBuf,
    },
    /// Export all nodes, edges, and rules to a destination directory.
    Export {
        db_dir: PathBuf,
        dest: PathBuf,
        format: ExportFormat,
    },
    /// Build (or incrementally sync) a graph of a git repository.
    IngestGit {
        db_dir: PathBuf,
        opts: ingest_git::IngestGitOpts,
    },
    /// Wire the /mushroom skill and MCP server into Claude Code / Cursor.
    Install(install::InstallOpts),
    /// Undo what `install` wrote (manifest-driven).
    Uninstall(install::InstallOpts),
    /// 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,
    },
    /// Bring the store up to date with the repository the `GitSync` marker
    /// names: the commits since the marker, then the dirty working tree.
    Sync {
        db_dir: PathBuf,
        /// Print the report as one JSON object instead of the plain digest.
        /// The MCP `sync` tool runs this binary and reads that object, so the
        /// counts reach an assistant without being parsed back out of prose.
        json: bool,
    },
    /// Re-extract named files only. Body of the PostToolUse hook, which reads
    /// the paths off a payload on stdin when none are given on the command line.
    Touch {
        db_dir: Option<PathBuf>,
        auto: bool,
        files: Vec<PathBuf>,
    },
    /// Summarise the repository the store was built from: clusters, key
    /// files, owners, what is hot, and what is worth asking about.
    Map {
        db_dir: PathBuf,
        /// Print the [`core_api::repograph::RepoMap`] as JSON instead of the
        /// rendered digest.
        json: bool,
    },
    /// Everything the graph knows about one file or symbol.
    Context {
        db_dir: PathBuf,
        target: String,
    },
    /// What else the named files reach: co-change partners, importers, and the
    /// symbols other files call.
    Impact {
        db_dir: PathBuf,
        files: Vec<String>,
    },
    /// Who has written a file, and when.
    Owners {
        db_dir: PathBuf,
        path: String,
    },
    /// What links two nodes, with the evidence behind each link.
    Why {
        db_dir: PathBuf,
        a: String,
        b: String,
    },
    Version,
    Help,
}

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

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

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

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

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

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

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

Usage:
  mushroomdb install [--platform claude-code|cursor|codex|all] [--project|--user] [--db <path>]
                     [--command <path>] [--no-git-hooks] [--no-prewarm]
  mushroomdb uninstall [--platform claude-code|cursor|codex|all] [--project|--user] [--db <path>]
  mushroomdb doctor [--project|--user] [--platform claude-code|cursor|codex|all]
                     verify an install: config entry, store, hooks, git hooks, and a real
                     stdio handshake with the configured MCP command; exits 1 on any `fail`
  mushroomdb serve <db-dir> [--addr 127.0.0.1:8080] [--token <secret>] [--ui <dist-dir>] [--no-ui] [--demo-if-empty] [--snapshot-every <secs>]
  mushroomdb mcp <db-dir>|--auto
  mushroomdb stats <db-dir>
  mushroomdb demo <db-dir>
  mushroomdb recall <db-dir>|--auto   hook body: reads a prompt payload on stdin, prints related graph facts
  mushroomdb sync <db-dir> [--json] re-sync the repo the store was built from: new commits, then the dirty working tree
  mushroomdb map <db-dir> [--json] summarise the graphed repository: clusters, key files, owners, hot files
                                   --json prints the computed map instead of the rendered digest
  mushroomdb context <db-dir> <target>   one file or symbol from every side: signature, source, callers,
                                   callees, importers, co-change partners, commits, notes
                                   <target> is a file path, a symbol key, or a bare symbol name
  mushroomdb impact <db-dir> <file>...   what changing these files reaches: partners, importers,
                                   and the symbols other files call
  mushroomdb owners <db-dir> <path>      top author and share, who else knows it, last touch, last 4 quarters
  mushroomdb why <db-dir> <a> <b>        every rule edge between two nodes with its evidence, or the
                                   shortest path between them
  mushroomdb touch <db-dir>|--auto [<file>...]
                                   re-extract just these files; with no <file> reads them from a
                                   PostToolUse payload on stdin (hook body)
  mushroomdb suggest <db-dir>
  mushroomdb asof <db-dir> --commit N [--query \"MATCH ...\"]
  mushroomdb query <db-dir> [--query \"MATCH ...\"] <cypher…>
  mushroomdb snapshot <db-dir> [--keep-wal]
  mushroomdb migrate <db-dir>
  mushroomdb verify <db-dir>       validate CRC32 integrity of every snapshot section
  mushroomdb backup <db-dir> <dest>   process-local consistent copy of the database to <dest>
                                      WARNING: unsafe against a concurrently running serve process;
                                      use POST /backup on the HTTP server for live-serve backups
  mushroomdb export <db-dir> <dest> --format jsonl|parquet|graphml   export all data
                                      graphml writes one file: <dest>/graph.graphml if <dest> is an
                                      existing directory, otherwise <dest> is the file path itself
                                      (nodes + edges only; rules have no GraphML analogue)
  mushroomdb ingest-git <db-dir> <repo-dir> [--exclude <pattern>]... [--max-commits-per-file N]
                        [--recurse-submodules] [--prs] [--no-structure] [--no-docs] [--ensure-gitignore]
                                   graph a git repo (authors, commits, files, symbols, imports, calls, mentions); re-run to sync
                                   --recurse-submodules also walks each initialised submodule
                                   --prs links merged pull requests via gh (skipped when gh is unavailable)
                                   --no-structure skips the working-tree pass (no hashes, symbols, imports or calls)
                                   --no-docs skips Markdown bodies, headings and mentions
                                   --ensure-gitignore adds the database directory to the repo's .gitignore
                                   with no --exclude the defaults apply: target/ node_modules/ dist/ .git/ *.lock *.min.js
  mushroomdb schema apply <db-dir> <schema.json>
  mushroomdb algo pagerank <db-dir> [--top N] [--dir out|in|both]
  mushroomdb algo wcc <db-dir> [--top N]
  mushroomdb algo degree <db-dir> [--top N] [--dir out|in|both]
  mushroomdb algo communities <db-dir> [--edge-type T]... [--weight-prop P] [--min-weight X] [--top N]
  mushroomdb --version
  mushroomdb --help

Default serve address is 127.0.0.1:8080. Non-loopback --addr requires --token or MUSHROOMDB_TOKEN.
install defaults: --platform auto-detect; scope auto (project inside a git checkout, else user);
the MCP entry runs `npx -y mushroomdb@<version>` unless a `mushroomdb` on PATH is this binary, or
--command names one (a relative --command or --db is anchored to the current directory).
--no-git-hooks skips the post-commit/checkout/merge sync hooks; --no-prewarm skips fetching the
pinned package once. 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.
--auto resolves the database as $CLAUDE_PROJECT_DIR/mushroom-memory, else ./mushroom-memory in a
git checkout, 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 git_hooks = true;
    let mut prewarm = true;
    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
            };
            // 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" {
            git_hooks = 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}"));
        }
    }
    Ok(install::InstallOpts {
        platform,
        scope,
        db,
        command,
        git_hooks,
        prewarm,
    })
}

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

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_dir_or_auto("mcp", &args[1..]).map(|(db_dir, auto)| Command::Mcp { db_dir, auto })
        }
        "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..]),
        "schema" => parse_schema(&args[1..]),
        "migrate" => parse_one_dir("migrate", &args[1..]).map(|db_dir| Command::Migrate { db_dir }),
        "verify" => parse_one_dir("verify", &args[1..]).map(|db_dir| Command::Verify { db_dir }),
        "backup" => parse_backup(&args[1..]),
        "export" => parse_export(&args[1..]),
        "recall" => parse_dir_or_auto("recall", &args[1..])
            .map(|(db_dir, auto)| Command::Recall { db_dir, auto }),
        "sync" => parse_dir_with_json("sync", &args[1..])
            .map(|(db_dir, json)| Command::Sync { db_dir, json }),
        "map" => parse_dir_with_json("map", &args[1..])
            .map(|(db_dir, json)| Command::Map { db_dir, json }),
        "context" => {
            parse_positional("context", &args[1..], 1, 1).map(|(db_dir, rest)| Command::Context {
                db_dir,
                target: rest[0].clone(),
            })
        }
        "impact" => parse_positional("impact", &args[1..], 1, usize::MAX)
            .map(|(db_dir, files)| Command::Impact { db_dir, files }),
        "owners" => {
            parse_positional("owners", &args[1..], 1, 1).map(|(db_dir, rest)| Command::Owners {
                db_dir,
                path: rest[0].clone(),
            })
        }
        "why" => parse_positional("why", &args[1..], 2, 2).map(|(db_dir, rest)| Command::Why {
            db_dir,
            a: rest[0].clone(),
            b: rest[1].clone(),
        }),
        "touch" => parse_touch(&args[1..]),
        "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),
        "doctor" => parse_doctor_cmd(&args[1..]).map(Command::Doctor),
        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 tls_cert: Option<PathBuf> = None;
    let mut tls_key: Option<PathBuf> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--addr" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --addr".to_string())?;
            addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--addr=") {
            addr = val.parse().map_err(|_| format!("invalid address: {val}"))?;
            i += 1;
        } else if a == "--ui" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --ui".to_string())?;
            ui = ServeUi::Filesystem(PathBuf::from(val));
            saw_ui = true;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--ui=") {
            ui = ServeUi::Filesystem(PathBuf::from(val));
            saw_ui = true;
            i += 1;
        } else if a == "--no-ui" {
            ui = ServeUi::None;
            saw_no_ui = true;
            i += 1;
        } else if a == "--demo-if-empty" {
            demo_if_empty = true;
            i += 1;
        } else if a == "--token" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --token".to_string())?;
            token = Some(val.to_string());
            i += 2;
        } else if let Some(val) = a.strip_prefix("--token=") {
            token = Some(val.to_string());
            i += 1;
        } else if a == "--role-token" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --role-token".to_string())?;
            let (tok, role) = parse_role_token(val)?;
            role_tokens.push((tok, role));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--role-token=") {
            let (tok, role) = parse_role_token(val)?;
            role_tokens.push((tok, role));
            i += 1;
        } else if a == "--snapshot-every" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --snapshot-every".to_string())?;
            snapshot_every = Some(parse_snapshot_every(val)?);
            i += 2;
        } else if let Some(val) = a.strip_prefix("--snapshot-every=") {
            snapshot_every = Some(parse_snapshot_every(val)?);
            i += 1;
        } else if a == "--tls-cert" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --tls-cert".to_string())?;
            tls_cert = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--tls-cert=") {
            tls_cert = Some(PathBuf::from(val));
            i += 1;
        } else if a == "--tls-key" {
            let val = args
                .get(i + 1)
                .copied()
                .ok_or_else(|| "missing value for --tls-key".to_string())?;
            tls_key = Some(PathBuf::from(val));
            i += 2;
        } else if let Some(val) = a.strip_prefix("--tls-key=") {
            tls_key = Some(PathBuf::from(val));
            i += 1;
        } else if a.starts_with('-') {
            return Err(format!("unexpected flag: {a}"));
        } else if db_dir.is_none() {
            db_dir = Some(PathBuf::from(a));
            i += 1;
        } else {
            return Err(format!("unexpected extra argument: {a}"));
        }
    }
    if saw_ui && saw_no_ui {
        return Err("cannot combine --ui and --no-ui".to_string());
    }
    match (&tls_cert, &tls_key) {
        (Some(_), None) => return Err("--tls-cert requires --tls-key".to_string()),
        (None, Some(_)) => return Err("--tls-key requires --tls-cert".to_string()),
        _ => {}
    }
    let db_dir = db_dir.ok_or_else(|| "serve requires <db-dir>".to_string())?;
    Ok(Command::Serve {
        db_dir,
        addr,
        ui,
        demo_if_empty,
        token,
        role_tokens,
        snapshot_every,
        tls_cert,
        tls_key,
    })
}

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

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

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

fn parse_asof(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut commit: Option<u64> = None;
    let mut query: Option<String> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        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.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())?;
    let commit = commit.ok_or_else(|| "asof requires --commit N".to_string())?;
    Ok(Command::AsOf {
        db_dir,
        commit,
        query,
    })
}

/// Execute an as-of query at the given commit and print results.
pub fn run_asof(db_dir: &Path, commit: u64, query: Option<&str>) -> Result<String, CliError> {
    let total = wal_commit_count_at(db_dir)?;
    let db = GraphDb::open_at(db_dir, commit)?;
    let mut out = String::new();
    let _ = writeln!(out, "as-of commit {} of {}", commit, total);
    if let Some(cypher) = query {
        let params = BTreeMap::new();
        let rs = db.query(cypher, &params)?;
        out.push_str(&format_result_set(&rs));
    }
    Ok(out)
}

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

/// Run a Cypher read or write and print columns/rows like [`run_asof`].
pub fn run_query(db_dir: &Path, cypher: &str) -> Result<String, CliError> {
    let params = BTreeMap::new();
    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;
    let mut keep_wal = false;
    let mut archive_wal = false;
    let mut retention: Option<u32> = None;
    let mut i = 0;
    while i < args.len() {
        let a = args[i];
        if a == "--keep-wal" {
            keep_wal = true;
            i += 1;
        } else if a == "--archive-wal" {
            archive_wal = true;
            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,
        keep_wal,
        archive_wal,
        retention,
    })
}

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

    if from_ver == Some(current) {
        return Ok(format!("already current (V{current})\n"));
    }

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

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

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

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

/// Open `dir` and write `snapshot.bin`. Default truncates the WAL.
pub fn run_snapshot(
    db_dir: &Path,
    keep_wal: bool,
    archive_wal: bool,
    retention: Option<u32>,
) -> Result<String, CliError> {
    let mut db = GraphDb::open(db_dir)?;
    if archive_wal {
        db.set_wal_archive_retention(retention);
        db.snapshot_with(SnapshotOptions {
            archive_wal: true,
            keep_wal: false,
        })?;
    } else if keep_wal {
        db.snapshot_with(SnapshotOptions {
            keep_wal: true,
            archive_wal: false,
        })?;
    } else {
        db.snapshot()?;
    }
    Ok(format!(
        "snapshot written: {}\n",
        db_dir.join("snapshot.bin").display()
    ))
}

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

fn parse_schema_apply(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut schema_file = None;
    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 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())?;
    let schema_file =
        schema_file.ok_or_else(|| "schema apply requires <schema.json>".to_string())?;
    Ok(Command::SchemaApply {
        db_dir,
        schema_file,
    })
}

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

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

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

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

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

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

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

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

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

/// Body of `mushroomdb map <db-dir> [--json]`.
///
/// Opens read-only, with both write paths off: a map is a question, and asking
/// it must never migrate a snapshot, rewrite a torn WAL tail, or make a writer
/// wait on the cross-process lock.
pub fn run_map(db_dir: &Path, json: bool) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    let map = repograph::repo_map(&db, &repograph::MapOptions::default());
    if json {
        let mut out = serde_json::to_string_pretty(&map)
            .map_err(|e| CliError(format!("serialise map: {e}")))?;
        out.push('\n');
        return Ok(out);
    }
    Ok(repograph::render_map(&map))
}

/// 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.
fn open_for_reading(db_dir: &Path) -> Result<structure::Db, CliError> {
    Ok(GraphDb::open_with_options(
        db_dir,
        core_api::OpenOptions {
            auto_migrate: false,
            repair_wal: false,
            read_only: true,
        },
    )?)
}

/// Body of `mushroomdb context <db-dir> <target>`.
///
/// The source is quoted from the repository the `GitSync` marker names, which
/// is the checkout the store was built from.
pub fn run_context(db_dir: &Path, target: &str) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    Ok(repograph::render_context(&repograph::context(
        &db, None, target,
    )))
}

/// Body of `mushroomdb impact <db-dir> <file>...`.
///
/// The files named are taken to be the change, so each is reported and every
/// partner that is one of them is marked `modified`.
pub fn run_impact(db_dir: &Path, files: &[String]) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    let modified: BTreeSet<String> = files.iter().cloned().collect();
    let report = repograph::impact(&db, files, &modified, &repograph::ImpactOptions::default());
    Ok(repograph::render_impact(&report))
}

/// Body of `mushroomdb owners <db-dir> <path>`.
pub fn run_owners(db_dir: &Path, path: &str) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    match repograph::owners(&db, path, None) {
        Some(report) => Ok(repograph::render_owners(&report)),
        None => Err(CliError(format!("no file in the store at {path}"))),
    }
}

/// Body of `mushroomdb why <db-dir> <a> <b>`.
pub fn run_why(db_dir: &Path, a: &str, b: &str) -> Result<String, CliError> {
    let db = open_for_reading(db_dir)?;
    Ok(repograph::render_why(&repograph::why(&db, a, b)))
}

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

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

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

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

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

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

/// `touch [<db-dir>|--auto] [<file>...]`. The first positional is the database
/// unless `--auto` already named it, in which case every positional is a file.
fn parse_touch(args: &[&str]) -> Result<Command, String> {
    let mut db_dir = None;
    let mut auto = false;
    let mut files = Vec::new();
    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_none() && !auto {
            db_dir = Some(PathBuf::from(*a));
        } else {
            files.push(PathBuf::from(*a));
        }
    }
    if db_dir.is_none() && !auto {
        return Err("touch requires <db-dir> or --auto".into());
    }
    if db_dir.is_some() && auto {
        return Err("touch: --auto takes no <db-dir>".into());
    }
    Ok(Command::Touch {
        db_dir,
        auto,
        files,
    })
}

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

/// `<cmd> <db-dir> [--json]`, shared by `map` and `sync`.
fn parse_dir_with_json(cmd: &str, args: &[&str]) -> Result<(PathBuf, bool), String> {
    let mut db_dir = None;
    let mut json = false;
    for a in args {
        if *a == "--json" {
            json = 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));
        }
    }
    let db_dir = db_dir.ok_or_else(|| format!("{cmd} requires <db-dir>"))?;
    Ok((db_dir, json))
}

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);
    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();
        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,
        })?;
        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,
        })?;
        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,
        })?;
        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,
        })?;
        // Name lookup for `mushroomdb recall`. Adds no nodes or edges.
        for (label, field) in [("Org", "name"), ("Project", "name"), ("Person", "name")] {
            w.enable_fulltext(label, field)?;
        }
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    buf
}

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

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

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

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

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

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

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

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

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

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

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

        let cases = [
            Case {
                args: &[],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("no-args → Help, got {other:?}"),
                },
            },
            Case {
                args: &["--help"],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("--help → Help, got {other:?}"),
                },
            },
            Case {
                args: &["-h"],
                check: |r| match r {
                    Ok(Command::Help) => {}
                    other => panic!("-h → Help, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(addr, default_bind());
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        assert!(role_tokens.is_empty());
                        assert_eq!(snapshot_every, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve <dir> → Serve default addr, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr", "127.0.0.1:8080"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(
                            addr,
                            "127.0.0.1:8080".parse::<std::net::SocketAddr>().unwrap()
                        );
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        assert!(role_tokens.is_empty());
                        assert_eq!(snapshot_every, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve --addr after dir, got {other:?}"),
                },
            },
            Case {
                args: &["serve", "/tmp/demo-db", "--addr=127.0.0.1:9090"],
                check: |r| match r {
                    Ok(Command::Serve {
                        db_dir,
                        addr,
                        ui,
                        demo_if_empty,
                        token,
                        role_tokens,
                        snapshot_every,
                        tls_cert,
                        tls_key,
                    }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                        assert_eq!(
                            addr,
                            "127.0.0.1:9090".parse::<std::net::SocketAddr>().unwrap()
                        );
                        assert_eq!(ui, super::ServeUi::Embedded);
                        assert!(!demo_if_empty);
                        assert_eq!(token, None);
                        let _ = role_tokens; // empty, not asserted
                        assert_eq!(snapshot_every, None);
                        assert_eq!(tls_cert, None);
                        assert_eq!(tls_key, None);
                    }
                    other => panic!("serve --addr=VALUE, got {other:?}"),
                },
            },
            Case {
                args: &["mcp", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Mcp { db_dir, auto }) => {
                        assert_eq!(db_dir, Some(PathBuf::from("/tmp/demo-db")));
                        assert!(!auto);
                    }
                    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:?}"),
                },
            },
        ];

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

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

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

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

    #[test]
    fn parse_snapshot_and_query() {
        match parse_args(&["snapshot", "/tmp/db"]).unwrap() {
            Command::Snapshot { keep_wal, .. } => assert!(!keep_wal),
            other => panic!("{other:?}"),
        }
        match parse_args(&["snapshot", "/tmp/db", "--keep-wal"]).unwrap() {
            Command::Snapshot { keep_wal, .. } => assert!(keep_wal),
            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}"
        );
    }

    #[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",
            "--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; enabling it adds no nodes, edges or rules.
        let db = SharedDb::open(&dir).expect("reopen demo");
        assert_eq!(
            db.read().fulltext_pairs(),
            vec![
                ("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, false, false, 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);
    }

    #[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").expect("query");
        assert!(out.contains("columns:"), "got {out}");
        assert!(out.contains("id=alice"), "got {out}");
        let _ = run_query(&dir, "CREATE (n:Person {id: 'bob'})").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);
    }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    #[test]
    fn map_parses_a_dir_and_an_optional_json_flag() {
        assert_eq!(
            parse_args(&["map", "/tmp/db"]).unwrap(),
            Command::Map {
                db_dir: PathBuf::from("/tmp/db"),
                json: false,
            }
        );
        // The flag may come before or after the directory.
        let want = Command::Map {
            db_dir: PathBuf::from("/tmp/db"),
            json: true,
        };
        assert_eq!(parse_args(&["map", "/tmp/db", "--json"]).unwrap(), want);
        assert_eq!(parse_args(&["map", "--json", "/tmp/db"]).unwrap(), want);
        assert!(parse_args(&["map"]).is_err(), "<db-dir> is required");
        assert!(parse_args(&["map", "/tmp/db", "/tmp/other"]).is_err());
        assert!(parse_args(&["map", "/tmp/db", "--nope"]).is_err());
        assert!(usage().contains("mushroomdb map <db-dir> [--json]"));
    }

    #[test]
    fn the_graph_tools_take_a_dir_and_their_keys() {
        assert_eq!(
            parse_args(&["context", "/tmp/db", "src/db.rs#open"]).unwrap(),
            Command::Context {
                db_dir: PathBuf::from("/tmp/db"),
                target: "src/db.rs#open".to_string(),
            }
        );
        assert_eq!(
            parse_args(&["impact", "/tmp/db", "a.rs", "b.rs"]).unwrap(),
            Command::Impact {
                db_dir: PathBuf::from("/tmp/db"),
                files: vec!["a.rs".to_string(), "b.rs".to_string()],
            }
        );
        assert_eq!(
            parse_args(&["owners", "/tmp/db", "a.rs"]).unwrap(),
            Command::Owners {
                db_dir: PathBuf::from("/tmp/db"),
                path: "a.rs".to_string(),
            }
        );
        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!["context", "/tmp/db"],
            vec!["context", "/tmp/db", "a", "b"],
            vec!["impact", "/tmp/db"],
            vec!["owners", "/tmp/db"],
            vec!["why", "/tmp/db", "a"],
            vec!["why", "/tmp/db", "a", "b", "c"],
            vec!["why", "/tmp/db", "-a", "b"],
            vec!["context"],
        ] {
            assert!(parse_args(&args).is_err(), "{args:?} must not parse");
        }
        for line in [
            "mushroomdb context <db-dir> <target>",
            "mushroomdb impact <db-dir> <file>...",
            "mushroomdb owners <db-dir> <path>",
            "mushroomdb why <db-dir> <a> <b>",
        ] {
            assert!(usage().contains(line), "usage is missing {line:?}");
        }
    }

    /// 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
            }
        );
        assert_eq!(
            parse_args(&["recall", "--auto"]).unwrap(),
            Command::Recall {
                db_dir: None,
                auto: true
            }
        );
        for cmd in ["mcp", "recall", "touch"] {
            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"));
    }

    #[test]
    fn sync_and_touch_parse() {
        assert_eq!(
            parse_args(&["sync", "/tmp/db"]).unwrap(),
            Command::Sync {
                db_dir: PathBuf::from("/tmp/db"),
                json: false,
            }
        );
        assert_eq!(
            parse_args(&["sync", "/tmp/db", "--json"]).unwrap(),
            Command::Sync {
                db_dir: PathBuf::from("/tmp/db"),
                json: true,
            }
        );
        assert!(parse_args(&["sync"]).is_err(), "db-dir is required");

        // Positional form: the first path is the database, the rest are files.
        assert_eq!(
            parse_args(&["touch", "/tmp/db", "src/a.rs", "src/b.rs"]).unwrap(),
            Command::Touch {
                db_dir: Some(PathBuf::from("/tmp/db")),
                auto: false,
                files: vec![PathBuf::from("src/a.rs"), PathBuf::from("src/b.rs")],
            }
        );
        // With --auto every positional is a file.
        assert_eq!(
            parse_args(&["touch", "--auto", "src/a.rs"]).unwrap(),
            Command::Touch {
                db_dir: None,
                auto: true,
                files: vec![PathBuf::from("src/a.rs")],
            }
        );
        // No files at all is the hook form: the paths arrive on stdin.
        assert_eq!(
            parse_args(&["touch", "--auto"]).unwrap(),
            Command::Touch {
                db_dir: None,
                auto: true,
                files: vec![],
            }
        );
        assert!(usage().contains("mushroomdb sync <db-dir>"));
        assert!(usage().contains("mushroomdb touch"));
    }

    #[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"));
    }
}