transcription/tests/test_meta_contract_guards.py

"""Regression guards for canonical documentation and instruction contracts."""

from __future__ import annotations

import re
from pathlib import Path

from transcription.config import Settings

PROJECT_ROOT = Path(__file__).resolve().parents[1]


ACTIVE_CONTRACT_FILES = (
    "AGENTS.md",
    ".github/agents/python-reviewer.agent.md",
    ".github/prompts/review-python-architecture.prompt.md",
    ".github/instructions/documentation-sync.instructions.md",
    ".github/instructions/services.instructions.md",
    ".github/instructions/ui.instructions.md",
    ".github/instructions/error-handling.instructions.md",
    ".github/instructions/providers.instructions.md",
    ".github/instructions/tests.instructions.md",
    ".github/skills/python-code-reviewer/skill.md",
    ".github/skills/evidence-provenance-auditor/skill.md",
    ".github/skills/test-effectiveness-auditor/skill.md",
    "docs/index.md",
    "docs/architecture.md",
    "docs/requirements.md",
    "docs/schema.md",
    "docs/error_handling.md",
    "docs/ui/README.md",
)


LEGACY_REFERENCE_MARKERS = (
    "docs-v4x-archive",
    "docs/ver4/history.md",
    "docs/ver4.0",
    "docs/ver4.1",
    "docs/ver4.2",
    "docs/ver4.3",
    "docs/ver4.4",
    "docs/ver4.5",
    "docs/ver4.6",
    "docs/ver4.7",
)


def _read(relative_path: str) -> str:
    return (PROJECT_ROOT / relative_path).read_text(encoding="utf-8")


BASELINE_SOURCE = "docs/index.md"

# Docs that legitimately discuss versions other than the current baseline: the roadmap plans
# future work, and migration/deployment notes describe historical phases by name.
BASELINE_SCAN_EXCLUSIONS = frozenset(
    {
        "docs/roadmap_plan.md",
        "docs/data_migration.md",
        "docs/cloudflare_tunnel_access.md",
        "docs/production-runbook.md",
        "docs/backup_restore.md",
    }
)

_BASELINE_DECLARATION = re.compile(r"Current Baseline:\s*V(\d+\.\d+)")
_BASELINE_CLAIM = re.compile(
    r"\b(?:Current Baseline:\s*|current\s+|active\s+|canonical\s+)V(\d+(?:\.\d+)?)",
    re.IGNORECASE,
)
_LEGACY_VERSION_REFERENCE = re.compile(r"\b[Vv](4(?:\.\d+)?|5(?:\.\d+)?|6\.0)\b|test_v42_evidence|test_v45_candidates")
LEGACY_VERSION_REFERENCE_EXCLUSIONS = frozenset(
    {
        ".github/workflows/quality-gate.yml",
        "docs/data_migration.md",
        "docs/requirements.md",
        "docs/reviews",
        "docs/roadmap_plan.md",
        "docs/schema.md",
        "src/transcription/db/operations.py",
        "tests/test_meta_contract_guards.py",
    }
)


def _declared_baseline() -> str:
    """Return the one baseline version the canonical doc set must agree on."""
    match = _BASELINE_DECLARATION.search(_read(BASELINE_SOURCE))
    assert match is not None, f"{BASELINE_SOURCE} must declare 'Current Baseline: V<major>.<minor>'"
    return match.group(1)


def _baseline_scanned_docs() -> list[Path]:
    docs_root = PROJECT_ROOT / "docs"
    return sorted(
        path
        for path in docs_root.rglob("*.md")
        # docs/reviews/** are dated, non-canonical snapshots and are pinned to the version
        # that was current when they were written.
        if "reviews" not in path.relative_to(docs_root).parts
        and path.relative_to(PROJECT_ROOT).as_posix() not in BASELINE_SCAN_EXCLUSIONS
    )


def test_canonical_docs_declare_one_consistent_baseline():
    """A stale baseline label makes canonical docs read as drift against current code.

    Every canonical doc that names the current baseline must name the same one, so a version
    bump cannot leave half the authority set describing a superseded release as current.
    """
    baseline = _declared_baseline()
    violations: dict[str, list[str]] = {}

    for path in _baseline_scanned_docs():
        relative = path.relative_to(PROJECT_ROOT).as_posix()
        text = path.read_text(encoding="utf-8")
        stale = {version for version in _BASELINE_CLAIM.findall(text) if version != baseline}
        if stale:
            violations[relative] = sorted(stale)

    assert violations == {}, (
        f"Canonical baseline is V{baseline} (declared in {BASELINE_SOURCE}), "
        f"but these docs claim another version is current: {violations}"
    )


