runtime_verify.py

"""RUNTIME LOOP (typed): verify a real statement about the codebase — the actual-use path.

The extractor (qwen) converts prose into TYPED FACTS instead of sentences, so the
deterministic layer checks structure directly and claim phrasing stops mattering — the
runtime smoke test showed the regex rules were phrasing-locked to the eval's own templates
(invented methods and delegate arity fell past them to the judge). Structural facts are
checked against the parsed index (lookup, zero variance); BEHAVIOR facts route to the judge
carrying their Class::Method anchor, so symbol-precise evidence always attaches.

Usage:
  python runtime_verify.py "<statement about the code>"
  python runtime_verify.py --file <path-to-text>
"""

from __future__ import annotations

import re
import sys
from pathlib import Path

from checker import (C_CALLGRAPH, C_CORPUS, C_CORPUS_ABSENT, C_HOP, C_INDEX,
                     JudgeChecker, Verdict, build_index, rt_match)
from corpus import CPP_EXTS, domain_roots, file_chunk_lists
from extract_claims import gen

EXTRACT = """Below is a statement about a C++ game codebase. Convert every checkable factual
claim it makes into TYPED FACTS, one per line, using EXACTLY these forms:

EXISTS: Class::Method                        (the statement says this method exists or tells you to call it)
RETURN_TYPE: Class::Method = <type>
AUTHORITY: Class::Method = server_only | client_ok
PURITY: Class::Method = read_only | modifies
DELEGATE_ARITY: Class::DelegateName = <number of parameters>
CALLS: Class::Method = <FunctionOrEventName>               (the statement says this method calls that function, or fires/broadcasts that event)
BEHAVIOR: Class::Method = <one short factual sentence>     (any code fact that fits no other form)

Rules:
- One atomic fact per line; never combine two facts on a line.
- Every line MUST carry the Class::Member anchor. If the statement does not name the class, write ?::Member.
- Extract ONLY facts the statement explicitly asserts. NEVER infer or assume authority, purity,
  or return types the statement does not state — an unmentioned property is not a fact.
- Only code facts. Skip opinions, plans, and vague statements. If there are none, output exactly: NONE

EXAMPLE STATEMENT: "Call UFooComponent::GetCount to read the count — it returns an int32, is safe to call on clients, and fires OnCountChanged, a one-parameter delegate. In Reset you should clear the cached list before rebuilding it."
EXAMPLE OUTPUT:
EXISTS: UFooComponent::GetCount
RETURN_TYPE: UFooComponent::GetCount = int32
AUTHORITY: UFooComponent::GetCount = client_ok
CALLS: UFooComponent::GetCount = OnCountChanged
DELEGATE_ARITY: UFooComponent::OnCountChanged = 1
BEHAVIOR: ?::Reset = clear the cached list before rebuilding it

STATEMENT:
{text}"""

FACT_RE = re.compile(r"^(EXISTS|RETURN_TYPE|AUTHORITY|PURITY|DELEGATE_ARITY|CALLS|BEHAVIOR):\s*"
                     r"([\w?]+)::(\w+)\s*(?:=\s*(.+?))?\s*$")


def extract_facts(text: str):
    out = gen(EXTRACT.format(text=text), num_predict=500)
    facts = []
    for line in out.splitlines():
        m = FACT_RE.match(line.strip())
        if not m:
            continue
        kind, cls, member, value = m.group(1), m.group(2), m.group(3), (m.group(4) or "").strip()
        # anchor hygiene: extraction sometimes leaks the prompt example's names (UFooComponent,
        # GetCount) or invents members the statement never mentioned — a fact must be traceable
        # to the input text or it isn't the statement's claim
        if member not in text:
            continue
        if cls != "?" and cls not in text and cls.removeprefix("U") not in text:
            cls = "?"
        if kind == "CALLS" and value.strip("() ") not in text:   # callee must be the statement's, not invented
            continue
        facts.append((kind, cls, member, value))
    return facts


def _dnorm(name: str) -> str:
    """Delegate names: claims say the property (OnItemAdded), the index keys the type
    (FOnRiftItemAdded) — compare with the F prefix and project infix stripped."""
    return re.sub(r"^F", "", name).replace("Rift", "")


