// SPDX-License-Identifier: MIT

/// Default TCP port the Cortex daemon binds to when no `--port` flag or
/// `CORTEX_PORT` env var is set. Prod-path consumers must reference this
/// constant rather than hard-coding `7437`. Test fixtures are exempt.
pub const DEFAULT_CORTEX_PORT: u16 = 7437;
const CLI_CAPABILITIES_CONTRACT_VERSION: &str = "1";

mod admin;
mod aging;
mod api_types;
mod auth;
mod budgets;
mod co_occurrence;
mod compaction;
mod compiler;
mod conflict;
mod crystallize;
mod daemon_lifecycle;
mod db;
mod embeddings;
mod eval;
mod export_data;
mod focus;
mod handlers;
mod hook_boot;
mod indexer;
mod mcp_proxy;
mod prompt_inject;
mod rate_limit;
mod rerank;
mod server;
mod service;
mod setup;
mod state;
mod tls;
mod transport;
mod workspace;

use chrono::{self, Utc};
use fs2::FileExt;
use serde_json::{json, Value};
use std::collections::HashSet;
use std::io::IsTerminal;
use std::io::Write as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

/// Install a panic hook that records every panic to `~/.cortex/panic.log` and
/// stderr before the runtime continues unwinding. Without this, dev-build
/// panics produce silent thread death (or process exit) with no breadcrumb,
/// which made the daemon's intermittent crashes nearly impossible to diagnose.
fn install_daemon_panic_hook(paths: &auth::CortexPaths) {
    static INSTALLED: AtomicBool = AtomicBool::new(false);
    if INSTALLED.swap(true, Ordering::SeqCst) {
        return;
    }
    let panic_log_path = paths.home.join("panic.log");
    let previous = std::panic::take_hook();
    std::panic::set_hook(Box::new(move |info| {
        let payload = info.payload();
        let message = if let Some(s) = payload.downcast_ref::<&str>() {
            (*s).to_string()
        } else if let Some(s) = payload.downcast_ref::<String>() {
            s.clone()
        } else {
            "<non-string panic payload>".to_string()
        };
        let location = info
            .location()
            .map(|loc| format!("{}:{}:{}", loc.file(), loc.line(), loc.column()))
            .unwrap_or_else(|| "<unknown location>".to_string());
        let backtrace = std::backtrace::Backtrace::force_capture();
        let entry = format!(
            "[{ts}] PANIC at {location}: {message}\n{backtrace}\n",
            ts = Utc::now().to_rfc3339(),
        );
        eprintln!("[cortex] {entry}");
        if let Some(parent) = panic_log_path.parent() {
            let _ = std::fs::create_dir_all(parent);
        }
        if let Ok(mut file) = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&panic_log_path)
        {
            let _ = file.write_all(entry.as_bytes());
        }
        previous(info);
    }));
}

const BACKUP_RETENTION_COUNT: usize = 3;
const BRIDGE_BACKUP_CLEANUP_SCHEMA_VERSION: i32 = 5;
const LOG_ROTATION_BYTES: u64 = 1024 * 1024;
const CONTROL_CENTER_LOCK_FILE: &str = "control-center.lock";
const CONTROL_CENTER_OWNER_TAG: &str = "control-center";
const SINGLE_DAEMON_TEST_BYPASS_ENV: &str = "CORTEX_SINGLE_DAEMON_TEST_BYPASS";
const SPAWN_PARENT_PID_ENV: &str = "CORTEX_SPAWN_PARENT_PID";
const ORPHAN_WATCH_INTERVAL_SECS: u64 = 2;
const DEFAULT_EMBED_BACKFILL_BATCH_SIZE: usize = 200;
const DEFAULT_EMBED_BACKFILL_MAX_BATCHES_PER_PASS: usize = 8;
const DEFAULT_EMBED_BACKFILL_INTERVAL_SECS: u64 = 120;
const DEFAULT_EMBED_BACKFILL_STARTUP_DRAIN_MAX_BATCHES: usize = 64;
const DEFAULT_STARTUP_INDEX_DELAY_SECS: u64 = 5;
const DEFAULT_STARTUP_AGING_DELAY_SECS: u64 = 20;
const DEFAULT_STARTUP_EMBED_DELAY_SECS: u64 = 30;
const DEFAULT_STARTUP_CRYSTALLIZE_DELAY_SECS: u64 = 45;
const DEFAULT_STARTUP_STORAGE_GOVERNOR_DELAY_SECS: u64 = 12;
const STARTUP_STORAGE_GOVERNOR_CATCHUP_PASSES: usize = 3;
const STARTUP_STORAGE_GOVERNOR_CATCHUP_INTERVAL_SECS: u64 = 90;
const BACKGROUND_DB_LOCK_RETRY_MS: u64 = 50;
const BACKGROUND_DB_LOCK_DEFAULT_MAX_WAIT_MS: u64 = 2_000;
const APP_MANAGED_STARTUP_HEAVY_DELAY_SECS: u64 = 45;
const APP_MANAGED_STARTUP_HEAVY_DELAY_MAX_SECS: u64 = 120;
const APP_MANAGED_AGING_STARTUP_OFFSET_SECS: u64 = 15;
const APP_MANAGED_EMBED_STARTUP_OFFSET_SECS: u64 = 30;
const APP_MANAGED_CRYSTALLIZE_STARTUP_OFFSET_SECS: u64 = 45;
const DEFAULT_IDLE_SHUTDOWN_CHECK_INTERVAL_SECS: u64 = 5;
const DEFAULT_IDLE_SHUTDOWN_MIN_UPTIME_SECS: u64 = 120;
const STARTUP_INDEX_DELAY_ENV: &str = "CORTEX_STARTUP_INDEX_DELAY_SECS";
const STARTUP_AGING_DELAY_ENV: &str = "CORTEX_STARTUP_AGING_DELAY_SECS";
const STARTUP_EMBED_DELAY_ENV: &str = "CORTEX_STARTUP_EMBED_DELAY_SECS";
const STARTUP_CRYSTALLIZE_DELAY_ENV: &str = "CORTEX_STARTUP_CRYSTALLIZE_DELAY_SECS";
const STARTUP_STORAGE_GOVERNOR_DELAY_ENV: &str = "CORTEX_STARTUP_STORAGE_GOVERNOR_DELAY_SECS";
const BACKGROUND_DB_LOCK_MAX_WAIT_MS_ENV: &str = "CORTEX_BACKGROUND_DB_LOCK_MAX_WAIT_MS";
const EMBED_BACKFILL_DRAIN_ON_STARTUP_ENV: &str = "CORTEX_EMBED_BACKFILL_DRAIN_ON_STARTUP";
const EMBED_BACKFILL_STARTUP_DRAIN_MAX_BATCHES_ENV: &str =
    "CORTEX_EMBED_BACKFILL_STARTUP_DRAIN_MAX_BATCHES";
const IDLE_SHUTDOWN_SECS_ENV: &str = "CORTEX_IDLE_SHUTDOWN_SECS";
const IDLE_SHUTDOWN_MIN_UPTIME_SECS_ENV: &str = "CORTEX_IDLE_SHUTDOWN_MIN_UPTIME_SECS";
const STARTUP_LOG_FILES: &[&str] = &[
    "daemon.log",
    "daemon.err.log",
    "daemon.out.log",
    "mcp-crash.log",
    "rust-daemon.err.log",
];

// ── Backup rotation helpers ───────────────────────────────────────────────

/// Check if a backup should be created (>24h since last backup).
fn should_backup(backup_dir: &Path) -> bool {
    let last_backup_file = backup_dir.join(".last_backup");
    if !last_backup_file.exists() {
        return true;
    }
    match std::fs::read_to_string(&last_backup_file) {
        Ok(ts) => {
            if let Ok(last_backup) = chrono::DateTime::parse_from_rfc3339(&ts) {
                let now = Utc::now();
                // Convert FixedOffset to UTC for subtraction
                let last_utc = last_backup.with_timezone(&Utc);
                let hours_since_last = (now - last_utc).num_hours();
                hours_since_last >= 24
            } else {
                true
            }
        }
        Err(_) => true,
    }
}

/// Rotate backups to keep only the most recent N.
fn rotate_backups(backup_dir: &Path, keep: usize) -> Result<usize, std::io::Error> {
    let mut backups: Vec<_> = std::fs::read_dir(backup_dir)
        .map(|entries| {
            entries
                .filter_map(|entry| entry.ok())
                .filter(|entry| {
                    entry.file_name().to_string_lossy().starts_with("cortex-")
                        && entry.file_name().to_string_lossy().ends_with(".db")
                        && !entry.file_name().to_string_lossy().contains(".corrupt")
                })
                .collect()
        })
        .unwrap_or_default();

    if backups.len() <= keep {
        return Ok(0);
    }

    // Sort by modification time (oldest first)
    backups.sort_by_key(|entry| entry.metadata().ok().and_then(|m| m.modified().ok()));

    let mut removed = 0usize;
    for backup in backups.iter().take(backups.len() - keep) {
        std::fs::remove_file(backup.path())?;
        removed += 1;
    }

    Ok(removed)
}

fn cleanup_backup_retention(backup_dir: &Path) -> usize {
    match rotate_backups(backup_dir, BACKUP_RETENTION_COUNT) {
        Ok(removed) => removed,
        Err(e) => {
            eprintln!("[cortex] Warning: backup rotation failed: {e}");
            0
        }
    }
}

fn cleanup_bridge_backups(home: &Path, schema_version: i32) -> bool {
    if schema_version < BRIDGE_BACKUP_CLEANUP_SCHEMA_VERSION {
        return false;
    }

    let bridge_backup_dir = home.join("bridge-backups");
    if !bridge_backup_dir.exists() {
        return false;
    }

    match std::fs::remove_dir_all(&bridge_backup_dir) {
        Ok(()) => {
            eprintln!("[cortex] Removed legacy bridge-backups for schema version {schema_version}");
            true
        }
        Err(e) => {
            eprintln!("[cortex] Warning: failed to remove legacy bridge-backups: {e}");
            false
        }
    }
}

fn cleanup_expired_rows(conn: &rusqlite::Connection, label: &str) {
    match db::delete_expired_entries(conn) {
        Ok(counts) if counts.memories_deleted > 0 || counts.decisions_deleted > 0 => {
            eprintln!(
                "[cortex] {label}: deleted {} expired memories and {} expired decisions",
                counts.memories_deleted, counts.decisions_deleted
            );
        }
        Ok(_) => {}
        Err(e) => eprintln!("[cortex] Warning: expired-row cleanup failed: {e}"),
    }
}

fn rotate_log_file(home: &Path, file_name: &str) -> Result<bool, std::io::Error> {
    let log_path = home.join(file_name);
    let metadata = match std::fs::metadata(&log_path) {
        Ok(metadata) => metadata,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(err) => return Err(err),
    };

    if metadata.len() <= LOG_ROTATION_BYTES {
        return Ok(false);
    }

    let rotated_path = home.join(format!("{file_name}.1"));
    if rotated_path.exists() {
        std::fs::remove_file(&rotated_path)?;
    }
    std::fs::rename(&log_path, &rotated_path)?;
    std::fs::File::create(&log_path)?;
    Ok(true)
}

fn rotate_startup_logs(home: &Path) -> usize {
    let mut rotated = 0usize;
    for file_name in STARTUP_LOG_FILES {
        match rotate_log_file(home, file_name) {
            Ok(true) => {
                rotated += 1;
                eprintln!("[cortex] Rotated log file {file_name}");
            }
            Ok(false) => {}
            Err(e) => {
                eprintln!("[cortex] Warning: failed to rotate {file_name}: {e}");
            }
        }
    }
    rotated
}

fn collect_backup_cleanup_files(backup_dir: &Path, keep: usize) -> Vec<(std::path::PathBuf, u64)> {
    let mut backups: Vec<_> = std::fs::read_dir(backup_dir)
        .map(|entries| {
            entries
                .filter_map(|entry| entry.ok())
                .filter(|entry| {
                    entry.file_name().to_string_lossy().starts_with("cortex-")
                        && entry.file_name().to_string_lossy().ends_with(".db")
                        && !entry.file_name().to_string_lossy().contains(".corrupt")
                })
                .collect()
        })
        .unwrap_or_default();

    if backups.len() <= keep {
        return Vec::new();
    }

    backups.sort_by_key(|entry| entry.metadata().ok().and_then(|m| m.modified().ok()));
    let remove_count = backups.len() - keep;
    backups
        .into_iter()
        .take(remove_count)
        .map(|entry| {
            let size = entry.metadata().map(|meta| meta.len()).unwrap_or(0);
            (entry.path(), size)
        })
        .collect()
}

fn format_cleanup_bytes(bytes: u64) -> String {
    const KB: f64 = 1024.0;
    const MB: f64 = KB * 1024.0;

    if bytes >= MB as u64 {
        format!("{:.1} MB", bytes as f64 / MB)
    } else if bytes >= KB as u64 {
        format!("{:.1} KB", bytes as f64 / KB)
    } else {
        format!("{bytes} B")
    }
}

fn path_size_bytes(path: &Path) -> u64 {
    match std::fs::metadata(path) {
        Ok(meta) if meta.is_file() => meta.len(),
        Ok(meta) if meta.is_dir() => std::fs::read_dir(path)
            .map(|entries| {
                entries
                    .filter_map(|entry| entry.ok())
                    .map(|entry| path_size_bytes(&entry.path()))
                    .sum()
            })
            .unwrap_or(0),
        _ => 0,
    }
}

fn run_backup_cleanup(backup_dir: &Path, dry_run: bool) -> Vec<String> {
    let candidates = collect_backup_cleanup_files(backup_dir, BACKUP_RETENTION_COUNT);
    let mut lines = Vec::new();
    for (path, size) in candidates {
        let target = format!(
            "backups/{}",
            path.file_name()
                .and_then(|name| name.to_str())
                .unwrap_or_default()
        );
        lines.push(format!("DELETE {target} ({})", format_cleanup_bytes(size)));
        if !dry_run {
            let _ = std::fs::remove_file(path);
        }
    }
    lines
}

fn run_log_cleanup(home: &Path, dry_run: bool) -> Vec<String> {
    let mut lines = Vec::new();
    for file_name in STARTUP_LOG_FILES {
        let log_path = home.join(file_name);
        let metadata = match std::fs::metadata(&log_path) {
            Ok(metadata) => metadata,
            Err(_) => continue,
        };
        if metadata.len() <= LOG_ROTATION_BYTES {
            continue;
        }

        lines.push(format!(
            "ROTATE {file_name} ({})",
            format_cleanup_bytes(metadata.len())
        ));

        if dry_run {
            continue;
        }

        let rotated_path = home.join(format!("{file_name}.1"));
        if rotated_path.exists() {
            let _ = std::fs::remove_file(&rotated_path);
        }
        if std::fs::rename(&log_path, &rotated_path).is_ok() {
            let _ = std::fs::File::create(&log_path);
        }
    }
    lines
}

fn run_bridge_backup_cleanup(home: &Path, schema_version: i32, dry_run: bool) -> Vec<String> {
    if schema_version < BRIDGE_BACKUP_CLEANUP_SCHEMA_VERSION {
        return Vec::new();
    }

    let bridge_dir = home.join("bridge-backups");
    if !bridge_dir.exists() {
        return Vec::new();
    }

    let size = path_size_bytes(&bridge_dir);
    let line = format!("DELETE bridge-backups/ ({})", format_cleanup_bytes(size));
    if !dry_run {
        let _ = std::fs::remove_dir_all(&bridge_dir);
    }
    vec![line]
}

fn run_stale_pid_cleanup(paths: &auth::CortexPaths, dry_run: bool) -> Vec<String> {
    let Some(pid) = auth::stale_pid_candidate(paths) else {
        return Vec::new();
    };

    let lines = vec![format!("DELETE cortex.pid (process {pid} not running)")];

    if !dry_run {
        let _ = auth::cleanup_stale_pid_lock(paths);
    }

    lines
}

/// Create a backup of the database file.
fn create_backup(db_path: &Path, backup_dir: &Path) -> Result<String, String> {
    std::fs::create_dir_all(backup_dir).map_err(|e| format!("create backup dir: {e}"))?;

    let timestamp = chrono::Local::now().format("%Y%m%d");
    let dest = backup_dir.join(format!("cortex-{timestamp}.db"));

    // Copy the DB file (not move - preserves original)
    std::fs::copy(db_path, &dest).map_err(|e| format!("copy db: {e}"))?;

    eprintln!("[cortex] Backup created: {}", dest.display());

    // Rotate old backups after creating a fresh backup.
    let _ = cleanup_backup_retention(backup_dir);

    // Update last backup timestamp
    let last_backup_file = backup_dir.join(".last_backup");
    let now_ts = chrono::Utc::now().to_rfc3339();
    if let Err(e) = std::fs::write(&last_backup_file, now_ts) {
        eprintln!("[cortex] Warning: failed to write last_backup timestamp: {e}");
    }

    Ok(dest.to_string_lossy().to_string())
}

