codex-mmo/tests/test_catalog_profiles.py

from __future__ import annotations

import hashlib
import http.server
import io
import json
import os
import shutil
import stat
import tarfile
import threading
import tomllib
import unittest
import urllib.error
import zipfile
from copy import deepcopy
from datetime import date
from pathlib import Path
from typing import Any
from unittest import mock

from common import ROOT, RuntimeSandbox
from mmo_catalog import (
    _http_bytes,
    _http_json,
    build_codex_discovery_overlay,
    catalog_summary,
    discover_codex,
    discover_opencode_zen,
    discover_openrouter,
    discover_zai,
    find_model,
    local_inventory_report,
    refresh_discovery,
    verify_catalog,
)
from mmo_catalog_data import (
    _validate_models,
    _validate_resource,
    _validate_route,
    validate_model_entry,
    validated_global_catalog,
)
from mmo_codex_home import _codex_provider_config
from mmo_guidance import (
    PROFILE_SKILL_NAME,
    PROFILE_SKILL_RELATIVE_PATH,
    agent_guidance_relative_path,
    compiled_guidance,
    coordination_capable_agents,
    profile_skill_text,
)
from mmo_inventory_snapshot import (
    _opencode_go_docs,
    build_opencode_go_snapshot,
    build_opencode_zen_snapshot,
    load_inventory_snapshots,
    model_record_fingerprint,
    opencode_zen_catalog_pricing,
    opencode_zen_reasoning,
    openrouter_catalog_pricing,
    openrouter_reasoning,
    route_catalog_key,
    validate_inventory_snapshot,
)
from mmo_profiles import (
    clone_profile,
    discover_profiles,
    install_profile_pack,
    load_settings,
    profile_summary,
    remove_profile,
    resolve_profile,
)
from mmo_schema import extract_json_document, validate_instance, validate_schema_definition
from mmo_snapshot import _snapshot_fingerprint, _switchyard_routes, compile_profile, load_snapshot
from mmo_tool_mcp import (
    codex_tool_mcp_server_config,
    load_tool_mcp_registry,
    validate_tool_mcp_server,
)
from mmo_util import (
    bounded_text,
    load_install_runtime,
    parse_env_file,
    read_json,
    read_toml,
    stable_hash,
    toml_dumps,
    validate_id,
)
from mmo_version import APP_SERVER_PROTOCOL_CODEX_VERSION, MMO_SCHEMA_VERSION

import scripts.generate_catalog as generate_catalog


class CatalogProfileTests(unittest.TestCase):
    def test_tool_mcp_registry_is_closed_layered_and_operator_bounded(self) -> None:
        with RuntimeSandbox() as box:
            first = box.config / "tool-mcp.d" / "10-base.toml"
            first.write_text(
                f"""schema_version = {MMO_SCHEMA_VERSION}

[tool_mcp_servers.firecrawl]
transport = "streamable_http"
url = "https://example.invalid/mcp"
bearer_token_env_var = "FIRECRAWL_API_KEY"
enabled_tools = ["search", "scrape"]
default_tools_approval_mode = "writes"

[tool_mcp_servers.firecrawl.tools.scrape]
approval_mode = "prompt"
""",
                encoding="utf-8",
            )
            second = box.config / "tool-mcp.d" / "20-override.toml"
            second.write_text(
                f"""schema_version = {MMO_SCHEMA_VERSION}

[tool_mcp_servers.firecrawl]
transport = "stdio"
command = "/bin/true"
args = ["--fixture"]
env_vars = ["IDA_MCP_TOKEN"]
enabled_tools = ["inspect"]
default_tools_approval_mode = "approve"
""",
                encoding="utf-8",
            )
            registry = load_tool_mcp_registry()
            self.assertEqual(registry["firecrawl"]["transport"], "stdio")
            self.assertEqual(registry["firecrawl"]["enabled_tools"], ["inspect"])
            self.assertNotIn("url", registry["firecrawl"])

            enabled = codex_tool_mcp_server_config(
                registry["firecrawl"],
                {"required": False, "enabled_tools": ["inspect"]},
            )
            self.assertTrue(enabled["enabled"])
            self.assertFalse(enabled["required"])
            self.assertEqual(enabled["disabled_tools"], [])
            disabled = codex_tool_mcp_server_config(registry["firecrawl"], None)
            self.assertFalse(disabled["enabled"])
            self.assertEqual(disabled["disabled_tools"], ["inspect"])

            invalid_cases = {
                "reserved": (
                    "mmo_mesh",
                    {
                        "transport": "stdio",
                        "command": "/bin/true",
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "approve",
                    },
                    "reserved",
                ),
                "oauth": (
                    "remote",
                    {
                        "transport": "streamable_http",
                        "url": "https://example.invalid/mcp",
                        "auth": "oauth",
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    "unsupported OAuth",
                ),
                "literal secret": (
                    "local",
                    {
                        "transport": "stdio",
                        "command": "/bin/true",
                        "env": {"API_KEY": "secret"},
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    "credential-bearing",
                ),
                "literal header secret": (
                    "remote",
                    {
                        "transport": "streamable_http",
                        "url": "https://example.invalid/mcp",
                        "http_headers": {"X-Service-Token": "secret"},
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    "credential-bearing",
                ),
                "unbounded": (
                    "local",
                    {
                        "transport": "stdio",
                        "command": "/bin/true",
                        "enabled_tools": [],
                        "default_tools_approval_mode": "prompt",
                    },
                    "cannot be empty",
                ),
                "relative command path": (
                    "local",
                    {
                        "transport": "stdio",
                        "command": "./tool-mcp",
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    "PATH executable name or an absolute path",
                ),
                "nul argument": (
                    "local",
                    {
                        "transport": "stdio",
                        "command": "/bin/true",
                        "args": ["before\x00after"],
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    "args cannot contain NUL",
                ),
                "malformed URI escape": (
                    "remote",
                    {
                        "transport": "streamable_http",
                        "url": "https://example.invalid/%6G",
                        "enabled_tools": ["inspect"],
                        "default_tools_approval_mode": "prompt",
                    },
                    r"valid HTTP\(S\) URL",
                ),
            }
            for label, (server_id, definition, message) in invalid_cases.items():
                with self.subTest(label=label), self.assertRaisesRegex(ValueError, message):
                    validate_tool_mcp_server(server_id, definition, label=label)

    def test_profile_tool_mcp_grants_resolve_and_only_narrow_operator_policy(self) -> None:
        with RuntimeSandbox() as box:
            (box.config / "tool-mcp.d" / "servers.toml").write_text(
                f"""schema_version = {MMO_SCHEMA_VERSION}

[tool_mcp_servers.firecrawl]
transport = "streamable_http"
url = "https://example.invalid/mcp"
bearer_token_env_var = "FIRECRAWL_API_KEY"
enabled_tools = ["search", "scrape"]
default_tools_approval_mode = "writes"
""",
                encoding="utf-8",
            )
            with (box.config / "credentials.env").open("a", encoding="utf-8") as credentials:
                credentials.write("FIRECRAWL_API_KEY=must-not-enter-snapshot\n")
            profile = box.root / "tool-mcp-profile"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
            profile_path = profile / "profile.toml"
            original = read_toml(profile_path)
            original["agents"]["integrator"]["tool_mcp_servers"] = {
                "firecrawl": {"enabled_tools": ["search"]}
            }
            profile_path.write_text(toml_dumps(original), encoding="utf-8")

            resolved = resolve_profile(profile)
            grant = resolved["agents"]["integrator"]["tool_mcp_servers"]["firecrawl"]
            self.assertTrue(grant["required"])
            self.assertEqual(grant["enabled_tools"], ["search"])
            self.assertEqual(resolved["capabilities"]["tool_mcp_servers"], ["firecrawl"])
            snapshot = compile_profile(profile)
            self.assertEqual(snapshot["manifest"]["tool_mcp_servers"], ["firecrawl"])
            serialized = json.dumps(snapshot["resolved"])
            self.assertIn("FIRECRAWL_API_KEY", serialized)
            self.assertNotIn("must-not-enter-snapshot", serialized)

            expanded = deepcopy(original)
            expanded["agents"]["integrator"]["tool_mcp_servers"]["firecrawl"] = {
                "enabled_tools": ["unknown"]
            }
            profile_path.write_text(toml_dumps(expanded), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "exceeds the operator allowlist"):
                resolve_profile(profile)

            missing = deepcopy(original)
            missing["agents"]["integrator"]["tool_mcp_servers"] = {"missing": {}}
            profile_path.write_text(toml_dumps(missing), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "undefined operator tool MCP server"):
                resolve_profile(profile)

    def test_model_lookup_requires_an_exact_route_qualified_key(self) -> None:
        with RuntimeSandbox():
            key = "codex_chatgpt_builtin__gpt_5_6_sol"
            self.assertEqual(find_model(key)["key"], key)
            for unqualified_or_inexact in (
                "gpt-5.6-sol",
                "deepseek-v4-flash",
                "OPENAI_CHATGPT__GPT_5_6_SOL",
            ):
                with self.subTest(value=unqualified_or_inexact):
                    with self.assertRaisesRegex(
                        FileNotFoundError, "route-qualified catalog model key"
                    ):
                        find_model(unqualified_or_inexact)

    def test_catalog_generator_reports_success_after_normal_regeneration(self) -> None:
        catalog_path = ROOT / "config" / "catalog.toml"
        inventory_path = ROOT / "config" / "upstream-inventory.json"
        outputs = {
            catalog_path: catalog_path.read_text(encoding="utf-8") + "# synthetic drift\n",
            inventory_path: inventory_path.read_text(encoding="utf-8"),
        }
        stdout = io.StringIO()
        with (
            mock.patch.object(generate_catalog, "_rendered_outputs", return_value=outputs),
            mock.patch.object(generate_catalog, "atomic_write_text") as write_catalog,
            mock.patch.object(generate_catalog, "atomic_write_json") as write_inventory,
            mock.patch("sys.stdout", stdout),
        ):
            self.assertEqual(generate_catalog.main([]), 0)
        report = json.loads(stdout.getvalue())
        self.assertTrue(report["passed"], report)
        self.assertEqual(report["changed"], ["config/catalog.toml"])
        write_catalog.assert_called_once()
        write_inventory.assert_called_once()

    def test_discovery_http_rejects_cross_origin_redirects(self) -> None:
        target_requests: list[str | None] = []

        class TargetHandler(http.server.BaseHTTPRequestHandler):
            def do_GET(self) -> None:
                target_requests.append(self.headers.get("Authorization"))
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(b'{"data":[]}')

            def log_message(self, _format: str, *_args: Any) -> None:
                pass

        target = http.server.ThreadingHTTPServer(("127.0.0.1", 0), TargetHandler)

        class RedirectHandler(http.server.BaseHTTPRequestHandler):
            def do_GET(self) -> None:
                self.send_response(302)
                self.send_header(
                    "Location",
                    f"http://127.0.0.1:{target.server_port}/models",
                )
                self.end_headers()

            def log_message(self, _format: str, *_args: Any) -> None:
                pass

        redirect = http.server.ThreadingHTTPServer(("127.0.0.1", 0), RedirectHandler)
        threads = [
            threading.Thread(target=server.serve_forever, daemon=True)
            for server in (target, redirect)
        ]
        for thread in threads:
            thread.start()
        try:
            with self.assertRaisesRegex(urllib.error.URLError, "cross-origin discovery redirect"):
                _http_json(
                    f"http://127.0.0.1:{redirect.server_port}/models",
                    2.0,
                    headers={"Authorization": "Bearer must-not-leak"},
                )
            self.assertEqual(target_requests, [])
        finally:
            redirect.shutdown()
            target.shutdown()
            redirect.server_close()
            target.server_close()

    def test_discovery_http_rejects_non_http_urls_before_opening_them(self) -> None:
        with RuntimeSandbox() as box:
            document = box.root / "local-models.json"
            document.write_text('{"data":[{"id":"must-not-be-read"}]}', encoding="utf-8")
            for url in (
                document.as_uri(),
                "https://example.invalid/\x00models",
                "https://example.invalid/\N{NO-BREAK SPACE}models",
            ):
                with self.subTest(url=url), self.assertRaisesRegex(ValueError, "absolute HTTP"):
                    _http_json(url, 2.0)

    def test_install_runtime_requires_a_json_object_root(self) -> None:
        with RuntimeSandbox() as box:
            runtime = box.root / "runtime.json"
            runtime.write_text("[]\n", encoding="utf-8")
            with (
                mock.patch("mmo_util.install_runtime_path", return_value=runtime),
                self.assertRaisesRegex(RuntimeError, "root must be an object"),
            ):
                load_install_runtime()

    def test_versioned_configuration_rejects_unknown_fields_and_missing_explicit_files(
        self,
    ) -> None:
        with RuntimeSandbox() as box:
            profile = box.root / "closed-schema-profile"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
            profile_path = profile / "profile.toml"
            original_profile = read_toml(profile_path)

            typo = deepcopy(original_profile)
            typo["agents"]["invariant_designer"]["network_acess"] = True
            profile_path.write_text(toml_dumps(typo), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "network_acess"):
                resolve_profile(profile)

            missing_metadata = deepcopy(original_profile)
            del missing_metadata["description"]
            profile_path.write_text(toml_dumps(missing_metadata), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "profile.description"):
                resolve_profile(profile)

            bad_approval = deepcopy(original_profile)
            bad_approval["agents"]["invariant_designer"]["approval_policy"] = "sometimes"
            profile_path.write_text(toml_dumps(bad_approval), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "approval_policy"):
                resolve_profile(profile)

            indexed_search = deepcopy(original_profile)
            indexed_search["agents"]["invariant_designer"]["web_search"] = "indexed"
            profile_path.write_text(toml_dumps(indexed_search), encoding="utf-8")
            self.assertEqual(
                resolve_profile(profile)["agents"]["invariant_designer"]["web_search"],
                "indexed",
            )

            missing = deepcopy(original_profile)
            missing["smoke"] = "missing-smoke.toml"
            profile_path.write_text(toml_dumps(missing), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "explicit profile smoke file is missing"):
                resolve_profile(profile)

            profile_path.write_text(toml_dumps(original_profile), encoding="utf-8")
            smoke_path = profile / "smoke.toml"
            smoke = read_toml(smoke_path)
            smoke["tasks"][0]["expected_pattern"] = "silently ignored before"
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "expected_pattern"):
                resolve_profile(profile)

            smoke = read_toml(smoke_path)
            del smoke["tasks"][0]["expected_pattern"]
            smoke["tasks"][0]["required_mcp_tools"] = ["missing.search"]
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "requires ungranted MCP tool"):
                resolve_profile(profile)

            outside_smoke = box.root / "outside-smoke.toml"
            outside_smoke.write_text(
                f"schema_version = {MMO_SCHEMA_VERSION}\ntasks = []\n",
                encoding="utf-8",
            )
            escaped = deepcopy(original_profile)
            escaped["smoke"] = "../outside-smoke.toml"
            profile_path.write_text(toml_dumps(escaped), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "smoke path escapes"):
                resolve_profile(profile)

            bad_catalog = {
                "schema_version": MMO_SCHEMA_VERSION,
                "routes": {
                    "typo_route": {
                        "driver": "switchyard",
                        "api_operator": "example",
                        "access_product": "example_api",
                        "wire_protocol": "openai_chat",
                        "billing_mode": "api",
                        "base_url": "https://example.invalid/v1",
                        "max_retry": 1,
                    }
                },
            }
            overlay = box.config / "catalog.d" / "typo.toml"
            overlay.write_text(toml_dumps(bad_catalog), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "max_retry"):
                validated_global_catalog()

            overlay.unlink()
            settings_path = box.config / "settings.toml"
            settings_path.write_text(
                settings_path.read_text(encoding="utf-8") + 'codex_bni = "typo"\n',
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "codex_bni"):
                load_settings()

    def test_smoke_required_mcp_tools_support_dotted_server_ids_without_ambiguity(self) -> None:
        with RuntimeSandbox() as box:
            (box.config / "tool-mcp.d" / "dotted.toml").write_text(
                f"""schema_version = {MMO_SCHEMA_VERSION}

[tool_mcp_servers."repo.search"]
transport = "stdio"
command = "/bin/true"
enabled_tools = ["query"]
default_tools_approval_mode = "approve"

[tool_mcp_servers.repo]
transport = "stdio"
command = "/bin/true"
enabled_tools = ["search.query"]
default_tools_approval_mode = "approve"
""",
                encoding="utf-8",
            )
            profile = box.root / "dotted-smoke-tool-profile"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
            profile_path = profile / "profile.toml"
            profile_data = read_toml(profile_path)
            profile_data["id"] = "dotted-smoke-tool-profile"
            profile_data["agents"]["integrator"]["tool_mcp_servers"] = {
                "repo.search": {"enabled_tools": ["query"]}
            }
            profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
            smoke_path = profile / "smoke.toml"
            smoke = read_toml(smoke_path)
            smoke["tasks"][0]["required_mcp_tools"] = ["repo.search.query"]
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")

            resolved = resolve_profile(profile)
            self.assertEqual(
                resolved["smoke"]["tasks"][0]["required_mcp_tools"],
                ["repo.search.query"],
            )

            profile_data["agents"]["integrator"]["tool_mcp_servers"]["repo"] = {
                "enabled_tools": ["search.query"]
            }
            profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "ambiguous across granted servers"):
                resolve_profile(profile)

    def test_smoke_can_require_only_the_runtime_owned_agent_mcp_tools_a_role_receives(
        self,
    ) -> None:
        with RuntimeSandbox() as box:
            profile = box.root / "agent-mcp-smoke-profile"
            shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
            profile_path = profile / "profile.toml"
            profile_data = read_toml(profile_path)
            profile_data["id"] = "agent-mcp-smoke-profile"
            profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
            smoke_path = profile / "smoke.toml"
            smoke = read_toml(smoke_path)
            smoke["tasks"][0]["required_mcp_tools"] = [
                "mmo_mesh.agents_spawn",
                "mmo_mesh.agents_wait",
                "mmo_mesh.agent_result",
                "mmo_mesh.agent_result_accept",
            ]
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")

            resolved = resolve_profile(profile)
            self.assertEqual(
                resolved["smoke"]["tasks"][0]["required_mcp_tools"],
                smoke["tasks"][0]["required_mcp_tools"],
            )

            smoke["tasks"][2]["required_mcp_tools"] = ["mmo_mesh.agents_spawn"]
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "requires ungranted MCP tool"):
                resolve_profile(profile)

    def test_shared_resource_lock_requires_one_capacity_definition(self) -> None:
        with RuntimeSandbox() as box:
            (box.config / "catalog.d" / "conflicting-resources.toml").write_text(
                f"""schema_version = {MMO_SCHEMA_VERSION}

[resources.first_pool]
lock_key = "shared:test-pool"
max_active = 1

[resources.second_pool]
lock_key = "shared:test-pool"
max_active = 2
""",
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "conflicting max_active"):
                validated_global_catalog()

    def test_current_schema_rejects_wrong_scalar_types_and_nonrelative_pack_paths(self) -> None:
        with RuntimeSandbox() as box:
            profile = box.root / "typed-profile"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
            profile_path = profile / "profile.toml"
            original = read_toml(profile_path)

            mutations = {
                "boolean schema version": ("schema_version", True, "schema_version"),
                "numeric profile id": ("id", 7, "profile id"),
                "non-semantic profile version": (
                    "version",
                    "8",
                    "active package version",
                ),
                "numeric agent description": (
                    "agents.invariant_designer.description",
                    7,
                    "agent invariant_designer.description",
                ),
                "boolean instructions": (
                    "agents.invariant_designer.instructions",
                    False,
                    "agent invariant_designer.instructions",
                ),
                "top-level instructions": (
                    "agents.invariant_designer.instructions",
                    "README.md",
                    r"agents/\*\.md",
                ),
                "boolean resource group": (
                    "agents.invariant_designer.resource_group",
                    False,
                    "agent invariant_designer.resource_group",
                ),
            }
            for label, (path, value, message) in mutations.items():
                with self.subTest(label=label):
                    changed = deepcopy(original)
                    cursor = changed
                    parts = path.split(".")
                    for part in parts[:-1]:
                        cursor = cursor[part]
                    cursor[parts[-1]] = value
                    profile_path.write_text(toml_dumps(changed), encoding="utf-8")
                    with self.assertRaisesRegex(ValueError, message):
                        resolve_profile(profile)

            absolute_catalog = deepcopy(original)
            absolute_catalog["catalog"] = str((profile / "catalog.toml").resolve())
            profile_path.write_text(toml_dumps(absolute_catalog), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "relative path"):
                resolve_profile(profile)

            profile_path.write_text(toml_dumps(original), encoding="utf-8")
            smoke_path = profile / "smoke.toml"
            smoke = read_toml(smoke_path)
            smoke["schema_version"] = True
            smoke_path.write_text(toml_dumps(smoke), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "unsupported smoke schema"):
                resolve_profile(profile)

            settings_path = box.config / "settings.toml"
            settings_path.write_text(
                settings_path.read_text(encoding="utf-8").replace(
                    str(box.base_codex_home), "relative-codex-home"
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "expand to an absolute path"):
                load_settings()

    def test_catalog_scalar_types_are_validated_before_generation(self) -> None:
        with self.assertRaisesRegex(ValueError, "route bad.driver"):
            _validate_route("bad", {"driver": ["switchyard"]})

        route_input = {
            "driver": "switchyard",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_chat",
            "billing_mode": "api",
            "base_url": "https://example.invalid/v1",
        }
        with self.assertRaisesRegex(ValueError, "credential_envs"):
            _validate_route("bad", {**route_input, "credential_envs": [True]})

        builtin_input = {
            "driver": "codex_builtin",
            "api_operator": "openai",
            "access_product": "chatgpt_codex",
            "wire_protocol": "codex_builtin",
            "billing_mode": "chatgpt_subscription",
            "provider_id": "openai",
            "auth": "chatgpt",
        }
        with self.assertRaisesRegex(ValueError, "built-in provider_id"):
            _validate_route("bad", {**builtin_input, "provider_id": "not-bundled"})
        with self.assertRaisesRegex(ValueError, "auth must be"):
            _validate_route("bad", {**builtin_input, "auth": "builtin"})

        route = _validate_route("example", route_input)
        self.assertEqual(route["max_retries"], 1)
        model_input = {"route": "example", "upstream_id": "example", "maker": "maker"}
        with self.assertRaisesRegex(ValueError, "exact route namespace 'example'"):
            validate_model_entry("ambiguous_model_key", model_input, {"example": route})
        with self.assertRaisesRegex(ValueError, "duplicate route/upstream binding"):
            _validate_models(
                {
                    "example__first": {
                        "route": "example",
                        "upstream_id": "same-model",
                        "maker": "maker",
                    },
                    "example__second": {
                        "route": "example",
                        "upstream_id": "same-model",
                        "maker": "maker",
                    },
                },
                {"example": route},
            )
        for field, value in (
            ("display_name", False),
            ("description", 7),
            ("availability", []),
            ("source", {"wrong": "shape"}),
            ("resource_group", False),
        ):
            with (
                self.subTest(model_field=field),
                self.assertRaisesRegex(ValueError, f"model example__bad.{field}"),
            ):
                validate_model_entry(
                    "example__bad",
                    {**model_input, field: value},
                    {"example": route},
                )

        for cost_value in (True, -1, float("inf"), "3.0"):
            with (
                self.subTest(cache_write_input_cost=cost_value),
                self.assertRaisesRegex(
                    ValueError,
                    "model example__bad.cache_write_input_cost_per_million",
                ),
            ):
                validate_model_entry(
                    "example__bad",
                    {
                        **model_input,
                        "cache_write_input_cost_per_million": cost_value,
                    },
                    {"example": route},
                )

        with self.assertRaisesRegex(ValueError, "unknown fields.*aliases"):
            validate_model_entry(
                "example__bad",
                {**model_input, "aliases": ["legacy-key"]},
                {"example": route},
            )

        for non_finite in (float("nan"), float("inf"), 10**400):
            with (
                self.subTest(openrouter_max_price=non_finite),
                self.assertRaisesRegex(ValueError, "max_price.prompt.*finite"),
            ):
                validate_model_entry(
                    "example__bad",
                    {
                        **model_input,
                        "route_policy": {"max_price": {"prompt": non_finite}},
                    },
                    {"example": route},
                )

        with self.assertRaisesRegex(ValueError, "resource bad.description"):
            _validate_resource("bad", {"description": False})

        function_only = validate_model_entry(
            "example__function_only",
            {
                **model_input,
                "supports_custom_tools": False,
            },
            {"example": route},
        )
        self.assertTrue(function_only["tool_calling"])
        self.assertFalse(function_only["supports_custom_tools"])

        with self.assertRaisesRegex(ValueError, "supports_custom_tools requires tool_calling"):
            validate_model_entry(
                "example__invalid_custom_tools",
                {
                    **model_input,
                    "tool_calling": False,
                    "supports_custom_tools": True,
                    "parallel_tool_calls": False,
                },
                {"example": route},
            )

    def test_profile_catalog_validates_unused_rows_and_resource_references(self) -> None:
        with RuntimeSandbox() as box:
            profile = box.root / "unused-invalid-catalog-row"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", profile)
            profile_path = profile / "profile.toml"
            data = read_toml(profile_path)
            data["id"] = "unused-invalid-catalog-row"
            data["catalog"] = "catalog.toml"
            profile_path.write_text(toml_dumps(data), encoding="utf-8")
            (profile / "catalog.toml").write_text(
                toml_dumps(
                    {
                        "schema_version": MMO_SCHEMA_VERSION,
                        "models": {
                            "codex_chatgpt_builtin__unused_invalid_resource": {
                                "route": "codex_chatgpt_builtin",
                                "upstream_id": "unused-invalid-resource",
                                "maker": "example",
                                "resource_group": "missing-resource",
                            }
                        },
                    }
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(
                ValueError, "unused_invalid_resource.*unknown resource group"
            ):
                resolve_profile(profile)

    def test_compiler_derives_weighted_concurrency_and_rejects_retired_spawn_quotas(self) -> None:
        with RuntimeSandbox() as box:
            weighted = box.root / "weighted-concurrency"
            shutil.copytree(ROOT / "profiles" / "incident-hypothesis-triage", weighted)
            profile_path = weighted / "profile.toml"
            data = read_toml(profile_path)
            data["id"] = "weighted-concurrency"
            data["catalog"] = "catalog.toml"
            data["agents"]["evidence_runner"]["resource_group"] = "weighted-test"
            data["agents"]["evidence_runner"]["resource_units"] = 1
            data["agents"]["causal_challenger"]["resource_group"] = "weighted-test"
            data["agents"]["causal_challenger"]["resource_units"] = 3
            data["coordination"]["max_active_agents"] = 4
            profile_path.write_text(toml_dumps(data), encoding="utf-8")
            (weighted / "catalog.toml").write_text(
                toml_dumps(
                    {
                        "schema_version": MMO_SCHEMA_VERSION,
                        "resources": {
                            "weighted-test": {
                                "lock_key": "test:weighted",
                                "max_active": 3,
                            }
                        },
                    }
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, r"feasible concurrency \(3\)"):
                resolve_profile(weighted)

            data["coordination"]["max_active_agents"] = 3
            data["agents"]["causal_challenger"]["resource_units"] = 4
            profile_path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "resource_units.*exceeds resource group"):
                resolve_profile(weighted)

            retired_quota = box.root / "retired-spawn-quota"
            shutil.copytree(ROOT / "profiles" / "incident-hypothesis-triage", retired_quota)
            profile_path = retired_quota / "profile.toml"
            data = read_toml(profile_path)
            data["id"] = "retired-spawn-quota"
            data["coordination"]["max_total_spawns"] = 2
            profile_path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "unknown fields: max_total_spawns"):
                resolve_profile(retired_quota)

            data = read_toml(retired_quota / "profile.toml")
            data["coordination"].pop("max_total_spawns")
            data["agents"]["evidence_runner"]["max_spawns"] = 2
            profile_path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "unknown fields: max_spawns"):
                resolve_profile(retired_quota)

    def test_gateway_hash_covers_every_switchyard_route_semantic(self) -> None:
        route = _validate_route(
            "example",
            {
                "driver": "switchyard",
                "api_operator": "example",
                "access_product": "example_api",
                "wire_protocol": "openai_responses",
                "billing_mode": "api",
                "base_url": "https://example.invalid/v1",
            },
        )
        model = validate_model_entry(
            "example__model",
            {
                "route": "example",
                "upstream_id": "example-model",
                "maker": "example",
                "context_window": 32_000,
                "tool_calling": True,
                "default_reasoning": "high",
            },
            {"example": route},
        )
        resolved = {"routes": {"example": route}, "models": {"example__model": model}}
        baseline_routes, baseline_ids, baseline_hash = _switchyard_routes(resolved)
        self.assertIsNotNone(baseline_hash)
        self.assertEqual(
            baseline_routes["routes"]["example__model"]["id"],
            baseline_ids["example__model"],
        )

        mutations = {
            "context_window": 64_000,
            "tool_calling": False,
            "default_reasoning": "none",
        }
        for field, value in mutations.items():
            with self.subTest(field=field):
                changed = deepcopy(resolved)
                changed["models"]["example__model"][field] = value
                _routes, changed_ids, changed_hash = _switchyard_routes(changed)
                self.assertNotEqual(changed_hash, baseline_hash)
                self.assertNotEqual(changed_ids, baseline_ids)

        changed = deepcopy(resolved)
        changed["models"]["example__model"]["extra_body"] = {"service_tier": "priority"}
        routes, changed_ids, changed_hash = _switchyard_routes(changed)
        self.assertEqual(
            routes["targets"]["example__model"]["extra_body"],
            {"service_tier": "priority"},
        )
        self.assertNotEqual(changed_hash, baseline_hash)
        self.assertNotEqual(changed_ids, baseline_ids)

    def test_model_extra_body_matches_switchyard_target_schema(self) -> None:
        switchyard = _validate_route(
            "example",
            {
                "driver": "switchyard",
                "api_operator": "example",
                "access_product": "example_api",
                "wire_protocol": "openai_responses",
                "billing_mode": "api",
                "base_url": "https://example.invalid/v1",
            },
        )
        model = validate_model_entry(
            "example__model",
            {
                "route": "example",
                "upstream_id": "example-model",
                "maker": "example",
                "extra_body": {"service_tier": "priority", "temperature": 0.25},
            },
            {"example": switchyard},
        )
        self.assertEqual(model["extra_body"]["service_tier"], "priority")

        builtin = _validate_route(
            "builtin",
            {
                "driver": "codex_builtin",
                "api_operator": "openai",
                "access_product": "chatgpt_codex",
                "wire_protocol": "codex_builtin",
                "billing_mode": "chatgpt_subscription",
                "provider_id": "openai",
                "auth": "chatgpt",
            },
        )
        with self.assertRaisesRegex(ValueError, "only by Switchyard"):
            validate_model_entry(
                "builtin__bad_model",
                {
                    "route": "builtin",
                    "upstream_id": "gpt-example",
                    "maker": "openai",
                    "extra_body": {"service_tier": "priority"},
                },
                {"builtin": builtin},
            )
        with self.assertRaisesRegex(ValueError, "non-finite"):
            validate_model_entry(
                "example__bad_model",
                {
                    "route": "example",
                    "upstream_id": "example-model",
                    "maker": "example",
                    "extra_body": {"temperature": float("nan")},
                },
                {"example": switchyard},
            )

    def test_bounded_text_includes_marker_inside_hard_limit(self) -> None:
        source = "0123456789" * 100
        for limit in (0, 1, 20, 21, 22, 23, 500):
            with self.subTest(limit=limit):
                result, truncated = bounded_text(source, limit)
                self.assertTrue(truncated)
                self.assertLessEqual(len(result), limit)
                self.assertEqual(len(result), limit)
        result, truncated = bounded_text("short", 5)
        self.assertEqual((result, truncated), ("short", False))

    def test_gateway_host_matches_switchyard_ip_and_local_security_boundary(self) -> None:
        with RuntimeSandbox() as box:
            settings_path = box.config / "settings.toml"
            original = settings_path.read_text(encoding="utf-8")
            for host, expected in (("127.0.0.1", "127.0.0.1"), ("::1", "::1")):
                with self.subTest(valid=host):
                    settings_path.write_text(
                        original.replace('gateway_host = "127.0.0.1"', f'gateway_host = "{host}"'),
                        encoding="utf-8",
                    )
                    self.assertEqual(load_settings()["gateway_host"], expected)

            for host in (
                "localhost",
                "[::1]",
                "fe80::1%eth0",
                "0.0.0.0",
                "192.0.2.1",
                "::",
            ):
                with self.subTest(invalid=host):
                    settings_path.write_text(
                        original.replace('gateway_host = "127.0.0.1"', f'gateway_host = "{host}"'),
                        encoding="utf-8",
                    )
                    with self.assertRaisesRegex(ValueError, "settings.gateway_host"):
                        load_settings()

    def test_route_base_urls_match_http_and_downstream_join_contract(self) -> None:
        common = {
            "driver": "switchyard",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_responses",
            "billing_mode": "api",
        }
        for base_url in (
            "https://example.invalid/v1",
            "HTTPS://EXAMPLE.invalid/v1",
            "http://localhost:8000",
            "http://[::1]:8000/v1/responses",
        ):
            with self.subTest(valid=base_url):
                route = _validate_route("valid_base", {**common, "base_url": base_url})
                self.assertEqual(route["base_url"], base_url)

        for base_url in (
            "https://",
            "ftp://example.invalid/v1",
            "https://user:password@example.invalid/v1",
            "https://example.invalid/v1?tenant=a",
            "https://example.invalid/v1?",
            "https://example.invalid/v1#fragment",
            "https://example.invalid/v1#",
            "https://bad host.invalid/v1",
            "https://example.invalid:99999/v1",
            "https://example.invalid/v1\\responses",
            "https://example.invalid/v1\nresponses",
            "https://example.invalid/%6G",
            "https://example.invalid/café",
            "http:/[::1]",
        ):
            for driver in ("switchyard", "codex_custom"):
                with (
                    self.subTest(invalid=base_url, driver=driver),
                    self.assertRaisesRegex(ValueError, r"valid HTTP\(S\) base_url"),
                ):
                    value = {
                        "driver": driver,
                        "api_operator": "example",
                        "access_product": "example_api",
                        "wire_protocol": "openai_responses",
                        "billing_mode": "api",
                        "base_url": base_url,
                    }
                    _validate_route("bad_base", value)

    def test_route_fields_are_rejected_when_inert_for_the_selected_driver(self) -> None:
        builtin = {
            "driver": "codex_builtin",
            "api_operator": "openai",
            "access_product": "chatgpt_codex",
            "wire_protocol": "codex_builtin",
            "billing_mode": "chatgpt_subscription",
            "provider_id": "openai",
            "auth": "chatgpt",
        }
        with self.assertRaisesRegex(ValueError, "unsupported by driver 'codex_builtin'"):
            _validate_route("builtin", {**builtin, "base_url": "https://example.invalid"})

        custom = {
            "driver": "codex_custom",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_responses",
            "billing_mode": "api",
            "base_url": "https://example.invalid/v1",
        }
        with self.assertRaisesRegex(ValueError, "unsupported by driver 'codex_custom'"):
            _validate_route("custom", {**custom, "auth": "bearer"})

        catalog_only = _validate_route(
            "inventory",
            {
                "driver": "catalog_only",
                "api_operator": "example",
                "access_product": "example_catalog",
                "wire_protocol": "anthropic_messages",
                "billing_mode": "catalog_only",
                "base_url": "https://example.invalid/api/anthropic",
                "auth": "bearer",
                "credential_envs": ["EXAMPLE_API_KEY"],
            },
        )
        self.assertEqual(catalog_only["auth"], "bearer")

    def test_codex_custom_retry_fields_remain_unsigned_toml_integers(self) -> None:
        common = {
            "driver": "codex_custom",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_responses",
            "billing_mode": "api",
            "base_url": "https://example.invalid/v1",
        }
        valid = _validate_route(
            "retry_route",
            {
                **common,
                "request_max_retries": 0,
                "stream_max_retries": 100,
                "stream_idle_timeout_ms": 1_000,
            },
        )
        self.assertEqual(valid["request_max_retries"], 0)
        self.assertEqual(valid["stream_max_retries"], 100)
        self.assertEqual(valid["stream_idle_timeout_ms"], 1_000)

        defaults = _validate_route("retry_defaults", common)
        self.assertEqual(defaults["request_max_retries"], 1)
        self.assertEqual(defaults["stream_max_retries"], 1)
        self.assertEqual(defaults["stream_idle_timeout_ms"], 600_000)

        for field in (
            "request_max_retries",
            "stream_max_retries",
        ):
            for invalid in (True, -1, "1", 2**63):
                with (
                    self.subTest(field=field, invalid=invalid),
                    self.assertRaisesRegex(ValueError, "must be an integer between"),
                ):
                    _validate_route("bad_retry", {**common, field: invalid})

        for invalid in (True, -1, 0, 999, "1000", 2**63):
            with (
                self.subTest(field="stream_idle_timeout_ms", invalid=invalid),
                self.assertRaisesRegex(ValueError, "must be an integer between"),
            ):
                _validate_route("bad_retry", {**common, "stream_idle_timeout_ms": invalid})

        with self.assertRaisesRegex(ValueError, "name must be a non-empty string"):
            _validate_route("bad_name", {**common, "name": 7})

    def test_route_headers_require_exact_v2_downstream_names(self) -> None:
        switchyard_input = {
            "driver": "switchyard",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_chat",
            "billing_mode": "api",
            "base_url": "https://example.invalid/v1",
        }
        switchyard = _validate_route(
            "header_switchyard",
            {**switchyard_input, "extra_headers": {"X-Static": "catalog-value"}},
        )
        self.assertEqual(switchyard["extra_headers"], {"X-Static": "catalog-value"})
        routes, _route_ids, _gateway_hash = _switchyard_routes(
            {"routes": {"header_switchyard": switchyard}, "models": {}}
        )
        self.assertEqual(
            routes["llm_clients"]["header_switchyard"]["extra_headers"],
            {"X-Static": "catalog-value"},
        )
        self.assertNotIn("http_headers", routes["llm_clients"]["header_switchyard"])

        custom_input = {
            "driver": "codex_custom",
            "api_operator": "example",
            "access_product": "example_api",
            "wire_protocol": "openai_responses",
            "billing_mode": "api",
            "base_url": "https://example.invalid/v1",
        }
        custom = _validate_route(
            "header_codex",
            {
                **custom_input,
                "http_headers": {"X-Static": "catalog-value"},
                "env_http_headers": {"X-Secret": "EXAMPLE_SECRET"},
            },
        )
        self.assertEqual(custom["http_headers"], {"X-Static": "catalog-value"})
        self.assertEqual(custom["env_http_headers"], {"X-Secret": "EXAMPLE_SECRET"})
        snapshot = {
            "resolved": {
                "models": {
                    "header_codex__custom_model": {
                        "route": "header_codex",
                        "upstream_id": "custom-model",
                    }
                },
                "routes": {"header_codex": custom},
            }
        }
        _model_id, provider_id, provider_tables, _flags = _codex_provider_config(
            snapshot, {"model": "header_codex__custom_model"}, None
        )
        self.assertEqual(
            provider_tables[provider_id]["http_headers"],
            {"X-Static": "catalog-value"},
        )
        self.assertEqual(
            provider_tables[provider_id]["env_http_headers"],
            {"X-Secret": "EXAMPLE_SECRET"},
        )
        self.assertEqual(provider_tables[provider_id]["stream_idle_timeout_ms"], 3_600_000)
        _model_id, provider_id, provider_tables, _flags = _codex_provider_config(
            snapshot,
            {
                "model": "header_codex__custom_model",
                "stall_warning_seconds": 21_600,
                "finalization_grace_seconds": 900,
            },
            None,
        )
        self.assertEqual(provider_tables[provider_id]["stream_idle_timeout_ms"], 43_200_000)

        for route, field in (
            (switchyard_input, "default_headers"),
            (switchyard_input, "http_headers"),
            (switchyard_input, "env_headers"),
            (custom_input, "default_headers"),
            (custom_input, "env_headers"),
        ):
            with (
                self.subTest(field=field, driver=route["driver"]),
                self.assertRaisesRegex(ValueError, "unknown fields|unsupported"),
            ):
                _validate_route(
                    "retired_header_alias",
                    {**route, field: {"X-Secret": "EXAMPLE_SECRET"}},
                )
        for control in ("\x00", "\x01", "\x0b", "\x1f", "\x7f", "\r", "\n"):
            with (
                self.subTest(control=control),
                self.assertRaisesRegex(ValueError, "prohibited control character"),
            ):
                _validate_route(
                    "bad_header_value",
                    {
                        **custom_input,
                        "http_headers": {"X-Static": f"safe{control}unsafe"},
                    },
                )
        accepted = _validate_route(
            "valid_header_values",
            {**custom_input, "http_headers": {"X-Static": "tab\tand café"}},
        )
        self.assertEqual(accepted["http_headers"], {"X-Static": "tab\tand café"})
        with self.assertRaisesRegex(ValueError, "duplicate case-insensitive"):
            _validate_route(
                "duplicate_header_case",
                {
                    **custom_input,
                    "http_headers": {"X-Trace": "first", "x-trace": "second"},
                },
            )
        with self.assertRaisesRegex(ValueError, "both http_headers and env_http_headers"):
            _validate_route(
                "duplicate_header_sources",
                {
                    **custom_input,
                    "http_headers": {"X-Trace": "literal"},
                    "env_http_headers": {"x-trace": "TRACE_HEADER"},
                },
            )

    def test_complete_route_qualified_catalog_baselines(self) -> None:
        with RuntimeSandbox():
            summary = catalog_summary()
            self.assertEqual(summary["routes"], 21)
            self.assertEqual(summary["models"], 560)
            self.assertEqual(summary["agent_compatible_models"], 462)
            self.assertEqual(summary["resources"], 9)
            self.assertGreaterEqual(len(summary["models_by_maker"]), 40)
            self.assertEqual(
                set(summary["models_by_api_operator"]),
                {"local", "openai", "opencode", "openrouter", "zai"},
            )

            report = local_inventory_report()
            self.assertTrue(report["passed"], report)
            self.assertEqual(report["source_errors"], [])
            self.assertEqual(
                {key: value["actual_count"] for key, value in report["inventories"].items()},
                {
                    "openai-codex": 6,
                    "opencode-go": 29,
                    "opencode-zen": 64,
                    "openrouter": 422,
                    "zai-api": 35,
                    "zai-coding-plan": 3,
                },
            )

            catalog = validated_global_catalog()
            identities = [
                (model["route"], model["upstream_id"]) for model in catalog["models"].values()
            ]
            self.assertEqual(len(identities), len(set(identities)))
            self.assertTrue(
                all(
                    key.startswith(f"{model['route']}__") and model["maker"]
                    for key, model in catalog["models"].items()
                )
            )
            self.assertEqual(
                catalog["routes"]["opencode_go_openai_chat"]["credential_envs"],
                ["OPENCODE_API_KEY"],
            )
            self.assertEqual(
                catalog["routes"]["openrouter_openai_chat"]["credential_envs"],
                ["OPENROUTER_API_KEY"],
            )
            self.assertEqual(
                catalog["routes"]["zai_coding_responses"]["credential_envs"],
                ["ZAI_CODING_API_KEY"],
            )
            self.assertEqual(
                catalog["routes"]["zai_coding_responses"]["base_url"],
                "https://api.z.ai/api/v1",
            )

            glm = find_model("zai_coding_responses__glm_5_3")
            self.assertEqual(glm["route"], "zai_coding_responses")
            self.assertEqual(glm["upstream_id"], "glm-5.3")
            self.assertEqual(glm["maker"], "zai")
            self.assertEqual(glm["reasoning_levels"], ["low", "high", "max"])
            self.assertEqual(glm["default_reasoning"], "max")
            self.assertNotIn("aliases", glm)
            for removed_key in (
                "zai__glm_5_3",
                "zai_coding_chat__glm_5_1",
                "zai_coding_chat__glm_5_2",
                "openai_chatgpt__gpt_5_6_sol",
                "opencode_go__deepseek_v4_flash",
                "openrouter__qwen_qwen3_8_27b",
                "llama_cpp_local__qwen3_5_9b",
            ):
                with self.subTest(removed_key=removed_key):
                    with self.assertRaises(FileNotFoundError):
                        find_model(removed_key)

            local_qwen = find_model("llama_cpp_local_openai_chat__qwen3_5_9b")
            self.assertEqual(local_qwen["context_window"], 32_768)
            self.assertEqual(local_qwen["max_output_tokens"], 8_192)
            self.assertEqual(local_qwen["reasoning_levels"], ["none"])

            flash = find_model("opencode_go_openai_chat__deepseek_v4_flash")
            self.assertEqual(flash["maker"], "deepseek")
            self.assertEqual(flash["reasoning_levels"], ["low", "high", "max"])
            self.assertTrue(flash["structured_output"])
            self.assertNotIn("input_cost_per_million", flash)

            with self.assertRaises(FileNotFoundError):
                find_model("opencode_go_responses__muse_spark_1_2")
            contributor = find_model("opencode_go_responses__muse_spark_1_2_contributor")
            self.assertEqual(contributor["availability"], "current")
            self.assertTrue(contributor["agent_compatible"])
            self.assertTrue(contributor["tool_calling"])
            vision = find_model("opencode_go_openai_chat__deepseek_v4_flash_vision_exp")
            self.assertEqual(vision["modalities"], ["text"])
            self.assertTrue(vision["agent_compatible"])
            self.assertNotIn("input_cost_per_million", vision)
            ox_alpha = find_model("opencode_go_openai_chat__ox_alpha_free")
            self.assertTrue(ox_alpha["agent_compatible"])
            self.assertNotIn("input_cost_per_million", ox_alpha)

            sol = find_model("codex_chatgpt_builtin__gpt_5_6_sol")
            self.assertEqual(sol["maker"], "openai")
            self.assertEqual(sol["context_window"], 272_000)
            self.assertEqual(
                sol["reasoning_levels"],
                ["low", "medium", "high", "xhigh", "max", "ultra"],
            )
            self.assertTrue(sol["supports_reasoning_summaries"])

            qwen = find_model("openrouter_openai_chat__qwen_qwen3_8_27b")
            self.assertEqual(qwen["maker"], "qwen")
            self.assertEqual(qwen["upstream_id"], "qwen/qwen3.8-27b")
            self.assertEqual(qwen["context_window"], 1_000_000)
            self.assertEqual(qwen["input_cost_per_million"], 0.4)
            self.assertEqual(qwen["output_cost_per_million"], 3.0)

            for profile_id in discover_profiles():
                resolved = resolve_profile(profile_id)
                for agent in resolved["agents"].values():
                    if agent["model"] == "zai_coding_responses__glm_5_3":
                        self.assertEqual(
                            resolved["models"][agent["model"]]["route"],
                            "zai_coding_responses",
                        )

    def test_opencode_zen_catalog_contracts(self) -> None:
        with RuntimeSandbox():
            catalog = validated_global_catalog()
            zen_models = {
                key: model
                for key, model in catalog["models"].items()
                if model.get("inventory") == "opencode-zen"
            }
            self.assertEqual(len(zen_models), 64)
            self.assertEqual(
                {
                    provider: sum(model["route"] == provider for model in zen_models.values())
                    for provider in (
                        "opencode_zen_responses",
                        "opencode_zen_openai_chat",
                        "opencode_zen_anthropic_messages",
                        "opencode_zen_google_catalog",
                    )
                },
                {
                    "opencode_zen_responses": 25,
                    "opencode_zen_openai_chat": 20,
                    "opencode_zen_anthropic_messages": 13,
                    "opencode_zen_google_catalog": 6,
                },
            )
            self.assertEqual(sum(model["agent_compatible"] for model in zen_models.values()), 58)
            self.assertEqual(sum(model["structured_output"] for model in zen_models.values()), 39)
            self.assertEqual(
                sum(model["reasoning_levels"] != ["none"] for model in zen_models.values()),
                46,
            )
            self.assertEqual(
                sum("input_cost_per_million" in model for model in zen_models.values()),
                50,
            )
            self.assertEqual(
                sum(
                    model["availability"].startswith("deprecated-") for model in zen_models.values()
                ),
                9,
            )

            sol = zen_models["opencode_zen_responses__gpt_5_6_sol"]
            self.assertEqual(sol["route"], "opencode_zen_responses")
            self.assertEqual(
                sol["reasoning_levels"],
                ["none", "low", "medium", "high", "xhigh", "max"],
            )
            self.assertEqual(sol["default_reasoning"], "none")
            self.assertNotIn("input_cost_per_million", sol)

            opus = zen_models["opencode_zen_anthropic_messages__claude_opus_5"]
            self.assertEqual(opus["route"], "opencode_zen_anthropic_messages")
            self.assertEqual(opus["reasoning_levels"], ["low", "medium", "high", "xhigh", "max"])
            self.assertEqual(opus["default_reasoning"], "medium")
            self.assertEqual(opus["input_cost_per_million"], 5.0)
            self.assertEqual(opus["output_cost_per_million"], 25.0)

            deepseek = zen_models["opencode_zen_openai_chat__deepseek_v4_flash"]
            self.assertEqual(deepseek["route"], "opencode_zen_openai_chat")
            self.assertEqual(deepseek["reasoning_levels"], ["none", "low", "high", "max"])
            self.assertEqual(deepseek["default_reasoning"], "none")
            self.assertNotIn("input_cost_per_million", deepseek)

            undocumented = zen_models["opencode_zen_openai_chat__deepseek_v4_flash_free"]
            self.assertEqual(undocumented["availability"], "live-undocumented")
            self.assertNotIn("input_cost_per_million", undocumented)

            contributor = zen_models["opencode_zen_responses__muse_spark_1_2_contributor_free"]
            self.assertEqual(contributor["input_cost_per_million"], 0.0)
            self.assertTrue(contributor["agent_compatible"])

            ox_alpha = zen_models["opencode_zen_openai_chat__x_preview_f_free"]
            self.assertEqual(ox_alpha["output_cost_per_million"], 0.0)
            self.assertTrue(ox_alpha["agent_compatible"])

            kimi = zen_models["opencode_zen_openai_chat__kimi_k2_5"]
            self.assertEqual(kimi["cached_input_cost_per_million"], 0.1)
            self.assertEqual(kimi["pricing_source"], "opencode-zen-docs-source")
            self.assertEqual(kimi["availability"], "deprecated-2026-08-05-live-listed")

            sonnet_4 = zen_models["opencode_zen_anthropic_messages__claude_sonnet_4"]
            self.assertEqual(sonnet_4["availability"], "deprecated-2026-06-15-live-listed")
            self.assertNotIn("input_cost_per_million", sonnet_4)

            gemini = zen_models["opencode_zen_google_catalog__gemini_3_7_flash"]
            self.assertEqual(gemini["route"], "opencode_zen_google_catalog")
            self.assertFalse(gemini["agent_compatible"])
            self.assertEqual(gemini["modalities"], ["text", "image", "video", "audio", "file"])
            self.assertEqual(gemini["reasoning_levels"], ["low", "medium", "high"])
            self.assertEqual(
                catalog["routes"]["opencode_zen_google_catalog"]["driver"], "catalog_only"
            )

    def test_common_inventory_snapshots_are_integrity_checked(self) -> None:
        snapshots = load_inventory_snapshots(ROOT / "config" / "inventory-snapshots")
        self.assertEqual(
            {snapshot["inventory"] for snapshot in snapshots},
            {
                "openai-codex",
                "opencode-go",
                "opencode-zen",
                "openrouter",
                "zai-api",
                "zai-coding-plan",
            },
        )
        self.assertEqual(
            {snapshot["inventory"]: snapshot["as_of"] for snapshot in snapshots},
            {
                "openai-codex": "2026-08-23",
                "opencode-go": "2026-08-23",
                "opencode-zen": "2026-08-23",
                "openrouter": "2026-08-23",
                "zai-api": "2026-08-16",
                "zai-coding-plan": "2026-08-16",
            },
        )
        for snapshot in snapshots:
            self.assertTrue(snapshot["models_sha256"])
        codex = next(snapshot for snapshot in snapshots if snapshot["inventory"] == "openai-codex")
        self.assertEqual(
            {record["catalog"]["capability_confidence"] for record in codex["models"].values()},
            {f"codex-{APP_SERVER_PROTOCOL_CODEX_VERSION}-baseline"},
        )
        self.assertLessEqual(
            len(
                route_catalog_key(
                    "a" * 40,
                    "provider/model-with-an-extremely-long-upstream-identifier" * 3,
                    set(),
                )
            ),
            64,
        )
        self.assertEqual(
            route_catalog_key("example_route", "Vendor/Model-X", set()),
            "example_route__vendor_model_x",
        )
        openrouter = next(
            snapshot for snapshot in snapshots if snapshot["inventory"] == "openrouter"
        )
        self.assertEqual(
            openrouter["discovery"]["endpoint_selections"],
            {
                "deepseek/deepseek-v4-pro": "parasail/fp8",
                "nvidia/nemotron-3-ultra-550b-a55b": "together",
            },
        )
        selected_policies = {
            record["catalog"]["upstream_id"]: record["catalog"].get("route_policy")
            for record in openrouter["models"].values()
            if record["catalog"]["upstream_id"] in openrouter["discovery"]["endpoint_selections"]
        }
        self.assertEqual(
            {key: value["only"][0] for key, value in selected_policies.items()},
            openrouter["discovery"]["endpoint_selections"],
        )
        self.assertTrue(
            all(
                policy["zdr"] and not policy["allow_fallbacks"]
                for policy in selected_policies.values()
            )
        )
        qwen = next(
            record
            for record in openrouter["models"].values()
            if record["catalog"]["upstream_id"] == "qwen/qwen3.8-27b"
        )
        self.assertEqual(
            qwen["evidence"]["architecture"]["input_modalities"],
            ["text", "image", "video"],
        )
        self.assertEqual(
            openrouter_catalog_pricing(qwen["evidence"]),
            {
                "input_cost_per_million": 0.4,
                "output_cost_per_million": 3.0,
                "cached_input_cost_per_million": 0.05,
            },
        )
        latest_alias = next(
            record
            for record in openrouter["models"].values()
            if record["catalog"]["upstream_id"] == "~deepseek/deepseek-v4-flash-latest"
        )
        self.assertEqual(latest_alias["catalog"]["maker"], "deepseek")
        tampered = deepcopy(openrouter)
        first_key, first = next(iter(tampered["models"].items()))
        first["catalog"]["context_window"] += 1
        with self.assertRaisesRegex(ValueError, "integrity mismatch"):
            validate_inventory_snapshot(tampered)
        wrong_namespace = deepcopy(openrouter)
        record = wrong_namespace["models"].pop(first_key)
        wrong_namespace["models"]["wrong_route__model"] = record
        wrong_namespace["models_sha256"] = stable_hash(wrong_namespace["models"])
        with self.assertRaisesRegex(ValueError, "exact route namespace 'openrouter_openai_chat'"):
            validate_inventory_snapshot(wrong_namespace)
        duplicate_binding = deepcopy(openrouter)
        duplicate_binding["models"]["openrouter_openai_chat__duplicate_binding"] = deepcopy(
            duplicate_binding["models"][first_key]
        )
        duplicate_binding["models_sha256"] = stable_hash(duplicate_binding["models"])
        with self.assertRaisesRegex(ValueError, "duplicate route/upstream binding"):
            validate_inventory_snapshot(duplicate_binding)
        self.assertEqual(
            openrouter_reasoning({"id": "example/omitted", "reasoning": {"mandatory": False}}),
            (["none"], "none"),
        )
        self.assertEqual(
            openrouter_reasoning(
                {
                    "id": "example/null",
                    "reasoning": {
                        "mandatory": False,
                        "default_enabled": True,
                        "supported_efforts": None,
                    },
                }
            ),
            (["none", "minimal", "low", "medium", "high", "xhigh", "max"], "medium"),
        )
        self.assertEqual(
            openrouter_reasoning(
                {
                    "id": "example/mandatory-null",
                    "reasoning": {"mandatory": True, "supported_efforts": None},
                }
            ),
            (["minimal", "low", "medium", "high", "xhigh", "max"], "medium"),
        )

        go = next(snapshot for snapshot in snapshots if snapshot["inventory"] == "opencode-go")
        qwen_go = next(
            record
            for record in go["models"].values()
            if record["catalog"]["upstream_id"] == "qwen3.6-plus"
        )
        self.assertEqual(qwen_go["catalog"]["route"], "opencode_go_anthropic_messages")
        self.assertEqual(
            qwen_go["evidence"]["protocol_resolution"],
            {
                "authority": "opencode-go-docs-source",
                "disagreement": True,
                "documented_npm": "@ai-sdk/anthropic",
                "models_dev_npm": "@ai-sdk/openai-compatible",
                "selected_npm": "@ai-sdk/anthropic",
            },
        )
        self.assertEqual(
            model_record_fingerprint(qwen_go)["endpoint_metadata"]["protocol_resolution"],
            qwen_go["evidence"]["protocol_resolution"],
        )

        qwen_docs = qwen_go["evidence"]["docs"]
        qwen_endpoint = qwen_docs["endpoint"]
        qwen_docs_document = "\n".join(
            [
                "## Endpoints",
                "",
                "| Model | Model ID | Endpoint | AI SDK Package |",
                "| --- | --- | --- | --- |",
                f"| {qwen_endpoint['name']} | {qwen_docs['id']} | "
                f"`{qwen_endpoint['url']}` | `{qwen_endpoint['npm']}` |",
                "",
                "## Usage limits",
                "",
                "| Model | Input | Output | Cached Read | Cached Write | Usage |",
                "| --- | --- | --- | --- | --- | --- |",
                *[
                    f"| {row['label']} | ${row['input']} | ${row['output']} | "
                    f"${row['cache_read']} | ${row['cache_write']} | ${row['usage']} |"
                    for row in qwen_docs["pricing"]
                ],
            ]
        )
        qwen_rebuilt = build_opencode_go_snapshot(
            {
                "object": "list",
                "data": [{**qwen_go["evidence"]["live"], "created": 1}],
            },
            {
                "opencode-go": {
                    "id": "opencode-go",
                    "api": "https://opencode.ai/zen/go/v1",
                    "env": ["OPENCODE_API_KEY"],
                    "models": {
                        "qwen3.6-plus": qwen_go["evidence"]["models_dev"],
                    },
                }
            },
            qwen_docs_document,
            as_of=go["as_of"],
            retrieved_at=go["captures"][0]["retrieved_at"],
            listing_sha256=go["captures"][0]["response_sha256"],
            models_dev_sha256=go["captures"][1]["response_sha256"],
            docs_sha256=go["captures"][2]["response_sha256"],
            listing_url=go["sources"]["opencode-go-models"],
            models_dev_url=go["sources"]["models-dev-opencode-go"],
            docs_url=go["sources"]["opencode-go-docs-source"],
        )
        rebuilt_qwen = next(iter(qwen_rebuilt["models"].values()))
        self.assertEqual(rebuilt_qwen["catalog"]["route"], "opencode_go_anthropic_messages")
        self.assertTrue(rebuilt_qwen["evidence"]["protocol_resolution"]["disagreement"])

        zen = next(snapshot for snapshot in snapshots if snapshot["inventory"] == "opencode-zen")
        self.assertTrue(
            all("created" not in record["evidence"]["live"] for record in zen["models"].values())
        )
        listing = {
            "object": "list",
            "data": [
                {**record["evidence"]["live"], "created": 1} for record in zen["models"].values()
            ],
        }
        models_dev: dict[str, Any] = {
            "opencode": {
                "id": "opencode",
                "api": "https://opencode.ai/zen/v1",
                "env": ["OPENCODE_API_KEY"],
                "models": {
                    record["evidence"]["models_dev"]["id"]: record["evidence"]["models_dev"]
                    for record in zen["models"].values()
                },
            }
        }
        endpoint_rows: list[str] = []
        pricing_rows: list[str] = []
        deprecation_rows: list[str] = []
        for record in zen["models"].values():
            docs = record["evidence"]["docs"]
            endpoint = docs.get("endpoint")
            if endpoint:
                endpoint_rows.append(
                    f"| {endpoint['name']} | {docs['id']} | `{endpoint['url']}` | "
                    f"`{endpoint['npm']}` |"
                )
            for price_row in docs.get("pricing", []):
                values = []
                for field in ("input", "output", "cache_read", "cache_write"):
                    value = price_row.get(field)
                    values.append("-" if value is None else f"${value}")
                pricing_rows.append(
                    f"| {price_row['label']} | {values[0]} | {values[1]} | "
                    f"{values[2]} | {values[3]} |"
                )
            if docs.get("deprecation_date"):
                parsed = date.fromisoformat(docs["deprecation_date"])
                display_date = f"{parsed.strftime('%B')} {parsed.day}, {parsed.year}"
                name = endpoint["name"] if endpoint else record["evidence"]["models_dev"]["name"]
                deprecation_rows.append(f"| {name} | {display_date} |")
        docs_document = "\n".join(
            [
                "## Endpoints",
                "",
                "| Model | Model ID | Endpoint | AI SDK Package |",
                "| --- | --- | --- | --- |",
                *endpoint_rows,
                "",
                "## Pricing",
                "",
                "| Model | Input | Output | Cached Read | Cached Write |",
                "| --- | --- | --- | --- | --- |",
                *pricing_rows,
                "",
                "### Deprecated models",
                "",
                "| Model | Deprecation date |",
                "| --- | --- |",
                *deprecation_rows,
            ]
        )
        rebuilt = build_opencode_zen_snapshot(
            listing,
            models_dev,
            docs_document,
            as_of=zen["as_of"],
            retrieved_at=zen["captures"][0]["retrieved_at"],
            listing_sha256=zen["captures"][0]["response_sha256"],
            models_dev_sha256=zen["captures"][1]["response_sha256"],
            docs_sha256=zen["captures"][2]["response_sha256"],
            listing_url=zen["sources"]["opencode-zen-models"],
            models_dev_url=zen["sources"]["models-dev-opencode-zen"],
            docs_url=zen["sources"]["opencode-zen-docs-source"],
        )
        self.assertEqual(rebuilt, zen)
        undocumented_docs = "\n".join(
            line for line in docs_document.splitlines() if "DeepSeek V4 Flash Free" not in line
        )
        undocumented = build_opencode_zen_snapshot(
            listing,
            models_dev,
            undocumented_docs,
            as_of=zen["as_of"],
            retrieved_at=zen["captures"][0]["retrieved_at"],
            listing_sha256=zen["captures"][0]["response_sha256"],
            models_dev_sha256=zen["captures"][1]["response_sha256"],
            docs_sha256=zen["captures"][2]["response_sha256"],
            listing_url=zen["sources"]["opencode-zen-models"],
            models_dev_url=zen["sources"]["models-dev-opencode-zen"],
            docs_url=zen["sources"]["opencode-zen-docs-source"],
        )
        undocumented_record = next(
            record
            for record in undocumented["models"].values()
            if record["catalog"]["upstream_id"] == "deepseek-v4-flash-free"
        )
        self.assertEqual(undocumented_record["catalog"]["availability"], "live-undocumented")
        self.assertTrue(undocumented_record["catalog"]["agent_compatible"])
        self.assertNotIn("input_cost_per_million", undocumented_record["catalog"])
        self.assertFalse(undocumented_record["evidence"]["docs"]["documented_endpoint"])
        conflicting_metadata = deepcopy(models_dev)
        conflicting_metadata["opencode"]["models"]["gpt-5.6-sol"]["provider"]["npm"] = (
            "@ai-sdk/openai-compatible"
        )
        with self.assertRaisesRegex(ValueError, "protocol mismatch"):
            build_opencode_zen_snapshot(
                listing,
                conflicting_metadata,
                docs_document,
                as_of=zen["as_of"],
                retrieved_at=zen["captures"][0]["retrieved_at"],
                listing_sha256=zen["captures"][0]["response_sha256"],
                models_dev_sha256=zen["captures"][1]["response_sha256"],
                docs_sha256=zen["captures"][2]["response_sha256"],
                listing_url=zen["sources"]["opencode-zen-models"],
                models_dev_url=zen["sources"]["models-dev-opencode-zen"],
                docs_url=zen["sources"]["opencode-zen-docs-source"],
            )
        self.assertEqual(
            opencode_zen_reasoning(
                {
                    "id": "toggle-and-effort",
                    "reasoning": True,
                    "reasoning_options": [
                        {"type": "toggle"},
                        {"type": "effort", "values": ["low", "high", "max"]},
                    ],
                }
            ),
            (["none", "low", "high", "max"], "none"),
        )
        self.assertEqual(
            opencode_zen_reasoning(
                {
                    "id": "budget-only",
                    "reasoning": True,
                    "reasoning_options": [{"type": "budget_tokens"}],
                }
            ),
            (["none"], "none"),
        )
        self.assertEqual(
            opencode_zen_catalog_pricing(
                {
                    "id": "tiered",
                    "pricing": [
                        {"label": "Tiered (≤ 200K tokens)", "input": 1, "output": 2},
                        {"label": "Tiered (> 200K tokens)", "input": 2, "output": 3},
                    ],
                }
            ),
            {},
        )
        self.assertEqual(
            opencode_zen_catalog_pricing(
                {
                    "id": "dash-valued",
                    "pricing": [{"label": "No scalar rate", "input": None, "output": None}],
                }
            ),
            {},
        )
        with self.assertRaisesRegex(ValueError, "output is required"):
            opencode_zen_catalog_pricing(
                {
                    "id": "partial-rate",
                    "pricing": [{"label": "Partial", "input": 1.0, "output": None}],
                }
            )
        go_docs = _opencode_go_docs(
            """## Endpoints

| Model | Model ID | Endpoint | AI SDK Package |
| --- | --- | --- | --- |
| DeepSeek V4 Flash | deepseek-v4-flash | https://opencode.ai/zen/go/v1/chat/completions | @ai-sdk/openai-compatible |

## Usage limits

| Model | Input | Output | Cached Read | Cached Write | Usage |
| --- | --- | --- | --- | --- | --- |
| DeepSeek V4 Flash (Off-Peak) | $0.22 | $0.66 | $0.007 | - | - |
| DeepSeek V4 Flash (Peak) | $0.44 | $1.32 | $0.014 | - | - |
"""
        )
        self.assertEqual(
            [row["label"] for row in go_docs["deepseek-v4-flash"]["pricing"]],
            ["DeepSeek V4 Flash (Off-Peak)", "DeepSeek V4 Flash (Peak)"],
        )
        self.assertEqual(
            opencode_zen_catalog_pricing(go_docs["deepseek-v4-flash"]),
            {},
        )

    def test_zai_coding_plan_responses_compiles_to_dedicated_codex_endpoint(self) -> None:
        with RuntimeSandbox():
            catalog = validated_global_catalog()
            routes, route_ids, gateway_hash = _switchyard_routes(
                {
                    "routes": {"zai_coding_responses": catalog["routes"]["zai_coding_responses"]},
                    "models": {
                        "zai_coding_responses__glm_5_3": catalog["models"][
                            "zai_coding_responses__glm_5_3"
                        ]
                    },
                }
            )
        self.assertIsNotNone(gateway_hash)
        self.assertEqual(
            routes["llm_clients"]["zai_coding_responses"],
            {
                "format": "openai_responses",
                "base_url": "https://api.z.ai/api/v1",
                "api_key_env": "ZAI_CODING_API_KEY",
                "max_retries": 1,
            },
        )
        self.assertEqual(
            routes["targets"]["zai_coding_responses__glm_5_3"],
            {"id": "glm-5.3", "llm_client": "zai_coding_responses"},
        )
        self.assertIn("zai_coding_responses__glm_5_3", route_ids)

    def test_local_inventory_report_detects_snapshot_catalog_drift(self) -> None:
        with RuntimeSandbox():
            tampered = deepcopy(validated_global_catalog())
            tampered["models"]["openrouter_openai_chat__qwen_qwen3_8_27b"]["context_window"] -= 1
            with mock.patch("mmo_catalog.validated_global_catalog", return_value=tampered):
                report = local_inventory_report()
        self.assertFalse(report["passed"], report)
        self.assertEqual(
            report["inventories"]["openrouter"]["catalog_record_mismatches"],
            ["openrouter_openai_chat__qwen_qwen3_8_27b"],
        )

    def test_local_inventory_report_requires_complete_snapshot_discovery_metadata(self) -> None:
        with RuntimeSandbox():
            snapshots = load_inventory_snapshots(ROOT / "config" / "inventory-snapshots")
            tampered = deepcopy(snapshots)
            openrouter = next(
                snapshot for snapshot in tampered if snapshot["inventory"] == "openrouter"
            )
            del openrouter["discovery"]["endpoint"]
            with mock.patch("mmo_catalog.load_inventory_snapshots", return_value=tampered):
                report = local_inventory_report()
        self.assertFalse(report["passed"], report)
        self.assertFalse(report["inventories"]["openrouter"]["snapshot_ok"])

    def test_openrouter_compiles_to_exact_switchyard_client_contract(self) -> None:
        with RuntimeSandbox():
            catalog = validated_global_catalog()
            routes, route_ids, gateway_hash = _switchyard_routes(
                {
                    "routes": {
                        "openrouter_openai_chat": catalog["routes"]["openrouter_openai_chat"]
                    },
                    "models": {
                        "openrouter_openai_chat__qwen_qwen3_8_27b": catalog["models"][
                            "openrouter_openai_chat__qwen_qwen3_8_27b"
                        ]
                    },
                }
            )
        self.assertIsNotNone(gateway_hash)
        self.assertEqual(
            routes["llm_clients"]["openrouter_openai_chat"],
            {
                "format": "openai_chat",
                "base_url": "https://openrouter.ai/api/v1",
                "api_key_env": "OPENROUTER_API_KEY",
                "extra_headers": {"X-OpenRouter-Metadata": "enabled"},
                "max_retries": 1,
            },
        )
        self.assertEqual(
            routes["targets"]["openrouter_openai_chat__qwen_qwen3_8_27b"],
            {
                "id": "qwen/qwen3.8-27b",
                "llm_client": "openrouter_openai_chat",
            },
        )
        self.assertIn("openrouter_openai_chat__qwen_qwen3_8_27b", route_ids)

    def test_opencode_zen_compiles_only_supported_switchyard_clients(self) -> None:
        model_keys = (
            "opencode_zen_responses__gpt_5_6_sol",
            "opencode_zen_openai_chat__deepseek_v4_flash",
            "opencode_zen_anthropic_messages__claude_opus_5",
            "opencode_zen_google_catalog__gemini_3_7_flash",
        )
        route_keys = (
            "opencode_zen_responses",
            "opencode_zen_openai_chat",
            "opencode_zen_anthropic_messages",
            "opencode_zen_google_catalog",
        )
        with RuntimeSandbox():
            catalog = validated_global_catalog()
            routes, route_ids, gateway_hash = _switchyard_routes(
                {
                    "routes": {key: catalog["routes"][key] for key in route_keys},
                    "models": {key: catalog["models"][key] for key in model_keys},
                }
            )
        self.assertIsNotNone(gateway_hash)
        for route, format_name in (
            ("opencode_zen_responses", "openai_responses"),
            ("opencode_zen_openai_chat", "openai_chat"),
            ("opencode_zen_anthropic_messages", "anthropic_messages"),
        ):
            self.assertEqual(
                routes["llm_clients"][route],
                {
                    "format": format_name,
                    "base_url": "https://opencode.ai/zen/v1",
                    "api_key_env": "OPENCODE_API_KEY",
                    "max_retries": 1,
                },
            )
        self.assertNotIn("opencode_zen_google_catalog", routes["llm_clients"])
        self.assertEqual(
            set(routes["targets"]),
            {
                "opencode_zen_responses__gpt_5_6_sol",
                "opencode_zen_openai_chat__deepseek_v4_flash",
                "opencode_zen_anthropic_messages__claude_opus_5",
            },
        )
        self.assertEqual(set(route_ids), set(routes["targets"]))

    def test_opencode_zen_discovery_uses_public_listing_without_credentials(self) -> None:
        response = {
            "object": "list",
            "data": [
                {"id": "gpt-5.6-sol", "object": "model"},
                {"id": "claude-opus-5", "object": "model"},
            ],
        }
        with mock.patch("mmo_catalog._http_json", return_value=response) as request:
            result = discover_opencode_zen(
                url="https://example.invalid/models",
                timeout=1.0,
            )
        self.assertTrue(result["passed"], result)
        self.assertEqual(result["models"], ["claude-opus-5", "gpt-5.6-sol"])
        self.assertNotIn("fake-opencode", json.dumps(result))
        self.assertEqual(request.call_args.args, ("https://example.invalid/models", 1.0))
        self.assertEqual(request.call_args.kwargs, {})

    def test_zai_authenticated_discovery_without_credential_leakage(self) -> None:
        with RuntimeSandbox():
            with mock.patch(
                "mmo_catalog._http_json",
                return_value={"data": [{"id": "glm-5.3"}, {"id": "glm-5-turbo"}]},
            ) as request:
                result = discover_zai(
                    "zai-coding-plan",
                    url="https://example.invalid/models",
                    api_key="explicit-zai",
                    timeout=1.0,
                )
            self.assertTrue(result["passed"], result)
            self.assertEqual(result["models"], ["glm-5-turbo", "glm-5.3"])
            self.assertNotIn("fake-zai", json.dumps(result))
            self.assertEqual(
                request.call_args.kwargs["headers"]["Authorization"],
                "Bearer explicit-zai",
            )
            with mock.patch("mmo_catalog._http_json") as custom_request:
                skipped = discover_zai(
                    "zai-coding-plan",
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertTrue(skipped["skipped"], skipped)
            self.assertFalse(skipped["authenticated_request"])
            custom_request.assert_not_called()

            with mock.patch(
                "mmo_catalog._http_json",
                return_value={"data": [{"id": "glm-5.3"}]},
            ) as official_request:
                official = discover_zai("zai-coding-plan", timeout=1.0)
            self.assertTrue(official["passed"], official)
            self.assertEqual(
                official_request.call_args.kwargs["headers"]["Authorization"],
                "Bearer fake-zai-coding",
            )

    def test_openrouter_discovery_uses_optional_credential_without_leakage(self) -> None:
        with RuntimeSandbox() as box:
            response = {
                "data": [{"id": "openai/gpt-5.6-sol"}, {"id": "openrouter/auto"}],
                "links": {"next": None},
                "total_count": 2,
            }
            with mock.patch("mmo_catalog._http_json", return_value=response) as request:
                result = discover_openrouter(timeout=1.0)
            self.assertTrue(result["passed"], result)
            self.assertTrue(result["authenticated_request"])
            self.assertEqual(result["models"], ["openai/gpt-5.6-sol", "openrouter/auto"])
            self.assertNotIn("fake-openrouter", json.dumps(result))
            self.assertEqual(
                request.call_args.kwargs["headers"]["Authorization"],
                "Bearer fake-openrouter",
            )

            with mock.patch("mmo_catalog._http_json", return_value=response) as custom_request:
                custom = discover_openrouter(
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertTrue(custom["passed"], custom)
            self.assertFalse(custom["authenticated_request"])
            self.assertIsNone(custom_request.call_args.kwargs["headers"])

            os.environ.pop("OPENROUTER_API_KEY")
            (box.config / "credentials.env").write_text(
                "ZAI_CODING_API_KEY=fake-zai-coding\n", encoding="utf-8"
            )
            with mock.patch("mmo_catalog._http_json", return_value=response) as public_request:
                public = discover_openrouter(
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertTrue(public["passed"], public)
            self.assertFalse(public["authenticated_request"])
            self.assertIsNone(public_request.call_args.kwargs["headers"])

            with (
                mock.patch("mmo_catalog._catalog_credentials") as credentials,
                mock.patch("mmo_catalog._http_json", return_value=response) as blank_request,
            ):
                explicit_public = discover_openrouter(api_key="", timeout=1.0)
            self.assertTrue(explicit_public["passed"], explicit_public)
            self.assertFalse(explicit_public["authenticated_request"])
            credentials.assert_not_called()
            self.assertIsNone(blank_request.call_args.kwargs["headers"])

            with (
                mock.patch("mmo_catalog._catalog_credentials") as credentials,
                mock.patch("mmo_catalog._http_json", return_value=response),
            ):
                custom_without_key = discover_openrouter(
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertTrue(custom_without_key["passed"], custom_without_key)
            credentials.assert_not_called()

            with mock.patch(
                "mmo_catalog._http_json",
                return_value={
                    "data": [{"id": "openrouter/auto"}],
                    "links": {"next": "https://example.invalid/models?offset=1"},
                    "total_count": 2,
                },
            ):
                incomplete = discover_openrouter(
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertFalse(incomplete["passed"], incomplete)
            self.assertIn("incomplete", incomplete["error"])

            with mock.patch(
                "mmo_catalog._http_json",
                return_value={"data": [{"id": "openrouter/auto"}]},
            ):
                missing_pagination = discover_openrouter(
                    url="https://example.invalid/models",
                    timeout=1.0,
                )
            self.assertFalse(missing_pagination["passed"], missing_pagination)
            self.assertIn("pagination metadata", missing_pagination["error"])

    def test_catalog_http_user_agent_tracks_the_package_version(self) -> None:
        response = mock.MagicMock()
        response.__enter__.return_value.read.return_value = b"{}"
        opener = mock.Mock()
        opener.open.return_value = response
        with mock.patch("mmo_catalog.urllib.request.build_opener", return_value=opener):
            self.assertEqual(_http_bytes("https://example.invalid/models", 1), b"{}")
        request = opener.open.call_args.args[0]
        self.assertEqual(request.get_header("User-agent"), "codex-mmo/8.0.0 catalog-discovery")

    def test_remote_verification_compares_full_inventory_fingerprints(self) -> None:
        snapshots = {
            snapshot["inventory"]: snapshot
            for snapshot in load_inventory_snapshots(ROOT / "config" / "inventory-snapshots")
        }

        def observed(
            inventory: str,
            _expected: dict[str, Any],
            **_kwargs: Any,
        ) -> dict[str, Any]:
            return deepcopy(snapshots[inventory])

        coding_ids = sorted(
            record["catalog"]["upstream_id"]
            for record in snapshots["zai-coding-plan"]["models"].values()
        )
        with (
            RuntimeSandbox(),
            mock.patch("mmo_catalog._build_observed_public_snapshot", side_effect=observed),
            mock.patch(
                "mmo_catalog.discover_zai",
                return_value={"passed": True, "skipped": False, "models": coding_ids},
            ),
            mock.patch(
                "mmo_catalog._verify_source_captures",
                return_value={"passed": True, "captures": []},
            ),
        ):
            report = verify_catalog(remote=True)
        # The reviewed Go snapshot deliberately remains a release blocker until
        # hy3-preview has a complete capability fingerprint.
        self.assertFalse(report["passed"], report)
        go = report["remote"]["opencode-go"]["comparison"]
        self.assertFalse(go["exact"])
        self.assertEqual(go["incomplete_evidence"][0]["upstream_id"], "hy3-preview")
        for inventory in ("opencode-zen", "openrouter"):
            comparison = report["remote"][inventory]["comparison"]
            self.assertTrue(comparison["exact"], comparison)
            self.assertEqual(comparison["fingerprint_mismatches"], [])
            self.assertEqual(comparison["incomplete_evidence"], [])
        self.assertTrue(report["remote"]["zai-coding-plan"]["full_fingerprint_verified"])

    def test_codex_runtime_discovery(self) -> None:
        with RuntimeSandbox() as box:
            result = discover_codex(binary=os.environ["MMO_CODEX_BIN"], home=box.base_codex_home)
            self.assertTrue(result["passed"], result)
            self.assertEqual(
                set(result["models"]),
                {
                    "codex-auto-review",
                    "gpt-5.2",
                    "gpt-5.3-codex-spark",
                    "gpt-5.4",
                    "gpt-5.4-mini",
                    "gpt-5.5",
                    "gpt-5.6-luna",
                    "gpt-5.6-sol",
                    "gpt-5.6-terra",
                },
            )
            self.assertEqual(build_codex_discovery_overlay(result)["models"], {})

            snapshot = next(
                item
                for item in load_inventory_snapshots(ROOT / "config" / "inventory-snapshots")
                if item["inventory"] == "openai-codex"
            )
            metadata: dict[str, Any] = {}
            for record in snapshot["models"].values():
                runtime = record["evidence"]["codex_runtime"]
                metadata[runtime["slug"]] = {
                    "slug": runtime["slug"],
                    "comp_hash": runtime["comp_hash"],
                    "context_window": runtime["context_window"],
                    "input_modalities": runtime["input_modalities"],
                    "supported_reasoning_levels": [
                        {"effort": level} for level in runtime["reasoning"]["levels"]
                    ],
                    "default_reasoning_level": runtime["reasoning"]["default"],
                    "supports_reasoning_summaries": runtime["reasoning"]["summaries"],
                    "supports_parallel_tool_calls": runtime["tools"]["parallel_tool_calls"],
                    "shell_type": runtime["tools"]["shell_type"],
                    "tool_mode": runtime["tools"]["tool_mode"],
                    "apply_patch_tool_type": runtime["tools"]["apply_patch_tool_type"],
                    "supports_search_tool": runtime["tools"]["search"],
                    "visibility": runtime["visibility"],
                    "supported_in_api": runtime["supported_in_api"],
                    "multi_agent_version": runtime["multi_agent_version"],
                    "service_tiers": runtime["service_tiers"],
                }
            exact_discovery = {**result, "metadata": metadata}
            with mock.patch("mmo_catalog.discover_codex", return_value=exact_discovery):
                report = verify_catalog(
                    include_codex=True,
                    codex_binary=os.environ["MMO_CODEX_BIN"],
                    codex_home=box.base_codex_home,
                )
            codex = report["remote"]["openai-codex"]
            self.assertTrue(report["passed"], report)
            self.assertEqual(codex["ignored_internal_models"], ["codex-auto-review"])
            self.assertEqual(
                codex["known_deprecated_models_observed"],
                ["gpt-5.2", "gpt-5.3-codex-spark"],
            )
            self.assertEqual(codex["comparison"]["unknown_to_catalog"], [])
            self.assertTrue(codex["full_fingerprint_verified"])

    def test_requested_codex_verification_fails_when_no_runtime_source_runs(self) -> None:
        with RuntimeSandbox() as box:
            empty_home = box.root / "empty codex home"
            empty_home.mkdir()
            report = verify_catalog(
                include_codex=True,
                codex_binary=str(box.root / "missing-codex"),
                codex_home=empty_home,
            )
            self.assertFalse(report["passed"], report)
            codex = report["remote"]["openai-codex"]
            self.assertFalse(codex["passed"])
            self.assertFalse(codex["runtime_source_ok"])

    def test_codex_overlay_install_requires_a_live_runtime_source(self) -> None:
        report = {
            "passed": False,
            "remote": {
                "openai-codex": {
                    "passed": True,
                    "runtime_source_ok": False,
                    "models": [],
                    "metadata": {},
                }
            },
        }
        with (
            RuntimeSandbox(),
            mock.patch("mmo_catalog.verify_catalog", return_value=report),
            self.assertRaisesRegex(RuntimeError, "live model catalog"),
        ):
            refresh_discovery(include_codex=True, install_codex_overlay=True)

    def test_all_profiles_resolve_compile_and_are_content_addressed(self) -> None:
        with RuntimeSandbox():
            profiles = discover_profiles()
            self.assertEqual(
                set(profiles),
                {
                    "adaptive-engineering",
                    "access-efficient-escalation-lab",
                    "bounded-research-organization-lab",
                    "codex-harness-team",
                    "competing-implementations-lab",
                    "contract-first-refactoring",
                    "high-confidence-debugging",
                    "incident-hypothesis-triage",
                    "research-backed-engineering",
                    "route-resilience-lab",
                    "secure-change",
                    "visual-engineering",
                },
            )
            modes = set()
            for profile_id in profiles:
                resolved = resolve_profile(profile_id)
                self.assertEqual(resolved["schema_version"], MMO_SCHEMA_VERSION)
                for agent in resolved["agents"].values():
                    self.assertNotIn("progress_interval_seconds", agent)
                    self.assertNotIn("default_timeout_seconds", agent)
                    self.assertNotIn("max_timeout_seconds", agent)
                    self.assertIn(agent["execution_mode"], {"turn", "goal"})
                    if agent["execution_mode"] == "goal":
                        self.assertLessEqual(
                            agent["goal_token_budget"], agent["max_goal_token_budget"]
                        )
                    else:
                        self.assertIsNone(agent["goal_token_budget"])
                        self.assertIsNone(agent["max_goal_token_budget"])
                    self.assertNotIn("execution_policy", agent)
                    self.assertGreaterEqual(agent["finalization_grace_seconds"], 30)
                modes.add(resolved["coordination"]["orchestration"])
                first = compile_profile(profile_id)
                second = compile_profile(profile_id)
                self.assertEqual(
                    first["manifest"]["snapshot_hash"], second["manifest"]["snapshot_hash"]
                )
                self.assertEqual(first["manifest"]["profile_id"], profile_id)
                directory = Path(first["directory"])
                self.assertFalse(bool(directory.stat().st_mode & 0o200))
                if first["manifest"]["gateway_required"]:
                    self.assertTrue((directory / "routes.toml").is_file())
                    routes = read_toml(directory / "routes.toml")
                    self.assertEqual(set(routes["routes"]), set(first["manifest"]["route_ids"]))
            self.assertEqual(modes, {"mcp", "hybrid"})
            self.assertTrue(
                any(
                    "native" in agent["backends"]
                    for profile_id in profiles
                    for agent in resolve_profile(profile_id)["agents"].values()
                )
            )

    def test_retired_wait_modes_and_native_strict_contract_claims_are_rejected(self) -> None:
        with RuntimeSandbox() as box:
            barrier = box.root / "barrier-profile"
            shutil.copytree(ROOT / "profiles" / "adaptive-engineering", barrier)
            path = barrier / "profile.toml"
            data = read_toml(path)
            data["id"] = "barrier-profile"
            data["coordination"]["wait_policy"] = "barrier"
            path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "coordination.wait_policy"):
                resolve_profile(barrier)

            native = box.root / "native-strict-profile"
            shutil.copytree(ROOT / "profiles" / "codex-harness-team", native)
            path = native / "profile.toml"
            data = read_toml(path)
            data["id"] = "native-strict-profile"
            data["agents"]["invariant_designer"]["contract_enforcement"] = "strict"
            path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "native-only agents cannot claim strict"):
                resolve_profile(native)

            noisy_writer_limit = box.root / "noisy-writer-limit"
            shutil.copytree(ROOT / "profiles" / "route-resilience-lab", noisy_writer_limit)
            path = noisy_writer_limit / "profile.toml"
            data = read_toml(path)
            data["id"] = "noisy-writer-limit"
            data["coordination"]["max_active_writers"] = 1
            path.write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, r"writable MCP capacity \(0\)"):
                resolve_profile(noisy_writer_limit)

    def test_goal_lifecycle_and_app_server_control_graph_are_bounded(self) -> None:
        with RuntimeSandbox() as box:
            valid = box.root / "valid-app-server-profile"
            shutil.copytree(ROOT / "profiles" / "adaptive-engineering", valid)
            path = valid / "profile.toml"
            data = read_toml(path)
            data["id"] = "valid-app-server-profile"
            worker = data["agents"]["implementation_specialist"]
            worker["execution_mode"] = "goal"
            worker["goal_token_budget"] = 100_000
            worker["max_goal_token_budget"] = 180_000
            worker["finalization_grace_seconds"] = 30
            path.write_text(toml_dumps(data), encoding="utf-8")
            resolved = resolve_profile(valid)
            self.assertEqual(
                resolved["agents"]["implementation_specialist"]["max_goal_token_budget"],
                180_000,
            )
            self.assertIn(
                "implementation_specialist",
                resolved["agents"]["orchestrator"]["controls"],
            )
            self.assertIn("repo_scout", resolved["agents"]["orchestrator"]["controls"])

            cases = (
                (
                    "inverted-goal-budget",
                    lambda profile: profile["agents"]["implementation_specialist"].update(
                        goal_token_budget=181_000,
                        max_goal_token_budget=180_000,
                    ),
                    "goal_token_budget exceeds max_goal_token_budget",
                ),
                (
                    "turn-with-goal-budget",
                    lambda profile: profile["agents"]["implementation_specialist"].update(
                        execution_mode="turn"
                    ),
                    "turn execution cannot declare goal token budgets",
                ),
                (
                    "native-goal-mode",
                    lambda profile: profile["agents"]["repo_scout"].update(
                        execution_mode="goal",
                        goal_token_budget=100_000,
                        max_goal_token_budget=180_000,
                    ),
                    "native participants must use turn execution",
                ),
                (
                    "unknown-control-action",
                    lambda profile: profile["agents"]["orchestrator"]["controls"][
                        "repo_scout"
                    ].update(actions=["inspect", "teleport"]),
                    "contains unsupported actions",
                ),
                (
                    "self-control-target",
                    lambda profile: profile["agents"]["orchestrator"]["controls"].update(
                        orchestrator={"actions": ["inspect"]}
                    ),
                    "an agent cannot grant itself controls",
                ),
                (
                    "root-fork-control-target",
                    lambda profile: profile["agents"]["implementation_specialist"][
                        "controls"
                    ].update(orchestrator={"actions": ["fork"]}),
                    "immutable root run cannot be a fork control target",
                ),
            )
            for profile_id, mutate, message in cases:
                with self.subTest(profile_id=profile_id):
                    profile = box.root / profile_id
                    shutil.copytree(ROOT / "profiles" / "adaptive-engineering", profile)
                    profile_path = profile / "profile.toml"
                    profile_data = read_toml(profile_path)
                    profile_data["id"] = profile_id
                    mutate(profile_data)
                    profile_path.write_text(toml_dumps(profile_data), encoding="utf-8")
                    with self.assertRaisesRegex(ValueError, message):
                        resolve_profile(profile)

    def test_retired_bundled_profile_ids_have_no_compatibility_aliases(self) -> None:
        retired = {
            "actor-critic",
            "codex-glm-deepseek",
            "codex-ultra-native",
            "glm-deepseek-qwen",
            "local-first",
            "proposal-debate-judge",
            "security-review-council",
            "single-model-parallel",
            "three-expert-council",
            "vision-code-verifier",
            "architecture-decision-council",
            "cost-aware-escalation",
            "reproduce-fix-verify",
            "single-model-native",
        }
        with RuntimeSandbox():
            for profile_id in retired:
                with self.subTest(profile_id=profile_id):
                    with self.assertRaises(FileNotFoundError):
                        resolve_profile(profile_id)

    def test_low_trust_policy_is_mechanical(self) -> None:
        with RuntimeSandbox():
            resolved = resolve_profile("access-efficient-escalation-lab")
            qwen = resolved["agents"]["literal_scout"]
            self.assertEqual(qwen["trust"], "low")
            self.assertEqual(qwen["permissions"], "read-only")
            self.assertEqual(qwen["verification"], "always")
            self.assertEqual(qwen["backends"], ["mcp"])
            self.assertEqual(qwen["max_active"], 1)
            forbidden = {"architecture", "implement", "debug", "review"}
            self.assertFalse(forbidden & set(qwen["allowed_task_kinds"]))

    def test_low_trust_and_multimodal_capabilities_reject_unsafe_profiles(self) -> None:
        with RuntimeSandbox() as box:
            low = box.root / "unsafe-low"
            import shutil

            shutil.copytree(ROOT / "profiles" / "access-efficient-escalation-lab", low)
            data = read_toml(low / "profile.toml")
            data["id"] = "unsafe-low"
            data["agents"]["literal_scout"]["max_active"] = 2
            (low / "profile.toml").write_text(toml_dumps(data), encoding="utf-8")
            with self.assertRaisesRegex(ValueError, "low-trust agents require max_active=1"):
                resolve_profile(low)

            lossy = box.root / "lossy-vision"
            shutil.copytree(ROOT / "profiles" / "visual-engineering", lossy)
            data = read_toml(lossy / "profile.toml")
            data["id"] = "lossy-vision"
            data["agents"]["visual_analyst"]["requires_tool_images"] = True
            (lossy / "profile.toml").write_text(toml_dumps(data), encoding="utf-8")
            (lossy / "catalog.toml").write_text(
                toml_dumps(
                    {
                        "schema_version": MMO_SCHEMA_VERSION,
                        "routes": {
                            "codex_chatgpt_builtin": {
                                "preserves_tool_media": False,
                                "tool_result_modalities": ["text"],
                            }
                        },
                    }
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "cannot consume image-bearing tool results"):
                resolve_profile(lossy)

    def test_snapshot_payload_is_cryptographically_verified(self) -> None:
        with RuntimeSandbox():
            snapshot = compile_profile("codex-harness-team")
            directory = Path(snapshot["directory"])
            target = directory / "instructions" / "invariant_designer.md"
            target.chmod(0o600)
            target.write_text(target.read_text(encoding="utf-8") + "tampered\n", encoding="utf-8")
            with self.assertRaisesRegex(RuntimeError, "snapshot payload integrity failure"):
                load_snapshot(snapshot["manifest"]["snapshot_hash"])

    def test_profile_guidance_is_deterministic_and_snapshot_owned(self) -> None:
        with RuntimeSandbox():
            resolved = resolve_profile("codex-harness-team")
            round_tripped = json.loads(json.dumps(resolved, sort_keys=True))
            self.assertEqual(compiled_guidance(resolved), compiled_guidance(round_tripped))
            self.assertEqual(
                stable_hash(_snapshot_fingerprint(resolved)),
                stable_hash(_snapshot_fingerprint(round_tripped)),
            )

            snapshot = compile_profile("codex-harness-team")
            directory = Path(snapshot["directory"])
            manifest = snapshot["manifest"]
            self.assertEqual(manifest["guidance_schema_version"], MMO_SCHEMA_VERSION)
            self.assertEqual(manifest["coordination_capable_agents"], ["integrator"])
            self.assertIn(PROFILE_SKILL_RELATIVE_PATH, manifest["payload_files"])
            for agent_id in snapshot["resolved"]["agents"]:
                self.assertIn(agent_guidance_relative_path(agent_id), manifest["payload_files"])

            skill = (directory / PROFILE_SKILL_RELATIVE_PATH).read_text(encoding="utf-8")
            self.assertLess(len(skill.splitlines()), 500)
            self.assertTrue(skill.startswith(f"---\nname: {PROFILE_SKILL_NAME}\n"))
            self.assertIn("Within at most three substantive task calls", skill)
            self.assertIn("`agents_spawn`", skill)
            self.assertIn("`agents_wait`", skill)
            self.assertIn("until `next_cursor` is null", skill)
            self.assertIn("Never bypass this lifecycle by opening MMO job result files", skill)
            self.assertIn("Failed, stopped, and cancelled jobs", " ".join(skill.split()))
            for tool in (
                "agent_list",
                "agent_status",
                "agent_inspect",
                "agent_trace",
                "agent_steer",
                "agent_interrupt",
                "agent_pause",
                "agent_continue",
                "agent_detach",
                "agent_stop",
                "agent_finalize",
                "agent_compact",
                "agent_respond",
                "agent_set_effort",
                "agent_fork",
                "agent_cancel",
                "agent_result_accept",
                "agent_result_reject",
            ):
                self.assertIn(f"`{tool}`", skill)
            self.assertIn("native agents are durable child threads", skill)
            self.assertIn("Do not tell a model to watch a clock", skill)
            flat_skill = " ".join(skill.split())
            self.assertIn(
                "discover and inspect retained work before spawning replacements", flat_skill
            )
            self.assertIn("any predecessor root-thread ID", flat_skill)
            self.assertIn("starting at most one same-thread continuation", flat_skill)

            guidance = compiled_guidance(resolved)
            native_guidance = guidance[agent_guidance_relative_path("repo_scout")]
            self.assertIn("one durable Codex `turn`", native_guidance)
            self.assertNotIn("goal token budget", native_guidance)
            mcp_guidance = guidance[agent_guidance_relative_path("fresh_critic")]
            self.assertIn("one durable Codex `turn`", mcp_guidance)
            root_guidance = guidance[agent_guidance_relative_path("integrator")]
            self.assertIn("Execution mode: durable Codex `goal`", root_guidance)
            self.assertIn('call `update_goal` with `status="complete"`', root_guidance)
            self.assertIn("final assistant message does not finish an active goal", root_guidance)
            self.assertIn("Unix app-server host", skill)
            self.assertIn("token accounting", skill)

            adaptive = resolve_profile("adaptive-engineering")
            self.assertEqual(
                coordination_capable_agents(adaptive),
                ["adversarial_reviewer", "implementation_specialist", "orchestrator"],
            )
            adaptive_guidance = compiled_guidance(adaptive)
            self.assertIn(PROFILE_SKILL_RELATIVE_PATH, adaptive_guidance)
            control_only_guidance = adaptive_guidance[
                agent_guidance_relative_path("implementation_specialist")
            ]
            self.assertIn("control authority but no spawn authority", control_only_guidance)
            self.assertNotIn("leaf participant", control_only_guidance)
            self.assertIn(
                "This checkpoint applies only when the current role's `Direct children` cell",
                adaptive_guidance[PROFILE_SKILL_RELATIVE_PATH],
            )

            mcp_skill = profile_skill_text(resolve_profile("secure-change"))
            self.assertIn("Codex native agent delegation is not available", mcp_skill)
            self.assertIn("## Use the MCP lifecycle", mcp_skill)
            self.assertNotIn("Maximum concurrent native spawned threads", mcp_skill)

            native_resolved = deepcopy(resolved)
            native_resolved["coordination"]["orchestration"] = "native"
            native_skill = profile_skill_text(native_resolved)
            self.assertIn("`mmo_mesh` Agent MCP delegation is not available", native_skill)
            self.assertNotIn("## Use the MCP lifecycle", native_skill)
            self.assertNotIn("Maximum total MCP descendant spawns", native_skill)

            changed_guidance = dict(compiled_guidance(resolved))
            changed_guidance[PROFILE_SKILL_RELATIVE_PATH] += "\nChanged guidance.\n"
            with mock.patch("mmo_snapshot.compiled_guidance", return_value=changed_guidance):
                self.assertNotEqual(
                    stable_hash(_snapshot_fingerprint(resolved)),
                    manifest["snapshot_hash"],
                )

            leaf_only = deepcopy(resolved)
            for agent in leaf_only["agents"].values():
                agent["can_spawn"] = []
                agent["controls"] = {}
            self.assertNotIn(PROFILE_SKILL_RELATIVE_PATH, compiled_guidance(leaf_only))

    def test_compiled_guidance_tampering_is_rejected(self) -> None:
        with RuntimeSandbox():
            snapshot = compile_profile("codex-harness-team")
            skill = Path(snapshot["directory"]) / PROFILE_SKILL_RELATIVE_PATH
            skill.chmod(0o600)
            skill.write_text(skill.read_text(encoding="utf-8") + "tampered\n", encoding="utf-8")
            with self.assertRaisesRegex(RuntimeError, "snapshot content-address mismatch"):
                load_snapshot(snapshot["manifest"]["snapshot_hash"])

    def test_historical_snapshot_keeps_its_content_addressed_guidance(self) -> None:
        with RuntimeSandbox():
            snapshot = compile_profile("codex-harness-team")
            resolved = snapshot["resolved"]
            changed_guidance = dict(compiled_guidance(resolved))
            changed_guidance[PROFILE_SKILL_RELATIVE_PATH] += "\nNew release guidance.\n"

            with mock.patch("mmo_snapshot.compiled_guidance", return_value=changed_guidance):
                loaded = load_snapshot(snapshot["manifest"]["snapshot_hash"])

            self.assertEqual(
                loaded["manifest"]["snapshot_hash"],
                snapshot["manifest"]["snapshot_hash"],
            )
            self.assertNotEqual(
                changed_guidance[PROFILE_SKILL_RELATIVE_PATH],
                (Path(snapshot["directory"]) / PROFILE_SKILL_RELATIVE_PATH).read_text(
                    encoding="utf-8"
                ),
            )

    def test_snapshot_loader_rejects_retired_resolved_profile_schema(self) -> None:
        with RuntimeSandbox():
            snapshot = compile_profile("codex-harness-team")
            resolved_path = Path(snapshot["directory"]) / "resolved-profile.json"
            resolved_path.chmod(0o600)
            resolved = read_json(resolved_path)
            resolved["schema_version"] = 4
            resolved_path.write_text(
                json.dumps(resolved, indent=2, sort_keys=True) + "\n",
                encoding="utf-8",
            )
            with self.assertRaisesRegex(RuntimeError, "unsupported resolved profile schema"):
                load_snapshot(snapshot["manifest"]["snapshot_hash"])

    def test_snapshot_manifest_is_derived_and_cache_detects_added_files(self) -> None:
        with RuntimeSandbox():
            snapshot = compile_profile("codex-harness-team")
            directory = Path(snapshot["directory"])
            target = directory / "instructions" / "invariant_designer.md"
            manifest_path = directory / "manifest.json"
            target.chmod(0o600)
            manifest_path.chmod(0o600)
            target.write_text("attacker-controlled payload\n", encoding="utf-8")
            manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
            manifest["payload_files"]["instructions/invariant_designer.md"] = hashlib.sha256(
                target.read_bytes()
            ).hexdigest()
            manifest_path.write_text(
                json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8"
            )
            with self.assertRaisesRegex(RuntimeError, "snapshot manifest integrity failure"):
                load_snapshot(snapshot["manifest"]["snapshot_hash"])

            snapshot = compile_profile("codex-harness-team", force=True)
            directory = Path(snapshot["directory"])
            load_snapshot(snapshot["manifest"]["snapshot_hash"])
            instructions = directory / "instructions"
            instructions.chmod(0o700)
            (instructions / "unexpected.md").write_text("unexpected\n", encoding="utf-8")
            with self.assertRaisesRegex(RuntimeError, "snapshot payload integrity failure"):
                load_snapshot(snapshot["manifest"]["snapshot_hash"])

    def test_schema_definition_and_json_contract_edges(self) -> None:
        invalid = {
            "type": ["object", "object"],
            "properties": {"value": {"type": "number", "minimum": 2, "maximum": 1}},
            "required": ["value", "value"],
            "additionalProperties": "no",
            "enum": [],
        }
        errors = validate_schema_definition(invalid)
        self.assertTrue(any("duplicate types" in item for item in errors), errors)
        self.assertTrue(any("minimum exceeds maximum" in item for item in errors), errors)
        self.assertTrue(any("duplicate property" in item for item in errors), errors)
        self.assertTrue(any("additionalProperties" in item for item in errors), errors)
        self.assertTrue(any("non-empty" in item for item in errors), errors)
        self.assertTrue(validate_instance(True, {"const": 1}))
        self.assertFalse(validate_instance(1.0, {"type": "integer"}))
        self.assertTrue(validate_instance(1.5, {"type": "integer"}))
        self.assertFalse(validate_schema_definition({"minItems": 1.0}))
        self.assertTrue(validate_schema_definition({"minItems": 1.5}))
        self.assertTrue(validate_schema_definition({"pattern": b"bytes"}))
        self.assertFalse(validate_schema_definition(True))
        self.assertFalse(validate_instance({"anything": True}, True))
        self.assertTrue(validate_instance({"anything": True}, False))
        self.assertTrue(validate_instance([1], {"items": False}))
        self.assertFalse(validate_instance([], {"items": False}))
        huge_integer = 10**1000
        self.assertFalse(validate_schema_definition({"minimum": huge_integer}))
        self.assertFalse(
            validate_instance(
                huge_integer,
                {"type": "number", "minimum": huge_integer - 1},
            )
        )
        self.assertTrue(
            any("unique" in item for item in validate_schema_definition({"enum": [1, 1.0]}))
        )
        self.assertFalse(validate_instance(1.0, {"const": 1}))
        self.assertTrue(validate_instance([1, 1.0], {"uniqueItems": True}))
        self.assertFalse(validate_instance([True, 1], {"uniqueItems": True}))
        self.assertTrue(validate_schema_definition({"const": {1: "not JSON"}}))
        self.assertEqual(extract_json_document('{"value": NaN}')[0], None)
        self.assertEqual(extract_json_document('{"value": 1e400}')[0], None)
        self.assertEqual(extract_json_document('{"value": 1, "value": 2}')[0], None)
        self.assertEqual(extract_json_document('{"value": "\\ud800"}')[0], None)

        valid_formats = {
            "date": ("0001-01-01", "0400-02-29", "1582-10-10"),
            "date-time": (
                "1963-06-19t08:30:06.283185z",
                "1998-12-31T23:59:60Z",
                "1998-12-31T15:59:60.123-08:00",
                "1999-01-01T00:59:60+01:00",
                "1985-04-12T00:59:59.999999999999999Z",
            ),
            "uri": (
                "http://087.10.0.1/",
                "http://999.999.999.999/",
                "ldap://[2001:db8::7]/c=GB?objectClass?one",
                "mailto:John.Doe@example.com",
                "urn:oasis:names:specification:docbook:dtd:xml:4.1.2",
            ),
        }
        invalid_formats = {
            "date": ("2021-02-29", "2023-W13-2", "2020-01-01Z"),
            "date-time": (
                "1998-12-31T23:58:60Z",
                "1998-12-30T23:59:60Z",
                "1999-01-15T00:59:60+01:00",
                "1990-12-31T15:59:59-24:00",
                "1985-04-12T23:20:50+01",
                "1985-04-12T23:20:50Z\n",
            ),
            "uri": (
                "//foo.bar/path",
                "http://example.com/%6G",
                "https://example.org/foobar®.txt",
                "http://example.com:abc/path",
                "http://[::ffff:01.2.3.4]",
                "http:/[::1]",
            ),
        }
        for format_name, values in valid_formats.items():
            for value in values:
                with self.subTest(format_name=format_name, valid=value):
                    self.assertEqual(validate_instance(value, {"format": format_name}), [])
        for format_name, values in invalid_formats.items():
            for value in values:
                with self.subTest(format_name=format_name, invalid=value):
                    self.assertTrue(validate_instance(value, {"format": format_name}))

    def test_featured_evidence_contracts_reject_contradictory_or_untyped_claims(self) -> None:
        visual = read_json(
            ROOT / "profiles" / "visual-engineering" / "contracts" / "visual-review.json"
        )
        visual_pass = {
            "verdict": "pass",
            "image_artifacts": [
                {
                    "kind": "reference",
                    "relative_path": "reference.png",
                    "sha256": "a" * 64,
                    "viewport_width": None,
                },
                {
                    "kind": "render",
                    "relative_path": "render.png",
                    "sha256": "b" * 64,
                    "viewport_width": 1440,
                },
            ],
            "mismatches": [],
            "blockers": [],
        }
        self.assertEqual(validate_instance(visual_pass, visual), [])
        contradictory_visual = {
            **visual_pass,
            "mismatches": [
                {
                    "gate": "perceptual",
                    "severity": "high",
                    "region": "header",
                    "evidence": "reference and render differ",
                }
            ],
            "blockers": ["browser capture missing"],
        }
        self.assertTrue(validate_instance(contradictory_visual, visual))

        security = read_json(
            ROOT / "profiles" / "secure-change" / "contracts" / "security-validation.json"
        )
        unevidenced_security = {
            "finding_id": "finding-1",
            "verdict": "confirmed",
            "commands": [],
            "artifacts": [],
            "limitations": [],
        }
        self.assertTrue(validate_instance(unevidenced_security, security))
        self.assertEqual(
            validate_instance(
                {
                    **unevidenced_security,
                    "commands": [
                        {
                            "command": "python reproduce.py",
                            "exit_code": 0,
                            "observation": "boundary reproduced",
                        }
                    ],
                },
                security,
            ),
            [],
        )

        research = read_json(
            ROOT / "profiles" / "research-backed-engineering" / "contracts" / "research.json"
        )
        invalid_research = {
            "status": "supported",
            "question": "Which current contract applies?",
            "claims": [
                {
                    "claim": "current behavior",
                    "source_url": "not a URI",
                    "source_date": "2026-13-40",
                    "retrieved_at": "2026-08-16",
                    "authority": "primary",
                    "evidence": "documented behavior",
                    "inference": False,
                }
            ],
            "conflicts": [],
            "limitations": [],
        }
        errors = validate_instance(invalid_research, research)
        self.assertGreaterEqual(sum("format" in item for item in errors), 3, errors)

    def test_discovery_overlay_keys_and_catalog_only_defaults_are_conservative(self) -> None:
        with RuntimeSandbox() as box:
            metadata = {
                slug: {
                    "context_window": 200_000,
                    "input_modalities": ["text"],
                    "supported_reasoning_levels": [{"effort": "high"}],
                }
                for slug in (
                    "future+model",
                    "future model",
                    "模型-" + "x" * 100,
                )
            }
            metadata["malformed"] = {"context_window": "unknown"}
            metadata["malformed_reasoning"] = {
                "context_window": 200_000,
                "input_modalities": ["text"],
                "supported_reasoning_levels": 7,
            }
            metadata["duplicate_modalities"] = {
                "context_window": 200_000,
                "input_modalities": ["text", "text"],
                "supported_reasoning_levels": [{"effort": "high"}],
            }
            overlay = build_codex_discovery_overlay(
                {"models": [*metadata, "future+model"], "metadata": metadata}
            )
            self.assertEqual(len(overlay["models"]), 3)
            self.assertEqual(len(set(overlay["models"])), 3)
            for key, model in overlay["models"].items():
                validate_id(key, "discovered model key")
                self.assertTrue(key.startswith("codex_chatgpt_builtin__"))
                self.assertLessEqual(len(key), 64)
                self.assertFalse(model["agent_compatible"])
                self.assertFalse(model["tool_calling"])

            fragment = {
                "schema_version": MMO_SCHEMA_VERSION,
                "routes": {
                    "media_test": {
                        "driver": "catalog_only",
                        "api_operator": "example",
                        "access_product": "example_catalog",
                        "wire_protocol": "catalog_only",
                        "billing_mode": "catalog_only",
                        "transport_modalities": ["text", "image"],
                    }
                },
                "models": {
                    "media_test__model": {
                        "route": "media_test",
                        "upstream_id": "media-test-model",
                        "maker": "example",
                        "kind": "image_generation",
                        "context_window": 0,
                        "reasoning_levels": ["none"],
                        "default_reasoning": "none",
                        "modalities": ["text"],
                        "output_modalities": ["image"],
                        "unit_cost_usd": 10**1000,
                    }
                },
            }
            (box.config / "catalog.d" / "media.toml").write_text(
                toml_dumps(fragment), encoding="utf-8"
            )
            catalog = validated_global_catalog()
            provider = catalog["routes"]["media_test"]
            model = catalog["models"]["media_test__model"]
            self.assertFalse(provider["tool_calling"])
            self.assertFalse(provider["parallel_tool_calls"])
            self.assertFalse(model["tool_calling"])
            self.assertFalse(model["parallel_tool_calls"])
            self.assertEqual(model["unit_cost_usd"], 10**1000)

    def test_model_input_modalities_must_fit_the_provider_transport(self) -> None:
        route = _validate_route(
            "text_only",
            {
                "driver": "catalog_only",
                "api_operator": "example",
                "access_product": "example_catalog",
                "wire_protocol": "catalog_only",
                "billing_mode": "catalog_only",
                "transport_modalities": ["text"],
                "transport_output_modalities": ["text"],
            },
        )
        with self.assertRaisesRegex(ValueError, "cannot carry input modalities"):
            validate_model_entry(
                "text_only__vision",
                {
                    "route": "text_only",
                    "upstream_id": "vision",
                    "maker": "example",
                    "kind": "vision_chat",
                    "agent_compatible": False,
                    "modalities": ["text", "image"],
                    "output_modalities": ["text"],
                    "reasoning_levels": ["none"],
                    "default_reasoning": "none",
                },
                {"text_only": route},
            )

    def test_profile_summary_applies_runtime_bindings(self) -> None:
        with RuntimeSandbox():
            summary = profile_summary(
                "codex-harness-team",
                bindings={"invariant_designer": "codex_chatgpt_builtin__gpt_5_6_terra"},
            )
        self.assertEqual(
            summary["agents"]["invariant_designer"]["model"],
            "codex_chatgpt_builtin__gpt_5_6_terra",
        )

    def test_environment_parser_preserves_unicode_and_unknown_escapes(self) -> None:
        with RuntimeSandbox() as box:
            path = box.root / "unicode.env"
            path.write_text('UNICODE="café 世界"\nPATHISH="C:\\models\\qwen"\n', encoding="utf-8")
            self.assertEqual(
                parse_env_file(path),
                {"UNICODE": "café 世界", "PATHISH": r"C:\models\qwen"},
            )

        encoded = toml_dumps({"value": "before\x7fafter"})
        self.assertEqual(tomllib.loads(encoded), {"value": "before\x7fafter"})
        self.assertNotIn("\x7f", encoded)
        with self.assertRaisesRegex(ValueError, "Unicode scalar"):
            toml_dumps({"value": "\ud800"})

    def test_invalid_low_trust_profiles_are_rejected(self) -> None:
        cases = {
            "low-trust-spawns": ("can_spawn", ["routine_engineer"], "cannot spawn"),
            "low-trust-contract": ("contract_enforcement", "warn", "strict contract"),
            "low-trust-concurrency": ("max_active", 2, "max_active=1"),
        }
        with RuntimeSandbox():
            for profile_id, (field, value, message) in cases.items():
                with self.subTest(profile_id=profile_id):
                    destination = clone_profile("access-efficient-escalation-lab", profile_id)
                    path = destination / "profile.toml"
                    data = read_toml(path)
                    data["agents"]["literal_scout"][field] = value
                    path.write_text(toml_dumps(data), encoding="utf-8")
                    with self.assertRaisesRegex(ValueError, message):
                        resolve_profile(profile_id)

    def test_profile_clone_install_remove_and_archive_traversal_rejection(self) -> None:
        with RuntimeSandbox() as box:
            destination = clone_profile("access-efficient-escalation-lab", "custom-tiered")
            self.assertTrue(destination.is_dir())
            self.assertEqual(resolve_profile("custom-tiered")["profile"]["id"], "custom-tiered")
            remove_profile("custom-tiered")
            self.assertNotIn("custom-tiered", discover_profiles())

            literal_source = box.config / "profiles.d" / "literal-source"
            shutil.copytree(ROOT / "profiles" / "access-efficient-escalation-lab", literal_source)
            literal_manifest = literal_source / "profile.toml"
            literal_manifest.write_text(
                literal_manifest.read_text(encoding="utf-8").replace(
                    'id = "access-efficient-escalation-lab"',
                    "'id' = 'literal-source' # retained clone comment",
                    1,
                ),
                encoding="utf-8",
            )
            literal_clone = clone_profile("literal-source", "literal-clone")
            cloned_text = (literal_clone / "profile.toml").read_text(encoding="utf-8")
            self.assertIn('id = "literal-clone" # retained clone comment', cloned_text)
            self.assertEqual(resolve_profile("literal-clone")["profile"]["id"], "literal-clone")

            malicious = box.root / "bad.tar.gz"
            payload = box.root / "payload.txt"
            payload.write_text("bad", encoding="utf-8")
            with tarfile.open(malicious, "w:gz") as archive:
                archive.add(payload, arcname="../escape.txt")
            with self.assertRaises(ValueError):
                install_profile_pack(malicious)

            zip_link = box.root / "bad-link.zip"
            info = zipfile.ZipInfo("profile.toml")
            info.create_system = 3
            info.external_attr = (stat.S_IFLNK | 0o777) << 16
            with zipfile.ZipFile(zip_link, "w") as archive:
                archive.writestr(info, "target")
            with self.assertRaisesRegex(ValueError, "links and special files"):
                install_profile_pack(zip_link)

            alias = box.root / "bad-alias.zip"
            with zipfile.ZipFile(alias, "w") as archive:
                archive.writestr("pack/profile.toml", 'schema_version = 1\nid = "alias"\n')
                archive.writestr("pack/./profile.toml", 'schema_version = 1\nid = "alias"\n')
            with self.assertRaisesRegex(ValueError, "unsafe archive member"):
                install_profile_pack(alias)

    def test_profile_install_replaces_only_current_generation_content(self) -> None:
        with RuntimeSandbox() as box:
            source = box.root / "versioned-pack"
            shutil.copytree(ROOT / "profiles" / "access-efficient-escalation-lab", source)
            manifest = source / "profile.toml"
            text = manifest.read_text(encoding="utf-8").replace(
                'id = "access-efficient-escalation-lab"',
                'id = "versioned-pack"',
                1,
            )
            manifest.write_text(text, encoding="utf-8")

            self.assertEqual(install_profile_pack(source), "versioned-pack")
            self.assertEqual(install_profile_pack(source), "versioned-pack")

            readme = source / "README.md"
            readme.write_text(readme.read_text(encoding="utf-8") + "\nchanged\n", encoding="utf-8")
            with self.assertRaises(FileExistsError):
                install_profile_pack(source)
            self.assertEqual(install_profile_pack(source, replace=True), "versioned-pack")
            installed = box.config / "profiles.d" / "versioned-pack" / "README.md"
            self.assertTrue(installed.read_text(encoding="utf-8").endswith("\nchanged\n"))

            installed_manifest = installed.parent / "profile.toml"
            installed_manifest.write_text(
                installed_manifest.read_text(encoding="utf-8").replace(
                    "schema_version = 8", "schema_version = 7", 1
                ),
                encoding="utf-8",
            )
            with self.assertRaises(FileExistsError):
                install_profile_pack(source)
            self.assertEqual(install_profile_pack(source, replace=True), "versioned-pack")
            self.assertEqual(read_toml(installed_manifest)["schema_version"], MMO_SCHEMA_VERSION)

            manifest.write_text(
                manifest.read_text(encoding="utf-8").replace(
                    'version = "8.0.0"', 'version = "7.9.0"', 1
                ),
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ValueError, "active package version"):
                install_profile_pack(source, replace=True)


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