Merge 00fb0ab673 into b781d157d7

.circleci/config.yml

@@ -3241,6 +3241,17 @@ jobs:
             - run:
                 name: Build the candidate dashboard
                 command: cd ui/litellm-dashboard && NEXT_TELEMETRY_DISABLED=1 npm run build
+      - when:
+          condition:
+            equal: [mcp, << parameters.suite >>]
+          steps:
+            - install_node
+            - run:
+                name: Install pinned official MCP conformance runner and reference
+                command: |
+                  export MCP_CONFORMANCE_ROOT="/tmp/litellm-mcp-conformance"
+                  bash .circleci/scripts/install_mcp_conformance.sh
+                  echo 'export MCP_CONFORMANCE_ROOT="/tmp/litellm-mcp-conformance"' >> "$BASH_ENV"
       - start_postgres:
           image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5
           server_args: "-c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all -c pg_stat_statements.max=20000"
@@ -3249,6 +3260,13 @@ jobs:
           name: Run owned integration contracts
           command: bash .circleci/scripts/run_integration.sh << parameters.suite >> << parameters.mode >>
           no_output_timeout: 15m
+      - when:
+          condition:
+            equal: [mcp, << parameters.suite >>]
+          steps:
+            - codecov/upload:
+                file: test-results/integration-mcp/conformance-coverage.xml
+                flags: mcp-conformance
       - run:
           name: Stop owned database and Redis
           when: always

.circleci/scripts/install_mcp_conformance.sh

@@ -0,0 +1,35 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+root="${MCP_CONFORMANCE_ROOT:?Set MCP_CONFORMANCE_ROOT to a fresh directory}"
+test ! -e "$root"
+archive="$(mktemp)"
+trap 'rm -f "$archive"' EXIT
+curl --fail --location --silent --show-error \
+  'https://codeload.github.com/modelcontextprotocol/conformance/tar.gz/7169291ec0b68eb370fddcd9947313ab0d5e4156' \
+  --output "$archive"
+PYTHONPATH=tests .venv/bin/python - "$archive" <<'PY'
+import sys
+from pathlib import Path
+from integration._support.conformance import verify_archive
+
+verify_archive(Path(sys.argv[1]), "51c1e27027f36be5b5f067746eb3a0f240fc3fbb7adcbd64bc4a79d86db65cd8")
+PY
+mkdir -p "$root"
+tar -xzf "$archive" --strip-components=1 -C "$root"
+npm ci --ignore-scripts --prefix "$root"
+# The current reference misclassifies legacy initialize _meta as stateless traffic.
+# Use the last official legacy reference; leave its source and lockfile untouched.
+curl --fail --location --silent --show-error \
+  'https://codeload.github.com/modelcontextprotocol/conformance/tar.gz/8f3994c75ff1aed1e39f91cff9358e2bc2c81dcd' \
+  --output "$archive"
+PYTHONPATH=tests .venv/bin/python - "$archive" <<'PYREF'
+import sys
+from pathlib import Path
+from integration._support.conformance import verify_archive
+
+verify_archive(Path(sys.argv[1]), "181119bd222f29db208b2952537430fcc865f91b64f2ac3d7525a67f490ebaa1")
+PYREF
+mkdir "$root/legacy-reference"
+tar -xzf "$archive" --strip-components=1 -C "$root/legacy-reference"
+npm ci --ignore-scripts --prefix "$root/legacy-reference/examples/servers/typescript"

.circleci/scripts/run_integration.sh

@@ -221,6 +221,12 @@ if [ "${CIRCLE_NODE_TOTAL:-1}" -gt 1 ]; then
   printf '%s\n' "${node_files[@]}" > "$results/node-files.txt"
 fi
 
+test_command=(.venv/bin/python tests/integration/run.py "$suite" --results "$results" "${node_files[@]}")
+if [ "$suite" = mcp ]; then
+  test_command=(.venv/bin/python -m coverage run --rcfile=tests/integration/conformance_coverage.toml \
+    tests/integration/run.py "$suite" --results "$results" "${node_files[@]}")
+fi
+
 env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
   INTEGRATION_RUN_ID="$integration_identity" \
   DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \
@@ -234,7 +240,17 @@ env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \
   INTEGRATION_PROXY_DATABASE_URL="$INTEGRATION_PROXY_DATABASE_URL" \
   INTEGRATION_PROXY_READ_REPLICA_URL="$INTEGRATION_PROXY_READ_REPLICA_URL" \
   INTEGRATION_ROUTING="$INTEGRATION_ROUTING" \
-  .venv/bin/python tests/integration/run.py "$suite" --results "$results" "${node_files[@]}"
+  MCP_CONFORMANCE_ROOT="${MCP_CONFORMANCE_ROOT:-}" \
+  COVERAGE_FILE="$PWD/$results/.coverage-conformance" "${test_command[@]}"
+
+if [ "$suite" = mcp ]; then
+  export COVERAGE_FILE="$PWD/$results/.coverage-conformance"
+  .venv/bin/python -m coverage run --rcfile=tests/integration/conformance_coverage.toml \
+    -m pytest tests/unit/integration_support/test_conformance.py -q
+  .venv/bin/python -m coverage combine --rcfile=tests/integration/conformance_coverage.toml
+  .venv/bin/python -m coverage xml --rcfile=tests/integration/conformance_coverage.toml \
+    -o "$results/conformance-coverage.xml"
+fi
 
 if [ "${INTEGRATION_COVERAGE:-0}" = 1 ]; then
   for covered_pid in "$proxy_pid" "$peer_pid"; do

tests/integration/_support/conformance.py

