agentmux_srv/migrations/

runner.rs

// Copyright 2026, AgentMux Corp.
// SPDX-License-Identifier: Apache-2.0

//! Migration runner — invoked by `agentmux-srv migrate` (a CLI subcommand,
//! and the now-dead `agentmux-launcher::srv_spawner::run_migrate` subprocess
//! path) and, separately, in-process at every srv boot via
//! `run_pending_migrations`.
//!
//! Both paths report the SAME per-migration begin/done progress
//! (`ApplyProgress`) but over different channels: the CLI path emits
//! newline-delimited JSON to stdout (read by the dead subprocess path, kept
//! for that reason); `run_pending_migrations` emits `agentmux_common::
//! srv_stderr`'s `AGENTMUXSRV-MIGRATION-BEGIN`/`-END` lines to stderr — the
//! real, in-process boot path's own protocol, so the launcher's splash
//! screen can show live per-migration sub-rows instead of a silent clock.
//! Failures are written to `<home>/logs/migration-error.log`; the
//! process exits non-zero so the launcher can surface the error rather than
//! booting with a half-migrated data dir.

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::time::Instant;

use crate::backend::storage::store::Store;
use crate::registry::{resolve_global_shared_root, resolve_shared_store_path};

use super::{MigrationContext, MigrationScope, REGISTRY};

// ── Progress events (newline-delimited JSON → launcher reads these) ───────────

fn emit(event: &str, id: &str, extra: &str) {
    if extra.is_empty() {
        println!("{{\"event\":\"{}\",\"id\":\"{}\"}}", event, id);
    } else {
        println!("{{\"event\":\"{}\",\"id\":\"{}\",{}}}", event, id, extra);
    }
}

fn emit_summary(applied: usize, skipped: usize) {
    println!(
        "{{\"event\":\"complete\",\"applied\":{},\"skipped\":{}}}",
        applied, skipped
    );
}

/// The true, unconditional `~/.agentmux` root — `MigrationContext.home`.
///
/// Deliberately NOT derived from `resolve_shared_store_path()` (which
/// varies under isolated-auth mode, resolving to a channel-scoped path
/// instead of the global one) — nearly every registered migration builds
/// paths directly off `ctx.home` (registry/definitions/transcripts dirs),
/// and `backup_stores`/`write_error_log` below use it too. All of those
/// must stay anchored to the real global root regardless of isolation;
/// only `ctx.shared_store_path` itself is meant to vary. A single shared
/// helper (not one inline derivation per call site) so the two can't
/// drift apart. See `registry::resolve_global_shared_root`'s doc comment
/// and `docs/specs/SPEC_ISOLATED_AUTH_DEV_TESTING_2026_07_27.md`.
fn resolve_home() -> Option<PathBuf> {
    resolve_global_shared_root().and_then(|p| p.parent().map(Path::to_path_buf))
}

// ── Entry point ───────────────────────────────────────────────────────────────

/// Exit code of `migrate --verify` when at least one applied migration's
/// post-condition is a mismatch or its check errored. Distinct from 1 (a
/// migration RUN failed) so a caller can tell "your data is inconsistent"
/// from "the migrate command itself broke", and from 2, which clap uses for
/// a CLI usage error (a typoed flag must not read as a data finding —
/// reagent P2 on #3058). See docs/exe-return-codes.md.
pub const VERIFY_FAILED_EXIT_CODE: i32 = 3;

/// Called from `main.rs` when the `migrate` subcommand is active.
/// Returns the exit code (0 = success, 1 = failure, 3 = `--verify` found a
/// mismatch — see `VERIFY_FAILED_EXIT_CODE`).
pub fn run_migrate_command(data_dir: &Path, dry_run: bool, list: bool, verify: bool) -> i32 {
    // Unresolvable shared store path means we're in CI or an unusual env that
    // has no AGENTMUX_SHARED_DIR. Mirror the daemon's behaviour: treat as a
    // no-op and exit 0 so the launcher proceeds normally.
    let shared_store_path = match resolve_shared_store_path() {
        Some(p) => p,
        None => {
            emit_summary(0, 0);
            return 0;
        }
    };

    let home = match resolve_home() {
        Some(p) => p,
        None => {
            emit_summary(0, 0);
            return 0;
        }
    };

    // `--verify` is read-only by contract, so it dispatches BEFORE the store
    // opens below: `Store::open*` run schema setup, WAL configuration and
    // legacy-table adoption, and create the database file when it is missing
    // (codex P2 on #3058). A doctor must not do any of that — least of all
    // turn a typoed data dir into a fresh empty store that then "verifies".
    if verify {
        return cmd_verify(data_dir, &home, &shared_store_path);
    }

    // Ensure the shared/ directory exists — on a fresh install it has not been
    // created yet (the daemon does create_dir_all at startup, but migrate runs
    // before the daemon). SQLite cannot create parent directories itself.
    if let Some(parent) = shared_store_path.parent() {
        if let Err(e) = std::fs::create_dir_all(parent) {
            eprintln!("migration: failed to create shared dir {}: {}", parent.display(), e);
            return 1;
        }
    }

    // `--list` and `--dry-run` are read-only previews and deliberately do NOT
    // take the migration lock: an operator uses them precisely while another
    // process may be mid-migration, and a 30-minute wait would defeat that
    // (reagent P1 on #3062). They open the stores the ordinary way (schema
    // setup and all — the same thing the daemon does on every boot).
    if list || dry_run {
        let (shared_store, channel_store) = match open_stores_for_preview(&shared_store_path, data_dir) {
            Ok(pair) => pair,
            Err(msg) => {
                eprintln!("migration: {}", msg);
                return 1;
            }
        };
        if list {
            return cmd_list(&shared_store, channel_store.as_ref());
        }
        let pending = pending_of(REGISTRY, &shared_store, channel_store.as_ref());
        if pending.is_empty() {
            emit_summary(0, REGISTRY.len());
        }
        for m in &pending {
            println!("pending: {} — {}", m.id(), m.description());
        }
        return 0;
    }

    // Apply: the SAME locked batch the daemon runs (`apply_pending`), so the
    // CLI can never race a booting srv — and the lock is taken before the
    // stores are opened, so a daemon holding a write transaction longer than
    // `Store::open*`'s own 5 s busy timeout waits on the lock instead of
    // failing with a spurious "database is locked" (reagent P1 on #3062).
    // The only CLI-specific parts are the NDJSON progress lines the
    // launcher's `run_migrate` reader expects and the error log file.
    let progress = ApplyProgress {
        on_start: &|id, description| {
            emit("migration_start", id, &format!("\"description\":\"{}\"", description));
        },
        on_done: &|id, ms| emit("migration_done", id, &format!("\"duration_ms\":{}", ms)),
    };
    match apply_pending(REGISTRY, &home, &shared_store_path, data_dir, Some(&progress)) {
        Ok(outcome) => {
            emit_summary(outcome.applied, outcome.skipped);
            0
        }
        Err(msg) => {
            write_error_log(&home, &msg);
            eprintln!("{}", msg);
            1
        }
    }
}

/// Return the store that tracks applied state for a migration of the given scope.
/// Global migrations → shared store; Channel migrations → channel store.
fn tracking_store<'a>(
    scope: super::MigrationScope,
    shared: &'a Store,
    channel: Option<&'a Store>,
) -> Option<&'a Store> {
    match scope {
        super::MigrationScope::Global => Some(shared),
        super::MigrationScope::Channel => channel,
    }
}

fn cmd_list(shared_store: &Store, channel_store: Option<&Store>) -> i32 {
    let shared_applied = shared_store.migrations_list_applied().unwrap_or_default();
    let channel_applied = channel_store
        .and_then(|s| s.migrations_list_applied().ok())
        .unwrap_or_default();
    for m in REGISTRY.iter() {
        let applied_ids = match m.scope() {
            super::MigrationScope::Global => &shared_applied,
            super::MigrationScope::Channel => &channel_applied,
        };
        let status = if applied_ids.contains(&m.id().to_string()) { "applied" } else { "pending" };
        println!("{} [{}] [{}] — {}", m.id(), m.scope().as_str(), status, m.description());
    }
    0
}

// ── Backup ────────────────────────────────────────────────────────────────────

fn backup_stores(home: &Path, shared_store_path: &Path, data_dir: &Path) -> std::io::Result<()> {
    let version = env!("CARGO_PKG_VERSION");
    let ts = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    let backup_dir = home
        .join("shared")
        .join("backups")
        .join(format!("pre-migration-{}-{}", version, ts));
    std::fs::create_dir_all(&backup_dir)?;

    // shared store.db
    if shared_store_path.exists() {
        std::fs::copy(shared_store_path, backup_dir.join("store.db"))?;
    }

    // channel objects.db
    let objects_db = data_dir.join("db").join("objects.db");
    if objects_db.exists() {
        std::fs::copy(&objects_db, backup_dir.join("objects.db"))?;
    }

    prune_old_backups(home);
    Ok(())
}

