From eb79c486567d68cd87fd8b89843a17dc352bc3f1 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 15:59:34 -0700
Subject: [PATCH 1/7] fix(queue): verify opted-in package source validation

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 README.md                          |  18 ++
 actions/reconcile-queue/action.yml |  22 ++-
 scripts/reconcile-queue.py         | 279 ++++++++++++++++++++++++++-
 tests/test_reconcile_queue.py      | 298 +++++++++++++++++++++++++++++
 4 files changed, 611 insertions(+), 6 deletions(-)

diff --git a/README.md b/README.md
index b294d6f..2f661c9 100644
--- a/README.md
+++ b/README.md
@@ -113,6 +113,24 @@ admission remains enabled. Same-repository Dependabot updates to existing action
 SHA pins can enter automatically after full immutable workflow blobs prove that
 only pins and version comments changed. Runtime code, permissions, renamed or
 new workflows, and other automation changes still require deliberate admission.
+A package repository can opt in to dispatched source-PR result reporting with
+`package-source-checks`, a JSON object mapping each configured workflow filename
+to its required job names. Include every protected context exactly once and
+include `dco.yml` with `DCO sign-off`. Keep this map aligned with branch protection.
+The source adapter accepts only open, non-draft, same-repository PRs from the
+GitHub Actions bot on `automation/package-updates-<package>` branches. It proves
+one actual parent on protected default-branch history, changes limited to the
+selected `pkgs/by-name/<prefix>/<package>/` directory and `docs/catalog.md`, and
+workflow blobs identical to current protected main. It executes no PR code.
+
+Source statuses link actual successful required jobs only after every configured
+workflow and required job passes on the exact source SHA. A blocked approval run
+can be ignored only when its jobs endpoint proves zero jobs. Failed, pending,
+skipped, missing, stale and untrusted evidence never becomes success; newer
+attempts invalidate older source adapter results. Normal required checks and
+expected-head merge-queue admission still apply. This opt-in adds no token
+permissions and leaves existing synthetic queue result reporting unchanged.
+
 Set `base-sha-workflows: '["ci.yml"]'` for a workflow that declares a `base_sha`
 dispatch input. This supplies the queue commit's first parent for derivation
 comparison without teaching the shared library repository names.
diff --git a/actions/reconcile-queue/action.yml b/actions/reconcile-queue/action.yml
index f436211..3cccd78 100644
--- a/actions/reconcile-queue/action.yml
+++ b/actions/reconcile-queue/action.yml
@@ -7,6 +7,9 @@ inputs:
   automation-branch:
     description: Optional same-repository github-actions bot branch eligible for admission.
     default: ''
+  package-source-checks:
+    description: Opt-in JSON workflow-to-required-job map for same-repository package updater source PRs.
+    default: '{}'
   base-sha-workflows:
     description: JSON array of workflows accepting the base_sha dispatch input.
     default: '[]'
@@ -19,6 +22,7 @@ runs:
     - name: Validate queue inputs
       shell: bash
       env:
+        PACKAGE_SOURCE_CHECKS: ${{ inputs.package-source-checks }}
         AUTOMATION_BRANCH: ${{ inputs.automation-branch }}
         BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }}
         QUEUE_WORKFLOWS: ${{ inputs.workflows }}
@@ -33,19 +37,32 @@ runs:
           printf 'Invalid automation-branch input: %s\n' "$AUTOMATION_BRANCH" >&2
           exit 1
         fi
-        python3 - "$QUEUE_WORKFLOWS" "$BASE_SHA_WORKFLOWS" <<'PY'
+        python3 - "$QUEUE_WORKFLOWS" "$BASE_SHA_WORKFLOWS" "$PACKAGE_SOURCE_CHECKS" <<'PY'
         import json
         import re
         import sys
 
         workflow_pattern = re.compile(r"^[A-Za-z0-9._/-]+\.yml$")
-        for label, value in zip(("workflows", "base-sha-workflows"), sys.argv[1:]):
+        for label, value in zip(("workflows", "base-sha-workflows"), sys.argv[1:3]):
             entries = json.loads(value)
             if not isinstance(entries, list) or not all(
                 isinstance(entry, str) and workflow_pattern.fullmatch(entry)
                 for entry in entries
             ):
                 raise SystemExit(f"Invalid {label} input")
+        source = json.loads(sys.argv[3])
+        if not isinstance(source, dict) or any(
+            not workflow_pattern.fullmatch(key) or not isinstance(jobs, list) or not jobs
+            or any(not isinstance(job, str) or not job or len(job) > 256 for job in jobs)
+            or len(jobs) != len(set(jobs))
+            for key, jobs in source.items()
+        ):
+            raise SystemExit("Invalid package-source-checks input")
+        contexts = [job for jobs in source.values() for job in jobs]
+        if len(contexts) != len(set(contexts)):
+            raise SystemExit("Source required job names must be unique")
+        if source and (set(source) != set(json.loads(sys.argv[1])) or "DCO sign-off" not in source.get("dco.yml", [])):
+            raise SystemExit("Source checks must cover every workflow, including actual source DCO")
         PY
     - name: Set up queue validation Python
       uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7
@@ -66,6 +83,7 @@ runs:
         GH_TOKEN: ${{ github.token }}
         GH_REPO: ${{ github.repository }}
         DEFAULT_BRANCH: ${{ github.event.repository.default_branch }}
+        PACKAGE_SOURCE_CHECKS: ${{ inputs.package-source-checks }}
         AUTOMATION_BRANCH: ${{ inputs.automation-branch }}
         BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }}
         QUEUE_WORKFLOWS: ${{ inputs.workflows }}
diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index d36cc27..f1d4477 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -28,6 +28,8 @@
 BASE_SHA_WORKFLOWS = json.loads(os.environ.get("BASE_SHA_WORKFLOWS", "[]"))
 QUEUE_PREFIX = f"refs/heads/gh-readonly-queue/{DEFAULT_BRANCH}/"
 ADMISSION = "Merge queue admission"
+SOURCE_CHECKS = json.loads(os.environ.get("PACKAGE_SOURCE_CHECKS", "{}"))
+SOURCE_REPORT = "Mirrored package source job"
 PAGE_SIZE = 100
 GH = shutil.which("gh") or "/usr/bin/gh"
 
