pond-db 0.8.0

use std::{
    collections::HashMap,
    fs::{self, File},
    io::{self, IsTerminal},
    path::{Path, PathBuf},
    sync::Arc,
    time::Duration,
};

use chrono::{DateTime, Utc};

use anyhow::{Context, bail};
use clap::{ArgGroup, CommandFactory, Parser, Subcommand, ValueEnum};
use comfy_table::{
    Attribute, Cell, CellAlignment, ColumnConstraint, ContentArrangement, Table, presets::NOTHING,
};
use indicatif::{ProgressBar, ProgressStyle};
use pond::{
    PROTOCOL_VERSION, adapter,
    config::{self, Config, DEFAULT_CONFIG_TOML},
    embed::{BatchProgress, CandleEmbedder, EmbedSummary, EmbedWorker, Embedder, LazyEmbedder},
    handlers::{self, IngestSummary, SessionOutcome, SyncEvent, SyncStatus},
    sessions::{
        AdapterStats, CorpusStats, EmbeddingProgress, LanceArchiveCounts, LanceArchiveExport,
        LanceArchiveImport, MESSAGES_FTS_INDEX, MESSAGES_VECTOR_INDEX, OptimizeOutcome, RowTotals,
        Store,
    },
    substrate::{
        CheckFailure, CredsBinding, IndexStatus, MaintenancePolicy, OptimizeEvent,
        OptimizeProgressFn, PhaseOutcome, ResolvedStorage, StorageUrl, TableSizes,
        default_cleanup_older_than, index_lag_threshold,
    },
    transport::{self, AppState},
    wire::{
        self, ErrorEnvelope, GetEnvelope, GetRequest, GetResponse, GetResult, MessageView,
        PartKind, PartSummary, ProjectFilter, ResponseMode, ResponsePart, SearchEnvelope,
        SearchFilters, SearchModeWire, SearchRequest, SearchResponse, SearchResult, SearchSession,
        SessionFrom,
    },
};

// Bin-only subsystems: the interactive setup wizard and the OS-scheduler
// integration. Neither has a library caller, so they stay out of `pond::`.
mod init;
mod schedule;

#[derive(Debug, serde::Serialize, serde::Deserialize)]
struct PondArchiveManifest {
    archive_version: u32,
    pond_version: String,
    protocol_version: u16,
    created_at: DateTime<Utc>,
    rows: LanceArchiveCounts,
    source_versions: pond::sessions::LanceArchiveVersions,
    embedding_model: String,
    embedding_dim: usize,
}

/// CLI surface for `pond search --mode`. Maps 1:1 to `SearchModeWire`; kept
/// separate so the clap derive lives next to the rest of the CLI types.
#[derive(Debug, Clone, Copy, ValueEnum)]
pub enum CliSearchMode {
    Fts,
    Vector,
    Hybrid,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
enum SyncStage {
    Import,
    Embed,
    UpdateIndexes,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
enum ServeTransport {
    Http,
    Stdio,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
enum ExportFormat {
    Pond,
    Jsonl,
}

impl From<CliSearchMode> for SearchModeWire {
    fn from(mode: CliSearchMode) -> Self {
        match mode {
            CliSearchMode::Fts => SearchModeWire::Fts,
            CliSearchMode::Vector => SearchModeWire::Vector,
            CliSearchMode::Hybrid => SearchModeWire::Hybrid,
        }
    }
}

/// CLI surface for `pond get --response-mode`. Maps 1:1 to wire `ResponseMode`.
#[derive(Debug, Clone, Copy, ValueEnum)]
enum CliResponseMode {
    Conversational,
    Complete,
    Verbatim,
}

impl From<CliResponseMode> for ResponseMode {
    fn from(mode: CliResponseMode) -> Self {
        match mode {
            CliResponseMode::Conversational => ResponseMode::Conversational,
            CliResponseMode::Complete => ResponseMode::Complete,
            CliResponseMode::Verbatim => ResponseMode::Verbatim,
        }
    }
}

/// CLI surface for `pond sql --format`. Maps to `sql::Mode` / `sql::Format`.
/// Mirrors the MCP `pond_sql_query` `format` arg (text|json|parquet|ndjson).
#[derive(Debug, Clone, Copy, PartialEq, Eq, ValueEnum)]
enum CliSqlFormat {
    Text,
    Json,
    Ndjson,
    Parquet,
}

/// CLI surface for `pond get --from`. Maps 1:1 to wire `SessionFrom`.
#[derive(Debug, Clone, Copy, ValueEnum)]
enum CliSessionFrom {
    Start,
    End,
}

impl From<CliSessionFrom> for SessionFrom {
    fn from(value: CliSessionFrom) -> Self {
        match value {
            CliSessionFrom::Start => SessionFrom::Start,
            CliSessionFrom::End => SessionFrom::End,
        }
    }
}
use serde_json::{Value, json};
use tokio::io::AsyncWriteExt;
use tracing_subscriber::{EnvFilter, fmt};
use url::Url;

/// `SkipOracle` backed by a pre-loaded `Store::last_message_timestamps` map.
struct StoredWatermarks {
    map: HashMap<String, DateTime<Utc>>,
}

impl StoredWatermarks {
    fn new(map: HashMap<String, DateTime<Utc>>) -> Self {
        Self { map }
    }
}

impl pond::adapter::SkipOracle for StoredWatermarks {
    fn last_ingested_at(&self, session_id: &str) -> Option<DateTime<Utc>> {
        self.map.get(session_id).copied()
    }
}

/// Adapter clap can call to parse `--storage-path` / `POND_STORAGE_PATH`:
/// bare paths, `~/...`, `file://`, and the remote URL grammar
/// (spec.md#storage-url-grammar) including the fat `s3+https://` form.
fn parse_storage_path(input: &str) -> anyhow::Result<StorageUrl> {
    StorageUrl::parse(input)
}

/// First executable named `name` on `PATH`. Shared by the `pond init` MCP
/// probe and the scheduler's stable-binary resolution (`schedule::pond_bin`).
fn find_on_path(name: &str) -> Option<PathBuf> {
    let paths = std::env::var_os("PATH")?;
    std::env::split_paths(&paths).find_map(|dir| {
        let candidate = dir.join(name);
        is_executable(&candidate).then_some(candidate)
    })
}

#[cfg(unix)]
fn is_executable(path: &Path) -> bool {
    use std::os::unix::fs::PermissionsExt;
    std::fs::metadata(path)
        .map(|meta| meta.is_file() && meta.permissions().mode() & 0o111 != 0)
        .unwrap_or(false)
}

#[cfg(not(unix))]
fn is_executable(path: &Path) -> bool {
    path.is_file()
}

/// Help palette tied to `pond::output`: bold headers/usage, cyan literals,
/// dim placeholders. clap's styling types are anstyle re-exports, so this is
/// the same color stack the rest of the CLI renders with.
const STYLES: clap::builder::styling::Styles = clap::builder::styling::Styles::styled()
    .header(anstyle::Style::new().bold())
    .usage(anstyle::Style::new().bold())
    .literal(anstyle::AnsiColor::Cyan.on_default())
    .placeholder(anstyle::Style::new().dimmed());

/// `--version` detail line: crate version plus the release commit (stamped
/// into release builds via `POND_BUILD_COMMIT`) and the build target. `-V`
/// stays the short crate version.
static LONG_VERSION: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
    let target = format!("{}-{}", std::env::consts::ARCH, std::env::consts::OS);
    match option_env!("POND_BUILD_COMMIT") {
        Some(commit) if !commit.is_empty() => {
            format!("{} ({commit} {target})", env!("CARGO_PKG_VERSION"))
        }
        _ => format!("{} ({target})", env!("CARGO_PKG_VERSION")),
    }
});

/// Lossless storage and hybrid search for sessions from any AI agent client.
#[derive(Debug, Parser)]
#[command(
    name = "pond",
    version,
    long_version = LONG_VERSION.as_str(),
    styles = STYLES,
    max_term_width = 100,
    after_long_help = "\
Getting started:
  pond init                                  set up storage, sources, MCP, and scheduling
  pond sync                                  import new sessions, embed, update indexes
  pond search \"that auth refactor\"           find past work
  claude mcp add -s user pond -- pond mcp    register pond as an MCP server in Claude Code

Every command documents itself: `pond <command> --help` carries examples."
)]
struct Cli {
    #[command(subcommand)]
    command: Command,
    #[command(flatten)]
    #[command(next_help_heading = "Global options")]
    verbose: clap_verbosity_flag::Verbosity<clap_verbosity_flag::WarnLevel>,
}

/// Storage and config selectors shared by every data-touching command.
/// Flattened LAST in each variant: the struct-level `next_help_heading`
/// applies to every arg declared after the flatten point, so trailing it
/// keeps the per-command flags under the default heading.
#[derive(Debug, clap::Args)]
#[command(next_help_heading = "Global options")]
struct StoreArgs {
    /// Storage destination: a local path or remote URL.
    ///
    /// Accepts a bare path, `~/path`, `file://`, `s3://bucket/prefix`,
    /// `s3+https://host/bucket/prefix`, `gs://`, or `az://`. Default:
    /// `[storage].path` from config, then the platform data dir
    /// (`~/.local/share/pond`).
    #[arg(
        long,
        global = true,
        env = "POND_STORAGE_PATH",
        hide_env_values = true,
        value_parser = parse_storage_path,
        value_name = "URL"
    )]
    storage_path: Option<StorageUrl>,
    /// Config file to read (default: `~/.config/pond/config.toml`).
    #[arg(
        long,
        global = true,
        env = "POND_CONFIG",
        hide_env_values = true,
        value_name = "PATH"
    )]
    config: Option<PathBuf>,
}

