"""Strict restored-audit verification at reconciliation, resume and READY boundaries."""

from __future__ import annotations

import base64
import copy
import hashlib
import json
import re
import uuid
from datetime import UTC, datetime, timedelta
from pathlib import Path
from types import SimpleNamespace
from typing import Any
from unittest.mock import Mock

import pytest
from sqlalchemy import Column, Integer, MetaData, Table, create_engine
from sqlalchemy.orm import Session

from modelforge_api.domain.audit import AUDIT_EMPTY_LEGACY_PREFIX_SEAL
from modelforge_api.domain.recovery import (
    RESTORE_PHASE_ORDER,
    RecoveryFailureCode,
    RestoreAdvanceRequest,
    RestoreState,
)
from modelforge_api.persistence.models import Base
from modelforge_api.services.audit import (
    AUDIT_CURRENT_HASH_FORMAT,
    AUDIT_HASH_FORMAT_V1,
    AuditChainCheckpoint,
    AuditEventRecord,
    canonical_audit_payload_and_hash,
    canonical_audit_payload_text_and_hash,
    legacy_audit_prefix_seal,
    normalise_audit_timestamp,
)
from modelforge_api.services.recovery import RecoveryError, RecoveryService
from modelforge_api.services.recovery_fingerprint import (
    CURRENT_TRUTH_TABLES,
    FINGERPRINT_VERSION,
    LEGACY_FINGERPRINT_VERSION,
    MAX_ROWS_PER_TABLE,
    _canonical_row,
    _table_digest,
    audit_fingerprint_differences_are_compatible,
    classified_tables,
    fingerprint_compatibility,
    fingerprint_session,
)
from modelforge_api.services.recovery_postgres import (
    CommandResult,
    PostgresEngine,
    PostgresTarget,
    PostgresToolError,
)
from modelforge_api.settings import Settings

TARGET = PostgresTarget("postgres", 5432, "restored", "modelforge", None)


@pytest.fixture
def session() -> Session:
    engine = create_engine("sqlite+pysqlite:///:memory:")
    Base.metadata.create_all(engine)
    with Session(engine) as value:
        yield value


def _service(session: Session, tmp_path: Path) -> RecoveryService:
    key = base64.b64encode(b"modelforge-rc-audit-test-key!!"[:32]).decode()
    return RecoveryService(
        session,
        Settings(
            backup_root=tmp_path / "backups",
            backup_restore_root=tmp_path / "restore",
            backup_encryption_key=key,
            config_root=tmp_path / "config",
            database_url="postgresql+psycopg://modelforge:test@postgres:5432/modelforge",
        ),
    )


def _mixed_records() -> tuple[list[AuditEventRecord], AuditChainCheckpoint]:
    legacy_id = uuid.UUID("11111111-1111-1111-1111-111111111111")
    legacy_time = datetime(2026, 8, 28, 20, 0, 0, 123456, tzinfo=UTC)
    legacy_payload, legacy_hash = canonical_audit_payload_and_hash(
        correlation_id="legacy-request",
        actor_type="operator",
        actor_id="legacy-operator",
        action="LEGACY_APPROVAL",
        resource_type="revision",
        resource_id="revision-1",
        outcome="success",
        details={"approved": True},
        previous_event_hash=None,
        hash_format=AUDIT_HASH_FORMAT_V1,
    )
    legacy = AuditEventRecord(
        id=legacy_id,
        sequence=1,
        occurred_at=legacy_time,
        event_hash=legacy_hash,
        hash_format=AUDIT_HASH_FORMAT_V1,
        **legacy_payload,
    )

    v2_id = uuid.UUID("22222222-2222-2222-2222-222222222222")
    v2_time = legacy_time + timedelta(seconds=1)
    v2_payload, v2_canonical_payload, v2_hash = canonical_audit_payload_text_and_hash(
        correlation_id="v2-request",
        actor_type="operator",
        actor_id="v2-operator",
        action="V2_APPROVAL",
        resource_type="revision",
        resource_id="revision-2",
        outcome="success",
        details={"approved": True},
        previous_event_hash=legacy_hash,
        hash_format=AUDIT_CURRENT_HASH_FORMAT,
        event_id=v2_id,
        occurred_at=v2_time,
    )
    v2 = AuditEventRecord(
        id=v2_id,
        sequence=2,
        occurred_at=v2_time,
        event_hash=v2_hash,
        hash_format=AUDIT_CURRENT_HASH_FORMAT,
        canonical_payload=v2_canonical_payload,
        **v2_payload,
    )
    checkpoint = AuditChainCheckpoint(
        singleton_id=1,
        event_count=2,
        last_sequence=2,
        last_event_hash=v2_hash,
        hash_format=AUDIT_CURRENT_HASH_FORMAT,
        v2_start_sequence=2,
        legacy_prefix_count=1,
        legacy_prefix_seal=legacy_audit_prefix_seal([legacy]),
    )
    return [legacy, v2], checkpoint


