Add kb accept-unknown: judge an unknown knowledge commit through the gate (design v8 8.3)

doc/architecture/SPEC/kb_service.md

@@ -258,3 +258,20 @@ issue（同一报告只开一个），标签不存在时不带标签重试，绝
   其他任何回执结束这段连续失败。
 - 测试：`test_kb_upstream_facts.py`（声明抽取、符号定义、证明与复核的一致/不一致/不可达/伪造、真实 git 镜像、
   门禁在 L2 前失败、发布器一致时合并、不一致或不可达时不合并并在 24 小时后转人工、未配置上游不合并）。
+
+## 2026-09-29 v8：`kb accept-unknown`
+第二种处置未知提交的方式（与精确 revert 并列，design §8.3）：
+- `kb accept-unknown SHA`（至少 7 位、唯一前缀）只校验并记录请求（`accept_request:<sha>`，已处置或不是未知提交 → 退出码 2）；
+  `kb serve` 持有租约，在下一个 tick 由 `accept.accept_pending` 判定。
+- 判定：该提交自身的改动（第一父 → 提交）作为 `accept` 变更集走**完整**门禁（L1、L2、一致性、上游事实），要求当前通过的
+  评审校准；只能是单个启用且公开的仓库范围、全部为受治理页面、非根提交。服务的 revert PR 仍在创建中（`pr_requested`）
+  时等到下一 tick。
+- 只有 `pass` 才处置：`disposed:<sha> = <sha>`（提交在历史中即被信任）、记录 `state=accepted`、丢弃该提交的 `unrecorded`
+  intake 候选、已打开的 revert PR 转 `superseding` 并下发 `close`。暂停不自动解除：由人 `kb resume`。
+  其余结果（门禁未过、无校准、范围或路径不合格）不处置，转人工并写入请求记录。
+- 可恢复：通过后先把记录写为 `accepting`，清理（按 `ledger.event_by_external_id` 直接定位候选事件、revert 转 `superseding`）
+  完成后才写 `disposed:` 与 `accepted`；中途崩溃由下一 tick 完成清理且不再次调用评审。
+- 与外部 PR 一样，L1 发现该提交退役/删除的规则写入 `accept` 变更集的 `retirements`/`purges`，清理时经 `record_retirements`
+  （幂等 upsert）记入退役账本，下个发版的巡检据此 purge。
+- `kb status` 新增 `unknown_commits`：未处置的未知提交及其接受请求的状态。
+- 测试：`test_kb_accept_unknown.py`。

doc/features/knowledge-service.md

@@ -68,7 +68,7 @@ infermatrix-copilot kb rollback --to <sha>   # 回指到较早的快照（下一
 理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。他人的知识 PR 每天检查一次：结果写在该 PR 上唯一一条
 “知识质量门”评论里（原地更新），未通过则不合并，作者修改后次日复检。
 有人绕过门禁直接合并/推送知识变更时，服务立即暂停该仓库、停止激活新快照、告警，并开出撤回 PR（不自动合并）；
-合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁。发版巡检在上游新 release 后运行
+合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁；也可以用 `kb accept-unknown <sha>` 让该提交本身完整通过门禁后保留它（通过才算处置，仍需 `kb resume`）。发版巡检在上游新 release 后运行
 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。
 
 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。

src/infermatrix_copilot/kb_service/accept.py

