@@ -42,8 +42,8 @@
 }
 
 CORE_DEPENDENCIES = {
-    # 引擎边界：纯数据与退出码翻译，不执行进程（探测由调用方注入）。
-    "codeguard.engine": {"__future__", "dataclasses", "typing"},
+    # 引擎边界：纯数据与退出码翻译，不执行进程（探测与能力报告由调用方注入）。
+    "codeguard.engine": {"__future__", "dataclasses", "typing", "any"},
     "codeguard.cve_policy": {"__future__", "dataclasses"},
     "codeguard.cve_reports": {"__future__", "json", "math", "codeguard.cve_policy"},
     "codeguard.cve_scanners": {"__future__", "tempfile", "pathlib", "codeguard.cve_policy",
 
@@ -1,25 +1,42 @@
 """判定引擎边界：谁签发质量结论，以及各引擎退出码如何翻译成门禁动作。
 
 插件自带的 Legacy Python 引擎随包发布，可以签发 PASS/FAIL。
-Rust ``codeguard`` 组件是后续统一内核，但当前对所有质量路径固定返回退出码 3
-（检查未完成），能力矩阵全部条目为 ``gap``。在它自行解除未完成状态之前，
-不得用它签发任何质量结论。
+Rust ``codeguard`` 组件是后续统一内核，其就绪程度**由它自己声明**——
+``codeguard capabilities --format json`` 报出的 ``capability_inventory`` 里，
+每个「语言 × 平台 × 检查类别」单元的 ``status`` 取 ``implemented`` / ``gap`` /
+``not_applicable``。
 
-本模块只做数据与规则，不执行进程：探测结果由调用方注入，便于在不启动外部
-工具的情况下验证门禁语义。
+因此本模块**不写死**内核是否可用：写死会在内核演进后静默过期，让插件对外
+给出与事实相反的说法。权威性按类别逐一判断：
+
+- 该类别在内核自报中为 ``implemented`` → 该类别可由 Rust 签发；
+- 为 ``gap``、报告不可读、或内核缺席 → 一律仍由 Legacy 签发。
+
+这样交接是数据变化（内核自己把某个类别标为 implemented），而不是插件改代码。
+
+本模块只做数据与规则，不执行进程：探测与能力报告由调用方注入，便于在不启动
+外部工具的情况下验证门禁语义。
 """
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Literal
+from typing import Any, Literal
 
 EngineId = Literal["legacy", "rust"]
 Verdict = Literal["PASS", "FAIL", "UNVERIFIED", "INCOMPLETE"]
+CapabilityStatus = Literal["implemented", "gap", "not_applicable"]
+
+# 内核能力清单的协议标识；不匹配时视为"不可读"，按 gap 处理。
+INVENTORY_REPORT_TYPE = "capability_inventory"
 
 
 @dataclass(frozen=True)
 class EngineSpec:
-    """一个候选判定引擎的登记信息；权威性是显式事实，不由存在性推导。"""
+    """一个候选判定引擎的登记信息。
+
+    ``authoritative`` 是 Legacy 的静态事实（随插件发布即可签发）；
+    Rust 的权威性是**逐类别**的，见 :func:`category_authority`。
+    """
 
     id: EngineId
     label: str
@@ -39,11 +56,12 @@ class EngineSpec:
 RUST = EngineSpec(
     id="rust",
     label="Rust codeguard 组件（独立内核）",
+    # Rust 是否有权威性取决于它自报的能力，不在此写死。
     authoritative=False,
     version_command=("codeguard", "--version"),
     note=(
-        "对所有质量路径固定返回退出码 3（未完成），能力矩阵全部 gap；"
-        "只可作只读观察，命令面改用位置参数，无 fix/dockerfile/--lang/--ecosystem/--severity。"
+        "就绪程度由内核自报的 capability_inventory 决定；"
+        "命令面改用位置参数，无 fix/dockerfile/--lang/--ecosystem/--severity。"
     ),
 )
 
@@ -65,6 +83,18 @@ class GateOutcome:
     reason: str
 
 
+@dataclass(frozen=True)
+class CategoryAuthority:
+    """某个检查类别当前由谁签发，以及依据。"""
+
+    language: str
+    platform: str
+    category: str
+    engine: EngineId
+    status: CapabilityStatus | None
+    reason: str
+
+
 # Legacy：0 通过 / 1 无法验证 / 2 存在违规 / 3 参数错误。
 LEGACY_EXIT: dict[int, tuple[Verdict, bool]] = {
     0: ("PASS", False),
@@ -92,9 +122,7 @@ class GateOutcome:
 def translate(engine_id: EngineId, exit_code: int) -> GateOutcome:
     """把某引擎的退出码翻译成门禁动作；未知码一律按未验证处理。"""
     spec = ENGINES[engine_id]
-    verdict, blocks = _EXIT_TABLES[engine_id].get(
-        exit_code, ("UNVERIFIED", True)
-    )
+    verdict, blocks = _EXIT_TABLES[engine_id].get(exit_code, ("UNVERIFIED", True))
     reason = spec.note if not spec.authoritative else "Legacy 退出码语义"
     return GateOutcome(
         engine=engine_id,
@@ -105,11 +133,82 @@ def translate(engine_id: EngineId, exit_code: int) -> GateOutcome:
     )
 
 
+def _iter_cells(inventory: Any):
+    """遍历能力清单中的单元，兼容带过滤器的扁平 cells 与未过滤的 languages。"""
+    if not isinstance(inventory, dict):
+        return
+    if isinstance(inventory.get("cells"), list):
+        yield from (cell for cell in inventory["cells"] if isinstance(cell, dict))
+        return
+    for language in inventory.get("languages") or []:
+        if not isinstance(language, dict):
+            continue
+        for platform, categories in (language.get("platforms") or {}).items():
+            if not isinstance(categories, dict):
+                continue
+            for category, cell in categories.items():
+                if isinstance(cell, dict):
+                    yield {
+                        "language": language.get("language"),
+                        "platform": platform,
+                        "category": category,
+                        **cell,
+                    }
+
+
+def category_authority(
+    inventory: Any,
+    *,
+    language: str,
+    platform: str,
+    category: str,
+) -> CategoryAuthority:
+    """判断某个「语言 × 平台 × 类别」当前应由谁签发。
+
+    只有当内核自报该单元 ``implemented`` 时才交给 Rust；其余情况（gap、
+    not_applicable、清单不可读、找不到该单元、内核缺席）一律留在 Legacy。
+    读不到声明不等于通过——这是 fail closed。
+    """
+    readable = (
+        isinstance(inventory, dict)
+        and inventory.get("report_type") == INVENTORY_REPORT_TYPE
+    )
+    if not readable:
+        return CategoryAuthority(
+            language=language, platform=platform, category=category,
+            engine=LEGACY.id, status=None,
+            reason="内核能力清单不可读或协议不匹配，权威性留在 Legacy",
+        )
+
+    for cell in _iter_cells(inventory):
+        if (cell.get("language") == language and cell.get("platform") == platform
+                and cell.get("category") == category):
+            status = cell.get("status")
+            if status == "implemented":
+                return CategoryAuthority(
+                    language=language, platform=platform, category=category,
+                    engine=RUST.id, status="implemented",
+                    reason="内核自报该检查类别已实现",
+                )
+            return CategoryAuthority(
+                language=language, platform=platform, category=category,
+                engine=LEGACY.id,
+                status=status if status in ("gap", "not_applicable") else None,
+                reason=f"内核自报该类别为 {status or '未知'}，权威性留在 Legacy",
+            )
+
+    return CategoryAuthority(
+        language=language, platform=platform, category=category,
+        engine=LEGACY.id, status=None,
+        reason="内核能力清单中无该单元，按未证明处理",
+    )
+
+
 def resolve_authoritative(*, rust_available: bool) -> EngineSpec:
-    """选择签发质量结论的引擎。
+    """在**没有任何类别**自报为 implemented 时选择签发方。
 
-    Rust 二进制存在并不改变权威性：只要它仍对质量路径返回未完成，签发方就必须
-    是 Legacy。``rust_available`` 只影响调用方是否需要提示"检测到新内核"。
+    逐类别交接请用 :func:`category_authority`；本函数只回答全局兜底问题：
+    二进制存在本身不构成权威。
     """
     if rust_available and not RUST.authoritative:
         return LEGACY
 
@@ -15,7 +15,12 @@
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
 
-from codeguard.engine import ENGINES, resolve_authoritative
+from codeguard.engine import (
+    ENGINES,
+    INVENTORY_REPORT_TYPE,
+    _iter_cells,
+    resolve_authoritative,
+)
 
 PROBE_TIMEOUT = 10
 
@@ -55,8 +60,54 @@ def probe_rust() -> dict:
     }
 
 
+def probe_capability_inventory(path: str) -> dict:
+    """读取内核自报的能力清单。
+
+    权威性由此决定：插件不写死内核是否可用，而是读它自己声明的
+    implemented / gap。读不到就按 gap 处理（fail closed）。
+    """
+    try:
+        proc = subprocess.run(
+            [path, "capabilities", "all", "--format", "json"],
+            capture_output=True, text=True, timeout=PROBE_TIMEOUT, check=False,
+        )
+    except (OSError, subprocess.SubprocessError) as exc:
+        return {"readable": False, "reason": f"能力清单读取失败：{exc}"}
+    if proc.returncode != 0:
+        return {"readable": False, "reason": f"能力清单命令退出 {proc.returncode}"}
+    try:
+        payload = json.loads(proc.stdout)
+    except ValueError:
+        return {"readable": False, "reason": "能力清单不是 JSON"}
+    if not isinstance(payload, dict) or payload.get("report_type") != INVENTORY_REPORT_TYPE:
+        return {"readable": False, "reason": "缺少 capability_inventory 标记"}
+    return summarize_inventory(payload)
+
+
+def summarize_inventory(payload: dict) -> dict:
+    """把能力清单汇总成 implemented / gap 计数。"""
+    counts: dict[str, int] = {}
+    for cell in _iter_cells(payload):
+        status = cell.get("status")
+        key = status if status in ("implemented", "gap", "not_applicable") else "unknown"
+        counts[key] = counts.get(key, 0) + 1
+    return {
+        "readable": True,
+        "release_version": payload.get("release_version"),
+        "counts": counts,
+        "implemented": counts.get("implemented", 0),
+        "gap": counts.get("gap", 0),
+        "not_applicable": counts.get("not_applicable", 0),
+        "unknown": counts.get("unknown", 0),
+    }
+
+
 def build_report() -> dict:
     rust = probe_rust()
+    inventory = (
+        probe_capability_inventory(rust["path"]) if rust["available"]
+        else {"readable": False, "reason": "Rust 组件缺席"}
+    )
     authoritative = resolve_authoritative(rust_available=rust["available"])
     return {
         "report_type": "engine",
@@ -72,6 +123,7 @@ def build_report() -> dict:
             for spec in (ENGINES["legacy"], ENGINES["rust"])
         ],
         "rust_component": rust,
+        "rust_capability_inventory": inventory,
     }
 
 
@@ -83,16 +135,30 @@ def render_human(report: dict) -> str:
         lines.append(f"  - {entry['label']}：{mark}")
         lines.append(f"      {entry['note']}")
     rust = report["rust_component"]
+    inv = report["rust_capability_inventory"]
     lines.append("")
     if rust["available"]:
         lines.append(
             f"检测到 Rust 组件：{rust['path']}（cli_version={rust['cli_version']}，"
             f"target={rust['target']}，build_identity={rust['build_identity']}，"
             f"rulepack_compatibility={rust['rulepack_compatibility']}）"
         )
-        lines.append("  它当前不签发质量结论：检测到 ≠ 可用于门禁。")
     else:
         lines.append(f"未检测到 Rust 组件：{rust['reason']}")
+    lines.append("")
+    if inv.get("readable"):
+        lines.append(
+            f"内核自报能力（release={inv.get('release_version')}）："
+            f"implemented {inv['implemented']} / gap {inv['gap']} / "
+            f"not_applicable {inv['not_applicable']} / unknown {inv['unknown']}"
+        )
+        if inv["implemented"] == 0:
+            lines.append("  尚无任何类别自报已实现 → 全局签发方保持 Legacy。")
+        else:
+            lines.append("  已有类别自报已实现 → 该类别可由 Rust 签发，"
+                         "逐类别结论见 category_authority。")
+    else:
+        lines.append(f"内核能力清单不可读（{inv.get('reason')}）→ 按未证明处理，签发方保持 Legacy。")
     return "\n".join(lines)