fix(kb): independently review each subscription module card

Author: tzhouam

doc/architecture/SPEC/kb_service.md

@@ -432,13 +432,17 @@ x 必须为正。因阈值停下的调用（`stop_reason="max_budget"`）→ `Mo
   仓库入口页链接；`components/` 已存在却没有入口页时组直接放在仓库目录下。`_routes.yaml` 只追加（保留开头注释块，
   `routes_append_only` 校验不重排、不改名、不删除）。预算用完或超过卡片上限时，已写的卡片照常成 PR，剩下的模块进
   `unfinished`；报告里列出仍无路由的模块。测试：`test_kb_init_modules.py`。
-  显式 `--unlimited-subscription` 使用 `module_prompt_version: 2` 与独立的
+  显式 `--unlimited-subscription` 使用 `module_prompt_version: 3` 与独立的
   `SYSTEM_CARD_IMPL`，从固定 SHA 的共享索引提供带文件哈希、行号与完整性状态的实现片段；
   整个卡片输入仍受 40,000 UTF-8 字节限制，按剩余文件分配预算，能放下时提供全文，否则提供
   明确的首尾行段，未展示的行为保持未知。读取失败不使用替换字符补造源码；元数据本身超限时
   本模块保持未完成，不发起生成。卡片不能从目录名推断开发／运行时职责，也不能在未展示配置
   与分支时断言默认副作用；打印的手工操作须与实际执行区分。此模式仅保留每文件前四个签名，
   以留出实现片段预算；旧模式仍保留最多四十个签名，提示词和身份不变。
+  此模式每张候选地图卡再由独立 Codex 按实际引用文件的源码片段评审一次；沿用 prose 的
+  `faithful / does_not_weaken / non_contradictory` 三项，全部为 yes 才写入。真实结论、维度、
+  原因、模型与文本哈希保存在阶段记录；拒绝或无法评审只令本卡保持未知，其他组件继续。
+  未通过的卡片不增加模块路由覆盖，也不计入功能深度认可；后续功能／核心文件目标独立验收。
   测试：`test_kb_module_source_evidence.py`。
 
 ## 2026-09-30 kb init：阶段 3（deepen）与 shadow 开关

src/infermatrix_copilot/kb_service/init_modules.py

@@ -26,6 +26,7 @@
 from __future__ import annotations
 
 import copy
+import hashlib
 import json
 import re
 from dataclasses import dataclass
@@ -39,7 +40,8 @@
 from .init_stages import (
     _SLUG, ROUTES_NAME, _fence, _one_line, _page_frontmatter, _slug, _Stage, _title_of, review_route_line,
 )
-from .init_support import InitError, InitRecord, generate
+from .init_support import InitError, InitRecord, classify_verdict, generate, judge
+from .models import ModelUnavailable
 
 MAX_CARD_FILES = 200
 MAX_CARD_BYTES = 40_000
@@ -88,7 +90,7 @@
 """
 
 
-def _implementation_payload(payload: dict, index, pin: str) -> dict:
+def _implementation_payload(payload: dict, index, pin: str, *, max_bytes: int = MAX_CARD_BYTES) -> dict:
     """Share the card cap fairly; unread or cropped code never becomes whole-file evidence."""
     if index.identity.get("pin") != pin:
         raise ValueError("module implementation index differs from the fixed source SHA")
@@ -109,7 +111,7 @@ def _implementation_payload(payload: dict, index, pin: str) -> dict:
     def size():
         return len(_fence(out).encode("utf-8"))
 
-    if size() > MAX_CARD_BYTES:
+    if size() > max_bytes:
         raise ValueError("module metadata exceeds the implementation card byte cap; no evidence dispatched")
 
     def ranges(lines, count):
@@ -126,7 +128,7 @@ def ranges(lines, count):
 
     for offset, (item, lines) in enumerate(readable):
         before = size()
-        allocation = (MAX_CARD_BYTES - before) // (len(readable) - offset)
+        allocation = (max_bytes - before) // (len(readable) - offset)
         limit = before + allocation
         item.update(status="complete", ranges=ranges(lines, len(lines)))
         if size() <= limit:
@@ -360,7 +362,7 @@ class _Modules(_Stage):
     STAGE = "modules"
 
     def _input_options(self) -> dict:
-        return {"module_prompt_version": 2} if self.rt.unlimited_subscription else {}
+        return {"module_prompt_version": 3} if self.rt.unlimited_subscription else {}
 
     def _build(self, tree: Path) -> InitRecord:
         routes_path = f"{self.repo_dir}/{ROUTES_NAME}"
@@ -602,13 +604,50 @@ def items(name: str, allowed: set[str], text_key: str) -> list[tuple[str, str]]:
             group = _slug(PurePosixPath(key.rstrip("/")).name or self.lifecycle.repo)
         headings = {k: _one_line((data.get("headings") or {}).get(k)).replace("*", "") or v
                     for k, v in _DEFAULT_HEADINGS.items()}
-        return _Card(
+        card = _Card(
             module=key, prefixes=[], title=_one_line(data["title"]),
             purpose=self._d5_prose(str(data.get("purpose") or "")),
             entry_points=items("entry_points", in_module, "what"), key_files=items("key_files", in_module, "what"),
             docs=items("docs", doc_paths, "why"),
             signals=[s for s in (_one_line(x, 60) for x in data.get("signals") or []) if s][:MAX_ITEMS],
             group=group, group_title=_one_line(data.get("group_title")) or group, headings=headings)
+        if self.rt.unlimited_subscription and not self._review_card(card, listed):
+            return None
+        return card
+
+    def _review_card(self, card: _Card, listed: list[dict]) -> bool:
+        """One independent source review; a rejected card cannot halt other modules."""
+        from ..knowledge_service.l1 import Block
+
+        referenced = {path for path, _ in card.entry_points + card.key_files}
+        page = f"{self.root}/{card.group}/module-review.md"
+        text = self._card_text(card)
+        text_sha = hashlib.sha256(text.encode()).hexdigest()
+        key = f"module:{card.module}"
+        payload = {"repository": self.lifecycle.full_name, "module": card.module,
+                   "files": [entry for entry in listed if entry["path"] in referenced],
+                   "doc_files": [path for path, _ in card.docs],
+                   "source_scope": "Only shown implementation ranges support behavior. Directory names do not "
+                   "establish runtime roles; default side effects require shown configuration and branches. "
+                   "Printed/manual instructions do not establish execution; unshown steps remain unknown."}
+        # Reserve the actual card text and the existing judge envelope/indentation.
+        allowance = MAX_CARD_BYTES - len(_fence({"card": text}).encode()) - 2048
+        try:
+            evidence = _implementation_payload(payload, self.source_index, self.record.pin, max_bytes=allowance)
+            verdict = judge(self.rt, self.budget, self.lifecycle.init,
+                            Block("prose", page, "", "prose", text_sha), base={}, head={page: text},
+                            evidence=[evidence])
+            label, dimensions, reasons, model = (classify_verdict(verdict), verdict.dimensions,
+                                                 verdict.reasons, verdict.model)
+        except (ValueError, BudgetExhausted, ModelUnavailable) as exc:
+            label, dimensions, reasons, model = "unjudged", {}, {"source_review": str(exc)}, ""
+        self.record.verdicts[key] = {"kind": "module_prose", "page": page, "verdict": label, "dimensions": dimensions,
+                                     "reasons": reasons, "model": model, "text_sha": text_sha}
+        if label != "pass":
+            self.record.unfinished.append(f"module {card.module}: source review {label}; card remains unknown")
+            self.record.dropped.append({"rule_id": key, "module": card.module,
+                                        "why": "source review: " + json.dumps(reasons)})
+        return label == "pass"
 
     def _card_text(self, card: _Card) -> str:
         """The card page. Sections are bold labels, not headings: the

test/test_kb_init_modules.py

@@ -11,7 +11,7 @@
 
 import yaml
 
-from infermatrix_copilot.kb_service.init_modules import SYSTEM_CARD, _leading_doc, declaration, routes_append_only
+from infermatrix_copilot.kb_service.init_modules import SYSTEM_CARD, SYSTEM_CARD_IMPL, _leading_doc, declaration, routes_append_only
 from infermatrix_copilot.kb_service.init_stages import run_stage
 from infermatrix_copilot.kb_service.config import RepoLifecycle
 from infermatrix_copilot.kb_service.init_support import InitRecord
@@ -44,7 +44,7 @@ def __init__(self, *, card_cost=0.01, group="tooling", **kwargs):
         self.group = group
 
     def call_json(self, role, *, system, prompt, validate=None, max_budget_usd=None):
-        if system != SYSTEM_CARD:
+        if system not in (SYSTEM_CARD, SYSTEM_CARD_IMPL):
             reply = super().call_json(role, system=system, prompt=prompt, validate=validate,
                                       max_budget_usd=max_budget_usd)
             if system.startswith("You set up the review knowledge base"):

test/test_kb_module_source_evidence.py

@@ -12,6 +12,7 @@
 )
 from infermatrix_copilot.kb_service.init_stages import _fence, run_stage
 from infermatrix_copilot.kb_service.models import ModelRole
+from infermatrix_copilot.kb_service.judge_tuning import JUDGE_SYSTEM
 from test_kb_init_modules import CardGateway, _modules_lifecycle, _skeleton, _world_with_tools
 from test_kb_init_skeleton import _commit, _runtime, world  # noqa: F401
 
@@ -81,7 +82,7 @@ def test_metadata_cap_and_pinned_index_identity_fail_closed():
 def test_unlimited_prompt_identity_and_instructions_leave_legacy_unchanged():
     assert _Modules._input_options(SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=False))) == {}
     assert _Modules._input_options(SimpleNamespace(rt=SimpleNamespace(unlimited_subscription=True))) == {
-        "module_prompt_version": 2}
+        "module_prompt_version": 3}
     assert "never the full code" in SYSTEM_CARD and "source_pin" not in SYSTEM_CARD
     assert "never the full code" not in SYSTEM_CARD_IMPL
     assert all(text in SYSTEM_CARD_IMPL for text in
@@ -140,3 +141,76 @@ def call_json(self, role, *, system, prompt, **kwargs):
                        _modules_lifecycle(), "modules", dry_run=True, unlimited_subscription=True)
     assert record.status == "dry_run", record.problems
     assert any("key_files 'tools/lint/y.py' is not offered" in note for note in record.notes)
+
+
+@pytest.mark.parametrize("answer", ["no", "unsure", "unavailable", "subscription_unavailable"])
+def test_nonpass_source_review_drops_only_that_card_and_retains_real_unknown(world, answer):
+    from infermatrix_copilot.kb_service.models import ModelUnavailable
+    _world_with_tools(world)
+    _skeleton(world)
+
+    class SourceReviewGateway(ImplementationGateway):
+        judge_subscription_checks = 0
+
+        def subscription_billing(self, role):
+            if role.name == "judge":
+                self.judge_subscription_checks += 1
+                if answer == "subscription_unavailable" and self.judge_subscription_checks == 3:
+                    raise ModelUnavailable("scripted subscription recheck failure")
+            return True
+
+        def call_json(self, role, *, system, prompt, **kwargs):
+            if system == JUDGE_SYSTEM and 'tools/lint/' in prompt:
+                if answer == "unavailable":
+                    raise ModelUnavailable("scripted independent channel failure")
+                reply = super().call_json(role, system=system, prompt=prompt, **kwargs)
+                reply.data["dimensions"]["faithful"] = answer
+                reply.data["reasons"] = {"faithful": "runtime/conditional claim unsupported by shown source"}
+                return reply
+            return super().call_json(role, system=system, prompt=prompt, **kwargs)
+
+    gateway = SourceReviewGateway()
+    record = run_stage(_runtime(world, gateway, generator=ModelRole("generator", "zcode", "GLM-5.3")),
+                       _modules_lifecycle(), "modules", dry_run=True, unlimited_subscription=True)
+    assert record.status == "dry_run", record.problems
+    assert "tools/fmt/" in record.coverage["cards"] and "tools/lint/" not in record.coverage["cards"]
+    assert record.coverage["modules"]["after"] < 1 and "tools/lint/" in record.coverage["unrouted"]
+    rejected = record.verdicts["module:tools/lint/"]
+    assert rejected["kind"] == "module_prose" and rejected["verdict"] != "pass"
+    assert rejected["reasons"] and any("tools/lint/" in x for x in record.unfinished)
+    assert not record.depth and not record.discovery.get("acceptance")
+    assert len([call for call in gateway.calls if call["system"] == SYSTEM_CARD_IMPL]) == 2
+
+
+def test_independent_module_review_is_pinned_focused_bounded_and_three_yes(world):
+    _world_with_tools(world)
+    _commit(world["upstream"], {"tools/lint/unused.py": "def unused():\n    return private_unused_behavior()\n"},
+            "unreferenced source")
+    _skeleton(world)
+    gateway = ImplementationGateway()
+    record = run_stage(_runtime(world, gateway, generator=ModelRole("generator", "zcode", "GLM-5.3")),
+                       _modules_lifecycle(), "modules", dry_run=True, unlimited_subscription=True)
+    assert record.status == "dry_run", record.problems
+    reviews = [call for call in gateway.calls if call["system"] == JUDGE_SYSTEM]
+    assert len(reviews) == 2
+    for call in reviews:
+        assert call["role"] == "judge" and len(call["prompt"].encode()) <= MAX_CARD_BYTES
+        payload = json.loads(call["prompt"].split("\n", 1)[1].rsplit("</untrusted_data>", 1)[0])
+        assert set(payload["dimensions_to_answer"]) == {"faithful", "does_not_weaken", "non_contradictory"}
+        evidence = payload["evidence"][0]
+        assert evidence["source_pin"] == record.pin
+        assert {item["path"] for item in evidence["source_evidence"]} == {evidence["files"][0]["path"]}
+        assert "private_unused_behavior" not in call["prompt"] or "unused.py" in payload["change"]["after"]
+        assert "Printed/manual" in evidence["source_scope"]
+    assert all(verdict["model"] == "gpt-6-sol" and set(verdict["dimensions"].values()) == {"yes"}
+               for key, verdict in record.verdicts.items() if key.startswith("module:"))
+
+
+def test_legacy_module_cards_do_not_gain_an_independent_native_call(world):
+    _world_with_tools(world)
+    _skeleton(world)
+    gateway = CardGateway()
+    record = run_stage(_runtime(world, gateway), _modules_lifecycle(), "modules", dry_run=True)
+    assert record.status == "dry_run", record.problems
+    assert not any(call["system"] == JUDGE_SYSTEM for call in gateway.calls)
+    assert not record.verdicts