@@ -76,10 +78,17 @@ def admit(pr: dict[str, Any]) -> bool:
         Whether this trusted PR is now queued and may need a validation ref.
 
     """
-    trusted = pr["user"]["login"] == "dependabot[bot]" or (
-        bool(AUTOMATION_BRANCH)
-        and pr["user"]["login"] == "github-actions[bot]"
-        and pr["head"]["ref"] == AUTOMATION_BRANCH
+    source = bool(SOURCE_CHECKS) and source_identity(pr)
+    if source and not reconcile_source(pr):
+        return False
+    trusted = (
+        source
+        or pr["user"]["login"] == "dependabot[bot]"
+        or (
+            bool(AUTOMATION_BRANCH)
+            and pr["user"]["login"] == "github-actions[bot]"
+            and pr["head"]["ref"] == AUTOMATION_BRANCH
+        )
     )
     if (
         not trusted
@@ -378,6 +387,268 @@ def front_queue_entry() -> dict[str, Any] | None:
     return entries[0] if entries else None
 
 
+def source_identity(pr: dict[str, Any]) -> bool:
+    """Limit opt-in source reconciliation to the package updater's bot PRs.
+
+    Returns:
+        Whether identity and namespace are eligible for immutable inspection.
+
+    """
+    return (
+        pr.get("state") == "open"
+        and not pr.get("draft", True)
+        and pr.get("user", {}).get("id") == 41898282
+        and pr["user"].get("type") == "Bot"
+        and pr["user"].get("login") == "github-actions[bot]"
+        and (pr.get("head", {}).get("repo") or {}).get("full_name") == REPOSITORY
+        and (pr.get("base", {}).get("repo") or {}).get("full_name") == REPOSITORY
+        and pr["base"].get("ref") == DEFAULT_BRANCH
+        and re.fullmatch(
+            r"automation/package-updates-[a-z0-9][a-z0-9+._-]*",
+            pr["head"].get("ref", ""),
+        )
+        is not None
+        and re.fullmatch(r"[0-9a-f]{40}", pr["head"].get("sha", "")) is not None
+    )
+
+
+def source_contract(pr: dict[str, Any]) -> str | None:
+    """Prove the real source parent, owned paths and current workflow identities.
+
+    Returns:
+        The exact protected base, or None for an untrusted or stale source.
+
+    """
+    sha = pr["head"]["sha"]
+    commit = api(f"repos/{REPOSITORY}/git/commits/{sha}")
+    parents = commit.get("parents", [])
+    if len(parents) != 1:
+        return 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"]
+    comparison = api(f"repos/{REPOSITORY}/compare/{base}...{protected}")
+    if comparison.get("status") not in {"ahead", "identical"}:
+        return None
+    changes = api(f"repos/{REPOSITORY}/compare/{base}...{sha}")
+    files = changes.get("files", [])
+    package = pr["head"]["ref"].removeprefix("automation/package-updates-")
+    prefix = f"pkgs/by-name/{package[:2]}/{package}/"
+    if (
+        changes.get("status") != "ahead"
+        or changes.get("total_commits") != 1
+        or not files
+        or len(files) >= 300
+        or not any(item["filename"].startswith(prefix) for item in files)
+        or any(
+            item.get("status") == "renamed"
+            or not (
+                item["filename"].startswith(prefix)
+                or item["filename"] == "docs/catalog.md"
+            )
+            for item in files
+        )
+    ):
+        return None
+    trees = []
+    for revision in (sha, protected):
+        tree = api(f"repos/{REPOSITORY}/git/trees/{revision}?recursive=1")
+        if tree.get("truncated") is not False:
+            return None
+        trees.append({item["path"]: item for item in tree["tree"]})
+    for workflow in WORKFLOWS:
+        path = f".github/workflows/{workflow}"
+        old, current = (tree.get(path, {}) for tree in trees)
+        if (
+            not old.get("sha")
+            or old.get("sha") != current.get("sha")
+            or old.get("mode") != "100644"
+            or current.get("mode") != "100644"
+            or old.get("type") != "blob"
+            or current.get("type") != "blob"
+        ):
+            return None
+    return base
+
+
+def source_jobs(pr: dict[str, Any]) -> list[dict[str, Any]] | None:
+    """Read the latest actual validation attempt for every trusted workflow.
+
+    Returns:
+        Its real completed jobs, or None while evidence is missing or pending.
+
+    """
+    jobs: list[dict[str, Any]] = []
+    sha, branch = pr["head"]["sha"], pr["head"]["ref"]
+    for workflow in WORKFLOWS:
+        metadata = api(f"repos/{REPOSITORY}/actions/workflows/{workflow}")
+        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}"
+        )
+        if response["total_count"] > PAGE_SIZE:
+            return None
+        selected = None
+        selected_jobs = []
+        for run in response["workflow_runs"]:
+            if not (
+                run.get("workflow_id") == metadata["id"]
+                and run.get("path") == path
+                and run.get("head_sha") == sha
+                and run.get("head_branch") == branch
+                and run.get("repository", {}).get("full_name") == REPOSITORY
+                and run.get("head_repository", {}).get("full_name") == REPOSITORY
+                and run.get("event") in {"pull_request", "workflow_dispatch"}
+                and isinstance(run.get("id"), int)
+                and run["id"] > 0
+                and isinstance(run.get("run_attempt"), int)
+                and run["run_attempt"] > 0
+            ):
+                continue
+            batch = api(
+                f"repos/{REPOSITORY}/actions/runs/{run['id']}/jobs?filter=latest&per_page={PAGE_SIZE}"
+            )
+            if batch["total_count"] > PAGE_SIZE:
+                return None
+            if (
+                run.get("status") == "completed"
+                and run.get("conclusion") == "action_required"
+                and batch["total_count"] == 0
+                and batch["jobs"] == []
+            ):
+                continue
+            selected, selected_jobs = run, batch["jobs"]
+            break
+        if selected is None or selected.get("status") != "completed":
+            return None
+        history = api(
+            f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}"
+        )
+        if history["total_count"] > PAGE_SIZE:
+            return None
+        fresh = next(
+            (
+                run
+                for run in history["workflow_runs"]
+                if run.get("id") == selected["id"]
+            ),
+            {},
+        )
+        position = next(
+            (
+                index
+                for index, run in enumerate(history["workflow_runs"])
+                if run.get("id") == selected["id"]
+            ),
+            None,
+        )
+        if position is None:
+            return None
+        for newer in history["workflow_runs"][:position]:
+            if (
+                newer.get("head_sha") != sha
+                or newer.get("workflow_id") != metadata["id"]
+            ):
+                return None
+            batch = api(
+                f"repos/{REPOSITORY}/actions/runs/{newer['id']}/jobs?filter=latest&per_page={PAGE_SIZE}"
+            )
+            if not (
+                newer.get("status") == "completed"
+                and newer.get("conclusion") == "action_required"
+                and batch.get("total_count") == 0
+                and batch.get("jobs") == []
+            ):
+                return None
+        fresh_attempt = api(f"repos/{REPOSITORY}/actions/runs/{selected['id']}")
+        if fresh != selected or fresh_attempt.get("run_attempt") != selected.get(
+            "run_attempt"
+        ):
+            return None
+        if any(
+            fresh.get(key) != selected.get(key)
+            for key in ("head_sha", "run_attempt", "status", "conclusion")
+        ):
+            return None
+        if selected.get("conclusion") != "success" or any(
+            job.get("head_sha") != sha or job.get("status") != "completed"
+            for job in selected_jobs
+        ):
+            return None
+        for context in SOURCE_CHECKS[workflow]:
+            matches = [job for job in selected_jobs if job.get("name") == context]
+            if len(matches) != 1 or matches[0].get("conclusion") != "success":
+                return None
+            job = matches[0]
+            if not job.get("html_url", "").startswith(
+                f"https://github.com/{REPOSITORY}/actions/runs/{selected['id']}/job/"
+            ):
+                return None
+            jobs.append(job)
+    return jobs
+
+
+def reconcile_source(pr: dict[str, Any]) -> bool:
+    """Expose only verified source results, preserving failures and pending work.
+
+    Returns:
+        Whether the complete reviewed required source job set has passed.
+
+    """
+    ref, sha = f"refs/heads/{pr['head']['ref']}", pr["head"]["sha"]
+    if set(SOURCE_CHECKS) != set(WORKFLOWS) or "DCO sign-off" not in SOURCE_CHECKS.get(
+        "dco.yml", []
+    ):
+        return False
+    if not source_identity(pr):
+        return False
+    jobs = source_jobs(pr) if source_contract(pr) is not None else None
+    statuses = read_statuses(sha)
+    results = (
+        [
+            {"context": job["name"], "state": "success", "target_url": job["html_url"]}
+            for job in jobs
+        ]
+        if jobs
+        else []
+    )
+    for status in statuses.values():
+        if status.get("description", "").startswith(SOURCE_REPORT) and not jobs:
+            results.append(
+                {
+                    "context": status["context"],
+                    "state": "pending",
+                    "target_url": status["target_url"],
+                }
+            )
+    if jobs and source_contract(pr) is None:
+        return False
+    for result in results:
+        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
+        previous = statuses.get(result["context"], {})
+        if (
+            previous.get("state") == result["state"]
+            and previous.get("target_url") == result["target_url"]
+        ):
+            continue
+        api(
+            f"repos/{REPOSITORY}/statuses/{sha}",
+            "POST",
+            {**result, "description": f"{SOURCE_REPORT} result"},
+        )
+    return bool(jobs)
+
+
 def main() -> None:
     """Reconcile open automation PRs, then start missing queue validation.
 
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index f80253b..0597846 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -890,5 +890,303 @@ def test_rejects_invalid_run_id(self) -> None:
         api.assert_not_called()
 
 
