"""Operator-facing migration archive and rehearsal commands."""

from __future__ import annotations

import json
import secrets
import shutil
import subprocess
from datetime import UTC, datetime
from enum import StrEnum
from pathlib import Path
from typing import Annotated, Any
from urllib.parse import urlsplit, urlunsplit
from uuid import uuid4

import typer

from sibyl.cli.common import console, error, info, run_async, success, warn
from sibyl.config import settings
from sibyl.persistence.auth_archive import (
    AUTH_ARCHIVE_TABLES,
    export_auth_archive_payload,
    restore_auth_archive_payload,
)
from sibyl.persistence.content_archive import (
    export_content_archive_payload,
    restore_content_archive_payload,
)
from sibyl_core.migrate.archive import (
    AUTH_FILENAME,
    CONTENT_FILENAME,
    GRAPH_FILENAME,
    POSTGRES_FILENAME,
    auth_payload_from_archive,
    build_manifest,
    content_payload_from_archive,
    effective_graph_counts,
    graph_payload_from_archive,
    load_archive,
    validate_archive,
    write_archive,
)
from sibyl_core.migrate.merge import ArchiveMergeOptions, EntityCollisionPolicy, merge_archives

app = typer.Typer(
    name="migrate",
    help="Migration archives, verification, and rehearsal tooling",
    no_args_is_help=True,
)

DEFAULT_REHEARSAL_BASE_URL = "http://localhost:3334"
DEFAULT_REHEARSAL_BASELINES_DIR = Path("baselines")
DEFAULT_REHEARSAL_MANIFEST = Path(".moon/cache/baseline-runtime-manifest.json")
DEFAULT_REHEARSAL_EMAIL = "baseline-corpus@sibyl.dev"
DEFAULT_REHEARSAL_PASSWORD = "baseline-corpus-password-secure-123!"  # noqa: S105
DEFAULT_AUTH_FLOW_PASSWORD = "auth-flow-password-secure-123!"  # noqa: S105
DEFAULT_AUTH_FLOW_EMAIL_OUTBOX = Path(".moon/cache/auth-flow-email-outbox.jsonl")
DEFAULT_CUTOVER_BENCH_LABEL = "cutover-acceptance"
_AUTH_READ_ONLY_FUNCTION = "sibyl_reject_auth_rbac_write"
_AUTH_READ_ONLY_TRIGGER = "sibyl_auth_rbac_read_only"
DEFAULT_CLOUD_ENV_PATH = Path(".env.sibyl-cloud")


async def verify_graph_archive(*args: Any, **kwargs: Any) -> Any:
    from sibyl_core.migrate.verify import verify_graph_archive as _verify_graph_archive

    return await _verify_graph_archive(*args, **kwargs)


class AuthReadOnlyMode(StrEnum):
    freeze = "freeze"
    unfreeze = "unfreeze"


class ArchiveSourceType(StrEnum):
    surreal_archive = "surreal-archive"
    legacy_archive = "legacy-archive"


class ArchiveTargetMode(StrEnum):
    surreal = "surreal"
    postgres_rehearsal = "postgres-rehearsal"


class CloudCoordinationBackend(StrEnum):
    local = "local"
    redis = "redis"


def _dotenv_quote(value: str) -> str:
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def _dotenv_line(key: str, value: str) -> str:
    return f"{key}={_dotenv_quote(value)}"


def _render_cloud_env(
    *,
    surreal_url: str,
    surreal_username: str,
    surreal_password: str,
    public_url: str,
    coordination_backend: CloudCoordinationBackend,
    redis_host: str,
    redis_port: int,
    redis_password: str,
    canonical_org_id: str,
) -> str:
    lines = [
        "# Generated by: sibyld migrate cloud-env",
        f"# Generated at: {datetime.now(UTC).isoformat()}",
        "",
        _dotenv_line("SIBYL_ENVIRONMENT", "production"),
        _dotenv_line("SIBYL_STORE", "surreal"),
        _dotenv_line("SIBYL_AUTH_STORE", "surreal"),
        _dotenv_line("SIBYL_SURREAL_URL", surreal_url or "wss://<surreal-cloud-endpoint>/rpc"),
        _dotenv_line("SIBYL_SURREAL_USERNAME", surreal_username or "root"),
        _dotenv_line("SIBYL_SURREAL_PASSWORD", surreal_password or "<surreal-root-password>"),
        _dotenv_line("SIBYL_JWT_SECRET", secrets.token_hex(32)),
        _dotenv_line("SIBYL_PASSWORD_PEPPER", secrets.token_hex(32)),
        _dotenv_line("SIBYL_PUBLIC_URL", public_url or "https://<your-sibyl-host>"),
        _dotenv_line("SIBYL_COORDINATION_BACKEND", coordination_backend.value),
    ]
    if coordination_backend is CloudCoordinationBackend.redis:
        lines.extend(
            [
                _dotenv_line("SIBYL_REDIS_HOST", redis_host),
                _dotenv_line("SIBYL_REDIS_PORT", str(redis_port)),
                _dotenv_line("SIBYL_REDIS_PASSWORD", redis_password),
            ]
        )
    lines.extend(
        [
            "",
            "# Migration helper value; not read by Sibyl at runtime.",
            _dotenv_line("SIBYL_MIGRATION_CANONICAL_ORG_ID", canonical_org_id or str(uuid4())),
            "",
        ]
    )
    return "\n".join(lines)


def _print_cloud_env_next_steps(output: Path, canonical_org_id: str) -> None:
    info("Next steps:")
    typer.echo(f"  set -a; source {output}; set +a")
    typer.echo("  sibyld migrate export -o sibyl-$(hostname).tar.gz")
    typer.echo(
        "  sibyld migrate merge *.tar.gz "
        f"--canonical-org-id {canonical_org_id} "
        "-o sibyl-cloud-merged.tar.gz --dry-run"
    )
    typer.echo(
        "  sibyld migrate import sibyl-cloud-merged.tar.gz "
        "--source-type surreal-archive --target-mode surreal --dry-run"
    )


def _load_valid_archive(source: Path):
    try:
        archive = load_archive(source)
    except Exception as exc:
        error(f"Archive load failed: {exc}")
        raise typer.Exit(code=1) from exc

    errors = validate_archive(archive)
    if errors:
        for issue in errors:
            warn(issue)
        error("Archive validation failed")
        raise typer.Exit(code=1)

    return archive


def _resolve_org_id(requested_org_id: str, archive_org_id: str) -> str:
    effective_org_id = requested_org_id or archive_org_id
    if not effective_org_id:
        error("Operation requires --org-id or an archive manifest organization_id")
        raise typer.Exit(code=1)
    return effective_org_id


def _resolve_export_org_id(requested_org_id: str) -> str:
    if requested_org_id:
        return requested_org_id

    from sibyl.persistence import organization_runtime

    @run_async
    async def _list_org_ids() -> list[str]:
        return await organization_runtime.list_org_ids()

    org_ids = _list_org_ids()
    if len(org_ids) == 1:
        org_id = org_ids[0]
        info(f"No --org-id supplied; using the only organization: {org_id}")
        return org_id

    if not org_ids:
        error("No organizations found in the active auth runtime")
        info("Create an organization first, or pass --org-id if the graph exists elsewhere")
        raise typer.Exit(code=1)

    error("Multiple organizations found; pass --org-id to choose one")
    for org_id in org_ids[:10]:
        console.print(f"  [dim]{org_id}[/dim]")
    if len(org_ids) > 10:
        console.print(f"  [dim]... and {len(org_ids) - 10} more[/dim]")
    info("Rerun the export with --org-id ORG_UUID")
    raise typer.Exit(code=1)


def _print_verify_summary(result: object) -> None:
    expected_entities = getattr(result, "expected_entities", 0)
    actual_entities = getattr(result, "actual_entities", 0)
    expected_relationships = getattr(result, "expected_relationships", 0)
    actual_relationships = getattr(result, "actual_relationships", 0)
    expected_episodes = getattr(result, "expected_episodes", 0)
    actual_episodes = getattr(result, "actual_episodes", 0)
    expected_mentions = getattr(result, "expected_mentions", 0)
    actual_mentions = getattr(result, "actual_mentions", 0)
    validated_entity_ids = list(getattr(result, "validated_entity_ids", []))
    validated_episode_ids = list(getattr(result, "validated_episode_ids", []))
    errors = list(getattr(result, "errors", []))

    info(f"Entities: expected {expected_entities}, actual {actual_entities}")
    info(f"Relationships: expected {expected_relationships}, actual {actual_relationships}")
    info(f"Episodes: expected {expected_episodes}, actual {actual_episodes}")
    info(f"Mentions: expected {expected_mentions}, actual {actual_mentions}")
    if validated_entity_ids:
        info(f"Sampled entities: {len(validated_entity_ids)}")
    if validated_episode_ids:
        info(f"Sampled episodes: {len(validated_episode_ids)}")
    if errors:
        warn(f"Verification failed with {len(errors)} issue(s)")
        for issue in errors:
            console.print(f"  [dim]{issue}[/dim]")
        raise typer.Exit(code=1)


