MHRN-Space / src /dashboard /verification.py
ThomasHeisig's picture
Sync GitHub main 6bb97fb9a95926f8118bd61bec366af28afa0e5c (part 8)
c4e7f6b verified
Raw History Blame Contribute Delete
23.5 kB
"""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"<missing>")
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"<missing>"
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,
)