src/dashboard/verification.py

"""Shared verification utilities for dashboard status builders.

Both ``IntegrationStatusBuilder`` and ``GateStatusBuilder`` use these
functions so that ``/api/integration/status`` and ``/api/gate/status``
can never disagree about the same source tree.

Scientifically relevant paths (tree digest)::

    src/
    configs/
    research/schemas/
    pyproject.toml

A test-baseline change (docs, CHANGELOG, test_baseline.json itself) must
NOT invalidate the test status. Only changes to the scientifically
relevant source tree mark the baseline as stale.
"""

from __future__ import annotations

import hashlib
import json
import subprocess
from pathlib import Path
from typing import Any, cast

# ============================================================================
# Scientifically relevant source paths
# ============================================================================

# Source code and config changes must mark the baseline as stale.
SCIENTIFIC_PATHS: list[str] = [
    "src/",
    "configs/",
    "research/schemas/",
    "pyproject.toml",
]

# Test logic changes must also mark the baseline as stale — a changed test
# is a changed verification. But the baseline file itself must NOT
# invalidate itself (a file cannot stably contain its own digest).
TEST_PATHS: list[str] = ["tests/"]
_DIGEST_EXCLUDE_FILES: set[str] = {"tests/test_baseline.json"}
_DIGEST_EXCLUDE_SUFFIXES: tuple[str, ...] = (".pyc", ".pyo")
_DIGEST_EXCLUDE_DIRS: tuple[str, ...] = ("__pycache__",)

# Stable evidence boundaries.  Paths are repository-relative and deliberately
# explicit so a dashboard-only change does not stale storage evidence.
EVIDENCE_SCOPES: dict[str, tuple[str, ...]] = {
    "restore_determinism": (
        "src/core/",
        "src/storage/",
        "src/learning/",
        "src/homeostasis/",
        "configs/",
        "tests/test_restore_determinism_abc.py",
        "tests/test_production_restore.py",
    ),
    "structural_e2e": (
        "src/self_organization/",
        "src/manipulation/",
        "src/storage/",
        "tests/test_structural_e2e.py",
    ),
    "structural_live_loop": (
        "src/self_organization/",
        "src/manipulation/",
        "src/storage/",
        "src/controller/",
        "tests/test_structural_live_loop.py",
    ),
    "runtime_integration": (
        "src/controller/",
        "src/dashboard/operator_bridge.py",
        "src/main.py",
        "tests/test_single_listener.py",
    ),
    "dashboard": (
        "src/dashboard/",
        "tests/test_dashboard.py",
        "tests/test_dashboard_completion.py",
    ),
    "research": (
        "src/research/",
        "research/schemas/",
        "tests/test_research_registry.py",
        "tests/test_experiment_validity.py",
    ),
    "release": ("src/", "configs/", "research/schemas/", "tests/"),
}

# ============================================================================
# Test baseline evaluation
# ============================================================================


def read_test_baseline(repo_root: Path) -> dict[str, Any] | None:
    """Read ``tests/test_baseline.json``.

    Returns ``None`` if the file is missing or unparseable.
    """
    baseline_path = repo_root / "tests" / "test_baseline.json"
    if not baseline_path.exists():
        return None
    try:
        return cast(
            "dict[str, Any]", json.loads(baseline_path.read_text(encoding="utf-8"))
        )
    except Exception:
        return None


def compute_source_tree_digest(
    repo_root: Path,
    paths: list[str] | None = None,
) -> str | None:
    """Return a SHA-256 digest of the scientifically relevant source tree.

    The digest covers ``src/``, ``configs/``, ``research/schemas/``,
    ``pyproject.toml`` and ``tests/`` (excluding ``test_baseline.json``
    so the baseline file cannot invalidate itself).

    The digest is computed from file contents (not git blobs) so it works
    in a dirty working tree and does not require a clean git state.
    Returns ``None`` if no files were found.
    """
    all_paths = paths if paths is not None else SCIENTIFIC_PATHS + TEST_PATHS
    try:
        hasher = hashlib.sha256()
        found_any = False
        for rel in all_paths:
            target = repo_root / rel
            if target.is_file():
                rel_posix = rel.rstrip("/")
                if rel_posix in _DIGEST_EXCLUDE_FILES:
                    continue
                hasher.update(rel_posix.encode("utf-8"))
                hasher.update(b"\0")
                hasher.update(target.read_bytes())
                hasher.update(b"\0")
                found_any = True
            elif target.is_dir():
                for path in sorted(target.rglob("*")):
                    if not path.is_file():
                        continue
                    rel_path = path.relative_to(repo_root).as_posix()
                    if rel_path in _DIGEST_EXCLUDE_FILES:
                        continue
                    if rel_path.endswith(_DIGEST_EXCLUDE_SUFFIXES):
                        continue
                    if any(
                        part in _DIGEST_EXCLUDE_DIRS
                        for part in path.relative_to(repo_root).parts
                    ):
                        continue
                    hasher.update(rel_path.encode("utf-8"))
                    hasher.update(b"\0")
                    hasher.update(path.read_bytes())
                    hasher.update(b"\0")
                    found_any = True
        if not found_any:
            return None
        return hasher.hexdigest()
    except Exception:
        return None


