codex/scripts/mcp_conformance/review_regressions.py

#!/usr/bin/env python3
"""Run reviewer-derived MCP regressions against a real Codex app-server."""

import argparse
import json
import os
import shutil
import sys
import tempfile
import threading
import time
from contextlib import contextmanager, nullcontext
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import Iterator, Mapping, Sequence

_MODULE_DIR = Path(__file__).resolve().parent
if str(_MODULE_DIR) not in sys.path:
    sys.path.insert(0, str(_MODULE_DIR))

from run_codex_compliance import (  # noqa: E402 - direct scripts must first add their sibling directory.
    LEGACY_VERSION,
    MISMATCHED_DISCOVERY_ID_PROFILE,
    MODERN_VERSION,
    NULL_DISCOVERY_ID_PROFILE,
    REPEATED_CURSOR_PROFILE,
    REVIEW_EXACT_INTEGER,
    REVIEW_MRTR_INPUT_REQUEST_COUNT,
    REVIEW_PROFILE,
    SHIPPING_LEGACY_VERSION,
    SSE_COMMENT_FLOOD_PROFILE,
    SSE_CR_COMMENTS_PROFILE,
    TEST_SERVER_NAME,
    AppServerClient,
    AppServerError,
    CaseResult,
    ProtocolServer,
    _call_tool,
    _command_detail,
    _isolated_environment,
    _response_result,
    _run_command,
    make_http_server,
)
from server import (  # noqa: E402 - direct scripts must first add their sibling directory.
    CATALOG_MAX_PROFILE,
    CATALOG_OVER_LIMIT_PROFILE,
    MAX_CATALOG_ITEMS,
)

REVIEW_MODES = (SHIPPING_LEGACY_VERSION, LEGACY_VERSION, MODERN_VERSION)
REVIEW_REPORT_SCHEMA_VERSION = 1
REVIEW_BASELINE_KIND = "mcp-review-regression-baseline-v1"
REVIEWER = "codex-mcp-regression-suite"
LEGACY_ENVIRONMENT_SENTINEL = "review-legacy-protocol-environment"
MRTR_REQUEST_LIMIT = 64
CATALOG_BOUNDARY_PROFILES = (CATALOG_MAX_PROFILE, CATALOG_OVER_LIMIT_PROFILE)
CATALOG_BOUNDARY_TRANSPORTS = ("stdio", "http")
CATALOG_LIMIT_ERROR = (
    f"tools/list exceeded the catalog limit of {MAX_CATALOG_ITEMS} items"
)


@dataclass(frozen=True, order=True)
class _ReviewCheckIdentity:
    mode: str
    transport: str
    check_id: str


def _required_review_cases() -> set[tuple[str, str]]:
    cases = {(mode, f"review-stdio:{REVIEW_PROFILE}") for mode in REVIEW_MODES}
    cases.update(
        (MODERN_VERSION, f"review-http:{profile}")
        for profile in (
            REVIEW_PROFILE,
            REPEATED_CURSOR_PROFILE,
            MISMATCHED_DISCOVERY_ID_PROFILE,
            NULL_DISCOVERY_ID_PROFILE,
            SSE_CR_COMMENTS_PROFILE,
            SSE_COMMENT_FLOOD_PROFILE,
        )
    )
    cases.update(
        (mode, f"review-{transport}:{profile}")
        for mode in REVIEW_MODES
        for transport in CATALOG_BOUNDARY_TRANSPORTS
        for profile in CATALOG_BOUNDARY_PROFILES
    )
    return cases


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


def _review_registration_command(
    codex_binary: Path,
    server_script: Path,
    *,
    transport: str,
    mode: str,
    profile: str,
    http_url: str | None,
) -> list[str]:
    command = [str(codex_binary), "mcp", "add", TEST_SERVER_NAME]
    if transport == "stdio":
        protocol_environment = (
            MODERN_VERSION if mode == MODERN_VERSION else LEGACY_ENVIRONMENT_SENTINEL
        )
        command.extend(
            [
                "--env",
                f"CODEX_MCP_PROTOCOL_VERSION={protocol_environment}",
                "--",
                sys.executable,
                str(server_script),
                "--mode",
                mode,
                "--transport",
                "stdio",
                "--profile",
                profile,
            ]
        )
    else:
        if http_url is None:
            raise ValueError("review HTTP transport requires a fixture URL")
        command.extend(["--url", http_url])
    return command


def _review_inventory_entry(result: Mapping[str, object]) -> dict[str, object] | None:
    entries = result.get("data")
    if not isinstance(entries, list):
        return None
    for entry in entries:
        if isinstance(entry, dict) and entry.get("name") == TEST_SERVER_NAME:
            return entry
    return None


