kb init: check rule claims and evidence at a pinned upstream SHA, and adapter lifecycle flips

doc/architecture/SPEC/kb_service.md

@@ -1,6 +1,6 @@
 # kb_service/ —— 规范
 
-<!-- verified-against: 2026-09-29 -->
+<!-- verified-against: 2026-09-30 -->
 
 `知识服务核心：仓库配置、账本、outbox、CLI · refactor-status: new`
 
@@ -291,3 +291,10 @@ diff 时保留原摘录（证据不会因此变得更"好过"）。只扩展带
 `merged_pr` 事件的证据新增 `merge_commit_sha`（旧事件经 `pull()` 查询）。`run_gate(evidence_for=...)` 由 `gate_and_stage`
 与事实证明共用同一个观测者（一次镜像同步）。首个线上运行中"所给 diff 在…之前截断"是 L2 判 unsure/fail 的主要原因，
 而整批 10 个 PR 的证据（约 105 KB）对每条规则都重复发送；按规则取证据后约 11 KB + 所引 PR 的完整 diff。
+
+## 2026-09-30 adapter 生命周期开关检查（kb init）
+`lifecycle_flip.check_lifecycle_flip(base_yaml, head_yaml, allowed=)`：两版 manifest 解析后删去允许的键路径，
+其余必须完全相等（比较解析值，注释与格式可保留）；`FLIP_KEYS` = `knowledge_lifecycle.enabled/mode`，
+`CALIBRATION_KEYS` = `knowledge_lifecycle.calibration_set`。`check_flip_to_shadow` 另要求 head 为
+`enabled: true` 且 `mode` 为 shadow（缺省即 shadow）。知识 L1 的白名单只含 `knowledge/`，adapter 这一处改动由它单独校验；
+调用方还需在 head adapter 上跑 `config.parse_lifecycle`。测试：`test_kb_lifecycle_flip.py`。

doc/architecture/SPEC/knowledge_service.md

@@ -1,6 +1,6 @@
 # knowledge_service/ — provider curation components
 
-<!-- verified-against: 2026-09-29 -->
+<!-- verified-against: 2026-09-30 -->
 
 The provider owns knowledge curation beneath the public SDK v1 facade.
 `KnowledgeCurator` composes four domain components over one explicit work
@@ -72,3 +72,9 @@ merge group 逐段验证、暂停清单与 `human-approved` 审批核对已于 2
 `recheck(upstream, facts, observer)` 由发布器调用：仓库一致、SHA 为 40 位、每条事实重新观测后逐字段相等，
 签名了不成立的必需事实或格式错误均为问题；上游不可读（未知 SHA、网络、API 错误）抛 `FactsError`，由调用方重试。
 仅标准库。`gate_verifier.verify_change` 不再拒绝带事实的判定（由发布器的 `recheck` 复核）。
+
+## 2026-09-30 钉点声明（pinned_claims）
+`pinned_claims` 供 `kb init` 使用：`PinnedObserver` 在本地克隆上以固定钉点实现 `facts.Observer`（另有
+`is_ancestor`），`check_rules` 在钉点上观测规则声明且要求引用的 PR 已合并进钉点历史，`Evidence` /
+`check_evidence` 把行区间绑定到内容哈希。服务与发布器的 `facts` 路径不变。详见
+[pinned_claims.md](knowledge_service/pinned_claims.md)。

doc/architecture/SPEC/knowledge_service/pinned_claims.md