def _event_row(event: AuditEventRecord) -> dict[str, str]:
    return {
        "id": str(event.id),
        "sequence": str(event.sequence),
        "occurred_at": normalise_audit_timestamp(event.occurred_at),
        "correlation_id": event.correlation_id,
        "actor_type": event.actor_type,
        "actor_id": event.actor_id,
        "action": event.action,
        "resource_type": event.resource_type,
        "resource_id": event.resource_id or "",
        "resource_id_is_null": "t" if event.resource_id is None else "f",
        "outcome": event.outcome,
        "details": json.dumps(event.details, sort_keys=True, separators=(",", ":")),
        "previous_event_hash": event.previous_event_hash or "",
        "previous_event_hash_is_null": "t" if event.previous_event_hash is None else "f",
        "event_hash": event.event_hash,
        "hash_format": event.hash_format,
        "canonical_payload": event.canonical_payload or "",
        "canonical_payload_is_null": "t" if event.canonical_payload is None else "f",
    }


def _head_row(checkpoint: AuditChainCheckpoint) -> dict[str, str]:
    return {
        "singleton_id": str(checkpoint.singleton_id),
        "event_count": str(checkpoint.event_count),
        "last_sequence": str(checkpoint.last_sequence),
        "last_event_hash": checkpoint.last_event_hash or "",
        "last_event_hash_is_null": "t" if checkpoint.last_event_hash is None else "f",
        "hash_format": checkpoint.hash_format,
        "v2_start_sequence": str(checkpoint.v2_start_sequence),
        "legacy_prefix_count": str(checkpoint.legacy_prefix_count),
        "legacy_prefix_seal": checkpoint.legacy_prefix_seal,
    }


class SnapshotEngine:
    def __init__(
        self,
        events: list[dict[str, str]],
        checkpoint: dict[str, str],
        *,
        marker_count: str = "0",
    ) -> None:
        self.events = events
        self.checkpoint = checkpoint
        self.marker_count = marker_count
        self.operations: list[str] = []

    def query_rows(
        self,
        _target: PostgresTarget,
        sql: str,
        *,
        max_rows: int = 1000,
    ) -> list[dict[str, str]]:
        self.operations.append("query_rows")
        if "from audit_chain_heads" in sql:
            return [copy.deepcopy(self.checkpoint)]
        if "count(distinct id)" in sql:
            return [
                {
                    "event_count": str(len(self.events)),
                    "distinct_event_ids": str(
                        len({row["id"] for row in self.events})
                    ),
                    "distinct_sequences": str(
                        len({row["sequence"] for row in self.events})
                    ),
                }
            ]
        match = re.search(
            r"sequence > (-?\d+) or \(sequence = -?\d+ and "
            r"id > '([0-9a-f-]+)'::uuid\)",
            sql,
        )
        after = int(match.group(1)) if match is not None else None
        after_id = uuid.UUID(match.group(2)) if match is not None else None
        return [
            copy.deepcopy(row)
            for row in self.events
            if after is None
            or (int(row["sequence"]), uuid.UUID(row["id"])) > (after, after_id)
        ][:max_rows]

    def scalar(self, _target: PostgresTarget, _sql: str) -> str:
        self.operations.append("scalar")
        return self.marker_count


def _engine() -> SnapshotEngine:
    records, checkpoint = _mixed_records()
    return SnapshotEngine([_event_row(event) for event in records], _head_row(checkpoint))