def _tool_input_schema(
    entry: Mapping[str, object],
    tool_name: str,
) -> Mapping[str, object] | None:
    tools = entry.get("tools")
    if not isinstance(tools, dict):
        return None
    tool = tools.get(tool_name)
    if not isinstance(tool, dict):
        return None
    schema = tool.get("inputSchema")
    if not isinstance(schema, dict):
        schema = tool.get("input_schema")
    return schema if isinstance(schema, dict) else None


def _exact_integer_property(schema: Mapping[str, object] | None) -> bool:
    if schema is None:
        return False
    properties = schema.get("properties")
    if not isinstance(properties, dict):
        return False
    value = properties.get("value")
    return (
        isinstance(value, dict)
        and value.get("type") == "integer"
        and isinstance(value.get("minimum"), int)
        and not isinstance(value.get("minimum"), bool)
        and value.get("minimum") == REVIEW_EXACT_INTEGER
        and isinstance(value.get("default"), int)
        and not isinstance(value.get("default"), bool)
        and value.get("default") == REVIEW_EXACT_INTEGER
    )


def _elicitation_schema(request: Mapping[str, object]) -> Mapping[str, object] | None:
    for key in ("requestedSchema", "requested_schema", "schema"):
        schema = request.get(key)
        if isinstance(schema, dict):
            return schema
    for key in ("request", "elicitation", "params"):
        nested = request.get(key)
        if isinstance(nested, dict):
            found = _elicitation_schema(nested)
            if found is not None:
                return found
    return None


def _review_elicitation_content(params: Mapping[str, object]) -> Mapping[str, object]:
    schema = _elicitation_schema(params)
    if _exact_integer_property(schema):
        return {"value": REVIEW_EXACT_INTEGER}
    return {"value": "review-confirmed", "confirmation": "confirmed"}


def _mrtr_budget_is_bounded(
    request_count: int,
    *,
    elapsed_seconds: float,
    timeout_seconds: float,
) -> bool:
    return (
        0 <= request_count <= MRTR_REQUEST_LIMIT
        and elapsed_seconds <= timeout_seconds + 1
    )


def _initialize_review_client(
    case: CaseResult,
    client: AppServerClient,
    *,
    mode: str,
) -> bool:
    response, detail = _response_result(
        client.request(
            "initialize",
            {
                "clientInfo": {
                    "name": "mcp-review-regression-runner",
                    "title": "MCP review regression runner",
                    "version": "1.0.0",
                },
                "capabilities": {
                    "experimentalApi": True,
                    "requestAttestation": False,
                    "mcpServerOpenaiFormElicitation": True,
                },
            },
        )
    )
    if not case.check("review/app-server-initialize", response is not None, detail):
        return False
    client.notify("initialized")
    if mode != MODERN_VERSION:
        return True

    feature, feature_detail = _response_result(
        client.request(
            "experimentalFeature/enablement/set",
            {"enablement": {"mcp_2026_07_28": True}},
        )
    )
    enabled = (
        feature is not None
        and isinstance(feature.get("enablement"), dict)
        and feature["enablement"].get("mcp_2026_07_28") is True
    )
    return case.check(
        "review/modern-feature-enablement",
        enabled,
        "enabled the production modern MCP feature" if enabled else feature_detail,
    )


def _review_thread_id(
    case: CaseResult,
    client: AppServerClient,
    workspace: Path,
) -> str | None:
    response, detail = _response_result(
        client.request("thread/start", {"cwd": str(workspace), "ephemeral": True})
    )
    thread = response.get("thread") if response is not None else None
    thread_id = thread.get("id") if isinstance(thread, dict) else None
    if not case.check(
        "review/ephemeral-thread",
        isinstance(thread_id, str),
        "created isolated review thread" if isinstance(thread_id, str) else detail,
    ):
        return None
    return str(thread_id)