// Parsed once, matched once, immediately destructured - the size spread
// between variants (StorageUrl-carrying vs unit-like) has no runtime cost.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Subcommand)]
enum Command {
    /// Set up pond (idempotent: safe to re-run).
    ///
    /// Walks through storage, source adapters, MCP registration, and an
    /// optional sync schedule, then writes config.toml in one pass at the
    /// end. Re-running repairs or updates an existing setup; flags answer
    /// sections non-interactively.
    #[command(after_long_help = "Examples:
  pond init                                   interactive setup (or repair)
  pond init --yes                             accept defaults, no prompts
  pond init --storage-path s3://bucket/pond   preset storage, prompt for the rest
  pond init --adapters claude-code,codex-cli --yes
  pond init --schedule 1h --yes               also register an hourly sync")]
    Init(init::InitArgs),
    /// Make pond current: import, embed, update indexes.
    ///
    /// The everyday command: pulls fresh sessions from every enabled
    /// `[sources.*]` entry (or one named adapter), embeds the backlog, and
    /// folds new rows into the search indexes. With no sources configured it
    /// probes the machine and offers to enable what it finds.
    #[command(after_long_help = "Examples:
  pond sync                                  sync every enabled source
  pond sync claude-code                      sync one adapter (re-enables it if declined)
  pond sync codex-cli --source-dir ~/backup  one-off path override, config untouched
  pond sync --only embed                     run a single stage
  pond sync -y                               auto-accept newly detected sources")]
    Sync {
        /// Adapter name (claude-code, codex-cli, ...); default: every enabled source.
        adapter: Option<String>,
        /// One-off source-path override (requires <ADAPTER>).
        ///
        /// Bypasses `[sources.<adapter>]` and does not modify config.toml.
        #[arg(long, value_name = "DIR")]
        source_dir: Option<PathBuf>,
        /// Run exactly one stage: import, embed, or update-indexes.
        #[arg(long, value_enum)]
        only: Option<SyncStage>,
        /// Skip a stage. Can be passed multiple times.
        #[arg(long, value_enum)]
        skip: Vec<SyncStage>,
        /// Re-embed stale rows after an embedding model change.
        #[arg(long)]
        force_embed: bool,
        /// Auto-accept every probe prompt (for non-interactive runs).
        #[arg(long, short = 'y')]
        yes: bool,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Show pond health, data, and source status.
    ///
    /// Storage destination and size, row counts, index readiness, embedding
    /// backlog, configured sources, and the sync schedule. Anything off
    /// names its fix inline.
    #[command(after_long_help = "Examples:
  pond status              the one-screen overview
  pond status --adapters   per-adapter and per-project breakdown")]
    Status {
        /// Show one section per adapter, with project tables and index detail.
        ///
        /// Implies `--include-subagents` so the rollup reconciles with the
        /// storage row counts above.
        #[arg(long)]
        adapters: bool,
        /// Break out sub-agent sessions (e.g. `claude-code/general-purpose`).
        ///
        /// Default rolls sessions up to the main agent only.
        #[arg(long)]
        include_subagents: bool,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Search stored messages.
    ///
    /// Hybrid retrieval (semantic + full-text) when embeddings exist,
    /// full-text otherwise. Keep the query about concepts; scope with
    /// `--project`, `--from-date`, and friends instead of putting names in
    /// the query.
    #[command(after_long_help = "Examples:
  pond search \"lance compaction tuning\"
  pond search \"auth retry\" --project pond --limit 5
  pond search \"migration plan\" --from-date 2026-05-01 --format json")]
    Search {
        /// Free-text query. Semantic concepts work best; project names belong
        /// in `--project`.
        query: String,
        /// Tenant namespace. The personal pond has exactly one; leave at the
        /// default.
        #[arg(long, default_value = "local")]
        namespace: String,
        /// Max sessions to return.
        #[arg(long, default_value_t = 10)]
        limit: usize,
        /// Operator-only retrieval mode override.
        ///
        /// Production callers should omit this and let the server pick
        /// (hybrid when embeddings exist, FTS-only otherwise); benchmark and
        /// ablation harnesses use it to force one arm against the same corpus.
        #[arg(long, value_enum)]
        mode: Option<CliSearchMode>,
        /// Project filter: substring match by default.
        ///
        /// `--project pond` -> contains "pond". Prefix with `re:` for regex
        /// (`--project 're:^/Users/.*/x402'`); `lit:` escapes a literal value
        /// that would otherwise be parsed as a prefix.
        #[arg(long, value_parser = parse_project_filter)]
        project: Option<ProjectFilter>,
        /// Filter to one session (exact match) - search within a single,
        /// possibly long, session.
        #[arg(long, value_name = "ID")]
        session_id: Option<String>,
        /// Filter to one source agent, e.g. `claude-code` (or
        /// `claude-code/general-purpose` for a subagent).
        #[arg(long, value_name = "AGENT")]
        source_agent: Option<String>,
        /// Include subagent sessions (excluded by default).
        #[arg(long)]
        include_subagents: bool,
        /// ISO date (YYYY-MM-DD) lower bound, inclusive.
        #[arg(long)]
        from_date: Option<String>,
        /// ISO date (YYYY-MM-DD) upper bound, inclusive.
        #[arg(long)]
        to_date: Option<String>,
        /// Server-side score threshold; hits below this are dropped.
        ///
        /// Not an absence signal: present and absent content score in
        /// overlapping bands, so leave at 0 unless trimming an over-long tail.
        #[arg(long, default_value_t = 0.0)]
        min_score: f64,
        /// Print Lance query plans instead of search results.
        #[arg(long)]
        explain: bool,
        #[arg(long, value_enum, default_value_t = OutputFormat::Pretty)]
        format: OutputFormat,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Fetch a session or message.
    ///
    /// Returns readable transcripts by id: a whole session (`--session-id`)
    /// or one message with optional surrounding context (`--message-id`).
    /// Pagination cursors (`after-id:`) print at the end of truncated output.
    #[command(after_long_help = "Examples:
  pond get --session-id 58a96901-4a4f-40be-a3c1-62419ec8c580
  pond get --session-id <ID> --session-from end      most recent messages first
  pond get --message-id <ID> --context-depth 3       a hit plus its neighbors
  pond get --session-id <ID> --response-mode verbatim --format json")]
    #[command(group(ArgGroup::new("get_selector")
        .required(true)
        .args(["session_id", "message_id"])))]
    Get {
        /// Fetch an entire session by id. Mutually exclusive with `--message-id`.
        #[arg(long, value_name = "ID")]
        session_id: Option<String>,
        /// Fetch a single message by id. Mutually exclusive with `--session-id`.
        #[arg(long, value_name = "ID")]
        message_id: Option<String>,
        /// Tenant namespace. The personal pond has exactly one; leave at the
        /// default.
        #[arg(long, default_value = "local")]
        namespace: String,
        /// For `--message-id` mode: include this many sibling messages on each
        /// side (grep -C style). Ignored in session mode.
        #[arg(long, default_value_t = 0)]
        context_depth: usize,
        /// Cap on returned messages (session mode) or parts (message mode).
        #[arg(long, default_value_t = 20)]
        limit: usize,
        /// Depth: conversational, complete, or verbatim.
        ///
        /// conversational = text + part summaries; complete = all messages +
        /// summaries; verbatim = full parts inline. With `--message-id` it
        /// selects which siblings fill the context window.
        #[arg(
            long,
            value_enum,
            default_value_t = CliResponseMode::Conversational,
        )]
        response_mode: CliResponseMode,
        /// Session mode only: which end to read from.
        ///
        /// start = oldest first (default); end = most recent, e.g. to recover
        /// context after compaction.
        #[arg(
            long,
            value_enum,
            default_value_t = CliSessionFrom::Start,
            conflicts_with = "message_id"
        )]
        session_from: CliSessionFrom,
        /// Continuation anchor from a prior response: last message id (session)
        /// or last part id (message). Exclusive lower bound.
        #[arg(long, value_name = "ID")]
        after_id: Option<String>,
        #[arg(long, value_enum, default_value_t = OutputFormat::Pretty)]
        format: OutputFormat,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Run one read-only SQL query over the corpus.
    ///
    /// DataFusion / PostgreSQL-compatible SELECT/WITH over the sessions,
    /// messages, and parts tables; writes and side-effecting statements are
    /// rejected. Same surface as the `pond_sql_query` MCP tool - the MCP
    /// resource `schema://pond-sql` documents columns, indexed predicates,
    /// and pagination patterns.
    #[command(after_long_help = "Examples:
  pond sql \"SELECT count(*) FROM sessions\"
  pond sql \"SELECT session_id, ts, role FROM messages WHERE contains_tokens(search_text, 'occ retry') LIMIT 20\"
  pond sql \"SELECT * FROM messages\" --format parquet -o messages.parquet")]
    Sql {
        /// The SQL query. Wrap in quotes; remember to escape `$` in zsh/bash.
        sql: String,
        /// Output format. text/json/ndjson go to stdout; parquet requires
        /// `--output-file` (binary).
        #[arg(long, value_enum, default_value_t = CliSqlFormat::Text)]
        format: CliSqlFormat,
        /// Inline row cap for text/json output. Default 100, max 1000.
        /// Ignored for ndjson/parquet (which return every row).
        #[arg(long, default_value_t = 100)]
        limit: usize,
        /// Write the export bytes here instead of stdout (required for
        /// `--format parquet`; optional for ndjson). Ignored for text/json.
        #[arg(long, short = 'o')]
        output_file: Option<PathBuf>,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Run the HTTP API server (or MCP over stdio with --transport stdio).
    ///
    /// Serves the wire protocol over HTTP on --host:--port. Most agent
    /// setups want `pond mcp` instead; `serve` is for the HTTP transport and
    /// for supervised deployments.
    #[command(after_long_help = "Examples:
  pond serve                       HTTP on 127.0.0.1:9797
  pond serve --port 8080
  pond serve --transport stdio     same as `pond mcp`")]
    Serve {
        /// Wire transport: the HTTP API, or MCP over stdio.
        #[arg(long, value_enum, default_value_t = ServeTransport::Http)]
        transport: ServeTransport,
        /// Bind address for the HTTP transport.
        #[arg(
            long,
            env = "POND_HOST",
            hide_env_values = true,
            default_value = "127.0.0.1"
        )]
        host: String,
        /// Bind port for the HTTP transport.
        #[arg(
            long,
            env = "POND_PORT",
            hide_env_values = true,
            default_value_t = 9797
        )]
        port: u16,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Serve the MCP tools over stdio (for agent clients).
    ///
    /// Equivalent to `pond serve --transport stdio`. Register once per
    /// client; the tools are pond_search, pond_get, and pond_sql_query, with
    /// resources schema://pond, schema://pond-sql, and stats://pond.
    #[command(after_long_help = "Examples:
  claude mcp add -s user pond -- pond mcp    register in Claude Code
  codex mcp add pond -- pond mcp             register in Codex CLI")]
    Mcp {
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Manage the automatic sync schedule.
    ///
    /// Registers `pond sync -q` with the OS scheduler: launchd on macOS,
    /// systemd user timers (or a crontab fence) on Linux. `pond init` offers
    /// the same setup interactively.
    #[command(after_long_help = "Examples:
  pond schedule start              sync every hour
  pond schedule start --every 15m
  pond schedule status
  pond schedule logs
  pond schedule stop")]
    Schedule {
        #[command(subcommand)]
        command: schedule::ScheduleCmd,
    },
    /// Inspect, probe, and switch storage destinations.
    ///
    /// Bare `pond storage` shows the resolved destination, its creds
    /// binding, and per-table sizes. Subcommands probe a destination
    /// (check), switch with guided migration (use), and copy between
    /// destinations (migrate).
    #[command(after_long_help = "Examples:
  pond storage                                     where data lives, and how big
  pond storage check s3+https://host/bucket/pond   probe a destination end-to-end
  pond storage use s3+https://host/bucket/pond     migrate to it and switch config")]
    Storage {
        #[command(subcommand)]
        command: Option<StorageCmd>,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Export a compact restorable .pond archive.
    ///
    /// A .pond file is a zip of the canonical datasets plus a manifest;
    /// `pond import` restores it losslessly. `--format jsonl` streams the
    /// wire representation instead.
    #[command(after_long_help = "Examples:
  pond export                          writes pond-export.pond
  pond export -o backup/$(date +%F).pond
  pond export --format jsonl | jq .    stream the wire form")]
    Export {
        /// Output path. Default: `pond-export.pond` for archive format, stdout for JSONL.
        #[arg(long, short = 'o')]
        out: Option<PathBuf>,
        /// Archive format: a compact .pond file, or the JSONL wire stream.
        #[arg(long, value_enum, default_value_t = ExportFormat::Pond)]
        format: ExportFormat,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Restore a .pond archive.
    ///
    /// Idempotent union merge into the current destination: re-running the
    /// same archive inserts nothing new, and a populated destination unions
    /// rather than clobbers.
    #[command(after_long_help = "Examples:
  pond import backup/2026-06-01.pond")]
    Import {
        /// Path to the .pond archive to restore.
        archive: PathBuf,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Inspect configuration.
    ///
    /// Resolved values with provenance (show), the config file location
    /// (path), and the fully-annotated template (schema).
    #[command(flatten_help = true)]
    #[command(after_long_help = "Examples:
  pond config show                 every setting, its value, and where it came from
  pond config schema > ~/.config/pond/config.toml   start from the annotated template")]
    Config {
        #[command(subcommand)]
        command: ConfigCmd,
    },
    /// Generate shell completions.
    ///
    /// Writes the completion script for the given shell to stdout.
    #[command(after_long_help = "Install:
  bash:  pond completions bash > ~/.local/share/bash-completion/completions/pond
  zsh:   pond completions zsh > \"${fpath[1]}/_pond\"
  fish:  pond completions fish > ~/.config/fish/completions/pond.fish

Homebrew and nix packages ship these pre-installed.")]
    Completions {
        #[arg(value_enum)]
        shell: clap_complete::Shell,
    },
    /// Embed the backlog of un-embedded messages (spec.md#search). Idempotent:
    /// the backlog is every message with a null `vector`, so a re-run picks up
    /// exactly where the last one stopped. A model swap (rows embedded under
    /// a different model id) requires `--force`, which clears those rows and
    /// drops the IVF_PQ before the new vectors land.
    #[command(hide = true)]
    Embed {
        /// Optional cap on messages embedded this run (mostly for benchmarks).
        #[arg(long)]
        limit: Option<usize>,
        /// Allow re-embedding rows whose stored `embedding_model` does not
        /// match the configured model. Without this flag, such rows abort the
        /// run with a typed error so a model swap is never silent.
        #[arg(long)]
        force: bool,
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Inspect and maintain Lance indexes.
    #[command(hide = true)]
    Index {
        #[command(subcommand)]
        command: IndexCommand,
        #[command(flatten)]
        store: StoreArgs,
    },
}

// Parsed once, matched once, immediately destructured - the size spread
// between variants (StorageUrl-carrying vs unit-like) has no runtime cost.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Subcommand)]
enum StorageCmd {
    /// Probe a destination end-to-end.
    ///
    /// Parse, creds resolution, conditional put (the OCC primitive Lance's
    /// commit handler relies on), read-back, delete. Exit codes: 0 ok,
    /// 2 parse error, 3 no creds, 4 auth failed, 5 OCC unsupported.
    #[command(after_long_help = "Examples:
  pond storage check                                  probe the configured destination
  pond storage check s3+https://host/bucket/prefix    probe a candidate before switching")]
    Check {
        /// Destination URL. Defaults to the configured storage path.
        url: Option<String>,
    },
    /// Switch the configured destination, with guided migration.
    ///
    /// Probes the destination first, offers to copy existing data, verifies
    /// row counts reconcile, and only then updates `[storage].path` in
    /// config.toml. The previous destination is never modified.
    #[command(after_long_help = "Examples:
  pond storage use s3+https://host/bucket/pond      probe, offer migration, switch
  pond storage use ~/pond-data --migrate -y         non-interactive: copy, verify, switch
  pond storage use s3://bucket/pond --no-migrate    switch without copying")]
    Use {
        /// The new destination URL.
        url: String,
        /// Copy existing data to the destination before switching.
        #[arg(long, conflicts_with = "no_migrate")]
        migrate: bool,
        /// Switch without copying existing data.
        #[arg(long)]
        no_migrate: bool,
        /// Skip the confirmation prompt (copies data unless --no-migrate).
        #[arg(long, short = 'y')]
        yes: bool,
    },
    /// Copy canonical data between two storages.
    ///
    /// Idempotent union merge: re-runnable, resumable, valid onto a
    /// populated destination. Never deletes the source. `pond storage use`
    /// wraps this with a config switch; reach for migrate directly for
    /// one-off copies.
    #[command(after_long_help = "Examples:
  pond storage migrate --from ~/.local/share/pond --to s3://bucket/pond")]
    Migrate {
        /// Source storage URL.
        #[arg(long, value_parser = parse_storage_path)]
        from: StorageUrl,
        /// Destination storage URL.
        #[arg(long, value_parser = parse_storage_path)]
        to: StorageUrl,
    },
}

// Parsed once, matched once, immediately destructured - the size spread
// between variants (StorageUrl-carrying vs unit-like) has no runtime cost.
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Subcommand)]
enum ConfigCmd {
    /// Show the resolved configuration.
    ///
    /// Redacted values, per-field source (cli/env/file/default), and the
    /// active URL's creds binding.
    Show {
        #[command(flatten)]
        store: StoreArgs,
    },
    /// Print the path of the config file pond loads.
    Path {
        /// Config file to read (default: `~/.config/pond/config.toml`).
        #[arg(long, env = "POND_CONFIG", hide_env_values = true, value_name = "PATH")]
        config: Option<PathBuf>,
    },
    /// Print the fully-annotated config.toml template.
    Schema,
}

#[derive(Debug, Subcommand)]
enum IndexCommand {
    Status,
    Optimize {
        #[arg(long)]
        wait: bool,
        /// Override the manifest-retention window for this run. Accepts
        /// `Ns`/`Nm`/`Nh`/`Nd` (default: the configured `[maintenance]
        /// .cleanup_older_than`, or `1d`). The cleanup pass stays safe
        /// (`delete_unverified=false`), so this is OCC-coordinated and
        /// safe to run while the cron is active; `0s` reclaims every
        /// verified dead version up to the latest committed manifest.
        #[arg(long, value_parser = parse_retention)]
        cleanup_older_than: Option<chrono::Duration>,
    },
    Rebuild {
        intent: Option<String>,
    },
}

fn parse_retention(raw: &str) -> Result<chrono::Duration, String> {
    let trimmed = raw.trim();
    let split = trimmed
        .find(|c: char| !c.is_ascii_digit())
        .unwrap_or(trimmed.len());
    let (number, unit) = trimmed.split_at(split);
    let amount: i64 = number
        .parse()
        .map_err(|_| format!("retention {raw:?}: leading number is not an integer"))?;
    if amount < 0 {
        return Err(format!("retention {raw:?} must be non-negative"));
    }
    match unit {
        "s" => Ok(chrono::Duration::seconds(amount)),
        "m" => Ok(chrono::Duration::minutes(amount)),
        "h" => Ok(chrono::Duration::hours(amount)),
        "d" => Ok(chrono::Duration::days(amount)),
        other => Err(format!(
            "retention {raw:?}: unit {other:?} not recognized (use s/m/h/d)"
        )),
    }
}

/// Parse `--project <value>` into a `ProjectFilter`. `re:<pattern>` selects
/// regex; `lit:<text>` escapes a literal value that would otherwise be
/// parsed as a prefix; everything else is a substring match.
fn parse_project_filter(input: &str) -> Result<ProjectFilter, String> {
    if let Some(pattern) = input.strip_prefix("re:") {
        Ok(ProjectFilter::Regex(pattern.to_owned()))
    } else if let Some(literal) = input.strip_prefix("lit:") {
        Ok(ProjectFilter::Contains(literal.to_owned()))
    } else {
        Ok(ProjectFilter::Contains(input.to_owned()))
    }
}

/// Output mode for `pond search` and `pond get`. Pretty is the human default;
/// Json emits the wire envelope verbatim (including error envelopes), so
/// scripts can `--format json | jq ...` against the same surface as the HTTP
/// transport.
#[derive(Debug, Clone, Copy, ValueEnum)]
enum OutputFormat {
    Pretty,
    Json,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    human_panic::setup_panic!();

    let cli = Cli::parse();
    init_tracing(cli.verbose.tracing_level_filter());

    match cli.command {
        Command::Init(args) => init::run(args).await?,
        Command::Status {
            adapters,
            include_subagents,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            // --adapters needs sub-agents broken out so the per-adapter
            // rollup reconciles with the storage row counts above.
            let include_subagents = include_subagents || adapters;
            let loaded = Config::load(config_path(config))?;
            let (resolved, store) = open_store(storage_path, &loaded, false).await?;
            if !store.initialized().await? {
                render_empty_status("pond status", &resolved)?;
                output(&crate::schedule::status_line())?;
            } else {
                let (sizes, stats, index_status, embedding) = tokio::try_join!(
                    store.table_sizes(),
                    store.corpus_stats(include_subagents),
                    store.index_status(),
                    store.embedding_progress(),
                )?;
                render_status_header("pond status", &resolved, &sizes, &stats.totals)?;
                render_status_checks(&stats, &index_status, embedding, adapters)?;
                let probes = adapter::probe_unconfigured(&loaded.sources);
                if !probes.is_empty() {
                    let names: Vec<&str> = probes.iter().map(|c| c.name.as_str()).collect();
                    output_err(&pond::output::paint(
                        &format!(
                            "hint     {} unconfigured adapter(s): {} - run `pond sync` to enable",
                            probes.len(),
                            names.join(", "),
                        ),
                        pond::output::dim(),
                    ))?;
                }
            }
        }
        Command::Sync {
            adapter,
            source_dir,
            only,
            skip,
            force_embed,
            yes,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let config_file = config_path(config);
            let mut loaded = Config::load(&config_file)?;
            let (_, store) = open_store(storage_path, &loaded, true).await?;
            let stages = SyncStages::resolve(only, &skip)?;
            let mut summary = SyncRunSummary::default();
            // `--source-dir` is an explicit one-off bypass of `[sources.<name>]`
            // (resolve_sync_sources honors it directly), so skip the config-based
            // re-enable - otherwise a probe-less adapter with no config entry
            // (e.g. claude-ai-export) errors before the override is applied.
            if stages.import
                && source_dir.is_none()
                && let Some(name) = adapter.as_deref()
            {
                maybe_reenable_positional(&mut loaded, &config_file, name, yes).await?;
            }
            if stages.import {
                summary.ingest = Some(
                    run_import_stage(&store, &loaded, &config_file, adapter.clone(), source_dir)
                        .await?,
                );
            }
            if stages.import && adapter.is_none() {
                let extra = handle_probe_prompt(&store, &mut loaded, &config_file, yes).await?;
                match summary.ingest.as_mut() {
                    Some(existing) => existing.merge(&extra),
                    None => summary.ingest = Some(extra),
                }
            }
            if stages.embed {
                run_embed_stage(&store, force_embed).await?;
            }
            if stages.update_indexes {
                let policy = configured_maintenance_policy(&loaded, None)?;
                run_update_indexes_stage(&store, &policy).await?;
            }
            render_sync_summary(&store, &summary).await?;
        }
        Command::Embed {
            limit,
            force,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let config = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &config, true).await?;
            let summary = run_embed_stage_with_limit(&store, force, limit, "--force").await?;
            if !summary.cancelled && summary.messages > 0 {
                let policy = configured_maintenance_policy(&config, None)?;
                let outcome = run_update_indexes_stage(&store, &policy).await?;
                if outcome.any_indices_failed() {
                    std::process::exit(1);
                }
            }
        }
        Command::Serve {
            transport,
            host,
            port,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let config = Config::load(config_path(config))?;
            let store = Arc::new(open_store(storage_path, &config, true).await?.1);
            let embedder = Arc::new(LazyEmbedder::candle());
            let state = AppState {
                store,
                embedder,
                search: config.search.clone(),
            };
            match transport {
                ServeTransport::Http => {
                    output(&format!("serve: http listening on http://{host}:{port}"))?;
                    transport::http::serve(state, host, port).await?;
                }
                ServeTransport::Stdio => {
                    eprintln!("serve: stdio MCP ready; stdout is reserved for JSON-RPC");
                    transport::mcp::serve_stdio(state).await?;
                }
            }
        }
        Command::Mcp {
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let config = Config::load(config_path(config))?;
            let store = Arc::new(open_store(storage_path, &config, true).await?.1);
            // Lazy: idle `pond mcp` instances in every Claude Code session
            // stay light. The model load only happens once per process on the
            // first `pond_search` tool call that needs hybrid retrieval.
            let embedder = Arc::new(LazyEmbedder::candle());
            transport::mcp::serve_stdio(AppState {
                store,
                embedder,
                search: config.search.clone(),
            })
            .await?;
        }
        Command::Search {
            query,
            namespace,
            limit,
            mode,
            project,
            session_id,
            source_agent,
            include_subagents,
            from_date,
            to_date,
            min_score,
            explain,
            format,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            let embedder = LazyEmbedder::candle();
            let request = SearchRequest {
                protocol_version: PROTOCOL_VERSION,
                namespace: Some(namespace),
                query,
                mode_override: mode.map(SearchModeWire::from),
                filters: SearchFilters {
                    project,
                    session_id,
                    source_agent,
                    from_date,
                    to_date,
                    min_score,
                    include_subagents,
                },
                limit,
            };
            if explain {
                let plans = explain_search(&store, &embedder, &request, &loaded.search).await?;
                output(&plans)?;
                return Ok(());
            }
            let envelope = handlers::pond_search(&store, &embedder, request, &loaded.search).await;
            if !render_search_envelope(format, &envelope)? {
                std::process::exit(1);
            }
        }
        Command::Index {
            command,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            match command {
                IndexCommand::Status => {
                    let statuses = store.index_status().await?;
                    render_index_status(&statuses)?;
                }
                IndexCommand::Optimize {
                    wait,
                    cleanup_older_than,
                } => {
                    let policy = configured_maintenance_policy(&loaded, cleanup_older_than)?;
                    let (progress, bar) = optimize_progress_bar();
                    let outcome = store.optimize_indices(Some(progress), &policy).await?;
                    bar.finish_and_clear();
                    render_optimize_outcome(&outcome)?;
                    if wait {
                        wait_for_index_catchup(&store).await?;
                    }
                    let statuses = store.index_status().await?;
                    render_index_status(&statuses)?;
                    if outcome.any_indices_failed() {
                        std::process::exit(1);
                    }
                }
                IndexCommand::Rebuild { intent } => {
                    store.rebuild_indices(intent.as_deref()).await?;
                    let statuses = store.index_status().await?;
                    render_index_status(&statuses)?;
                }
            }
        }
        Command::Get {
            session_id,
            message_id,
            namespace,
            context_depth,
            limit,
            response_mode,
            session_from,
            after_id,
            format,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            let request = GetRequest {
                protocol_version: PROTOCOL_VERSION,
                namespace: Some(namespace),
                session_id,
                message_id,
                context_depth,
                limit,
                response_mode: ResponseMode::from(response_mode),
                session_from: SessionFrom::from(session_from),
                after_id,
            };
            let view_from = request.session_from;
            let envelope = handlers::pond_get(&store, request).await;
            if !render_get_envelope(format, &envelope, view_from)? {
                std::process::exit(1);
            }
        }
        Command::Config { command } => run_config_command(command).await?,
        Command::Schedule { command } => schedule::run(command)?,
        Command::Completions { shell } => {
            clap_complete::generate(shell, &mut Cli::command(), "pond", &mut io::stdout());
        }
        Command::Storage {
            command,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => run_storage_command(command, storage_path, config).await?,
        Command::Export {
            out,
            format,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            match format {
                ExportFormat::Pond => {
                    let path = out.unwrap_or_else(|| PathBuf::from("pond-export.pond"));
                    let summary = export_pond_archive(&store, &path).await?;
                    output(&format!(
                        "{} {}  sessions={} messages={} parts={}",
                        pond::output::paint("export:", pond::output::dim()),
                        path.display(),
                        summary.rows.sessions,
                        summary.rows.messages,
                        summary.rows.parts,
                    ))?;
                }
                ExportFormat::Jsonl => {
                    let summary = match out {
                        Some(path) => {
                            let file = tokio::fs::File::create(&path)
                                .await
                                .with_context(|| format!("failed to open {}", path.display()))?;
                            let mut writer = tokio::io::BufWriter::new(file);
                            let summary = handlers::pond_export(&store, None, &mut writer).await?;
                            writer.flush().await.context("export: flush")?;
                            summary
                        }
                        None => {
                            let mut stdout = tokio::io::stdout();
                            handlers::pond_export(&store, None, &mut stdout).await?
                        }
                    };
                    output(&format!(
                        "{} jsonl sessions={} messages={} parts={}",
                        pond::output::paint("export:", pond::output::dim()),
                        summary.sessions,
                        summary.messages,
                        summary.parts,
                    ))?;
                }
            }
        }
        Command::Import {
            archive,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            let summary = import_pond_archive(&store, &archive).await?;
            output(&format!(
                "{} sessions={} messages={} parts={} inserted_sessions={} inserted_messages={} inserted_parts={}",
                pond::output::paint("import:", pond::output::dim()),
                summary.rows.sessions,
                summary.rows.messages,
                summary.rows.parts,
                summary.inserted.sessions,
                summary.inserted.messages,
                summary.inserted.parts,
            ))?;
            output(&format!(
                "{} run `pond sync --only update-indexes` to rebuild search indexes",
                pond::output::paint("hint", pond::output::dim()),
            ))?;
        }
        Command::Sql {
            sql,
            format,
            limit,
            output_file,
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            if matches!(format, CliSqlFormat::Parquet) && output_file.is_none() {
                bail!(
                    "--format parquet requires --output-file <path> (binary, can't go to stdout)"
                );
            }
            let loaded = Config::load(config_path(config))?;
            let (_, store) = open_store(storage_path, &loaded, false).await?;
            let mode = match format {
                CliSqlFormat::Text => pond::sql::Mode::Inline,
                CliSqlFormat::Json => pond::sql::Mode::InlineJson,
                CliSqlFormat::Ndjson => pond::sql::Mode::Export(pond::sql::Format::Ndjson),
                CliSqlFormat::Parquet => pond::sql::Mode::Export(pond::sql::Format::Parquet),
            };
            let inline_rows = limit.min(pond::sql::MAX_INLINE_ROWS);
            let (sessions, messages, parts) = tokio::try_join!(
                store.dataset(pond::substrate::Table::Sessions),
                store.dataset(pond::substrate::Table::Messages),
                store.dataset(pond::substrate::Table::Parts),
            )?;
            let tables = pond::sql::Tables {
                sessions,
                messages,
                parts,
            };
            match pond::sql::run(&tables, &sql, mode, inline_rows).await {
                Ok(pond::sql::Outcome::Inline(text)) => {
                    output(&text)?;
                }
                Ok(pond::sql::Outcome::InlineJson(value)) => {
                    // Pretty-print for the CLI - the structured payload is for
                    // agents over MCP; humans reading stdout want it readable.
                    output(&serde_json::to_string_pretty(&value)?)?;
                }
                Ok(pond::sql::Outcome::Export {
                    bytes,
                    format: _,
                    rows,
                    columns: _,
                }) => match output_file {
                    Some(path) => {
                        fs::write(&path, &bytes)
                            .with_context(|| format!("write export to {}", path.display()))?;
                        output_err(&format!(
                            "{} {} row(s), {} bytes -> {}",
                            pond::output::paint("export:", pond::output::dim()),
                            rows,
                            bytes.len(),
                            path.display()
                        ))?;
                    }
                    None => {
                        use std::io::Write;
                        io::stdout().write_all(&bytes)?;
                    }
                },
                Err(pond::sql::SqlError::Query(message)) => {
                    output_err(&format!(
                        "{} {message}",
                        pond::output::paint("sql error:", pond::output::dim())
                    ))?;
                    std::process::exit(2);
                }
                Err(pond::sql::SqlError::Infra(error)) => {
                    return Err(error);
                }
            }
        }
    }

    Ok(())
}

fn init_tracing(cli_level: tracing::level_filters::LevelFilter) {
    // Lance's IVF_PQ builder warns once per empty centroid during merge
    // (rust/lance/src/index/vector/builder.rs: "partition N is empty, skipping").
    // It already handles the case - records a zero-sized partition and continues -
    // so the warning is benign log noise during index maintenance.
    //
    // `aws_config`'s IMDS region probe WARNs when there's no instance-metadata
    // endpoint (every non-EC2 host): a 1s connect timeout that doesn't affect
    // the explicit-creds path. Lance's AIMD throttle WARNs once per retry while
    // a probe target is unreachable - the failure itself is already surfaced in
    // the check/probe error. Silencing both keeps the storage probe's spinner
    // (and the init wizard) from being corrupted mid-render; `RUST_LOG`
    // (which replaces this whole filter) still opts back in.
    let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| {
        EnvFilter::new(format!(
            "{cli_level},lance::index::vector::builder=error,aws_config=error,lance::object_store::throttle=error"
        ))
    });
    fmt().with_env_filter(filter).with_writer(io::stderr).init();
}

#[allow(clippy::print_stdout)]
fn output(message: &str) -> anyhow::Result<()> {
    pond::output::line(message)
}

fn output_err(message: &str) -> anyhow::Result<()> {
    pond::output::line_err(message)
}

/// Partition `outcomes` into accepts/declines, persist both, and refresh
/// `loaded` from disk. Shared between the post-import probe sweep and the
/// positional re-enable path.
fn apply_outcomes(
    loaded: &mut Config,
    config_file: &Path,
    outcomes: &[adapter::PromptOutcome],
) -> anyhow::Result<usize> {
    let accepts: Vec<adapter::Candidate> = outcomes
        .iter()
        .filter(|o| o.enable)
        .map(|o| o.candidate.clone())
        .collect();
    let declines: Vec<&str> = outcomes
        .iter()
        .filter(|o| !o.enable)
        .map(|o| o.candidate.name.as_str())
        .collect();
    if !accepts.is_empty() {
        adapter::persist_accept(config_file, &accepts)?;
    }
    if !declines.is_empty() {
        adapter::persist_decline(config_file, &declines)?;
    }
    if !accepts.is_empty() || !declines.is_empty() {
        *loaded = Config::load(config_file)?;
    }
    Ok(accepts.len())
}

/// `pond sync <name>` positional override. When `<name>` is currently
/// absent or `enabled = false`, re-probe just that one adapter and prompt
/// (or auto-accept on `--yes`). Used to re-enable a previously-declined
/// adapter without editing `config.toml` by hand.
async fn maybe_reenable_positional(
    loaded: &mut Config,
    config_file: &Path,
    name: &str,
    auto_accept: bool,
) -> anyhow::Result<()> {
    use serde_json::Value;
    use std::io::IsTerminal;
    let present = loaded.sources.get(name);
    let is_enabled = present
        .and_then(|b| b.get("enabled").and_then(Value::as_bool))
        .unwrap_or(false);
    if is_enabled {
        return Ok(());
    }
    let candidates = adapter::discover(Some(name));
    if candidates.is_empty() {
        bail!("no `[sources.{name}]` and probe returned nothing; add the entry manually");
    }
    if !auto_accept && !std::io::stdin().is_terminal() {
        bail!(
            "source [{name}] is disabled and stdin is not a terminal; pass --yes or re-run on a TTY"
        );
    }
    let outcomes = adapter::prompt_each(&candidates, auto_accept)?;
    let accepted = apply_outcomes(loaded, config_file, &outcomes)?;
    if accepted == 0 {
        bail!("declined; nothing to sync");
    }
    Ok(())
}

/// After `pond sync`'s import stage, prompt the operator about any
/// freshly-detectable adapter that has no `[sources.<name>]` section yet.
/// `enabled = true`/`enabled = false` is persisted either way so the
/// decision sticks; only the positional `pond sync <name>` re-prompts a
/// previously-declined adapter. Non-TTY runs (and `--yes` runs without a
/// TTY) emit a one-line stderr hint and continue without writing - the
/// operator can re-run on a TTY to opt in/out.
async fn handle_probe_prompt(
    store: &Store,
    loaded: &mut Config,
    config_file: &Path,
    auto_accept: bool,
) -> anyhow::Result<IngestSummary> {
    use std::io::IsTerminal;
    let mut accumulated = IngestSummary::default();
    let candidates = adapter::probe_unconfigured(&loaded.sources);
    if candidates.is_empty() {
        return Ok(accumulated);
    }
    let interactive = std::io::stdin().is_terminal();
    if !interactive && !auto_accept {
        let names: Vec<&str> = candidates.iter().map(|c| c.name.as_str()).collect();
        output_err(&pond::output::paint(
            &format!(
                "hint     {} unconfigured adapter(s): {} - run `pond sync` on a TTY to enable",
                candidates.len(),
                names.join(", "),
            ),
            pond::output::dim(),
        ))?;
        return Ok(accumulated);
    }
    let outcomes = adapter::prompt_each(&candidates, auto_accept)?;
    apply_outcomes(loaded, config_file, &outcomes)?;
    for outcome in &outcomes {
        if outcome.enable && outcome.sync_now {
            let summary = run_import_stage(
                store,
                loaded,
                config_file,
                Some(outcome.candidate.name.clone()),
                None,
            )
            .await?;
            accumulated.merge(&summary);
        }
    }
    Ok(accumulated)
}

/// Open the store with an indicatif spinner ticking while
/// [`Store::open_with_options`] runs. Open itself is cheap; the spinner only
/// matters for visual consistency with other long-running commands.
async fn open_store_with_spinner(
    location: &Url,
    storage: HashMap<String, String>,
    caps: pond::substrate::RuntimeCaps,
) -> anyhow::Result<Store> {
    let spinner = ProgressBar::new_spinner();
    spinner.set_style(
        ProgressStyle::with_template("{spinner:.green} opening pond store... [{elapsed_precise}]")
            .unwrap_or_else(|_| ProgressStyle::default_spinner()),
    );
    spinner.enable_steady_tick(Duration::from_millis(120));
    let result = Store::open_with_options(location, storage, caps).await;
    spinner.finish_and_clear();
    result
}

/// Resolve `[runtime]` into the substrate's typed caps struct. A standalone
/// helper to keep every call site terse and to make sure no surface forgets
/// the caps.
fn runtime_caps(config: &Config) -> pond::substrate::RuntimeCaps {
    pond::substrate::RuntimeCaps::from_config(&config.runtime)
}

/// Resolve the storage destination: `--storage-path` / `POND_STORAGE_PATH`
/// (folded together by clap) > `[storage].path` > the platform-local default.
fn resolve_storage_location(
    explicit: Option<StorageUrl>,
    loaded: &Config,
) -> anyhow::Result<StorageUrl> {
    if let Some(storage) = explicit {
        return Ok(storage);
    }
    if let Some(path) = &loaded.storage.path {
        return StorageUrl::parse(path).context("invalid [storage].path in config");
    }
    let url = pond::config::default_storage_path(
        std::env::var_os("XDG_DATA_HOME").map(PathBuf::from),
        std::env::var_os("HOME").map(PathBuf::from),
    )?;
    StorageUrl::parse(url.as_str())
}

/// Resolve creds for the destination and open the store. One chokepoint so
/// every command resolves identically and no surface forgets the
/// unmatched-set warning (misbinding must never be silent).
async fn open_store(
    explicit: Option<StorageUrl>,
    loaded: &Config,
    spinner: bool,
) -> anyhow::Result<(ResolvedStorage, Store)> {
    let storage = resolve_storage_location(explicit, loaded)?;
    let resolved = storage.resolve(&loaded.creds)?;
    warn_unmatched_sets(&[&resolved], loaded)?;
    let store = if spinner {
        open_store_with_spinner(
            resolved.lance_url(),
            resolved.options.clone(),
            runtime_caps(loaded),
        )
        .await?
    } else {
        Store::open_with_options(
            resolved.lance_url(),
            resolved.options.clone(),
            runtime_caps(loaded),
        )
        .await?
    };
    Ok((resolved, store))
}

/// spec.md#creds-scope-match: a defined set that bound to none of this
/// invocation's URLs gets named on stderr - a wrong scope must surface
/// before the auth error it causes.
fn warn_unmatched_sets(resolved: &[&ResolvedStorage], loaded: &Config) -> anyhow::Result<()> {
    for name in pond::substrate::unmatched_creds_sets(resolved, &loaded.creds) {
        output_err(&pond::output::paint(
            &format!("hint     creds set [{name}] matched no storage URL in this invocation"),
            pond::output::dim(),
        ))?;
    }
    Ok(())
}

/// The config path: an explicit `--config` (or `POND_CONFIG`) wins; otherwise
/// `$XDG_CONFIG_HOME/pond/config.toml` (default `~/.config/pond/config.toml`),
/// regardless of where the storage path points. Config and data are different
/// XDG categories - they always live in different directories, even when both
/// are local.
fn config_path(explicit: Option<PathBuf>) -> PathBuf {
    if let Some(path) = explicit {
        return path;
    }
    pond::config::default_config_path(
        std::env::var_os("XDG_CONFIG_HOME").map(PathBuf::from),
        std::env::var_os("HOME").map(PathBuf::from),
    )
}

async fn run_config_command(command: ConfigCmd) -> anyhow::Result<()> {
    use pond::output::{dim, paint};
    match command {
        ConfigCmd::Schema => output(DEFAULT_CONFIG_TOML.trim_end()),
        ConfigCmd::Path { config } => output(&config_path(config).display().to_string()),
        ConfigCmd::Show {
            store: StoreArgs {
                storage_path,
                config,
            },
        } => {
            let path = config_path(config);
            let (loaded, figment) = Config::load_with_provenance(&path)?;
            // Contracted like every other human-facing path (`pond config
            // path` stays absolute for scripting).
            output(&format!(
                "{}  {}{}",
                paint("config ", dim()),
                pond::config::contract_home(&path).display(),
                if path.exists() {
                    ""
                } else {
                    "  (absent - defaults + env)"
                },
            ))?;

            // The active destination, its provenance, and its creds binding -
            // the line that makes a wrong scope match visible.
            let storage = resolve_storage_location(storage_path.clone(), &loaded)?;
            let resolved = storage.resolve(&loaded.creds)?;
            let storage_source = match &storage_path {
                // clap folds the env var into the flag; argv tells them apart
                // (value comparison can't - flag and env may carry the same
                // URL, and the flag wins per the precedence ladder).
                Some(_) => {
                    if std::env::args()
                        .any(|arg| arg == "--storage-path" || arg.starts_with("--storage-path="))
                    {
                        "cli"
                    } else {
                        "env"
                    }
                }
                None if loaded.storage.path.is_some() => classify_source(&figment, "storage.path"),
                None => "default",
            };
            output(&format!(
                "{}  {}  ({})  -> {}",
                paint("storage", dim()),
                resolved.display(),
                storage_source,
                describe_binding_with_source(&resolved.binding, &figment),
            ))?;
            // The one precedence ladder (spec.md#storage-env-mirror), shown
            // wherever sources are attributed.
            output(&paint(
                "ladder   cli flag > POND_* env > config file > ambient cloud chain > built-in defaults",
                dim(),
            ))?;
            output("")?;

            let mut table = new_table();
            table.set_header(
                ["setting", "value", "source"]
                    .map(|h| Cell::new(h).add_attributes(vec![Attribute::Dim, Attribute::Bold])),
            );
            let value = serde_json::to_value(&loaded).context("serialize config for display")?;
            let mut rows = Vec::new();
            flatten_config(String::new(), &value, &mut rows);
            for (key, raw) in rows {
                let source = classify_source(&figment, &key);
                table.add_row([
                    key.clone(),
                    redact_config_value(&key, &raw),
                    source.to_owned(),
                ]);
            }
            output(&table.to_string())?;
            warn_unmatched_sets(&[&resolved], &loaded)?;
            Ok(())
        }
    }
}

/// Map a [`CheckFailure`] onto the documented `pond storage check` exit
/// codes (3 no creds, 4 auth failed, 5 OCC unsupported, 1 other IO).
fn check_failure_exit_code(failure: &CheckFailure) -> i32 {
    match failure {
        CheckFailure::NoCreds { .. } => 3,
        CheckFailure::Auth { .. } => 4,
        CheckFailure::OccUnsupported { .. } => 5,
        CheckFailure::Io { .. } => 1,
    }
}

/// One dim `cause:` line under a probe failure's fix-naming lead. The full
/// untruncated chain stays reachable via `RUST_LOG=debug`.
fn render_check_cause(failure: &CheckFailure) -> anyhow::Result<()> {
    if let Some(cause) = failure.concise_cause() {
        output_err(&pond::output::paint(
            &format!("cause: {cause}"),
            pond::output::dim(),
        ))?;
    }
    tracing::debug!("full probe failure: {failure:?}");
    Ok(())
}

async fn run_storage_command(
    command: Option<StorageCmd>,
    storage_path: Option<StorageUrl>,
    config: Option<PathBuf>,
) -> anyhow::Result<()> {
    use pond::output::{dim, paint};
    let config_file = config_path(config);
    let loaded = Config::load(&config_file)?;
    match command {
        // Bare `pond storage`: where the data lives, under which creds, and
        // how big - the storage slice of `pond status`.
        None => {
            let (resolved, store) = open_store(storage_path, &loaded, false).await?;
            if !store.initialized().await? {
                render_empty_status("pond storage", &resolved)?;
            } else {
                let (sizes, stats) =
                    tokio::try_join!(store.table_sizes(), store.corpus_stats(false))?;
                render_status_header("pond storage", &resolved, &sizes, &stats.totals)?;
            }
            warn_unmatched_sets(&[&resolved], &loaded)?;
        }
        Some(StorageCmd::Check { url }) => {
            let storage = match url {
                Some(raw) => match StorageUrl::parse(&raw) {
                    Ok(parsed) => parsed,
                    Err(error) => {
                        output_err(&format!("check: parse error: {error:#}"))?;
                        std::process::exit(2);
                    }
                },
                None => resolve_storage_location(storage_path, &loaded)?,
            };
            let resolved = storage.resolve(&loaded.creds)?;
            output(&format!(
                "{}  {}",
                resolved.display(),
                paint(&format!("[{}]", resolved.binding.describe()), dim()),
            ))?;
            warn_unmatched_sets(&[&resolved], &loaded)?;
            match pond::substrate::storage_check(&resolved).await {
                Ok(()) => {
                    output(
                        "check: ok - conditional put (OCC), read-back, and delete all succeeded",
                    )?;
                }
                Err(failure) => {
                    // Distinct exit codes (documented in --help) so cron and
                    // CI can branch on the failure class.
                    output_err(&format!("check: {failure}"))?;
                    render_check_cause(&failure)?;
                    std::process::exit(check_failure_exit_code(&failure));
                }
            }
        }
        Some(StorageCmd::Use {
            url,
            migrate,
            no_migrate,
            yes,
        }) => {
            run_storage_use(
                &loaded,
                &config_file,
                storage_path,
                &url,
                migrate,
                no_migrate,
                yes,
            )
            .await?;
        }
        Some(StorageCmd::Migrate { from, to }) => {
            let from_resolved = from.resolve(&loaded.creds)?;
            let to_resolved = to.resolve(&loaded.creds)?;
            warn_unmatched_sets(&[&from_resolved, &to_resolved], &loaded)?;
            // Per-URL binding lines before any work: a wrong scope match must
            // be visible immediately, not after an auth error.
            let dim = pond::output::dim();
            output(&format!(
                "{} {}  {}",
                pond::output::paint("from:", dim),
                from_resolved.display(),
                pond::output::paint(&format!("[{}]", from_resolved.binding.describe()), dim),
            ))?;
            output(&format!(
                "{}   {}  {}",
                pond::output::paint("to:", dim),
                to_resolved.display(),
                pond::output::paint(&format!("[{}]", to_resolved.binding.describe()), dim),
            ))?;
            let from_store = Store::open_with_options(
                from_resolved.lance_url(),
                from_resolved.options.clone(),
                runtime_caps(&loaded),
            )
            .await?;
            let to_store = Store::open_with_options(
                to_resolved.lance_url(),
                to_resolved.options.clone(),
                runtime_caps(&loaded),
            )
            .await?;
            let imported = migrate_between_stores(&from_store, &to_store).await?;
            output(&format!(
                "{} sessions={} messages={} parts={} inserted_sessions={} inserted_messages={} inserted_parts={}",
                pond::output::paint("migrate:", dim),
                imported.rows.sessions,
                imported.rows.messages,
                imported.rows.parts,
                imported.inserted.sessions,
                imported.inserted.messages,
                imported.inserted.parts,
            ))?;
            output(&format!(
                "{} the source was not modified; run `pond sync --only update-indexes --storage-path {}` to build destination indexes",
                pond::output::paint("hint", dim),
                to_resolved.display(),
            ))?;
        }
    }
    Ok(())
}

/// The string `[storage].path` should carry for `url`: the display form
/// (contracted local path / verbatim remote URL) minus the trailing slash
/// Lance's `uri_to_url` appends to directory paths.
fn storage_config_value(url: &StorageUrl) -> String {
    let display = url.display();
    if url.is_local() && display.len() > 1 && display.ends_with('/') {
        display.trim_end_matches('/').to_owned()
    } else {
        display
    }
}

/// Set `[storage].path` in `doc` as a proper `[storage]` section (index
/// assignment on an absent key would synthesize the inline
/// `storage = { path = ... }` form instead).
fn set_storage_path(doc: &mut toml_edit::DocumentMut, path_value: &str) {
    use toml_edit::{Item, Table, value};
    match doc.get_mut("storage").and_then(Item::as_table_like_mut) {
        Some(storage) => {
            storage.insert("path", value(path_value));
        }
        None => {
            let mut storage = Table::new();
            storage.insert("path", value(path_value));
            doc.insert("storage", Item::Table(storage));
        }
    }
}

/// `pond storage use`: validate-then-activate. The destination is probed
/// end-to-end and (optionally) populated + verified BEFORE `[storage].path`
/// flips, so a typo or auth failure can never strand the config pointing at
/// a broken destination.
async fn run_storage_use(
    loaded: &Config,
    config_file: &Path,
    storage_path: Option<StorageUrl>,
    url: &str,
    migrate: bool,
    no_migrate: bool,
    yes: bool,
) -> anyhow::Result<()> {
    use pond::output::{dim, paint};

    let dest = match StorageUrl::parse(url) {
        Ok(parsed) => parsed,
        Err(error) => {
            output_err(&format!("use: parse error: {error:#}"))?;
            std::process::exit(2);
        }
    };
    let dest_resolved = dest.resolve(&loaded.creds)?;
    output(&format!(
        "{}  {}",
        dest_resolved.display(),
        paint(&format!("[{}]", dest_resolved.binding.describe()), dim()),
    ))?;
    warn_unmatched_sets(&[&dest_resolved], loaded)?;

    // 1. End-to-end probe first - same classes and exit codes as `check`.
    let spinner = ProgressBar::new_spinner();
    spinner.set_style(
        ProgressStyle::with_template("{spinner:.green} probing destination... [{elapsed_precise}]")
            .unwrap_or_else(|_| ProgressStyle::default_spinner()),
    );
    spinner.enable_steady_tick(Duration::from_millis(120));
    let check = pond::substrate::storage_check(&dest_resolved).await;
    spinner.finish_and_clear();
    if let Err(failure) = check {
        output_err(&format!(
            "use: destination failed the end-to-end check; config not changed: {failure}"
        ))?;
        render_check_cause(&failure)?;
        std::process::exit(check_failure_exit_code(&failure));
    }
    output("check: ok - conditional put (OCC), read-back, and delete all succeeded")?;

    // 2. Offer to copy existing data before the switch.
    let current = resolve_storage_location(storage_path, loaded)?;
    let already_configured = loaded.storage.path.as_deref() == Some(dest.canonical().as_str())
        || loaded.storage.path.as_deref().is_some_and(|path| {
            StorageUrl::parse(path)
                .map(|parsed| parsed.canonical() == dest.canonical())
                .unwrap_or(false)
        });
    if current.canonical() == dest.canonical() && already_configured {
        output(&format!(
            "{} {} is already the configured destination - nothing to change",
            paint("use:", dim()),
            dest.display(),
        ))?;
        return Ok(());
    }
    let current_resolved = current.resolve(&loaded.creds)?;
    let from_store = Store::open_with_options(
        current_resolved.lance_url(),
        current_resolved.options.clone(),
        runtime_caps(loaded),
    )
    .await?;
    let from_totals = from_store.corpus_stats(false).await?.totals;
    if from_totals.sessions > 0 && !no_migrate {
        let copy = if migrate || yes {
            true
        } else if io::stdin().is_terminal() {
            cliclack::confirm(format!(
                "Copy existing data ({} sessions) from {} first?",
                format_thousands(from_totals.sessions),
                current.display(),
            ))
            .initial_value(true)
            .interact()
            .context("migration prompt failed")?
        } else {
            bail!(
                "current storage has data and stdin is not a terminal; pass --migrate to copy it or --no-migrate to switch without copying"
            );
        };
        if copy {
            let to_store = Store::open_with_options(
                dest_resolved.lance_url(),
                dest_resolved.options.clone(),
                runtime_caps(loaded),
            )
            .await?;
            let imported = migrate_between_stores(&from_store, &to_store).await?;
            output(&format!(
                "{} sessions={} messages={} parts={} inserted_sessions={} inserted_messages={} inserted_parts={}",
                paint("migrate:", dim()),
                imported.rows.sessions,
                imported.rows.messages,
                imported.rows.parts,
                imported.inserted.sessions,
                imported.inserted.messages,
                imported.inserted.parts,
            ))?;
            let policy = configured_maintenance_policy(loaded, None)?;
            run_update_indexes_stage(&to_store, &policy).await?;
            // 3. Verify the copy reconciles BEFORE flipping config.
            let dest_totals = to_store.corpus_stats(false).await?.totals;
            if dest_totals.sessions < from_totals.sessions
                || dest_totals.messages < from_totals.messages
                || dest_totals.parts < from_totals.parts
            {
                bail!(
                    "destination row counts do not reconcile (source {}/{}/{} vs destination {}/{}/{} sessions/messages/parts); config not changed - rerun `pond storage use` to retry the copy",
                    from_totals.sessions,
                    from_totals.messages,
                    from_totals.parts,
                    dest_totals.sessions,
                    dest_totals.messages,
                    dest_totals.parts,
                );
            }
        }
    }

    // 4. Flip `[storage].path`, preserving the rest of the file verbatim.
    let existing = if config_file.exists() {
        fs::read_to_string(config_file)
            .with_context(|| format!("failed to read {}", config_file.display()))?
    } else {
        String::new()
    };
    let mut doc: toml_edit::DocumentMut = existing
        .parse()
        .with_context(|| format!("failed to parse {} as TOML", config_file.display()))?;
    let path_value = storage_config_value(&dest);
    set_storage_path(&mut doc, &path_value);
    if let Some(parent) = config_file.parent() {
        fs::create_dir_all(parent)
            .with_context(|| format!("failed to create {}", parent.display()))?;
    }
    fs::write(config_file, doc.to_string())
        .with_context(|| format!("failed to write {}", config_file.display()))?;
    output(&format!(
        "{} [storage].path = {} written to {}",
        paint("use:", dim()),
        path_value,
        config::display(&config::url_for_path(config_file)?),
    ))?;
    output(&format!(
        "{} previous data at {} is untouched; verify with `pond storage check` and `pond status`",
        paint("hint", dim()),
        current.display(),
    ))?;
    Ok(())
}

/// Map a figment provenance lookup onto the source column vocabulary.
fn classify_source(figment: &figment::Figment, key: &str) -> &'static str {
    match figment.find_metadata(key) {
        Some(md) if md.name.contains("POND_") => "env",
        Some(_) => "file",
        None => "default",
    }
}