@@ -0,0 +1,174 @@
+import hashlib
+import json
+import os
+import queue
+import signal
+import subprocess
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import TYPE_CHECKING, Final, Literal
+
+import httpx
+import psutil
+from pydantic import BaseModel, Field, JsonValue, TypeAdapter
+
+if TYPE_CHECKING:
+    from integration._support.mcp import McpPeer
+
+
+class ConformanceCheck(BaseModel):
+    id: str
+    status: Literal["SUCCESS", "FAILURE", "WARNING", "INFO"]
+    errorMessage: str | None = None
+    details: dict[str, JsonValue] = Field(default_factory=dict)
+
+
+def verify_archive(archive: Path, expected_sha256: str) -> None:
+    actual: Final = hashlib.sha256(archive.read_bytes()).hexdigest()
+    assert actual == expected_sha256, f"SHA-256 mismatch for {archive.name}: {actual}"
+
+
+def read_checks(directory: Path, scenario: str) -> tuple[ConformanceCheck, ...]:
+    reports: Final = tuple(directory.rglob("checks.json"))
+    assert len(reports) == 1, f"conformance {scenario}: expected one fresh report, found {len(reports)}"
+    checks: Final = TypeAdapter(tuple[ConformanceCheck, ...]).validate_json(reports[0].read_bytes())
+    assert len({check.id for check in checks}) == len(checks), f"conformance {scenario}: duplicate checks"
+    required: Final = {
+        "server-initialize": ("server-initialize", "server-session-id-visible-ascii", "wire-schema-valid"),
+        "tools-list": ("tools-list", "tools-name-format", "wire-schema-valid"),
+        "tools-call-image": ("tools-call-image", "wire-schema-valid"),
+    }.get(scenario, (scenario,))
+    for identity in required:
+        assert tuple(check.status for check in checks if check.id == identity) == ("SUCCESS",), (
+            f"conformance {scenario}: required check {identity} did not pass: {checks}"
+        )
+    assert all(check.status != "FAILURE" for check in checks), f"conformance {scenario}: failed checks: {checks}"
+    if scenario == "tools-call-simple-text":
+        result: Final = next(check for check in checks if check.id == scenario).details.get("result")
+        # The official scenario accepts any nonempty text, including tool errors.
+        assert isinstance(result, dict) and result.get("isError", False) is False, f"reference payload: {result}"
+        assert result.get("content") == [{"type": "text", "text": "This is a simple text response for testing."}], (
+            f"reference payload: {result}"
+        )
+    return checks
+
+
+def run_scenario(root: Path, url: str, scenario: str, directory: Path) -> tuple[ConformanceCheck, ...]:
+    directory.mkdir(parents=True, exist_ok=False)
+    with (directory / "runner.log").open("w") as log:
+        result: Final = subprocess.run(
+            [
+                "node",
+                "--import",
+                "tsx",
+                "src/index.ts",
+                "server",
+                "--url",
+                url,
+                "--scenario",
+                scenario,
+                "--spec-version",
+                "2025-11-25",
+                "--output-dir",
+                str(directory.resolve()),
+                "--timeout",
+                "30000",
+            ],
+            cwd=root,
+            stdout=log,
+            stderr=subprocess.STDOUT,
+            timeout=45,
+            check=False,
+        )
+    assert result.returncode == 0, (directory / "runner.log").read_text()
+    return read_checks(directory, scenario)
+
+
+@contextmanager
+def reference_server(root: Path, directory: Path, port: int) -> Iterator["McpPeer"]:
+    from integration._support.client import eventually
+    from integration._support.mcp import McpPeer
+    from integration._support.process import group_members, signal_group, stop_root_process
+
+    directory.mkdir(parents=True, exist_ok=True)
+    url: Final = f"http://127.0.0.1:{port}"
+    with (directory / "reference.log").open("w") as log:
+        process: Final = subprocess.Popen(
+            ["node", "--import", "tsx", "everything-server.ts"],
+            cwd=root / "legacy-reference/examples/servers/typescript",
+            env={"PATH": os.environ["PATH"], "PORT": str(port)},
+            stdout=log,
+            stderr=subprocess.STDOUT,
+            start_new_session=True,
+        )
+        try:
+            with httpx.Client(timeout=1, trust_env=False) as client:
+
+                def ready() -> bool:
+                    assert process.poll() is None, (directory / "reference.log").read_text()
+                    try:
+                        return client.get(url + "/mcp").status_code == 400
+                    except httpx.TransportError:
+                        return False
+
+                eventually(ready, bool, seconds=15)
+            yield McpPeer(url + "/mcp", queue.Queue())
+        finally:
+            stopped: Final = stop_root_process(process)
+            residual: Final = group_members(process.pid)
+            if residual:
+                signal_group(process.pid, signal.SIGTERM)
+                psutil.wait_procs(residual, timeout=5)
+            remaining: Final = group_members(process.pid)
+            if remaining:
+                signal_group(process.pid, signal.SIGKILL)
+                psutil.wait_procs(remaining, timeout=3)
+            process.wait(timeout=3)
+            assert not group_members(process.pid), "Official reference child survived cleanup"
+            assert stopped and not remaining, "Official reference required forced cleanup"
+
+
+@contextmanager
+def authenticated_endpoint(target: str, key: str, alias: str) -> Iterator[str]:
+    from integration._support.asgi import asgi_server
+    from starlette.applications import Starlette
+    from starlette.requests import Request
+    from starlette.responses import StreamingResponse
+    from starlette.routing import Mount
+    from starlette.types import Receive, Scope, Send
+
+    async def forward(scope: Scope, receive: Receive, send: Send) -> None:
+        request: Final = Request(scope, receive)
+        original: Final = await request.body()
+        body: Final = prefixed_request(original, alias)
+        headers: Final = tuple(
+            (name, value)
+            for name, value in request.headers.raw
+            if name.lower() != b"authorization" and (name.lower() != b"content-length" or body == original)
+        ) + ((b"authorization", f"Bearer {key}".encode()),)
+        async with httpx.AsyncClient(timeout=35, trust_env=False) as client:
+            async with client.stream(request.method, target, headers=headers, content=body) as response:
+                streamed: Final = StreamingResponse(response.aiter_raw(), status_code=response.status_code)
+                streamed.raw_headers = [
+                    (name, value)
+                    for name, value in response.headers.raw
+                    if name.lower() not in (b"transfer-encoding", b"connection")
+                ]
+                await streamed(scope, receive, send)
+
+    with asgi_server(Starlette(routes=[Mount("/mcp", app=forward)])) as url:
+        yield url + "/mcp/"
+
+
+def prefixed_request(body: bytes, alias: str) -> bytes:
+    try:
+        request: Final = json.loads(body)
+    except (ValueError, UnicodeDecodeError):
+        return body
+    if not isinstance(request, dict) or request.get("method") not in ("tools/call", "prompts/get"):
+        return body
+    params: Final = request.get("params")
+    if not isinstance(params, dict) or not isinstance(params.get("name"), str):
+        return body
+    return json.dumps({**request, "params": {**params, "name": f"{alias}-{params['name']}"}}).encode()

