From a478ae4938451159eb98cf5771e0461ab2e38d70 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 21:47:47 -0700
Subject: [PATCH 1/9] Batch source callback reads and reconcile completed
 source heads

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 scripts/reconcile-queue.py    | 341 +++++++++++++++++++++++++++++-----
 tests/test_reconcile_queue.py | 146 ++++++++++++++-
 2 files changed, 440 insertions(+), 47 deletions(-)

diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index e289874..20874c4 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -34,6 +34,10 @@
 SOURCE_ACTIVITY = os.environ.get("SOURCE_ACTIVITY", "")
 SOURCE_CACHE: dict[str, Any] = {}
 SOURCE_EVIDENCE: dict[str, dict[str, Any]] = {}
+SOURCE_CALLBACK = False
+SOURCE_SWEEP = False
+SOURCE_VIEW: dict[str, Any] = {}
+SOURCE_RUN_LISTS: dict[str, Any] = {}
 PAGE_SIZE = 100
 GH = shutil.which("gh") or "/usr/bin/gh"
 
@@ -86,11 +90,15 @@ def admit(pr: dict[str, Any]) -> bool:
     source_statuses = None
     if source:
         sha = pr["head"]["sha"]
-        if SOURCE_HEAD and SOURCE_HEAD != sha:
+        if SOURCE_HEAD and not SOURCE_SWEEP and SOURCE_HEAD != sha:
             return False