+class PackageSourceTests(unittest.TestCase):
+    """Exercise opted-in source proof with real REST response shapes."""
+
+    def setUp(self) -> None:
+        self.sha, self.base = "a" * 40, "b" * 40
+        self.branch = "automation/package-updates-example"
+        self.pr = {
+            "state": "open",
+            "draft": False,
+            "number": 7,
+            "node_id": "PR_7",
+            "user": {"login": "github-actions[bot]", "type": "Bot", "id": 41898282},
+            "head": {
+                "sha": self.sha,
+                "ref": self.branch,
+                "repo": {"full_name": queue.REPOSITORY},
+            },
+            "base": {
+                "ref": queue.DEFAULT_BRANCH,
+                "repo": {"full_name": queue.REPOSITORY},
+            },
+        }
+        self.checks = {"ci.yml": ["Build", "Lint"], "dco.yml": ["DCO sign-off"]}
+        self.runs = {}
+        self.jobs = {}
+        for index, workflow in enumerate(self.checks, 1):
+            self.runs[workflow] = [
+                {
+                    "id": index,
+                    "workflow_id": index,
+                    "path": f".github/workflows/{workflow}",
+                    "head_sha": self.sha,
+                    "head_branch": self.branch,
+                    "event": "workflow_dispatch",
+                    "repository": {"full_name": queue.REPOSITORY},
+                    "head_repository": {"full_name": queue.REPOSITORY},
+                    "status": "completed",
+                    "conclusion": "success",
+                    "run_attempt": 1,
+                }
+            ]
+            self.jobs[index] = [
+                {
+                    "id": index * 10 + number,
+                    "name": name,
+                    "head_sha": self.sha,
+                    "status": "completed",
+                    "conclusion": "success",
+                    "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{index}/job/{index * 10 + number}",
+                }
+                for number, name in enumerate(self.checks[workflow])
+            ]
+        self.paths = [
+            {"filename": "pkgs/by-name/ex/example/source.nix", "status": "modified"},
+            {"filename": "docs/catalog.md", "status": "modified"},
+        ]
+        self.parents = [{"sha": self.base}]
+        self.ancestry = "ahead"
+        self.tree = {
+            "truncated": False,
+            "tree": [
+                {
+                    "path": f".github/workflows/{workflow}",
+                    "sha": "c" * 40,
+                    "mode": "100644",
+                    "type": "blob",
+                }
+                for workflow in self.checks
+            ],
+        }
+        self.drift = False
+        self.head = self.sha
+        self.statuses = []
+        self.writes = []
+        self.fail_jobs_api = False
+        self.new_attempt = False
+        self.new_run = False
+        self.history_reads = 0
+
+    def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
+        if method != "GET":
+            self.writes.append((path, data))
+            return None
+        if "/git/commits/" in path:
+            return {"parents": self.parents}
+        if f"compare/{self.base}...{self.base}" in path:
+            return {"status": self.ancestry}
+        if "/compare/" in path:
+            return {"status": "ahead", "total_commits": 1, "files": self.paths}
+        if "/git/trees/" in path:
+            if self.drift and self.sha not in path:
+                return {"truncated": False, "tree": []}
+            return self.tree
+        if "/git/ref/heads/" in path:
+            return {"object": {"sha": self.head if self.branch in path else self.base}}
+        if path.endswith("/pulls/7"):
+            return self.pr
+        if "/status?" in path or path.endswith("/status"):
+            return {"statuses": self.statuses}
+        if "/files?" in path:
+            return self.paths
+        for index, workflow in enumerate(self.checks, 1):
+            if path.endswith(f"/workflows/{workflow}"):
+                return {
+                    "id": index,
+                    "path": f".github/workflows/{workflow}",
+                    "state": "active",
+                }
+            if f"/workflows/{workflow}/runs?" in path:
+                self.history_reads += 1
+                if self.new_run and self.history_reads == 2:
+                    self.runs[workflow].insert(
+                        0,
+                        {
+                            **self.runs[workflow][0],
+                            "id": 3,
+                            "status": "in_progress",
+                            "conclusion": None,
+                        },
+                    )
+                    self.jobs[3] = []
+                return {
+                    "total_count": len(self.runs[workflow]),
+                    "workflow_runs": self.runs[workflow],
+                }
+        if "/jobs?" in path:
+            if self.fail_jobs_api:
+                raise subprocess.CalledProcessError(1, ["gh", "api"])
+            run_id = int(path.split("/runs/")[1].split("/")[0])
+            return {"total_count": len(self.jobs[run_id]), "jobs": self.jobs[run_id]}
+        if "/actions/runs/" in path:
+            run_id = int(path.rsplit("/", 1)[1])
+            for runs in self.runs.values():
+                for run in runs:
+                    if run["id"] == run_id:
+                        return {**run, "run_attempt": 2} if self.new_attempt else run
+        raise AssertionError(path)
+
+    def _report(self) -> bool:
+        with (
+            patch.object(queue, "api", side_effect=self._api),
+            patch.object(queue, "SOURCE_CHECKS", self.checks),
+            patch.object(queue, "WORKFLOWS", list(self.checks)),
+        ):
+            return queue.reconcile_source(self.pr)
+
+    def test_real_required_jobs_become_linked_results(self) -> None:
+        self.assertTrue(self._report())
+        self.assertEqual(
+            [data["context"] for _, data in self.writes],
+            ["Build", "Lint", "DCO sign-off"],
+        )
+        self.assertTrue(all(data["state"] == "success" for _, data in self.writes))
+        self.assertEqual(self.writes[-1][1]["target_url"], self.jobs[2][0]["html_url"])
+
+    def test_actions_source_admission_requires_explicit_opt_in(self) -> None:
+        result = subprocess.CompletedProcess([], 0, '[{"bucket":"pass"}]')
+        with (
+            patch.object(queue, "api", side_effect=self._api),
+            patch.object(queue, "SOURCE_CHECKS", self.checks),
+            patch.object(queue, "WORKFLOWS", list(self.checks)),
+            patch.object(
+                queue, "graphql", return_value={"node": {"mergeQueueEntry": None}}
+            ) as graphql,
+            patch.object(queue.subprocess, "run", return_value=result),
+        ):
+            self.assertTrue(queue.admit(self.pr))
+            self.assertEqual(graphql.call_count, 2)
+            self.assertEqual(self.writes[-1][1]["context"], queue.ADMISSION)
+        self.writes.clear()
+        with (
+            patch.object(queue, "SOURCE_CHECKS", {}),
+            patch.object(queue, "api") as api,
+        ):
+            self.assertFalse(queue.admit(self.pr))
+            api.assert_not_called()
+
+    def test_wrong_author_fork_draft_and_namespace_are_ineligible(self) -> None:
+        for key, value in (("id", 9), ("type", "User"), ("login", "someone")):
+            with self.subTest(key=key):
+                old = self.pr["user"][key]
+                self.pr["user"][key] = value
+                self.assertFalse(queue.source_identity(self.pr))
+                self.pr["user"][key] = old
+        self.pr["head"]["repo"]["full_name"] = "attacker/example"
+        self.assertFalse(queue.source_identity(self.pr))
+        self.pr["head"]["repo"]["full_name"] = queue.REPOSITORY
+        self.pr["draft"] = True
+        self.assertFalse(queue.source_identity(self.pr))
+        self.pr["draft"] = False
+        self.pr["head"]["ref"] = "automation/package-updates/example"
+        self.assertFalse(queue.source_identity(self.pr))
+
+    def test_wrong_parent_ancestry_scope_and_workflow_are_rejected(self) -> None:
+        for invalid in (
+            "parent",
+            "ancestry",
+            "scope",
+            "rename",
+            "workflow",
+            "truncated",
+        ):
+            with self.subTest(invalid=invalid):
+                self.setUp()
+                if invalid == "parent":
+                    self.parents.append({"sha": "d" * 40})
+                if invalid == "ancestry":
+                    self.ancestry = "diverged"
+                if invalid == "scope":
+                    self.paths.append({"filename": ".github/workflows/dco.yml"})
+                if invalid == "rename":
+                    self.paths[0]["status"] = "renamed"
+                if invalid == "workflow":
+                    self.drift = True
+                if invalid == "truncated":
+                    self.tree["truncated"] = True
+                self.assertFalse(self._report())
+                self.assertEqual(self.writes, [])
+
+    def test_real_failed_pending_skipped_missing_and_wrong_head_jobs_never_pass(
+        self,
+    ) -> None:
+        for invalid in (
+            "failure",
+            "pending",
+            "skipped",
+            "missing",
+            "head",
+            "url",
+            "attempt",
+        ):
+            with self.subTest(invalid=invalid):
+                self.setUp()
+                if invalid == "failure":
+                    self.runs["dco.yml"][0]["conclusion"] = "failure"
+                if invalid == "pending":
+                    self.runs["dco.yml"][0]["status"] = "in_progress"
+                if invalid == "skipped":
+                    self.jobs[2][0]["conclusion"] = "skipped"
+                if invalid == "missing":
+                    self.jobs[2] = []
+                if invalid == "head":
+                    self.jobs[2][0]["head_sha"] = "other"
+                if invalid == "url":
+                    self.jobs[2][0]["html_url"] = "https://attacker.example/job"
+                if invalid == "attempt":
+                    self.new_attempt = True
+                if invalid == "new_run":
+                    self.new_run = True
+                self.assertFalse(self._report())
+                self.assertEqual(self.writes, [])
+
+    def test_zero_job_approval_run_does_not_suppress_real_dispatch(self) -> None:
+        self.runs["dco.yml"].insert(
+            0,
+            {
+                **self.runs["dco.yml"][0],
+                "id": 3,
+                "event": "pull_request",
+                "conclusion": "action_required",
+            },
+        )
+        self.jobs[3] = []
+        self.assertTrue(self._report())
+        self.jobs[3] = [{**self.jobs[2][0], "conclusion": "failure"}]
+        self.writes.clear()
+        self.assertFalse(self._report())
+        self.assertEqual(self.writes, [])
+
+    def test_api_error_does_not_write_or_enqueue(self) -> None:
+        self.fail_jobs_api = True
+        with self.assertRaises(subprocess.CalledProcessError):
+            self._report()
+        self.assertEqual(self.writes, [])
+
+    def test_failed_new_attempt_invalidates_previous_source_success(self) -> None:
+        self.statuses = [
+            {
+                "context": "DCO sign-off",
+                "state": "success",
+                "target_url": self.jobs[2][0]["html_url"],
+                "description": f"{queue.SOURCE_REPORT} result",
+            }
+        ]
+        self.runs["dco.yml"][0]["conclusion"] = "failure"
+        self.assertFalse(self._report())
+        self.assertEqual([data["state"] for _, data in self.writes], ["pending"])
+
+    def test_changed_ref_and_missing_dco_configuration_do_not_report(self) -> None:
+        self.head = "d" * 40
+        self.assertFalse(self._report())
+        self.assertEqual(self.writes, [])
+        self.setUp()
+        self.checks["dco.yml"] = ["other"]
+        self.assertFalse(self._report())
+        self.assertEqual(self.writes, [])
+
+
 if __name__ == "__main__":
     unittest.main()