def test_recovery_strictly_accepts_a_valid_mixed_legacy_v2_chain(
    session: Session, tmp_path: Path
) -> None:
    subject = _service(session, tmp_path)
    subject.engine = _engine()  # type: ignore[assignment]

    checkpoint = subject._verify_restored_audit_chain(TARGET)

    assert checkpoint.event_count == 2
    assert checkpoint.v2_start_sequence == 2


@pytest.mark.parametrize(
    ("event_index", "field", "replacement"),
    [
        (0, "id", "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"),
        (0, "occurred_at", "2026-08-29T20:00:00.123456Z"),
        (1, "id", "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"),
        (1, "occurred_at", "2026-08-29T20:00:01.123456Z"),
    ],
)
def test_recovery_rejects_id_and_timestamp_tamper_in_legacy_and_v2(
    session: Session,
    tmp_path: Path,
    event_index: int,
    field: str,
    replacement: str,
) -> None:
    engine = _engine()
    engine.events[event_index][field] = replacement
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]

    with pytest.raises(RecoveryError) as captured:
        subject._verify_restored_audit_chain(TARGET)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value


@pytest.mark.parametrize("retained", [1, 0])
def test_recovery_rejects_tail_and_complete_history_deletion(
    session: Session, tmp_path: Path, retained: int
) -> None:
    engine = _engine()
    engine.events = engine.events[:retained]
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]

    with pytest.raises(RecoveryError) as captured:
        subject._verify_restored_audit_chain(TARGET)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
    assert "checkpoint" in str(captured.value.details).lower()


def test_corruption_blocks_before_already_applied_marker_lookup(
    session: Session, tmp_path: Path
) -> None:
    engine = _engine()
    engine.events.pop()
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]
    record = SimpleNamespace(id=uuid.uuid4())
    plan = SimpleNamespace(
        database_destination="postgresql://modelforge:test@postgres:5432/restored",
        backup_set=SimpleNamespace(backup_id="backup-1"),
    )

    with pytest.raises(RecoveryError) as captured:
        subject._reconcile_restored(record, plan)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
    assert "scalar" not in engine.operations


def test_valid_resume_checks_chain_before_returning_already_applied(
    session: Session, tmp_path: Path
) -> None:
    engine = _engine()
    engine.marker_count = "1"
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]
    record = SimpleNamespace(id=uuid.uuid4())
    plan = SimpleNamespace(
        database_destination="postgresql://modelforge:test@postgres:5432/restored",
        backup_set=SimpleNamespace(backup_id="backup-1"),
    )

    result = subject._reconcile_restored(record, plan)

    assert result["reconciliation"] == "ALREADY_APPLIED"
    assert engine.operations.index("query_rows") < engine.operations.index("scalar")


def test_corruption_is_an_independent_ready_gate_before_fingerprints(
    session: Session, tmp_path: Path
) -> None:
    engine = _engine()
    engine.events.clear()
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]
    record = SimpleNamespace()
    plan = SimpleNamespace(
        database_destination="postgresql://modelforge:test@postgres:5432/restored"
    )

    with pytest.raises(RecoveryError) as captured:
        subject._validate_restored(record, plan)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value


def test_ready_resume_reverifies_even_when_every_phase_duration_is_present(
    session: Session, tmp_path: Path
) -> None:
    subject = _service(session, tmp_path)
    _records, checkpoint = _mixed_records()
    verifier = Mock(return_value=checkpoint)
    subject._verify_restored_audit_chain = verifier  # type: ignore[method-assign]
    subject._journal = Mock()  # type: ignore[method-assign]
    subject.audit = SimpleNamespace(write=Mock())  # type: ignore[assignment]

    plan = SimpleNamespace(
        database_destination="postgresql://modelforge:test@postgres:5432/restored",
        backup_set=SimpleNamespace(backup_id="backup-1"),
        state="PREFLIGHT_PASSED",
    )
    record = SimpleNamespace(
        id=uuid.uuid4(),
        state=RestoreState.VALIDATING.value,
        plan=plan,
        phase_durations={phase.value: 0.1 for phase in RESTORE_PHASE_ORDER},
        rto_seconds=None,
        rpo_seconds=0.0,
        ready_at=None,
    )
    subject._operation_row = lambda _operation_id: record  # type: ignore[method-assign]
    subject._operation_response = lambda value: value  # type: ignore[method-assign]

    result = subject.advance_restore(
        record.id,
        RestoreAdvanceRequest(actor="operator", reason="resume final READY control"),
    )

    assert result.state == RestoreState.READY.value
    verifier.assert_called_once()
    verified_target = verifier.call_args.args[0]
    assert (verified_target.host, verified_target.database, verified_target.user) == (
        TARGET.host,
        TARGET.database,
        TARGET.user,
    )
    assert record.phase_durations[RestoreState.READY.value] >= 0


def _v2_chain_rows(
    count: int,
    *,
    duplicate_id_at: int | None = None,
    duplicate_sequence_at: int | None = None,
) -> tuple[list[dict[str, str]], AuditChainCheckpoint]:
    rows: list[dict[str, str]] = []
    previous_hash: str | None = None
    started = datetime(2026, 8, 30, 12, 0, tzinfo=UTC)
    for index in range(1, count + 1):
        event_id = uuid.UUID(
            int=(index - 1 if duplicate_id_at == index else index)
        )
        occurred_at = started + timedelta(microseconds=index)
        payload, canonical_payload, event_hash = canonical_audit_payload_text_and_hash(
            correlation_id=f"request-{index}",
            actor_type="operator",
            actor_id="pagination-control",
            action="VALID_EVENT",
            resource_type="audit-test",
            resource_id=str(index),
            outcome="success",
            details={"index": index},
            previous_event_hash=previous_hash,
            hash_format=AUDIT_CURRENT_HASH_FORMAT,
            event_id=event_id,
            occurred_at=occurred_at,
        )
        record = AuditEventRecord(
            id=event_id,
            sequence=index,
            occurred_at=occurred_at,
            event_hash=event_hash,
            hash_format=AUDIT_CURRENT_HASH_FORMAT,
            canonical_payload=canonical_payload,
            **payload,
        )
        row = _event_row(record)
        if duplicate_sequence_at == index:
            row["sequence"] = str(index - 1)
        rows.append(row)
        previous_hash = event_hash
    checkpoint = AuditChainCheckpoint(
        singleton_id=1,
        event_count=count,
        last_sequence=count,
        last_event_hash=rows[-1]["event_hash"],
        hash_format=AUDIT_CURRENT_HASH_FORMAT,
        v2_start_sequence=1,
        legacy_prefix_count=0,
        legacy_prefix_seal=AUDIT_EMPTY_LEGACY_PREFIX_SEAL,
    )
    return rows, checkpoint


def test_valid_501_row_chain_crosses_the_composite_cursor_boundary(
    session: Session, tmp_path: Path
) -> None:
    rows, checkpoint = _v2_chain_rows(501)
    engine = SnapshotEngine(rows, _head_row(checkpoint))
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]

    verified = subject._verify_restored_audit_chain(TARGET)

    assert verified.event_count == 501
    assert engine.operations.count("query_rows") == 4


def test_composite_cursor_does_not_skip_the_501st_duplicate_sequence(
    session: Session, tmp_path: Path
) -> None:
    rows, checkpoint = _v2_chain_rows(501, duplicate_sequence_at=501)
    engine = SnapshotEngine(rows, _head_row(checkpoint))
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]

    with pytest.raises(RecoveryError) as captured:
        subject._verify_restored_audit_chain(TARGET)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
    assert engine.operations.count("query_rows") == 4  # head, identity counts, both pages
    assert any(
        "not unique" in violation or "expected 501" in violation
        for violation in captured.value.details["violations"]
    )


def test_strict_recovery_rejects_a_valid_hash_chain_with_duplicate_event_uuid(
    session: Session, tmp_path: Path
) -> None:
    rows, checkpoint = _v2_chain_rows(501, duplicate_id_at=501)
    engine = SnapshotEngine(rows, _head_row(checkpoint))
    subject = _service(session, tmp_path)
    subject.engine = engine  # type: ignore[assignment]

    with pytest.raises(RecoveryError) as captured:
        subject._verify_restored_audit_chain(TARGET)

    assert captured.value.code == RecoveryFailureCode.AUDIT_CHAIN_CORRUPT.value
    assert engine.operations.count("query_rows") == 4
    violations = captured.value.details["violations"]
    assert any("UUID" in violation and "unique" in violation for violation in violations)
    assert any("UUID" in violation and "duplicated" in violation for violation in violations)