#[tokio::main]
async fn main() {
    let args: Vec<String> = std::env::args().collect();
    let mode = args.get(1).map(|s| s.as_str()).unwrap_or("");
    let paths = auth::CortexPaths::resolve_from_args(&args);
    if let Ok(current_exe) = std::env::current_exe() {
        if is_disallowed_startup_binary_path(&current_exe) {
            eprintln!(
                "[cortex] Refusing to run from disallowed runtime path: {}",
                current_exe.display()
            );
            std::process::exit(1);
        }
    }

    match mode {
        "" | "--help" | "-h" | "help" => {
            print_usage_and_exit(0);
        }
        "--version" | "-V" | "version" => {
            println!("cortex {}", env!("CARGO_PKG_VERSION"));
        }
        "capabilities" => {
            if args.iter().any(|arg| arg == "--json") {
                println!(
                    "{}",
                    serde_json::to_string_pretty(&cli_capabilities_payload()).unwrap()
                );
            } else {
                println!("{}", cli_capabilities_summary());
            }
        }
        "robot-docs" => {
            let subcmd = args.get(2).map(String::as_str).unwrap_or("guide");
            match subcmd {
                "" | "guide" | "--help" | "-h" => println!("{}", cli_robot_docs_guide()),
                other => {
                    eprintln!("{}", unknown_robot_docs_subcommand_message(other));
                    std::process::exit(1);
                }
            }
        }

        // ── HTTP daemon (standalone or via service) ─────────────────
        "serve" => {
            #[cfg(unix)]
            async fn sigterm_future() {
                use tokio::signal::unix::{signal, SignalKind};
                let mut sigterm =
                    signal(SignalKind::terminate()).expect("Failed to register SIGTERM handler");
                sigterm.recv().await;
            }
            #[cfg(not(unix))]
            async fn sigterm_future() {
                std::future::pending::<()>().await;
            }

            run_daemon(paths.clone(), async {
                tokio::select! {
                    _ = tokio::signal::ctrl_c() => {
                        eprintln!("[cortex] Received Ctrl+C, shutting down...");
                    }
                    _ = sigterm_future() => {
                        eprintln!("[cortex] Received SIGTERM, shutting down...");
                    }
                }
            })
            .await;
        }

        // ── MCP stdio transport ─────────────────────────────────────
        "mcp" => {
            let remaining = &args[2..];
            let agent = parse_flag_value(remaining, "--agent");
            let (base_url, api_key, local_owner_mode) = resolve_client_target(remaining, &paths);
            if let Err(e) = ensure_remote_target_has_api_key(&base_url, api_key.as_deref(), &paths)
            {
                eprintln!("[cortex-mcp] {e}");
                std::process::exit(1);
            }
            if local_owner_mode {
                apply_path_env(&paths);
                match ensure_daemon(&paths, agent.as_deref(), false, false).await {
                    Ok(()) => {}
                    Err(e) => {
                        eprintln!("[cortex-mcp] {e}");
                        std::process::exit(1);
                    }
                }
            }
            if let Err(e) = mcp_proxy::run(&base_url, api_key.as_deref(), agent.as_deref()).await {
                eprintln!("[cortex-mcp] {e}");
                std::process::exit(1);
            }
        }

        "paths" => {
            if args.iter().any(|a| a == "--json") {
                println!("{}", paths.to_json());
            } else {
                eprintln!("Usage: cortex paths --json");
                std::process::exit(1);
            }
        }

        "boot" => {
            let remaining: Vec<String> = args[2..].to_vec();
            if let Err(e) = run_boot_cli(&paths, &remaining).await {
                eprintln!("Error: {e}");
                std::process::exit(1);
            }
        }

        "plugin" => {
            let subcmd = args.get(2).map(|s| s.as_str()).unwrap_or("");
            match subcmd {
                "ensure-daemon" => {
                    let agent = parse_flag_value(&args[3..], "--agent");
                    apply_path_env(&paths);
                    if let Err(e) = ensure_daemon(&paths, agent.as_deref(), true, true).await {
                        eprintln!("Error: {e}");
                        std::process::exit(1);
                    }
                }
                "mcp" => {
                    let remaining = &args[3..];
                    let (base_url, api_key, local_owner_mode) =
                        resolve_client_target(remaining, &paths);
                    let agent = parse_flag_value(remaining, "--agent");
                    if let Err(e) =
                        ensure_remote_target_has_api_key(&base_url, api_key.as_deref(), &paths)
                    {
                        eprintln!("[cortex-plugin] {e}");
                        std::process::exit(1);
                    }
                    if local_owner_mode {
                        apply_path_env(&paths);
                        match ensure_daemon(&paths, agent.as_deref(), false, true).await {
                            Ok(()) => {}
                            Err(e) => {
                                eprintln!("[cortex-plugin] {e}");
                                std::process::exit(1);
                            }
                        }
                    }
                    if let Err(e) =
                        mcp_proxy::run(&base_url, api_key.as_deref(), agent.as_deref()).await
                    {
                        eprintln!("[cortex-plugin] {e}");
                        std::process::exit(1);
                    }
                }
                _ => {
                    eprintln!("Usage: cortex plugin <ensure-daemon|mcp>");
                    std::process::exit(1);
                }
            }
        }

        // ── Hook: SessionStart (replaces brain-boot.js) ─────────────
        "hook-boot" => {
            let agent = args
                .get(2)
                .and_then(|a| {
                    if a == "--agent" {
                        args.get(3).map(|s| s.as_str())
                    } else {
                        Some(a.as_str())
                    }
                })
                .unwrap_or("claude-opus");
            hook_boot::run_boot(agent).await;
        }

        // ── Hook: Statusline one-liner ──────────────────────────────
        "hook-status" => {
            hook_boot::run_status().await;
        }

        // ── Windows Service lifecycle ───────────────────────────────
        "service" => {
            let subcmd = args.get(2).cloned().unwrap_or_default();
            if let Err(err) = tokio::task::spawn_blocking(move || match subcmd.as_str() {
                "install" => service::install(),
                "uninstall" => service::uninstall(),
                "start" => service::start(),
                "stop" => service::stop(),
                "status" => service::status(),
                "ensure" => service::ensure(),
                _ => {
                    eprintln!("Usage: cortex service <install|uninstall|start|stop|status|ensure>");
                }
            })
            .await
            {
                eprintln!("[cortex] Service command task failed: {err}");
                std::process::exit(1);
            }
        }

        // ── Windows Service entry point (called by SCM) ─────────────
        "service-run" => {
            service::dispatch_service();
        }

        // ── System prompt injector CLI ──────────────────────────────
        "prompt-inject" => {
            let remaining: Vec<String> = args[2..].to_vec();
            prompt_inject::run(&remaining).await;
        }

        // ── Setup: detect AI tools, configure, verify ──────────────
        "setup" => {
            let remaining: Vec<String> = args[2..].to_vec();
            if remaining.iter().any(|a| a == "--team") {
                let dry_run = remaining.iter().any(|a| a == "--dry-run");
                setup::run_setup_team(&remaining, dry_run).await;
            } else {
                setup::run_setup().await;
            }
        }

        // ── Migrate: alias for setup --team with dry-run support ───
        "migrate" => {
            let remaining: Vec<String> = args[2..].to_vec();
            let dry_run = remaining.iter().any(|a| a == "--dry-run");
            setup::run_setup_team(&remaining, dry_run).await;
        }

        // ── Data export/import CLI ──────────────────────────────────
        "export" => {
            let remaining: Vec<String> = args[2..].to_vec();
            run_export_cli(&paths, &remaining);
        }
        "import" => {
            let remaining: Vec<String> = args[2..].to_vec();
            run_import_cli(&paths, &remaining);
        }
        "sync" => {
            let remaining: Vec<String> = args[2..].to_vec();
            run_sync_cli(&paths, &remaining);
        }
        "eval" => {
            let remaining: Vec<String> = args[2..].to_vec();
            run_eval_cli(&paths, &remaining);
        }
        "doctor" => {
            run_doctor_cli(&paths);
        }
        "reindex" => {
            let json_output = args.iter().any(|a| a == "--json");
            run_reindex_cli(&paths, json_output);
        }
        "re-embed" | "reembed" => {
            let mut remaining: Vec<String> = args[2..].to_vec();
            if !remaining.iter().any(|arg| arg == "--until-exhausted") {
                remaining.push("--until-exhausted".to_string());
            }
            run_embeddings_drain_cli(&paths, &remaining).await;
        }
        "recrystallize" => {
            let json_output = args.iter().any(|a| a == "--json");
            run_recrystallize_cli(&paths, json_output).await;
        }
        "cleanup" => {
            let dry_run = args.iter().any(|a| a == "--dry-run");
            let include_events = args.iter().any(|a| a == "--events");
            let max_event_passes = match parse_flag_usize(&args[2..], "--max-passes") {
                Ok(Some(value)) => value.clamp(1, 12),
                Ok(None) => 3,
                Err(err) => {
                    eprintln!("Error: {err}");
                    std::process::exit(1);
                }
            };
            run_cleanup_cli(&paths, dry_run, include_events, max_event_passes);
        }
        "embeddings" => {
            let remaining: Vec<String> = args[2..].to_vec();
            run_embeddings_cli(&paths, &remaining).await;
        }

        // ── Backup/restore CLI ────────────────────────────────────
        "backup" => {
            let db_path = paths.db.clone();
            let home_dir = paths.home.clone();
            // Force checkpoint before backup for consistency
            let conn = match db::open(&db_path) {
                Ok(c) => c,
                Err(e) => {
                    eprintln!("Error: failed to open database: {e}");
                    std::process::exit(1);
                }
            };
            db::checkpoint_wal_best_effort(&conn);
            drop(conn);

            let backup_dir = home_dir.join("backups");
            match create_backup(&db_path, &backup_dir) {
                Ok(path) => {
                    println!("Backup created: {path}");
                }
                Err(e) => {
                    eprintln!("Error: {e}");
                    std::process::exit(1);
                }
            }
        }
        "restore" => {
            let restore_file = match args.get(2) {
                Some(f) => f.clone(),
                None => {
                    eprintln!("Usage: cortex restore <backup-file.db>");
                    eprintln!("       cortex restore <backup-file.db> --skip-verification");
                    eprintln!();
                    eprintln!("Example: cortex restore ~/.cortex/backups/cortex-20260407.db");
                    std::process::exit(1);
                }
            };

            let skip_verification = args.iter().any(|a| a == "--skip-verification");

            // Check if daemon is running by checking PID file
            let paths_check = auth::CortexPaths::resolve();
            let daemon_running = paths_check.pid.exists();

            if daemon_running {
                eprintln!(
                    "[cortex] Warning: Daemon PID file exists at {}",
                    paths_check.pid.display()
                );
                eprintln!(
                    "[cortex] Please stop the daemon first with: Ctrl+C or kill the daemon process"
                );
                eprintln!("[cortex] Continuing restore anyway...");
                std::thread::sleep(Duration::from_millis(500));
            }

            let db_path = paths.db.clone();
            let home_dir = paths.home.clone();

            // Create a pre-restore backup
            let timestamp = chrono::Local::now().format("%Y%m%dT%H%M%S");
            let pre_backup = home_dir.join(format!("cortex.pre-restore.{}.db", timestamp));

            eprintln!(
                "[cortex] Creating pre-restore backup at: {}",
                pre_backup.display()
            );
            if let Err(e) = std::fs::copy(&db_path, &pre_backup) {
                eprintln!("[cortex] Error: failed to create pre-restore backup: {e}");
                eprintln!("[cortex] Restore cancelled for safety");
                std::process::exit(1);
            }

            // Restore from backup file
            eprintln!("[cortex] Restoring from: {}", restore_file);
            if let Err(e) = std::fs::copy(&restore_file, &db_path) {
                eprintln!("[cortex] Error: failed to restore backup: {e}");
                eprintln!(
                    "[cortex] Pre-restore backup preserved at: {}",
                    pre_backup.display()
                );
                std::process::exit(1);
            }

            // Verify integrity of restored DB
            if !skip_verification {
                eprintln!("[cortex] Verifying integrity of restored database...");
                match db::open(&db_path) {
                    Ok(conn) => {
                        if !db::verify_integrity(&conn).unwrap_or(false) {
                            eprintln!("[cortex] Error: restored database failed integrity check!");
                            eprintln!("[cortex] Rolling back to pre-restore backup...");
                            if let Err(e) = std::fs::copy(&pre_backup, &db_path) {
                                eprintln!(
                                    "[cortex] Critical: rollback failed! DB may be corrupted: {e}"
                                );
                            } else {
                                eprintln!("[cortex] Rollback complete");
                            }
                            std::process::exit(1);
                        }
                        eprintln!("[cortex] Integrity check passed");
                    }
                    Err(e) => {
                        eprintln!("[cortex] Error: failed to open restored database: {e}");
                        eprintln!("[cortex] Rolling back to pre-restore backup...");
                        if let Err(e) = std::fs::copy(&pre_backup, &db_path) {
                            eprintln!(
                                "[cortex] Critical: rollback failed! DB may be corrupted: {e}"
                            );
                        } else {
                            eprintln!("[cortex] Rollback complete");
                        }
                        std::process::exit(1);
                    }
                }
            }

            eprintln!(
                "[cortex] Restore complete. Pre-restore backup preserved at: {}",
                pre_backup.display()
            );
            eprintln!("[cortex] You can now restart the daemon with: cortex serve");
        }

        // ── User management CLI ────────────────────────────────────
        "user" => {
            let subcmd = args.get(2).map(|s| s.as_str()).unwrap_or("");
            match subcmd {
                "add" => {
                    let username = match args.get(3) {
                        Some(u) => u.clone(),
                        None => {
                            eprintln!(
                                "Usage: cortex user add <username> [--role member|admin] [--display-name \"...\"]"
                            );
                            std::process::exit(1);
                        }
                    };
                    let mut role = "member".to_string();
                    let mut display_name: Option<String> = None;
                    let mut i = 4usize;
                    while i < args.len() {
                        match args[i].as_str() {
                            "--role" => {
                                if let Some(v) = args.get(i + 1) {
                                    role = v.clone();
                                    i += 1;
                                }
                            }
                            "--display-name" => {
                                if let Some(v) = args.get(i + 1) {
                                    display_name = Some(v.clone());
                                    i += 1;
                                }
                            }
                            _ => {}
                        }
                        i += 1;
                    }
                    let mut body = serde_json::json!({
                        "username": username,
                        "role": role,
                    });
                    if let Some(dn) = display_name {
                        body["display_name"] = serde_json::json!(dn);
                    }
                    match admin_request("POST", "/admin/user/add", Some(body)).await {
                        Ok(json) => {
                            let api_key = json_str(&json, "api_key");
                            let key_masked = api_key_output_masked();
                            println!("User created:");
                            println!("  Username:  {}", json_str(&json, "username"));
                            println!("  User ID:   {}", json_field(&json, "user_id"));
                            println!("  Role:      {}", json_str(&json, "role"));
                            println!("  API Key:   {}", format_api_key_for_output(&api_key));
                            if key_masked {
                                println!(
                                    "  NOTE: API key is masked because stdout is non-interactive."
                                );
                                println!(
                                    "        Re-run this command in a terminal to display the full key."
                                );
                            }
                            println!();
                            println!("Save the API key -- it cannot be retrieved later.");
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "rotate-key" => {
                    let username = match args.get(3) {
                        Some(u) => u.clone(),
                        None => {
                            eprintln!("Usage: cortex user rotate-key <username>");
                            std::process::exit(1);
                        }
                    };
                    let body = serde_json::json!({ "username": username });
                    match admin_request("POST", "/admin/user/rotate-key", Some(body)).await {
                        Ok(json) => {
                            let api_key = json_str(&json, "api_key");
                            let key_masked = api_key_output_masked();
                            println!("API key rotated for '{}':", json_str(&json, "username"));
                            println!("  New API Key: {}", format_api_key_for_output(&api_key));
                            if key_masked {
                                println!(
                                    "  NOTE: API key is masked because stdout is non-interactive."
                                );
                                println!(
                                    "        Re-run this command in a terminal to display the full key."
                                );
                            }
                            println!();
                            println!("Save the API key -- it cannot be retrieved later.");
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "remove" => {
                    let username = match args.get(3) {
                        Some(u) => u.clone(),
                        None => {
                            eprintln!("Usage: cortex user remove <username>");
                            std::process::exit(1);
                        }
                    };
                    if !confirm_action(&format!("Remove user '{username}'?")) {
                        eprintln!("Cancelled.");
                        std::process::exit(0);
                    }
                    let body = serde_json::json!({ "username": username });
                    match admin_request("POST", "/admin/user/remove", Some(body)).await {
                        Ok(json) => {
                            println!("Removed user '{}'", json_str(&json, "removed"));
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "list" => match admin_request("GET", "/admin/users", None).await {
                    Ok(json) => {
                        let users = json["users"].as_array();
                        match users {
                            Some(arr) if !arr.is_empty() => {
                                println!(
                                    "{:<6} {:<20} {:<20} {:<10} CREATED",
                                    "ID", "USERNAME", "DISPLAY NAME", "ROLE"
                                );
                                println!("{}", "-".repeat(80));
                                for u in arr {
                                    println!(
                                        "{:<6} {:<20} {:<20} {:<10} {}",
                                        json_field(u, "id"),
                                        json_str(u, "username"),
                                        json_str_or(u, "display_name", "-"),
                                        json_str(u, "role"),
                                        json_str_or(u, "created_at", "-"),
                                    );
                                }
                                println!();
                                println!("{} user(s)", arr.len());
                            }
                            _ => println!("No users found."),
                        }
                    }
                    Err(e) => {
                        eprintln!("Error: {e}");
                        std::process::exit(1);
                    }
                },
                _ => {
                    eprintln!("Usage: cortex user <add|rotate-key|remove|list>");
                    std::process::exit(1);
                }
            }
        }

        // ── Team management CLI ────────────────────────────────────
        "team" => {
            let subcmd = args.get(2).map(|s| s.as_str()).unwrap_or("");
            match subcmd {
                "create" => {
                    let name = match args.get(3) {
                        Some(n) => n.clone(),
                        None => {
                            eprintln!("Usage: cortex team create <name>");
                            std::process::exit(1);
                        }
                    };
                    let body = serde_json::json!({ "name": name });
                    match admin_request("POST", "/admin/team/create", Some(body)).await {
                        Ok(json) => {
                            println!("Team created:");
                            println!("  Name:    {}", json_str(&json, "name"));
                            println!("  Team ID: {}", json_field(&json, "team_id"));
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "add" => {
                    let team_name = match args.get(3) {
                        Some(t) => t.clone(),
                        None => {
                            eprintln!(
                                "Usage: cortex team add <team> <username> [--role member|admin]"
                            );
                            std::process::exit(1);
                        }
                    };
                    let username = match args.get(4) {
                        Some(u) => u.clone(),
                        None => {
                            eprintln!(
                                "Usage: cortex team add <team> <username> [--role member|admin]"
                            );
                            std::process::exit(1);
                        }
                    };
                    let mut role = "member".to_string();
                    let mut i = 5usize;
                    while i < args.len() {
                        if args[i] == "--role" {
                            if let Some(v) = args.get(i + 1) {
                                role = v.clone();
                                i += 1;
                            }
                        }
                        i += 1;
                    }
                    let body = serde_json::json!({
                        "team_name": team_name,
                        "username": username,
                        "role": role,
                    });
                    match admin_request("POST", "/admin/team/add-member", Some(body)).await {
                        Ok(json) => {
                            println!(
                                "Added '{}' to team '{}' as {}",
                                json_str(&json, "username"),
                                json_str(&json, "team"),
                                json_str(&json, "role"),
                            );
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "remove" => {
                    let team_name = match args.get(3) {
                        Some(t) => t.clone(),
                        None => {
                            eprintln!("Usage: cortex team remove <team> <username>");
                            std::process::exit(1);
                        }
                    };
                    let username = match args.get(4) {
                        Some(u) => u.clone(),
                        None => {
                            eprintln!("Usage: cortex team remove <team> <username>");
                            std::process::exit(1);
                        }
                    };
                    if !confirm_action(&format!("Remove '{username}' from team '{team_name}'?")) {
                        eprintln!("Cancelled.");
                        std::process::exit(0);
                    }
                    let body = serde_json::json!({
                        "team_name": team_name,
                        "username": username,
                    });
                    match admin_request("POST", "/admin/team/remove-member", Some(body)).await {
                        Ok(json) => {
                            let removed = &json["removed"];
                            println!(
                                "Removed '{}' from team '{}'",
                                json_str(removed, "username"),
                                json_str(removed, "team"),
                            );
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "list" => match admin_request("GET", "/admin/teams", None).await {
                    Ok(json) => {
                        let teams = json["teams"].as_array();
                        match teams {
                            Some(arr) if !arr.is_empty() => {
                                println!("{:<6} {:<30} {:<10} CREATED", "ID", "NAME", "MEMBERS");
                                println!("{}", "-".repeat(70));
                                for t in arr {
                                    println!(
                                        "{:<6} {:<30} {:<10} {}",
                                        json_field(t, "id"),
                                        json_str(t, "name"),
                                        json_field(t, "member_count"),
                                        json_str_or(t, "created_at", "-"),
                                    );
                                }
                                println!();
                                println!("{} team(s)", arr.len());
                            }
                            _ => println!("No teams found."),
                        }
                    }
                    Err(e) => {
                        eprintln!("Error: {e}");
                        std::process::exit(1);
                    }
                },
                _ => {
                    eprintln!("Usage: cortex team <create|add|remove|list>");
                    std::process::exit(1);
                }
            }
        }

        // ── Admin management CLI ───────────────────────────────────
        "admin" => {
            let subcmd = args.get(2).map(|s| s.as_str()).unwrap_or("");
            match subcmd {
                "list-unowned" => match admin_request("GET", "/admin/unowned", None).await {
                    Ok(json) => {
                        let unowned = json["unowned"].as_object();
                        match unowned {
                            Some(map) if !map.is_empty() => {
                                println!("{:<25} UNOWNED ROWS", "TABLE");
                                println!("{}", "-".repeat(40));
                                let mut total: i64 = 0;
                                for (table, count) in map {
                                    let n = count.as_i64().unwrap_or(0);
                                    total += n;
                                    println!("{:<25} {}", table, n);
                                }
                                println!("{}", "-".repeat(40));
                                println!("{:<25} {}", "TOTAL", total);
                            }
                            _ => println!("No unowned data found."),
                        }
                    }
                    Err(e) => {
                        eprintln!("Error: {e}");
                        std::process::exit(1);
                    }
                },
                "assign-owner" => {
                    let mut from_user: Option<String> = None;
                    let mut to_user: Option<String> = None;
                    let mut table: Option<String> = None;
                    let mut i = 3usize;
                    while i < args.len() {
                        match args[i].as_str() {
                            "--from" => {
                                if let Some(v) = args.get(i + 1) {
                                    from_user = Some(v.clone());
                                    i += 1;
                                }
                            }
                            "--to" => {
                                if let Some(v) = args.get(i + 1) {
                                    to_user = Some(v.clone());
                                    i += 1;
                                }
                            }
                            "--table" => {
                                if let Some(v) = args.get(i + 1) {
                                    table = Some(v.clone());
                                    i += 1;
                                }
                            }
                            _ => {}
                        }
                        i += 1;
                    }
                    let Some(to) = to_user else {
                        eprintln!(
                            "Usage: cortex admin assign-owner [--from <user>] --to <user> [--table <table>]"
                        );
                        std::process::exit(1);
                    };
                    let mut body = serde_json::json!({ "to_user": to });
                    if let Some(from) = from_user {
                        body["from_user"] = serde_json::json!(from);
                    }
                    if let Some(t) = table {
                        body["table"] = serde_json::json!(t);
                    }
                    match admin_request("POST", "/admin/assign-owner", Some(body)).await {
                        Ok(json) => {
                            let assigned = json["assigned"].as_object();
                            match assigned {
                                Some(map) if !map.is_empty() => {
                                    println!("{:<25} ROWS ASSIGNED", "TABLE");
                                    println!("{}", "-".repeat(40));
                                    let mut total: i64 = 0;
                                    for (tbl, count) in map {
                                        let n = count.as_i64().unwrap_or(0);
                                        total += n;
                                        println!("{:<25} {}", tbl, n);
                                    }
                                    println!("{}", "-".repeat(40));
                                    println!("{:<25} {}", "TOTAL", total);
                                }
                                _ => println!("No rows assigned."),
                            }
                        }
                        Err(e) => {
                            eprintln!("Error: {e}");
                            std::process::exit(1);
                        }
                    }
                }
                "stats" => match admin_request("GET", "/admin/stats", None).await {
                    Ok(json) => {
                        println!("Cortex Admin Stats");
                        println!("{}", "=".repeat(50));
                        println!();
                        println!(
                            "Users: {}    Teams: {}    DB Size: {}",
                            json_field(&json, "user_count"),
                            json_field(&json, "team_count"),
                            json_str_or(&json, "db_size_mb", "?"),
                        );
                        println!();

                        if let Some(tables) = json["tables"].as_object() {
                            println!("{:<25} ROWS", "TABLE");
                            println!("{}", "-".repeat(40));
                            for (tbl, count) in tables {
                                println!("{:<25} {}", tbl, count);
                            }
                        }

                        if let Some(per_user) = json["per_user"].as_array() {
                            if !per_user.is_empty() {
                                println!();
                                println!("Per-User Breakdown:");
                                println!(
                                    "  {:<20} {:<10} {:<10} CRYSTALS",
                                    "USERNAME", "MEMORIES", "DECISIONS"
                                );
                                println!("  {}", "-".repeat(55));
                                for u in per_user {
                                    println!(
                                        "  {:<20} {:<10} {:<10} {}",
                                        json_str(u, "username"),
                                        json_field(u, "memories"),
                                        json_field(u, "decisions"),
                                        json_field(u, "crystals"),
                                    );
                                }
                            }
                        }
                    }
                    Err(e) => {
                        eprintln!("Error: {e}");
                        std::process::exit(1);
                    }
                },
                "budgets" => {
                    run_admin_budgets_cli(&paths, &args[3..]);
                }
                "rollback" => {
                    run_admin_rollback_cli(&paths, &args[3..]);
                }
                _ => {
                    eprintln!(
                        "Usage: cortex admin <list-unowned|assign-owner|stats|budgets|rollback>"
                    );
                    std::process::exit(1);
                }
            }
        }
        other => {
            eprintln!("{}", unknown_cli_command_message(other));
            std::process::exit(1);
        }
    }
}

fn run_admin_budgets_cli(paths: &auth::CortexPaths, args: &[String]) {
    let subcmd = args.first().map(String::as_str).unwrap_or("");
    let json_output = args.iter().any(|arg| arg == "--json");
    match subcmd {
        "status" => {
            let status = budgets::BudgetConfigStatus::load_from_home(&paths.home);
            let payload = status.to_health_json(0);
            if json_output {
                println!("{}", serde_json::to_string_pretty(&payload).unwrap());
                return;
            }
            print_budget_status_human(&payload);
        }
        "validate" => {
            let Some(path) = parse_flag_value(args, "--path") else {
                eprintln!("Usage: cortex admin budgets validate --path <file> [--json]");
                std::process::exit(1);
            };
            let status = budgets::BudgetConfigStatus::load_from_path(path);
            let mut payload = status.to_health_json(0);
            if !status.config_loaded && status.error.is_none() {
                payload["error"] = json!({
                    "code": "not_found",
                    "message": "budget config file was not found",
                    "endpoint": null,
                    "field": null
                });
            }
            if json_output {
                println!("{}", serde_json::to_string_pretty(&payload).unwrap());
            } else {
                print_budget_status_human(&payload);
            }
            if payload["error"].is_object() {
                std::process::exit(1);
            }
        }
        _ => {
            eprintln!("Usage: cortex admin budgets <status|validate --path <file>> [--json]");
            std::process::exit(1);
        }
    }
}

fn print_budget_status_human(payload: &Value) {
    println!("Cortex Budget Governance");
    println!("{}", "=".repeat(50));
    println!("Source: {}", json_str(payload, "source"));
    println!("Config loaded: {}", json_field(payload, "configLoaded"));
    println!("Enabled: {}", json_field(payload, "enabled"));
    if let Some(error) = payload.get("error").and_then(Value::as_object) {
        println!(
            "Error: {} ({})",
            error
                .get("message")
                .and_then(Value::as_str)
                .unwrap_or("unknown error"),
            error
                .get("code")
                .and_then(Value::as_str)
                .unwrap_or("unknown")
        );
        return;
    }
    if let Some(endpoints) = payload.get("endpoints").and_then(Value::as_object) {
        if endpoints.is_empty() {
            println!("Endpoints: unlimited");
            return;
        }
        println!();
        println!("{:<12} {:<10} WINDOW", "ENDPOINT", "LIMIT");
        println!("{}", "-".repeat(36));
        for (endpoint, budget) in endpoints {
            println!(
                "{:<12} {:<10} {}s",
                endpoint,
                budget.get("limit").and_then(Value::as_u64).unwrap_or(0),
                budget
                    .get("windowSeconds")
                    .and_then(Value::as_u64)
                    .unwrap_or(0)
            );
        }
    }
}

fn cli_usage_text() -> String {
    format!(
        r#"Cortex v{} -- Universal AI Memory Daemon

Usage:
  cortex <command>
  cortex help
  cortex capabilities --json
  cortex robot-docs guide

Options:
  --version, -V      Print CLI version
  --help, -h         Print this help

Agent surfaces:
  capabilities --json  Print a deterministic machine-readable CLI contract
  robot-docs guide     Print a short operator guide for coding agents

Setup:
  setup              First-run setup: detect AI tools, configure, verify
  setup --team       Team-mode setup + schema migration + owner API key
  migrate            Alias for setup --team (solo -> team migration)
  migrate --dry-run  Preview migration without modifying the database

Daemon:
  serve [--bind <addr>]  HTTP daemon on :{} (default bind 127.0.0.1)
  mcp [--url <base>] [--api-key <key>] [--agent <name>]  MCP stdio
  paths --json       Print resolved Cortex paths + port + bind as JSON
  boot [--agent <name>] [--budget <n>] [--json] [--url <base>] [--api-key <key>]
  plugin ensure-daemon [--agent <name>]  Ensure daemon is running, then print port
  plugin mcp [--url <base>] [--api-key <key>] [--agent <name>]

Hooks:
  hook-boot [AGENT]  SessionStart hook (default: claude-opus)
  hook-status        Statusline one-liner

Tools:
  prompt-inject      Inject Cortex context into system prompt files
  export             Export data (--format json|sql, --out <file>)
  import             Import JSON data (--file <path>, optional --user <username>)
  sync export        Export changeset JSON (--out <file>, optional --since <iso>)
  sync import        Import a sync changeset (--file <path>, optional --user/--visibility)
  sync watch         Watched-folder sync loop (--dir <path>, optional --once)
  eval               Run retrieval evaluation; supports --json and regression flags
  doctor             Validate DB schema, migrations, integrity, and FTS health
  reindex [--json]   Fully rebuild FTS indexes from canonical rows
  re-embed [...]     Alias for embeddings drain --until-exhausted
  recrystallize [--json]  Rebuild crystal graph and embeddings
  cleanup [--dry-run] [--events] [--max-passes <n>]
  backup             Create manual backup (stores in ~/.cortex/backups/)
  restore <file>     Restore from backup file (daemon must be stopped)
  admin rollback --session-id <id> [--apply] [--json]

Embeddings:
  embeddings status [--json]  Show active-model embedding backlog counts
  embeddings drain [--batch-size <n>] [--max-batches <n>] [--lock-wait-ms <n>] [--until-exhausted] [--json]

User Management (team mode):
  user add <name>    Add user [--role member|admin] [--display-name "..."]
  user rotate-key <name>  Rotate a user's API key
  user remove <name> Remove user (with confirmation)
  user list          List all users

Team Management (team mode):
  team create <name> Create a team
  team add <team> <user>  Add member [--role member|admin]
  team remove <team> <user>  Remove member (with confirmation)
  team list          List all teams

Admin (team mode):
  admin list-unowned List rows without an owner
  admin assign-owner [--from <user>] --to <user> [--table <t>]
  admin stats        Database and per-user statistics
  admin budgets status [--json]
  admin budgets validate --path <file> [--json]

Service:
  service install    Register as Windows Service (manual start by default)
  service uninstall  Remove Windows Service
  service start      Start the service
  service stop       Stop the service
  service status     Check service status
  service ensure     Ensure service is installed, running, and healthy

Troubleshooting:
  cortex doctor      Validate DB schema, migrations, integrity, and FTS state
  cortex boot        Preferred local boot path (auto-adds auth + SSRF headers)
  HTTP 403           Add header: X-Cortex-Request: true
  HTTP 401           Use Authorization: Bearer <token> from ~/.cortex/cortex.token
  MCP not visible    Restart the client after adding MCP servers; they do not hot-attach mid-session
  App-hosted daemon  Restart the daemon from Cortex Control Center instead of stopping/starting it manually
  More help          See Info/connecting.md for full connection and auth examples
"#,
        env!("CARGO_PKG_VERSION"),
        DEFAULT_CORTEX_PORT
    )
}

fn cli_capabilities_payload() -> Value {
    json!({
        "schema_version": 1,
        "contract_version": CLI_CAPABILITIES_CONTRACT_VERSION,
        "tool": {
            "name": "cortex",
            "version": env!("CARGO_PKG_VERSION"),
            "default_port": DEFAULT_CORTEX_PORT,
            "default_bind": "127.0.0.1"
        },
        "agent_entrypoints": [
            {
                "name": "help",
                "command": "cortex help",
                "output": "human",
                "side_effects": "none"
            },
            {
                "name": "capabilities",
                "command": "cortex capabilities --json",
                "output": "json",
                "side_effects": "none"
            },
            {
                "name": "robot-docs",
                "command": "cortex robot-docs guide",
                "output": "text",
                "side_effects": "none"
            }
        ],
        "commands": {
            "serve": {
                "usage": "cortex serve [--bind <addr>] [--port <n>]",
                "purpose": "Run the HTTP daemon",
                "output": "logs",
                "side_effects": "starts_daemon"
            },
            "mcp": {
                "usage": "cortex mcp [--url <base>] [--api-key <key>] [--agent <name>]",
                "purpose": "Run the MCP stdio proxy",
                "output": "stdio_json_rpc",
                "side_effects": "may_ensure_local_daemon"
            },
            "paths": {
                "usage": "cortex paths --json",
                "purpose": "Print resolved paths, port, and bind configuration",
                "output": "json",
                "side_effects": "none"
            },
            "boot": {
                "usage": "cortex boot [--agent <name>] [--budget <n>] [--json]",
                "purpose": "Preferred local boot path with auth and SSRF headers",
                "output": "human_or_json",
                "side_effects": "may_ensure_local_daemon"
            },
            "doctor": {
                "usage": "cortex doctor",
                "purpose": "Validate database schema, migrations, integrity, and FTS health",
                "output": "human",
                "side_effects": "reads_database"
            },
            "admin budgets": {
                "usage": "cortex admin budgets status [--json]",
                "purpose": "Inspect or validate budget governance configuration",
                "output": "human_or_json",
                "side_effects": "none"
            },
            "admin rollback": {
                "usage": "cortex admin rollback --session-id <id> [--apply] [--json]",
                "purpose": "Dry-run or apply soft-delete rollback for one session",
                "output": "human_or_json",
                "side_effects": "mutates_database_with_--apply"
            }
        },
        "environment": {
            "CORTEX_HOME": "Overrides the Cortex home directory",
            "CORTEX_PORT": "Overrides the daemon port",
            "CORTEX_BIND": "Overrides daemon bind address; defaults to localhost",
            "CORTEX_API_KEY": "Supplies API key for remote client commands",
            "CORTEX_API_BASE": "Supplies base URL for remote client commands",
            "NO_COLOR": "Requests plain output where color is supported"
        },
        "exit_codes": {
            "0": "success",
            "1": "user_input_or_runtime_error"
        },
        "dangerous_operations": [
            {
                "command": "cortex restore <file>",
                "gate": "requires explicit backup file and warns when daemon appears active"
            },
            {
                "command": "cortex admin rollback --session-id <id> --apply",
                "gate": "dry-run by default; --apply required to mutate"
            },
            {
                "command": "cortex user remove <name>",
                "gate": "interactive confirmation"
            },
            {
                "command": "cortex team remove <team> <user>",
                "gate": "interactive confirmation"
            }
        ],
        "recommended_agent_flow": [
            "Run `cortex capabilities --json` to discover supported surfaces.",
            "Use `cortex paths --json` before reading or writing Cortex files.",
            "Use `cortex boot --json` for local attachment when a daemon may be needed.",
            "Use JSON flags where available and treat non-zero exit as retryable only after inspecting stderr."
        ]
    })
}

fn cli_capabilities_summary() -> String {
    format!(
        "Cortex agent capabilities\n\
         JSON contract: cortex capabilities --json\n\
         Agent guide: cortex robot-docs guide\n\
         Core JSON commands: paths --json, boot --json, reindex --json, recrystallize --json, embeddings status --json, admin budgets status --json\n\
         Default daemon endpoint: http://127.0.0.1:{}\n\
         Exit codes: 0 success, 1 user-input or runtime error",
        DEFAULT_CORTEX_PORT
    )
}

fn cli_robot_docs_guide() -> &'static str {
    r#"Cortex robot guide

Discovery:
  cortex capabilities --json
  cortex help
  cortex paths --json

Local attach:
  cortex boot --json
  cortex mcp --agent codex

Health checks:
  cortex doctor
  cortex embeddings status --json
  cortex admin budgets status --json

Maintenance:
  cortex backup
  cortex cleanup --dry-run
  cortex reindex --json
  cortex recrystallize --json

Danger gates:
  cortex restore <file> warns if a daemon appears active.
  cortex admin rollback --session-id <id> is dry-run unless --apply is present.
  cortex user remove and cortex team remove ask for confirmation.

Output contract:
  Prefer commands with --json when present.
  Treat stderr as diagnostic text.
  Treat exit code 0 as success and exit code 1 as user-input or runtime failure.
"#
}

fn top_level_command_suggestion(command: &str) -> Option<&'static str> {
    let normalized = command.trim().to_ascii_lowercase().replace(['_', '-'], "");
    match normalized.as_str() {
        "capability" | "capabilitiesjson" | "caps" => Some("cortex capabilities --json"),
        "robotdoc" | "robotdocs" | "agentdoc" | "agentdocs" | "docs" => {
            Some("cortex robot-docs guide")
        }
        "path" => Some("cortex paths --json"),
        "budget" | "budgets" => Some("cortex admin budgets status --json"),
        "rollback" => Some("cortex admin rollback --session-id <id> [--apply] [--json]"),
        _ => None,
    }
}

fn unknown_cli_command_message(command: &str) -> String {
    let prefix = if command.starts_with('-') {
        format!("Unknown option: {command}")
    } else {
        format!("Unknown command: {command}")
    };
    match top_level_command_suggestion(command) {
        Some(suggestion) => format!(
            "{prefix}\nDid you mean: `{suggestion}`?\nRun `cortex help` or `cortex capabilities --json` for supported commands."
        ),
        None => format!(
            "{prefix}\nRun `cortex help` or `cortex capabilities --json` for supported commands."
        ),
    }
}

fn unknown_robot_docs_subcommand_message(subcommand: &str) -> String {
    format!("Unknown robot-docs command: {subcommand}\nDid you mean: `cortex robot-docs guide`?")
}

fn print_usage_and_exit(code: i32) -> ! {
    let usage = cli_usage_text();
    if code == 0 {
        print!("{usage}");
    } else {
        eprint!("{usage}");
    }
    std::process::exit(code);
}

fn event_type_count(conn: &rusqlite::Connection, event_type: &str) -> i64 {
    conn.query_row(
        "SELECT COUNT(*) FROM events WHERE type = ?1",
        rusqlite::params![event_type],
        |row| row.get::<_, i64>(0),
    )
    .unwrap_or(0)
}

fn top_event_type_counts(conn: &rusqlite::Connection, limit: usize) -> Vec<(String, i64)> {
    let mut statement = match conn.prepare(
        "SELECT type, COUNT(*) AS cnt FROM events GROUP BY type ORDER BY cnt DESC LIMIT ?1",
    ) {
        Ok(stmt) => stmt,
        Err(_) => return Vec::new(),
    };

    let rows = match statement.query_map([limit as i64], |row| {
        Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?))
    }) {
        Ok(rows) => rows,
        Err(_) => return Vec::new(),
    };

    rows.filter_map(Result::ok).collect()
}

fn run_event_compaction_cleanup(
    db_path: &Path,
    dry_run: bool,
    max_passes: usize,
) -> Result<Vec<String>, String> {
    if !db_path.exists() {
        return Ok(vec![
            "EVENTS skip: database file is missing; nothing to compact".to_string(),
        ]);
    }

    let conn = db::open(db_path).map_err(|e| format!("events cleanup open db: {e}"))?;
    db::configure(&conn).map_err(|e| format!("events cleanup configure db: {e}"))?;

    let before_nonboot = compaction::non_boot_event_count(&conn);
    let before_decision_stored = event_type_count(&conn, "decision_stored");
    let mut lines = vec![format!(
        "EVENTS before: pressure={} nonboot_rows={} decision_stored_rows={} (soft={} hard={})",
        compaction::classify_event_pressure(before_nonboot),
        before_nonboot,
        before_decision_stored,
        compaction::EVENT_NONBOOT_SOFT_LIMIT_ROWS,
        compaction::EVENT_NONBOOT_HARD_LIMIT_ROWS,
    )];

    let top_before = top_event_type_counts(&conn, 5);
    if !top_before.is_empty() {
        lines.push("EVENTS top types before:".to_string());
        for (event_type, count) in top_before {
            lines.push(format!("  {event_type:<24} {count}"));
        }
    }

    if dry_run {
        lines.push(format!(
            "EVENTS dry-run only: rerun with `cortex cleanup --events --max-passes {max_passes}` to apply compaction"
        ));
        return Ok(lines);
    }

    for pass in 1..=max_passes.max(1) {
        let nonboot_before_pass = compaction::non_boot_event_count(&conn);
        let maybe_result = compaction::run_compaction_governor(&conn);
        let Some(result) = maybe_result else {
            lines.push(format!(
                "EVENTS pass {pass}: no additional compaction needed (pressure={})",
                compaction::classify_event_pressure(nonboot_before_pass)
            ));
            break;
        };

        let nonboot_after_pass = compaction::non_boot_event_count(&conn);
        let pressure_after_pass = compaction::classify_event_pressure(nonboot_after_pass);
        lines.push(format!(
            "EVENTS pass {pass}: pruned events={} benchmark={} archived={} expired={} feedback={} | nonboot {} -> {} ({pressure_after_pass})",
            result.events_pruned,
            result.benchmark_pruned,
            result.archived_text_stripped,
            result.expired_pruned,
            result.feedback_aggregated,
            nonboot_before_pass,
            nonboot_after_pass,
        ));

        if nonboot_after_pass >= nonboot_before_pass || pressure_after_pass == "normal" {
            break;
        }
    }

    let after_nonboot = compaction::non_boot_event_count(&conn);
    let after_decision_stored = event_type_count(&conn, "decision_stored");
    lines.push(format!(
        "EVENTS after: pressure={} nonboot_rows={} decision_stored_rows={}",
        compaction::classify_event_pressure(after_nonboot),
        after_nonboot,
        after_decision_stored,
    ));

    let top_after = top_event_type_counts(&conn, 5);
    if !top_after.is_empty() {
        lines.push("EVENTS top types after:".to_string());
        for (event_type, count) in top_after {
            lines.push(format!("  {event_type:<24} {count}"));
        }
    }

    Ok(lines)
}

fn run_cleanup_cli(
    paths: &auth::CortexPaths,
    dry_run: bool,
    include_events: bool,
    max_event_passes: usize,
) {
    let schema_version = if paths.db.exists() {
        db::open(&paths.db)
            .and_then(|conn| db::current_schema_user_version(&conn))
            .unwrap_or_default()
    } else {
        0
    };

    let mut lines = Vec::new();
    lines.extend(run_backup_cleanup(&paths.home.join("backups"), dry_run));
    lines.extend(run_log_cleanup(&paths.home, dry_run));
    lines.extend(run_bridge_backup_cleanup(
        &paths.home,
        schema_version,
        dry_run,
    ));
    lines.extend(run_stale_pid_cleanup(paths, dry_run));
    if include_events {
        match run_event_compaction_cleanup(&paths.db, dry_run, max_event_passes) {
            Ok(event_lines) => lines.extend(event_lines),
            Err(err) => lines.push(format!("EVENTS cleanup failed: {err}")),
        }
    }

    if lines.is_empty() {
        println!("No cleanup actions needed");
        return;
    }

    for line in lines {
        println!("{line}");
    }
}

fn run_doctor_cli(paths: &auth::CortexPaths) {
    let db_path = paths.db.clone();
    println!("[doctor] db_path={}", db_path.display());

    let conn = match db::open(&db_path) {
        Ok(v) => v,
        Err(e) => {
            eprintln!("[doctor] FAIL open: {e}");
            std::process::exit(1);
        }
    };
    if let Err(e) = db::configure(&conn) {
        eprintln!("[doctor] FAIL configure: {e}");
        std::process::exit(1);
    }

    let expected_tables = [
        "memories",
        "decisions",
        "embeddings",
        "events",
        "co_occurrence",
        "locks",
        "activities",
        "messages",
        "sessions",
        "tasks",
        "feed",
        "feed_acks",
        "context_cache",
        "recall_feedback",
        "schema_migrations",
        "focus_sessions",
        "memory_clusters",
        "cluster_members",
        "memories_fts",
        "decisions_fts",
    ];

    let missing_tables: Vec<&str> = expected_tables
        .iter()
        .copied()
        .filter(|table| !db::table_exists(&conn, table))
        .collect();
    if missing_tables.is_empty() {
        println!(
            "[doctor] OK tables: {}/{}",
            expected_tables.len(),
            expected_tables.len()
        );
    } else {
        println!(
            "[doctor] FAIL tables missing: {}",
            missing_tables.join(", ")
        );
    }

    let (schema_current, pending_versions) = match db::pending_migration_versions(&conn) {
        Ok(pending) => (pending.is_empty(), pending),
        Err(e) => {
            println!("[doctor] FAIL schema status: {e}");
            (false, vec![])
        }
    };
    if schema_current {
        let expected_versions: HashSet<&'static str> = db::migration_definitions()
            .iter()
            .map(|(version, _)| *version)
            .collect();
        let (applied, marker_rows) = db::applied_migration_versions(&conn)
            .map(|versions| {
                let schema_applied = versions
                    .iter()
                    .filter(|version| expected_versions.contains(version.as_str()))
                    .count();
                let non_schema_markers = versions.len().saturating_sub(schema_applied);
                (schema_applied, non_schema_markers)
            })
            .unwrap_or((0, 0));
        println!(
            "[doctor] OK schema current: {applied}/{} migrations applied",
            db::migration_definitions().len()
        );
        if marker_rows > 0 {
            println!("[doctor] INFO schema markers: {marker_rows} non-schema row(s) ignored");
        }
    } else if !pending_versions.is_empty() {
        println!(
            "[doctor] FAIL schema pending: {}",
            pending_versions.join(", ")
        );
    }

    let integrity_ok = match db::verify_integrity(&conn) {
        Ok(true) => {
            println!("[doctor] OK integrity_check");
            true
        }
        Ok(false) => {
            println!("[doctor] FAIL integrity_check");
            false
        }
        Err(e) => {
            println!("[doctor] FAIL integrity_check error: {e}");
            false
        }
    };

    let fts_trigger_names = [
        "memories_fts_ai",
        "memories_fts_ad",
        "memories_fts_au",
        "decisions_fts_ai",
        "decisions_fts_ad",
        "decisions_fts_au",
    ];
    let fts_tables_ok =
        db::table_exists(&conn, "memories_fts") && db::table_exists(&conn, "decisions_fts");
    let fts_queries_ok = conn
        .query_row("SELECT COUNT(*) FROM memories_fts", [], |row| {
            row.get::<_, i64>(0)
        })
        .is_ok()
        && conn
            .query_row("SELECT COUNT(*) FROM decisions_fts", [], |row| {
                row.get::<_, i64>(0)
            })
            .is_ok();
    let fts_triggers_ok = fts_trigger_names.iter().all(|name| {
        conn.query_row(
            "SELECT 1 FROM sqlite_master WHERE type='trigger' AND name=?1 LIMIT 1",
            rusqlite::params![name],
            |_| Ok(()),
        )
        .is_ok()
    });
    let fts_ok = fts_tables_ok && fts_queries_ok && fts_triggers_ok;
    if fts_ok {
        println!("[doctor] OK fts indexes");
    } else {
        println!("[doctor] FAIL fts indexes");
    }

    let nonboot_event_rows = compaction::non_boot_event_count(&conn);
    let decision_stored_rows = event_type_count(&conn, "decision_stored");
    let event_pressure = compaction::classify_event_pressure(nonboot_event_rows);
    println!(
        "[doctor] EVENT pressure={} nonboot_rows={} decision_stored_rows={} (soft={} hard={})",
        event_pressure,
        nonboot_event_rows,
        decision_stored_rows,
        compaction::EVENT_NONBOOT_SOFT_LIMIT_ROWS,
        compaction::EVENT_NONBOOT_HARD_LIMIT_ROWS,
    );
    let top_event_types = top_event_type_counts(&conn, 5);
    if !top_event_types.is_empty() {
        println!("[doctor] EVENT top types:");
        for (event_type, count) in top_event_types {
            println!("  {:<24} {}", event_type, count);
        }
    }
    if event_pressure != "normal" {
        println!(
            "[doctor] WARN elevated event pressure detected; run `cortex cleanup --events --dry-run` to preview one-time remediation."
        );
    }

    let all_ok = missing_tables.is_empty() && schema_current && integrity_ok && fts_ok;
    if all_ok {
        println!("[doctor] GREEN");
        return;
    }

    println!("[doctor] RED");
    std::process::exit(1);
}

/// `cortex admin rollback --session-id <id> [--apply] [--json]`
///
/// Soft-deletes every memory + decision the session's agent wrote from
/// session start onward. Dry-run by default (reports counts, writes
/// nothing). Recall already filters `status='active'` so flipped rows
/// stop surfacing immediately.
///
/// When applied, writes a persistent `session.rolled_back` entry to the
/// `events` table so SSE subscribers + audit trails see the action. The
/// CLI runs offline; no live daemon connection is required.
fn run_admin_rollback_cli(paths: &auth::CortexPaths, args: &[String]) {
    let mut session_id: Option<String> = None;
    let mut apply = false;
    let mut json_output = false;
    let mut i = 0usize;
    while i < args.len() {
        match args[i].as_str() {
            "--session-id" | "--session" => {
                if let Some(v) = args.get(i + 1) {
                    session_id = Some(v.clone());
                    i += 1;
                }
            }
            "--apply" => apply = true,
            "--json" => json_output = true,
            "--help" | "-h" => {
                println!(
                    "Usage: cortex admin rollback --session-id <id> [--apply] [--json]\n\
                     \n\
                     Soft-deletes every memory + decision written by the session's\n\
                     agent since the session started. Dry-run by default; pass\n\
                     --apply to write. Idempotent."
                );
                std::process::exit(0);
            }
            other => {
                eprintln!("Unknown flag: {other}");
                eprintln!("Usage: cortex admin rollback --session-id <id> [--apply] [--json]");
                std::process::exit(1);
            }
        }
        i += 1;
    }

    let Some(session_id) = session_id else {
        eprintln!("Usage: cortex admin rollback --session-id <id> [--apply] [--json]");
        std::process::exit(1);
    };

    let conn = match db::open(&paths.db) {
        Ok(c) => c,
        Err(err) => {
            eprintln!("Error: failed to open database for rollback: {err}");
            std::process::exit(1);
        }
    };
    if let Err(err) = db::configure(&conn) {
        eprintln!("Error: failed to configure database for rollback: {err}");
        std::process::exit(1);
    }
    // Ensure tables exist on a freshly-created DB so `--session-id`
    // returns "not found" instead of "no such table".
    if let Err(err) = db::initialize_schema(&conn) {
        eprintln!("Error: failed to initialize schema: {err}");
        std::process::exit(1);
    }
    db::run_pending_migrations(&conn);

    let stats = match admin::rollback_session_by_id(&conn, &session_id, apply) {
        Ok(s) => s,
        Err(err) => {
            eprintln!("Error: rollback failed: {err}");
            std::process::exit(1);
        }
    };

    // On apply, persist a `session.rolled_back` event so SSE/audit see it
    // even when the daemon was offline at rollback time.
    if apply && !stats.agent.is_empty() {
        let payload = json!({
            "session_id": stats.session_id,
            "agent": stats.agent,
            "session_started_at": stats.session_started_at,
            "memories_affected": stats.memories_affected,
            "decisions_affected": stats.decisions_affected,
            "already_rolled_back": stats.already_rolled_back,
        });
        let _ = conn.execute(
            "INSERT INTO events(type, data, source_agent) VALUES (?1, ?2, ?3)",
            rusqlite::params!["session.rolled_back", payload.to_string(), stats.agent,],
        );
    }

    if json_output {
        println!(
            "{}",
            json!({
                "rollback": true,
                "applied": stats.applied,
                "session_id": stats.session_id,
                "agent": stats.agent,
                "session_started_at": stats.session_started_at,
                "memories_affected": stats.memories_affected,
                "decisions_affected": stats.decisions_affected,
                "already_rolled_back": stats.already_rolled_back,
            })
        );
    } else if stats.agent.is_empty() {
        eprintln!(
            "Session not found: '{session_id}'. The sessions table is keyed\n\
             by agent + current session_id; expired / superseded sessions\n\
             cannot be rolled back by id alone."
        );
        std::process::exit(1);
    } else {
        let label = if stats.applied { "applied" } else { "dry-run" };
        println!(
            "[rollback {label}] session={} agent={} started_at={}",
            stats.session_id, stats.agent, stats.session_started_at
        );
        println!(
            "  memories to flip: {}    decisions to flip: {}",
            stats.memories_affected, stats.decisions_affected
        );
        if stats.already_rolled_back {
            println!("  note: session already rolled back previously; nothing to do.");
        }
        if !stats.applied {
            println!("  Dry-run only. Pass --apply to persist.");
        }
    }

    std::process::exit(0);
}

fn run_reindex_cli(paths: &auth::CortexPaths, json_output: bool) {
    let conn = match db::open(&paths.db) {
        Ok(conn) => conn,
        Err(err) => {
            eprintln!("Error: failed to open database for reindex: {err}");
            std::process::exit(1);
        }
    };
    if let Err(err) = db::configure(&conn) {
        eprintln!("Error: failed to configure database for reindex: {err}");
        std::process::exit(1);
    }

    let memories_base = conn
        .query_row(
            "SELECT COUNT(*) FROM memories WHERE status = 'active'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .unwrap_or(0);
    let decisions_base = conn
        .query_row(
            "SELECT COUNT(*) FROM decisions WHERE status = 'active'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .unwrap_or(0);

    if let Err(err) = db::reindex_fts(&conn) {
        eprintln!("Error: failed to rebuild FTS indexes: {err}");
        std::process::exit(1);
    }

    let memories_fts = conn
        .query_row("SELECT COUNT(*) FROM memories_fts", [], |row| {
            row.get::<_, i64>(0)
        })
        .unwrap_or(0);
    let decisions_fts = conn
        .query_row("SELECT COUNT(*) FROM decisions_fts", [], |row| {
            row.get::<_, i64>(0)
        })
        .unwrap_or(0);

    if json_output {
        println!(
            "{}",
            json!({
                "reindexed": true,
                "counts": {
                    "memories_base": memories_base,
                    "memories_fts": memories_fts,
                    "decisions_base": decisions_base,
                    "decisions_fts": decisions_fts,
                }
            })
        );
        return;
    }

    println!("Reindex complete");
    println!("memories: base={memories_base}, fts={memories_fts}");
    println!("decisions: base={decisions_base}, fts={decisions_fts}");
}

async fn run_recrystallize_cli(paths: &auth::CortexPaths, json_output: bool) {
    let (state, _shutdown_rx) =
        state::initialize(paths, false).expect("Failed to initialize state for recrystallize");

    let result_payload = {
        let conn = state.db.lock().await;

        let crystals_before: i64 = conn
            .query_row("SELECT COUNT(*) FROM memory_clusters", [], |row| {
                row.get::<_, i64>(0)
            })
            .unwrap_or(0);
        let members_before: i64 = conn
            .query_row("SELECT COUNT(*) FROM cluster_members", [], |row| {
                row.get::<_, i64>(0)
            })
            .unwrap_or(0);
        let embeddings_before: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM embeddings WHERE target_type = 'crystal'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .unwrap_or(0);

        let removed_embeddings = conn
            .execute("DELETE FROM embeddings WHERE target_type = 'crystal'", [])
            .unwrap_or(0);
        let removed_crystals = conn.execute("DELETE FROM memory_clusters", []).unwrap_or(0);

        let brain_sender = Some(state.brain_firing.clone());
        let pass = crystallize::run_crystallize_pass_with_brain(
            &conn,
            state.embedding_engine.as_deref(),
            state.default_owner_id,
            &brain_sender,
        );

        let crystals_after: i64 = conn
            .query_row("SELECT COUNT(*) FROM memory_clusters", [], |row| {
                row.get::<_, i64>(0)
            })
            .unwrap_or(0);
        let members_after: i64 = conn
            .query_row("SELECT COUNT(*) FROM cluster_members", [], |row| {
                row.get::<_, i64>(0)
            })
            .unwrap_or(0);
        let embeddings_after: i64 = conn
            .query_row(
                "SELECT COUNT(*) FROM embeddings WHERE target_type = 'crystal'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .unwrap_or(0);

        json!({
            "recrystallized": true,
            "owner_filter": state.default_owner_id,
            "removed": {
                "crystals": removed_crystals,
                "members": members_before,
                "embeddings": removed_embeddings
            },
            "before": {
                "crystals": crystals_before,
                "members": members_before,
                "embeddings": embeddings_before
            },
            "pass": {
                "clusters_found": pass.clusters_found,
                "crystals_created": pass.crystals_created,
                "crystals_updated": pass.crystals_updated,
                "entries_consolidated": pass.entries_consolidated
            },
            "after": {
                "crystals": crystals_after,
                "members": members_after,
                "embeddings": embeddings_after
            }
        })
    };

    if json_output {
        println!("{result_payload}");
        return;
    }

    println!("Recrystallize complete");
    println!(
        "removed: crystals={}, members={}, crystal_embeddings={}",
        result_payload["removed"]["crystals"]
            .as_i64()
            .unwrap_or_default(),
        result_payload["removed"]["members"]
            .as_i64()
            .unwrap_or_default(),
        result_payload["removed"]["embeddings"]
            .as_i64()
            .unwrap_or_default()
    );
    println!(
        "pass: clusters_found={}, created={}, updated={}, consolidated={}",
        result_payload["pass"]["clusters_found"]
            .as_u64()
            .unwrap_or_default(),
        result_payload["pass"]["crystals_created"]
            .as_u64()
            .unwrap_or_default(),
        result_payload["pass"]["crystals_updated"]
            .as_u64()
            .unwrap_or_default(),
        result_payload["pass"]["entries_consolidated"]
            .as_u64()
            .unwrap_or_default()
    );
    println!(
        "after: crystals={}, members={}, crystal_embeddings={}",
        result_payload["after"]["crystals"]
            .as_i64()
            .unwrap_or_default(),
        result_payload["after"]["members"]
            .as_i64()
            .unwrap_or_default(),
        result_payload["after"]["embeddings"]
            .as_i64()
            .unwrap_or_default()
    );
}

async fn run_embeddings_cli(paths: &auth::CortexPaths, args: &[String]) {
    let subcmd = args.first().map(|s| s.as_str()).unwrap_or("");
    match subcmd {
        "status" => {
            let json_output = args.iter().any(|arg| arg == "--json");
            run_embeddings_status_cli(paths, json_output).await;
        }
        "drain" => {
            run_embeddings_drain_cli(paths, &args[1..]).await;
        }
        _ => {
            eprintln!(
                "Usage: cortex embeddings <status|drain> [--json] [--batch-size <n>] [--max-batches <n>] [--lock-wait-ms <n>] [--until-exhausted] [--max-iterations <n>]"
            );
            std::process::exit(1);
        }
    }
}

async fn run_embeddings_status_cli(paths: &auth::CortexPaths, json_output: bool) {
    let (state, _shutdown_rx) =
        state::initialize(paths, false).expect("Failed to initialize state for embeddings status");
    let Some(engine) = state.embedding_engine.clone() else {
        eprintln!(
            "[embeddings] No embedding model is currently loaded. Run `cortex serve` once to trigger model download, then retry."
        );
        std::process::exit(1);
    };

    let model_key = engine.model_key().to_string();
    let (backlog_memories, backlog_decisions) = {
        let conn = state.db.lock().await;
        count_unembedded_targets_for_model(&conn, &model_key)
    };
    let backlog_total = backlog_memories + backlog_decisions;

    if json_output {
        println!(
            "{}",
            json!({
                "model": model_key,
                "backlog": {
                    "memories": backlog_memories,
                    "decisions": backlog_decisions,
                    "total": backlog_total
                }
            })
        );
    } else {
        println!("Embeddings status");
        println!("model: {model_key}");
        println!(
            "backlog: memories={}, decisions={}, total={}",
            backlog_memories, backlog_decisions, backlog_total
        );
    }
}

async fn run_embeddings_drain_cli(paths: &auth::CortexPaths, args: &[String]) {
    let batch_size = match parse_flag_usize(args, "--batch-size") {
        Ok(Some(value)) => value.clamp(1, 10_000),
        Ok(None) => parse_env_usize(
            "CORTEX_EMBED_BACKFILL_BATCH_SIZE",
            DEFAULT_EMBED_BACKFILL_BATCH_SIZE,
        )
        .clamp(1, 10_000),
        Err(err) => {
            eprintln!("Error: {err}");
            std::process::exit(1);
        }
    };
    let max_batches_per_pass = match parse_flag_usize(args, "--max-batches") {
        Ok(Some(value)) => value.clamp(1, 10_000),
        Ok(None) => parse_env_usize(
            "CORTEX_EMBED_BACKFILL_MAX_BATCHES_PER_PASS",
            DEFAULT_EMBED_BACKFILL_MAX_BATCHES_PER_PASS,
        )
        .clamp(1, 10_000),
        Err(err) => {
            eprintln!("Error: {err}");
            std::process::exit(1);
        }
    };
    let lock_wait_ms = match parse_flag_usize(args, "--lock-wait-ms") {
        Ok(Some(value)) => value.clamp(100, 60_000),
        Ok(None) => (background_db_lock_max_wait().as_millis() as usize).clamp(100, 60_000),
        Err(err) => {
            eprintln!("Error: {err}");
            std::process::exit(1);
        }
    };
    let max_iterations = match parse_flag_usize(args, "--max-iterations") {
        Ok(Some(value)) => value.clamp(1, 1024),
        Ok(None) => 32,
        Err(err) => {
            eprintln!("Error: {err}");
            std::process::exit(1);
        }
    };
    let until_exhausted = args.iter().any(|arg| arg == "--until-exhausted");
    let json_output = args.iter().any(|arg| arg == "--json");
    let lock_wait = Duration::from_millis(lock_wait_ms as u64);

    let (state, _shutdown_rx) =
        state::initialize(paths, false).expect("Failed to initialize state for embeddings drain");
    let Some(engine) = state.embedding_engine.clone() else {
        eprintln!(
            "[embeddings] No embedding model is currently loaded. Run `cortex serve` once to trigger model download, then retry."
        );
        std::process::exit(1);
    };
    let model_key = engine.model_key().to_string();

    let mut iterations_ran = 0usize;
    let mut queued_total = 0usize;
    let mut computed_total = 0usize;
    let mut passes_ran = 0usize;
    let mut exhausted = false;

    while iterations_ran < max_iterations {
        iterations_ran += 1;
        let pass = build_embeddings_async(
            engine.clone(),
            &state.db,
            batch_size,
            max_batches_per_pass,
            lock_wait,
        )
        .await;
        queued_total += pass.queued_total;
        computed_total += pass.computed_total;
        passes_ran += pass.passes_ran;
        exhausted = pass.exhausted;

        if pass.exhausted || pass.queued_total == 0 || !until_exhausted {
            break;
        }
    }

    let (remaining_memories, remaining_decisions) = {
        let conn = state.db.lock().await;
        count_unembedded_targets_for_model(&conn, &model_key)
    };
    let remaining_total = remaining_memories + remaining_decisions;
    exhausted = exhausted || remaining_total == 0;

    if json_output {
        println!(
            "{}",
            json!({
                "model": model_key,
                "batch_size": batch_size,
                "max_batches_per_pass": max_batches_per_pass,
                "lock_wait_ms": lock_wait_ms,
                "until_exhausted": until_exhausted,
                "max_iterations": max_iterations,
                "iterations_ran": iterations_ran,
                "queued_total": queued_total,
                "computed_total": computed_total,
                "passes_ran": passes_ran,
                "remaining": {
                    "memories": remaining_memories,
                    "decisions": remaining_decisions,
                    "total": remaining_total
                },
                "exhausted": exhausted
            })
        );
    } else {
        println!("Embeddings drain");
        println!("model: {model_key}");
        println!(
            "drain: queued={}, built={}, passes={}, iterations={}",
            queued_total, computed_total, passes_ran, iterations_ran
        );
        println!(
            "remaining: memories={}, decisions={}, total={}",
            remaining_memories, remaining_decisions, remaining_total
        );
        println!("exhausted: {exhausted}");
    }

    if until_exhausted && !exhausted {
        eprintln!(
            "[embeddings] backlog still pending after {} iteration(s); rerun with higher --max-iterations or --max-batches",
            iterations_ran
        );
        std::process::exit(2);
    }
}

fn run_sync_cli(paths: &auth::CortexPaths, args: &[String]) {
    let Some(command) = args.first().map(|value| value.as_str()) else {
        eprintln!("Usage: cortex sync <export|import|watch> [options]");
        std::process::exit(1);
    };

    match command {
        "export" => run_sync_export_cli(paths, &args[1..]),
        "import" => run_sync_import_cli(paths, &args[1..]),
        "watch" => run_sync_watch_cli(paths, &args[1..]),
        _ => {
            eprintln!("Usage: cortex sync <export|import|watch> [options]");
            std::process::exit(1);
        }
    }
}

fn run_eval_cli(paths: &auth::CortexPaths, args: &[String]) {
    let json_output = args.iter().any(|arg| arg == "--json");
    let fail_on_regression = args.iter().any(|arg| arg == "--fail-on-regression");
    let baseline_file = parse_flag_value(args, "--baseline-file");
    let max_regression = match parse_flag_value(args, "--max-regression") {
        Some(raw) => {
            let parsed = raw
                .trim()
                .parse::<f64>()
                .map_err(|_| format!("invalid value for --max-regression: '{raw}'"))
                .unwrap_or_else(|err| {
                    eprintln!("{err}");
                    std::process::exit(1);
                });
            if !(0.0..=1.0).contains(&parsed) {
                eprintln!("--max-regression must be between 0.0 and 1.0");
                std::process::exit(1);
            }
            parsed
        }
        None => 0.10,
    };
    let window_days = match parse_flag_usize(args, "--window-days") {
        Ok(Some(value)) => value.min(180) as i64,
        Ok(None) => 30,
        Err(err) => {
            eprintln!("Invalid --window-days value: {err}");
            std::process::exit(1);
        }
    };

    let conn = match open_cli_connection(&paths.db) {
        Ok(conn) => conn,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };
    let mut snapshot = eval::build_eval_snapshot(&conn, window_days);
    let regression_gate = baseline_file.as_deref().map(|path| {
        let baseline_raw = std::fs::read_to_string(path).unwrap_or_else(|err| {
            eprintln!("Failed to read baseline snapshot file '{path}': {err}");
            std::process::exit(1);
        });
        let baseline_json: Value = serde_json::from_str(&baseline_raw).unwrap_or_else(|err| {
            eprintln!("Invalid baseline snapshot JSON in '{path}': {err}");
            std::process::exit(1);
        });
        eval::build_eval_regression_gate(&snapshot, &baseline_json, max_regression)
    });
    if let (Some(gate), Value::Object(map)) = (regression_gate.clone(), &mut snapshot) {
        map.insert("regressionGate".to_string(), gate);
    }
    if json_output {
        println!(
            "{}",
            serde_json::to_string_pretty(&snapshot).unwrap_or_else(|_| "{}".to_string())
        );
        if fail_on_regression
            && regression_gate
                .as_ref()
                .and_then(|gate| gate.get("ok"))
                .and_then(Value::as_bool)
                == Some(false)
        {
            std::process::exit(2);
        }
        return;
    }

    let totals = snapshot.get("totals").cloned().unwrap_or_else(|| json!({}));
    let window = snapshot.get("window").cloned().unwrap_or_else(|| json!({}));
    let signals = snapshot
        .get("signals")
        .cloned()
        .unwrap_or_else(|| json!({}));
    println!("Eval snapshot ({window_days}d)");
    println!(
        "active: memories={}, decisions={}, open_conflicts={}",
        totals
            .get("activeMemories")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0),
        totals
            .get("activeDecisions")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0),
        totals
            .get("openConflicts")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0)
    );
    println!(
        "window: conflicts={}, resolutions={}, recalls={}",
        window
            .get("recentConflicts")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0),
        window
            .get("recentResolutions")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0),
        window
            .get("recentRecallQueries")
            .and_then(serde_json::Value::as_i64)
            .unwrap_or(0)
    );
    println!(
        "signals: conflict_burden={:.4}, decay_burden={:.4}, resolution_velocity={:.4}, contradiction_rate={:.4}",
        signals
            .get("conflictBurden")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("decayBurden")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("resolutionVelocity")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("contradictionRate")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0)
    );
    println!(
        "task: success_rate={:.4}, first_pass={:.4}, median_time_ms={:.2}, retry_count={:.4}",
        signals
            .get("taskSuccessRate")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("firstPassSuccess")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("medianTimeToValidResultMs")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("retryCount")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0)
    );
    println!(
        "memory_quality: stale_hit_rate={:.4}, low_trust_hit_rate={:.4}, consensus_precision={:.4}",
        signals
            .get("staleMemoryHitRate")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("lowTrustHitRate")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0),
        signals
            .get("consensusPromotionPrecision")
            .and_then(serde_json::Value::as_f64)
            .unwrap_or(0.0)
    );
    if let Some(gate) = regression_gate {
        let gate_ok = gate.get("ok").and_then(Value::as_bool).unwrap_or(true);
        println!(
            "regression_gate: ok={}, max_regression={:.3}",
            gate_ok, max_regression
        );
        if fail_on_regression && !gate_ok {
            std::process::exit(2);
        }
    }
}

fn run_export_cli(paths: &auth::CortexPaths, args: &[String]) {
    let mut format = "json".to_string();
    let mut out_path: Option<String> = None;

    let mut i = 0usize;
    while i < args.len() {
        match args[i].as_str() {
            "--format" => {
                if let Some(v) = args.get(i + 1) {
                    format = v.to_string();
                    i += 1;
                }
            }
            "--out" => {
                if let Some(v) = args.get(i + 1) {
                    out_path = Some(v.to_string());
                    i += 1;
                }
            }
            _ => {}
        }
        i += 1;
    }

    let Some(export_format) = export_data::ExportFormat::parse(&format) else {
        eprintln!("Usage: cortex export --format json|sql [--out <path>]");
        std::process::exit(1);
    };

    let conn = match open_cli_connection(&paths.db) {
        Ok(conn) => conn,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };

    let output = match export_format {
        export_data::ExportFormat::Json => {
            let value = export_data::export_json_value(&conn);
            serde_json::to_string_pretty(&value).unwrap_or_else(|_| "{}".to_string())
        }
        export_data::ExportFormat::Sql => export_data::export_sql_text(&conn),
    };

    if let Some(path) = out_path {
        if let Err(e) = std::fs::write(&path, output) {
            eprintln!("Failed to write export file {path}: {e}");
            std::process::exit(1);
        }
        eprintln!("Exported to {path}");
    } else {
        println!("{output}");
    }
}

fn run_sync_export_cli(paths: &auth::CortexPaths, args: &[String]) {
    let out_path = parse_flag_value(args, "--out");
    let since_override = parse_flag_value(args, "--since");
    if let Some(since) = since_override.as_deref() {
        if chrono::DateTime::parse_from_rfc3339(since).is_err() {
            eprintln!(
                "Invalid --since value '{since}'. Use RFC3339 (for example 2026-04-19T00:00:00Z)."
            );
            std::process::exit(1);
        }
    }

    let cursor_file = parse_flag_value(args, "--cursor-file").map(PathBuf::from);
    let since = resolve_sync_since(since_override.as_deref(), cursor_file.as_deref());
    let conn = match open_cli_connection(&paths.db) {
        Ok(conn) => conn,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };

    let value = export_data::export_json_changeset_value(&conn, since.as_deref());
    let output = serde_json::to_string_pretty(&value).unwrap_or_else(|_| "{}".to_string());

    if let Some(path) = out_path {
        if let Err(e) = std::fs::write(&path, output) {
            eprintln!("Failed to write sync export file {path}: {e}");
            std::process::exit(1);
        }
        eprintln!("Sync export written to {path}");
    } else {
        println!("{output}");
    }

    if let Some(cursor_path) = cursor_file {
        if let Some(cursor) = value.get("cursor").and_then(serde_json::Value::as_str) {
            if let Err(err) = write_sync_cursor_file(&cursor_path, cursor) {
                eprintln!("{err}");
                std::process::exit(1);
            }
        }
    }
}

fn run_import_cli(paths: &auth::CortexPaths, args: &[String]) {
    let parsed = parse_import_cli_args(
        args,
        "Usage: cortex import --file <path> [--user <username>] [--visibility private|team|shared]",
    );
    let parsed = match parsed {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };
    let counts = match import_payload_from_file(paths, &parsed, "import-cli") {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };
    println!(
        "{{\"imported\":{{\"memories\":{},\"decisions\":{}}}}}",
        counts.memories, counts.decisions
    );
}

fn run_sync_import_cli(paths: &auth::CortexPaths, args: &[String]) {
    let parsed = parse_import_cli_args(
        args,
        "Usage: cortex sync import --file <path> [--user <username>] [--visibility private|team|shared]",
    );
    let parsed = match parsed {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };
    let counts = match import_payload_from_file(paths, &parsed, "sync-import-cli") {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };
    println!(
        "{{\"imported\":{{\"memories\":{},\"decisions\":{}}}}}",
        counts.memories, counts.decisions
    );
}

fn run_sync_watch_cli(paths: &auth::CortexPaths, args: &[String]) {
    let Some(dir_raw) = parse_flag_value(args, "--dir") else {
        eprintln!(
            "Usage: cortex sync watch --dir <path> [--interval-seconds <n>] [--once] [--user <username>] [--visibility private|team|shared] [--since <iso>] [--cursor-file <path>]"
        );
        std::process::exit(1);
    };
    let watch_dir = PathBuf::from(dir_raw);
    if !watch_dir.exists() {
        if let Err(e) = std::fs::create_dir_all(&watch_dir) {
            eprintln!(
                "Failed to create sync watch directory {}: {e}",
                watch_dir.display()
            );
            std::process::exit(1);
        }
    }
    if !watch_dir.is_dir() {
        eprintln!(
            "Sync watch path must be a directory: {}",
            watch_dir.display()
        );
        std::process::exit(1);
    }

    let interval_seconds = match parse_flag_usize(args, "--interval-seconds") {
        Ok(Some(value)) => value as u64,
        Ok(None) => 15,
        Err(err) => {
            eprintln!("Invalid --interval-seconds: {err}");
            std::process::exit(1);
        }
    };
    let once = args.iter().any(|arg| arg == "--once");
    let username = parse_flag_value(args, "--user");
    let visibility =
        parse_flag_value(args, "--visibility").unwrap_or_else(|| "private".to_string());
    if !matches!(visibility.as_str(), "private" | "team" | "shared") {
        eprintln!("Invalid --visibility value '{visibility}'. Use private|team|shared.");
        std::process::exit(1);
    }

    let mut bootstrap_since = parse_flag_value(args, "--since");
    if let Some(since) = bootstrap_since.as_deref() {
        if chrono::DateTime::parse_from_rfc3339(since).is_err() {
            eprintln!(
                "Invalid --since value '{since}'. Use RFC3339 (for example 2026-04-19T00:00:00Z)."
            );
            std::process::exit(1);
        }
    }

    let state_id = sync_watch_state_id(&watch_dir);
    let state_root = paths.home.join("runtime").join("sync-watch");
    let seen_file = state_root.join(format!("{state_id}.seen"));
    let default_cursor = state_root.join(format!("{state_id}.cursor"));
    let cursor_file = parse_flag_value(args, "--cursor-file")
        .map(PathBuf::from)
        .unwrap_or(default_cursor);

    let local_site_id = match ensure_sync_site_id(paths) {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };

    let mut seen = match load_sync_seen_set(&seen_file) {
        Ok(value) => value,
        Err(err) => {
            eprintln!("{err}");
            std::process::exit(1);
        }
    };

    loop {
        let candidates = match collect_sync_watch_import_candidates(&watch_dir, &local_site_id) {
            Ok(value) => value,
            Err(err) => {
                eprintln!("{err}");
                std::process::exit(1);
            }
        };

        let mut seen_dirty = false;
        for candidate in candidates {
            let Some(name) = candidate.file_name().and_then(|value| value.to_str()) else {
                continue;
            };
            if seen.contains(name) {
                continue;
            }
            let import_options = ImportCliArgs {
                file_path: candidate.clone(),
                username: username.clone(),
                visibility: visibility.clone(),
            };
            match import_payload_from_file(paths, &import_options, "sync-watch-import") {
                Ok(counts) => {
                    eprintln!(
                        "[sync watch] imported {} (memories={}, decisions={})",
                        candidate.display(),
                        counts.memories,
                        counts.decisions
                    );
                    seen.insert(name.to_string());
                    seen_dirty = true;
                }
                Err(err) => {
                    eprintln!(
                        "[sync watch] import skipped for {}: {}",
                        candidate.display(),
                        err
                    );
                }
            }
        }
        if seen_dirty {
            if let Err(err) = write_sync_seen_set(&seen_file, &seen) {
                eprintln!("{err}");
                std::process::exit(1);
            }
        }

        let since = read_sync_cursor_file(&cursor_file).or_else(|| bootstrap_since.take());
        let conn = match open_cli_connection(&paths.db) {
            Ok(conn) => conn,
            Err(err) => {
                eprintln!("{err}");
                std::process::exit(1);
            }
        };
        let changeset = export_data::export_json_changeset_value(&conn, since.as_deref());
        let memories_count = changeset
            .get("memories_count")
            .and_then(serde_json::Value::as_u64)
            .unwrap_or(0);
        let decisions_count = changeset
            .get("decisions_count")
            .and_then(serde_json::Value::as_u64)
            .unwrap_or(0);
        let total = memories_count + decisions_count;
        if total > 0 {
            let filename = format!(
                "changeset-{}-{}.json",
                local_site_id,
                chrono::Utc::now().format("%Y%m%dT%H%M%S%3fZ")
            );
            let out_path = watch_dir.join(filename);
            let output =
                serde_json::to_string_pretty(&changeset).unwrap_or_else(|_| "{}".to_string());
            if let Err(err) = std::fs::write(&out_path, output) {
                eprintln!(
                    "Failed to write sync watch export {}: {err}",
                    out_path.display()
                );
                std::process::exit(1);
            }
            eprintln!(
                "[sync watch] exported {} (memories={}, decisions={})",
                out_path.display(),
                memories_count,
                decisions_count
            );
        }
        if let Some(cursor) = changeset.get("cursor").and_then(serde_json::Value::as_str) {
            if let Err(err) = write_sync_cursor_file(&cursor_file, cursor) {
                eprintln!("{err}");
                std::process::exit(1);
            }
        }

        if once {
            break;
        }
        std::thread::sleep(Duration::from_secs(interval_seconds.max(1)));
    }
}

#[derive(Debug, Clone)]
struct ImportCliArgs {
    file_path: PathBuf,
    username: Option<String>,
    visibility: String,
}

fn parse_import_cli_args(args: &[String], usage: &str) -> Result<ImportCliArgs, String> {
    let mut file_path: Option<String> = None;
    let mut username: Option<String> = None;
    let mut visibility = "private".to_string();

    let mut i = 0usize;
    while i < args.len() {
        match args[i].as_str() {
            "--file" => {
                if let Some(v) = args.get(i + 1) {
                    file_path = Some(v.to_string());
                    i += 1;
                }
            }
            "--user" => {
                if let Some(v) = args.get(i + 1) {
                    username = Some(v.to_string());
                    i += 1;
                }
            }
            "--visibility" => {
                if let Some(v) = args.get(i + 1) {
                    visibility = v.to_string();
                    i += 1;
                }
            }
            _ => {}
        }
        i += 1;
    }

    let Some(file_path) = file_path else {
        return Err(usage.to_string());
    };
    if !matches!(visibility.as_str(), "private" | "team" | "shared") {
        return Err(format!(
            "Invalid --visibility value '{visibility}'. Use private|team|shared."
        ));
    }

    Ok(ImportCliArgs {
        file_path: PathBuf::from(file_path),
        username,
        visibility,
    })
}

fn open_cli_connection(db_path: &Path) -> Result<rusqlite::Connection, String> {
    let conn = db::open(db_path)
        .map_err(|e| format!("Failed to open database at {}: {e}", db_path.display()))?;
    db::configure(&conn).map_err(|e| format!("Failed to configure database: {e}"))?;
    db::initialize_schema(&conn).map_err(|e| format!("Failed to initialize schema: {e}"))?;
    crystallize::migrate_crystal_tables(&conn);
    Ok(conn)
}

fn import_payload_from_file(
    paths: &auth::CortexPaths,
    parsed: &ImportCliArgs,
    source_agent_fallback: &str,
) -> Result<export_data::ImportCounts, String> {
    let file_display = parsed.file_path.display().to_string();
    let raw = std::fs::read_to_string(&parsed.file_path)
        .map_err(|e| format!("Cannot read import file {file_display}: {e}"))?;
    let payload: export_data::ImportPayload =
        serde_json::from_str(&raw).map_err(|e| format!("Import file is not valid JSON: {e}"))?;

    let mut conn = open_cli_connection(&paths.db)?;
    let team_mode = db::current_mode(&conn) == "team";
    if parsed.username.is_some() && !team_mode {
        return Err("--user import requires team mode. Run: cortex setup --team".to_string());
    }

    let owner_id = if team_mode {
        if let Some(user) = parsed.username.as_ref() {
            match conn.query_row(
                "SELECT id FROM users WHERE username = ?1",
                rusqlite::params![user.clone()],
                |row| row.get::<_, i64>(0),
            ) {
                Ok(id) => Some(id),
                Err(_) => {
                    return Err(format!(
                        "Unknown user '{user}'. Create the user before import."
                    ));
                }
            }
        } else {
            conn.query_row(
                "SELECT value FROM config WHERE key = 'owner_user_id' LIMIT 1",
                [],
                |row| row.get::<_, String>(0),
            )
            .ok()
            .and_then(|v| v.parse::<i64>().ok())
            .or_else(|| {
                conn.query_row(
                    "SELECT id FROM users ORDER BY CASE role WHEN 'owner' THEN 0 ELSE 1 END, id ASC LIMIT 1",
                    [],
                    |row| row.get::<_, i64>(0),
                )
                .ok()
            })
        }
    } else {
        None
    };
    if team_mode && owner_id.is_none() {
        return Err(
            "Team mode import requires a target owner. Run `cortex setup --team` first."
                .to_string(),
        );
    }

    let options = export_data::ImportOptions {
        owner_id,
        visibility: if team_mode {
            Some(parsed.visibility.clone())
        } else {
            None
        },
        source_agent_fallback: source_agent_fallback.to_string(),
    };
    export_data::import_payload(&mut conn, &payload, &options)
}

fn resolve_sync_since(override_since: Option<&str>, cursor_file: Option<&Path>) -> Option<String> {
    override_since
        .map(str::to_string)
        .or_else(|| cursor_file.and_then(read_sync_cursor_file))
}

fn read_sync_cursor_file(path: &Path) -> Option<String> {
    std::fs::read_to_string(path).ok().and_then(|raw| {
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_string())
        }
    })
}

fn write_sync_cursor_file(path: &Path, cursor: &str) -> Result<(), String> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| {
            format!(
                "Failed to prepare cursor directory {}: {e}",
                parent.display()
            )
        })?;
    }
    std::fs::write(path, format!("{cursor}\n"))
        .map_err(|e| format!("Failed to write cursor file {}: {e}", path.display()))
}

fn ensure_sync_site_id(paths: &auth::CortexPaths) -> Result<String, String> {
    let site_id_path = paths.home.join("site_id");
    if let Ok(existing) = std::fs::read_to_string(&site_id_path) {
        let candidate = sanitize_sync_site_id(existing.trim());
        if !candidate.is_empty() {
            return Ok(candidate);
        }
    }

    if let Some(parent) = site_id_path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| {
            format!(
                "Failed to create sync site-id directory {}: {e}",
                parent.display()
            )
        })?;
    }
    let created = uuid::Uuid::new_v4().to_string();
    std::fs::write(&site_id_path, format!("{created}\n")).map_err(|e| {
        format!(
            "Failed to persist sync site-id {}: {e}",
            site_id_path.display()
        )
    })?;
    Ok(created)
}

fn sanitize_sync_site_id(raw: &str) -> String {
    raw.chars()
        .filter(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_'))
        .collect()
}

fn sync_watch_state_id(watch_dir: &Path) -> String {
    use std::hash::{Hash, Hasher};
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    watch_dir.to_string_lossy().to_string().hash(&mut hasher);
    format!("{:016x}", hasher.finish())
}

fn is_sync_changeset_file_name(name: &str) -> bool {
    name.starts_with("changeset-") && name.ends_with(".json")
}

fn collect_sync_watch_import_candidates(
    watch_dir: &Path,
    local_site_id: &str,
) -> Result<Vec<PathBuf>, String> {
    let mut files = Vec::new();
    let local_prefix = format!("changeset-{local_site_id}-");
    let entries = std::fs::read_dir(watch_dir).map_err(|e| {
        format!(
            "Failed to read sync watch directory {}: {e}",
            watch_dir.display()
        )
    })?;
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|value| value.to_str()) else {
            continue;
        };
        if !is_sync_changeset_file_name(name) {
            continue;
        }
        if name.starts_with(&local_prefix) {
            continue;
        }
        files.push(path);
    }
    files.sort();
    Ok(files)
}

fn load_sync_seen_set(path: &Path) -> Result<HashSet<String>, String> {
    let mut seen = HashSet::new();
    if !path.exists() {
        return Ok(seen);
    }
    let raw = std::fs::read_to_string(path)
        .map_err(|e| format!("Failed to read sync watch state {}: {e}", path.display()))?;
    for line in raw.lines() {
        let trimmed = line.trim();
        if !trimmed.is_empty() {
            seen.insert(trimmed.to_string());
        }
    }
    Ok(seen)
}

fn write_sync_seen_set(path: &Path, seen: &HashSet<String>) -> Result<(), String> {
    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| {
            format!(
                "Failed to create sync watch state directory {}: {e}",
                parent.display()
            )
        })?;
    }
    let mut rows: Vec<&str> = seen.iter().map(String::as_str).collect();
    rows.sort_unstable();
    std::fs::write(path, rows.join("\n"))
        .map_err(|e| format!("Failed to write sync watch state {}: {e}", path.display()))
}

