test_loop.py

"""Sub-step 0 plumbing check: round-trip the slot contract over a REAL MCP transport.

Spawns server.py over stdio, calls all three ops, and asserts:
  - the seam works (host <-> cartridge over MCP),
  - energy SEPARATES an obvious in-domain vs out-of-domain split,
  - predict is shaped-but-stubbed.

This isolates "the pipe works" from "the energy has resolution" (sub-step 1), so later
noise is never blamed on the transport.

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

IN_DOMAIN = "The Actor Pawn ticks and the Blueprint UFunction replicated component"
OUT_DOMAIN = "the quick brown fox jumps over the lazy sleeping dog at noon"


def _payload(result) -> dict:
    # FastMCP returns the dict as JSON text content; parse it back.
    return json.loads(result.content[0].text)


async def main() -> int:
    params = StdioServerParameters(command=sys.executable, args=[str(HERE / "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", "assess", "predict"} <= tools, f"missing tools: {tools}"

            desc = _payload(await session.call_tool("describe", {}))
            assert {"id", "domain", "version", "scope"} <= desc.keys(), desc

            hot = _payload(await session.call_tool("assess", {"raw_input": IN_DOMAIN}))
            cold = _payload(await session.call_tool("assess", {"raw_input": OUT_DOMAIN}))
            assert hot["in_domain"] is True, hot
            assert cold["in_domain"] is False, cold
            assert hot["energy"] < cold["energy"], (hot, cold)

            roll = _payload(await session.call_tool("predict", {"raw_input": IN_DOMAIN}))
            assert roll["stubbed"] is True and roll["steps"] == [], roll

    print("PASS -- seam works over MCP")
    print(f"  cartridge : {desc['id']} ({desc['domain']})")
    print(f"  in-domain : energy={hot['energy']}  in_domain={hot['in_domain']}")
    print(f"  out-domain: energy={cold['energy']}  in_domain={cold['in_domain']}")
    print(f"  separation: {cold['energy'] - hot['energy']:.4f}")
    print(f"  predict   : stubbed (shape reserved)")
    return 0


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