-        source_statuses = read_statuses(sha)
-        if not SOURCE_HEAD and any(
-            status["context"] == ADMISSION for status in source_statuses.values()
+        source_statuses = SOURCE_VIEW[sha] if SOURCE_CALLBACK else read_statuses(sha)
+        if (
+            not SOURCE_HEAD
+            and not SOURCE_SWEEP
+            and any(
+                status["context"] == ADMISSION for status in source_statuses.values()
+            )
         ):
             state = graphql(
                 "query($id:ID!) { node(id:$id) { ... on PullRequest { mergeQueueEntry { id } } } }",
@@ -98,28 +106,30 @@ def admit(pr: dict[str, Any]) -> bool:
             )["node"]
             return bool(state["mergeQueueEntry"])
         if not reconcile_source(pr, source_statuses):
-            if SOURCE_HEAD == sha and any(
-                status["context"] == ADMISSION and status.get("state") == "success"
-                for status in source_statuses.values()
+            if (
+                (SOURCE_HEAD == sha or SOURCE_SWEEP)
+                and any(
+                    status["context"] == ADMISSION and status.get("state") == "success"
+                    for status in source_statuses.values()
+                )
+                and source_lease(pr)
             ):
-                fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}")
-                current = api(
-                    f"repos/{REPOSITORY}/git/ref/heads/{quote(pr['head']['ref'], safe='/')}"
+                api(
+                    f"repos/{REPOSITORY}/statuses/{sha}",
+                    "POST",
+                    {
+                        "state": "pending",
+                        "context": ADMISSION,
+                        "description": "Source validation changed; current successful jobs required",
+                    },
                 )
-                if (
-                    source_identity(fresh)
-                    and fresh["head"]["sha"] == sha
-                    and current["object"]["sha"] == sha
-                ):
-                    api(
-                        f"repos/{REPOSITORY}/statuses/{sha}",
-                        "POST",
-                        {
-                            "state": "pending",
-                            "context": ADMISSION,
-                            "description": "Source validation changed; current successful jobs required",
-                        },
-                    )
+            return False
+        if (
+            SOURCE_CALLBACK
+            and source_statuses.get(ADMISSION, {}).get("state") == "success"
+        ):
+            # A current already-admitted source needs no queue scan or write.
+            # Its queue validation completion callbacks own queue progression.
             return False
     trusted = (
         source
@@ -162,13 +172,17 @@ def admit(pr: dict[str, Any]) -> bool:
     if state["mergeQueueEntry"]:
         if (
             source
-            and SOURCE_HEAD == sha
+            and (SOURCE_HEAD == sha or SOURCE_SWEEP)
             and any(
                 status["context"] == ADMISSION and status.get("state") != "success"
                 for status in (source_statuses or {}).values()
             )
         ):
-            if not source_runs_current(pr) or source_contract(pr) is None:
+            if (
+                not source_runs_current(pr)
+                or source_contract(pr) is None
+                or not source_lease(pr)
+            ):
                 reconcile_source(pr, allow_success=False)
                 return False
             api(
@@ -603,11 +617,164 @@ def source_read(path: str) -> Any:  # ruff: ignore[any-type]
         The authoritative response; current refs and latest runs are never cached.
 
     """
+    if SOURCE_CALLBACK and "/actions/workflows/" in path and path not in SOURCE_CACHE:
+        inventory_path = f"repos/{REPOSITORY}/actions/workflows?per_page={PAGE_SIZE}"
+        if inventory_path not in SOURCE_CACHE:
+            inventory = api(inventory_path)
+            workflows = inventory.get("workflows")
+            count = inventory.get("total_count")
+            if (
+                not isinstance(count, int)
+                or not isinstance(workflows, list)
+                or not 0 <= count < PAGE_SIZE
+                or count != len(workflows)
+            ):
+                raise ValueError("Workflow inventory is incomplete")
+            SOURCE_CACHE[inventory_path] = inventory
+            for item in workflows:
+                filename = item.get("path", "").removeprefix(".github/workflows/")
+                if filename in WORKFLOWS:
+                    key = f"repos/{REPOSITORY}/actions/workflows/{filename}"
+                    if key in SOURCE_CACHE and SOURCE_CACHE[key] != item:
+                        raise ValueError("Workflow identity changed within callback")
+                    SOURCE_CACHE[key] = item
+        if path not in SOURCE_CACHE:
+            raise ValueError("Required source workflow is absent")
     if path not in SOURCE_CACHE:
         SOURCE_CACHE[path] = api(path)
     return SOURCE_CACHE[path]
 
 
+def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
+    """Read source PR identity, live ref and latest statuses in one fresh view."""
+    owner, name = REPOSITORY.split("/", 1)
+    fields = (
+        "number id isDraft baseRefName headRefName headRefOid "
+        "author { __typename login ... on Bot { databaseId } } "
+        "headRepository { nameWithOwner } "
+        "headRef { name repository { nameWithOwner } target { oid ... on Commit { "
+        "status { contexts { context state targetUrl description "
+        "creator { __typename login ... on Bot { databaseId } } } } } } }"
+    )
+    if number is None:
+        selection = (
+            "pullRequests(first:100,states:OPEN,baseRefName:$branch) { "
+            "pageInfo { hasNextPage } nodes { " + fields + " } }"
+        )
+        result = graphql(
+            "query($owner:String!,$name:String!,$branch:String!) { "
+            "repository(owner:$owner,name:$name) { " + selection + " } }",
+            owner=owner,
+            name=name,
+            branch=DEFAULT_BRANCH,
+        )["repository"]["pullRequests"]
+        if result["pageInfo"]["hasNextPage"]:
+            raise ValueError("Source snapshot exceeds bounded PR inventory")
+        nodes = result["nodes"]
+    else:
+        result = graphql(
+            "query($owner:String!,$name:String!,$number:Int!) { "
+            "repository(owner:$owner,name:$name) { pullRequest(number:$number) { "
+            "state " + fields + " } } }",
+            owner=owner,
+            name=name,
+            number=number,
+        )["repository"]["pullRequest"]
+        nodes = [result] if result and result["state"] == "OPEN" else []
+    pulls = []
+    for node in nodes:
+        actor = node.get("author") or {}
+        ref = node.get("headRef")
+        if not ref:
+            continue
+        target = ref["target"]
+        if (
+            ref["name"] != node["headRefName"]
+            or ref["repository"]["nameWithOwner"] != REPOSITORY
+            or target["oid"] != node["headRefOid"]
+        ):
+            raise ValueError("Inconsistent live source ref snapshot")
+
+        def bot(value: dict[str, Any]) -> dict[str, Any]:
+            return {
+                "type": value.get("__typename"),
+                "id": value.get("databaseId"),
+                "login": "github-actions[bot]"
+                if (
+                    value.get("__typename") == "Bot"
+                    and value.get("databaseId") == 41898282
+                    and value.get("login") == "github-actions"
+                )
+                else value.get("login"),
+            }
+
+        statuses = {}
+        for status in (target.get("status") or {}).get("contexts", []):
+            context = status["context"]
+            if context in statuses:
+                raise ValueError("Duplicate source status context")
+            statuses[context] = {
+                "context": context,
+                "state": status["state"].lower(),
+                "target_url": status["targetUrl"],
+                "description": status["description"],
+                "creator": bot(status.get("creator") or {}),
+            }
+        pr = {
+            "number": node["number"],
+            "node_id": node["id"],
+            "state": "open",
+            "draft": node["isDraft"],
+            "user": bot(actor),
+            "head": {
+                "sha": target["oid"],
+                "ref": node["headRefName"],
+                "repo": {
+                    "full_name": (node.get("headRepository") or {}).get("nameWithOwner")
+                },
+            },
+            "base": {"ref": node["baseRefName"], "repo": {"full_name": REPOSITORY}},
+        }
+        SOURCE_VIEW[target["oid"]] = statuses
+        pulls.append(pr)
+    return pulls
+
+
+def source_lease(pr: dict[str, Any]) -> bool:
+    """Recheck the exact live identity before an immutable-SHA write batch."""
+    if SOURCE_CALLBACK:
+        fresh = source_snapshot(pr["number"])
+        return (
+            len(fresh) == 1
+            and source_identity(fresh[0])
+            and fresh[0]["head"] == pr["head"]
+        )
+    fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}")
+    if not source_identity(fresh) or fresh["head"]["sha"] != pr["head"]["sha"]:
+        return False
+    ref = quote(pr["head"]["ref"], safe="/")
+    return (
+        api(f"repos/{REPOSITORY}/git/ref/heads/{ref}")["object"]["sha"]
+        == pr["head"]["sha"]
+    )
+
+
+def source_run_inventory(pr: dict[str, Any]) -> None:
+    """Refresh the bounded exact-head run inventory; never persist latest state."""
+    sha = pr["head"]["sha"]
+    result = api(f"repos/{REPOSITORY}/actions/runs?head_sha={sha}&per_page={PAGE_SIZE}")
+    runs = result.get("workflow_runs")
+    count = result.get("total_count")
+    if (
+        not isinstance(count, int)
+        or not isinstance(runs, list)
+        or not 0 <= count < PAGE_SIZE
+        or count != len(runs)
+    ):
+        raise ValueError("Source run inventory is incomplete")
+    SOURCE_RUN_LISTS[sha] = result
+
+
 def source_contract(pr: dict[str, Any]) -> str | None:
     """Prove the real source parent, owned paths and current workflow identities.
 
@@ -684,8 +851,12 @@ def source_run(
     path = f".github/workflows/{workflow}"
     if metadata.get("path") != path or metadata.get("state") != "active":
         return None
-    response = api(
-        f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}"
+    response = (
+        SOURCE_RUN_LISTS[sha]
+        if SOURCE_CALLBACK
+        else api(
+            f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}"
+        )
     )
     if (
         not isinstance(response.get("total_count"), int)
@@ -729,6 +900,8 @@ def source_runs_current(pr: dict[str, Any]) -> bool:
         Whether every collected run still has its exact inspected identity/state.
 
     """
+    if SOURCE_CALLBACK:
+        source_run_inventory(pr)
     for workflow, expected in SOURCE_EVIDENCE[pr["head"]["sha"]]["runs"].items():
         selected = source_run(pr, workflow)
         if selected is None or any(
@@ -753,6 +926,7 @@ def source_jobs(
     SOURCE_EVIDENCE[sha] = evidence
     jobs: list[dict[str, Any]] = []
     failures = {"failure", "timed_out", "cancelled"}
+    complete = True
     for workflow in WORKFLOWS:
         selected = source_run(pr, workflow)
         if selected is None:
@@ -760,7 +934,8 @@ def source_jobs(
         run, batch = selected
         evidence["runs"][workflow] = run
         if run.get("status") != "completed":
-            return None
+            complete = False
+            continue
         previous = (
             [statuses.get(context, {}) for context in SOURCE_CHECKS[workflow]]
             if statuses
@@ -829,9 +1004,10 @@ def source_jobs(
         if run.get("conclusion") != "success" or any(
             job.get("conclusion") != "success" for job in current_jobs
         ):
-            return None
+            complete = False
+            continue
         jobs.extend(current_jobs)
-    return jobs
+    return jobs if complete else None
 
 
 def reconcile_source(
@@ -852,7 +1028,7 @@ def reconcile_source(
         or not source_identity(pr)
     ):
         return False
-    sha, ref = pr["head"]["sha"], f"refs/heads/{pr['head']['ref']}"
+    sha = pr["head"]["sha"]
     statuses = read_statuses(sha) if statuses is None else statuses
     SOURCE_EVIDENCE[sha] = {"runs": {}, "failures": []}
     initial_event = SOURCE_HEAD == sha and SOURCE_ACTIVITY in {
@@ -864,6 +1040,40 @@ def reconcile_source(
         for status in statuses.values()
     ):
         return False
+    ready = True
+    if SOURCE_CALLBACK and not initial_event:
+        source_run_inventory(pr)
+        selected = [source_run(pr, workflow) for workflow in WORKFLOWS]
+        ready = all(item and item[0].get("status") == "completed" for item in selected)
+        failed_run = any(
+            item
+            and item[0].get("status") == "completed"
+            and item[0].get("conclusion") in {"failure", "timed_out", "cancelled"}
+            for item in selected
+        )
+        if ready and statuses.get(ADMISSION, {}).get("state") == "success":
+            # A write-free duplicate callback can reuse only exact current-run
+            # attestations. Changed attempts always take the full contract path.
+            attested = all(
+                status.get("state") == "success"
+                and status.get("creator", {}).get("id") == 41898282
+                and status.get("description")
+                == f"{SOURCE_REPORT} result; attempt {item[0]['run_attempt']}"
+                and re.fullmatch(
+                    re.escape(
+                        f"https://github.com/{REPOSITORY}/actions/runs/{item[0]['id']}/job/"
+                    )
+                    + r"[0-9]+",
+                    status.get("target_url", ""),
+                )
+                for workflow, item in zip(WORKFLOWS, selected, strict=True)
+                for context in SOURCE_CHECKS[workflow]
+                for status in [statuses.get(context, {})]
+            )
+            if attested and allow_success:
+                return True
+        if not ready and not failed_run:
+            initial_event = True
     jobs = (
         source_jobs(pr, statuses)
         if not initial_event and source_contract(pr) is not None
@@ -912,6 +1122,20 @@ def reconcile_source(
                 "target_url": status["target_url"],
                 "attempt": 0,
             }
+    changed = [
+        result
+        for result in results.values()
+        if any(
+            statuses.get(result["context"], {}).get(key) != value
+            for key, value in {
+                "state": result["state"],
+                "target_url": result["target_url"],
+                "description": f"{SOURCE_REPORT} result; attempt {result['attempt']}",
+            }.items()
+        )
+    ]
+    if changed and not source_lease(pr):
+        return False
     wrote = False
     for result in results.values():
         previous = statuses.get(result["context"], {})
@@ -922,12 +1146,6 @@ def reconcile_source(
             and previous.get("description") == description
         ):
             continue
-        fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}")
-        if not source_identity(fresh) or fresh["head"]["sha"] != sha:
-            return False
-        current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}")
-        if current["object"]["sha"] != sha:
-            return False
         wrote = True
         api(
             f"repos/{REPOSITORY}/statuses/{sha}",
@@ -939,22 +1157,34 @@ def reconcile_source(
                 "description": description,
             },
         )
-    if jobs and wrote and (not source_runs_current(pr) or source_contract(pr) is None):
+    if (
+        jobs
+        and wrote
+        and (
+            not source_runs_current(pr)
+            or source_contract(pr) is None
+            or not source_lease(pr)
+        )
+    ):
         reconcile_source(pr, allow_success=False)
         return False
     return bool(jobs)
 
 
-def main() -> None:
+def reconcile_main() -> None:
     """Reconcile open automation PRs, then start missing queue validation.
 
     Raises:
         subprocess.CalledProcessError: An API error other than a deleted ref occurred.
 
     """
-    global SOURCE_HEAD, SOURCE_ACTIVITY
+    global SOURCE_HEAD, SOURCE_ACTIVITY, SOURCE_CALLBACK, SOURCE_SWEEP
     SOURCE_CACHE.clear()
     SOURCE_EVIDENCE.clear()
+    SOURCE_VIEW.clear()
+    SOURCE_RUN_LISTS.clear()
+    SOURCE_CALLBACK = False
+    SOURCE_SWEEP = False
     run_id = os.environ.get("SOURCE_RUN_ID", "")
     try:
         notification = source_notification(
@@ -969,20 +1199,29 @@ def main() -> None:
         wait_for_source_run(run_id)
     else:
         SOURCE_HEAD, SOURCE_ACTIVITY = notification
+        SOURCE_CALLBACK = True
+        SOURCE_SWEEP = SOURCE_ACTIVITY == "completed"
 
     # Pagination is bounded by GitHub's repository PR limit, not a waiting loop.
     queue_expected = False
     page = 1
     while True:
-        pulls = api(
-            f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}"
-            f"&per_page={PAGE_SIZE}&page={page}"
-        )
+        if SOURCE_CALLBACK:
+            pulls = source_snapshot()
+        else:
+            pulls = api(
+                f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}"
+                f"&per_page={PAGE_SIZE}&page={page}"
+            )
         for pr in pulls:
+            if SOURCE_CALLBACK and not source_identity(pr):
+                continue
             queue_expected = admit(pr) or queue_expected
-        if len(pulls) < PAGE_SIZE:
+        if SOURCE_CALLBACK or len(pulls) < PAGE_SIZE:
             break
         page += 1
+    if SOURCE_CALLBACK and (not SOURCE_SWEEP or not queue_expected):
+        return
 
     # GITHUB_TOKEN enqueue events do not start merge_group workflows. Explicit
     # dispatch is supported. Discover queue refs and validate their exact SHA.
@@ -1025,6 +1264,16 @@ def main() -> None:
         advance_after_publication(published_passes)
 
 
+def main() -> None:
+    """Run one reconciliation with no mutable callback state leaking to another pass."""
+    global SOURCE_CALLBACK, SOURCE_SWEEP
+    try:
+        reconcile_main()
+    finally:
+        SOURCE_CALLBACK = False
+        SOURCE_SWEEP = False
+
+
 def advance_after_publication(published: set[str]) -> None:
     """Wait once for GitHub's asynchronous merge to expose its next live front."""
     for attempt in range(12):
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index 5923add..8e5c650 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -25,6 +25,14 @@
 spec.loader.exec_module(queue)
 
 
+def snapshot_fixture(fixture: Any) -> list[dict[str, Any]]:
+    """Expose the same immutable-head statuses as the GraphQL projection."""
+    queue.SOURCE_VIEW[fixture.sha] = {
+        status["context"]: status for status in fixture.statuses
+    }
+    return [fixture.pr]
+
+
 class ReconcileTests(unittest.TestCase):
     """Keep CI and queue failures observable without executing PR code."""
 
@@ -1008,6 +1016,19 @@ def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
         if method != "GET":
             self.writes.append((path, data))
             return None
+        if "/actions/workflows?" in path:
+            workflows = [
+                {
+                    "id": index,
+                    "path": f".github/workflows/{workflow}",
+                    "state": "active",
+                }
+                for index, workflow in enumerate(self.checks, 1)
+            ]
+            return {"total_count": len(workflows), "workflows": workflows}
+        if "/actions/runs?head_sha=" in path:
+            runs = [run for group in self.runs.values() for run in group]
+            return {"total_count": len(runs), "workflow_runs": runs}
         if "/git/commits/" in path:
             return {"parents": self.parents}
         if f"compare/{self.base}...{self.base}" in path:
@@ -1478,7 +1499,7 @@ def test_source_event_inspects_only_its_exact_head(self) -> None:
 
     def test_rerun_event_invalidates_even_an_admitted_source(self) -> None:
         counts = self._case("admitted_event")
-        self.assertEqual(counts, {"get": 35, "write": 17, "jobs": 0})
+        self.assertEqual(counts, {"get": 5, "write": 17, "jobs": 0})
 
     def test_initial_requested_event_with_no_attestation_stops_immediately(
         self,
@@ -1759,6 +1780,11 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:
             patch.object(queue, "SOURCE_HEAD", ""),
             patch.object(queue, "SOURCE_ACTIVITY", ""),
             patch.object(queue, "wait_for_source_run") as wait,
+            patch.object(
+                queue,
+                "source_snapshot",
+                side_effect=lambda *args: snapshot_fixture(self.fixture),
+            ),
             patch.object(queue, "graphql") as graphql,
         ):
             queue.main()
@@ -1934,3 +1960,121 @@ def front() -> dict[str, Any]:
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class SourceSnapshotTests(unittest.TestCase):
+    """Check live-ref projection and batch leases independently of native jobs."""
+
+    def node(self) -> dict[str, Any]:
+        return {
+            "number": 7,
+            "id": "PR7",
+            "isDraft": False,
+            "baseRefName": "main",
+            "headRefName": "automation/package-updates-example",
+            "headRefOid": "a" * 40,
+            "author": {
+                "__typename": "Bot",
+                "databaseId": 41898282,
+                "login": "github-actions",
+            },
+            "headRepository": {"nameWithOwner": queue.REPOSITORY},
+            "headRef": {
+                "name": "automation/package-updates-example",
+                "repository": {"nameWithOwner": queue.REPOSITORY},
+                "target": {"oid": "a" * 40, "status": {"contexts": []}},
+            },
+        }
+
+    def snapshot(
+        self, node: dict[str, Any], truncated: bool = False
+    ) -> list[dict[str, Any]]:
+        response = {
+            "repository": {
+                "pullRequests": {
+                    "pageInfo": {"hasNextPage": truncated},
+                    "nodes": [node],
+                }
+            }
+        }
+        with (
+            patch.object(queue, "graphql", return_value=response),
+            patch.object(queue, "SOURCE_VIEW", {}),
+        ):
+            return queue.source_snapshot()
+
+    def test_exact_bot_projection_preserves_source_identity(self) -> None:
+        self.assertTrue(queue.source_identity(self.snapshot(self.node())[0]))
+
+    def test_login_alone_does_not_prove_bot_identity(self) -> None:
+        node = self.node()
+        node["author"]["__typename"] = "User"
+        self.assertFalse(queue.source_identity(self.snapshot(node)[0]))
+
+    def test_inconsistent_live_head_and_cross_repository_ref_rejected(self) -> None:
+        for mutation in ("oid", "repository", "name"):
+            node = self.node()
+            if mutation == "oid":
+                node["headRef"]["target"]["oid"] = "b" * 40
+            elif mutation == "repository":
+                node["headRef"]["repository"]["nameWithOwner"] = "other/repository"
+            else:
+                node["headRef"]["name"] = "other-branch"
+            with self.subTest(mutation=mutation), self.assertRaises(ValueError):
+                self.snapshot(node)
+
+    def test_truncated_pull_inventory_is_not_silently_accepted(self) -> None:
+        with self.assertRaises(ValueError):
+            self.snapshot(self.node(), truncated=True)
+
+    def test_duplicate_status_context_rejected(self) -> None:
+        node = self.node()
+        value = {
+            "context": "DCO sign-off",
+            "state": "SUCCESS",
+            "targetUrl": "url",
+            "description": "real",
+            "creator": node["author"],
+        }
+        node["headRef"]["target"]["status"]["contexts"] = [value, value]
+        with self.assertRaises(ValueError):
+            self.snapshot(node)
+
+    def test_missing_deleted_ref_is_not_an_admission_lease(self) -> None:
+        node = self.node()
+        node["headRef"] = None
+        self.assertEqual(self.snapshot(node), [])
+
+    def test_aggregate_run_inventory_rejects_truncation_and_count_mismatch(
+        self,
+    ) -> None:
+        for response in (
+            {"total_count": 100, "workflow_runs": []},
+            {"total_count": 1, "workflow_runs": []},
+        ):
+            with (
+                patch.object(queue, "api", return_value=response),
+                self.assertRaises(ValueError),
+            ):
+                queue.source_run_inventory({"head": {"sha": "a" * 40}})
+
+    def test_partial_ci_does_not_hide_real_terminal_dco_failure(self) -> None:
+        fixture = PackageSourceTests()
+        fixture.setUp()
+        fixture.runs["ci.yml"][0].update(status="in_progress", conclusion=None)
+        run = fixture.runs["dco.yml"][0]
+        run["conclusion"] = "failure"
+        fixture.jobs[run["id"]][0]["conclusion"] = "failure"
+        with (
+            patch.object(queue, "api", side_effect=fixture._api),
+            patch.object(queue, "SOURCE_CHECKS", fixture.checks),
+            patch.object(queue, "WORKFLOWS", list(fixture.checks)),
+            patch.object(queue, "SOURCE_CALLBACK", False),
+            patch.object(queue, "SOURCE_CACHE", {}),
+            patch.object(queue, "SOURCE_EVIDENCE", {}),
+        ):
+            self.assertIsNone(queue.source_jobs(fixture.pr))
+            self.assertEqual(
+                queue.SOURCE_EVIDENCE[fixture.sha]["failures"][0]["html_url"],
+                fixture.jobs[run["id"]][0]["html_url"],
+            )

From 174094ea02831f00b4bc34590e2eac5fd71a5eab Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 21:58:25 -0700
Subject: [PATCH 2/9] Coalesce source readiness reads and bound callback
 content writes

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 .github/workflows/reconcile-merge-queue.yml |   2 +-
 actions/queue-completion/action.yml         |   6 +-
 scripts/reconcile-queue.py                  | 226 +++++++++++++++--
 tests/test_reconcile_queue.py               | 266 +++++++++++++++++++-
 tests/test_source_callback_action.py        |  38 ++-
 5 files changed, 504 insertions(+), 34 deletions(-)

diff --git a/.github/workflows/reconcile-merge-queue.yml b/.github/workflows/reconcile-merge-queue.yml
index 9cb90c5..ee16c79 100644
--- a/.github/workflows/reconcile-merge-queue.yml
+++ b/.github/workflows/reconcile-merge-queue.yml
@@ -23,7 +23,7 @@ jobs:
       github.event_name != 'workflow_run' ||
       contains(fromJSON('["pull_request", "merge_group"]'), github.event.workflow_run.event)
     runs-on: ubuntu-24.04
-    timeout-minutes: 5
+    timeout-minutes: 15
     permissions:
       checks: read # Read validation evidence for the queued commit.
       actions: write # Dispatch trusted queue validation.
diff --git a/actions/queue-completion/action.yml b/actions/queue-completion/action.yml
index c4efb65..2d62eba 100644
--- a/actions/queue-completion/action.yml
+++ b/actions/queue-completion/action.yml
@@ -15,15 +15,15 @@ runs:
         DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}
         SOURCE_ACTIVITY: ${{ inputs.source-activity }}
       run: |
-        args=(-f source_run_id="$GITHUB_RUN_ID")
+        args=(-f ref="$DEFAULT_BRANCH" -f "inputs[source_run_id]=$GITHUB_RUN_ID")
         case "$SOURCE_ACTIVITY" in
           '') ;;
           in_progress|completed)
-            args+=(-f source_run_attempt="$GITHUB_RUN_ATTEMPT" -f source_activity="$SOURCE_ACTIVITY") ;;
+            args+=(-f "inputs[source_run_attempt]=$GITHUB_RUN_ATTEMPT" -f "inputs[source_activity]=$SOURCE_ACTIVITY") ;;
           *) echo '::error::Invalid package-source notification phase'; exit 1 ;;
         esac
         for attempt in 1 2 3; do
-          if gh workflow run reconcile-merge-queue.yml --ref "$DEFAULT_BRANCH" "${args[@]}"; then
+          if gh api "repos/$GH_REPO/actions/workflows/reconcile-merge-queue.yml/dispatches" --method POST "${args[@]}"; then
             exit 0
           fi
           echo "::warning::Queue completion notification attempt $attempt failed"
diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index 20874c4..549f8dc 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -38,6 +38,8 @@
 SOURCE_SWEEP = False
 SOURCE_VIEW: dict[str, Any] = {}
 SOURCE_RUN_LISTS: dict[str, Any] = {}
+SOURCE_PREFLIGHT: dict[str, Any] = {}
+SOURCE_LAST_WRITE: float | None = None
 PAGE_SIZE = 100
 GH = shutil.which("gh") or "/usr/bin/gh"
 
@@ -49,6 +51,20 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:  # ruff: ig
         The decoded response, or None for an empty response.
 
     """
+    global SOURCE_LAST_WRITE
+    writes = method != "GET" and (
+        path != "graphql"
+        or (
+            isinstance(data, dict) and str(data.get("query", "")).startswith("mutation")
+        )
+    )
+    if SOURCE_CALLBACK and writes:
+        # A six-source sweep publishes 102 statuses. Pace all content writes,
+        # including enqueue/dispatch, leaving room for 60 notifier requests.
+        now = time.monotonic()
+        if SOURCE_LAST_WRITE is not None:
+            time.sleep(max(0.0, 4.0 - (now - SOURCE_LAST_WRITE)))
+        SOURCE_LAST_WRITE = time.monotonic()
     command = [GH, "api", path, "--method", method]
     if data is not None:
         command += ["--input", "-"]
@@ -92,7 +108,9 @@ def admit(pr: dict[str, Any]) -> bool:
         sha = pr["head"]["sha"]
         if SOURCE_HEAD and not SOURCE_SWEEP and SOURCE_HEAD != sha:
             return False
-        source_statuses = SOURCE_VIEW[sha] if SOURCE_CALLBACK else read_statuses(sha)
+        source_statuses = SOURCE_VIEW.get(sha)
+        if source_statuses is None:
+            source_statuses = read_statuses(sha)
         if (
             not SOURCE_HEAD
             and not SOURCE_SWEEP
@@ -545,15 +563,22 @@ def valid(producer: dict[str, Any]) -> bool:
             )
         if run["run_attempt"] != int(attempt):
             raise ValueError("Source callback producer attempt moved backwards")
-    try:
-        ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}")
-    except subprocess.CalledProcessError as error:
-        if "HTTP 404" not in (error.stderr or ""):
-            raise
-        raise ObsoleteSourceNotification(
-            "Verified source producer branch no longer exists"
-        ) from error
-    current_sha = ref.get("object", {}).get("sha")
+    if SOURCE_CALLBACK:
+        current_sha = source_ref(branch)
+        if current_sha is None:
+            raise ObsoleteSourceNotification(
+                "Verified source producer branch no longer exists"
+            )
+    else:
+        try:
+            ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}")
+        except subprocess.CalledProcessError as error:
+            if "HTTP 404" not in (error.stderr or ""):
+                raise
+            raise ObsoleteSourceNotification(
+                "Verified source producer branch no longer exists"
+            ) from error
+        current_sha = ref.get("object", {}).get("sha")
     if (
         not isinstance(current_sha, str)
         or re.fullmatch(r"[0-9a-f]{40}", current_sha) is None
@@ -585,6 +610,15 @@ def front_queue_entry() -> dict[str, Any] | None:
     return entries[0] if entries else None
 
 
+def source_bot(actor: dict[str, Any]) -> bool:
+    """Require the canonical GitHub Actions bot identity, never login alone."""
+    return (
+        actor.get("id") == 41898282
+        and actor.get("type") == "Bot"
+        and actor.get("login") == "github-actions[bot]"
+    )
+
+
 def source_identity(pr: dict[str, Any]) -> bool:
     """Limit opt-in source reconciliation to the package updater's bot PRs.
 
@@ -635,7 +669,10 @@ def source_read(path: str) -> Any:  # ruff: ignore[any-type]
                 filename = item.get("path", "").removeprefix(".github/workflows/")
                 if filename in WORKFLOWS:
                     key = f"repos/{REPOSITORY}/actions/workflows/{filename}"
-                    if key in SOURCE_CACHE and SOURCE_CACHE[key] != item:
+                    if key in SOURCE_CACHE and any(
+                        SOURCE_CACHE[key].get(field) != item.get(field)
+                        for field in ("id", "path", "state")
+                    ):
                         raise ValueError("Workflow identity changed within callback")
                     SOURCE_CACHE[key] = item
         if path not in SOURCE_CACHE:
@@ -684,6 +721,20 @@ def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
     pulls = []
     for node in nodes:
         actor = node.get("author") or {}
+        if not (
+            actor.get("__typename") == "Bot"
+            and actor.get("databaseId") == 41898282
+            and actor.get("login") == "github-actions"
+            and (node.get("headRepository") or {}).get("nameWithOwner") == REPOSITORY
+            and node.get("baseRefName") == DEFAULT_BRANCH
+            and re.fullmatch(
+                r"automation/package-updates-[a-z0-9][a-z0-9+._-]*",
+                node.get("headRefName", ""),
+            )
+            is not None
+            and not node.get("isDraft", True)
+        ):
+            continue
         ref = node.get("headRef")
         if not ref:
             continue
@@ -740,6 +791,24 @@ def bot(value: dict[str, Any]) -> dict[str, Any]:
     return pulls
 
 
+def source_ref(branch: str) -> str | None:
+    """Read the producer's exact current branch without a REST request."""
+    owner, name = REPOSITORY.split("/", 1)
+    ref = graphql(
+        "query($owner:String!,$name:String!,$branch:String!) { "
+        "repository(owner:$owner,name:$name) { ref(qualifiedName:$branch) { "
+        "name repository { nameWithOwner } target { oid } } } }",
+        owner=owner,
+        name=name,
+        branch=f"refs/heads/{branch}",
+    )["repository"]["ref"]
+    if ref is None:
+        return None
+    if ref["name"] != branch or ref["repository"]["nameWithOwner"] != REPOSITORY:
+        raise ValueError("Producer ref identity does not match")
+    return ref["target"]["oid"]
+
+
 def source_lease(pr: dict[str, Any]) -> bool:
     """Recheck the exact live identity before an immutable-SHA write batch."""
     if SOURCE_CALLBACK:
@@ -775,6 +844,79 @@ def source_run_inventory(pr: dict[str, Any]) -> None:
     SOURCE_RUN_LISTS[sha] = result
 
 
+def source_readiness(pulls: list[dict[str, Any]]) -> None:
+    """Batch fresh bounded check-suite projections for read-only readiness.
+
+    This projection never authorizes result publication or a new admission.
+    Missing, truncated or approval-required evidence takes the full REST path.
+    """
+    sources = [pr for pr in pulls if source_identity(pr)]
+    if not sources:
+        return
+    owner, name = REPOSITORY.split("/", 1)
+    fields = (
+        "oid checkSuites(first:100) { pageInfo { hasNextPage } nodes { "
+        "status conclusion repository { nameWithOwner } commit { oid } "
+        "branch { name repository { nameWithOwner } } workflowRun { "
+        "databaseId runAttempt event file { path } workflow { databaseId state } } } }"
+    )
+    selections = " ".join(
+        f'head{index}:object(expression:"{pr["head"]["sha"]}") {{ ... on Commit {{ {fields} }} }}'
+        for index, pr in enumerate(sources)
+    )
+    result = graphql(
+        "query($owner:String!,$name:String!) { repository(owner:$owner,name:$name) { "
+        + selections
+        + " } }",
+        owner=owner,
+        name=name,
+    )["repository"]
+    for index, pr in enumerate(sources):
+        sha = pr["head"]["sha"]
+        item = result.get(f"head{index}")
+        if not item or item.get("oid") != sha:
+            continue
+        suites = item.get("checkSuites") or {}
+        if suites.get("pageInfo", {}).get("hasNextPage") is not False:
+            continue
+        selected = {}
+        for suite in suites.get("nodes", []):
+            run = suite.get("workflowRun") or {}
+            workflow = run.get("file") or {}
+            filename = workflow.get("path", "").removeprefix(".github/workflows/")
+            if filename not in WORKFLOWS:
+                continue
+            metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{filename}")
+            branch = suite.get("branch") or {}
+            if not (
+                suite.get("repository", {}).get("nameWithOwner") == REPOSITORY
+                and suite.get("commit", {}).get("oid") == sha
+                and branch.get("name") == pr["head"]["ref"]
+                and branch.get("repository", {}).get("nameWithOwner") == REPOSITORY
+                and run.get("workflow", {}).get("databaseId") == metadata["id"]
+                and run["workflow"].get("state") == "ACTIVE"
+                and workflow.get("path") == f".github/workflows/{filename}"
+                and run.get("event") in {"pull_request", "workflow_dispatch"}
+                and isinstance(run.get("databaseId"), int)
+                and run["databaseId"] > 0
+                and isinstance(run.get("runAttempt"), int)
+                and run["runAttempt"] > 0
+            ):
+                continue
+            record = {
+                "id": run["databaseId"],
+                "run_attempt": run["runAttempt"],
+                "status": suite["status"].lower(),
+                "conclusion": (suite.get("conclusion") or "").lower(),
+            }
+            if filename not in selected or record["id"] > selected[filename]["id"]:
+                selected[filename] = record
+        if set(selected) == set(WORKFLOWS) and not any(
+            run["conclusion"] == "action_required" for run in selected.values()
+        ):
+            SOURCE_PREFLIGHT[sha] = selected
+
+
 def source_contract(pr: dict[str, Any]) -> str | None:
     """Prove the real source parent, owned paths and current workflow identities.
 
@@ -930,7 +1072,8 @@ def source_jobs(
     for workflow in WORKFLOWS:
         selected = source_run(pr, workflow)
         if selected is None:
-            return None
+            complete = False
+            continue
         run, batch = selected
         evidence["runs"][workflow] = run
         if run.get("status") != "completed":
@@ -949,7 +1092,7 @@ def source_jobs(
             and all(
                 status.get("state") == "success"
                 and status.get("description") == description
-                and status.get("creator", {}).get("id") == 41898282
+                and source_bot(status.get("creator", {}))
                 and re.fullmatch(
                     re.escape(prefix) + r"[0-9]+", status.get("target_url", "")
                 )
@@ -1040,6 +1183,38 @@ def reconcile_source(
         for status in statuses.values()
     ):
         return False
+    projected = SOURCE_PREFLIGHT.get(sha, {}) if SOURCE_CALLBACK else {}
+    if projected and not initial_event:
+        ready = all(run["status"] == "completed" for run in projected.values())
+        failed_run = any(
+            run["conclusion"] in {"failure", "timed_out", "cancelled"}
+            for run in projected.values()
+        )
+        if not ready and not failed_run:
+            initial_event = True
+        elif (
+            ready
+            and statuses.get(ADMISSION, {}).get("state") == "success"
+            and allow_success
+            and all(
+                run["conclusion"] == "success"
+                and status.get("state") == "success"
+                and source_bot(status.get("creator", {}))
+                and status.get("description")
+                == f"{SOURCE_REPORT} result; attempt {run['run_attempt']}"
+                and re.fullmatch(
+                    re.escape(
+                        f"https://github.com/{REPOSITORY}/actions/runs/{run['id']}/job/"
+                    )
+                    + r"[0-9]+",
+                    status.get("target_url", ""),
+                )
+                for workflow, run in projected.items()
+                for context in SOURCE_CHECKS[workflow]
+                for status in [statuses.get(context, {})]
+            )
+        ):
+            return True
     ready = True
     if SOURCE_CALLBACK and not initial_event:
         source_run_inventory(pr)
@@ -1055,8 +1230,9 @@ def reconcile_source(
             # A write-free duplicate callback can reuse only exact current-run
             # attestations. Changed attempts always take the full contract path.
             attested = all(
-                status.get("state") == "success"
-                and status.get("creator", {}).get("id") == 41898282
+                item[0].get("conclusion") == "success"
+                and status.get("state") == "success"
+                and source_bot(status.get("creator", {}))
                 and status.get("description")
                 == f"{SOURCE_REPORT} result; attempt {item[0]['run_attempt']}"
                 and re.fullmatch(
@@ -1178,14 +1354,22 @@ def reconcile_main() -> None:
         subprocess.CalledProcessError: An API error other than a deleted ref occurred.
 
     """
-    global SOURCE_HEAD, SOURCE_ACTIVITY, SOURCE_CALLBACK, SOURCE_SWEEP
+    global \
+        SOURCE_HEAD, \
+        SOURCE_ACTIVITY, \
+        SOURCE_CALLBACK, \
+        SOURCE_SWEEP, \
+        SOURCE_LAST_WRITE
     SOURCE_CACHE.clear()
     SOURCE_EVIDENCE.clear()
     SOURCE_VIEW.clear()
     SOURCE_RUN_LISTS.clear()
+    SOURCE_PREFLIGHT.clear()
+    SOURCE_LAST_WRITE = None
     SOURCE_CALLBACK = False
     SOURCE_SWEEP = False
     run_id = os.environ.get("SOURCE_RUN_ID", "")
+    SOURCE_CALLBACK = bool(run_id and SOURCE_CHECKS)
     try:
         notification = source_notification(
             run_id,
@@ -1196,7 +1380,12 @@ def reconcile_main() -> None:
         print(f"::warning::Ignoring obsolete verified source callback: {error}")
         return
     if notification is None:
+        SOURCE_CALLBACK = False
         wait_for_source_run(run_id)
+        if run_id and SOURCE_CHECKS:
+            # Queue callbacks also scan admitted source PRs. Share one fresh
+            # exact-head status view instead of six independent REST reads.
+            source_snapshot()
     else:
         SOURCE_HEAD, SOURCE_ACTIVITY = notification
         SOURCE_CALLBACK = True
@@ -1208,6 +1397,8 @@ def reconcile_main() -> None:
     while True:
         if SOURCE_CALLBACK:
             pulls = source_snapshot()
+            if SOURCE_SWEEP:
+                source_readiness(pulls)
         else:
             pulls = api(
                 f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}"
@@ -1272,6 +1463,9 @@ def main() -> None:
     finally:
         SOURCE_CALLBACK = False
         SOURCE_SWEEP = False
+        SOURCE_VIEW.clear()
+        SOURCE_PREFLIGHT.clear()
+        SOURCE_RUN_LISTS.clear()
 
 
 def advance_after_publication(published: set[str]) -> None:
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index 8e5c650..b7b1daa 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -1260,7 +1260,11 @@ def _attest(self) -> None:
                 "state": "success",
                 "target_url": job["html_url"],
                 "description": f"{queue.SOURCE_REPORT} result; attempt 1",
-                "creator": {"id": 41898282},
+                "creator": {
+                    "id": 41898282,
+                    "type": "Bot",
+                    "login": "github-actions[bot]",
+                },
             }
             for jobs in self.jobs.values()
             for job in jobs
@@ -1310,7 +1314,16 @@ def test_rerun_during_status_publication_is_bounded_and_invalidates_new_greens(
         def api(path: str, method: str = "GET", data: object = None) -> Any:
             result = self._api(path, method, data)
             if method == "POST":
-                self.statuses.append({**data, "creator": {"id": 41898282}})
+                self.statuses.append(
+                    {
+                        **data,
+                        "creator": {
+                            "id": 41898282,
+                            "type": "Bot",
+                            "login": "github-actions[bot]",
+                        },
+                    }
+                )
                 if len(self.writes) == 1:
                     self.runs["ci.yml"][0].update(
                         run_attempt=2, status="in_progress", conclusion=None
@@ -1657,6 +1670,11 @@ def api(
                     patch.object(queue, "SOURCE_EVIDENCE", {}),
                     patch.object(queue, "admit") as admit,
                     patch.object(queue, "dispatch_entry") as dispatch,
+                    patch.object(
+                        queue,
+                        "source_ref",
+                        return_value=None if reason == "deleted" else self.fixture.head,
+                    ),
                     patch.object(queue, "graphql") as graphql,
                     contextlib.redirect_stdout(output),
                 ):
@@ -1780,6 +1798,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:
             patch.object(queue, "SOURCE_HEAD", ""),
             patch.object(queue, "SOURCE_ACTIVITY", ""),
             patch.object(queue, "wait_for_source_run") as wait,
+            patch.object(queue, "source_ref", return_value=self.fixture.sha),
             patch.object(
                 queue,
                 "source_snapshot",
@@ -2009,7 +2028,7 @@ def test_exact_bot_projection_preserves_source_identity(self) -> None:
     def test_login_alone_does_not_prove_bot_identity(self) -> None:
         node = self.node()
         node["author"]["__typename"] = "User"
-        self.assertFalse(queue.source_identity(self.snapshot(node)[0]))
+        self.assertEqual(self.snapshot(node), [])
 
     def test_inconsistent_live_head_and_cross_repository_ref_rejected(self) -> None:
         for mutation in ("oid", "repository", "name"):
@@ -2040,6 +2059,40 @@ def test_duplicate_status_context_rejected(self) -> None:
         with self.assertRaises(ValueError):
             self.snapshot(node)
 
+    def test_human_fork_does_not_abort_two_owned_source_heads(self) -> None:
+        first = self.node()
+        second = self.node()
+        second["number"] = 8
+        second["id"] = "PR8"
+        second["headRefOid"] = "b" * 40
+        second["headRef"]["target"]["oid"] = "b" * 40
+        fork = self.node()
+        fork["author"] = {"__typename": "User", "login": "contributor"}
+        fork["headRepository"]["nameWithOwner"] = "contributor/fork"
+        fork["headRef"]["repository"]["nameWithOwner"] = "contributor/fork"
+        response = {
+            "repository": {
+                "pullRequests": {
+                    "pageInfo": {"hasNextPage": False},
+                    "nodes": [fork, first, second],
+                }
+            }
+        }
+        with (
+            patch.dict(os.environ, {"SOURCE_RUN_ID": "123"}),
+            patch.object(queue, "SOURCE_CHECKS", {"ci.yml": ["Build"]}),
+            patch.object(
+                queue, "source_notification", return_value=("a" * 40, "completed")
+            ),
+            patch.object(queue, "source_readiness"),
+            patch.object(queue, "graphql", return_value=response),
+            patch.object(queue, "admit", return_value=False) as admit,
+        ):
+            queue.main()
+        self.assertEqual(
+            [call.args[0]["number"] for call in admit.call_args_list], [7, 8]
+        )
+
     def test_missing_deleted_ref_is_not_an_admission_lease(self) -> None:
         node = self.node()
         node["headRef"] = None
@@ -2078,3 +2131,210 @@ def test_partial_ci_does_not_hide_real_terminal_dco_failure(self) -> None:
                 queue.SOURCE_EVIDENCE[fixture.sha]["failures"][0]["html_url"],
                 fixture.jobs[run["id"]][0]["html_url"],
             )
+
+
+class SourceReadinessTests(unittest.TestCase):
+    """Keep the cheap readiness path read-only and fail closed on partial evidence."""
+
+    def setUp(self) -> None:
+        self.fixture = PackageSourceTests()
+        self.fixture.setUp()
+        self.fixture._attest()
+        self.fixture.statuses.append({"context": queue.ADMISSION, "state": "success"})
+        self.suites = []
+        for runs in self.fixture.runs.values():
+            run = runs[0]
+            self.suites.append(
+                {
+                    "status": "COMPLETED",
+                    "conclusion": "SUCCESS",
+                    "repository": {"nameWithOwner": queue.REPOSITORY},
+                    "commit": {"oid": self.fixture.sha},
+                    "branch": {
+                        "name": self.fixture.branch,
+                        "repository": {"nameWithOwner": queue.REPOSITORY},
+                    },
+                    "workflowRun": {
+                        "databaseId": run["id"],
+                        "runAttempt": run["run_attempt"],
+                        "event": run["event"],
+                        "file": {"path": run["path"]},
+                        "workflow": {
+                            "databaseId": run["workflow_id"],
+                            "state": "ACTIVE",
+                        },
+                    },
+                }
+            )
+
+    def response(self, truncated: bool = False) -> dict[str, Any]:
+        return {
+            "repository": {
+                "head0": {
+                    "oid": self.fixture.sha,
+                    "checkSuites": {
+                        "pageInfo": {"hasNextPage": truncated},
+                        "nodes": self.suites,
+                    },
+                }
+            }
+        }
+
+    def test_readiness_rejects_truncated_missing_and_tampered_suite_identity(
+        self,
+    ) -> None:
+        for mode in (
+            "truncated",
+            "missing",
+            "branch",
+            "repository",
+            "workflow",
+            "head",
+            "approval",
+        ):
+            self.setUp()
+            if mode == "missing":
+                self.suites.pop()
+            elif mode == "branch":
+                self.suites[0]["branch"]["name"] = "wrong"
+            elif mode == "repository":
+                self.suites[0]["repository"]["nameWithOwner"] = "other/repo"
+            elif mode == "workflow":
+                self.suites[0]["workflowRun"]["workflow"]["databaseId"] = 99
+            elif mode == "head":
+                self.suites[0]["commit"]["oid"] = "b" * 40
+            elif mode == "approval":
+                self.suites[0]["conclusion"] = "ACTION_REQUIRED"
+            with (
+                self.subTest(mode=mode),
+                patch.object(
+                    queue, "graphql", return_value=self.response(mode == "truncated")
+                ),
+                patch.object(queue, "api", side_effect=self.fixture._api),
+                patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+                patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+                patch.object(queue, "SOURCE_CACHE", {}),
+                patch.object(queue, "SOURCE_PREFLIGHT", {}),
+            ):
+                queue.source_readiness([self.fixture.pr])
+                self.assertNotIn(self.fixture.sha, queue.SOURCE_PREFLIGHT)
+
+    def report(self, projected: bool = True) -> bool:
+        statuses = {value["context"]: value for value in self.fixture.statuses}
+        with (
+            patch.object(queue, "api", side_effect=self.fixture._api),
+            patch.object(queue, "graphql", return_value=self.response()),
+            patch.object(queue, "source_lease", return_value=True),
+            patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+            patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+            patch.object(queue, "SOURCE_CACHE", {}),
+            patch.object(queue, "SOURCE_PREFLIGHT", {}),
+            patch.object(queue, "SOURCE_EVIDENCE", {}),
+            patch.object(queue, "SOURCE_RUN_LISTS", {}),
+            patch.object(queue, "SOURCE_CALLBACK", True),
+            patch.object(queue, "SOURCE_HEAD", ""),
+        ):
+            if projected:
+                queue.source_readiness([self.fixture.pr])
+            return queue.reconcile_source(self.fixture.pr, statuses)
+
+    def test_failed_current_run_with_old_greens_cannot_take_either_duplicate_shortcut(
+        self,
+    ) -> None:
+        for projected in (False, True):
+            self.setUp()
+            run = self.fixture.runs["dco.yml"][0]
+            run["conclusion"] = "failure"
+            self.suites[-1]["conclusion"] = "FAILURE"
+            self.fixture.jobs[run["id"]][0]["conclusion"] = "failure"
+            with self.subTest(projected=projected):
+                self.assertFalse(self.report(projected))
+                failures = [
+                    data
+                    for _path, data in self.fixture.writes
+                    if data["state"] == "failure"
+                ]
+                self.assertEqual(len(failures), 1)
+                self.assertEqual(
+                    failures[0]["target_url"],
+                    self.fixture.jobs[run["id"]][0]["html_url"],
+                )
+
+    def test_missing_ci_producer_does_not_hide_later_failed_dco(self) -> None:
+        self.fixture.runs["ci.yml"] = []
+        self.suites.pop(0)
+        run = self.fixture.runs["dco.yml"][0]
+        run["conclusion"] = "failure"
+        self.fixture.jobs[run["id"]][0]["conclusion"] = "failure"
+        self.assertFalse(self.report())
+        self.assertEqual(
+            [
+                data["target_url"]
+                for _path, data in self.fixture.writes
+                if data["state"] == "failure"
+            ],
+            [self.fixture.jobs[run["id"]][0]["html_url"]],
+        )
+
+    def test_known_pending_ci_and_failed_dco_preserve_actual_failure_url(self) -> None:
+        run = self.fixture.runs["ci.yml"][0]
+        run.update(status="in_progress", conclusion=None)
+        self.suites[0].update(status="IN_PROGRESS", conclusion=None)
+        run = self.fixture.runs["dco.yml"][0]
+        run["conclusion"] = "failure"
+        self.suites[-1]["conclusion"] = "FAILURE"
+        self.fixture.jobs[run["id"]][0]["conclusion"] = "failure"
+        self.assertFalse(self.report())
+        self.assertEqual(
+            [
+                data["target_url"]
+                for _path, data in self.fixture.writes
+                if data["state"] == "failure"
+            ],
+            [self.fixture.jobs[run["id"]][0]["html_url"]],
+        )
+
+    def test_new_attempt_invalidates_old_managed_greens(self) -> None:
+        self.fixture.runs["ci.yml"][0].update(
+            run_attempt=2, status="in_progress", conclusion=None
+        )
+        self.suites[0].update(status="IN_PROGRESS", conclusion=None)
+        self.suites[0]["workflowRun"]["runAttempt"] = 2
+        self.assertFalse(self.report())
+        self.assertTrue(self.fixture.writes)
+        self.assertEqual(
+            {data["state"] for _path, data in self.fixture.writes}, {"pending"}
+        )
+
+    def test_duplicate_completed_callback_does_not_fetch_jobs_or_publish(self) -> None:
+        self.fixture.fail_jobs_api = True
+        self.assertTrue(self.report())
+        self.assertEqual(self.fixture.writes, [])
+
+
+class SourcePacingTests(unittest.TestCase):
+    """Read operations do not consume the paced source content-write slots."""
+
+    def test_actual_transport_paces_rest_and_enqueue_but_not_graphql_reads(
+        self,
+    ) -> None:
+        now = [0.0]
+        sleeps = []
+
+        def sleep(seconds: float) -> None:
+            sleeps.append(seconds)
+            now[0] += seconds
+
+        response = subprocess.CompletedProcess([], 0, "{}")
+        with (
+            patch.object(queue, "SOURCE_CALLBACK", True),
+            patch.object(queue, "SOURCE_LAST_WRITE", None),
+            patch.object(queue.time, "monotonic", side_effect=lambda: now[0]),
+            patch.object(queue.time, "sleep", side_effect=sleep),
+            patch.object(queue.subprocess, "run", return_value=response),
+        ):
+            queue.api("repos/example/statuses/sha", "POST", {})
+            queue.api("graphql", "POST", {"query": "query {viewer {login}}"})
+            queue.api("graphql", "POST", {"query": "mutation { enqueuePullRequest }"})
+            queue.api("repos/example/dispatches", "POST", {})
+        self.assertEqual(sleeps, [4.0, 4.0])
diff --git a/tests/test_source_callback_action.py b/tests/test_source_callback_action.py
index 78c2137..ffba471 100644
--- a/tests/test_source_callback_action.py
+++ b/tests/test_source_callback_action.py
@@ -14,7 +14,7 @@
 class SourceCallbackActionTests(unittest.TestCase):
     """Keep existing queue callers compatible and bind source attempt notifications."""
 
-    def invoke(self, activity: str) -> tuple[int, list[str], str]:
+    def invoke(self, activity: str, exit_code: int = 0) -> tuple[int, list[str], str]:
         action = yaml.load(
             (
                 Path(__file__).parents[1] / "actions/queue-completion/action.yml"
@@ -29,8 +29,12 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]:
                 f"#!{sys.executable}\nimport json, os, sys\n"
                 "from pathlib import Path\n"
                 "Path(os.environ['CALLBACK_RECEIPT']).write_text(json.dumps(sys.argv[1:]))\n"
+                "sys.exit(int(os.environ['CALLBACK_EXIT_CODE']))\n"
             )
             gh.chmod(0o755)
+            sleeper = root / "sleep"
+            sleeper.write_text("#!/bin/sh\nexit 0\n")
+            sleeper.chmod(0o755)
             result = subprocess.run(
                 ["bash", "-e", "-c", action["runs"]["steps"][0]["run"]],
                 env={
@@ -38,6 +42,8 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]:
                     "PATH": f"{root}:{os.environ['PATH']}",
                     "CALLBACK_RECEIPT": str(receipt),
                     "DEFAULT_BRANCH": "main",
+                    "GH_REPO": "nix-forge/example",
+                    "CALLBACK_EXIT_CODE": str(exit_code),
                     "GITHUB_RUN_ID": "123",
                     "GITHUB_RUN_ATTEMPT": "2",
                     "SOURCE_ACTIVITY": activity,
@@ -49,7 +55,7 @@ def invoke(self, activity: str) -> tuple[int, list[str], str]:
             return (
                 result.returncode,
                 json.loads(receipt.read_text()) if receipt.exists() else [],
-                result.stderr,
+                result.stdout + result.stderr,
             )
 
     def test_legacy_queue_payload_is_unchanged(self) -> None:
@@ -58,13 +64,14 @@ def test_legacy_queue_payload_is_unchanged(self) -> None:
         self.assertEqual(
             args,
             [
-                "workflow",
-                "run",
-                "reconcile-merge-queue.yml",
-                "--ref",
-                "main",
+                "api",
+                "repos/nix-forge/example/actions/workflows/reconcile-merge-queue.yml/dispatches",
+                "--method",
+                "POST",
                 "-f",
-                "source_run_id=123",
+                "ref=main",
+                "-f",
+                "inputs[source_run_id]=123",
             ],
         )
 
@@ -73,11 +80,20 @@ def test_source_start_and_completion_bind_actual_attempt(self) -> None:
             with self.subTest(activity=activity):
                 status, args, stderr = self.invoke(activity)
                 self.assertEqual(status, 0, stderr)
-                self.assertIn("source_run_id=123", args)
-                self.assertIn("source_run_attempt=2", args)
-                self.assertIn(f"source_activity={activity}", args)
+                self.assertIn("inputs[source_run_id]=123", args)
+                self.assertIn("inputs[source_run_attempt]=2", args)
+                self.assertIn(f"inputs[source_activity]={activity}", args)
 
     def test_invalid_phase_cannot_dispatch(self) -> None:
         status, args, stderr = self.invoke("completed --ref unreviewed")
         self.assertNotEqual(status, 0, stderr)
         self.assertEqual(args, [])
+
+    def test_failed_notification_keeps_bounded_retries_and_visible_failure(
+        self,
+    ) -> None:
+        status, args, output = self.invoke("completed", exit_code=1)
+        self.assertEqual(status, 1)
+        self.assertEqual(output.count("notification attempt"), 3)
+        self.assertEqual(args[0], "api")
+        self.assertIn("inputs[source_activity]=completed", args)

From 711d08dc64340ff093509d3843c64cf7634613bf Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:05:05 -0700
Subject: [PATCH 3/9] Use the test interpreter for the callback retry sleep
 fixture

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 tests/test_source_callback_action.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_source_callback_action.py b/tests/test_source_callback_action.py
index ffba471..ae0373d 100644
--- a/tests/test_source_callback_action.py
+++ b/tests/test_source_callback_action.py
@@ -33,7 +33,7 @@ def invoke(self, activity: str, exit_code: int = 0) -> tuple[int, list[str], str
             )
             gh.chmod(0o755)
             sleeper = root / "sleep"
-            sleeper.write_text("#!/bin/sh\nexit 0\n")
+            sleeper.write_text(f"#!{sys.executable}\n")
             sleeper.chmod(0o755)
             result = subprocess.run(
                 ["bash", "-e", "-c", action["runs"]["steps"][0]["run"]],

From 31d290e8a1a5356dcdf2798c0ed27aaa5a7fb934 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:21:40 -0700
Subject: [PATCH 4/9] Bound package queue reads while preserving every native
 result

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 scripts/reconcile-queue.py    | 190 +++++++++++++++----
 tests/test_reconcile_queue.py | 339 +++++++++++++++++++++++++++++++++-
 2 files changed, 493 insertions(+), 36 deletions(-)

diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index 549f8dc..0a54c10 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -40,6 +40,7 @@
 SOURCE_RUN_LISTS: dict[str, Any] = {}
 SOURCE_PREFLIGHT: dict[str, Any] = {}
 SOURCE_LAST_WRITE: float | None = None
+SOURCE_PRODUCER_COMPLETE = False
 PAGE_SIZE = 100
 GH = shutil.which("gh") or "/usr/bin/gh"
 
@@ -58,7 +59,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:  # ruff: ig
             isinstance(data, dict) and str(data.get("query", "")).startswith("mutation")
         )
     )
-    if SOURCE_CALLBACK and writes:
+    if (SOURCE_CALLBACK or SOURCE_CHECKS) and writes:
         # A six-source sweep publishes 102 statuses. Pace all content writes,
         # including enqueue/dispatch, leaving room for 60 notifier requests.
         now = time.monotonic()
@@ -78,7 +79,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:  # ruff: ig
     return json.loads(result.stdout) if result.stdout.strip() else None
 
 
-def graphql(query: str, **variables: str) -> dict[str, Any]:
+def graphql(query: str, **variables: str | int) -> dict[str, Any]:
     """Send a parameterized GraphQL operation and reject API errors.
 
     Returns:
@@ -481,11 +482,14 @@ def source_notification(
         ObsoleteSourceNotification: A verified attempt or branch has advanced or vanished.
 
     """
+    global SOURCE_PRODUCER_COMPLETE
+    SOURCE_PRODUCER_COMPLETE = False
     if not run_id or not SOURCE_CHECKS:
         return None
     if not run_id.isascii() or not run_id.isdecimal() or int(run_id) <= 0:
         raise ValueError("Source run ID must be a positive ASCII integer")
     run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}")
+    SOURCE_PRODUCER_COMPLETE = run.get("status") == "completed"
     branch = run.get("head_branch", "")
     if not isinstance(branch, str) or not branch.startswith(
         "automation/package-updates-"
@@ -550,7 +554,7 @@ def valid(producer: dict[str, Any]) -> bool:
                 "Older source callback does not match its immutable producer"
             )
         raise ObsoleteSourceNotification("Producer has a newer validated attempt")
-    if activity == "completed":
+    if activity == "completed" and not SOURCE_PRODUCER_COMPLETE:
         wait_for_source_run(run_id)
         run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}")
         if not valid(run) or run["head_sha"] != sha:
@@ -651,7 +655,7 @@ def source_read(path: str) -> Any:  # ruff: ignore[any-type]
         The authoritative response; current refs and latest runs are never cached.
 
     """
-    if SOURCE_CALLBACK and "/actions/workflows/" in path and path not in SOURCE_CACHE:
+    if SOURCE_CHECKS and "/actions/workflows/" in path and path not in SOURCE_CACHE:
         inventory_path = f"repos/{REPOSITORY}/actions/workflows?per_page={PAGE_SIZE}"
         if inventory_path not in SOURCE_CACHE:
             inventory = api(inventory_path)
@@ -682,16 +686,18 @@ def source_read(path: str) -> Any:  # ruff: ignore[any-type]
     return SOURCE_CACHE[path]
 
 
-def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
+def source_snapshot(
+    number: int | None = None, *, sources_only: bool = True
+) -> list[dict[str, Any]]:
     """Read source PR identity, live ref and latest statuses in one fresh view."""
     owner, name = REPOSITORY.split("/", 1)
     fields = (
         "number id isDraft baseRefName headRefName headRefOid "
-        "author { __typename login ... on Bot { databaseId } } "
+        "author { __typename login ... on Bot { databaseId } ... on User { databaseId } } "
         "headRepository { nameWithOwner } "
         "headRef { name repository { nameWithOwner } target { oid ... on Commit { "
         "status { contexts { context state targetUrl description "
-        "creator { __typename login ... on Bot { databaseId } } } } } } }"
+        "creator { __typename login ... on Bot { databaseId } ... on User { databaseId } } } } } } }"
     )
     if number is None:
         selection = (
@@ -721,7 +727,7 @@ def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
     pulls = []
     for node in nodes:
         actor = node.get("author") or {}
-        if not (
+        eligible_source = (
             actor.get("__typename") == "Bot"
             and actor.get("databaseId") == 41898282
             and actor.get("login") == "github-actions"
@@ -733,13 +739,17 @@ def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
             )
             is not None
             and not node.get("isDraft", True)
-        ):
+        )
+        if sources_only and not eligible_source:
             continue
         ref = node.get("headRef")
         if not ref:
             continue
         target = ref["target"]
         if (
+            eligible_source
+            or (node.get("headRepository") or {}).get("nameWithOwner") == REPOSITORY
+        ) and (
             ref["name"] != node["headRefName"]
             or ref["repository"]["nameWithOwner"] != REPOSITORY
             or target["oid"] != node["headRefOid"]
@@ -747,16 +757,16 @@ def source_snapshot(number: int | None = None) -> list[dict[str, Any]]:
             raise ValueError("Inconsistent live source ref snapshot")
 
         def bot(value: dict[str, Any]) -> dict[str, Any]:
+            login = value.get("login")
+            if value.get("__typename") == "Bot" and (
+                (value.get("databaseId"), login)
+                in {(41898282, "github-actions"), (49699333, "dependabot")}
+            ):
+                login += "[bot]"
             return {
                 "type": value.get("__typename"),
                 "id": value.get("databaseId"),
-                "login": "github-actions[bot]"
-                if (
-                    value.get("__typename") == "Bot"
-                    and value.get("databaseId") == 41898282
-                    and value.get("login") == "github-actions"
-                )
-                else value.get("login"),
+                "login": login,
             }
 
         statuses = {}
@@ -778,7 +788,7 @@ def bot(value: dict[str, Any]) -> dict[str, Any]:
             "draft": node["isDraft"],
             "user": bot(actor),
             "head": {
-                "sha": target["oid"],
+                "sha": node["headRefOid"],
                 "ref": node["headRefName"],
                 "repo": {
                     "full_name": (node.get("headRepository") or {}).get("nameWithOwner")
@@ -932,9 +942,18 @@ def source_contract(pr: dict[str, Any]) -> str | None:
     base = parents[0]["sha"]
     if re.fullmatch(r"[0-9a-f]{40}", base) is None:
         return None
-    protected = api(
-        f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}"
-    )["object"]["sha"]
+    protected = (
+        source_ref(DEFAULT_BRANCH)
+        if SOURCE_CALLBACK
+        else api(f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}")[
+            "object"
+        ]["sha"]
+    )
+    if (
+        not isinstance(protected, str)
+        or re.fullmatch(r"[0-9a-f]{40}", protected) is None
+    ):
+        return None
     comparison = source_read(f"repos/{REPOSITORY}/compare/{base}...{protected}")
     if comparison.get("status") not in {"ahead", "identical"}:
         return None
@@ -1301,7 +1320,8 @@ def reconcile_source(
     changed = [
         result
         for result in results.values()
-        if any(
+        if not source_bot(statuses.get(result["context"], {}).get("creator", {}))
+        or any(
             statuses.get(result["context"], {}).get(key) != value
             for key, value in {
                 "state": result["state"],
@@ -1317,7 +1337,8 @@ def reconcile_source(
         previous = statuses.get(result["context"], {})
         description = f"{SOURCE_REPORT} result; attempt {result['attempt']}"
         if (
-            previous.get("state") == result["state"]
+            source_bot(previous.get("creator", {}))
+            and previous.get("state") == result["state"]
             and previous.get("target_url") == result["target_url"]
             and previous.get("description") == description
         ):
@@ -1365,7 +1386,7 @@ def reconcile_main() -> None:
     SOURCE_VIEW.clear()
     SOURCE_RUN_LISTS.clear()
     SOURCE_PREFLIGHT.clear()
-    SOURCE_LAST_WRITE = None
+    SOURCE_LAST_WRITE = time.monotonic()
     SOURCE_CALLBACK = False
     SOURCE_SWEEP = False
     run_id = os.environ.get("SOURCE_RUN_ID", "")
@@ -1379,13 +1400,15 @@ def reconcile_main() -> None:
     except ObsoleteSourceNotification as error:
         print(f"::warning::Ignoring obsolete verified source callback: {error}")
         return
+    callback_pulls = None
     if notification is None:
         SOURCE_CALLBACK = False
-        wait_for_source_run(run_id)
+        if not SOURCE_PRODUCER_COMPLETE:
+            wait_for_source_run(run_id)
         if run_id and SOURCE_CHECKS:
             # Queue callbacks also scan admitted source PRs. Share one fresh
             # exact-head status view instead of six independent REST reads.
-            source_snapshot()
+            callback_pulls = source_snapshot(sources_only=False)
     else:
         SOURCE_HEAD, SOURCE_ACTIVITY = notification
         SOURCE_CALLBACK = True
@@ -1399,6 +1422,8 @@ def reconcile_main() -> None:
             pulls = source_snapshot()
             if SOURCE_SWEEP:
                 source_readiness(pulls)
+        elif callback_pulls is not None:
+            pulls = callback_pulls
         else:
             pulls = api(
                 f"repos/{REPOSITORY}/pulls?state=open&base={quote(DEFAULT_BRANCH, safe='')}"
@@ -1408,7 +1433,7 @@ def reconcile_main() -> None:
             if SOURCE_CALLBACK and not source_identity(pr):
                 continue
             queue_expected = admit(pr) or queue_expected
-        if SOURCE_CALLBACK or len(pulls) < PAGE_SIZE:
+        if SOURCE_CALLBACK or callback_pulls is not None or len(pulls) < PAGE_SIZE:
             break
         page += 1
     if SOURCE_CALLBACK and (not SOURCE_SWEEP or not queue_expected):
@@ -1511,10 +1536,18 @@ def dispatch_entry(entry: dict[str, Any]) -> bool:
     if not ref.startswith(QUEUE_PREFIX):
         return False
     base = api(f"repos/{REPOSITORY}/git/commits/{sha}")["parents"][0]["sha"]
+    inventory = queue_run_inventory(ref, sha) if SOURCE_CHECKS else None
     validation_runs: list[dict[str, Any]] = []
     for workflow in WORKFLOWS:
-        runs = api(
-            f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page=1"
+        runs = (
+            {
+                "total_count": int(workflow in inventory),
+                "workflow_runs": [inventory[workflow]] if workflow in inventory else [],
+            }
+            if inventory is not None
+            else api(
+                f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page=1"
+            )
         )
         if runs["total_count"]:
             latest_run = runs["workflow_runs"][0]
@@ -1540,6 +1573,38 @@ def dispatch_entry(entry: dict[str, Any]) -> bool:
     return publish_dispatch_results(ref, sha, validation_runs)
 
 
+def queue_run_inventory(ref: str, sha: str) -> dict[str, Any]:
+    """Read complete exact-head queue producers without five repeated lists."""
+    source_run_inventory({"head": {"sha": sha}})
+    result = {}
+    for workflow in WORKFLOWS:
+        metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{workflow}")
+        if (
+            metadata.get("path") != f".github/workflows/{workflow}"
+            or metadata.get("state") != "active"
+        ):
+            raise ValueError("Queue workflow identity is not active")
+        for run in SOURCE_RUN_LISTS[sha]["workflow_runs"]:
+            if run.get("workflow_id") != metadata["id"]:
+                continue
+            if not (
+                run.get("path") == metadata["path"]
+                and run.get("head_sha") == sha
+                and run.get("head_branch") == ref.removeprefix("refs/heads/")
+                and run.get("repository", {}).get("full_name") == REPOSITORY
+                and run.get("head_repository", {}).get("full_name") == REPOSITORY
+                and run.get("event") in {"merge_group", "workflow_dispatch"}
+                and type(run.get("id")) is int
+                and run["id"] > 0
+                and type(run.get("run_attempt")) is int
+                and run["run_attempt"] > 0
+            ):
+                raise ValueError("Queue producer identity does not match its exact ref")
+            result[workflow] = {**run, "queue_workflow": workflow}
+            break
+    return result
+
+
 def read_statuses(sha: str) -> dict[str, Any]:
     """Read the latest commit status for every context.
 
@@ -1586,6 +1651,7 @@ def latest_dispatch_jobs(
     """
     jobs: list[dict[str, Any]] = []
     for run in runs:
+        run_jobs: list[dict[str, Any]] = []
         page = 1
         while True:
             batch = api(
@@ -1593,9 +1659,18 @@ def latest_dispatch_jobs(
                 f"?filter=latest&per_page={PAGE_SIZE}&page={page}"
             )["jobs"]
             jobs.extend(batch)
+            run_jobs.extend(batch)
             if len(batch) < PAGE_SIZE:
                 break
             page += 1
+        if SOURCE_CHECKS:
+            if any(
+                job.get("run_id") != run["id"]
+                or job.get("run_attempt") != run["run_attempt"]
+                for job in run_jobs
+            ):
+                return None
+            continue
         # A separate dispatch or rerun can replace the evidence being inspected.
         fresh = api(
             f"repos/{REPOSITORY}/actions/workflows/{run['queue_workflow']}/runs"
@@ -1608,13 +1683,35 @@ def latest_dispatch_jobs(
         }
         if any(current.get(key) != value for key, value in expected.items()):
             return None
+    if SOURCE_CHECKS and not queue_runs_current(sha, runs):
+        return None
     if any(job["head_sha"] != sha or job["status"] != "completed" for job in jobs):
         return None
     return [job for job in jobs if job["name"] != "Queue completion callback"]
 
 
+def queue_runs_current(sha: str, runs: list[dict[str, Any]]) -> bool:
+    """Recheck every collected producer after collection or publication."""
+    if not runs:
+        return False
+    current = queue_run_inventory(f"refs/heads/{runs[0]['head_branch']}", sha)
+    return all(
+        expected["queue_workflow"] in current
+        and all(
+            current[expected["queue_workflow"]].get(key) == expected.get(key)
+            for key in ("id", "run_attempt", "head_sha", "status", "conclusion")
+        )
+        for expected in runs
+    )
+
+
 def invalidate_results(
-    ref: str, sha: str, latest: dict[str, Any], reported: set[str] | None = None
+    ref: str,
+    sha: str,
+    latest: dict[str, Any],
+    reported: set[str] | None = None,
+    *,
+    leased: bool = False,
 ) -> None:
     """Invalidate old adapter results absent from the latest verified jobs."""
     for status in latest.values():
@@ -1630,6 +1727,7 @@ def invalidate_results(
                     "state": "pending",
                     "target_url": status["target_url"],
                 },
+                leased=leased,
             )
 
 
@@ -1661,7 +1759,13 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) ->
     # success still requires the complete configured workflow set to pass.
     jobs = latest_dispatch_jobs(sha, evidence) if evidence else None
     if not jobs:
-        invalidate_results(ref, sha, latest)
+        leased = bool(SOURCE_CHECKS) and any(
+            status.get("description", "").startswith("Mirrored queue job")
+            for status in latest.values()
+        )
+        if leased and not queue_ref_current(ref, sha):
+            return False
+        invalidate_results(ref, sha, latest, leased=leased)
         return False
     results = {
         job["name"]: "success" if job["conclusion"] == "success" else "failure"
@@ -1669,8 +1773,12 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) ->
         if job["conclusion"] in failures
         or (all_passed and job["conclusion"] == "success")
     }
-    invalidate_results(ref, sha, latest, set(results))
+    leased = bool(SOURCE_CHECKS)
+    if leased and not queue_ref_current(ref, sha):
+        return False
+    invalidate_results(ref, sha, latest, set(results), leased=leased)
     published_success = False
+    published_any = False
     for job in jobs:
         # Skipped and neutral jobs provide no passing or failing evidence.
         if job["name"] in results:
@@ -1683,8 +1791,16 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) ->
                     "state": results[job["name"]],
                     "target_url": job["html_url"],
                 },
+                leased=leased,
             )
             published_success |= changed and results[job["name"]] == "success"
+            published_any |= changed
+    if leased and published_any:
+        if not queue_ref_current(ref, sha):
+            return False
+        if not queue_runs_current(sha, runs):
+            invalidate_results(ref, sha, latest, leased=True)
+            return False
     source_required = {name for names in SOURCE_CHECKS.values() for name in names}
     return (
         published_success
@@ -1699,6 +1815,8 @@ def publish_status(
     sha: str,
     latest: dict[str, Any],
     result: dict[str, str],
+    *,
+    leased: bool = False,
 ) -> bool:
     """Publish one actual job result once, while its queue ref still matches.
 
@@ -1710,8 +1828,7 @@ def publish_status(
     previous = latest.get(name, {})
     if previous.get("state") == state and previous.get("target_url") == target_url:
         return False
-    current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}")
-    if current["object"]["sha"] != sha:
+    if not leased and not queue_ref_current(ref, sha):
         return False
     api(
         f"repos/{REPOSITORY}/statuses/{sha}",
@@ -1726,8 +1843,15 @@ def publish_status(
         },
     )
     print(f"Reported {name}: {state} for {sha}")
+    latest[name] = {**result, "description": "Mirrored queue job result"}
     return True
 
 
+def queue_ref_current(ref: str, sha: str) -> bool:
+    """Lease only this exact immutable queue commit, never a replacement head."""
+    current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}")
+    return current["object"]["sha"] == sha
+
+
 if __name__ == "__main__":
     main()
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index b7b1daa..a1150bf 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -1343,6 +1343,61 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:
             ["success"] * 3 + ["pending"] * 3,
         )
 