fn parse_flag_value(args: &[String], flag: &str) -> Option<String> {
    args.iter()
        .position(|a| a == flag)
        .and_then(|idx| args.get(idx + 1))
        .cloned()
}

fn env_trimmed(key: &str) -> Option<String> {
    std::env::var(key)
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

fn parse_truthy_flag(value: &str) -> bool {
    matches!(
        value.trim().to_ascii_lowercase().as_str(),
        "1" | "true" | "yes" | "on"
    )
}

fn single_daemon_test_bypass_enabled() -> bool {
    cfg!(debug_assertions)
        && std::env::var(SINGLE_DAEMON_TEST_BYPASS_ENV)
            .ok()
            .is_some_and(|value| parse_truthy_flag(&value))
}

fn normalize_option(value: Option<&str>) -> Option<String> {
    value
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(str::to_string)
}

fn local_daemon_base_url(paths: &auth::CortexPaths) -> String {
    transport::local_http_base_url(paths)
}

fn is_local_client_base_url(base_url: &str, paths: &auth::CortexPaths) -> bool {
    transport::is_local_http_base_url(base_url, paths)
}

fn resolve_client_target_inputs(
    override_url: Option<&str>,
    override_api_key: Option<&str>,
    env_base_url: Option<&str>,
    env_api_key: Option<&str>,
    default_base_url: &str,
) -> (String, Option<String>, bool) {
    let resolved_base_url =
        normalize_option(override_url).or_else(|| normalize_option(env_base_url));
    let resolved_api_key =
        normalize_option(override_api_key).or_else(|| normalize_option(env_api_key));
    let local_owner_mode = resolved_base_url.is_none() && resolved_api_key.is_none();
    let base_url = resolved_base_url.unwrap_or_else(|| default_base_url.to_string());
    (base_url, resolved_api_key, local_owner_mode)
}

fn resolve_client_target(
    args: &[String],
    paths: &auth::CortexPaths,
) -> (String, Option<String>, bool) {
    let override_url = parse_flag_value(args, "--url");
    let override_api_key = parse_flag_value(args, "--api-key");
    let env_base_url = env_trimmed("CORTEX_API_BASE").or_else(|| env_trimmed("CORTEX_BASE_URL"));
    let env_api_key = env_trimmed("CORTEX_API_KEY");
    resolve_client_target_inputs(
        override_url.as_deref(),
        override_api_key.as_deref(),
        env_base_url.as_deref(),
        env_api_key.as_deref(),
        &local_daemon_base_url(paths),
    )
}

fn ensure_remote_target_has_api_key(
    base_url: &str,
    api_key: Option<&str>,
    paths: &auth::CortexPaths,
) -> Result<(), String> {
    let parsed = reqwest::Url::parse(base_url).map_err(|_| {
        format!("Invalid Cortex target URL '{base_url}'. Use an absolute http:// or https:// URL.")
    })?;
    if !matches!(parsed.scheme(), "http" | "https") {
        return Err(format!(
            "Unsupported Cortex target URL scheme '{}' in '{base_url}'. Use http or https.",
            parsed.scheme()
        ));
    }
    if parsed.host_str().is_none() {
        return Err(format!(
            "Invalid Cortex target URL '{base_url}': missing host."
        ));
    }
    if !parsed.username().is_empty() || parsed.password().is_some() {
        return Err(
            "Cortex target URL must not include embedded credentials; pass --api-key instead."
                .to_string(),
        );
    }
    if parsed.query().is_some() || parsed.fragment().is_some() {
        return Err(
            "Cortex target URL must not include query parameters or fragments.".to_string(),
        );
    }

    if api_key.is_none() && !is_local_client_base_url(base_url, paths) {
        return Err(format!(
            "Remote Cortex target '{base_url}' requires an API key. Pass --api-key <key> or set CORTEX_API_KEY."
        ));
    }
    Ok(())
}

fn apply_path_env(paths: &auth::CortexPaths) {
    std::env::set_var("CORTEX_HOME", &paths.home);
    std::env::set_var("CORTEX_DB", &paths.db);
    std::env::set_var("CORTEX_PORT", paths.port.to_string());
    std::env::set_var("CORTEX_BIND", &paths.bind);
    match &paths.ipc_endpoint {
        Some(endpoint) => std::env::set_var("CORTEX_IPC_ENDPOINT", endpoint),
        None => std::env::remove_var("CORTEX_IPC_ENDPOINT"),
    }
}

fn parse_flag_usize(args: &[String], flag: &str) -> Result<Option<usize>, String> {
    let Some(idx) = args.iter().position(|a| a == flag) else {
        return Ok(None);
    };

    let raw = args
        .get(idx + 1)
        .ok_or_else(|| format!("missing value for {flag}"))?;
    let value = raw
        .parse::<usize>()
        .map_err(|_| format!("invalid value for {flag}: '{raw}'"))?;
    if value == 0 {
        return Err(format!("{flag} must be >= 1"));
    }
    Ok(Some(value))
}

fn parse_env_usize(key: &str, default: usize) -> usize {
    std::env::var(key)
        .ok()
        .and_then(|raw| raw.trim().parse::<usize>().ok())
        .filter(|value| *value > 0)
        .unwrap_or(default)
}

fn parse_env_u64(key: &str, default: u64) -> u64 {
    std::env::var(key)
        .ok()
        .and_then(|raw| raw.trim().parse::<u64>().ok())
        .filter(|value| *value > 0)
        .unwrap_or(default)
}

#[cfg(not(windows))]
use daemon_lifecycle::issue_owner_token_for_spawn;
use daemon_lifecycle::{
    daemon_healthy, is_cortex_health_payload, readiness_state_from_payload,
    validate_spawned_owner_claim, wait_for_health, DAEMON_OWNER_TOKEN_ENV,
    SPAWN_PARENT_START_TIME_ENV,
};
const DAEMON_STARTUP_WAIT_SECS: u64 = 90;
const DEFAULT_BOOT_BUDGET: usize = 600;
const DEFAULT_DAEMON_LOCK_WAIT_SECS: u64 = 15;
const DAEMON_LOCK_RETRY_INTERVAL_MS: u64 = 100;
const DAEMON_LOCK_HANDOFF_GRACE_SECS: u64 = 3;
const DAEMON_LOCAL_SPAWN_ENV: &str = "CORTEX_DAEMON_OWNER_LOCAL_SPAWN";
const APP_REQUIRED_ENV: &str = "CORTEX_APP_REQUIRED";
const APP_CLIENT_ENV: &str = "CORTEX_APP_CLIENT";
const APP_MANAGED_STARTUP_DELAY_ENV: &str = "CORTEX_APP_MANAGED_STARTUP_DELAY_SECS";

fn read_auth_token_from_path(token_path: &std::path::Path) -> Option<String> {
    std::fs::read_to_string(token_path).ok().and_then(|token| {
        let trimmed = token.trim();
        if trimmed.is_empty() {
            None
        } else {
            Some(trimmed.to_string())
        }
    })
}

fn resolve_boot_auth_header(
    token_path: &std::path::Path,
    api_key: Option<&str>,
    allow_local_token_fallback: bool,
) -> Option<String> {
    if let Some(api_key) = api_key {
        let trimmed = api_key.trim();
        if !trimmed.is_empty() {
            return Some(format!("Bearer {trimmed}"));
        }
    }
    if allow_local_token_fallback {
        return read_auth_token_from_path(token_path).map(|token| format!("Bearer {token}"));
    }
    None
}

async fn request_boot_payload(
    paths: &auth::CortexPaths,
    base_url: &str,
    token_path: &std::path::Path,
    api_key: Option<&str>,
    allow_local_token_fallback: bool,
    agent: &str,
    budget: usize,
) -> Result<serde_json::Value, String> {
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .map_err(|e| format!("create boot client: {e}"))?;

    let mut boot_url = reqwest::Url::parse(&format!("{}/boot", base_url.trim_end_matches('/')))
        .map_err(|e| format!("invalid boot URL '{base_url}': {e}"))?;
    boot_url
        .query_pairs_mut()
        .append_pair("agent", agent)
        .append_pair("budget", &budget.to_string());

    let mut headers = vec![
        ("x-cortex-request".to_string(), "true".to_string()),
        ("x-source-agent".to_string(), agent.to_string()),
    ];
    if let Some(auth) = resolve_boot_auth_header(token_path, api_key, allow_local_token_fallback) {
        headers.push(("authorization".to_string(), auth));
    }

    let (status, body) = transport::request_url_with_local_ipc_fallback(
        &client,
        "GET",
        boot_url.as_ref(),
        paths,
        &headers,
        None,
        Duration::from_secs(10),
    )
    .await
    .map_err(|e| format!("boot request failed: {e}"))?;
    if !status.is_success() {
        let detail = body.trim();
        return if detail.is_empty() {
            Err(format!("boot returned {status}"))
        } else {
            Err(format!("boot returned {status}: {detail}"))
        };
    }

    serde_json::from_str::<serde_json::Value>(&body)
        .map_err(|e| format!("parse boot response failed: {e}"))
}

async fn run_boot_cli(paths: &auth::CortexPaths, args: &[String]) -> Result<(), String> {
    let agent = parse_flag_value(args, "--agent").unwrap_or_else(|| "cli".to_string());
    let agent = agent.trim();
    if agent.is_empty() {
        return Err("agent cannot be empty".to_string());
    }

    let budget = parse_flag_usize(args, "--budget")?.unwrap_or(DEFAULT_BOOT_BUDGET);
    let json_output = args.iter().any(|arg| arg == "--json");
    let (base_url, api_key, local_owner_mode) = resolve_client_target(args, paths);
    ensure_remote_target_has_api_key(&base_url, api_key.as_deref(), paths)?;

    if local_owner_mode {
        // Boot CLI does not own daemon lifecycle and must not auto-spawn.
        ensure_daemon(paths, None, false, false).await?;
    }

    let local_target_identity_valid = if local_owner_mode {
        false
    } else if is_local_client_base_url(&base_url, paths) {
        daemon_healthy(paths).await
    } else {
        false
    };
    let allow_local_token_fallback = local_owner_mode || local_target_identity_valid;
    let payload = request_boot_payload(
        paths,
        &base_url,
        &paths.token,
        api_key.as_deref(),
        allow_local_token_fallback,
        agent,
        budget,
    )
    .await?;

    if json_output {
        println!(
            "{}",
            serde_json::to_string_pretty(&payload)
                .map_err(|e| format!("serialize boot response failed: {e}"))?
        );
    } else {
        let boot_prompt = payload
            .get("bootPrompt")
            .and_then(|value| value.as_str())
            .ok_or_else(|| "boot response missing bootPrompt".to_string())?;
        println!("{boot_prompt}");
    }
    Ok(())
}

async fn boot_agent(paths: &auth::CortexPaths, agent: &str) -> Result<(), String> {
    let base_url = local_daemon_base_url(paths);
    request_boot_payload(paths, &base_url, &paths.token, None, true, agent, 200)
        .await
        .map(|_| ())
}

/// Hold the singleton daemon lock before startup so duplicate `serve`
/// invocations cannot rotate the shared auth token and then die on bind.
fn daemon_lock_wait_timeout() -> Duration {
    let secs = std::env::var("CORTEX_DAEMON_LOCK_WAIT_SECS")
        .ok()
        .and_then(|value| value.parse::<u64>().ok())
        .unwrap_or(DEFAULT_DAEMON_LOCK_WAIT_SECS);
    Duration::from_secs(secs.max(1))
}

#[derive(Debug)]
struct RuntimeLockGuards {
    _scoped: std::fs::File,
    _global: Option<std::fs::File>,
}

fn try_acquire_runtime_locks(paths: &auth::CortexPaths) -> Result<RuntimeLockGuards, String> {
    let scoped = auth::acquire_daemon_lock(paths)?;
    let global = if single_daemon_test_bypass_enabled() {
        None
    } else {
        match auth::acquire_global_daemon_lock() {
            Ok(lock) => Some(lock),
            Err(err) => {
                drop(scoped);
                return Err(err);
            }
        }
    };
    Ok(RuntimeLockGuards {
        _scoped: scoped,
        _global: global,
    })
}

fn acquire_runtime_lock(paths: &auth::CortexPaths) -> Result<RuntimeLockGuards, String> {
    let _ = auth::cleanup_stale_pid_lock(paths);
    if std::env::var("CORTEX_WAIT_FOR_DAEMON_LOCK")
        .ok()
        .is_some_and(|value| value == "1")
    {
        let deadline = std::time::Instant::now() + daemon_lock_wait_timeout();
        let last_err = loop {
            match try_acquire_runtime_locks(paths) {
                Ok(lock) => return Ok(lock),
                Err(err) => {
                    let _ = auth::cleanup_stale_pid_lock(paths);
                    if std::time::Instant::now() >= deadline {
                        break err;
                    }
                    std::thread::sleep(Duration::from_millis(DAEMON_LOCK_RETRY_INTERVAL_MS));
                }
            }
        };

        // Sleep/wake edge: lock ownership can hand off shortly after timeout due scheduler jitter.
        let grace_deadline =
            std::time::Instant::now() + Duration::from_secs(DAEMON_LOCK_HANDOFF_GRACE_SECS);
        while std::time::Instant::now() < grace_deadline {
            let _ = auth::cleanup_stale_pid_lock(paths);
            if let Ok(lock) = try_acquire_runtime_locks(paths) {
                return Ok(lock);
            }
            std::thread::sleep(Duration::from_millis(DAEMON_LOCK_RETRY_INTERVAL_MS));
        }
        return Err(last_err);
    }
    try_acquire_runtime_locks(paths)
}

fn daemon_owner_tag_from_env() -> Option<String> {
    std::env::var("CORTEX_DAEMON_OWNER")
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

fn daemon_owner_token_from_env() -> Option<String> {
    std::env::var(DAEMON_OWNER_TOKEN_ENV)
        .ok()
        .map(|value| value.trim().to_string())
        .filter(|value| !value.is_empty())
}

fn spawn_parent_pid_from_env() -> Option<u32> {
    std::env::var(SPAWN_PARENT_PID_ENV)
        .ok()
        .and_then(|value| value.trim().parse::<u32>().ok())
}

fn spawn_parent_start_time_from_env() -> Option<u64> {
    std::env::var(SPAWN_PARENT_START_TIME_ENV)
        .ok()
        .and_then(|value| value.trim().parse::<u64>().ok())
}

fn should_watch_spawn_parent(owner_tag: Option<&str>) -> bool {
    owner_tag
        .map(|owner| !owner.eq_ignore_ascii_case(CONTROL_CENTER_OWNER_TAG))
        .unwrap_or(true)
}

fn is_control_center_owner(owner_tag: Option<&str>) -> bool {
    owner_tag
        .map(|owner| owner.eq_ignore_ascii_case(CONTROL_CENTER_OWNER_TAG))
        .unwrap_or(false)
}

fn parse_env_u64_nonnegative(key: &str, default: u64) -> u64 {
    std::env::var(key)
        .ok()
        .and_then(|raw| raw.trim().parse::<u64>().ok())
        .unwrap_or(default)
}

fn app_managed_startup_heavy_delay(owner_tag: Option<&str>) -> Duration {
    if !is_control_center_owner(owner_tag) {
        return Duration::from_secs(0);
    }
    let secs = parse_env_u64_nonnegative(
        APP_MANAGED_STARTUP_DELAY_ENV,
        APP_MANAGED_STARTUP_HEAVY_DELAY_SECS,
    )
    .min(APP_MANAGED_STARTUP_HEAVY_DELAY_MAX_SECS);
    Duration::from_secs(secs)
}

#[derive(Clone, Copy)]
struct StartupSchedule {
    index: Duration,
    aging: Duration,
    embed: Duration,
    crystallize: Duration,
    storage_governor_initial: Duration,
}

fn startup_delay_from_env(key: &str, default: u64) -> Duration {
    Duration::from_secs(parse_env_u64_nonnegative(key, default).min(3_600))
}

fn startup_schedule(owner_tag: Option<&str>) -> StartupSchedule {
    let zero = Duration::from_secs(0);
    let heavy = app_managed_startup_heavy_delay(owner_tag);
    let index = if heavy > zero {
        heavy
    } else {
        startup_delay_from_env(STARTUP_INDEX_DELAY_ENV, DEFAULT_STARTUP_INDEX_DELAY_SECS)
    };
    let aging = if heavy > zero {
        heavy + Duration::from_secs(APP_MANAGED_AGING_STARTUP_OFFSET_SECS)
    } else {
        startup_delay_from_env(STARTUP_AGING_DELAY_ENV, DEFAULT_STARTUP_AGING_DELAY_SECS)
    };
    let embed = if heavy > zero {
        heavy + Duration::from_secs(APP_MANAGED_EMBED_STARTUP_OFFSET_SECS)
    } else {
        startup_delay_from_env(STARTUP_EMBED_DELAY_ENV, DEFAULT_STARTUP_EMBED_DELAY_SECS)
    };
    let crystallize = if heavy > zero {
        heavy + Duration::from_secs(APP_MANAGED_CRYSTALLIZE_STARTUP_OFFSET_SECS)
    } else {
        startup_delay_from_env(
            STARTUP_CRYSTALLIZE_DELAY_ENV,
            DEFAULT_STARTUP_CRYSTALLIZE_DELAY_SECS,
        )
    };
    let storage_governor_initial = startup_delay_from_env(
        STARTUP_STORAGE_GOVERNOR_DELAY_ENV,
        DEFAULT_STARTUP_STORAGE_GOVERNOR_DELAY_SECS,
    );
    StartupSchedule {
        index,
        aging,
        embed,
        crystallize,
        storage_governor_initial,
    }
}

fn background_db_lock_max_wait() -> Duration {
    let max_wait_ms = parse_env_u64_nonnegative(
        BACKGROUND_DB_LOCK_MAX_WAIT_MS_ENV,
        BACKGROUND_DB_LOCK_DEFAULT_MAX_WAIT_MS,
    )
    .clamp(100, 60_000);
    Duration::from_millis(max_wait_ms)
}

async fn acquire_background_db_lock<'a>(
    db: &'a std::sync::Arc<tokio::sync::Mutex<rusqlite::Connection>>,
    task_name: &str,
    max_wait: Duration,
) -> Option<tokio::sync::MutexGuard<'a, rusqlite::Connection>> {
    let started = std::time::Instant::now();
    loop {
        if let Ok(conn) = db.try_lock() {
            return Some(conn);
        }
        if started.elapsed() >= max_wait {
            eprintln!(
                "[cortex] Skipping {task_name}: DB lock busy for {}ms",
                started.elapsed().as_millis()
            );
            return None;
        }
        tokio::time::sleep(Duration::from_millis(BACKGROUND_DB_LOCK_RETRY_MS)).await;
    }
}

