crtx 0.1.1

//! Schema migration CLI surfaces.
//!
//! The schema v2 command is deliberately pre-cutover while
//! `cortex_core::SCHEMA_VERSION` remains `1`. Dry-run exposes fixture evidence
//! without mutation; the backup-manifest path may run staged expand/backfill,
//! append the schema boundary event, and verify the mixed chain, but still
//! refuses schema v2 cutover.

use std::path::{Path, PathBuf};

use chrono::{DateTime, Utc};
use clap::{Args, Subcommand};
use cortex_core::{
    compose_policy_outcomes, Event, KeyLifecycleState, PolicyContribution, PolicyDecision,
    PolicyOutcome, SchemaMigrationV1ToV2Payload, TrustTier, SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND,
};
use cortex_ledger::{
    audit::verify_schema_migration_v1_to_v2_boundary, canonical_payload_bytes, verify_chain,
    JsonlLog, SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
    SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
    SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
};
use cortex_store::migrate_v2::{
    fixture_verification_result_hash as store_fixture_verification_result_hash,
    staged_execution_plan, V2DryRunPlan, V2MigrationStageStatus, SCHEMA_V2_EXPAND_SQL,
};
use cortex_store::repo::{
    authority::{key_state_policy_decision_test_allow, principal_state_policy_decision_test_allow},
    AuthorityRepo, KeyTimelineRecord, PrincipalTimelineRecord,
};
use cortex_store::Pool;
use ed25519_dalek::{Signature, Signer, Verifier, VerifyingKey};
use serde::Deserialize;

use crate::cmd::open_default_store;
use crate::cmd::temporal::{revalidate_operator_temporal_authority, revalidation_failed_invariant};
use crate::exit::Exit;
use crate::output::{self, Envelope};
use crate::paths::DataLayout;

/// Atomic-cutover prerequisites tracked across the schema v2 transition.
///
/// All six gates landed together in the atomic Phase B commit (ADR 0018).
/// The slice is retained as documentation: a non-empty list here would indicate
/// a regression in cutover authority. Tests assert it is empty.
const MISSING_ATOMIC_CUTOVER_PREREQUISITES: &[&str] = &[];

/// Schema version a pre-v2 backup manifest must declare.
///
/// The blessed pre-v2 backup is taken before the schema v2 cutover commit and
/// is always at schema v1. This constant remains `1` even after
/// `cortex_core::SCHEMA_VERSION` advances; rolling back to a future v3 will
/// require a separate pre-cutover-v3 backup kind, not a reinterpretation of the
/// v1 manifest contract.
const PRE_V2_BACKUP_SCHEMA_VERSION: u16 = 1;

const RESERVED_BACKUP_MANIFEST_CUTOVER_APPROVAL_FIELDS: &[&str] = &[
    "cutover_approved",
    "operator_approval",
    "operator_approval_attestation",
    "schema_cutover_approved",
    "schema_version_cutover_approved",
    "default_v2_cutover_ready",
    "unattended_migrate",
];

/// Migration subcommands.
#[derive(Debug, Subcommand)]
pub enum MigrateSub {
    /// Preview the schema v2 staging plan for the current v1 store.
    V2(MigrateV2Args),
    /// Produce an Ed25519-signed operator attestation envelope authorising the
    /// next v1 -> v2 boundary append for the current store (ADR 0010 §1-§2,
    /// Gate 5 punch list #17).
    ///
    /// The boundary triple is read from the current v1 store using the same
    /// preflight that `migrate v2 --dry-run` runs; operators do not have to
    /// transcribe hashes by hand. The signing key is loaded from a 32-byte
    /// raw Ed25519 seed file (`--signing-seed`). The resulting envelope is
    /// passed to `cortex migrate v2 --operator-attestation <PATH>` to unlock
    /// the cutover gate.
    SignOperatorAttestation(SignOperatorAttestationArgs),
    /// Seed the temporal authority store with an operator key + principal
    /// for the CI v1-to-v2 cutover drill **only**.
    ///
    /// The cutover surface revalidates the supplied `--operator-key-id`
    /// against the durable `authority_key_timeline` and refuses with
    /// `migrate.v2.operator_temporal_authority.revalidation_failed
    /// (key_unknown, principal_unknown)` if the operator is not
    /// registered. Production operators register their key through a
    /// separate, fully attested admission ceremony; this surface exists
    /// solely so the CI drill can boot an ephemeral per-fixture operator
    /// without the production ceremony.
    ///
    /// This command is **hard-gated**: it refuses unless the environment
    /// variable `CORTEX_DRILL_ALLOW_SEED_OPERATOR=1` is set on the
    /// invoking process. Operators MUST NOT export that variable in any
    /// production shell — the gate exists so the surface cannot be
    /// reached accidentally outside the drill harness.
    SeedDrillOperator(SeedDrillOperatorArgs),
}

/// Arguments for `cortex migrate seed-drill-operator`.
///
/// All fields except `--operator-key-id` have defaults so the drill harness
/// can invoke this with one argument and remain readable.
#[derive(Debug, Args)]
pub struct SeedDrillOperatorArgs {
    /// Operator key id to register in the key timeline. Matches the
    /// `operator_key_id` field the cutover envelope was signed with.
    #[arg(long, value_name = "STRING")]
    pub operator_key_id: String,
    /// Principal id this key belongs to. Defaults to
    /// `<operator_key_id>-principal`.
    #[arg(long, value_name = "STRING")]
    pub principal_id: Option<String>,
}

/// Arguments for `cortex migrate sign-operator-attestation`.
#[derive(Debug, Args)]
pub struct SignOperatorAttestationArgs {
    /// Output path for the JSON envelope.
    #[arg(long, value_name = "PATH")]
    pub output: PathBuf,
    /// Path to a file containing the 32-byte raw Ed25519 signing seed
    /// (binary, exactly 32 bytes — operators may produce one with
    /// `head -c 32 /dev/urandom > seed` and protect it on disk).
    #[arg(long, value_name = "PATH")]
    pub signing_seed: PathBuf,
    /// Operator-facing key id. Free-form (matches `cortex-identity`); the
    /// envelope binds this id into the canonical signing input.
    #[arg(long, value_name = "STRING", default_value = "cortex-operator")]
    pub operator_key_id: String,
}

/// Arguments for `cortex migrate v2`.
#[derive(Debug, Args)]
pub struct MigrateV2Args {
    /// Inspect the current v1 store and print the v2 migration plan without mutation.
    #[arg(long)]
    pub dry_run: bool,
    /// Backup manifest proving a blessed pre-v2 backup exists before any future mutation.
    #[arg(long, value_name = "PATH")]
    pub backup_manifest: Option<PathBuf>,
    /// Non-interactive cutover approval path. Unlocks only when a valid
    /// `--operator-attestation` is also supplied (ADR 0010 operator
    /// attestation gate at the migration authority root).
    #[arg(long)]
    pub unattended_migrate: bool,
    /// Path to an Ed25519-signed operator attestation envelope authorising
    /// the v1 -> v2 boundary append (ADR 0010 §1-§2). The envelope binds the
    /// operator's signing key to the `previous_v1_head_hash`,
    /// `migration_script_digest`, and `fixture_verification_result_hash` so a
    /// captured envelope cannot be re-used against a different boundary.
    #[arg(long, value_name = "PATH")]
    pub operator_attestation: Option<PathBuf>,
    /// Re-mirror the v1->v2 boundary row from JSONL into SQLite. Used when
    /// the cutover transaction rolled back after the JSONL fsync but before
    /// the SQLite mirror committed (B1 partial-mutation recovery,
    /// RED_TEAM_FINDINGS phase B). Idempotent: succeeds if the boundary is
    /// already mirrored. Mutually exclusive with `--dry-run`,
    /// `--backup-manifest`, and `--unattended-migrate`.
    #[arg(long)]
    pub resume_mirror: bool,
}

#[derive(Debug, Deserialize)]
struct BackupManifest {
    kind: String,
    schema_version: u16,
    sqlite_store: String,
    jsonl_mirror: String,
    tool_version: String,
    backup_timestamp: String,
    #[serde(default)]
    table_row_counts: Option<BackupManifestTableRowCounts>,
}

/// Pre-v2 backup-manifest table-count payload required by the post-migrate
/// count-mismatch refusal helper.
///
/// Cortex backup manifests written by this binary always include this field.
/// Older or hand-written manifests that omit it deserialize as `None` and are
/// rejected by [`validate_backup_manifest`] so cutover cannot proceed without
/// the pre-migrate count baseline.
#[derive(Debug, Clone, Copy, Deserialize)]
struct BackupManifestTableRowCounts {
    events: u64,
    traces: u64,
    episodes: u64,
    memories: u64,
}

/// Run a migration subcommand.
pub fn run(sub: MigrateSub) -> Exit {
    match sub {
        MigrateSub::V2(args) => {
            let dry_run = args.dry_run;
            let exit = run_v2(args);
            if output::json_enabled() {
                emit_migrate_v2_envelope(dry_run, exit)
            } else {
                exit
            }
        }
        MigrateSub::SignOperatorAttestation(args) => run_sign_operator_attestation(args),
        MigrateSub::SeedDrillOperator(args) => run_seed_drill_operator(args),
    }
}

/// Environment variable that must be set to `1` to unlock the
/// `cortex migrate seed-drill-operator` surface. The gate exists so the
/// surface cannot be reached accidentally outside the CI v1-to-v2 cutover
/// drill harness.
const DRILL_SEED_OPERATOR_ENV_GATE: &str = "CORTEX_DRILL_ALLOW_SEED_OPERATOR";

/// Stable invariant emitted when the drill-only seed surface refuses
/// because the gate environment variable is not set.
const SEED_DRILL_OPERATOR_GATE_INVARIANT: &str = "migrate.seed_drill_operator.gate_not_set";

/// Implementation for `cortex migrate seed-drill-operator`. This is a
/// drill-only surface — see the doc-comment on
/// [`MigrateSub::SeedDrillOperator`].
fn run_seed_drill_operator(args: SeedDrillOperatorArgs) -> Exit {
    match std::env::var(DRILL_SEED_OPERATOR_ENV_GATE) {
        Ok(value) if value == "1" => {}
        _ => {
            eprintln!(
                "cortex migrate seed-drill-operator: {SEED_DRILL_OPERATOR_GATE_INVARIANT}: \
                 refused; this surface is hard-gated and requires \
                 {DRILL_SEED_OPERATOR_ENV_GATE}=1 (CI drill harness only). \
                 no state was changed."
            );
            return Exit::PreconditionUnmet;
        }
    }

    if args.operator_key_id.trim().is_empty() {
        eprintln!(
            "cortex migrate seed-drill-operator: refused; --operator-key-id must not be empty. \
             no state was changed."
        );
        return Exit::Usage;
    }

    let pool = match open_default_store("migrate seed-drill-operator") {
        Ok(pool) => pool,
        Err(exit) => {
            eprintln!("cortex migrate seed-drill-operator: refused; no state was changed.");
            return exit;
        }
    };

    let principal_id = args
        .principal_id
        .clone()
        .unwrap_or_else(|| format!("{}-principal", args.operator_key_id));
    // Backdate `effective_at` to the Unix epoch so the drill's seeded operator
    // is unambiguously active at any later `signed_at` the cutover envelope
    // carries. The revalidation logic checks the key state at the envelope's
    // signed_at AND at "now"; both timestamps are necessarily after the epoch,
    // so a single seed call covers both checks regardless of whether the seed
    // runs before or after `sign-operator-attestation`.
    let effective_at = DateTime::<Utc>::from_timestamp(0, 0).expect("epoch is a valid UTC time");
    let repo = AuthorityRepo::new(&pool);

    let principal_record = PrincipalTimelineRecord {
        principal_id: principal_id.clone(),
        trust_tier: TrustTier::Operator,
        effective_at,
        trust_review_due_at: None,
        removed_at: None,
        audit_ref: None,
    };
    if let Err(err) = repo.append_principal_state(
        &principal_record,
        &principal_state_policy_decision_test_allow(),
    ) {
        eprintln!(
            "cortex migrate seed-drill-operator: failed to register principal `{principal_id}`: {err}. \
             no state was changed (or only the principal row was inserted)."
        );
        return Exit::Internal;
    }

    let key_record = KeyTimelineRecord {
        key_id: args.operator_key_id.clone(),
        principal_id: principal_id.clone(),
        state: KeyLifecycleState::Active,
        effective_at,
        reason: Some("v1-to-v2-drill: seed-drill-operator surface".into()),
        audit_ref: None,
    };
    if let Err(err) = repo.append_key_state(&key_record, &key_state_policy_decision_test_allow()) {
        eprintln!(
            "cortex migrate seed-drill-operator: failed to register key `{}` for principal `{principal_id}`: {err}.",
            args.operator_key_id,
        );
        return Exit::Internal;
    }

    println!("cortex migrate seed-drill-operator: ok");
    println!("operator_key_id={}", args.operator_key_id);
    println!("principal_id={principal_id}");
    println!("trust_tier=operator");
    println!("key_state=active");
    println!("effective_at={}", effective_at.to_rfc3339());

    Exit::Ok
}

