diff --git a/docs/knowledge-model-policy.md b/docs/knowledge-model-policy.md index 74f5d81e..2c7bdc64 100644 --- a/docs/knowledge-model-policy.md +++ b/docs/knowledge-model-policy.md @@ -189,14 +189,41 @@ fails, the run reports `publication-failed`; the next sweep attempts the same single write once more. A head can also carry statuses from other contexts, so the read is bounded by -both 50 pages and a deadline. The deadline belongs to the sweep, not to one -head: targets are processed serially under one workflow timeout, so a fresh -budget per head would let a long sweep run out of time before a later target's -failure is published. Each history request is also capped by the time remaining, -and 15 seconds are held back so that publication still fits. Exhausting either -bound is reported as `history-unavailable`, which publishes the failure -immediately, so a long scan cannot consume the workflow's deadline before -anything replaces an earlier success. +both 50 pages and time. Time is split rather than shared as one absolute +deadline. Each target receives an equal part of what is left in the sweep, +capped at 30 seconds, so a target that finishes early returns its unused time +to the later ones. The sweep itself is bounded by 120 seconds per attempt, and +one 15-second reserve is withheld from the sweep so publication still fits after +the last read. Each history request is additionally capped by the share still +remaining. Exhausting the page bound or the target's share is reported as +`history-unavailable`, which publishes the failure immediately and lets the +remaining targets continue. + +The former design gave every target one absolute 60-second deadline: targets are +processed serially, so the first ones spent it and each later target then failed +closed, and a whole-sweep retry replayed the same prefix. + +Retries now carry progress. Each attempt writes the targets it settled to a plan +file, and the next attempt inherits them instead of re-invalidating and +re-publishing the same verdicts, so an exhausted tail cannot replace a verdict an +earlier attempt already published. A carried target is trusted only after one +cheap re-read shows it is still open against the policy branch at the same head, +base, state and declaration. Any change to those falls back to full evaluation, +and the plan is discarded outright when the trusted base differs, because a +different base means different loaded code and policy. + +Carrying is limited to settled verdicts: `compliant`, `noncompliant` and the two +capacity states. Work reported as rerunnable - unreadable evidence, a failed +write, an advanced or entered target - is never inherited, so a genuine failure +is retried and stays visible rather than being optimistically cached. + +The premise stored with a verdict must be the one that produced it. A target +reassessed during the publication pass is therefore stored with the snapshot, +approval comments and ownership that reassessment used, not with the earlier +evaluation's evidence; otherwise a changed-then-reverted approval would make the +stored premise match again and the stale success would be reused. If the open-PR +collection cannot be read, no target is carried, because a target read alone +cannot prove it does not share its head. At or above the cap nothing can be written at all. The run reports `budget-saturated` without attempting a write. An earlier `success` may still be @@ -232,6 +259,22 @@ does not close the window between the last metadata read and the status write, a it cannot act when GitHub supplies no event. Repository protection plus a required check is what turns these states into an actual merge block. +```mermaid +flowchart LR + subgraph before["Before"] + A1["One 60 s deadline"] --> A2["Earlier targets spend it"] + A2 --> A3["Later targets: history-unavailable"] + A3 --> A4["Retry replays the same prefix"] + A4 --> A2 + end + subgraph after["After"] + B1["120 s sweep bound"] --> B2["Share = remaining / targets left, max 30 s"] + B2 --> B3["Early target returns its unused share"] + B3 --> B4["Settled targets written to the plan"] + B4 --> B5["Retry re-checks the premise, then evaluates only what is left"] + end +``` + When the trusted base advances mid-sweep, the checker returns a retryable result promptly instead of re-evaluating against its own stale checkout. It exits `2`, and trusted `tools/run_knowledge_model_policy.sh` refetches `main` and runs the @@ -253,7 +296,7 @@ Each result carries `kind` alongside the status `state`: | `advanced` | The evaluated head was superseded; nothing is written to the new head | | `publication-failed` | Status write failed; the verdict was not recorded | | `invalidation-failed` | The earlier result could not be replaced by `pending`; no verdict was published | -| `history-unavailable` | The head's status history could not be read, so the capacity margin is unproven; an explicit failure replaces the old result immediately, before any other target is processed | +| `history-unavailable` | The head's status history could not be read within its share of the sweep, so the capacity margin is unproven; an explicit failure replaces the old result immediately, before any other target is processed | | `budget-exhausted` | The head has used its reserved status capacity; one failure is recorded there, no success is published and a new head is required | | `budget-saturated` | The head already holds 1,000 statuses for this context; nothing can be written, so an earlier success may remain the newest entry | diff --git a/tests/test_knowledge_model_policy.py b/tests/test_knowledge_model_policy.py index d9336329..ca6014d4 100644 --- a/tests/test_knowledge_model_policy.py +++ b/tests/test_knowledge_model_policy.py @@ -17,8 +17,11 @@ EXIT_NONCOMPLIANT, EXIT_RERUN, HISTORY_PAGES, + HISTORY_RESERVE, STATUS_LIMIT, STATUS_RESERVE, + SWEEP_SECONDS, + TARGET_SECONDS, GitHub, approved, assemble_file_pages, @@ -32,9 +35,11 @@ json_block, matching, model_errors, + plan_file, refresh_event, snapshot, summarize, + target_deadline, within_budget, ) @@ -335,6 +340,11 @@ def test_workflow_never_executes_pr_code_and_refreshes_metadata_and_base(): # The runner refetches trusted main and reruns only for the stale-base exit code. assert "git fetch --quiet origin main" in runner assert "--policy-sha" in runner and 'status" -ne 2' in runner + # Each attempt shares one plan file, so a retry inherits the targets an earlier attempt + # settled instead of re-publishing them and starving the tail. + assert '--plan "$plan"' in runner + # Three attempts, each bounded by the sweep plus one target's ceiling, still fit the job. + assert 3 * (SWEEP_SECONDS + TARGET_SECONDS) < 10 * 60 assert "pull_request.head" not in json.dumps(value) assert "knowledge-model-policy" not in ( ROOT / ".github/workflows/ci-ready.yml" @@ -1251,34 +1261,272 @@ def test_api_slow_history_scan_stops_before_the_workflow_deadline(api_fixture, p seed_history(state, 400, context="other-ci") # A page cap does not bound elapsed time, so slow successful responses must trip the time # budget on their own. - monkeypatch.setattr("tools.check_knowledge_model_policy.HISTORY_SECONDS", 0) + monkeypatch.setattr("tools.check_knowledge_model_policy.TARGET_SECONDS", 0) result = check_pr(api, 123, policy, BASE, False) assert result["kind"] == "history-unavailable" assert latest_status(state)["state"] == "failure" -def test_api_history_deadline_is_shared_across_the_sweep(api_fixture, policy, monkeypatch): +def test_target_deadline_gives_each_target_an_equal_share(monkeypatch): + clock = {"now": 1000.0} + monkeypatch.setattr("tools.check_knowledge_model_policy.time", + SimpleNamespace(monotonic=lambda: clock["now"])) + deadline = clock["now"] + SWEEP_SECONDS + # The reserve is withheld once for publication, and the rest is split between the targets + # still to be evaluated rather than being available to whichever target runs first. + share = (SWEEP_SECONDS - HISTORY_RESERVE) / 4 + assert share < TARGET_SECONDS + assert target_deadline(deadline, 4) == clock["now"] + share + # Fewer remaining targets get a larger share, but never more than the per-target ceiling. + assert target_deadline(deadline, 2) == clock["now"] + TARGET_SECONDS + assert target_deadline(deadline, 1) == clock["now"] + TARGET_SECONDS + # A spent sweep leaves no share, so the read fails closed rather than extending the run. + clock["now"] = deadline - HISTORY_RESERVE + assert target_deadline(deadline, 1) <= clock["now"] + + +def test_api_a_slow_target_cannot_starve_the_next(api_fixture, policy, monkeypatch): api, state = api_fixture other = "c" * 40 - state["prs"][124] = {**pr(), "number": 124, "head": {"sha": other}} + state["prs"][124] = {**pr({**declared(), "head_sha": other}), "number": 124, + "head": {"sha": other}} + # A multi-page read is needed for the target to outlive its own share: the check happens + # before each page, not after the whole scan. + seed_history(state, 400, context="other-ci") clock = {"now": 1000.0} - # A slow but successful read on the first target consumes nearly the whole budget; the second - # target then has no time left, so it must report the failure instead of starting a new scan. monkeypatch.setattr("tools.check_knowledge_model_policy.time", SimpleNamespace(monotonic=lambda: clock["now"])) def slow(route): - if route.endswith("/statuses") or "/statuses?" in route: - clock["now"] += 55 + # The read never completes within this target's own share, but the sweep itself still has + # time left, so the next target must be evaluated instead of inheriting a dead deadline. + if route.endswith("/statuses"): + clock["now"] += 20 state["before_request"] = slow results = check_prs(api, policy, BASE, False, [123, 124]) - assert [r["kind"] for r in results] == ["compliant", "history-unavailable"] - assert results[1]["rerun"] is True - assert [r["state"] for r in results] == ["success", "failure"] - # The second target's failure is published, so its earlier state cannot survive the sweep. + assert [r["kind"] for r in results] == ["history-unavailable", "compliant"] + assert results[0]["rerun"] is True + assert [r["state"] for r in results] == ["failure", "success"] + # The slow target still fails closed: its earlier state cannot survive the sweep. assert next(s for s in reversed(state["statuses"]) - if s["sha"] == other)["state"] == "failure" + if s["sha"] == HEAD)["state"] == "failure" + assert next(s for s in reversed(state["statuses"]) + if s["sha"] == other)["state"] == "success" + + +def test_api_every_target_is_evaluated_past_the_former_shared_window(api_fixture, policy, + monkeypatch): + api, state = api_fixture + numbers = list(range(200, 218)) + for number in numbers: + head = f"{number:040x}" + state["prs"][number] = {**pr({**declared(), "head_sha": head}), "number": number, + "head": {"sha": head}} + clock = {"now": 1000.0} + monkeypatch.setattr("tools.check_knowledge_model_policy.time", + SimpleNamespace(monotonic=lambda: clock["now"])) + + def ordinary(route): + # Four seconds of ordinary metadata work per target: well over the 45 seconds the former + # single deadline left for the whole sweep once its 15-second reserve was removed. + if route.endswith("/files"): + clock["now"] += 4 + + state["before_request"] = ordinary + results = check_prs(api, policy, BASE, False, numbers) + assert len(results) == len(numbers) + assert [r["kind"] for r in results] == ["compliant"] * len(numbers) + assert clock["now"] - 1000.0 > SWEEP_SECONDS / 2 + for number in numbers: + assert next(s for s in reversed(state["statuses"]) + if s["sha"] == f"{number:040x}")["state"] == "success" + + +def test_api_retries_keep_completed_targets_instead_of_rewriting_them(api_fixture, policy, + monkeypatch): + api, state = api_fixture + numbers = list(range(200, 218)) + for number in numbers: + head = f"{number:040x}" + state["prs"][number] = {**pr({**declared(), "head_sha": head}), "number": number, + "head": {"sha": head}} + clock = {"now": 1000.0} + monkeypatch.setattr("tools.check_knowledge_model_policy.time", + SimpleNamespace(monotonic=lambda: clock["now"])) + + def ordinary(route): + # Eight seconds of ordinary metadata work per target: one sweep can reach about 14 of the + # 18 targets before its effective window closes, so the tail needs another attempt. + if route.endswith("/files"): + clock["now"] += 8 + + state["before_request"] = ordinary + plan = {"policy_sha": BASE, "completed": {}} + attempts, posts = [], [] + for _ in range(3): + state["requests"].clear() + results = check_prs(api, policy, BASE, False, numbers, plan) + attempts.append(results) + posts.append([route for method, route in state["requests"] if method == "POST"]) + if exit_code(results) != EXIT_RERUN: + break + # The local runner loop restarts the clock for each attempt. + clock["now"] = 1000.0 + + # The first sweep starves its own tail; the second finishes exactly the work the first left. + assert len(attempts) == 2 + assert [r["kind"] for r in attempts[0]].count("history-unavailable") > 0 + # A later attempt spends its budget on the unfinished targets only; the earlier verdicts are + # reported from the plan and are never re-invalidated or rewritten. + assert not any(kind == "history-unavailable" for kind in + [r["kind"] for r in attempts[1]]) + for number in numbers: + assert next(s for s in reversed(state["statuses"]) + if s["sha"] == f"{number:040x}")["state"] == "success" + completed_first = [r["pull_request"] for r in attempts[0] if r["kind"] == "compliant"] + assert completed_first + # Those targets are neither re-invalidated nor rewritten by the second attempt. + assert not any(route.endswith(f"{number:040x}") + for number in completed_first for route in posts[1]) + # Every target is still reported once per attempt, including the carried ones. + assert [r["pull_request"] for r in attempts[1]] == numbers + + +def test_api_a_carried_target_with_changed_premises_is_evaluated_again(api_fixture, policy): + api, state = api_fixture + first = check_pr(api, 123, policy, BASE, False) + assert first["kind"] == "compliant" + plan = {"policy_sha": BASE, "completed": {}} + results = check_prs(api, policy, BASE, False, [123], plan) + assert [r["kind"] for r in results] == ["compliant"] + # An edited declaration keeps the same head but invalidates the carried verdict. + state["prs"][123]["body"] = block(declared([attribution("unknown", "unknown")])) + results = check_prs(api, policy, BASE, False, [123], plan) + assert [r["kind"] for r in results] == ["noncompliant"] + assert latest_status(state)["state"] == "failure" + + +def test_api_a_revoked_approval_is_not_carried_as_a_settled_success(api_fixture, policy): + api, state = api_fixture + value = declared([attribution("unknown", "unknown")]) + value["exception"] = {"paths": [KNOWLEDGE], "reason": "Bounded exception"} + state["prs"][123] = pr(value) + state["comments"][123] = [approval({KNOWLEDGE})] + plan = {"policy_sha": BASE, "completed": {}} + assert check_prs(api, policy, BASE, False, [123], plan)[0]["kind"] == "compliant" + # Unchanged premises are carried rather than re-evaluated. + assert check_prs(api, policy, BASE, False, [123], plan)[0]["kind"] == "compliant" + # The PR snapshot is untouched, but the approval that produced the verdict is revoked. + state["comments"][123] = [approval({KNOWLEDGE}, decision="revoke")] + results = check_prs(api, policy, BASE, False, [123], plan) + assert [r["kind"] for r in results] == ["noncompliant"] + assert results[0]["rerun"] is False + assert latest_status(state)["state"] == "failure" + + +def test_api_a_new_shared_head_peer_invalidates_a_carried_success(api_fixture, policy): + api, state = api_fixture + plan = {"policy_sha": BASE, "completed": {}} + assert check_prs(api, policy, BASE, False, [123], plan)[0]["kind"] == "compliant" + assert check_prs(api, policy, BASE, False, [123], plan)[0]["kind"] == "compliant" + # A peer on another base can share this head without ever entering the target list, so + # ownership has to be re-derived from the open-PR collection rather than from the target. + state["prs"][900] = {**pr(), "number": 900, "head": {"sha": HEAD}, + "base": {"sha": BASE, "ref": "release"}} + results = check_prs(api, policy, BASE, False, [123], plan) + assert [r["kind"] for r in results] == ["noncompliant"] + assert "shared heads are ineligible" in results[0]["errors"][0] + assert latest_status(state)["state"] == "failure" + + +def test_api_a_reassessed_verdict_carries_the_premises_it_actually_used(api_fixture, policy): + api, state = api_fixture + value = declared([attribution("unknown", "unknown")]) + value["exception"] = {"paths": [KNOWLEDGE], "reason": "Bounded exception"} + state["prs"][123] = pr(value) + plan = {"policy_sha": BASE, "completed": {}} + # Ordinary discussion but no valid approval at evaluation time, so evaluation refuses. The + # approval then appears before the publication pass, which revalidates the comments it read, + # reassesses once, and publishes success from the new approval. + discussion = {"id": 1, "user": {"login": "commenter"}, "body": "Discussing the change"} + state["comments"][123] = [discussion] + reads = {"comments": 0} + original = api.pages + + def pending_approval(route): + if route == "issues/123/comments": + reads["comments"] += 1 + if reads["comments"] > 1: + return [discussion, approval({KNOWLEDGE})] + return original(route) + + api.pages = pending_approval + results = check_prs(api, policy, BASE, False, [123], plan) + api.pages = original + assert [r["kind"] for r in results] == ["compliant"] + # The stored premise is the approval the reassessment actually used, not the empty list the + # earlier evaluation read. + assert plan["completed"]["123"]["comments"] == [discussion, approval({KNOWLEDGE})] + # The approval is removed again. The stored premise must be the one the reassessment used, + # so reverting to "no approval" is a change and the settled success cannot be carried. + state["comments"][123] = [] + results = check_prs(api, policy, BASE, False, [123], plan) + assert [r["kind"] for r in results] == ["noncompliant"] + assert latest_status(state)["state"] == "failure" + + +def test_api_an_unreadable_collection_never_carries_a_settled_success(api_fixture, policy): + api, state = api_fixture + plan = {"policy_sha": BASE, "completed": {}} + assert check_prs(api, policy, BASE, False, [123], plan)[0]["kind"] == "compliant" + # The open-PR listing is unreadable, so the target alone cannot prove unique ownership: the + # carry must be refused rather than treated as an unshared head. + state["fail"] = lambda route: route == "pulls" + results = check_prs(api, policy, BASE, False, [123], plan) + assert results[0]["kind"] == "unverified" + assert results[0]["rerun"] is True + assert exit_code(results) == EXIT_RERUN + + +def test_api_carry_revalidation_tolerates_an_unreadable_target(api_fixture, policy): + api, state = api_fixture + state["prs"][124] = {**pr({**declared(), "head_sha": "c" * 40}), "number": 124, + "head": {"sha": "c" * 40}} + plan = {"policy_sha": BASE, "completed": {}} + assert len(check_prs(api, policy, BASE, False, None, plan)) == 2 + # One carried target cannot be read during revalidation. That is a fallback, not an + # invocation-level error, and the other targets must still be evaluated. + original = api.get + + def unreadable(route): + if route == "pulls/123": + raise HTTPError(route, 503, "Unavailable", {}, None) + return original(route) + + api.get = unreadable + results = check_prs(api, policy, BASE, False, None, plan) + assert len(results) == 2 + assert {r["pull_request"] for r in results} == {123, 124} + + +def test_plan_is_discarded_when_the_trusted_base_changes(tmp_path): + path = tmp_path / "plan.json" + path.write_text(json.dumps({"policy_sha": "a" * 40, + "completed": {"123": {"head": "b" * 40}}}), encoding="utf-8") + # A different trusted base means different loaded code and policy, so no verdict carries over. + assert plan_file(path, "c" * 40) == {"policy_sha": "c" * 40, "completed": {}} + assert plan_file(path, "a" * 40)["completed"] == {"123": {"head": "b" * 40}} + # A corrupt or unreadable plan is treated the same way, never as completed work. + path.write_text("not json", encoding="utf-8") + assert plan_file(path, "a" * 40) == {"policy_sha": "a" * 40, "completed": {}} + + +def test_runner_keeps_one_plan_across_attempts(): + runner = (ROOT / "tools/run_knowledge_model_policy.sh").read_text(encoding="utf-8") + # The runner owns the retry loop and passes one plan file to every attempt. + assert '--plan "$plan"' in runner + assert "rm -f \"$plan\"" in runner def test_api_other_contexts_do_not_consume_the_budget(api_fixture, policy): diff --git a/tools/check_knowledge_model_policy.py b/tools/check_knowledge_model_policy.py index c4efa1d5..f3470ec4 100644 --- a/tools/check_knowledge_model_policy.py +++ b/tools/check_knowledge_model_policy.py @@ -31,11 +31,13 @@ # recorded by this text and leaves it in place instead of spending another reserved slot. BUDGET_DESCRIPTION = "Status history is nearly full; push a new head" # A head can carry statuses from other contexts, so the page count alone does not bound the scan. -# The deadline is set once per sweep, not per head: targets are processed serially under one -# workflow timeout, so a fresh budget for each read would let a long sweep die before a later -# target's failure is published. The reserve keeps time free for that publication. +# Each target gets its own share of the time still left in the sweep, capped by a per-target +# ceiling: one absolute deadline shared by every target let the earlier targets consume the whole +# window, after which each later target failed closed on every retry. The sweep deadline remains +# as the outer bound, and the reserve keeps time free for publication after the last read. HISTORY_PAGES = 50 -HISTORY_SECONDS = 60 +TARGET_SECONDS = 30 +SWEEP_SECONDS = 120 HISTORY_RESERVE = 15 # Outcome kinds, reported separately so evaluator noise is never read as a rejection. KINDS = ("compliant", "noncompliant", "evaluator-error", "unverified", "deferred", "closed", @@ -54,6 +56,9 @@ "invalidation-failed", "history-unavailable"} # A lifecycle kind forbids publication: the captured head is no longer the current target. NO_PUBLISH = {"closed", "advanced"} +# Kinds that settle a target for this attempt. A later attempt trusts one only after re-checking +# that the target is still open against the policy branch at the same head. +COMPLETED_KINDS = {"compliant", "noncompliant", "closed", "budget-exhausted", "budget-saturated"} def summarize(results): @@ -330,12 +335,15 @@ def history(self, head, deadline): # GitHub treats the context name case-insensitively, so counting only the exact spelling # would miss equivalent entries and could spend the last free slots on a pending and a # success. The list is newest first, which makes entry 0 the head's effective status. + # + # The deadline is already one target's share of the sweep, so this method only stops when + # that share is spent; the publication reserve is withheld by the caller. key = CONTEXT.casefold() found = [] page = 1 while True: remaining = deadline - time.monotonic() - if page > HISTORY_PAGES or remaining <= HISTORY_RESERVE: + if page > HISTORY_PAGES or remaining <= 0: raise ValueError("Status history scan exceeded its budget") query = urlencode({"per_page": 100, "page": page}) items, links = self.request(f"commits/{head}/statuses?{query}", @@ -561,6 +569,17 @@ def invalidated(api, head, dry_run, deadline): return None, True +def target_deadline(deadline, remaining_targets): + # One absolute budget for the whole sweep starved every target after the earlier ones had + # spent it, and a retry replayed the same prefix. Each target instead gets an equal share of + # the time left, capped so one slow head cannot hold the sweep: a target that finishes early + # returns its unused time to the later ones. Inside the publication reserve the share is + # negative, which makes the read fail closed immediately. + now = time.monotonic() + share = (deadline - HISTORY_RESERVE - now) / max(remaining_targets, 1) + return now + min(TARGET_SECONDS, share) + + def lifecycle_result(result, scan_error): # Invalidate an old success when the current declaration was not validly assessed. The # check still publishes a verdict; only an unknowable lifecycle is reported without a write. @@ -573,10 +592,63 @@ def lifecycle_result(result, scan_error): return {**result, "rerun": False} -def check_prs(api, policy, policy_sha, dry_run, numbers=None): - # One deadline for the whole sweep: history reads share it, so a long scan on an early target - # cannot leave a later target without time to publish its own failure. - deadline = time.monotonic() + HISTORY_SECONDS +def plan_file(path, policy_sha): + # The plan is scoped to one trusted base and one runner invocation. A different base means the + # loaded policy and checker are different code, so earlier verdicts may not describe it. + if path is None or not path.exists(): + return {"policy_sha": policy_sha, "completed": {}} + try: + value = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=unique_object) + except (OSError, ValueError, TypeError): + return {"policy_sha": policy_sha, "completed": {}} + if not isinstance(value, dict) or value.get("policy_sha") != policy_sha: + return {"policy_sha": policy_sha, "completed": {}} + if not isinstance(value.get("completed"), dict): + return {"policy_sha": policy_sha, "completed": {}} + return value + + +def write_plan(path, plan): + if path is None: + return + path.write_text(json.dumps(plan, ensure_ascii=True, sort_keys=True), encoding="utf-8") + + +def still_current(api, number, entry, policy, open_prs, scan_error): + # Premise revalidation for work another attempt already published. The verdict depended on + # evidence outside the PR snapshot, so a carried entry must prove all of it still holds: the + # metadata snapshot, the approvals that were read, and head ownership across the open PRs. A + # stored premise that is missing cannot be revalidated, so it falls back to full evaluation. + if "comments" not in entry or "ownership" not in entry: + return False + if scan_error is not None: + # Without the open-PR collection the target alone looks like a unique owner, which is the + # same assumption `lifecycle_result` refuses for a fresh verdict. + return False + try: + live = api.get(f"pulls/{number}") + snapshot = admission_snapshot(live) + except API_ERRORS: + return False + if snapshot != tuple(entry.get("snapshot") or ()) or not addressable(live, policy): + return False + if head_owner(open_prs or [live], number, live["head"]["sha"]) != entry["ownership"]: + # A peer appearing or leaving changes eligibility without changing this PR. + return False + if entry["comments"]: + try: + if api.pages(f"issues/{number}/comments") != entry["comments"]: + # An approval added, edited or revoked changes the verdict it produced. + return False + except API_ERRORS: + return False + return True + + +def check_prs(api, policy, policy_sha, dry_run, numbers=None, plan=None): + # The sweep deadline bounds the whole evaluation pass and keeps publication time free; each + # target's read is bounded separately by its share of what remains. + deadline = time.monotonic() + SWEEP_SECONDS scan_error = None try: open_prs = api.pages("pulls?state=open") @@ -590,10 +662,27 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None): "kind": "unverified", "description": "Lifecycle metadata unavailable", "errors": [evaluation_failure(scan_error)["errors"][0]]}] numbers = [pr["number"] for pr in open_prs if addressable(pr, policy)] + # Work another attempt already settled is not repeated: re-running it would spend the same + # bounded time and, in a later attempt's exhausted tail, replace a published verdict with a + # failure. A carried target is trusted only after its own premise is re-confirmed here. + completed = {} if plan is None else plan["completed"] + carried = {} + pending = list(numbers) + for number in list(numbers): + entry = completed.get(str(number)) + if not isinstance(entry, dict): + continue + head = entry.get("head") + if isinstance(head, str) and SHA.fullmatch(head): + if not still_current(api, number, entry, policy, open_prs, scan_error): + continue + carried[number] = entry + pending.remove(number) + numbers = pending # One evaluation pass. Stale work is returned as deferred so the outer runner, which can # refetch trusted main, owns retries; this checker never repairs its own loaded policy. records = {} - for number in numbers: + for index, number in enumerate(numbers): try: pr = api.get(f"pulls/{number}") except API_ERRORS as error: @@ -610,7 +699,8 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None): "ownership": None, "publish": False, "result": closed_result("Not open against the policy branch")} continue - failure, publish = invalidated(api, head, dry_run, deadline) + failure, publish = invalidated(api, head, dry_run, + target_deadline(deadline, len(numbers) - index)) if failure is not None: # Without a published invalidation an old success would still be the head's effective # state, so this target gets no verdict and asks the runner for a fresh attempt. @@ -695,11 +785,16 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None): ownership = None if scan_error is not None else head_owner(open_prs, number, head) if premise or ownership != record["ownership"]: # Reassess against current metadata and the current open-PR collection: a closed - # peer can end head sharing, and a moved body or approval changes the verdict. - result, _ = read_pr_result(api, live, policy, policy_sha, - open_prs if open_prs else [live]) + # peer can end head sharing, and a moved body or approval changes the verdict. The + # premises recorded afterwards must be the ones this reassessment used, or the plan + # would pair a new verdict with evidence that no longer produced it. + result, comments = read_pr_result(api, live, policy, policy_sha, + open_prs if open_prs else [live]) + record["comments"] = comments + record["snapshot"] = admission_snapshot(live) ownership = (None if scan_error is not None else head_owner(open_prs, number, head)) + record["ownership"] = ownership if result["kind"] in NO_PUBLISH: # The reassessment observed that this head is no longer the target. published[number] = {**result, "head": head, @@ -723,6 +818,30 @@ def check_prs(api, policy, policy_sha, dry_run, numbers=None): evaluation_failure(error)["errors"][0]]} published[number] = {**result, "head": head, "rerun": result["kind"] in RERUN_KINDS} + # Carried verdicts are reported as they were published by the earlier attempt. Nothing is + # rewritten for them here: their head was re-confirmed as current at the start of this run. + for number, entry in carried.items(): + if number not in published: + published[number] = {**entry["result"], "head": entry["head"], "rerun": False} + if plan is not None: + # The next attempt inherits only what this one settled, together with the premise the + # verdict was assessed against: the head alone would let an edited declaration or a moved + # base be carried as if it had been evaluated. + plan["completed"] = {str(number): entry for number, entry in carried.items()} + for number, item in published.items(): + settled = plan["completed"].get(str(number), {}) + record = records.get(number) + head_value = record["head"] if record else settled.get("head") + snapshot_value = record["snapshot"] if record else settled.get("snapshot") + comments_value = record["comments"] if record else settled.get("comments") + ownership_value = record["ownership"] if record else settled.get("ownership") + if (item["kind"] in COMPLETED_KINDS and snapshot_value is not None + and comments_value is not None): + plan["completed"][str(number)] = { + "head": head_value, "snapshot": snapshot_value, + "comments": comments_value, "ownership": ownership_value, + "result": {key: item[key] for key in + ("state", "kind", "description", "errors")}} results = [] for number in sorted(published): item = published[number] @@ -754,6 +873,7 @@ def main(argv=None): parser.add_argument("--policy-sha", required=True) parser.add_argument("--pull-request", type=int) parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--plan", type=Path) args = parser.parse_args(argv) if not SHA.fullmatch(args.policy_sha): raise ValueError("Expected the trusted policy checkout SHA") @@ -772,7 +892,9 @@ def main(argv=None): event = json.loads(Path(args.event_path).read_text(encoding="utf-8")) # Every trigger shares one writer, so a superseded queued event still covers all open PRs. numbers = None if refresh_event(event, os.environ["GITHUB_EVENT_NAME"]) else [] - results = check_prs(api, policy, args.policy_sha, args.dry_run, numbers) + plan = plan_file(args.plan, args.policy_sha) + results = check_prs(api, policy, args.policy_sha, args.dry_run, numbers, plan) + write_plan(args.plan, plan) for result in results: print(json.dumps(result, ensure_ascii=True)) print(json.dumps(summarize(results), ensure_ascii=True)) diff --git a/tools/run_knowledge_model_policy.sh b/tools/run_knowledge_model_policy.sh index c8927d34..9f7f97fe 100644 --- a/tools/run_knowledge_model_policy.sh +++ b/tools/run_knowledge_model_policy.sh @@ -2,18 +2,26 @@ # Run the knowledge admission checker against the current trusted main. # Exit 2 means the work could not be completed (stale base, unreadable evidence); this runner # is the single retry owner, so it refetches main and runs the checker again. +# A plan file carries the targets an attempt already settled, so a retry spends its bounded time +# on the remaining work instead of re-publishing the same verdicts and starving the tail again. +# The checker scopes the plan to the trusted base; a different base discards it. set -uo pipefail attempts=3 +plan="${RUNNER_TEMP:-${TMPDIR:-/tmp}}/knowledge-model-policy-plan-${GITHUB_RUN_ID:-local}.json" +rm -f "$plan" for attempt in $(seq 1 "$attempts"); do git fetch --quiet origin main || exit 1 git checkout --quiet origin/main || exit 1 status=0 - python tools/check_knowledge_model_policy.py --policy-sha "$(git rev-parse HEAD)" "$@" || status=$? + python tools/check_knowledge_model_policy.py --policy-sha "$(git rev-parse HEAD)" \ + --plan "$plan" "$@" || status=$? if [ "$status" -ne 2 ]; then + rm -f "$plan" exit "$status" fi echo "Admission work incomplete; retrying against current main (${attempt}/${attempts})" >&2 done +rm -f "$plan" echo "Admission work still incomplete after ${attempts} attempts" >&2 exit 1