fn process_pid_start_time(pid: u32) -> Option<u64> {
    let mut system = sysinfo::System::new_all();
    let target = sysinfo::Pid::from_u32(pid);
    system.refresh_processes(sysinfo::ProcessesToUpdate::Some(&[target]), true);
    system.process(target).map(|process| process.start_time())
}

fn process_pid_identity_matches(pid: u32, expected_start_time: u64) -> bool {
    process_pid_start_time(pid)
        .map(|actual_start_time| actual_start_time == expected_start_time)
        .unwrap_or(false)
}

fn spawn_parent_orphan_watch_task<F>(
    shutdown_tx: std::sync::Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>>,
    parent_pid: u32,
    parent_start_time: u64,
    watch_interval: Duration,
    identity_matches: F,
) -> tokio::task::JoinHandle<()>
where
    F: Fn(u32, u64) -> bool + Send + Sync + 'static,
{
    let identity_matches = std::sync::Arc::new(identity_matches);
    tokio::spawn(async move {
        let mut interval = tokio::time::interval(watch_interval);
        interval.tick().await;
        loop {
            interval.tick().await;
            if !(identity_matches)(parent_pid, parent_start_time) {
                eprintln!(
                    "[cortex] Spawn parent process {parent_pid} exited or was recycled; shutting down daemon"
                );
                if let Some(tx) = shutdown_tx.lock().await.take() {
                    let _ = tx.send(());
                }
                break;
            }
        }
    })
}