def _run_normal_review_checks(
    case: CaseResult,
    client: AppServerClient,
    *,
    mode: str,
    workspace: Path,
) -> None:
    inventory, inventory_detail = _response_result(
        client.request("mcpServerStatus/list", {"detail": "full"})
    )
    entry = _review_inventory_entry(inventory) if inventory is not None else None
    if not case.check(
        "review/server-discovery",
        entry is not None,
        "discovered the adversarial review fixture"
        if entry is not None
        else inventory_detail,
    ):
        return
    assert entry is not None

    tools = entry.get("tools")
    if not isinstance(tools, dict):
        tools = {}
    if mode == MODERN_VERSION:
        required_page_two = {
            "review_integer_elicitation",
            "review_large_integer",
            "review_mrtr_cap",
            "review_protocol_env",
        }
        missing = sorted(required_page_two - set(tools))
        case.check(
            "review/modern-paginated-tools-page-two",
            not missing,
            "all protected second-page tools reached the app-server inventory"
            if not missing
            else "second-page tools missing: " + ", ".join(missing),
        )

    schema = _tool_input_schema(entry, "review_large_integer")
    exact_schema = _exact_integer_property(schema)
    case.check(
        "review/exact-large-integer-tool-schema",
        exact_schema,
        "preserved integer minimum and default 9007199254740993"
        if exact_schema
        else "the app-server rounded or omitted the 2^53+1 tool schema",
    )

    thread_id = _review_thread_id(case, client, workspace)
    if thread_id is None:
        return

    integer, integer_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="review_large_integer",
        arguments={"value": REVIEW_EXACT_INTEGER},
    )
    content = integer.get("structuredContent") if integer is not None else None
    exact_value = (
        isinstance(content, dict)
        and isinstance(content.get("value"), int)
        and not isinstance(content.get("value"), bool)
        and content.get("value") == REVIEW_EXACT_INTEGER
    )
    case.check(
        "review/exact-large-integer-tool-round-trip",
        exact_value,
        "round-tripped 9007199254740993 without floating-point conversion"
        if exact_value
        else integer_detail,
    )

    if mode != MODERN_VERSION:
        environment, environment_detail = _call_tool(
            client,
            thread_id=thread_id,
            tool="review_protocol_env",
            arguments={},
        )
        observed = (
            environment.get("structuredContent") if environment is not None else None
        )
        preserved = (
            isinstance(observed, dict)
            and observed.get("value") == LEGACY_ENVIRONMENT_SENTINEL
        )
        case.check(
            "review/legacy-reserved-stdio-environment-preserved",
            preserved,
            "forwarded the explicitly configured legacy protocol environment unchanged"
            if preserved
            else environment_detail,
        )
        return

    before = len(client.elicitation_requests)
    elicitation, elicitation_detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="review_integer_elicitation",
        arguments={},
    )
    requests = client.elicitation_requests[before:]
    exact_elicitation = any(
        _exact_integer_property(_elicitation_schema(request)) for request in requests
    )
    case.check(
        "review/exact-large-integer-elicitation-schema",
        exact_elicitation,
        "app-server preserved 9007199254740993 in the actual elicitation schema"
        if exact_elicitation
        else "the production elicitation schema rounded or omitted the 2^53+1 integer",
    )
    elicitation_content = (
        elicitation.get("structuredContent") if elicitation is not None else None
    )
    completed = (
        isinstance(elicitation_content, dict)
        and elicitation_content.get("value") == REVIEW_EXACT_INTEGER
    )
    case.check(
        "review/exact-large-integer-elicitation-round-trip",
        completed,
        "completed elicitation using the exact 2^53+1 integer"
        if completed
        else elicitation_detail,
    )

    mrtr_before = len(client.elicitation_requests)
    started = time.monotonic()
    mrtr_result = None
    try:
        mrtr_result, mrtr_detail = _call_tool(
            client,
            thread_id=thread_id,
            tool="review_mrtr_cap",
            arguments={},
        )
    except AppServerError as exc:
        mrtr_detail = str(exc)
    elapsed = time.monotonic() - started
    observed = len(client.elicitation_requests) - mrtr_before
    bounded = mrtr_result is None and _mrtr_budget_is_bounded(
        observed,
        elapsed_seconds=elapsed,
        timeout_seconds=client.timeout_seconds,
    )
    case.check(
        "review/mrtr-input-requests-bounded-to-64",
        bounded,
        (
            f"rejected {REVIEW_MRTR_INPUT_REQUEST_COUNT} simultaneous input requests; "
            f"observed {observed} elicitations in {elapsed:.2f}s"
            if bounded
            else (
                f"observed {observed} elicitation requests for a "
                f"{REVIEW_MRTR_INPUT_REQUEST_COUNT}-request response in "
                f"{elapsed:.2f}s; {mrtr_detail}"
            )
        ),
    )


def _run_malformed_discovery_check(
    case: CaseResult,
    client: AppServerClient,
    *,
    profile: str,
) -> None:
    name = (
        "review/reject-null-discovery-response-id"
        if profile == NULL_DISCOVERY_ID_PROFILE
        else "review/reject-mismatched-discovery-response-id"
    )
    try:
        inventory, detail = _response_result(
            client.request("mcpServerStatus/list", {"detail": "full"})
        )
    except AppServerError as exc:
        case.check(name, True, f"rejected malformed discovery response: {exc}")
        return
    entry = _review_inventory_entry(inventory) if inventory is not None else None
    tools = entry.get("tools") if entry is not None else None
    rejected = (
        inventory is None or entry is None or not isinstance(tools, dict) or not tools
    )
    case.check(
        name,
        rejected,
        "rejected the uncorrelated discovery response without silently downgrading"
        if rejected
        else "accepted an uncorrelated discovery response and exposed server tools",
    )