@@ -0,0 +1,159 @@
+"""Accepting an unknown knowledge commit instead of reverting it (design v8 §8.3).
+
+``kb accept-unknown <sha>`` is the second way to dispose of a commit that
+reached ``knowledge/`` on main without passing the gate: the commit's own
+change (its first parent to the commit) is judged by the FULL gate (L1, L2,
+consistency, upstream facts, a current judge calibration) as a change set of
+kind ``accept``. Only a pass disposes of it: the commit becomes trusted, its
+open revert PR is closed, and its queued intake candidate is dropped. Anything
+else leaves it undisposed and tells people why.
+
+The CLI only records the request; the scheduler, which holds the ledger lease
+for its whole life, judges it on its next tick.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+
+from .audit import DISPOSED, UNKNOWN, _governed, disposals
+
+ACCEPT = "accept_request:"   # "*" cursors: requests waiting for the scheduler
+KIND = "accept"
+
+
+class AcceptError(ValueError):
+    """The request cannot be recorded (no such unknown commit, already disposed...)."""
+
+
+def resolve(ledger, ref: str) -> str:
+    """The full SHA of an open unknown commit from a unique prefix (7+ hex)."""
+    if not re.fullmatch(r"[0-9a-f]{7,40}", ref):
+        raise AcceptError(f"not a commit SHA: {ref!r}")
+    names = [n[len(UNKNOWN):] for n in ledger.cursors_with_prefix("*", UNKNOWN)]
+    matches = [sha for sha in names if sha.startswith(ref)]
+    if len(matches) != 1:
+        raise AcceptError(f"{ref} matches {len(matches)} unknown commits")
+    return matches[0]
+
+
+def request_accept(ledger, ref: str, *, at: float) -> str:
+    sha = resolve(ledger, ref)
+    disposed = {name[len(DISPOSED):] for name in ledger.cursors_with_prefix("*", DISPOSED)}
+    if sha in disposed:
+        raise AcceptError(f"{sha[:12]} is already disposed of")
+    ledger.set_cursor("*", ACCEPT + sha, json.dumps({"at": at, "state": "requested"}))
+    return sha
+
+
+def accept_pending(rt) -> list[str]:
+    """Judge every recorded request (needs the scheduler's lease). Returns events."""
+    if not getattr(rt, "lease_owner", None):
+        return []
+    events = []
+    for name, raw in sorted(rt.ledger.cursors_with_prefix("*", ACCEPT).items()):
+        request = json.loads(raw)
+        if request.get("state") != "requested":
+            continue
+        sha = name[len(ACCEPT):]
+        outcome, detail = _accept(rt, sha)
+        rt.ledger.set_cursor("*", name, json.dumps({**request, "state": outcome, "detail": detail,
+                                                    "decided_at": rt.clock()}))
+        events.append(f"accept {sha[:12]} {outcome}: {detail}")
+    return events
+
+
+def _accept(rt, sha: str) -> tuple[str, str]:
+    from .gate import run_gate
+    from .runtime import calibration_current
+
+    raw = rt.ledger.get_cursor("*", UNKNOWN + sha)
+    if raw is None:
+        return "refused", "not an unknown commit"
+    record = json.loads(raw)
+    if record.get("state") == "accepting":  # passed; interrupted before its cleanup finished
+        return _finish(rt, sha, record)
+    if sha in disposals(rt):
+        return "done", "already disposed of"
+    paths, parents, scopes = record.get("paths") or [], record.get("parents") or [], record.get("scopes") or []
+    lifecycle = rt.registry.get(scopes[0]) if len(scopes) == 1 else None
+
+    def refuse(why: str) -> tuple[str, str]:
+        rt.ledger.enqueue_human(lifecycle.repo if lifecycle else next(iter(rt.registry)),
+                                f"accept-unknown {sha[:12]} refused: {why}")
+        return "refused", why
+
+    if lifecycle is None or not lifecycle.enabled or not lifecycle.publishes:
+        return refuse(f"it touches {scopes}: only a change inside one enabled, public repository can be accepted")
+    if not parents or any(not _governed(p) for p in paths):
+        return refuse("it touches paths the gate does not govern (or is a root commit): revert it by hand")
+    revert = record.get("revert_changeset")
+    if revert and rt.ledger.changeset(revert)["status"] == "pr_requested":
+        return "requested", "its revert PR is being opened; judged again next tick"
+    if not calibration_current(rt, lifecycle):
+        return refuse("the judge calibration is missing or outdated")
+    first = parents[0]
+    base, head = rt.knowledge.knowledge_files(first), rt.knowledge.knowledge_files(sha)
+    changes = [_change(e) for e in rt.knowledge.raw_manifest(first, sha)]
+    decision = run_gate(base=base, head=head, changes=changes, external_texts=rt.knowledge.external_texts(sha),
+                        evidence=[{"source_reference": f"commit {sha}", "title": record.get("reason", ""),
+                                   "body": rt.knowledge.diff_text(first, sha, paths)[:8000]}],
+                        gateway=rt.gateway, judge=rt.judge, release=rt.release_for(lifecycle.repo),
+                        repo_dir=lifecycle.knowledge_dir, protected_rules=lifecycle.protected_rules,
+                        retire_ratio=lifecycle.retire_ratio, max_files=lifecycle.max_files,
+                        facts=rt.upstream_facts(lifecycle))
+    from .external import _lifecycle_bookkeeping
+
+    changeset_id = rt.ledger.new_changeset_id(lifecycle.repo, KIND)
+    rt.save_changeset_files(changeset_id, {"base_sha": first, "files": {}, "deleted": [], "evidence": []})
+    passed = decision.status == "pass"
+    rt.ledger.stage_intake(rt.lease_owner, lifecycle.repo, changeset_id, kind=KIND,
+                           status="accepted" if passed else "failed", verdicts=[], human_reason="",
+                           drafted_events=[], detail={
+                               "accepts": sha, "base_sha": first, "operations": [], "judge": rt.judge.label(),
+                               "decision": decision.to_dict(), "release": rt.release_for(lifecycle.repo),
+                               # what L1 found it retiring/purging (as for external PRs): the
+                               # retirement ledger drives the next release's purge
+                               **_lifecycle_bookkeeping(decision.l1, head)})
+    rt.trace("decision", context={"repo": lifecycle.repo, "changeset_id": changeset_id, "playbook": "kb-audit",
+                                  "step": "accept-unknown"},
+             result={"status": decision.status, "sha": sha, "reasons": decision.reasons})
+    if not passed:
+        return refuse(f"the gate did not pass it ({decision.status}): " + "; ".join(decision.reasons)[:400])
+    # recorded first, disposed of last: an interrupted cleanup is finished by
+    # the next tick (without judging again), never skipped
+    record = {**record, "state": "accepting", "accepted_by": changeset_id, "repo": lifecycle.repo}
+    rt.ledger.set_cursor("*", UNKNOWN + sha, json.dumps(record))
+    return _finish(rt, sha, record)
+
+
+def _finish(rt, sha: str, record: dict) -> tuple[str, str]:
+    """Idempotent: record its retirements, drop the candidate, close the revert
+    PR, then dispose."""
+    from .merge import _request_close, record_retirements
+
+    changeset_id, repo = record["accepted_by"], record["repo"]
+    record_retirements(rt, rt.ledger.changeset(changeset_id))   # upserts: safe to repeat
+    event = rt.ledger.event_by_external_id(repo, "unrecorded", sha)
+    if event is not None and event["status"] == "pending":
+        rt.ledger.set_event_status(event["id"], "superseded", f"accepted as {changeset_id}")
+    revert = record.get("revert_changeset")
+    if revert:
+        pending = rt.ledger.changeset(revert)
+        if pending["status"] == "revert_open":
+            rt.ledger.update_changeset(revert, status="superseding", detail={
+                **pending["detail"], "rebuilt_as": changeset_id,
+                "superseded_because": f"{sha[:12]} was accepted through the gate"})
+            _request_close(rt, repo, rt.ledger.changeset(revert))
+    # the commit disposes of itself: trusted wherever it is part of the history
+    rt.ledger.set_cursor("*", DISPOSED + sha, sha)
+    rt.ledger.set_cursor("*", UNKNOWN + sha, json.dumps({**record, "state": "accepted"}))
+    return "accepted", f"passed the gate as {changeset_id}; run `kb resume --repo {repo}` to resume merging"
+
+
+def _change(entry: dict):
+    from ..knowledge_service.l1 import Change
+
+    status = entry["status"] if entry["status"] in "AMD" else "M"
+    return Change(entry["path"], status, entry["old_mode"], entry["new_mode"])

src/infermatrix_copilot/kb_service/cli.py

@@ -4,7 +4,8 @@
     kb status                         repositories, modes, generations, queues
     kb pause  (--repo R | --all) --reason TEXT
     kb resume (--repo R | --all)
-    kb control                        refresh the signed control record and hold list now
+    kb control                        refresh the signed control record now
+    kb accept-unknown SHA             judge an unknown knowledge commit through the gate instead of reverting it
     kb run --playbook kb-intake --repo R   run one knowledge playbook once
     kb calibrate --repo R             run the judge over the repository's calibration set
     kb serve [--once]                 the scheduler (holds the single-writer lease)
@@ -80,6 +81,24 @@ def _outbox(state_dir: Path, ledger):
     return Outbox(state_dir, _signing_key(), ledger, clock=time.time)
 
 
+def _unknown_status(ledger) -> dict:
+    """Unknown knowledge commits not disposed of, with any accept request."""
+    from .accept import ACCEPT
+    from .audit import DISPOSED, UNKNOWN
+
+    disposed = {n[len(DISPOSED):] for n in ledger.cursors_with_prefix("*", DISPOSED)}
+    out = {}
+    for name, raw in ledger.cursors_with_prefix("*", UNKNOWN).items():
+        sha = name[len(UNKNOWN):]
+        if sha in disposed:
+            continue
+        record = json.loads(raw)
+        request = ledger.get_cursor("*", ACCEPT + sha)
+        out[sha] = {"state": record.get("state"), "reason": record.get("reason"),
+                    "accept": json.loads(request) if request else None}
+    return out
+
+
 def _refresh(state_dir: Path, ledger, registry) -> None:
     _outbox(state_dir, ledger).transition(lambda: None)
 
@@ -99,6 +118,8 @@ def main(argv: list[str] | None = None) -> int:
         if name == "pause":
             cmd.add_argument("--reason", required=True)
     sub.add_parser("control")
+    accept = sub.add_parser("accept-unknown")
+    accept.add_argument("sha")
     run = sub.add_parser("run")
     run.add_argument("--playbook", required=True)
     run.add_argument("--repo", required=True)
@@ -160,6 +181,7 @@ def main(argv: list[str] | None = None) -> int:
                      "human_queue": len(ledger.human_queue(row["repo"])) if row["repo"] != "*" else None}
                     for row in ledger.all_repo_states()
                 ],
