scripts.freshness_guard

Reference › scripts › freshness_guard

The freshness guard: it detects, and refreshes, each class of stale data behind the 2026-08-04 incident. tests/scripts/test_freshness_guard.py imports it.

Invocation: main takes --check. With it, the guard only detects and exits 1 if anything is stale, refreshing nothing. Without it, the guard refreshes what it finds stale and exits 1 only if some check could not be brought to an ok state.

Six checks run in turn, each given a refresh flag and returning whether the data was stale and whether it is ok: check_ast_runtime_merge (the AST snapshot merged with the runtime call graph, see build_runtime_callgraph), check_prolog_kbs (the Prolog knowledge bases built from the snapshot), check_durations_sink (the test durations file), check_observer_registry (the test observer registry in a run directory), check_packer_cache (the context packer cache) and check_capture_age (the age of the runtime call-graph capture, detect only). _report prints one status line per check. A checks argument lets a caller supply its own checks.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

No internal imports and no importers.

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/freshness_guard.py

"""Freshness guard — detect and refresh every stale-data class the 2026-08-04
run history hit (Mohamed ruling: "no staleness, every run is fresh").

Each check is a (detect, refresh) pair over one artifact. Run all via
``python scripts/freshness_guard.py`` (exit 0 = everything fresh or refreshed,
exit 1 = a refresh failed); ``--check`` reports without refreshing (non-zero
when anything is stale). ``scripts/run_tests.py`` invokes this before the test
phase so no gate run ever consumes stale inputs.

The classes, each traced to a measured incident that day:

1. **AST snapshot missing its runtime-call merge** (`plans/ast-ultimate.json`):
   a bare ``ast-sync`` regeneration drops every function's ``runtime_calls``
   field, silently un-counting DI/dispatch edges (derive_rtm fell 1055->575).
   Detect: zero ``runtime_calls`` entries while ``plans/runtime-callgraph.json``
   exists. Refresh: ``build_runtime_callgraph.py reaugment`` (in-place).

2. **Prolog KBs older than the snapshot** (``.cache/trace_intel/*.pl``,
   ``.claude/run/codebase_kb/facts.pl``): stale structural facts made
   derive_rtm abort. Detect: facts file missing or mtime < snapshot mtime.
   Refresh: delegate to ``scripts.trace_intel.ensure_kb`` (its own mtime logic,
   single source of truth) for the trace_intel pair; reindex the code_intel KB
   via ``scripts/code_intel.py reindex`` only when stale.

3. **Duration-sink bloat** (`reports/test-durations.ndjson`): grew to 267MB /
   1.78M lines including July records of deleted tests (a 256s entry for a
   decommissioned module distorted every pole analysis and cost ~4.4s per
   runner invocation). Detect: any record older than the retention window OR
   referencing a test file that no longer exists. Refresh: compact in place,
   keeping only fresh records of existing files.

4. **Observer-registry ghosts** (``$XDG_RUNTIME_DIR/orchestrata/
   pytest-sessions/*.json``): sessions whose process died without finalizing
   stay ``status=running`` forever, corrupting live counts ("11 running" with
   zero live pids). Detect: status==running and pid dead. Refresh: mark
   ``status=killed`` (facts preserved — never faked to "finished").

5. **Packer duration-cache dead entries** (``.cache/test_durations.json``):
   per-module durations for modules that no longer exist skew duration-based
   bin packing (worker idle). Detect: keys whose module path is gone.
   Refresh: drop those keys.

6. **Runtime capture age** (`plans/runtime-callgraph.json`): report-only —
   a fresh capture requires an instrumented suite run (per-batch env, NOT a
   global PYTEST_ADDOPTS, which leaks into nested pytest and broke 6 modules
   on 2026-08-04). The guard reports age so the operator schedules a capture
   run; it never silently re-captures.
"""

from __future__ import annotations

import argparse
import json
import os
import subprocess
import sys
import time
from collections.abc import Callable, Sequence
from pathlib import Path