def _get_pg_env() -> dict[str, str]:
    """Build environment for retained archive rehearsal pg tools."""
    import os

    env = os.environ.copy()
    env["PGPASSWORD"] = settings.postgres_password.get_secret_value()
    return env


def _get_pg_connection_args() -> list[str]:
    """Build psql connection args for retained archive rehearsal."""
    return [
        "-h",
        settings.postgres_host,
        "-p",
        str(settings.postgres_port),
        "-U",
        settings.postgres_user,
        "-d",
        settings.postgres_db,
    ]


def _find_pg_tool(tool: str) -> str:
    """Find PostgreSQL client tools for explicit archive rehearsal."""
    for path in (
        f"/opt/homebrew/opt/libpq/bin/{tool}",
        f"/opt/homebrew/opt/postgresql@18/bin/{tool}",
        f"/opt/homebrew/opt/postgresql@17/bin/{tool}",
        f"/opt/homebrew/opt/postgresql@16/bin/{tool}",
        f"/usr/local/opt/libpq/bin/{tool}",
        f"/usr/local/opt/postgresql@18/bin/{tool}",
        f"/usr/local/opt/postgresql@17/bin/{tool}",
        f"/usr/local/opt/postgresql@16/bin/{tool}",
    ):
        if Path(path).exists():
            return path
    return shutil.which(tool) or tool