fn process_looks_like_cortex_daemon(process: &sysinfo::Process) -> bool {
    let cmd: Vec<String> = process
        .cmd()
        .iter()
        .map(|arg| arg.to_string_lossy().to_ascii_lowercase())
        .collect();
    if cmd.is_empty() {
        return false;
    }
    let has_daemon_role = cmd.iter().any(|arg| arg == "serve" || arg == "service-run")
        || cmd
            .windows(2)
            .any(|pair| pair[0] == "service" && pair[1] == "run");
    if !has_daemon_role {
        return false;
    }

    let exe_is_cortex = process
        .exe()
        .and_then(|path| path.file_stem().or(path.file_name()))
        .map(|name| name.to_string_lossy().eq_ignore_ascii_case("cortex"))
        .unwrap_or(false);
    let cmd_is_cortex = cmd
        .first()
        .map(|first| first.contains("cortex"))
        .unwrap_or(false);
    exe_is_cortex || cmd_is_cortex
}

fn detect_other_cortex_daemon_process() -> Option<(u32, String, String)> {
    let current_pid = std::process::id();
    let mut system = sysinfo::System::new_all();
    system.refresh_processes(sysinfo::ProcessesToUpdate::All, true);
    for (pid, process) in system.processes() {
        let pid_u32 = pid.as_u32();
        if pid_u32 == current_pid {
            continue;
        }
        if !process_looks_like_cortex_daemon(process) {
            continue;
        }
        let exe = process
            .exe()
            .map(|path| path.display().to_string())
            .unwrap_or_else(|| "<unknown>".to_string());
        let cmd = process
            .cmd()
            .iter()
            .map(|arg| arg.to_string_lossy().into_owned())
            .collect::<Vec<_>>()
            .join(" ");
        return Some((pid_u32, exe, cmd));
    }
    None
}

fn spawned_owner_requires_parent_pid(owner_tag: Option<&str>) -> bool {
    owner_tag
        .map(|owner| should_watch_spawn_parent(Some(owner)))
        .unwrap_or(false)
}

fn validate_spawned_owner_runtime_claim(
    paths: &auth::CortexPaths,
    owner_tag: Option<&str>,
    parent_pid: Option<u32>,
    parent_start_time: Option<u64>,
    owner_token: Option<&str>,
) -> Result<(), String> {
    if spawned_owner_requires_parent_pid(owner_tag) && parent_pid.is_none() {
        return Err(format!(
            "owner '{}' requires {} linkage",
            owner_tag.unwrap_or("unknown"),
            SPAWN_PARENT_PID_ENV
        ));
    }
    if spawned_owner_requires_parent_pid(owner_tag) && parent_start_time.is_none() {
        return Err(format!(
            "owner '{}' requires {} linkage",
            owner_tag.unwrap_or("unknown"),
            SPAWN_PARENT_START_TIME_ENV
        ));
    }

    if let (Some(parent_pid), Some(parent_start_time)) = (parent_pid, parent_start_time) {
        let Some(actual_start_time) = process_pid_start_time(parent_pid) else {
            return Err(format!(
                "spawn parent process {parent_pid} is not running during ownership claim validation"
            ));
        };
        if actual_start_time != parent_start_time {
            return Err(format!(
                "spawn parent start-time mismatch for pid {parent_pid} (env={parent_start_time}, actual={actual_start_time})"
            ));
        }
    }

    validate_spawned_owner_claim(paths, owner_tag, parent_pid, owner_token)
}

async fn startup_single_daemon_preflight(paths: &auth::CortexPaths) -> Result<(), String> {
    if let Some((pid, exe, cmd)) = detect_other_cortex_daemon_process() {
        if single_daemon_test_bypass_enabled() {
            eprintln!(
                "[cortex] Warning: bypassing single-daemon process preflight for debug test run (detected pid={pid}, exe={exe}, cmd=\"{cmd}\")"
            );
        } else {
            return Err(format!(
                "daemon startup denied: Cortex already has an active daemon process (pid={pid}, exe={exe}, cmd=\"{cmd}\")"
            ));
        }
    }

    let bind_addr = paths.bind.trim();
    let bind_error = match std::net::TcpListener::bind((bind_addr, paths.port)) {
        Ok(listener) => {
            drop(listener);
            return Ok(());
        }
        Err(err) => err,
    };

    let readiness_url = format!("{}/readiness", local_daemon_base_url(paths));
    let health_url = format!("{}/health", local_daemon_base_url(paths));
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(2))
        .build()
        .map_err(|err| format!("daemon startup preflight: build HTTP client: {err}"))?;
    let (mut status, mut body) = match transport::request_url_with_local_ipc_fallback(
        &client,
        "GET",
        &readiness_url,
        paths,
        &[],
        None,
        Duration::from_secs(2),
    )
    .await
    {
        Ok((status, body)) => (status.as_u16(), body),
        Err(readiness_err) => {
            // Backward compatibility for daemons that do not expose /readiness yet.
            match transport::request_url_with_local_ipc_fallback(
                &client,
                "GET",
                &health_url,
                paths,
                &[],
                None,
                Duration::from_secs(2),
            )
            .await
            {
                Ok((status, body)) => (status.as_u16(), body),
                Err(health_err) => {
                    return Err(format!(
                        "daemon startup denied: cannot bind {bind_addr}:{} ({bind_error}) and readiness probe at {readiness_url} failed ({readiness_err}); fallback health probe at {health_url} also failed ({health_err})",
                        paths.port
                    ));
                }
            }
        }
    };

    if let Some(ready) = readiness_state_from_payload(status, &body, Some(paths.port), Some(paths))
    {
        return if ready {
            Err(format!(
                "daemon startup denied: canonical Cortex instance is already ready on port {}",
                paths.port
            ))
        } else {
            Err(format!(
                "daemon startup denied: canonical Cortex instance is already starting on port {}",
                paths.port
            ))
        };
    }
    if readiness_state_from_payload(status, &body, Some(paths.port), None).is_some() {
        return Err(format!(
            "daemon startup denied: port {} is served by a different Cortex runtime identity",
            paths.port
        ));
    }

    // Fallback for legacy daemons (or intermediaries that do not proxy readiness):
    // probe /health and apply canonical identity checks there.
    if let Ok((health_status, health_body)) = transport::request_url_with_local_ipc_fallback(
        &client,
        "GET",
        &health_url,
        paths,
        &[],
        None,
        Duration::from_secs(2),
    )
    .await
    {
        status = health_status.as_u16();
        body = health_body;
    }

    if is_cortex_health_payload(status, &body, Some(paths.port), Some(paths)) {
        return Err(format!(
            "daemon startup denied: canonical Cortex instance is already healthy on port {}",
            paths.port
        ));
    }
    if is_cortex_health_payload(status, &body, Some(paths.port), None) {
        return Err(format!(
            "daemon startup denied: port {} is served by a different Cortex runtime identity",
            paths.port
        ));
    }

    Err(format!(
        "daemon startup denied: cannot bind {bind_addr}:{} ({bind_error}); readiness probe at {readiness_url} returned non-canonical payload (HTTP {status})",
        paths.port
    ))
}

fn app_init_required_client_name(agent: Option<&str>) -> String {
    env_trimmed(APP_CLIENT_ENV)
        .or_else(|| normalize_option(agent))
        .unwrap_or_else(|| "client".to_string())
}

fn app_init_required_error(paths: &auth::CortexPaths, agent: Option<&str>) -> String {
    let client = app_init_required_client_name(agent);
    format!(
        "APP_INIT_REQUIRED: {client} is attach-only and cannot start the daemon automatically on port {}. Start Cortex Control Center and initialize the app-managed daemon, then retry.",
        paths.port
    )
}

fn local_spawn_allowed_for_request(allow_service_ensure: bool) -> bool {
    if !allow_service_ensure {
        return false;
    }
    let app_client_marked = env_trimmed(APP_CLIENT_ENV).is_some();
    let local_spawn_raw = std::env::var(DAEMON_LOCAL_SPAWN_ENV).ok();
    let local_spawn_disabled = local_spawn_raw
        .as_ref()
        .is_some_and(|value| !parse_truthy_flag(value));
    let app_required = std::env::var(APP_REQUIRED_ENV)
        .ok()
        .is_some_and(|value| parse_truthy_flag(&value));
    // Fail closed for app-marked clients when no explicit local spawn policy exists.
    // This prevents partial registration env contracts from silently re-enabling local spawn.
    if app_client_marked && local_spawn_raw.is_none() {
        return false;
    }
    !(local_spawn_disabled || app_required)
}

fn control_center_lock_path(paths: &auth::CortexPaths) -> PathBuf {
    paths.home.join("runtime").join(CONTROL_CENTER_LOCK_FILE)
}

fn is_lock_contention_error(err: &std::io::Error) -> bool {
    if matches!(
        err.kind(),
        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::PermissionDenied
    ) {
        return true;
    }
    if cfg!(windows) {
        return matches!(err.raw_os_error(), Some(32 | 33));
    }
    false
}

fn control_center_is_active(paths: &auth::CortexPaths) -> Result<bool, String> {
    let lock_path = control_center_lock_path(paths);
    let lock_file = match std::fs::OpenOptions::new()
        .create(false)
        .read(true)
        .write(true)
        .open(&lock_path)
    {
        Ok(file) => file,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => return Ok(false),
        Err(err) if is_lock_contention_error(&err) => return Ok(true),
        Err(err) => {
            return Err(format!(
                "open control-center lock {}: {err}",
                lock_path.display()
            ));
        }
    };

    match lock_file.try_lock_exclusive() {
        Ok(()) => {
            let _ = lock_file.unlock();
            Ok(false)
        }
        Err(err) if is_lock_contention_error(&err) => Ok(true),
        Err(err) => Err(format!(
            "probe control-center lock {}: {err}",
            lock_path.display()
        )),
    }
}

async fn ensure_daemon(
    paths: &auth::CortexPaths,
    agent: Option<&str>,
    emit_port: bool,
    allow_service_ensure: bool,
) -> Result<(), String> {
    std::fs::create_dir_all(&paths.home).map_err(|e| format!("create home dir: {e}"))?;
    let _ = auth::migrate_legacy_db(paths)?;
    let local_spawn_allowed = local_spawn_allowed_for_request(allow_service_ensure);
    let control_center_active_snapshot = if local_spawn_allowed {
        control_center_is_active(paths).ok()
    } else {
        None
    };

    let lock = auth::acquire_daemon_lock(paths);

    match lock {
        Ok(_guard) => {
            if daemon_healthy(paths).await {
                // already healthy
            } else if local_spawn_allowed {
                if control_center_active_snapshot == Some(true) {
                    return Err(app_init_required_error(paths, agent));
                }
                match control_center_is_active(paths) {
                    Ok(true) => return Err(app_init_required_error(paths, agent)),
                    Ok(false) => {}
                    Err(err) => {
                        return Err(format!(
                            "{} (control-center lock probe failed: {})",
                            app_init_required_error(paths, agent),
                            err
                        ));
                    }
                }
                #[cfg(windows)]
                {
                    if !ensure_service_ready_async().await {
                        return Err(format!(
                            "daemon is not healthy on port {} and Windows service ensure failed. Run `cortex service ensure` manually.",
                            paths.port
                        ));
                    }
                }
                #[cfg(not(windows))]
                {
                    ensure_local_plugin_spawn_async(paths, agent).await?;
                }
            } else {
                return Err(app_init_required_error(paths, agent));
            }
        }
        Err(_) => {
            if !wait_for_health(paths, Duration::from_secs(DAEMON_STARTUP_WAIT_SECS)).await {
                if local_spawn_allowed {
                    if control_center_active_snapshot == Some(true) {
                        return Err(app_init_required_error(paths, agent));
                    }
                    match control_center_is_active(paths) {
                        Ok(true) => return Err(app_init_required_error(paths, agent)),
                        Ok(false) => {}
                        Err(err) => {
                            return Err(format!(
                                "{} (control-center lock probe failed: {})",
                                app_init_required_error(paths, agent),
                                err
                            ));
                        }
                    }
                    #[cfg(windows)]
                    {
                        if ensure_service_ready_async().await {
                            // proceed
                        } else {
                            return Err(format!(
                                "daemon is not healthy on port {} and Windows service ensure failed while daemon lock was held.",
                                paths.port
                            ));
                        }
                    }
                    #[cfg(not(windows))]
                    {
                        return Err(format!(
                            "daemon is not healthy on port {} and another process still holds the daemon lock. Retry after the in-flight startup finishes.",
                            paths.port
                        ));
                    }
                } else {
                    return Err(app_init_required_error(paths, agent));
                }
            }
        }
    }

    if let Some(agent) = agent {
        if let Err(e) = boot_agent(paths, agent).await {
            eprintln!("[cortex-plugin] Warning: boot call failed for agent '{agent}': {e}");
        }
    }

    if emit_port {
        println!("{}", paths.port);
    }
    Ok(())
}

#[cfg(windows)]
async fn ensure_service_ready_async() -> bool {
    tokio::task::spawn_blocking(service::ensure_ready)
        .await
        .unwrap_or(false)
}

#[cfg(not(windows))]
fn plugin_owner_tag(agent: Option<&str>) -> String {
    let normalized = agent
        .unwrap_or("plugin")
        .trim()
        .chars()
        .map(|ch| {
            if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' {
                ch.to_ascii_lowercase()
            } else {
                '-'
            }
        })
        .collect::<String>()
        .trim_matches('-')
        .to_string();
    if normalized.is_empty() {
        "plugin".to_string()
    } else {
        format!("plugin-{normalized}")
    }
}

fn normalized_path_for_guard(path: &Path) -> String {
    path.to_string_lossy()
        .replace('\\', "/")
        .to_ascii_lowercase()
}

fn path_is_under_root(path: &Path, root: &Path) -> bool {
    let normalized_path = normalized_path_for_guard(path);
    let mut normalized_root = normalized_path_for_guard(root);
    if !normalized_root.ends_with('/') {
        normalized_root.push('/');
    }
    normalized_path == normalized_root.trim_end_matches('/')
        || normalized_path.starts_with(&normalized_root)
}

fn is_disallowed_startup_binary_path(path: &Path) -> bool {
    let normalized = normalized_path_for_guard(path);
    let file_name = path
        .file_name()
        .and_then(|name| name.to_str())
        .unwrap_or_default()
        .to_ascii_lowercase();

    if file_name.starts_with("cortex-daemon-run") {
        return true;
    }
    if normalized.contains("/daemon-lifecycle-runtime/") {
        return true;
    }

    let mut temp_roots = vec![std::env::temp_dir()];
    if let Ok(temp) = std::env::var("TEMP") {
        temp_roots.push(PathBuf::from(temp));
    }
    if let Ok(tmp) = std::env::var("TMP") {
        temp_roots.push(PathBuf::from(tmp));
    }
    temp_roots
        .iter()
        .any(|root| !root.as_os_str().is_empty() && path_is_under_root(path, root))
}

#[cfg(not(windows))]
async fn ensure_local_plugin_spawn_async(
    paths: &auth::CortexPaths,
    agent: Option<&str>,
) -> Result<(), String> {
    let current_exe = std::env::current_exe().map_err(|e| format!("resolve cortex binary: {e}"))?;
    if is_disallowed_startup_binary_path(&current_exe) {
        return Err(format!(
            "refusing to launch daemon from disallowed runtime path: {}",
            current_exe.display()
        ));
    }
    let parent_pid = std::process::id();
    let parent_start = process_pid_start_time(parent_pid)
        .ok_or_else(|| format!("resolve spawn parent start time for pid {parent_pid}"))?;
    let owner_tag = plugin_owner_tag(agent);
    let owner_token = issue_owner_token_for_spawn(paths, &owner_tag, parent_pid)
        .map_err(|e| format!("issue owner token: {e}"))?;

    let mut cmd = std::process::Command::new(current_exe);
    cmd.arg("serve")
        .arg("--home")
        .arg(paths.home.display().to_string())
        .arg("--db")
        .arg(paths.db.display().to_string())
        .arg("--port")
        .arg(paths.port.to_string())
        .arg("--bind")
        .arg(paths.bind.as_str())
        .env("CORTEX_DAEMON_OWNER", &owner_tag)
        .env("CORTEX_DAEMON_OWNER_SOURCE", "plugin-local")
        .env("CORTEX_DAEMON_OWNER_AGENT", agent.unwrap_or("plugin"))
        .env("CORTEX_DAEMON_OWNER_MODE", "local-plugin")
        .env(SPAWN_PARENT_PID_ENV, parent_pid.to_string())
        .env(SPAWN_PARENT_START_TIME_ENV, parent_start.to_string())
        .env(DAEMON_OWNER_TOKEN_ENV, owner_token)
        .stdin(std::process::Stdio::null())
        .stdout(std::process::Stdio::null())
        .stderr(std::process::Stdio::null());

    cmd.spawn()
        .map_err(|e| format!("spawn local daemon from plugin mode: {e}"))?;

    if wait_for_health(paths, Duration::from_secs(DAEMON_STARTUP_WAIT_SECS)).await {
        Ok(())
    } else {
        Err(format!(
            "daemon spawn started but health is still unavailable on port {}",
            paths.port
        ))
    }
}

// ── Admin CLI helpers ───────────────────────────────────────────────────────

fn read_auth_token() -> Result<String, String> {
    let token_path = auth::CortexPaths::resolve().token;
    std::fs::read_to_string(&token_path)
        .map(|v| v.trim().to_string())
        .map_err(|_| {
            format!(
                "Cannot read auth token at {}. Is the daemon running?",
                token_path.display()
            )
        })
}

async fn admin_request(
    method: &str,
    path: &str,
    body: Option<serde_json::Value>,
) -> Result<serde_json::Value, String> {
    let paths = auth::CortexPaths::resolve();
    let token = read_auth_token()?;
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(10))
        .build()
        .map_err(|e| format!("create admin client: {e}"))?;
    let base_url = local_daemon_base_url(&paths);
    let payload = body.map(|value| value.to_string());
    let mut headers = vec![
        ("authorization".to_string(), format!("Bearer {token}")),
        ("x-cortex-request".to_string(), "true".to_string()),
    ];
    if payload.is_some() {
        headers.push(("content-type".to_string(), "application/json".to_string()));
    }
    let (status, body_text) = transport::request_with_local_ipc_fallback(
        &client,
        method,
        &base_url,
        path,
        &paths,
        &headers,
        payload.as_deref(),
        Duration::from_secs(10),
    )
    .await
    .map_err(|e| {
        if e.to_ascii_lowercase().contains("connect") {
            "Cortex daemon not running. Start with: cortex serve".to_string()
        } else {
            format!("Request failed: {e}")
        }
    })?;
    if status.as_u16() == 403 {
        return Err("Admin commands require team mode. Run: cortex setup --team".to_string());
    }
    if status.as_u16() == 404 {
        return Err("Endpoint not found. Is the daemon up to date?".to_string());
    }
    let json: serde_json::Value = serde_json::from_str(&body_text).map_err(|_| {
        if body_text.is_empty() {
            format!("Empty response from daemon (HTTP {status})")
        } else {
            format!("Unexpected response (HTTP {status}): {body_text}")
        }
    })?;
    if !status.is_success() {
        let msg = json
            .get("error")
            .and_then(|v| v.as_str())
            .unwrap_or("Unknown error");
        return Err(msg.to_string());
    }
    Ok(json)
}

fn confirm_action(prompt: &str) -> bool {
    eprint!("{prompt} [y/N] ");
    let mut input = String::new();
    if std::io::stdin().read_line(&mut input).is_err() {
        return false;
    }
    matches!(input.trim().to_lowercase().as_str(), "y" | "yes")
}

fn json_str(val: &serde_json::Value, key: &str) -> String {
    val.get(key)
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_string()
}