def test_legacy_version_markers_are_scoped_to_allowed_historical_files():
    """Version labels from older project baselines should not linger in active files.

    Historical references are allowed only where the old version is the subject matter:
    roadmap planning, dated reviews, stable requirement IDs, and explicit migration or
    compatibility notes.
    """
    scanned: list[Path] = []
    for root in (
        PROJECT_ROOT / ".github",
        PROJECT_ROOT / "docs",
        PROJECT_ROOT / "src",
        PROJECT_ROOT / "tests",
    ):
        scanned.extend(path for path in root.rglob("*") if path.suffix in {".md", ".py", ".yml", ".yaml"})
    scanned.extend((PROJECT_ROOT / "AGENTS.md", PROJECT_ROOT / ".pre-commit-config.yaml"))

    violations: dict[str, list[str]] = {}
    for path in sorted(set(scanned)):
        relative = path.relative_to(PROJECT_ROOT).as_posix()
        if any(
            relative == excluded or relative.startswith(f"{excluded}/")
            for excluded in LEGACY_VERSION_REFERENCE_EXCLUSIONS
        ):
            continue
        matches = sorted(
            {match.group(0) for match in _LEGACY_VERSION_REFERENCE.finditer(path.read_text(encoding="utf-8"))}
        )
        if matches:
            violations[relative] = matches

    assert violations == {}, f"Legacy version markers remain in active files: {violations}"


def test_active_contract_files_are_present():
    """Guard the guard: ensure all expected authority files are scanned."""
    missing = [path for path in ACTIVE_CONTRACT_FILES if not (PROJECT_ROOT / path).exists()]
    assert missing == []


def test_every_github_contract_file_is_scanned():
    """A new instruction, skill, agent, or prompt must not escape these guards by default.

    Files added under .github without being listed here would never be checked for legacy
    authority references or broken repo paths, which is how stale guidance accumulates.
    """
    github_root = PROJECT_ROOT / ".github"
    discovered = {
        path.relative_to(PROJECT_ROOT).as_posix()
        for path in github_root.rglob("*.md")
        if not path.is_relative_to(github_root / "ISSUE_TEMPLATE")
    }
    unscanned = sorted(discovered - set(ACTIVE_CONTRACT_FILES))
    assert unscanned == [], f"Add these to ACTIVE_CONTRACT_FILES so they are guarded: {unscanned}"


def test_no_legacy_authority_references_in_active_contract_files():
    """Active contracts must not route authority through removed revision trees."""
    violations: dict[str, list[str]] = {}
    for relative_path in ACTIVE_CONTRACT_FILES:
        text = _read(relative_path)
        found = [marker for marker in LEGACY_REFERENCE_MARKERS if marker in text]
        if found:
            violations[relative_path] = found
    assert violations == {}


# Matches repo-relative references such as `docs/schema.md`, `./docs/reviews`, or
# `../../src/transcription/db/models.py` (the leading `../` is consumed by the lookbehind).
_REPO_PATH_REFERENCE = re.compile(r"(?<![\w.-])((?:\./)?(?:docs|src|tests)/[\w./-]+)")
_TEST_NODE_REFERENCE = re.compile(r"((?:\./)?tests/[\w./-]+\.py)::([\w:]+)")


def _normalize_reference(raw: str) -> str:
    return raw.removeprefix("./").rstrip(".,;:)")


def test_contract_files_reference_only_existing_repo_paths():
    """Every docs/src/tests path named in a contract file must exist.

    This generalizes the legacy-marker list: instead of enumerating retired trees one by one,
    any reference that no longer resolves fails, whether it is a renamed doc, a moved module,
    or a deleted test.
    """
    violations: dict[str, list[str]] = {}

    for relative_path in ACTIVE_CONTRACT_FILES:
        text = _read(relative_path)
        broken = set()
        for match in _REPO_PATH_REFERENCE.finditer(text):
            reference = _normalize_reference(match.group(1))
            # Glob patterns describe a family of paths, not one target.
            if "*" in reference or "<" in reference:
                continue
            candidate = reference.rstrip("/")
            if not (PROJECT_ROOT / candidate).exists():
                broken.add(reference)
        if broken:
            violations[relative_path] = sorted(broken)

    assert violations == {}, f"Contract files reference paths that do not exist: {violations}"


def test_contract_files_reference_only_existing_test_names():
    """A cited test node id must still name a real test.

    Contract files point at specific guards as proof a rule is enforced. A renamed test turns
    that citation into a claim of enforcement that no longer holds.
    """
    violations: dict[str, list[str]] = {}

    for relative_path in ACTIVE_CONTRACT_FILES:
        text = _read(relative_path)
        broken = set()
        for match in _TEST_NODE_REFERENCE.finditer(text):
            test_file = _normalize_reference(match.group(1))
            target = PROJECT_ROOT / test_file
            if not target.exists():
                continue  # Reported by the path-existence guard.
            source = target.read_text(encoding="utf-8")
            missing_names = [name for name in match.group(2).split("::") if name not in source]
            if missing_names:
                broken.add(f"{test_file}::{match.group(2)}")
        if broken:
            violations[relative_path] = sorted(broken)

    assert violations == {}, f"Contract files cite test names that no longer exist: {violations}"