fn prune_old_backups(home: &Path) {
    let backups_dir = home.join("shared").join("backups");
    let cutoff = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
        .saturating_sub(30 * 24 * 60 * 60); // 30 days

    let Ok(entries) = std::fs::read_dir(&backups_dir) else { return };
    for entry in entries.flatten() {
        let path = entry.path();
        if !path.is_dir() { continue; }
        let name = path.file_name().unwrap_or_default().to_string_lossy().to_string();
        // parse timestamp from "pre-migration-<version>-<ts>"
        if let Some(ts_str) = name.rsplit('-').next() {
            if let Ok(ts) = ts_str.parse::<u64>() {
                if ts < cutoff {
                    let _ = std::fs::remove_dir_all(&path);
                }
            }
        }
    }
}

// ── In-process migration runner (used by srv startup) ────────────────────────

/// Apply all pending migrations in-process before srv opens its stores.
///
/// Called unconditionally at srv startup so the shared store is fully backfilled
/// before `id_store` binds to it. This prevents apparent data loss (empty shared
/// store) on first boot after an upgrade. Fast-path: returns `Ok(0)` immediately
/// when all migrations are already applied.
///
/// Unlike `run_migrate_command` this writes progress via `tracing` rather than
/// stdout JSON and is meant to be called from within the daemon process rather
/// than a subprocess. Returns the number of migrations applied. On error, returns
/// `Err` — callers should fall back to the per-channel store rather than binding
/// `id_store` to an un-backfilled shared store.
///
/// `data_dir` must be the wave data dir (parent of `db/`), not the db dir.
pub fn run_pending_migrations(data_dir: &Path) -> Result<usize, String> {
    let shared_store_path = match resolve_shared_store_path() {
        Some(p) => p,
        None => {
            tracing::info!("run_pending_migrations: shared store path unresolvable — nothing to do");
            return Ok(0);
        }
    };

    let home = match resolve_home() {
        Some(p) => p,
        None => return Err("run_pending_migrations: cannot resolve global shared root".to_string()),
    };

    // Per-migration BEGIN/END stderr lines, inside the AGENTMUXSRV-MIGRATING
    // window `bootstrap/stores.rs` already opens. Without this, srv's launcher-
    // supervised splash shows a bare, silently-running "Backend startup"
    // clock for the whole migration batch — up to 30 minutes
    // (MIGRATION_LOCK_WAIT) — with no indication anything unusual is
    // happening inside it. See
    // docs/reports/REPORT_SPLASH_SCREEN_ARCHITECTURE_RETHINK_2026_09_14.md
    // §2.1. Unlike `run_migrate_command`'s stdout JSON (read by the
    // now-dead subprocess path), this goes to STDERR — the channel every
    // real daemon-boot marker (ESTART, MIGRATING, MIGRATION-FAILED) already
    // uses — via `agentmux_common::srv_stderr`, which both real consumers
    // (the launcher and, in `task dev:standalone`, the CEF host sidecar)
    // already parse the rest of this protocol from. The launcher parses
    // these two lines (`srv_spawner.rs`); a supervisor that doesn't (the
    // CEF host sidecar today) just logs them like any other stderr line.
    let progress = ApplyProgress {
        on_start: &|id, description| {
            eprintln!("{}", agentmux_common::srv_stderr::migration_begin_line(id, description));
        },
        on_done: &|id, ms| {
            eprintln!("{}", agentmux_common::srv_stderr::migration_end_line(id, ms));
        },
    };

    apply_pending(REGISTRY, &home, &shared_store_path, data_dir, Some(&progress)).map(|o| o.applied)
}

// ── Cross-process migration lock (Phase 3, F6) ────────────────────────────────
//
// Two `agentmux-srv` processes against the same data dir (a crash-restart
// race, or a misconfigured channel) could both see a migration as pending
// and both run `up()`: SQLite's own locking serializes individual statements,
// not the runner's check → run → mark-applied sequence (classic TOCTOU).
//
// The lock is a dedicated one-table SQLite file next to the shared store
// with `BEGIN IMMEDIATE` held for the whole batch. That IS a cross-process
// advisory lock — SQLite takes an OS-level file lock — with three properties
// a hand-rolled lockfile does not get for free: it blocks (busy_timeout)
// instead of failing, it is released by the OS when the holder dies (no
// stale-lockfile cleanup), and it is portable to Windows. It is NOT taken on
// the real stores because migrations open their own connections to those
// (m0007 opens the channel store itself); a transaction held there would
// deadlock them.

/// File name of the lock database, created beside the shared store.
pub(super) const MIGRATION_LOCK_FILE: &str = "migrations.lock.db";

/// How long a second runner waits for the first to finish before giving up.
/// Matches the launcher's extended ESTART deadline for a long migration
/// batch (`srv_spawner.rs` / `sidecar.rs`: 30 minutes).
const MIGRATION_LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(30 * 60);

/// Held for the duration of a migration batch. Dropping it releases the lock.
pub(super) struct MigrationLock {
    _conn: rusqlite::Connection,
}

impl MigrationLock {
    /// Acquire the batch lock at `path`, waiting up to `wait` for another
    /// holder. `Err` only if the lock database cannot be opened or the wait
    /// expires — both fatal for a migration run, by design.
    pub(super) fn acquire(path: &Path, wait: std::time::Duration) -> Result<Self, String> {
        let conn = rusqlite::Connection::open(path)
            .map_err(|e| format!("open migration lock {}: {}", path.display(), e))?;
        conn.busy_timeout(wait)
            .map_err(|e| format!("migration lock busy_timeout: {}", e))?;
        // The table exists only so the file is a well-formed database on
        // first use; the lock is the RESERVED lock BEGIN IMMEDIATE takes.
        conn.execute_batch("CREATE TABLE IF NOT EXISTS lock (id INTEGER PRIMARY KEY); BEGIN IMMEDIATE;")
            .map_err(|e| format!("acquire migration lock {} (another runner may still hold it): {}", path.display(), e))?;
        Ok(Self { _conn: conn })
    }
}

/// Open the shared + channel stores the way every writer does (creating the
/// channel dir/db on a fresh install — see the comment inside). Shared by the
/// locked batch and the lock-free `--list`/`--dry-run` previews.
fn open_stores_for_preview(shared_store_path: &Path, data_dir: &Path) -> Result<(Store, Option<Store>), String> {
    let shared_store = Store::open_shared(shared_store_path)
        .map_err(|e| format!("open shared store: {}", e))?;
    let channel_store_path = data_dir.join("db").join("objects.db");
    if let Some(parent) = channel_store_path.parent() {
        std::fs::create_dir_all(parent).map_err(|e| format!("create channel db dir: {}", e))?;
    }
    // Channel store (objects.db) tracks channel-scoped migrations independently
    // per channel — MigrationScope::Channel migrations record here, not in shared.
    // Always open/create it so that seed migrations (e.g. m0008_default_bundle)
    // run on fresh install. Skipping it on a fresh install once left the
    // Channel-scoped migrations (0002/0003/0007) out of the pending list, so
    // they were never marked applied; when mstore then created objects.db,
    // count_pending_migrations at ESTART reported them pending and fired the
    // "Migration failed" UI warning on every fresh first launch.
    // Data-transformation migrations guard themselves with
    // `if !ctx.channel_store_path.exists()` and are no-ops on an empty db.
    let channel_store = Store::open(&channel_store_path)
        .map(Some)
        .map_err(|e| format!("open channel store: {}", e))?;
    Ok((shared_store, channel_store))
}

/// The migrations in `regs` whose tracking store does not record them applied.
fn pending_of<'r>(
    regs: &'r [&'r (dyn super::Migration + Sync)],
    shared: &Store,
    channel: Option<&Store>,
) -> Vec<&'r (dyn super::Migration + Sync)> {
    regs.iter()
        .copied()
        .filter(|m| {
            tracking_store(m.scope(), shared, channel).map_or(false, |s| !s.migration_is_applied(m.id()))
        })
        .collect()
}

/// Per-migration progress hooks. Two real implementations: the CLI's NDJSON
/// stdout lines (`run_migrate_command`), and — since this PR —
/// `run_pending_migrations`'s `AGENTMUXSRV-MIGRATION-BEGIN`/`-END` stderr
/// lines, which the launcher turns into live splash sub-rows. Both also
/// still get the unconditional `tracing` lines `apply_pending` emits
/// regardless of whether a progress hook is passed at all.
pub(super) struct ApplyProgress<'a> {
    pub on_start: &'a dyn Fn(&str, &str),
    pub on_done: &'a dyn Fn(&str, u64),
}

/// What a batch did. `skipped` = already applied at the time of the run.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) struct ApplyOutcome {
    pub applied: usize,
    pub skipped: usize,
}

/// The locked migration batch — the ONE apply path, used by the daemon
/// (`run_pending_migrations`) and by `agentmux-srv migrate`. Generic over the
/// registry so the concurrency contract can be tested with a fake migration.
///
/// Order matters and is the point of Phase 3: the cross-process lock is
/// taken FIRST, before the stores are even opened, so that check → run →
/// mark is one critical section and a second runner blocks on the lock
/// (30 min) rather than tripping `Store::open*`'s 5 s busy timeout against
/// a migrating peer.
pub(super) fn apply_pending(
    regs: &[&(dyn super::Migration + Sync)],
    home: &Path,
    shared_store_path: &Path,
    data_dir: &Path,
    progress: Option<&ApplyProgress<'_>>,
) -> Result<ApplyOutcome, String> {
    if let Some(parent) = shared_store_path.parent() {
        if let Err(e) = std::fs::create_dir_all(parent) {
            return Err(format!("run_pending_migrations: create shared dir: {}", e));
        }
    }

    // Two locks, one beside each store this batch can write, always in the
    // same order (shared, then channel) so two runners can never deadlock
    // on each other. One lock beside the shared store is not enough: two
    // processes can share a `data_dir` (hence the same `objects.db`) while
    // resolving DIFFERENT shared stores — isolated-auth instances with
    // different `AGENTMUX_INSTANCE_DIR`s do exactly that — and would then
    // hold different shared-side locks while racing on the same
    // channel-scoped migrations (codex P1 on #3062).
    let shared_lock_path = shared_store_path
        .parent()
        .map(|p| p.join(MIGRATION_LOCK_FILE))
        .unwrap_or_else(|| PathBuf::from(MIGRATION_LOCK_FILE));
    let _shared_lock = MigrationLock::acquire(&shared_lock_path, MIGRATION_LOCK_WAIT)
        .map_err(|e| format!("run_pending_migrations: {}", e))?;
    let channel_db_dir = data_dir.join("db");
    if let Err(e) = std::fs::create_dir_all(&channel_db_dir) {
        return Err(format!("run_pending_migrations: create channel db dir: {}", e));
    }
    let _channel_lock = MigrationLock::acquire(&channel_db_dir.join(MIGRATION_LOCK_FILE), MIGRATION_LOCK_WAIT)
        .map_err(|e| format!("run_pending_migrations: {}", e))?;

    let (shared_store, channel_store) = open_stores_for_preview(shared_store_path, data_dir)
        .map_err(|e| format!("run_pending_migrations: {}", e))?;
    let channel_store_path = channel_db_dir.join("objects.db");

    let pending = pending_of(regs, &shared_store, channel_store.as_ref());
    if pending.is_empty() {
        return Ok(ApplyOutcome { applied: 0, skipped: regs.len() });
    }
    let skipped = regs.len() - pending.len();

    let ctx = super::MigrationContext {
        home: home.to_path_buf(),
        data_dir: data_dir.to_path_buf(),
        shared_store_path: shared_store_path.to_path_buf(),
        channel_store_path: channel_store_path.clone(),
    };

    if let Err(e) = backup_stores(home, shared_store_path, data_dir) {
        return Err(format!("run_pending_migrations: backup failed: {}", e));
    }

    let mut applied = 0;
    for m in &pending {
        let t = Instant::now();
        tracing::info!(id = m.id(), description = m.description(), "run_pending_migrations: applying");
        if let Some(p) = progress {
            (p.on_start)(m.id(), m.description());
        }
        match m.up(&ctx) {
            Ok(()) => {
                let ms = t.elapsed().as_millis() as u64;
                let scope = m.scope().as_str();
                let tracking = tracking_store(m.scope(), &shared_store, channel_store.as_ref())
                    .ok_or_else(|| format!("run_pending_migrations: no tracking store for {} (channel store missing)", m.id()))?;
                if let Err(e) = tracking.migration_mark_applied(m.id(), scope, ms) {
                    return Err(format!("run_pending_migrations: failed to record {} as applied: {}", m.id(), e));
                }
                tracing::info!(id = m.id(), duration_ms = ms, "run_pending_migrations: applied");
                if let Some(p) = progress {
                    (p.on_done)(m.id(), ms);
                }
                applied += 1;
            }
            Err(e) => {
                return Err(format!("run_pending_migrations: migration {} failed: {}", m.id(), e));
            }
        }
    }

    Ok(ApplyOutcome { applied, skipped })
}

// ── Pending count (used by srv startup before migration and for ESTART) ──────

/// Why a pending count could not be determined. Distinguished from "zero
/// pending" on purpose: a caller deciding whether the schema is current
/// must be able to tell "nothing to do" from "I could not tell", because
/// the safe reaction to each is opposite.
#[derive(Debug)]
pub enum PendingCountError {
    /// No shared store path resolves (CI, or an env with no
    /// `AGENTMUX_SHARED_DIR`). Global-scope migrations cannot be counted.
    SharedStoreUnresolvable,
    /// The shared store exists but would not open.
    SharedStoreUnreadable { path: PathBuf, error: String },
    /// The channel store would not open. Channel-scoped migrations cannot be
    /// counted. Carries the global-scope count that WAS readable, so the
    /// lenient wrapper can keep reporting it — but it is still an error,
    /// because a partial count must not be mistaken for a complete one.
    ChannelStoreUnreadable { path: PathBuf, error: String, global_pending: usize },
}

impl PendingCountError {
    /// The pending count that WAS readable despite this error — the
    /// global-scope half when only the channel store failed, otherwise none
    /// at all.
    ///
    /// For callers that must not block on an unreadable store (the
    /// `AGENTMUXSRV-MIGRATING` hint, which only sizes a timeout, and the
    /// `ESTART` report, which runs after the migration batch). A caller
    /// *deciding* anything on the count must handle the error instead — see
    /// [`try_count_pending_migrations`].
    pub fn readable_pending(&self) -> usize {
        match self {
            Self::ChannelStoreUnreadable { global_pending, .. } => *global_pending,
            Self::SharedStoreUnresolvable | Self::SharedStoreUnreadable { .. } => 0,
        }
    }
}

impl std::fmt::Display for PendingCountError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::SharedStoreUnresolvable => write!(f, "could not resolve the shared store path"),
            Self::SharedStoreUnreadable { path, error } => {
                write!(f, "failed to open the shared store at {}: {}", path.display(), error)
            }
            Self::ChannelStoreUnreadable { path, error, .. } => {
                write!(f, "failed to open the channel store at {}: {}", path.display(), error)
            }
        }
    }
}

/// Number of REGISTRY migrations not yet applied, or why that could not be
/// determined.
///
/// Opens stores read-write (SQLite has no read-only open for WAL mode); this
/// may create `objects.db` if it does not exist. `data_dir` must be the wave
/// data dir (parent of `db/`), not the db dir itself.
///
/// WHY THIS EXISTS ALONGSIDE [`count_pending_migrations`]. That function
/// reports `0` on every error, which is the right default for its caller:
/// the migration run is still ahead of it, so a bad count costs nothing.
/// It is the wrong default for a caller using the count to decide whether
/// the schema is current — there, a transient open failure reading as
/// "nothing pending" means proceeding against a store that may be one or
/// many migrations behind. `SPEC_FAST_STARTUP_UPGRADE_OWNS_MIGRATIONS_AND_UPDATES_2026_09_15.md`
/// §4.1 (invariant S1b) requires that gate to fail closed, which it cannot
/// do without a fallible check to fail closed *on*.
pub fn try_count_pending_migrations(data_dir: &Path) -> Result<usize, PendingCountError> {
    let shared_store_path =
        resolve_shared_store_path().ok_or(PendingCountError::SharedStoreUnresolvable)?;
    let shared_store = Store::open_shared(&shared_store_path).map_err(|e| {
        PendingCountError::SharedStoreUnreadable {
            path: shared_store_path.clone(),
            error: e.to_string(),
        }
    })?;
    let pending_in = |channel: Option<&Store>| {
        REGISTRY
            .iter()
            .filter(|m| {
                let tracking = tracking_store(m.scope(), &shared_store, channel);
                tracking.map_or(false, |s| !s.migration_is_applied(m.id()))
            })
            .count()
    };
    let channel_store_path = data_dir.join("db").join("objects.db");
    let channel_store = Store::open(&channel_store_path).map_err(|e| {
        PendingCountError::ChannelStoreUnreadable {
            path: channel_store_path.clone(),
            error: e.to_string(),
            // Global-scope migrations track in the shared store, which opened
            // fine — that half of the count is still valid.
            global_pending: pending_in(None),
        }
    })?;
    Ok(pending_in(Some(&channel_store)))
}

