ue_verify_server.py

"""UE 5.8 engine verifier as a local MCP server — cartridge #2 (one cartridge = one process).

Same runtime loop as rift-verify (verify_server.py), pointed at the Unreal Engine 5.8 source.
The UE env is baked in BELOW, before any project import, because corpus.domain_roots() /
CARTRIDGE_* are read at TypedChecker/JudgeChecker construction time — so it must be set first.

HARD RULE (MCP stdio): nothing prints to stdout — stdout is the JSONRPC wire. Progress in the
project code already routes to stderr; keep it that way.

Run standalone (stdio):  .venv/Scripts/python.exe ue_verify_server.py
"""

from __future__ import annotations

import os

# --- domain selection: MUST precede project imports ---
os.environ.setdefault("CARTRIDGE_DOMAIN", "ue")
os.environ.setdefault("CARTRIDGE_FILTER", "UFUNCTION")
# no MAXFILES cap: the FULL Runtime tree (Aaron 2026-07-01) — the UFUNCTION filter is the
# only gate. Embed cache MUST be pre-warmed before registering (never cold-embed at startup).

import json
import time
from datetime import datetime, timezone
from pathlib import Path

from mcp.server.fastmcp import FastMCP

import index_cache
from checker import JudgeChecker
from corpus import domain_roots
from runtime_verify import TypedChecker, verify

mcp = FastMCP("ue-verify")

HERE = Path(__file__).resolve().parent
LOG = HERE / "ue_field_log.jsonl"

TYPED = TypedChecker()
JUDGE = JudgeChecker()
_index_built = time.time()


@mcp.tool()
def describe() -> dict:
    """What this verifier checks against: the indexed UE 5.8 engine domain and its size."""
    return {
        "id": "ue-verify",
        "domain": [str(r) for r in domain_roots()],
        "classes_indexed": len(TYPED.idx),
        "methods_indexed": sum(len(e["methods"]) for e in TYPED.idx.values()),
        "implementations_indexed": len(TYPED.calls),
        "index_built": datetime.fromtimestamp(_index_built, timezone.utc).isoformat(),
        "verdicts": "flag = contradicted by source | pass = confirmed | abstain = cannot verify",
        "confidence": "1.0 extracted = parsed-index fact | <1.0 inferred = corpus/call-graph/"
                      "judge evidence (rubric in checker.py) | 0.0 ambiguous = abstain",
    }


@mcp.tool()
def refresh_index() -> dict:
    """Force a rebuild of the engine index/corpus/call-graph from current source. The
    content-keyed parse cache re-parses only changed files, so this is cheap even at
    engine scale. Use after an engine upgrade or source edit (no git hooks here — the
    UE install is not a git clone)."""
    global _index_built
    index_cache.invalidate()
    TYPED.refresh()
    JUDGE.refresh()
    _index_built = time.time()
    return {"refreshed_at": datetime.now(timezone.utc).isoformat(),
            "files_reparsed": index_cache.last_stats["parsed"],
            "files_cached": index_cache.last_stats["cached"],
            "classes_indexed": len(TYPED.idx)}


@mcp.tool()
def verify_statement(text: str) -> dict:
    """Verify a statement about the Unreal Engine 5.8 source (indexed Runtime modules).

    Extracts the typed factual claims the statement makes (method existence, return types,
    authority/purity specifiers, delegate arities, documented behaviour) and checks each
    against parsed engine source. Per fact: flag = source contradicts it, pass = confirmed,
    abstain = cannot be verified. Use before acting on any factual assertion about UE APIs."""
    results = verify(text, TYPED, JUDGE)
    summary = {k: sum(r["verdict"] == k for r in results) for k in ("flag", "abstain", "pass")}
    with LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": datetime.now(timezone.utc).isoformat(),
                            "text": text, "results": results}) + "\n")
    return {"facts": results, "summary": summary}


if __name__ == "__main__":
    mcp.run()