fn emit_migrate_v2_envelope(dry_run: bool, exit: Exit) -> Exit {
    let transcript = MIGRATE_TRANSCRIPT.with(|cell| std::mem::take(&mut *cell.borrow_mut()));
    let payload = serde_json::json!({
        "dry_run": dry_run,
        "schema_version_target": cortex_core::SCHEMA_VERSION,
        "transcript": transcript,
    });
    let envelope = Envelope::new("cortex.migrate.v2", exit, payload);
    output::emit(&envelope, exit)
}

thread_local! {
    static MIGRATE_TRANSCRIPT: std::cell::RefCell<Vec<serde_json::Value>> =
        const { std::cell::RefCell::new(Vec::new()) };
}

fn record_migrate_event(label: &str, value: serde_json::Value) {
    if !output::json_enabled() {
        return;
    }
    MIGRATE_TRANSCRIPT.with(|cell| {
        cell.borrow_mut().push(serde_json::json!({
            "label": label,
            "value": value,
        }));
    });
}

fn run_sign_operator_attestation(args: SignOperatorAttestationArgs) -> Exit {
    let layout = match DataLayout::resolve(None, None) {
        Ok(layout) => layout,
        Err(exit) => return exit,
    };
    let pool = match open_default_store("migrate sign-operator-attestation") {
        Ok(pool) => pool,
        Err(exit) => {
            eprintln!("cortex migrate sign-operator-attestation: refused; no state was changed.");
            return exit;
        }
    };
    let plan = match cortex_store::migrate_v2::dry_run_plan(&pool) {
        Ok(plan) => plan,
        Err(err) => {
            eprintln!(
                "cortex migrate sign-operator-attestation: failed to build dry-run plan: {err}. no state was changed."
            );
            return Exit::PreconditionUnmet;
        }
    };
    if !plan.is_ready() {
        for failure in &plan.failures {
            eprintln!(
                "cortex migrate sign-operator-attestation: {}: {}",
                failure.invariant(),
                failure.detail()
            );
        }
        eprintln!(
            "cortex migrate sign-operator-attestation: dry-run refused; no envelope was produced."
        );
        eprintln!(
            "cortex migrate sign-operator-attestation: hint: run `cortex migrate v2 \
             --backup-manifest <path>` to upgrade, or `cortex doctor --repair` to apply pending \
             migrations"
        );
        return Exit::SchemaMismatch;
    }
    let boundary = match boundary_preflight(&layout, &plan) {
        Ok(boundary) => boundary,
        Err(exit) => return exit,
    };

    let seed = match std::fs::read(&args.signing_seed) {
        Ok(seed) => seed,
        Err(err) => {
            eprintln!(
                "cortex migrate sign-operator-attestation: signing seed `{}` could not be read: {err}.",
                args.signing_seed.display()
            );
            return Exit::PreconditionUnmet;
        }
    };
    if seed.len() != 32 {
        eprintln!(
            "cortex migrate sign-operator-attestation: signing seed `{}` must be exactly 32 bytes; got {}.",
            args.signing_seed.display(),
            seed.len()
        );
        return Exit::PreconditionUnmet;
    }
    let mut seed_array = [0u8; 32];
    seed_array.copy_from_slice(&seed);
    let signing_key = ed25519_dalek::SigningKey::from_bytes(&seed_array);
    let verifying_key = signing_key.verifying_key();

    let signed_at = Utc::now();
    let signed_at_rfc3339 = signed_at.to_rfc3339();
    let envelope_unsigned = OperatorAttestationEnvelopeForSigning {
        schema_version: OPERATOR_ATTESTATION_ENVELOPE_SCHEMA_VERSION,
        purpose: OPERATOR_ATTESTATION_ENVELOPE_PURPOSE,
        operator_key_id: &args.operator_key_id,
        signed_at_rfc3339: &signed_at_rfc3339,
        previous_v1_head_hash: &boundary.previous_v1_head_hash,
        migration_script_digest: &boundary.migration_script_digest,
        fixture_verification_result_hash: &boundary.fixture_verification_result_hash,
    };
    let signing_input = envelope_unsigned.signing_input();
    let signature = signing_key.sign(&signing_input);

    let envelope_json = serde_json::json!({
        "schema_version": OPERATOR_ATTESTATION_ENVELOPE_SCHEMA_VERSION,
        "purpose": OPERATOR_ATTESTATION_ENVELOPE_PURPOSE,
        "operator_verifying_key_hex": lowercase_hex(verifying_key.as_bytes()),
        "operator_key_id": args.operator_key_id,
        "signed_at": signed_at_rfc3339,
        "boundary": {
            "previous_v1_head_hash": boundary.previous_v1_head_hash,
            "migration_script_digest": boundary.migration_script_digest,
            "fixture_verification_result_hash": boundary.fixture_verification_result_hash,
        },
        "signature_hex": lowercase_hex(&signature.to_bytes()),
    });

    let serialized = match serde_json::to_string_pretty(&envelope_json) {
        Ok(s) => s,
        Err(err) => {
            eprintln!(
                "cortex migrate sign-operator-attestation: failed to serialise envelope: {err}."
            );
            return Exit::Internal;
        }
    };
    if let Err(err) = std::fs::write(&args.output, serialized) {
        eprintln!(
            "cortex migrate sign-operator-attestation: failed to write `{}`: {err}.",
            args.output.display()
        );
        return Exit::Internal;
    }

    println!("cortex migrate sign-operator-attestation: ok");
    println!("operator_attestation_path={}", args.output.display());
    println!("operator_key_id={}", args.operator_key_id);
    println!("signed_at={signed_at_rfc3339}");
    println!(
        "boundary_previous_v1_head_hash={}",
        boundary.previous_v1_head_hash
    );
    println!(
        "migration_script_digest={}",
        boundary.migration_script_digest
    );
    println!(
        "fixture_verification_result_hash={}",
        boundary.fixture_verification_result_hash
    );

    Exit::Ok
}

/// Borrowed view of the envelope's signed fields used by both the signer
/// (`run_sign_operator_attestation`) and the verifier
/// (`operator_attestation_signing_input`). Keeping the signed-input builder
/// in one place makes the canonical bytes definitionally identical across
/// signer and verifier.
struct OperatorAttestationEnvelopeForSigning<'a> {
    schema_version: u16,
    purpose: &'a str,
    operator_key_id: &'a str,
    signed_at_rfc3339: &'a str,
    previous_v1_head_hash: &'a str,
    migration_script_digest: &'a str,
    fixture_verification_result_hash: &'a str,
}

impl OperatorAttestationEnvelopeForSigning<'_> {
    fn signing_input(&self) -> Vec<u8> {
        const DOMAIN_TAG_OPERATOR_ATTESTATION_MIGRATION: u8 = 0x20;
        let mut out = Vec::new();
        out.push(DOMAIN_TAG_OPERATOR_ATTESTATION_MIGRATION);
        out.extend_from_slice(&self.schema_version.to_be_bytes());
        push_lp(&mut out, self.purpose.as_bytes());
        push_lp(&mut out, self.operator_key_id.as_bytes());
        push_lp(&mut out, self.signed_at_rfc3339.as_bytes());
        push_lp(&mut out, self.previous_v1_head_hash.as_bytes());
        push_lp(&mut out, self.migration_script_digest.as_bytes());
        push_lp(&mut out, self.fixture_verification_result_hash.as_bytes());
        out
    }
}

fn lowercase_hex(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        out.push_str(&format!("{b:02x}"));
    }
    out
}

