test_mcp.py

"""Tests for `probbit mcp`, the Model Context Protocol server, over pipes with a dependency-free client (stdlib unittest; run:
python3 python/test_mcp.py). Both eras: a legacy client opens with `initialize`; a 2026-07-28 client puts the protocol version
and its capabilities in every request's `_meta`. Every tool answer is compared with the CLI's own output for the same input."""
import json, os, subprocess, tempfile, unittest
import probbit

EX = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "examples")
MODERN = {"io.modelcontextprotocol/protocolVersion": "2026-07-28", "io.modelcontextprotocol/clientCapabilities": {},
          "io.modelcontextprotocol/clientInfo": {"name": "test_mcp", "version": "1"}}
VOLATILE = ("ms", "telemetry", "phases")


def cli(args, doc=None):
    p = subprocess.run([probbit.find_binary(), *args], input=json.dumps(doc) if doc is not None else "", capture_output=True, encoding="utf-8")
    return p.returncode, p.stdout


def stable(d):
    return {k: v for k, v in d.items() if k not in VOLATILE}


class Client:
    def __init__(self):
        self.p = subprocess.Popen([probbit.find_binary(), "mcp"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                  encoding="utf-8", bufsize=1)
        self.n = 0

    def send_raw(self, line):
        self.p.stdin.write(line + "\n"); self.p.stdin.flush()

    def recv(self):
        line = self.p.stdout.readline()
        assert line.endswith("\n") and "\n" not in line[:-1], repr(line)
        return json.loads(line)

    def request(self, method, params=None, meta=None):
        self.n += 1
        msg = {"jsonrpc": "2.0", "id": self.n, "method": method}
        if params is not None or meta is not None:
            msg["params"] = dict(params or {})
            if meta is not None:
                msg["params"]["_meta"] = meta
        self.send_raw(json.dumps(msg))
        r = self.recv()
        assert r.get("id") == self.n and r.get("jsonrpc") == "2.0", r
        return r

    def close(self):
        self.p.stdin.close()
        code = self.p.wait(timeout=20)
        rest, err = self.p.stdout.read(), self.p.stderr.read()
        self.p.stdout.close(); self.p.stderr.close()
        return code, rest, err


class Mcp(unittest.TestCase):
    def setUp(self):
        self.c = Client()

    def tearDown(self):
        if self.c.p.poll() is None:
            self.c.close()

    def legacy(self, version="2025-06-18"):
        r = self.c.request("initialize", {"protocolVersion": version, "capabilities": {}, "clientInfo": {"name": "test_mcp", "version": "1"}})
        self.c.send_raw(json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}))  # a notification: no answer
        return r["result"]

    def call(self, name, arguments, meta=None):
        r = self.c.request("tools/call", {"name": name, "arguments": arguments}, meta)
        self.assertIn("result", r, r)
        return r["result"]

    def test_invalid_envelope_params_do_not_execute_and_server_recovers(self):
        for msg, code in [
            ({"jsonrpc": "2.0", "id": 5, "method": "ping", "params": []}, -32602),
            ({"jsonrpc": "2.0", "id": 5, "method": "ping", "params": {"_meta": False}}, -32602),
            ({"jsonrpc": "2.0", "id": 0.5, "method": "ping"}, -32600),
            ({"jsonrpc": "2.0", "id": 5, "method": "tools/list", "params": {"_meta": dict(MODERN, **{"io.modelcontextprotocol/clientCapabilities": False})}}, -32602),
        ]:
            self.c.send_raw(json.dumps(msg))
            self.assertEqual(self.c.recv()["error"]["code"], code)
        self.assertIn("result", self.c.request("ping"))

    def test_fuzz_explicit_synthetic_source_replays_through_strict_persona_tool(self):
        self.legacy()
        persona = {"probbit_persona": 1, "identity": {"name": "Fixture", "version": "1"},
                   "traits": [{"id": "action", "levels": ["retry", "ask"], "logw": [1, 0]}],
                   "inputs": [{"id": "praise", "kind": "flag"}, {"id": "criticism", "kind": "flag"}],
                   "learning": {"from": ["praise", "criticism"], "traits": ["action"], "rate": 3, "step_cap": 2, "total_cap": 2},
                   "reward_from": ["env"]}
        args = {"persona": persona, "never": {"then": {"action": ["retry"]}}, "seeds": [0], "scripts": 0, "depth": 3, "threads": 1}
        raw = self.call("probbit_persona_fuzz", args)["structuredContent"]
        self.assertIsNone(raw["properties"][0]["shortest"]["replay"])
        f = self.call("probbit_persona_fuzz", dict(args, fixture_src="env:synthetic"))
        self.assertFalse(f["isError"])
        shortest = f["structuredContent"]["properties"][0]["shortest"]
        state = self.call("probbit_persona_init", {"persona": persona, "seed": 0})["structuredContent"]
        for event in shortest["script"]:
            turn = self.call("probbit_persona_turn", {"persona": persona, "state": state, "inputs": event})
            self.assertFalse(turn["isError"])
            state = turn["structuredContent"]["state"]
        self.assertEqual(turn["structuredContent"]["stance"], shortest["stance"])
        self.assertTrue(self.call("probbit_persona_fuzz", dict(args, fixture_src="env:production"))["isError"])
        bad = self.call("probbit_persona_turn", {"persona": persona, "state": state, "inputs": {"praise": True, "src": "self"}})
        self.assertTrue(bad["isError"])

    def test_legacy_handshake_lists_eight_tools(self):
        init = self.legacy()
        self.assertEqual(init["protocolVersion"], "2025-06-18"); self.assertEqual(init["serverInfo"]["name"], "probbit")
        self.assertIn("tools", init["capabilities"]); self.assertNotIn("resultType", init)
        tools = self.c.request("tools/list")["result"]["tools"]
        self.assertEqual([t["name"] for t in tools], ["probbit_decide", "probbit_run", "probbit_stats", "probbit_demo", "probbit_evaluate", "probbit_persona_init", "probbit_persona_turn", "probbit_persona_fuzz", "probbit_live_event", "probbit_live_verify"])
        for t in tools:
            self.assertEqual(t["inputSchema"]["type"], "object"); self.assertTrue(t["description"])
        self.assertIn("flags", tools[0]["inputSchema"]["properties"]); self.assertIn("vars", tools[1]["inputSchema"]["properties"])
        ev = tools[4]["inputSchema"]
        self.assertEqual(sorted(ev["properties"]["probbit"]["properties"]["rules"]["properties"]), sorted(["pairs", "caps", "all_different", "implies", "tables", "precedes", "linear"]))
        self.assertIn("cap", ev["$defs"]); self.assertIn("program", ev["properties"]["flags"]["properties"])
        for t, key in ((tools[6], "inputs"), (tools[8], "event")):  # goal signals (drives) are an object, the other inputs scalars
            g = t["inputSchema"]["properties"][key]["properties"]["goals"]
            self.assertEqual(sorted(g["additionalProperties"]["properties"]), ["cue", "deadline_hours", "novelty", "progress", "setback", "win"])
        self.assertEqual(self.c.request("ping")["result"], {})
        self.assertEqual(self.legacy("1999-01-01")["protocolVersion"], "2025-11-25")  # unknown -> the newest handshake revision

    def test_tools_answer_exactly_as_the_cli(self):
        self.legacy("2025-11-25")
        demo = self.call("probbit_demo", {"tasks": 12, "seed": 7})
        code, out = cli(["demo", "--tasks", "12", "--seed", "7"])
        self.assertFalse(demo["isError"]); self.assertEqual(demo["content"][0]["text"], out.rstrip("\n"))
        doc = demo["structuredContent"]
        d = self.call("probbit_decide", dict(doc, flags={"sweeps": 200, "polish_ms": 0, "summary": True}))
        code, out = cli(["decide", "--sweeps", "200", "--polish-ms", "0", "--summary"], doc)
        self.assertFalse(d["isError"]); self.assertEqual(stable(d["structuredContent"]), stable(json.loads(out)))
        self.assertEqual(json.loads(d["content"][0]["text"]), d["structuredContent"])
        with open(os.path.join(EX, "knapsack-20.json")) as f:
            prog = json.load(f)
        r = self.call("probbit_run", dict(prog, flags={"op": "sample", "sweeps": 300, "polish_ms": 0, "collective": False}))
        code, out = cli(["run", "--op", "sample", "--sweeps", "300", "--polish-ms", "0", "--collective", "off"], prog)
        self.assertEqual(stable(r["structuredContent"]), stable(json.loads(out)))
        s = self.call("probbit_stats", {"sweeps": 50})
        self.assertIn("self_test", s["structuredContent"]); self.assertFalse(s["isError"])

    def test_answers_that_say_no_are_not_errors_and_errors_are(self):
        self.legacy()
        inf = self.call("probbit_run", {"probbit_ir": 1, "values": ["a"], "vars": [{"id": "x"}, {"id": "y"}], "caps": [{"value": "a", "limit": 1}]})
        self.assertFalse(inf["isError"]); self.assertEqual(inf["structuredContent"]["verdict"], "infeasible")
        bad = self.call("probbit_run", {"probbit_ir": 1, "values": ["a"], "vars": [{"id": "x", "zz": 1}]})
        self.assertTrue(bad["isError"]); self.assertEqual(bad["structuredContent"]["error"]["code"], "schema")
        flag = self.call("probbit_decide", {"workers": [{"id": "A", "cap": 1}], "tasks": [{"scores": {"A": 1}}], "flags": {"bogus": 1}})
        self.assertTrue(flag["isError"]); self.assertIn("unknown argument", flag["content"][0]["text"]); self.assertNotIn("structuredContent", flag)
        self.assertEqual(self.c.request("tools/call", {"name": "nope", "arguments": {}})["error"]["code"], -32602)
        self.assertEqual(self.c.request("tools/call", {"name": "probbit_demo", "arguments": [1]})["error"]["code"], -32602)
        self.assertEqual(self.c.request("resources/list")["error"]["code"], -32601)

    def test_evaluate_answers_exactly_as_the_cli(self):
        self.legacy("2025-11-25")
        with open(os.path.join(EX, "evaluate", "support-12.json")) as f:
            req = json.load(f)
        for flags, args in (({}, []), ({"summary": True}, ["--summary"]), ({"op": "sample", "sweeps": 2000, "polish_ms": 0}, ["--op", "sample", "--sweeps", "2000", "--polish-ms", "0"])):
            r = self.call("probbit_evaluate", dict(req, flags=flags))
            code, out = cli(["evaluate", *args], req)
            self.assertFalse(r["isError"]); self.assertEqual(code, 0); self.assertEqual(stable(r["structuredContent"]), stable(json.loads(out)))
            self.assertEqual(r["structuredContent"]["answers"]["team"]["probbit"]["value"], "technical")  # the judge said billing; the rules move it
        prog = self.call("probbit_evaluate", dict(req, flags={"program": True}))
        code, out = cli(["evaluate", "--program"], req)
        self.assertEqual(prog["content"][0]["text"], out.rstrip("\n")); self.assertEqual(prog["structuredContent"]["probbit_ir"], 1)
        refused = self.call("probbit_evaluate", dict(req, flags={"op": "sample", "sweeps": 400, "polish_ms": 0}))  # exit 3: an answer
        self.assertFalse(refused["isError"]); self.assertEqual(refused["structuredContent"]["verdict"], "refused")
        self.assertTrue(all(not a["probbit"]["released"] for a in refused["structuredContent"]["answers"].values()))
        bad = self.call("probbit_evaluate", {"questions": {"a": {"type": "boolean"}}})
        self.assertTrue(bad["isError"]); self.assertEqual(bad["structuredContent"]["error"], {"code": "value", "path": "questions.a.type", "message": bad["structuredContent"]["error"]["message"]})

    def test_persona_tools_answer_exactly_as_the_cli(self):
        self.legacy("2025-11-25")
        path = os.path.join(EX, "persona", "tutor.yaml")
        with open(os.path.join(EX, "persona", "tutor.json")) as f:
            doc = json.load(f)
        init = self.call("probbit_persona_init", {"persona": doc, "seed": 2})
        code, out = cli(["persona", "init", path, "--seed", "2"])
        self.assertFalse(init["isError"]); self.assertEqual(code, 0); self.assertEqual(init["content"][0]["text"], out.rstrip("\n"))
        state = init["structuredContent"]
        import tempfile
        with tempfile.TemporaryDirectory() as d:
            sp = os.path.join(d, "state.json")
            with open(sp, "w") as f:
                f.write(out)
            for inputs in ({"confused": True, "error": True}, {"loss": True, "sentiment": "negative"}, {}):
                r = self.call("probbit_persona_turn", {"persona_path": path, "state": state, "inputs": inputs})
                code, out = cli(["persona", "turn", path, "--state", sp, "--inputs", json.dumps(inputs)])
                self.assertFalse(r["isError"]); self.assertEqual(code, 0)
                self.assertEqual(r["structuredContent"]["stance"], json.loads(out))  # MCP answer == CLI answer
                with open(sp) as f:
                    self.assertEqual(r["structuredContent"]["state"], json.load(f))  # and the same next state
                self.assertEqual(json.loads(r["content"][0]["text"]), r["structuredContent"])
                state = r["structuredContent"]["state"]
        refused = self.call("probbit_persona_turn", {"persona": dict(doc, engine={"op": "sample", "sweeps": 8}), "state": self.call("probbit_persona_init", {"persona": dict(doc, engine={"op": "sample", "sweeps": 8})})["structuredContent"], "inputs": {}})
        self.assertFalse(refused["isError"]); self.assertIn(refused["structuredContent"]["stance"]["status"], ("refused", "partial", "fallback"))  # a refusal is an answer
        bad = self.call("probbit_persona_turn", {"persona": doc, "state": state, "inputs": {"stakes": 2}})
        self.assertTrue(bad["isError"]); self.assertEqual(bad["structuredContent"]["error"]["code"], "persona"); self.assertEqual(bad["structuredContent"]["error"]["path"], "inputs.stakes")
        for args, where in (({"persona": doc, "persona_path": path, "state": state}, "arguments"), ({"persona": doc, "state": state, "extra": 1}, "arguments.extra"),
                            ({"persona": doc}, "arguments.state"), ({"persona": [1], "state": state}, "arguments.persona")):
            e = self.call("probbit_persona_turn", args)
            self.assertTrue(e["isError"]); self.assertEqual(e["structuredContent"]["error"]["path"], where)

    def test_readings_never_come_through_the_agents_channel(self):
        # senses (docs/persona.md 2.11, S7): both turn tools refuse an event that carries readings for a persona with a senses
        # block; the same event without readings is answered; a persona without the block lists `senses` in ignored, as before
        self.legacy("2025-11-25")
        path = os.path.join(EX, "persona", "sensed.yaml")
        state = self.call("probbit_persona_init", {"persona_path": path, "seed": 3})["structuredContent"]
        for tool, key in (("probbit_persona_turn", "inputs"), ("probbit_live_event", "event")):
            args = {"persona_path": path, "state": state, key: {"senses": {"tool_fail": True}, "src": "env:runner"}}
            e = self.call(tool, args)
            self.assertTrue(e["isError"]); self.assertEqual(e["structuredContent"]["error"]["path"], "arguments.%s.senses" % key)
            ok = self.call(tool, dict(args, **{key: {"praise": True, "src": "human:ana"}}))
            self.assertFalse(ok["isError"])
        tutor = os.path.join(EX, "persona", "tutor.yaml")
        t0 = self.call("probbit_persona_init", {"persona_path": tutor, "seed": 3})["structuredContent"]
        r = self.call("probbit_persona_turn", {"persona_path": tutor, "state": t0, "inputs": {"senses": {"tool_fail": True}}})
        self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["stance"]["ignored"], ["senses"])

    def test_needs_amounts_never_come_through_the_agents_channel(self):
        # needs (docs/persona.md 2.12, RFC 0002 security review): both turn tools refuse an event that carries needs amounts for a
        # persona with a needs block, from any source; the same event without amounts is answered; without the block `needs` is
        # listed in ignored, as before
        self.legacy("2025-11-25")
        path = os.path.join(EX, "persona", "body.yaml")
        state = self.call("probbit_persona_init", {"persona_path": path, "seed": 3})["structuredContent"]
        for tool, key in (("probbit_persona_turn", "inputs"), ("probbit_live_event", "event")):
            for amount in ({"context": 0.6}, {"contact": {"amount": 0.8, "src": "human:principal"}}):
                args = {"persona_path": path, "state": state, key: {"needs": amount, "src": "env:host"}}
                e = self.call(tool, args)
                self.assertTrue(e["isError"]); self.assertEqual(e["structuredContent"]["error"]["path"], "arguments.%s.needs" % key)
            ok = self.call(tool, dict(args, **{key: {"elapsed_hours": 1, "src": "env:host"}}))
            self.assertFalse(ok["isError"])
        tutor = os.path.join(EX, "persona", "tutor.yaml")
        t0 = self.call("probbit_persona_init", {"persona_path": tutor, "seed": 3})["structuredContent"]
        r = self.call("probbit_persona_turn", {"persona_path": tutor, "state": t0, "inputs": {"needs": {"context": 0.6}}})
        self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["stance"]["ignored"], ["needs"])

    def test_persona_fuzz_answers_exactly_as_the_cli(self):
        self.legacy("2025-11-25")
        tutor = os.path.join(EX, "..", "probbit-cli", "tests", "fixtures", "persona", "tutor-0.5.0.yaml")
        rule = "{when: {sentiment: negative}, then: {humour: {at_most: light}}}"
        r = self.call("probbit_persona_fuzz", {"persona_path": tutor, "never": rule, "seeds": "0-4", "scripts": 10})
        code, out = cli(["persona", "fuzz", tutor, "--never", rule, "--seeds", "0-4", "--scripts", "10", "--json"])
        self.assertFalse(r["isError"]); self.assertEqual(code, 1)  # a counterexample: an answer (exit 1 on the CLI)
        self.assertEqual(r["content"][0]["text"], out.rstrip("\n"))  # MCP answer == CLI answer, byte for byte
        self.assertTrue(r["structuredContent"]["found"])
        self.assertEqual(r["structuredContent"]["properties"][0]["shortest"]["script"], [{"sentiment": "negative"}])
        with open(os.path.join(EX, "persona", "tutor.json"), encoding="utf-8") as f:
            doc = json.load(f)
        inline = self.call("probbit_persona_fuzz", {"persona": doc, "never": {"when": {"loss": True}, "then": {"humour": ["none"]}}, "seeds": [0, 1], "scripts": 5})
        self.assertFalse(inline["isError"]); self.assertFalse(inline["structuredContent"]["found"])  # a habit never breaks
        for args, where in (({"persona_path": tutor}, "arguments"), ({"persona_path": tutor, "never": rule, "seeds": "9-1"}, "arguments.seeds"),
                            ({"persona_path": tutor, "never": rule, "extra": 1}, "arguments.extra"), ({"persona_path": tutor, "never": {"when": {"sentimentx": "negative"}, "then": {"humour": ["none"]}}}, "arguments.never.when.sentimentx")):
            e = self.call("probbit_persona_fuzz", args)
            self.assertTrue(e["isError"]); self.assertEqual(e["structuredContent"]["error"]["path"], where)

    def test_live_event_logs_a_strand_the_cli_verifies(self):
        self.legacy("2025-11-25")
        tools = self.c.request("tools/list")["result"]["tools"]
        for tool in tools:
            self.assertEqual(tool["annotations"]["readOnlyHint"], tool["name"] != "probbit_live_event")
        event_tool = next(t for t in tools if t["name"] == "probbit_live_event")
        self.assertFalse(event_tool["annotations"]["destructiveHint"])
        path = os.path.join(EX, "persona", "tutor.yaml")
        with tempfile.TemporaryDirectory() as d:
            strand = os.path.join(d, "pip.strand")
            r = self.call("probbit_live_event", {"persona_path": path, "seed": 2, "event": {"loss": True}, "strand_path": strand})
            self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["stance"]["stance"]["humour"]["level"], "none")
            state = r["structuredContent"]["state"]
            r = self.call("probbit_live_event", {"persona_path": path, "state": state, "event": {"praise": True, "elapsed_hours": 6.5}, "strand_path": strand})
            self.assertEqual(r["structuredContent"]["strand"]["events"], 2)
            code, out = cli(["live", "verify", strand])
            self.assertEqual(code, 0); self.assertEqual(json.loads(out)["last_line"], r["structuredContent"]["strand"]["head"])  # MCP writes, the CLI replays
            v = self.call("probbit_live_verify", {"strand_path": strand})
            self.assertFalse(v["isError"]); self.assertEqual(v["content"][0]["text"], out.rstrip("\n"))
            with open(strand, encoding="utf-8") as f:
                text = f.read()
            bad = self.call("probbit_live_verify", {"strand": text.replace('"elapsed_hours":6.5', '"elapsed_hours":7')})
            self.assertFalse(bad["isError"]); self.assertEqual(bad["structuredContent"], {"ok": False, "line": 3, "diverges": "the stance differs"})  # an answer
            for args, where in (({"persona_path": path, "seed": 2, "strand_path": strand}, "state"), ({"persona_path": path, "extra": 1}, "arguments.extra"),
                                ({"persona_path": path, "event": {"elapsed_hours": -1}}, "arguments.event.elapsed_hours")):
                e = self.call("probbit_live_event", args)
                self.assertTrue(e["isError"]); self.assertEqual(e["structuredContent"]["error"]["path"], where)

    def test_live_verify_answers_unsupported_for_a_newer_engines_strand(self):
        # docs/persona.md 5.7: a block or a line kind this engine does not know is an answer with its own key (unsupported), not a
        # tool error and not a divergence; the text is the CLI's stdout (exit 5); serverInfo names this engine's version
        init = self.legacy("2025-11-25")
        code, out = cli(["version"])
        self.assertEqual((code, "probbit " + init["serverInfo"]["version"]), (0, out.strip()))
        st = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "probbit-cli", "tests", "fixtures", "strands")
        for name, line, field in (("next-load.strand", 1, "load"), ("next-rest.strand", 8, "rest")):
            path = os.path.join(st, name)
            code, text = cli(["live", "verify", path])
            self.assertEqual(code, 5)
            v = self.call("probbit_live_verify", {"strand_path": path})
            self.assertFalse(v["isError"]); self.assertEqual(v["content"][0]["text"], text.rstrip("\n"))
            self.assertEqual(v["structuredContent"], {"ok": False, "line": line, "unsupported": field, "engine": out.strip(),
                                                      "strand_engine": "probbit 0.11.0-dev"})
            with open(path, encoding="utf-8") as f:
                self.assertEqual(self.call("probbit_live_verify", {"strand": f.read()})["structuredContent"], v["structuredContent"])

    def test_persona_and_live_tools_pass_goal_signals(self):
        # drives (docs/persona.md 2.9): `goals` passes through probbit_persona_turn and probbit_live_event unchanged
        self.legacy("2025-11-25")
        fx = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "probbit-cli", "tests", "fixtures", "persona")
        path = os.path.join(fx, "drives-adversary.json")
        with open(os.path.join(fx, "drives-adversary-script.json"), encoding="utf-8") as f:
            script = json.load(f)["turns"][:5]
        with open(os.path.join(fx, "drives-adversary-replay.jsonl"), encoding="utf-8") as f:
            gold = [json.loads(x) for x in f.read().splitlines()][:5]
        state = self.call("probbit_persona_init", {"persona_path": path, "seed": 4})["structuredContent"]
        for inputs, g in zip(script, gold):
            r = self.call("probbit_persona_turn", {"persona_path": path, "state": state, "inputs": inputs})
            self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["stance"], g)  # MCP answer == the CLI's replay
            state = r["structuredContent"]["state"]
        with tempfile.TemporaryDirectory() as d:
            strand = os.path.join(d, "goals.strand")
            r = self.call("probbit_live_event", {"persona_path": path, "seed": 4, "event": {"goals": {"chores": {"deadline_hours": 5}}, "security": True}, "strand_path": strand})
            self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["stance"]["pursue"]["goal"], "chores")
            r = self.call("probbit_live_event", {"persona_path": path, "state": r["structuredContent"]["state"], "event": {"goals": {"fun": {"win": 1.5, "cue": True}}, "elapsed_hours": 2}, "strand_path": strand})
            self.assertFalse(r["isError"]); self.assertEqual(r["structuredContent"]["strand"]["events"], 2)
            code, out = cli(["live", "verify", strand])
            self.assertEqual(code, 0); self.assertEqual(json.loads(out)["last_line"], r["structuredContent"]["strand"]["head"])  # MCP writes, the CLI replays
            with open(strand, encoding="utf-8") as f:
                self.assertIn('"goals":{"fun":{"cue":true,"win":1.5}}', f.read())

    def test_structured_content_from_2025_06_18_on(self):
        self.legacy("2024-11-05")
        r = self.call("probbit_demo", {"tasks": 3})
        self.assertNotIn("structuredContent", r); json.loads(r["content"][0]["text"])

    def test_modern_requests_carry_their_metadata(self):
        d = self.c.request("server/discover", meta=MODERN)["result"]
        self.assertEqual(d["resultType"], "complete"); self.assertIn("2026-07-28", d["supportedVersions"])
        self.assertEqual(d["_meta"]["io.modelcontextprotocol/serverInfo"]["name"], "probbit"); self.assertIn("tools", d["capabilities"])
        t = self.c.request("tools/list", meta=MODERN)["result"]
        self.assertEqual(t["resultType"], "complete"); self.assertEqual(len(t["tools"]), 10)
        r = self.call("probbit_demo", {"tasks": 4, "seed": 2}, MODERN)
        self.assertEqual(r["resultType"], "complete"); self.assertEqual(len(r["structuredContent"]["tasks"]), 4)
        e = self.c.request("tools/list", meta={"io.modelcontextprotocol/protocolVersion": "1999-01-01", "io.modelcontextprotocol/clientCapabilities": {}})["error"]
        self.assertEqual(e["code"], -32022); self.assertEqual(e["data"]["requested"], "1999-01-01"); self.assertIn("2026-07-28", e["data"]["supported"])
        e = self.c.request("tools/list", meta={"io.modelcontextprotocol/protocolVersion": "2026-07-28"})["error"]
        self.assertEqual(e["code"], -32602)  # clientCapabilities is required
        self.assertEqual(self.c.request("tools/list")["error"]["code"], -32602)  # no _meta and no initialize before it

    def test_framing_batches_and_shutdown(self):
        self.c.send_raw("" + json.dumps({"jsonrpc": "2.0", "id": "a", "method": "ping"}))  # a byte-order mark on the first line
        self.assertEqual(self.c.recv(), {"jsonrpc": "2.0", "id": "a", "result": {}})
        self.c.send_raw("this is not json")
        e = self.c.recv(); self.assertEqual((e["id"], e["error"]["code"]), (None, -32700))
        self.c.send_raw(json.dumps({"jsonrpc": "2.0", "method": "ping"}))  # notification-shaped ping: no answer
        self.c.send_raw(json.dumps([{"jsonrpc": "2.0", "id": 7, "method": "ping"}, {"jsonrpc": "2.0", "method": "notifications/initialized"}]))
        b = self.c.recv(); self.assertEqual(b, [{"jsonrpc": "2.0", "id": 7, "result": {}}])
        self.c.send_raw(json.dumps({"jsonrpc": "1.0", "id": 8, "method": "ping"}))
        self.assertEqual(self.c.recv()["error"]["code"], -32600)
        code, rest, err = self.c.close()
        self.assertEqual(code, 0); self.assertEqual(rest, ""); self.assertIn("ready on stdio", err)


if __name__ == "__main__":
    unittest.main()