"""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 platform import subprocess import threading from dataclasses import dataclass 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__",) _INSPECTION_CACHE_LOCK = threading.RLock() _INSPECTION_CACHE: dict[tuple[str, tuple[str, ...], str], "SourceTreeInspection"] = {} _INSPECTION_CACHE_MAX_ENTRIES = 32 @dataclass(frozen=True, slots=True) class SourceTreeInspection: """Deterministic source-freeze details used by gates and diagnostics.""" digest: str | None dirty_relevant_paths: tuple[str, ...] untracked_relevant_paths: tuple[str, ...] missing_relevant_paths: tuple[str, ...] mismatching_files: tuple[str, ...] platform: str line_ending_normalization_mode: str git_available: bool def _is_digest_file(path: Path, repo_root: Path) -> bool: """Return whether a filesystem path belongs to the digest scope.""" if not path.is_file(): return False relative = path.relative_to(repo_root).as_posix() if relative in _DIGEST_EXCLUDE_FILES or relative.endswith(_DIGEST_EXCLUDE_SUFFIXES): return False return not any( part in _DIGEST_EXCLUDE_DIRS or part.endswith(".egg-info") for part in path.relative_to(repo_root).parts ) def _filesystem_digest_paths(repo_root: Path, paths: list[str]) -> list[str]: """Return digest paths in the historical, reproducible traversal order.""" result: list[str] = [] for relative in paths: target = repo_root / relative if target.is_file(): if _is_digest_file(target, repo_root): result.append(relative.rstrip("/")) elif target.is_dir(): result.extend( path.relative_to(repo_root).as_posix() for path in sorted( target.rglob("*"), key=lambda item: item.relative_to(repo_root).parts, ) if _is_digest_file(path, repo_root) ) return result def _git_output( repo_root: Path, arguments: list[str], input_data: bytes = b"" ) -> bytes | None: """Run a bounded, read-only Git command.""" try: result = subprocess.run( ["git", *arguments], cwd=str(repo_root), input=input_data, capture_output=True, timeout=10, check=False, ) except (OSError, subprocess.SubprocessError): return None if result.returncode != 0: return None return result.stdout def _git_scope_paths( repo_root: Path, paths: list[str], *, untracked: bool ) -> list[str]: arguments = ["ls-files", "-z"] if untracked: arguments.extend(["--others", "--exclude-standard"]) arguments.extend(["--", *paths]) output = _git_output(repo_root, arguments) if output is None: return [] return [ item.decode("utf-8", "surrogateescape") for item in output.split(b"\0") if item ] def _git_changed_paths(repo_root: Path, paths: list[str]) -> set[str]: output = _git_output(repo_root, ["diff", "HEAD", "--name-only", "-z", "--", *paths]) if output is None: return set() return { item.decode("utf-8", "surrogateescape") for item in output.split(b"\0") if item } def _git_index_blobs( repo_root: Path, requested_paths: set[str] | None = None ) -> dict[str, bytes]: """Read stage-zero index blobs in one Git batch operation.""" listing = _git_output(repo_root, ["ls-files", "-s", "-z"]) if listing is None: return {} entries: list[tuple[str, str]] = [] for raw_entry in listing.split(b"\0"): if not raw_entry: continue header, separator, raw_path = raw_entry.partition(b"\t") fields = header.split() if not separator or len(fields) != 3 or fields[2] != b"0": continue relative = raw_path.decode("utf-8", "surrogateescape") if requested_paths is not None and relative not in requested_paths: continue entries.append( ( relative, fields[1].decode("ascii"), ) ) if not entries: return {} request = b"".join(f"{object_id}\n".encode("ascii") for _, object_id in entries) output = _git_output(repo_root, ["cat-file", "--batch"], request) if output is None: return {} blobs: dict[str, bytes] = {} offset = 0 for relative, object_id in entries: header_end = output.find(b"\n", offset) if header_end < 0: break fields = output[offset:header_end].split() offset = header_end + 1 if len(fields) != 3 or fields[0].decode("ascii", "ignore") != object_id: break size = int(fields[2]) blobs[relative] = output[offset : offset + size] offset += size + 1 return blobs def _git_text_paths(repo_root: Path, paths: list[str]) -> set[str]: """Resolve Git's text attribute for paths using one batch query.""" if not paths: return set() input_data = ("\n".join(paths) + "\n").encode("utf-8", "surrogateescape") output = _git_output(repo_root, ["check-attr", "--stdin", "text"], input_data) if output is None: return set() text_paths: set[str] = set() for raw_line in output.splitlines(): try: relative, _attribute, value = raw_line.decode( "utf-8", "surrogateescape" ).rsplit(": ", 2) except ValueError: continue if value in {"set", "auto"}: text_paths.add(relative) return text_paths def _canonical_text_bytes(data: bytes) -> bytes: """Apply Git's portable text line-ending representation without decoding.""" return data.replace(b"\r\n", b"\n") def _digest_bytes(path: Path, relative: str, text_paths: set[str]) -> bytes: data = path.read_bytes() if relative in text_paths or (relative not in text_paths and b"\0" not in data): return _canonical_text_bytes(data) return data def _source_state_fingerprint(repo_root: Path, paths: list[str]) -> str | None: """Return a cheap content-bound key for one Git working-tree scope. The key includes HEAD, the complete tracked diff against HEAD and relevant untracked file bytes. It therefore changes for staged, unstaged and untracked edits while avoiding a full clean-tree blob scan on every dashboard poll. """ head = _git_output(repo_root, ["rev-parse", "HEAD"]) if head is None: return None diff = _git_output(repo_root, ["diff", "--binary", "HEAD", "--", *paths]) if diff is None: return None untracked = _git_scope_paths(repo_root, paths, untracked=True) hasher = hashlib.sha256() hasher.update(head) hasher.update(b"\0") hasher.update(diff) hasher.update(b"\0") for relative in sorted(untracked): path = repo_root / relative if not _is_digest_file(path, repo_root): continue hasher.update(relative.encode("utf-8", "surrogateescape")) hasher.update(b"\0") try: hasher.update(path.read_bytes()) except OSError: hasher.update(b"") hasher.update(b"\0") return hasher.hexdigest() def _cached_source_tree_inspection( repo_root: Path, paths: list[str] | None = None ) -> SourceTreeInspection: """Reuse an inspection only while its exact Git/content fingerprint matches.""" all_paths = paths if paths is not None else SCIENTIFIC_PATHS + TEST_PATHS fingerprint = _source_state_fingerprint(repo_root, all_paths) if fingerprint is None: return inspect_source_tree(repo_root, all_paths) key = (str(repo_root.resolve()), tuple(all_paths), fingerprint) with _INSPECTION_CACHE_LOCK: cached = _INSPECTION_CACHE.get(key) if cached is not None: return cached inspection = inspect_source_tree(repo_root, all_paths) with _INSPECTION_CACHE_LOCK: if len(_INSPECTION_CACHE) >= _INSPECTION_CACHE_MAX_ENTRIES: _INSPECTION_CACHE.clear() _INSPECTION_CACHE[key] = inspection return inspection def inspect_source_tree( repo_root: Path, paths: list[str] | None = None, ) -> SourceTreeInspection: """Inspect and hash the source freeze with Git-aware text canonicalization.""" all_paths = paths if paths is not None else SCIENTIFIC_PATHS + TEST_PATHS filesystem_paths = _filesystem_digest_paths(repo_root, all_paths) tracked_paths = _git_scope_paths(repo_root, all_paths, untracked=False) untracked_paths = _git_scope_paths(repo_root, all_paths, untracked=True) git_available = bool( tracked_paths or _git_output(repo_root, ["rev-parse", "--git-dir"]) ) changed_paths: set[str] = ( _git_changed_paths(repo_root, all_paths) if git_available else set() ) relevant_untracked = sorted( path for path in untracked_paths if _is_digest_file(repo_root / path, repo_root) ) tracked_set = { path for path in tracked_paths if _is_digest_file(repo_root / path, repo_root) } filesystem_set = set(filesystem_paths) missing_paths = sorted(tracked_set - filesystem_set) dirty_paths = sorted(path for path in changed_paths if path in tracked_set) mismatching = tuple( sorted(set(dirty_paths) | set(relevant_untracked) | set(missing_paths)) ) text_paths: set[str] = ( _git_text_paths( repo_root, sorted(set(filesystem_paths) | set(relevant_untracked)) ) if git_available else set() ) index_blobs = _git_index_blobs(repo_root, tracked_set) if git_available else {} hasher = hashlib.sha256() found_any = False ordered_paths = list(filesystem_paths) ordered_paths.extend(path for path in missing_paths if path not in filesystem_set) for relative in ordered_paths: path = repo_root / relative if ( relative in index_blobs and relative not in dirty_paths and relative not in missing_paths ): data = index_blobs[relative] elif path.is_file(): data = _digest_bytes(path, relative, text_paths) else: data = b"" hasher.update(relative.encode("utf-8", "surrogateescape")) hasher.update(b"\0") hasher.update(data) hasher.update(b"\0") found_any = True return SourceTreeInspection( digest=hasher.hexdigest() if found_any else None, dirty_relevant_paths=tuple(dirty_paths), untracked_relevant_paths=tuple(relevant_untracked), missing_relevant_paths=tuple(missing_paths), mismatching_files=mismatching, platform=platform.platform(), line_ending_normalization_mode="git-text-crlf-to-lf;binary-byte-exact", git_available=git_available, ) def canonical_source_file_bytes(repo_root: Path, relative: str) -> bytes: """Return one working-tree file in the source-freeze representation.""" path = repo_root / relative if not path.is_file(): raise FileNotFoundError(relative) text_paths = _git_text_paths(repo_root, [relative]) return _digest_bytes(path, relative, text_paths) def source_digest_paths(repo_root: Path, paths: list[str] | None = None) -> list[str]: """Return the included repository-relative paths in digest order.""" return _filesystem_digest_paths( repo_root, paths if paths is not None else SCIENTIFIC_PATHS + TEST_PATHS ) def git_source_blob(repo_root: Path, relative: str) -> bytes | None: """Read one HEAD blob for human-readable source-digest diagnostics.""" return _git_output(repo_root, ["show", f"HEAD:{relative}"]) # 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). Tracked clean files use their canonical Git index blobs. Working-tree changes and relevant untracked files are included after Git-compatible text line-ending normalization; binary bytes remain exact. Returns ``None`` if no files were found. """ try: return _cached_source_tree_inspection(repo_root, paths).digest except (OSError, ValueError, subprocess.SubprocessError): return None def compute_legacy_raw_source_tree_digest( repo_root: Path, paths: list[str] | None = None, ) -> str | None: """Compute the pre-fix raw-byte digest for migration diagnostics only.""" all_paths = paths if paths is not None else SCIENTIFIC_PATHS + TEST_PATHS try: hasher = hashlib.sha256() found_any = False for relative in _filesystem_digest_paths(repo_root, all_paths): hasher.update(relative.encode("utf-8", "surrogateescape")) hasher.update(b"\0") hasher.update((repo_root / relative).read_bytes()) hasher.update(b"\0") found_any = True return hasher.hexdigest() if found_any else None except OSError: 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, dirty_relevant_paths: tuple[str, ...] = (), untracked_relevant_paths: tuple[str, ...] = (), mismatching_files: tuple[str, ...] = (), stale_reason: str | None = None, platform_name: str | None = None, line_ending_normalization_mode: str = "git-text-crlf-to-lf;binary-byte-exact", ) -> 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 self.dirty_relevant_paths = dirty_relevant_paths self.untracked_relevant_paths = untracked_relevant_paths self.mismatching_files = mismatching_files self.stale_reason = stale_reason self.platform = platform_name or platform.platform() self.line_ending_normalization_mode = line_ending_normalization_mode 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) inspection = _cached_source_tree_inspection(repo_root) current_tree_digest = inspection.digest 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, dirty_relevant_paths=inspection.dirty_relevant_paths, untracked_relevant_paths=inspection.untracked_relevant_paths, mismatching_files=inspection.mismatching_files, platform_name=inspection.platform, line_ending_normalization_mode=inspection.line_ending_normalization_mode, ) # --- 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 stale_reason: str | None = "missing_tree_digest" if tested_tree_digest is not None and current_tree_digest is not None: local_tree_changed = bool(inspection.mismatching_files) stale = tested_tree_digest != current_tree_digest or local_tree_changed if not stale: stale_reason = None elif inspection.dirty_relevant_paths or inspection.missing_relevant_paths: stale_reason = "working_tree_content_diff" elif inspection.untracked_relevant_paths: stale_reason = "untracked_relevant_path" elif tested_tree_digest == compute_legacy_raw_source_tree_digest(repo_root): stale_reason = "legacy_platform_digest_mismatch" else: stale_reason = "verification_artifact_scope_mismatch" 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, dirty_relevant_paths=inspection.dirty_relevant_paths, untracked_relevant_paths=inspection.untracked_relevant_paths, mismatching_files=inspection.mismatching_files, stale_reason=stale_reason, platform_name=inspection.platform, line_ending_normalization_mode=inspection.line_ending_normalization_mode, )