def _run_repeated_cursor_check(case: CaseResult, client: AppServerClient) -> None:
    started = time.monotonic()
    try:
        inventory, _ = _response_result(
            client.request("mcpServerStatus/list", {"detail": "full"})
        )
        entry = _review_inventory_entry(inventory) if inventory is not None else None
        tools = entry.get("tools") if entry is not None else None
        rejected = (
            inventory is None
            or entry is None
            or not isinstance(tools, dict)
            or not tools
        )
        detail = (
            "rejected the repeated pagination cursor within the startup deadline"
            if rejected
            else "accepted or exposed a catalog with a repeated pagination cursor"
        )
    except AppServerError as exc:
        rejected = True
        detail = f"terminated repeated-cursor pagination: {exc}"
    elapsed = time.monotonic() - started
    case.check(
        "review/repeated-pagination-cursor-bounded",
        rejected and elapsed <= client.timeout_seconds + 2,
        f"{detail}; elapsed={elapsed:.2f}s",
    )


def _run_catalog_boundary_checks(
    case: CaseResult,
    client: AppServerClient,
    *,
    profile: str,
    workspace: Path,
) -> None:
    if profile not in CATALOG_BOUNDARY_PROFILES:
        raise ValueError(f"unknown catalog boundary profile: {profile}")

    at_limit = profile == CATALOG_MAX_PROFILE
    check_prefix = (
        "review/catalog-at-limit" if at_limit else "review/catalog-over-limit"
    )
    expected_status = "ready" if at_limit else "failed"
    event_index = len(client.events)
    if _review_thread_id(case, client, workspace) is None:
        return

    try:
        event = client.wait_for_notification(
            "mcpServer/startupStatus/updated",
            predicate=lambda params: (
                params.get("name") == TEST_SERVER_NAME
                and params.get("status") in {"ready", "failed"}
            ),
            after_event_index=event_index,
        )
    except AppServerError as exc:
        case.check(
            f"{check_prefix}/startup-{expected_status}",
            False,
            f"did not observe the required {expected_status} startup: {exc}",
        )
        return

    params = event.get("params")
    case.check(
        f"{check_prefix}/startup-{expected_status}",
        isinstance(params, dict) and params.get("status") == expected_status,
        f"observed the expected {expected_status} MCP server startup"
        if isinstance(params, dict) and params.get("status") == expected_status
        else (
            "observed MCP server startup "
            f"{params.get('status')!r}; expected {expected_status!r}"
            if isinstance(params, dict)
            else "MCP server startup notification did not contain parameters"
        ),
    )
    if not isinstance(params, dict):
        return

    if not at_limit:
        error = params.get("error")
        case.check(
            f"{check_prefix}/exact-limit-error",
            isinstance(error, str) and CATALOG_LIMIT_ERROR in error,
            error
            if isinstance(error, str)
            else f"startup failure did not report {CATALOG_LIMIT_ERROR!r}",
        )

    inventory, detail = _response_result(
        client.request("mcpServerStatus/list", {"detail": "full"})
    )
    entry = _review_inventory_entry(inventory) if inventory is not None else None
    if not case.check(
        f"{check_prefix}/configured-server",
        entry is not None,
        "retained the configured catalog-boundary server"
        if entry is not None
        else detail,
    ):
        return
    assert entry is not None

    server_info = entry.get("serverInfo")
    expected_server_info = (
        isinstance(server_info, dict) if at_limit else server_info is None
    )
    case.check(
        f"{check_prefix}/server-info",
        expected_server_info,
        "exposed initialized server information"
        if at_limit and expected_server_info
        else "left rejected server information null"
        if not at_limit and expected_server_info
        else "catalog server information did not match its startup status",
    )

    tools = entry.get("tools")
    actual_count = len(tools) if isinstance(tools, dict) else None
    expected_count = MAX_CATALOG_ITEMS if at_limit else 0
    case.check(
        f"{check_prefix}/tool-count",
        isinstance(tools, dict) and actual_count == expected_count,
        f"discovered exactly {expected_count} tools"
        if actual_count == expected_count
        else f"expected {expected_count} discovered tools, observed {actual_count}",
    )