class TypedChecker:
    """Deterministic check of a typed fact against the parsed API index — pure lookup."""
    name = "typed"

    def __init__(self):
        self.refresh()

    def refresh(self) -> None:
        """(Re)build index/corpus/call-graph from the parse caches — cost scales with the diff."""
        import index_cache
        self.idx = build_index()
        self.corpus = "\n".join(c for f in file_chunk_lists(domain_roots(), CPP_EXTS) for c in f)
        self.calls = index_cache.calls_index(domain_roots())   # (owner, method) -> (called, fired)

    def _resolve(self, cls: str, member: str, kind: str):
        """Return the owning class for member, honouring the claimed class if it checks out,
        else falling back to a unique owner elsewhere in the index."""
        table = "delegates" if kind == "DELEGATE_ARITY" else "methods"
        def members(c):
            entries = self.idx.get(c, {}).get(table, {})
            if kind == "DELEGATE_ARITY":
                return {dn for dn in entries if _dnorm(dn) == _dnorm(member)}
            return {member} if member in entries else set()
        if cls != "?" and members(cls):
            return cls
        owners = {c for c in self.idx if members(c)}
        return owners.pop() if len(owners) == 1 else None

    def check(self, kind, cls, member, value) -> Verdict:
        n = self.name
        owner = self._resolve(cls, member, kind)

        if kind == "EXISTS":
            if owner:
                return Verdict("pass", f"{owner}::{member} is in the UFUNCTION index", n,
                               C_INDEX, "extracted")
            if cls in self.idx:
                # real class anchor: member may be a plain (non-UFUNCTION) method — corpus can
                # confirm it; genuinely absent everywhere = invented method
                if f"{member}(" in self.corpus:
                    return Verdict("pass", f"{member}() appears in source", n, C_CORPUS, "inferred")
                return Verdict("flag", f"no method named {member}() anywhere in source", n,
                               C_CORPUS_ABSENT, "inferred")
            # fabricated/unresolvable class anchor (APawn::, PlayerState::, ?::, leaked example
            # names): a bare corpus substring cannot verify membership of THAT class — passing
            # here dresses an unverifiable claim in false green
            return Verdict("abstain", f"class {cls} not in index — existence unverifiable", n)

        if kind == "CALLS":
            callee = value.strip("() ").split()[0] if value.strip() else ""
            if not callee:
                return Verdict("abstain", "CALLS fact carries no callee", n)
            if owner is None:                       # non-UFUNCTION methods live in defs only
                impl_owners = {o for (o, m) in self.calls if m == member}
                owner = (cls if cls in impl_owners
                         else impl_owners.pop() if len(impl_owners) == 1 else None)
            body = self.calls.get((owner, member)) if owner else None
            if body is None:
                return Verdict("abstain", f"no implementation of {cls}::{member} indexed", n)
            called, fired = body
            if callee in called or any(_dnorm(callee) == _dnorm(f) for f in fired):
                return Verdict("pass", f"{owner}::{member} calls/broadcasts {callee} directly", n,
                               C_CALLGRAPH, "extracted")
            # one hop: a direct callee whose own implementation (any owner) reaches it
            for c in sorted(called):
                for (o2, m2), (cc, fd) in self.calls.items():
                    if m2 == c and (callee in cc or any(_dnorm(callee) == _dnorm(f) for f in fd)):
                        return Verdict("pass", f"{owner}::{member} reaches {callee} via {c}()", n,
                                       C_HOP, "inferred")
            return Verdict("flag", f"{owner}::{member} implementation never calls or broadcasts "
                                   f"{callee} (directly or via one hop)", n, C_CALLGRAPH, "extracted")

        if owner is None:
            return Verdict("abstain", f"{cls}::{member} not resolvable in index", n)

        if kind == "RETURN_TYPE":
            rt = self.idx[owner]["methods"][member][0]
            verdict, reason = rt_match(value, rt)
            conf, basis = (C_INDEX, "extracted") if verdict != "abstain" else (0.0, "ambiguous")
            return Verdict(verdict, reason, n, conf, basis)

        if kind == "AUTHORITY":
            specs = self.idx[owner]["methods"][member][1]
            server_only = bool({"BlueprintAuthorityOnly", "Server"} & specs)
            says_client = "client" in value.lower()
            ok = (not server_only) if says_client else server_only
            return Verdict("pass" if ok else "flag",
                           f"server_only={server_only}, claimed {value}", n, C_INDEX, "extracted")

        if kind == "PURITY":
            rt, specs, const = self.idx[owner]["methods"][member]
            read_only = "BlueprintPure" in specs or const
            says_modifies = "modif" in value.lower()
            ok = (not read_only) if says_modifies else read_only
            return Verdict("pass" if ok else "flag",
                           f"read_only={read_only}, claimed {value}", n, C_INDEX, "extracted")

        if kind == "DELEGATE_ARITY":
            arities = {dn: a for dn, a in self.idx[owner]["delegates"].items()
                       if _dnorm(dn) == _dnorm(member)}
            m = re.search(r"\d+", value)
            if not arities or not m:
                return Verdict("abstain", f"{member} arity unknown", n)
            dn, actual = next(iter(arities.items()))
            claimed = int(m.group())
            return Verdict("pass" if claimed == actual else "flag",
                           f"{dn}: claims {claimed}, actually {actual} params", n,
                           C_INDEX, "extracted")

        return Verdict("abstain", f"no deterministic check for {kind}", n)