+                "unknown_commits": _unknown_status(ledger),
             }
             json.dump(report, sys.stdout, indent=2, sort_keys=True)
             print()
@@ -183,6 +205,17 @@ def main(argv: list[str] | None = None) -> int:
         if args.command == "control":
             _refresh(state_dir, ledger, registry)
             return 0
+        if args.command == "accept-unknown":
+            from .accept import AcceptError, request_accept
+
+            try:
+                sha = request_accept(ledger, args.sha, at=time.time())
+            except AcceptError as exc:
+                print(str(exc), file=sys.stderr)
+                return 2
+            print(f"accept requested for {sha}: `kb serve` judges it through the gate on its next tick "
+                  "(`kb status` and the human queue show the outcome)")
+            return 0
         if args.command == "run":
             from ..config import Settings
             from .runner import run_playbook

src/infermatrix_copilot/kb_service/ledger.py

@@ -311,6 +311,11 @@ def record_event(self, repo: str, source: str, external_id: str, payload: dict)
                 return None
             return int(cur.lastrowid)
 
+    def event_by_external_id(self, repo: str, source: str, external_id: str) -> dict[str, Any] | None:
+        row = self._conn.execute("SELECT * FROM events WHERE repo=? AND source=? AND external_id=?",
+                                 (repo, source, external_id)).fetchone()
+        return {**dict(row), "payload": json.loads(row["payload"])} if row else None
+
     def events(self, repo: str, status: str, *, limit: int = 100) -> list[dict[str, Any]]:
         rows = self._conn.execute(
             "SELECT * FROM events WHERE repo=? AND status=? ORDER BY id LIMIT ?",

src/infermatrix_copilot/kb_service/scheduler.py

@@ -27,6 +27,7 @@
 from . import merge
 from .activate import ActivationError, activate, activation_lock
 from .archive import make_archive
+from .accept import accept_pending
 from .audit import audit_main, open_unknown, provenance_problems
 from .report import flush_reports
 from .companion import publish_companion
@@ -91,8 +92,13 @@ def tick(self) -> None:
                     self._record("*", "audit", finding=finding)
             except Exception as exc:
                 self._record("*", "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
+        try:  # `kb accept-unknown` requests: an unknown commit judged through the gate
+            for event in accept_pending(rt):
+                self._record("*", "accept", detail=event)
+        except Exception as exc:
+            self._record("*", "error", error=repr(exc), trace=traceback.format_exc()[-2000:])
         if self._due("external", self.intake_every) and not merge.is_paused(rt.ledger, "*"):
-            try:  # knowledge PRs the service did not open (source 4, human approvals)
+            try:  # knowledge PRs the service did not open (source 4)
                 for event in poll_external(rt):
                     self._record("*", "external", detail=event)
             except Exception as exc: