Refine and recheck knowledge changes the gate rejected (design v8)

doc/architecture/SPEC/kb_service.md

@@ -218,3 +218,14 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
   拒绝（转人工，要求 `main` 允许发布器直接合并）；只有经 GraphQL 确认既不在合并队列中（必要时 `dequeuePullRequest`）也没有自动合并，才记为拒绝，否则保留意图，由恢复流程继续取消或接收
   后来发生的合并。存在未结的合并意图时该项跳过，直到恢复流程从 GitHub 得到结果。
 - external 的状态集合由 `merge.IN_FLIGHT` 派生，合并中的外部 PR 不会被重复评审。
+
+## 2026-09-29 v8：精炼—复检
+`refine`：质量门拒绝的服务自身变更（`intake`/`rebuild`/`refine` 类，暂存时 `failed` 且判定为 `fail`，或发布器本地门禁因变更本身
+拒绝 → `refine_needed`）从不合并。调度器（持有租约、仓库未暂停）把全部拒绝理由（本地门禁问题、L1 问题、L2 各维度理由、
+一致性冲突、判定理由）连同原操作交给生成器，在当前 main 上生成精炼后的变更集（kind `refine`，`refine_round`、`refine_history`），
+整体重跑 L1 + L2 + 一致性；通过则发布，原变更记为 `refined`（有 PR 则 `superseding` 并关闭旧 PR）。至多 `MAX_REFINES = 2`
+轮；仍失败 → `refine_exhausted`，人工队列一条，附每一轮理由，之后不再调用生成器。精炼按证据标记恢复（任意状态，含 `companion_pending`），崩溃后不会重复暂存；生成器拿到全部原始证据；重建沿用精炼轮次，
+上限不会因上下文变化而重置；精炼一次处理整个变更集，操作数上限为 6 × 事件数（且不少于原操作数）。
+服务自己的 PR 在任何状态下（`Ledger.changesets_for_pr`）都不会被当作他人的 PR 重新评审。
+外部 PR 被本地门禁拒绝仍为 `gate_failed` 转人工（他人 PR 的处理见后续）；仓库设置导致的拒绝（`main` 必须允许发布器直接合并）
+也转人工，精炼无法修复。

doc/features/knowledge-service.md

@@ -63,7 +63,8 @@ infermatrix-copilot kb rollback --to <sha>   # 回指到较早的快照（下一
 
 合并流程（v8）：`open_pr` 回执 → 为该 PR 的精确 head 签发判定，随 `merge` 项交给 GPU 盒发布器 → 发布器在
 `main` 与 PR head 的合并树上运行本地门禁，通过才 `gh pr merge`（每仓库同时至多 1 个）→ 合并后激活新快照。
-本地门禁拒绝的变更从不合并：上下文变化则在当前 main 上重建，变更本身的问题转人工。发版巡检在上游新 release 后运行
+本地门禁拒绝的变更从不合并：上下文变化则在当前 main 上重建；服务自己的变更被质量门或本地门禁拒绝时，由生成器按
+理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。发版巡检在上游新 release 后运行
 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。
 
 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。

src/infermatrix_copilot/kb_service/external.py

@@ -141,10 +141,11 @@ def _judged(rt, number: int, head: str, path: str) -> bool:
 
 
 def _ours(rt, number: int) -> bool:
+    """A PR the service opened, in ANY status (refine_needed, refine_exhausted,
+    superseding...): it is never judged again as someone else's PR."""
     for lifecycle in rt.registry.values():
-        for changeset in rt.ledger.changesets(lifecycle.repo, _ALL):
-            if changeset["pr_number"] == number and changeset["kind"] != KIND:
-                return True
+        if any(cs["kind"] != KIND for cs in rt.ledger.changesets_for_pr(lifecycle.repo, number)):
+            return True
     return False
 
 

src/infermatrix_copilot/kb_service/intake.py