+    def test_noncanonical_creator_is_replaced_once_after_actual_jobs_prove_success(
+        self,
+    ) -> None:
+        for creator in (
+            {"id": 9, "type": "User", "login": "developer"},
+            {"id": 9, "type": "Bot", "login": "other[bot]"},
+            {"id": 41898282, "type": "User", "login": "github-actions[bot]"},
+            {"id": 41898282, "type": "Bot", "login": "other[bot]"},
+        ):
+            self.setUp()
+            self._attest()
+            self.statuses[0]["creator"] = creator
+            calls = []
+
+            def api(
+                path: str,
+                method: str = "GET",
+                data: object = None,
+                *,
+                records: list[tuple[str, str]] = calls,
+            ) -> Any:
+                records.append((method, path))
+                result = self._api(path, method, data)
+                if method == "POST":
+                    previous = next(
+                        value
+                        for value in self.statuses
+                        if value["context"] == data["context"]
+                    )
+                    previous.update(data)
+                    previous["creator"] = {
+                        "id": 41898282,
+                        "type": "Bot",
+                        "login": "github-actions[bot]",
+                    }
+                return result
+
+            with (
+                self.subTest(creator=creator),
+                patch.object(queue, "api", side_effect=api),
+                patch.object(queue, "SOURCE_CACHE", {}),
+                patch.object(queue, "SOURCE_EVIDENCE", {}),
+                patch.object(queue, "SOURCE_CHECKS", self.checks),
+                patch.object(queue, "WORKFLOWS", list(self.checks)),
+                patch.object(queue, "SOURCE_CALLBACK", False),
+            ):
+                self.assertTrue(queue.reconcile_source(self.pr))
+                self.assertEqual(len(self.writes), 1)
+                self.assertTrue(any("/jobs?" in path for _method, path in calls))
+                self.writes.clear()
+                calls.clear()
+                self.assertTrue(queue.reconcile_source(self.pr))
+                self.assertEqual(self.writes, [])
+                self.assertFalse(any("/jobs?" in path for _method, path in calls))
+
     def test_untrusted_cached_status_cannot_replace_actual_job_evidence(self) -> None:
         self._attest()
         self.statuses[-1]["creator"]["id"] = 9
