ue_test_verify_loop.py

"""Plumbing check for ue_verify_server.py: round-trip the UE cart over a REAL MCP transport.

Spawns ue_verify_server.py over stdio (full-corpus scope is baked into the server), then:
  - describe: index is non-trivial (full Runtime tree, not the old 60-file slice),
  - a known UE truth passes (AsyncLoadPrimaryAsset return type, validated 2026-07-01),
  - a wrong-return-type fake on the same method flags,
  - an invented method flags,
  - the calls landed in ue_field_log.jsonl.
A PASS with clean protocol traffic = zero JSONRPC parse errors (stdout stayed pure wire).

Run:  .venv/Scripts/python.exe ue_test_verify_loop.py
"""

from __future__ import annotations

import asyncio
import json
import sys
from pathlib import Path

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

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

TRUTH = "UAsyncActionLoadPrimaryAssetBase::AsyncLoadPrimaryAsset returns UAsyncActionLoadPrimaryAsset*."
FAKE_RETURN = "UAsyncActionLoadPrimaryAssetBase::AsyncLoadPrimaryAsset returns bool."
FAKE_INVENTED = "UAsyncActionLoadPrimaryAssetBase::TotallyInventedMethodXYZ exists and returns an int32."


def _payload(result) -> dict:
    return json.loads(result.content[0].text)


def _sum(facts) -> dict:
    return {k: sum(f["verdict"] == k for f in facts) for k in ("flag", "abstain", "pass")}


async def main() -> int:
    log_before = LOG.stat().st_size if LOG.exists() else 0

    params = StdioServerParameters(command=sys.executable, args=[str(HERE / "ue_verify_server.py")])
    async with stdio_client(params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            tools = {t.name for t in (await session.list_tools()).tools}
            assert {"describe", "verify_statement"} <= tools, f"missing tools: {tools}"

            desc = _payload(await session.call_tool("describe", {}))
            assert desc["classes_indexed"] > 100, f"index looks capped, not full-tree: {desc}"
            assert desc["methods_indexed"] > 1000, f"index looks capped, not full-tree: {desc}"

            outcomes = {}
            for label, stmt in [("truth", TRUTH), ("fake-return", FAKE_RETURN),
                                ("fake-invented", FAKE_INVENTED)]:
                out = _payload(await session.call_tool("verify_statement", {"text": stmt}))
                assert out["facts"], f"{label}: no facts extracted"
                outcomes[label] = (out["facts"], _sum(out["facts"]))

            assert outcomes["truth"][1]["flag"] == 0, \
                f"known truth flagged: {outcomes['truth'][0]}"
            assert outcomes["truth"][1]["pass"] >= 1, \
                f"known truth did not pass: {outcomes['truth'][0]}"
            assert outcomes["fake-return"][1]["flag"] >= 1, \
                f"wrong-return fake not flagged: {outcomes['fake-return'][0]}"
            assert outcomes["fake-invented"][1]["flag"] >= 1, \
                f"invented method not flagged: {outcomes['fake-invented'][0]}"

    log_after = LOG.stat().st_size if LOG.exists() else 0
    assert log_after > log_before, "calls were not appended to ue_field_log.jsonl"

    print("PASS -- UE cart runtime loop works over MCP")
    print(f"  index    : {desc['classes_indexed']} classes, {desc['methods_indexed']} methods")
    for label, (facts, summary) in outcomes.items():
        print(f"  {label}: {summary}")
        for f in facts:
            print(f"    {f['verdict'].upper():7} ({f['checker']}) {f['fact']}")
    print(f"  field log: +{log_after - log_before} bytes")
    return 0


if __name__ == "__main__":
    raise SystemExit(asyncio.run(main()))