/// Binding line for `pond config show`: the set name, how it matched, and
/// which layer defined it ("creds work (scope match, file)").
fn describe_binding_with_source(binding: &CredsBinding, figment: &figment::Figment) -> String {
    match binding {
        CredsBinding::Set { name, .. } => {
            let source = classify_source(figment, &format!("creds.{name}"));
            let base = binding.describe();
            format!("{}, {source})", base.trim_end_matches(')'))
        }
        other => other.describe(),
    }
}

/// Flatten the serialized config into dotted (key, value) leaf rows,
/// skipping `null`s (unset optionals) and empty containers.
fn flatten_config(prefix: String, value: &Value, rows: &mut Vec<(String, String)>) {
    match value {
        Value::Null => {}
        Value::Object(map) => {
            for (key, child) in map {
                let path = if prefix.is_empty() {
                    key.clone()
                } else {
                    format!("{prefix}.{key}")
                };
                flatten_config(path, child, rows);
            }
        }
        Value::String(text) => rows.push((prefix, text.clone())),
        other => rows.push((prefix, other.to_string())),
    }
}

/// spec.md#storage-redaction: any field whose name contains key / secret /
/// token / password prints `********` regardless of length - including keys
/// inside `extra`. The `_file` / `_command` variants print their literal
/// value: the path or command IS the safe part.
fn redact_config_value(key: &str, value: &str) -> String {
    let field = key.rsplit('.').next().unwrap_or(key).to_ascii_lowercase();
    if field.ends_with("_file") || field.ends_with("_command") {
        return value.to_owned();
    }
    let sensitive = ["key", "secret", "token", "password"];
    if sensitive.iter().any(|needle| field.contains(needle)) {
        return "********".to_owned();
    }
    value.to_owned()
}

