Only conflicts involving the changed rules fail the consistency check

doc/architecture/SPEC/kb_service.md

@@ -275,3 +275,10 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
   （幂等 upsert）记入退役账本，下个发版的巡检据此 purge。
 - `kb status` 新增 `unknown_commits`：未处置的未知提交及其接受请求的状态。
 - 测试：`test_kb_accept_unknown.py`。
+
+## 2026-09-29 一致性冲突只算本次改动的
+一致性评审看整个 owner 目录，但只有**涉及本次改动的规则**（新增、编辑、退役、被替代的规则 ID）的冲突才使变更失败；
+两条都未被改动的规则之间的冲突已经在 main 上，记入该目录的 `preexisting`，不计入变更（精炼也无法修复它，否则该目录
+的任何变更都会被永久挡住并白白消耗精炼轮次）。没有写出规则 ID 的冲突仍算本次改动的（失败即关闭）。评审提示中带
+`changed_rule_ids`。校准不传 `changed`，保持整目录严格判定。首个线上运行即遇到此情况：configuration 目录中 CONF-1a..4a 在
+两页重复、DIFF-2s/2t、SERV-4o/4r5 在 main 上已冲突，导致两批 intake 全部 `refine_exhausted`。

src/infermatrix_copilot/kb_service/gate.py

@@ -169,7 +169,12 @@ def validate(data: dict) -> None:
 
 
 def check_consistency(head: dict[str, str], owner_dirs: list[str], *, gateway: ModelGateway,
-                      judge: ModelRole) -> list[dict]:
+                      judge: ModelRole, changed: set[str] | None = None) -> list[dict]:
+    """``changed``: the rule IDs the change adds, edits, retires or supersedes.
+    Only a conflict naming one of them is the change's; a conflict between
+    rules it does not touch was already on main, is kept as ``preexisting``,
+    and never fails the change (refining could not fix it). A conflict that
+    names no rule counts against the change. None: every conflict counts."""
     from ..knowledge_service.l1 import _sha as sha256_text
 
     results = []
