#!/usr/bin/env python3
"""host.py 方法 ↔ protocol/ schema 映射钉死测试（引擎升级 0.166.0 批）。

钉死两件事：
1. host.py 实际触及的方法集合（发出的请求 + 处理的入站通知）必须
   等于下表；host.py 新增方法时本测试先红，逼升级者同步更新映射，
   而不是悄悄带一个 schema 里没有的方法上线。
2. 下表每个方法在 protocol/ 两 bundle 里存在，且 params 引用与
   0.166.0 重导结果一致；上游删方法或改 params 引用时本测试先红。

方法数勘误：2026-11-05 对齐表写“17 个方法/通知”并不准确——
实查 host.py 触及 21 个方法字符串：18 个发出请求（含 initialize
握手）+ 3 个入站通知；对齐表列出的 18 项漏了 initialize、
app/list、mcpServerStatus/list 三个。本测试以实查 21 项为准。
其中 17 个应用请求（不含 initialize）与对齐表原意一致，只是
对齐表把其中 app/list、mcpServerStatus/list 两项漏列了。

本测试只读 host.py 文本与 protocol/ 两 bundle，不 import host，
不启动任何 app-server 进程，不做模型调用。
"""

import json
import os
import re
import unittest

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

# method -> v2 ClientRequest 的 params 引用；None 表示该请求的
# params 在 schema 里是 {"type": "null"}（config/mcpServer/reload）。
EXPECTED_REQUESTS = {
    "initialize": "InitializeParams",
    "thread/start": "ThreadStartParams",
    "thread/resume": "ThreadResumeParams",
    "thread/fork": "ThreadForkParams",
    "thread/list": "ThreadListParams",
    "thread/read": "ThreadReadParams",
    "thread/name/set": "ThreadSetNameParams",
    "thread/archive": "ThreadArchiveParams",
    "thread/unarchive": "ThreadUnarchiveParams",
    "thread/compact/start": "ThreadCompactStartParams",
    "turn/start": "TurnStartParams",
    "turn/interrupt": "TurnInterruptParams",
    "review/start": "ReviewStartParams",
    "skills/list": "SkillsListParams",
    "app/list": "AppsListParams",
    "mcpServerStatus/list": "ListMcpServerStatusParams",
    "mcpServer/oauth/login": "McpServerOauthLoginParams",
    "config/mcpServer/reload": None,
}

# method -> (bundle, container, params 引用)。turn/completed 与
# thread/tokenUsage/updated 是服务端发来的通知（v2 ServerNotification）；
# mcpServer/elicitation/request 是服务端发来的请求，活在 v1 bundle
# 的 ServerRequest 里（v2 bundle 没有 ServerRequest 定义）。
EXPECTED_INBOUND = {
    "turn/completed": ("v2", "ServerNotification", "TurnCompletedNotification"),
    "thread/tokenUsage/updated": (
        "v2", "ServerNotification", "ThreadTokenUsageUpdatedNotification"),
    "mcpServer/elicitation/request": (
        "v1", "ServerRequest", "McpServerElicitationRequestParams"),
}


def _load_bundle(name):
    path = os.path.join(ROOT, "protocol", name)
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def _method_map(bundle, container):
    out = {}
    for entry in bundle["definitions"].get(container, {}).get("oneOf", []):
        props = entry.get("properties", {})
        method = (props.get("method", {}).get("enum") or [None])[0]
        if not method:
            continue
        params = props.get("params", {})
        ref = params.get("$ref", "")
        out[method] = ref.split("/")[-1] if ref else params
    return out


class TestHostProtocolMapping(unittest.TestCase):
    def test_host_method_set_matches_pin(self):
        with open(os.path.join(ROOT, "host.py"), encoding="utf-8") as f:
            src = f.read()
        requests = set(re.findall(r'\.request\(\s*"([^"]+)"', src))
        inbound = set(
            re.findall(r'\.get\("method"\)\s*[=!]=\s*"([^"]+)"', src))
        self.assertEqual(requests, set(EXPECTED_REQUESTS),
                         "host.py 发出的请求集合变了，先更新映射表再合")
        self.assertEqual(inbound, set(EXPECTED_INBOUND),
                         "host.py 处理的入站通知集合变了，先更新映射表再合")

    def test_requests_in_v2_schema(self):
        bundle = _load_bundle("codex_app_server_protocol.v2.schemas.json")
        methods = _method_map(bundle, "ClientRequest")
        for method, ref in EXPECTED_REQUESTS.items():
            self.assertIn(method, methods, f"v2 schema 缺少 host 请求 {method}")
            if ref is None:
                self.assertEqual(methods[method], {"type": "null"},
                                 f"{method} 的 params 应为 null 类型")
            else:
                self.assertEqual(methods[method], ref,
                                 f"{method} 的 params 引用变了")

    def test_inbound_in_schema(self):
        bundles = {
            "v1": _load_bundle("codex_app_server_protocol.schemas.json"),
            "v2": _load_bundle("codex_app_server_protocol.v2.schemas.json"),
        }
        for method, (bundle_name, container, ref) in EXPECTED_INBOUND.items():
            methods = _method_map(bundles[bundle_name], container)
            self.assertIn(method, methods,
                          f"{bundle_name} schema {container} 缺少 {method}")
            self.assertEqual(methods[method], ref,
                             f"{method} 的 params 引用变了")

    def test_bundles_parse(self):
        for name in ("codex_app_server_protocol.schemas.json",
                     "codex_app_server_protocol.v2.schemas.json"):
            bundle = _load_bundle(name)
            self.assertIn("$schema", bundle)
            self.assertIn("definitions", bundle)
            self.assertGreater(len(bundle["definitions"]), 0)


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