REPO_ROOT = Path(__file__).resolve().parent.parent
DURATIONS_NDJSON = REPO_ROOT / "reports" / "test-durations.ndjson"
AST_SNAPSHOT = REPO_ROOT / "plans" / "ast-ultimate.json"
RUNTIME_CALLGRAPH = REPO_ROOT / "plans" / "runtime-callgraph.json"
PACKER_CACHE = REPO_ROOT / ".cache" / "test_durations.json"
CODE_INTEL_KB = REPO_ROOT / ".claude" / "run" / "codebase_kb" / "facts.pl"
RETENTION_DAYS = 7.0
CAPTURE_MAX_AGE_DAYS = 7.0


def _report(name: str, status: str, detail: str) -> None:
    print(f"[freshness] {name}: {status} — {detail}")


def check_ast_runtime_merge(
    refresh: bool,
    *,
    ast_snapshot: Path | None = None,
    runtime_callgraph: Path | None = None,
) -> tuple[bool, bool]:
    """(1) ast-ultimate.json must carry runtime_calls when a capture exists."""
    snapshot = ast_snapshot if ast_snapshot is not None else AST_SNAPSHOT
    callgraph = (
        runtime_callgraph if runtime_callgraph is not None else RUNTIME_CALLGRAPH
    )
    if not snapshot.exists() or not callgraph.exists():
        _report("ast-runtime-merge", "SKIP", "snapshot or capture absent")
        return False, True
    data = json.loads(snapshot.read_text(encoding="utf-8"))
    n = 0
    for mod in data.get("modules", []):
        for fn in mod.get("functions", []):
            n += len(fn.get("runtime_calls", []))
        for cls in mod.get("classes", []):
            for meth in cls.get("methods", []):
                n += len(meth.get("runtime_calls", []))
        if n:
            break
    if n:
        _report("ast-runtime-merge", "FRESH", "runtime_calls present")
        return False, True
    if not refresh:
        _report("ast-runtime-merge", "STALE", "0 runtime_calls in snapshot")
        return True, True
    proc = subprocess.run(
        [
            sys.executable,
            str(REPO_ROOT / "scripts" / "build_runtime_callgraph.py"),
            "reaugment",
        ],
        capture_output=True,
        text=True,
        timeout=600,
        check=False,
        cwd=REPO_ROOT,
    )
    ok = proc.returncode == 0
    _report(
        "ast-runtime-merge",
        "REFRESHED" if ok else "REFRESH-FAILED",
        (
            (proc.stdout.strip().splitlines() or ["reaugment"])[-1]
            if ok
            else proc.stderr[-300:]
        ),
    )
    return True, ok


def check_prolog_kbs(
    refresh: bool, *, ast_snapshot: Path | None = None
) -> tuple[bool, bool]:
    """(2) trace_intel + code_intel KBs no older than the AST snapshot."""
    snapshot = ast_snapshot if ast_snapshot is not None else AST_SNAPSHOT
    if not snapshot.exists():
        _report("prolog-kbs", "SKIP", "no snapshot")
        return False, True
    snap_m = snapshot.stat().st_mtime
    sys.path.insert(0, str(REPO_ROOT / "scripts"))
    import trace_intel as t  # noqa: PLC0415  (path-dependent script import)

    stale_paths = [
        p
        for p in (t.STRUCTURAL_FACTS, t.TRACEABILITY_FACTS, CODE_INTEL_KB)
        if not p.exists() or p.stat().st_mtime < snap_m
    ]
    if not stale_paths:
        _report("prolog-kbs", "FRESH", "all KBs newer than snapshot")
        return False, True
    if not refresh:
        _report("prolog-kbs", "STALE", ", ".join(str(p) for p in stale_paths))
        return True, True
    t.ensure_kb(no_refresh=False)
    if CODE_INTEL_KB in stale_paths:
        proc = subprocess.run(
            [sys.executable, str(REPO_ROOT / "scripts" / "code_intel.py"), "reindex"],
            capture_output=True,
            text=True,
            timeout=600,
            check=False,
            cwd=REPO_ROOT,
        )
        if proc.returncode != 0:
            _report("prolog-kbs", "REFRESH-FAILED", proc.stderr[-300:])
            return True, False
    _report("prolog-kbs", "REFRESHED", f"{len(stale_paths)} KB file(s) rebuilt")
    return True, True