def _fingerprint(
    version: str,
    *,
    include_checkpoint: bool,
) -> dict[str, object]:
    table_names = set(classified_tables())
    if not include_checkpoint:
        table_names.remove("audit_chain_heads")
    tables: dict[str, object] = {
        name: {
            "row_count": 0,
            "digest": hashlib.sha256(name.encode()).hexdigest(),
            "status": "COMPLETE",
            "redacted_columns": [],
        }
        for name in sorted(table_names)
    }
    tables["audit_events"] = {
        "row_count": 12,
        "digest": "a" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    if include_checkpoint:
        tables["audit_chain_heads"] = {
            "row_count": 1,
            "digest": "b" * 64,
            "status": "COMPLETE",
            "redacted_columns": [],
        }
    return {
        "version": version,
        "tables": tables,
        "digest": hashlib.sha256(version.encode()).hexdigest(),
        "groups": {},
        "physical_tables": sorted(table_names | CURRENT_TRUTH_TABLES),
        "unexpected_tables": [],
    }


def _with_legitimate_reconciliation_delta(
    source: dict[str, object], *, introduce_checkpoint: bool
) -> dict[str, object]:
    restored = copy.deepcopy(source)
    restored["version"] = FINGERPRINT_VERSION
    restored["digest"] = "f" * 64
    restored["unexpected_tables"] = []
    restored["physical_tables"] = sorted(classified_tables() | CURRENT_TRUTH_TABLES)
    tables = restored["tables"]
    assert isinstance(tables, dict)
    tables["audit_events"] = {
        "row_count": 13,
        "digest": "c" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    tables["audit_chain_heads"] = {
        "row_count": 1,
        "digest": "d" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    if not introduce_checkpoint:
        tables["operational_alerts"] = {
            "row_count": 0,
            "digest": "e" * 64,
            "status": "COMPLETE",
            "redacted_columns": [],
        }
    return restored


def test_schema_0022_fingerprint_allows_exact_checkpoint_introduction_only() -> None:
    source = _fingerprint(
        LEGACY_FINGERPRINT_VERSION,
        include_checkpoint=False,
    )
    source.pop("unexpected_tables")  # m15.1 predates explicit physical-table reporting
    source.pop("physical_tables")
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=True
    )
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260828_0022"
    )

    assert contract.compatible is True
    assert contract.mode == "SCHEMA_0022_TO_0024"
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is True
    )


def test_schema_0024_fingerprint_allows_only_exact_reconciliation_deltas() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )

    assert contract.compatible is True
    assert contract.mode == "CURRENT_STRICT"
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is True
    )


def test_current_fingerprint_missing_its_checkpoint_remains_strictly_incompatible() -> None:
    source = _fingerprint(
        FINGERPRINT_VERSION,
        include_checkpoint=False,
    )
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=True
    )
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )

    assert contract.compatible is False
    assert contract.mode == "INCOMPATIBLE"
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is False
    )


@pytest.mark.parametrize("attack", ["missing_models", "attacker_table"])
def test_fingerprint_contract_rejects_sparse_or_unexpected_tables(attack: str) -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    tables = source["tables"]
    assert isinstance(tables, dict)
    if attack == "missing_models":
        tables.pop("models")
    else:
        tables["attacker_shadow"] = {
            "row_count": 1,
            "digest": "9" * 64,
            "status": "COMPLETE",
            "redacted_columns": [],
        }

    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )

    assert contract.compatible is False
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is False
    )


def test_semantic_fingerprint_exposes_an_unexpected_physical_table(
    session: Session,
) -> None:
    source = fingerprint_session(session)
    session.connection().exec_driver_sql(
        "CREATE TABLE attacker_shadow (payload TEXT NOT NULL)"
    )
    restored = fingerprint_session(session)

    assert source["unexpected_tables"] == []
    assert restored["unexpected_tables"] == ["attacker_shadow"]
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )
    assert contract.compatible is False