/// Return the number of REGISTRY migrations that have not yet been applied,
/// reporting 0 on any error so startup is never blocked.
///
/// **A `0` from this function does not mean the schema is current** — it may
/// equally mean the stores could not be read. Callers that need to tell those
/// apart must use [`try_count_pending_migrations`]; this one exists for the
/// reporting path (`AGENTMUXSRV-ESTART`'s `pending_migrations:` field) where
/// the migration run has already happened and an unreadable store is not a
/// decision point. `data_dir` must be the wave data dir (parent of `db/`),
/// not the db dir itself.
pub fn count_pending_migrations(data_dir: &Path) -> usize {
    match try_count_pending_migrations(data_dir) {
        Ok(n) => n,
        Err(e) => {
            // Unchanged from before this function grew a fallible sibling:
            // an unreadable channel store still reports the global-scope
            // count rather than dropping to 0.
            let fallback = e.readable_pending();
            tracing::warn!("count_pending_migrations: {} — reporting {}", e, fallback);
            fallback
        }
    }
}

// ── Error log ─────────────────────────────────────────────────────────────────

fn write_error_log(home: &Path, msg: &str) {
    let log_dir = home.join("logs");
    let _ = std::fs::create_dir_all(&log_dir);
    let path = log_dir.join("migration-error.log");
    let content = format!("{}\n", msg);
    let _ = std::fs::write(path, content);
}

// ── Verify (doctor pass) ──────────────────────────────────────────────────────
//
// SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03 Phase 1, second bullet: "applied"
// in `db_migrations` means `up()` returned Ok — nothing re-checks that it
// wrote anything. `migrate --verify` asks each APPLIED migration for its own
// post-condition (`Migration::verify`) and reports per migration. Unapplied
// migrations are skipped: there is nothing to verify yet, and `--list`
// already shows them.
//
// Read-only by contract (codex P2 on #3058): nothing on this path calls
// `Store::open*`. Those run schema setup, WAL configuration and legacy-table
// adoption, and create the file when missing. This path opens plain SQLite
// connections with SQLITE_OPEN_READ_ONLY, so a typoed data dir yields "no
// store here", never a freshly created empty database that then "verifies".

/// One line of the verify report.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerifyRecord {
    pub id: &'static str,
    pub scope: &'static str,
    pub outcome: super::VerifyOutcome,
}

/// Which migration ids each tracking store records as applied, read through
/// a FALLIBLE path. `Err` means the store exists but its tracking state
/// could not be read (not a database, malformed table, locked): every
/// migration of that scope then reports `VerifyOutcome::Error` instead of
/// being silently skipped as "unapplied" and letting the pass exit 0 —
/// codex P1 on #3058, which `Store::migration_is_applied`'s
/// `unwrap_or(false)` would have caused. A store file that does not exist,
/// or one predating the `db_migrations` table, has nothing recorded and is
/// `Ok(empty)`.
pub struct AppliedIds {
    pub global: Result<HashSet<String>, String>,
    pub channel: Result<HashSet<String>, String>,
}

impl AppliedIds {
    fn for_scope(&self, scope: MigrationScope) -> &Result<HashSet<String>, String> {
        match scope {
            MigrationScope::Global => &self.global,
            MigrationScope::Channel => &self.channel,
        }
    }
}

/// Open an existing SQLite file read-only. `Ok(None)` when the file is
/// absent — never creates one (the whole point of this path).
pub(super) fn open_readonly(path: &Path) -> Result<Option<rusqlite::Connection>, String> {
    if !path.exists() {
        return Ok(None);
    }
    rusqlite::Connection::open_with_flags(path, rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY)
        .map(Some)
        .map_err(|e| format!("open {} read-only: {}", path.display(), e))
}

/// `SELECT id FROM db_migrations` through a read-only connection. A missing
/// file or a missing table is `Ok(empty)` — nothing was ever recorded there.
/// Any other failure is `Err`, which `verify_applied` turns into an
/// `Error` outcome for every migration of that scope.
pub(super) fn read_applied_ids(path: &Path) -> Result<HashSet<String>, String> {
    let Some(conn) = open_readonly(path)? else {
        return Ok(HashSet::new());
    };
    let mut stmt = match conn.prepare("SELECT id FROM db_migrations") {
        Ok(s) => s,
        Err(rusqlite::Error::SqliteFailure(_, Some(msg))) if msg.contains("no such table") => {
            return Ok(HashSet::new());
        }
        Err(e) => return Err(format!("read db_migrations in {}: {}", path.display(), e)),
    };
    stmt.query_map([], |row| row.get::<_, String>(0))
        .and_then(|rows| rows.collect::<Result<HashSet<String>, _>>())
        .map_err(|e| format!("read db_migrations in {}: {}", path.display(), e))
}

/// Run `verify()` for every migration in `regs` whose tracking store records
/// it as applied. Generic over the registry slice so the harness can be
/// tested with fakes; production passes `REGISTRY`.
pub fn verify_applied(
    regs: &[&(dyn super::Migration + Sync)],
    ctx: &MigrationContext,
    applied: &AppliedIds,
) -> Vec<VerifyRecord> {
    let mut out = Vec::new();
    for m in regs {
        let outcome = match applied.for_scope(m.scope()) {
            Err(e) => super::VerifyOutcome::Error(format!("cannot read applied state: {}", e)),
            Ok(ids) if ids.contains(m.id()) => m.verify(ctx),
            Ok(_) => continue,
        };
        out.push(VerifyRecord { id: m.id(), scope: m.scope().as_str(), outcome });
    }
    out
}

/// Exit code for a verify report: 0 unless any record is a mismatch or an
/// error. `NotVerifiable` is not a failure — it is the honest default.
pub fn verify_exit_code(records: &[VerifyRecord]) -> i32 {
    if records.iter().any(|r| r.outcome.is_failure()) {
        VERIFY_FAILED_EXIT_CODE
    } else {
        0
    }
}

/// One `migration_verify` NDJSON line. `detail` goes through serde_json so
/// control characters (a multi-line SQLite error, say) cannot break the
/// line-delimited stream a consumer tails (reagent P2 on #3058).
fn verify_event_line(r: &VerifyRecord) -> String {
    let detail = serde_json::to_string(r.outcome.detail()).unwrap_or_else(|_| "\"\"".to_string());
    format!(
        "{{\"event\":\"migration_verify\",\"id\":\"{}\",\"scope\":\"{}\",\"status\":\"{}\",\"detail\":{}}}",
        r.id,
        r.scope,
        r.outcome.label(),
        detail
    )
}

// ── Doctor report ─────────────────────────────────────────────────────────────
//
// SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03.md Phase 1c: the `--verify` pass
// and the applied/pending listing, as ONE machine-readable structure that both
// `agentmux-srv migrate --verify` (printed below) and the live instance's
// `GET /api/v1/muxspect/migrations` (JSON) are built from. One producer, so
// the CLI you run against a stopped install and the `muxspect migrations`
// you run from inside a running one can never disagree about what
// "verified" means. Read-only end to end — same `open_readonly` path as
// `--verify`, never opens a `Store`, never creates a file.

/// The complete migration doctor report for one data dir.
#[derive(Debug, Clone, serde::Serialize)]
pub struct DoctorReport {
    pub data_dir: String,
    pub shared_store: String,
    pub channel_store: String,
    /// One entry per tracking store whose `db_migrations` could not be read
    /// (not a database, malformed, locked). Every migration of that scope
    /// reads `unknown` / `error` below; this names the cause once.
    pub tracking_errors: Vec<String>,
    pub migrations: Vec<DoctorRow>,
    pub summary: DoctorSummary,
    /// What `migrate --verify` exits with for this state: `0`, or
    /// [`VERIFY_FAILED_EXIT_CODE`] when any check is a mismatch or error.
    pub exit_code: i32,
}

/// One registry migration, in registry order.
#[derive(Debug, Clone, serde::Serialize)]
pub struct DoctorRow {
    pub id: &'static str,
    pub scope: &'static str,
    pub description: &'static str,
    /// `applied` | `pending` | `unknown` — the last when the scope's
    /// tracking store could not be read (see `tracking_errors`).
    pub state: &'static str,
    /// [`super::VerifyOutcome::label`] for an applied migration: `ok` |
    /// `MISMATCH` | `error` | `n/a`. Absent for a pending one — there is no
    /// post-condition to check until it has run.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub verify: Option<&'static str>,
    /// The verdict's evidence or reason (counts, the SQLite error, …).
    #[serde(skip_serializing_if = "String::is_empty")]
    pub detail: String,
}