def check_durations_sink(
    refresh: bool,
    now: float | None = None,
    *,
    sink_path: Path | None = None,
    repo_root: Path | None = None,
) -> tuple[bool, bool]:
    """(3) compact the duration sink: recent records of existing files only."""
    sink = sink_path if sink_path is not None else DURATIONS_NDJSON
    root = repo_root if repo_root is not None else REPO_ROOT
    if not sink.exists():
        _report("durations-sink", "SKIP", "absent")
        return False, True
    now = now if now is not None else time.time()
    cutoff = now - RETENTION_DAYS * 86400
    kept: list[str] = []
    dropped = 0
    seen_files: dict[str, bool] = {}
    with sink.open(encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                dropped += 1
                continue
            if not isinstance(obj, dict):
                # Valid JSON but the wrong shape (e.g. a bare list/scalar) —
                # treat the same as malformed: drop, never crash the compactor.
                dropped += 1
                continue
            ts = obj.get("ts")
            if not isinstance(ts, (int, float)) or ts < cutoff:
                dropped += 1
                continue
            if obj.get("kind") == "run":
                # A run-identity HEADER names no file to existence-check: it
                # has no "test" key, so before this branch existed
                # test = obj.get("test", "") -> "" -> fpath="" ->
                # (root / "").exists() == root.exists() == True. Headers
                # survived compaction ONLY by that pathlib accident. Made
                # explicit here: headers are exempt from the file-existence
                # check by design (they name no file); the ts-freshness
                # check above is the only retention rule that applies to
                # them.
                kept.append(line)
                continue
            test = obj.get("test", "")
            fpath = test.split("::", 1)[0]
            exists = seen_files.get(fpath)
            if exists is None:
                exists = (root / fpath).exists()
                seen_files[fpath] = exists
            if not exists:
                dropped += 1
                continue
            kept.append(line)
    if not dropped:
        _report("durations-sink", "FRESH", f"{len(kept)} records, all current")
        return False, True
    if not refresh:
        _report(
            "durations-sink",
            "STALE",
            f"{dropped} stale of {dropped + len(kept)} records",
        )
        return True, True
    tmp = sink.with_suffix(".ndjson.tmp")
    tmp.write_text("\n".join(kept) + ("\n" if kept else ""), encoding="utf-8")
    os.replace(tmp, sink)
    _report(
        "durations-sink",
        "REFRESHED",
        f"compacted: kept {len(kept)}, dropped {dropped} "
        f"({sink.stat().st_size // 1024}KB now)",
    )
    return True, True


def check_observer_registry(
    refresh: bool, *, run_dir: Path | None = None
) -> tuple[bool, bool]:
    """(4) mark dead-pid running sessions as killed."""
    if run_dir is None:
        run_dir = (
            Path(os.environ.get("XDG_RUNTIME_DIR", f"/run/user/{os.getuid()}"))
            / "orchestrata"
            / "pytest-sessions"
        )
    if not run_dir.exists():
        _report("observer-registry", "SKIP", "no registry dir")
        return False, True
    ghosts = 0
    for meta_path in run_dir.glob("*.json"):
        try:
            meta = json.loads(meta_path.read_text(encoding="utf-8"))
        except OSError, json.JSONDecodeError:
            continue
        if meta.get("status") != "running":
            continue
        pid = meta.get("pid")
        alive = False
        if isinstance(pid, int):
            try:
                os.kill(pid, 0)
                alive = True
            except ProcessLookupError:
                alive = False
            except PermissionError:
                alive = True
        if alive:
            continue
        ghosts += 1
        if refresh:
            meta["status"] = "killed"
            meta_path.write_text(json.dumps(meta), encoding="utf-8")
    if not ghosts:
        _report("observer-registry", "FRESH", "no dead-pid running sessions")
        return False, True
    _report(
        "observer-registry",
        "REFRESHED" if refresh else "STALE",
        f"{ghosts} ghost session(s) {'marked killed' if refresh else 'found'}",
    )
    return True, True


def check_packer_cache(
    refresh: bool,
    *,
    cache_path: Path | None = None,
    repo_root: Path | None = None,
) -> tuple[bool, bool]:
    """(5) drop packer-cache entries for modules that no longer exist."""
    cache_file = cache_path if cache_path is not None else PACKER_CACHE
    root = repo_root if repo_root is not None else REPO_ROOT
    if not cache_file.exists():
        _report("packer-cache", "SKIP", "absent")
        return False, True
    try:
        cache = json.loads(cache_file.read_text(encoding="utf-8"))
    except OSError, json.JSONDecodeError:
        _report(
            "packer-cache",
            "STALE",
            "unreadable — removing" if refresh else "unreadable",
        )
        if refresh:
            cache_file.unlink(missing_ok=True)
        return True, True
    if not isinstance(cache, dict):
        _report("packer-cache", "STALE", "wrong shape")
        if refresh:
            cache_file.unlink(missing_ok=True)
        return True, True
    dead = [k for k in cache if not (root / k.split("::", 1)[0]).exists()]
    if not dead:
        _report("packer-cache", "FRESH", f"{len(cache)} entries, all current")
        return False, True
    if refresh:
        for k in dead:
            del cache[k]
        cache_file.write_text(
            json.dumps(cache, indent=0, sort_keys=True), encoding="utf-8"
        )
    _report(
        "packer-cache",
        "REFRESHED" if refresh else "STALE",
        f"{len(dead)} dead-module entr{'ies' if len(dead) != 1 else 'y'}"
        f" {'pruned' if refresh else 'found'}",
    )
    return True, True


def check_capture_age(
    _refresh: bool,
    now: float | None = None,
    *,
    runtime_callgraph: Path | None = None,
) -> tuple[bool, bool]:
    """(6) report-only: age of the runtime capture."""
    callgraph = (
        runtime_callgraph if runtime_callgraph is not None else RUNTIME_CALLGRAPH
    )
    if not callgraph.exists():
        _report("runtime-capture", "SKIP", "no capture file")
        return False, True
    now = now if now is not None else time.time()
    age_days = (now - callgraph.stat().st_mtime) / 86400
    if age_days <= CAPTURE_MAX_AGE_DAYS:
        _report("runtime-capture", "FRESH", f"{age_days:.1f} days old")
        return False, True
    _report(
        "runtime-capture",
        "STALE",
        f"{age_days:.1f} days old (> {CAPTURE_MAX_AGE_DAYS:.0f}d) — schedule an "
        "instrumented suite run (per-batch ORCH_RCG_DIR env; never global "
        "PYTEST_ADDOPTS, it leaks into nested pytest)",
    )
    return True, True  # report-only: stale but never a refresh failure


CHECKS = [
    check_ast_runtime_merge,
    check_prolog_kbs,
    check_durations_sink,
    check_observer_registry,
    check_packer_cache,
    check_capture_age,
]


def main(
    argv: list[str] | None = None,
    *,
    checks: Sequence[Callable[[bool], tuple[bool, bool]]] | None = None,
) -> int:
    parser = argparse.ArgumentParser(description=(__doc__ or "").splitlines()[0])
    parser.add_argument(
        "--check",
        action="store_true",
        help="Detect only; exit 1 if anything is stale, refresh nothing.",
    )
    args = parser.parse_args(argv)
    any_stale = False
    all_ok = True
    for check in checks if checks is not None else CHECKS:
        stale, ok = check(not args.check)
        any_stale = any_stale or stale
        all_ok = all_ok and ok
    if args.check:
        return 1 if any_stale else 0
    return 0 if all_ok else 1


if __name__ == "__main__":
    sys.exit(main())