@@ -143,10 +143,10 @@ def draft_prompt(repo: str, evidence: dict, files: dict[str, str], repo_dir: str
                   "reason", "evidence", "page_title")
 
 
-def _validate_reply(data: dict) -> None:
+def _validate_reply(data: dict, max_operations: int = MAX_OPS_PER_EVENT) -> None:
     operations = data.get("operations")
-    if not isinstance(operations, list) or len(operations) > MAX_OPS_PER_EVENT:
-        raise ValueError(f"operations must be a list of at most {MAX_OPS_PER_EVENT}")
+    if not isinstance(operations, list) or len(operations) > max_operations:
+        raise ValueError(f"operations must be a list of at most {max_operations}")
     for item in operations:
         if not isinstance(item, dict) or item.get("kind") not in INTAKE_KINDS:
             raise ValueError(f"operation kind must be one of {INTAKE_KINDS}")
@@ -162,7 +162,7 @@ def _validate_reply(data: dict) -> None:
 
 def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict,
                   files: dict[str, str], gateway: ModelGateway, generator: ModelRole,
-                  release: str, today: str) -> Draft:
+                  release: str, today: str, max_operations: int = MAX_OPS_PER_EVENT) -> Draft:
     prompt = draft_prompt(repo, evidence, files, repo_dir)
     attempts: list[dict] = []
     feedback = ""
@@ -172,7 +172,7 @@ def draft_changes(*, repo: str, repo_dir: str, event_id: int, evidence: dict,
         try:
             with trace_context(attempt=attempt):
                 reply = gateway.call_json(generator, system=SYSTEM, prompt=prompt + feedback,
-                                          validate=_validate_reply)
+                                          validate=lambda data: _validate_reply(data, max_operations))
         except ModelUnavailable as exc:
             if "failed its schema" not in str(exc):
                 raise  # the model itself is unavailable: the event waits

src/infermatrix_copilot/kb_service/ledger.py