def test_fingerprint_rejects_large_non_audit_row_loss() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    source_tables = source["tables"]
    assert isinstance(source_tables, dict)
    source_tables["projects"] = {
        "row_count": 20,
        "digest": "1" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    restored_tables = restored["tables"]
    assert isinstance(restored_tables, dict)
    restored_tables["projects"] = {
        "row_count": 1,
        "digest": "2" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )

    assert contract.compatible is True
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is False
    )


def test_fingerprint_rejects_same_count_different_audit_digest() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    restored_tables = restored["tables"]
    assert isinstance(restored_tables, dict)
    restored_tables["audit_events"] = {
        "row_count": 12,
        "digest": "c" * 64,
        "status": "COMPLETE",
        "redacted_columns": [],
    }
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )

    assert contract.compatible is True
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is False
    )


def test_fingerprint_helper_cannot_reuse_a_contract_after_table_mutation() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )
    restored_tables = restored["tables"]
    assert isinstance(restored_tables, dict)
    restored_tables.pop("models")

    assert contract.compatible is True
    assert (
        audit_fingerprint_differences_are_compatible(source, restored, contract)
        is False
    )


def test_bounded_table_fingerprint_can_never_authorize_ready() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    source_tables = source["tables"]
    assert isinstance(source_tables, dict)
    source_tables["models"] = {
        "row_count": MAX_ROWS_PER_TABLE,
        "digest": "7" * 64,
        "status": "BOUNDED",
        "redacted_columns": [],
    }
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )

    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )
    assert contract.compatible is False


def test_more_than_250k_rows_is_observed_as_bounded_not_a_complete_digest() -> None:
    table = Table("simulated_large_table", MetaData(), Column("value", Integer()))

    class SimulatedRows:
        def yield_per(self, _size: int) -> Any:
            for value in range(MAX_ROWS_PER_TABLE + 1):
                yield (value,)

    class SimulatedConnection:
        def execute(self, _statement: Any) -> SimulatedRows:
            return SimulatedRows()

    entry = _table_digest(SimulatedConnection(), table)  # type: ignore[arg-type]

    assert entry["row_count"] == MAX_ROWS_PER_TABLE
    assert entry["status"] == "BOUNDED"


def test_length_prefixed_typed_rows_have_no_delimiter_or_null_type_collision() -> None:
    assert _canonical_row(("a\x1fb", "c")) != _canonical_row(("a", "b\x1fc"))
    assert _canonical_row((None,)) != _canonical_row(("\x00",))
    assert _canonical_row((1,)) != _canonical_row(("1",))


def test_fingerprint_rejects_dropped_current_truth_table_presence() -> None:
    source = _fingerprint(FINGERPRINT_VERSION, include_checkpoint=True)
    restored = _with_legitimate_reconciliation_delta(
        source, introduce_checkpoint=False
    )
    physical = restored["physical_tables"]
    assert isinstance(physical, list)
    physical.remove("host_telemetry_latest")

    contract = fingerprint_compatibility(
        source, restored, source_schema_revision="20260830_0024"
    )
    assert contract.compatible is False


def test_postgres_query_rows_parses_csv_safely_and_enforces_its_bound() -> None:
    engine = PostgresEngine()
    engine._run = lambda *_args, **_kwargs: CommandResult(  # type: ignore[method-assign]
        command="psql",
        returncode=0,
        stdout='id,details\r\n1,"{""message"":""a,b""}"\r\n',
        stderr="",
        duration_seconds=0.01,
    )
    assert engine.query_rows(TARGET, "select bounded", max_rows=1) == [
        {"id": "1", "details": '{"message":"a,b"}'}
    ]

    engine._run = lambda *_args, **_kwargs: CommandResult(  # type: ignore[method-assign]
        command="psql",
        returncode=0,
        stdout="id\r\n1\r\n2\r\n",
        stderr="",
        duration_seconds=0.01,
    )
    with pytest.raises(PostgresToolError, match="more than 1"):
        engine.query_rows(TARGET, "select unexpectedly_unbounded", max_rows=1)