def test_canonical_authority_references_are_present():
    """Critical instruction and skill files must keep canonical references explicit."""
    required_fragments = {
        "AGENTS.md": (
            "docs/index.md",
            "docs/invariant/*",
            "uv run",
            ".github/agents/",
            ".github/prompts/",
        ),
        ".github/instructions/documentation-sync.instructions.md": (
            "docs/index.md",
            "docs/schema.md",
            "docs/error_handling.md",
        ),
        ".github/instructions/services.instructions.md": (
            "docs/",
            "./error-handling.instructions.md",
            "src/transcription/db/models.py",
            "docs/schema.md",
            "append-only",
        ),
        ".github/instructions/ui.instructions.md": (
            "docs/",
            "./error-handling.instructions.md",
            "src/transcription/db/models.py",
            "docs/schema.md",
        ),
        ".github/instructions/error-handling.instructions.md": (
            "docs/error_handling.md",
            "docs/requirements.md",
        ),
        ".github/instructions/providers.instructions.md": (
            "docs/invariant/ai_evidence_and_provenance.md",
            "docs/schema.md",
            "tests/test_provider_boundaries.py",
            "SAFE_RESPONSE_HEADERS",
        ),
        ".github/instructions/tests.instructions.md": (
            "AGENTS.md",
            "docs/invariant/*",
            "tests/test_meta_contract_guards.py",
            "tests/test_config_isolation.py",
            "asyncio_mode",
        ),
        ".github/skills/python-code-reviewer/skill.md": (
            "docs/*",
            "docs/schema.md",
            ".github/instructions/error-handling.instructions.md",
        ),
        ".github/skills/evidence-provenance-auditor/skill.md": (
            "docs/schema.md",
            "docs/requirements.md",
            "docs/error_handling.md",
        ),
        ".github/skills/test-effectiveness-auditor/skill.md": (
            "docs/invariant/*",
            "tests/test_meta_contract_guards.py",
        ),
    }

    missing: dict[str, list[str]] = {}
    for relative_path, fragments in required_fragments.items():
        text = _read(relative_path)
        absent = [fragment for fragment in fragments if fragment not in text]
        if absent:
            missing[relative_path] = absent
    assert missing == {}


def _declared_env_production_example_keys(*, include_commented: bool) -> set[str]:
    text = _read(".env.production.example")
    pattern = r"^\s*#?\s*([A-Z0-9_]+)\s*=" if include_commented else r"^\s*([A-Z0-9_]+)\s*="
    return {match.group(1) for match in re.finditer(pattern, text, flags=re.MULTILINE)}


def _active_env_production_example_values() -> dict[str, str]:
    text = _read(".env.production.example")
    return {key: value.strip() for key, value in re.findall(r"^\s*([A-Z0-9_]+)\s*=\s*(.*)$", text, flags=re.MULTILINE)}


def _normalize_env_path_value(value: str | None) -> str | None:
    if value is None:
        return None
    normalized = value.strip().replace("\\", "/")
    while normalized.startswith("./"):
        normalized = normalized[2:]
    return normalized


def test_env_production_example_keys_match_runtime_settings_contract():
    """Guard against .env.production.example drift from Settings keys."""
    declared = _declared_env_production_example_keys(include_commented=True)
    settings_keys = {field.upper() for field in Settings.model_fields if field != "database"}
    database_keys = {
        "DATABASE__DRIVER",
        "DATABASE__PATH",
        "DATABASE__HOST",
        "DATABASE__PORT",
        "DATABASE__DATABASE",
        "DATABASE__USER",
        "DATABASE__PASSWORD",
    }
    deployment_helper_keys = {
        "CLOUDFLARE_TUNNEL_TOKEN",
        "RUNTIME_SETTINGS_ENV_FILE",
        "BACKUP_DIR",
        "BACKUP_RETENTION_DAYS",
    }
    allowed = settings_keys | database_keys | deployment_helper_keys

    missing = sorted(allowed - declared)
    unknown = sorted(declared - allowed)
    assert missing == []
    assert unknown == []


def test_env_production_example_builds_settings_and_keeps_production_defaults():
    """The production template must parse into Settings and preserve intended production posture."""
    active = _active_env_production_example_values()
    Settings(_env_file=PROJECT_ROOT / ".env.production.example", _cli_parse_args=False)

    assert active["ENVIRONMENT"] == "production"
    assert active["RUN_EMBEDDED_WORKER"] == "false"
    assert active["DATABASE__DRIVER"] == "postgres"