@@ -1563,7 +1618,35 @@ def test_delayed_start_revalidates_completed_attempt(self) -> None:
         self.assertEqual(result, (self.fixture.sha, "completed"))
         wait.assert_not_called()
 
+    def test_already_completed_producer_reuses_actual_metadata_without_wait(
+        self,
+    ) -> None:
+        reads = []
+
+        def api(path: str, method: str = "GET", data: object = None) -> Any:
+            reads.append(path)
+            return self.fixture._api(path, method, data)
+
+        with (
+            patch.object(queue, "api", side_effect=api),
+            patch.object(queue, "wait_for_source_run") as wait,
+            patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+            patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+            patch.object(queue, "SOURCE_CACHE", {}),
+            patch.object(queue, "SOURCE_CALLBACK", False),
+        ):
+            self.assertEqual(
+                queue.source_notification(self.run_id, "1", "completed"),
+                (self.fixture.sha, "completed"),
+            )
+        wait.assert_not_called()
+        self.assertEqual(
+            sum(path.endswith("/actions/runs/" + self.run_id) for path in reads), 1
+        )
+
     def test_new_attempt_during_completion_cannot_reuse_old_identity(self) -> None:
+        self.run.update(status="in_progress", conclusion=None)
+
         def rerun(_run_id: str) -> None:
             self.run.update(status="in_progress", conclusion=None, run_attempt=2)
 
