index_cache.py

"""Content-keyed parse cache — the freshness spine (Graphify pattern, adopted 2026-07-19).

Header parse products (owner / methods / delegates / docs) and .cpp implementation offsets
are cached per file. Validation is (mtime, size) fast-path with a sha1 fallback: a
touched-but-identical file re-keys without re-parsing; a changed file re-parses ALONE.
Startup and refresh cost scale with the diff, not the corpus — the mechanism that makes a
cart impossible to serve stale (fact-553 class of failure: the rift index silently aged
out against osprey-AIE).

Cache files are keyed by domain AND a hash of the root list, so repointing the corpus or
running tests against a temp corpus never collides with an existing cache.

Bodies are NOT stored in the cache (UE-scale JSON death) — only line offsets; impls()
slices bodies from the source text at load. In-process results are memoized so
TypedChecker and JudgeChecker share one load per process; invalidate() drops the memo
(the on-disk caches stay — they are the point).
"""

from __future__ import annotations

import hashlib
import json
import os
import re
from pathlib import Path

from corpus import iter_files
from generate_claims import method_docs, parse_header

_HERE = Path(__file__).resolve().parent
_BODY_CAP = 400          # brace-matched body line cap (pathological unclosed braces)

# stats of the most recent cache load, for tests and describe(): {"parsed": n, "cached": m}
last_stats: dict = {"parsed": 0, "cached": 0}

_memo: dict = {}


def invalidate() -> None:
    """Drop in-process memos so the next call re-validates against disk."""
    _memo.clear()


def _domain() -> str:
    return "ue" if os.environ.get("CARTRIDGE_DOMAIN") == "ue" else "rift"


def _cache_path(kind: str, roots: list[Path]) -> Path:
    roots_key = hashlib.sha1("\x00".join(sorted(str(r) for r in roots)).encode()).hexdigest()[:8]
    return _HERE / f".index_cache_{kind}_{_domain()}_{roots_key}.json"


def _load(cache_path: Path, paths: list[Path], parse_fn) -> dict:
    """path-str -> entry. Entry carries mtime/size/sha1 + parse_fn's products."""
    cache: dict = {}
    if cache_path.exists():
        try:
            cache = json.loads(cache_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError):
            cache = {}
    out: dict = {}
    parsed = cached = 0
    dirty = False
    for p in paths:
        key = str(p)
        try:
            st = p.stat()
        except OSError:
            continue
        e = cache.get(key)
        if e and e["mtime"] == st.st_mtime and e["size"] == st.st_size:
            out[key] = e
            cached += 1
            continue
        try:
            text = p.read_text(encoding="utf-8", errors="ignore")
        except OSError:
            continue
        sha = hashlib.sha1(text.encode("utf-8", errors="ignore")).hexdigest()
        if e and e.get("sha1") == sha:            # touched but identical: re-key, no re-parse
            e["mtime"], e["size"] = st.st_mtime, st.st_size
            out[key] = e
            cached += 1
            dirty = True
            continue
        e = parse_fn(text)
        e.update({"mtime": st.st_mtime, "size": st.st_size, "sha1": sha})
        out[key] = e
        parsed += 1
        dirty = True
    if dirty or len(out) != len(cache):           # also rewrites when files were removed
        tmp = cache_path.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(out), encoding="utf-8")
        tmp.replace(cache_path)
    last_stats.update({"parsed": parsed, "cached": cached})
    return out


# --- headers: owner / methods / delegates / docs ---

def _parse_header_products(text: str) -> dict:
    owner, methods, delegates = parse_header(text)
    return {
        "owner": owner,
        "methods": [[rt, name, sorted(specs), const] for rt, name, specs, const in methods],
        "delegates": [[dn, ar] for dn, ar in delegates],
        "docs": {name: doc for (_, name), doc in method_docs(text).items()},
    }


def headers(roots: list[Path]) -> dict:
    key = ("headers", _domain(), str(_cache_path("headers", roots)))
    if key not in _memo:
        _memo[key] = _load(_cache_path("headers", roots), iter_files(roots, (".h",)),
                           _parse_header_products)
    return _memo[key]


def docs_index(roots: list[Path]) -> dict:
    """(owner, method) -> doc-comment text. Same shape JudgeChecker.docs always had."""
    out = {}
    for e in headers(roots).values():
        if e["owner"]:
            for name, doc in e["docs"].items():
                out[(e["owner"], name)] = doc
    return out


# --- impls: (owner, method) -> brace-matched body from .cpp ---

_IMPL_RE = re.compile(r"\b(U\w+)::(\w+)\s*\(")


def _parse_impl_offsets(text: str) -> dict:
    lines = text.splitlines()
    defs = []                                     # [owner, name, start_line, line_count]
    seen = set()
    for i, line in enumerate(lines):
        dm = _IMPL_RE.search(line)
        if not dm or (dm.group(1), dm.group(2)) in seen:
            continue
        # skip declarations/calls: a DEFINITION opens its brace before any ";" appears.
        # (The old window heuristic — ");" within 3 lines — false-skipped every body whose
        # first statement is a call, silently dropping those impls from judge evidence.)
        window = "\n".join(lines[i:i + 6])
        brace, semi = window.find("{"), window.find(";")
        if brace == -1 or (semi != -1 and semi < brace):
            continue
        depth, started, end = 0, False, i
        for j in range(i, min(i + _BODY_CAP, len(lines))):
            depth += lines[j].count("{") - lines[j].count("}")
            if "{" in lines[j]:
                started = True
            end = j
            if started and depth <= 0:
                break
        seen.add((dm.group(1), dm.group(2)))
        defs.append([dm.group(1), dm.group(2), i, end - i + 1])
    return {"defs": defs}


def impls(roots: list[Path]) -> dict:
    """(owner, method) -> full brace-matched implementation body (sliced at load)."""
    key = ("impls", _domain(), str(_cache_path("impls", roots)))
    if key in _memo:
        return _memo[key]
    entries = _load(_cache_path("impls", roots), iter_files(roots, (".cpp",)),
                    _parse_impl_offsets)
    out: dict = {}
    for path, e in entries.items():
        if not e["defs"]:
            continue
        try:
            lines = Path(path).read_text(encoding="utf-8", errors="ignore").splitlines()
        except OSError:
            continue
        for owner, name, start, count in e["defs"]:
            out.setdefault((owner, name), "\n".join(lines[start:start + count]))
    _memo[key] = out
    return out


# --- calls: (owner, method) -> (called function names, Broadcast()ed delegate names) ---

_CALL_RE = re.compile(r"\b([A-Za-z_]\w+)\s*\(")
_FIRE_RE = re.compile(r"(\w+)(?:\.|->)Broadcast\s*\(")
_CPP_KEYWORDS = {"if", "for", "while", "switch", "return", "sizeof", "new", "delete",
                 "catch", "defined", "static_cast", "dynamic_cast", "const_cast",
                 "reinterpret_cast", "check", "checkf", "ensure", "ensureMsgf",
                 "UFUNCTION", "UPROPERTY", "UE_LOG", "TEXT"}


def calls_index(roots: list[Path]) -> dict:
    key = ("calls", _domain(), str(_cache_path("impls", roots)))
    if key in _memo:
        return _memo[key]
    out = {}
    for (owner, name), body in impls(roots).items():
        called = {c for c in _CALL_RE.findall(body) if c not in _CPP_KEYWORDS and c != name}
        fired = set(_FIRE_RE.findall(body))
        out[(owner, name)] = (called, fired)
    _memo[key] = out
    return out