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 export;

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

/// 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";

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

/// Parsed `mushroomdb` invocation.
#[derive(Debug, Clone, PartialEq, Eq)]
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>,
    },
    Mcp {
        db_dir: PathBuf,
    },
    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).
    Algo {
        db_dir: PathBuf,
        subcmd: AlgoSubcmd,
        /// Print only the top N results (0 = all).
        top: usize,
    },
    /// 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,
    },
    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 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>
  mushroomdb stats <db-dir>
  mushroomdb demo <db-dir>
  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   export all data
  mushroomdb schema apply <db-dir> <schema.json>
  mushroomdb algo pagerank <db-dir> [--top N]
  mushroomdb algo wcc <db-dir> [--top N]
  mushroomdb algo degree <db-dir> [--top N]
  mushroomdb --help

Default serve address is 127.0.0.1:8080. Non-loopback --addr requires --token or MUSHROOMDB_TOKEN.
"
}

/// 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),
        "serve" => parse_serve(&args[1..]),
        "mcp" => parse_one_dir("mcp", &args[1..]).map(|db_dir| Command::Mcp { db_dir }),
        "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..]),
        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 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.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());
    }
    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,
    })
}

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,
        },
    )?;
    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> {
    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 or parquet"))?;
            i += 2;
        } else if let Some(val) = a.strip_prefix("--format=") {
            format = ExportFormat::parse(val)
                .ok_or_else(|| format!("unknown format '{val}'; expected jsonl or parquet"))?;
            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)?,
        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
    ))
}

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".to_string());
    }
    let subcmd = match args[0] {
        "pagerank" => AlgoSubcmd::Pagerank,
        "wcc" => AlgoSubcmd::Wcc,
        "degree" => AlgoSubcmd::Degree,
        other => {
            return Err(format!(
                "unknown algo subcommand: {other}; expected pagerank | wcc | degree"
            ))
        }
    };
    let rest = &args[1..];
    let mut db_dir = None;
    let mut top: usize = 20;
    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.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,
    })
}

/// Run a graph algorithm and return a formatted string.
pub fn run_algo(db_dir: &Path, subcmd: &AlgoSubcmd, top: usize) -> Result<String, CliError> {
    let db = GraphDb::open(db_dir)?;
    match subcmd {
        AlgoSubcmd::Pagerank => {
            let config = 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: AlgoDir::Both,
                ..DegreeConfig::default()
            };
            let report = db.degree_centrality(&config);
            Ok(format_degree(&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
}

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,
        })?;
    }

    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,
                    }) => {
                        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);
                    }
                    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,
                    }) => {
                        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);
                    }
                    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,
                    }) => {
                        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);
                    }
                    other => panic!("serve --addr=VALUE, got {other:?}"),
                },
            },
            Case {
                args: &["mcp", "/tmp/demo-db"],
                check: |r| match r {
                    Ok(Command::Mcp { db_dir }) => {
                        assert_eq!(db_dir, PathBuf::from("/tmp/demo-db"));
                    }
                    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",
            ]
        );

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

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