Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 2)
094b823 verified | """ | |
| Benchmark coverage — "how much of Aureon OS is pinned?" made falsifiable. | |
| ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ | |
| The end goal is a full 100% benchmark of the Aureon OS. This module is the | |
| measuring instrument for that march: it reconciles the COMMITTED Tier-A | |
| benchmark report (``tests/benchmarks/report.json`` — every row names the | |
| module it pins) against the real filesystem (every top-level package under | |
| ``aureon/`` and every ``.py`` module inside it) and reports, with nothing | |
| invented: | |
| - which domains carry at least one Tier-A pin (covered), | |
| - which domains carry NONE (uncovered — the gap list is the roadmap), | |
| - how many unique modules are pinned, out of how many exist. | |
| The **ratchet** makes progress one-way: a committed baseline | |
| (``docs/BENCHMARK_COVERAGE.json``) records the covered-domain set and the | |
| pinned-module count; a later derivation may only grow both. Any regression is | |
| named (which domain fell out, which count went down) — never smoothed over. | |
| Honest by construction: everything is derived from the committed report and | |
| the disk. A missing report is an honest empty coverage with the blocker | |
| named, never a guessed number. A stale or superseded report is never promoted | |
| to a current measurement, even if its historical rows all say ``passed``. | |
| Gary Leckey · Aureon Institute | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import logging | |
| from dataclasses import dataclass, field | |
| from pathlib import Path | |
| from typing import Any, Dict, List | |
| logger = logging.getLogger("aureon.analytics.benchmark_coverage") | |
| _REPO_ROOT = Path(__file__).resolve().parents[2] | |
| _REPORT = _REPO_ROOT / "tests" / "benchmarks" / "report.json" | |
| _BASELINE = _REPO_ROOT / "docs" / "BENCHMARK_COVERAGE.json" | |
| _SKIP = {"__pycache__"} | |
| def _fs_domains(repo_root: Path) -> Dict[str, int]: | |
| """Real top-level packages under ``aureon/`` → recursive ``.py`` module count.""" | |
| out: Dict[str, int] = {} | |
| aureon = repo_root / "aureon" | |
| try: | |
| for child in sorted(aureon.iterdir()): | |
| if child.is_dir() and child.name not in _SKIP and (child / "__init__.py").exists(): | |
| out[child.name] = sum(1 for p in child.rglob("*.py") | |
| if "__pycache__" not in p.parts) | |
| except OSError as exc: # noqa: BLE001 — a missing tree is an honest empty, never a crash | |
| logger.debug("benchmark_coverage: cannot list aureon/ (%s)", exc) | |
| return out | |
| def _domain_of(module: str) -> str | None: | |
| """``aureon/<domain>/...`` → domain; anything else → None (outside the package).""" | |
| parts = module.split("/") | |
| if len(parts) >= 3 and parts[0] == "aureon": | |
| return parts[1] | |
| return None | |
| class BenchmarkCoverage: | |
| status: str # "measured" | "stale_superseded" | "honest_unavailable" | |
| blocker: str = "" | |
| benchmarks: int = 0 | |
| all_rows_passed: bool = False | |
| pinned_modules: List[str] = field(default_factory=list) # unique, sorted | |
| missing_modules: List[str] = field(default_factory=list) # named in report, absent on disk | |
| domains: Dict[str, Dict[str, Any]] = field(default_factory=dict) | |
| outside_aureon: List[str] = field(default_factory=list) # pins outside aureon/ (e.g. scripts/) | |
| covered_domains: List[str] = field(default_factory=list) | |
| uncovered_domains: List[str] = field(default_factory=list) | |
| domain_coverage_fraction: float | None = None | |
| total_modules: int = 0 | |
| module_pin_count: int = 0 | |
| def to_dict(self) -> Dict[str, Any]: | |
| return { | |
| "status": self.status, "blocker": self.blocker, | |
| "benchmarks": self.benchmarks, "all_rows_passed": self.all_rows_passed, | |
| "pinned_modules": self.pinned_modules, | |
| "missing_modules": self.missing_modules, | |
| "domains": self.domains, "outside_aureon": self.outside_aureon, | |
| "covered_domains": self.covered_domains, | |
| "uncovered_domains": self.uncovered_domains, | |
| "domain_coverage_fraction": self.domain_coverage_fraction, | |
| "total_modules": self.total_modules, | |
| "module_pin_count": self.module_pin_count, | |
| } | |
| def build_coverage(repo_root: Path | None = None, | |
| report_path: Path | None = None) -> BenchmarkCoverage: | |
| """Derive the coverage map from the committed Tier-A report + the disk.""" | |
| root = Path(repo_root) if repo_root else _REPO_ROOT | |
| rp = Path(report_path) if report_path else (root / "tests" / "benchmarks" / "report.json") | |
| try: | |
| report = json.loads(rp.read_text(encoding="utf-8")) | |
| except (OSError, ValueError) as exc: | |
| return BenchmarkCoverage(status="honest_unavailable", | |
| blocker=f"committed report unreadable: {exc}") | |
| if not isinstance(report, dict): | |
| return BenchmarkCoverage( | |
| status="honest_unavailable", | |
| blocker="committed report must contain a JSON object", | |
| ) | |
| report_status = report.get("report_status") | |
| production_ready = report.get("production_ready") | |
| current_effect_claim = report.get("current_effect_claim") | |
| if not ( | |
| report_status == "CURRENT" | |
| and production_ready is True | |
| and current_effect_claim is True | |
| ): | |
| return BenchmarkCoverage( | |
| status="stale_superseded", | |
| blocker=( | |
| "committed report is not current release evidence: " | |
| f"report_status={report_status!r}, " | |
| f"production_ready={production_ready is True}, " | |
| f"current_effect_claim={current_effect_claim is True}" | |
| ), | |
| ) | |
| rows = report.get("tier_a", []) | |
| if not isinstance(rows, list) or not all(isinstance(row, dict) for row in rows): | |
| return BenchmarkCoverage( | |
| status="honest_unavailable", | |
| blocker="current report tier_a must be a list of objects", | |
| ) | |
| if not rows: | |
| return BenchmarkCoverage(status="honest_unavailable", | |
| blocker="committed report carries no tier_a rows") | |
| mods = sorted({str(r.get("module", "")) for r in rows if r.get("module")}) | |
| missing = [m for m in mods if not (root / m).exists()] | |
| fs = _fs_domains(root) | |
| domains: Dict[str, Dict[str, Any]] = {} | |
| for d, count in fs.items(): | |
| pinned = sorted(m for m in mods if _domain_of(m) == d) | |
| domains[d] = {"modules": count, "pinned": pinned, "covered": bool(pinned)} | |
| outside = sorted(m for m in mods if _domain_of(m) is None) | |
| covered = sorted(d for d, v in domains.items() if v["covered"]) | |
| uncovered = sorted(d for d, v in domains.items() if not v["covered"]) | |
| return BenchmarkCoverage( | |
| status="measured", | |
| benchmarks=len(rows), | |
| all_rows_passed=all(bool(r.get("passed")) for r in rows), | |
| pinned_modules=mods, | |
| missing_modules=missing, | |
| domains=domains, | |
| outside_aureon=outside, | |
| covered_domains=covered, | |
| uncovered_domains=uncovered, | |
| domain_coverage_fraction=(round(len(covered) / len(fs), 4) if fs else None), | |
| total_modules=sum(fs.values()), | |
| module_pin_count=len(mods), | |
| ) | |
| def ratchet_check(live: BenchmarkCoverage, | |
| baseline: Dict[str, Any] | None) -> Dict[str, Any]: | |
| """One-way progress: live coverage may only GROW relative to the baseline. | |
| Regressions are NAMED — which domain fell out of coverage, which count went | |
| down. No baseline yet → ratchet passes trivially (first measurement seeds it). | |
| """ | |
| if not baseline: | |
| return {"ok": True, "regressions": [], "note": "no baseline — first measurement seeds it"} | |
| regressions: List[str] = [] | |
| base_covered = set(baseline.get("covered_domains", [])) | |
| lost = sorted(base_covered - set(live.covered_domains)) | |
| if lost: | |
| regressions.append(f"domains fell out of coverage: {lost}") | |
| base_pins = int(baseline.get("module_pin_count", 0)) | |
| if live.module_pin_count < base_pins: | |
| regressions.append( | |
| f"pinned-module count fell: {live.module_pin_count} < baseline {base_pins}") | |
| base_benchmarks = int(baseline.get("benchmarks", 0)) | |
| if live.benchmarks < base_benchmarks: | |
| regressions.append( | |
| f"benchmark count fell: {live.benchmarks} < baseline {base_benchmarks}") | |
| return {"ok": not regressions, "regressions": regressions, "note": ""} | |
| def load_baseline(path: Path | None = None) -> Dict[str, Any] | None: | |
| p = Path(path) if path else _BASELINE | |
| try: | |
| return json.loads(p.read_text(encoding="utf-8")) | |
| except (OSError, ValueError): | |
| return None | |
| def write_coverage(cov: BenchmarkCoverage, out_md: Path, out_json: Path) -> None: | |
| out_json.write_text(json.dumps(cov.to_dict(), indent=2), encoding="utf-8") | |
| lines = [ | |
| "# Benchmark coverage — the march to 100%", "", | |
| "> Derived from the committed Tier-A report + the filesystem. Nothing invented;", | |
| "> the uncovered list is the roadmap, the ratchet makes progress one-way.", "", | |
| f"- Status: **{cov.status}**", | |
| *([f"- Blocker: {cov.blocker}"] if cov.blocker else []), | |
| f"- Tier-A benchmarks: **{cov.benchmarks}** (all passed: {cov.all_rows_passed})", | |
| f"- Unique modules pinned: **{cov.module_pin_count}** of {cov.total_modules} on disk", | |
| f"- Domain coverage: **{len(cov.covered_domains)}/{len(cov.domains)}**" | |
| f" ({cov.domain_coverage_fraction})", "", | |
| "## Per-domain", "", | |
| "| Domain | Modules | Pinned | Covered |", "|---|---|---|---|", | |
| ] | |
| for d, v in sorted(cov.domains.items()): | |
| lines.append(f"| {d} | {v['modules']} | {len(v['pinned'])} | " | |
| f"{'yes' if v['covered'] else '**no**'} |") | |
| if cov.outside_aureon: | |
| lines += ["", "## Pins outside `aureon/`", ""] | |
| lines += [f"- `{m}`" for m in cov.outside_aureon] | |
| lines += ["", "## Uncovered domains (the roadmap)", ""] | |
| lines += [f"- `{d}` — {cov.domains[d]['modules']} modules, zero pins" | |
| for d in cov.uncovered_domains] or ["(none — 100% domain coverage)"] | |
| lines.append("") | |
| out_md.write_text("\n".join(lines), encoding="utf-8") | |
| def main(argv: List[str] | None = None) -> int: | |
| import argparse | |
| ap = argparse.ArgumentParser(description=__doc__) | |
| ap.add_argument("--out", type=Path, default=_REPO_ROOT / "docs" / "BENCHMARK_COVERAGE.md") | |
| args = ap.parse_args(argv) | |
| cov = build_coverage() | |
| ratchet = ratchet_check(cov, load_baseline()) | |
| write_coverage(cov, args.out, args.out.with_suffix(".json")) | |
| print(f"benchmarks={cov.benchmarks} pinned_modules={cov.module_pin_count}" | |
| f"/{cov.total_modules} domains={len(cov.covered_domains)}/{len(cov.domains)}" | |
| f" ratchet_ok={ratchet['ok']}") | |
| for r in ratchet["regressions"]: | |
| print(f"REGRESSION: {r}") | |
| return 0 if (cov.status == "measured" and ratchet["ok"]) else 1 | |
| __all__ = ["BenchmarkCoverage", "build_coverage", "ratchet_check", | |
| "load_baseline", "write_coverage", "main"] | |
| if __name__ == "__main__": # pragma: no cover | |
| raise SystemExit(main()) | |