def _restore_pg_sql(sql_content: str, clean: bool) -> None:
    """Restore a retained postgres.sql payload during explicit rehearsal."""
    if clean:
        drop_sql = """
DO $$ DECLARE
    r RECORD;
BEGIN
    FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname = 'public') LOOP
        EXECUTE 'DROP TABLE IF EXISTS ' || quote_ident(r.tablename) || ' CASCADE';
    END LOOP;
    FOR r IN (SELECT typname FROM pg_type t JOIN pg_namespace n ON t.typnamespace = n.oid
              WHERE n.nspname = 'public' AND t.typtype = 'e') LOOP
        EXECUTE 'DROP TYPE IF EXISTS ' || quote_ident(r.typname) || ' CASCADE';
    END LOOP;
END $$;
DROP TABLE IF EXISTS alembic_version CASCADE;

"""
        sql_content = drop_sql + sql_content

    cmd = [_find_pg_tool("psql"), *_get_pg_connection_args(), "--quiet", "--set", "ON_ERROR_STOP=1"]
    result = subprocess.run(  # noqa: S603 - trusted psql command
        cmd,
        env=_get_pg_env(),
        input=sql_content,
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        error(f"PostgreSQL archive restore failed: {result.stderr}")
        raise typer.Exit(code=1)

    success("PostgreSQL archive restored")


def _quote_pg_ident(identifier: str) -> str:
    return '"' + identifier.replace('"', '""') + '"'


def _build_auth_readonly_sql(mode: AuthReadOnlyMode) -> str:
    statements = ["BEGIN;"]
    quoted_function = _quote_pg_ident(_AUTH_READ_ONLY_FUNCTION)
    quoted_trigger = _quote_pg_ident(_AUTH_READ_ONLY_TRIGGER)

    if mode is AuthReadOnlyMode.freeze:
        statements.extend(
            [
                f"CREATE OR REPLACE FUNCTION {quoted_function}()",
                "RETURNS trigger",
                "LANGUAGE plpgsql",
                "AS $$",
                "BEGIN",
                "    RAISE EXCEPTION",
                "        'Sibyl legacy auth/RBAC tables are read-only after SurrealDB cutover'",
                "        USING ERRCODE = '25006';",
                "END;",
                "$$;",
            ]
        )

    for table in AUTH_ARCHIVE_TABLES:
        quoted_table = _quote_pg_ident(table)
        statements.append(f"DROP TRIGGER IF EXISTS {quoted_trigger} ON {quoted_table};")
        if mode is AuthReadOnlyMode.freeze:
            statements.extend(
                [
                    f"CREATE TRIGGER {quoted_trigger}",
                    f"BEFORE INSERT OR UPDATE OR DELETE OR TRUNCATE ON {quoted_table}",
                    f"FOR EACH STATEMENT EXECUTE FUNCTION {quoted_function}();",
                ]
            )

    if mode is AuthReadOnlyMode.unfreeze:
        statements.append(f"DROP FUNCTION IF EXISTS {quoted_function}();")

    statements.append("COMMIT;")
    return "\n".join(statements) + "\n"


def _apply_auth_readonly_sql(sql_content: str) -> None:
    cmd = [_find_pg_tool("psql"), *_get_pg_connection_args(), "--quiet", "--set", "ON_ERROR_STOP=1"]
    result = subprocess.run(  # noqa: S603 - trusted psql command
        cmd,
        env=_get_pg_env(),
        input=sql_content,
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        error(f"Legacy auth/RBAC read-only SQL failed: {result.stderr}")
        raise typer.Exit(code=1)


def _warn_if_database_dump_payload_skipped(*, archive: object, restore_database_dump: bool) -> None:
    if restore_database_dump:
        return
    archive_files = getattr(archive, "files", {})
    if POSTGRES_FILENAME not in archive_files:
        return
    warn(
        "Archive includes the database dump sidecar (postgres.sql), but database dump restore is disabled"
    )
    info(
        "Pass --restore-database-dump when you want the database dump restored before graph import"
    )


def _warn_if_auth_payload_skipped(*, archive: object, restore_auth: bool) -> None:
    archive_files = getattr(archive, "files", {})
    if AUTH_FILENAME not in archive_files:
        return
    if not restore_auth:
        warn("Archive includes auth.json, but auth restore is disabled")
        info("Pass --restore-auth when you want the Surreal auth snapshot imported")
        return
    if settings.uses_relational_auth:
        warn("Archive includes auth.json, but SIBYL_AUTH_STORE is not surreal")
        info("The active auth store is still relational; auth.json will be skipped")


def _warn_if_content_payload_skipped(*, archive: object, restore_content: bool) -> None:
    archive_files = getattr(archive, "files", {})
    if CONTENT_FILENAME not in archive_files:
        return
    if not restore_content:
        warn("Archive includes content.json, but content restore is disabled")
        info("Pass --restore-content when you want the Surreal content snapshot imported")
        return
    if settings.store != "surreal":
        warn("Archive includes content.json, but SIBYL_STORE is not surreal")
        info("The active runtime is not using Surreal; content.json will be skipped")


def _payload_decision(*, present: bool, enabled: bool, blocked: bool = False) -> str:
    if not present:
        return "missing"
    if not enabled:
        return "ignored"
    if blocked:
        return "skipped by target runtime"
    return "restore"


def _unsupported_payload_names(archive: object) -> list[str]:
    manifest = getattr(archive, "manifest", None)
    files = getattr(manifest, "files", {}) if manifest is not None else {}
    supported_kinds = {"auth", "content", "database_dump", "graph"}
    supported_names = {AUTH_FILENAME, CONTENT_FILENAME, GRAPH_FILENAME, POSTGRES_FILENAME}
    unsupported: list[str] = []
    for name, file_manifest in files.items():
        kind = getattr(file_manifest, "kind", "other")
        if name not in supported_names or kind not in supported_kinds:
            unsupported.append(name)
    return sorted(unsupported)


def _print_archive_restore_review(
    *,
    archive: object,
    source_type: ArchiveSourceType,
    target_mode: ArchiveTargetMode,
    restore_database_dump: bool,
    restore_graph: bool,
    restore_auth: bool,
    restore_content: bool,
) -> None:
    archive_files = getattr(archive, "files", {})
    manifest = getattr(archive, "manifest", None)
    manifest_source_store = getattr(manifest, "source_store", "unknown")
    source_path = getattr(archive, "source", "unknown")

    info("Archive restore review:")
    info(f"  source: {source_path}")
    info(f"  manifest source store: {manifest_source_store}")
    info(f"  declared source type: {source_type.value}")
    info(f"  target mode: {target_mode.value}")
    info(
        "  graph.json: "
        + _payload_decision(present=GRAPH_FILENAME in archive_files, enabled=restore_graph)
    )
    info(
        "  auth.json: "
        + _payload_decision(
            present=AUTH_FILENAME in archive_files,
            enabled=restore_auth,
            blocked=settings.uses_relational_auth,
        )
    )
    info(
        "  content.json: "
        + _payload_decision(
            present=CONTENT_FILENAME in archive_files,
            enabled=restore_content,
            blocked=settings.store != "surreal",
        )
    )
    info(
        "  postgres.sql: "
        + _payload_decision(
            present=POSTGRES_FILENAME in archive_files,
            enabled=restore_database_dump,
        )
    )
    unsupported = _unsupported_payload_names(archive)
    if unsupported:
        warn("Unsupported archive payloads will be ignored: " + ", ".join(unsupported))


def _require_database_dump_restore_policy(
    *,
    restore_database_dump: bool,
    source_type: ArchiveSourceType,
    target_mode: ArchiveTargetMode,
) -> None:
    if not restore_database_dump:
        return
    if source_type is not ArchiveSourceType.legacy_archive:
        error("Restoring postgres.sql requires --source-type legacy-archive")
        info("postgres.sql is a historical migration payload, not a default restore path")
        raise typer.Exit(code=1)
    if target_mode is not ArchiveTargetMode.postgres_rehearsal:
        error("Restoring postgres.sql requires --target-mode postgres-rehearsal")
        info("This prevents accidental writes to ambient PostgreSQL services")
        raise typer.Exit(code=1)


async def _replay_baseline(
    *,
    base_url: str,
    baselines_dir: Path,
    email: str,
    password: str,
    manifest_path: Path,
) -> None:
    from tools.baselines.replay import replay_all

    await replay_all(
        base_url=base_url,
        baselines_dir=baselines_dir,
        email=email,
        password=password,
        manifest_path=manifest_path,
    )


def _resolve_auth_flow_email(auth_flow_email: str) -> str:
    if auth_flow_email.strip():
        return auth_flow_email.strip()
    timestamp = datetime.now(UTC).strftime("%Y%m%d%H%M%S")
    suffix = uuid4().hex[:8]
    return f"auth-flow-{timestamp}-{suffix}@sibyl.dev"


def _normalized_auth_flow_base_url(value: str) -> str:
    parsed = urlsplit(value.rstrip("/"))
    scheme = parsed.scheme.lower()
    hostname = (parsed.hostname or "").lower()
    netloc = hostname
    if parsed.port is not None:
        netloc = f"{netloc}:{parsed.port}"
    elif parsed.netloc and "@" in parsed.netloc:
        netloc = parsed.netloc.lower()
    return urlunsplit((scheme, netloc, parsed.path.rstrip("/"), "", ""))


def _reject_same_auth_flow_compare_target(
    *,
    postgres_base_url: str,
    surreal_base_url: str,
    allow_same_base_url: bool,
) -> None:
    if allow_same_base_url:
        return
    if _normalized_auth_flow_base_url(postgres_base_url) != _normalized_auth_flow_base_url(
        surreal_base_url
    ):
        return

    error("auth-flow-compare requires distinct Postgres and Surreal base URLs")
    info("Start both runtimes and pass --postgres-base-url and --surreal-base-url")
    info("Use --allow-same-base-url only for local harness debugging")
    raise typer.Exit(code=1)


async def _run_auth_flow_gate(
    *,
    base_url: str,
    auth_flow_email: str,
    auth_flow_password: str,
    email_outbox_path: Path,
) -> None:
    from sibyl.cli.auth_flow import replay_auth_flow

    resolved_email = _resolve_auth_flow_email(auth_flow_email)
    result = await replay_auth_flow(
        base_url=base_url,
        email=resolved_email,
        password=auth_flow_password,
        email_outbox_path=email_outbox_path,
    )
    info(f"Auth flow exercised {len(result.steps)} steps for {result.primary_email}")


async def _run_auth_flow_compare(
    *,
    postgres_base_url: str,
    surreal_base_url: str,
    postgres_auth_flow_email: str,
    surreal_auth_flow_email: str,
    auth_flow_password: str,
    postgres_email_outbox_path: Path,
    surreal_email_outbox_path: Path,
) -> None:
    from sibyl.cli.auth_flow import compare_auth_flow_results, replay_auth_flow

    postgres_email = _resolve_auth_flow_email(postgres_auth_flow_email)
    surreal_email = _resolve_auth_flow_email(surreal_auth_flow_email)
    postgres_result = await replay_auth_flow(
        base_url=postgres_base_url,
        email=postgres_email,
        password=auth_flow_password,
        email_outbox_path=postgres_email_outbox_path,
    )
    surreal_result = await replay_auth_flow(
        base_url=surreal_base_url,
        email=surreal_email,
        password=auth_flow_password,
        email_outbox_path=surreal_email_outbox_path,
    )
    compare_auth_flow_results(
        left_label="postgres",
        left=postgres_result,
        right_label="surreal",
        right=surreal_result,
    )
    info(
        "Auth flow semantic equivalence passed "
        f"postgres_steps={len(postgres_result.steps)} surreal_steps={len(surreal_result.steps)}"
    )


def _run_moon_task(task: list[str], *, label: str) -> None:
    moon = shutil.which("moon")
    if moon is None:
        error("moon executable not found in PATH")
        raise typer.Exit(code=1)

    result = subprocess.run(  # noqa: S603 - trusted moon task invocation
        [moon, "run", *task],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        error(f"{label} failed")
        if result.stdout.strip():
            console.print(f"[dim]{result.stdout.strip()}[/dim]")
        if result.stderr.strip():
            console.print(f"[dim]{result.stderr.strip()}[/dim]")
        raise typer.Exit(code=1)

    success(f"{label} passed")


def _print_cutover_plan(
    *,
    run_auth_flow: bool,
    run_baseline: bool,
    run_bench_live_smoke: bool,
    run_bench_live: bool,
    reopen_writes: bool,
    manifest_path: Path,
) -> None:
    info("Cutover plan:")
    steps = [
        "Confirm legacy writes are frozen",
        "Import archive into the Surreal runtime",
        "Verify imported counts and sample entities",
    ]
    if run_auth_flow:
        steps.append("Run auth-flow acceptance harness")
    if run_baseline:
        steps.append("Replay deterministic runtime baseline")
    if run_bench_live_smoke:
        steps.append("Run bench-live-smoke acceptance check")
    if run_bench_live:
        steps.append("Run bench-live artifact capture")
    steps.append("Operator freezes legacy auth/RBAC writes before reopening")
    if reopen_writes:
        steps.append("Reopen writes on SurrealDB after operator acknowledgment")
    for index, step in enumerate(steps, start=1):
        info(f"  {index}. {step}")
    if run_baseline and not manifest_path.exists():
        warn(f"Baseline manifest not found yet: {manifest_path}")


def _print_writes_remain_frozen_notice() -> None:
    warn("Writes remain frozen. Rollback is still supported at this point.")
    info("Freeze legacy auth/RBAC writes before reopening:")
    typer.echo("  moon run auth-readonly -- --mode freeze --apply --yes")
    info(
        "Rerun with --reopen-writes --acknowledge-no-instant-rollback "
        "after final operator sign-off."
    )


def _require_cutover_surreal_runtime() -> None:
    if settings.store != "surreal":
        error("Cutover must run with SIBYL_STORE=surreal on the target runtime")
        raise typer.Exit(code=1)
    if settings.uses_relational_auth:
        error("Cutover must run with SIBYL_AUTH_STORE=surreal on the target runtime")
        info("Use import or rehearse for mixed-mode debugging before the cutover gate")
        raise typer.Exit(code=1)


async def _run_cutover_acceptance(
    *,
    archive: object,
    organization_id: str,
    sample_size: int,
    run_auth_flow: bool,
    auth_flow_email: str,
    auth_flow_password: str,
    email_outbox_path: Path,
    run_baseline: bool,
    base_url: str,
    baselines_dir: Path,
    email: str,
    password: str,
    manifest_path: Path,
    run_bench_live_smoke: bool,
    run_bench_live: bool,
    bench_label: str,
) -> None:
    result = await verify_graph_archive(
        archive,
        organization_id=organization_id,
        sample_size=sample_size,
    )
    _print_verify_summary(result)
    success("Archive verification passed")

    if run_auth_flow:
        info(f"Running auth flow harness against {base_url}...")
        await _run_auth_flow_gate(
            base_url=base_url,
            auth_flow_email=auth_flow_email,
            auth_flow_password=auth_flow_password,
            email_outbox_path=email_outbox_path,
        )
        success("Auth flow harness passed")

    if run_baseline:
        info(f"Replaying deterministic baseline against {base_url}...")
        await _replay_baseline(
            base_url=base_url,
            baselines_dir=baselines_dir,
            email=email,
            password=password,
            manifest_path=manifest_path,
        )
        success("Baseline replay passed")

    if run_bench_live_smoke:
        info("Running bench-live-smoke acceptance check...")
        _run_moon_task(["bench-live-smoke"], label="bench-live-smoke")

    if run_bench_live:
        info("Running bench-live acceptance capture...")
        _run_moon_task(
            [
                "bench-live",
                "--",
                "--label",
                bench_label,
                "--metadata",
                "store=surreal",
                "--metadata",
                "mode=cutover",
            ],
            label="bench-live",
        )

    success("Acceptance suite passed while writes remain frozen")


def _load_graph_export(org_id: str) -> tuple[dict[str, object], bytes]:
    from dataclasses import asdict

    from sibyl_core.tools.admin import create_backup

    @run_async
    async def _export() -> tuple[dict[str, object], bytes]:
        result = await create_backup(organization_id=org_id)
        if not result.success or result.backup_data is None:
            msg = result.message or "graph export failed"
            raise RuntimeError(msg)
        payload = asdict(result.backup_data)
        encoded = json.dumps(payload, indent=2, default=str).encode("utf-8")
        return payload, encoded

    return _export()


def _load_auth_export() -> tuple[dict[str, object], bytes]:
    @run_async
    async def _export() -> tuple[dict[str, object], bytes]:
        payload = await export_auth_archive_payload()
        encoded = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
        return payload, encoded

    return _export()


def _load_content_export() -> tuple[dict[str, object], bytes]:
    @run_async
    async def _export() -> tuple[dict[str, object], bytes]:
        payload = await export_content_archive_payload()
        encoded = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8")
        return payload, encoded

    return _export()


def _load_runtime_exports(
    *,
    include_auth: bool,
    include_content: bool,
) -> tuple[
    tuple[dict[str, object], bytes] | None,
    tuple[dict[str, object], bytes] | None,
]:
    @run_async
    async def _export() -> tuple[
        tuple[dict[str, object], bytes] | None,
        tuple[dict[str, object], bytes] | None,
    ]:
        auth_export: tuple[dict[str, object], bytes] | None = None
        content_export: tuple[dict[str, object], bytes] | None = None

        if include_auth:
            payload = await export_auth_archive_payload()
            auth_export = (
                payload,
                json.dumps(payload, indent=2, sort_keys=True).encode("utf-8"),
            )

        if include_content:
            payload = await export_content_archive_payload()
            content_export = (
                payload,
                json.dumps(payload, indent=2, sort_keys=True).encode("utf-8"),
            )

        return auth_export, content_export

    return _export()


def _restore_auth_payload(payload: dict[str, object], *, clean: bool) -> bool:
    @run_async
    async def _restore() -> bool:
        try:
            result = await restore_auth_archive_payload(payload, clean=clean)
            if result.success:
                success(
                    f"  Auth restored: {result.rows_restored} rows across {result.tables_restored} tables"
                )
            else:
                warn(f"  Auth restore completed with errors: {len(result.errors)}")
                for issue in result.errors[:10]:
                    console.print(f"  [dim]{issue}[/dim]")
            return result.success
        except Exception as exc:
            error(f"  Auth restore failed: {exc}")
            return False

    return _restore()


def _restore_content_payload(payload: dict[str, object], *, clean: bool) -> bool:
    @run_async
    async def _restore() -> bool:
        try:
            result = await restore_content_archive_payload(payload, clean=clean)
            if result.success:
                success(
                    "  Content restored: "
                    f"{result.rows_restored} rows across {result.tables_restored} tables"
                )
            else:
                warn(f"  Content restore completed with errors: {len(result.errors)}")
                for issue in result.errors[:10]:
                    console.print(f"  [dim]{issue}[/dim]")
            return result.success
        except Exception as exc:
            error(f"  Content restore failed: {exc}")
            return False

    return _restore()


def _restore_graph_payload(backup_dict: dict[str, object], org_id: str, *, clean: bool) -> bool:
    """Restore graph data from a decoded archive payload."""

    @run_async
    async def _restore() -> bool:
        from sibyl.cli.db import _coerce_graph_backup_data, _prepare_graph_runtime_async
        from sibyl_core.tools.admin import restore_backup

        try:
            backup_data = _coerce_graph_backup_data(backup_dict, org_id)
            await _prepare_graph_runtime_async(org_id, clean=clean)

            result = await restore_backup(
                backup_data,
                organization_id=org_id,
                skip_existing=not clean,
            )

            if result.success:
                success(
                    f"  Graph restored: {result.entities_restored} entities, "
                    f"{result.relationships_restored} relationships"
                )
            else:
                warn(f"  Graph restore completed with errors: {len(result.errors)}")

            return result.success
        except Exception as exc:
            error(f"  Graph restore failed: {exc}")
            return False

    return _restore()


def _bootstrap_surreal_runtimes(*, clean: bool) -> None:
    """Bootstrap SCHEMAFULL tables + indexes in surreal auth and content namespaces.

    Runs unconditionally during import so namespaces are queryable even when the
    archive carries no auth or content payload to restore. Without this, surreal
    lazily creates SCHEMALESS tables on first insert and search surfaces fail
    with "table does not exist" or return zero results.
    """
    from sibyl.persistence.surreal.auth import build_surreal_auth_client
    from sibyl.persistence.surreal.content import build_surreal_content_client
    from sibyl_core.backends.surreal import bootstrap_auth_schema, bootstrap_content_schema

    @run_async
    async def _bootstrap() -> None:
        if not settings.uses_relational_auth:
            info("Bootstrapping Surreal auth schema...")
            await bootstrap_auth_schema(build_surreal_auth_client(), reset=clean)
        if settings.store == "surreal":
            info("Bootstrapping Surreal content schema...")
            await bootstrap_content_schema(build_surreal_content_client(), reset=clean)

    _bootstrap()


@app.command("check")
def check_archive(
    source: Annotated[Path, typer.Argument(help="Archive .tar.gz or directory to inspect")],
) -> None:
    """Validate an archive and print its manifest summary."""
    archive = _load_valid_archive(source)

    manifest = archive.manifest
    info(f"Archive: {source}")
    info(f"Version: {manifest.version}")
    info(f"Source store: {manifest.source_store}")
    info(f"Organization: {manifest.organization_id or 'unknown'}")
    info(f"Created: {manifest.created_at or 'unknown'}")

    for name, file_manifest in sorted(manifest.files.items()):
        info(
            f"  {name} ({file_manifest.kind}, {file_manifest.size_bytes} bytes, {file_manifest.sha256[:12]})"
        )

    graph_payload = graph_payload_from_archive(archive)
    if graph_payload is not None:
        effective_counts = effective_graph_counts(graph_payload)
        info(
            "Graph counts: "
            f"{graph_payload.get('entity_count', len(graph_payload.get('entities', [])))} entities, "
            f"{graph_payload.get('relationship_count', len(graph_payload.get('relationships', [])))} relationships, "
            f"{graph_payload.get('episode_count', len(graph_payload.get('episodes', [])))} episodes, "
            f"{graph_payload.get('mention_count', len(graph_payload.get('mentions', [])))} mentions"
        )
        if effective_counts["relationship_count"] != int(
            graph_payload.get("relationship_count", len(graph_payload.get("relationships", [])))
        ) or effective_counts["mention_count"] != int(
            graph_payload.get("mention_count", len(graph_payload.get("mentions", [])))
        ):
            info(
                "Effective restore counts: "
                f"{effective_counts['entity_count']} entities, "
                f"{effective_counts['relationship_count']} relationships, "
                f"{effective_counts['episode_count']} episodes, "
                f"{effective_counts['mention_count']} mentions"
            )

    auth_payload = auth_payload_from_archive(archive)
    if auth_payload is not None:
        row_counts = auth_payload.get("row_counts", {})
        total_rows = int(auth_payload.get("total_rows") or sum(int(v) for v in row_counts.values()))
        info(f"Auth tables: {len(row_counts)} tables, {total_rows} rows")

    content_payload = content_payload_from_archive(archive)
    if content_payload is not None:
        row_counts = content_payload.get("row_counts", {})
        total_rows = int(
            content_payload.get("total_rows") or sum(int(v) for v in row_counts.values())
        )
        info(f"Content tables: {len(row_counts)} tables, {total_rows} rows")
    success("Archive validation passed")


@app.command("cloud-env")
def cloud_env(
    output: Annotated[
        Path,
        typer.Option("--output", "-o", help="Env file to write"),
    ] = DEFAULT_CLOUD_ENV_PATH,
    surreal_url: Annotated[
        str,
        typer.Option("--surreal-url", help="Surreal Cloud endpoint, usually wss://.../rpc"),
    ] = "",
    surreal_username: Annotated[
        str,
        typer.Option("--surreal-username", help="SurrealDB root/system username"),
    ] = "root",
    surreal_password: Annotated[
        str,
        typer.Option("--surreal-password", help="SurrealDB root/system password"),
    ] = "",
    public_url: Annotated[
        str,
        typer.Option("--public-url", help="Public Sibyl URL for callbacks and cookies"),
    ] = "",
    coordination_backend: Annotated[
        CloudCoordinationBackend,
        typer.Option(
            "--coordination-backend",
            help="Use redis for multiple active Sibyl API instances",
        ),
    ] = CloudCoordinationBackend.local,
    redis_host: Annotated[
        str,
        typer.Option("--redis-host", help="Redis/Valkey host when coordination is redis"),
    ] = "127.0.0.1",
    redis_port: Annotated[
        int,
        typer.Option("--redis-port", help="Redis/Valkey port when coordination is redis"),
    ] = 6381,
    redis_password: Annotated[
        str,
        typer.Option("--redis-password", help="Redis/Valkey password when coordination is redis"),
    ] = "",
    canonical_org_id: Annotated[
        str,
        typer.Option("--canonical-org-id", help="Canonical merged organization UUID"),
    ] = "",
    print_only: Annotated[
        bool,
        typer.Option("--print", help="Print env content instead of writing a file"),
    ] = False,
    force: Annotated[
        bool,
        typer.Option("--force", help="Overwrite an existing env file"),
    ] = False,
) -> None:
    """Generate a Surreal Cloud env file and consolidation checklist."""
    resolved_org_id = canonical_org_id or str(uuid4())
    payload = _render_cloud_env(
        surreal_url=surreal_url,
        surreal_username=surreal_username,
        surreal_password=surreal_password,
        public_url=public_url,
        coordination_backend=coordination_backend,
        redis_host=redis_host,
        redis_port=redis_port,
        redis_password=redis_password,
        canonical_org_id=resolved_org_id,
    )

    if print_only:
        console.print(payload)
    else:
        if output.exists() and not force:
            error(f"{output} already exists; pass --force to overwrite it")
            raise typer.Exit(code=1)
        output.write_text(payload, encoding="utf-8")
        output.chmod(0o600)
        success(f"Cloud env written to {output}")

    if not surreal_url:
        warn("Fill in SIBYL_SURREAL_URL with the Surreal Cloud WebSocket endpoint")
    if not surreal_password:
        warn("Fill in SIBYL_SURREAL_PASSWORD before running migration commands")
    _print_cloud_env_next_steps(output, resolved_org_id)


@app.command("merge")
def merge_archive_sources(
    sources: Annotated[
        list[Path],
        typer.Argument(help="Archive .tar.gz files or directories to merge"),
    ],
    output: Annotated[
        Path,
        typer.Option("--output", "-o", help="Output merged archive path"),
    ] = Path("sibyl_merged_migration.tar.gz"),
    canonical_org_id: Annotated[
        str,
        typer.Option("--canonical-org-id", help="Canonical organization UUID for merged data"),
    ] = "",
    canonical_org_name: Annotated[
        str,
        typer.Option(
            "--canonical-org-name",
            help="Canonical organization display name; defaults to the first source org name",
        ),
    ] = "",
    canonical_org_slug: Annotated[
        str,
        typer.Option(
            "--canonical-org-slug",
            help="Canonical organization slug; defaults to the first source org slug",
        ),
    ] = "",
    entity_collision_policy: Annotated[
        str,
        typer.Option(
            "--entity-collision-policy",
            help="Entity merge policy: merge-by-type-name or keep-all",
        ),
    ] = EntityCollisionPolicy.MERGE_BY_TYPE_NAME.value,
    drop_volatile_auth: Annotated[
        bool,
        typer.Option(
            "--drop-volatile-auth/--keep-volatile-auth",
            help="Drop sessions, reset tokens, login history, and device auth requests",
        ),
    ] = True,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Build and validate the merged archive without writing it"),
    ] = False,
) -> None:
    """Merge multiple migration archives into one canonical organization archive."""
    if not sources:
        error("At least one source archive is required")
        raise typer.Exit(code=1)
    if not canonical_org_id.strip():
        error("--canonical-org-id is required")
        raise typer.Exit(code=1)

    try:
        collision_policy = EntityCollisionPolicy(entity_collision_policy)
    except ValueError as exc:
        allowed = ", ".join(policy.value for policy in EntityCollisionPolicy)
        error(f"Invalid --entity-collision-policy. Expected one of: {allowed}")
        raise typer.Exit(code=1) from exc

    archives = [_load_valid_archive(source) for source in sources]
    try:
        result = merge_archives(
            archives,
            options=ArchiveMergeOptions(
                canonical_org_id=canonical_org_id,
                canonical_org_name=canonical_org_name,
                canonical_org_slug=canonical_org_slug,
                entity_collision_policy=collision_policy,
                drop_volatile_auth=drop_volatile_auth,
            ),
        )
    except ValueError as exc:
        error(str(exc))
        raise typer.Exit(code=1) from exc

    errors = validate_archive(result.archive)
    if errors:
        for issue in errors:
            warn(issue)
        error("Merged archive validation failed")
        raise typer.Exit(code=1)

    info(f"Merged {result.source_count} archive(s)")
    info(f"Source organizations: {', '.join(result.source_org_ids) or 'unknown'}")
    if result.graph_counts:
        info(
            "Graph counts: "
            f"{result.graph_counts['entities']} entities, "
            f"{result.graph_counts['relationships']} relationships, "
            f"{result.graph_counts['episodes']} episodes, "
            f"{result.graph_counts['mentions']} mentions"
        )
        if result.entity_alias_count:
            info(f"Merged entity aliases: {result.entity_alias_count}")
    if result.user_alias_count:
        info(f"Merged user aliases: {result.user_alias_count}")
    if result.auth_row_counts:
        info(f"Auth rows: {sum(result.auth_row_counts.values())}")
    if result.content_row_counts:
        info(f"Content rows: {sum(result.content_row_counts.values())}")
    if dry_run:
        success("Merged archive dry run passed")
        return

    write_archive(output, manifest=result.archive.manifest, files=result.archive.files)
    success(f"Merged archive written to {output}")


@app.command("export")
def export_archive(
    output: Annotated[
        Path,
        typer.Option("--output", "-o", help="Output archive path"),
    ] = Path("sibyl_migration.tar.gz"),
    org_id: Annotated[
        str,
        typer.Option("--org-id", help="Organization UUID for graph export"),
    ] = "",
    include_database_dump: Annotated[
        bool,
        typer.Option(
            "--include-database-dump/--no-include-database-dump",
            "--include-postgres/--no-include-postgres",
            help="Deprecated no-op; retained postgres.sql payloads are import-only",
        ),
    ] = False,
    include_graph: Annotated[
        bool,
        typer.Option("--include-graph/--skip-graph", help="Include graph runtime export"),
    ] = True,
    include_auth: Annotated[
        bool,
        typer.Option("--include-auth/--skip-auth", help="Include auth/RBAC snapshot"),
    ] = True,
    include_content: Annotated[
        bool,
        typer.Option(
            "--include-content/--skip-content", help="Include content/operations snapshot"
        ),
    ] = True,
) -> None:
    """Export a manifest-driven migration archive from the active store."""
    del include_database_dump

    if not include_graph and not include_auth and not include_content:
        error(
            "Select at least one supported payload: "
            "--include-graph, --include-auth, or --include-content"
        )
        raise typer.Exit(code=1)

    if include_graph:
        org_id = _resolve_export_org_id(org_id)

    files: dict[str, bytes] = {}
    file_metadata: dict[str, dict[str, object]] = {}
    archive_metadata: dict[str, object] = {}

    if include_graph:
        info(f"Exporting graph runtime from {settings.store} store...")
        graph_payload, graph_bytes = _load_graph_export(org_id)
        effective_counts = effective_graph_counts(graph_payload)
        files[GRAPH_FILENAME] = graph_bytes
        file_metadata[GRAPH_FILENAME] = {
            "kind": "graph",
            "entity_count": int(graph_payload.get("entity_count", 0)),
            "relationship_count": int(graph_payload.get("relationship_count", 0)),
            "episode_count": int(graph_payload.get("episode_count", 0)),
            "mention_count": int(graph_payload.get("mention_count", 0)),
            "effective_relationship_count": effective_counts["relationship_count"],
            "effective_mention_count": effective_counts["mention_count"],
        }
        archive_metadata["graph_entity_count"] = int(graph_payload.get("entity_count", 0))
        archive_metadata["graph_relationship_count"] = int(
            graph_payload.get("relationship_count", 0)
        )
        archive_metadata["graph_episode_count"] = int(graph_payload.get("episode_count", 0))
        archive_metadata["graph_mention_count"] = int(graph_payload.get("mention_count", 0))
        archive_metadata["graph_effective_relationship_count"] = effective_counts[
            "relationship_count"
        ]
        archive_metadata["graph_effective_mention_count"] = effective_counts["mention_count"]

    auth_export: tuple[dict[str, object], bytes] | None = None
    content_export: tuple[dict[str, object], bytes] | None = None
    if include_auth and include_content:
        auth_export, content_export = _load_runtime_exports(
            include_auth=True,
            include_content=True,
        )

    if include_auth:
        info(f"Exporting auth/RBAC snapshot from {settings.auth_store} auth runtime...")
        if auth_export is None:
            auth_export = _load_auth_export()
        auth_payload, auth_bytes = auth_export
        row_counts = dict(auth_payload.get("row_counts", {}))
        files[AUTH_FILENAME] = auth_bytes
        file_metadata[AUTH_FILENAME] = {
            "kind": "auth",
            "table_count": len(row_counts),
            "total_rows": int(auth_payload.get("total_rows", 0)),
        }
        archive_metadata["auth_table_count"] = len(row_counts)
        archive_metadata["auth_total_rows"] = int(auth_payload.get("total_rows", 0))

    if include_content:
        info(f"Exporting content/operations snapshot from {settings.store} content runtime...")
        if content_export is None:
            content_export = _load_content_export()
        content_payload, content_bytes = content_export
        row_counts = dict(content_payload.get("row_counts", {}))
        files[CONTENT_FILENAME] = content_bytes
        file_metadata[CONTENT_FILENAME] = {
            "kind": "content",
            "table_count": len(row_counts),
            "total_rows": int(content_payload.get("total_rows", 0)),
        }
        archive_metadata["content_table_count"] = len(row_counts)
        archive_metadata["content_total_rows"] = int(content_payload.get("total_rows", 0))

    manifest = build_manifest(
        organization_id=org_id,
        source_store=settings.store,
        files=files,
        file_metadata=file_metadata,
        metadata=archive_metadata,
    )
    write_archive(output, manifest=manifest, files=files)

    success(f"Migration archive written to {output}")


@app.command("import")
def import_archive(
    source: Annotated[Path, typer.Argument(help="Archive .tar.gz or directory to import")],
    source_type: Annotated[
        ArchiveSourceType,
        typer.Option("--source-type", help="Required archive source type for restore policy"),
    ],
    target_mode: Annotated[
        ArchiveTargetMode,
        typer.Option("--target-mode", help="Required restore target mode for policy checks"),
    ],
    org_id: Annotated[
        str,
        typer.Option("--org-id", help="Organization UUID override"),
    ] = "",
    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Print the restore review without writing data"),
    ] = False,
    clean: Annotated[
        bool,
        typer.Option("--clean", help="Clear the target graph before import"),
    ] = False,
    restore_database_dump: Annotated[
        bool,
        typer.Option(
            "--restore-database-dump",
            "--restore-postgres",
            help="Restore the database dump sidecar (postgres.sql) before graph import",
        ),
    ] = False,
    restore_graph: Annotated[
        bool,
        typer.Option("--restore-graph/--skip-graph", help="Restore graph payload"),
    ] = True,
    restore_auth: Annotated[
        bool,
        typer.Option(
            "--restore-auth/--skip-auth", help="Restore auth payload into Surreal auth storage"
        ),
    ] = True,
    restore_content: Annotated[
        bool,
        typer.Option(
            "--restore-content/--skip-content",
            help="Restore content payload into Surreal content storage",
        ),
    ] = True,
) -> None:
    """Import a manifest archive into the active store."""
    archive = _load_valid_archive(source)
    effective_org_id = (
        _resolve_org_id(org_id, archive.manifest.organization_id) if restore_graph else ""
    )

    _require_database_dump_restore_policy(
        restore_database_dump=restore_database_dump,
        source_type=source_type,
        target_mode=target_mode,
    )
    if restore_database_dump and POSTGRES_FILENAME not in archive.files:
        error("Archive does not contain the database dump sidecar (postgres.sql)")
        raise typer.Exit(code=1)

    if restore_graph and GRAPH_FILENAME not in archive.files:
        error("Archive does not contain graph.json")
        raise typer.Exit(code=1)
    _warn_if_database_dump_payload_skipped(
        archive=archive,
        restore_database_dump=restore_database_dump,
    )
    _warn_if_auth_payload_skipped(archive=archive, restore_auth=restore_auth)
    _warn_if_content_payload_skipped(archive=archive, restore_content=restore_content)
    _print_archive_restore_review(
        archive=archive,
        source_type=source_type,
        target_mode=target_mode,
        restore_database_dump=restore_database_dump,
        restore_graph=restore_graph,
        restore_auth=restore_auth,
        restore_content=restore_content,
    )

    if dry_run:
        success("Archive import dry run complete")
        return

    if not yes:
        warn("This will import archive data into the active runtime.")
        if not typer.confirm("Continue?"):
            info("Cancelled")
            return

    if restore_database_dump:
        info("Restoring database dump sidecar...")
        _restore_pg_sql(archive.files[POSTGRES_FILENAME].decode("utf-8"), clean)

    _bootstrap_surreal_runtimes(clean=clean)

    if restore_auth and AUTH_FILENAME in archive.files and not settings.uses_relational_auth:
        info("Restoring auth payload into Surreal auth storage...")
        payload = auth_payload_from_archive(archive)
        if payload is None or not _restore_auth_payload(payload, clean=clean):
            error("Auth import failed")
            raise typer.Exit(code=1)

    if restore_content and CONTENT_FILENAME in archive.files and settings.store == "surreal":
        info("Restoring content payload into Surreal content storage...")
        payload = content_payload_from_archive(archive)
        if payload is None or not _restore_content_payload(payload, clean=clean):
            error("Content import failed")
            raise typer.Exit(code=1)

    if restore_graph:
        info(f"Restoring graph payload into {settings.store} store...")
        payload = json.loads(archive.files[GRAPH_FILENAME].decode("utf-8"))
        if not _restore_graph_payload(payload, effective_org_id, clean=clean):
            error("Graph import failed")
            raise typer.Exit(code=1)

    success("Archive import complete")


@app.command("verify")
def verify_archive(
    source: Annotated[Path, typer.Argument(help="Archive .tar.gz or directory to verify")],
    org_id: Annotated[
        str,
        typer.Option("--org-id", help="Organization UUID override"),
    ] = "",
    sample_size: Annotated[
        int,
        typer.Option("--sample-size", help="How many entity IDs to spot-check"),
    ] = 10,
) -> None:
    """Verify an archive against the active runtime."""
    archive = _load_valid_archive(source)
    effective_org_id = _resolve_org_id(org_id, archive.manifest.organization_id)

    @run_async
    async def _verify() -> None:
        result = await verify_graph_archive(
            archive,
            organization_id=effective_org_id,
            sample_size=sample_size,
        )
        _print_verify_summary(result)
        success("Archive verification passed")

    _verify()


@app.command("auth-flow")
def auth_flow(
    base_url: Annotated[
        str,
        typer.Option("--base-url", help="Base URL for auth-flow replay"),
    ] = DEFAULT_REHEARSAL_BASE_URL,
    auth_flow_email: Annotated[
        str,
        typer.Option(
            "--auth-flow-email",
            help="Auth-flow user email; generated when omitted",
        ),
    ] = "",
    auth_flow_password: Annotated[
        str,
        typer.Option("--auth-flow-password", help="Auth-flow user password"),
    ] = DEFAULT_AUTH_FLOW_PASSWORD,
    email_outbox_path: Annotated[
        Path,
        typer.Option(
            "--email-outbox-path",
            help="JSONL outbox path used to consume the password reset token",
        ),
    ] = DEFAULT_AUTH_FLOW_EMAIL_OUTBOX,
) -> None:
    """Run the deterministic auth-flow acceptance harness against a live API."""

    @run_async
    async def _auth_flow() -> None:
        info(f"Running auth flow harness against {base_url}...")
        await _run_auth_flow_gate(
            base_url=base_url,
            auth_flow_email=auth_flow_email,
            auth_flow_password=auth_flow_password,
            email_outbox_path=email_outbox_path,
        )
        success("Auth flow harness passed")

    _auth_flow()


@app.command("auth-flow-compare")
def auth_flow_compare(
    postgres_base_url: Annotated[
        str,
        typer.Option("--postgres-base-url", help="Base URL for the Postgres auth runtime"),
    ] = DEFAULT_REHEARSAL_BASE_URL,
    surreal_base_url: Annotated[
        str,
        typer.Option("--surreal-base-url", help="Base URL for the Surreal auth runtime"),
    ] = DEFAULT_REHEARSAL_BASE_URL,
    postgres_auth_flow_email: Annotated[
        str,
        typer.Option(
            "--postgres-auth-flow-email",
            help="Postgres replay user email; generated when omitted",
        ),
    ] = "",
    surreal_auth_flow_email: Annotated[
        str,
        typer.Option(
            "--surreal-auth-flow-email",
            help="Surreal replay user email; generated when omitted",
        ),
    ] = "",
    auth_flow_password: Annotated[
        str,
        typer.Option("--auth-flow-password", help="Auth-flow user password"),
    ] = DEFAULT_AUTH_FLOW_PASSWORD,
    postgres_email_outbox_path: Annotated[
        Path,
        typer.Option(
            "--postgres-email-outbox-path",
            help="JSONL outbox path for the Postgres replay reset token",
        ),
    ] = DEFAULT_AUTH_FLOW_EMAIL_OUTBOX,
    surreal_email_outbox_path: Annotated[
        Path,
        typer.Option(
            "--surreal-email-outbox-path",
            help="JSONL outbox path for the Surreal replay reset token",
        ),
    ] = DEFAULT_AUTH_FLOW_EMAIL_OUTBOX,
    allow_same_base_url: Annotated[
        bool,
        typer.Option(
            "--allow-same-base-url",
            help="Allow comparing one API to itself for local harness debugging",
        ),
    ] = False,
) -> None:
    """Compare normalized auth-flow semantics across Postgres and Surreal runtimes."""

    @run_async
    async def _auth_flow_compare() -> None:
        _reject_same_auth_flow_compare_target(
            postgres_base_url=postgres_base_url,
            surreal_base_url=surreal_base_url,
            allow_same_base_url=allow_same_base_url,
        )
        info("Running auth flow semantic comparison...")
        await _run_auth_flow_compare(
            postgres_base_url=postgres_base_url,
            surreal_base_url=surreal_base_url,
            postgres_auth_flow_email=postgres_auth_flow_email,
            surreal_auth_flow_email=surreal_auth_flow_email,
            auth_flow_password=auth_flow_password,
            postgres_email_outbox_path=postgres_email_outbox_path,
            surreal_email_outbox_path=surreal_email_outbox_path,
        )
        success("Auth flow semantic comparison passed")

    _auth_flow_compare()


@app.command("auth-readonly")
def auth_readonly(
    mode: Annotated[
        AuthReadOnlyMode,
        typer.Option(
            "--mode",
            help="Generate SQL to freeze or unfreeze legacy auth/RBAC writes",
        ),
    ] = AuthReadOnlyMode.freeze,
    output: Annotated[
        Path | None,
        typer.Option("--output", "-o", help="Write SQL to a file instead of stdout"),
    ] = None,
    apply_sql: Annotated[
        bool,
        typer.Option("--apply", help="Apply the SQL to the configured PostgreSQL database"),
    ] = False,
    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
) -> None:
    """Generate or apply the legacy auth/RBAC read-only guard."""
    sql_content = _build_auth_readonly_sql(mode)

    if output is not None:
        output.write_text(sql_content, encoding="utf-8")
        success(f"Auth/RBAC {mode.value} SQL written to {output}")
    else:
        typer.echo(sql_content, nl=False)

    if not apply_sql:
        if output is not None:
            info("Rerun with --apply during the cutover write-freeze when ready.")
        return

    if not yes:
        warn("This changes write behavior on the configured PostgreSQL database.")
        if not typer.confirm("Continue?"):
            info("Cancelled")
            return

    _apply_auth_readonly_sql(sql_content)
    success(f"Legacy auth/RBAC {mode.value} guard applied")


@app.command("rehearse")
def rehearse_archive(
    source: Annotated[Path, typer.Argument(help="Archive .tar.gz or directory to rehearse")],
    source_type: Annotated[
        ArchiveSourceType,
        typer.Option("--source-type", help="Required archive source type for restore policy"),
    ],
    target_mode: Annotated[
        ArchiveTargetMode,
        typer.Option("--target-mode", help="Required restore target mode for policy checks"),
    ],
    org_id: Annotated[
        str,
        typer.Option("--org-id", help="Organization UUID override"),
    ] = "",
    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Print the rehearsal restore review without writing data"),
    ] = False,
    clean: Annotated[
        bool,
        typer.Option("--clean", help="Clear the target graph before import"),
    ] = True,
    restore_database_dump: Annotated[
        bool,
        typer.Option(
            "--restore-database-dump",
            "--restore-postgres",
            help="Restore the database dump sidecar (postgres.sql) before graph import",
        ),
    ] = False,
    restore_auth: Annotated[
        bool,
        typer.Option(
            "--restore-auth/--skip-auth", help="Restore auth payload into Surreal auth storage"
        ),
    ] = True,
    restore_content: Annotated[
        bool,
        typer.Option(
            "--restore-content/--skip-content",
            help="Restore content payload into Surreal content storage",
        ),
    ] = True,
    run_baseline: Annotated[
        bool,
        typer.Option(
            "--run-baseline/--skip-baseline", help="Replay the deterministic runtime baseline"
        ),
    ] = True,
    run_auth_flow: Annotated[
        bool,
        typer.Option(
            "--run-auth-flow/--skip-auth-flow", help="Run the auth-flow acceptance harness"
        ),
    ] = True,
    base_url: Annotated[
        str,
        typer.Option("--base-url", help="Base URL for auth-flow and baseline replay"),
    ] = DEFAULT_REHEARSAL_BASE_URL,
    auth_flow_email: Annotated[
        str,
        typer.Option(
            "--auth-flow-email",
            help="Auth-flow user email; generated when omitted",
        ),
    ] = "",
    auth_flow_password: Annotated[
        str,
        typer.Option("--auth-flow-password", help="Auth-flow user password"),
    ] = DEFAULT_AUTH_FLOW_PASSWORD,
    email_outbox_path: Annotated[
        Path,
        typer.Option(
            "--email-outbox-path",
            help="JSONL outbox path used to consume the password reset token",
        ),
    ] = DEFAULT_AUTH_FLOW_EMAIL_OUTBOX,
    baselines_dir: Annotated[
        Path,
        typer.Option("--baselines-dir", help="Directory containing baseline case files"),
    ] = DEFAULT_REHEARSAL_BASELINES_DIR,
    manifest_path: Annotated[
        Path,
        typer.Option(
            "--manifest-path", help="Runtime baseline manifest from `moon run baseline-seed`"
        ),
    ] = DEFAULT_REHEARSAL_MANIFEST,
    email: Annotated[
        str,
        typer.Option("--email", help="Baseline user email"),
    ] = DEFAULT_REHEARSAL_EMAIL,
    password: Annotated[
        str,
        typer.Option("--password", help="Baseline user password"),
    ] = DEFAULT_REHEARSAL_PASSWORD,
    sample_size: Annotated[
        int,
        typer.Option("--sample-size", help="How many entity IDs to spot-check during verify"),
    ] = 10,
) -> None:
    """Run an import + verify + baseline smoke rehearsal against the active store."""
    archive = _load_valid_archive(source)
    effective_org_id = _resolve_org_id(org_id, archive.manifest.organization_id)

    _require_database_dump_restore_policy(
        restore_database_dump=restore_database_dump,
        source_type=source_type,
        target_mode=target_mode,
    )
    if restore_database_dump and POSTGRES_FILENAME not in archive.files:
        error("Archive does not contain the database dump sidecar (postgres.sql)")
        raise typer.Exit(code=1)
    if GRAPH_FILENAME not in archive.files:
        error("Archive does not contain graph.json")
        raise typer.Exit(code=1)
    _warn_if_database_dump_payload_skipped(
        archive=archive,
        restore_database_dump=restore_database_dump,
    )
    _warn_if_auth_payload_skipped(archive=archive, restore_auth=restore_auth)
    _warn_if_content_payload_skipped(archive=archive, restore_content=restore_content)
    _print_archive_restore_review(
        archive=archive,
        source_type=source_type,
        target_mode=target_mode,
        restore_database_dump=restore_database_dump,
        restore_graph=True,
        restore_auth=restore_auth,
        restore_content=restore_content,
    )
    if dry_run:
        success("Migration rehearsal dry run complete")
        return

    if run_baseline and not manifest_path.exists():
        error(f"Baseline manifest not found: {manifest_path}")
        raise typer.Exit(code=1)

    if not yes:
        warn("This will import archive data and run rehearsal checks against the active runtime.")
        if not typer.confirm("Continue?"):
            info("Cancelled")
            return

    if restore_database_dump:
        info("Restoring database dump sidecar...")
        _restore_pg_sql(archive.files[POSTGRES_FILENAME].decode("utf-8"), clean)

    _bootstrap_surreal_runtimes(clean=clean)

    if restore_auth and AUTH_FILENAME in archive.files and not settings.uses_relational_auth:
        info("Restoring auth payload into Surreal auth storage...")
        payload = auth_payload_from_archive(archive)
        if payload is None or not _restore_auth_payload(payload, clean=clean):
            error("Auth import failed")
            raise typer.Exit(code=1)

    if restore_content and CONTENT_FILENAME in archive.files and settings.store == "surreal":
        info("Restoring content payload into Surreal content storage...")
        payload = content_payload_from_archive(archive)
        if payload is None or not _restore_content_payload(payload, clean=clean):
            error("Content import failed")
            raise typer.Exit(code=1)

    info(f"Restoring graph payload into {settings.store} store...")
    payload = json.loads(archive.files[GRAPH_FILENAME].decode("utf-8"))
    if not _restore_graph_payload(payload, effective_org_id, clean=clean):
        error("Graph import failed")
        raise typer.Exit(code=1)

    @run_async
    async def _rehearse() -> None:
        result = await verify_graph_archive(
            archive,
            organization_id=effective_org_id,
            sample_size=sample_size,
        )
        _print_verify_summary(result)
        success("Archive verification passed")

        if run_auth_flow:
            info(f"Running auth flow harness against {base_url}...")
            await _run_auth_flow_gate(
                base_url=base_url,
                auth_flow_email=auth_flow_email,
                auth_flow_password=auth_flow_password,
                email_outbox_path=email_outbox_path,
            )
            success("Auth flow harness passed")

        if run_baseline:
            info(f"Replaying deterministic baseline against {base_url}...")
            await _replay_baseline(
                base_url=base_url,
                baselines_dir=baselines_dir,
                email=email,
                password=password,
                manifest_path=manifest_path,
            )
            success("Baseline replay passed")

        success("Migration rehearsal passed")

    _rehearse()