/// `pond storage migrate` core: export the source's clean datasets into local
/// staging, then merge-import into the destination. Idempotency comes from
/// `lance-deterministic-pk` + merge-insert: a rerun inserts nothing, and a
/// populated destination unions rather than clobbers.
async fn migrate_between_stores(from: &Store, to: &Store) -> anyhow::Result<LanceArchiveImport> {
    let staging = tempfile::Builder::new()
        .prefix("pond-migrate-")
        .tempdir()
        .context("failed to create migrate staging dir")?;
    let data_dir = staging.path().join("data");
    let spinner = ProgressBar::new_spinner();
    spinner.set_style(
        ProgressStyle::with_template("{spinner:.green} {msg} [{elapsed_precise}]")
            .unwrap_or_else(|_| ProgressStyle::default_spinner()),
    );
    spinner.enable_steady_tick(Duration::from_millis(120));
    spinner.set_message("migrate: copying from source...");
    let _exported = from.export_clean_lance_datasets(&data_dir).await?;
    spinner.set_message("migrate: merging into destination...");
    let imported = to.import_clean_lance_datasets(&data_dir).await;
    spinner.finish_and_clear();
    imported
}

#[derive(Debug, Default)]
struct SyncStages {
    import: bool,
    embed: bool,
    update_indexes: bool,
}