fn run_v2(args: MigrateV2Args) -> Exit {
    if args.resume_mirror {
        // B1 follow-up (RED_TEAM_FINDINGS phase B): the `--resume-mirror`
        // flag exists solely to re-mirror a durable JSONL boundary row into
        // SQLite when the cutover transaction rolled back after the JSONL
        // fsync. It is NOT a cutover surface — no boundary is ever appended,
        // no backup-manifest baseline is consumed, no operator attestation
        // is collected (the boundary already exists and was authored by a
        // prior, separately-attested cutover attempt). The flag is mutually
        // exclusive with the cutover-surface flags so an operator cannot
        // accidentally promote a recovery into a fresh cutover.
        if args.dry_run
            || args.backup_manifest.is_some()
            || args.unattended_migrate
            || args.operator_attestation.is_some()
        {
            eprintln!(
                "cortex migrate v2: --resume-mirror is a partial-mutation recovery surface and cannot be combined with --dry-run, --backup-manifest, --unattended-migrate, or --operator-attestation. no state was changed."
            );
            return Exit::Usage;
        }
        return run_v2_resume_mirror();
    }

    if args.unattended_migrate && args.operator_attestation.is_none() {
        // Per ADR 0026 §4 (hard wall) and ADR 0010 §1, the migration
        // authority root requires a fresh operator attestation. Until the
        // operator supplies one via `--operator-attestation`, the
        // `--unattended-migrate` path stays refused with a clearer reason.
        eprintln!(
            "cortex migrate v2: --unattended-migrate requires --operator-attestation <PATH> with a valid Ed25519-signed operator attestation envelope (ADR 0010 §1-§2). Operator attestation cannot be substituted by --unattended-migrate alone. no state was changed."
        );
        emit_cutover_readiness_stderr();
        return Exit::PreconditionUnmet;
    }

    if !args.dry_run {
        let Some(backup_manifest) = args.backup_manifest.as_deref() else {
            eprintln!(
                "cortex migrate v2: cutover requires --backup-manifest <path> pointing at a blessed pre-v2 backup. rerun with --dry-run to preview, or supply a backup manifest emitted by `cortex backup --output`. no state was changed."
            );
            emit_cutover_readiness_stderr();
            return Exit::Usage;
        };
        let backup_manifest = match validate_backup_manifest(backup_manifest) {
            Ok(manifest) => manifest,
            Err(exit) => return exit,
        };

        let layout = match DataLayout::resolve(None, None) {
            Ok(layout) => layout,
            Err(exit) => return exit,
        };
        let mut pool = match open_default_store("migrate v2") {
            Ok(pool) => pool,
            Err(exit) => {
                eprintln!("cortex migrate v2: full migration refused; no state was changed.");
                return exit;
            }
        };

        // R1 (RED_TEAM_FINDINGS phase B): refuse a `cortex_pre_v2_backup`
        // manifest pointed at a live store that already has v2 rows outside
        // the boundary. `cortex backup` decides manifest kind purely by
        // JSONL boundary presence; a fresh-v2 store with no boundary and
        // all-v2 rows still gets kind=cortex_pre_v2_backup. Letting that
        // manifest drive the cutover would append a boundary on top of v2
        // rows the boundary falsely claims were v1.
        if let Err(exit) =
            enforce_backup_manifest_consistent_with_live_store(&pool, &backup_manifest.path)
        {
            return exit;
        }
        let stage_plan = match staged_execution_plan(&pool) {
            Ok(plan) => plan,
            Err(err) => {
                eprintln!(
                    "cortex migrate v2: failed to build staged execution plan: {err}. no state was changed."
                );
                return Exit::PreconditionUnmet;
            }
        };
        if !stage_plan.dry_run.is_ready() {
            for failure in &stage_plan.dry_run.failures {
                eprintln!(
                    "cortex migrate v2: {}: {}",
                    failure.invariant(),
                    failure.detail()
                );
            }
            eprintln!("cortex migrate v2: full migration refused; no state was changed.");
            eprintln!(
                "cortex migrate v2: hint: run `cortex doctor --repair` to apply pending \
                 migrations, or retry `cortex migrate v2 --backup-manifest <path>` once the \
                 store preconditions are satisfied"
            );
            return Exit::SchemaMismatch;
        }
        let boundary = match boundary_preflight(&layout, &stage_plan.dry_run) {
            Ok(boundary) => boundary,
            Err(exit) => return exit,
        };

        // Compose the ADR 0026 policy decision for the v1 -> v2 boundary
        // append (Gate 5 punch list #17). The three contributors compose
        // ADR 0019 (authority class), ADR 0010 (attestation), and ADR 0023
        // (current-use temporal authority). If `--operator-attestation` was
        // supplied, verify the envelope against the just-computed boundary
        // payload before signalling `Allow` for the attestation contributor.
        let migration_policy = match build_migration_policy_decision(
            &pool,
            args.operator_attestation.as_deref(),
            &boundary,
        ) {
            Ok(decision) => decision,
            Err(exit) => return exit,
        };

        for stage in &stage_plan.stages {
            eprintln!(
                "cortex migrate v2: stage={} status={} mutates_store={} enables_cutover={}",
                stage.name,
                stage_status_label(stage.status),
                stage.mutates_store,
                stage.enables_cutover
            );
        }

        // B1 (RED_TEAM_FINDINGS phase B): wrap the SQLite-side cutover body in a
        // single explicit transaction. Before this fix the cutover ran a fan-out
        // of unbatched `pool.execute(...)` calls (expand/backfill, mirror,
        // re-backfill, readiness) with no enclosing tx — a failure between
        // any of those steps left the SQLite store partially mutated and
        // operators had to restore from blessed backup (ADR 0033 §3) just to
        // retry. With this guard, every SQLite mutation participates in the
        // same atomic unit: any in-tx failure rolls back to the pre-cutover
        // SQLite snapshot and leaves only the JSONL boundary row durable.
        //
        // The JSONL boundary append itself stays OUTSIDE the SQLite tx by
        // doctrine: JSONL is the canonical durable record (ADR 0033 §1 +
        // BUILD_SPEC §7) and the SQLite mirror is a peer for queryability.
        // Recovery from "JSONL-ahead-of-SQLite" partial commit is wired
        // through `cortex migrate v2 --resume-mirror` (see
        // `run_v2_resume_mirror` below).
        // Capture the migrate timestamp ONCE so the manifest body, its
        // BLAKE3 digest, the boundary row's `source_attestation_json` anchor,
        // and the on-disk manifest are all built from the same value. Drift
        // here would break the byte-identity guarantee between the in-chain
        // digest and the on-disk manifest (Decision #6).
        let migrate_timestamp = Utc::now();
        let cutover_outcome = run_v2_cutover_tx(
            &mut pool,
            &layout,
            &boundary,
            &migration_policy,
            &backup_manifest.table_row_counts,
            &PostMigrateManifestInputs {
                pre_v2_backup_manifest_path: backup_manifest.path.clone(),
                migrate_timestamp,
            },
        );
        let CutoverTxOutcome {
            expand_report,
            boundary_head,
            post_migrate_manifest_blake3,
            post_migrate_manifest_body,
        } = match cutover_outcome {
            Ok(outcome) => outcome,
            Err(exit) => return exit,
        };

        eprintln!(
            "cortex migrate v2: expand_backfill_added_columns={}",
            expand_report.added_columns.len()
        );
        for table in &expand_report.created_tables {
            eprintln!("cortex migrate v2: expand_backfill_created_table={table}");
        }
        eprintln!(
            "cortex migrate v2: legacy_event_attestations_backfilled={}",
            expand_report.legacy_event_attestations_backfilled
        );
        eprintln!(
            "cortex migrate v2: boundary_previous_v1_head_hash={}",
            boundary.previous_v1_head_hash
        );
        eprintln!("cortex migrate v2: boundary_event_hash={boundary_head}");
        eprintln!("cortex migrate v2: boundary_event_kind={SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND}");
        eprintln!("cortex migrate v2: boundary_audit=ok");
        eprintln!("cortex migrate v2: post_migrate_mixed_chain_audit=ok");
        eprintln!("cortex migrate v2: post_cutover_audit_dispatch=available");
        eprintln!(
            "cortex migrate v2: post_migrate_row_count_refusal=ok pre_events={} pre_traces={} pre_episodes={} pre_memories={}",
            backup_manifest.table_row_counts.events,
            backup_manifest.table_row_counts.traces,
            backup_manifest.table_row_counts.episodes,
            backup_manifest.table_row_counts.memories,
        );

        eprintln!("cortex migrate v2: cutover_authority=ok");
        eprintln!("cortex migrate v2: cutover_approved=true");
        eprintln!("cortex migrate v2: cutover_guard=committed");
        emit_cutover_readiness_stderr();
        eprintln!(
            "cortex migrate v2: backup_manifest_table_row_counts events={} traces={} episodes={} memories={}",
            backup_manifest.table_row_counts.events,
            backup_manifest.table_row_counts.traces,
            backup_manifest.table_row_counts.episodes,
            backup_manifest.table_row_counts.memories,
        );

        // Emit post-migrate manifest (ADR 0018 §"Acceptance gate", schema v2
        // atomic cutover commit). Operators carry this artifact alongside the
        // pre-v2 backup so rollback can compare row counts and boundary state.
        //
        // Decision #6 / RED_TEAM_FINDINGS D2: the manifest is now tamper-
        // evident. `post_migrate_manifest_body` was canonicalised and digested
        // inside `run_v2_cutover_tx`; the digest is anchored on the boundary
        // row's `source_attestation_json` column atomically with the cutover.
        // Here we just insert `manifest_blake3` into the body and serialize.
        // The verifier (`cortex restore verify-post-migrate-manifest`) splits
        // `manifest_blake3` back out, recomputes the digest against the
        // remaining fields, and refuses on mismatch.
        let post_manifest_path = backup_manifest
            .path
            .with_file_name("POST_V2_MIGRATE_MANIFEST");
        let mut post_manifest = post_migrate_manifest_body;
        if let Some(obj) = post_manifest.as_object_mut() {
            obj.insert(
                POST_MIGRATE_MANIFEST_DIGEST_FIELD.to_string(),
                serde_json::Value::String(post_migrate_manifest_blake3.clone()),
            );
        } else {
            // Defensive: `build_post_migrate_manifest_body` always returns an
            // object; if a refactor breaks that, fail closed rather than write
            // a non-object manifest that the verifier could not parse.
            eprintln!(
                "cortex migrate v2: internal error — post-migrate manifest body is not a JSON object; refusing to write tamper-evident manifest."
            );
            return Exit::Internal;
        }
        if let Err(err) = std::fs::write(
            &post_manifest_path,
            serde_json::to_string_pretty(&post_manifest).unwrap_or_default(),
        ) {
            eprintln!(
                "cortex migrate v2: failed to write post-migrate manifest `{}`: {err}.",
                post_manifest_path.display()
            );
            return Exit::Internal;
        }
        eprintln!(
            "cortex migrate v2: post_migrate_manifest={}",
            post_manifest_path.display()
        );
        eprintln!("cortex migrate v2: post_migrate_manifest_blake3={post_migrate_manifest_blake3}");
        eprintln!("cortex migrate v2: {POST_MIGRATE_MANIFEST_CHAIN_ANCHORED_INVARIANT}=ok");
        eprintln!(
            "cortex migrate v2: schema cutover complete. SCHEMA_VERSION={} active. New writes are v2-only; v1 binaries opening this store will refuse with Exit::SchemaMismatch. Rollback is restore-from-blessed-pre-v2-backup (ADR 0033 §3); divergent migrations use ADR 0033 §4 fork isolation.",
            cortex_core::SCHEMA_VERSION
        );
        record_migrate_event(
            "cutover_summary",
            serde_json::json!({
                "schema_version_active": cortex_core::SCHEMA_VERSION,
                "boundary_event_hash": boundary_head,
                "boundary_previous_v1_head_hash": boundary.previous_v1_head_hash,
                "migration_script_digest": boundary.migration_script_digest,
                "fixture_verification_result_hash": boundary.fixture_verification_result_hash,
                "post_migrate_manifest": post_manifest_path.display().to_string(),
                "post_migrate_manifest_blake3": post_migrate_manifest_blake3,
                "cutover_authority": "ok",
                "cutover_approved": true,
                "cutover_guard": "committed",
                "post_migrate_row_count_refusal": "ok",
            }),
        );
        return Exit::Ok;
    }

    if let Some(backup_manifest) = args.backup_manifest.as_deref() {
        eprintln!(
            "cortex migrate v2: --backup-manifest `{}` is for the future mutating path and is not accepted with --dry-run. no state was changed.",
            backup_manifest.display()
        );
        return Exit::Usage;
    }

    let layout = match DataLayout::resolve(None, None) {
        Ok(layout) => layout,
        Err(exit) => return exit,
    };
    let pool = match open_default_store("migrate v2") {
        Ok(pool) => pool,
        Err(exit) => {
            eprintln!("cortex migrate v2: dry-run refused; no state was changed.");
            return exit;
        }
    };
    let plan = match cortex_store::migrate_v2::dry_run_plan(&pool) {
        Ok(plan) => plan,
        Err(err) => {
            eprintln!(
                "cortex migrate v2: failed to build dry-run plan: {err}. no state was changed."
            );
            return Exit::PreconditionUnmet;
        }
    };

    if !plan.is_ready() {
        for failure in &plan.failures {
            eprintln!(
                "cortex migrate v2: {}: {}",
                failure.invariant(),
                failure.detail()
            );
        }
        eprintln!("cortex migrate v2: dry-run refused; no state was changed.");
        eprintln!(
            "cortex migrate v2: hint: run `cortex doctor --repair` to apply pending migrations, \
             or retry `cortex migrate v2 --backup-manifest <path>` once the store preconditions \
             are satisfied"
        );
        return Exit::SchemaMismatch;
    }

    let boundary = match boundary_preflight(&layout, &plan) {
        Ok(boundary) => boundary,
        Err(exit) => return exit,
    };

    let json = output::json_enabled();
    if !json {
        println!("cortex migrate v2: dry-run ok");
        println!(
            "expected_schema_version={}",
            plan.fixture.expected_schema_version
        );
        println!(
            "observed_row_schema_versions={:?}",
            plan.fixture.observed_row_schema_versions
        );
        match &plan.fixture.event_chain_head {
            Some(head) => println!("event_chain_head={head}"),
            None => println!("event_chain_head=<none>"),
        }
        println!("jsonl_event_chain_head={}", boundary.previous_v1_head_hash);
        for migration in &plan.fixture.applied_migrations {
            println!("applied_migration={migration}");
        }
        for count in &plan.fixture.table_counts {
            println!("table={} rows={}", count.table, count.rows);
        }
        for step in &plan.steps {
            println!("step={step}");
        }
        println!("boundary_event_kind={SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND}");
        println!("schema_version_target=2");
        println!(
            "boundary_previous_v1_head_hash={}",
            boundary.previous_v1_head_hash
        );
        println!(
            "migration_script_digest={}",
            boundary.migration_script_digest
        );
        println!(
            "fixture_verification_result_hash={}",
            boundary.fixture_verification_result_hash
        );
        println!("operator_attestation_mode=dry_run_not_collected");
        println!("boundary_preflight_ready=true");
        println!("cutover_authority=ok");
        println!("cutover_approved=false");
        println!("cutover_guard=requires_backup_manifest");
        emit_cutover_readiness_stdout();
        println!("no state was changed");
    } else {
        record_migrate_event(
            "dry_run_summary",
            serde_json::json!({
                "expected_schema_version": plan.fixture.expected_schema_version,
                "observed_row_schema_versions": plan.fixture.observed_row_schema_versions,
                "event_chain_head": plan.fixture.event_chain_head,
                "jsonl_event_chain_head": boundary.previous_v1_head_hash,
                "applied_migrations": plan.fixture.applied_migrations,
                "table_counts": plan
                    .fixture
                    .table_counts
                    .iter()
                    .map(|count| serde_json::json!({
                        "table": count.table,
                        "rows": count.rows,
                    }))
                    .collect::<Vec<_>>(),
                "steps": plan.steps,
                "boundary_event_kind": SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND,
                "boundary_previous_v1_head_hash": boundary.previous_v1_head_hash,
                "migration_script_digest": boundary.migration_script_digest,
                "fixture_verification_result_hash": boundary.fixture_verification_result_hash,
                "operator_attestation_mode": "dry_run_not_collected",
                "boundary_preflight_ready": true,
                "cutover_authority": "ok",
                "cutover_approved": false,
                "cutover_guard": "requires_backup_manifest",
            }),
        );
    }

    Exit::Ok
}

/// Result of the atomic SQLite-side cutover transaction (Slice 1 / B1).
struct CutoverTxOutcome {
    expand_report: cortex_store::migrate_v2::V2ExpandBackfillReport,
    boundary_head: String,
    /// BLAKE3 digest of the canonical-serialized post-migrate manifest body
    /// (RED_TEAM_FINDINGS D2, Decision #6). Computed inside the cutover tx
    /// against the manifest payload built with the known `boundary_event_hash`,
    /// then stamped onto the boundary row's `source_attestation_json` column
    /// so it is anchored in the v2 audit chain atomically with the cutover.
    /// The same digest is written into the on-disk manifest's `manifest_blake3`
    /// field. The two are byte-identical.
    post_migrate_manifest_blake3: String,
    /// The full manifest payload (without `manifest_blake3`) used to compute
    /// the digest. The CLI then inserts `manifest_blake3` and writes it to
    /// disk after the cutover tx commits.
    post_migrate_manifest_body: serde_json::Value,
}

/// Errors raised inside the in-tx cutover body that should map to a CLI exit
/// code without propagating a partial SQLite mutation. Every variant rolls
/// back the surrounding transaction; the operator-facing message lives next
/// to the `match` arm in [`run_v2_cutover_tx`].
enum CutoverTxError {
    ExpandBackfill(cortex_store::StoreError),
    JsonlAppend(cortex_ledger::JsonlError),
    BoundaryMirror(cortex_store::StoreError),
    RebackfillAttestations(cortex_store::StoreError),
    BoundaryAuditFailures(Vec<cortex_ledger::audit::SchemaMigrationBoundaryFailure>),
    BoundaryAuditError(cortex_ledger::JsonlError),
    ChainAuditFailures(usize),
    ChainAuditError(cortex_ledger::JsonlError),
    PostMigrateCountMismatch(Vec<cortex_store::verify::PostMigrateCountMismatchFailure>),
    PostMigrateCountError(cortex_store::StoreError),
    CutoverReadiness(cortex_store::StoreError),
    TxBegin(cortex_store::StoreError),
    TxCommit(rusqlite::Error),
    /// Failed to stamp the post-migrate manifest digest into the boundary
    /// row's `source_attestation_json` column (Decision #6).
    PostMigrateManifestAnchor(rusqlite::Error),
}