@@ -0,0 +1,37 @@
+# knowledge_service/pinned_claims.py —— 规范
+
+<!-- verified-against: 2026-09-30 -->
+
+`LOC ~200 · 在钉住的上游 SHA 上核验规则声明与代码证据（kb init §9.1） · refactor-status: new`
+
+## 职责
+- `PinnedObserver(repo_dir, repository, pin, pull=)`：在本地 git 仓库（bare 或非 bare）上实现
+  `facts.Observer`，`head()` 固定返回钉住的 SHA；`top_level` / `path_exists` / `file_text` 读 git
+  对象，`pull(n)` 默认走 `gh api repos/{repo}/pulls/{n}`（可注入；任何查询失败——没有 `gh`、网络错误、非对象回答——都归一为 `FactsError`，原有的 `FactsError` 原样抛出）。另加协议之外的
+  `is_ancestor(sha)`：`sha` 是钉点或其祖先（非 40 位 SHA、镜像中不存在的对象、非 commit 对象为 False；候选对象损坏（`cat-file --batch-check` 同样输出 missing 但 stderr 报解包错误），或 git 读不到钉点历史（`merge-base` 此时也以 1 退出、只在 stderr 报 `Could not read`）时抛 `FactsError`，不把读失败当成"合入在钉点之后"）。钉点必须是 40 位
+  且在仓库中存在的提交，否则构造即抛 `FactsError`。
+- `check_rules(rule_texts, observer)`：对每条（活跃）规则用 `facts.claims_in` 抽取声明，按
+  `facts.observe` 在钉点观测；每条声明必须成立，`^[PR #N]` 还必须已合并进钉点历史
+  （`is_ancestor(merge_commit_sha)`）。返回以规则 ID 开头的问题列表。
+- `Evidence(path, start, end, sha256)`、`evidence_for(observer, path, start, end)`、
+  `check_evidence(entries, observer)`：把钉点上某文件的行区间（1 起、闭区间）绑定到内容哈希，
+  并能复核：文件存在、区间在文件内、哈希一致。
+
+## 公开契约
+哈希口径：`str.splitlines()` 切行（`\n`、`\r\n`、缺末尾换行视为相同），取 `start..end` 行以
+`\n` 连接再补一个 `\n`，UTF-8 编码后 SHA-256。`Evidence.to_dict()/from_dict()` 用于 init 记录。
+
+## 不变量
+- 与服务端 `facts.attest` 的声明抽取完全一致（同一 `claims_in`），只是观测点从上游 head 换成钉点。
+- 不修改 `facts.Observer` 协议，不影响服务与发布器。
+- 上游不可读（git 失败、未知提交、PR 查询失败）抛 `FactsError`，不当作"声明不成立"。
+
+## 边界 —— 不属于这里
+不写任何文件、不调用模型、不决定 init 各阶段如何处理问题（阻断或进清单由调用方决定）。
+
+## 依赖（允许）
+标准库 + `facts`；外部命令仅 `git` 与（默认 `pull`）`gh`。
+
+## 测试
+`test_kb_pinned_claims.py`（临时 git 仓库：钉点前后文件、符号、PR 合并先后与未合并、证据区间与
+哈希、换行风格无关、非法钉点）。

src/infermatrix_copilot/kb_service/lifecycle_flip.py

