R4 P0: C2S 门禁设计收口与吸收清单验真

.github/workflows/e2e.yml

@@ -116,6 +116,11 @@ jobs:
       - name: Bring up Redis test service
         run: docker compose -f docker-compose.test.yml up -d redis --wait
 
+      - name: C2S gate matrix contract
+        run: |
+          python3 scripts/check_c2s_gate_matrix.py
+          python3 -m unittest scripts/tests/check_c2s_gate_matrix_test.py
+
       - name: Schema stage (build + check + test + generate)
         working-directory: agent/packages/schema
         run: |

docs/plans-skeleton/plan-refactor-master-v1.md

@@ -98,7 +98,7 @@
 
 1. 9 条轨道全部归档（各自 bot 场景常绿 + 吸收 plan 全部归档/验伪结案）；
 2. 三个 2 万行级 god file（inventory/mod.rs、client_request_handler.rs、persistence/mod.rs）不复存在，最大单文件 < 3000 行；
-3. `qi_current` 裸写编译不过；client 无未登记的会话态 store；113 C2S 变体全部有显式 GateSpec/no_gate 声明；28 旁路 channel 收编或豁免登记；
+3. `qi_current` 裸写编译不过；client 无未登记的会话态 store；届时现行 `ClientRequestV1` 全部变体均有显式 GateSpec/no_gate 声明（2026-08-03 P0 基线为 104，新增变体自动纳入）；28 旁路 channel 收编或豁免登记；
 4. bot 场景数从 ~30 增至 ≥80，CI e2e 是唯一主门禁且无已知假绿。
 5. `flash-review` label 下 open issue 全部显式处置（fixed / dup / 验伪关闭 / 促升 skeleton，见 §10），无静默积压。
 

docs/plans-skeleton/plan-refactor-wire-s2c-v1.md

@@ -22,7 +22,7 @@
 - ⬜ P1 emit builder + scope 落地：builder 上线，vfx/audio/env 三类先挂 scope（跨维 bleed 立灭）；跨位面切换时 env/season 全量重发。
 - ⬜ P2 client 桥接层收敛：枚举前缀剥离收敛到单点（含 forge-session 修复）；`ServerDataRouter` 注册表整备（分域注册文件，不再单个 1547 行 switch 追加）。
 - ⬜ P3 旁路归一批次：28 channel 逐批收编入 server_data envelope 或登记豁免（资源包/握手类可豁免）；删除散装 receiver。
-- ⬜ P4 契约 pin 全量化：双向 sample 对拍测试补齐（113 C2S + 144 S2C 每变体至少一条正反 sample，schema 改动连 sample 一起改）；emit 迁移到 builder 的长尾批次。
+- ⬜ P4 契约 pin 全量化：双向 sample 对拍测试补齐（届时现行 `ClientRequestV1` 全部变体，2026-08-03 基线 104；以及届时现行 `ServerDataType` 全部变体，2026-07-27 侦察基线 144；每变体至少一条正反 sample，schema 改动连 sample 一起改）；emit 迁移到 builder 的长尾批次。
 - ⬜ P5 bot 验收 + 吸收 plan 批量归档。
 
 ## 吸收清单（短名省略 plan-bughunt- 前缀与 -v1 后缀）