@@ -187,7 +192,8 @@ def check_consistency(head: dict[str, str], owner_dirs: list[str], *, gateway: M
                     rules.append({"page": path, "rule_id": section.rule_id,
                                   "supersedes": footer.supersedes,
                                   "text": section.body_without_footer})
-        prompt = "<untrusted_data>\n" + json.dumps({"directory": directory, "rules": rules},
+        prompt = "<untrusted_data>\n" + json.dumps({"directory": directory, "rules": rules,
+                                                   **({"changed_rule_ids": sorted(changed)} if changed is not None else {})},
                                                    ensure_ascii=False, indent=1).replace("<", "\\u003c") \
             + "\n</untrusted_data>\n"
 
@@ -202,13 +208,47 @@ def validate(data: dict) -> None:
             verdict, conflicts = reply.data["verdict"], reply.data.get("conflicts") or []
         except ModelUnavailable as exc:
             verdict, conflicts = "unsure", [["", "", str(exc)]]
+        preexisting: list = []
+        if verdict == "conflict" and changed is not None and conflicts:
+            ours = [c for c in conflicts if _names_changed(c, changed)]
+            preexisting = [c for c in conflicts if not _names_changed(c, changed)]
+            conflicts = ours
+            if not ours:
+                verdict = "consistent"
         results.append({
-            "owner_dir": directory, "verdict": verdict, "conflicts": conflicts,
+            "owner_dir": directory, "verdict": verdict, "conflicts": conflicts, "preexisting": preexisting,
             "pages": {p: sha256_text(t) for p, t in pages.items()},
         })
     return results
 
 
+def _names_changed(conflict, changed: set[str]) -> bool:
+    """Whether a reported conflict involves a changed rule. Anything malformed,
+    or naming no rule (empty, null, not a string), counts as the change's."""
+    if not isinstance(conflict, (list, tuple)) or len(conflict) < 2:
+        return True
+    ids = conflict[:2]
+    if not all(isinstance(i, str) and i.strip() for i in ids):
+        return True
+    return any(i.strip() in changed for i in ids)
+
+
+def _changed_rule_ids(blocks, base: dict[str, str], head: dict[str, str]) -> set[str]:
+    """The rule IDs of every changed rule block, with the ``###`` rules nested
+    inside them (on either side of the change)."""
+    out: set[str] = set()
+    for block in blocks:
+        if block.kind != "rule":
+            continue
+        out.add(block.rule_id)
+        for files in (base, head):
+            try:
+                out.update(Page.parse(files[block.path]).rule(block.rule_id).nested_rule_ids)
+            except (KeyError, LifecycleError):
+                continue
+    return out
+
+
 def run_gate(*, base: dict[str, str], head: dict[str, str], changes: list[Change],
              external_texts: dict[str, str], evidence: list[dict], gateway: ModelGateway,
              judge: ModelRole, release: str, repo_dir: str,
@@ -255,7 +295,8 @@ def run_gate(*, base: dict[str, str], head: dict[str, str], changes: list[Change
                                                for b in blocks if b.verdict == "fail"], l1, blocks,
                             upstream=upstream, facts=signed_facts)
     owner_dirs = sorted({_owner_dir(b.path) for b in l1.blocks})
-    consistency = check_consistency(head, owner_dirs, gateway=gateway, judge=judge)
+    consistency = check_consistency(head, owner_dirs, gateway=gateway, judge=judge,
+                                    changed=_changed_rule_ids(l1.blocks, base, head))
     if any(item["verdict"] == "conflict" for item in consistency):
         return GateDecision("fail", reasons + ["change set is inconsistent"], l1, blocks, consistency,
                             upstream, signed_facts)

test/test_kb_intake_gate.py

@@ -559,3 +559,42 @@ def answer(role, prompt):
     assert run_intake(rt, lifecycle) is None
     drafted = rt.ledger.events("demo", "drafted")
     assert len(drafted) == 1 and drafted[0]["detail"] == "demo-intake-usurped"
+
+
+def _conflicting(pairs, seen=None):
+    def answer(role, prompt):
+        if '"directory"' in prompt:
+            if seen is not None:
+                seen.append(prompt)
+            return {"verdict": "conflict", "conflicts": [[a, b, "they disagree"] for a, b in pairs]}
+        return _judge_all("yes")(role, prompt)
+    return answer
+
+
+ADD = [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #11"))]
+
+
+def test_a_conflict_already_on_main_does_not_fail_the_change():
+    decision = _gate(ADD, _conflicting([("DEMO-1a", "OLD-9z")]))        # neither rule is the change's
+    assert decision.status == "pass", decision.reasons
+    (item,) = decision.consistency
+    assert item["verdict"] == "consistent" and item["conflicts"] == []
+    assert item["preexisting"] == [["DEMO-1a", "OLD-9z", "they disagree"]]
+    decision = _gate(ADD, _conflicting([("DEMO-1a", "OLD-9z"), ("DEMO-2a", "DEMO-1a")]))
+    assert decision.status == "fail"                                    # the new rule is in it: ours
+    assert decision.consistency[0]["conflicts"] == [["DEMO-2a", "DEMO-1a", "they disagree"]]
+    assert _gate(ADD, _conflicting([("", "")])).status == "fail"        # names no rule: counted against it
+    assert _gate(ADD, _conflicting([(None, None)])).status == "fail"    # null IDs: likewise
+
+
+def test_a_rule_nested_in_a_changed_rule_is_part_of_the_change():
+    text = _rule("DEMO-2a", "PR #11") + "\n### DEMO-2a1 — a nested case\n\n- 强制：嵌套规则。 ^[PR #11]\n"
+    nested = [Op("add", PAGE, "DEMO-2a", text)]
+    assert _gate(nested, _conflicting([("DEMO-2a1", "DEMO-1a")])).status == "fail"
+
+
+def test_the_consistency_judge_is_told_which_rules_changed():
+    seen: list[str] = []
+    _gate(ADD, _conflicting([], seen))
+    data = json.loads(seen[0].split("<untrusted_data>\n", 1)[1].rsplit("\n</untrusted_data>", 1)[0])
+    assert data["changed_rule_ids"] == ["DEMO-2a"]