diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index bcd296b5..44666f8d 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/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` 必须允许发布器直接合并） +也转人工，精炼无法修复。 diff --git a/doc/features/knowledge-service.md b/doc/features/knowledge-service.md index 760ecc90..35607035 100644 --- a/doc/features/knowledge-service.md +++ b/doc/features/knowledge-service.md @@ -63,7 +63,8 @@ infermatrix-copilot kb rollback --to # 回指到较早的快照（下一 合并流程（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`。 diff --git a/src/infermatrix_copilot/kb_service/external.py b/src/infermatrix_copilot/kb_service/external.py index 27218d86..75bbc181 100644 --- a/src/infermatrix_copilot/kb_service/external.py +++ b/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 diff --git a/src/infermatrix_copilot/kb_service/intake.py b/src/infermatrix_copilot/kb_service/intake.py index bd7dd36e..e67e8593 100644 --- a/src/infermatrix_copilot/kb_service/intake.py +++ b/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 diff --git a/src/infermatrix_copilot/kb_service/ledger.py b/src/infermatrix_copilot/kb_service/ledger.py index 58075370..693314ff 100644 --- a/src/infermatrix_copilot/kb_service/ledger.py +++ b/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: diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index 8f8b997e..36131586 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/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) diff --git a/src/infermatrix_copilot/kb_service/refine.py b/src/infermatrix_copilot/kb_service/refine.py new file mode 100644 index 00000000..875cf58b --- /dev/null +++ b/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 diff --git a/src/infermatrix_copilot/kb_service/runtime.py b/src/infermatrix_copilot/kb_service/runtime.py index 1cac4d6f..0e724ac5 100644 --- a/src/infermatrix_copilot/kb_service/runtime.py +++ b/src/infermatrix_copilot/kb_service/runtime.py @@ -280,7 +280,7 @@ def draft_key_for_event(repo: str, event_id) -> str: def gate_and_stage(rt: KbRuntime, lifecycle: RepoLifecycle, owner: str, *, kind: str, base: dict, base_sha: str, external: dict, operations, result, evidence: list[dict], event_ids: list[int], release: str, force_human: str = "", - hold: bool = False, draft_keys: list[str] = ()) -> str: + hold: bool = False, draft_keys: list[str] = (), extra_detail: dict | None = None) -> str: """Run the quality gate on one change set and stage it (files first, then every ledger write in one transaction fenced on the lease). Used by intake, sweep and purge. ``force_human`` routes a passing change set to people @@ -315,6 +315,7 @@ def gate_and_stage(rt: KbRuntime, lifecycle: RepoLifecycle, owner: str, *, kind: drafted_events=event_ids, detail={ "operations": operations_json(operations), "event_ids": event_ids, + **(extra_detail or {}), "base_sha": base_sha, "release": release, "decision": decision.to_dict(), "generator": rt.generator.label(), "judge": rt.judge.label(), }) diff --git a/test/test_kb_external.py b/test/test_kb_external.py index 5bb849dd..77a5b24c 100644 --- a/test/test_kb_external.py +++ b/test/test_kb_external.py @@ -268,3 +268,15 @@ def test_rules_an_external_pr_retires_enter_the_retirement_ledger(tmp_path): merge.advance(rt, lifecycle) row = rt.ledger._conn.execute("SELECT rule_id, page FROM retirements WHERE repo='demo'").fetchall() assert [tuple(r) for r in row] == [("DEMO-1a", PAGE)] + + +def test_our_own_prs_are_never_judged_as_external_in_any_status(tmp_path): + rt, lifecycle, origin = _setup(tmp_path) + for number, status in ((31, "refine_needed"), (32, "refine_exhausted"), (33, "superseding")): + ours = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.lease_owner, "demo", ours, detail={}, status=status, verdicts=[], + human_reason="", drafted_events=[]) + rt.ledger.update_changeset(ours, pr_number=number) + _open_human_pr(rt, origin, number, _add_rule_files()) + assert poll_external(rt) == [] + assert not _items(tmp_path, "merge") diff --git a/test/test_kb_flow.py b/test/test_kb_flow.py index 28faa55c..c50a6d21 100644 --- a/test/test_kb_flow.py +++ b/test/test_kb_flow.py @@ -169,10 +169,15 @@ def refused(**ack): assert refused(error="local gate: context changed since the verdict was judged: x", problems=["context changed since the verdict was judged: x"])["status"] == "rebuild_needed" rt.ledger.update_changeset(changeset_id, status="pr_open") - # the change itself: never merged, people decide + # the change itself: never merged; our own change is refined (test_kb_refine) changeset = refused(error="local gate: L1 dangling_reference", problems=["L1 dangling_reference a b"]) - assert changeset["status"] == "gate_failed" + assert changeset["status"] == "refine_needed" assert changeset["detail"]["gate_problems"] == ["L1 dangling_reference a b"] + rt.ledger.update_changeset(changeset_id, status="pr_open") + # how the repository is set up: refining cannot help, people decide + changeset = refused(error="local gate: queued", problems=["PR #42 was queued ...: main must allow a direct " + "merge by the publisher"]) + assert changeset["status"] == "gate_failed" assert any("local gate refused" in row["reason"] for row in rt.ledger.human_queue("demo")) diff --git a/test/test_kb_rebuild.py b/test/test_kb_rebuild.py index 528d6ab2..1852f23c 100644 --- a/test/test_kb_rebuild.py +++ b/test/test_kb_rebuild.py @@ -43,13 +43,15 @@ def test_transient_refusals_are_signed_again(tmp_path): assert len(_items(tmp_path, "merge")) == 1 # the answered item left the outbox; this is the new one -def test_a_real_problem_goes_to_people_once(tmp_path): +def test_a_real_problem_is_refined_never_retried_as_is(tmp_path): rt, lifecycle, changeset_id, _ = _posted(tmp_path) _fail(rt, "L1 duplicate_rule_id DEMO-2a") - assert rt.ledger.changeset(changeset_id)["status"] == "gate_failed" - assert len(rt.ledger.human_queue("demo")) == 1 - assert merge.advance(rt, lifecycle) == [] - assert _items(tmp_path, "merge") == [] # never retried as is + changeset = rt.ledger.changeset(changeset_id) + assert changeset["status"] == "refine_needed" # refined next pass (test_kb_refine) + assert changeset["detail"]["gate_problems"] == ["L1 duplicate_rule_id DEMO-2a"] + assert rt.ledger.human_queue("demo") == [] + assert merge.advance(rt, lifecycle) == [] # no lease: nothing refined, nothing retried + assert _items(tmp_path, "merge") == [] def test_a_context_change_rebuilds_on_current_main_and_replaces_the_pr(tmp_path): @@ -111,7 +113,10 @@ def test_a_rebuild_that_fails_its_own_gate_leaves_the_pr_to_people(tmp_path): rt.gateway = ScriptedGateway(_generator_then_judge("no")) _fail(rt, "context changed since the verdict was judged: repos/demo/core/_index.md") rt.lease_owner = rt.ledger.acquire_lease("scheduler") - assert merge.advance(rt, lifecycle) == [f"rebuild_failed {changeset_id}"] + events = merge.advance(rt, lifecycle) + assert events[0] == f"rebuild_failed {changeset_id}" + # the rebuild the gate rejected is a change of our own: it is refined next + assert any(e.startswith("refined demo-rebuild-") for e in events[1:]) assert rt.ledger.changeset(changeset_id)["status"] == "rebuild_failed" assert _items(tmp_path, "close") == [] and rt.ledger.human_queue("demo") diff --git a/test/test_kb_refine.py b/test/test_kb_refine.py new file mode 100644 index 00000000..ebf28ca8 --- /dev/null +++ b/test/test_kb_refine.py @@ -0,0 +1,214 @@ +"""Refine and recheck: a change the gate rejected is never merged; it is refined and judged again.""" + +from __future__ import annotations + +import json + +from infermatrix_copilot.kb_service import merge +from infermatrix_copilot.kb_service.refine import MAX_REFINES +from infermatrix_copilot.kb_service.runtime import collect_events, run_intake +from test_kb_flow import _ack, _flow_runtime, _items, _open_pr +from test_kb_intake_gate import PAGE, ScriptedGateway, _judge_all, _rule + +HEAD = "d" * 40 + + +class Judges: + """The generator adds a rule; the judge answers from a script, one verdict per judging round.""" + + def __init__(self, verdicts): + self.verdicts = list(verdicts) # "yes"/"no" per round of block judging + self.prompts: list[str] = [] + + def __call__(self, role, prompt): + if role.name == "generator": + self.prompts.append(prompt) + return {"operations": [{"kind": "add", "page": PAGE, "rule_id": "DEMO-2a", + "section_markdown": _rule("DEMO-2a", "PR #11")}], "rationale": "r"} + if '"directory"' in prompt: + return {"verdict": "consistent", "conflicts": []} + value = self.verdicts[0] if self.verdicts else "yes" + answer = _judge_all(value)(role, prompt) + if value == "no": + answer["reasons"] = {d: "the claim is not in the evidence" for d in answer["dimensions"]} + return answer + + def next_round(self): + if self.verdicts: + self.verdicts.pop(0) + + +def _runtime(tmp_path, verdicts): + rt, lifecycle = _flow_runtime(tmp_path) + judges = Judges(verdicts) + rt.gateway = ScriptedGateway(judges) + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + return rt, lifecycle, judges + + +def test_a_rejected_intake_is_refined_with_the_reasons_and_judged_again(tmp_path): + rt, lifecycle, judges = _runtime(tmp_path, ["no", "yes"]) + collect_events(rt, lifecycle) + first = run_intake(rt, lifecycle) + assert rt.ledger.changeset(first)["status"] == "failed" + judges.next_round() + events = merge.advance(rt, lifecycle) + (event,) = [e for e in events if e.startswith("refined")] + new_id = event.rsplit(" ", 1)[1] + assert "gate_feedback" in judges.prompts[-1] and "not in the evidence" in judges.prompts[-1] + old, new = rt.ledger.changeset(first), rt.ledger.changeset(new_id) + assert old["status"] == "refined" and old["detail"]["refined_as"] == new_id + assert new["kind"] == "refine" and new["detail"]["refine_round"] == 1 + assert new["status"] == "pr_requested" # passed the whole gate again, then handed over + assert merge.advance(rt, lifecycle) == [] # refined once, not again + + +def test_a_change_still_rejected_after_two_refinements_stays_unmerged_for_people(tmp_path): + rt, lifecycle, judges = _runtime(tmp_path, ["no"] * (MAX_REFINES + 1)) + collect_events(rt, lifecycle) + run_intake(rt, lifecycle) + for _ in range(MAX_REFINES + 1): + merge.advance(rt, lifecycle) + exhausted = rt.ledger.changesets("demo", ("refine_exhausted",)) + assert len(exhausted) == 1 and exhausted[0]["detail"]["refine_round"] == MAX_REFINES + assert len(exhausted[0]["detail"]["refine_history"]) == MAX_REFINES + 1 + (row,) = rt.ledger.human_queue("demo") + assert f"still rejected after {MAX_REFINES} refinements" in row["reason"] and "round 0" in row["reason"] + assert not _items(tmp_path, "open_pr") and not _items(tmp_path, "merge") + calls = len(judges.prompts) + merge.advance(rt, lifecycle) + assert len(judges.prompts) == calls # no more paid generator calls + + +def test_a_pr_the_local_gate_refused_is_replaced_by_a_refined_one(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + changeset_id, publisher = _open_pr(tmp_path, rt, lifecycle) + _ack(tmp_path, rt, publisher, kind="open_pr", changeset_id=changeset_id, ok=True, pr=42, head_sha=HEAD) + rt.github.prs[42] = {"state": "open", "merged": False, "head": {"sha": HEAD}} + merge.advance(rt, lifecycle) + _ack(tmp_path, rt, publisher, kind="merge", changeset_id=changeset_id, ok=False, + error="local gate: L1 dangling_reference repos/demo/x.md DEMO-1a", + problems=["L1 dangling_reference repos/demo/x.md DEMO-1a"]) + assert rt.ledger.changeset(changeset_id)["status"] == "refine_needed" + assert rt.ledger.human_queue("demo") == [] # not people yet: refined first + judges = Judges(["yes"]) + rt.gateway = ScriptedGateway(judges) + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + events = merge.advance(rt, lifecycle) + assert any(e.startswith(f"refined {changeset_id} as ") for e in events) + assert "dangling_reference" in judges.prompts[-1] + old = rt.ledger.changeset(changeset_id) + assert old["status"] == "superseding" + (close,) = _items(tmp_path, "close") + assert close["body"]["pr"] == 42 + assert rt.ledger.changeset(old["detail"]["refined_as"])["status"] == "pr_requested" + + +def test_a_refinement_is_never_staged_twice(tmp_path): + rt, lifecycle, judges = _runtime(tmp_path, ["no", "yes"]) + collect_events(rt, lifecycle) + first = run_intake(rt, lifecycle) + judges.next_round() + real = rt.ledger.update_changeset + crashed = {"n": 0} + + def crash_once(cs_id, **fields): + if cs_id == first and fields.get("status") == "refined" and not crashed["n"]: + crashed["n"] += 1 + raise RuntimeError("crash after staging the refinement") + return real(cs_id, **fields) + rt.ledger.update_changeset = crash_once + import pytest + with pytest.raises(RuntimeError): + merge.advance(rt, lifecycle) + rt.ledger.update_changeset = real + merge.advance(rt, lifecycle) + assert len([cs for cs in rt.ledger.changesets("demo", ("pr_requested", "gated", "failed")) + if cs["kind"] == "refine"]) == 1 + + +def test_an_external_pr_refused_by_the_local_gate_goes_to_people(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + owner = rt.ledger.acquire_lease("t") + cs = rt.ledger.new_changeset_id("demo", "external") + rt.ledger.stage_intake(owner, "demo", cs, kind="external", status="merge_requested", verdicts=[], + human_reason="", drafted_events=[], detail={}) + rt.ledger.update_changeset(cs, pr_number=7, head_sha=HEAD, pending_item={"id": "m1", "kind": "merge", + "expires_at": rt.clock() + 60}) + merge.apply_acks(rt, [{"item_id": "m1", "kind": "merge", "changeset_id": cs, "ok": False, + "error": "local gate: L1 x", "problems": ["L1 x"]}]) + assert rt.ledger.changeset(cs)["status"] == "gate_failed" + assert rt.ledger.human_queue("demo") + assert json.dumps(rt.ledger.changeset(cs)["detail"]["gate_problems"]) == '["L1 x"]' + + +def _failed_intake(tmp_path, verdicts): + rt, lifecycle, judges = _runtime(tmp_path, verdicts) + collect_events(rt, lifecycle) + first = run_intake(rt, lifecycle) + assert rt.ledger.changeset(first)["status"] == "failed" + return rt, lifecycle, judges, first + + +def test_every_piece_of_evidence_reaches_the_refinement(tmp_path): + rt, lifecycle, judges, first = _failed_intake(tmp_path, ["no"]) + data = rt.load_changeset_files(first) + data["evidence"] = [{"source_reference": "PR #11", "title": "Bound the queue", "changed_files": ["a.py"]}, + {"source_reference": "PR #12", "title": "Retry the upload", "changed_files": ["b.py"]}] + rt.save_changeset_files(first, data) + merge.advance(rt, lifecycle) + prompt = judges.prompts[-1] + assert "Bound the queue" in prompt and "Retry the upload" in prompt and "PR #12" in prompt + + +def test_a_rebuild_keeps_the_refine_limit(tmp_path): + rt, lifecycle, judges, first = _failed_intake(tmp_path, ["no"]) + detail = rt.ledger.changeset(first)["detail"] + rt.ledger.update_changeset(first, status="rebuild_needed", pr_number=42, head_sha=HEAD, detail={ + **detail, "refine_round": MAX_REFINES, "refine_history": [{"changeset": "a", "reasons": ["x"]}, + {"changeset": "b", "reasons": ["y"]}]}) + new_id = merge.rebuild(rt, lifecycle, rt.ledger.changeset(first), "context changed since ...") + assert new_id is None # the rebuild was rejected by its own gate + (rebuilt,) = [cs for cs in rt.ledger.changesets_of_kind("demo", "rebuild")] + assert rebuilt["detail"]["refine_round"] == MAX_REFINES # the lineage carried over + calls = len(judges.prompts) + merge.advance(rt, lifecycle) + assert rt.ledger.changeset(rebuilt["id"])["status"] == "refine_exhausted" # no third refinement + assert len(judges.prompts) == calls + + +def test_a_refinement_staged_before_a_crash_is_found_in_any_status(tmp_path): + from infermatrix_copilot.kb_service.refine import REFINE_MARK + + rt, lifecycle, judges, first = _failed_intake(tmp_path, ["no"]) + staged = rt.ledger.new_changeset_id("demo", "refine") + rt.save_changeset_files(staged, {"base_sha": "x", "files": {}, "deleted": [], + "evidence": [{"source_reference": REFINE_MARK + first}]}) + rt.ledger.stage_intake(rt.lease_owner, "demo", staged, kind="refine", status="companion_pending", + verdicts=[], human_reason="", drafted_events=[], detail={}) + calls = len(judges.prompts) + merge.advance(rt, lifecycle) + assert len(judges.prompts) == calls # not drafted again + assert rt.ledger.changeset(first)["detail"]["refined_as"] == staged + + +def test_a_refinement_may_carry_as_many_operations_as_the_change_set(tmp_path): + rt, lifecycle, judges, first = _failed_intake(tmp_path, ["no"]) + data = rt.load_changeset_files(first) + data["evidence"] = [{"source_reference": f"PR #{n}", "title": f"event {n}"} for n in (11, 12)] + rt.save_changeset_files(first, data) + many = [{"kind": "add", "page": PAGE, "rule_id": f"DEMO-{n}a", "section_markdown": _rule(f"DEMO-{n}a", "PR #11")} + for n in range(3, 10)] # 7 operations: more than one event may carry + + class Many(Judges): + def __call__(self, role, prompt): + if role.name == "generator" and "gate_feedback" in prompt: + self.prompts.append(prompt) + return {"operations": many, "rationale": "r"} + return super().__call__(role, prompt) + judges = Many(["yes"]) + rt.gateway = ScriptedGateway(judges) + events = merge.advance(rt, lifecycle) + (event,) = [e for e in events if e.startswith(f"refined {first} as ")] + refined = rt.ledger.changeset(event.rsplit(" ", 1)[1]) + assert len(refined["detail"]["operations"]) == 7