From 0e5f3264eec4f3664ced917201ed24f74ecc8fac Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:10:58 -0700
Subject: [PATCH 2/7] fix(queue): preserve source failures and bound API
 polling

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 README.md                          |  12 +-
 actions/reconcile-queue/action.yml |   4 +
 scripts/reconcile-queue.py         | 393 +++++++++++++++++++----------
 tests/test_reconcile_queue.py      | 258 ++++++++++++++++++-
 4 files changed, 533 insertions(+), 134 deletions(-)

diff --git a/README.md b/README.md
index 2f661c9..239a8ef 100644
--- a/README.md
+++ b/README.md
@@ -127,8 +127,16 @@ Source statuses link actual successful required jobs only after every configured
 workflow and required job passes on the exact source SHA. A blocked approval run
 can be ignored only when its jobs endpoint proves zero jobs. Failed, pending,
 skipped, missing, stale and untrusted evidence never becomes success; newer
-attempts invalidate older source adapter results. Normal required checks and
-expected-head merge-queue admission still apply. This opt-in adds no token
+attempts invalidate older source adapter results. Actual failed required jobs
+receive failure statuses linking the current failed job. Normal required checks and
+expected-head merge-queue admission still apply. Unchanged completed runs reuse
+only this bot's previously verified job attestations for the same run and attempt;
+latest-run metadata remains live. Immutable Git objects and workflow identity are
+cached only within one reconciliation. Opted-in callers should also reconcile
+`workflow_run` requested and in-progress events to invalidate rerun results early.
+These events inspect only their exact source head; initial events without prior
+results do not poll validation, and unrelated scans skip already-admitted heads.
+Keep hourly scheduled recovery for missed events. This opt-in adds no token
 permissions and leaves existing synthetic queue result reporting unchanged.
 
 Set `base-sha-workflows: '["ci.yml"]'` for a workflow that declares a `base_sha`
diff --git a/actions/reconcile-queue/action.yml b/actions/reconcile-queue/action.yml
index 3cccd78..6d94fba 100644
--- a/actions/reconcile-queue/action.yml
+++ b/actions/reconcile-queue/action.yml
@@ -27,6 +27,8 @@ runs:
         BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }}
         QUEUE_WORKFLOWS: ${{ inputs.workflows }}
         SOURCE_RUN_ID: ${{ inputs.source-run-id }}
+        SOURCE_HEAD_SHA: ${{ github.event.workflow_run.head_sha }}
+        SOURCE_ACTIVITY: ${{ github.event.action }}
       run: |
         set -euo pipefail
         if [[ -n "$SOURCE_RUN_ID" && ! "$SOURCE_RUN_ID" =~ ^[0-9]+$ ]]; then
@@ -88,4 +90,6 @@ runs:
         BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }}
         QUEUE_WORKFLOWS: ${{ inputs.workflows }}
         SOURCE_RUN_ID: ${{ inputs.source-run-id }}
+        SOURCE_HEAD_SHA: ${{ github.event.workflow_run.head_sha }}
+        SOURCE_ACTIVITY: ${{ github.event.action }}
       run: python3 "$ACTION_PATH/../../scripts/reconcile-queue.py"
diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index f1d4477..d8cd552 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -30,6 +30,10 @@
 ADMISSION = "Merge queue admission"
 SOURCE_CHECKS = json.loads(os.environ.get("PACKAGE_SOURCE_CHECKS", "{}"))
 SOURCE_REPORT = "Mirrored package source job"
+SOURCE_HEAD = os.environ.get("SOURCE_HEAD_SHA", "")
+SOURCE_ACTIVITY = os.environ.get("SOURCE_ACTIVITY", "")
+SOURCE_CACHE: dict[str, Any] = {}
+SOURCE_EVIDENCE: dict[str, dict[str, Any]] = {}
 PAGE_SIZE = 100
 GH = shutil.which("gh") or "/usr/bin/gh"
 
@@ -79,8 +83,22 @@ def admit(pr: dict[str, Any]) -> bool:
 
     """
     source = bool(SOURCE_CHECKS) and source_identity(pr)
-    if source and not reconcile_source(pr):
-        return False
+    source_statuses = None
+    if source:
+        sha = pr["head"]["sha"]
+        if SOURCE_HEAD 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()
+        ):
+            state = graphql(
+                "query($id:ID!) { node(id:$id) { ... on PullRequest { mergeQueueEntry { id } } } }",
+                id=pr["node_id"],
+            )["node"]
+            return bool(state["mergeQueueEntry"])
+        if not reconcile_source(pr, source_statuses):
+            return False
     trusted = (
         source
         or pr["user"]["login"] == "dependabot[bot]"
@@ -123,7 +141,11 @@ def admit(pr: dict[str, Any]) -> bool:
         return True
     # Once admitted, a rejected head needs a fix or deliberate requeue.
     # Never keep rebuilding the same failing merge group indefinitely.
-    statuses = api(f"repos/{REPOSITORY}/commits/{sha}/status")["statuses"]
+    statuses = (
+        list(source_statuses.values())
+        if source_statuses is not None
+        else api(f"repos/{REPOSITORY}/commits/{sha}/status")["statuses"]
+    )
     if any(status["context"] == ADMISSION for status in statuses):
         print(f"PR #{number}: already admitted this head; leave rejection for review")
         return False
@@ -412,6 +434,18 @@ def source_identity(pr: dict[str, Any]) -> bool:
     )
 
 
+def source_read(path: str) -> Any:  # ruff: ignore[any-type]
+    """Cache immutable source objects and stable workflow identity within one pass.
+
+    Returns:
+        The authoritative response; current refs and latest runs are never cached.
+
+    """
+    if path not in SOURCE_CACHE:
+        SOURCE_CACHE[path] = api(path)
+    return SOURCE_CACHE[path]
+
+
 def source_contract(pr: dict[str, Any]) -> str | None:
     """Prove the real source parent, owned paths and current workflow identities.
 
