scripts/mcp_conformance/run_codex_compliance.py

#!/usr/bin/env python3
"""Black-box MCP compliance runner for a supplied Codex binary."""

import argparse
import importlib.util
import json
import os
import queue
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from collections import deque
from contextlib import contextmanager, nullcontext
from dataclasses import asdict, dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Iterator, Mapping, Sequence

_MODULE_DIR = Path(__file__).resolve().parent
if str(_MODULE_DIR) not in sys.path:
    # Some execution environments set PYTHONSAFEPATH. Add only this
    # trusted package directory so direct `python path/to/script.py` usage
    # continues to work alongside the Bazel entry point.
    sys.path.insert(0, str(_MODULE_DIR))

from official_conformance import (  # noqa: E402 - direct scripts must first add their sibling directory.
    OFFICIAL_CONFORMANCE_GIT_REF,
    OFFICIAL_CONFORMANCE_REPOSITORY,
    OfficialScenarioResult,
    default_conformance_command,
    run_official_mode,
    scenarios_for_mode,
)

_FIXTURE_MODULE_PATH = Path(__file__).resolve().with_name("server.py")
_FIXTURE_SPEC = importlib.util.spec_from_file_location(
    "_mcp_spec_test_fixture_server",
    _FIXTURE_MODULE_PATH,
)
if _FIXTURE_SPEC is None or _FIXTURE_SPEC.loader is None:
    raise RuntimeError(f"could not load MCP fixture module: {_FIXTURE_MODULE_PATH}")
_FIXTURE_MODULE = importlib.util.module_from_spec(_FIXTURE_SPEC)
sys.modules[_FIXTURE_SPEC.name] = _FIXTURE_MODULE
_FIXTURE_SPEC.loader.exec_module(_FIXTURE_MODULE)

LEGACY_VERSION = _FIXTURE_MODULE.LEGACY_VERSION
SHIPPING_LEGACY_VERSION = _FIXTURE_MODULE.SHIPPING_LEGACY_VERSION
MODERN_VERSION = _FIXTURE_MODULE.MODERN_VERSION
MISMATCHED_DISCOVERY_ID_PROFILE = _FIXTURE_MODULE.MISMATCHED_DISCOVERY_ID_PROFILE
NULL_DISCOVERY_ID_PROFILE = _FIXTURE_MODULE.NULL_DISCOVERY_ID_PROFILE
REPEATED_CURSOR_PROFILE = _FIXTURE_MODULE.REPEATED_CURSOR_PROFILE
REVIEW_EXACT_INTEGER = _FIXTURE_MODULE.REVIEW_EXACT_INTEGER
REVIEW_MRTR_INPUT_REQUEST_COUNT = _FIXTURE_MODULE.REVIEW_MRTR_INPUT_REQUEST_COUNT
REVIEW_PROFILE = _FIXTURE_MODULE.REVIEW_PROFILE
RESOURCE_URIS = _FIXTURE_MODULE.RESOURCE_URIS
SERVER_NAME = _FIXTURE_MODULE.SERVER_NAME
SERVER_VERSION = _FIXTURE_MODULE.SERVER_VERSION
SSE_COMMENT_FLOOD_PROFILE = _FIXTURE_MODULE.SSE_COMMENT_FLOOD_PROFILE
SSE_CR_COMMENTS_PROFILE = _FIXTURE_MODULE.SSE_CR_COMMENTS_PROFILE
ProtocolServer = _FIXTURE_MODULE.ProtocolServer
make_http_server = _FIXTURE_MODULE.make_http_server

REPORT_SCHEMA_VERSION = 4
COMPACT_REGRESSION_BASELINE_KIND = "mcp-conformance-regression-baseline-v1"
REQUIRED_REGRESSION_MODES = (
    SHIPPING_LEGACY_VERSION,
    LEGACY_VERSION,
    MODERN_VERSION,
)
TEST_SERVER_NAME = "mcp_spec_fixture"
TEST_META_KEY = "com.openai/mcp-spec-test"
CHECK_ORDER = (
    "mcp_add",
    "mcp_get",
    "app_server_initialize",
    "modern_feature_enablement",
    "inventory",
    "ephemeral_thread",
    "echo_tool",
    "unicode_resource_read",
    "per_request_metadata",
    "request_scoped_notifications",
    "multi_round_trip_request",
    "http_header_mirroring",
    "app_server_protocol",
    "case_runtime",
    "mcp_remove",
    "isolated_config_cleanup",
)


@dataclass
class CommandResult:
    returncode: int
    stdout: str
    stderr: str


@dataclass
class CheckResult:
    name: str
    success: bool
    detail: str
    status: str | None = None
    source: str = "supplemental"
    scenario: str | None = None
    check_id: str | None = None
    category: str | None = None

    def __post_init__(self) -> None:
        if self.status is None:
            self.status = "PASS" if self.success else "FAIL"


@dataclass
class CaseResult:
    transport: str
    mode: str
    success: bool = False
    duration_seconds: float = 0.0
    checks: list[CheckResult] = field(default_factory=list)
    diagnostics: str | None = None

    def check(self, name: str, success: bool, detail: str) -> bool:
        self.checks.append(
            CheckResult(
                name=name,
                success=success,
                detail=detail,
                status="PASS" if success else "FAIL",
            )
        )
        return success

    def finish(self, started_at: float) -> None:
        self.duration_seconds = round(time.monotonic() - started_at, 3)
        self.success = bool(self.checks) and all(check.success for check in self.checks)


@dataclass(frozen=True, order=True)
class _RegressionCheckIdentity:
    mode: str
    transport: str
    source: str
    scenario: str
    check_id: str


class AppServerError(RuntimeError):
    pass