impl SyncStages {
    fn resolve(only: Option<SyncStage>, skip: &[SyncStage]) -> anyhow::Result<Self> {
        let mut stages = match only {
            Some(SyncStage::Import) => Self {
                import: true,
                ..Self::default()
            },
            Some(SyncStage::Embed) => Self {
                embed: true,
                ..Self::default()
            },
            Some(SyncStage::UpdateIndexes) => Self {
                update_indexes: true,
                ..Self::default()
            },
            None => Self {
                import: true,
                embed: true,
                update_indexes: true,
            },
        };
        for stage in skip {
            match stage {
                SyncStage::Import => stages.import = false,
                SyncStage::Embed => stages.embed = false,
                SyncStage::UpdateIndexes => stages.update_indexes = false,
            }
        }
        if !(stages.import || stages.embed || stages.update_indexes) {
            bail!("no sync stages selected");
        }
        Ok(stages)
    }
}

/// Only the import recap survives to `render_sync_summary`; embed and index
/// each print their own line as they finish, so their outcomes aren't threaded
/// back here.
#[derive(Debug, Default)]
struct SyncRunSummary {
    ingest: Option<IngestSummary>,
}

async fn run_import_stage(
    store: &Store,
    loaded: &Config,
    config_file: &Path,
    adapter: Option<String>,
    source_dir: Option<PathBuf>,
) -> anyhow::Result<IngestSummary> {
    let sources = resolve_sync_sources(loaded, config_file, adapter.as_deref(), source_dir)?;
    if sources.is_empty() {
        let disabled = loaded.disabled_source_names();
        let label = pond::output::paint("import:", pond::output::dim());
        if disabled.is_empty() {
            output(&format!(
                "{label} no sources configured. Run `pond sync` on a TTY to detect adapters, or add `[sources.<name>]` blocks to {}.",
                config_file.display(),
            ))?;
        } else {
            output(&format!(
                "{label} no enabled sources. Found {} disabled: {}. Add `enabled = true` to the section in {}, or re-enable interactively with `pond sync <name>`.",
                disabled.len(),
                disabled.join(", "),
                config_file.display(),
            ))?;
        }
        return Ok(IngestSummary::default());
    }
    let watermarks = StoredWatermarks::new(store.session_last_ingested_at().await?);
    let mut total = IngestSummary::default();
    for (name, blob) in sources {
        let summary = sync_with_progress(store, &name, blob, &watermarks).await?;
        total.merge(&summary);
    }
    Ok(total)
}

async fn run_embed_stage(store: &Store, force: bool) -> anyhow::Result<EmbedSummary> {
    run_embed_stage_with_limit(store, force, None, "--force-embed").await
}

async fn run_embed_stage_with_limit(
    store: &Store,
    force: bool,
    limit: Option<usize>,
    force_hint: &'static str,
) -> anyhow::Result<EmbedSummary> {
    let stale = store.stale_embedding_count().await?;
    if stale > 0 {
        if !force {
            bail!(
                "{stale} message(s) embedded under a different model id; pass \
                 `{force_hint}` to re-embed (the vector index will be rebuilt under \
                 the configured model {:?})",
                pond::embed::model_id(),
            );
        }
        output(&pond::output::paint(
            &format!(
                "embed: --force: re-embedding {} stale-model row(s) after dropping IVF_PQ",
                format_thousands(stale as u64),
            ),
            pond::output::yellow(),
        ))?;
        store.drop_vector_index().await?;
    }

    let progress = store.embedding_progress().await?;
    let backlog = progress.total.saturating_sub(progress.embedded);
    let bar_total = match limit {
        Some(cap) => backlog.min(cap),
        None => backlog,
    };
    if bar_total == 0 && stale == 0 {
        // No backlog and no stale rows: stage is silent. The summary's
        // `indexes` line will confirm "semantic ready" downstream.
        return Ok(EmbedSummary::default());
    }

    let embedder = CandleEmbedder::load()?;
    let device = embedder.device().to_owned();
    let bar = ProgressBar::with_draw_target(
        Some(bar_total as u64),
        indicatif::ProgressDrawTarget::stderr_with_hz(8),
    );
    bar.set_style(
        ProgressStyle::with_template(
            "semantic embedding [{elapsed_precise}] [{bar:24}] {pos}/{len} messages  {wide_msg}",
        )
        .unwrap_or_else(|_| ProgressStyle::default_bar())
        .progress_chars("##-"),
    );
    bar.enable_steady_tick(Duration::from_millis(120));
    let cancel = Arc::new(std::sync::atomic::AtomicBool::new(false));
    {
        let cancel = cancel.clone();
        tokio::spawn(async move {
            let _ = tokio::signal::ctrl_c().await;
            cancel.store(true, std::sync::atomic::Ordering::Relaxed);
            eprintln!("\ninterrupted; flushing window (Ctrl-C again to abort)...");
            let _ = tokio::signal::ctrl_c().await;
            std::process::exit(130);
        });
    }
    let started = std::time::Instant::now();
    let bar_for_callback = bar.clone();
    let mut worker = EmbedWorker::new(store, &embedder)
        .with_cancel(cancel)
        .with_progress(move |progress: BatchProgress| {
            let secs = started.elapsed().as_secs_f64().max(0.001);
            let rate = progress.total_messages as f64 / secs;
            bar_for_callback.set_position(progress.total_messages as u64);
            bar_for_callback.set_message(format!("{rate:.0} msgs/s"));
        });
    if force {
        worker = worker.include_stale();
    }
    if let Some(limit) = limit {
        worker = worker.with_limit(limit);
    }
    let summary = worker.run().await?;
    bar.finish_and_clear();
    if summary.messages > 0 {
        let label = if summary.cancelled {
            "semantic embedding (interrupted)"
        } else {
            "semantic embedding"
        };
        output(&format!(
            "{}  +{} messages  ({})",
            pond::output::paint(label, pond::output::dim()),
            format_thousands(summary.messages as u64),
            device,
        ))?;
    }
    Ok(summary)
}

/// Resolve the `MaintenancePolicy` for one optimize/sync invocation: start
/// from `[maintenance]` (or the in-process defaults), then apply the CLI
/// `--cleanup-older-than` override when present.
fn configured_maintenance_policy(
    config: &Config,
    cleanup_override: Option<chrono::Duration>,
) -> anyhow::Result<MaintenancePolicy> {
    let compaction_fragment_cap = config
        .maintenance
        .compaction_fragment_cap
        .unwrap_or(pond::substrate::DEFAULT_COMPACTION_FRAGMENT_CAP);
    let configured_cleanup = config
        .maintenance
        .cleanup_older_than
        .as_deref()
        .map(parse_retention)
        .transpose()
        .map_err(|err| anyhow::anyhow!("invalid [maintenance].cleanup_older_than: {err}"))?;
    let cleanup_older_than = cleanup_override
        .or(configured_cleanup)
        .unwrap_or_else(default_cleanup_older_than);
    // Readers pin a manifest version per request; cleanup reclaiming a pinned
    // version's files breaks in-flight reads on object-store backends.
    if cleanup_older_than < chrono::Duration::hours(1) {
        anyhow::bail!(
            "cleanup retention below the 1h floor; set [maintenance].cleanup_older_than \
             (or --cleanup-older-than) to 1h or longer"
        );
    }
    Ok(MaintenancePolicy {
        compaction_fragment_cap,
        cleanup_older_than,
    })
}

async fn run_update_indexes_stage(
    store: &Store,
    policy: &MaintenancePolicy,
) -> anyhow::Result<OptimizeOutcome> {
    let (progress, bar) = optimize_progress_bar();
    let outcome = store.optimize_indices(Some(progress), policy).await?;
    bar.finish_and_clear();
    // No `index:` recap line: the `render_sync_summary` (or `pond status`)
    // `indexes  text + semantic ready` line is the single source of truth
    // for index health. Hints below still fire on real conflicts / failures.
    render_optimize_hints(&outcome)?;
    Ok(outcome)
}

/// Final recap of a `pond sync` run. Three-line shape:
///
/// ```text
///
/// indexes   text + semantic ready
/// added     +44 sessions, +2,337 messages
/// stored    8,460 sessions, 172,710 messages
///
/// (messages = searchable text rows; use -v for full counts)
/// ```
///
/// `added` is suppressed when both deltas are zero (a no-op sync still
/// always prints the `stored` line so an operator can confirm corpus
/// state without re-running `pond status`). The trailing disclaimer
/// goes to stderr per the rust-cli/book result-vs-meta discipline.
async fn render_sync_summary(store: &Store, summary: &SyncRunSummary) -> anyhow::Result<()> {
    use pond::output::{dim, paint};

    let (sessions_added, messages_added) = match &summary.ingest {
        Some(ingest) => (
            ingest.sessions_inserted as u64,
            ingest.messages_inserted_searchable as u64,
        ),
        None => (0, 0),
    };

    let (index_status, embedding, stats) = tokio::try_join!(
        store.index_status(),
        store.embedding_progress(),
        store.corpus_stats(false),
    )?;
    let health = classify_index_health(&index_status, index_lag_threshold(), &embedding);

    output("")?;
    output(&render_indexes_line(&health))?;
    if sessions_added + messages_added > 0 {
        output(&format!(
            "{}     +{} sessions, +{} messages",
            paint("added", dim()),
            format_thousands(sessions_added),
            format_thousands(messages_added),
        ))?;
    }
    output(&format!(
        "{}    {} sessions, {} messages",
        paint("stored", dim()),
        format_thousands(stats.totals.sessions),
        format_thousands(embedding.total as u64),
    ))?;
    output_err("")?;
    output_err(&paint(
        "(messages = searchable text rows; use -v for full counts)",
        dim(),
    ))?;
    Ok(())
}

async fn export_pond_archive(store: &Store, path: &Path) -> anyhow::Result<LanceArchiveExport> {
    if let Some(parent) = path.parent()
        && !parent.as_os_str().is_empty()
    {
        fs::create_dir_all(parent)
            .with_context(|| format!("failed to create {}", parent.display()))?;
    }
    let temp = tempfile::Builder::new()
        .prefix("pond-export-")
        .tempdir()
        .context("failed to create export staging dir")?;
    let data_dir = temp.path().join("data");
    let summary = store.export_clean_lance_datasets(&data_dir).await?;
    let manifest = PondArchiveManifest {
        archive_version: 1,
        pond_version: env!("CARGO_PKG_VERSION").to_owned(),
        protocol_version: PROTOCOL_VERSION,
        created_at: Utc::now(),
        rows: summary.rows,
        source_versions: summary.source_versions,
        embedding_model: pond::embed::model_id().to_owned(),
        embedding_dim: pond::sessions::embedding_dim(),
    };
    let manifest_json =
        serde_json::to_vec_pretty(&manifest).context("failed to serialize archive manifest")?;
    fs::write(temp.path().join("manifest.json"), manifest_json)
        .context("failed to write archive manifest")?;
    zip_directory(temp.path(), path)?;
    Ok(summary)
}