@@ -420,7 +454,7 @@ def source_contract(pr: dict[str, Any]) -> str | None:
 
     """
     sha = pr["head"]["sha"]
-    commit = api(f"repos/{REPOSITORY}/git/commits/{sha}")
+    commit = source_read(f"repos/{REPOSITORY}/git/commits/{sha}")
     parents = commit.get("parents", [])
     if len(parents) != 1:
         return None
@@ -430,10 +464,10 @@ def source_contract(pr: dict[str, Any]) -> str | None:
     protected = api(
         f"repos/{REPOSITORY}/git/ref/heads/{quote(DEFAULT_BRANCH, safe='')}"
     )["object"]["sha"]
-    comparison = api(f"repos/{REPOSITORY}/compare/{base}...{protected}")
+    comparison = source_read(f"repos/{REPOSITORY}/compare/{base}...{protected}")
     if comparison.get("status") not in {"ahead", "identical"}:
         return None
-    changes = api(f"repos/{REPOSITORY}/compare/{base}...{sha}")
+    changes = source_read(f"repos/{REPOSITORY}/compare/{base}...{sha}")
     files = changes.get("files", [])
     package = pr["head"]["ref"].removeprefix("automation/package-updates-")
     prefix = f"pkgs/by-name/{package[:2]}/{package}/"
@@ -455,7 +489,7 @@ def source_contract(pr: dict[str, Any]) -> str | None:
         return None
     trees = []
     for revision in (sha, protected):
-        tree = api(f"repos/{REPOSITORY}/git/trees/{revision}?recursive=1")
+        tree = source_read(f"repos/{REPOSITORY}/git/trees/{revision}?recursive=1")
         if tree.get("truncated") is not False:
             return None
         trees.append({item["path"]: item for item in tree["tree"]})
@@ -474,178 +508,275 @@ def source_contract(pr: dict[str, Any]) -> str | None:
     return base
 
 
-def source_jobs(pr: dict[str, Any]) -> list[dict[str, Any]] | None:
-    """Read the latest actual validation attempt for every trusted workflow.
+def source_run(
+    pr: dict[str, Any], workflow: str
+) -> tuple[dict[str, Any], dict[str, Any] | None] | None:
+    """Select the latest real exact-head run, proving approval-only exclusions.
 
     Returns:
-        Its real completed jobs, or None while evidence is missing or pending.
+        Current run metadata and optional already-read jobs, or None.
 
     """
-    jobs: list[dict[str, Any]] = []
     sha, branch = pr["head"]["sha"], pr["head"]["ref"]
-    for workflow in WORKFLOWS:
-        metadata = api(f"repos/{REPOSITORY}/actions/workflows/{workflow}")
-        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}"
-        )
-        if response["total_count"] > PAGE_SIZE:
-            return None
-        selected = None
-        selected_jobs = []
-        for run in response["workflow_runs"]:
-            if not (
-                run.get("workflow_id") == metadata["id"]
-                and run.get("path") == path
-                and run.get("head_sha") == sha
-                and run.get("head_branch") == branch
-                and run.get("repository", {}).get("full_name") == REPOSITORY
-                and run.get("head_repository", {}).get("full_name") == REPOSITORY
-                and run.get("event") in {"pull_request", "workflow_dispatch"}
-                and isinstance(run.get("id"), int)
-                and run["id"] > 0
-                and isinstance(run.get("run_attempt"), int)
-                and run["run_attempt"] > 0
-            ):
-                continue
+    metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{workflow}")
+    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}"
+    )
+    if (
+        not isinstance(response.get("total_count"), int)
+        or not 0 <= response["total_count"] <= PAGE_SIZE
+        or not isinstance(response.get("workflow_runs"), list)
+    ):
+        return None
+    for run in response["workflow_runs"]:
+        if not (
+            run.get("workflow_id") == metadata["id"]
+            and run.get("path") == path
+            and run.get("head_sha") == sha
+            and run.get("head_branch") == branch
+            and run.get("repository", {}).get("full_name") == REPOSITORY
+            and run.get("head_repository", {}).get("full_name") == REPOSITORY
+            and run.get("event") in {"pull_request", "workflow_dispatch"}
+            and isinstance(run.get("id"), int)
+            and run["id"] > 0
+            and isinstance(run.get("run_attempt"), int)
+            and run["run_attempt"] > 0
+        ):
+            continue
+        batch = None
+        if (
+            run.get("status") == "completed"
+            and run.get("conclusion") == "action_required"
+        ):
             batch = api(
                 f"repos/{REPOSITORY}/actions/runs/{run['id']}/jobs?filter=latest&per_page={PAGE_SIZE}"
             )
-            if batch["total_count"] > PAGE_SIZE:
-                return None
-            if (
-                run.get("status") == "completed"
-                and run.get("conclusion") == "action_required"
-                and batch["total_count"] == 0
-                and batch["jobs"] == []
-            ):
+            if batch.get("total_count") == 0 and batch.get("jobs") == []:
                 continue
-            selected, selected_jobs = run, batch["jobs"]
-            break
-        if selected is None or selected.get("status") != "completed":
+        return run.copy(), batch
+    return None
+
+
+def source_runs_current(pr: dict[str, Any]) -> bool:
+    """Reject any producer run or attempt that changed after collection.
+
+    Returns:
+        Whether every collected run still has its exact inspected identity/state.
+
+    """
+    for workflow, expected in SOURCE_EVIDENCE[pr["head"]["sha"]]["runs"].items():
+        selected = source_run(pr, workflow)
+        if selected is None or any(
+            selected[0].get(key) != expected.get(key)
+            for key in ("id", "run_attempt", "head_sha", "status", "conclusion")
+        ):
+            return False
+    return True
+
+
+def source_jobs(
+    pr: dict[str, Any], statuses: dict[str, Any] | None = None
+) -> list[dict[str, Any]] | None:
+    """Collect actual current jobs, or reuse our verified unchanged-run attestation.
+
+    Returns:
+        All required successful jobs, or None for incomplete/failed evidence.
+
+    """
+    sha = pr["head"]["sha"]
+    evidence: dict[str, Any] = {"runs": {}, "failures": []}
+    SOURCE_EVIDENCE[sha] = evidence
+    jobs: list[dict[str, Any]] = []
+    failures = {"failure", "timed_out", "cancelled"}
+    for workflow in WORKFLOWS:
+        selected = source_run(pr, workflow)
+        if selected is None:
             return None
-        history = api(
-            f"repos/{REPOSITORY}/actions/workflows/{workflow}/runs?head_sha={sha}&per_page={PAGE_SIZE}"
-        )
-        if history["total_count"] > PAGE_SIZE:
+        run, batch = selected
+        evidence["runs"][workflow] = run
+        if run.get("status") != "completed":
             return None