class AppServerClient:
    """Minimal JSONL client for the Codex app-server protocol."""

    def __init__(
        self,
        codex_binary: Path,
        *,
        env: Mapping[str, str],
        cwd: Path,
        timeout_seconds: float,
        elicitation_content: (
            Callable[[Mapping[str, object]], Mapping[str, object]] | None
        ) = None,
    ) -> None:
        self.timeout_seconds = timeout_seconds
        self._elicitation_content = elicitation_content
        self._next_id = 1
        self._messages: queue.Queue[dict[str, object] | None] = queue.Queue()
        self._write_lock = threading.Lock()
        self._stderr: deque[str] = deque(maxlen=200)
        self._parse_errors: deque[str] = deque(maxlen=20)
        self.events: list[dict[str, object]] = []
        self.elicitation_requests: list[dict[str, object]] = []
        self.process = subprocess.Popen(
            [str(codex_binary), "app-server"],
            cwd=cwd,
            env=dict(env),
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )
        self._stdout_thread = threading.Thread(
            target=self._read_stdout,
            name="codex-app-server-stdout",
            daemon=True,
        )
        self._stderr_thread = threading.Thread(
            target=self._read_stderr,
            name="codex-app-server-stderr",
            daemon=True,
        )
        self._stdout_thread.start()
        self._stderr_thread.start()

    def __enter__(self) -> "AppServerClient":
        return self

    def __exit__(self, *_: object) -> None:
        self.close()

    def _read_stdout(self) -> None:
        assert self.process.stdout is not None
        for line in self.process.stdout:
            stripped = line.strip()
            if not stripped:
                continue
            try:
                message = json.loads(stripped)
            except json.JSONDecodeError:
                self._parse_errors.append(stripped[:500])
                continue
            if isinstance(message, dict):
                self._messages.put(message)
        self._messages.put(None)

    def _read_stderr(self) -> None:
        assert self.process.stderr is not None
        for line in self.process.stderr:
            self._stderr.append(line.rstrip())

    def _send(self, message: Mapping[str, object]) -> None:
        if self.process.poll() is not None:
            raise AppServerError(
                f"Codex app-server exited with code {self.process.returncode}"
            )
        assert self.process.stdin is not None
        encoded = json.dumps(message, ensure_ascii=False, separators=(",", ":"))
        with self._write_lock:
            self.process.stdin.write(encoded + "\n")
            self.process.stdin.flush()

    def notify(self, method: str) -> None:
        self._send({"method": method})

    def request(
        self,
        method: str,
        params: Mapping[str, object] | None,
    ) -> dict[str, object]:
        request_id = self._next_id
        self._next_id += 1
        message: dict[str, object] = {"id": request_id, "method": method}
        if params is not None:
            message["params"] = dict(params)
        self._send(message)

        deadline = time.monotonic() + self.timeout_seconds
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise AppServerError(
                    f"timed out waiting for app-server response to {method}"
                )
            try:
                message = self._messages.get(timeout=remaining)
            except queue.Empty as exc:
                raise AppServerError(
                    f"timed out waiting for app-server response to {method}"
                ) from exc
            if message is None:
                raise AppServerError(
                    f"Codex app-server closed stdout while handling {method}"
                )
            if message.get("id") == request_id and "method" not in message:
                return message
            if "id" in message and isinstance(message.get("method"), str):
                self._handle_server_request(message)
            else:
                self.events.append(message)

    def wait_for_notification(
        self,
        method: str,
        *,
        predicate: Callable[[Mapping[str, object]], bool] | None = None,
        after_event_index: int = 0,
    ) -> dict[str, object]:
        for event in self.events[after_event_index:]:
            if event.get("method") != method:
                continue
            params = event.get("params")
            if isinstance(params, dict) and (predicate is None or predicate(params)):
                return event

        deadline = time.monotonic() + self.timeout_seconds
        while True:
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                raise AppServerError(
                    f"timed out waiting for app-server notification {method}"
                )
            try:
                message = self._messages.get(timeout=remaining)
            except queue.Empty as exc:
                raise AppServerError(
                    f"timed out waiting for app-server notification {method}"
                ) from exc
            if message is None:
                raise AppServerError(
                    f"Codex app-server closed stdout while waiting for {method}"
                )
            if "id" in message and isinstance(message.get("method"), str):
                self._handle_server_request(message)
                continue
            self.events.append(message)
            if message.get("method") != method:
                continue
            params = message.get("params")
            if isinstance(params, dict) and (predicate is None or predicate(params)):
                return message

    def _handle_server_request(self, message: dict[str, object]) -> None:
        self.events.append(message)
        request_id = message.get("id")
        method = message.get("method")
        if method == "mcpServer/elicitation/request":
            params = message.get("params")
            if isinstance(params, dict):
                self.elicitation_requests.append(params)
            else:
                params = {}
            content = (
                dict(self._elicitation_content(params))
                if self._elicitation_content is not None
                else {"confirmation": "confirmed"}
            )
            self._send(
                {
                    "id": request_id,
                    "result": {
                        "action": "accept",
                        "content": content,
                        "_meta": None,
                    },
                }
            )
            return
        self._send(
            {
                "id": request_id,
                "error": {
                    "code": -32601,
                    "message": f"unsupported test client request: {method}",
                },
            }
        )

    def diagnostic_text(self) -> str:
        parts: list[str] = []
        if self._parse_errors:
            parts.append("non-JSON stdout: " + " | ".join(self._parse_errors))
        startup_failures = [
            event
            for event in self.events
            if event.get("method") == "mcpServer/startupStatus/updated"
            and isinstance(event.get("params"), dict)
            and event["params"].get("status") not in ("starting", "ready")
        ]
        if startup_failures:
            parts.append(
                "startup events: "
                + json.dumps(startup_failures[-3:], ensure_ascii=False)
            )
        if self._stderr:
            parts.append("stderr:\n" + "\n".join(self._stderr))
        return "\n".join(parts)[-8_000:]

    def close(self) -> None:
        if self.process.poll() is None:
            self.process.terminate()
            try:
                self.process.wait(timeout=3)
            except subprocess.TimeoutExpired:
                self.process.kill()
                self.process.wait(timeout=3)
        for stream in (self.process.stdin, self.process.stdout, self.process.stderr):
            if stream is not None:
                stream.close()
        self._stdout_thread.join(timeout=1)
        self._stderr_thread.join(timeout=1)


def _run_command(
    command: Sequence[str],
    *,
    env: Mapping[str, str],
    cwd: Path,
    timeout_seconds: float,
) -> CommandResult:
    try:
        completed = subprocess.run(
            list(command),
            cwd=cwd,
            env=dict(env),
            text=True,
            capture_output=True,
            timeout=timeout_seconds,
            check=False,
        )
        return CommandResult(
            completed.returncode,
            completed.stdout.strip(),
            completed.stderr.strip(),
        )
    except subprocess.TimeoutExpired as exc:
        stdout = (
            exc.stdout.decode() if isinstance(exc.stdout, bytes) else exc.stdout or ""
        )
        stderr = (
            exc.stderr.decode() if isinstance(exc.stderr, bytes) else exc.stderr or ""
        )
        return CommandResult(
            124,
            stdout.strip(),
            (stderr + f"\ncommand timed out after {timeout_seconds:g}s").strip(),
        )


def _response_result(
    response: Mapping[str, object],
) -> tuple[dict[str, object] | None, str]:
    error = response.get("error")
    if isinstance(error, dict):
        return None, json.dumps(error, ensure_ascii=False, sort_keys=True)
    result = response.get("result")
    if not isinstance(result, dict):
        return None, "response did not contain an object result"
    return result, "ok"


def _command_detail(result: CommandResult) -> str:
    detail = result.stdout or result.stderr or f"exit code {result.returncode}"
    return detail[-2_000:]


def _result_detail(
    result: Mapping[str, object] | None,
    fallback: str,
) -> str:
    if result is None:
        return fallback
    return (
        "unexpected result: "
        + json.dumps(result, ensure_ascii=False, sort_keys=True, default=str)
    )[-2_000:]


def _isolated_environment(codex_home: Path) -> dict[str, str]:
    env = dict(os.environ)
    env["CODEX_HOME"] = str(codex_home)
    # Direct app-server MCP calls do not need a model or credentials.
    for name in ("CODEX_API_KEY", "CODEX_ACCESS_TOKEN", "OPENAI_API_KEY"):
        env.pop(name, None)
    return env


@contextmanager
def _running_http_fixture(mode: str) -> Iterator[str]:
    httpd = make_http_server(
        ProtocolServer(mode),
        "127.0.0.1",
        0,
        log_requests=False,
    )
    thread = threading.Thread(
        target=httpd.serve_forever,
        name=f"mcp-fixture-{mode}",
        daemon=True,
    )
    thread.start()
    try:
        _, port = httpd.server_address
        yield f"http://127.0.0.1:{port}/mcp"
    finally:
        httpd.shutdown()
        httpd.server_close()
        thread.join(timeout=3)


def _registration_command(
    codex_binary: Path,
    server_script: Path,
    *,
    transport: str,
    mode: str,
    http_url: str | None,
) -> list[str]:
    command = [str(codex_binary), "mcp", "add", TEST_SERVER_NAME]
    if transport == "stdio":
        if mode == MODERN_VERSION:
            command.extend(["--env", f"CODEX_MCP_PROTOCOL_VERSION={MODERN_VERSION}"])
        return [
            *command,
            "--",
            sys.executable,
            str(server_script),
            "--mode",
            mode,
            "--transport",
            "stdio",
        ]
    assert http_url is not None
    return [*command, "--url", http_url]


def _validate_registration(
    config: Mapping[str, object],
    *,
    transport: str,
    mode: str,
    http_url: str | None,
) -> tuple[bool, str]:
    if config.get("name") != TEST_SERVER_NAME or config.get("enabled") is not True:
        return False, "registered server name or enabled state was incorrect"
    value = config.get("transport")
    if not isinstance(value, dict):
        return False, "registered server did not contain a transport object"
    if transport == "stdio":
        args = value.get("args")
        if value.get("type") != "stdio" or not isinstance(args, list):
            return False, f"unexpected stdio transport: {value!r}"
        if mode not in args or "stdio" not in args:
            return False, f"stdio registration omitted mode or transport: {args!r}"
        env = value.get("env")
        modern_opt_in = (
            isinstance(env, dict)
            and env.get("CODEX_MCP_PROTOCOL_VERSION") == MODERN_VERSION
        )
        if mode == MODERN_VERSION and not modern_opt_in:
            return False, "modern stdio registration omitted its protocol opt-in"
        if mode in (SHIPPING_LEGACY_VERSION, LEGACY_VERSION) and modern_opt_in:
            return (
                False,
                "legacy stdio registration unexpectedly enabled the modern protocol",
            )
    elif value.get("type") != "streamable_http" or value.get("url") != http_url:
        return False, f"unexpected HTTP transport: {value!r}"
    return True, f"registered {value.get('type')} transport"