async fn import_pond_archive(store: &Store, path: &Path) -> anyhow::Result<LanceArchiveImport> {
    let temp = tempfile::Builder::new()
        .prefix("pond-import-")
        .tempdir()
        .context("failed to create import staging dir")?;
    unzip_archive(path, temp.path())?;
    let manifest_path = temp.path().join("manifest.json");
    let manifest_bytes = fs::read(&manifest_path)
        .with_context(|| format!("failed to read {}", manifest_path.display()))?;
    let manifest: PondArchiveManifest =
        serde_json::from_slice(&manifest_bytes).context("failed to parse archive manifest")?;
    if manifest.archive_version != 1 {
        bail!(
            "unsupported .pond archive version {}; supported: 1",
            manifest.archive_version
        );
    }
    if manifest.protocol_version != PROTOCOL_VERSION {
        bail!(
            "unsupported .pond protocol version {}; supported: {}",
            manifest.protocol_version,
            PROTOCOL_VERSION
        );
    }
    if manifest.embedding_dim != pond::sessions::embedding_dim() {
        bail!(
            "archive embedding_dim={} does not match configured embedding_dim={}",
            manifest.embedding_dim,
            pond::sessions::embedding_dim()
        );
    }
    store
        .import_clean_lance_datasets(&temp.path().join("data"))
        .await
}

fn zip_directory(source: &Path, dest: &Path) -> anyhow::Result<()> {
    let file =
        File::create(dest).with_context(|| format!("failed to create {}", dest.display()))?;
    let mut zip = zip::ZipWriter::new(file);
    let file_options = zip::write::SimpleFileOptions::default()
        .compression_method(zip::CompressionMethod::Deflated)
        .unix_permissions(0o644);
    let dir_options = zip::write::SimpleFileOptions::default()
        .compression_method(zip::CompressionMethod::Stored)
        .unix_permissions(0o755);
    for entry in walkdir::WalkDir::new(source).sort_by_file_name() {
        let entry = entry.context("failed to walk archive staging dir")?;
        let path = entry.path();
        let rel = path
            .strip_prefix(source)
            .context("failed to compute archive relative path")?;
        if rel.as_os_str().is_empty() {
            continue;
        }
        let name = rel
            .to_string_lossy()
            .replace(std::path::MAIN_SEPARATOR, "/");
        if entry.file_type().is_dir() {
            zip.add_directory(format!("{name}/"), dir_options)
                .with_context(|| format!("failed to add archive directory {name}"))?;
            continue;
        }
        zip.start_file(&name, file_options)
            .with_context(|| format!("failed to add archive file {name}"))?;
        let mut input =
            File::open(path).with_context(|| format!("failed to open {}", path.display()))?;
        io::copy(&mut input, &mut zip)
            .with_context(|| format!("failed to write archive file {name}"))?;
    }
    zip.finish().context("failed to finalize archive")?;
    Ok(())
}

fn unzip_archive(source: &Path, dest: &Path) -> anyhow::Result<()> {
    let file = File::open(source)
        .with_context(|| format!("failed to open archive {}", source.display()))?;
    let mut archive = zip::ZipArchive::new(file).context("failed to read .pond archive")?;
    for i in 0..archive.len() {
        let mut entry = archive
            .by_index(i)
            .context("failed to read archive entry")?;
        let Some(enclosed) = entry.enclosed_name() else {
            bail!("archive contains unsafe path {}", entry.name());
        };
        let outpath = dest.join(enclosed);
        if entry.is_dir() {
            fs::create_dir_all(&outpath)
                .with_context(|| format!("failed to create {}", outpath.display()))?;
            continue;
        }
        if let Some(parent) = outpath.parent() {
            fs::create_dir_all(parent)
                .with_context(|| format!("failed to create {}", parent.display()))?;
        }
        let mut output = File::create(&outpath)
            .with_context(|| format!("failed to create {}", outpath.display()))?;
        io::copy(&mut entry, &mut output)
            .with_context(|| format!("failed to extract {}", outpath.display()))?;
    }
    Ok(())
}

/// Resolve which (adapter, path) pairs `pond sync` should drive in this run.
///
/// Precedence:
/// 1. `--source-dir <path>` with `<adapter>` set: one-off run, no config writes.
/// 2. `<adapter>` set, `[sources.<adapter>].path` present: use that.
/// 3. `<adapter>` set, no config entry: run per-adapter discovery (one
///    candidate), prompt to add it, persist, then use it.
/// 4. No `<adapter>`, `[sources]` non-empty: sync every entry.
/// 5. No `<adapter>`, empty `[sources]`: discover across every adapter,
///    prompt, persist, then sync the picks.
fn resolve_sync_sources(
    config: &Config,
    config_file: &Path,
    name: Option<&str>,
    source_dir: Option<PathBuf>,
) -> anyhow::Result<Vec<(String, Value)>> {
    if let Some(source_dir) = source_dir {
        let name = name.ok_or_else(|| {
            anyhow::anyhow!("--source-dir requires an explicit <adapter> positional argument")
        })?;
        let known = adapter::known_names();
        if !known.contains(&name) {
            bail!("unknown adapter {name:?}; known: {}", known.join(", "));
        }
        // `--source-dir` is a filesystem-shaped override. Adapters that need
        // a richer config blob can't use this path; they must edit config.toml.
        return Ok(vec![(name.to_owned(), json!({ "path": source_dir }))]);
    }

    if let Some(name) = name {
        let known = adapter::known_names();
        if !known.contains(&name) {
            bail!("unknown adapter {name:?}; known: {}", known.join(", "));
        }
        if let Some(blob) = config.sources.get(name) {
            return Ok(vec![(name.to_owned(), blob.clone())]);
        }
        let candidates = adapter::discover(Some(name));
        let picks =
            adapter::prompt_and_persist(config_file, &candidates, io::stdin().is_terminal())?;
        return Ok(picks.into_iter().map(|c| (c.name, c.config)).collect());
    }

    if !config.sources.is_empty() {
        return config.resolve_sources(None);
    }
    let candidates = adapter::discover(None);
    let picks = adapter::prompt_and_persist(config_file, &candidates, io::stdin().is_terminal())?;
    Ok(picks.into_iter().map(|c| (c.name, c.config)).collect())
}

/// Run one adapter's ingest pass into `store` with a live progress bar and
/// one greppable log line per finished (or skipped) session.
async fn sync_with_progress(
    store: &Store,
    name: &str,
    config: Value,
    oracle: &dyn pond::adapter::SkipOracle,
) -> anyhow::Result<IngestSummary> {
    let factory = adapter::by_name(name).ok_or_else(|| {
        anyhow::anyhow!(
            "unknown adapter {name:?}; known: {}",
            adapter::known_names().join(", "),
        )
    })?;
    let adapter = factory.open(config)?;

    // `stderr_with_hz(8)` (indicatif 0.18.4) lowers the redraw rate from the
    // 20Hz default so SIGWINCH-triggered terminal reflows have time to
    // settle between renders. `{wide_msg}` truncates the (variable-length)
    // message instead of wrapping past the column count, which would leave
    // the previous render's tail orphaned in scrollback when the user
    // resizes mid-run (indicatif#144, #695, microsoft/terminal#6932).
    let bar =
        ProgressBar::with_draw_target(Some(0), indicatif::ProgressDrawTarget::stderr_with_hz(8));
    bar.set_style(
        ProgressStyle::with_template(
            "sync {prefix} [{elapsed_precise}] [{bar:12}] {pos}/{len} sessions  {wide_msg}",
        )
        .unwrap_or_else(|_| ProgressStyle::default_bar())
        .progress_chars("##-"),
    );
    // Pad to the widest adapter name so stacked bars align.
    bar.set_prefix(format!("{name:<12}"));
    bar.enable_steady_tick(Duration::from_millis(250));

    let mut messages: u64 = 0;
    let mut errors: u64 = 0;
    let mut drops: u64 = 0;
    let started = std::time::Instant::now();
    let bar_ref = &bar;

    let summary = handlers::ingest_adapter(store, adapter.as_ref(), oracle, |event| match event {
        SyncEvent::Discovered { total } => {
            if let Some(total) = total {
                bar_ref.set_length(total as u64);
            }
        }
        SyncEvent::SessionDone(outcome) => {
            // Map the four-class status to a compact bar tag + a tracing
            // status label. `dropped` is shown for Partial sessions so the
            // operator can see when one of the bar's "ok-ish" sessions
            // actually has missing events.
            let dropped_count: usize;
            let optional_reason: Option<String>;
            let status_label: &str;
            match &outcome.status {
                SyncStatus::Ok => {
                    status_label = "ok";
                    dropped_count = 0;
                    optional_reason = None;
                }
                SyncStatus::Partial {
                    dropped_events,
                    first_drop_reason,
                } => {
                    drops += *dropped_events as u64;
                    status_label = "partial";
                    dropped_count = *dropped_events;
                    optional_reason = Some(match first_drop_reason {
                        Some(reason) => {
                            format!("dropped {dropped_events} event(s) mid-session: {reason}")
                        }
                        None => format!("dropped {dropped_events} event(s) mid-session"),
                    });
                }
                SyncStatus::Skipped { reason } => {
                    errors += 1;
                    status_label = "skipped";
                    dropped_count = 0;
                    optional_reason = Some(reason.clone());
                }
                SyncStatus::Rejected { reason } => {
                    errors += 1;
                    status_label = "rejected";
                    dropped_count = 0;
                    optional_reason = Some(reason.clone());
                }
                SyncStatus::Fresh => {
                    status_label = "fresh";
                    dropped_count = 0;
                    optional_reason = None;
                }
                SyncStatus::Empty => {
                    // Sidecar/metadata files shrink the denominator so the
                    // bar lands at `N/N sessions` instead of leaving a gap.
                    let len = bar_ref.length().unwrap_or(0);
                    bar_ref.set_length(len.saturating_sub(1));
                    status_label = "empty";
                    dropped_count = 0;
                    optional_reason = None;
                }
            }
            messages += outcome.messages as u64;
            // Only surface the non-`ok`/`fresh` cases as scroll-back lines;
            // the bulk are routine successes already counted by the bar's
            // pos/len/msg counters. `pond::sync` at INFO still carries the
            // full per-session detail at `-v` verbosity.
            if !matches!(
                outcome.status,
                SyncStatus::Ok | SyncStatus::Fresh | SyncStatus::Empty
            ) {
                bar_ref.println(format_sync_line(name, &outcome, optional_reason.as_deref()));
            }
            match optional_reason.as_deref() {
                None => tracing::info!(
                    target: "pond::sync",
                    adapter = name,
                    status = status_label,
                    project = outcome.project.as_deref().unwrap_or("-"),
                    session = outcome.session_id.as_deref().unwrap_or("-"),
                    messages = outcome.messages,
                    dropped = dropped_count,
                    "session done"
                ),
                Some(reason) => tracing::info!(
                    target: "pond::sync",
                    adapter = name,
                    status = status_label,
                    project = outcome.project.as_deref().unwrap_or("-"),
                    session = outcome.session_id.as_deref().unwrap_or("-"),
                    messages = outcome.messages,
                    dropped = dropped_count,
                    %reason,
                    "session done"
                ),
            }
            if !matches!(outcome.status, SyncStatus::Empty) {
                // Empty already shrunk `len`; ticking `pos` would over-count.
                bar_ref.inc(1);
            }
            bar_ref.set_message(format_bar_message(
                messages,
                drops,
                errors,
                started.elapsed(),
            ));
        }
    })
    .await?;

    let tail = format_sync_outcome(&summary, drops, errors);
    bar.finish_with_message(tail);
    Ok(summary)
}

/// Frozen per-adapter bar tail after `sync_with_progress` finishes. Replaces
/// the in-flight throughput display (`N msgs / X msgs/s`) with the outcome
/// counts so scroll-back tells the story of what landed, not what was
/// decoded. Empty/sidecar files are intentionally not surfaced here -
/// per design they are part of normal operation, not a signal to the user.
fn format_sync_outcome(summary: &IngestSummary, drops: u64, errors: u64) -> String {
    let new_sessions = summary.sessions_inserted as u64;
    let new_messages = summary.messages_inserted_searchable as u64;
    let mut tail = if new_sessions == 0 && new_messages == 0 {
        "up to date".to_owned()
    } else {
        format!(
            "+{} sessions (+{} messages)",
            format_thousands(new_sessions),
            format_thousands(new_messages),
        )
    };
    if drops > 0 {
        tail.push_str(&format!("  {} dropped", format_thousands(drops)));
    }
    if errors > 0 {
        tail.push_str(&format!("  {} err", format_thousands(errors)));
    }
    tail
}

/// One greppable per-session log line. Examples:
///
/// ```text
/// [00:04:32] claude-code ok    project=/Users/you/Projects/app  session=58a96901-4a4f-40be-a3c1-62419ec8c580  msgs=512
/// [00:04:33] claude-code skip  /Users/you/.../58a96901-....jsonl: empty jsonl session
/// ```
fn format_sync_line(adapter: &str, outcome: &SessionOutcome, reason: Option<&str>) -> String {
    use pond::output::{dim, green, paint, red, yellow};

    let (raw_tag, tag_style) = match &outcome.status {
        SyncStatus::Ok => ("ok  ", green()),
        SyncStatus::Partial { .. } => ("part", yellow()),
        SyncStatus::Skipped { .. } => ("skip", red()),
        SyncStatus::Rejected { .. } => ("rej ", red()),
        SyncStatus::Fresh => ("fresh", green()),
        SyncStatus::Empty => ("empty", dim()),
    };
    let tag = paint(raw_tag, tag_style);
    if matches!(outcome.status, SyncStatus::Fresh) {
        let ts = chrono::Local::now().format("%H:%M:%S");
        let session = outcome.session_id.as_deref().unwrap_or("-");
        return format!("[{ts}] {adapter} {tag}  session={session}  (cached)");
    }
    let ts = chrono::Local::now().format("%H:%M:%S");
    let project = outcome.project.as_deref().unwrap_or("-");
    let session = outcome.session_id.as_deref().unwrap_or("-");
    match reason {
        None => format!(
            "[{ts}] {adapter} {tag}  project={project}  session={session}  msgs={}",
            outcome.messages,
        ),
        Some(reason) => format!("[{ts}] {adapter} {tag}  {reason}"),
    }
}

fn format_bar_message(messages: u64, drops: u64, errors: u64, elapsed: Duration) -> String {
    let secs = elapsed.as_secs_f64().max(0.001);
    let msg_per_sec = (messages as f64) / secs;
    let mut out = format!(
        "{} msgs  {:.0} msgs/s",
        format_thousands(messages),
        msg_per_sec,
    );
    // Only surface the trouble counters once they're nonzero; a clean run
    // stays short enough to fit the bar without truncation.
    if drops > 0 {
        out.push_str(&format!("  {} dropped", format_thousands(drops)));
    }
    if errors > 0 {
        out.push_str(&format!("  {} err", format_thousands(errors)));
    }
    out
}

/// Render an integer with thousands separators (`12_345_678` -> `"12,345,678"`).
fn format_thousands(value: u64) -> String {
    let raw = value.to_string();
    let mut out = String::with_capacity(raw.len() + raw.len() / 3);
    for (idx, ch) in raw.chars().rev().enumerate() {
        if idx > 0 && idx % 3 == 0 {
            out.push(',');
        }
        out.push(ch);
    }
    out.chars().rev().collect()
}

/// Pretty-print a byte count: `2_589_934_592 -> "2.41 GiB"`. Plain function
/// rather than a humansize-crate add: the spec is small, deterministic, and
/// the dep would land just to format one line of `pond status`.
fn format_bytes(bytes: u64) -> String {
    const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];
    if bytes < 1024 {
        return format!("{bytes} B");
    }
    let mut value = bytes as f64;
    let mut unit = 0;
    while value >= 1024.0 && unit < UNITS.len() - 1 {
        value /= 1024.0;
        unit += 1;
    }
    if value >= 100.0 {
        format!("{:.0} {}", value, UNITS[unit])
    } else if value >= 10.0 {
        format!("{:.1} {}", value, UNITS[unit])
    } else {
        format!("{:.2} {}", value, UNITS[unit])
    }
}