/// Execute the atomic SQLite-side cutover (B1): expand/backfill, JSONL
/// boundary append, in-tx boundary mirror, post-mirror re-backfill, read-only
/// audit and readiness checks, then commit.
///
/// The function opens one explicit SQLite transaction up front and commits
/// only when every in-tx step has succeeded. Any in-tx failure drops the
/// transaction (rusqlite auto-rolls-back on drop) so the SQLite store
/// observed by the next process is identical to the pre-cutover snapshot.
/// The JSONL append itself is performed mid-tx because the canonical
/// durable record must exist before the mirror can copy it; if the mirror
/// or any subsequent in-tx step fails, the JSONL row is the only durable
/// trace and operators recover by re-running migrate (the existing
/// `boundary_preflight` refuses on the durable boundary; a follow-up
/// `--resume-mirror` flag is tracked as a TODO).
fn run_v2_cutover_tx(
    pool: &mut cortex_store::Pool,
    layout: &DataLayout,
    boundary: &BoundaryPreflight,
    migration_policy: &cortex_core::PolicyDecision,
    pre_counts_payload: &BackupManifestTableRowCounts,
    manifest_inputs: &PostMigrateManifestInputs,
) -> Result<CutoverTxOutcome, Exit> {
    let tx = match pool.transaction() {
        Ok(tx) => tx,
        Err(err) => {
            let err = cortex_store::StoreError::from(err);
            return Err(emit_cutover_tx_failure(
                CutoverTxError::TxBegin(err),
                boundary,
            ));
        }
    };

    let expand_report =
        match cortex_store::migrate_v2::apply_expand_backfill_skeleton(&tx, Utc::now()) {
            Ok(report) => report,
            Err(err) => {
                return Err(emit_cutover_tx_failure(
                    CutoverTxError::ExpandBackfill(err),
                    boundary,
                ));
            }
        };

    // JSONL boundary append. Stays OUTSIDE the SQLite tx by doctrine
    // (BUILD_SPEC §7: JSONL is canonical, SQLite mirror is a peer). If the
    // append succeeds and any later in-tx step fails, the JSONL row is
    // durable and the SQLite tx rolls back on drop; operators recover by
    // running `cortex migrate v2 --resume-mirror`, which re-mirrors the
    // durable JSONL boundary into a fresh SQLite snapshot inside its own
    // atomic transaction (see `run_v2_resume_mirror`).
    let (boundary_head, boundary_event) =
        match JsonlLog::open(&layout.event_log_path).and_then(|mut log| {
            log.append_schema_migration_v1_to_v2_with_event(
                SchemaMigrationV1ToV2Payload::new(
                    boundary.previous_v1_head_hash.clone(),
                    boundary.migration_script_digest.clone(),
                    None,
                    boundary.fixture_verification_result_hash.clone(),
                ),
                migration_policy,
            )
        }) {
            Ok((head, event)) => (head, event),
            Err(err) => {
                return Err(emit_cutover_tx_failure(
                    CutoverTxError::JsonlAppend(err),
                    boundary,
                ));
            }
        };

    // Mirror the boundary row into SQLite using the same transaction
    // (B1). Without this, the previous code opened a second tx inside
    // `mirror_single_event_into_sqlite` and committed independently of the
    // surrounding expand/backfill, defeating atomicity.
    if let Err(err) =
        cortex_store::mirror::mirror_single_event_into_sqlite_in_tx(&tx, &boundary_event)
    {
        return Err(emit_cutover_tx_failure(
            CutoverTxError::BoundaryMirror(err),
            boundary,
        ));
    }

    // Re-run the legacy attestation backfill so the just-mirrored boundary
    // row carries a `LegacyUnattested` marker. The earlier
    // `apply_expand_backfill_skeleton` ran before the boundary mirror, so
    // the boundary row would otherwise have `source_attestation_json IS NULL`
    // and the default-v2 cutover readiness gate would refuse.
    //
    // B2 (RED_TEAM_FINDINGS phase B): the boundary row is the *announcement*
    // of v2, not a legacy v1 import. The verified operator attestation
    // envelope above (ADR 0010 §1-§2, Gate 5 punch list #17) authorises the
    // boundary append at the migration authority root, but is not yet
    // materialised onto the boundary row's `source_attestation_json`
    // column. The `legacy_unattested` stamp is a narrow placeholder; the
    // rationale and the future-slice plan live on
    // `backfill_legacy_event_attestations` in `cortex-store::migrate_v2`
    // so the choice is durable next to the SQL that writes it.
    if let Err(err) = cortex_store::migrate_v2::backfill_legacy_event_attestations(&tx, Utc::now())
    {
        return Err(emit_cutover_tx_failure(
            CutoverTxError::RebackfillAttestations(err),
            boundary,
        ));
    }

    // Decision #6 / RED_TEAM_FINDINGS D2: stamp the post-migrate manifest
    // BLAKE3 digest onto the just-mirrored boundary row's
    // `source_attestation_json` column inside the same atomic cutover
    // transaction. The digest is the BLAKE3 of the canonical-serialized
    // manifest contents EXCLUDING the `manifest_blake3` field itself; the
    // CLI writes the on-disk manifest with `manifest_blake3` set to the same
    // hex string after the tx commits. The verifier
    // (`cortex restore verify-post-migrate-manifest`) recomputes and refuses
    // on mismatch.
    //
    // The merge uses SQLite's `json_set` so the existing
    // `{state, value}` shape produced by `backfill_legacy_event_attestations`
    // is preserved (`SourceAttestation` deserialization tolerates the extra
    // top-level field). Stamping happens after `backfill_legacy_event_attestations`
    // so an empty boundary row is impossible at this point — the boundary row
    // is guaranteed to already have a non-null `source_attestation_json`.
    let post_migrate_manifest_body = build_post_migrate_manifest_body(
        &boundary_head,
        boundary,
        pre_counts_payload,
        manifest_inputs,
    );
    let post_migrate_manifest_blake3 = canonical_blake3_hex(&post_migrate_manifest_body);
    if let Err(err) = tx.execute(
        "UPDATE events
         SET source_attestation_json = json_set(
             source_attestation_json,
             '$.post_migrate_manifest_blake3',
             ?1
         )
         WHERE id = ?2;",
        rusqlite::params![
            post_migrate_manifest_blake3.as_str(),
            boundary_event.id.to_string(),
        ],
    ) {
        return Err(emit_cutover_tx_failure(
            CutoverTxError::PostMigrateManifestAnchor(err),
            boundary,
        ));
    }

    // Read-only post-mirror audits. They participate in the same SQLite tx
    // for consistency, but no row is written.
    match verify_schema_migration_v1_to_v2_boundary(&layout.event_log_path, true) {
        Ok(report) if report.ok() => {}
        Ok(report) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::BoundaryAuditFailures(report.failures),
                boundary,
            ));
        }
        Err(err) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::BoundaryAuditError(err),
                boundary,
            ));
        }
    }
    match verify_chain(&layout.event_log_path) {
        Ok(report) if report.ok() => {}
        Ok(report) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::ChainAuditFailures(report.failures.len()),
                boundary,
            ));
        }
        Err(err) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::ChainAuditError(err),
                boundary,
            ));
        }
    }

    // Post-migrate count-mismatch refusal (ADR 0033 §1, Phase A4). Runs
    // against the in-tx connection so the count compare sees the boundary
    // row and any backfill changes that will only become durable on commit.
    let pre_counts = cortex_store::verify::PreV2BackupRowCounts {
        events: pre_counts_payload.events,
        traces: pre_counts_payload.traces,
        episodes: pre_counts_payload.episodes,
        memories: pre_counts_payload.memories,
    };
    match cortex_store::verify::verify_post_migrate_row_counts(&tx, &pre_counts) {
        Ok(failures) if failures.is_empty() => {}
        Ok(failures) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::PostMigrateCountMismatch(failures),
                boundary,
            ));
        }
        Err(err) => {
            return Err(emit_cutover_tx_failure(
                CutoverTxError::PostMigrateCountError(err),
                boundary,
            ));
        }
    }

    // Default schema-v2 cutover readiness gate (ADR 0033 §1 + §5).
    if let Err(err) = cortex_store::migrate_v2::require_default_v2_cutover_readiness(&tx) {
        return Err(emit_cutover_tx_failure(
            CutoverTxError::CutoverReadiness(err),
            boundary,
        ));
    }

    if let Err(err) = tx.commit() {
        return Err(emit_cutover_tx_failure(
            CutoverTxError::TxCommit(err),
            boundary,
        ));
    }

    Ok(CutoverTxOutcome {
        expand_report,
        boundary_head,
        post_migrate_manifest_blake3,
        post_migrate_manifest_body,
    })
}

/// Inputs supplied by the cutover-surface caller for the post-migrate
/// manifest's tamper-evident digest (Decision #6 / RED_TEAM_FINDINGS D2).
///
/// The migrate timestamp is captured ONCE at the cutover entry point and
/// threaded here so the manifest body, its BLAKE3 digest, and the on-disk
/// JSON are all built from the same value.
pub(crate) struct PostMigrateManifestInputs {
    pub(crate) pre_v2_backup_manifest_path: PathBuf,
    pub(crate) migrate_timestamp: DateTime<Utc>,
}

/// Key under which the canonical-bytes BLAKE3 digest is embedded in the
/// post-migrate manifest. This is the field the verifier strips before
/// recomputing the digest.
pub(crate) const POST_MIGRATE_MANIFEST_DIGEST_FIELD: &str = "manifest_blake3";

/// Schema-version target stamped on the post-migrate manifest body. This is
/// the manifest envelope version (Decision #6), not `cortex_core::SCHEMA_VERSION`,
/// so the field stays stable across future store-schema bumps until we
/// explicitly version the manifest envelope itself.
pub(crate) const POST_MIGRATE_MANIFEST_ENVELOPE_VERSION: u16 = 1;

/// Stable invariant tokens for the post-migrate manifest tamper-evident
/// digest verifier (Decision #6 / RED_TEAM_FINDINGS D2). Both
/// `cortex migrate v2` and `cortex restore verify-post-migrate-manifest`
/// emit these, so operator scripts grepping for them see a single token set.
pub(crate) const POST_MIGRATE_MANIFEST_CHAIN_ANCHORED_INVARIANT: &str =
    "migrate.post_manifest.chain_anchor.appended";

/// Build the post-migrate manifest body WITHOUT the `manifest_blake3` field.
///
/// The same function is used by the verifier to reproduce the canonical
/// bytes before hashing, so callers must NOT mutate the returned value in a
/// way that would diverge from this canonical shape. The verifier strips
/// `manifest_blake3` from the on-disk JSON, canonicalises the remaining
/// object, and compares the digest against the recorded value; if the shape
/// drifts here without a corresponding verifier update, verification will
/// fail closed for every legitimate manifest.
fn build_post_migrate_manifest_body(
    boundary_head: &str,
    boundary: &BoundaryPreflight,
    pre_counts: &BackupManifestTableRowCounts,
    inputs: &PostMigrateManifestInputs,
) -> serde_json::Value {
    // ADR 0037 §5 amendment: `cortex migrate v2` cutover runs entirely
    // under `local_unsigned` runtime mode. The cutover transaction
    // verifies the boundary event's hash chain plus the SQLite mirror
    // before commit (see `staged post-migration audit`), so a
    // successful manifest carries `proof_state=full_chain_verified` and
    // `authority_class=verified` (the operator attestation envelope
    // binds the cutover to an operator-signed root of trust). The
    // composed `claim_ceiling` clamps to `signed_local_ledger` — the
    // operator attestation lifts the class but the runtime mode caps
    // the ceiling at `local_unsigned` because the cutover itself does
    // not anchor externally.
    serde_json::json!({
        "kind": "cortex_post_v2_migrate",
        "manifest_envelope_version": POST_MIGRATE_MANIFEST_ENVELOPE_VERSION,
        "schema_version": cortex_core::SCHEMA_VERSION,
        "migrate_timestamp": inputs.migrate_timestamp.to_rfc3339(),
        "boundary_event_hash": boundary_head,
        "boundary_previous_v1_head_hash": boundary.previous_v1_head_hash,
        "migration_script_digest": boundary.migration_script_digest,
        "fixture_verification_result_hash": boundary.fixture_verification_result_hash,
        "pre_v2_backup_manifest": inputs.pre_v2_backup_manifest_path.display().to_string(),
        "pre_v2_table_row_counts": {
            "events": pre_counts.events,
            "traces": pre_counts.traces,
            "episodes": pre_counts.episodes,
            "memories": pre_counts.memories,
        },
        "schema_v1_to_v2_event_boundary_delta":
            cortex_store::verify::SCHEMA_V1_TO_V2_EVENT_BOUNDARY_DELTA,
        "tool_version": env!("CARGO_PKG_VERSION"),
        "runtime_mode": cortex_core::RuntimeMode::LocalUnsigned,
        "proof_state": cortex_core::ClaimProofState::FullChainVerified,
        "claim_ceiling": cortex_core::ClaimCeiling::LocalUnsigned,
        "authority_class": cortex_core::AuthorityClass::Verified,
    })
}