def compute_scope_digest(repo_root: Path, scope: str) -> str | None:
    """Return the content digest for one named evidence scope."""
    paths = EVIDENCE_SCOPES.get(scope)
    if paths is None:
        raise ValueError(f"unknown evidence scope: {scope}")
    return compute_source_tree_digest(repo_root, list(paths))


def artifact_digest_matches(repo_root: Path, artifact: dict[str, Any]) -> bool:
    """Validate scoped artifact digests, falling back to legacy digests."""
    scope = artifact.get("scope")
    scope_digest = artifact.get("scope_digest")
    if isinstance(scope, str) and isinstance(scope_digest, str):
        try:
            return scope_digest == compute_scope_digest(repo_root, scope)
        except ValueError:
            return False
    legacy_digest = artifact.get("tested_tree_digest")
    return isinstance(
        legacy_digest, str
    ) and legacy_digest == compute_source_tree_digest(repo_root)


def current_git_head(repo_root: Path) -> str | None:
    """Return the current git HEAD SHA, or None if unavailable."""
    try:
        result = subprocess.run(
            ["git", "rev-parse", "HEAD"],
            capture_output=True,
            text=True,
            timeout=5,
            cwd=str(repo_root),
        )
        if result.returncode == 0:
            return result.stdout.strip()
    except Exception:
        pass
    return None


class BaselineEvaluation:
    """Result of evaluating the test baseline against the current tree."""

    def __init__(
        self,
        *,
        stale: bool,
        available: bool,
        passed: int,
        failed: int,
        skipped: int,
        collection_errors: int,
        tested_commit: str | None,
        current_commit: str | None,
        tested_tree_digest: str | None,
        current_tree_digest: str | None,
    ) -> None:
        self.stale = stale
        self.available = available
        self.passed = passed
        self.failed = failed
        self.skipped = skipped
        self.collection_errors = collection_errors
        self.tested_commit = tested_commit
        self.current_commit = current_commit
        self.tested_tree_digest = tested_tree_digest
        self.current_tree_digest = current_tree_digest


def evaluate_test_baseline(repo_root: Path) -> BaselineEvaluation:
    """Evaluate ``tests/test_baseline.json`` against the current source tree.

    Handles both the new format (``full_suite`` + ``full_collection``) and
    the legacy format (``verified_subset``) so that old baselines do not
    silently report zero counts.

    Returns a :class:`BaselineEvaluation` with ``stale=True`` when the
    scientifically relevant source tree changed since the baseline.
    """
    baseline = read_test_baseline(repo_root)
    current_commit = current_git_head(repo_root)
    current_tree_digest = compute_source_tree_digest(repo_root)

    if baseline is None:
        return BaselineEvaluation(
            stale=True,
            available=False,
            passed=0,
            failed=0,
            skipped=0,
            collection_errors=1,
            tested_commit=None,
            current_commit=current_commit,
            tested_tree_digest=None,
            current_tree_digest=current_tree_digest,
        )

    # --- Counts: support both new (full_suite) and legacy (verified_subset) ---
    full_suite = baseline.get("full_suite", {})
    full_collection = baseline.get("full_collection", {})
    legacy_subset = baseline.get("verified_subset", {})

    if full_suite:
        passed = int(full_suite.get("passed", 0))
        failed = int(full_suite.get("failed", 0))
        skipped = int(full_suite.get("skipped", 0))
    elif legacy_subset:
        passed = int(legacy_subset.get("passed", 0))
        failed = int(legacy_subset.get("failed", 0))
        skipped = int(legacy_subset.get("skipped", 0))
    else:
        passed = 0
        failed = 0
        skipped = 0

    if full_collection:
        collection_errors = int(full_collection.get("collection_errors", 1))
    else:
        collection_errors = 0 if passed > 0 else 1

    tested_commit = baseline.get("tested_commit")
    tested_tree_digest = baseline.get("tested_tree_digest")

    stale = True
    if tested_tree_digest is not None and current_tree_digest is not None:
        stale = tested_tree_digest != current_tree_digest

    return BaselineEvaluation(
        stale=stale,
        available=True,
        passed=passed,
        failed=failed,
        skipped=skipped,
        collection_errors=collection_errors,
        tested_commit=tested_commit,
        current_commit=current_commit,
        tested_tree_digest=tested_tree_digest,
        current_tree_digest=current_tree_digest,
    )