diff --git a/doc/architecture/SPEC/kb_service.md b/doc/architecture/SPEC/kb_service.md index b26932bd..1d5f77d4 100644 --- a/doc/architecture/SPEC/kb_service.md +++ b/doc/architecture/SPEC/kb_service.md @@ -95,3 +95,16 @@ stdlib + PyYAML + `cryptography`(`kb` extra)+ `.adapters` + `.knowledge_serv 巡检为发版 + 页面,均带随机后缀),只有被采纳的那次尝试(`draft_key#attempt`)列入之后 `decision` 记录的 `draft_keys`,被拒绝的尝试与失败的重试不继承判定。记录的每个字符串(含键、model、usage)都经过脱敏。 - CLI:`kb traces`、`kb replay --record ID --model P:M[:E]`、`kb export --out FILE`。 + +## 2026-09-28 过期判定:重签、重建或转人工 +`merge.advance` 记录每次签发判定的时间(`verdict_issued_at`)与入队时间(`queued_at`),并处理不再能通过门禁的判定: +- 判定签发超过 48 小时(有效期 72 小时):`verdict_posted` 直接回到 `pr_open` 重签;`queued` 以及入队 2 小时仍未合并的 + PR **不按计时释放队列名额**,而是先下发 `pause`(出队 + 转 draft),其回执把它送回 `pr_open`,再重签并重新入队。 +- 尚未过期的 `enqueue` 项未回执时不重签(其回执决定:进入 `queued` 后走上面的出队路径)。 +- PR 阶段 `kb-gate` 失败(只看签发当前判定之后发布的状态),按描述(验证器的第一个问题)分类:暂停清单/不可达/ + 验证器错误、我们自己暂停留下的 `kb:hold` 等瞬时问题 → 等待;判定过期或 + 缺失 → 重签;上下文改变、一致性页面变化、无法干净合并、清单不符 → **重建**:只有持有租约的调度器执行,在当前 main + 上重新应用同样的操作并重新过质量门,暂存为 `rebuild` 变更集(证据带 `rebuild of <旧 id>` 标记,与暂存原子写入, + 中断后再次执行会找到它而不会重复暂存)。只有重建**通过**质量门才取代旧 PR:旧变更集进入 `superseding`,`close` 项 + 过期即重发,直到观测到 PR 已关闭才记为 `superseded`(不计入被人工推翻的熔断);重建未通过、操作无法再应用或没有 + 可重放的操作 → `rebuild_failed` 并转人工,旧 PR 保持打开。其他失败(变更本身有问题)→ `gate_failed` 并转人工一次。 diff --git a/src/infermatrix_copilot/kb_service/merge.py b/src/infermatrix_copilot/kb_service/merge.py index d0a2eec1..3a615a1e 100644 --- a/src/infermatrix_copilot/kb_service/merge.py +++ b/src/infermatrix_copilot/kb_service/merge.py @@ -25,6 +25,19 @@ VERDICT_MARKER = "" IN_FLIGHT = ("pr_requested", "pr_open", "verdict_posted", "queued") +# a verdict must first verify within 72 hours: re-sign well before that +RESIGN_AFTER = 48 * 3600 +# a queued PR that neither merged nor left the queue by then was dropped by a +# failing merge group (GitHub does not tell us): re-sign and re-enqueue +QUEUE_STALL = 2 * 3600 +# kb-gate failures, by what the service can do about them (matched against the +# status description, which is the verifier's first problem) +TRANSIENT_GATE = ("hold list", "paused by the knowledge service", "global pause", "unreachable", + "verifier error", "fetched head does not match", + "carries kb:hold") # left by our own pause until the re-signed verdict lifts it +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") KNOWLEDGE_REPO_ENV = "KB_KNOWLEDGE_REPOSITORY" DEFAULT_KNOWLEDGE_REPOSITORY = "JiusiServe/InferMatrixCopilot" @@ -102,6 +115,8 @@ def apply_acks(rt, acks: list[dict]) -> None: if kind == "open_pr": fields.update(pr_number=int(ack["pr"]), head_sha=str(ack["head_sha"]), branch=str(ack.get("branch", ""))) + elif kind == "enqueue": + fields["detail"] = {**changeset["detail"], "queued_at": rt.clock()} rt.ledger.update_changeset(changeset_id, **fields) @@ -136,11 +151,123 @@ def _pr_state(rt, number: int) -> dict: def _gate_status(rt, sha: str) -> str: + return _gate(rt, sha)[0] + + +def _gate(rt, sha: str, *, since: float = 0.0) -> tuple[str, str]: + """(state, description) of the latest kb-gate status on ``sha``; a status + posted before ``since`` (the current verdict) is not about it: ("", "").""" combined = rt.github.get(f"/repos/{knowledge_repository()}/commits/{sha}/status") for status in combined.get("statuses", []): if status.get("context") == "kb-gate": - return str(status.get("state") or "") - return "" + posted = _epoch(status.get("updated_at") or status.get("created_at")) + if since and posted and posted < since: + return "", "" + return str(status.get("state") or ""), str(status.get("description") or "") + return "", "" + + +def _epoch(value) -> float: + if not value: + return 0.0 + import datetime as dt + + try: + return dt.datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp() + except ValueError: + return 0.0 + + +def _enqueue_outstanding(rt, changeset: dict) -> bool: + """A signed enqueue that the publisher may still execute (unexpired).""" + pending = changeset.get("pending_item") or {} + return pending.get("kind") == "enqueue" and float(pending.get("expires_at", 0)) > rt.clock() + + +def _resign(rt, changeset: dict, why: str) -> None: + """Back to pr_open: the next pass signs a fresh verdict for the same head.""" + rt.ledger.update_changeset(changeset["id"], status="pr_open", pending_item=None) + _outcome(rt, changeset, "resigned", reason=why) + + +REBUILD_MARK = "rebuild of " + + +def _rebuild_of(rt, repo: str, old_id: str) -> str | None: + """A change set already staged as the rebuild of ``old_id`` (its evidence + carries the mark, written atomically with the staging), so an interrupted + rebuild is resumed instead of staged twice.""" + for candidate in rt.ledger.changesets(repo, ("gated", "failed", "human", *IN_FLIGHT)): + if candidate["kind"] != "rebuild": + continue + try: + evidence = rt.load_changeset_files(candidate["id"]).get("evidence") or [] + except (OSError, ValueError): + continue + if any(str(item.get("source_reference") or "") == REBUILD_MARK + old_id for item in evidence): + return candidate["id"] + return None + + +def rebuild(rt, lifecycle, changeset: dict, why: str) -> str | None: + """Rebuild a change set whose verdict can no longer verify on the current + base (context pages moved, or the PR no longer applies): re-apply its + operations to current main and gate them again. Only a rebuild that passes + the gate replaces the PR; the old one then waits in ``superseding`` until + its close is observed. Anything else goes to people with the old PR left + open. Needs the scheduler's lease; returns the new change set id or None.""" + from ..knowledge_service.lifecycle import LifecycleError + from ..knowledge_service.ops import KnowledgeOperation, apply_operations + from .runtime import gate_and_stage, publish + + owner = getattr(rt, "lease_owner", None) + if not owner: + return None + + def to_people(reason: str) -> None: + rt.ledger.update_changeset(changeset["id"], status="rebuild_failed") + rt.ledger.enqueue_human(lifecycle.repo, reason, changeset["id"]) + + detail = changeset["detail"] + raw = detail.get("operations") or [] + if not raw: # a mechanical change set (index fixes): nothing to re-apply + to_people(f"kb-gate failed ({why}); no operations to rebuild") + return None + new_id = _rebuild_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) + operations = [KnowledgeOperation.from_dict(item) for item in raw] + try: + result = apply_operations(base, operations, release=detail["release"], today=rt.today()) + except LifecycleError as exc: + to_people(f"cannot rebuild on current main: {exc}") + return None + evidence = [*(rt.load_changeset_files(changeset["id"]).get("evidence") or []), + {"source_reference": REBUILD_MARK + changeset["id"], "title": why}] + 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=[]) + 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 + to_people(f"rebuild {new_id} on current main did not pass the gate ({staged['status']})") + return None + rt.ledger.update_changeset(changeset["id"], status="superseding", + detail={**detail, "rebuilt_as": new_id, "superseded_because": why}) + _outcome(rt, changeset, "superseded", reason=why, replaced_by=new_id) + _request_close(rt, lifecycle.repo, rt.ledger.changeset(changeset["id"])) + publish(rt, lifecycle, new_id) + return new_id + + +def _request_close(rt, repo: str, changeset: dict) -> None: + issue_once(rt, repo, changeset, "close", { + "changeset_id": changeset["id"], "pr": int(changeset["pr_number"]), + "reason": (f"Superseded by change set {changeset['detail'].get('rebuilt_as')}, rebuilt on " + f"current main ({changeset['detail'].get('superseded_because', '')}).")}) def advance(rt, lifecycle) -> list[str]: @@ -149,12 +276,24 @@ def advance(rt, lifecycle) -> list[str]: if rt.outbox is None: return events paused = is_paused(rt.ledger, lifecycle.repo) - queued = rt.ledger.changesets(lifecycle.repo, ("queued",)) - for changeset in rt.ledger.changesets(lifecycle.repo, IN_FLIGHT): + # the one queue slot per repository is taken by a queued PR AND by any PR + # whose signed enqueue is still executable (not yet acked) + queued = rt.ledger.changesets(lifecycle.repo, ("queued",)) + [ + cs for cs in rt.ledger.changesets(lifecycle.repo, ("verdict_posted",)) if _enqueue_outstanding(rt, cs)] + for changeset in rt.ledger.changesets(lifecycle.repo, (*IN_FLIGHT, "superseding")): number = changeset["pr_number"] if number is None: continue pr: dict[str, Any] = _pr_state(rt, int(number)) + if changeset["status"] == "superseding": + # the replaced PR stays tracked until GitHub shows it closed; the + # close item is re-issued whenever the previous one expired + if pr.get("state") == "closed" or pr.get("merged"): + rt.ledger.update_changeset(changeset["id"], status="superseded", pending_item=None) + events.append(f"superseded {changeset['id']}") + else: + _request_close(rt, lifecycle.repo, changeset) + continue if pr.get("merged"): rt.ledger.update_changeset(changeset["id"], status="merged", merge_sha=str(pr.get("merge_commit_sha") or "")) @@ -176,6 +315,41 @@ def advance(rt, lifecycle) -> list[str]: continue if paused: continue # the pause itself is issued by `kb pause` / breakers + signed_at = float(changeset["detail"].get("verdict_issued_at") or 0) + stale_verdict = bool(signed_at) and rt.clock() - signed_at > RESIGN_AFTER + queued_at = float(changeset["detail"].get("queued_at") or 0) + if changeset["status"] == "queued" and ( + stale_verdict or (queued_at and rt.clock() - queued_at > QUEUE_STALL)): + # never free the queue slot on a timer: dequeue (+ draft) first; the + # pause ack returns it to pr_open, then it is re-signed and re-queued + if issue_once(rt, lifecycle.repo, changeset, "pause", { + "changeset_id": changeset["id"], "pr": int(number), + "reason": "re-signing: the verdict is ageing or the PR stalled in the merge queue"}): + events.append(f"requeue {changeset['id']}") + continue + if changeset["status"] == "verdict_posted" and _enqueue_outstanding(rt, changeset): + continue # its ack decides: queued (then the queued path applies) or refused + if changeset["status"] == "verdict_posted" and stale_verdict: + _resign(rt, changeset, "verdict close to its issue window") + events.append(f"resign {changeset['id']}") + continue + if changeset["status"] == "verdict_posted": + state, description = _gate(rt, changeset["head_sha"], since=signed_at) + if state == "failure" and not any(m in description for m in TRANSIENT_GATE): + if any(m in description for m in RESIGN_GATE): + _resign(rt, changeset, description) + events.append(f"resign {changeset['id']}") + elif any(m in description for m in REBUILD_GATE): + if not getattr(rt, "lease_owner", None): + continue # only the scheduler (holding the lease) rebuilds + new_id = rebuild(rt, lifecycle, changeset, description) + events.append(f"rebuilt {changeset['id']} as {new_id}" if new_id + else f"rebuild_failed {changeset['id']}") + else: # a real problem with the change itself: people decide + rt.ledger.update_changeset(changeset["id"], status="gate_failed") + rt.ledger.enqueue_human(lifecycle.repo, f"kb-gate failed: {description}", changeset["id"]) + events.append(f"gate_failed {changeset['id']}") + continue if changeset["status"] == "pr_open": pending = changeset.get("pending_item") if pending and pending.get("kind") == "post_verdict" and float(pending["expires_at"]) > rt.clock(): @@ -185,6 +359,9 @@ def advance(rt, lifecycle) -> list[str]: "changeset_id": changeset["id"], "pr": int(number), "head_sha": changeset["head_sha"], "comment": f"{VERDICT_MARKER}\n```json\n{json.dumps(envelope, ensure_ascii=False)}\n```\n", }): + current = rt.ledger.changeset(changeset["id"]) + rt.ledger.update_changeset(changeset["id"], detail={ + **current["detail"], "verdict_issued_at": envelope["payload"]["issued_at"]}) events.append(f"verdict issued {changeset['id']}") elif changeset["status"] == "verdict_posted" and not queued: if _gate_status(rt, changeset["head_sha"]) == "success": diff --git a/test/test_kb_rebuild.py b/test/test_kb_rebuild.py new file mode 100644 index 00000000..4ff211ed --- /dev/null +++ b/test/test_kb_rebuild.py @@ -0,0 +1,219 @@ +"""Stale verdicts: re-sign, rebuild on current main, or hand to people.""" + +from __future__ import annotations + +from infermatrix_copilot.kb_service import merge +from test_kb_flow import KnowledgeGitHub, _ack, _flow_runtime, _items, _open_pr + +HEAD = "d" * 40 + + +class GateGitHub(KnowledgeGitHub): + """kb-gate statuses with descriptions (the verifier's first problem).""" + + def __init__(self): + super().__init__() + self.descriptions: dict[str, str] = {} + + def _answer(self, url): + if "/commits/" in url and url.endswith("/status"): + sha = url.split("/commits/")[1].split("/")[0] + state = self.statuses.get(sha) + return {"statuses": [{"context": "kb-gate", "state": state, + "description": self.descriptions.get(sha, "")}] if state else []} + return super()._answer(url) + + +def _posted(tmp_path): + rt, lifecycle = _flow_runtime(tmp_path) + github = GateGitHub() + github.prs = rt.github.prs + rt.github = github + 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}} + assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] + _ack(tmp_path, rt, publisher, kind="post_verdict", changeset_id=changeset_id, ok=True) + assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" + return rt, lifecycle, changeset_id, publisher + + +def _fail(rt, description): + rt.github.statuses[HEAD] = "failure" + rt.github.descriptions[HEAD] = description + + +def test_the_signing_time_is_recorded_and_an_old_verdict_is_resigned(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + signed_at = rt.ledger.changeset(changeset_id)["detail"]["verdict_issued_at"] + assert signed_at == rt.clock() + start = rt.clock() + rt.clock = lambda: start + merge.RESIGN_AFTER + 1 + assert merge.advance(rt, lifecycle) == [f"resign {changeset_id}"] + assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] # a fresh verdict, same head + assert len(_items(tmp_path, "post_verdict")) >= 1 + + +def test_an_expired_or_missing_verdict_is_resigned(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + _fail(rt, "no valid signed verdict for PR #42 at dddddddddddd (rejected: verdict is outside its issue window)") + assert merge.advance(rt, lifecycle) == [f"resign {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" + + +def test_transient_gate_failures_wait(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + _fail(rt, "hold list: hold list is stale (the knowledge service may be down)") + assert merge.advance(rt, lifecycle) == [] + assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" + + +def test_a_real_problem_goes_to_people_once(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + _fail(rt, "L1 duplicate_rule_id DEMO-2a") + assert merge.advance(rt, lifecycle) == [f"gate_failed {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "gate_failed" + assert rt.ledger.human_queue("demo") + assert merge.advance(rt, lifecycle) == [] + + +def test_a_context_change_rebuilds_on_current_main_and_replaces_the_pr(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + _fail(rt, "context changed since the verdict was judged: repos/demo/core/_index.md") + assert merge.advance(rt, lifecycle) == [] # outside the scheduler: no lease, no rebuild + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + events = merge.advance(rt, lifecycle) + assert len(events) == 1 and events[0].startswith(f"rebuilt {changeset_id} as ") + new_id = events[0].rsplit(" ", 1)[1] + old, new = rt.ledger.changeset(changeset_id), rt.ledger.changeset(new_id) + assert old["status"] == "superseding" and new["status"] == "pr_requested" + assert new["detail"]["operations"] == old["detail"]["operations"] + assert new["detail"]["base_sha"] == rt.knowledge.main_sha() + (close,) = _items(tmp_path, "close") + assert close["body"]["pr"] == 42 and new_id in close["body"]["reason"] + assert any(item["body"]["changeset_id"] == new_id for item in _items(tmp_path, "open_pr")) + # the replaced PR stays tracked until GitHub shows it closed + assert merge.advance(rt, lifecycle) == [] + start = rt.clock() + rt.clock = lambda: start + 25 * 3600 # the close item expired unanswered + merge.advance(rt, lifecycle) + assert len(_items(tmp_path, "close")) == 2 + rt.github.prs[42] = {"state": "closed", "merged": False, "head": {"sha": HEAD}} + assert f"superseded {changeset_id}" in merge.advance(rt, lifecycle) + assert rt.ledger.changeset(changeset_id)["status"] == "superseded" + assert rt.ledger.changesets("demo", ("closed",)) == [] # not counted as overturned by people + + +def test_an_interrupted_rebuild_resumes_instead_of_staging_twice(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + old = rt.ledger.changeset(changeset_id) + real_update = rt.ledger.update_changeset + calls = {"n": 0} + + def crash_once(cs_id, **fields): + if cs_id == changeset_id and fields.get("status") == "superseding" and not calls["n"]: + calls["n"] += 1 + raise RuntimeError("crash after staging the rebuild") + return real_update(cs_id, **fields) + + rt.ledger.update_changeset = crash_once + import pytest + with pytest.raises(RuntimeError): + merge.rebuild(rt, lifecycle, old, "context changed since the verdict was judged: x") + rebuilt = rt.ledger.changesets("demo", ("gated", "pr_requested")) + assert len(rebuilt) == 1 + new_id = merge.rebuild(rt, lifecycle, rt.ledger.changeset(changeset_id), "context changed since ...") + assert new_id == rebuilt[0]["id"] + assert len([cs for cs in rt.ledger.changesets("demo", ("gated", "pr_requested")) if cs["kind"] == "rebuild"]) == 1 + + +def test_a_rebuild_that_fails_its_own_gate_leaves_the_pr_to_people(tmp_path): + from test_kb_intake_gate import ScriptedGateway, _generator_then_judge + + rt, lifecycle, changeset_id, _ = _posted(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}"] + assert rt.ledger.changeset(changeset_id)["status"] == "rebuild_failed" + assert _items(tmp_path, "close") == [] and rt.ledger.human_queue("demo") + + +def test_a_rule_that_no_longer_applies_goes_to_people(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + detail = rt.ledger.changeset(changeset_id)["detail"] + detail["operations"] = [{"kind": "edit_same_meaning", "page": "repos/demo/core/rules.md", + "rule_id": "DEMO-9z", "section_markdown": "## DEMO-9z — gone\n"}] + rt.ledger.update_changeset(changeset_id, detail=detail) + _fail(rt, "does not merge cleanly into the current base") + rt.lease_owner = rt.ledger.acquire_lease("scheduler") + assert merge.advance(rt, lifecycle) == [f"rebuild_failed {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "rebuild_failed" + assert rt.ledger.human_queue("demo") + + +def test_a_queued_pr_that_stalls_is_resigned_and_frees_the_queue_slot(tmp_path): + rt, lifecycle, changeset_id, publisher = _posted(tmp_path) + rt.github.statuses[HEAD] = "success" + assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] + _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) + assert rt.ledger.changeset(changeset_id)["status"] == "queued" + start = rt.clock() + rt.clock = lambda: start + merge.QUEUE_STALL + 1 + assert merge.advance(rt, lifecycle) == [f"requeue {changeset_id}"] + assert rt.ledger.changeset(changeset_id)["status"] == "queued" # the slot is held until dequeued + assert merge.advance(rt, lifecycle) == [] # the pause item is pending + (pause,) = _items(tmp_path, "pause") + assert pause["body"]["pr"] == 42 + _ack(tmp_path, rt, publisher, kind="pause", changeset_id=changeset_id, ok=True) + assert rt.ledger.changeset(changeset_id)["status"] == "pr_open" + assert merge.advance(rt, lifecycle) == [f"verdict issued {changeset_id}"] + + +def test_an_outstanding_enqueue_blocks_re_signing_until_its_ack(tmp_path): + rt, lifecycle, changeset_id, publisher = _posted(tmp_path) + rt.github.statuses[HEAD] = "success" + assert merge.advance(rt, lifecycle) == [f"enqueue issued {changeset_id}"] + detail = rt.ledger.changeset(changeset_id)["detail"] + rt.ledger.update_changeset(changeset_id, detail={**detail, "verdict_issued_at": rt.clock() - merge.RESIGN_AFTER - 1}) + _fail(rt, "no valid signed verdict for PR #42") + assert merge.advance(rt, lifecycle) == [] # not re-signed under a live enqueue + assert rt.ledger.changeset(changeset_id)["pending_item"]["kind"] == "enqueue" + _ack(tmp_path, rt, publisher, kind="enqueue", changeset_id=changeset_id, ok=True) # a delayed ack + assert rt.ledger.changeset(changeset_id)["status"] == "queued" + assert merge.advance(rt, lifecycle) == [f"requeue {changeset_id}"] # the queued path dequeues first + + +def test_our_own_hold_label_and_statuses_older_than_the_verdict_never_escalate(tmp_path): + rt, lifecycle, changeset_id, _ = _posted(tmp_path) + _fail(rt, "PR #42 carries kb:hold") + assert merge.advance(rt, lifecycle) == [] + _fail(rt, "L1 duplicate_rule_id DEMO-2a") + + class Old(GateGitHub): + def _answer(self, url): + answer = super()._answer(url) + for status in answer.get("statuses", []) if isinstance(answer, dict) else []: + status["updated_at"] = "2020-01-01T00:00:00Z" # posted before this verdict + return answer + + old = Old() + old.prs, old.statuses, old.descriptions = rt.github.prs, rt.github.statuses, rt.github.descriptions + rt.github = old + assert merge.advance(rt, lifecycle) == [] + assert rt.ledger.changeset(changeset_id)["status"] == "verdict_posted" + + +def test_an_outstanding_enqueue_holds_the_single_queue_slot(tmp_path): + rt, lifecycle, first, publisher = _posted(tmp_path) + rt.github.statuses[HEAD] = "success" + assert merge.advance(rt, lifecycle) == [f"enqueue issued {first}"] + second = rt.ledger.new_changeset_id("demo", "intake") + rt.ledger.stage_intake(rt.ledger.acquire_lease("t2"), "demo", second, detail={}, status="verdict_posted", + kind="intake", verdicts=[], human_reason="", drafted_events=[]) + rt.ledger.update_changeset(second, pr_number=43, head_sha="e" * 40) + rt.github.prs[43] = {"state": "open", "merged": False, "head": {"sha": "e" * 40}} + rt.github.statuses["e" * 40] = "success" + assert merge.advance(rt, lifecycle) == [] # the first's enqueue is still executable + assert [i["body"]["changeset_id"] for i in _items(tmp_path, "enqueue")] == [first]