@app.command("cutover")
def cutover_archive(
    source: Annotated[Path, typer.Argument(help="Archive .tar.gz or directory to cut over from")],
    source_type: Annotated[
        ArchiveSourceType,
        typer.Option("--source-type", help="Required archive source type for restore policy"),
    ],
    target_mode: Annotated[
        ArchiveTargetMode,
        typer.Option("--target-mode", help="Required restore target mode for policy checks"),
    ],
    org_id: Annotated[
        str,
        typer.Option("--org-id", help="Organization UUID override"),
    ] = "",
    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation")] = False,
    dry_run: Annotated[
        bool,
        typer.Option("--dry-run", help="Print the cutover plan without importing data"),
    ] = False,
    write_freeze_confirmed: Annotated[
        bool,
        typer.Option(
            "--write-freeze-confirmed",
            help="Acknowledge that legacy writes are frozen before cutover begins",
        ),
    ] = False,
    clean: Annotated[
        bool,
        typer.Option("--clean", help="Clear the target graph before import"),
    ] = True,
    restore_database_dump: Annotated[
        bool,
        typer.Option(
            "--restore-database-dump",
            "--restore-postgres",
            help="Restore the database dump sidecar (postgres.sql) before graph import",
        ),
    ] = False,
    restore_auth: Annotated[
        bool,
        typer.Option(
            "--restore-auth/--skip-auth", help="Restore auth payload into Surreal auth storage"
        ),
    ] = True,
    restore_content: Annotated[
        bool,
        typer.Option(
            "--restore-content/--skip-content",
            help="Restore content payload into Surreal content storage",
        ),
    ] = True,
    run_baseline: Annotated[
        bool,
        typer.Option(
            "--run-baseline/--skip-baseline", help="Replay the deterministic runtime baseline"
        ),
    ] = True,
    run_auth_flow: Annotated[
        bool,
        typer.Option(
            "--run-auth-flow/--skip-auth-flow", help="Run the auth-flow acceptance harness"
        ),
    ] = True,
    run_bench_live_smoke: Annotated[
        bool,
        typer.Option(
            "--run-bench-live-smoke", help="Run the live smoke bench after baseline replay"
        ),
    ] = False,
    run_bench_live: Annotated[
        bool,
        typer.Option(
            "--run-bench-live", help="Run the artifact-producing live bench after acceptance smoke"
        ),
    ] = False,
    bench_label: Annotated[
        str,
        typer.Option("--bench-label", help="Label used when running bench-live"),
    ] = DEFAULT_CUTOVER_BENCH_LABEL,
    base_url: Annotated[
        str,
        typer.Option("--base-url", help="Base URL for auth-flow and baseline replay"),
    ] = DEFAULT_REHEARSAL_BASE_URL,
    auth_flow_email: Annotated[
        str,
        typer.Option(
            "--auth-flow-email",
            help="Auth-flow user email; generated when omitted",
        ),
    ] = "",
    auth_flow_password: Annotated[
        str,
        typer.Option("--auth-flow-password", help="Auth-flow user password"),
    ] = DEFAULT_AUTH_FLOW_PASSWORD,
    email_outbox_path: Annotated[
        Path,
        typer.Option(
            "--email-outbox-path",
            help="JSONL outbox path used to consume the password reset token",
        ),
    ] = DEFAULT_AUTH_FLOW_EMAIL_OUTBOX,
    baselines_dir: Annotated[
        Path,
        typer.Option("--baselines-dir", help="Directory containing baseline case files"),
    ] = DEFAULT_REHEARSAL_BASELINES_DIR,
    manifest_path: Annotated[
        Path,
        typer.Option(
            "--manifest-path", help="Runtime baseline manifest from `moon run baseline-seed`"
        ),
    ] = DEFAULT_REHEARSAL_MANIFEST,
    email: Annotated[
        str,
        typer.Option("--email", help="Baseline user email"),
    ] = DEFAULT_REHEARSAL_EMAIL,
    password: Annotated[
        str,
        typer.Option("--password", help="Baseline user password"),
    ] = DEFAULT_REHEARSAL_PASSWORD,
    sample_size: Annotated[
        int,
        typer.Option("--sample-size", help="How many entity IDs to spot-check during verify"),
    ] = 10,
    reopen_writes: Annotated[
        bool,
        typer.Option(
            "--reopen-writes",
            help="Mark the acceptance gate complete and permit writes on SurrealDB",
        ),
    ] = False,
    acknowledge_no_instant_rollback: Annotated[
        bool,
        typer.Option(
            "--acknowledge-no-instant-rollback",
            help="Acknowledge that rollback is no longer promised once writes reopen on SurrealDB",
        ),
    ] = False,
) -> None:
    """Run the explicit Surreal cutover acceptance gate on a validated archive."""
    _require_cutover_surreal_runtime()

    archive = _load_valid_archive(source)
    effective_org_id = _resolve_org_id(org_id, archive.manifest.organization_id)

    _require_database_dump_restore_policy(
        restore_database_dump=restore_database_dump,
        source_type=source_type,
        target_mode=target_mode,
    )
    if restore_database_dump and POSTGRES_FILENAME not in archive.files:
        error("Archive does not contain the database dump sidecar (postgres.sql)")
        raise typer.Exit(code=1)
    if GRAPH_FILENAME not in archive.files:
        error("Archive does not contain graph.json")
        raise typer.Exit(code=1)
    _warn_if_database_dump_payload_skipped(
        archive=archive,
        restore_database_dump=restore_database_dump,
    )
    _warn_if_auth_payload_skipped(archive=archive, restore_auth=restore_auth)
    _warn_if_content_payload_skipped(archive=archive, restore_content=restore_content)
    _print_archive_restore_review(
        archive=archive,
        source_type=source_type,
        target_mode=target_mode,
        restore_database_dump=restore_database_dump,
        restore_graph=True,
        restore_auth=restore_auth,
        restore_content=restore_content,
    )

    warn("Rollback is supported only until writes reopen on SurrealDB.")
    warn("This command does not unfreeze or freeze writes for you; it enforces the operator gate.")

    if dry_run:
        _print_cutover_plan(
            run_auth_flow=run_auth_flow,
            run_baseline=run_baseline,
            run_bench_live_smoke=run_bench_live_smoke,
            run_bench_live=run_bench_live,
            reopen_writes=reopen_writes,
            manifest_path=manifest_path,
        )
        success("Cutover dry run complete")
        return

    if not write_freeze_confirmed:
        error("Cutover requires --write-freeze-confirmed before import begins")
        raise typer.Exit(code=1)
    if run_baseline and not manifest_path.exists():
        error(f"Baseline manifest not found: {manifest_path}")
        raise typer.Exit(code=1)

    if not yes:
        warn("This will import the archive and run acceptance checks while writes remain frozen.")
        if not typer.confirm("Continue?"):
            info("Cancelled")
            return

    if restore_database_dump:
        info("Restoring database dump sidecar...")
        _restore_pg_sql(archive.files[POSTGRES_FILENAME].decode("utf-8"), clean)

    _bootstrap_surreal_runtimes(clean=clean)

    if restore_auth and AUTH_FILENAME in archive.files and not settings.uses_relational_auth:
        info("Importing auth payload into the Surreal auth runtime...")
        payload = auth_payload_from_archive(archive)
        if payload is None or not _restore_auth_payload(payload, clean=clean):
            error("Auth import failed")
            raise typer.Exit(code=1)

    if restore_content and CONTENT_FILENAME in archive.files and settings.store == "surreal":
        info("Importing content payload into the Surreal content runtime...")
        payload = content_payload_from_archive(archive)
        if payload is None or not _restore_content_payload(payload, clean=clean):
            error("Content import failed")
            raise typer.Exit(code=1)

    info("Importing graph payload into the Surreal runtime...")
    payload = json.loads(archive.files[GRAPH_FILENAME].decode("utf-8"))
    if not _restore_graph_payload(payload, effective_org_id, clean=clean):
        error("Graph import failed")
        raise typer.Exit(code=1)
    run_async(_run_cutover_acceptance)(
        archive=archive,
        organization_id=effective_org_id,
        sample_size=sample_size,
        run_auth_flow=run_auth_flow,
        auth_flow_email=auth_flow_email,
        auth_flow_password=auth_flow_password,
        email_outbox_path=email_outbox_path,
        run_baseline=run_baseline,
        base_url=base_url,
        baselines_dir=baselines_dir,
        email=email,
        password=password,
        manifest_path=manifest_path,
        run_bench_live_smoke=run_bench_live_smoke,
        run_bench_live=run_bench_live,
        bench_label=bench_label,
    )

    if not reopen_writes:
        _print_writes_remain_frozen_notice()
        return

    if not acknowledge_no_instant_rollback:
        error("Refusing to reopen writes without --acknowledge-no-instant-rollback")
        raise typer.Exit(code=1)

    warn("Rollback is no longer promised once writes reopen on SurrealDB.")
    success("Acceptance gate complete. Writes may now be reopened on the Surreal runtime.")