async fn explain_search(
    store: &Store,
    embedder: &LazyEmbedder,
    request: &SearchRequest,
    search: &config::SearchConfig,
) -> anyhow::Result<String> {
    handlers::explain_search_plan(store, embedder, request.clone(), search)
        .await
        .map_err(|envelope| anyhow::anyhow!("{envelope:?}"))
}

async fn wait_for_index_catchup(store: &Store) -> anyhow::Result<()> {
    let deadline = std::time::Instant::now() + Duration::from_secs(600);
    loop {
        let statuses = store.index_status().await?;
        if statuses.iter().all(|status| status.unindexed_rows == 0) {
            return Ok(());
        }
        if std::time::Instant::now() >= deadline {
            anyhow::bail!("timed out waiting for indexes to catch up");
        }
        tokio::time::sleep(Duration::from_secs(2)).await;
    }
}

/// Build the spinner + progress callback pair for index maintenance.
/// `PhaseStart` updates the spinner so the operator sees what's running live;
/// per-phase timing lands at `-vv` (debug) verbosity rather than the default
/// output, which carries one `indexes` line in the sync/status footer instead.
fn optimize_progress_bar() -> (OptimizeProgressFn, ProgressBar) {
    let bar = ProgressBar::new_spinner();
    bar.set_style(
        ProgressStyle::with_template("{spinner:.green} {elapsed_precise} {wide_msg}")
            .unwrap_or_else(|_| ProgressStyle::default_spinner()),
    );
    bar.enable_steady_tick(Duration::from_millis(120));
    let bar_for_callback = bar.clone();
    let callback: OptimizeProgressFn = Box::new(move |event| match event {
        OptimizeEvent::PhaseStart {
            table,
            phase,
            detail,
        } => {
            let label = match detail {
                Some(d) => format!("{} {} ({d})", table.as_str(), phase.label()),
                None => format!("{} {}", table.as_str(), phase.label()),
            };
            bar_for_callback.set_message(label);
        }
        OptimizeEvent::PhaseDone {
            table,
            phase,
            elapsed_ms,
        } => {
            tracing::debug!(
                target: "pond::sync",
                table = table.as_str(),
                phase = phase.label(),
                elapsed_ms,
                "index phase done"
            );
        }
    });
    (callback, bar)
}

fn render_optimize_outcome(outcome: &OptimizeOutcome) -> anyhow::Result<()> {
    use pond::output::{bold, paint};
    let mut table = new_table();
    table.set_header(vec!["table", "indices", "compaction"]);
    for entry in &outcome.tables {
        table.add_row(vec![
            Cell::new(entry.table.as_str()),
            phase_cell(&entry.indices, "indices"),
            phase_cell(&entry.compaction, "compaction"),
        ]);
    }
    output(&paint("index maintenance", bold()))?;
    output(&table.to_string())?;
    render_optimize_hints(outcome)
}

/// Deferral and failure lines only, no per-table table. Used by `pond sync`,
/// whose summary line already carries per-table status; the table would just
/// repeat it.
fn render_optimize_hints(outcome: &OptimizeOutcome) -> anyhow::Result<()> {
    use pond::output::{dim, paint, red, yellow};
    for entry in &outcome.tables {
        if matches!(entry.compaction, PhaseOutcome::SkippedConflict) {
            output(&format!(
                "{}  compaction on {} deferred: concurrent writer; rerun once it finishes",
                paint("hint", dim()),
                entry.table.as_str(),
            ))?;
        }
    }
    for entry in &outcome.tables {
        if let PhaseOutcome::Failed(error) = &entry.indices {
            output(&paint(
                &format!("error  indices on {}: {error:#}", entry.table.as_str()),
                red(),
            ))?;
        }
        if let PhaseOutcome::Failed(error) = &entry.compaction {
            output(&paint(
                &format!("error  compaction on {}: {error:#}", entry.table.as_str()),
                yellow(),
            ))?;
        }
    }
    Ok(())
}

fn phase_cell(outcome: &PhaseOutcome, _phase: &str) -> Cell {
    use pond::output::{dim, paint, red, yellow};
    match outcome {
        PhaseOutcome::Ok => Cell::new("ok"),
        PhaseOutcome::Noop => Cell::new(paint("-", dim())),
        PhaseOutcome::NotAttempted => Cell::new(paint("-", dim())),
        PhaseOutcome::SkippedConflict => Cell::new(paint("skipped (conflict)", yellow())),
        PhaseOutcome::Failed(_) => Cell::new(paint("failed", red())),
    }
}

fn render_index_status(statuses: &[IndexStatus]) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint, yellow};
    let mut table = new_table();
    table.set_header(vec![
        "table",
        "intent",
        "exists",
        "fragments",
        "unindexed rows",
    ]);
    for status in statuses {
        let unindexed = format_thousands(status.unindexed_rows as u64);
        let unindexed_cell = if status.unindexed_rows == 0 {
            Cell::new(unindexed)
        } else {
            Cell::new(paint(&unindexed, yellow()))
        };
        table.add_row(vec![
            Cell::new(status.table.as_str()),
            Cell::new(&status.intent_name),
            Cell::new(if status.exists { "yes" } else { "no" }),
            Cell::new(status.fragments_covered.to_string()),
            unindexed_cell.set_alignment(CellAlignment::Right),
        ]);
    }
    output(&paint("index status", bold()))?;
    output(&table.to_string())?;
    if statuses.iter().any(|status| status.unindexed_rows > 0) {
        output(&format!(
            "{}  run `pond sync --only update-indexes` to fold trailing fragments",
            paint("hint", dim()),
        ))?;
    }
    Ok(())
}

/// Title + storage-destination line, shared by the populated header and the
/// empty-store render so both `pond status` and `pond storage` open the same way.
fn render_status_storage_line(title: &str, resolved: &ResolvedStorage) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};
    output(&paint(title, bold()))?;
    output(&format!(
        "{}  {}  {}",
        paint("storage", dim()),
        resolved.display(),
        paint(&format!("[{}]", resolved.binding.describe()), dim()),
    ))?;
    Ok(())
}

/// Storage configured but never synced (no tables): the storage line plus a
/// pointer at `pond sync`, instead of erroring on the first table describe.
fn render_empty_status(title: &str, resolved: &ResolvedStorage) -> anyhow::Result<()> {
    use pond::output::{dim, paint};
    render_status_storage_line(title, resolved)?;
    output(&format!(
        "{}    no data yet - run `pond sync` to import sessions",
        paint("stored", dim()),
    ))?;
    Ok(())
}

fn render_status_header(
    title: &str,
    resolved: &ResolvedStorage,
    sizes: &TableSizes,
    totals: &RowTotals,
) -> anyhow::Result<()> {
    render_status_storage_line(title, resolved)?;

    let mut table = new_table();
    let total_bytes = sizes.sessions + sizes.messages + sizes.parts + sizes.other;
    let rows = [
        (
            "sessions",
            sizes.sessions,
            Some(totals.sessions),
            sizes.sessions_data,
        ),
        (
            "messages",
            sizes.messages,
            Some(totals.messages),
            sizes.messages_data,
        ),
        ("parts", sizes.parts, Some(totals.parts), sizes.parts_data),
        ("other", sizes.other, None, Default::default()),
    ];
    for (label, bytes, rows_opt, data) in rows {
        // Surface superseded data versions only when they matter: the gap
        // self-heals once manifests age past the cleanup retention window.
        let dead_note = data
            .dead()
            .filter(|dead| *dead > 64 * 1024 * 1024 && *dead * 10 > data.on_disk)
            .map(|dead| format!("{} pending cleanup", format_bytes(dead)))
            .unwrap_or_default();
        table.add_row(vec![
            Cell::new(format!("  {label}")),
            Cell::new(format_bytes(bytes)).set_alignment(CellAlignment::Right),
            Cell::new(
                rows_opt
                    .map(|n| format!("{} rows", format_thousands(n)))
                    .unwrap_or_default(),
            )
            .set_alignment(CellAlignment::Right),
            Cell::new(dead_note).set_alignment(CellAlignment::Right),
        ]);
    }
    table.add_row(vec![
        Cell::new("  total").add_attribute(Attribute::Bold),
        Cell::new(format_bytes(total_bytes))
            .set_alignment(CellAlignment::Right)
            .add_attribute(Attribute::Bold),
        Cell::new(""),
        Cell::new(""),
    ]);
    output(&table.to_string())?;
    Ok(())
}

/// Render the checks that can take longer on a large corpus. The command
/// prints storage first, then calls this once the bounded scans finish.
fn render_status_checks(
    stats: &CorpusStats,
    index_status: &[IndexStatus],
    embedding: EmbeddingProgress,
    adapters: bool,
) -> anyhow::Result<()> {
    use pond::output::{dim, paint, yellow};

    output("")?;
    let health = classify_index_health(index_status, index_lag_threshold(), &embedding);
    output(&render_indexes_line(&health))?;
    output(&format!(
        "{}    {} sessions, {} messages",
        paint("stored", dim()),
        format_thousands(stats.totals.sessions),
        format_thousands(embedding.total as u64),
    ))?;
    if adapters {
        output("")?;
        output(&paint("index detail", dim()))?;
        for status in index_status {
            let line = format!(
                "  {}.{}  exists={}  fragments={}  unindexed={}",
                status.table.as_str(),
                status.intent_name,
                if status.exists { "yes" } else { "no" },
                status.fragments_covered,
                format_thousands(status.unindexed_rows as u64),
            );
            if status.unindexed_rows == 0 {
                output(&line)?;
            } else {
                output(&paint(&line, yellow()))?;
            }
        }
    }

    if !adapters {
        output(&format!(
            "{}    {} adapter(s); pass `--adapters` for project tables",
            paint("sources", dim()),
            stats.adapters.len(),
        ))?;
        output(&crate::schedule::status_line())?;
    } else {
        // Render adapters in registry order so the layout matches the discovery
        // picker; adapters present in the data but not in the registry append at
        // the bottom (defensive: catches deleted adapters whose data is still on
        // disk).
        let mut by_name: std::collections::HashMap<&str, &AdapterStats> = stats
            .adapters
            .iter()
            .map(|stat| (stat.adapter.as_str(), stat))
            .collect();
        for factory in adapter::registry() {
            if let Some(stat) = by_name.remove(factory.name()) {
                render_adapter_block(stat)?;
            }
        }
        for stat in by_name.values() {
            render_adapter_block(stat)?;
        }
    }
    output_err("")?;
    output_err(&paint(
        "(messages = searchable text rows; use -v for full counts)",
        dim(),
    ))?;
    Ok(())
}

#[derive(Debug, Clone)]
enum IndexHealthState {
    NotBuilt,
    Ready,
    Pending(u64),
}

#[derive(Debug, Clone)]
struct IndexHealth {
    text: IndexHealthState,
    semantic: IndexHealthState,
}

/// `Ready` means the substrate's lag guard is intentionally batching; queries
/// fall through to the brute-force scan over a small remainder. `Pending(N)`
/// means the trailing fragment count crossed the threshold and a fold is owed.
fn classify_index_health(
    statuses: &[IndexStatus],
    lag_threshold: usize,
    embedding: &EmbeddingProgress,
) -> IndexHealth {
    use IndexHealthState::*;

    fn classify_one(status: &IndexStatus, lag_threshold: usize) -> IndexHealthState {
        if !status.exists {
            return NotBuilt;
        }
        if status.unindexed_rows == 0 || status.unindexed_fragments < lag_threshold {
            Ready
        } else {
            Pending(status.unindexed_rows as u64)
        }
    }

    let mut text = NotBuilt;
    let mut semantic = NotBuilt;
    for status in statuses {
        match status.intent_name.as_str() {
            MESSAGES_FTS_INDEX => text = classify_one(status, lag_threshold),
            MESSAGES_VECTOR_INDEX => semantic = classify_one(status, lag_threshold),
            _ => {}
        }
    }
    // Semantic search misses unembedded rows even when IVF_PQ's own
    // unindexed-fragments check passes.
    let embed_backlog = embedding.total.saturating_sub(embedding.embedded);
    if embed_backlog > 0 && matches!(semantic, Ready) {
        semantic = Pending(embed_backlog as u64);
    }
    IndexHealth { text, semantic }
}

fn render_indexes_line(health: &IndexHealth) -> String {
    use IndexHealthState::*;
    use pond::output::{dim, paint, yellow};

    let body = match (&health.text, &health.semantic) {
        (Ready, Ready) => "text + semantic ready".to_owned(),
        _ => {
            let text_part = match &health.text {
                Ready => "text ready".to_owned(),
                Pending(n) => format!("text {} pending", format_thousands(*n)),
                NotBuilt => "text not built".to_owned(),
            };
            let semantic_part = match &health.semantic {
                Ready => "semantic ready".to_owned(),
                Pending(n) => format!("semantic {} pending", format_thousands(*n)),
                NotBuilt => "semantic below activation threshold".to_owned(),
            };
            format!("{text_part} . {semantic_part}")
        }
    };
    let any_pending = matches!(health.text, Pending(_)) || matches!(health.semantic, Pending(_));
    let label = if any_pending {
        paint("indexes", yellow())
    } else {
        paint("indexes", dim())
    };
    format!("{label}   {body}")
}

fn render_adapter_block(stat: &AdapterStats) -> anyhow::Result<()> {
    use pond::output::{bold, cyan, paint};

    output("")?;
    output(&format!(
        "{}  {} sessions  {} messages  {} projects",
        paint(&stat.adapter, cyan().bold()),
        paint(&format_thousands(stat.sessions), bold()),
        paint(&format_thousands(stat.messages), bold()),
        paint(&format_thousands(stat.projects.len() as u64), bold()),
    ))?;
    if stat.projects.is_empty() {
        return Ok(());
    }
    let mut table = new_table();
    table.set_header(vec![
        Cell::new("project")
            .add_attribute(Attribute::Bold)
            .add_attribute(Attribute::Dim),
        Cell::new("sessions")
            .set_alignment(CellAlignment::Right)
            .add_attribute(Attribute::Bold)
            .add_attribute(Attribute::Dim),
        Cell::new("messages")
            .set_alignment(CellAlignment::Right)
            .add_attribute(Attribute::Bold)
            .add_attribute(Attribute::Dim),
    ]);
    for project in &stat.projects {
        let label = project.project.as_str();
        table.add_row(vec![
            Cell::new(label),
            Cell::new(format_thousands(project.sessions)).set_alignment(CellAlignment::Right),
            Cell::new(format_thousands(project.messages)).set_alignment(CellAlignment::Right),
        ]);
    }
    // Let the project column flex; right-size the numeric columns to their
    // content so the long path takes the remaining width and truncates with
    // an ellipsis on narrow terminals.
    if let Some(col) = table.column_mut(1) {
        col.set_constraint(ColumnConstraint::ContentWidth);
    }
    if let Some(col) = table.column_mut(2) {
        col.set_constraint(ColumnConstraint::ContentWidth);
    }
    output(&table.to_string())?;
    Ok(())
}

/// House style for `pond status` tables: borderless, dynamic-width, no inner
/// rules. Centralized so future tabular commands match without copy-paste.
fn new_table() -> Table {
    let mut table = Table::new();
    table
        .load_preset(NOTHING)
        .set_content_arrangement(ContentArrangement::Dynamic);
    table
}

/// Dispatch an envelope through the chosen format. Returns `true` when the
/// envelope was a `Success` (callers exit non-zero on `false`). JSON mode
/// always emits the envelope to stdout so scripts can pipe both success and
/// error bodies through `jq`; pretty mode routes errors to stderr so stdout
/// stays parseable.
fn render_search_envelope(format: OutputFormat, envelope: &SearchEnvelope) -> anyhow::Result<bool> {
    match format {
        OutputFormat::Json => {
            output(
                &serde_json::to_string_pretty(envelope)
                    .context("serialize search envelope as JSON")?,
            )?;
            Ok(matches!(envelope, SearchEnvelope::Success(_)))
        }
        OutputFormat::Pretty => match envelope {
            SearchEnvelope::Success(response) => {
                render_search_pretty(response)?;
                Ok(true)
            }
            SearchEnvelope::Error(error) => {
                render_error_pretty(error);
                Ok(false)
            }
        },
    }
}