#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize)]
pub struct DoctorSummary {
    pub total: usize,
    pub applied: usize,
    pub pending: usize,
    pub unknown: usize,
    /// Applied migrations whose `verify()` ran — equals `applied` unless a
    /// tracking store was unreadable.
    pub checked: usize,
    pub ok: usize,
    pub mismatch: usize,
    pub error: usize,
    pub not_verifiable: usize,
}

/// Paths and tracking state the doctor works from. Reads both tracking
/// stores through the fallible read-only path; never opens a `Store`.
fn doctor_inputs(data_dir: &Path, home: &Path, shared_store_path: &Path) -> (MigrationContext, AppliedIds) {
    let channel_store_path = data_dir.join("db").join("objects.db");
    let ctx = MigrationContext {
        home: home.to_path_buf(),
        data_dir: data_dir.to_path_buf(),
        shared_store_path: shared_store_path.to_path_buf(),
        channel_store_path: channel_store_path.clone(),
    };
    let applied = AppliedIds {
        global: read_applied_ids(shared_store_path),
        channel: read_applied_ids(&channel_store_path),
    };
    (ctx, applied)
}

/// Assemble the report from an already-run verify pass, so `cmd_verify` can
/// print the per-record NDJSON lines from the same `records` the report
/// summarises rather than verifying twice.
fn build_doctor_report(
    regs: &[&(dyn super::Migration + Sync)],
    ctx: &MigrationContext,
    applied: &AppliedIds,
    records: &[VerifyRecord],
) -> DoctorReport {
    let mut summary = DoctorSummary { total: regs.len(), ..Default::default() };
    let mut migrations = Vec::with_capacity(regs.len());
    for m in regs {
        let record = records.iter().find(|r| r.id == m.id());
        let state = match applied.for_scope(m.scope()) {
            Err(_) => "unknown",
            Ok(ids) if ids.contains(m.id()) => "applied",
            Ok(_) => "pending",
        };
        match state {
            "applied" => summary.applied += 1,
            "pending" => summary.pending += 1,
            _ => summary.unknown += 1,
        }
        let (verify, detail) = match record {
            Some(r) => {
                summary.checked += 1;
                match r.outcome {
                    super::VerifyOutcome::Ok(_) => summary.ok += 1,
                    super::VerifyOutcome::Mismatch(_) => summary.mismatch += 1,
                    super::VerifyOutcome::Error(_) => summary.error += 1,
                    super::VerifyOutcome::NotVerifiable => summary.not_verifiable += 1,
                }
                (Some(r.outcome.label()), r.outcome.detail().to_string())
            }
            None => (None, String::new()),
        };
        migrations.push(DoctorRow {
            id: m.id(),
            scope: m.scope().as_str(),
            description: m.description(),
            state,
            verify,
            detail,
        });
    }
    let tracking_errors = [&applied.global, &applied.channel]
        .into_iter()
        .filter_map(|r| r.as_ref().err().cloned())
        .collect();
    DoctorReport {
        data_dir: ctx.data_dir.display().to_string(),
        shared_store: ctx.shared_store_path.display().to_string(),
        channel_store: ctx.channel_store_path.display().to_string(),
        tracking_errors,
        migrations,
        summary,
        exit_code: verify_exit_code(records),
    }
}

/// The doctor report for `data_dir`, against an explicit registry (tests
/// pass fakes; production passes `REGISTRY`). Read-only.
pub fn doctor_report(
    regs: &[&(dyn super::Migration + Sync)],
    data_dir: &Path,
    home: &Path,
    shared_store_path: &Path,
) -> DoctorReport {
    let (ctx, applied) = doctor_inputs(data_dir, home, shared_store_path);
    let records = verify_applied(regs, &ctx, &applied);
    build_doctor_report(regs, &ctx, &applied, &records)
}

/// [`doctor_report`] for the running instance's own registry and resolved
/// shared store / home — what `GET /api/v1/muxspect/migrations` serves.
/// `Err` when the shared root can't be resolved at all (no
/// `AGENTMUX_SHARED_DIR`, CI), the same condition under which
/// `run_migrate_command` treats migrations as a no-op.
pub fn doctor_report_for_instance(data_dir: &Path) -> Result<DoctorReport, String> {
    let shared_store_path = resolve_shared_store_path()
        .ok_or_else(|| "shared store path is unresolvable (AGENTMUX_SHARED_DIR / home not set)".to_string())?;
    let home = resolve_home().ok_or_else(|| "AgentMux home is unresolvable".to_string())?;
    Ok(doctor_report(REGISTRY, data_dir, &home, &shared_store_path))
}

fn cmd_verify(data_dir: &Path, home: &Path, shared_store_path: &Path) -> i32 {
    let (ctx, applied) = doctor_inputs(data_dir, home, shared_store_path);
    let records = verify_applied(REGISTRY, &ctx, &applied);
    for r in &records {
        let detail = r.outcome.detail();
        if detail.is_empty() {
            println!("verify: {} [{}] [{}]", r.id, r.scope, r.outcome.label());
        } else {
            println!("verify: {} [{}] [{}] — {}", r.id, r.scope, r.outcome.label(), detail);
        }
        println!("{}", verify_event_line(r));
    }
    let report = build_doctor_report(REGISTRY, &ctx, &applied, &records);
    let s = &report.summary;
    println!(
        "verify: {} applied checked — {} ok, {} mismatch, {} error, {} not verifiable (data dir {})",
        s.checked,
        s.ok,
        s.mismatch,
        s.error,
        s.not_verifiable,
        data_dir.display()
    );
    println!(
        "{{\"event\":\"verify_complete\",\"checked\":{},\"ok\":{},\"mismatch\":{},\"error\":{},\"not_verifiable\":{}}}",
        s.checked, s.ok, s.mismatch, s.error, s.not_verifiable
    );
    report.exit_code
}

/// `SELECT COUNT(*) FROM <table>` on a (read-only) connection, treating a
/// missing table as 0 rows — a store from before the table existed has
/// nothing in it, which is exactly the answer a post-condition wants. Any
/// other SQLite error is surfaced. `table` must be a literal identifier from
/// the calling migration, never user input (it is interpolated, not bound).
pub(super) fn table_count(conn: &rusqlite::Connection, table: &str) -> Result<i64, String> {
    match conn.query_row(&format!("SELECT COUNT(*) FROM {}", table), [], |row| row.get::<_, i64>(0)) {
        Ok(n) => Ok(n),
        Err(rusqlite::Error::SqliteFailure(_, Some(msg))) if msg.contains("no such table") => Ok(0),
        Err(e) => Err(format!("count {}: {}", table, e)),
    }
}