scripts/check_c2s_gate_matrix.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import re
+import sys
+from collections import Counter
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+ENUM_PATH = ROOT / "server/src/schema/client_request.rs"
+PLAN_PATH = ROOT / "docs/plan-refactor-c2s-gate-v1.md"
+ENUM_SERDE_RE = re.compile(
+    r'#\[serde\(([^]]*)\)\]\s*pub enum ClientRequestV1\s*\{', re.MULTILINE
+)
+MATRIX_RE = re.compile(r"^\|\s*(\d+)\s*\|\s*`([^`]+)`\s*\|")
+VARIANT_DECL_RE = re.compile(r"^([A-Z][A-Za-z0-9_]*)\s*(.*)$")
+
+
+def _without_line_comment(line: str) -> str:
+    return line.split("//", 1)[0]
+
+
+def _leading_attributes(code: str) -> tuple[list[str], str]:
+    attributes: list[str] = []
+    while code.startswith("#["):
+        bracket_depth = 0
+        in_string = False
+        escaped = False
+        end = None
+        for index, char in enumerate(code):
+            if in_string:
+                if escaped:
+                    escaped = False
+                elif char == "\\":
+                    escaped = True
+                elif char == '"':
+                    in_string = False
+                continue
+            if char == '"':
+                in_string = True
+            elif char == "[":
+                bracket_depth += 1
+            elif char == "]":
+                bracket_depth -= 1
+                if bracket_depth == 0:
+                    end = index + 1
+                    break
+        if end is None:
+            raise RuntimeError(f"unterminated ClientRequestV1 attribute: {code!r}")
+        attributes.append(code[:end])
+        code = code[end:].lstrip()
+    if code.startswith("#"):
+        raise RuntimeError(f"unsupported ClientRequestV1 attribute syntax: {code!r}")
+    return attributes, code
+
+
+def parse_enum_variants(source: str) -> list[str]:
+    serde = ENUM_SERDE_RE.search(source)
+    if not serde or not re.search(r'\btag\s*=\s*"type"', serde.group(1)):
+        raise RuntimeError("ClientRequestV1 must use serde tag = \"type\"")
+    if not re.search(r'\brename_all\s*=\s*"snake_case"', serde.group(1)):
+        raise RuntimeError("ClientRequestV1 must use serde rename_all = \"snake_case\"")
+
+    lines = source.splitlines()
+    variants: list[str] = []
+    inside = False
+    depth = 0
+    tuple_depth = 0
+    pending_attributes: list[str] = []
+
+    for line in lines:
+        if not inside:
+            if line == "pub enum ClientRequestV1 {":
+                inside = True
+                depth = 1
+            continue
+
+        code = _without_line_comment(line).strip()
+        if depth == 1 and tuple_depth:
+            tuple_depth += code.count("(") - code.count(")")
+            if tuple_depth < 0:
+                raise RuntimeError(f"unbalanced tuple variant syntax: {line!r}")
+            continue
+        if depth == 1 and code == "}":
+            depth = 0
+            break
+        if depth == 1 and code:
+            attributes, code = _leading_attributes(code)
+            pending_attributes.extend(attributes)
+            if not code:
+                continue
+            if any("serde" in attribute and "rename" in attribute for attribute in pending_attributes):
+                raise RuntimeError("ClientRequestV1 variant-level serde rename is unsupported")
+            pending_attributes.clear()
+            match = VARIANT_DECL_RE.match(code)
+            if not match:
+                raise RuntimeError(f"unsupported ClientRequestV1 syntax: {line!r}")
+            suffix = match.group(2).strip()
+            if not suffix or suffix[0] not in "{(,":
+                raise RuntimeError(f"unsupported ClientRequestV1 variant syntax: {line!r}")
+            variants.append(match.group(1))
+            if suffix[0] == "(":
+                tuple_depth = suffix.count("(") - suffix.count(")")
+                if tuple_depth < 0:
+                    raise RuntimeError(f"unbalanced tuple variant syntax: {line!r}")
+
+        depth += code.count("{") - code.count("}")
+        if depth == 0:
+            break
+
+    if not inside or depth != 0 or not variants:
+        raise RuntimeError(f"cannot parse ClientRequestV1 from {ENUM_PATH}")
+    return variants
+
+
+def enum_variants() -> list[str]:
+    return parse_enum_variants(ENUM_PATH.read_text(encoding="utf-8"))
+
+
+def matrix_variants() -> tuple[list[int], list[str]]:
+    rows: list[tuple[int, str]] = []
+    for line in PLAN_PATH.read_text(encoding="utf-8").splitlines():
+        if match := MATRIX_RE.match(line):
+            rows.append((int(match.group(1)), match.group(2)))
+    if not rows:
+        raise RuntimeError(f"cannot parse C2S matrix from {PLAN_PATH}")
+    return [number for number, _ in rows], [variant for _, variant in rows]
+
+
+def duplicates(values: list[str]) -> list[str]:
+    return sorted(value for value, count in Counter(values).items() if count > 1)
+
+
+def first_order_mismatch(left: list[str], right: list[str]) -> tuple[int, str, str] | None:
+    for index in range(max(len(left), len(right))):
+        left_value = left[index] if index < len(left) else "<missing>"
+        right_value = right[index] if index < len(right) else "<missing>"
+        if left_value != right_value:
+            return index, left_value, right_value
+    return None
+
+
+def main() -> int:
+    errors: list[str] = []
+    try:
+        enum = enum_variants()
+        numbers, matrix = matrix_variants()
+    except RuntimeError as error:
+        print(f"C2S gate matrix check failed:\n- {error}", file=sys.stderr)
+        return 1
+
+    expected_numbers = list(range(1, len(matrix) + 1))
+    if numbers != expected_numbers:
+        errors.append(f"matrix numbering is not contiguous: {numbers}")
+
+    for label, values in (("enum", enum), ("matrix", matrix)):
+        duplicate_values = duplicates(values)
+        if duplicate_values:
+            errors.append(f"duplicate {label} variants: {duplicate_values}")
+
+    enum_matrix_missing = [variant for variant in enum if variant not in matrix]
+    matrix_enum_extra = [variant for variant in matrix if variant not in enum]
+    if enum_matrix_missing:
+        errors.append(f"missing matrix variants: {enum_matrix_missing}")
+    if matrix_enum_extra:
+        errors.append(f"extra matrix variants: {matrix_enum_extra}")
+    if set(matrix) != set(enum):
+        errors.append("matrix and Rust enum variant sets differ")
+
+    mismatch = first_order_mismatch(enum, matrix)
+    if mismatch:
+        errors.append(
+            "first enum/matrix order mismatch at row "
+            f"{mismatch[0] + 1}: enum={mismatch[1]} matrix={mismatch[2]}"
+        )
+
+    if errors:
+        print("C2S gate matrix check failed:", file=sys.stderr)
+        for error in errors:
+            print(f"- {error}", file=sys.stderr)
+        return 1
+
+    print(f"C2S gate matrix matches all {len(enum)} Rust ClientRequestV1 variants")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/tests/check_c2s_gate_matrix_test.py

