test_verify_loop.py

"""Plumbing check for verify_server.py: round-trip the runtime loop over a REAL MCP transport.

Spawns verify_server.py over stdio, calls describe + verify_statement with a statement
containing one validated fake (AddItem returns bool) and one validated truth (SaveInventory
2-second debounce), and asserts:
  - the seam works (host <-> verifier over MCP),
  - the fake is flagged and the truth is not,
  - the call landed in field_log.jsonl (the field-prose dataset accumulates).

Run:  .venv/Scripts/python.exe 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 / "field_log.jsonl"

STATEMENT = ("URiftInventoryComponent::AddItem returns a bool indicating success. "
             "SaveInventory uses a 2-second debounce so rapid mutations don't each trigger a save.")


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


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

    params = StdioServerParameters(command=sys.executable, args=[str(HERE / "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", "learn_from_log", "reverify_learned"} <= tools, \
                f"missing tools: {tools}"

            desc = _payload(await session.call_tool("describe", {}))
            assert desc["classes_indexed"] > 0 and desc["methods_indexed"] > 0, desc

            out = _payload(await session.call_tool("verify_statement", {"text": STATEMENT}))
            facts, summary = out["facts"], out["summary"]
            assert facts, "no facts extracted"

            flagged = [f for f in facts if f["verdict"] == "flag"]
            assert any("bool" in f["fact"] and "AddItem" in f["fact"] for f in flagged), \
                f"known fake not flagged: {facts}"
            assert not any("SaveInventory" in f["fact"] and f["verdict"] == "flag"
                           and "debounce" in f["fact"].lower() for f in facts), \
                f"known truth flagged: {facts}"

            # learning tools: shape + hot-reload counter (ownership abstains must PARK, never learn)
            learn = _payload(await session.call_tool("learn_from_log", {}))
            assert {"learned", "parked", "skipped", "active_learned_facts"} <= learn.keys(), learn
            assert not any("lives on" in i["fact"] for i in learn["learned"]), \
                f"ownership claim leaked into learned store: {learn['learned']}"
            rev = _payload(await session.call_tool("reverify_learned", {}))
            assert {"reverified", "retired", "active_learned_facts"} <= rev.keys(), rev

    log_after = LOG.stat().st_size if LOG.exists() else 0
    assert log_after > log_before, "call was not appended to field_log.jsonl"

    print("PASS -- runtime loop works over MCP")
    print(f"  index    : {desc['classes_indexed']} classes, {desc['methods_indexed']} methods")
    print(f"  verdicts : {summary}")
    for f in facts:
        print(f"    {f['verdict'].upper():7} ({f['checker']}) {f['fact']}")
    print(f"  field log: +{log_after - log_before} bytes")
    print(f"  learning : {len(learn['learned'])} learned, {len(learn['parked'])} parked, "
          f"{learn['skipped']} known | reverified {rev['reverified']}, retired {len(rev['retired'])} "
          f"| {rev['active_learned_facts']} active learned fact(s)")
    return 0


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