fn json_str_or(val: &serde_json::Value, key: &str, default: &str) -> String {
    val.get(key)
        .and_then(|v| v.as_str())
        .unwrap_or(default)
        .to_string()
}

fn json_field(val: &serde_json::Value, key: &str) -> String {
    match val.get(key) {
        Some(v) if v.is_string() => v.as_str().unwrap_or("").to_string(),
        Some(v) => v.to_string(),
        None => "-".to_string(),
    }
}

fn api_key_output_masked() -> bool {
    !std::io::stdout().is_terminal()
}

fn format_api_key_for_output(api_key: &str) -> String {
    if !api_key_output_masked() {
        return api_key.to_string();
    }
    mask_secret_for_logs(api_key)
}

fn mask_secret_for_logs(secret: &str) -> String {
    const PREFIX: usize = 8;
    const SUFFIX: usize = 4;
    let chars: Vec<char> = secret.chars().collect();
    if chars.is_empty() {
        return String::new();
    }
    if chars.len() <= PREFIX + SUFFIX {
        return "*".repeat(chars.len().max(4));
    }
    let prefix: String = chars.iter().take(PREFIX).collect();
    let suffix: String = chars
        .iter()
        .skip(chars.len().saturating_sub(SUFFIX))
        .collect();
    format!("{prefix}...{suffix}")
}

// ── Shared daemon logic (used by `serve` and `service-run`) ─────────────────

/// Run the full Cortex daemon. The `extra_shutdown` future is an additional
/// shutdown trigger beyond the HTTP /shutdown endpoint:
/// - `serve` passes Ctrl+C / SIGTERM
/// - `service-run` passes the SCM stop signal
pub(crate) async fn run_daemon(
    paths: auth::CortexPaths,
    extra_shutdown: impl std::future::Future<Output = ()> + Send + 'static,
) {
    let _daemon_lock = match acquire_runtime_lock(&paths) {
        Ok(lock) => lock,
        Err(err) => {
            if daemon_healthy(&paths).await {
                eprintln!(
                    "[cortex] Daemon already healthy on port {}; exiting cleanly.",
                    paths.port
                );
                return;
            }
            eprintln!("[cortex] FATAL: {err}");
            eprintln!(
                "[cortex] Reuse the existing daemon instead of launching a second `cortex serve`."
            );
            std::process::exit(1);
        }
    };

    let db_path = paths.db.clone();
    eprintln!(
        "[cortex] Starting Cortex v{} (Rust)...",
        env!("CARGO_PKG_VERSION")
    );
    eprintln!("[cortex] DB: {}", db_path.display());

    install_daemon_panic_hook(&paths);

    let daemon_owner = daemon_owner_tag_from_env();
    let parent_pid = spawn_parent_pid_from_env();
    let parent_start_time = spawn_parent_start_time_from_env();
    let owner_token = daemon_owner_token_from_env();
    if let Err(reason) = validate_spawned_owner_runtime_claim(
        &paths,
        daemon_owner.as_deref(),
        parent_pid,
        parent_start_time,
        owner_token.as_deref(),
    ) {
        eprintln!("[cortex] FATAL: invalid spawned owner claim ({reason}); refusing startup");
        std::process::exit(1);
    }
    if let Err(reason) = startup_single_daemon_preflight(&paths).await {
        eprintln!("[cortex] FATAL: {reason}");
        std::process::exit(1);
    }

    let (state, shutdown_rx) = state::initialize(&paths, true).expect("Failed to initialize state");

    if should_watch_spawn_parent(daemon_owner.as_deref()) {
        if let (Some(parent_pid), Some(parent_start_time)) = (parent_pid, parent_start_time) {
            let _watcher = spawn_parent_orphan_watch_task(
                state.shutdown_tx.clone(),
                parent_pid,
                parent_start_time,
                Duration::from_secs(ORPHAN_WATCH_INTERVAL_SECS),
                process_pid_identity_matches,
            );
        }
    }

    let startup_schedule = startup_schedule(daemon_owner.as_deref());
    let background_lock_wait = background_db_lock_max_wait();
    eprintln!(
        "[cortex] Startup scheduling: index={}s, aging={}s, embeddings={}s, crystallize={}s, storage_governor={}s",
        startup_schedule.index.as_secs(),
        startup_schedule.aging.as_secs(),
        startup_schedule.embed.as_secs(),
        startup_schedule.crystallize.as_secs(),
        startup_schedule.storage_governor_initial.as_secs()
    );

    if let Some(parent) = paths.pid.parent() {
        std::fs::create_dir_all(parent).ok();
    }
    std::fs::write(&paths.pid, std::process::id().to_string()).ok();

    let token_path = paths.token.clone();
    let pid_path = paths.pid.clone();
    let backup_dir = paths.home.join("backups");
    eprintln!("[cortex] Auth token at {}", token_path.display());
    eprintln!(
        "[cortex] PID {} written to {}",
        std::process::id(),
        pid_path.display()
    );
    let cleaned_backups = cleanup_backup_retention(&backup_dir);
    eprintln!("[cortex] Cleaned {cleaned_backups} old backups, kept {BACKUP_RETENTION_COUNT}");
    let rotated_logs = rotate_startup_logs(&paths.home);
    if rotated_logs > 0 {
        eprintln!("[cortex] Rotated {rotated_logs} oversized log files");
    }

    // ── Recover WAL on startup ──────────────────────────────────────
    // Run WAL checkpoint to recover any pending writes from a previous crash.
    // This ensures committed transactions are flushed to the main DB file.
    {
        let conn = state.db.lock().await;
        if let Err(e) = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") {
            eprintln!("[cortex] WAL recovery warning: {e}");
        } else {
            eprintln!("[cortex] WAL recovery complete");
        }
    }

    // ── Schema migrations (idempotent) ──────────────────────────────
    let schema_version = {
        let conn = state.db.lock().await;
        let applied = db::run_pending_migrations(&conn);
        if applied > 0 {
            eprintln!("[cortex] Applied {applied} schema migrations");
        }
        db::current_schema_user_version(&conn).unwrap_or(0)
    };
    let _ = cleanup_bridge_backups(&paths.home, schema_version);

    // ── Startup indexing + decay (non-blocking) ─────────────────────
    // This used to run inline before the server bound its port, which could
    // delay startup significantly on large source trees.
    {
        let db_index = state.db.clone();
        let home = state.home.clone();
        let owner_id = state.default_owner_id;
        let startup_delay = startup_schedule.index;
        let lock_wait = background_lock_wait;
        tokio::spawn(async move {
            if startup_delay > Duration::from_secs(0) {
                tokio::time::sleep(startup_delay).await;
            }
            let started = std::time::Instant::now();
            if let Some(conn) =
                acquire_background_db_lock(&db_index, "startup indexing", lock_wait).await
            {
                let indexed = indexer::index_all(&conn, &home, owner_id);
                let decayed = indexer::decay_pass(&conn);
                eprintln!(
                    "[cortex] Startup indexing complete: indexed {indexed}, decayed {decayed} scores in {}ms",
                    started.elapsed().as_millis()
                );
            }
        });
    }

    // ── Background embedding builder ────────────────────────────────
    if let Some(engine) = state.embedding_engine.clone() {
        let db = state.db.clone();
        let batch_size = parse_env_usize(
            "CORTEX_EMBED_BACKFILL_BATCH_SIZE",
            DEFAULT_EMBED_BACKFILL_BATCH_SIZE,
        )
        .clamp(1, 10_000);
        let max_batches_per_pass = parse_env_usize(
            "CORTEX_EMBED_BACKFILL_MAX_BATCHES_PER_PASS",
            DEFAULT_EMBED_BACKFILL_MAX_BATCHES_PER_PASS,
        )
        .clamp(1, 1000);
        let interval_secs = parse_env_u64(
            "CORTEX_EMBED_BACKFILL_INTERVAL_SECS",
            DEFAULT_EMBED_BACKFILL_INTERVAL_SECS,
        )
        .clamp(5, 86_400);
        let drain_on_startup = std::env::var(EMBED_BACKFILL_DRAIN_ON_STARTUP_ENV)
            .ok()
            .map(|value| parse_truthy_flag(&value))
            .unwrap_or(false);
        let startup_drain_max_batches = parse_env_usize(
            EMBED_BACKFILL_STARTUP_DRAIN_MAX_BATCHES_ENV,
            DEFAULT_EMBED_BACKFILL_STARTUP_DRAIN_MAX_BATCHES,
        )
        .clamp(1, 10_000);
        let startup_delay = startup_schedule.embed;
        let lock_wait = background_lock_wait;
        let startup_max_batches_per_pass = if startup_delay > Duration::from_secs(0) {
            max_batches_per_pass.min(2)
        } else {
            max_batches_per_pass
        };
        tokio::spawn(async move {
            if startup_delay > Duration::from_secs(0) {
                tokio::time::sleep(startup_delay).await;
            }
            let startup_pass = build_embeddings_async(
                engine.clone(),
                &db,
                batch_size,
                startup_max_batches_per_pass,
                lock_wait,
            )
            .await;
            if startup_pass.queued_total > 0 && !startup_pass.exhausted {
                if drain_on_startup {
                    let drain_pass = build_embeddings_async(
                        engine.clone(),
                        &db,
                        batch_size,
                        startup_drain_max_batches,
                        lock_wait,
                    )
                    .await;
                    if drain_pass.exhausted {
                        eprintln!(
                            "[embeddings] Startup drain completed backlog in {} batches",
                            drain_pass.passes_ran
                        );
                    } else if drain_pass.queued_total > 0 {
                        eprintln!(
                            "[embeddings] Startup drain reached cap with backlog still pending (passes={}, queued={})",
                            drain_pass.passes_ran, drain_pass.queued_total
                        );
                    }
                } else {
                    eprintln!(
                        "[embeddings] Startup pass left backlog pending; set {}=1 to run a one-time extended drain",
                        EMBED_BACKFILL_DRAIN_ON_STARTUP_ENV
                    );
                }
            }
            let mut interval = tokio::time::interval(Duration::from_secs(interval_secs));
            interval.tick().await; // skip first immediate tick
            loop {
                interval.tick().await;
                build_embeddings_async(
                    engine.clone(),
                    &db,
                    batch_size,
                    max_batches_per_pass,
                    lock_wait,
                )
                .await;
            }
        });
    } else {
        let models_dir = paths.models.clone();
        tokio::spawn(async move {
            if let Some(dir) = embeddings::ensure_model_downloaded_in(&models_dir).await {
                eprintln!(
                    "[embeddings] Model ready at {} -- restart to activate",
                    dir.display()
                );
            }
        });
    }

    // ── Background WAL checkpoint every 10s (crash-safe) ──────────────
    {
        let db_wal = state.db.clone();
        let db_path = db_path.clone();
        let home_dir = paths.home.clone();
        let lock_wait = background_lock_wait.min(Duration::from_millis(750));
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
            interval.tick().await; // skip first immediate tick
            loop {
                interval.tick().await;

                // Checkpoint WAL first to ensure consistency
                let Some(conn) =
                    acquire_background_db_lock(&db_wal, "wal checkpoint", lock_wait).await
                else {
                    continue;
                };
                {
                    db::checkpoint_wal_best_effort(&conn);
                }

                // Check if daily backup is needed
                let backup_dir = home_dir.join("backups");
                if should_backup(&backup_dir) {
                    if let Err(e) = create_backup(&db_path, &backup_dir) {
                        eprintln!("[cortex] Backup failed: {e}");
                    }
                }
            }
        });
    }

    // ── Background quick_check every 30 minutes ────────────────────────
    // Runs PRAGMA quick_check (B-tree only) to catch corruption that develops
    // during runtime.  On failure, sets db_corrupted so /health reflects it.
    {
        let db_qc = state.db_read.clone();
        let db_corrupted_flag = state.db_corrupted.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(30 * 60));
            interval.tick().await; // skip first tick -- startup integrity_check already ran
            loop {
                interval.tick().await;
                let conn = db_qc.lock().await;
                if db::quick_check(&conn) {
                    // Clear the flag if a previous check had set it (e.g. after manual repair).
                    db_corrupted_flag.store(false, std::sync::atomic::Ordering::SeqCst);
                } else {
                    eprintln!(
                        "[cortex] WARNING: runtime PRAGMA quick_check FAILED -- \
                         database may be corrupted. Restart the daemon to trigger auto-repair. \
                         /health endpoint now shows degraded=true, db_corrupted=true."
                    );
                    db_corrupted_flag.store(true, std::sync::atomic::Ordering::SeqCst);
                }
            }
        });
    }

    // ── Background aging pass every 6 hours ──────────────────────────
    {
        let db_aging = state.db.clone();
        let startup_delay = startup_schedule.aging;
        let lock_wait = background_lock_wait;
        tokio::spawn(async move {
            if startup_delay > Duration::from_secs(0) {
                tokio::time::sleep(startup_delay).await;
            }
            // Run initial aging pass on startup
            if let Some(conn) =
                acquire_background_db_lock(&db_aging, "initial aging pass", lock_wait).await
            {
                let (compressed, archived) = aging::run_aging_pass(&conn);
                if compressed > 0 || archived > 0 {
                    eprintln!(
                        "[cortex] Initial aging: {compressed} compressed, {archived} archived"
                    );
                }
                cleanup_expired_rows(&conn, "Initial expired cleanup");
            }
            // Then run every 6 hours
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(6 * 3600));
            interval.tick().await;
            loop {
                interval.tick().await;
                if let Some(conn) =
                    acquire_background_db_lock(&db_aging, "aging pass", lock_wait).await
                {
                    aging::run_aging_pass(&conn);
                    cleanup_expired_rows(&conn, "Expired cleanup");
                }
            }
        });
    }

    // ── Background storage governor ────────────────────────────────────
    {
        let db_compaction = state.db.clone();
        let startup_delay = startup_schedule.storage_governor_initial;
        let lock_wait = background_lock_wait;
        tokio::spawn(async move {
            if startup_delay > Duration::from_secs(0) {
                tokio::time::sleep(startup_delay).await;
            }
            // Catch-up passes soon after startup to relieve event pressure early.
            for pass in 0..STARTUP_STORAGE_GOVERNOR_CATCHUP_PASSES {
                if let Some(conn) = acquire_background_db_lock(
                    &db_compaction,
                    "startup storage governor",
                    lock_wait,
                )
                .await
                {
                    let ran = compaction::run_compaction_governor_startup(&conn).is_some();
                    if !ran {
                        break;
                    }
                }
                if pass + 1 < STARTUP_STORAGE_GOVERNOR_CATCHUP_PASSES {
                    tokio::time::sleep(Duration::from_secs(
                        STARTUP_STORAGE_GOVERNOR_CATCHUP_INTERVAL_SECS,
                    ))
                    .await;
                }
            }

            let mut interval = tokio::time::interval(std::time::Duration::from_secs(30 * 60));
            interval.tick().await;
            loop {
                interval.tick().await;
                if let Some(conn) =
                    acquire_background_db_lock(&db_compaction, "storage governor", lock_wait).await
                {
                    let _ = compaction::run_compaction_governor(&conn);
                }
            }
        });
    }

    // ── Background crystallization pass every 2 hours ─────────────
    {
        let db_crystal = state.db.clone();
        let engine_crystal = state.embedding_engine.clone();
        let crystal_owner_id = state.default_owner_id;
        let brain_crystal: crystallize::BrainFiringSender = Some(state.brain_firing.clone());
        let initial_delay = startup_schedule.crystallize;
        let lock_wait = background_lock_wait;
        tokio::spawn(async move {
            // Initial pass on startup (after embeddings are built, with app-managed delay if needed)
            tokio::time::sleep(initial_delay).await;
            if let Some(conn) =
                acquire_background_db_lock(&db_crystal, "initial crystallization", lock_wait).await
            {
                let result = crystallize::run_crystallize_pass_with_brain(
                    &conn,
                    engine_crystal.as_deref(),
                    crystal_owner_id,
                    &brain_crystal,
                );
                if result.crystals_created > 0 || result.crystals_updated > 0 {
                    eprintln!(
                        "[cortex] Initial crystallization: {} created, {} updated",
                        result.crystals_created, result.crystals_updated
                    );
                }
            }
            // Then run every 2 hours
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(2 * 3600));
            interval.tick().await;
            loop {
                interval.tick().await;
                if let Some(conn) =
                    acquire_background_db_lock(&db_crystal, "crystallization pass", lock_wait).await
                {
                    crystallize::run_crystallize_pass_with_brain(
                        &conn,
                        engine_crystal.as_deref(),
                        crystal_owner_id,
                        &brain_crystal,
                    );
                }
            }
        });
    }

    // ── Background rate limiter cleanup every 5 minutes ────────────
    {
        let rl = state.rate_limiter.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(300));
            interval.tick().await;
            loop {
                interval.tick().await;
                rl.cleanup().await;
            }
        });
    }

    let idle_shutdown_secs = std::env::var(IDLE_SHUTDOWN_SECS_ENV)
        .ok()
        .and_then(|raw| raw.trim().parse::<u64>().ok())
        .unwrap_or(0);
    let idle_min_uptime_secs = parse_env_u64(
        IDLE_SHUTDOWN_MIN_UPTIME_SECS_ENV,
        DEFAULT_IDLE_SHUTDOWN_MIN_UPTIME_SECS,
    )
    .clamp(1, 86_400);
    if idle_shutdown_secs > 0 {
        eprintln!(
            "[cortex] Idle shutdown enabled (timeout={}s, min_uptime={}s)",
            idle_shutdown_secs, idle_min_uptime_secs
        );
    }

    let readiness_signal = state.readiness.clone();
    let db_for_shutdown = state.db.clone();
    let state_for_idle_shutdown = state.clone();
    let router = server::build_router(state, paths.port);

    // Combine shutdown sources: HTTP /shutdown, extra (Ctrl+C or SCM stop)
    let shutdown_future = async move {
        let idle_shutdown_future = async move {
            if idle_shutdown_secs == 0 {
                std::future::pending::<()>().await;
                return;
            }
            tokio::time::sleep(Duration::from_secs(idle_min_uptime_secs)).await;
            let mut interval = tokio::time::interval(Duration::from_secs(
                DEFAULT_IDLE_SHUTDOWN_CHECK_INTERVAL_SECS,
            ));
            interval.tick().await;
            loop {
                interval.tick().await;
                let idle_for = state_for_idle_shutdown.idle_for_secs();
                if idle_for >= idle_shutdown_secs {
                    eprintln!(
                        "[cortex] Idle shutdown threshold reached (idle={}s >= {}s)",
                        idle_for, idle_shutdown_secs
                    );
                    break;
                }
            }
        };

        tokio::select! {
            _ = shutdown_rx => {
                eprintln!("[cortex] Shutdown requested via HTTP");
            }
            _ = extra_shutdown => {}
            _ = idle_shutdown_future => {}
        }
    };

    server::run(
        router,
        &paths.bind,
        paths.port,
        paths.ipc_endpoint.clone(),
        &db_path,
        Some(readiness_signal),
        shutdown_future,
    )
    .await;

    // WAL checkpoint + cleanup
    eprintln!("[cortex] Flushing database...");
    {
        let conn = db_for_shutdown.lock().await;
        if let Err(e) = conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);") {
            eprintln!("[cortex] Warning: WAL checkpoint failed: {e}");
        }
    }

    let _ = std::fs::remove_file(&pid_path);
    eprintln!("[cortex] Shutdown complete.");
}

/// Build embeddings for all un-embedded memories and decisions.
/// IMPORTANT: Does NOT hold the DB lock during ONNX inference.
/// Reads IDs/text in a short lock, embeds in memory (no lock), then writes in batches.
type EmbeddingBackfillRows = Vec<(i64, String)>;
type EmbeddingBackfillTargets = (EmbeddingBackfillRows, EmbeddingBackfillRows);

#[derive(Clone, Copy, Debug, Default)]
struct EmbeddingBackfillPassResult {
    queued_total: usize,
    computed_total: usize,
    passes_ran: usize,
    exhausted: bool,
}

fn backfill_batch_may_have_more(
    memory_count: usize,
    decision_count: usize,
    batch_size: usize,
) -> bool {
    memory_count >= batch_size || decision_count >= batch_size
}

fn collect_unembedded_targets_for_model(
    conn: &rusqlite::Connection,
    model_key: &str,
    limit: usize,
) -> EmbeddingBackfillTargets {
    let mem: EmbeddingBackfillRows = conn
        .prepare(
            "SELECT m.id, m.text FROM memories m \
             WHERE m.status = 'active' \
                AND NOT EXISTS (\
                    SELECT 1 FROM embeddings e \
                    WHERE e.target_type = 'memory' \
                      AND e.target_id = m.id \
                      AND LOWER(COALESCE(e.model, '')) = ?1\
                ) \
             ORDER BY m.id ASC \
             LIMIT ?2",
        )
        .and_then(|mut stmt| {
            stmt.query_map(rusqlite::params![model_key, limit as i64], |row| {
                Ok((row.get(0)?, row.get(1)?))
            })
            .map(|rows| rows.filter_map(|r| r.ok()).collect())
        })
        .unwrap_or_default();

    let dec: EmbeddingBackfillRows = conn
        .prepare(
            "SELECT d.id, d.decision FROM decisions d \
             WHERE d.status = 'active' \
                AND NOT EXISTS (\
                    SELECT 1 FROM embeddings e \
                    WHERE e.target_type = 'decision' \
                      AND e.target_id = d.id \
                      AND LOWER(COALESCE(e.model, '')) = ?1\
                ) \
             ORDER BY d.id ASC \
             LIMIT ?2",
        )
        .and_then(|mut stmt| {
            stmt.query_map(rusqlite::params![model_key, limit as i64], |row| {
                Ok((row.get(0)?, row.get(1)?))
            })
            .map(|rows| rows.filter_map(|r| r.ok()).collect())
        })
        .unwrap_or_default();

    (mem, dec)
}

fn count_unembedded_targets_for_model(
    conn: &rusqlite::Connection,
    model_key: &str,
) -> (usize, usize) {
    let memory_count = conn
        .query_row(
            "SELECT COUNT(*) FROM memories m \
             WHERE m.status = 'active' \
               AND NOT EXISTS (\
                   SELECT 1 FROM embeddings e \
                   WHERE e.target_type = 'memory' \
                     AND e.target_id = m.id \
                     AND LOWER(COALESCE(e.model, '')) = ?1\
               )",
            rusqlite::params![model_key],
            |row| row.get::<_, i64>(0),
        )
        .unwrap_or(0)
        .max(0) as usize;

    let decision_count = conn
        .query_row(
            "SELECT COUNT(*) FROM decisions d \
             WHERE d.status = 'active' \
               AND NOT EXISTS (\
                   SELECT 1 FROM embeddings e \
                   WHERE e.target_type = 'decision' \
                     AND e.target_id = d.id \
                     AND LOWER(COALESCE(e.model, '')) = ?1\
               )",
            rusqlite::params![model_key],
            |row| row.get::<_, i64>(0),
        )
        .unwrap_or(0)
        .max(0) as usize;

    (memory_count, decision_count)
}

