From 3359d7a450cd70553b9e3b12f8e64c5034f20d92 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Tue, 29 Sep 2026 15:51:33 +0800 Subject: [PATCH] Add kb accept-unknown: judge an unknown knowledge commit through the gate (design v8 8.3) Until now an unknown commit (knowledge that reached main without passing the gate) could only be disposed of by an exact revert. The design also allows keeping it, but only when its governed content fully passes the gate. This adds that path. - `kb accept-unknown SHA` takes a unique prefix of 7+ hex characters. It only validates the request and records it as accept_request:. It exits 2 for something that is not an unknown commit, is ambiguous, or is already disposed of. `kb serve` holds the ledger lease, so the scheduler judges the request on its next tick (accept.accept_pending). - The commit's own change (first parent to commit) goes through the full gate as a change set of kind "accept": L1, L2, consistency and upstream facts, with a current judge calibration required. It must lie inside one enabled public repository scope, touch only governed pages, and not be a root commit. While the service's revert PR is still being opened, the request waits for the next tick. - Only a pass disposes of the commit. disposed: is set to the commit itself, so it is trusted wherever it is part of the history, and the record is marked accepted. Its "unrecorded" intake candidate is dropped. An open revert PR goes to superseding and gets a close item. The pause is not lifted: `kb resume` stays a person's call. - Acceptance can be resumed. After a pass, the record is first written as "accepting". The disposal and "accepted" are written only after the cleanup; the cleanup finds the candidate event directly (new ledger.event_by_external_id, no scan) and closes the revert. A crash in between is finished by the next tick without calling the judge again. - Like external PRs, the rules L1 finds the commit retiring or purging are kept on the accept change set. The cleanup records them in the retirement ledger (idempotent upserts), so the next release sweep purges them. - Any other result leaves the commit undisposed, tells people why, and is kept on the request record. - `kb status` now lists open unknown commits with their accept request. - Also: a stale comment in the scheduler ("human approvals") and the help line for `kb control` (there is no hold list any more). Tests (test_kb_accept_unknown.py): - an accepted commit is disposed of; its revert PR is closed and its candidate dropped; activation and the audit pass it; it is decided once; - a request waits while its revert PR is being opened; - a commit the gate rejects stays unknown, with its revert kept and people told; - no acceptance without a current calibration; - a non-governed change cannot be accepted; - requests are validated when recorded; - nothing is judged without the lease; - an interrupted acceptance is finished without judging again; - the candidate is found behind more than 1000 pending events; - an accepted retirement becomes purgeable at the next release; - `kb status` lists open unknown commits. Co-Authored-By: Claude Opus 5.5 --- doc/architecture/SPEC/kb_service.md | 17 ++ doc/features/knowledge-service.md | 2 +- src/infermatrix_copilot/kb_service/accept.py | 159 +++++++++++++++ src/infermatrix_copilot/kb_service/cli.py | 35 +++- src/infermatrix_copilot/kb_service/ledger.py | 5 + .../kb_service/scheduler.py | 8 +- test/test_kb_accept_unknown.py | 182 ++++++++++++++++++ 7 files changed, 405 insertions(+), 3 deletions(-) create mode 100644 src/infermatrix_copilot/kb_service/accept.py create mode 100644 test/test_kb_accept_unknown.py diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index eb6c8f22..aa039fb3 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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:`，已处置或不是未知提交 → 退出码 2）； + `kb serve` 持有租约，在下一个 tick 由 `accept.accept_pending` 判定。 +- 判定：该提交自身的改动（第一父 → 提交）作为 `accept` 变更集走**完整**门禁（L1、L2、一致性、上游事实），要求当前通过的 + 评审校准；只能是单个启用且公开的仓库范围、全部为受治理页面、非根提交。服务的 revert PR 仍在创建中（`pr_requested`） + 时等到下一 tick。 +- 只有 `pass` 才处置：`disposed: = `（提交在历史中即被信任）、记录 `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`。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 208de2c1..5cbfa80b 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -68,7 +68,7 @@ infermatrix-copilot kb rollback --to # 回指到较早的快照（下一 理由精炼后整体复检，至多 2 轮，仍失败则留在人工队列（不合并）。他人的知识 PR 每天检查一次：结果写在该 PR 上唯一一条 “知识质量门”评论里（原地更新），未通过则不合并，作者修改后次日复检。 有人绕过门禁直接合并/推送知识变更时，服务立即暂停该仓库、停止激活新快照、告警，并开出撤回 PR（不自动合并）； -合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁。发版巡检在上游新 release 后运行 +合并精确的撤回 PR 并 `kb resume` 后恢复，被撤回的内容会作为新候选重新过门禁；也可以用 `kb accept-unknown ` 让该提交本身完整通过门禁后保留它（通过才算处置，仍需 `kb resume`）。发版巡检在上游新 release 后运行 T1/T2/T3 与 purge，每个规则页一个变更集，全部经过质量门。 评审服务读取知识：将 `KNOWLEDGE_ROOT` 指向 `$KB_STATE_DIR/active`。 diff --git a/src/infermatrix_copilot/kb_service/accept.py b/src/infermatrix_copilot/kb_service/accept.py new file mode 100644 index 00000000..54029b9f --- /dev/null +++ b/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 `` 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"]) diff --git a/src/infermatrix_copilot/kb_service/cli.py b/src/infermatrix_copilot/kb_service/cli.py index ba2aee27..7249add4 100644 --- a/src/infermatrix_copilot/kb_service/cli.py +++ b/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 diff --git a/src/infermatrix_copilot/kb_service/ledger.py b/src/infermatrix_copilot/kb_service/ledger.py index b59a6ec1..eb0c9142 100644 --- a/src/infermatrix_copilot/kb_service/ledger.py +++ b/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 ?", diff --git a/src/infermatrix_copilot/kb_service/scheduler.py b/src/infermatrix_copilot/kb_service/scheduler.py index 9aec70cd..dc33357e 100644 --- a/src/infermatrix_copilot/kb_service/scheduler.py +++ b/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: diff --git a/test/test_kb_accept_unknown.py b/test/test_kb_accept_unknown.py new file mode 100644 index 00000000..f18eda43 --- /dev/null +++ b/test/test_kb_accept_unknown.py @@ -0,0 +1,182 @@ +"""`kb accept-unknown`: an unknown knowledge commit judged through the gate instead of reverted.""" + +from __future__ import annotations + +import json + +import pytest + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.accept import ACCEPT, AcceptError, accept_pending, request_accept +from infermatrix_copilot.kb_service.audit import DISPOSED, UNKNOWN, audit_main, open_unknown, provenance_problems +from infermatrix_copilot.knowledge_service.ops import KnowledgeOperation as Op, apply_operations +from test_kb_audit import _land, _setup +from test_kb_flow import _items +from test_kb_intake_gate import PAGE, ScriptedGateway, _calibrate, _judge_all, _rule +from test_kb_provenance import _open_revert + + +def _sneak_rule(origin): + """A reasonable rule pushed straight to main, around the gate.""" + page = origin / "knowledge" / PAGE + before = page.read_text() + after = apply_operations({PAGE: before}, [Op("add", PAGE, "DEMO-2a", _rule("DEMO-2a", "PR #11"))], + release="v1", today="2026-09-29").files[PAGE] + return _land(origin, f"knowledge/{PAGE}", after, "add a rule directly"), before + + +def _unknown(tmp_path, *, judge="yes"): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt.gateway = ScriptedGateway(_judge_all(judge)) + _calibrate(tmp_path, rt, lifecycle) + sha, before = _sneak_rule(origin) + audit_main(rt, scheduler._pause) + assert open_unknown(rt) == [sha] + return rt, rt.registry["demo"], scheduler, origin, sha, before + + +def _state(rt, sha): + return json.loads(rt.ledger.get_cursor("*", ACCEPT + sha)) + + +def test_an_accepted_commit_is_disposed_of_and_its_revert_closed(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + assert request_accept(rt.ledger, sha[:10], at=rt.clock()) == sha + (event,) = accept_pending(rt) + assert "requested" in event and _state(rt, sha)["state"] == "requested" # its revert PR is being opened + revert_id, _head = _open_revert(rt, origin, sha, before) + (event,) = accept_pending(rt) + assert f"accept {sha[:12]} accepted" in event and "kb resume --repo demo" in event + assert rt.ledger.get_cursor("*", DISPOSED + sha) == sha and open_unknown(rt) == [] + (accepted,) = rt.ledger.changesets_of_kind("demo", "accept") + assert accepted["status"] == "accepted" and accepted["detail"]["accepts"] == sha + assert accepted["detail"]["decision"]["status"] == "pass" + revert = rt.ledger.changeset(revert_id) + assert revert["status"] == "superseding" + assert [i["body"]["pr"] for i in _items(tmp_path, "close")] == [90] + assert rt.ledger.events("demo", "pending") == [] # the candidate is dropped + assert provenance_problems(rt, rt.knowledge.fetch()) == [] # activation may pass it now + assert audit_main(rt, scheduler._pause) == [] + assert accept_pending(rt) == [] # decided once + + +def test_a_commit_the_gate_rejects_stays_unknown(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path, judge="no") + _open_revert(rt, origin, sha, before) + request_accept(rt.ledger, sha, at=rt.clock()) + (event,) = accept_pending(rt) + assert "refused" in event and "the gate did not pass it (fail)" in event + assert open_unknown(rt) == [sha] and rt.ledger.get_cursor("*", DISPOSED + sha) is None + assert any("accept-unknown" in row["reason"] for row in rt.ledger.human_queue("demo")) + (rejected,) = rt.ledger.changesets_of_kind("demo", "accept") + assert rejected["status"] == "failed" + assert rt.ledger.changesets_of_kind("demo", "revert")[0]["status"] == "revert_open" # the revert stays + + +def test_no_acceptance_without_a_current_judge_calibration(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + _open_revert(rt, origin, sha, before) + _calibrate(tmp_path, rt, lifecycle, passed=False) + request_accept(rt.ledger, sha, at=rt.clock()) + (event,) = accept_pending(rt) + assert "calibration" in event and open_unknown(rt) == [sha] + assert rt.ledger.changesets_of_kind("demo", "accept") == [] # nothing was judged + + +def test_a_change_outside_the_governed_pages_cannot_be_accepted(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + _calibrate(tmp_path, rt, lifecycle) + sha = _land(origin, "knowledge/repos/demo/tools/x.py", "print(1)\n", "a script, directly") + audit_main(rt, scheduler._pause) + request_accept(rt.ledger, sha, at=rt.clock()) + (event,) = accept_pending(rt) + assert "does not govern" in event and open_unknown(rt) == [sha] + + +def test_requests_are_validated_when_recorded(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + with pytest.raises(AcceptError, match="not a commit SHA"): + request_accept(rt.ledger, "HEAD", at=0) + with pytest.raises(AcceptError, match="matches 0 unknown commits"): + request_accept(rt.ledger, "0" * 40, at=0) + rt.ledger.set_cursor("*", DISPOSED + sha, "f" * 40) + with pytest.raises(AcceptError, match="already disposed"): + request_accept(rt.ledger, sha, at=0) + assert rt.ledger.get_cursor("*", UNKNOWN + sha) # the record itself is untouched + + +def test_without_the_lease_nothing_is_judged(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + request_accept(rt.ledger, sha, at=rt.clock()) + rt.lease_owner = None + assert accept_pending(rt) == [] and _state(rt, sha)["state"] == "requested" + + +def test_kb_status_lists_open_unknown_commits_with_their_request(tmp_path, monkeypatch, capsys): + from infermatrix_copilot.kb_service.cli import _unknown_status + + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + request_accept(rt.ledger, sha, at=1.0) + status = _unknown_status(rt.ledger) + assert status[sha]["state"] == "reverting" and status[sha]["accept"]["state"] == "requested" + rt.ledger.set_cursor("*", DISPOSED + sha, sha) + assert _unknown_status(rt.ledger) == {} + assert merge.is_paused(rt.ledger, "demo") # resuming stays a person's call + + +def test_an_interrupted_acceptance_is_finished_without_judging_again(tmp_path): + rt, lifecycle, scheduler, origin, sha, before = _unknown(tmp_path) + revert_id, _head = _open_revert(rt, origin, sha, before) + request_accept(rt.ledger, sha, at=rt.clock()) + real = rt.ledger.update_changeset + + def crash(changeset_id, **fields): + if changeset_id == revert_id: + raise RuntimeError("the process died here") + return real(changeset_id, **fields) + rt.ledger.update_changeset = crash + with pytest.raises(RuntimeError): + accept_pending(rt) + rt.ledger.update_changeset = real + assert rt.ledger.get_cursor("*", DISPOSED + sha) is None # not disposed of half-way + calls = len(rt.gateway.calls) + (event,) = accept_pending(rt) + assert "accepted" in event and len(rt.gateway.calls) == calls # finished, not judged again + assert rt.ledger.changeset(revert_id)["status"] == "superseding" + assert rt.ledger.get_cursor("*", DISPOSED + sha) == sha + assert len(rt.ledger.changesets_of_kind("demo", "accept")) == 1 + + +def test_the_candidate_is_dropped_however_many_events_wait(tmp_path): + rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt.gateway = ScriptedGateway(_judge_all("yes")) + _calibrate(tmp_path, rt, lifecycle) + for n in range(1001): + rt.ledger.record_event("demo", "merged_pr", str(n), {}) + sha, before = _sneak_rule(origin) + audit_main(rt, scheduler._pause) + _open_revert(rt, origin, sha, before) + request_accept(rt.ledger, sha, at=rt.clock()) + accept_pending(rt) + assert rt.ledger.event_by_external_id("demo", "unrecorded", sha)["status"] == "superseded" + + +def test_an_accepted_retirement_becomes_purgeable_next_release(tmp_path): + from dataclasses import replace + + rt, lifecycle, scheduler, origin = _setup(tmp_path) + rt.gateway = ScriptedGateway(_judge_all("yes")) + rt.release_label = lambda repo: "v1" + _calibrate(tmp_path, rt, lifecycle) + rt.registry["demo"] = replace(rt.registry["demo"], retire_ratio=1.0) + _land(origin, "skills/x.md", "no citations here\n", "drop the citation") # else it needs a companion + before = (origin / "knowledge" / PAGE).read_text() + retired = apply_operations({PAGE: before}, [Op("retire", PAGE, "DEMO-1a", reason="upstream-removed", + evidence="PR #13")], release="v1", today="2026-09-29").files[PAGE] + sha = _land(origin, f"knowledge/{PAGE}", retired, "retire a rule directly") + audit_main(rt, scheduler._pause) + _open_revert(rt, origin, sha, before) + request_accept(rt.ledger, sha, at=rt.clock()) + (event,) = accept_pending(rt) + assert "accepted" in event, event + assert [r["rule_id"] for r in rt.ledger.purge_eligible("demo", "v2")] == ["DEMO-1a"]