#[cfg(test)]
mod lock_tests {
    use super::*;
    use crate::migrations::{Migration, MigrationError};
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[test]
    fn lock_is_exclusive_across_connections_and_released_on_drop() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join(MIGRATION_LOCK_FILE);
        let first = MigrationLock::acquire(&path, std::time::Duration::from_millis(100)).expect("first holder");
        // A second acquirer waits out its budget and fails while the first holds.
        let err = MigrationLock::acquire(&path, std::time::Duration::from_millis(200))
            .err()
            .expect("second acquire must fail while the first is held");
        assert!(err.contains("another runner may still hold it"), "{}", err);
        drop(first);
        MigrationLock::acquire(&path, std::time::Duration::from_millis(100)).expect("free after drop");
    }

    /// A Global migration that records how many times `up()` ran and is slow
    /// enough that two runners started together both pass the "is it
    /// pending?" check before either can mark it applied — unless the
    /// critical section holds.
    static UP_RUNS: AtomicUsize = AtomicUsize::new(0);
    struct SlowOnce;
    impl Migration for SlowOnce {
        fn id(&self) -> &'static str { "9900_slow_once" }
        fn scope(&self) -> MigrationScope { MigrationScope::Global }
        fn description(&self) -> &'static str { "counts its own runs" }
        fn up(&self, _ctx: &MigrationContext) -> Result<(), MigrationError> {
            UP_RUNS.fetch_add(1, Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(300));
            Ok(())
        }
    }
    static SLOW_ONCE: SlowOnce = SlowOnce;

    /// Channel-scoped twin of `SlowOnce`, for the different-shared-store case.
    static CHANNEL_UP_RUNS: AtomicUsize = AtomicUsize::new(0);
    struct SlowOnceChannel;
    impl Migration for SlowOnceChannel {
        fn id(&self) -> &'static str { "9901_slow_once_channel" }
        fn scope(&self) -> MigrationScope { MigrationScope::Channel }
        fn description(&self) -> &'static str { "counts its own runs (channel scope)" }
        fn up(&self, _ctx: &MigrationContext) -> Result<(), MigrationError> {
            CHANNEL_UP_RUNS.fetch_add(1, Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(300));
            Ok(())
        }
    }
    static SLOW_ONCE_CHANNEL: SlowOnceChannel = SlowOnceChannel;

    #[test]
    fn runners_with_different_shared_stores_but_one_data_dir_still_serialize() {
        // codex P1 on #3062: isolated-auth instances resolve different shared
        // stores yet can share a data dir, i.e. the same objects.db and the
        // same channel-scoped migrations. The shared-side lock alone would let
        // both through; the channel-side lock must catch it.
        let dir = tempfile::tempdir().unwrap();
        let home = dir.path().to_path_buf();
        let data_dir = home.join("data");
        let shared_a = home.join("iso-a").join("shared").join("store.db");
        let shared_b = home.join("iso-b").join("shared").join("store.db");
        CHANNEL_UP_RUNS.store(0, Ordering::SeqCst);

        let results: Vec<Result<ApplyOutcome, String>> = std::thread::scope(|s| {
            let handles: Vec<_> = [shared_a.clone(), shared_b.clone()]
                .into_iter()
                .map(|shared| {
                    let (home, data_dir) = (home.clone(), data_dir.clone());
                    s.spawn(move || {
                        let regs: Vec<&(dyn Migration + Sync)> = vec![&SLOW_ONCE_CHANNEL];
                        apply_pending(&regs, &home, &shared, &data_dir, None)
                    })
                })
                .collect();
            handles.into_iter().map(|h| h.join().expect("runner thread")).collect()
        });

        for r in &results {
            assert!(r.is_ok(), "both runners must succeed: {:?}", results);
        }
        let applied: usize = results.iter().map(|r| r.as_ref().unwrap().applied).sum();
        assert_eq!(applied, 1, "exactly one runner applied it: {:?}", results);
        assert_eq!(CHANNEL_UP_RUNS.load(Ordering::SeqCst), 1, "channel-scoped up() ran once despite two shared stores");
    }

    #[test]
    fn previews_do_not_wait_on_the_migration_lock() {
        // reagent P1 on #3062: `--list` / `--dry-run` must stay usable while
        // another process is mid-migration. Hold the lock, then take the same
        // read path the previews use and require it to come back at once.
        let dir = tempfile::tempdir().unwrap();
        let shared = dir.path().join("shared").join("store.db");
        std::fs::create_dir_all(shared.parent().unwrap()).unwrap();
        let data_dir = dir.path().join("data");
        let _held = MigrationLock::acquire(&shared.parent().unwrap().join(MIGRATION_LOCK_FILE), std::time::Duration::from_millis(100))
            .expect("hold the lock");

        let before = UP_RUNS.load(Ordering::SeqCst);
        let started = std::time::Instant::now();
        let (shared_store, channel_store) = open_stores_for_preview(&shared, &data_dir).expect("preview opens stores");
        let regs: Vec<&(dyn Migration + Sync)> = vec![&SLOW_ONCE];
        let pending = pending_of(&regs, &shared_store, channel_store.as_ref());
        assert_eq!(pending.len(), 1, "nothing applied yet, so the fake is pending");
        assert!(started.elapsed() < std::time::Duration::from_secs(5), "preview must not block on the lock");
        assert_eq!(UP_RUNS.load(Ordering::SeqCst), before, "preview never runs up()");
    }

    #[test]
    fn two_concurrent_runners_on_one_data_dir_apply_a_migration_exactly_once() {
        // Phase 3 acceptance (SPEC_MIGRATION_SYSTEM_HARDENING_2026_08_03):
        // "two concurrent run_pending_migrations calls against the same
        // fresh data dir; assert the migration's up() body executes exactly
        // once." Both runners must also succeed: the loser sees the winner's
        // db_migrations row once it gets the lock and applies nothing.
        let dir = tempfile::tempdir().unwrap();
        let home = dir.path().to_path_buf();
        let shared = home.join("shared").join("store.db");
        let data_dir = home.join("data");
        UP_RUNS.store(0, Ordering::SeqCst);

        let results: Vec<Result<usize, String>> = std::thread::scope(|s| {
            let handles: Vec<_> = (0..2)
                .map(|_| {
                    let (home, shared, data_dir) = (home.clone(), shared.clone(), data_dir.clone());
                    s.spawn(move || {
                        let regs: Vec<&(dyn Migration + Sync)> = vec![&SLOW_ONCE];
                        apply_pending(&regs, &home, &shared, &data_dir, None).map(|o| o.applied)
                    })
                })
                .collect();
            handles.into_iter().map(|h| h.join().expect("runner thread")).collect()
        });

        for r in &results {
            assert!(r.is_ok(), "both runners must succeed: {:?}", results);
        }
        let applied: usize = results.iter().map(|r| *r.as_ref().unwrap()).sum();
        assert_eq!(applied, 1, "exactly one runner applied it: {:?}", results);
        assert_eq!(UP_RUNS.load(Ordering::SeqCst), 1, "up() body must execute exactly once");

        // And the record is durable: a third run has nothing to do.
        let regs: Vec<&(dyn Migration + Sync)> = vec![&SLOW_ONCE];
        assert_eq!(apply_pending(&regs, &home, &shared, &data_dir, None).map(|o| o.applied).unwrap(), 0);
        assert_eq!(UP_RUNS.load(Ordering::SeqCst), 1);
    }
}

#[cfg(test)]
mod verify_tests {
    use super::*;
    use crate::migrations::{Migration, MigrationError, VerifyOutcome};

    /// A migration whose `verify()` returns a canned outcome. `up()` is never
    /// called by the verify harness.
    struct Fake {
        id: &'static str,
        scope: MigrationScope,
        outcome: VerifyOutcome,
    }
    impl Migration for Fake {
        fn id(&self) -> &'static str { self.id }
        fn scope(&self) -> MigrationScope { self.scope }
        fn description(&self) -> &'static str { "fake" }
        fn up(&self, _ctx: &MigrationContext) -> Result<(), MigrationError> { Ok(()) }
        fn verify(&self, _ctx: &MigrationContext) -> VerifyOutcome { self.outcome.clone() }
    }

    /// Default `verify()` — a migration that never opted in.
    struct Silent;
    impl Migration for Silent {
        fn id(&self) -> &'static str { "9998_silent" }
        fn scope(&self) -> MigrationScope { MigrationScope::Global }
        fn description(&self) -> &'static str { "no verify impl" }
        fn up(&self, _ctx: &MigrationContext) -> Result<(), MigrationError> { Ok(()) }
    }

    fn ctx_in(dir: &tempfile::TempDir) -> MigrationContext {
        MigrationContext {
            home: dir.path().to_path_buf(),
            data_dir: dir.path().join("data"),
            shared_store_path: dir.path().join("shared").join("store.db"),
            channel_store_path: dir.path().join("data").join("db").join("objects.db"),
        }
    }

    fn ids(list: &[&str]) -> Result<HashSet<String>, String> {
        Ok(list.iter().map(|s| s.to_string()).collect())
    }

    #[test]
    fn only_applied_migrations_are_verified_and_unapplied_are_skipped() {
        let dir = tempfile::tempdir().unwrap();
        let applied = Fake { id: "9001_applied", scope: MigrationScope::Global, outcome: VerifyOutcome::Ok("fine".into()) };
        let pending = Fake { id: "9002_pending", scope: MigrationScope::Global, outcome: VerifyOutcome::Mismatch("would fail".into()) };
        let state = AppliedIds { global: ids(&["9001_applied"]), channel: ids(&[]) };

        let regs: Vec<&(dyn Migration + Sync)> = vec![&applied, &pending];
        let records = verify_applied(&regs, &ctx_in(&dir), &state);
        assert_eq!(records.len(), 1, "the pending one has nothing to verify yet");
        assert_eq!(records[0].id, "9001_applied");
        assert_eq!(records[0].outcome, VerifyOutcome::Ok("fine".into()));
        assert_eq!(verify_exit_code(&records), 0);
    }