@@ -400,6 +400,18 @@ def changesets(self, repo: str, statuses: tuple[str, ...]) -> list[dict[str, Any
             (repo, *statuses)).fetchall()
         return [_changeset_row(r) for r in rows]
 
+    def changesets_for_pr(self, repo: str, number: int) -> list[dict[str, Any]]:
+        """Every change set of ``repo`` that has PR ``number``, whatever its status."""
+        rows = self._conn.execute(
+            "SELECT * FROM changesets WHERE repo=? AND pr_number=? ORDER BY created_at", (repo, number)).fetchall()
+        return [_changeset_row(r) for r in rows]
+
+    def changesets_of_kind(self, repo: str, kind: str) -> list[dict[str, Any]]:
+        """Every change set of ``kind`` in ``repo``, whatever its status."""
+        rows = self._conn.execute(
+            "SELECT * FROM changesets WHERE repo=? AND kind=? ORDER BY created_at", (repo, kind)).fetchall()
+        return [_changeset_row(r) for r in rows]
+
     def record_verdict(self, repo: str, *, layer: str, verdict: str, changeset_id: str | None = None,
                        block_id: str = "", model: str = "", model_version: str = "",
                        detail: dict | None = None) -> None:

src/infermatrix_copilot/kb_service/merge.py

@@ -48,6 +48,9 @@
 RESIGN_GATE = ("outside its issue window", "no valid signed verdict")
 REBUILD_GATE = ("context changed since the verdict", "pages changed since the consistency judgement",
                 "does not merge cleanly", "signed patch manifest does not match")
+# a refused merge caused by how the repository is set up, not by the change:
+# refining cannot fix it, people must (it would only loop)
+CONFIG_MERGE = ("must allow a direct merge",)
 # a refused local-gate merge that is about the moment, not the change: sign again
 TRANSIENT_MERGE = ("main kept moving", "not merging:", "head of PR", "head moved", "git ", "gh ",
                    "did not merge", "is not open", "verdict: ")
@@ -242,11 +245,19 @@ def _apply_merge_ack(rt, changeset: dict, ack: dict) -> None:
     elif any(m in text for m in TRANSIENT_MERGE) and not ack.get("problems"):
         rt.ledger.update_changeset(changeset_id, status="pr_open", pending_item=None)
     else:  # the local gate refused the change itself: it is not merged
-        rt.ledger.update_changeset(changeset_id, status="gate_failed", pending_item=None,
-                                   detail={**changeset["detail"], "gate_problems": (ack.get("problems") or [])[:50]})
-        rt.ledger.enqueue_human(repo, f"local gate refused PR #{changeset['pr_number']}: {text[:400]}",
-                                changeset_id)
-        _outcome(rt, changeset, "gate_failed", problems=(ack.get("problems") or [])[:20])
+        from .refine import REFINABLE
+
+        problems = (ack.get("problems") or [])[:50]
+        if changeset["kind"] in REFINABLE and not any(m in text for m in CONFIG_MERGE):
+            # our own change: refined with these reasons and checked again
+            rt.ledger.update_changeset(changeset_id, status="refine_needed", pending_item=None,
+                                       detail={**changeset["detail"], "gate_problems": problems})
+        else:
+            rt.ledger.update_changeset(changeset_id, status="gate_failed", pending_item=None,
+                                       detail={**changeset["detail"], "gate_problems": problems})
+            rt.ledger.enqueue_human(repo, f"local gate refused PR #{changeset['pr_number']}: {text[:400]}",
+                                    changeset_id)
+        _outcome(rt, changeset, "gate_failed", problems=problems[:20])
 
 
 def sign_verdict(rt, changeset: dict) -> dict:
@@ -405,7 +416,11 @@ def to_people(reason: str) -> None:
         new_id = gate_and_stage(rt, lifecycle, owner, kind="rebuild", base=base, base_sha=base_sha,
                                 external=external, operations=operations, result=result,
                                 evidence=evidence, event_ids=[], release=detail["release"],
-                                draft_keys=[])
+                                draft_keys=[],
+                                # a rebuild continues the change's refine lineage: the
+                                # limit never resets because context moved meanwhile
+                                extra_detail={k: detail[k] for k in ("refine_round", "refine_history")
+                                              if k in detail})
     staged = rt.ledger.changeset(new_id)
     if staged["status"] not in ("gated", *IN_FLIGHT):
         # the rebuild itself did not pass the gate: nothing replaces the PR
@@ -594,6 +609,10 @@ def advance(rt, lifecycle) -> list[str]:
                     **current["detail"], "verdict_issued_at": envelope["payload"]["issued_at"]})
                 merging = [changeset]
                 events.append(f"merge issued {changeset['id']}")
+    from .refine import refine_pending
+
+    if not paused:
+        events += refine_pending(rt, lifecycle)
     return events + verify_merged(rt, lifecycle)
 
 

src/infermatrix_copilot/kb_service/refine.py