tests/integration/_support/process.py

@@ -103,6 +103,8 @@ def owned_proxy_process(
         "LITELLM_MASTER_KEY": gateway.key,
         "LITELLM_SALT_KEY": os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt"),
         "STORE_MODEL_IN_DB": "True",
+        "COVERAGE_PROCESS_CONFIG": "",
+        "COVERAGE_PROCESS_START": "",
         **overrides,
     }
     output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(directory)))

tests/integration/conformance_coverage.toml

@@ -0,0 +1,9 @@
+[tool.coverage.run]
+branch = true
+parallel = true
+relative_files = true
+patch = ["subprocess"]
+include = ["tests/integration/_support/conformance.py", "tests/integration/_support/process.py"]
+
+[tool.coverage.report]
+show_missing = true

tests/integration/mcp/test_mcp_official_conformance.py

@@ -0,0 +1,144 @@
+import asyncio
+import json
+import os
+import uuid
+from pathlib import Path
+from typing import Final
+
+import httpx
+import pytest
+from integration._support.client import Gateway
+from integration._support.conformance import authenticated_endpoint, reference_server, run_scenario
+from integration._support.mcp import official_client_outcomes, register_mcp
+from integration._support.wire import Reply, wire_server
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+from mcp.types import ImageContent
+
+
+@pytest.mark.parametrize("name", ("server-initialize", "tools-list", "tools-call-image"))
+def test_official_scenario_through_gateway(gateway: Gateway, tmp_path: Path, unused_tcp_port: int, name: str) -> None:
+    root: Final = Path(os.environ["MCP_CONFORMANCE_ROOT"])
+    output: Final = Path(os.environ.get("INTEGRATION_RESULTS_DIR", str(tmp_path))) / f"conformance-{uuid.uuid4().hex}"
+    with reference_server(root, output, unused_tcp_port) as reference, gateway.scenario() as scenario:
+        alias: Final = "official" + uuid.uuid4().hex[:8]
+        identity: Final = register_mcp(scenario, reference, alias, mcp_info={"protocol_version": "2025-11-25"})
+        key: Final = scenario.key(object_permission={"mcp_servers": [identity]})
+        direct: Final = run_scenario(root, reference.url, name, output / "direct")
+        endpoint: Final = str(gateway.client.base_url).rstrip("/") + f"/{alias}/mcp"
+        with authenticated_endpoint(endpoint, key, alias) as authenticated:
+            proxied: Final = run_scenario(root, authenticated, name, output / "gateway")
+        if name == "tools-call-image":
+
+            async def check_image(url: str, tool: str, token: str | None = None) -> None:
+                async with httpx.AsyncClient(headers={"Authorization": f"Bearer {token}"} if token else {}) as http:
+                    async with streamable_http_client(url, http_client=http) as streams:
+                        async with ClientSession(streams[0], streams[1]) as session:
+                            await session.initialize()
+                            result: Final = await session.call_tool(tool, {})
+                            assert result.is_error is False and len(result.content) == 1, result
+                            content: Final = result.content[0]
+                            assert isinstance(content, ImageContent) and content.mime_type == "image/png", content
+                            assert content.data == (
+                                "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlE"
+                                "QVR42mP8z8DwHwAFBQIAX8jx0gAAAABJRU5ErkJggg=="
+                            ), content
+
+            asyncio.run(check_image(reference.url, "test_image_content"))
+            asyncio.run(check_image(endpoint, f"{alias}-test_image_content", key))
+        assert tuple(check.status for check in direct if check.id == name) == ("SUCCESS",)
+        assert tuple(check.status for check in proxied if check.id == name) == ("SUCCESS",)
+        listed, called = official_client_outcomes(gateway, key, f"/{alias}/mcp", f"{alias}-test_simple_text", {})
+        assert f"{alias}-test_simple_text" in listed.tools, listed
+        assert called.ok and called.text == "This is a simple text response for testing.", called
+
+
+@pytest.mark.parametrize("status", (200, 403))
+def test_conformance_bridge_preserves_headers_payload_and_error_status(status: int) -> None:
+    body: Final = {"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "test_image_content"}}
+    response_body: Final = b'data: {"jsonrpc":"2.0","id":7,"error":{"code":-32000,"message":"denied"}}\n\n'
+    with wire_server(
+        lambda request: Reply(
+            status=status, chunks=(response_body[:12], response_body[12:]), content_type="text/event-stream"
+        )
+    ) as upstream:
+        with authenticated_endpoint(upstream.url + "/mcp", "test-key", "official") as endpoint:
+            response: Final = httpx.post(
+                endpoint,
+                json=body,
+                headers={
+                    "Host": "127.0.0.1",
+                    "Origin": "https://untrusted.example",
+                    "Authorization": "Bearer replaced",
+                    "MCP-Protocol-Version": "2025-03-26",
+                    "Mcp-Session-Id": "same-session",
+                },
+            )
+        assert response.status_code == status and response.content == response_body
+        requests: Final = upstream.drain()
+        assert len(requests) == 1
+        received: Final = requests[0]
+        assert json.loads(received.body) == {**body, "params": {"name": "official-test_image_content"}}
+        assert {
+            name: received.headers[name]
+            for name in ("host", "origin", "authorization", "mcp-protocol-version", "mcp-session-id")
+        } == {
+            "host": "127.0.0.1",
+            "origin": "https://untrusted.example",
+            "authorization": "Bearer test-key",
+            "mcp-protocol-version": "2025-03-26",
+            "mcp-session-id": "same-session",
+        }
+
+
+def test_official_runner_rejects_unknown_scenario(tmp_path: Path) -> None:
+    output: Final = tmp_path / "unknown-scenario"
+    with pytest.raises(AssertionError):
+        run_scenario(Path(os.environ["MCP_CONFORMANCE_ROOT"]), "http://127.0.0.1:1/mcp", "missing-scenario", output)
+    assert "missing-scenario" in (output / "runner.log").read_text()
+
+
+def test_stalled_reference_is_killed_and_cannot_report_clean_teardown(tmp_path: Path, unused_tcp_port: int) -> None:
+    import signal
+    from contextlib import ExitStack
+
+    import psutil
+
+    children: Final = frozenset(child.pid for child in psutil.Process().children())
+    with ExitStack() as cleanup:
+        cleanup.enter_context(reference_server(Path(os.environ["MCP_CONFORMANCE_ROOT"]), tmp_path, unused_tcp_port))
+        started: Final = tuple(child for child in psutil.Process().children() if child.pid not in children)
+        assert len(started) == 1, started
+        victim: Final = started[0]
+        victim.send_signal(signal.SIGSTOP)
+        with pytest.raises(AssertionError, match="forced cleanup"):
+            cleanup.close()
+    assert not psutil.pid_exists(victim.pid), "Stopped official reference survived forced cleanup"
+
+
+def test_reference_children_are_stopped_after_the_root_exits(tmp_path: Path, unused_tcp_port: int) -> None:
+    import psutil
+
+    source: Final = tmp_path / "legacy-reference/examples/servers/typescript"
+    source.mkdir(parents=True)
+    (source / "node_modules").symlink_to(
+        Path(os.environ["MCP_CONFORMANCE_ROOT"]) / "legacy-reference/examples/servers/typescript/node_modules",
+        target_is_directory=True,
+    )
+    (source / "everything-server.ts").write_text(
+        "import http from 'node:http';\n"
+        "import { spawn } from 'node:child_process';\n"
+        "import { writeFileSync } from 'node:fs';\n"
+        "const child = spawn(process.execPath, ['-e', 'setInterval(() => {}, 1000)'], {stdio: 'ignore'});\n"
+        "writeFileSync('child.pid', String(child.pid));\n"
+        "http.createServer((req, res) => { res.writeHead(400); res.end(); }).listen(Number(process.env.PORT));\n"
+        "process.on('SIGTERM', () => process.exit(0));\n"
+    )
+    with reference_server(tmp_path, tmp_path / "logs", unused_tcp_port):
+        child: Final = psutil.Process(int((source / "child.pid").read_text()))
+    try:
+        assert not child.is_running(), "Reference child survived a clean root exit"
+    finally:
+        if child.is_running():
+            child.kill()
+            psutil.wait_procs((child,), timeout=3)

tests/unit/integration_support/test_conformance.py

@@ -0,0 +1,183 @@
+import hashlib
+import json
+import os
+from pathlib import Path
+from typing import Final
+from unittest.mock import Mock, patch
+
+import pytest
+from tests.integration._support.conformance import read_checks, verify_archive, prefixed_request
+from pydantic import ValidationError
+
+
+@pytest.mark.parametrize("explicit", (False, True))
+def test_owned_proxy_isolates_automatic_coverage_unless_requested(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, explicit: bool
+) -> None:
+    monkeypatch.syspath_prepend(str(Path(__file__).resolve().parents[2]))
+    from integration._support.client import Gateway
+    from integration._support.process import owned_proxy_process
+
+    monkeypatch.setenv("COVERAGE_PROCESS_CONFIG", "parent-config")
+    monkeypatch.setenv("COVERAGE_PROCESS_START", "parent.toml")
+    overrides: Final = {"COVERAGE_PROCESS_CONFIG": "child-config"} if explicit else {}
+    gateway: Final = Gateway(Mock(), "owner-key", "http://upstream")
+    with (
+        patch("integration._support.process.subprocess.Popen") as spawn,
+        patch("integration._support.process.httpx.Client") as client,
+        patch("integration._support.process.group_members", return_value=()),
+    ):
+        spawn.return_value.poll.return_value = None
+        client.return_value.__enter__.return_value.get.return_value.status_code = 200
+        with owned_proxy_process(gateway, tmp_path, overrides):
+            child: Final = spawn.call_args.kwargs["env"]
+            assert child["COVERAGE_PROCESS_CONFIG"] == ("child-config" if explicit else "")
+            assert child["COVERAGE_PROCESS_START"] == ""
+            assert child["LITELLM_MASTER_KEY"] == gateway.key
+    assert os.environ["COVERAGE_PROCESS_CONFIG"] == "parent-config"
+
+
+def test_changed_archive_is_rejected(tmp_path: Path) -> None:
+    archive: Final = tmp_path / "reference.tar.gz"
+    archive.write_bytes(b"changed reference")
+    with pytest.raises(AssertionError, match="SHA-256"):
+        verify_archive(archive, hashlib.sha256(b"approved reference").hexdigest())
+
+
+def test_matching_archive_is_accepted(tmp_path: Path) -> None:
+    archive: Final = tmp_path / "reference.tar.gz"
+    archive.write_bytes(b"approved reference")
+    assert verify_archive(archive, "5a870d3ee9520e3a912c735d007d7b40f6d3cbf8b61d909ebb8361f423d4ba1e") is None
+
+
+@pytest.mark.parametrize(
+    "checks",
+    (
+        [],
+        [{"id": "tools-list", "status": "FAILURE"}],
+        [{"id": "tools-list", "status": "WARNING"}],
+        [{"id": "wire-schema", "status": "SUCCESS"}],
+        [{"id": "tools-list", "status": "SUCCESS"}, {"id": "wire-schema", "status": "FAILURE"}],
+        [{"id": "tools-list", "status": "SUCCESS"}, {"id": "tools-list", "status": "SUCCESS"}],
+    ),
+)
+def test_incomplete_or_failed_scenario_cannot_pass(tmp_path: Path, checks: list[dict[str, str]]) -> None:
+    (tmp_path / "checks.json").write_text(json.dumps(checks))
+    with pytest.raises(AssertionError, match="conformance"):
+        read_checks(tmp_path, "tools-list")
+
+
+def test_skipped_scenario_without_a_report_cannot_pass(tmp_path: Path) -> None:
+    with pytest.raises(AssertionError, match="conformance"):
+        read_checks(tmp_path, "tools-list")
+
+
+def test_multiple_reports_cannot_supply_a_stale_pass(tmp_path: Path) -> None:
+    for name in ("previous", "current"):
+        directory: Final = tmp_path / name
+        directory.mkdir()
+        (directory / "checks.json").write_text('[{"id":"tools-list","status":"SUCCESS"}]')
+    with pytest.raises(AssertionError, match="conformance"):
+        read_checks(tmp_path, "tools-list")
+
+
+def test_passing_scenario_keeps_nonbinding_diagnostics(tmp_path: Path) -> None:
+    (tmp_path / "checks.json").write_text(
+        json.dumps(
+            [
+                {"id": identity, "status": "INFO" if identity == "advisory" else "SUCCESS"}
+                for identity in ("tools-list", "tools-name-format", "wire-schema-valid", "advisory")
+            ]
+        )
+    )
+    checks: Final = read_checks(tmp_path, "tools-list")
+    assert checks[-1].id == "advisory" and checks[-1].status == "INFO"
+
+
+def test_unrecognized_result_status_cannot_pass(tmp_path: Path) -> None:
+    (tmp_path / "checks.json").write_text('[{"id":"tools-list","status":"SKIPPED"}]')
+    with pytest.raises(ValidationError, match="status"):
+        read_checks(tmp_path, "tools-list")
+
+
+@pytest.mark.parametrize(
+    "result",
+    (
+        {"content": [{"type": "text", "text": "Tool not found"}], "isError": True},
+        {"content": [{"type": "text", "text": "wrong upstream payload"}]},
+    ),
+)
+def test_official_text_success_cannot_hide_an_error_or_wrong_payload(tmp_path: Path, result: dict[str, object]) -> None:
+    (tmp_path / "checks.json").write_text(
+        json.dumps([{"id": "tools-call-simple-text", "status": "SUCCESS", "details": {"result": result}}])
+    )
+    with pytest.raises(AssertionError, match="reference payload"):
+        read_checks(tmp_path, "tools-call-simple-text")
+
+
+def test_official_text_success_requires_the_reference_payload(tmp_path: Path) -> None:
+    (tmp_path / "checks.json").write_text(
+        json.dumps(
+            [
+                {
+                    "id": "tools-call-simple-text",
+                    "status": "SUCCESS",
+                    "details": {
+                        "result": {
+                            "content": [{"type": "text", "text": "This is a simple text response for testing."}],
+                            "isError": False,
+                        }
+                    },
+                }
+            ]
+        )
+    )
+    assert read_checks(tmp_path, "tools-call-simple-text")[0].status == "SUCCESS"
+
+
+@pytest.mark.parametrize("method", ("tools/call", "prompts/get"))
+def test_only_fixture_name_is_translated_to_advertised_name(method: str) -> None:
+    request: Final = {
+        "jsonrpc": "2.0",
+        "id": 4,
+        "method": method,
+        "params": {"name": "test_tool", "_meta": {"progressToken": "keep"}},
+    }
+    changed: Final = json.loads(prefixed_request(json.dumps(request).encode(), "official"))
+    assert changed == {**request, "params": {"name": "official-test_tool", "_meta": {"progressToken": "keep"}}}
+    assert "arguments" not in changed["params"]
+
+
+@pytest.mark.parametrize(
+    "body",
+    (
+        b"malformed JSON",
+        b"[]",
+        b'{"method":"initialize","params":{"protocolVersion":"2025-03-26"}}',
+        b'{"method":"tools/call","params":{"name":17}}',
+    ),
+)
+def test_name_integration_preserves_other_requests(body: bytes) -> None:
+    assert prefixed_request(body, "official") == body
+
+
+@pytest.mark.parametrize(
+    "scenario,identities",
+    (
+        ("server-initialize", ("server-initialize", "server-session-id-visible-ascii", "wire-schema-valid")),
+        ("tools-list", ("tools-list", "tools-name-format", "wire-schema-valid")),
+        ("tools-call-image", ("tools-call-image", "wire-schema-valid")),
+    ),
+)
+def test_missing_secondary_checks_cannot_report_complete_conformance(
+    tmp_path: Path, scenario: str, identities: tuple[str, ...]
+) -> None:
+    report: Final = tmp_path / "checks.json"
+    for missing in identities:
+        report.write_text(
+            json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities if identity != missing])
+        )
+        with pytest.raises(AssertionError, match="conformance"):
+            read_checks(tmp_path, scenario)
+    report.write_text(json.dumps([{"id": identity, "status": "SUCCESS"} for identity in identities]))
+    assert len(read_checks(tmp_path, scenario)) == len(identities)