@@ -0,0 +1,125 @@
+"""Check that an adapter manifest edit changes only named lifecycle keys.
+
+``kb init`` touches one file outside ``knowledge/``: the adapter manifest, and
+only to switch the repository's knowledge service on (``enabled`` + ``mode``)
+or to name its calibration set. The knowledge L1 never sees that file (its
+whitelist is ``knowledge/``), so this check validates it on its own: both
+documents are parsed, the allowed key paths are removed from both, and what is
+left must be equal. Formatting and comments are free to stay as they are; the
+comparison is on parsed values.
+
+The caller additionally runs ``config.parse_lifecycle`` on the head adapter
+(it needs an adapter object, not text) so the new values are also valid.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+import yaml
+
+LIFECYCLE = "knowledge_lifecycle"
+FLIP_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "enabled"), (LIFECYCLE, "mode")})
+CALIBRATION_KEYS: frozenset[tuple[str, ...]] = frozenset({(LIFECYCLE, "calibration_set")})
+
+
+def _load(text: str, which: str) -> tuple[Any, list[str]]:
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError as exc:
+        return None, [f"{which} manifest is not valid YAML: {exc}"]
+    if not isinstance(data, dict):
+        return None, [f"{which} manifest is not a mapping"]
+    return data, []
+
+
+def _copy(value: Any, active: frozenset[int] = frozenset()) -> Any:
+    """A tree copy with YAML aliases UNshared: ``deepcopy`` keeps ``*alias``
+    nodes shared, so removing an allowed key through one path would also hide
+    a change made through another. A recursive alias is refused."""
+    if isinstance(value, (dict, list)):
+        if id(value) in active:
+            raise ValueError("manifest contains a recursive alias")
+        active = active | {id(value)}
+        if isinstance(value, dict):
+            return {key: _copy(item, active) for key, item in value.items()}
+        return [_copy(item, active) for item in value]
+    return value
+
+
+def _without(data: dict, paths: Iterable[tuple[str, ...]]) -> dict:
+    out = _copy(data)
+    for path in paths:
+        node: Any = out
+        for key in path[:-1]:
+            node = node.get(key) if isinstance(node, dict) else None
+            if node is None:
+                break
+        if isinstance(node, dict):
+            node.pop(path[-1], None)
+    return out
+
+
+def _typed(mapping: dict) -> dict[tuple[str, Any], Any]:
+    # 1, 1.0 and True are equal (and collide) as Python dict keys; YAML keeps them apart
+    return {(type(key).__name__, key): value for key, value in mapping.items()}
+
+
+def _diff_keys(base: Any, head: Any, prefix: str = "") -> list[str]:
+    """Paths where ``base`` and ``head`` differ, type-sensitively: YAML ``1``,
+    ``1.0`` and ``true`` are different values even though Python compares
+    them equal (and ``parse_lifecycle`` would stringify them differently)."""
+    where = prefix or "<root>"
+    if isinstance(base, dict) and isinstance(head, dict):
+        typed_base, typed_head = _typed(base), _typed(head)
+        out = []
+        for key in sorted(set(typed_base) | set(typed_head), key=str):
+            child = f"{prefix}.{key[1]}" if prefix else str(key[1])
+            if key not in typed_base or key not in typed_head:
+                out.append(child)
+            else:
+                out.extend(_diff_keys(typed_base[key], typed_head[key], child))
+        return out
+    if isinstance(base, list) and isinstance(head, list):
+        if len(base) != len(head):
+            return [where]
+        out = []
+        for index, (left, right) in enumerate(zip(base, head)):
+            out.extend(_diff_keys(left, right, f"{where}[{index}]"))
+        return out
+    same = type(base) is type(head) and (base == head or (base != base and head != head))  # .nan
+    return [] if same else [where]
+
+
+def check_lifecycle_flip(base_yaml: str, head_yaml: str, *,
+                         allowed: Iterable[tuple[str, ...]]) -> list[str]:
+    """Problems when the manifest changed anything besides the ``allowed``
+    key paths (empty list: the edit is confined to them)."""
+    allowed = list(allowed)
+    base, problems = _load(base_yaml, "base")
+    head, more = _load(head_yaml, "head")
+    problems += more
+    if problems:
+        return problems
+    try:
+        changed = _diff_keys(_without(base, allowed), _without(head, allowed))
+    except ValueError as exc:
+        return [str(exc)]
+    return [f"manifest key {key} changed; only "
+            f"{', '.join('.'.join(p) for p in sorted(allowed))} may change" for key in changed]
+
+
+def check_flip_to_shadow(base_yaml: str, head_yaml: str) -> list[str]:
+    """The init lifecycle flip: only ``enabled``/``mode`` change, and the head
+    turns the service on in shadow mode."""
+    problems = check_lifecycle_flip(base_yaml, head_yaml, allowed=FLIP_KEYS)
+    if problems:
+        return problems
+    section = (yaml.safe_load(head_yaml) or {}).get(LIFECYCLE)
+    if not isinstance(section, dict):
+        return [f"head manifest has no {LIFECYCLE} mapping"]
+    if section.get("enabled") is not True:
+        problems.append(f"{LIFECYCLE}.enabled must be true")
+    if section.get("mode", "shadow") != "shadow":  # parse_lifecycle's default
+        problems.append(f"{LIFECYCLE}.mode must be shadow")
+    return problems