@@ -1613,6 +1696,8 @@ def test_future_attempt_is_a_real_metadata_error(self) -> None:
             self.notification("in_progress", "2")
 
     def test_head_mutation_during_wait_is_a_real_error(self) -> None:
+        self.run.update(status="in_progress", conclusion=None)
+
         def mutate(_run_id: str) -> None:
             self.run["head_sha"] = "e" * 40
 
@@ -1924,6 +2009,9 @@ def test_actual_sixteen_status_publication_reveals_next_front_later(self) -> Non
         fixture.statuses = []
         old = {"ref": queue.QUEUE_PREFIX + "old", "object": {"sha": fixture.sha}}
         new = {"ref": queue.QUEUE_PREFIX + "next", "object": {"sha": "d" * 40}}
+        for runs in fixture.runs.values():
+            for run in runs:
+                run["head_branch"] = old["ref"].removeprefix("refs/heads/")
         statuses = []
         dispatches = []
         ticks = 0
@@ -1947,7 +2035,7 @@ def api(path: str, method: str = "GET", data: object = None) -> Any:
                 }
             if "/git/commits/" in path:
                 return {"parents": [{"sha": fixture.base}]}
-            if "/actions/workflows/" in path and "head_sha=" + "d" * 40 in path:
+            if "/actions/" in path and "head_sha=" + "d" * 40 in path:
                 return {"total_count": 0, "workflow_runs": []}
             return fixture._api(path, method, data)
 