-        fresh = next(
-            (
-                run
-                for run in history["workflow_runs"]
-                if run.get("id") == selected["id"]
-            ),
-            {},
-        )
-        position = next(
-            (
-                index
-                for index, run in enumerate(history["workflow_runs"])
-                if run.get("id") == selected["id"]
-            ),
-            None,
+        previous = (
+            [statuses.get(context, {}) for context in SOURCE_CHECKS[workflow]]
+            if statuses
+            else []
         )
-        if position is None:
-            return None
-        for newer in history["workflow_runs"][:position]:
-            if (
-                newer.get("head_sha") != sha
-                or newer.get("workflow_id") != metadata["id"]
-            ):
-                return None
+        description = f"{SOURCE_REPORT} result; attempt {run['run_attempt']}"
+        prefix = f"https://github.com/{REPOSITORY}/actions/runs/{run['id']}/job/"
+        if (
+            run.get("conclusion") == "success"
+            and previous
+            and all(
+                status.get("state") == "success"
+                and status.get("description") == description
+                and status.get("creator", {}).get("id") == 41898282
+                and re.fullmatch(
+                    re.escape(prefix) + r"[0-9]+", status.get("target_url", "")
+                )
+                for status in previous
+            )
+        ):
+            jobs.extend(
+                {
+                    "name": status["context"],
+                    "html_url": status["target_url"],
+                    "attempt": run["run_attempt"],
+                }
+                for status in previous
+            )
+            continue
+        if batch is None:
             batch = api(
-                f"repos/{REPOSITORY}/actions/runs/{newer['id']}/jobs?filter=latest&per_page={PAGE_SIZE}"
+                f"repos/{REPOSITORY}/actions/runs/{run['id']}/jobs?filter=latest&per_page={PAGE_SIZE}"
             )
-            if not (
-                newer.get("status") == "completed"
-                and newer.get("conclusion") == "action_required"
-                and batch.get("total_count") == 0
-                and batch.get("jobs") == []
-            ):
-                return None
-        fresh_attempt = api(f"repos/{REPOSITORY}/actions/runs/{selected['id']}")
-        if fresh != selected or fresh_attempt.get("run_attempt") != selected.get(
-            "run_attempt"
+        if (
+            not isinstance(batch.get("total_count"), int)
+            or not 0 <= batch["total_count"] <= PAGE_SIZE
+            or not isinstance(batch.get("jobs"), list)
+            or batch["total_count"] != len(batch["jobs"])
         ):
             return None
+        selected_jobs = batch["jobs"]
         if any(
-            fresh.get(key) != selected.get(key)
-            for key in ("head_sha", "run_attempt", "status", "conclusion")
-        ):
-            return None
-        if selected.get("conclusion") != "success" or any(
             job.get("head_sha") != sha or job.get("status") != "completed"
             for job in selected_jobs
         ):
             return None
+        current_jobs = []
         for context in SOURCE_CHECKS[workflow]:
             matches = [job for job in selected_jobs if job.get("name") == context]
-            if len(matches) != 1 or matches[0].get("conclusion") != "success":
+            if len(matches) != 1:
                 return None
             job = matches[0]
-            if not job.get("html_url", "").startswith(
-                f"https://github.com/{REPOSITORY}/actions/runs/{selected['id']}/job/"
+            if (
+                not isinstance(job.get("id"), int)
+                or job["id"] <= 0
+                or job.get("html_url") != f"{prefix}{job['id']}"
             ):
                 return None
-            jobs.append(job)
+            record = {**job, "attempt": run["run_attempt"]}
+            current_jobs.append(record)
+            if job.get("conclusion") in failures:
+                evidence["failures"].append(record)
+        if run.get("conclusion") != "success" or any(
+            job.get("conclusion") != "success" for job in current_jobs
+        ):
+            return None
+        jobs.extend(current_jobs)
     return jobs
 
 
-def reconcile_source(pr: dict[str, Any]) -> bool:
-    """Expose only verified source results, preserving failures and pending work.
+def reconcile_source(
+    pr: dict[str, Any],
+    statuses: dict[str, Any] | None = None,
+    *,
+    allow_success: bool = True,
+) -> bool:
+    """Report real source outcomes and keep producer races from admission.
 
     Returns:
-        Whether the complete reviewed required source job set has passed.
+        Whether complete required source evidence remained successful and current.
 
     """
-    ref, sha = f"refs/heads/{pr['head']['ref']}", pr["head"]["sha"]
-    if set(SOURCE_CHECKS) != set(WORKFLOWS) or "DCO sign-off" not in SOURCE_CHECKS.get(
-        "dco.yml", []
+    if (
+        set(SOURCE_CHECKS) != set(WORKFLOWS)
+        or "DCO sign-off" not in SOURCE_CHECKS.get("dco.yml", [])
+        or not source_identity(pr)
     ):
         return False
-    if not source_identity(pr):
+    sha, ref = pr["head"]["sha"], f"refs/heads/{pr['head']['ref']}"
+    statuses = read_statuses(sha) if statuses is None else statuses
+    SOURCE_EVIDENCE[sha] = {"runs": {}, "failures": []}
+    initial_event = SOURCE_HEAD == sha and SOURCE_ACTIVITY in {
+        "requested",
+        "in_progress",
+    }
+    if initial_event and not any(
+        status.get("description", "").startswith(SOURCE_REPORT)
+        for status in statuses.values()
+    ):
         return False
-    jobs = source_jobs(pr) if source_contract(pr) is not None else None
-    statuses = read_statuses(sha)
-    results = (
-        [
-            {"context": job["name"], "state": "success", "target_url": job["html_url"]}
-            for job in jobs
-        ]
-        if jobs
-        else []
+    jobs = (
+        source_jobs(pr, statuses)
+        if not initial_event and source_contract(pr) is not None
+        else None
     )
-    for status in statuses.values():
-        if status.get("description", "").startswith(SOURCE_REPORT) and not jobs:
-            results.append(
-                {
-                    "context": status["context"],
-                    "state": "pending",
-                    "target_url": status["target_url"],
-                }
-            )
+    if SOURCE_EVIDENCE[sha]["runs"] and not source_runs_current(pr):
+        source_jobs(pr, statuses)
+        if SOURCE_EVIDENCE[sha]["runs"] and not source_runs_current(pr):
+            SOURCE_EVIDENCE[sha]["failures"] = []
+        jobs = None
+    if not allow_success:
+        jobs = None
     if jobs and source_contract(pr) is None:
-        return False
-    for result in results:
-        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
+        jobs = None
+        SOURCE_EVIDENCE[sha]["failures"] = []
+    failed = SOURCE_EVIDENCE[sha]["failures"]
+    results = {
+        job["name"]: {
+            "context": job["name"],
+            "state": "success",
+            "target_url": job["html_url"],
+            "attempt": job["attempt"],
+        }
+        for job in jobs or []
+    }
+    results.update(
+        {
+            job["name"]: {
+                "context": job["name"],
+                "state": "failure",
+                "target_url": job["html_url"],
+                "attempt": job["attempt"],
+            }
+            for job in failed
+        }
+    )
+    for status in statuses.values():
+        if (
+            status.get("description", "").startswith(SOURCE_REPORT)
+            and not jobs
+            and status["context"] not in results
+        ):
+            results[status["context"]] = {
+                "context": status["context"],
+                "state": "pending",
+                "target_url": status["target_url"],
+                "attempt": 0,
+            }
+    wrote = False
+    for result in results.values():
         previous = statuses.get(result["context"], {})
+        description = f"{SOURCE_REPORT} result; attempt {result['attempt']}"
         if (
             previous.get("state") == result["state"]
             and previous.get("target_url") == result["target_url"]
+            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}",
             "POST",
-            {**result, "description": f"{SOURCE_REPORT} result"},
+            {
+                "context": result["context"],
+                "state": result["state"],
+                "target_url": result["target_url"],
+                "description": description,
+            },
         )
+    if jobs and wrote and (not source_runs_current(pr) or source_contract(pr) is None):
+        reconcile_source(pr, allow_success=False)
+        return False
     return bool(jobs)
 
 
@@ -656,6 +787,8 @@ def main() -> None:
         subprocess.CalledProcessError: An API error other than a deleted ref occurred.
 
     """
+    SOURCE_CACHE.clear()
+    SOURCE_EVIDENCE.clear()
     wait_for_source_run(os.environ.get("SOURCE_RUN_ID", ""))
 
     # Pagination is bounded by GitHub's repository PR limit, not a waiting loop.
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index 0597846..ea23852 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -968,6 +968,7 @@ def setUp(self) -> None:
         self.new_attempt = False
         self.new_run = False
         self.history_reads = 0
+        self.workflow_reads = {}
 
     def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
         if method != "GET":
@@ -985,7 +986,7 @@ def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
             return self.tree
         if "/git/ref/heads/" in path:
             return {"object": {"sha": self.head if self.branch in path else self.base}}
-        if path.endswith("/pulls/7"):
+        if path.endswith(f"/pulls/{self.pr['number']}"):
             return self.pr
         if "/status?" in path or path.endswith("/status"):
             return {"statuses": self.statuses}
@@ -1000,7 +1001,15 @@ def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
                 }
             if f"/workflows/{workflow}/runs?" in path:
                 self.history_reads += 1
-                if self.new_run and self.history_reads == 2:
+                self.workflow_reads[workflow] = self.workflow_reads.get(workflow, 0) + 1
+                if self.new_attempt and self.workflow_reads[workflow] == 2:
+                    self.runs[workflow][0] = {
+                        **self.runs[workflow][0],
+                        "run_attempt": 2,
+                        "status": "in_progress",
+                        "conclusion": None,
+                    }
+                if self.new_run and self.workflow_reads[workflow] == 2:
                     self.runs[workflow].insert(
                         0,
                         {
@@ -1031,6 +1040,8 @@ def _api(self, path: str, method: str = "GET", data: object = None) -> Any:
     def _report(self) -> bool:
         with (
             patch.object(queue, "api", side_effect=self._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)),
         ):
@@ -1049,6 +1060,8 @@ def test_actions_source_admission_requires_explicit_opt_in(self) -> None:
         result = subprocess.CompletedProcess([], 0, '[{"bucket":"pass"}]')
         with (
             patch.object(queue, "api", side_effect=self._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(
@@ -1178,6 +1191,90 @@ def test_failed_new_attempt_invalidates_previous_source_success(self) -> None:
         self.assertFalse(self._report())
         self.assertEqual([data["state"] for _, data in self.writes], ["pending"])
 
+    def _attest(self) -> None:
+        self.statuses = [
+            {
+                "context": job["name"],
+                "state": "success",
+                "target_url": job["html_url"],
+                "description": f"{queue.SOURCE_REPORT} result; attempt 1",
+                "creator": {"id": 41898282},
+            }
+            for jobs in self.jobs.values()
+            for job in jobs
+        ]
+
+    def test_real_failed_new_job_replaces_old_success_with_actual_failure_url(
+        self,
+    ) -> None:
+        self._attest()
+        self.runs["dco.yml"][0].update(run_attempt=2, conclusion="failure")
+        self.jobs[2][0].update(
+            id=25,
+            conclusion="failure",
+            html_url=f"https://github.com/{queue.REPOSITORY}/actions/runs/2/job/25",
+        )
+        self.assertFalse(self._report())
+        actual = {data["context"]: data for _, data in self.writes}
+        self.assertEqual(actual["DCO sign-off"]["state"], "failure")
+        self.assertEqual(
+            actual["DCO sign-off"]["target_url"], self.jobs[2][0]["html_url"]
+        )
+        self.assertEqual(actual["Build"]["state"], "pending")
+
+    def test_producer_rerun_after_collection_invalidates_greens_and_refuses_admission(
+        self,
+    ) -> None:
+        self._attest()
+        self.new_attempt = True
+        with (
+            patch.object(queue, "api", side_effect=self._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, "graphql") as graphql,
+            patch.object(queue.subprocess, "run") as command,
+        ):
+            self.assertFalse(queue.admit(self.pr))
+        self.assertEqual([data["state"] for _, data in self.writes], ["pending"] * 3)
+        graphql.assert_not_called()
+        command.assert_not_called()
+
+    def test_rerun_during_status_publication_is_bounded_and_invalidates_new_greens(
+        self,
+    ) -> None:
+        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}})
+                if len(self.writes) == 1:
+                    self.runs["ci.yml"][0].update(
+                        run_attempt=2, status="in_progress", conclusion=None
+                    )
+            return result
+
+        with (
+            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)),
+        ):
+            self.assertFalse(queue.reconcile_source(self.pr))
+        self.assertEqual(
+            [data["state"] for _, data in self.writes],
+            ["success"] * 3 + ["pending"] * 3,
+        )
+
+    def test_untrusted_cached_status_cannot_replace_actual_job_evidence(self) -> None:
+        self._attest()
+        self.statuses[-1]["creator"]["id"] = 9
+        self.fail_jobs_api = True
+        with self.assertRaises(subprocess.CalledProcessError):
+            self._report()
+        self.assertEqual(self.writes, [])
+
     def test_changed_ref_and_missing_dco_configuration_do_not_report(self) -> None:
         self.head = "d" * 40
         self.assertFalse(self._report())
@@ -1188,5 +1285,162 @@ def test_changed_ref_and_missing_dco_configuration_do_not_report(self) -> None:
         self.assertEqual(self.writes, [])
 
 
+class SourceBudgetTests(unittest.TestCase):
+    """Count API requests for six callers with the actual sixteen-job contract."""
+
+    def _fixtures(self) -> list[PackageSourceTests]:
+        checks = {
+            "ci.yml": [
+                "Build (x86_64-linux)",
+                "Build (aarch64-linux)",
+                "Build (aarch64-darwin)",
+                "Lint and security",
+                "Flake checks (x86_64-linux)",
+                "Flake checks (aarch64-linux)",
+                "Flake checks (aarch64-darwin)",
+                "NUR compatibility / NUR (locked)",
+                "NUR compatibility / NUR (unstable)",
+                "Flake lock health",
+                "Build (x86_64-linux 2/2)",
+                "Build (aarch64-linux 2/2)",
+            ],
+            "codeql.yml": ["Analyze (python) / analyze"],
+            "dependency-review.yml": ["dependency-review / review"],
+            "dco.yml": ["DCO sign-off"],
+            "fuzz.yml": ["Fuzz font path validation"],
+        }
+        fixtures = []
+        for number in range(7, 13):
+            fixture = PackageSourceTests()
+            fixture.setUp()
+            fixture.checks = checks
+            fixture.sha = f"{number:040x}"
+            fixture.head = fixture.sha
+            fixture.branch = f"automation/package-updates-example{number}"
+            fixture.pr["number"] = number
+            fixture.pr["head"].update(sha=fixture.sha, ref=fixture.branch)
+            fixture.paths[0]["filename"] = f"pkgs/by-name/ex/example{number}/source.nix"
+            fixture.tree["tree"] = [
+                {
+                    "path": f".github/workflows/{workflow}",
+                    "sha": "c" * 40,
+                    "mode": "100644",
+                    "type": "blob",
+                }
+                for workflow in checks
+            ]
+            fixture.runs, fixture.jobs = {}, {}
+            for workflow_id, workflow in enumerate(checks, 1):
+                run_id = number * 10 + workflow_id
+                fixture.runs[workflow] = [
+                    {
+                        "id": run_id,
+                        "workflow_id": workflow_id,
+                        "path": f".github/workflows/{workflow}",
+                        "head_sha": fixture.sha,
+                        "head_branch": fixture.branch,
+                        "event": "workflow_dispatch",
+                        "repository": {"full_name": queue.REPOSITORY},
+                        "head_repository": {"full_name": queue.REPOSITORY},
+                        "status": "completed",
+                        "conclusion": "success",
+                        "run_attempt": 1,
+                    }
+                ]
+                fixture.jobs[run_id] = [
+                    {
+                        "id": run_id * 100 + index,
+                        "name": name,
+                        "head_sha": fixture.sha,
+                        "status": "completed",
+                        "conclusion": "success",
+                        "html_url": f"https://github.com/{queue.REPOSITORY}/actions/runs/{run_id}/job/{run_id * 100 + index}",
+                    }
+                    for index, name in enumerate(checks[workflow])
+                ]
+            fixture._attest()
+            fixtures.append(fixture)
+        return fixtures
+
+    def _case(self, mode: str) -> dict[str, int]:
+        fixtures = self._fixtures()
+        counts = {"get": 0, "write": 0, "jobs": 0}
+        if mode in {"queued", "admitted_event"}:
+            for fixture in fixtures:
+                fixture.statuses.append(
+                    {"context": queue.ADMISSION, "state": "success"}
+                )
+        if mode == "initial_event":
+            fixtures[0].statuses = []
+
+        def api(path: str, method: str = "GET", data: object = None) -> Any:
+            counts["get" if method == "GET" else "write"] += 1
+            if "/jobs?" in path:
+                counts["jobs"] += 1
+            for fixture in fixtures:
+                if (
+                    fixture.sha in path
+                    or fixture.branch in path
+                    or path.endswith(f"/pulls/{fixture.pr['number']}")
+                    or f"/pulls/{fixture.pr['number']}/files?" in path
+                ):
+                    return fixture._api(path, method, data)
+                for run_id in fixture.jobs:
+                    if f"/runs/{run_id}/" in path or path.endswith(f"/runs/{run_id}"):
+                        return fixture._api(path, method, data)
+            return fixtures[0]._api(path, method, data)
+
+        def graphql(query: str, **variables: str) -> dict[str, Any]:
+            counts["get"] += 1
+            return {"node": {"mergeQueueEntry": {"id": "queue"}}}
+
+        head = fixtures[0].sha if mode.endswith("event") else ""
+        activity = (
+            "in_progress"
+            if mode in {"admitted_event", "initial_event"}
+            else "completed"
+        )
+        with (
+            patch.object(queue, "api", side_effect=api),
+            patch.object(queue, "graphql", side_effect=graphql),
+            patch.object(queue, "SOURCE_CACHE", {}),
+            patch.object(queue, "SOURCE_EVIDENCE", {}),
+            patch.object(queue, "SOURCE_CHECKS", fixtures[0].checks),
+            patch.object(queue, "WORKFLOWS", list(fixtures[0].checks)),
+            patch.object(queue, "SOURCE_HEAD", head),
+            patch.object(queue, "SOURCE_ACTIVITY", activity),
+        ):
+            for fixture in fixtures:
+                if mode == "idle":
+                    self.assertTrue(queue.reconcile_source(fixture.pr))
+                else:
+                    queue.admit(fixture.pr)
+        return counts
+
+    def test_six_idle_source_prs_fit_hourly_recovery_budget(self) -> None:
+        counts = self._case("idle")
+        self.assertLessEqual(counts["get"], 110)
+        self.assertEqual(counts["write"], 0)
+        self.assertEqual(counts["jobs"], 0)
+
+    def test_six_admitted_sources_skip_unrelated_polling(self) -> None:
+        self.assertEqual(self._case("queued"), {"get": 12, "write": 0, "jobs": 0})
+
+    def test_source_event_inspects_only_its_exact_head(self) -> None:
+        counts = self._case("completed_event")
+        self.assertLessEqual(counts["get"], 25)
+        self.assertEqual(counts["write"], 0)
+        self.assertEqual(counts["jobs"], 0)
+
+    def test_rerun_event_invalidates_even_an_admitted_source(self) -> None:
+        counts = self._case("admitted_event")
+        self.assertEqual(counts, {"get": 33, "write": 16, "jobs": 0})
+
+    def test_initial_requested_event_with_no_attestation_stops_immediately(
+        self,
+    ) -> None:
+        self.assertEqual(self._case("initial_event"), {"get": 1, "write": 0, "jobs": 0})
+
+
 if __name__ == "__main__":
     unittest.main()

From dc803bb28ea7ae9f2c576d2d03bb78a7f497e72b Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:14:55 -0700
Subject: [PATCH 3/7] fix(queue): bind source jobs to their exact run attempt

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

diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py
index d8cd552..2fc8752 100644
--- a/scripts/reconcile-queue.py
+++ b/scripts/reconcile-queue.py
@@ -641,7 +641,10 @@ def source_jobs(
             return None
         selected_jobs = batch["jobs"]
         if any(
-            job.get("head_sha") != sha or job.get("status") != "completed"
+            job.get("head_sha") != sha
+            or job.get("status") != "completed"
+            or job.get("run_id") != run["id"]
+            or job.get("run_attempt") != run["run_attempt"]
             for job in selected_jobs
         ):
             return None
diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index ea23852..5c7500b 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -934,6 +934,8 @@ def setUp(self) -> None:
             self.jobs[index] = [
                 {
                     "id": index * 10 + number,
+                    "run_id": index,
+                    "run_attempt": 1,
                     "name": name,
                     "head_sha": self.sha,
                     "status": "completed",
@@ -1152,6 +1154,10 @@ def test_real_failed_pending_skipped_missing_and_wrong_head_jobs_never_pass(
                     self.new_attempt = True
                 if invalid == "new_run":
                     self.new_run = True
+                if invalid == "job_attempt":
+                    self.jobs[2][0]["run_attempt"] = 2
+                if invalid == "job_run":
+                    self.jobs[2][0]["run_id"] = 9
                 self.assertFalse(self._report())
                 self.assertEqual(self.writes, [])
 
@@ -1211,6 +1217,7 @@ def test_real_failed_new_job_replaces_old_success_with_actual_failure_url(
         self.runs["dco.yml"][0].update(run_attempt=2, conclusion="failure")
         self.jobs[2][0].update(
             id=25,
+            run_attempt=2,
             conclusion="failure",
             html_url=f"https://github.com/{queue.REPOSITORY}/actions/runs/2/job/25",
         )
@@ -1350,6 +1357,8 @@ def _fixtures(self) -> list[PackageSourceTests]:
                 fixture.jobs[run_id] = [
                     {
                         "id": run_id * 100 + index,
+                        "run_id": run_id,
+                        "run_attempt": 1,
                         "name": name,
                         "head_sha": fixture.sha,
                         "status": "completed",

From 16505e21727ab06ae091bfa6bbaa6f140244c262 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:24:25 -0700
Subject: [PATCH 4/7] docs(release): describe verified package source
 automation

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 docs/releases/v2.11.0.md | 41 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 41 insertions(+)
 create mode 100644 docs/releases/v2.11.0.md

diff --git a/docs/releases/v2.11.0.md b/docs/releases/v2.11.0.md
new file mode 100644
index 0000000..95c2bb6
--- /dev/null
+++ b/docs/releases/v2.11.0.md
@@ -0,0 +1,41 @@
+# Nix Forge CI v2.11.0
+
+## Changelog
+
+- Add opt-in reporting and normal queue admission for verified package source PRs
+  whose explicitly dispatched validation is missing from GitHub's PR rollup.
+- Prove the exact GitHub Actions bot identity, same-repository package branch,
+  actual parent ancestry, selected package/catalog paths, and current protected
+  workflow blobs before trusting source validation. No PR code is executed.
+- Report actual failed required jobs with their exact job URLs. Missing, running,
+  skipped, stale, or untrusted evidence never becomes success. Producer changes
+  during collection or publication invalidate old results and refuse admission.
+- Reuse verified unchanged-run attestations and cache only immutable Git data and
+  stable workflow identity within one reconciliation. Latest-run state remains
+  live; exact-head events avoid polling unrelated source PRs.
+
+Existing callers keep source reporting disabled and retain their current queue
+behavior. Update general shared references coherently to the v2.11.0 release
+commit. Repository checks and SLSA builders retain separately reviewed pins.
+The package caller opts in through `package-source-checks`, mapping every
+configured workflow to all of its protected required job names, including actual
+source `DCO sign-off`. Add `requested`, `in_progress`, and `completed` workflow-run
+notifications and retain hourly scheduled recovery. Regenerate existing source
+branches from reviewed protected main so their workflow blobs match the caller.
+
+The change adds no token permissions and passes no privileged token to PR builds.
+Expected-head queue admission and full merge-group validation remain required.
+Regression coverage includes source identity and ownership, workflow drift,
+approval runs with zero jobs, actual job run/attempt provenance, failure URLs,
+producer races, stale-result invalidation, and six-PR request budgets. All 133
+repository tests and offline workflow/security checks passed locally; required
+hosted PR and merge-group validation and representative caller checks must pass
+before the release tag is published.
+
+The functional source was reviewed at
+`dc803bb28ea7ae9f2c576d2d03bb78a7f497e72b`. The signed tag and release manifest identify
+the final reviewed source commit including these notes. Publication uses
+`nix-forge/ci/.github/workflows/release.yml`, which verifies the archive checksum,
+all provenance subjects, exact pinned builder identity, and protected-main tag
+ancestry. Restore the v2.10.0 pin and remove the source opt-in for rollback.
+No support end date is scheduled for compatible v2.x consumers.

From 9599b9b621a2d85994d3e5a1a718c91dd2e84d8f Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:25:32 -0700
Subject: [PATCH 5/7] docs(release): scope the source execution guarantee

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

diff --git a/docs/releases/v2.11.0.md b/docs/releases/v2.11.0.md
index 95c2bb6..1272d36 100644
--- a/docs/releases/v2.11.0.md
+++ b/docs/releases/v2.11.0.md
@@ -6,7 +6,7 @@
   whose explicitly dispatched validation is missing from GitHub's PR rollup.
 - Prove the exact GitHub Actions bot identity, same-repository package branch,
   actual parent ancestry, selected package/catalog paths, and current protected
-  workflow blobs before trusting source validation. No PR code is executed.
+  workflow blobs before trusting source validation. The reconciler executes no PR code.
 - Report actual failed required jobs with their exact job URLs. Missing, running,
   skipped, stale, or untrusted evidence never becomes success. Producer changes
   during collection or publication invalidate old results and refuse admission.

From 34b59014c18efb8401cbea90d6b5e98bb53c58f5 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:27:44 -0700
Subject: [PATCH 6/7] docs(site): index the package automation release notes

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 site/mkdocs.yml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/site/mkdocs.yml b/site/mkdocs.yml
index 74fe4c1..59324ff 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.11.0 release notes: releases/v2.11.0.md
   - v2.10.0 release notes: releases/v2.10.0.md
   - v2.9.0 release notes: releases/v2.9.0.md
   - v2.8.0 release notes: releases/v2.8.0.md

From f1395bba39d0d0aeb689f5faacea8d7a2398eca4 Mon Sep 17 00:00:00 2001
From: IanHollow <72767437+IanHollow@users.noreply.github.com>
Date: Mon, 12 Oct 2026 16:35:37 -0700
Subject: [PATCH 7/7] test(queue): enumerate source run provenance threats

Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com>
---
 tests/test_reconcile_queue.py | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py
index 5c7500b..e96779a 100644
--- a/tests/test_reconcile_queue.py
+++ b/tests/test_reconcile_queue.py
@@ -1135,6 +1135,9 @@ def test_real_failed_pending_skipped_missing_and_wrong_head_jobs_never_pass(
             "head",
             "url",
             "attempt",
+            "new_run",
+            "job_attempt",
+            "job_run",
         ):
             with self.subTest(invalid=invalid):
                 self.setUp()