def _run_sse_check(
    case: CaseResult,
    client: AppServerClient,
    workspace: Path,
    *,
    profile: str,
) -> None:
    inventory, detail = _response_result(
        client.request("mcpServerStatus/list", {"detail": "full"})
    )
    entry = _review_inventory_entry(inventory) if inventory is not None else None
    if not case.check(
        "review/sse-cr-discovery",
        entry is not None,
        "discovered the CR-only SSE fixture" if entry is not None else detail,
    ):
        return
    thread_id = _review_thread_id(case, client, workspace)
    if thread_id is None:
        return
    result, detail = _call_tool(
        client,
        thread_id=thread_id,
        tool="progress",
        arguments={},
        meta={"progressToken": "review-cr-only-sse"},
    )
    content = result.get("content") if result is not None else None
    accepted = isinstance(content, list) and bool(content)
    flood = profile == SSE_COMMENT_FLOOD_PROFILE
    case.check(
        "review/sse-comment-flood-excluded-from-event-size-limit"
        if flood
        else "review/sse-carriage-return-and-comment-framing",
        accepted,
        "accepted more than 8 MiB of valid SSE comment keepalives"
        if accepted and flood
        else "accepted valid CR-only SSE lines and ignored comment keepalives"
        if accepted
        else detail,
    )