def verify(text: str, typed: TypedChecker, judge: JudgeChecker) -> list[dict]:
    """The runtime loop as a callable: prose -> typed facts -> per-fact verdict dicts.
    Shared by the CLI below and the MCP server (verify_server.py)."""
    results = []
    # statement-level guard: ownership/location claims are inherently null (a spawnable
    # component can live on ANY actor) — surface the abstain deterministically, whether or
    # not extraction happens to emit the claim as a fact this run
    if JudgeChecker.OWNERSHIP_RE.search(text):
        results.append({"fact": "OWNERSHIP/LOCATION claim in statement",
                        "verdict": "abstain",
                        "reason": "ownership/location is a design decision — a spawnable component "
                                  "can live on any actor; not verifiable from source",
                        "checker": "typed", "confidence": 0.0, "basis": "ambiguous"})
    exists: dict = {}
    for kind, cls, member, value in extract_facts(text):
        # every anchored fact implies its member exists — check once per anchor, in code,
        # so an invented method is caught regardless of which fact type extraction chose
        a = (cls, member)
        if kind != "DELEGATE_ARITY" and a not in exists:   # delegates aren't methods — arity check resolves them itself
            exists[a] = typed.check("EXISTS", cls, member, "")
            results.append({"fact": f"EXISTS: {cls}::{member}", "verdict": exists[a].verdict,
                            "reason": exists[a].reason, "checker": exists[a].checker,
                            "confidence": exists[a].confidence, "basis": exists[a].basis})
        if a not in exists:
            exists[a] = Verdict("pass", "delegate anchor", "typed", 1.0, "extracted")
        if kind == "EXISTS":
            continue                    # covered by the anchor check above
        if exists[a].verdict == "flag":
            continue                    # member doesn't exist — dependent facts are moot
        if kind == "BEHAVIOR":
            owner = typed._resolve(cls, member, kind) or cls
            v = judge.check(f"{owner}::{member}: {value}")
        else:
            v = typed.check(kind, cls, member, value)
        shown = f"{kind}: {cls}::{member}" + (f" = {value}" if value else "")
        results.append({"fact": shown, "verdict": v.verdict, "reason": v.reason,
                        "checker": v.checker, "confidence": v.confidence, "basis": v.basis})
    return results


def main() -> int:
    args = sys.argv[1:]
    if not args:
        print(__doc__)
        return 1
    text = Path(args[1]).read_text(encoding="utf-8") if args[0] == "--file" else " ".join(args)

    results = verify(text, TypedChecker(), JudgeChecker())
    if not results:
        print("no checkable claims extracted")
        return 0

    order = {"flag": 0, "abstain": 1, "pass": 2}
    for r in sorted(results, key=lambda r: order[r["verdict"]]):
        print(f"  {r['verdict'].upper():7} [{r['basis']} {r['confidence']:.2f}] ({r['checker']}) {r['fact']}")
        print(f"          — {r['reason']}")

    n = {k: sum(r["verdict"] == k for r in results) for k in order}
    print(f"\n  {n['flag']} flagged | {n['abstain']} unverifiable | {n['pass']} confirmed")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())