fn render_get_envelope(
    format: OutputFormat,
    envelope: &GetEnvelope,
    session_from: SessionFrom,
) -> anyhow::Result<bool> {
    match format {
        OutputFormat::Json => {
            output(
                &serde_json::to_string_pretty(envelope)
                    .context("serialize get envelope as JSON")?,
            )?;
            Ok(matches!(envelope, GetEnvelope::Success(_)))
        }
        OutputFormat::Pretty => match envelope {
            GetEnvelope::Success(response) => {
                render_get_pretty(response, session_from)?;
                Ok(true)
            }
            GetEnvelope::Error(error) => {
                render_error_pretty(error);
                Ok(false)
            }
        },
    }
}

fn render_search_pretty(response: &SearchResponse) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};

    output(&format!(
        "{} {} matched {}  {} returned {}  {} {} searchable in scope",
        paint("search:", dim()),
        paint(&format_thousands(response.matched_total as u64), bold()),
        if response.matched_total == 1 {
            "message"
        } else {
            "messages"
        },
        paint(&format_thousands(response.sessions.len() as u64), bold()),
        if response.sessions.len() == 1 {
            "session"
        } else {
            "sessions"
        },
        paint("of", dim()),
        paint(
            &format_thousands(response.searchable_in_scope as u64),
            bold()
        ),
    ))?;
    if response.sessions.is_empty() {
        // spec.md#search-absence-honesty: name the recovery when the scope
        // itself is empty - the filters, not the query, produced the zero.
        if response.searchable_in_scope == 0 {
            output(
                "scope is empty: the filters exclude every searchable message; widen or drop \
                 project/date filters",
            )?;
        }
        return Ok(());
    }
    for (idx, session) in response.sessions.iter().enumerate() {
        output("")?;
        render_search_session(idx + 1, session)?;
    }
    Ok(())
}

fn render_search_session(rank: usize, session: &SearchSession) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};

    let best_score = session
        .matches
        .first()
        .map(|hit| hit.score)
        .unwrap_or_default();
    output(&format!(
        "{}  best={}  {}/{} matched",
        paint(&format!("[{rank}]"), dim()),
        paint(&format!("{best_score:.4}"), bold()),
        paint(
            &format_thousands(session.matched_message_count as u64),
            bold(),
        ),
        paint(
            &format_thousands(session.session_messages_count as u64),
            bold(),
        ),
    ))?;
    output(&format!(
        "    {}  {}  {}",
        paint(&session.project, dim()),
        paint(&session.source_agent, dim()),
        paint(&session.session_id, dim()),
    ))?;
    for hit in &session.matches {
        render_search_match(hit)?;
    }
    Ok(())
}

fn render_search_match(hit: &SearchResult) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};
    output(&format!(
        "    {}  {}  {}  {}",
        paint(
            &hit.timestamp.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
            dim(),
        ),
        paint_role(hit.role.as_str()),
        paint(&format!("{:.4}", hit.score), bold()),
        paint(&hit.message_id, dim()),
    ))?;
    render_hit_text(&hit.text)?;
    Ok(())
}

fn render_hit_text(text: &str) -> anyhow::Result<()> {
    use pond::output::{dim, paint};
    let prefix = paint(">", dim());
    for line in text.lines() {
        output(&format!("    {prefix} {line}"))?;
    }
    Ok(())
}

fn render_session_header(session: &pond::wire::GetSession) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};
    output(&format!(
        "{} {}  source={}  project={}",
        paint("session", dim()),
        paint(&session.id, bold()),
        session.source_agent,
        session.project.as_str(),
    ))?;
    output(&format!(
        "{} {}",
        paint("created:", dim()),
        paint(
            &session.created_at.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
            dim(),
        ),
    ))
}

fn render_get_pretty(response: &GetResponse, session_from: SessionFrom) -> anyhow::Result<()> {
    use pond::output::{bold, dim, paint};

    render_session_header(&response.session)?;
    match &response.result {
        GetResult::Session {
            messages,
            messages_remaining,
        } => {
            for (idx, message) in messages.iter().enumerate() {
                output("")?;
                let parts = message.parts.as_deref().unwrap_or(&[]);
                render_message_view(idx + 1, message, parts, false)?;
            }
            output("")?;
            // Tail page: the remaining messages are *earlier*, before this page;
            // after_id only pages forward, so label them "earlier" and omit the
            // dead-end cursor (the start path keeps the forward after-id cursor).
            let tail = matches!(session_from, SessionFrom::End);
            let mut footer = format!(
                "{} {} messages",
                paint("(total:", dim()),
                paint(&format_thousands(messages.len() as u64), bold()),
            );
            if *messages_remaining > 0 {
                footer.push_str(&format!(
                    " {} {}",
                    paint(&format_thousands(*messages_remaining as u64), bold()),
                    paint(if tail { "earlier" } else { "remaining [more]" }, dim()),
                ));
            }
            footer.push_str(&paint(")", dim()));
            output(&footer)?;
            if *messages_remaining > 0 {
                if tail {
                    output(&paint(
                        "session-from: start to read from the beginning",
                        dim(),
                    ))?;
                } else if let Some(last) = messages.last() {
                    output(&format!("{} {}", paint("after-id:", dim()), last.id))?;
                }
            }
        }
        GetResult::Message {
            target,
            target_parts,
            target_parts_remaining,
            siblings,
        } => {
            // Interleave the target with its siblings in timestamp order so the
            // thread reads top-to-bottom; the target carries its full parts.
            let mut thread: Vec<(&MessageView, bool)> =
                siblings.iter().map(|view| (view, false)).collect();
            thread.push((target, true));
            thread.sort_by_key(|(view, _)| view.timestamp);
            for (idx, (view, is_target)) in thread.iter().enumerate() {
                output("")?;
                let parts = if *is_target {
                    target_parts.as_slice()
                } else {
                    &[]
                };
                render_message_view(idx + 1, view, parts, *is_target)?;
            }
            if *target_parts_remaining > 0 {
                output("")?;
                output(&format!(
                    "{} {} parts remaining {}",
                    paint("(target:", dim()),
                    paint(&format_thousands(*target_parts_remaining as u64), bold()),
                    paint("[more])", dim()),
                ))?;
                if let Some(last) = target_parts.last() {
                    output(&format!("{} {}", paint("after-id:", dim()), last.id))?;
                }
            }
        }
    }
    Ok(())
}

/// Render one message view: header, text/content, then either full parts (when
/// supplied - verbatim session parts or a message-mode target) or the compact
/// part summaries otherwise.
fn render_message_view(
    rank: usize,
    view: &MessageView,
    full_parts: &[ResponsePart],
    is_target: bool,
) -> anyhow::Result<()> {
    use pond::output::{dim, paint};

    let marker = if is_target {
        paint("  <- target", dim())
    } else {
        String::new()
    };
    output(&format!(
        "{}  {}  {}  {}{marker}",
        paint(&format!("[{rank}]"), dim()),
        paint(
            &view.timestamp.format("%Y-%m-%dT%H:%M:%SZ").to_string(),
            dim(),
        ),
        paint_role(view.role.as_str()),
        paint(&view.id, dim()),
    ))?;
    // When full parts are present they are the complete content; rendering
    // `text` (a search_text projection of those same parts) too would just
    // double the body.
    if full_parts.is_empty() {
        if let Some(text) = &view.text {
            render_hit_text(text)?;
        }
        if let Some(content) = &view.content {
            render_hit_text(content)?;
        }
        for summary in &view.parts_summary {
            render_part_summary(summary)?;
        }
    } else {
        for part in full_parts {
            render_part(part)?;
        }
    }
    Ok(())
}

fn render_part_summary(summary: &PartSummary) -> anyhow::Result<()> {
    use pond::output::{dim, paint};
    let mut line = format!("[{}]", summary.kind);
    if let Some(label) = &summary.label {
        line.push(' ');
        line.push_str(label);
    }
    if let Some(call_id) = &summary.call_id {
        line.push_str(&format!(" call_id={call_id}"));
    }
    output(&format!("    {}", paint(&line, dim())))
}

fn render_part(part: &ResponsePart) -> anyhow::Result<()> {
    use pond::output::{dim, paint, yellow};

    let prefix = paint(">", dim());
    match &part.kind {
        // `Option<String>`: render only what's there. A `None` text part
        // means the source row carried no text field; printing nothing is
        // the faithful representation - no "<unresolved>" placeholder.
        PartKind::Text { text } => {
            if let Some(text) = text {
                for line in text.lines() {
                    output(&format!("    {prefix} {line}"))?;
                }
            }
        }
        PartKind::Reasoning { text } => {
            let tag = paint("[reasoning]", dim());
            if let Some(text) = text {
                for line in text.lines() {
                    output(&format!("    {tag} {prefix} {line}"))?;
                }
            }
        }
        PartKind::File {
            media_type,
            file_name,
            ..
        } => {
            output(&format!(
                "    {} media_type={} file_name={}",
                paint("[file]", yellow()),
                media_type.as_deref().unwrap_or("-"),
                file_name.as_deref().unwrap_or("-"),
            ))?;
        }
        // For tool_call / tool_result: omit the field entirely when None.
        // Concretely: a tool_result with no resolvable name prints as
        // `[tool_result] call_id=toolu_01...` (no name token), not
        // `[tool_result] unknown call_id=toolu_01...` (which lied) and
        // not `[tool_result] - call_id=toolu_01...` (which translates).
        PartKind::ToolCall { call_id, name, .. } => {
            let name_token = name.as_deref().map(|n| format!(" {n}")).unwrap_or_default();
            let call_id_token = call_id
                .as_deref()
                .map(|id| format!(" call_id={id}"))
                .unwrap_or_default();
            output(&format!(
                "    {}{name_token}{call_id_token}",
                paint("[tool_call]", yellow()),
            ))?;
        }
        PartKind::ToolResult {
            call_id,
            name,
            is_failure,
            ..
        } => {
            let name_token = name.as_deref().map(|n| format!(" {n}")).unwrap_or_default();
            let call_id_token = call_id
                .as_deref()
                .map(|id| format!(" call_id={id}"))
                .unwrap_or_default();
            output(&format!(
                "    {}{name_token}{call_id_token}{}",
                paint("[tool_result]", yellow()),
                if *is_failure { " (failure)" } else { "" },
            ))?;
        }
        PartKind::ToolApprovalRequest {
            approval_id,
            tool_call_id,
        } => {
            output(&format!(
                "    {} approval_id={approval_id} tool_call_id={tool_call_id}",
                paint("[approval_request]", yellow()),
            ))?;
        }
        PartKind::ToolApprovalResponse {
            approval_id,
            approved,
            reason,
        } => {
            let suffix = reason
                .as_deref()
                .map(|r| format!(" reason={r}"))
                .unwrap_or_default();
            output(&format!(
                "    {} approval_id={approval_id} approved={approved}{suffix}",
                paint("[approval_response]", yellow()),
            ))?;
        }
    }
    Ok(())
}

fn render_error_pretty(error: &ErrorEnvelope) {
    use pond::output::{bold, dim, paint, red};

    let code = match error.error.code {
        wire::ErrorCode::ValidationFailed => "validation_failed",
        wire::ErrorCode::VersionUnsupported => "version_unsupported",
        wire::ErrorCode::NotFound => "not_found",
        wire::ErrorCode::NamespaceUnknown => "namespace_unknown",
        wire::ErrorCode::StorageUnavailable => "storage_unavailable",
        wire::ErrorCode::Conflict => "conflict",
        wire::ErrorCode::Internal => "internal",
    };
    eprintln!(
        "{} {} {}",
        paint("error", red().bold()),
        paint(code, bold()),
        error.error.message,
    );
    let details_present = !error.error.details.is_null()
        && !error
            .error
            .details
            .as_object()
            .map(|map| map.is_empty())
            .unwrap_or(false);
    if details_present {
        eprintln!(
            "{}",
            paint(&format!("  details: {}", error.error.details), dim()),
        );
    }
}

fn paint_role(role: &str) -> String {
    use pond::output::{cyan, dim, green, paint, yellow};
    let style = match role {
        "user" => green(),
        "assistant" => cyan(),
        "tool" => yellow(),
        _ => dim(),
    };
    paint(role, style)
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used, clippy::unwrap_used)]

    use super::*;

    #[test]
    fn redaction_masks_secret_fields_but_spares_file_and_command_variants() {
        // spec.md#storage-redaction: name-based masking, including `extra`.
        assert_eq!(
            redact_config_value("creds.work.access_key_id", "AKIA"),
            "********"
        );
        assert_eq!(
            redact_config_value("creds.work.secret_access_key", "s"),
            "********"
        );
        assert_eq!(
            redact_config_value("creds.work.extra.sas_token", "t"),
            "********"
        );
        assert_eq!(
            redact_config_value("creds.work.extra.request_timeout", "60s"),
            "60s"
        );
        // The path / command IS the safe part.
        assert_eq!(
            redact_config_value("creds.work.access_key_id_file", "/k"),
            "/k"
        );
        assert_eq!(
            redact_config_value("creds.work.secret_access_key_command", "op read x"),
            "op read x"
        );
        assert_eq!(redact_config_value("creds.work.region", "fsn1"), "fsn1");
    }

    #[test]
    fn flatten_config_emits_dotted_leaves_and_skips_nulls() {
        let value = serde_json::json!({
            "storage": {"path": "/p"},
            "search": {"nprobes": null},
            "creds": {"work": {"region": "r", "extra": {"a": "b"}}},
        });
        let mut rows = Vec::new();
        flatten_config(String::new(), &value, &mut rows);
        assert!(rows.contains(&("storage.path".to_owned(), "/p".to_owned())));
        assert!(rows.contains(&("creds.work.extra.a".to_owned(), "b".to_owned())));
        assert!(rows.iter().all(|(key, _)| key != "search.nprobes"));
    }

    #[test]
    fn storage_location_resolves_flag_then_config_then_default() {
        let mut loaded = Config::default();
        // Built-in default: the platform-local dir.
        let resolved = resolve_storage_location(None, &loaded).unwrap();
        assert!(resolved.is_local());
        // `[storage].path` beats the default.
        loaded.storage.path = Some("s3://from-config/p".to_owned());
        let resolved = resolve_storage_location(None, &loaded).unwrap();
        assert_eq!(resolved.lance_url().as_str(), "s3://from-config/p");
        // The CLI/env flag (folded by clap) beats the file.
        let flag = StorageUrl::parse("s3://from-flag/p").unwrap();
        let resolved = resolve_storage_location(Some(flag), &loaded).unwrap();
        assert_eq!(resolved.lance_url().as_str(), "s3://from-flag/p");
    }

    #[test]
    fn cli_parses_and_subcommands_are_wired() {
        // The canonical clap derive self-check: catches conflicting flags,
        // broken groups, and missing value parsers at test time.
        Cli::command().debug_assert();
    }

    // Long-help snapshots for the root and every visible subcommand. The
    // help text IS the agent-facing API surface (the docs promise that
    // `pond <cmd> --help` carries examples), so a wording change must show
    // up in review as a snapshot diff. `max_term_width = 100` on the root
    // keeps wrapping deterministic regardless of the invoking terminal, and
    // long help never embeds the version string, so release bumps don't
    // churn these.
    #[test]
    fn help_snapshots() {
        let mut root = Cli::command();
        root.build();
        insta::assert_snapshot!("help_root", root.render_long_help().to_string());
        let visible: Vec<String> = root
            .get_subcommands()
            .filter(|sub| !sub.is_hide_set() && sub.get_name() != "help")
            .map(|sub| sub.get_name().to_owned())
            .collect();
        for name in visible {
            let sub = root
                .find_subcommand_mut(&name)
                .expect("visible subcommand exists");
            insta::assert_snapshot!(format!("help_{name}"), sub.render_long_help().to_string());
        }
    }
}