/// Compute the BLAKE3 digest of a JSON value in `blake3:<hex>` form using
/// the canonical encoder shared with the ledger payload-hash path.
///
/// The encoder sorts object keys recursively and emits no whitespace, so the
/// digest is byte-stable across implementations that produce semantically
/// equivalent JSON. This is the canonical-serialization scheme called out in
/// Decision #6 (sorted keys, no trailing whitespace).
pub(crate) fn canonical_blake3_hex(value: &serde_json::Value) -> String {
    let bytes = canonical_payload_bytes(value);
    format!("blake3:{}", blake3::hash(&bytes).to_hex())
}

/// Stable invariant name surfaced when the atomic cutover transaction
/// rolled back. Operator scripts and tests detect partial-mutation rollback
/// by grepping for this token.
const CUTOVER_TX_ROLLBACK_INVARIANT: &str = "migrate.v2.cutover_tx.rolled_back";

/// Map an in-tx failure to operator-facing stderr lines and a CLI exit.
/// Dropping the surrounding `Transaction` after this returns rolls back the
/// SQLite mutations; the JSONL append (if it ran) stays durable.
fn emit_cutover_tx_failure(err: CutoverTxError, boundary: &BoundaryPreflight) -> Exit {
    match err {
        CutoverTxError::TxBegin(err) => {
            eprintln!(
                "cortex migrate v2: failed to open SQLite cutover transaction: {err}. cutover refused; no state was changed."
            );
            Exit::PreconditionUnmet
        }
        CutoverTxError::ExpandBackfill(err) => {
            eprintln!(
                "cortex migrate v2: expand/backfill stage failed inside cutover transaction: {err}. boundary event was not appended; cutover remains disabled."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::PreconditionUnmet
        }
        CutoverTxError::JsonlAppend(err) => {
            eprintln!(
                "cortex migrate v2: boundary append failed: {err}. post-migration audit and cutover remain disabled."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::PreconditionUnmet
        }
        CutoverTxError::BoundaryMirror(err) => {
            eprintln!(
                "cortex migrate v2: boundary mirror into SQLite failed after JSONL append: {err}. cutover refused; the JSONL boundary row is durable — run `cortex migrate v2 --resume-mirror` to re-mirror it into SQLite, or restore from blessed pre-v2 backup."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
        CutoverTxError::RebackfillAttestations(err) => {
            eprintln!(
                "cortex migrate v2: boundary attestation backfill failed: {err}. cutover refused."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
        CutoverTxError::BoundaryAuditFailures(failures) => {
            for failure in &failures {
                eprintln!(
                    "cortex migrate v2: {}: {:?}",
                    failure.invariant, failure.detail
                );
            }
            eprintln!(
                "cortex migrate v2: boundary audit failed after append; post-migration audit and cutover remain disabled."
            );
            eprintln!(
                "cortex migrate v2: hint: the JSONL boundary row was appended but failed \
                 validation; run `cortex migrate v2 --resume-mirror` to retry the cutover, or \
                 restore from the blessed pre-v2 backup"
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::SchemaMismatch
        }
        CutoverTxError::BoundaryAuditError(err) => {
            eprintln!(
                "cortex migrate v2: boundary audit failed after append: {err}. post-migration audit and cutover remain disabled."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::PreconditionUnmet
        }
        CutoverTxError::ChainAuditFailures(count) => {
            eprintln!(
                "cortex migrate v2: staged post-migration audit found {count} hash-chain failure(s); cutover remains disabled."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
        CutoverTxError::ChainAuditError(err) => {
            eprintln!(
                "cortex migrate v2: staged post-migration audit failed: {err}. cutover remains disabled."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::PreconditionUnmet
        }
        CutoverTxError::PostMigrateCountMismatch(failures) => {
            for failure in &failures {
                eprintln!(
                    "cortex migrate v2: {}: {}",
                    failure.invariant(),
                    failure.detail()
                );
            }
            eprintln!(
                "cortex migrate v2: post-migrate row counts drifted from the pre-v2 backup manifest baseline. Restore from the blessed pre-v2 backup or use ADR 0033 §4 fork isolation; in-place down-migration is forbidden."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
        CutoverTxError::PostMigrateCountError(err) => {
            eprintln!(
                "cortex migrate v2: post-migrate row-count verification failed: {err}. Cutover refused; restore from blessed pre-v2 backup."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::PreconditionUnmet
        }
        CutoverTxError::CutoverReadiness(err) => {
            eprintln!(
                "cortex migrate v2: default-v2 cutover readiness gate failed: {err}. Cutover refused; restore from blessed pre-v2 backup."
            );
            eprintln!(
                "cortex migrate v2: hint: run `cortex migrate v2 --backup-manifest <path>` to \
                 reattempt the upgrade, or restore from the blessed pre-v2 backup if the store \
                 is unrecoverable"
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::SchemaMismatch
        }
        CutoverTxError::TxCommit(err) => {
            eprintln!(
                "cortex migrate v2: cutover transaction commit failed: {err}. cutover refused; rollback applied — no SQLite mutation is durable. If the JSONL boundary row appended successfully, run `cortex migrate v2 --resume-mirror` to re-mirror it into SQLite; otherwise restore from blessed pre-v2 backup."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
        CutoverTxError::PostMigrateManifestAnchor(err) => {
            eprintln!(
                "cortex migrate v2: failed to anchor post-migrate manifest BLAKE3 digest onto the boundary row's source_attestation_json column: {err}. cutover refused; rollback applied (Decision #6 / RED_TEAM_FINDINGS D2)."
            );
            emit_cutover_tx_rollback_line(boundary);
            Exit::IntegrityFailure
        }
    }
}

fn emit_cutover_tx_rollback_line(boundary: &BoundaryPreflight) {
    eprintln!(
        "cortex migrate v2: {CUTOVER_TX_ROLLBACK_INVARIANT}: SQLite cutover transaction rolled back; pre-cutover SQLite snapshot is intact (previous_v1_head_hash={}).",
        boundary.previous_v1_head_hash
    );
}

/// Stable invariant emitted when `--resume-mirror` successfully re-mirrors the
/// JSONL boundary row into SQLite and commits the recovery transaction. Operator
/// scripts and tests detect successful partial-mutation recovery by grepping
/// for this token.
const RESUME_MIRROR_COMPLETED_INVARIANT: &str = "migrate.v2.resume_mirror.completed";

/// Stable invariant emitted when `--resume-mirror` finds the boundary row
/// already present in SQLite. The flag is idempotent by design (re-running
/// against a fully-migrated store is a no-op), so this is success, not
/// failure: the recovery target was already reached. Exit::Ok with this
/// invariant on stderr.
const RESUME_MIRROR_BOUNDARY_ALREADY_MIRRORED_INVARIANT: &str =
    "migrate.v2.resume_mirror.boundary_already_mirrored";

/// Stable invariant emitted when `--resume-mirror` is invoked against a store
/// whose JSONL log has no `schema_migration.v1_to_v2` boundary row. That shape
/// is the wrong recovery surface: either no cutover ran yet (operator should
/// run a normal `cortex migrate v2 --backup-manifest` cutover) or both stores
/// were rolled back from blessed backup (no recovery needed). Either way,
/// `--resume-mirror` refuses closed without mutation.
const RESUME_MIRROR_NO_JSONL_BOUNDARY_INVARIANT: &str =
    "migrate.v2.resume_mirror.no_jsonl_boundary";

/// Stable invariant surfaced (warn-level) when `--resume-mirror` had to replay
/// the schema-v2 expand/backfill skeleton inside its recovery transaction
/// because the cutover tx rolled back the DDL.
///
/// BUG_HUNT_2026-05-12 BH-4: prior to this surface, `run_v2_resume_mirror`
/// assumed the schema-v2 expand DDL (notably the `events.source_attestation_json`
/// column) was durable from a prior cutover-tx execution. That assumption fails
/// in exactly the scenario the recovery path is designed for: a fresh-store
/// cutover where `apply_expand_backfill_skeleton` runs the ALTER inside the
/// cutover tx, then a later in-tx step refuses and rolls the whole tx back,
/// including the ALTER. The JSONL boundary append survives (it lives outside
/// the SQLite tx by doctrine), but the column it requires is gone — and the
/// resume-mirror legacy-attestation backfill writes into a non-existent column.
///
/// The recovery surface is now self-healing: it replays the (idempotent)
/// `apply_expand_backfill_skeleton` at the head of its own tx, identical to
/// the cutover ordering, then proceeds with the mirror INSERT. When the column
/// was already present the replay is a no-op (`add_column_if_missing` /
/// `create_table_if_missing` guards skip every DDL statement); when it had
/// been rolled back, this invariant is surfaced so operators can tell which
/// scenario fired.
const RESUME_MIRROR_SCHEMA_SKELETON_REPLAY_INVARIANT: &str =
    "migrate.v2.resume_mirror.schema_skeleton_replayed";

/// Re-mirror the durable JSONL `schema_migration.v1_to_v2` boundary row into
/// SQLite when the original cutover transaction rolled back after the JSONL
/// fsync but before the SQLite mirror committed (B1 partial-mutation
/// recovery, RED_TEAM_FINDINGS phase B).
///
/// The recovery preconditions are narrow on purpose:
///   1. JSONL has exactly one boundary row (verified via
///      `verify_schema_migration_v1_to_v2_boundary`).
///   2. SQLite does NOT already carry that boundary row (idempotency probe
///      keyed by the boundary's durable `event_hash`).
///
/// On those preconditions the function opens a single SQLite transaction,
/// inserts the boundary row via `mirror_single_event_into_sqlite_in_tx`,
/// re-runs `backfill_legacy_event_attestations` so the just-mirrored row
/// carries a `legacy_unattested` marker, runs
/// `require_default_v2_cutover_readiness` against the in-tx view, and commits.
///
/// Doctrine guardrails (ADR 0033):
///   * `--resume-mirror` is forward-recovery — it never appends a new
///     boundary row, never bumps `SCHEMA_VERSION`, and is not a down-migration.
///   * Idempotent: running `--resume-mirror` on a fully-migrated store is a
///     no-op with `RESUME_MIRROR_BOUNDARY_ALREADY_MIRRORED_INVARIANT`.
///   * Wrong-surface refusal: running `--resume-mirror` against a pre-cutover
///     store refuses closed with `RESUME_MIRROR_NO_JSONL_BOUNDARY_INVARIANT`.
///
/// **BH-4 framing — defensive cover, not a currently-reachable bug.** The
/// recovery transaction replays the (idempotent) schema-v2 expand/backfill
/// skeleton at its head before INSERTing the mirror row (see
/// [`RESUME_MIRROR_SCHEMA_SKELETON_REPLAY_INVARIANT`]). The scenario this
/// defends against — a cutover tx that ran `ALTER TABLE events ADD COLUMN
/// source_attestation_json` and then rolled back, leaving the JSONL boundary
/// durable but the column absent — is **not CLI-reachable today**:
/// `open_default_store` runs `verify_schema_version` on every entry, and
/// that probe refuses closed (`SchemaMismatch`) before `--resume-mirror`
/// can reach the replay path. The replay is therefore defensive cover for
/// **future code paths** that might bypass `verify_schema_version` (e.g.
/// an offline-recovery surface, a forensic CLI that opens the store
/// without the standard verify, or a refactor of `open_default_store` that
/// loosens the precondition). Future readers tuning their threat model
/// should treat the replay as "library-grade idempotence" rather than
/// "active bug fix"; if you remove it, you must first prove no future
/// caller can land in this surface with a rolled-back-internal ALTER.
fn run_v2_resume_mirror() -> Exit {
    let layout = match DataLayout::resolve(None, None) {
        Ok(layout) => layout,
        Err(exit) => return exit,
    };
    let mut pool = match open_default_store("migrate v2 --resume-mirror") {
        Ok(pool) => pool,
        Err(exit) => {
            eprintln!("cortex migrate v2 --resume-mirror: resume refused; no state was changed.");
            return exit;
        }
    };

    // Precondition 1: JSONL has exactly one boundary row. Required-mode audit
    // is the right surface — missing or duplicate boundary rows are both
    // structurally invalid for resume.
    let boundary_report = match verify_schema_migration_v1_to_v2_boundary(
        &layout.event_log_path,
        true,
    ) {
        Ok(report) => report,
        Err(err) => {
            eprintln!(
                    "cortex migrate v2 --resume-mirror: failed to inspect JSONL boundary: {err}. no state was changed."
                );
            return Exit::PreconditionUnmet;
        }
    };
    if boundary_report.boundary_rows.is_empty() {
        eprintln!(
            "cortex migrate v2 --resume-mirror: {RESUME_MIRROR_NO_JSONL_BOUNDARY_INVARIANT}: JSONL log `{}` does not carry a schema_migration.v1_to_v2 boundary row. --resume-mirror is the wrong recovery surface; if no cutover has run, use `cortex migrate v2 --backup-manifest <PATH> --operator-attestation <PATH>`; if both stores were rolled back from a blessed pre-v2 backup, no recovery is needed. no state was changed.",
            layout.event_log_path.display()
        );
        return Exit::PreconditionUnmet;
    }
    if !boundary_report.ok() {
        for failure in &boundary_report.failures {
            eprintln!(
                "cortex migrate v2 --resume-mirror: {}: {:?}",
                failure.invariant, failure.detail
            );
        }
        eprintln!(
            "cortex migrate v2 --resume-mirror: JSONL boundary audit failed; resume refused. no state was changed."
        );
        eprintln!(
            "cortex migrate v2 --resume-mirror: hint: if the boundary row is structurally \
             corrupt, restore from the blessed pre-v2 backup; if this is a fresh v2 store with \
             no boundary, run `cortex migrate v2 --backup-manifest <path>` instead"
        );
        return Exit::SchemaMismatch;
    }

    // Walk the JSONL log to extract the typed boundary `Event`. We need the
    // sealed event itself (not just its hash) so the mirror INSERT writes
    // byte-identical bytes into SQLite. The boundary report above guarantees
    // exactly one matching row exists; here we find it.
    let boundary_event = match load_boundary_event(&layout.event_log_path) {
        Ok(event) => event,
        Err(exit) => return exit,
    };

    // Precondition 2: SQLite must NOT already carry the boundary row. The
    // mirror helper itself is idempotent (identical row -> no-op), but we
    // probe up front so the operator-facing diagnostic is precise: the
    // recovery target is already reached.
    match boundary_row_present_in_sqlite(&pool, &boundary_event.event_hash) {
        Ok(true) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: {RESUME_MIRROR_BOUNDARY_ALREADY_MIRRORED_INVARIANT}: SQLite already carries the schema_migration.v1_to_v2 boundary row at event_hash={}. resume is a no-op; no state was changed.",
                boundary_event.event_hash
            );
            record_migrate_event(
                "resume_mirror_summary",
                serde_json::json!({
                    "invariant": RESUME_MIRROR_BOUNDARY_ALREADY_MIRRORED_INVARIANT,
                    "boundary_event_hash": boundary_event.event_hash,
                    "mirrored": false,
                    "no_op": true,
                }),
            );
            return Exit::Ok;
        }
        Ok(false) => {}
        Err(err) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: failed to probe SQLite for boundary row: {err}. no state was changed."
            );
            return Exit::PreconditionUnmet;
        }
    }

    // All preconditions hold; perform the recovery inside a single explicit
    // SQLite transaction so the mirror INSERT, the legacy-attestation
    // re-backfill, and the readiness gate participate in one atomic unit
    // (same B1 doctrine as the cutover transaction itself).
    let tx = match pool.transaction() {
        Ok(tx) => tx,
        Err(err) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: failed to open SQLite recovery transaction: {err}. no state was changed."
            );
            return Exit::PreconditionUnmet;
        }
    };

    // BUG_HUNT_2026-05-12 BH-4: replay the schema-v2 expand/backfill skeleton
    // at the head of the recovery tx, identical to the cutover ordering. The
    // cutover (`run_v2_cutover_tx`) calls `apply_expand_backfill_skeleton`
    // INSIDE its tx; if any later in-tx step refuses, the rollback includes
    // the ALTER TABLE statements that added `events.source_attestation_json`
    // and the side tables `mirror_single_event_into_sqlite_in_tx` /
    // `backfill_legacy_event_attestations` rely on. Without this replay, the
    // resume-mirror path runs `UPDATE events SET source_attestation_json = ...`
    // against a column that does not exist and fails with a raw SQLite error.
    //
    // Idempotency contract: `apply_expand_backfill_skeleton` guards each
    // `ALTER TABLE ADD COLUMN` with `add_column_if_missing` (PRAGMA
    // table_info check) and each `CREATE TABLE` with `create_table_if_missing`
    // (sqlite_master lookup). The internal backfills (`backfill_legacy_event_attestations`,
    // `backfill_episode_summary_spans`, `backfill_memory_summary_spans`,
    // `backfill_context_pack_advisories`, `backfill_memory_salience_defaults`)
    // all use `WHERE ... IS NULL` predicates so re-running them on an
    // already-backfilled store is a no-op. The common case (cutover persisted
    // the DDL but mirror failed downstream) sees zero columns added and zero
    // tables created.
    let expand_report = match cortex_store::migrate_v2::apply_expand_backfill_skeleton(
        &tx,
        Utc::now(),
    ) {
        Ok(report) => report,
        Err(err) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: schema v2 expand/backfill replay failed inside recovery transaction: {err}. transaction rolled back; no state was changed."
            );
            return Exit::IntegrityFailure;
        }
    };

    // Diagnose which BH-4 scenario fired so operators can tell whether the
    // DDL was rolled back by the cutover tx (replay added columns/tables) or
    // the cutover persisted the DDL and only the mirror downstream failed
    // (replay is a no-op). Both converge on the same recovery completion
    // invariant below; the warn-level token here is the distinguishing
    // signal.
    if !expand_report.added_columns.is_empty() || !expand_report.created_tables.is_empty() {
        eprintln!(
            "cortex migrate v2 --resume-mirror: {RESUME_MIRROR_SCHEMA_SKELETON_REPLAY_INVARIANT}: cutover transaction rolled back the schema v2 expand DDL; replayed inside recovery transaction. added_columns={:?} created_tables={:?}",
            expand_report.added_columns,
            expand_report.created_tables
        );
    }

    if let Err(err) =
        cortex_store::mirror::mirror_single_event_into_sqlite_in_tx(&tx, &boundary_event)
    {
        eprintln!(
            "cortex migrate v2 --resume-mirror: boundary mirror INSERT failed inside recovery transaction: {err}. transaction rolled back; no state was changed."
        );
        return Exit::IntegrityFailure;
    }

    if let Err(err) = cortex_store::migrate_v2::backfill_legacy_event_attestations(&tx, Utc::now())
    {
        eprintln!(
            "cortex migrate v2 --resume-mirror: legacy attestation backfill failed inside recovery transaction: {err}. transaction rolled back; no state was changed."
        );
        return Exit::IntegrityFailure;
    }

    if let Err(err) = cortex_store::migrate_v2::require_default_v2_cutover_readiness(&tx) {
        eprintln!(
            "cortex migrate v2 --resume-mirror: default-v2 cutover readiness gate failed inside recovery transaction: {err}. transaction rolled back; no state was changed."
        );
        eprintln!(
            "cortex migrate v2 --resume-mirror: hint: run `cortex migrate v2 \
             --backup-manifest <path>` to reattempt the upgrade from scratch, or restore from \
             the blessed pre-v2 backup"
        );
        return Exit::SchemaMismatch;
    }

    if let Err(err) = tx.commit() {
        eprintln!(
            "cortex migrate v2 --resume-mirror: recovery transaction commit failed: {err}. no SQLite mutation is durable; no state was changed."
        );
        return Exit::IntegrityFailure;
    }

    eprintln!(
        "cortex migrate v2 --resume-mirror: {RESUME_MIRROR_COMPLETED_INVARIANT}: boundary mirrored into SQLite and readiness gate green. boundary_event_hash={} jsonl_path={}",
        boundary_event.event_hash,
        layout.event_log_path.display()
    );
    record_migrate_event(
        "resume_mirror_summary",
        serde_json::json!({
            "invariant": RESUME_MIRROR_COMPLETED_INVARIANT,
            "boundary_event_hash": boundary_event.event_hash,
            "boundary_event_kind": SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND,
            "mirrored": true,
            "no_op": false,
            "schema_skeleton_replayed": !expand_report.added_columns.is_empty()
                || !expand_report.created_tables.is_empty(),
            "schema_skeleton_added_columns": expand_report.added_columns,
            "schema_skeleton_created_tables": expand_report.created_tables,
        }),
    );
    Exit::Ok
}

/// Walk the JSONL log and return the typed `schema_migration.v1_to_v2`
/// boundary event. Caller has already proven via
/// [`verify_schema_migration_v1_to_v2_boundary`] that exactly one boundary row
/// exists; this function rescans to materialise the sealed [`Event`] needed
/// by the mirror INSERT.
fn load_boundary_event(event_log_path: &Path) -> Result<Event, Exit> {
    let log = match JsonlLog::open(event_log_path) {
        Ok(log) => log,
        Err(err) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: failed to open JSONL log `{}`: {err}. no state was changed.",
                event_log_path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    let iter = match log.iter() {
        Ok(iter) => iter,
        Err(err) => {
            eprintln!(
                "cortex migrate v2 --resume-mirror: failed to iterate JSONL log: {err}. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    let mut boundary: Option<Event> = None;
    for item in iter {
        let event = match item {
            Ok(event) => event,
            Err(err) => {
                eprintln!(
                    "cortex migrate v2 --resume-mirror: failed to decode JSONL row: {err}. no state was changed."
                );
                return Err(Exit::PreconditionUnmet);
            }
        };
        if event
            .payload
            .get("kind")
            .and_then(serde_json::Value::as_str)
            == Some(SCHEMA_MIGRATION_V1_TO_V2_EVENT_KIND)
        {
            if boundary.is_some() {
                // verify_schema_migration_v1_to_v2_boundary would already
                // have refused this case, but defend against future drift.
                eprintln!(
                    "cortex migrate v2 --resume-mirror: JSONL log carries more than one schema_migration.v1_to_v2 row; resume refused. no state was changed."
                );
                eprintln!(
                    "cortex migrate v2 --resume-mirror: hint: this is a structural integrity \
                     issue — restore from the blessed pre-v2 backup to recover a clean store"
                );
                return Err(Exit::SchemaMismatch);
            }
            boundary = Some(event);
        }
    }
    boundary.ok_or_else(|| {
        // Mirrors the verify_schema_migration_v1_to_v2_boundary refusal but
        // is reachable only if the two scans disagree, which would itself be
        // a structural bug worth surfacing.
        eprintln!(
            "cortex migrate v2 --resume-mirror: boundary audit reported a boundary row but the rescan found none. no state was changed."
        );
        Exit::PreconditionUnmet
    })
}

/// Idempotency probe: does the SQLite `events` table already carry a row
/// whose `event_hash` matches the durable JSONL boundary?
///
/// Keyed by `event_hash` because the boundary's row id is fresh on every
/// cutover attempt (the JSONL boundary append generates a new `EventId`),
/// but the hash is stable across the JSONL fsync. Probing by hash also
/// matches `mirror_single_event_into_sqlite_in_tx`'s own no-op contract
/// (identical row -> `AlreadyPresent`), so this preflight cannot disagree
/// with the in-tx mirror outcome.
fn boundary_row_present_in_sqlite(
    pool: &cortex_store::Pool,
    event_hash: &str,
) -> Result<bool, cortex_store::StoreError> {
    let count: u64 = pool.query_row(
        "SELECT COUNT(*) FROM events WHERE event_hash = ?1;",
        rusqlite::params![event_hash],
        |row| row.get(0),
    )?;
    Ok(count > 0)
}

struct ValidBackupManifest {
    path: PathBuf,
    table_row_counts: BackupManifestTableRowCounts,
}

/// Stable invariant emitted when a `cortex_pre_v2_backup` manifest is fed to
/// `cortex migrate v2 --backup-manifest` against a live store that already
/// has v2 rows outside the boundary (R1, RED_TEAM_FINDINGS phase B).
const BACKUP_MANIFEST_PRE_V2_KIND_BUT_V2_ROWS_PRESENT_INVARIANT: &str =
    "migrate.v2.backup_manifest.pre_v2_kind_but_v2_rows_present";

/// R1 (RED_TEAM_FINDINGS phase B): refuse a `cortex_pre_v2_backup` manifest
/// when the live store carries any row with `schema_version >= 2` outside
/// the schema-migration boundary row.
///
/// `cortex backup` decides manifest kind purely by JSONL boundary presence
/// (`cortex-cli::cmd::backup::write_manifest`), so a fresh-v2 store with no
/// boundary still emits `kind=cortex_pre_v2_backup`. Feeding that manifest
/// to the cutover would let `migrate v2` append a boundary on top of v2
/// rows that the boundary falsely claims were v1 — the misuse path R1
/// describes. This guard refuses closed before any mutation.
///
/// The boundary row itself is excluded from the count so a legitimate
/// re-run against an already-cut-over store still trips `boundary_preflight`
/// (which is the right surface for "boundary already exists" refusal),
/// not this guard.
fn enforce_backup_manifest_consistent_with_live_store(
    pool: &cortex_store::Pool,
    manifest_path: &std::path::Path,
) -> Result<(), Exit> {
    let counts = match cortex_store::verify::count_post_v2_rows_outside_boundary(pool) {
        Ok(counts) => counts,
        Err(err) => {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` consistency check failed to read live store: {err}. no state was changed.",
                manifest_path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    if counts.is_empty() {
        return Ok(());
    }

    eprintln!(
        "cortex migrate v2: {BACKUP_MANIFEST_PRE_V2_KIND_BUT_V2_ROWS_PRESENT_INVARIANT}: backup manifest `{}` declares kind=cortex_pre_v2_backup but the live store has {} row(s) with schema_version >= 2 outside the schema_migration.v1_to_v2 boundary (events_post_v2={}, traces_post_v2={}). Refusing to corrupt a v2 store with a fake v1 migration. no state was changed.",
        manifest_path.display(),
        counts.total(),
        counts.events_post_v2,
        counts.traces_post_v2,
    );
    Err(Exit::PreconditionUnmet)
}

fn validate_backup_manifest(path: &std::path::Path) -> Result<ValidBackupManifest, Exit> {
    if !path.is_file() {
        eprintln!(
            "cortex migrate v2: backup manifest `{}` was not found; no state was changed.",
            path.display()
        );
        return Err(Exit::PreconditionUnmet);
    }

    let raw = match std::fs::read_to_string(path) {
        Ok(raw) => raw,
        Err(err) => {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` could not be read: {err}. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    let manifest_value: serde_json::Value = match serde_json::from_str(&raw) {
        Ok(manifest_value) => manifest_value,
        Err(err) => {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` is not a valid backup manifest: {err}. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    reject_reserved_cutover_approval_fields(path, &manifest_value)?;
    let manifest: BackupManifest = match serde_json::from_value(manifest_value) {
        Ok(manifest) => manifest,
        Err(err) => {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` is not a valid backup manifest: {err}. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    };

    if manifest.kind != "cortex_pre_v2_backup" {
        eprintln!(
            "cortex migrate v2: backup manifest `{}` has invalid kind `{}`; expected cortex_pre_v2_backup. no state was changed.",
            path.display(),
            manifest.kind
        );
        return Err(Exit::PreconditionUnmet);
    }
    if manifest.schema_version != PRE_V2_BACKUP_SCHEMA_VERSION {
        eprintln!(
            "cortex migrate v2: backup manifest `{}` has schema_version {}; expected {}. no state was changed.",
            path.display(),
            manifest.schema_version,
            PRE_V2_BACKUP_SCHEMA_VERSION
        );
        return Err(Exit::SchemaMismatch);
    }
    for (field, value) in [
        ("sqlite_store", manifest.sqlite_store.as_str()),
        ("jsonl_mirror", manifest.jsonl_mirror.as_str()),
        ("tool_version", manifest.tool_version.as_str()),
    ] {
        if value.trim().is_empty() {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` has empty `{field}`. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    }
    for (field, value) in [
        ("sqlite_store", manifest.sqlite_store.as_str()),
        ("jsonl_mirror", manifest.jsonl_mirror.as_str()),
    ] {
        if let Err(reason) = validate_manifest_artifact_ref(value) {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` has invalid `{field}` artifact reference `{value}`: {reason}. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
        let artifact_path = manifest_artifact_path(path, value);
        if !artifact_path.is_file() {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` references missing `{field}` artifact `{}`. no state was changed.",
                path.display(),
                artifact_path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    }
    if chrono::DateTime::parse_from_rfc3339(&manifest.backup_timestamp).is_err() {
        eprintln!(
            "cortex migrate v2: backup manifest `{}` has invalid `backup_timestamp`; expected RFC3339. no state was changed.",
            path.display()
        );
        return Err(Exit::PreconditionUnmet);
    }

    let Some(table_row_counts) = manifest.table_row_counts else {
        eprintln!(
            "cortex migrate v2: backup manifest `{}` is missing required `table_row_counts` field. Regenerate the backup with this binary's `cortex backup --output` so the post-migrate count-mismatch refusal helper has a pre-migrate baseline. no state was changed.",
            path.display()
        );
        return Err(Exit::PreconditionUnmet);
    };

    Ok(ValidBackupManifest {
        path: path.to_path_buf(),
        table_row_counts,
    })
}

fn reject_reserved_cutover_approval_fields(
    path: &std::path::Path,
    manifest: &serde_json::Value,
) -> Result<(), Exit> {
    let Some(object) = manifest.as_object() else {
        return Ok(());
    };

    for field in RESERVED_BACKUP_MANIFEST_CUTOVER_APPROVAL_FIELDS {
        if object.contains_key(*field) {
            eprintln!(
                "cortex migrate v2: backup manifest `{}` contains reserved cutover approval field `{field}`; backup manifests cannot approve schema cutover. no state was changed.",
                path.display()
            );
            return Err(Exit::PreconditionUnmet);
        }
    }

    Ok(())
}

fn emit_cutover_readiness_stdout() {
    for line in cutover_readiness_lines() {
        println!("{line}");
    }
}

fn emit_cutover_readiness_stderr() {
    for line in cutover_readiness_lines() {
        eprintln!("cortex migrate v2: {line}");
    }
}

fn cutover_readiness_lines() -> Vec<String> {
    vec![
        "default_v2_persistence_ready=true".to_string(),
        "default_v2_write_enabled=true".to_string(),
        "default_v2_cutover_ready=true".to_string(),
        "cutover_readiness=ready".to_string(),
        format!(
            "cutover_readiness_missing_gates={}",
            MISSING_ATOMIC_CUTOVER_PREREQUISITES.len()
        ),
        // After ADR 0026 punch list #17 landed the operator attestation
        // contributor at the migration authority root,
        // `--unattended-migrate` is supported only when paired with a valid
        // `--operator-attestation <PATH>` envelope (ADR 0010 §1-§2). The flag
        // reports the supported-with-attestation shape so operator scripts
        // can detect cutover-readiness without parsing the refusal text.
        "unattended_migrate_supported=requires_operator_attestation".to_string(),
    ]
}

fn validate_manifest_artifact_ref(artifact: &str) -> Result<(), &'static str> {
    let artifact_path = std::path::Path::new(artifact);
    if artifact_path.is_absolute() {
        return Err("absolute paths are not accepted");
    }
    if artifact_path
        .components()
        .any(|component| matches!(component, std::path::Component::ParentDir))
    {
        return Err("parent-directory traversal is not accepted");
    }
    Ok(())
}

fn manifest_artifact_path(manifest_path: &std::path::Path, artifact: &str) -> PathBuf {
    let artifact_path = std::path::Path::new(artifact);
    manifest_path
        .parent()
        .unwrap_or_else(|| std::path::Path::new("."))
        .join(artifact_path)
}

struct BoundaryPreflight {
    previous_v1_head_hash: String,
    migration_script_digest: String,
    fixture_verification_result_hash: String,
}

fn boundary_preflight(layout: &DataLayout, plan: &V2DryRunPlan) -> Result<BoundaryPreflight, Exit> {
    match verify_schema_migration_v1_to_v2_boundary(&layout.event_log_path, false) {
        Ok(report) if !report.failures.is_empty() => {
            for failure in &report.failures {
                eprintln!(
                    "cortex migrate v2: {}: {:?}",
                    failure.invariant, failure.detail
                );
            }
            eprintln!(
                "cortex migrate v2: existing schema boundary audit failed; no state was changed."
            );
            return Err(Exit::SchemaMismatch);
        }
        Ok(report) if !report.boundary_rows.is_empty() => {
            eprintln!(
                "cortex migrate v2: schema_migration.v1_to_v2 boundary already exists; no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
        Ok(_) => {}
        Err(err) => {
            eprintln!(
                "cortex migrate v2: failed to inspect existing schema boundary events: {err}. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
    }

    let jsonl_head = match JsonlLog::open(&layout.event_log_path) {
        Ok(log) => log.head().map(str::to_owned),
        Err(err) => {
            eprintln!(
                "cortex migrate v2: failed to inspect JSONL event head: {err}. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
    };
    let previous_v1_head_hash = match (jsonl_head, plan.fixture.event_chain_head.clone()) {
        (Some(jsonl), Some(sqlite)) if jsonl != sqlite => {
            eprintln!(
                "cortex migrate v2: boundary preflight found mismatched event heads: jsonl={jsonl}, sqlite={sqlite}. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
        (Some(jsonl), _) => jsonl,
        (None, Some(sqlite)) => {
            eprintln!(
                "cortex migrate v2: boundary preflight found SQLite head {sqlite} but no JSONL event head. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
        (None, None) => {
            eprintln!(
                "cortex migrate v2: boundary preflight requires a current v1 event_chain_head. no state was changed."
            );
            return Err(Exit::PreconditionUnmet);
        }
    };

    let migration_script_digest = migration_bundle_digest();
    let fixture_verification_result_hash =
        store_fixture_verification_result_hash(plan, &previous_v1_head_hash);

    Ok(BoundaryPreflight {
        previous_v1_head_hash,
        migration_script_digest,
        fixture_verification_result_hash,
    })
}

fn migration_bundle_digest() -> String {
    let mut input = String::new();
    input.push_str("cortex-schema-v2-migration-bundle\n");
    input.push_str("schema_v2_expand_sql\n");
    input.push_str(SCHEMA_V2_EXPAND_SQL);
    input.push_str("\ncli_tooling=cortex migrate v2 --dry-run preflight\n");
    format!("blake3:{}", blake3::hash(input.as_bytes()).to_hex())
}

fn stage_status_label(status: V2MigrationStageStatus) -> &'static str {
    match status {
        V2MigrationStageStatus::Ready => "ready",
        V2MigrationStageStatus::Pending => "pending",
        V2MigrationStageStatus::Blocked => "blocked",
    }
}

/// Stable schema version for the operator-attestation envelope at the
/// migration authority boundary. Bumping this requires an ADR update because
/// the envelope's canonical signing bytes change shape.
const OPERATOR_ATTESTATION_ENVELOPE_SCHEMA_VERSION: u16 = 1;

/// Discriminator string baked into the envelope so a captured envelope cannot
/// be replayed against a different surface (e.g. a future RESTORE_INTENT
/// envelope).
const OPERATOR_ATTESTATION_ENVELOPE_PURPOSE: &str = "cortex.schema_migration.v1_to_v2";

/// On-disk shape of the Ed25519-signed operator attestation envelope authorising
/// the v1 -> v2 migration boundary append.
///
/// Encoded as a single JSON document (`--operator-attestation <PATH>`). The
/// canonical signing input is a deterministic, length-prefixed binary
/// encoding of the envelope's non-signature fields — see
/// [`operator_attestation_signing_input`] for the exact framing. The
/// signature binds the operator's verifying key to the proposed boundary
/// payload triple `(previous_v1_head_hash, migration_script_digest,
/// fixture_verification_result_hash)` so a captured envelope cannot be
/// re-used against a different boundary.
#[derive(Debug, Clone, Deserialize)]
struct OperatorAttestationEnvelope {
    schema_version: u16,
    purpose: String,
    /// Hex-encoded Ed25519 public key (32 raw bytes, lowercase, no separator).
    operator_verifying_key_hex: String,
    /// Logical operator key identifier (free-form; matches `cortex-identity`).
    operator_key_id: String,
    /// RFC3339 timestamp the envelope was signed at.
    signed_at: DateTime<Utc>,
    /// Boundary payload triple that this attestation authorises. MUST match
    /// the just-computed boundary preflight values; mismatches refuse the
    /// envelope.
    boundary: OperatorAttestationBoundary,
    /// Hex-encoded Ed25519 signature (64 raw bytes, lowercase) over the
    /// canonical signing input.
    signature_hex: String,
}

#[derive(Debug, Clone, Deserialize)]
struct OperatorAttestationBoundary {
    previous_v1_head_hash: String,
    migration_script_digest: String,
    fixture_verification_result_hash: String,
}

/// Errors raised while loading or verifying an operator attestation envelope.
#[derive(Debug)]
enum OperatorAttestationError {
    NotFound(PathBuf),
    Read {
        path: PathBuf,
        source: std::io::Error,
    },
    Decode {
        path: PathBuf,
        source: serde_json::Error,
    },
    UnknownSchema {
        path: PathBuf,
        found: u16,
        expected: u16,
    },
    WrongPurpose {
        path: PathBuf,
        found: String,
        expected: &'static str,
    },
    MalformedVerifyingKey {
        path: PathBuf,
        detail: String,
    },
    MalformedSignature {
        path: PathBuf,
        detail: String,
    },
    BoundaryMismatch {
        path: PathBuf,
        field: &'static str,
        envelope: String,
        observed: String,
    },
    SignatureRejected {
        path: PathBuf,
    },
}

impl std::fmt::Display for OperatorAttestationError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::NotFound(path) => write!(
                f,
                "operator attestation `{}` was not found",
                path.display()
            ),
            Self::Read { path, source } => write!(
                f,
                "operator attestation `{}` could not be read: {source}",
                path.display()
            ),
            Self::Decode { path, source } => write!(
                f,
                "operator attestation `{}` is not a valid JSON envelope: {source}",
                path.display()
            ),
            Self::UnknownSchema {
                path,
                found,
                expected,
            } => write!(
                f,
                "operator attestation `{}` declares schema_version {found}; expected {expected}",
                path.display()
            ),
            Self::WrongPurpose {
                path,
                found,
                expected,
            } => write!(
                f,
                "operator attestation `{}` has purpose `{found}`; expected `{expected}`",
                path.display()
            ),
            Self::MalformedVerifyingKey { path, detail } => write!(
                f,
                "operator attestation `{}` has a malformed operator_verifying_key_hex field: {detail}",
                path.display()
            ),
            Self::MalformedSignature { path, detail } => write!(
                f,
                "operator attestation `{}` has a malformed signature_hex field: {detail}",
                path.display()
            ),
            Self::BoundaryMismatch {
                path,
                field,
                envelope,
                observed,
            } => write!(
                f,
                "operator attestation `{}` boundary field `{field}` mismatch: envelope=`{envelope}`, observed=`{observed}`",
                path.display()
            ),
            Self::SignatureRejected { path } => write!(
                f,
                "operator attestation `{}` Ed25519 signature did not verify under the declared operator key",
                path.display()
            ),
        }
    }
}

/// Decode a lowercase hex string into bytes. Returns `Err` with a stable
/// reason on any non-hex character or odd length.
fn decode_lowercase_hex(input: &str) -> Result<Vec<u8>, String> {
    if input.len() % 2 != 0 {
        return Err(format!("odd hex length {}", input.len()));
    }
    let mut out = Vec::with_capacity(input.len() / 2);
    let bytes = input.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        let hi = hex_nibble(bytes[i])
            .ok_or_else(|| format!("invalid hex byte `{}` at offset {i}", bytes[i] as char))?;
        let lo = hex_nibble(bytes[i + 1]).ok_or_else(|| {
            format!(
                "invalid hex byte `{}` at offset {}",
                bytes[i + 1] as char,
                i + 1
            )
        })?;
        out.push((hi << 4) | lo);
        i += 2;
    }
    Ok(out)
}

fn hex_nibble(byte: u8) -> Option<u8> {
    match byte {
        b'0'..=b'9' => Some(byte - b'0'),
        b'a'..=b'f' => Some(byte - b'a' + 10),
        _ => None,
    }
}

/// Build the canonical signing input bytes for an operator attestation
/// envelope. Length-prefixed framing with big-endian length prefixes, in the
/// fixed order `(schema_version, purpose, operator_key_id, signed_at,
/// previous_v1_head_hash, migration_script_digest,
/// fixture_verification_result_hash)`. A 1-byte domain tag (`0x20`) prefixes
/// the bytes so the operator-attestation domain is structurally disjoint
/// from `cortex-core::canonical` (which reserves `0x10` for the Ed25519
/// attestation preimage and `0x11` for the rotation envelope).
fn operator_attestation_signing_input(env: &OperatorAttestationEnvelope) -> Vec<u8> {
    let signed_at_rfc3339 = env.signed_at.to_rfc3339();
    OperatorAttestationEnvelopeForSigning {
        schema_version: env.schema_version,
        purpose: env.purpose.as_str(),
        operator_key_id: env.operator_key_id.as_str(),
        signed_at_rfc3339: &signed_at_rfc3339,
        previous_v1_head_hash: env.boundary.previous_v1_head_hash.as_str(),
        migration_script_digest: env.boundary.migration_script_digest.as_str(),
        fixture_verification_result_hash: env.boundary.fixture_verification_result_hash.as_str(),
    }
    .signing_input()
}

fn push_lp(out: &mut Vec<u8>, bytes: &[u8]) {
    out.extend_from_slice(&(bytes.len() as u64).to_be_bytes());
    out.extend_from_slice(bytes);
}

/// Load an operator attestation envelope from disk and verify its signature
/// against the just-computed boundary preflight.
fn load_and_verify_operator_attestation(
    path: &Path,
    boundary: &BoundaryPreflight,
) -> Result<OperatorAttestationEnvelope, OperatorAttestationError> {
    if !path.is_file() {
        return Err(OperatorAttestationError::NotFound(path.to_path_buf()));
    }
    let raw = std::fs::read_to_string(path).map_err(|source| OperatorAttestationError::Read {
        path: path.to_path_buf(),
        source,
    })?;
    let envelope: OperatorAttestationEnvelope =
        serde_json::from_str(&raw).map_err(|source| OperatorAttestationError::Decode {
            path: path.to_path_buf(),
            source,
        })?;

    if envelope.schema_version != OPERATOR_ATTESTATION_ENVELOPE_SCHEMA_VERSION {
        return Err(OperatorAttestationError::UnknownSchema {
            path: path.to_path_buf(),
            found: envelope.schema_version,
            expected: OPERATOR_ATTESTATION_ENVELOPE_SCHEMA_VERSION,
        });
    }
    if envelope.purpose != OPERATOR_ATTESTATION_ENVELOPE_PURPOSE {
        return Err(OperatorAttestationError::WrongPurpose {
            path: path.to_path_buf(),
            found: envelope.purpose.clone(),
            expected: OPERATOR_ATTESTATION_ENVELOPE_PURPOSE,
        });
    }

    if envelope.boundary.previous_v1_head_hash != boundary.previous_v1_head_hash {
        return Err(OperatorAttestationError::BoundaryMismatch {
            path: path.to_path_buf(),
            field: "previous_v1_head_hash",
            envelope: envelope.boundary.previous_v1_head_hash.clone(),
            observed: boundary.previous_v1_head_hash.clone(),
        });
    }
    if envelope.boundary.migration_script_digest != boundary.migration_script_digest {
        return Err(OperatorAttestationError::BoundaryMismatch {
            path: path.to_path_buf(),
            field: "migration_script_digest",
            envelope: envelope.boundary.migration_script_digest.clone(),
            observed: boundary.migration_script_digest.clone(),
        });
    }
    if envelope.boundary.fixture_verification_result_hash
        != boundary.fixture_verification_result_hash
    {
        return Err(OperatorAttestationError::BoundaryMismatch {
            path: path.to_path_buf(),
            field: "fixture_verification_result_hash",
            envelope: envelope.boundary.fixture_verification_result_hash.clone(),
            observed: boundary.fixture_verification_result_hash.clone(),
        });
    }

    let verifying_key_bytes =
        decode_lowercase_hex(&envelope.operator_verifying_key_hex).map_err(|detail| {
            OperatorAttestationError::MalformedVerifyingKey {
                path: path.to_path_buf(),
                detail,
            }
        })?;
    if verifying_key_bytes.len() != 32 {
        return Err(OperatorAttestationError::MalformedVerifyingKey {
            path: path.to_path_buf(),
            detail: format!(
                "expected 32 verifying key bytes, got {}",
                verifying_key_bytes.len()
            ),
        });
    }
    let mut key_array = [0u8; 32];
    key_array.copy_from_slice(&verifying_key_bytes);
    let verifying_key = VerifyingKey::from_bytes(&key_array).map_err(|err| {
        OperatorAttestationError::MalformedVerifyingKey {
            path: path.to_path_buf(),
            detail: err.to_string(),
        }
    })?;

    let signature_bytes = decode_lowercase_hex(&envelope.signature_hex).map_err(|detail| {
        OperatorAttestationError::MalformedSignature {
            path: path.to_path_buf(),
            detail,
        }
    })?;
    if signature_bytes.len() != 64 {
        return Err(OperatorAttestationError::MalformedSignature {
            path: path.to_path_buf(),
            detail: format!("expected 64 signature bytes, got {}", signature_bytes.len()),
        });
    }
    let mut sig_array = [0u8; 64];
    sig_array.copy_from_slice(&signature_bytes);
    let signature = Signature::from_bytes(&sig_array);

    let signing_input = operator_attestation_signing_input(&envelope);
    verifying_key
        .verify(&signing_input, &signature)
        .map_err(|_| OperatorAttestationError::SignatureRejected {
            path: path.to_path_buf(),
        })?;

    Ok(envelope)
}

/// Compose the ADR 0026 policy decision for the v1 -> v2 boundary append
/// (Gate 5 punch list #17). Each contributor reflects a doctrine root:
///
/// - `ledger.schema_migration.authority_class` — ADR 0019 §3: a non-operator
///   key cannot mint a v2 boundary. Today the CLI gates the boundary path
///   behind `--backup-manifest` + `--operator-attestation`; supplying a
///   verified attestation is taken as proof the proposing principal sits in
///   the `Operator` authority class. Future tier-admin scoping (ADR 0019 §7)
///   can refine this contributor without changing the rule id.
/// - `ledger.schema_migration.attestation_required` — ADR 0010 §1-§2: a
///   fresh Ed25519-signed operator attestation MUST be supplied over the
///   boundary payload triple. Absent or invalid attestation votes `Reject`.
/// - `ledger.schema_migration.current_use_temporal_authority` — ADR 0023 §2
///   / §5: the signing key state at attestation time MUST be `Active`.
///   Phase 2.6 closure
///   (`docs/design/PHASE_2_6_temporal_authority_revalidation_audit.md`)
///   wires this contributor through
///   [`AuthorityRepo::revalidate`](cortex_store::repo::AuthorityRepo::revalidate)
///   against the durable `authority_key_timeline` and
///   `authority_principal_timeline` rows for the operator-supplied key.
///   A revoked / retired / sub-`Operator` key votes `Reject` here and
///   fails the cutover closed before any boundary mutation.
fn build_migration_policy_decision(
    pool: &Pool,
    attestation_path: Option<&Path>,
    boundary: &BoundaryPreflight,
) -> Result<PolicyDecision, Exit> {
    let Some(path) = attestation_path else {
        eprintln!(
            "cortex migrate v2: cutover requires --operator-attestation <PATH> with a valid Ed25519-signed operator attestation envelope authorising the v1 -> v2 boundary (ADR 0010 §1-§2, ADR 0026 §4). no state was changed."
        );
        return Err(Exit::PreconditionUnmet);
    };

    let envelope = match load_and_verify_operator_attestation(path, boundary) {
        Ok(envelope) => envelope,
        Err(err) => {
            eprintln!("cortex migrate v2: {err}. no state was changed.");
            return Err(Exit::PreconditionUnmet);
        }
    };

    let attestation_reason = format!(
        "operator attestation envelope `{}` verified under key {} signed at {}",
        path.display(),
        envelope.operator_key_id,
        envelope.signed_at.to_rfc3339(),
    );

    let authority_reason = format!(
        "operator attestation envelope authorises ADR 0019 §3 operator authority class for key {}",
        envelope.operator_key_id
    );

    // Phase 2.6 closure: derive the current-use temporal authority
    // contributor from `AuthorityRepo::revalidate` against the durable
    // key/principal timeline rather than from a literal `Allow`.
    // `minimum_trust_tier = Operator` per the per-surface table in the
    // audit §6.2 — schema cutover is a doctrine root (ADR 0026 §4 hard
    // wall) and cannot run under a sub-`Operator` key.
    let temporal_authority = match revalidate_operator_temporal_authority(
        pool,
        SCHEMA_MIGRATION_CURRENT_USE_TEMPORAL_AUTHORITY_RULE_ID,
        &envelope.operator_key_id,
        envelope.signed_at,
        TrustTier::Operator,
    ) {
        Ok(contribution) => contribution,
        Err(err) => {
            let invariant = revalidation_failed_invariant("migrate.v2");
            eprintln!(
                "cortex migrate v2: {invariant}: failed to read authority timeline for key {}: {err}. no state was changed.",
                envelope.operator_key_id,
            );
            return Err(Exit::PreconditionUnmet);
        }
    };

    let contributions = vec![
        PolicyContribution::new(
            SCHEMA_MIGRATION_AUTHORITY_CLASS_RULE_ID,
            PolicyOutcome::Allow,
            authority_reason,
        )
        .expect("static contribution shape is valid"),
        PolicyContribution::new(
            SCHEMA_MIGRATION_ATTESTATION_REQUIRED_RULE_ID,
            PolicyOutcome::Allow,
            attestation_reason,
        )
        .expect("static contribution shape is valid"),
        temporal_authority.contribution(),
    ];

    let decision = compose_policy_outcomes(contributions, None);
    if !matches!(
        decision.final_outcome,
        PolicyOutcome::Allow | PolicyOutcome::Warn | PolicyOutcome::BreakGlass
    ) {
        let invariant = revalidation_failed_invariant("migrate.v2");
        if !temporal_authority.report.valid_now {
            eprintln!(
                "cortex migrate v2: {invariant}: operator temporal authority current use blocked for key {} (reasons: {}). no state was changed.",
                temporal_authority.report.key_id,
                temporal_authority
                    .report
                    .reasons
                    .iter()
                    .map(|reason| reason.wire_str())
                    .collect::<Vec<_>>()
                    .join(","),
            );
        }
        eprintln!(
            "cortex migrate v2: ADR 0026 composition for the v1 -> v2 boundary refused with outcome {:?}. no state was changed.",
            decision.final_outcome
        );
        return Err(Exit::PreconditionUnmet);
    }

    eprintln!(
        "cortex migrate v2: operator_attestation_path={}",
        path.display()
    );
    eprintln!(
        "cortex migrate v2: operator_attestation_key_id={}",
        envelope.operator_key_id
    );
    eprintln!(
        "cortex migrate v2: operator_attestation_signed_at={}",
        envelope.signed_at.to_rfc3339()
    );
    eprintln!("cortex migrate v2: operator_attestation_verified=true");
    eprintln!(
        "cortex migrate v2: operator_temporal_authority_revalidated=true key_id={} valid_now={}",
        temporal_authority.report.key_id, temporal_authority.report.valid_now,
    );

    Ok(decision)
}