def _validate_inventory(
    result: Mapping[str, object],
    *,
    mode: str,
) -> tuple[bool, str]:
    entries = result.get("data")
    if not isinstance(entries, list):
        return False, "mcpServerStatus/list result did not contain data"
    entry = next(
        (
            item
            for item in entries
            if isinstance(item, dict) and item.get("name") == TEST_SERVER_NAME
        ),
        None,
    )
    if not isinstance(entry, dict):
        return False, f"{TEST_SERVER_NAME!r} was absent from MCP status"

    missing: list[str] = []
    server_info = entry.get("serverInfo")
    if (
        not isinstance(server_info, dict)
        or server_info.get("name") != SERVER_NAME
        or server_info.get("version") != SERVER_VERSION
    ):
        missing.append("server identity")
    tools = entry.get("tools")
    expected_tools = {"echo", "client_metadata", "progress"}
    if mode == MODERN_VERSION:
        expected_tools.add("request_input")
    if not isinstance(tools, dict):
        missing.append("tool inventory")
    else:
        absent_tools = sorted(expected_tools - set(tools))
        if absent_tools:
            missing.append("tools " + ", ".join(absent_tools))
    resources = entry.get("resources")
    resource_uris = (
        {
            item.get("uri")
            for item in resources
            if isinstance(item, dict) and isinstance(item.get("uri"), str)
        }
        if isinstance(resources, list)
        else set()
    )
    absent_resources = sorted(set(RESOURCE_URIS) - resource_uris)
    if absent_resources:
        missing.append("paginated resources " + ", ".join(absent_resources))
    if missing:
        return False, "missing " + "; ".join(missing)
    return True, "identity, tools, and both paginated resources discovered"


def _validate_echo(
    result: Mapping[str, object],
    sentinel: str,
) -> tuple[bool, str]:
    structured = result.get("structuredContent")
    if isinstance(structured, dict) and structured.get("text") == sentinel:
        return True, "echo returned the exact sentinel"
    return False, f"unexpected echo result: {dict(result)!r}"


def _ratio(passed: int, total: int) -> dict[str, int | float]:
    percentage = round((passed / total) * 100, 1) if total else 0.0
    return {
        "passed": passed,
        "total": total,
        "percentage": percentage,
    }


def _check_ratio(checks: Sequence[CheckResult]) -> dict[str, int | float]:
    applicable = [check for check in checks if check.status != "SKIP"]
    return _ratio(
        sum(check.success for check in applicable),
        len(applicable),
    )


def _scenario_ratio(
    checks: Sequence[CheckResult],
    *,
    category: str,
) -> dict[str, int | float]:
    by_scenario: dict[str, list[CheckResult]] = {}
    for check in checks:
        if (
            check.scenario is None
            or check.category != category
            or check.source not in {"official", "harness"}
        ):
            continue
        by_scenario.setdefault(check.scenario, []).append(check)
    return _ratio(
        sum(
            all(check.success for check in scenario_checks)
            for scenario_checks in by_scenario.values()
        ),
        len(by_scenario),
    )


def _summarize_modes(cases: Sequence[CaseResult]) -> list[dict[str, object]]:
    summaries: list[dict[str, object]] = []
    modes = dict.fromkeys(case.mode for case in cases)
    for mode in modes:
        mode_cases = [case for case in cases if case.mode == mode]
        checks = [check for case in mode_cases for check in case.checks]
        official_checks = [check for check in checks if check.source == "official"]
        official_non_auth_checks = [
            check for check in official_checks if check.category == "non-auth"
        ]
        official_auth_checks = [
            check for check in official_checks if check.category == "auth"
        ]
        harness_checks = [check for check in checks if check.source == "harness"]
        supplemental_checks = [
            check for check in checks if check.source == "supplemental"
        ]
        summaries.append(
            {
                "mode": mode,
                "checks": _check_ratio(checks),
                "officialChecks": _check_ratio(official_checks),
                "officialNonAuthChecks": _check_ratio(official_non_auth_checks),
                "officialAuthChecks": _check_ratio(official_auth_checks),
                "officialNonAuthScenarios": _scenario_ratio(
                    checks,
                    category="non-auth",
                ),
                "officialAuthScenarios": _scenario_ratio(
                    checks,
                    category="auth",
                ),
                "harnessChecks": _check_ratio(harness_checks),
                "supplementalChecks": _check_ratio(supplemental_checks),
                "transportCases": _ratio(
                    sum(case.success for case in mode_cases),
                    len(mode_cases),
                ),
            }
        )
    return summaries


def _build_test_matrix(cases: Sequence[CaseResult]) -> dict[str, object]:
    columns = [
        {
            "key": f"{case.transport}:{case.mode}",
            "transport": case.transport,
            "mode": case.mode,
        }
        for case in cases
    ]
    checks_by_case = {
        f"{case.transport}:{case.mode}": {check.name: check for check in case.checks}
        for case in cases
    }
    observed = {check.name for case in cases for check in case.checks}
    ordered_names = [name for name in CHECK_ORDER if name in observed]
    ordered_names.extend(sorted(observed - set(ordered_names)))
    rows: list[dict[str, object]] = []
    for name in ordered_names:
        results: dict[str, str] = {}
        for column in columns:
            key = str(column["key"])
            check = checks_by_case[key].get(name)
            results[key] = "N/A" if check is None else check.status
        rows.append({"test": name, "results": results})
    return {"columns": columns, "rows": rows}


def _call_tool(
    client: AppServerClient,
    *,
    thread_id: str,
    tool: str,
    arguments: Mapping[str, object],
    meta: Mapping[str, object] | None = None,
) -> tuple[dict[str, object] | None, str]:
    params: dict[str, object] = {
        "threadId": thread_id,
        "server": TEST_SERVER_NAME,
        "tool": tool,
        "arguments": dict(arguments),
    }
    if meta is not None:
        params["_meta"] = dict(meta)
    return _response_result(client.request("mcpServer/tool/call", params))


def _exercise_app_server(
    case: CaseResult,
    client: AppServerClient,
    *,
    transport: str,
    mode: str,
    workspace: Path,
    enable_modern_feature: bool,
) -> None:
    initialize, initialize_detail = _response_result(
        client.request(
            "initialize",
            {
                "clientInfo": {
                    "name": "mcp-spec-compliance-runner",
                    "title": "MCP spec compliance runner",
                    "version": "1.0.0",
                },
                "capabilities": {
                    "experimentalApi": True,
                    "requestAttestation": False,
                    "mcpServerOpenaiFormElicitation": True,
                },
            },
        )
    )
    if not case.check(
        "app_server_initialize",
        initialize is not None,
        initialize_detail,
    ):
        return
    client.notify("initialized")

    if mode == MODERN_VERSION and enable_modern_feature:
        feature_result, feature_detail = _response_result(
            client.request(
                "experimentalFeature/enablement/set",
                {"enablement": {"mcp_2026_07_28": True}},
            )
        )
        enabled = (
            feature_result is not None
            and isinstance(feature_result.get("enablement"), dict)
            and feature_result["enablement"].get("mcp_2026_07_28") is True
        )
        if not case.check(
            "modern_feature_enablement",
            enabled,
            (
                "runtime feature mcp_2026_07_28 enabled"
                if enabled
                else _result_detail(feature_result, feature_detail)
            ),
        ):
            return

    inventory, inventory_detail = _response_result(
        client.request("mcpServerStatus/list", {})
    )
    if inventory is None:
        case.check("inventory", False, inventory_detail)
    else:
        valid, detail = _validate_inventory(inventory, mode=mode)
        case.check("inventory", valid, detail)

    thread_result, thread_detail = _response_result(
        client.request(
            "thread/start",
            {"cwd": str(workspace), "ephemeral": True},
        )
    )
    thread_id: str | None = None
    if thread_result is not None:
        thread_value = thread_result.get("thread")
        if isinstance(thread_value, dict) and isinstance(thread_value.get("id"), str):
            thread_id = str(thread_value["id"])
    if not case.check(
        "ephemeral_thread",
        thread_id is not None,
        "ephemeral thread created" if thread_id is not None else thread_detail,
    ):
        return
    assert thread_id is not None

    sentinel = f"codex-mcp-{transport}-{mode}"
    echo, echo_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="echo",
        arguments={"text": sentinel},
        meta={TEST_META_KEY: sentinel},
    )
    if echo is None:
        case.check("echo_tool", False, echo_detail)
    else:
        valid, detail = _validate_echo(echo, sentinel)
        case.check("echo_tool", valid, detail)

    resource, resource_detail = _response_result(
        client.request(
            "mcpServer/resource/read",
            {
                "threadId": thread_id,
                "server": TEST_SERVER_NAME,
                "uri": RESOURCE_URIS[1],
            },
        )
    )
    expected_resource_text = f"fixture contents for {RESOURCE_URIS[1]}"
    contents = resource.get("contents") if resource is not None else None
    resource_ok = (
        isinstance(contents, list)
        and len(contents) == 1
        and isinstance(contents[0], dict)
        and contents[0].get("text") == expected_resource_text
    )
    case.check(
        "unicode_resource_read",
        resource_ok,
        (
            "read the Unicode resource"
            if resource_ok
            else _result_detail(resource, resource_detail)
        ),
    )

    if mode != MODERN_VERSION:
        return

    metadata, metadata_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="client_metadata",
        arguments={},
        meta={TEST_META_KEY: sentinel},
    )
    structured = metadata.get("structuredContent") if metadata is not None else None
    metadata_ok = (
        isinstance(structured, dict)
        and structured.get("io.modelcontextprotocol/protocolVersion") == MODERN_VERSION
        and isinstance(
            structured.get("io.modelcontextprotocol/clientCapabilities"),
            dict,
        )
        and structured.get(TEST_META_KEY) == sentinel
    )
    case.check(
        "per_request_metadata",
        metadata_ok,
        (
            "required and caller metadata were preserved"
            if metadata_ok
            else _result_detail(metadata, metadata_detail)
        ),
    )

    progress, progress_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="progress",
        arguments={},
        meta={
            "progressToken": "fixture-progress",
            "io.modelcontextprotocol/logLevel": "info",
        },
    )
    progress_content = progress.get("content") if progress is not None else None
    progress_ok = isinstance(progress_content, list) and bool(progress_content)
    case.check(
        "request_scoped_notifications",
        progress_ok,
        (
            "progress and log notifications were consumed before the result"
            if progress_ok
            else _result_detail(progress, progress_detail)
        ),
    )

    elicitations_before = len(client.elicitation_requests)
    mrtr, mrtr_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="request_input",
        arguments={},
    )
    mrtr_structured = mrtr.get("structuredContent") if mrtr is not None else None
    new_elicitations = client.elicitation_requests[elicitations_before:]
    mrtr_ok = (
        isinstance(mrtr_structured, dict)
        and mrtr_structured.get("confirmation") == "confirmed"
        and any(
            item.get("serverName") == TEST_SERVER_NAME and item.get("mode") == "form"
            for item in new_elicitations
        )
    )
    case.check(
        "multi_round_trip_request",
        mrtr_ok,
        (
            "elicitation was surfaced, answered, retried, and completed"
            if mrtr_ok
            else _result_detail(mrtr, mrtr_detail)
        ),
    )

    if transport == "http":
        arguments = {
            "region": "us-west1",
            "attempt": 3,
            "enabled": True,
            "greeting": "Hello, 世界",
        }
        mirrored, mirrored_detail = _call_tool(
            client,
            thread_id=thread_id,
            tool="header_echo",
            arguments=arguments,
        )
        mirrored_structured = (
            mirrored.get("structuredContent") if mirrored is not None else None
        )
        mirrored_ok = mirrored_structured == arguments
        case.check(
            "http_header_mirroring",
            mirrored_ok,
            (
                "method, name, scalar, and Base64 headers matched"
                if mirrored_ok
                else _result_detail(mirrored, mirrored_detail)
            ),
        )