def _run_review_case(
    codex_binary: Path,
    server_script: Path,
    *,
    mode: str,
    profile: str,
    transport: str,
    case_home: Path,
    timeout_seconds: float,
) -> CaseResult:
    started = time.monotonic()
    case = CaseResult(transport=f"review-{transport}:{profile}", mode=mode)
    case_home.mkdir(parents=True, exist_ok=True)
    workspace = case_home / "workspace"
    workspace.mkdir(exist_ok=True)
    env = _isolated_environment(case_home)
    registered = False
    fixture = (
        _running_review_http_fixture(mode, profile)
        if transport == "http"
        else nullcontext(None)
    )

    try:
        if mode == MODERN_VERSION:
            feature = _run_command(
                [str(codex_binary), "features", "enable", "mcp_2026_07_28"],
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            if not case.check(
                "review/modern-feature-configuration",
                feature.returncode == 0,
                _command_detail(feature),
            ):
                return case

        with fixture as http_url:
            registration = _run_command(
                _review_registration_command(
                    codex_binary,
                    server_script,
                    transport=transport,
                    mode=mode,
                    profile=profile,
                    http_url=http_url,
                ),
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            registered = case.check(
                "review/mcp-registration",
                registration.returncode == 0,
                _command_detail(registration),
            )
            if not registered:
                return case

            with AppServerClient(
                codex_binary,
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
                elicitation_content=_review_elicitation_content,
            ) as client:
                if not _initialize_review_client(case, client, mode=mode):
                    return case
                if profile in (
                    MISMATCHED_DISCOVERY_ID_PROFILE,
                    NULL_DISCOVERY_ID_PROFILE,
                ):
                    _run_malformed_discovery_check(case, client, profile=profile)
                elif profile == REPEATED_CURSOR_PROFILE:
                    _run_repeated_cursor_check(case, client)
                elif profile in CATALOG_BOUNDARY_PROFILES:
                    _run_catalog_boundary_checks(
                        case, client, profile=profile, workspace=workspace
                    )
                elif profile in (SSE_CR_COMMENTS_PROFILE, SSE_COMMENT_FLOOD_PROFILE):
                    _run_sse_check(case, client, workspace, profile=profile)
                else:
                    _run_normal_review_checks(
                        case, client, mode=mode, workspace=workspace
                    )
    except (AppServerError, OSError, ValueError) as exc:
        case.check("review/client-runtime", False, str(exc))
    finally:
        if registered:
            removal = _run_command(
                [str(codex_binary), "mcp", "remove", TEST_SERVER_NAME],
                env=env,
                cwd=workspace,
                timeout_seconds=timeout_seconds,
            )
            case.check(
                "review/isolated-registration-cleanup",
                removal.returncode == 0,
                _command_detail(removal),
            )
        case.finish(started)
    return case


def run_review_regressions(
    codex_binary: Path,
    *,
    modes: Sequence[str] = REVIEW_MODES,
    server_script: Path | None = None,
    timeout_seconds: float = 8,
    artifact_parent: Path | None = None,
    keep_artifacts: bool = False,
) -> dict[str, object]:
    server_script = server_script or _MODULE_DIR / "server.py"
    root = Path(tempfile.mkdtemp(prefix="codex-mcp-review-", dir=artifact_parent))
    cases: list[CaseResult] = []
    try:
        for mode in modes:
            cases.append(
                _run_review_case(
                    codex_binary,
                    server_script,
                    mode=mode,
                    profile=REVIEW_PROFILE,
                    transport="stdio",
                    case_home=root / f"review-stdio-{mode}",
                    timeout_seconds=timeout_seconds,
                )
            )
            if mode != MODERN_VERSION:
                continue
            for profile in (
                REVIEW_PROFILE,
                REPEATED_CURSOR_PROFILE,
                MISMATCHED_DISCOVERY_ID_PROFILE,
                NULL_DISCOVERY_ID_PROFILE,
                SSE_CR_COMMENTS_PROFILE,
                SSE_COMMENT_FLOOD_PROFILE,
            ):
                cases.append(
                    _run_review_case(
                        codex_binary,
                        server_script,
                        mode=mode,
                        profile=profile,
                        transport="http",
                        case_home=root / f"review-http-{mode}-{profile}",
                        timeout_seconds=timeout_seconds,
                    )
                )

        for mode in modes:
            for transport in CATALOG_BOUNDARY_TRANSPORTS:
                for profile in CATALOG_BOUNDARY_PROFILES:
                    cases.append(
                        _run_review_case(
                            codex_binary,
                            server_script,
                            mode=mode,
                            profile=profile,
                            transport=transport,
                            case_home=root / f"review-{transport}-{mode}-{profile}",
                            timeout_seconds=timeout_seconds,
                        )
                    )

        checks = [check for case in cases for check in case.checks]
        failures = [check for check in checks if not check.success]
        return {
            "schemaVersion": REVIEW_REPORT_SCHEMA_VERSION,
            "success": bool(cases) and all(case.success for case in cases),
            "codexBinary": str(codex_binary),
            "reviewer": REVIEWER,
            "modes": list(modes),
            "summary": {
                "passed": len(checks) - len(failures),
                "failed": len(failures),
                "total": len(checks),
                "casesPassed": sum(case.success for case in cases),
                "casesTotal": len(cases),
            },
            "cases": [asdict(case) for case in cases],
            "artifacts": str(root) if keep_artifacts else None,
        }
    finally:
        if not keep_artifacts:
            shutil.rmtree(root, ignore_errors=True)


def _required_review_checks(
    report: Mapping[str, object],
    *,
    label: str,
    errors: list[str],
) -> dict[_ReviewCheckIdentity, bool]:
    if report.get("schemaVersion") != REVIEW_REPORT_SCHEMA_VERSION:
        errors.append(
            f"{label} does not use reviewer report schema "
            f"{REVIEW_REPORT_SCHEMA_VERSION}"
        )
    if report.get("reviewer") != REVIEWER:
        errors.append(f"{label} does not identify the production reviewer probes")

    compact = report.get("baselineKind") is not None
    modes = report.get("requiredModes" if compact else "modes")
    if modes != list(REVIEW_MODES):
        errors.append(f"{label} does not run all three required MCP protocol modes")

    expected_cases = _required_review_cases()
    observed_cases: set[tuple[str, str]] = set()
    identities: dict[_ReviewCheckIdentity, bool] = {}

    if compact:
        if report.get("baselineKind") != REVIEW_BASELINE_KIND:
            errors.append(f"{label} uses an unsupported reviewer baseline format")

        raw_cases = report.get("requiredCases")
        if not isinstance(raw_cases, list):
            errors.append(f"{label} does not contain required reviewer cases")
            raw_cases = []
        for case in raw_cases:
            if not isinstance(case, dict):
                errors.append(f"{label} contains a malformed reviewer case")
                continue
            mode = case.get("mode")
            transport = case.get("transport")
            if not isinstance(mode, str) or not isinstance(transport, str):
                errors.append(f"{label} contains a malformed reviewer case")
                continue
            identity = (mode, transport)
            if identity in observed_cases:
                errors.append(f"{label} contains a duplicate reviewer case")
                continue
            observed_cases.add(identity)

        raw_checks = report.get("checks")
        if not isinstance(raw_checks, dict):
            errors.append(f"{label} does not contain reviewer check identities")
            raw_checks = {}
        for bucket, success in (("passing", True), ("failing", False)):
            records = raw_checks.get(bucket)
            if not isinstance(records, list):
                errors.append(f"{label} does not contain {bucket} reviewer checks")
                continue
            for record in records:
                if not isinstance(record, dict):
                    errors.append(
                        f"{label} contains a malformed {bucket} reviewer check"
                    )
                    continue
                mode = record.get("mode")
                transport = record.get("transport")
                check_id = record.get("check_id")
                if (
                    not isinstance(mode, str)
                    or not isinstance(transport, str)
                    or not isinstance(check_id, str)
                    or not check_id.startswith("review/")
                    or (mode, transport) not in expected_cases
                ):
                    errors.append(
                        f"{label} contains a malformed {bucket} reviewer check"
                    )
                    continue
                identity = _ReviewCheckIdentity(mode, transport, check_id)
                if identity in identities:
                    errors.append(f"{label} contains a duplicate reviewer check")
                    continue
                identities[identity] = success
    else:
        raw_cases = report.get("cases")
        if not isinstance(raw_cases, list):
            errors.append(f"{label} does not contain production reviewer cases")
            raw_cases = []
        for case in raw_cases:
            if not isinstance(case, dict):
                errors.append(f"{label} contains a malformed reviewer case")
                continue
            mode = case.get("mode")
            transport = case.get("transport")
            if (
                not isinstance(mode, str)
                or not isinstance(transport, str)
                or (mode, transport) not in expected_cases
            ):
                errors.append(f"{label} contains an unexpected reviewer case")
                continue
            case_identity = (mode, transport)
            if case_identity in observed_cases:
                errors.append(f"{label} contains a duplicate reviewer case")
                continue
            observed_cases.add(case_identity)

            checks = case.get("checks")
            if not isinstance(checks, list) or not checks:
                errors.append(
                    f"{label} has no reviewer checks for {transport} / {mode}"
                )
                continue
            case_success = True
            for check in checks:
                if not isinstance(check, dict):
                    errors.append(f"{label} contains a malformed reviewer check")
                    continue
                check_id = check.get("name")
                success = check.get("success")
                if (
                    not isinstance(check_id, str)
                    or not check_id.startswith("review/")
                    or not isinstance(success, bool)
                ):
                    errors.append(f"{label} contains a malformed reviewer check")
                    continue
                identity = _ReviewCheckIdentity(mode, transport, check_id)
                if identity in identities:
                    errors.append(f"{label} contains a duplicate reviewer check")
                    continue
                identities[identity] = success
                case_success = case_success and success
            if case.get("success") is not case_success:
                errors.append(
                    f"{label} reports inconsistent reviewer case success "
                    f"for {transport} / {mode}"
                )

    missing_cases = sorted(expected_cases - observed_cases)
    unexpected_cases = sorted(observed_cases - expected_cases)
    for mode, transport in missing_cases:
        errors.append(f"{label} is missing required reviewer case {transport} / {mode}")
    for mode, transport in unexpected_cases:
        errors.append(f"{label} contains unexpected reviewer case {transport} / {mode}")

    summary = report.get("summary")
    passing = sum(identities.values())
    failed = len(identities) - passing
    cases_passed = sum(
        all(
            success
            for identity, success in identities.items()
            if (identity.mode, identity.transport) == case
        )
        and any((identity.mode, identity.transport) == case for identity in identities)
        for case in observed_cases
    )
    expected_summary = {
        "passed": passing,
        "failed": failed,
        "total": len(identities),
        "casesPassed": cases_passed,
        "casesTotal": len(observed_cases),
    }
    if not isinstance(summary, dict) or any(
        summary.get(name) != value for name, value in expected_summary.items()
    ):
        errors.append(f"{label} reports inconsistent reviewer check totals")
    if not compact and report.get("success") is not (
        bool(observed_cases) and failed == 0
    ):
        errors.append(f"{label} reports inconsistent overall reviewer success")
    return identities


def _compact_review_regression_baseline(
    report: Mapping[str, object],
) -> dict[str, object]:
    errors: list[str] = []
    checks = _required_review_checks(report, label="baseline", errors=errors)
    if errors:
        raise ValueError("; ".join(errors))

    summary = report.get("summary")
    assert isinstance(summary, dict)
    return {
        "baselineKind": REVIEW_BASELINE_KIND,
        "schemaVersion": REVIEW_REPORT_SCHEMA_VERSION,
        "reviewer": REVIEWER,
        "requiredModes": list(REVIEW_MODES),
        "requiredCases": [
            {"mode": mode, "transport": transport}
            for mode, transport in sorted(_required_review_cases())
        ],
        "summary": {
            key: summary[key]
            for key in ("passed", "failed", "total", "casesPassed", "casesTotal")
        },
        "checks": {
            "passing": [
                asdict(identity)
                for identity, success in sorted(checks.items())
                if success
            ],
            "failing": [
                asdict(identity)
                for identity, success in sorted(checks.items())
                if not success
            ],
        },
    }


def _evaluate_review_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_review_checks(
        baseline_report, label="baseline", errors=errors
    )
    candidate_checks = _required_review_checks(report, label="candidate", errors=errors)
    known_failures: list[dict[str, object]] = []
    new_failures: list[dict[str, object]] = []
    missing_checks: list[dict[str, object]] = []
    fixed_checks: list[dict[str, object]] = []

    for identity, success in sorted(candidate_checks.items()):
        if success:
            continue
        if baseline_checks.get(identity) is False:
            known_failures.append(asdict(identity))
        else:
            new_failures.append(asdict(identity))

    for identity, baseline_success in sorted(baseline_checks.items()):
        candidate_success = candidate_checks.get(identity)
        if candidate_success is None:
            missing_checks.append(asdict(identity))
        elif not baseline_success and candidate_success:
            fixed_checks.append(asdict(identity))

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


def _write_review_json(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 _parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run reviewer-derived MCP client regression tests against Codex."
    )
    parser.add_argument("codex_binary", type=Path)
    parser.add_argument("--mode", choices=("all", *REVIEW_MODES), default="all")
    parser.add_argument("--timeout", type=float, default=8)
    parser.add_argument("--server-script", type=Path, default=_MODULE_DIR / "server.py")
    parser.add_argument("--report", type=Path)
    parser.add_argument(
        "--baseline-report",
        type=Path,
        help="Compare every production reviewer check with a reviewed baseline.",
    )
    parser.add_argument(
        "--write-baseline",
        type=Path,
        help="Write a compact reviewer baseline from the completed full matrix.",
    )
    parser.add_argument(
        "--extract-baseline",
        type=Path,
        help="Extract a compact reviewer baseline from --baseline-report and exit.",
    )
    parser.add_argument("--artifact-parent", type=Path)
    parser.add_argument("--keep-artifacts", action="store_true")
    parser.add_argument("--json", action="store_true")
    return parser.parse_args(argv)


def _print_report(report: Mapping[str, object]) -> None:
    print(
        "Codex MCP reviewer regressions: "
        + ("PASS" if report.get("success") is True else "FAIL")
    )
    print(f"Binary: {report.get('codexBinary')}")
    cases = report.get("cases")
    if isinstance(cases, list):
        for case in cases:
            if not isinstance(case, dict):
                continue
            print(f"\n{case.get('transport')} / {case.get('mode')}")
            for check in case.get("checks", []):
                if isinstance(check, dict):
                    status = "PASS" if check.get("success") else "FAIL"
                    print(f"  {status} {check.get('name')}: {check.get('detail')}")
    summary = report.get("summary")
    if isinstance(summary, dict):
        print(
            f"\nSummary: {summary.get('passed')}/{summary.get('total')} checks; "
            f"{summary.get('casesPassed')}/{summary.get('casesTotal')} cases"
        )
    gate = report.get("regressionGate")
    if isinstance(gate, dict):
        status = "PASS" if gate.get("success") is True else "FAIL"
        known = gate.get("knownFailures")
        new = gate.get("newFailures")
        missing = gate.get("missingChecks")
        fixed = gate.get("fixedChecks")
        print(
            f"Regression gate: {status}; "
            f"{len(known) if isinstance(known, list) else 0} known, "
            f"{len(new) if isinstance(new, list) else 0} new, "
            f"{len(missing) if isinstance(missing, list) else 0} missing, "
            f"{len(fixed) if isinstance(fixed, list) else 0} fixed"
        )
        errors = gate.get("configurationErrors")
        if isinstance(errors, list):
            for error in errors:
                print(f"  Configuration error: {error}")


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()
    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 not server_script.is_file():
        print(f"error: review fixture does not exist: {server_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, UnicodeError, json.JSONDecodeError) as exc:
            print(
                f"error: cannot read reviewer baseline {baseline_path}: {exc}",
                file=sys.stderr,
            )
            return 2
        if not isinstance(loaded_baseline, dict):
            print(
                f"error: reviewer baseline 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_review_json(
                output_path, _compact_review_regression_baseline(baseline_report)
            )
        except (OSError, ValueError) as exc:
            print(f"error: cannot extract reviewer baseline: {exc}", file=sys.stderr)
            return 2
        print(f"Reviewer regression baseline: {output_path}")
        return 0

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

    modes = REVIEW_MODES if args.mode == "all" else (args.mode,)
    report = run_review_regressions(
        codex_binary,
        modes=modes,
        server_script=server_script,
        timeout_seconds=args.timeout,
        artifact_parent=parent,
        keep_artifacts=args.keep_artifacts,
    )
    if baseline_report is not None:
        report["regressionGate"] = _evaluate_review_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_review_json(output_path, _compact_review_regression_baseline(report))
        except (OSError, ValueError) as exc:
            print(f"error: cannot write reviewer baseline: {exc}", file=sys.stderr)
            return 2
    if args.report is not None:
        path = args.report.expanduser().resolve()
        _write_review_json(path, report)
    if args.json:
        print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
    else:
        _print_report(report)
        if args.report is not None:
            print(f"JSON report: {args.report.expanduser().resolve()}")
        if args.write_baseline is not None:
            print(f"Reviewer regression baseline: {args.write_baseline.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.get("success") is True else 1


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