    #[test]
    fn channel_scoped_migrations_consult_the_channel_state_not_the_global_one() {
        let dir = tempfile::tempdir().unwrap();
        let m = Fake { id: "9003_channel", scope: MigrationScope::Channel, outcome: VerifyOutcome::Ok("ok".into()) };
        let regs: Vec<&(dyn Migration + Sync)> = vec![&m];
        // Recorded in the GLOBAL set only: wrong tracking store for a Channel
        // migration, so it must read as unapplied.
        let wrong = AppliedIds { global: ids(&["9003_channel"]), channel: ids(&[]) };
        assert!(verify_applied(&regs, &ctx_in(&dir), &wrong).is_empty());
        let right = AppliedIds { global: ids(&[]), channel: ids(&["9003_channel"]) };
        assert_eq!(verify_applied(&regs, &ctx_in(&dir), &right).len(), 1);
    }

    #[test]
    fn unreadable_tracking_state_is_an_error_for_every_migration_of_that_scope() {
        // codex P1 on #3058: a broken db_migrations must not read as "nothing
        // applied" and let the pass exit 0.
        let dir = tempfile::tempdir().unwrap();
        let g = Fake { id: "9004_global", scope: MigrationScope::Global, outcome: VerifyOutcome::Ok("never reached".into()) };
        let c = Fake { id: "9005_channel", scope: MigrationScope::Channel, outcome: VerifyOutcome::Ok("fine".into()) };
        let regs: Vec<&(dyn Migration + Sync)> = vec![&g, &c];
        let state = AppliedIds { global: Err("file is not a database".into()), channel: ids(&["9005_channel"]) };
        let records = verify_applied(&regs, &ctx_in(&dir), &state);
        assert_eq!(records.len(), 2);
        assert!(matches!(&records[0].outcome, VerifyOutcome::Error(e) if e.contains("not a database")));
        assert_eq!(records[1].outcome, VerifyOutcome::Ok("fine".into()));
        assert_eq!(verify_exit_code(&records), VERIFY_FAILED_EXIT_CODE);
    }

    #[test]
    fn a_migration_without_a_verify_impl_reports_not_verifiable_and_passes() {
        let dir = tempfile::tempdir().unwrap();
        let regs: Vec<&(dyn Migration + Sync)> = vec![&Silent];
        let state = AppliedIds { global: ids(&["9998_silent"]), channel: ids(&[]) };
        let records = verify_applied(&regs, &ctx_in(&dir), &state);
        assert_eq!(records[0].outcome, VerifyOutcome::NotVerifiable);
        assert_eq!(verify_exit_code(&records), 0, "not verifiable is honest, not a failure");
    }

    #[test]
    fn mismatch_or_error_fails_the_pass_with_a_code_distinct_from_run_failure_and_clap() {
        let mk = |o: VerifyOutcome| VerifyRecord { id: "x", scope: "global", outcome: o };
        assert_eq!(verify_exit_code(&[mk(VerifyOutcome::Ok("a".into())), mk(VerifyOutcome::NotVerifiable)]), 0);
        assert_eq!(verify_exit_code(&[mk(VerifyOutcome::Ok("a".into())), mk(VerifyOutcome::Mismatch("gone".into()))]), VERIFY_FAILED_EXIT_CODE);
        assert_eq!(verify_exit_code(&[mk(VerifyOutcome::Error("locked".into()))]), VERIFY_FAILED_EXIT_CODE);
        assert_ne!(VERIFY_FAILED_EXIT_CODE, 1, "1 = a migration run failed");
        assert_ne!(VERIFY_FAILED_EXIT_CODE, 2, "2 = clap usage error (bad flag)");
    }

    #[test]
    fn read_applied_ids_reads_real_tracking_state_and_never_creates_a_store() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("shared").join("store.db");
        // Absent file: nothing recorded, and still absent afterwards.
        assert_eq!(read_applied_ids(&path).unwrap(), HashSet::new());
        assert!(!path.exists(), "verify must never create a database");