def _run_case(
    codex_binary: Path,
    server_script: Path,
    *,
    transport: str,
    mode: str,
    case_home: Path,
    timeout_seconds: float,
    enable_modern_feature: bool,
) -> CaseResult:
    started_at = time.monotonic()
    case = CaseResult(transport=transport, mode=mode)
    case_home.mkdir(parents=True)
    workspace = case_home / "workspace"
    workspace.mkdir()
    env = _isolated_environment(case_home)
    client: AppServerClient | None = None
    registered = False

    http_context = (
        _running_http_fixture(mode) if transport == "http" else nullcontext(None)
    )
    try:
        with http_context as http_url:
            add = _run_command(
                _registration_command(
                    codex_binary,
                    server_script,
                    transport=transport,
                    mode=mode,
                    http_url=http_url,
                ),
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            registered = case.check(
                "mcp_add",
                add.returncode == 0,
                _command_detail(add),
            )
            if registered:
                get = _run_command(
                    [
                        str(codex_binary),
                        "mcp",
                        "get",
                        TEST_SERVER_NAME,
                        "--json",
                    ],
                    env=env,
                    cwd=workspace,
                    timeout_seconds=timeout_seconds,
                )
                config: dict[str, object] | None = None
                if get.returncode == 0:
                    try:
                        decoded = json.loads(get.stdout)
                        if isinstance(decoded, dict):
                            config = decoded
                    except json.JSONDecodeError:
                        pass
                if config is None:
                    case.check("mcp_get", False, _command_detail(get))
                else:
                    valid, detail = _validate_registration(
                        config,
                        transport=transport,
                        mode=mode,
                        http_url=http_url,
                    )
                    case.check("mcp_get", valid, detail)

                client = AppServerClient(
                    codex_binary,
                    env=env,
                    cwd=workspace,
                    timeout_seconds=timeout_seconds,
                )
                try:
                    _exercise_app_server(
                        case,
                        client,
                        transport=transport,
                        mode=mode,
                        workspace=workspace,
                        enable_modern_feature=enable_modern_feature,
                    )
                except (AppServerError, OSError) as exc:
                    case.check("app_server_protocol", False, str(exc))
                finally:
                    client.close()
                    diagnostic_text = client.diagnostic_text()
                    if diagnostic_text and any(
                        not check.success for check in case.checks
                    ):
                        case.diagnostics = diagnostic_text
    except OSError as exc:
        case.check("case_runtime", False, str(exc))
    finally:
        if client is not None and client.process.poll() is None:
            client.close()
        if registered:
            remove = _run_command(
                [str(codex_binary), "mcp", "remove", TEST_SERVER_NAME],
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            case.check(
                "mcp_remove",
                remove.returncode == 0,
                _command_detail(remove),
            )
            listed = _run_command(
                [str(codex_binary), "mcp", "list", "--json"],
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            empty = False
            if listed.returncode == 0:
                try:
                    empty = json.loads(listed.stdout) == []
                except json.JSONDecodeError:
                    pass
            case.check(
                "isolated_config_cleanup",
                empty,
                "no MCP servers remained" if empty else _command_detail(listed),
            )
        case.finish(started_at)
    return case


def _official_check_result(
    result: OfficialScenarioResult,
    *,
    index: int,
    duplicate: bool,
) -> CheckResult:
    check = result.checks[index]
    if check.status == "SUCCESS":
        status = "PASS"
        success = True
    elif check.status in {"SKIPPED", "INFO"}:
        status = "SKIP"
        success = True
    else:
        status = "FAIL"
        success = False
    suffix = f"{check.name}#{index + 1}" if duplicate else check.name
    name = f"official/{result.scenario}/{suffix}"
    detail = check.error_message or check.description
    return CheckResult(
        name=name,
        success=success,
        detail=detail,
        status=status,
        source="official",
        scenario=result.scenario,
        check_id=check.check_id,
        category="auth" if result.scenario.startswith("auth/") else "non-auth",
    )


def _run_official_case(
    codex_binary: Path,
    adapter_script: Path,
    *,
    conformance_command: Sequence[str],
    mode: str,
    scenarios: Sequence[str],
    case_home: Path,
    timeout_seconds: float,
    enable_modern_feature: bool,
    require_automatic_auth: bool = False,
) -> CaseResult:
    started_at = time.monotonic()
    case = CaseResult(transport="official-http", mode=mode)
    case_home.mkdir(parents=True)
    env = _isolated_environment(case_home)
    env["CODEX_CONFORMANCE_TIMEOUT"] = str(timeout_seconds)
    env["CODEX_CONFORMANCE_ENABLE_MODERN_FEATURE"] = (
        "1" if enable_modern_feature else "0"
    )
    env["CODEX_CONFORMANCE_REQUIRE_AUTOMATIC_AUTH"] = (
        "1" if require_automatic_auth else "0"
    )
    results = run_official_mode(
        conformance_command=conformance_command,
        adapter_script=adapter_script,
        codex_binary=codex_binary,
        mode=mode,
        scenarios=scenarios,
        output_dir=case_home / "official-results",
        timeout_seconds=timeout_seconds,
        base_env=env,
    )

    diagnostics: list[str] = []
    for result in results:
        counts: dict[str, int] = {}
        for check in result.checks:
            counts[check.name] = counts.get(check.name, 0) + 1
        for index, check in enumerate(result.checks):
            case.checks.append(
                _official_check_result(
                    result,
                    index=index,
                    duplicate=counts[check.name] > 1,
                )
            )

        case.checks.append(
            CheckResult(
                name=f"harness/{result.scenario}/codex-adapter",
                success=result.adapter_success,
                detail=result.adapter_detail,
                status="PASS" if result.adapter_success else "FAIL",
                source="harness",
                scenario=result.scenario,
                category=(
                    "auth" if result.scenario.startswith("auth/") else "non-auth"
                ),
            )
        )
        if (
            result.adapter_success
            and not result.success
            and not any(
                check.status in {"FAILURE", "WARNING"} for check in result.checks
            )
        ):
            case.checks.append(
                CheckResult(
                    name=f"harness/{result.scenario}/official-runner",
                    success=False,
                    detail=result.runner_detail or "official runner failed",
                    status="FAIL",
                    source="harness",
                    scenario=result.scenario,
                    category=(
                        "auth" if result.scenario.startswith("auth/") else "non-auth"
                    ),
                )
            )
        if not result.success and result.runner_detail:
            diagnostics.append(f"{result.scenario}:\n{result.runner_detail}")

    if not results:
        case.checks.append(
            CheckResult(
                name="harness/official-scenario-selection",
                success=False,
                detail=f"no official scenarios selected for {mode}",
                status="FAIL",
                source="harness",
            )
        )
    if mode == MODERN_VERSION and "auth/offline-access-scope" in scenarios:
        case.checks.extend(
            _cimd_registration_checks(
                codex_binary,
                adapter_script,
                conformance_command=conformance_command,
                case_home=case_home / "cimd-registration",
                timeout_seconds=timeout_seconds,
                enable_modern_feature=enable_modern_feature,
            )
        )
    if diagnostics:
        case.diagnostics = "\n\n".join(diagnostics)[-16_000:]
    case.finish(started_at)
    return case


def _cimd_registration_checks(
    codex_binary: Path,
    adapter_script: Path,
    *,
    conformance_command: Sequence[str],
    case_home: Path,
    timeout_seconds: float,
    enable_modern_feature: bool,
) -> list[CheckResult]:
    checks: list[CheckResult] = []
    case_home.mkdir(parents=True)
    for check_name, client_registration in (
        ("auto_cimd", None),
        ("forced_cimd", "cimd"),
    ):
        env = _isolated_environment(case_home / check_name)
        env["CODEX_CONFORMANCE_TIMEOUT"] = str(timeout_seconds)
        env["CODEX_CONFORMANCE_ENABLE_MODERN_FEATURE"] = (
            "1" if enable_modern_feature else "0"
        )
        env["CODEX_CONFORMANCE_REQUIRE_AUTOMATIC_AUTH"] = "1"
        if client_registration is not None:
            env["CODEX_CONFORMANCE_CLIENT_REGISTRATION"] = client_registration
        results = run_official_mode(
            conformance_command=conformance_command,
            adapter_script=adapter_script,
            codex_binary=codex_binary,
            mode=MODERN_VERSION,
            scenarios=("auth/offline-access-scope",),
            output_dir=case_home / check_name / "official-results",
            timeout_seconds=timeout_seconds,
            base_env=env,
        )
        result = results[0] if len(results) == 1 else None
        success = result is not None and result.success
        checks.append(
            CheckResult(
                name=f"supplemental/auth/offline-access-scope/{check_name}",
                success=success,
                detail=(
                    result.adapter_detail or result.runner_detail
                    if result is not None
                    else "supplemental offline-access-scope scenario did not run"
                ),
                source="supplemental",
                scenario="auth/offline-access-scope",
                check_id=check_name,
                category="auth",
            )
        )
    return checks


def run_compliance(
    codex_binary: Path,
    *,
    server_script: Path,
    adapter_script: Path,
    conformance_command: Sequence[str],
    official_scenarios: Sequence[str] | None,
    modes: Sequence[str],
    transports: Sequence[str],
    timeout_seconds: float,
    artifact_parent: Path | None,
    keep_artifacts: bool,
    enable_modern_feature: bool = True,
    include_auth: bool = True,
    require_automatic_auth: bool = False,
) -> tuple[dict[str, object], Path | None]:
    started_at = datetime.now(timezone.utc)
    run_root = Path(
        tempfile.mkdtemp(
            prefix="codex-mcp-compliance-",
            dir=artifact_parent,
        )
    )
    version_env = _isolated_environment(run_root)
    version = _run_command(
        [str(codex_binary), "--version"],
        env=version_env,
        cwd=run_root,
        timeout_seconds=timeout_seconds,
    )

    cases: list[CaseResult] = []
    selected_scenarios: dict[str, tuple[str, ...]] = {
        mode: scenarios_for_mode(
            mode,
            official_scenarios,
            include_auth=include_auth,
        )
        for mode in modes
    }

    # Transport order is intentional: supplemental local stdio first, then the
    # official suite's loopback HTTP server.
    for transport in transports:
        for mode in modes:
            if transport == "stdio":
                case_home = run_root / f"stdio-{mode}"
                cases.append(
                    _run_case(
                        codex_binary,
                        server_script,
                        transport="stdio",
                        mode=mode,
                        case_home=case_home,
                        timeout_seconds=timeout_seconds,
                        enable_modern_feature=enable_modern_feature,
                    )
                )
            else:
                if not selected_scenarios[mode]:
                    continue
                case_home = run_root / f"official-http-{mode}"
                cases.append(
                    _run_official_case(
                        codex_binary,
                        adapter_script,
                        conformance_command=conformance_command,
                        mode=mode,
                        scenarios=selected_scenarios[mode],
                        case_home=case_home,
                        timeout_seconds=timeout_seconds,
                        enable_modern_feature=enable_modern_feature,
                        require_automatic_auth=require_automatic_auth,
                    )
                )

    passed = sum(case.success for case in cases)
    finished_at = datetime.now(timezone.utc)
    report: dict[str, object] = {
        "schemaVersion": REPORT_SCHEMA_VERSION,
        "success": version.returncode == 0 and passed == len(cases),
        "startedAt": started_at.isoformat(),
        "finishedAt": finished_at.isoformat(),
        "codexBinary": str(codex_binary),
        "codexVersion": version.stdout or None,
        "modernFeatureEnablement": enable_modern_feature,
        "automaticAuthRequired": require_automatic_auth,
        "officialConformance": {
            "repository": OFFICIAL_CONFORMANCE_REPOSITORY,
            "gitRef": OFFICIAL_CONFORMANCE_GIT_REF,
            "scope": (
                "all versioned client scenarios"
                if include_auth
                else "versioned non-auth client scenarios"
            ),
            "authenticationIncluded": include_auth,
            "scenarios": selected_scenarios,
        },
        "versionCheck": {
            "success": version.returncode == 0,
            "detail": _command_detail(version),
        },
        "summary": {
            "passed": passed,
            "failed": len(cases) - passed,
            "total": len(cases),
        },
        "modeSummaries": _summarize_modes(cases),
        "testMatrix": _build_test_matrix(cases),
        "cases": [asdict(case) for case in cases],
        "artifacts": str(run_root) if keep_artifacts else None,
    }

    retained: Path | None = run_root if keep_artifacts else None
    if not keep_artifacts:
        shutil.rmtree(run_root, ignore_errors=True)
    return report, retained


def _regression_check_identity(
    mode: str,
    transport: str,
    check: Mapping[str, object],
) -> _RegressionCheckIdentity | None:
    source = check.get("source")
    scenario = check.get("scenario")
    name = check.get("name")
    check_id = check.get("check_id")
    if source not in {"official", "harness", "supplemental"}:
        return None
    if not isinstance(name, str) or not name:
        return None
    if scenario is not None and (not isinstance(scenario, str) or not scenario):
        return None
    if check_id is not None and (not isinstance(check_id, str) or not check_id):
        return None

    # The upstream runner numbers repeated request observations by arrival
    # order. Neither that number nor the number of passing retries is a stable
    # protocol assertion, so prefer the upstream check ID when it exists.
    stable_name = name
    stem, separator, suffix = name.rpartition("#")
    if separator and stem and suffix.isdecimal():
        stable_name = stem
    return _RegressionCheckIdentity(
        mode=mode,
        transport=transport,
        source=source,
        scenario=scenario if isinstance(scenario, str) else "",
        check_id=check_id if isinstance(check_id, str) else stable_name,
    )


def _required_regression_checks(
    report: Mapping[str, object],
    *,
    label: str,
    errors: list[str],
) -> dict[_RegressionCheckIdentity, set[str]]:
    if report.get("schemaVersion") != REPORT_SCHEMA_VERSION:
        errors.append(f"{label} does not use report schema {REPORT_SCHEMA_VERSION}")

    version_check = report.get("versionCheck")
    if not isinstance(version_check, dict) or version_check.get("success") is not True:
        errors.append(f"{label} does not contain a successful Codex version check")
    if report.get("modernFeatureEnablement") is not True:
        errors.append(f"{label} did not enable the modern MCP feature")

    official = report.get("officialConformance")
    selected: Mapping[str, object] = {}
    if not isinstance(official, dict):
        errors.append(f"{label} does not contain official conformance metadata")
    else:
        if official.get("repository") != OFFICIAL_CONFORMANCE_REPOSITORY:
            errors.append(f"{label} uses a different upstream conformance repository")
        if official.get("gitRef") != OFFICIAL_CONFORMANCE_GIT_REF:
            errors.append(
                f"{label} uses a different pinned upstream conformance revision"
            )
        if official.get("authenticationIncluded") is not True:
            errors.append(f"{label} does not include the required OAuth scenarios")
        scenarios = official.get("scenarios")
        if isinstance(scenarios, dict):
            selected = scenarios
        else:
            errors.append(f"{label} does not contain a versioned scenario catalog")

    baseline_kind = report.get("baselineKind")
    if baseline_kind is not None:
        if baseline_kind != COMPACT_REGRESSION_BASELINE_KIND:
            errors.append(f"{label} uses an unsupported compact baseline format")
            return {}
        if report.get("requiredModes") != list(REQUIRED_REGRESSION_MODES):
            errors.append(
                f"{label} does not require the shipping, intermediate, and modern MCP versions"
            )
        transports = report.get("transports")
        if (
            not isinstance(transports, list)
            or len(transports) != 2
            or set(transports) != {"stdio", "official-http"}
        ):
            errors.append(f"{label} does not require both stdio and official HTTP")

        raw_identities = report.get("checks")
        if not isinstance(raw_identities, dict):
            errors.append(f"{label} does not contain compact baseline check identities")
            return {}

        identities: dict[_RegressionCheckIdentity, set[str]] = {}
        for bucket, status in (("passing", "PASS"), ("failing", "FAIL")):
            records = raw_identities.get(bucket)
            if not isinstance(records, list):
                errors.append(f"{label} does not contain {bucket} baseline identities")
                continue
            seen: set[_RegressionCheckIdentity] = set()
            for record in records:
                if not isinstance(record, dict):
                    errors.append(
                        f"{label} contains a malformed {bucket} baseline identity"
                    )
                    continue
                mode = record.get("mode")
                transport = record.get("transport")
                source = record.get("source")
                scenario = record.get("scenario")
                check_id = record.get("check_id")
                if (
                    mode not in REQUIRED_REGRESSION_MODES
                    or transport not in {"stdio", "official-http"}
                    or source not in {"official", "harness", "supplemental"}
                    or not isinstance(scenario, str)
                    or not isinstance(check_id, str)
                    or not check_id
                ):
                    errors.append(
                        f"{label} contains a malformed {bucket} baseline identity"
                    )
                    continue
                identity = _RegressionCheckIdentity(
                    mode, transport, source, scenario, check_id
                )
                if identity in seen:
                    errors.append(
                        f"{label} contains a duplicate {bucket} baseline identity"
                    )
                    continue
                seen.add(identity)
                identities.setdefault(identity, set()).add(status)

        for mode in REQUIRED_REGRESSION_MODES:
            expected_scenarios = scenarios_for_mode(mode, include_auth=True)
            actual_scenarios = selected.get(mode)
            if (
                not isinstance(actual_scenarios, (list, tuple))
                or tuple(actual_scenarios) != expected_scenarios
            ):
                errors.append(
                    f"{label} does not run the complete authenticated {mode} scenario catalog"
                )
            for transport in ("stdio", "official-http"):
                if not any(
                    identity.mode == mode and identity.transport == transport
                    for identity in identities
                ):
                    errors.append(
                        f"{label} is missing the required {transport} case for {mode}"
                    )
            observed = {
                identity.scenario
                for identity in identities
                if identity.mode == mode
                and identity.transport == "official-http"
                and identity.source in {"official", "harness"}
            }
            missing = sorted(set(expected_scenarios) - observed)
            unexpected = sorted(observed - set(expected_scenarios))
            if missing:
                errors.append(
                    f"{label} did not observe {mode} scenarios: {', '.join(missing)}"
                )
            if unexpected:
                errors.append(
                    f"{label} observed unexpected {mode} scenarios: {', '.join(unexpected)}"
                )
        return identities

    case_list = report.get("cases")
    cases: dict[tuple[str, str], Mapping[str, object]] = {}
    if not isinstance(case_list, list):
        errors.append(f"{label} does not contain transport cases")
        case_list = []
    for case in case_list:
        if not isinstance(case, dict):
            errors.append(f"{label} contains a malformed transport case")
            continue
        mode = case.get("mode")
        transport = case.get("transport")
        if mode not in REQUIRED_REGRESSION_MODES:
            continue
        if not isinstance(mode, str) or not isinstance(transport, str):
            errors.append(f"{label} contains a malformed required transport case")
            continue
        key = (mode, transport)
        if key in cases:
            errors.append(f"{label} contains a duplicate {transport} case for {mode}")
            continue
        cases[key] = case

    identities: dict[_RegressionCheckIdentity, set[str]] = {}
    for mode in REQUIRED_REGRESSION_MODES:
        expected_scenarios = scenarios_for_mode(mode, include_auth=True)
        actual_scenarios = selected.get(mode)
        if (
            not isinstance(actual_scenarios, (list, tuple))
            or tuple(actual_scenarios) != expected_scenarios
        ):
            errors.append(
                f"{label} does not run the complete authenticated {mode} scenario catalog"
            )

        for transport in ("stdio", "official-http"):
            case = cases.get((mode, transport))
            if case is None:
                errors.append(
                    f"{label} is missing the required {transport} case for {mode}"
                )
                continue
            raw_checks = case.get("checks")
            if not isinstance(raw_checks, list) or not raw_checks:
                errors.append(f"{label} has no {transport} checks for {mode}")
                continue

            observed_scenarios: set[str] = set()
            for check in raw_checks:
                if not isinstance(check, dict):
                    errors.append(
                        f"{label} contains a malformed {transport} check for {mode}"
                    )
                    continue
                identity = _regression_check_identity(mode, transport, check)
                status = check.get("status")
                success = check.get("success")
                if (
                    identity is None
                    or status not in {"PASS", "FAIL", "SKIP"}
                    or not isinstance(success, bool)
                    or success != (status != "FAIL")
                ):
                    errors.append(
                        f"{label} contains a malformed {transport} check for {mode}"
                    )
                    continue
                identities.setdefault(identity, set()).add(status)
                if transport == "official-http" and identity.source in {
                    "official",
                    "harness",
                }:
                    observed_scenarios.add(identity.scenario)

            if transport == "official-http":
                expected = set(expected_scenarios)
                missing = sorted(expected - observed_scenarios)
                unexpected = sorted(observed_scenarios - expected)
                if missing:
                    errors.append(
                        f"{label} did not observe {mode} scenarios: {', '.join(missing)}"
                    )
                if unexpected:
                    errors.append(
                        f"{label} observed unexpected {mode} scenarios: {', '.join(unexpected)}"
                    )

    return identities


def _compact_regression_baseline(report: Mapping[str, object]) -> dict[str, object]:
    errors: list[str] = []
    identities = _required_regression_checks(report, label="baseline", errors=errors)
    automatic_auth = report.get("automaticAuthRequired")
    if not isinstance(automatic_auth, bool):
        errors.append("baseline does not record its production OAuth policy")
    if any(
        identity.mode == SHIPPING_LEGACY_VERSION and "FAIL" in statuses
        for identity, statuses in identities.items()
    ):
        errors.append("baseline contains a shipping MCP regression")
    if errors:
        raise ValueError("; ".join(errors))

    official = report.get("officialConformance")
    assert isinstance(official, dict)
    scenarios = official.get("scenarios")
    assert isinstance(scenarios, dict)
    return {
        "baselineKind": COMPACT_REGRESSION_BASELINE_KIND,
        "schemaVersion": REPORT_SCHEMA_VERSION,
        "requiredModes": list(REQUIRED_REGRESSION_MODES),
        "transports": ["stdio", "official-http"],
        "versionCheck": {"success": True},
        "modernFeatureEnablement": True,
        "automaticAuthRequired": automatic_auth,
        "officialConformance": {
            "repository": OFFICIAL_CONFORMANCE_REPOSITORY,
            "gitRef": OFFICIAL_CONFORMANCE_GIT_REF,
            "authenticationIncluded": True,
            "scenarios": {mode: scenarios[mode] for mode in REQUIRED_REGRESSION_MODES},
        },
        "checks": {
            "passing": [
                asdict(identity)
                for identity, statuses in sorted(identities.items())
                if "PASS" in statuses
            ],
            "failing": [
                asdict(identity)
                for identity, statuses in sorted(identities.items())
                if "FAIL" in statuses
            ],
        },
    }


def _write_json_file(path: Path, value: Mapping[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def _write_compact_regression_baseline(
    report: Mapping[str, object],
    path: Path,
) -> None:
    _write_json_file(path, _compact_regression_baseline(report))


def _evaluate_regression_gate(
    report: Mapping[str, object],
    baseline_report: Mapping[str, object],
    *,
    baseline_path: Path | None = None,
) -> dict[str, object]:
    errors: list[str] = []
    baseline_checks = _required_regression_checks(
        baseline_report, label="baseline", errors=errors
    )
    candidate_checks = _required_regression_checks(
        report, label="candidate", errors=errors
    )

    baseline_auth = baseline_report.get("automaticAuthRequired")
    candidate_auth = report.get("automaticAuthRequired")
    if not isinstance(baseline_auth, bool) or not isinstance(candidate_auth, bool):
        errors.append(
            "baseline and candidate must record their production OAuth policy"
        )
    elif baseline_auth != candidate_auth:
        errors.append("baseline and candidate use different production OAuth policies")

    new_failures: list[dict[str, object]] = []
    missing_checks: list[dict[str, object]] = []
    known_failures: list[dict[str, object]] = []
    fixed_checks: list[dict[str, object]] = []

    for identity, statuses in sorted(candidate_checks.items()):
        baseline_statuses = baseline_checks.get(identity, set())
        if "FAIL" not in statuses:
            continue
        if identity.mode == SHIPPING_LEGACY_VERSION:
            errors.append(f"shipping MCP check failed: {identity.check_id}")
        if "FAIL" in baseline_statuses:
            known_failures.append(asdict(identity))
        else:
            new_failures.append(asdict(identity))

    for identity, baseline_statuses in sorted(baseline_checks.items()):
        candidate_statuses = candidate_checks.get(identity, set())
        if identity.mode == SHIPPING_LEGACY_VERSION and "FAIL" in baseline_statuses:
            errors.append(
                f"baseline shipping MCP check is not passing: {identity.check_id}"
            )
        if "PASS" in baseline_statuses and "PASS" not in candidate_statuses:
            missing_checks.append(asdict(identity))
        elif (
            "FAIL" in baseline_statuses
            and "FAIL" not in candidate_statuses
            and "PASS" not in candidate_statuses
        ):
            missing_checks.append(asdict(identity))
        elif "FAIL" in baseline_statuses and "FAIL" not in candidate_statuses:
            fixed_checks.append(asdict(identity))

    return {
        "success": not errors and not new_failures and not missing_checks,
        "requiredModes": list(REQUIRED_REGRESSION_MODES),
        "baselineReport": str(baseline_path) if baseline_path is not None else None,
        "configurationErrors": errors,
        "newFailures": new_failures,
        "missingChecks": missing_checks,
        "knownFailures": known_failures,
        "fixedChecks": fixed_checks,
    }


def _format_ratio(value: Mapping[str, object]) -> str:
    passed = value.get("passed", 0)
    total = value.get("total", 0)
    if total == 0:
        return "0/0 (N/A)"
    percentage = value.get("percentage", 0)
    return f"{passed}/{total} ({percentage:g}%)"


def _print_table(headers: Sequence[str], rows: Sequence[Sequence[str]]) -> None:
    widths = [
        max(len(headers[index]), *(len(row[index]) for row in rows))
        for index in range(len(headers))
    ]

    def print_row(row: Sequence[str], separator: str = " | ") -> None:
        print(
            separator.join(
                value.ljust(widths[index]) for index, value in enumerate(row)
            )
        )

    print_row(headers)
    print("-+-".join("-" * width for width in widths))
    for row in rows:
        print_row(row)


def _print_human_report(report: Mapping[str, object]) -> None:
    success = report.get("success") is True
    print(f"Codex MCP compliance: {'PASS' if success else 'FAIL'}")
    print(f"Binary: {report.get('codexBinary')}")
    print(f"Version: {report.get('codexVersion') or 'unknown'}")
    failure_details: list[tuple[str, str, str]] = []
    cases = report.get("cases")
    if isinstance(cases, list):
        for case in cases:
            if not isinstance(case, dict):
                continue
            label = f"{case.get('transport')} / {case.get('mode')}"
            print(f"  {'PASS' if case.get('success') else 'FAIL'} {label}")
            checks = case.get("checks")
            if not isinstance(checks, list):
                continue
            for check in checks:
                if isinstance(check, dict) and check.get("success") is not True:
                    failure_details.append(
                        (
                            label,
                            str(check.get("name")),
                            str(check.get("detail")),
                        )
                    )

    mode_summaries = report.get("modeSummaries")
    if isinstance(mode_summaries, list) and mode_summaries:
        print("\nPass rates by protocol mode")
        summary_rows: list[list[str]] = []
        for item in mode_summaries:
            if not isinstance(item, dict):
                continue
            checks = item.get("checks")
            official_checks = item.get("officialChecks")
            official_non_auth_checks = item.get("officialNonAuthChecks")
            official_auth_checks = item.get("officialAuthChecks")
            official_non_auth_scenarios = item.get("officialNonAuthScenarios")
            official_auth_scenarios = item.get("officialAuthScenarios")
            harness_checks = item.get("harnessChecks")
            supplemental_checks = item.get("supplementalChecks")
            transport_cases = item.get("transportCases")
            if not all(
                isinstance(value, dict)
                for value in (
                    checks,
                    official_checks,
                    official_non_auth_checks,
                    official_auth_checks,
                    official_non_auth_scenarios,
                    official_auth_scenarios,
                    harness_checks,
                    supplemental_checks,
                    transport_cases,
                )
            ):
                continue
            summary_rows.append(
                [
                    str(item.get("mode")),
                    _format_ratio(official_non_auth_scenarios),
                    _format_ratio(official_auth_scenarios),
                    _format_ratio(official_checks),
                    _format_ratio(harness_checks),
                    _format_ratio(supplemental_checks),
                    _format_ratio(checks),
                    _format_ratio(transport_cases),
                ]
            )
        _print_table(
            [
                "Mode",
                "Non-auth scenarios",
                "Auth scenarios",
                "Official assertions",
                "Harness",
                "Supplemental",
                "All checks",
                "Transport cases",
            ],
            summary_rows,
        )

    test_matrix = report.get("testMatrix")
    if isinstance(test_matrix, dict):
        columns = test_matrix.get("columns")
        rows = test_matrix.get("rows")
        if isinstance(columns, list) and isinstance(rows, list) and columns:
            matrix_headers = ["Test"]
            column_keys: list[str] = []
            for column in columns:
                if not isinstance(column, dict):
                    continue
                key = str(column.get("key"))
                column_keys.append(key)
                matrix_headers.append(f"{column.get('transport')} {column.get('mode')}")
            matrix_rows: list[list[str]] = []
            for row in rows:
                if not isinstance(row, dict) or not isinstance(
                    row.get("results"), dict
                ):
                    continue
                results = row["results"]
                matrix_rows.append(
                    [
                        str(row.get("test")),
                        *(str(results.get(key, "N/A")) for key in column_keys),
                    ]
                )
            print("\nPer-test results")
            _print_table(matrix_headers, matrix_rows)

    if failure_details:
        print("\nFailure details")
        for label, name, detail in failure_details:
            print(f"  {label} / {name}: {detail}")

    summary = report.get("summary")
    if isinstance(summary, dict):
        print(f"Summary: {summary.get('passed')}/{summary.get('total')} cases passed")
    gate = report.get("regressionGate")
    if isinstance(gate, dict):
        passed = gate.get("success") is True
        known = gate.get("knownFailures")
        known_count = len(known) if isinstance(known, list) else 0
        print(
            f"Regression gate: {'PASS' if passed else 'FAIL'} "
            f"({known_count} acknowledged baseline failures)"
        )
        for category in ("configurationErrors", "newFailures", "missingChecks"):
            values = gate.get(category)
            if isinstance(values, list):
                for value in values:
                    print(f"  {category}: {value}")
    artifacts = report.get("artifacts")
    if isinstance(artifacts, str):
        print(f"Artifacts: {artifacts}")


def _parse_args(argv: Sequence[str] | None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Run the full versioned official MCP client conformance suite against a "
            "supplied Codex binary, plus the supplemental local stdio suite."
        )
    )
    parser.add_argument("codex_binary", type=Path)
    parser.add_argument(
        "--mode",
        choices=("all", SHIPPING_LEGACY_VERSION, LEGACY_VERSION, MODERN_VERSION),
        default="all",
        help="Protocol era to test (default: both).",
    )
    parser.add_argument(
        "--transport",
        choices=("all", "stdio", "http"),
        default="all",
        help=(
            "Transport to test: supplemental stdio, official localhost HTTP, "
            "or both (default: both)."
        ),
    )
    parser.add_argument(
        "--server-script",
        type=Path,
        default=Path(__file__).resolve().with_name("server.py"),
        help="Path to the MCP fixture server.py.",
    )
    parser.add_argument(
        "--adapter-script",
        type=Path,
        default=Path(__file__).resolve().with_name("codex_conformance_adapter.py"),
        help="Path to the Codex adapter used by the official client suite.",
    )
    parser.add_argument(
        "--conformance-cli",
        type=Path,
        help=(
            "Use an existing official conformance executable. By default the "
            "runner bootstraps the pinned upstream Git commit with npx."
        ),
    )
    parser.add_argument(
        "--official-scenario",
        action="append",
        help=(
            "Run only this official versioned scenario. Repeat to select more "
            "(default: every applicable scenario for the selected mode)."
        ),
    )
    parser.add_argument(
        "--auth",
        action=argparse.BooleanOptionalAction,
        default=True,
        help=(
            "Include official OAuth authorization scenarios in HTTP coverage "
            "(default: enabled). Use --no-auth for a faster non-auth slice."
        ),
    )
    parser.add_argument(
        "--require-automatic-auth",
        action="store_true",
        help=(
            "Require Codex to recover from OAuth scope escalation and "
            "authorization-server migration without harness-injected re-login; "
            "also exercise production client-metadata selection."
        ),
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=45.0,
        help="Per-command and per-request timeout in seconds.",
    )
    parser.add_argument(
        "--report",
        type=Path,
        help="Also write the complete JSON report to this path.",
    )
    parser.add_argument(
        "--baseline-report",
        type=Path,
        help=(
            "Require the complete shipping, intermediate, and modern authenticated "
            "HTTP and stdio matrices to preserve every passing check from this report. "
            "Known baseline failures remain visible and new failures fail the gate."
        ),
    )
    parser.add_argument(
        "--write-baseline",
        type=Path,
        help="Write a compact, deterministic regression baseline from the completed run.",
    )
    parser.add_argument(
        "--extract-baseline",
        type=Path,
        help=(
            "Write a compact, deterministic baseline from --baseline-report "
            "and exit without rerunning conformance."
        ),
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Print the report as JSON instead of the human summary.",
    )
    parser.add_argument(
        "--artifact-parent",
        type=Path,
        help="Parent directory for isolated temporary Codex homes.",
    )
    parser.add_argument(
        "--keep-artifacts",
        action="store_true",
        help="Keep isolated Codex homes and include their path in the report.",
    )
    parser.add_argument(
        "--enable-modern-feature",
        action=argparse.BooleanOptionalAction,
        default=True,
        help=(
            "Configure mcp_2026_07_28 before MCP startup and verify it through "
            "the app-server runtime setter for modern cases (default: enabled)."
        ),
    )
    return parser.parse_args(argv)


def main(argv: Sequence[str] | None = None) -> int:
    args = _parse_args(argv)
    codex_binary = args.codex_binary.expanduser().resolve()
    server_script = args.server_script.expanduser().resolve()
    adapter_script = args.adapter_script.expanduser().resolve()
    modes = (
        (SHIPPING_LEGACY_VERSION, LEGACY_VERSION, MODERN_VERSION)
        if args.mode == "all"
        else (args.mode,)
    )
    transports = ("stdio", "http") if args.transport == "all" else (args.transport,)
    if not codex_binary.is_file() or not os.access(codex_binary, os.X_OK):
        print(f"error: Codex binary is not executable: {codex_binary}", file=sys.stderr)
        return 2
    if "stdio" in transports and not server_script.is_file():
        print(f"error: fixture server does not exist: {server_script}", file=sys.stderr)
        return 2
    if "http" in transports and not adapter_script.is_file():
        print(
            f"error: official conformance adapter does not exist: {adapter_script}",
            file=sys.stderr,
        )
        return 2
    if args.timeout <= 0:
        print("error: --timeout must be positive", file=sys.stderr)
        return 2
    if args.extract_baseline is not None and args.baseline_report is None:
        print("error: --extract-baseline requires --baseline-report", file=sys.stderr)
        return 2
    if args.extract_baseline is not None and args.write_baseline is not None:
        print(
            "error: --extract-baseline cannot be combined with --write-baseline",
            file=sys.stderr,
        )
        return 2

    baseline_report: dict[str, object] | None = None
    baseline_path: Path | None = None
    if args.baseline_report is not None:
        baseline_path = args.baseline_report.expanduser().resolve()
        try:
            loaded_baseline = json.loads(baseline_path.read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            print(
                f"error: cannot read baseline report {baseline_path}: {exc}",
                file=sys.stderr,
            )
            return 2
        if not isinstance(loaded_baseline, dict):
            print(
                f"error: baseline report must be a JSON object: {baseline_path}",
                file=sys.stderr,
            )
            return 2
        baseline_report = loaded_baseline

    if args.extract_baseline is not None:
        assert baseline_report is not None
        output_path = args.extract_baseline.expanduser().resolve()
        try:
            _write_compact_regression_baseline(baseline_report, output_path)
        except (OSError, ValueError) as exc:
            print(
                f"error: cannot extract regression baseline {output_path}: {exc}",
                file=sys.stderr,
            )
            return 2
        print(f"Regression baseline: {output_path}")
        return 0

    conformance_command: list[str]
    if args.conformance_cli is None:
        conformance_command = default_conformance_command()
    else:
        conformance_cli = args.conformance_cli.expanduser().resolve()
        if not conformance_cli.is_file():
            print(
                f"error: official conformance CLI does not exist: {conformance_cli}",
                file=sys.stderr,
            )
            return 2
        if os.access(conformance_cli, os.X_OK):
            conformance_command = [str(conformance_cli)]
        elif conformance_cli.suffix in {".js", ".mjs"} and shutil.which("node"):
            conformance_command = [str(shutil.which("node")), str(conformance_cli)]
        else:
            print(
                "error: --conformance-cli must be executable (or a JavaScript "
                f"file runnable with node): {conformance_cli}",
                file=sys.stderr,
            )
            return 2

    try:
        for mode in modes:
            scenarios_for_mode(
                mode,
                args.official_scenario,
                include_auth=args.auth,
            )
    except ValueError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2

    artifact_parent: Path | None = None
    if args.artifact_parent is not None:
        artifact_parent = args.artifact_parent.expanduser().resolve()
        artifact_parent.mkdir(parents=True, exist_ok=True)

    report, _ = run_compliance(
        codex_binary,
        server_script=server_script,
        adapter_script=adapter_script,
        conformance_command=conformance_command,
        official_scenarios=args.official_scenario,
        modes=modes,
        transports=transports,
        timeout_seconds=args.timeout,
        artifact_parent=artifact_parent,
        keep_artifacts=args.keep_artifacts,
        enable_modern_feature=args.enable_modern_feature,
        include_auth=args.auth,
        require_automatic_auth=args.require_automatic_auth,
    )
    if baseline_report is not None:
        report["regressionGate"] = _evaluate_regression_gate(
            report,
            baseline_report,
            baseline_path=baseline_path,
        )
    if args.write_baseline is not None:
        output_path = args.write_baseline.expanduser().resolve()
        try:
            _write_compact_regression_baseline(report, output_path)
        except (OSError, ValueError) as exc:
            print(
                f"error: cannot write regression baseline {output_path}: {exc}",
                file=sys.stderr,
            )
            return 2

    if args.report is not None:
        report_path = args.report.expanduser().resolve()
        _write_json_file(report_path, report)
    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
    else:
        _print_human_report(report)
        if args.report is not None:
            print(f"JSON report: {args.report.expanduser().resolve()}")
    if baseline_report is not None:
        gate = report.get("regressionGate")
        return 0 if isinstance(gate, dict) and gate.get("success") is True else 1
    return 0 if report["success"] else 1


if __name__ == "__main__":
    raise SystemExit(main())