@@ -2006,7 +2094,11 @@ def node(self) -> dict[str, Any]:
         }
 
     def snapshot(
-        self, node: dict[str, Any], truncated: bool = False
+        self,
+        node: dict[str, Any],
+        truncated: bool = False,
+        *,
+        sources_only: bool = True,
     ) -> list[dict[str, Any]]:
         response = {
             "repository": {
@@ -2020,7 +2112,78 @@ def snapshot(
             patch.object(queue, "graphql", return_value=response),
             patch.object(queue, "SOURCE_VIEW", {}),
         ):
-            return queue.source_snapshot()
+            return queue.source_snapshot(sources_only=sources_only)
+
+    def test_generic_snapshot_preserves_rest_bot_human_fork_and_draft_fields(
+        self,
+    ) -> None:
+        for kind in ("dependabot", "actions", "human", "fork", "draft"):
+            node = self.node()
+            if kind in {"dependabot", "draft"}:
+                node["author"].update(databaseId=49699333, login="dependabot")
+                node["headRefName"] = "dependabot/example"
+                node["isDraft"] = kind == "draft"
+                expected_login = "dependabot[bot]"
+            elif kind == "actions":
+                node["headRefName"] = "automation/update"
+                expected_login = "github-actions[bot]"
+            else:
+                node["author"] = {
+                    "__typename": "User",
+                    "databaseId": 1,
+                    "login": "developer",
+                }
+                node["headRefName"] = "feature/change"
+                expected_login = "developer"
+                if kind == "fork":
+                    node["headRepository"]["nameWithOwner"] = "developer/fork"
+                    node["headRef"]["repository"]["nameWithOwner"] = "developer/fork"
+            node["headRef"]["name"] = node["headRefName"]
+            with self.subTest(kind=kind):
+                pr = self.snapshot(node, sources_only=False)[0]
+                self.assertEqual(pr["user"]["login"], expected_login)
+                self.assertEqual(pr["user"]["id"], node["author"]["databaseId"])
+                self.assertEqual(pr["draft"], node["isDraft"])
+                self.assertEqual(
+                    pr["head"]["repo"]["full_name"],
+                    node["headRepository"]["nameWithOwner"],
+                )
+                self.assertEqual(pr["head"]["sha"], node["headRefOid"])
+                self.assertEqual(pr["head"]["ref"], node["headRefName"])
+                self.assertEqual(pr["base"]["ref"], node["baseRefName"])
+
+    def test_generic_dependabot_snapshot_still_admits_with_real_required_policy(
+        self,
+    ) -> None:
+        node = self.node()
+        node["author"].update(databaseId=49699333, login="dependabot")
+        node["headRefName"] = "dependabot/example"
+        node["headRef"]["name"] = node["headRefName"]
+        pr = self.snapshot(node, sources_only=False)[0]
+
+        def api(path: str, method: str = "GET", data: object = None) -> Any:
+            if "/files?" in path:
+                return []
+            if path.endswith("/status"):
+                return {"statuses": []}
+            if method == "POST":
+                return None
+            raise AssertionError(path)
+
+        with (
+            patch.object(queue, "SOURCE_CHECKS", {}),
+            patch.object(queue, "api", side_effect=api),
+            patch.object(
+                queue, "graphql", return_value={"node": {"mergeQueueEntry": None}}
+            ) as gql,
+            patch.object(
+                queue.subprocess,
+                "run",
+                return_value=subprocess.CompletedProcess([], 0, '[{"bucket":"pass"}]'),
+            ),
+        ):
+            self.assertTrue(queue.admit(pr))
+        self.assertIn("enqueuePullRequest", gql.call_args.args[0])
 
     def test_exact_bot_projection_preserves_source_identity(self) -> None:
         self.assertTrue(queue.source_identity(self.snapshot(self.node())[0]))
@@ -2225,6 +2388,7 @@ def report(self, projected: bool = True) -> bool:
             patch.object(queue, "api", side_effect=self.fixture._api),
             patch.object(queue, "graphql", return_value=self.response()),
             patch.object(queue, "source_lease", return_value=True),
+            patch.object(queue, "source_ref", return_value=self.fixture.base),
             patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
             patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
             patch.object(queue, "SOURCE_CACHE", {}),
@@ -2338,3 +2502,172 @@ def sleep(seconds: float) -> None:
             queue.api("graphql", "POST", {"query": "mutation { enqueuePullRequest }"})
             queue.api("repos/example/dispatches", "POST", {})
         self.assertEqual(sleeps, [4.0, 4.0])
+
+
+class SourceQueueBatchTests(unittest.TestCase):
+    """Keep all nineteen actual queue results bound to fresh producers and refs."""
+
+    def setUp(self) -> None:
+        self.fixture = SourceBudgetTests()._fixtures()[0]
+        self.fixture.branch = "gh-readonly-queue/main/pr-7-base"
+        self.fixture.statuses = []
+        self.ref = "refs/heads/" + self.fixture.branch
+        for runs in self.fixture.runs.values():
+            runs[0]["head_branch"] = self.fixture.branch
+        ci = self.fixture.runs["ci.yml"][0]
+        for position, name in enumerate(
+            (
+                "Discover CI inputs",
+                "Flake lock (flake.lock) / check",
+                "Flake lock (flake/dev/flake.lock) / check",
+            ),
+            50,
+        ):
+            self.fixture.jobs[ci["id"]].append(
+                {
+                    "id": ci["id"] * 100 + position,
+                    "run_id": ci["id"],
+                    "run_attempt": 1,
+                    "name": name,
+                    "head_sha": self.fixture.sha,
+                    "status": "completed",
+                    "conclusion": "success",
+                    "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{ci['id']}/job/{ci['id'] * 100 + position}",
+                }
+            )
+        self.runs = [
+            {**runs[0], "queue_workflow": workflow}
+            for workflow, runs in self.fixture.runs.items()
+        ]
+        self.reads = []
+
+    def report(self, mutation: str = "") -> bool:
+        posts = 0
+
+        def api(path: str, method: str = "GET", data: object = None) -> Any:
+            nonlocal posts
+            if method == "GET":
+                self.reads.append(path)
+            if mutation == "changed_before" and "/git/ref/" in path:
+                self.fixture.head = "b" * 40
+            result = self.fixture._api(path, method, data)
+            if method == "POST":
+                posts += 1
+                if posts == 1:
+                    if mutation == "changed_after":
+                        self.fixture.head = "b" * 40
+                    elif mutation == "new_attempt":
+                        self.fixture.runs["ci.yml"][0].update(
+                            run_attempt=2, status="in_progress", conclusion=None
+                        )
+                    elif mutation == "new_run":
+                        self.fixture.runs["ci.yml"][0].update(
+                            id=9999, status="in_progress", conclusion=None
+                        )
+            return result
+
+        with (
+            patch.object(queue, "api", side_effect=api),
+            patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+            patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+            patch.object(queue, "SOURCE_CACHE", {}),
+            patch.object(queue, "SOURCE_RUN_LISTS", {}),
+        ):
+            return queue.publish_dispatch_results(self.ref, self.fixture.sha, self.runs)
+
+    def test_all_nineteen_results_keep_exact_job_links_with_two_ref_leases(
+        self,
+    ) -> None:
+        self.assertTrue(self.report())
+        results = [data for _path, data in self.fixture.writes]
+        expected = {
+            job["name"]: job["html_url"]
+            for jobs in self.fixture.jobs.values()
+            for job in jobs
+        }
+        self.assertEqual(len(results), 19)
+        self.assertEqual(
+            {data["context"]: data["target_url"] for data in results}, expected
+        )
+        self.assertEqual({data["state"] for data in results}, {"success"})
+        self.assertEqual(sum("/git/ref/" in path for path in self.reads), 2)
+        self.assertEqual(
+            sum("/actions/runs?head_sha=" in path for path in self.reads), 2
+        )
+
+    def test_changed_queue_ref_before_batch_writes_nothing(self) -> None:
+        self.assertFalse(self.report("changed_before"))
+        self.assertEqual(self.fixture.writes, [])
+
+    def test_changed_ref_after_batch_only_writes_old_immutable_sha_and_cannot_advance(
+        self,
+    ) -> None:
+        self.assertFalse(self.report("changed_after"))
+        self.assertEqual(len(self.fixture.writes), 19)
+        self.assertTrue(
+            all(
+                path.endswith("/" + self.fixture.sha)
+                for path, _data in self.fixture.writes
+            )
+        )
+
+    def test_new_attempt_or_producer_after_batch_invalidates_every_published_success(
+        self,
+    ) -> None:
+        for mutation in ("new_attempt", "new_run"):
+            self.setUp()
+            with self.subTest(mutation=mutation):
+                self.assertFalse(self.report(mutation))
+                results = [data for _path, data in self.fixture.writes]
+                self.assertEqual(
+                    [data["state"] for data in results],
+                    ["success"] * 19 + ["pending"] * 19,
+                )
+
+    def test_actual_failed_job_preserves_failure_link_and_no_success(self) -> None:
+        run = self.fixture.runs["ci.yml"][0]
+        run["conclusion"] = "failure"
+        self.runs[0]["conclusion"] = "failure"
+        job = self.fixture.jobs[run["id"]][0]
+        job["conclusion"] = "failure"
+        self.assertFalse(self.report())
+        self.assertEqual(
+            [data["state"] for _path, data in self.fixture.writes], ["failure"]
+        )
+        self.assertEqual(self.fixture.writes[0][1]["target_url"], job["html_url"])
+
+    def test_wrong_job_run_or_attempt_never_produces_success(self) -> None:
+        for field in ("run_id", "run_attempt"):
+            self.setUp()
+            self.fixture.jobs[self.runs[0]["id"]][0][field] = 999
+            with self.subTest(field=field):
+                self.assertFalse(self.report())
+                self.assertEqual(self.fixture.writes, [])
+
+    def test_aggregate_rejects_tampered_path_branch_repository_and_truncated_inventory(
+        self,
+    ) -> None:
+        for field, value in (
+            ("path", ".github/workflows/wrong.yml"),
+            ("head_branch", "untrusted"),
+            ("repository", {"full_name": "other/repo"}),
+            ("head_repository", {"full_name": "other/repo"}),
+        ):
+            self.setUp()
+            self.fixture.runs["ci.yml"][0][field] = value
+            with (
+                self.subTest(field=field),
+                patch.object(queue, "api", side_effect=self.fixture._api),
+                patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+                patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+                patch.object(queue, "SOURCE_CACHE", {}),
+                self.assertRaises(ValueError),
+            ):
+                queue.queue_run_inventory(self.ref, self.fixture.sha)
+        with (
+            patch.object(
+                queue, "api", return_value={"total_count": 100, "workflow_runs": []}
+            ),
+            self.assertRaises(ValueError),
+        ):
+            queue.queue_run_inventory(self.ref, self.fixture.sha)

From fffe5bca4fdb7d9438a6adeace3cfbd4d5f8cc5b Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:27:04 -0700
Subject: [PATCH 5/9] Document package callback budgets and v2.13 migration

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 docs/releases/v2.13.0.md | 49 ++++++++++++++++++++++++++++++++++++++++
 site/mkdocs.yml          |  1 +
 2 files changed, 50 insertions(+)
 create mode 100644 docs/releases/v2.13.0.md

diff --git a/docs/releases/v2.13.0.md b/docs/releases/v2.13.0.md
new file mode 100644
index 0000000..d60070b
--- /dev/null
+++ b/docs/releases/v2.13.0.md
@@ -0,0 +1,49 @@
+# Nix Forge CI v2.13.0
+
+## Changelog
+
+- Reconcile all eligible package-source heads on a completion notification.
+  This recovers ready heads when workflow concurrency replaces a pending callback.
+- Read current PR identity, source refs and latest statuses in a GraphQL batch.
+  Use a bounded native check-suite projection to defer healthy partial producers
+  and recognize unchanged already-admitted results without writes.
+- Preserve full immutable source, workflow, current-run and actual-job proofs
+  for every new result or admission. Real failures retain their actual job URLs
+  when another producer is running or absent. Changed attempts invalidate old
+  source results. Truncated, missing and approval-only projections take the
+  complete validation path.
+- Replace callback workflow-name lookup with one direct dispatch request.
+  Preserve legacy queue callback inputs and bounded visible notification retries.
+- Pace opted-in package reconciliation content writes four seconds apart, including status,
+  enqueue and validation dispatch requests. Package-source callers must allow
+  15 minutes in their own reconciler jobs for a paced six-head sweep.
+
+A finite six-package fixture executes the actual reconciler for source
+notifications and all 30 queue completion callbacks. It preserves all 19 queue
+results and includes notification and candidate-dispatch requests. That fixture
+uses 831 REST requests. A separate 99-request source-inspection allowance brings
+the normal accounting model to 930 requests. Live read-only GraphQL queries for
+the PR snapshot and six distinct native source heads each cost one point.
+
+GitHub's standard repository token allows
+[1,000 REST requests per hour](https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api).
+Repeated reruns, updater requests and delayed producer completion consume
+additional requests. Completion callbacks retain their bounded wait when the
+producer is still running. These counts do not promise capacity for unlimited
+reruns or simultaneous unrelated automation. Real native validation can span
+multiple quota windows.
+
+Existing source start notifications remain asynchronous. They do not acknowledge
+that the reconciler has completed its status writes. Source completion
+notifications remain blocking after all actual validation jobs.
+
+No token permissions or required native job gates change. The independently
+reviewed release-builder and repository-checks pins remain separate. Consumers
+without package-source reconciliation retain their existing callback contract.
+
+The 189 repository regression tests passed locally. Required hosted validation
+and the protected normal merge queue must pass before publication. The signed
+tag and verified release manifest identify the protected source commit.
+Publication verifies the archive checksum and all release provenance subjects
+against the separately pinned SLSA builder. Restore the previous reviewed v2.x
+pin to roll back. Compatible v2.x consumers have no scheduled support end date.
diff --git a/site/mkdocs.yml b/site/mkdocs.yml
index 6cdcf29..e04fc53 100644
--- a/site/mkdocs.yml
+++ b/site/mkdocs.yml
@@ -36,6 +36,7 @@ nav:
   - OpenSSF baseline: openssf-baseline.md
   - Releases: releases.md
   - Release runbook: releases/README.md
+  - v2.13.0 release notes: releases/v2.13.0.md
   - v2.12.0 release notes: releases/v2.12.0.md
   - v2.11.0 release notes: releases/v2.11.0.md
   - v2.10.0 release notes: releases/v2.10.0.md

From 7124679ad5b97029bf2522c23fd1bc5e336217c7 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:27:29 -0700
Subject: [PATCH 6/9] Wrap the package callback release note

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 docs/releases/v2.13.0.md | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/docs/releases/v2.13.0.md b/docs/releases/v2.13.0.md
index d60070b..ce660e0 100644
--- a/docs/releases/v2.13.0.md
+++ b/docs/releases/v2.13.0.md
@@ -14,9 +14,10 @@
   complete validation path.
 - Replace callback workflow-name lookup with one direct dispatch request.
   Preserve legacy queue callback inputs and bounded visible notification retries.
-- Pace opted-in package reconciliation content writes four seconds apart, including status,
-  enqueue and validation dispatch requests. Package-source callers must allow
-  15 minutes in their own reconciler jobs for a paced six-head sweep.
+- Pace opted-in package reconciliation content writes four seconds apart,
+  including status, enqueue and validation dispatch requests. Package-source
+  callers must allow 15 minutes in their own reconciler jobs for a paced
+  six-head sweep.
 
 A finite six-package fixture executes the actual reconciler for source
 notifications and all 30 queue completion callbacks. It preserves all 19 queue

From e3d2a2754ae21030275bd9a2552e3c6eff3fdadf Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:28:57 -0700
Subject: [PATCH 7/9] State the automation model and native API accounting
 limits

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 docs/releases/v2.13.0.md | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/docs/releases/v2.13.0.md b/docs/releases/v2.13.0.md
index ce660e0..4dc3785 100644
--- a/docs/releases/v2.13.0.md
+++ b/docs/releases/v2.13.0.md
@@ -23,16 +23,19 @@ A finite six-package fixture executes the actual reconciler for source
 notifications and all 30 queue completion callbacks. It preserves all 19 queue
 results and includes notification and candidate-dispatch requests. That fixture
 uses 831 REST requests. A separate 99-request source-inspection allowance brings
-the normal accounting model to 930 requests. Live read-only GraphQL queries for
-the PR snapshot and six distinct native source heads each cost one point.
+the normal automation-layer accounting model to 930 requests. Live read-only
+GraphQL queries for the PR snapshot and six distinct native source heads each
+cost one point.
 
 GitHub's standard repository token allows
 [1,000 REST requests per hour](https://docs.github.com/en/rest/using-the-rest-api/rate-limits-for-the-rest-api).
-Repeated reruns, updater requests and delayed producer completion consume
-additional requests. Completion callbacks retain their bounded wait when the
-producer is still running. These counts do not promise capacity for unlimited
-reruns or simultaneous unrelated automation. Real native validation can span
-multiple quota windows.
+Updater publication, native validation tools, repeated reruns and delayed
+producer completion consume additional requests. Completion callbacks retain
+their bounded wait when the producer is still running. The six native queue
+candidates execute serially, so these measured calls need not share one hour.
+These counts do not guarantee capacity for arbitrary bursts. The production
+rollout must verify the current single-package update through native validation
+and the protected queue.
 
 Existing source start notifications remain asynchronous. They do not acknowledge
 that the reconciler has completed its status writes. Source completion

From 025fedb2b0b33f423627eeb3ba027f022da6a7c3 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:33:22 -0700
Subject: [PATCH 8/9] Ignore unrelated same-commit runs when checking queue
 producers

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 scripts/reconcile-queue.py    |   7 ++
 tests/test_reconcile_queue.py | 131 +++++++++++++++++++++++++++++++++-
 2 files changed, 137 insertions(+), 1 deletion(-)

diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index 0a54c10..4b66c52 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -1587,6 +1587,13 @@ def queue_run_inventory(ref: str, sha: str) -> dict[str, Any]:
         for run in SOURCE_RUN_LISTS[sha]["workflow_runs"]:
             if run.get("workflow_id") != metadata["id"]:
                 continue
+            # After a normal merge, the same commit can have a main-push run
+            # while its old queue ref still exists. That run is not evidence
+            # for this candidate; keep inspecting its own queue producers.
+            if run.get("event") not in {"merge_group", "workflow_dispatch"} or run.get(
+                "head_branch"
+            ) != ref.removeprefix("refs/heads/"):
+                continue
             if not (
                 run.get("path") == metadata["path"]
                 and run.get("head_sha") == sha
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index a1150bf..9084785 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -4,6 +4,7 @@
 
 import base64
 import contextlib
+import copy
 import importlib.util
 import io
 import os
@@ -2644,14 +2645,142 @@ def test_wrong_job_run_or_attempt_never_produces_success(self) -> None:
                 self.assertFalse(self.report())
                 self.assertEqual(self.fixture.writes, [])
 
+    def test_same_sha_main_push_does_not_replace_real_queue_codeql_evidence(
+        self,
+    ) -> None:
+        run = self.fixture.runs["codeql.yml"][0]
+        main_push = {
+            **run,
+            "id": run["id"] + 1000,
+            "event": "push",
+            "head_branch": "main",
+        }
+        self.fixture.runs["codeql.yml"].insert(0, main_push)
+        self.assertTrue(self.report())
+        links = [data["target_url"] for _path, data in self.fixture.writes]
+        self.assertEqual(len(links), 19)
+        self.assertTrue(all(f"/runs/{main_push['id']}/" not in link for link in links))
+
+    def test_lingering_refs_with_main_pushes_do_not_block_new_live_front_dispatch(
+        self,
+    ) -> None:
+        old = self.fixture
+        older = copy.deepcopy(old)
+        older.sha = "c" * 40
+        older.head = older.sha
+        older.branch = "gh-readonly-queue/main/older"
+        new_jobs = {}
+        for runs in older.runs.values():
+            run = runs[0]
+            jobs = older.jobs[run["id"]]
+            run.update(
+                id=run["id"] + 1000, head_sha=older.sha, head_branch=older.branch
+            )
+            for job in jobs:
+                job.update(id=job["id"] + 100000, run_id=run["id"], head_sha=older.sha)
+                job["html_url"] = (
+                    f"https://github.com/{queue.REPOSITORY}/actions/runs/{run['id']}/job/{job['id']}"
+                )
+            new_jobs[run["id"]] = jobs
+        older.jobs = new_jobs
+        for fixture in (old, older):
+            run = fixture.runs["codeql.yml"][0]
+            fixture.runs["codeql.yml"].insert(
+                0,
+                {
+                    **run,
+                    "id": run["id"] + 10000,
+                    "event": "push",
+                    "head_branch": "main",
+                },
+            )
+            fixture.statuses = [
+                {
+                    "context": job["name"],
+                    "state": "success",
+                    "target_url": job["html_url"],
+                    "description": "Mirrored queue job result",
+                }
+                for jobs in fixture.jobs.values()
+                for job in jobs
+            ]
+        new_sha = "d" * 40
+        new_ref = queue.QUEUE_PREFIX + "next"
+        refs = [
+            {"ref": "refs/heads/" + fixture.branch, "object": {"sha": fixture.sha}}
+            for fixture in (old, older)
+        ] + [{"ref": new_ref, "object": {"sha": new_sha}}]
+        dispatched = []
+
+        def api(path: str, method: str = "GET", data: object = None) -> Any:
+            if method == "POST":
+                self.assertTrue(path.endswith("/dispatches"))
+                dispatched.append(data)
+                return None
+            if "/git/matching-refs/" in path:
+                return refs
+            if "head_sha=" + new_sha in path:
+                return {"total_count": 0, "workflow_runs": []}
+            if "/git/commits/" in path:
+                return {"parents": [{"sha": old.base}]}
+            if path.endswith(new_ref.removeprefix("refs/")):
+                return {"object": {"sha": new_sha}}
+            if (
+                older.sha in path
+                or older.branch in path
+                or any(f"/runs/{run_id}/" in path for run_id in older.jobs)
+            ):
+                return older._api(path, method, data)
+            return old._api(path, method, data)
+
+        response = {
+            "repository": {
+                "pullRequests": {"pageInfo": {"hasNextPage": False}, "nodes": []}
+            }
+        }
+        with (
+            patch.dict(
+                os.environ,
+                {
+                    "SOURCE_RUN_ID": str(self.runs[1]["id"]),
+                    "SOURCE_RUN_ATTEMPT": "",
+                    "SOURCE_NOTIFICATION_ACTIVITY": "",
+                },
+            ),
+            patch.object(queue, "api", side_effect=api),
+            patch.object(queue, "graphql", return_value=response),
+            patch.object(queue, "SOURCE_CHECKS", old.checks),
+            patch.object(queue, "WORKFLOWS", list(old.checks)),
+            patch.object(queue, "SOURCE_HEAD", ""),
+            patch.object(queue, "SOURCE_ACTIVITY", ""),
+        ):
+            queue.main()
+        self.assertEqual(len(dispatched), 5)
+        self.assertEqual(
+            {data["ref"] for data in dispatched}, {new_ref.removeprefix("refs/heads/")}
+        )
+        self.assertEqual(old.writes + older.writes, [])
+
+    def test_nonqueue_branch_producer_is_not_candidate_evidence(self) -> None:
+        self.fixture.runs["ci.yml"][0]["head_branch"] = "main"
+        with (
+            patch.object(queue, "api", side_effect=self.fixture._api),
+            patch.object(queue, "SOURCE_CHECKS", self.fixture.checks),
+            patch.object(queue, "WORKFLOWS", list(self.fixture.checks)),
+            patch.object(queue, "SOURCE_CACHE", {}),
+        ):
+            self.assertNotIn(
+                "ci.yml", queue.queue_run_inventory(self.ref, self.fixture.sha)
+            )
+
     def test_aggregate_rejects_tampered_path_branch_repository_and_truncated_inventory(
         self,
     ) -> None:
         for field, value in (
             ("path", ".github/workflows/wrong.yml"),
-            ("head_branch", "untrusted"),
             ("repository", {"full_name": "other/repo"}),
             ("head_repository", {"full_name": "other/repo"}),
+            ("run_attempt", 0),
         ):
             self.setUp()
             self.fixture.runs["ci.yml"][0][field] = value

From f192bd4ee43228b0214c4e048e69ab177561b31c Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 22:34:55 -0700
Subject: [PATCH 9/9] Record the same-commit queue regression coverage

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 docs/releases/v2.13.0.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/docs/releases/v2.13.0.md b/docs/releases/v2.13.0.md
index 4dc3785..cd57e15 100644
--- a/docs/releases/v2.13.0.md
+++ b/docs/releases/v2.13.0.md
@@ -45,7 +45,7 @@ No token permissions or required native job gates change. The independently
 reviewed release-builder and repository-checks pins remain separate. Consumers
 without package-source reconciliation retain their existing callback contract.
 
-The 189 repository regression tests passed locally. Required hosted validation
+The 192 repository regression tests passed locally. Required hosted validation
 and the protected normal merge queue must pass before publication. The signed
 tag and verified release manifest identify the protected source commit.
 Publication verifies the archive checksum and all release provenance subjects