        std::fs::create_dir_all(path.parent().unwrap()).unwrap();
        let store = Store::open_shared(&path).expect("open shared");
        store.migration_mark_applied("9010_a", "global", 1).unwrap();
        store.migration_mark_applied("9011_b", "global", 1).unwrap();
        drop(store);
        let got = read_applied_ids(&path).unwrap();
        assert!(got.contains("9010_a") && got.contains("9011_b"));
        assert_eq!(got.len(), 2);
    }

    #[test]
    fn read_applied_ids_errors_on_a_file_that_is_not_a_database() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("objects.db");
        std::fs::write(&path, b"definitely not sqlite\n").unwrap();
        let err = read_applied_ids(&path).unwrap_err();
        assert!(err.contains("objects.db"), "names the store: {}", err);
    }

    #[test]
    fn table_count_treats_a_missing_table_as_zero_rows() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("objects.db");
        drop(Store::open(&path).expect("create channel store"));
        let conn = open_readonly(&path).unwrap().expect("exists");
        assert_eq!(table_count(&conn, "db_table_that_never_existed").unwrap(), 0);
        assert!(table_count(&conn, "db_migrations").is_ok());
    }

    // ---- Doctor report (Phase 1c — the producer `muxspect migrations` serves) ----

    #[test]
    fn doctor_report_on_a_fresh_data_dir_lists_everything_pending_and_creates_nothing() {
        let dir = tempfile::tempdir().unwrap();
        let data_dir = dir.path().join("data");
        let shared = dir.path().join("shared").join("store.db");
        let report = doctor_report(REGISTRY, &data_dir, dir.path(), &shared);

        assert_eq!(report.migrations.len(), REGISTRY.len());
        assert!(report.migrations.iter().all(|r| r.state == "pending" && r.verify.is_none()));
        assert_eq!(report.summary.total, REGISTRY.len());
        assert_eq!(report.summary.pending, REGISTRY.len());
        assert_eq!(report.summary.checked, 0);
        assert!(report.tracking_errors.is_empty());
        assert_eq!(report.exit_code, 0);
        // Read-only, like --verify: a doctor must never conjure a store.
        assert!(!shared.exists() && !data_dir.exists(), "doctor created files");
        // Serialises — the handler returns this as JSON.
        let json = serde_json::to_value(&report).unwrap();
        assert_eq!(json["summary"]["pending"], REGISTRY.len());
        assert!(json["migrations"][0].get("verify").is_none(), "pending rows omit `verify`");
    }

    #[test]
    fn doctor_report_reports_applied_rows_with_their_verdict_and_counts_them() {
        let dir = tempfile::tempdir().unwrap();
        let data_dir = dir.path().join("data");
        let channel = data_dir.join("db").join("objects.db");
        std::fs::create_dir_all(channel.parent().unwrap()).unwrap();
        let shared = dir.path().join("shared").join("store.db");
        {
            // A fresh channel store with 0007 recorded applied: no legacy
            // rows, empty target — consistent, so the verdict is `ok`.
            let store = Store::open(&channel).expect("create channel store");
            store.migration_mark_applied("0007_agents_consolidate", "channel", 1).unwrap();
        }
        let report = doctor_report(REGISTRY, &data_dir, dir.path(), &shared);
        let row = report.migrations.iter().find(|r| r.id == "0007_agents_consolidate").unwrap();
        assert_eq!(row.state, "applied");
        assert_eq!(row.verify, Some("ok"));
        assert!(row.detail.contains("no legacy rows"), "{}", row.detail);
        assert_eq!(report.summary.applied, 1);
        assert_eq!(report.summary.checked, 1);
        assert_eq!(report.summary.ok, 1);
        assert_eq!(report.summary.pending, REGISTRY.len() - 1);
        assert_eq!(report.exit_code, 0);
    }

    #[test]
    fn doctor_report_marks_a_scope_unknown_when_its_tracking_store_is_unreadable_and_exits_3() {
        let dir = tempfile::tempdir().unwrap();
        let data_dir = dir.path().join("data");
        let shared = dir.path().join("shared").join("store.db");
        std::fs::create_dir_all(shared.parent().unwrap()).unwrap();
        std::fs::write(&shared, b"definitely not sqlite\n").unwrap();
        let report = doctor_report(REGISTRY, &data_dir, dir.path(), &shared);

        let global_rows: Vec<_> = report.migrations.iter().filter(|r| r.scope == "global").collect();
        assert!(!global_rows.is_empty());
        assert!(global_rows.iter().all(|r| r.state == "unknown" && r.verify == Some("error")));
        assert!(report.migrations.iter().filter(|r| r.scope == "channel").all(|r| r.state == "pending"));
        assert_eq!(report.summary.unknown, global_rows.len());
        assert_eq!(report.summary.error, global_rows.len());
        // The cause is named once, not once per migration.
        assert_eq!(report.tracking_errors.len(), 1, "{:?}", report.tracking_errors);
        assert!(report.tracking_errors[0].contains("store.db"));
        assert_eq!(report.exit_code, VERIFY_FAILED_EXIT_CODE);
    }

    #[test]
    fn verify_event_line_is_valid_json_even_with_control_characters_in_detail() {
        let r = VerifyRecord {
            id: "9020_x",
            scope: "channel",
            outcome: VerifyOutcome::Error("line one\nline \"two\"\t\\end".into()),
        };
        let line = verify_event_line(&r);
        let parsed: serde_json::Value = serde_json::from_str(&line).expect("one JSON object per line");
        assert_eq!(parsed["event"], "migration_verify");
        assert_eq!(parsed["status"], "error");
        assert_eq!(parsed["detail"], "line one\nline \"two\"\t\\end");
        assert!(!line.contains('\n'), "must stay on one line");
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    // Process-global env access — shared with registry::paths and
    // migrations::m0011_shared_store_backfill's tests, which mutate the SAME
    // AGENTMUX_ISOLATED_AUTH/AGENTMUX_INSTANCE_DIR vars. A module-local lock
    // only serializes tests within this file; Cargo runs a crate's tests in
    // one multi-threaded process, so a local-only lock still let this
    // module's tests race against those two (reagent/codex on PR #2318).
    use crate::test_support::ISOLATED_AUTH_ENV_LOCK as ENV_LOCK;

    fn clear() {
        std::env::remove_var("AGENTMUX_HOME_OVERRIDE");
        std::env::remove_var("AGENTMUX_SHARED_DIR");
        std::env::remove_var("AGENTMUX_ISOLATED_AUTH");
        std::env::remove_var("AGENTMUX_INSTANCE_DIR");
        std::env::remove_var("AGENTMUX_CHANNEL");
    }

    // ── Pending count: lenient vs. fail-closed ───────────────────────────

    #[test]
    fn an_unreadable_shared_store_is_an_error_not_a_zero() {
        // The trap this pair of functions exists to separate: a caller
        // deciding whether the schema is current cannot act on `0`, because
        // `0` is also what an unreadable store produces.
        // SPEC_FAST_STARTUP_UPGRADE_OWNS_MIGRATIONS_AND_UPDATES_2026_09_15 S1b.
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        clear();
        // A home with no `shared/` directory: SQLite will not create the
        // parent, so the shared store cannot be opened.
        let home = tempfile::tempdir().expect("tempdir");
        std::env::set_var("AGENTMUX_HOME_OVERRIDE", home.path());
        let data_dir = home.path().join("data");

        let strict = try_count_pending_migrations(&data_dir);
        let lenient = count_pending_migrations(&data_dir);
        clear();

        match strict {
            Err(PendingCountError::SharedStoreUnreadable { .. }) => {}
            Err(other) => panic!("expected SharedStoreUnreadable, got {other}"),
            Ok(n) => panic!("an unreadable shared store must not read as {n} pending"),
        }
        assert_eq!(lenient, 0, "the lenient wrapper still reports 0, as it always did");
    }

    #[test]
    fn the_lenient_wrapper_keeps_reporting_the_global_half_it_could_read() {
        // An unreadable CHANNEL store is the partial case: global-scope
        // migrations track in the shared store, which opened fine. The
        // fallible check still refuses to answer (the total is incomplete);
        // the lenient one reports the half it could read, exactly as it did
        // before it grew a fallible sibling.
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        clear();
        let home = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir_all(home.path().join("shared")).expect("shared dir");
        std::env::set_var("AGENTMUX_HOME_OVERRIDE", home.path());
        // No `db/` under it, so the channel store cannot open.
        let data_dir = home.path().join("no-such-data-dir");

        let strict = try_count_pending_migrations(&data_dir);
        let lenient = count_pending_migrations(&data_dir);
        let global_pending = match &strict {
            Err(PendingCountError::ChannelStoreUnreadable { global_pending, .. }) => *global_pending,
            other => {
                let msg = format!("precondition: expected ChannelStoreUnreadable, got {other:?}");
                clear();
                panic!("{msg}");
            }
        };
        clear();

        assert_eq!(
            lenient, global_pending,
            "the lenient wrapper must keep reporting the readable global-scope count"
        );
    }

    #[test]
    fn readable_pending_keeps_the_global_half_but_claims_nothing_otherwise() {
        // What the non-blocking callers fall back to. A shared-store failure
        // has no readable half at all — reporting anything but 0 there would
        // be inventing a number.
        assert_eq!(
            PendingCountError::ChannelStoreUnreadable {
                path: std::path::PathBuf::from("/tmp/objects.db"),
                error: "locked".into(),
                global_pending: 3,
            }
            .readable_pending(),
            3
        );
        assert_eq!(PendingCountError::SharedStoreUnresolvable.readable_pending(), 0);
        assert_eq!(
            PendingCountError::SharedStoreUnreadable {
                path: std::path::PathBuf::from("/tmp/store.db"),
                error: "locked".into(),
            }
            .readable_pending(),
            0
        );
    }

    #[test]
    fn a_pending_count_error_says_which_store_and_why() {
        // The blocking upgrade state shows this text to a human, so it has to
        // name the store and carry the underlying reason rather than being a
        // bare "could not count".
        let e = PendingCountError::SharedStoreUnreadable {
            path: std::path::PathBuf::from("/tmp/store.db"),
            error: "disk I/O error".to_string(),
        };
        let msg = e.to_string();
        assert!(msg.contains("store.db"), "{msg}");
        assert!(msg.contains("disk I/O error"), "{msg}");
    }

    /// The regression test that would have caught the `ctx.home` /
    /// `shared_store_path` coupling bug before it shipped: `resolve_home()`
    /// must return the SAME value whether or not isolated-auth is active,
    /// even though `resolve_shared_store_path()` itself deliberately
    /// returns a DIFFERENT (channel-scoped) path under isolation. Every
    /// other Global migration, plus backups and the error log, anchor to
    /// `home` and must never silently move just because one channel opted
    /// into isolated auth.
    #[test]
    fn home_is_invariant_to_isolated_auth() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        clear();
        std::env::set_var("AGENTMUX_HOME_OVERRIDE", "/tmp/test-home");

        let home_default = resolve_home().unwrap();
        assert_eq!(home_default, PathBuf::from("/tmp/test-home"));

        let shared_store_path_default = resolve_shared_store_path().unwrap();

        std::env::set_var("AGENTMUX_ISOLATED_AUTH", "1");
        std::env::set_var("AGENTMUX_INSTANCE_DIR", "/tmp/test-home/dev/some-branch");
        let home_isolated = resolve_home().unwrap();
        let shared_store_path_isolated = resolve_shared_store_path().unwrap();

        assert_eq!(
            home_default, home_isolated,
            "resolve_home() must be invariant to AGENTMUX_ISOLATED_AUTH/AGENTMUX_INSTANCE_DIR"
        );
        // Meanwhile resolve_shared_store_path() DOES vary — confirms the two
        // functions have genuinely diverged as designed, not that isolation
        // silently does nothing.
        assert_ne!(
            shared_store_path_default, shared_store_path_isolated,
            "resolve_shared_store_path() must actually change under isolation"
        );

        clear();
    }

    /// Same invariant as `home_is_invariant_to_isolated_auth`, with a
    /// non-"stable" channel and an instance dir present. Auth is shared by
    /// default on every channel now (SPEC_SHARED_AUTH_ACROSS_CHANNELS_2026_10_03.md),
    /// so the channel alone changes nothing: neither the home nor the store path.
    #[test]
    fn home_and_store_are_invariant_to_the_channel_when_auth_is_shared() {
        let _g = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
        clear();
        std::env::set_var("AGENTMUX_HOME_OVERRIDE", "/tmp/test-home");

        let home_default = resolve_home().unwrap();
        let shared_store_path_default = resolve_shared_store_path().unwrap();

        // No AGENTMUX_ISOLATED_AUTH set at all — only a non-"stable" channel.
        std::env::set_var("AGENTMUX_CHANNEL", "dev-some-branch");
        std::env::set_var("AGENTMUX_INSTANCE_DIR", "/tmp/test-home/dev/some-branch");
        let home_isolated = resolve_home().unwrap();
        let shared_store_path_isolated = resolve_shared_store_path().unwrap();

        assert_eq!(
            home_default, home_isolated,
            "resolve_home() must not depend on the channel"
        );
        assert_eq!(
            shared_store_path_default, shared_store_path_isolated,
            "a non-stable channel alone must not move the shared store"
        );

        // An explicit opt-in still isolates, and still leaves the home alone.
        std::env::set_var("AGENTMUX_ISOLATED_AUTH", "1");
        assert_eq!(resolve_home().unwrap(), home_default);
        assert_ne!(resolve_shared_store_path().unwrap(), shared_store_path_default);

        clear();
    }
}