@@ -0,0 +1,158 @@
+"""Refine and recheck: what happens to a change the quality gate rejected.
+
+A change the gate rejects is never merged (design v8, D12). For the service's
+own change sets the generator gets the rejection reasons, proposes a refined
+change on current main, and the WHOLE gate runs again (L1, L2 per block,
+consistency; nothing from the rejected judgement is reused). At most
+``MAX_REFINES`` refinements per original change; after that it stays unmerged
+and goes to people with every round's reasons.
+
+Triggers: a change set the gate failed when it was staged (``failed`` with a
+``fail`` decision), and a PR the publisher's local gate refused because of the
+change itself (``refine_needed``). A refinement of a change that already has a
+PR replaces it the way a rebuild does: the old PR waits in ``superseding``
+until its close is observed, the refined change gets a PR of its own.
+"""
+
+from __future__ import annotations
+
+import uuid
+
+MAX_REFINES = 2
+REFINE_MARK = "refine of "
+REFINABLE = ("intake", "rebuild", "refine")
+
+
+def reasons_of(changeset: dict) -> list[str]:
+    """Everything the gate said against this change, most specific first."""
+    detail = changeset["detail"]
+    reasons = [str(p) for p in detail.get("gate_problems") or []]
+    decision = detail.get("decision") or {}
+    for issue in decision.get("l1_issues") or []:
+        reasons.append(f"L1 {issue.get('code')} {issue.get('path', '')} {issue.get('detail', '')}".strip())
+    for block in decision.get("blocks") or []:
+        if block.get("verdict") == "fail":
+            why = "; ".join(f"{k}: {v}" for k, v in (block.get("reasons") or {}).items())
+            reasons.append(f"L2 rejected {block.get('rule_id') or block.get('path')}: {why}".strip())
+    for item in decision.get("consistency") or []:
+        if item.get("verdict") == "conflict":
+            reasons.append(f"consistency conflict in {item.get('owner_dir')}: {item.get('conflicts')}")
+    reasons += [str(r) for r in decision.get("reasons") or []]
+    return list(dict.fromkeys(r for r in reasons if r))[:30]
+
+
+def needs_refinement(changeset: dict) -> bool:
+    detail = changeset["detail"]
+    if changeset["kind"] not in REFINABLE or detail.get("refined_as"):
+        return False
+    if changeset["status"] == "refine_needed":
+        return True
+    return changeset["status"] == "failed" and (detail.get("decision") or {}).get("status") == "fail"
+
+
+def _refine_of(rt, repo: str, old_id: str) -> str | None:
+    """A refinement already staged for ``old_id`` (resumed, never staged twice),
+    in whatever status staging left it (companion_pending included)."""
+    for candidate in rt.ledger.changesets_of_kind(repo, "refine"):
+        try:
+            evidence = rt.load_changeset_files(candidate["id"]).get("evidence") or []
+        except (OSError, ValueError):
+            continue
+        if any(str(item.get("source_reference") or "") == REFINE_MARK + old_id for item in evidence):
+            return candidate["id"]
+    return None
+
+
+def refine(rt, lifecycle, changeset: dict) -> str | None:
+    """Refine one rejected change set; returns the refined change set's id or
+    None (sent to people). Needs the scheduler's lease."""
+    from ..knowledge_service.lifecycle import LifecycleError
+    from ..trace_store import accepted_key, trace_context
+    from .intake import MAX_OPS_PER_EVENT, draft_changes
+    from .merge import _outcome, _request_close
+    from .models import ModelUnavailable
+    from .runtime import gate_and_stage, publish
+
+    owner = getattr(rt, "lease_owner", None)
+    if not owner:
+        return None
+    detail = changeset["detail"]
+    reasons = reasons_of(changeset) or ["the quality gate rejected the change"]
+    round_ = int(detail.get("refine_round") or 0) + 1
+    history = [*(detail.get("refine_history") or []), {"changeset": changeset["id"], "reasons": reasons}]
+
+    def to_people(why: str) -> None:
+        rt.ledger.update_changeset(changeset["id"], status="refine_exhausted",
+                                   detail={**detail, "refine_history": history})
+        lines = [f"round {i}: {h['changeset']}: " + "; ".join(h["reasons"])[:300] for i, h in enumerate(history)]
+        rt.ledger.enqueue_human(lifecycle.repo, f"{why}; not merged. " + " | ".join(lines)[:1800], changeset["id"])
+        _outcome(rt, changeset, "refine_exhausted", rounds=len(history))
+
+    if round_ > MAX_REFINES:
+        to_people(f"still rejected after {MAX_REFINES} refinements")
+        return None
+    new_id = _refine_of(rt, lifecycle.repo, changeset["id"])
+    if new_id is None:
+        base_sha = rt.knowledge.fetch()
+        base = rt.knowledge.knowledge_files(base_sha)
+        external = rt.knowledge.external_texts(base_sha)
+        evidence = rt.load_changeset_files(changeset["id"]).get("evidence") or []
+        release, today = rt.release_for(lifecycle.repo), rt.today()
+        # ALL the evidence: an intake change set can combine several events,
+        # and a refinement must not silently drop the rules of later ones
+        prompt_evidence = {"title": "; ".join(str(e.get("title") or "") for e in evidence if e.get("title")),
+                           "body": "\n\n".join(str(e.get("body") or "") for e in evidence if e.get("body")),
+                           "changed_files": sorted({f for e in evidence for f in e.get("changed_files") or []}),
+                           "evidence": evidence,
+                           "previous_operations": detail.get("operations") or [],
+                           "gate_feedback": reasons,
+                           "instruction": "The quality gate rejected the previous operations for the reasons in "
+                                          "gate_feedback. Propose corrected operations that fix exactly those "
+                                          "problems, or none if the change should not be made."}
+        key, holder = f"refine:{lifecycle.repo}:{changeset['id']}:{uuid.uuid4().hex[:8]}", {}
+        try:
+            with trace_context(draft_key=key, _accepted=holder, step="refine", refine_of=changeset["id"]):
+                # the whole change set is refined at once: as many operations as
+                # the events it combines may carry
+                limit = max(MAX_OPS_PER_EVENT * max(1, len(evidence)), len(detail.get("operations") or []))
+                draft = draft_changes(repo=lifecycle.repo, repo_dir=lifecycle.knowledge_dir, event_id=0,
+                                      evidence=prompt_evidence, files=base, gateway=rt.gateway,
+                                      generator=rt.generator, release=release, today=today,
+                                      max_operations=limit)
+        except ModelUnavailable:
+            return None  # the pinned generator is unavailable: tried again next pass
+        except LifecycleError as exc:
+            to_people(f"refinement could not be drafted: {exc}")
+            return None
+        if draft.rejected or not draft.operations or draft.result is None:
+            to_people("the generator proposed no valid refinement")
+            return None
+        mark = {"source_reference": REFINE_MARK + changeset["id"], "title": "; ".join(reasons)[:300]}
+        new_id = gate_and_stage(rt, lifecycle, owner, kind="refine", base=base, base_sha=base_sha,
+                                external=external, operations=draft.operations, result=draft.result,
+                                evidence=[*evidence, mark], event_ids=[], release=release,
+                                draft_keys=[accepted_key(key, holder)] if accepted_key(key, holder) else [],
+                                extra_detail={"refine_round": round_, "refine_history": history})
+    open_pr = changeset.get("pr_number") is not None
+    rt.ledger.update_changeset(changeset["id"], status="superseding" if open_pr else "refined",
+                               detail={**detail, "refined_as": new_id, "refine_history": history,
+                                       "superseded_because": "refined after the quality gate rejected it"})
+    _outcome(rt, changeset, "refined", replaced_by=new_id, round=round_)
+    if open_pr:
+        _request_close(rt, lifecycle.repo, rt.ledger.changeset(changeset["id"]))
+    if rt.ledger.changeset(new_id)["status"] == "gated":
+        publish(rt, lifecycle, new_id)
+    return new_id
+
+
+def refine_pending(rt, lifecycle) -> list[str]:
+    """One pass: refine every rejected change set of this repository."""
+    events: list[str] = []
+    if not getattr(rt, "lease_owner", None):
+        return events
+    for changeset in rt.ledger.changesets(lifecycle.repo, ("failed", "refine_needed")):
+        if not needs_refinement(changeset):
+            continue
+        new_id = refine(rt, lifecycle, changeset)
+        events.append(f"refined {changeset['id']} as {new_id}" if new_id else f"refine_stopped {changeset['id']}")
+    return events