@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import importlib.util
+import io
+import unittest
+from contextlib import redirect_stderr
+from pathlib import Path
+from unittest.mock import patch
+
+CHECKER = Path(__file__).resolve().parents[1] / "check_c2s_gate_matrix.py"
+spec = importlib.util.spec_from_file_location("check_c2s_gate_matrix", CHECKER)
+assert spec and spec.loader
+checker = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(checker)
+
+ENUM_PREFIX = '#[serde(deny_unknown_fields, tag = "type", rename_all = "snake_case")]\n'
+
+
+class ParserTests(unittest.TestCase):
+    def test_accepts_struct_unit_and_tuple_variants(self) -> None:
+        source = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    StructVariant {
+        value: u8,
+    },
+    UnitVariant,
+    TupleVariant(u8),
+}
+"""
+        self.assertEqual(
+            checker.parse_enum_variants(source),
+            ["StructVariant", "UnitVariant", "TupleVariant"],
+        )
+
+    def test_accepts_multiline_tuple_variant(self) -> None:
+        source = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    TupleVariant(
+        u8,
+        String,
+    ),
+}
+"""
+        self.assertEqual(checker.parse_enum_variants(source), ["TupleVariant"])
+
+    def test_fails_closed_on_unknown_top_level_syntax(self) -> None:
+        source = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    #[cfg(test)]
+    Unsupported = 1,
+}
+"""
+        with self.assertRaisesRegex(RuntimeError, "unsupported ClientRequestV1 variant syntax"):
+            checker.parse_enum_variants(source)
+
+    def test_fails_closed_on_serde_wire_renames(self) -> None:
+        camel = '#[serde(tag = "type", rename_all = "camelCase")]\npub enum ClientRequestV1 { Variant, }'
+        with self.assertRaisesRegex(RuntimeError, 'rename_all = "snake_case"'):
+            checker.parse_enum_variants(camel)
+
+        renamed = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    #[serde(rename = "other")]
+    Variant,
+}
+"""
+        with self.assertRaisesRegex(RuntimeError, "variant-level serde rename"):
+            checker.parse_enum_variants(renamed)
+
+        stacked = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    #[serde(rename = "other")]
+    #[doc = "variant docs"]
+    Variant,
+}
+"""
+        with self.assertRaisesRegex(RuntimeError, "variant-level serde rename"):
+            checker.parse_enum_variants(stacked)
+
+        inline = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    #[serde(rename = "other")] Variant,
+}
+"""
+        with self.assertRaisesRegex(RuntimeError, "variant-level serde rename"):
+            checker.parse_enum_variants(inline)
+
+        inline_cfg = ENUM_PREFIX + """pub enum ClientRequestV1 {
+    #[cfg(test)] Variant,
+}
+"""
+        self.assertEqual(checker.parse_enum_variants(inline_cfg), ["Variant"])
+
+    def test_main_rejects_each_contract_drift(self) -> None:
+        baseline = {
+            "enum_variants": ["Alpha", "Beta"],
+            "matrix_variants": ([1, 2], ["Alpha", "Beta"]),
+        }
+        cases = {
+            "mismatched sets": {"matrix_variants": ([1, 2], ["Alpha", "Gamma"])},
+            "reordered variants": {"matrix_variants": ([1, 2], ["Beta", "Alpha"])},
+            "duplicate enum variants": {"enum_variants": ["Alpha", "Alpha"]},
+            "duplicate matrix variants": {"matrix_variants": ([1, 2], ["Alpha", "Alpha"])},
+            "non-contiguous rows": {"matrix_variants": ([1, 3], ["Alpha", "Beta"])},
+        }
+        for label, override in cases.items():
+            values = baseline | override
+            with self.subTest(label=label), patch.object(
+                checker, "enum_variants", return_value=values["enum_variants"]
+            ), patch.object(
+                checker, "matrix_variants", return_value=values["matrix_variants"]
+            ), redirect_stderr(io.StringIO()):
+                self.assertEqual(checker.main(), 1)
+
+    def test_main_accepts_matching_enum_and_matrix(self) -> None:
+        with patch.object(
+            checker, "enum_variants", return_value=["Alpha", "Beta"]
+        ), patch.object(
+            checker, "matrix_variants", return_value=([1, 2], ["Alpha", "Beta"])
+        ):
+            self.assertEqual(checker.main(), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()