async fn build_embeddings_async(
    engine: std::sync::Arc<embeddings::EmbeddingEngine>,
    db: &std::sync::Arc<tokio::sync::Mutex<rusqlite::Connection>>,
    batch_size: usize,
    max_batches_per_pass: usize,
    lock_wait: Duration,
) -> EmbeddingBackfillPassResult {
    let model_key = engine.model_key();
    let mut result = EmbeddingBackfillPassResult::default();

    for _ in 0..max_batches_per_pass {
        let (unembedded_mem, unembedded_dec) = {
            let Some(conn) =
                acquire_background_db_lock(db, "embedding backfill scan", lock_wait).await
            else {
                break;
            };
            collect_unembedded_targets_for_model(&conn, model_key, batch_size)
        };

        let memory_count = unembedded_mem.len();
        let decision_count = unembedded_dec.len();
        let total = memory_count + decision_count;
        if total == 0 {
            result.exhausted = true;
            break;
        }
        result.passes_ran += 1;
        result.queued_total += total;

        let mut computed_batch = 0usize;
        let mut mem_results: Vec<(i64, Vec<u8>)> = Vec::new();
        for (id, text) in &unembedded_mem {
            if let Some(vec) = engine.clone().embed_async(text.clone()).await {
                mem_results.push((*id, embeddings::vector_to_blob(&vec)));
                computed_batch += 1;
            }
        }

        let mut dec_results: Vec<(i64, Vec<u8>)> = Vec::new();
        for (id, text) in &unembedded_dec {
            if let Some(vec) = engine.clone().embed_async(text.clone()).await {
                dec_results.push((*id, embeddings::vector_to_blob(&vec)));
                computed_batch += 1;
            }
        }

        {
            let Some(conn) =
                acquire_background_db_lock(db, "embedding backfill persist", lock_wait).await
            else {
                break;
            };
            for (id, blob) in &mem_results {
                let _ = conn.execute(
                    "INSERT OR REPLACE INTO embeddings (target_type, target_id, vector, model) \
                     VALUES ('memory', ?1, ?2, ?3)",
                    rusqlite::params![id, blob, model_key],
                );
            }
            for (id, blob) in &dec_results {
                let _ = conn.execute(
                    "INSERT OR REPLACE INTO embeddings (target_type, target_id, vector, model) \
                     VALUES ('decision', ?1, ?2, ?3)",
                    rusqlite::params![id, blob, model_key],
                );
            }
        }

        result.computed_total += computed_batch;
        if !backfill_batch_may_have_more(memory_count, decision_count, batch_size) {
            result.exhausted = true;
            break;
        }
    }

    if result.queued_total > 0 {
        eprintln!(
            "[embeddings] Built {}/{} embeddings this pass (passes={}, batch_size={}, max_batches={}, exhausted={})",
            result.computed_total,
            result.queued_total,
            result.passes_ran,
            batch_size,
            max_batches_per_pass,
            result.exhausted
        );
    }
    result
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::fs;
    use std::io::{ErrorKind, Read, Write};
    use std::net::TcpListener;
    use std::path::PathBuf;
    use std::process::{Command, Stdio};
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex, MutexGuard, OnceLock};
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

    static TEST_ENV_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
    const SPAWN_PARENT_TEST_CHILD_ENV: &str = "CORTEX_SPAWN_PARENT_TEST_CHILD";
    const CONTROL_CENTER_LOCK_TEST_CHILD_ENV: &str = "CORTEX_CONTROL_CENTER_LOCK_TEST_CHILD";
    const CONTROL_CENTER_LOCK_TEST_HOME_ENV: &str = "CORTEX_CONTROL_CENTER_LOCK_TEST_HOME";
    const CONTROL_CENTER_LOCK_TEST_READY_ENV: &str = "CORTEX_CONTROL_CENTER_LOCK_TEST_READY";
    const CONTROL_CENTER_LOCK_TEST_HOLD_MS_ENV: &str = "CORTEX_CONTROL_CENTER_LOCK_TEST_HOLD_MS";

    fn openapi_spec_path() -> PathBuf {
        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("..")
            .join("specs")
            .join("cortex-openapi.yaml")
    }

    struct ScopedEnvVar {
        key: &'static str,
    }

    impl ScopedEnvVar {
        fn set(key: &'static str, value: &str) -> Self {
            std::env::set_var(key, value);
            Self { key }
        }
    }

    impl Drop for ScopedEnvVar {
        fn drop(&mut self) {
            std::env::remove_var(self.key);
        }
    }

    fn env_guard() -> MutexGuard<'static, ()> {
        match TEST_ENV_LOCK.get_or_init(|| Mutex::new(())).lock() {
            Ok(guard) => guard,
            Err(poisoned) => poisoned.into_inner(),
        }
    }

    fn temp_test_dir(name: &str) -> std::path::PathBuf {
        let unique = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        std::env::temp_dir().join(format!("cortex_{name}_{unique}"))
    }

    fn run_preflight(paths: &auth::CortexPaths) -> Result<(), String> {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("build tokio runtime")
            .block_on(startup_single_daemon_preflight(paths))
    }

    fn run_ensure_daemon(
        paths: &auth::CortexPaths,
        agent: Option<&str>,
        emit_port: bool,
        allow_service_ensure: bool,
    ) -> Result<(), String> {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .expect("build tokio runtime")
            .block_on(ensure_daemon(paths, agent, emit_port, allow_service_ensure))
    }

    fn spawn_response_server(
        listener: TcpListener,
        status_line: &str,
        content_type: &str,
        body: String,
        max_requests: usize,
    ) -> std::thread::JoinHandle<()> {
        let status_line = status_line.to_string();
        let content_type = content_type.to_string();
        let max_requests = max_requests.max(1);
        std::thread::spawn(move || {
            let _ = listener.set_nonblocking(true);
            let deadline = Instant::now() + Duration::from_secs(3);
            let mut served = 0_usize;
            loop {
                match listener.accept() {
                    Ok((mut stream, _)) => {
                        let mut request_buffer = [0_u8; 2048];
                        let _ = stream.read(&mut request_buffer);
                        let response = format!(
                            "HTTP/1.1 {status_line}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                            body.len(),
                            body
                        );
                        let _ = stream.write_all(response.as_bytes());
                        let _ = stream.flush();
                        served += 1;
                        if served >= max_requests {
                            break;
                        }
                    }
                    Err(err) if err.kind() == ErrorKind::WouldBlock => {
                        if Instant::now() >= deadline {
                            break;
                        }
                        std::thread::sleep(Duration::from_millis(25));
                    }
                    Err(_) => break,
                }
            }
        })
    }

    fn spawn_single_response_server(
        listener: TcpListener,
        status_line: &str,
        content_type: &str,
        body: String,
    ) -> std::thread::JoinHandle<()> {
        spawn_response_server(listener, status_line, content_type, body, 1)
    }

    fn spawn_multi_response_server(
        listener: TcpListener,
        status_line: &str,
        content_type: &str,
        body: String,
        max_requests: usize,
    ) -> std::thread::JoinHandle<()> {
        spawn_response_server(listener, status_line, content_type, body, max_requests)
    }

    #[test]
    fn cli_usage_exposes_agent_entrypoints() {
        let usage = cli_usage_text();
        assert!(usage.contains("cortex capabilities --json"));
        assert!(usage.contains("cortex robot-docs guide"));
        assert!(usage.contains("Agent surfaces:"));
    }

    #[test]
    fn cli_capabilities_payload_has_agent_contract() {
        let payload = cli_capabilities_payload();
        assert_eq!(
            payload["contract_version"],
            CLI_CAPABILITIES_CONTRACT_VERSION
        );
        assert_eq!(
            payload["tool"]["default_port"].as_u64(),
            Some(DEFAULT_CORTEX_PORT as u64)
        );
        assert_eq!(payload["commands"]["paths"]["output"], "json");
        assert_eq!(payload["exit_codes"]["0"], "success");
    }

    #[test]
    fn robot_docs_guide_is_paste_ready_for_agents() {
        let guide = cli_robot_docs_guide();
        assert!(guide.contains("cortex capabilities --json"));
        assert!(guide.contains("cortex boot --json"));
        assert!(guide.contains("Danger gates:"));
        assert!(guide.contains("Treat exit code 0 as success"));
    }

    #[test]
    fn unknown_command_message_suggests_likely_agent_surface() {
        let message = unknown_cli_command_message("capability");
        assert!(message.contains("Unknown command: capability"));
        assert!(message.contains("Did you mean: `cortex capabilities --json`?"));
        assert!(message.contains("cortex help"));
    }

    #[test]
    fn spawned_owner_parent_probe_child_process() {
        if std::env::var(SPAWN_PARENT_TEST_CHILD_ENV).ok().as_deref() != Some("1") {
            return;
        }
        std::thread::sleep(Duration::from_millis(300));
    }

    #[test]
    fn control_center_lock_holder_child_process() {
        if std::env::var(CONTROL_CENTER_LOCK_TEST_CHILD_ENV)
            .ok()
            .as_deref()
            != Some("1")
        {
            return;
        }
        let home = std::env::var(CONTROL_CENTER_LOCK_TEST_HOME_ENV)
            .expect("control-center lock test home env missing");
        let ready_file = std::env::var(CONTROL_CENTER_LOCK_TEST_READY_ENV)
            .expect("control-center lock ready marker env missing");
        let hold_ms = std::env::var(CONTROL_CENTER_LOCK_TEST_HOLD_MS_ENV)
            .ok()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(1500);
        let lock_path = PathBuf::from(home)
            .join("runtime")
            .join(CONTROL_CENTER_LOCK_FILE);
        if let Some(parent) = lock_path.parent() {
            std::fs::create_dir_all(parent).expect("create lock parent dir");
        }
        let lock_file = std::fs::OpenOptions::new()
            .create(true)
            .read(true)
            .write(true)
            .truncate(false)
            .open(&lock_path)
            .expect("open lock file");
        lock_file
            .try_lock_exclusive()
            .expect("acquire control-center lock");
        std::fs::write(ready_file, b"locked").expect("write lock ready marker");
        std::thread::sleep(Duration::from_millis(hold_ms));
    }

    fn wait_for_control_center_lock(paths: &auth::CortexPaths, timeout: Duration) -> bool {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            if control_center_is_active(paths).unwrap_or(false) {
                return true;
            }
            std::thread::sleep(Duration::from_millis(25));
        }
        false
    }

    #[test]
    fn backfill_batch_may_have_more_only_when_a_table_hits_limit() {
        assert!(!backfill_batch_may_have_more(0, 0, 32));
        assert!(!backfill_batch_may_have_more(31, 8, 32));
        assert!(!backfill_batch_may_have_more(8, 31, 32));
        assert!(backfill_batch_may_have_more(32, 8, 32));
        assert!(backfill_batch_may_have_more(8, 32, 32));
        assert!(backfill_batch_may_have_more(32, 32, 32));
    }

    #[test]
    fn collect_unembedded_targets_for_model_rebuilds_mismatched_embeddings() {
        let conn = rusqlite::Connection::open_in_memory().expect("open sqlite");
        crate::db::configure(&conn).expect("configure sqlite");
        crate::db::initialize_schema(&conn).expect("initialize schema");
        crate::db::run_pending_migrations(&conn);

        conn.execute(
            "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
             VALUES (?1, ?2, 'note', 'active', 1.0, datetime('now'), datetime('now'))",
            rusqlite::params!["legacy memory", "memory::legacy"],
        )
        .expect("insert memory legacy");
        let memory_legacy_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
             VALUES (?1, ?2, 'note', 'active', 1.0, datetime('now'), datetime('now'))",
            rusqlite::params!["current memory", "memory::current"],
        )
        .expect("insert memory current");
        let memory_current_id = conn.last_insert_rowid();

        conn.execute(
            "INSERT INTO decisions (decision, context, source_agent, status, score, merged_count, quality, created_at, updated_at)
             VALUES (?1, ?2, 'tester', 'active', 1.0, 0, 70, datetime('now'), datetime('now'))",
            rusqlite::params!["legacy decision", "ctx::legacy"],
        )
        .expect("insert decision legacy");
        let decision_legacy_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO decisions (decision, context, source_agent, status, score, merged_count, quality, created_at, updated_at)
             VALUES (?1, ?2, 'tester', 'active', 1.0, 0, 70, datetime('now'), datetime('now'))",
            rusqlite::params!["current decision", "ctx::current"],
        )
        .expect("insert decision current");
        let decision_current_id = conn.last_insert_rowid();

        let sample_blob = crate::embeddings::vector_to_blob(&[0.1, 0.2, 0.3]);
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('memory', ?1, ?2, 'other-model')",
            rusqlite::params![memory_legacy_id, sample_blob.clone()],
        )
        .expect("insert legacy memory embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('memory', ?1, ?2, 'all-MiniLM-L6-v2')",
            rusqlite::params![memory_current_id, sample_blob.clone()],
        )
        .expect("insert current memory embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('decision', ?1, ?2, 'OTHER-MODEL')",
            rusqlite::params![decision_legacy_id, sample_blob.clone()],
        )
        .expect("insert legacy decision embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('decision', ?1, ?2, 'all-minilm-l6-v2')",
            rusqlite::params![decision_current_id, sample_blob],
        )
        .expect("insert current decision embedding");

        let (memories, decisions) =
            collect_unembedded_targets_for_model(&conn, "all-minilm-l6-v2", 256);
        let memory_ids: std::collections::HashSet<i64> =
            memories.iter().map(|(id, _)| *id).collect();
        let decision_ids: std::collections::HashSet<i64> =
            decisions.iter().map(|(id, _)| *id).collect();

        assert!(
            memory_ids.contains(&memory_legacy_id),
            "mismatched memory model should be queued for re-embedding"
        );
        assert!(
            !memory_ids.contains(&memory_current_id),
            "matching memory model should not be queued"
        );
        assert!(
            decision_ids.contains(&decision_legacy_id),
            "mismatched decision model should be queued for re-embedding"
        );
        assert!(
            !decision_ids.contains(&decision_current_id),
            "matching decision model should not be queued"
        );
    }

    #[test]
    fn collect_unembedded_targets_for_model_respects_limit_per_table() {
        let conn = rusqlite::Connection::open_in_memory().expect("open sqlite");
        crate::db::configure(&conn).expect("configure sqlite");
        crate::db::initialize_schema(&conn).expect("initialize schema");
        crate::db::run_pending_migrations(&conn);

        for idx in 0..3 {
            conn.execute(
                "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
                 VALUES (?1, ?2, 'note', 'active', 1.0, datetime('now'), datetime('now'))",
                rusqlite::params![format!("memory-{idx}"), format!("memory::{idx}")],
            )
            .expect("insert memory");
        }
        for idx in 0..3 {
            conn.execute(
                "INSERT INTO decisions (decision, context, status, score, merged_count, quality, created_at, updated_at)
                 VALUES (?1, ?2, 'active', 1.0, 0, 70, datetime('now'), datetime('now'))",
                rusqlite::params![format!("decision-{idx}"), format!("decision::{idx}")],
            )
            .expect("insert decision");
        }

        let (memories, decisions) =
            collect_unembedded_targets_for_model(&conn, "all-minilm-l6-v2", 1);
        assert_eq!(memories.len(), 1, "memory queue should honor LIMIT");
        assert_eq!(decisions.len(), 1, "decision queue should honor LIMIT");
        assert_eq!(memories[0].0, 1, "memory selection should be deterministic");
        assert_eq!(
            decisions[0].0, 1,
            "decision selection should be deterministic"
        );
    }

    #[test]
    fn count_unembedded_targets_for_model_reports_model_specific_backlog() {
        let conn = rusqlite::Connection::open_in_memory().expect("open sqlite");
        crate::db::configure(&conn).expect("configure sqlite");
        crate::db::initialize_schema(&conn).expect("initialize schema");
        crate::db::run_pending_migrations(&conn);

        conn.execute(
            "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
             VALUES (?1, ?2, 'note', 'active', 1.0, datetime('now'), datetime('now'))",
            rusqlite::params!["memory-backlog", "tests::count"],
        )
        .expect("insert active backlog memory");
        let memory_backlog_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
             VALUES (?1, ?2, 'note', 'active', 1.0, datetime('now'), datetime('now'))",
            rusqlite::params!["memory-current", "tests::count"],
        )
        .expect("insert active current memory");
        let memory_current_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO memories (text, source, type, status, score, created_at, updated_at)
             VALUES (?1, ?2, 'note', 'archived', 1.0, datetime('now'), datetime('now'))",
            rusqlite::params!["memory-archived", "tests::count"],
        )
        .expect("insert archived memory");

        conn.execute(
            "INSERT INTO decisions (decision, context, source_agent, status, score, merged_count, quality, created_at, updated_at)
             VALUES (?1, ?2, 'tester', 'active', 1.0, 0, 70, datetime('now'), datetime('now'))",
            rusqlite::params!["decision-backlog", "tests::count"],
        )
        .expect("insert active backlog decision");
        let decision_backlog_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO decisions (decision, context, source_agent, status, score, merged_count, quality, created_at, updated_at)
             VALUES (?1, ?2, 'tester', 'active', 1.0, 0, 70, datetime('now'), datetime('now'))",
            rusqlite::params!["decision-current", "tests::count"],
        )
        .expect("insert active current decision");
        let decision_current_id = conn.last_insert_rowid();
        conn.execute(
            "INSERT INTO decisions (decision, context, source_agent, status, score, merged_count, quality, created_at, updated_at)
             VALUES (?1, ?2, 'tester', 'archived', 1.0, 0, 70, datetime('now'), datetime('now'))",
            rusqlite::params!["decision-archived", "tests::count"],
        )
        .expect("insert archived decision");

        let sample_blob = crate::embeddings::vector_to_blob(&[0.1, 0.2, 0.3]);
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('memory', ?1, ?2, 'other-model')",
            rusqlite::params![memory_backlog_id, sample_blob.clone()],
        )
        .expect("insert legacy memory embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('memory', ?1, ?2, 'all-minilm-l6-v2')",
            rusqlite::params![memory_current_id, sample_blob.clone()],
        )
        .expect("insert current memory embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('decision', ?1, ?2, 'other-model')",
            rusqlite::params![decision_backlog_id, sample_blob.clone()],
        )
        .expect("insert legacy decision embedding");
        conn.execute(
            "INSERT INTO embeddings (target_type, target_id, vector, model) VALUES ('decision', ?1, ?2, 'all-MiniLM-L6-v2')",
            rusqlite::params![decision_current_id, sample_blob],
        )
        .expect("insert current decision embedding");

        let (memory_count, decision_count) =
            count_unembedded_targets_for_model(&conn, "all-minilm-l6-v2");
        assert_eq!(
            memory_count, 1,
            "exactly one active memory should be pending"
        );
        assert_eq!(
            decision_count, 1,
            "exactly one active decision should be pending"
        );
    }

    #[test]
    fn rotate_backups_keeps_three_most_recent_files() {
        let backup_dir = temp_test_dir("backup_rotation");
        fs::create_dir_all(&backup_dir).unwrap();

        for idx in 0..5 {
            let path = backup_dir.join(format!("cortex-2026040{}.db", idx + 1));
            fs::write(&path, format!("backup-{idx}")).unwrap();
            std::thread::sleep(Duration::from_millis(20));
        }

        let removed = rotate_backups(&backup_dir, BACKUP_RETENTION_COUNT).unwrap();
        assert_eq!(removed, 2);

        let mut remaining: Vec<String> = fs::read_dir(&backup_dir)
            .unwrap()
            .filter_map(|entry| entry.ok())
            .map(|entry| entry.file_name().to_string_lossy().to_string())
            .collect();
        remaining.sort();

        assert_eq!(
            remaining,
            vec![
                "cortex-20260403.db".to_string(),
                "cortex-20260404.db".to_string(),
                "cortex-20260405.db".to_string(),
            ]
        );

        let _ = fs::remove_dir_all(&backup_dir);
    }

    #[test]
    fn cleanup_bridge_backups_requires_schema_version_five_or_higher() {
        let home_dir = temp_test_dir("bridge_backups");
        let bridge_dir = home_dir.join("bridge-backups");
        fs::create_dir_all(&bridge_dir).unwrap();
        fs::write(bridge_dir.join("legacy.txt"), "legacy").unwrap();

        assert!(!cleanup_bridge_backups(&home_dir, 4));
        assert!(bridge_dir.exists());

        assert!(cleanup_bridge_backups(&home_dir, 5));
        assert!(!bridge_dir.exists());

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn rotate_log_file_replaces_existing_rotation_and_creates_fresh_log() {
        let home_dir = temp_test_dir("log_rotation");
        fs::create_dir_all(&home_dir).unwrap();

        let log_path = home_dir.join("daemon.log");
        let rotated_path = home_dir.join("daemon.log.1");
        fs::write(&rotated_path, "old-rotation").unwrap();
        fs::write(&log_path, vec![b'x'; (LOG_ROTATION_BYTES as usize) + 1]).unwrap();

        assert!(rotate_log_file(&home_dir, "daemon.log").unwrap());
        assert!(log_path.exists());
        assert_eq!(fs::metadata(&log_path).unwrap().len(), 0);
        assert_eq!(
            fs::metadata(&rotated_path).unwrap().len(),
            LOG_ROTATION_BYTES + 1
        );

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn run_backup_cleanup_dry_run_reports_files_without_deleting_them() {
        let backup_dir = temp_test_dir("backup_cleanup_dry_run");
        fs::create_dir_all(&backup_dir).unwrap();

        for idx in 0..4 {
            let path = backup_dir.join(format!("cortex-2026040{}.db", idx + 1));
            fs::write(&path, format!("backup-{idx}")).unwrap();
            std::thread::sleep(Duration::from_millis(20));
        }

        let lines = run_backup_cleanup(&backup_dir, true);
        assert_eq!(lines.len(), 1);
        assert!(lines[0].starts_with("DELETE backups/"));
        assert_eq!(fs::read_dir(&backup_dir).unwrap().count(), 4);

        let _ = fs::remove_dir_all(&backup_dir);
    }

    #[test]
    fn acquire_runtime_lock_rejects_duplicate_serve_startup() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("runtime_lock");
        fs::create_dir_all(&home_dir).unwrap();
        let global_lock_home = temp_test_dir("runtime_lock_global");
        fs::create_dir_all(&global_lock_home).unwrap();
        let global_lock_home_str = global_lock_home.to_string_lossy().to_string();
        let _global_lock_home = ScopedEnvVar::set("CORTEX_GLOBAL_LOCK_HOME", &global_lock_home_str);

        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let first_lock = acquire_runtime_lock(&paths).unwrap();
        let err = acquire_runtime_lock(&paths).unwrap_err();

        assert!(err.contains("another cortex instance"));

        drop(first_lock);
        let _ = fs::remove_dir_all(&home_dir);
        let _ = fs::remove_dir_all(&global_lock_home);
    }

    #[test]
    fn control_center_lock_detection_reports_cross_process_holder() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("control_center_lock_detection");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        let ready_file = home_dir.join("control-center-lock-ready");

        assert!(
            !control_center_is_active(&paths).expect("probe lock without holder"),
            "lock should not appear active before holder starts"
        );

        let current_exe = std::env::current_exe().expect("resolve current test binary path");
        let mut child = Command::new(current_exe)
            .arg("--exact")
            .arg("tests::control_center_lock_holder_child_process")
            .arg("--nocapture")
            .env(CONTROL_CENTER_LOCK_TEST_CHILD_ENV, "1")
            .env(CONTROL_CENTER_LOCK_TEST_HOME_ENV, &home_str)
            .env(
                CONTROL_CENTER_LOCK_TEST_READY_ENV,
                ready_file.to_string_lossy().to_string(),
            )
            .env(CONTROL_CENTER_LOCK_TEST_HOLD_MS_ENV, "30000")
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn lock-holder child");

        let ready_deadline = Instant::now() + Duration::from_secs(5);
        while !ready_file.exists() {
            if Instant::now() >= ready_deadline {
                let _ = child.kill();
                let _ = child.wait();
                panic!("lock-holder child never reported readiness");
            }
            std::thread::sleep(Duration::from_millis(25));
        }

        assert!(
            wait_for_control_center_lock(&paths, Duration::from_secs(3)),
            "lock should appear active while child holds cross-process lock"
        );

        let status = child.wait().expect("wait lock-holder child");
        assert!(
            status.success(),
            "lock-holder child should exit successfully"
        );

        assert!(
            !control_center_is_active(&paths).expect("probe lock after holder exits"),
            "lock should be released after child exits"
        );

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

    #[test]
    fn acquire_runtime_lock_waits_for_handoff_when_enabled() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("runtime_lock_handoff");
        fs::create_dir_all(&home_dir).unwrap();
        let global_lock_home = temp_test_dir("runtime_lock_handoff_global");
        fs::create_dir_all(&global_lock_home).unwrap();
        let global_lock_home_str = global_lock_home.to_string_lossy().to_string();
        let _global_lock_home = ScopedEnvVar::set("CORTEX_GLOBAL_LOCK_HOME", &global_lock_home_str);

        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let first_lock = acquire_runtime_lock(&paths).unwrap();
        let _wait_lock_flag = ScopedEnvVar::set("CORTEX_WAIT_FOR_DAEMON_LOCK", "1");
        let _wait_secs_flag = ScopedEnvVar::set("CORTEX_DAEMON_LOCK_WAIT_SECS", "1");

        let releaser = std::thread::spawn(move || {
            std::thread::sleep(Duration::from_millis(300));
            drop(first_lock);
        });

        let second_lock = acquire_runtime_lock(&paths).expect("lock handoff should succeed");
        drop(second_lock);
        releaser.join().unwrap();

        let _ = fs::remove_dir_all(&home_dir);
        let _ = fs::remove_dir_all(&global_lock_home);
    }

    #[test]
    fn acquire_runtime_lock_rejects_concurrent_startup_burst() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("runtime_lock_burst");
        fs::create_dir_all(&home_dir).unwrap();
        let global_lock_home = temp_test_dir("runtime_lock_burst_global");
        fs::create_dir_all(&global_lock_home).unwrap();
        let global_lock_home_str = global_lock_home.to_string_lossy().to_string();
        let _global_lock_home = ScopedEnvVar::set("CORTEX_GLOBAL_LOCK_HOME", &global_lock_home_str);

        let home_str = home_dir.to_string_lossy().to_string();
        let paths = Arc::new(auth::CortexPaths::resolve_with_overrides(
            Some(&home_str),
            None,
            Some(7437),
            None,
        ));

        let first_lock = acquire_runtime_lock(&paths).expect("first runtime lock must succeed");
        let workers: Vec<_> = (0..12)
            .map(|_| {
                let worker_paths = Arc::clone(&paths);
                std::thread::spawn(move || acquire_runtime_lock(&worker_paths).is_err())
            })
            .collect();
        let failures = workers
            .into_iter()
            .map(|worker| worker.join().expect("join worker"))
            .filter(|failed| *failed)
            .count();
        assert_eq!(
            failures, 12,
            "all concurrent startups should fail while runtime lock is held"
        );

        drop(first_lock);
        let second_lock =
            acquire_runtime_lock(&paths).expect("lock should be reacquired after release");
        drop(second_lock);

        let _ = fs::remove_dir_all(&home_dir);
        let _ = fs::remove_dir_all(&global_lock_home);
    }

    #[test]
    fn run_stale_pid_cleanup_keeps_lock_file() {
        let home_dir = temp_test_dir("stale_pid_cleanup");
        fs::create_dir_all(&home_dir).unwrap();

        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        fs::write(&paths.pid, "999999").unwrap();
        fs::write(&paths.lock, "lock-held").unwrap();

        let dry_run = run_stale_pid_cleanup(&paths, true);
        assert_eq!(
            dry_run,
            vec!["DELETE cortex.pid (process 999999 not running)"]
        );
        assert!(paths.pid.exists());
        assert!(paths.lock.exists());

        let apply = run_stale_pid_cleanup(&paths, false);
        assert_eq!(
            apply,
            vec!["DELETE cortex.pid (process 999999 not running)"]
        );
        assert!(!paths.pid.exists());
        assert!(paths.lock.exists());

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn spawned_owner_requires_parent_pid_only_for_non_control_center_owner() {
        assert!(spawned_owner_requires_parent_pid(Some("cli-mcp")));
        assert!(spawned_owner_requires_parent_pid(Some("plugin-claude")));
        assert!(!spawned_owner_requires_parent_pid(Some("control-center")));
        assert!(!spawned_owner_requires_parent_pid(None));
    }

    #[test]
    fn is_control_center_owner_is_case_insensitive() {
        assert!(is_control_center_owner(Some("control-center")));
        assert!(is_control_center_owner(Some("CoNtRoL-CeNtEr")));
        assert!(!is_control_center_owner(Some("plugin-claude")));
        assert!(!is_control_center_owner(None));
    }

    #[test]
    fn app_managed_startup_heavy_delay_only_applies_to_control_center_owner() {
        let _env_guard = env_guard();
        std::env::remove_var(APP_MANAGED_STARTUP_DELAY_ENV);
        assert_eq!(
            app_managed_startup_heavy_delay(Some("control-center")),
            Duration::from_secs(APP_MANAGED_STARTUP_HEAVY_DELAY_SECS)
        );
        assert_eq!(
            app_managed_startup_heavy_delay(Some("plugin-claude")),
            Duration::from_secs(0)
        );

        let _startup_delay = ScopedEnvVar::set(APP_MANAGED_STARTUP_DELAY_ENV, "0");
        assert_eq!(
            app_managed_startup_heavy_delay(Some("control-center")),
            Duration::from_secs(0)
        );

        drop(_startup_delay);
        let _excessive_delay = ScopedEnvVar::set(APP_MANAGED_STARTUP_DELAY_ENV, "777");
        assert_eq!(
            app_managed_startup_heavy_delay(Some("control-center")),
            Duration::from_secs(APP_MANAGED_STARTUP_HEAVY_DELAY_MAX_SECS)
        );
    }

    #[test]
    fn startup_schedule_uses_non_app_defaults_for_plugin_owner() {
        let _env_guard = env_guard();
        let _app_delay = ScopedEnvVar::set(APP_MANAGED_STARTUP_DELAY_ENV, "");
        let _index_delay = ScopedEnvVar::set(STARTUP_INDEX_DELAY_ENV, "");
        let _aging_delay = ScopedEnvVar::set(STARTUP_AGING_DELAY_ENV, "");
        let _embed_delay = ScopedEnvVar::set(STARTUP_EMBED_DELAY_ENV, "");
        let _crystallize_delay = ScopedEnvVar::set(STARTUP_CRYSTALLIZE_DELAY_ENV, "");
        let _storage_delay = ScopedEnvVar::set(STARTUP_STORAGE_GOVERNOR_DELAY_ENV, "");

        let schedule = startup_schedule(Some("plugin-claude"));
        assert_eq!(
            schedule.index,
            Duration::from_secs(DEFAULT_STARTUP_INDEX_DELAY_SECS)
        );
        assert_eq!(
            schedule.aging,
            Duration::from_secs(DEFAULT_STARTUP_AGING_DELAY_SECS)
        );
        assert_eq!(
            schedule.embed,
            Duration::from_secs(DEFAULT_STARTUP_EMBED_DELAY_SECS)
        );
        assert_eq!(
            schedule.crystallize,
            Duration::from_secs(DEFAULT_STARTUP_CRYSTALLIZE_DELAY_SECS)
        );
        assert_eq!(
            schedule.storage_governor_initial,
            Duration::from_secs(DEFAULT_STARTUP_STORAGE_GOVERNOR_DELAY_SECS)
        );
    }

    #[test]
    fn startup_schedule_applies_app_managed_offsets_for_control_center() {
        let _env_guard = env_guard();
        let _app_delay = ScopedEnvVar::set(APP_MANAGED_STARTUP_DELAY_ENV, "10");
        let _index_delay = ScopedEnvVar::set(STARTUP_INDEX_DELAY_ENV, "1");
        let _aging_delay = ScopedEnvVar::set(STARTUP_AGING_DELAY_ENV, "1");
        let _embed_delay = ScopedEnvVar::set(STARTUP_EMBED_DELAY_ENV, "1");
        let _crystallize_delay = ScopedEnvVar::set(STARTUP_CRYSTALLIZE_DELAY_ENV, "1");
        let _storage_delay = ScopedEnvVar::set(STARTUP_STORAGE_GOVERNOR_DELAY_ENV, "7");

        let schedule = startup_schedule(Some("control-center"));
        assert_eq!(schedule.index, Duration::from_secs(10));
        assert_eq!(
            schedule.aging,
            Duration::from_secs(10 + APP_MANAGED_AGING_STARTUP_OFFSET_SECS)
        );
        assert_eq!(
            schedule.embed,
            Duration::from_secs(10 + APP_MANAGED_EMBED_STARTUP_OFFSET_SECS)
        );
        assert_eq!(
            schedule.crystallize,
            Duration::from_secs(10 + APP_MANAGED_CRYSTALLIZE_STARTUP_OFFSET_SECS)
        );
        assert_eq!(schedule.storage_governor_initial, Duration::from_secs(7));
    }

    #[test]
    fn background_db_lock_wait_env_is_clamped() {
        let _env_guard = env_guard();
        let _small = ScopedEnvVar::set(BACKGROUND_DB_LOCK_MAX_WAIT_MS_ENV, "1");
        assert_eq!(background_db_lock_max_wait(), Duration::from_millis(100));
        drop(_small);

        let _large = ScopedEnvVar::set(BACKGROUND_DB_LOCK_MAX_WAIT_MS_ENV, "70000");
        assert_eq!(background_db_lock_max_wait(), Duration::from_millis(60_000));
    }

    #[test]
    fn spawned_owner_runtime_claim_requires_parent_linkage_for_plugin_owner() {
        let home_dir = temp_test_dir("owner_runtime_parent");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let err =
            validate_spawned_owner_runtime_claim(&paths, Some("plugin-claude"), None, None, None)
                .unwrap_err();
        assert!(err.contains(SPAWN_PARENT_PID_ENV));

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

    #[test]
    fn spawned_owner_runtime_claim_rejects_missing_owner_token_when_parent_set() {
        let home_dir = temp_test_dir("owner_runtime_token");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let parent_pid = std::process::id();
        let parent_start_time =
            process_pid_start_time(parent_pid).expect("current process start time should resolve");
        let err = validate_spawned_owner_runtime_claim(
            &paths,
            Some("plugin-claude"),
            Some(parent_pid),
            Some(parent_start_time),
            None,
        )
        .unwrap_err();
        assert!(err.contains("missing ownership token"));

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

    #[test]
    fn spawned_owner_runtime_claim_rejects_dead_parent_process() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("owner_runtime_dead_parent");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let current_exe = std::env::current_exe().expect("resolve current test binary path");
        let mut child = Command::new(current_exe)
            .arg("--exact")
            .arg("tests::spawned_owner_parent_probe_child_process")
            .arg("--nocapture")
            .env(SPAWN_PARENT_TEST_CHILD_ENV, "1")
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn parent probe child");
        let parent_pid = child.id();
        let deadline = Instant::now() + Duration::from_secs(5);
        let parent_start_time = loop {
            if let Some(start_time) = process_pid_start_time(parent_pid) {
                break start_time;
            }
            if Instant::now() >= deadline {
                let _ = child.kill();
                let _ = child.wait();
                panic!("failed to resolve child start time");
            }
            std::thread::sleep(Duration::from_millis(25));
        };
        let status = child.wait().expect("wait on parent probe child");
        assert!(
            status.success(),
            "parent probe child should exit successfully"
        );

        let err = validate_spawned_owner_runtime_claim(
            &paths,
            Some("plugin-claude"),
            Some(parent_pid),
            Some(parent_start_time),
            None,
        )
        .unwrap_err();
        assert!(err.contains("not running during ownership claim validation"));

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

    #[tokio::test]
    async fn spawn_parent_orphan_watch_task_triggers_shutdown_when_parent_identity_breaks() {
        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
        let shared_shutdown_tx = Arc::new(tokio::sync::Mutex::new(Some(shutdown_tx)));
        let probe_count = Arc::new(AtomicUsize::new(0));
        let probe_count_for_task = Arc::clone(&probe_count);
        let watcher = spawn_parent_orphan_watch_task(
            Arc::clone(&shared_shutdown_tx),
            4242,
            123,
            Duration::from_millis(5),
            move |_, _| probe_count_for_task.fetch_add(1, Ordering::SeqCst) == 0,
        );

        tokio::time::timeout(Duration::from_millis(250), shutdown_rx)
            .await
            .expect("watcher should signal shutdown when parent identity breaks")
            .expect("shutdown channel should deliver signal");
        watcher
            .await
            .expect("spawn-parent watcher task should exit cleanly");
        assert!(
            probe_count.load(Ordering::SeqCst) >= 2,
            "watcher should probe parent identity more than once"
        );
        assert!(
            shared_shutdown_tx.lock().await.is_none(),
            "shutdown sender should be consumed after parent identity break"
        );
    }

    #[tokio::test]
    async fn spawn_parent_orphan_watch_task_detects_real_parent_exit() {
        let mut parent_probe_child = {
            let _env_guard = env_guard();
            let current_exe = std::env::current_exe().expect("resolve current test binary path");
            Command::new(current_exe)
                .arg("--exact")
                .arg("tests::spawned_owner_parent_probe_child_process")
                .arg("--nocapture")
                .env(SPAWN_PARENT_TEST_CHILD_ENV, "1")
                .stdout(Stdio::null())
                .stderr(Stdio::null())
                .spawn()
                .expect("spawn parent probe child")
        };
        let parent_pid = parent_probe_child.id();
        let deadline = Instant::now() + Duration::from_secs(5);
        let parent_start_time = loop {
            if let Some(start_time) = process_pid_start_time(parent_pid) {
                break start_time;
            }
            if Instant::now() >= deadline {
                let _ = parent_probe_child.kill();
                let _ = parent_probe_child.wait();
                panic!("failed to resolve child process start time");
            }
            std::thread::sleep(Duration::from_millis(20));
        };

        let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>();
        let shared_shutdown_tx = Arc::new(tokio::sync::Mutex::new(Some(shutdown_tx)));
        let watcher = spawn_parent_orphan_watch_task(
            Arc::clone(&shared_shutdown_tx),
            parent_pid,
            parent_start_time,
            Duration::from_millis(20),
            process_pid_identity_matches,
        );

        let status = parent_probe_child
            .wait()
            .expect("wait on parent probe child");
        assert!(
            status.success(),
            "parent probe child should exit successfully"
        );
        tokio::time::timeout(Duration::from_secs(2), shutdown_rx)
            .await
            .expect("watcher should observe real parent process exit")
            .expect("shutdown signal should be delivered");
        watcher
            .await
            .expect("spawn-parent watcher task should exit cleanly");
        assert!(
            shared_shutdown_tx.lock().await.is_none(),
            "shutdown sender should be consumed after real parent exit"
        );
    }

    #[test]
    fn spawned_owner_runtime_claim_allows_unspawned_control_center_mode() {
        let home_dir = temp_test_dir("owner_runtime_unspawned");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        validate_spawned_owner_runtime_claim(&paths, Some("control-center"), None, None, None)
            .expect("direct control-center owner mode should remain compatible");

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

    #[test]
    fn spawned_owner_runtime_claim_rejects_missing_parent_start_time_when_parent_set() {
        let home_dir = temp_test_dir("owner_runtime_parent_start");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let parent_pid = std::process::id();
        let err = validate_spawned_owner_runtime_claim(
            &paths,
            Some("plugin-claude"),
            Some(parent_pid),
            None,
            None,
        )
        .unwrap_err();
        assert!(err.contains(SPAWN_PARENT_START_TIME_ENV));

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

    #[test]
    fn spawned_owner_runtime_claim_rejects_parent_start_time_mismatch() {
        let home_dir = temp_test_dir("owner_runtime_parent_start_mismatch");
        std::fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let parent_pid = std::process::id();
        let err = validate_spawned_owner_runtime_claim(
            &paths,
            Some("plugin-claude"),
            Some(parent_pid),
            Some(0),
            Some("invalid-token"),
        )
        .unwrap_err();
        assert!(err.contains("start-time mismatch"));

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

    #[test]
    fn parse_flag_usize_validates_and_parses_values() {
        let args = vec![
            "--agent".to_string(),
            "codex".to_string(),
            "--budget".to_string(),
            "900".to_string(),
        ];
        assert_eq!(parse_flag_usize(&args, "--budget").unwrap(), Some(900));

        let missing_value = vec!["--budget".to_string()];
        assert!(parse_flag_usize(&missing_value, "--budget")
            .unwrap_err()
            .contains("missing value"));

        let invalid_value = vec!["--budget".to_string(), "abc".to_string()];
        assert!(parse_flag_usize(&invalid_value, "--budget")
            .unwrap_err()
            .contains("invalid value"));

        let zero_value = vec!["--budget".to_string(), "0".to_string()];
        assert!(parse_flag_usize(&zero_value, "--budget")
            .unwrap_err()
            .contains("must be >= 1"));
    }

    #[test]
    fn sync_changeset_filename_filter_is_strict() {
        assert!(is_sync_changeset_file_name(
            "changeset-abc-20260419T101112000Z.json"
        ));
        assert!(is_sync_changeset_file_name("changeset-node-1.json"));
        assert!(!is_sync_changeset_file_name("changeset-node-1.txt"));
        assert!(!is_sync_changeset_file_name("metrics.json"));
    }

    #[test]
    fn resolve_sync_since_prefers_override_then_cursor_file() {
        let home_dir = temp_test_dir("sync_cursor_resolution");
        fs::create_dir_all(&home_dir).expect("create temp home");
        let cursor_file = home_dir.join("cursor.txt");

        let override_since = "2026-04-19T00:00:00Z";
        assert_eq!(
            resolve_sync_since(Some(override_since), Some(&cursor_file)),
            Some(override_since.to_string())
        );
        assert_eq!(resolve_sync_since(None, Some(&cursor_file)), None);

        write_sync_cursor_file(&cursor_file, "2026-04-20T00:00:00Z").expect("write cursor");
        assert_eq!(
            resolve_sync_since(None, Some(&cursor_file)),
            Some("2026-04-20T00:00:00Z".to_string())
        );

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

    #[test]
    fn sync_watch_candidate_collection_skips_local_site_files() {
        let watch_dir = temp_test_dir("sync_watch_candidates");
        fs::create_dir_all(&watch_dir).expect("create watch dir");
        fs::write(
            watch_dir.join("changeset-local-site-20260419T100000000Z.json"),
            "{}",
        )
        .expect("write local file");
        fs::write(
            watch_dir.join("changeset-remote-site-20260419T100001000Z.json"),
            "{}",
        )
        .expect("write remote file");
        fs::write(watch_dir.join("notes.txt"), "ignore").expect("write noise file");

        let files = collect_sync_watch_import_candidates(&watch_dir, "local-site")
            .expect("collect sync watch files");
        assert_eq!(files.len(), 1);
        let name = files[0]
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or_default()
            .to_string();
        assert_eq!(name, "changeset-remote-site-20260419T100001000Z.json");

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

    #[test]
    fn event_type_count_helpers_return_expected_rows() {
        let conn = rusqlite::Connection::open_in_memory().expect("open in-memory db");
        conn.execute(
            "CREATE TABLE events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                type TEXT NOT NULL,
                data TEXT,
                source_agent TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
            )",
            [],
        )
        .expect("create events table");
        conn.execute(
            "INSERT INTO events (type, data, source_agent) VALUES ('decision_stored', '{}', 'test')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO events (type, data, source_agent) VALUES ('decision_stored', '{}', 'test')",
            [],
        )
        .unwrap();
        conn.execute(
            "INSERT INTO events (type, data, source_agent) VALUES ('recall_query', '{}', 'test')",
            [],
        )
        .unwrap();

        assert_eq!(event_type_count(&conn, "decision_stored"), 2);
        assert_eq!(event_type_count(&conn, "recall_query"), 1);
        assert_eq!(event_type_count(&conn, "missing"), 0);

        let top = top_event_type_counts(&conn, 2);
        assert_eq!(top.len(), 2);
        assert_eq!(top[0], ("decision_stored".to_string(), 2));
        assert_eq!(top[1], ("recall_query".to_string(), 1));
    }

    #[test]
    fn run_event_compaction_cleanup_dry_run_reports_preview_lines() {
        let home_dir = temp_test_dir("event_cleanup_dry_run");
        fs::create_dir_all(&home_dir).expect("create temp home");
        let db_path = home_dir.join("cortex.db");
        let conn = rusqlite::Connection::open(&db_path).expect("open db");
        db::configure(&conn).expect("configure db");
        conn.execute(
            "CREATE TABLE events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                type TEXT NOT NULL,
                data TEXT,
                source_agent TEXT,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP
            )",
            [],
        )
        .expect("create events table");
        for _ in 0..5 {
            conn.execute(
                "INSERT INTO events (type, data, source_agent) VALUES ('decision_stored', '{}', 'test')",
                [],
            )
            .expect("insert event");
        }
        drop(conn);

        let lines =
            run_event_compaction_cleanup(&db_path, true, 2).expect("event cleanup dry run lines");
        assert!(
            lines.iter().any(|line| line.starts_with("EVENTS before:")),
            "missing before summary: {lines:?}"
        );
        assert!(
            lines.iter().any(|line| line.contains("dry-run only")),
            "missing dry-run hint: {lines:?}"
        );

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn resolve_boot_auth_header_prefers_api_key_and_falls_back_to_token_file() {
        let home_dir = temp_test_dir("boot_auth");
        fs::create_dir_all(&home_dir).unwrap();
        let token_path = home_dir.join("cortex.token");
        fs::write(&token_path, "local-token").unwrap();

        let explicit = resolve_boot_auth_header(&token_path, Some("ctx_remote"), true);
        assert_eq!(explicit, Some("Bearer ctx_remote".to_string()));

        let fallback = resolve_boot_auth_header(&token_path, None, true);
        assert_eq!(fallback, Some("Bearer local-token".to_string()));

        fs::write(&token_path, "   ").unwrap();
        assert_eq!(resolve_boot_auth_header(&token_path, None, true), None);
        assert_eq!(resolve_boot_auth_header(&token_path, None, false), None);

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn parse_truthy_flag_accepts_expected_values() {
        assert!(parse_truthy_flag("1"));
        assert!(parse_truthy_flag("true"));
        assert!(parse_truthy_flag("YES"));
        assert!(parse_truthy_flag(" on "));
        assert!(!parse_truthy_flag("0"));
        assert!(!parse_truthy_flag("false"));
        assert!(!parse_truthy_flag(""));
    }

    #[test]
    fn single_daemon_test_bypass_flag_respects_debug_gate() {
        let _env_guard = env_guard();
        std::env::remove_var(SINGLE_DAEMON_TEST_BYPASS_ENV);
        assert!(!single_daemon_test_bypass_enabled());

        let _bypass = ScopedEnvVar::set(SINGLE_DAEMON_TEST_BYPASS_ENV, "1");
        assert_eq!(single_daemon_test_bypass_enabled(), cfg!(debug_assertions));
    }

    #[test]
    fn local_spawn_policy_fails_closed_for_marked_app_client_without_spawn_contract() {
        let _env_guard = env_guard();
        std::env::remove_var(APP_REQUIRED_ENV);
        std::env::remove_var(DAEMON_LOCAL_SPAWN_ENV);
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "codex");
        assert!(
            !local_spawn_allowed_for_request(true),
            "app-marked clients should fail closed when spawn policy is missing"
        );
    }

    #[test]
    fn local_spawn_policy_allows_explicit_opt_in_for_marked_app_client() {
        let _env_guard = env_guard();
        std::env::remove_var(APP_REQUIRED_ENV);
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "codex");
        let _local_spawn = ScopedEnvVar::set(DAEMON_LOCAL_SPAWN_ENV, "1");
        assert!(
            local_spawn_allowed_for_request(true),
            "explicit local spawn opt-in should allow startup when app-required is unset"
        );
    }

    #[test]
    fn local_spawn_policy_app_required_overrides_local_spawn_opt_in() {
        let _env_guard = env_guard();
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "codex");
        let _local_spawn = ScopedEnvVar::set(DAEMON_LOCAL_SPAWN_ENV, "1");
        let _app_required = ScopedEnvVar::set(APP_REQUIRED_ENV, "1");
        assert!(
            !local_spawn_allowed_for_request(true),
            "app-required must force attach-only behavior even when local spawn is enabled"
        );
    }

    #[test]
    fn local_spawn_policy_respects_allow_service_ensure_short_circuit() {
        let _env_guard = env_guard();
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "codex");
        let _local_spawn = ScopedEnvVar::set(DAEMON_LOCAL_SPAWN_ENV, "1");
        std::env::remove_var(APP_REQUIRED_ENV);
        assert!(
            !local_spawn_allowed_for_request(false),
            "service-ensure gate must disable local spawn regardless of env opt-ins"
        );
    }

    #[test]
    fn ensure_daemon_app_required_policy_returns_machine_readable_error() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("app_required_policy");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let _app_required = ScopedEnvVar::set(APP_REQUIRED_ENV, "1");
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "codex");
        let err = run_ensure_daemon(&paths, Some("codex"), false, false).unwrap_err();
        assert!(err.contains("APP_INIT_REQUIRED"));
        assert!(err.contains("codex"));
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn ensure_daemon_respects_local_spawn_disable_flag() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("local_spawn_disabled_policy");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let _local_spawn = ScopedEnvVar::set(DAEMON_LOCAL_SPAWN_ENV, "0");
        let _app_client = ScopedEnvVar::set(APP_CLIENT_ENV, "claude");
        let err = run_ensure_daemon(&paths, Some("claude"), false, true).unwrap_err();
        assert!(err.contains("APP_INIT_REQUIRED"));
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn ensure_daemon_blocks_local_spawn_when_control_center_lock_is_held() {
        let _env_guard = env_guard();
        let home_dir = temp_test_dir("control_center_lock_blocks_spawn");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let _home_env = ScopedEnvVar::set("HOME", &home_str);
        let _userprofile_env = ScopedEnvVar::set("USERPROFILE", &home_str);
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        let ready_file = home_dir.join("control-center-lock-ready");
        std::env::remove_var(APP_REQUIRED_ENV);
        std::env::remove_var(APP_CLIENT_ENV);
        std::env::remove_var(DAEMON_LOCAL_SPAWN_ENV);

        let current_exe = std::env::current_exe().expect("resolve current test binary path");
        let mut child = Command::new(current_exe)
            .arg("--exact")
            .arg("tests::control_center_lock_holder_child_process")
            .arg("--nocapture")
            .env(CONTROL_CENTER_LOCK_TEST_CHILD_ENV, "1")
            .env(CONTROL_CENTER_LOCK_TEST_HOME_ENV, &home_str)
            .env(
                CONTROL_CENTER_LOCK_TEST_READY_ENV,
                ready_file.to_string_lossy().to_string(),
            )
            .env(CONTROL_CENTER_LOCK_TEST_HOLD_MS_ENV, "2000")
            .stdout(Stdio::null())
            .stderr(Stdio::null())
            .spawn()
            .expect("spawn lock-holder child");

        let ready_deadline = Instant::now() + Duration::from_secs(5);
        while !ready_file.exists() {
            if Instant::now() >= ready_deadline {
                let _ = child.kill();
                let _ = child.wait();
                panic!("lock-holder child never reported readiness");
            }
            std::thread::sleep(Duration::from_millis(25));
        }

        assert!(
            wait_for_control_center_lock(&paths, Duration::from_secs(3)),
            "control-center lock should be active before ensure_daemon call"
        );

        let err = run_ensure_daemon(&paths, Some("claude"), false, true).unwrap_err();
        assert!(
            err.contains("APP_INIT_REQUIRED"),
            "control-center lock should force attach-only error: {err}"
        );

        let _ = child.kill();
        let _ = child.wait();
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn ensure_daemon_attach_only_policy_holds_under_concurrent_app_clients() {
        let _env_guard = env_guard();
        let _app_required = ScopedEnvVar::set(APP_REQUIRED_ENV, "1");
        std::env::remove_var(APP_CLIENT_ENV);
        std::env::remove_var(DAEMON_LOCAL_SPAWN_ENV);

        let agents = ["codex", "claude", "gpt5"];
        let workers: Vec<_> = agents
            .iter()
            .map(|agent| {
                let agent_name = (*agent).to_string();
                std::thread::spawn(move || {
                    let home_dir = temp_test_dir(&format!("app_required_concurrent_{agent_name}"));
                    fs::create_dir_all(&home_dir).expect("create temp home");
                    let home_str = home_dir.to_string_lossy().to_string();
                    let paths = auth::CortexPaths::resolve_with_overrides(
                        Some(&home_str),
                        None,
                        Some(7437),
                        None,
                    );
                    let err = run_ensure_daemon(&paths, Some(&agent_name), false, false)
                        .expect_err("attach-only clients should not spawn daemon");
                    let _ = fs::remove_dir_all(&home_dir);
                    (agent_name, err)
                })
            })
            .collect();

        for worker in workers {
            let (agent_name, err) = worker.join().expect("join worker");
            assert!(
                err.contains("APP_INIT_REQUIRED"),
                "missing machine-readable attach-only marker for {agent_name}: {err}"
            );
            assert!(
                err.contains(&agent_name),
                "attach-only error should identify requesting agent {agent_name}: {err}"
            );
        }
    }

    #[test]
    fn ensure_daemon_attach_only_policy_holds_under_cross_surface_concurrency() {
        let _env_guard = env_guard();
        let _app_required = ScopedEnvVar::set(APP_REQUIRED_ENV, "1");
        std::env::remove_var(APP_CLIENT_ENV);
        std::env::remove_var(DAEMON_LOCAL_SPAWN_ENV);

        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("reserve port");
        let port = listener.local_addr().expect("listener addr").port();
        drop(listener);

        let home_dir = temp_test_dir("app_required_cross_surface_shared_home");
        fs::create_dir_all(&home_dir).expect("create temp home");
        let home_str = home_dir.to_string_lossy().to_string();
        let pid_path = home_dir.join("cortex.pid");

        let workers = vec![
            ("cli-codex".to_string(), Some("codex".to_string()), false),
            (
                "plugin-claude".to_string(),
                Some("claude-code".to_string()),
                true,
            ),
            ("direct-cli".to_string(), None, false),
        ];

        let handles: Vec<_> = workers
            .into_iter()
            .map(|(label, agent, allow_service_ensure)| {
                let worker_home = home_str.clone();
                std::thread::spawn(move || {
                    let paths = auth::CortexPaths::resolve_with_overrides(
                        Some(&worker_home),
                        None,
                        Some(port),
                        None,
                    );
                    let err =
                        run_ensure_daemon(&paths, agent.as_deref(), false, allow_service_ensure)
                            .expect_err(
                                "cross-surface attach-only callers should not spawn daemon",
                            );
                    (label, err)
                })
            })
            .collect();

        for handle in handles {
            let (label, err) = handle.join().expect("join worker");
            assert!(
                err.contains("APP_INIT_REQUIRED"),
                "cross-surface attach-only marker missing for {label}: {err}"
            );
        }
        assert!(
            !pid_path.exists(),
            "attach-only cross-surface contention should not create daemon pid file"
        );
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn disallowed_startup_binary_path_blocks_runtime_wrappers_and_temp_paths() {
        let wrapper = PathBuf::from(
            "C:/repo/daemon-rs/target/debug/daemon-lifecycle-runtime/cortex-daemon-run.exe",
        );
        assert!(is_disallowed_startup_binary_path(&wrapper));

        let wrapper_name_only = PathBuf::from("C:/repo/cortex-daemon-run");
        assert!(is_disallowed_startup_binary_path(&wrapper_name_only));

        let binary_name = if cfg!(windows) {
            "cortex.exe"
        } else {
            "cortex"
        };
        let temp_candidate = std::env::temp_dir().join("cortex").join(binary_name);
        assert!(is_disallowed_startup_binary_path(&temp_candidate));

        let safe = PathBuf::from("C:/cortex-test/example/.cortex/bin/cortex.exe");
        assert!(!is_disallowed_startup_binary_path(&safe));
    }

    #[test]
    fn resolve_client_target_inputs_prefers_cli_over_env_values() {
        let (base_url, api_key, local_owner_mode) = resolve_client_target_inputs(
            Some("https://cli.example"),
            Some("ctx_cli"),
            Some("https://env.example"),
            Some("ctx_env"),
            "http://127.0.0.1:7437",
        );
        assert_eq!(base_url, "https://cli.example");
        assert_eq!(api_key.as_deref(), Some("ctx_cli"));
        assert!(!local_owner_mode);
    }

    #[test]
    fn resolve_client_target_inputs_uses_env_and_disables_local_owner_mode() {
        let (base_url, api_key, local_owner_mode) = resolve_client_target_inputs(
            None,
            None,
            Some("https://100.101.102.103:7437"),
            Some("ctx_remote"),
            "http://127.0.0.1:7437",
        );
        assert_eq!(base_url, "https://100.101.102.103:7437");
        assert_eq!(api_key.as_deref(), Some("ctx_remote"));
        assert!(!local_owner_mode);
    }

    #[test]
    fn local_daemon_base_url_uses_loopback_for_wildcard_bind() {
        let home_dir = temp_test_dir("bind_wildcard");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let mut paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        paths.bind = "0.0.0.0".to_string();
        assert_eq!(local_daemon_base_url(&paths), "http://127.0.0.1:7437");

        paths.bind = "100.64.0.12".to_string();
        assert_eq!(local_daemon_base_url(&paths), "http://100.64.0.12:7437");

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn startup_preflight_rejects_non_canonical_health_payload() {
        let _env_guard = env_guard();
        let _bypass = ScopedEnvVar::set(SINGLE_DAEMON_TEST_BYPASS_ENV, "1");
        let home_dir = temp_test_dir("startup_preflight_noncanonical");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let mut paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        paths.bind = "127.0.0.1".to_string();

        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind test listener");
        let port = listener.local_addr().expect("resolve listener addr").port();
        paths.port = port;
        // Isolate this preflight fixture from any live local daemon IPC endpoint.
        paths.ipc_endpoint = None;
        let server =
            spawn_multi_response_server(listener, "404 Not Found", "text/plain", "nope".into(), 2);

        let err = run_preflight(&paths).unwrap_err();
        assert!(
            err.contains("non-canonical payload"),
            "unexpected preflight error: {err}"
        );

        server.join().expect("join response server");
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn startup_preflight_rejects_different_cortex_runtime_identity() {
        let _env_guard = env_guard();
        let _bypass = ScopedEnvVar::set(SINGLE_DAEMON_TEST_BYPASS_ENV, "1");
        let home_dir = temp_test_dir("startup_preflight_wrong_runtime");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let mut paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        paths.bind = "127.0.0.1".to_string();

        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind test listener");
        let port = listener.local_addr().expect("resolve listener addr").port();
        paths.port = port;
        // Isolate this preflight fixture from any live local daemon IPC endpoint.
        paths.ipc_endpoint = None;

        let payload = serde_json::json!({
            "status": "ready",
            "ready": true,
            "runtime": {
                "port": port,
                "token_path": "C:/other/cortex.token",
                "pid_path": "C:/other/cortex.pid",
                "db_path": "C:/other/cortex.db"
            },
            "stats": {
                "home": "C:/other"
            }
        })
        .to_string();
        let server = spawn_single_response_server(listener, "200 OK", "application/json", payload);

        let err = run_preflight(&paths).unwrap_err();
        assert!(
            err.contains("different Cortex runtime identity"),
            "unexpected preflight error: {err}"
        );

        server.join().expect("join response server");
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn startup_preflight_rejects_canonical_ready_readiness_state() {
        let _env_guard = env_guard();
        let _bypass = ScopedEnvVar::set(SINGLE_DAEMON_TEST_BYPASS_ENV, "1");
        let home_dir = temp_test_dir("startup_preflight_ready_state");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let mut paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);
        paths.bind = "127.0.0.1".to_string();

        let listener = TcpListener::bind(("127.0.0.1", 0)).expect("bind test listener");
        let port = listener.local_addr().expect("resolve listener addr").port();
        paths.port = port;
        // Isolate this preflight fixture from any live local daemon IPC endpoint.
        paths.ipc_endpoint = None;
        let payload = serde_json::json!({
            "status": "ready",
            "ready": true,
            "runtime": {
                "port": port,
                "token_path": paths.token.display().to_string(),
                "pid_path": paths.pid.display().to_string(),
                "db_path": paths.db.display().to_string()
            },
            "stats": {
                "home": paths.home.display().to_string()
            }
        })
        .to_string();
        let server = spawn_single_response_server(listener, "200 OK", "application/json", payload);

        let err = run_preflight(&paths).unwrap_err();
        assert!(
            err.contains("canonical Cortex instance is already ready"),
            "unexpected preflight error: {err}"
        );

        server.join().expect("join response server");
        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn is_local_client_base_url_accepts_configured_bind_host() {
        let home_dir = temp_test_dir("local_client_base");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let mut paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        paths.bind = "100.64.0.12".to_string();
        assert!(is_local_client_base_url("http://100.64.0.12:7437", &paths));
        assert!(!is_local_client_base_url("http://100.64.0.12:9999", &paths));
        assert!(!is_local_client_base_url(
            "https://example.com:7437",
            &paths
        ));

        paths.bind = "0.0.0.0".to_string();
        assert!(is_local_client_base_url("http://127.0.0.1:7437", &paths));

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn remote_target_without_api_key_is_rejected() {
        let home_dir = temp_test_dir("remote_target_auth_required");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let err =
            ensure_remote_target_has_api_key("https://100.64.0.12:7437", None, &paths).unwrap_err();
        assert!(err.contains("requires an API key"));

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn local_target_without_api_key_is_allowed() {
        let home_dir = temp_test_dir("local_target_no_key");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        assert!(ensure_remote_target_has_api_key("http://127.0.0.1:7437", None, &paths).is_ok());

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn invalid_client_target_url_is_rejected_cleanly() {
        let home_dir = temp_test_dir("invalid_client_target_url");
        fs::create_dir_all(&home_dir).unwrap();
        let home_str = home_dir.to_string_lossy().to_string();
        let paths =
            auth::CortexPaths::resolve_with_overrides(Some(&home_str), None, Some(7437), None);

        let invalid_scheme =
            ensure_remote_target_has_api_key("ftp://example.com", Some("ctx_key"), &paths)
                .unwrap_err();
        assert!(invalid_scheme.contains("Unsupported Cortex target URL scheme"));

        let embedded_creds = ensure_remote_target_has_api_key(
            "https://user:pass@example.com",
            Some("ctx_key"),
            &paths,
        )
        .unwrap_err();
        assert!(embedded_creds.contains("must not include embedded credentials"));

        let query_url = ensure_remote_target_has_api_key(
            "https://example.com?debug=1",
            Some("ctx_key"),
            &paths,
        )
        .unwrap_err();
        assert!(query_url.contains("must not include query parameters"));

        let _ = fs::remove_dir_all(&home_dir);
    }

    #[test]
    fn openapi_spec_version_matches_cargo_pkg_version() {
        let spec = fs::read_to_string(openapi_spec_path()).expect("read OpenAPI spec");
        assert!(
            spec.contains(&format!("version: {}", env!("CARGO_PKG_VERSION"))),
            "OpenAPI version must match Cargo package version"
        );
    }

    #[test]
    fn openapi_spec_declares_readiness_recall_explain_and_stats_paths() {
        let spec = fs::read_to_string(openapi_spec_path()).expect("read OpenAPI spec");
        assert!(spec.contains("/readiness:"), "missing /readiness in spec");
        assert!(
            spec.contains("/recall/explain:"),
            "missing /recall/explain in spec"
        );
        assert!(spec.contains("/stats:"), "missing /stats in spec");
        assert!(spec.contains("/boot/audit:"), "missing /boot/audit in spec");
    }
}
