From 3adc64070665935f8f7da0b9a2c714e57c5ad099 Mon Sep 17 00:00:00 2001 From: IanHollow <72767437+IanHollow@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:13:14 -0700 Subject: [PATCH 1/5] Recover source validation callbacks and asynchronous queue advance Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com> --- README.md | 32 ++- actions/queue-completion/action.yml | 14 +- actions/reconcile-queue/action.yml | 10 + scripts/reconcile-queue.py | 224 ++++++++++++++-- tests/test_reconcile_queue.py | 370 ++++++++++++++++++++++++++- tests/test_source_callback_action.py | 80 ++++++ 6 files changed, 707 insertions(+), 23 deletions(-) create mode 100644 tests/test_source_callback_action.py diff --git a/README.md b/README.md index 239a8ef..1387ba7 100644 --- a/README.md +++ b/README.md @@ -105,6 +105,29 @@ mirrors skipped validation as success. `actions/queue-completion` notifies this trusted reconciler from dispatched queue runs. Retain the caller's existing `Queue completion callback` name and event guard. +Package-source callers can also notify the reconciler when validation starts and +finishes. Add optional `source_run_attempt` and `source_activity` dispatch inputs +to the default-branch reconciler, forwarding them to `source-run-attempt` and +`source-activity` on `actions/reconcile-queue`. Source callback jobs use +`actions/queue-completion` with `source-activity: in_progress` at start and +`source-activity: completed` after all validation jobs. Restrict these jobs to +same-repository dispatched `automation/package-updates-` branches and retain +only their existing `actions: write` permission. Queue callbacks can omit the +new input and keep their original payload. + +The reconciler validates the producer run's repository, workflow identity, +path, branch, head and attempt before using its metadata. Start notifications +invalidate previous mirrored jobs and admission without waiting for builds; +completion notifications wait briefly for their own producer to finish and +revalidate its actual attempt. A delayed start for a completed attempt reads +the completed results. An already queued source PR cannot bypass this check. +Statuses never replace the required native jobs or relax branch protection. +After publishing new complete queue success, reconciliation waits at most 55 +seconds for GitHub's asynchronous merge to expose a different live front ref. +It verifies that ref and dispatches its missing validation through the existing +handler once. Running or failed candidates do not start this additional poll; +there is no recursive recovery dispatch. + The actions read the default branch from GitHub's repository event metadata. Use that same default branch in the caller's callback guard. A package updater can opt in to admission with `automation-branch`; it accepts only that exact @@ -128,8 +151,8 @@ 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. 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 +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 @@ -173,8 +196,9 @@ release. Do not provide privileged tokens to PR build steps. `actions/discover-lockfiles` emits a JSON `lockfiles` output for a workflow matrix. It requires a tracked `flake.lock` for each tracked `flake.nix` in the caller repository. Deleting only the lockfile fails discovery; deleting the entire -partition removes its matrix entry. Added, renamed and deleted partitions change that inventory -without edits to the workflow. Submodules own their lockfile health jobs. +partition removes its matrix entry. Added, renamed and deleted partitions change +that inventory without edits to the workflow. Submodules own their lockfile +health jobs. Require a stable aggregate job that waits for the complete matrix and fails for any failed, cancelled or skipped matrix. Protect that aggregate instead of individual lockfile names, so adding coverage cannot be bypassed and deleting a diff --git a/actions/queue-completion/action.yml b/actions/queue-completion/action.yml index 944e9e3..c4efb65 100644 --- a/actions/queue-completion/action.yml +++ b/actions/queue-completion/action.yml @@ -1,5 +1,9 @@ name: Notify queue completion description: Notify the trusted default-branch reconciler after manually dispatched validation. +inputs: + source-activity: + description: Optional package-source notification phase, in_progress or completed. Requires matching reconciler dispatch inputs. + default: '' runs: using: composite steps: @@ -9,9 +13,17 @@ runs: GH_TOKEN: ${{ github.token }} GH_REPO: ${{ github.repository }} DEFAULT_BRANCH: ${{ github.event.repository.default_branch }} + SOURCE_ACTIVITY: ${{ inputs.source-activity }} run: | + args=(-f 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") ;; + *) 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" -f source_run_id="$GITHUB_RUN_ID"; then + if gh workflow run reconcile-merge-queue.yml --ref "$DEFAULT_BRANCH" "${args[@]}"; then exit 0 fi echo "::warning::Queue completion notification attempt $attempt failed" diff --git a/actions/reconcile-queue/action.yml b/actions/reconcile-queue/action.yml index 6d94fba..c6a2ae9 100644 --- a/actions/reconcile-queue/action.yml +++ b/actions/reconcile-queue/action.yml @@ -16,6 +16,12 @@ inputs: source-run-id: description: Optional dispatch completion notification. default: '' + source-run-attempt: + description: Producer run attempt for a package-source start or completion callback. + default: '' + source-activity: + description: Package-source callback phase, in_progress or completed. + default: '' runs: using: composite steps: @@ -27,6 +33,8 @@ runs: BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }} QUEUE_WORKFLOWS: ${{ inputs.workflows }} SOURCE_RUN_ID: ${{ inputs.source-run-id }} + SOURCE_RUN_ATTEMPT: ${{ inputs.source-run-attempt }} + SOURCE_NOTIFICATION_ACTIVITY: ${{ inputs.source-activity }} SOURCE_HEAD_SHA: ${{ github.event.workflow_run.head_sha }} SOURCE_ACTIVITY: ${{ github.event.action }} run: | @@ -90,6 +98,8 @@ runs: BASE_SHA_WORKFLOWS: ${{ inputs.base-sha-workflows }} QUEUE_WORKFLOWS: ${{ inputs.workflows }} SOURCE_RUN_ID: ${{ inputs.source-run-id }} + SOURCE_RUN_ATTEMPT: ${{ inputs.source-run-attempt }} + SOURCE_NOTIFICATION_ACTIVITY: ${{ inputs.source-activity }} 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 2fc8752..cdfc055 100644 --- a/scripts/reconcile-queue.py +++ b/scripts/reconcile-queue.py @@ -98,6 +98,28 @@ 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() + ): + fresh = api(f"repos/{REPOSITORY}/pulls/{pr['number']}") + current = api( + f"repos/{REPOSITORY}/git/ref/heads/{quote(pr['head']['ref'], safe='/')}" + ) + 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 trusted = ( source @@ -138,6 +160,26 @@ def admit(pr: dict[str, Any]) -> bool: id=pr["node_id"], )["node"] if state["mergeQueueEntry"]: + if ( + source + and SOURCE_HEAD == sha + 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: + reconcile_source(pr, allow_success=False) + return False + api( + f"repos/{REPOSITORY}/statuses/{sha}", + "POST", + { + "state": "success", + "context": ADMISSION, + "description": "This PR head remains queued with current successful source jobs", + }, + ) return True # Once admitted, a rejected head needs a fix or deliberate requeue. # Never keep rebuilding the same failing merge group indefinitely. @@ -389,6 +431,89 @@ def wait_for_source_run(run_id: str) -> None: print(f"Source run {run_id} is still running; reconcile other ready entries") +def source_notification( + run_id: str, attempt: str, activity: str +) -> tuple[str, str] | None: + """Bind an explicit source callback to its real current producer identity. + + Returns: + Exact producer head and current activity, or None for legacy queue callbacks. + + Raises: + ValueError: Source notification metadata is missing, tampered or stale. + TypeError: Producer workflow path is not a string. + + """ + 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}") + branch = run.get("head_branch", "") + if not isinstance(branch, str) or not branch.startswith( + "automation/package-updates-" + ): + if activity or attempt: + raise ValueError( + "Explicit source callback requires a package updater branch" + ) + return None + if ( + activity not in {"in_progress", "completed"} + or not attempt.isascii() + or not attempt.isdecimal() + or int(attempt) <= 0 + or re.fullmatch(r"automation/package-updates-[a-z0-9][a-z0-9+._-]*", branch) + is None + ): + raise ValueError("Invalid source callback activity, attempt or branch") + path = run.get("path", "") + if not isinstance(path, str): + raise TypeError("Source callback workflow path is invalid") + workflow = path.removeprefix(".github/workflows/") + if workflow not in WORKFLOWS or workflow not in SOURCE_CHECKS: + raise ValueError("Source callback workflow is outside the required contract") + metadata = source_read(f"repos/{REPOSITORY}/actions/workflows/{workflow}") + + def valid(producer: dict[str, Any]) -> bool: + return ( + type(producer.get("id")) is int + and producer["id"] == int(run_id) + and type(producer.get("workflow_id")) is int + and type(metadata.get("id")) is int + and producer["workflow_id"] == metadata.get("id") + and metadata.get("state") == "active" + and metadata.get("path") == f".github/workflows/{workflow}" + and producer.get("path") == metadata.get("path") + and producer.get("repository", {}).get("full_name") == REPOSITORY + and producer.get("head_repository", {}).get("full_name") == REPOSITORY + and producer.get("event") == "workflow_dispatch" + and producer.get("head_branch") == branch + and isinstance(producer.get("head_sha"), str) + and re.fullmatch(r"[0-9a-f]{40}", producer["head_sha"]) is not None + and type(producer.get("run_attempt")) is int + and producer["run_attempt"] == int(attempt) + and producer.get("status") + in {"queued", "in_progress", "completed", "waiting", "pending", "requested"} + ) + + if not valid(run): + raise ValueError("Source callback producer identity does not match") + sha = run["head_sha"] + if activity == "completed": + wait_for_source_run(run_id) + run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}") + if not valid(run) or run["head_sha"] != sha: + raise ValueError( + "Source callback producer changed while awaiting completion" + ) + ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}") + if ref.get("object", {}).get("sha") != sha: + raise ValueError("Source callback branch advanced beyond its producer") + # A delayed start callback must not invalidate an already completed attempt. + return sha, "completed" if run["status"] == "completed" else "in_progress" + + def front_queue_entry() -> dict[str, Any] | None: """Read the live front entry without assuming the queue ref naming scheme. @@ -790,9 +915,19 @@ def main() -> None: subprocess.CalledProcessError: An API error other than a deleted ref occurred. """ + global SOURCE_HEAD, SOURCE_ACTIVITY SOURCE_CACHE.clear() SOURCE_EVIDENCE.clear() - wait_for_source_run(os.environ.get("SOURCE_RUN_ID", "")) + run_id = os.environ.get("SOURCE_RUN_ID", "") + notification = source_notification( + run_id, + os.environ.get("SOURCE_RUN_ATTEMPT", ""), + os.environ.get("SOURCE_NOTIFICATION_ACTIVITY", ""), + ) + if notification is None: + wait_for_source_run(run_id) + else: + SOURCE_HEAD, SOURCE_ACTIVITY = notification # Pagination is bounded by GitHub's repository PR limit, not a waiting loop. queue_expected = False @@ -835,21 +970,60 @@ def main() -> None: ) break time.sleep(5) + published_passes = set() for entry in refs: try: - dispatch_entry(entry) + if dispatch_entry(entry): + published_passes.add(entry["object"]["sha"]) except subprocess.CalledProcessError as error: # ruff: ignore[try-except-in-loop] # Each ref can disappear independently while GitHub advances the queue. if "HTTP 404" not in (error.stderr or ""): raise print(f"Queue ref disappeared: {entry['ref']}") + if published_passes: + advance_after_publication(published_passes) -def dispatch_entry(entry: dict[str, Any]) -> None: - """Start missing workflows for one queue ref if it still exists.""" +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): + front = front_queue_entry() + head = front.get("headCommit") if front else None + if head and head.get("oid") not in published: + refs = api( + f"repos/{REPOSITORY}/git/matching-refs/{QUEUE_PREFIX.removeprefix('refs/')}" + ) + matching = [ + entry + for entry in refs + if entry["ref"].startswith(QUEUE_PREFIX) + and entry["object"]["sha"] == head["oid"] + ] + if len(matching) == 1: + try: + dispatch_entry(matching[0]) + except subprocess.CalledProcessError as error: + if "HTTP 404" not in (error.stderr or ""): + raise + print("Next queue ref disappeared before validation dispatch") + return + if attempt < 11: + time.sleep(5) + print( + "::warning::No new verified front ref after publishing success; recovery discovery is bounded" + ) + + +def dispatch_entry(entry: dict[str, Any]) -> bool: + """Start missing queue workflows and report whether success was published. + + Returns: + Whether fresh passing results justify one bounded next-front discovery. + + """ ref, sha = entry["ref"], entry["object"]["sha"] if not ref.startswith(QUEUE_PREFIX): - return + return False base = api(f"repos/{REPOSITORY}/git/commits/{sha}")["parents"][0]["sha"] validation_runs: list[dict[str, Any]] = [] for workflow in WORKFLOWS: @@ -877,7 +1051,7 @@ def dispatch_entry(entry: dict[str, Any]) -> None: payload, ) print(f"Dispatched {workflow} for {sha}") - publish_dispatch_results(ref, sha, validation_runs) + return publish_dispatch_results(ref, sha, validation_runs) def read_statuses(sha: str) -> dict[str, Any]: @@ -973,11 +1147,16 @@ def invalidate_results( ) -def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> None: - """Expose verified dispatch results to the merge queue's status evaluator.""" +def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> bool: + """Expose verified dispatch results and identify newly published complete passes. + + Returns: + Whether new actual success results may let GitHub advance the live queue. + + """ dispatched = [run for run in runs if run["event"] == "workflow_dispatch"] if not dispatched: - return + return False latest = read_statuses(sha) all_passed = workflow_set_passed(sha, runs) failures = {"failure", "timed_out", "cancelled"} @@ -997,7 +1176,7 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> jobs = latest_dispatch_jobs(sha, evidence) if evidence else None if not jobs: invalidate_results(ref, sha, latest) - return + return False results = { job["name"]: "success" if job["conclusion"] == "success" else "failure" for job in jobs @@ -1005,10 +1184,11 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> or (all_passed and job["conclusion"] == "success") } invalidate_results(ref, sha, latest, set(results)) + published_success = False for job in jobs: # Skipped and neutral jobs provide no passing or failing evidence. if job["name"] in results: - publish_status( + changed = publish_status( ref, sha, latest, @@ -1018,6 +1198,14 @@ def publish_dispatch_results(ref: str, sha: str, runs: list[dict[str, Any]]) -> "target_url": job["html_url"], }, ) + published_success |= changed and results[job["name"]] == "success" + source_required = {name for names in SOURCE_CHECKS.values() for name in names} + return ( + published_success + and all_passed + and all(state == "success" for state in results.values()) + and source_required.issubset(results) + ) def publish_status( @@ -1025,15 +1213,20 @@ def publish_status( sha: str, latest: dict[str, Any], result: dict[str, str], -) -> None: - """Publish one actual job result once, while its queue ref still matches.""" +) -> bool: + """Publish one actual job result once, while its queue ref still matches. + + Returns: + Whether this exact current queue commit received a new result. + + """ name, state, target_url = result["context"], result["state"], result["target_url"] previous = latest.get(name, {}) if previous.get("state") == state and previous.get("target_url") == target_url: - return + return False current = api(f"repos/{REPOSITORY}/git/ref/{ref.removeprefix('refs/')}") if current["object"]["sha"] != sha: - return + return False api( f"repos/{REPOSITORY}/statuses/{sha}", "POST", @@ -1047,6 +1240,7 @@ def publish_status( }, ) print(f"Reported {name}: {state} for {sha}") + return True if __name__ == "__main__": diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py index e96779a..3d3da76 100644 --- a/tests/test_reconcile_queue.py +++ b/tests/test_reconcile_queue.py @@ -341,7 +341,7 @@ def test_queue_removed_during_discovery_does_not_dispatch_stale_refs(self) -> No patch.object(queue, "api", side_effect=self.api), patch.object(queue, "admit", return_value=True), patch.object(queue, "front_queue_entry", return_value=None), - patch.object(queue, "dispatch_entry") as dispatch, + patch.object(queue, "dispatch_entry", return_value=False) as dispatch, patch.object(queue.time, "sleep") as sleep, ): queue.main() @@ -370,7 +370,7 @@ def delayed_api(path: str, method: str = "GET", data: object = None) -> Any: "front_queue_entry", return_value={"headCommit": {"oid": "group"}}, ), - patch.object(queue, "dispatch_entry") as dispatch, + patch.object(queue, "dispatch_entry", return_value=False) as dispatch, patch.object(queue.time, "sleep") as sleep, ): queue.main() @@ -827,6 +827,36 @@ def test_separate_newer_run_invalidates_previous_success(self) -> None: self.report() self.assertEqual([item["state"] for item in self.writes], ["pending"]) + def test_only_complete_new_success_can_start_advance_poll(self) -> None: + with patch.object(queue, "api", side_effect=self.api): + self.assertTrue( + queue.publish_dispatch_results( + self.ref, "group", [self.dispatch_run, self.other] + ) + ) + self.dispatch_run["conclusion"] = "failure" + self.current = self.dispatch_run.copy() + self.jobs[0]["conclusion"] = "failure" + with patch.object(queue, "api", side_effect=self.api): + self.assertFalse( + queue.publish_dispatch_results( + self.ref, "group", [self.dispatch_run, self.other] + ) + ) + + def test_missing_protected_source_job_cannot_start_advance_poll(self) -> None: + with ( + patch.object(queue, "api", side_effect=self.api), + patch.object( + queue, "SOURCE_CHECKS", {"ci.yml": ["Lint", "Absent required build"]} + ), + ): + self.assertFalse( + queue.publish_dispatch_results( + self.ref, "group", [self.dispatch_run, self.other] + ) + ) + def test_existing_identical_result_is_not_republished(self) -> None: """Repeated recovery calls reuse the same result.""" self.statuses = [ @@ -1446,7 +1476,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": 33, "write": 16, "jobs": 0}) + self.assertEqual(counts, {"get": 35, "write": 17, "jobs": 0}) def test_initial_requested_event_with_no_attestation_stops_immediately( self, @@ -1454,5 +1484,339 @@ def test_initial_requested_event_with_no_attestation_stops_immediately( self.assertEqual(self._case("initial_event"), {"get": 1, "write": 0, "jobs": 0}) +class SourceNotificationTests(unittest.TestCase): + """Bind token-created notifications to actual immutable producer metadata.""" + + def setUp(self) -> None: + self.fixture = SourceBudgetTests()._fixtures()[0] + self.fixture.statuses.append({"context": queue.ADMISSION, "state": "success"}) + self.run = self.fixture.runs["ci.yml"][0] + self.run_id = str(self.run["id"]) + + def notification(self, activity: str, attempt: str = "1") -> tuple[str, str] | None: + with ( + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + ): + return queue.source_notification(self.run_id, attempt, activity) + + def test_start_does_not_wait_for_native_validation(self) -> None: + self.run.update(status="in_progress", conclusion=None, run_attempt=2) + with patch.object(queue, "wait_for_source_run") as wait: + result = self.notification("in_progress", "2") + self.assertEqual(result, (self.fixture.sha, "in_progress")) + wait.assert_not_called() + + def test_completion_waits_for_own_producer_then_revalidates(self) -> None: + self.run.update(status="in_progress", conclusion=None) + + def finish(run_id: str) -> None: + self.assertEqual(run_id, self.run_id) + self.run.update(status="completed", conclusion="success") + + with patch.object(queue, "wait_for_source_run", side_effect=finish) as wait: + result = self.notification("completed") + self.assertEqual(result, (self.fixture.sha, "completed")) + wait.assert_called_once_with(self.run_id) + + def test_delayed_start_revalidates_completed_attempt(self) -> None: + with patch.object(queue, "wait_for_source_run") as wait: + result = self.notification("in_progress") + self.assertEqual(result, (self.fixture.sha, "completed")) + wait.assert_not_called() + + def test_new_attempt_during_completion_cannot_reuse_old_identity(self) -> None: + def rerun(_run_id: str) -> None: + self.run.update(status="in_progress", conclusion=None, run_attempt=2) + + with ( + patch.object(queue, "wait_for_source_run", side_effect=rerun), + self.assertRaises(ValueError), + ): + self.notification("completed") + self.assertEqual(self.fixture.writes, []) + + def test_tampered_producer_fields_never_bind(self) -> None: + updates = [ + {"repository": {"full_name": "other/repository"}}, + {"head_repository": {"full_name": "other/repository"}}, + {"workflow_id": 1234}, + {"path": ".github/workflows/unapproved.yml"}, + {"event": "push"}, + {"head_sha": "not-a-sha"}, + {"head_branch": "automation/package-updates-../other"}, + {"run_attempt": 2}, + {"workflow_id": True}, + ] + for change in updates: + with self.subTest(change=change): + original = self.run.copy() + self.run.update(change) + with self.assertRaises(ValueError): + self.notification("in_progress") + self.run.clear() + self.run.update(original) + self.assertEqual(self.fixture.writes, []) + + def test_missing_phase_or_attempt_is_not_arbitrary_source_dispatch(self) -> None: + for activity, attempt in [("", "1"), ("completed", ""), ("success", "1")]: + with ( + self.subTest(activity=activity, attempt=attempt), + self.assertRaises(ValueError), + ): + self.notification(activity, attempt) + + def test_current_branch_must_still_name_producer_head(self) -> None: + self.fixture.head = "d" * 40 + with self.assertRaises(ValueError): + self.notification("in_progress") + + def test_legacy_queue_callback_keeps_original_contract(self) -> None: + self.run["head_branch"] = "gh-readonly-queue/main/pr-7-example" + with patch.object(queue, "wait_for_source_run") as wait: + self.assertIsNone(self.notification("", "")) + wait.assert_not_called() + + def admit(self, activity: str) -> bool: + with ( + patch.object(queue, "api", side_effect=self.fixture._api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_HEAD", self.fixture.sha), + patch.object(queue, "SOURCE_ACTIVITY", activity), + patch.object( + queue, + "graphql", + return_value={"node": {"mergeQueueEntry": {"id": "existing"}}}, + ), + ): + return queue.admit(self.fixture.pr) + + def test_admitted_rerun_start_invalidates_sixteen_jobs_and_admission(self) -> None: + self.run.update(status="in_progress", conclusion=None, run_attempt=2) + self.assertFalse(self.admit("in_progress")) + values = [data for _path, data in self.fixture.writes] + self.assertEqual(len(values), 17) + self.assertEqual({item["state"] for item in values}, {"pending"}) + self.assertEqual( + {item["context"] for item in values}, + {queue.ADMISSION} + | {name for names in self.fixture.checks.values() for name in names}, + ) + + def test_admitted_completion_reports_failed_real_job(self) -> None: + self.run["conclusion"] = "failure" + self.fixture.jobs[self.run["id"]][0]["conclusion"] = "failure" + self.assertFalse(self.admit("completed")) + values = [data for _path, data in self.fixture.writes] + failures = [item for item in values if item["state"] == "failure"] + self.assertEqual(len(failures), 1) + self.assertEqual( + failures[0]["target_url"], self.fixture.jobs[self.run["id"]][0]["html_url"] + ) + self.assertFalse(any(item["state"] == "success" for item in values)) + + def test_queued_completion_restores_only_after_all_actual_jobs(self) -> None: + for previous in self.fixture.statuses: + previous["state"] = "pending" + self.assertTrue(self.admit("completed")) + admission = [ + data + for _path, data in self.fixture.writes + if data["context"] == queue.ADMISSION + ] + self.assertEqual([value["state"] for value in admission], ["success"]) + self.assertEqual( + sum(data["state"] == "success" for _path, data in self.fixture.writes), 17 + ) + + def test_main_binds_start_before_admitted_source_shortcut(self) -> None: + self.run.update(status="in_progress", conclusion=None, run_attempt=2) + + def api(path: str, method: str = "GET", data: object = None) -> Any: + if "/pulls?" in path: + return [self.fixture.pr] + if "/git/matching-refs/" in path: + return [] + return self.fixture._api(path, method, data) + + with ( + patch.dict( + os.environ, + { + "SOURCE_RUN_ID": self.run_id, + "SOURCE_RUN_ATTEMPT": "2", + "SOURCE_NOTIFICATION_ACTIVITY": "in_progress", + }, + ), + patch.object(queue, "api", side_effect=api), + patch.object(queue, "SOURCE_CACHE", {}), + patch.object(queue, "SOURCE_EVIDENCE", {}), + patch.object(queue, "SOURCE_CHECKS", self.fixture.checks), + patch.object(queue, "WORKFLOWS", list(self.fixture.checks)), + patch.object(queue, "SOURCE_HEAD", ""), + patch.object(queue, "SOURCE_ACTIVITY", ""), + patch.object(queue, "wait_for_source_run") as wait, + patch.object(queue, "graphql") as graphql, + ): + queue.main() + wait.assert_not_called() + graphql.assert_not_called() + self.assertEqual(len(self.fixture.writes), 17) + self.assertEqual( + {data["state"] for _path, data in self.fixture.writes}, {"pending"} + ) + + +class QueueAdvanceTests(unittest.TestCase): + """Reproduce GitHub exposing the next ref only after success publication.""" + + def setUp(self) -> None: + self.old = { + "ref": "refs/heads/gh-readonly-queue/main/old", + "object": {"sha": "passed"}, + } + self.next = { + "ref": "refs/heads/gh-readonly-queue/main/next", + "object": {"sha": "next"}, + } + + def test_front_changes_fourteen_seconds_after_publication(self) -> None: + fronts = [{"headCommit": {"oid": "passed"}}] * 3 + [ + {"headCommit": {"oid": "next"}} + ] + with ( + patch.object(queue, "front_queue_entry", side_effect=fronts), + patch.object(queue, "api", return_value=[self.old, self.next]) as api, + patch.object(queue, "dispatch_entry", return_value=False) as dispatch, + patch.object(queue.time, "sleep") as sleep, + ): + queue.advance_after_publication({"passed"}) + self.assertEqual(sleep.call_count, 3) + dispatch.assert_called_once_with(self.next) + api.assert_called_once() + + def test_reused_old_head_never_dispatches_again(self) -> None: + with ( + patch.object( + queue, + "front_queue_entry", + return_value={"headCommit": {"oid": "passed"}}, + ) as front, + patch.object(queue, "api") as api, + patch.object(queue, "dispatch_entry", return_value=False) as dispatch, + patch.object(queue.time, "sleep") as sleep, + ): + queue.advance_after_publication({"passed"}) + self.assertEqual(front.call_count, 12) + self.assertEqual(sleep.call_count, 11) + api.assert_not_called() + dispatch.assert_not_called() + + def test_missing_or_unverified_ref_never_dispatches(self) -> None: + for refs in ( + [], + [self.old], + [{"ref": "refs/heads/untrusted", "object": {"sha": "next"}}], + [self.next, self.next], + ): + with ( + self.subTest(refs=refs), + patch.object( + queue, + "front_queue_entry", + return_value={"headCommit": {"oid": "next"}}, + ) as front, + patch.object(queue, "api", return_value=refs), + patch.object(queue, "dispatch_entry", return_value=False) as dispatch, + patch.object(queue.time, "sleep"), + ): + queue.advance_after_publication({"passed"}) + self.assertEqual(front.call_count, 12) + dispatch.assert_not_called() + + def test_main_waits_only_after_real_new_publication(self) -> None: + def api(path: str, method: str = "GET", data: object = None) -> Any: + if "/pulls?" in path: + return [] + if "/matching-refs/" in path: + return [self.old] + raise AssertionError(path) + + for published in (False, True): + with ( + self.subTest(published=published), + patch.object(queue, "api", side_effect=api), + patch.object(queue, "dispatch_entry", return_value=published), + patch.object(queue, "advance_after_publication") as advance, + ): + queue.main() + if published: + advance.assert_called_once_with({"passed"}) + else: + advance.assert_not_called() + + def test_actual_sixteen_status_publication_reveals_next_front_later(self) -> None: + fixture = SourceBudgetTests()._fixtures()[0] + fixture.statuses = [] + old = {"ref": queue.QUEUE_PREFIX + "old", "object": {"sha": fixture.sha}} + new = {"ref": queue.QUEUE_PREFIX + "next", "object": {"sha": "d" * 40}} + statuses = [] + dispatches = [] + ticks = 0 + + def api(path: str, method: str = "GET", data: object = None) -> Any: + if method == "POST" and "/statuses/" in path: + statuses.append(data) + return None + if method == "POST" and path.endswith("/dispatches"): + dispatches.append((path, data)) + return None + if "/pulls?" in path: + return [] + if "/matching-refs/" in path: + return [old] if ticks < 4 else [old, new] + if "/git/ref/heads/gh-readonly-queue/" in path: + return { + "object": { + "sha": fixture.sha if path.endswith("/old") else "d" * 40 + } + } + if "/git/commits/" in path: + return {"parents": [{"sha": fixture.base}]} + if "/actions/workflows/" in path and "head_sha=" + "d" * 40 in path: + return {"total_count": 0, "workflow_runs": []} + return fixture._api(path, method, data) + + def front() -> dict[str, Any]: + nonlocal ticks + self.assertEqual(len(statuses), 16) + self.assertEqual({item["state"] for item in statuses}, {"success"}) + ticks += 1 + return {"headCommit": {"oid": fixture.sha if ticks < 4 else "d" * 40}} + + with ( + patch.object(queue, "api", side_effect=api), + patch.object(queue, "WORKFLOWS", list(fixture.checks)), + patch.object(queue, "SOURCE_CHECKS", fixture.checks), + patch.object(queue, "BASE_SHA_WORKFLOWS", ["ci.yml", "dco.yml"]), + patch.object(queue, "front_queue_entry", side_effect=front), + patch.object(queue.time, "sleep") as sleep, + ): + queue.main() + self.assertEqual(sleep.call_count, 3) + self.assertEqual(len(dispatches), 5) + self.assertTrue( + all( + payload["ref"] == new["ref"].removeprefix("refs/heads/") + for _path, payload in dispatches + ) + ) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_source_callback_action.py b/tests/test_source_callback_action.py new file mode 100644 index 0000000..d06ece7 --- /dev/null +++ b/tests/test_source_callback_action.py @@ -0,0 +1,80 @@ +"""Exercise the callback's actual shell payload without calling GitHub.""" + +import json +import os +import subprocess +import tempfile +import unittest +from pathlib import Path + +import yaml + + +class SourceCallbackActionTests(unittest.TestCase): + """Keep existing queue callers compatible and bind source attempt notifications.""" + + def invoke(self, activity: str) -> tuple[int, list[str]]: + action = yaml.load( + ( + Path(__file__).parents[1] / "actions/queue-completion/action.yml" + ).read_text(), + Loader=yaml.BaseLoader, + ) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + receipt = root / "arguments.json" + gh = root / "gh" + gh.write_text( + "#!/usr/bin/env python3\nimport json, os, sys\n" + "from pathlib import Path\n" + "Path(os.environ['CALLBACK_RECEIPT']).write_text(json.dumps(sys.argv[1:]))\n" + ) + gh.chmod(0o755) + result = subprocess.run( + ["bash", "-e", "-c", action["runs"]["steps"][0]["run"]], + env={ + **os.environ, + "PATH": f"{root}:{os.environ['PATH']}", + "CALLBACK_RECEIPT": str(receipt), + "DEFAULT_BRANCH": "main", + "GITHUB_RUN_ID": "123", + "GITHUB_RUN_ATTEMPT": "2", + "SOURCE_ACTIVITY": activity, + }, + capture_output=True, + text=True, + check=False, + ) + return result.returncode, json.loads( + receipt.read_text() + ) if receipt.exists() else [] + + def test_legacy_queue_payload_is_unchanged(self) -> None: + status, args = self.invoke("") + self.assertEqual(status, 0) + self.assertEqual( + args, + [ + "workflow", + "run", + "reconcile-merge-queue.yml", + "--ref", + "main", + "-f", + "source_run_id=123", + ], + ) + + def test_source_start_and_completion_bind_actual_attempt(self) -> None: + for activity in ("in_progress", "completed"): + with self.subTest(activity=activity): + status, args = self.invoke(activity) + self.assertEqual(status, 0) + self.assertIn("source_run_id=123", args) + self.assertIn("source_run_attempt=2", args) + self.assertIn(f"source_activity={activity}", args) + + def test_invalid_phase_cannot_dispatch(self) -> None: + status, args = self.invoke("completed --ref unreviewed") + self.assertNotEqual(status, 0) + self.assertEqual(args, []) From 8cc10aa7a626997ef82bb46b6c3e91e0544869ef Mon Sep 17 00:00:00 2001 From: IanHollow <72767437+IanHollow@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:23:11 -0700 Subject: [PATCH 2/5] Ignore obsolete source callbacks and finish empty queues quietly Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com> --- README.md | 7 +- scripts/reconcile-queue.py | 63 +++++++++++++++--- tests/test_reconcile_queue.py | 120 +++++++++++++++++++++++++++++++++- 3 files changed, 175 insertions(+), 15 deletions(-) diff --git a/README.md b/README.md index 1387ba7..c224859 100644 --- a/README.md +++ b/README.md @@ -120,12 +120,15 @@ path, branch, head and attempt before using its metadata. Start notifications invalidate previous mirrored jobs and admission without waiting for builds; completion notifications wait briefly for their own producer to finish and revalidate its actual attempt. A delayed start for a completed attempt reads -the completed results. An already queued source PR cannot bypass this check. +the completed results. Verified obsolete attempts or advanced/deleted producer +branches stop without any status or queue writes; tampered identities and future +attempts remain errors. An already queued source PR cannot bypass this check. Statuses never replace the required native jobs or relax branch protection. After publishing new complete queue success, reconciliation waits at most 55 seconds for GitHub's asynchronous merge to expose a different live front ref. It verifies that ref and dispatches its missing validation through the existing -handler once. Running or failed candidates do not start this additional poll; +handler once. An empty queue returns immediately without a warning. Running or +failed candidates do not start this additional poll; there is no recursive recovery dispatch. The actions read the default branch from GitHub's repository event metadata. diff --git a/scripts/reconcile-queue.py b/scripts/reconcile-queue.py index cdfc055..e289874 100644 --- a/scripts/reconcile-queue.py +++ b/scripts/reconcile-queue.py @@ -431,6 +431,10 @@ def wait_for_source_run(run_id: str) -> None: print(f"Source run {run_id} is still running; reconcile other ready entries") +class ObsoleteSourceNotification(RuntimeError): + """A verified producer callback no longer describes current source work.""" + + def source_notification( run_id: str, attempt: str, activity: str ) -> tuple[str, str] | None: @@ -442,6 +446,7 @@ def source_notification( Raises: ValueError: Source notification metadata is missing, tampered or stale. TypeError: Producer workflow path is not a string. + ObsoleteSourceNotification: A verified attempt or branch has advanced or vanished. """ if not run_id or not SOURCE_CHECKS: @@ -492,7 +497,7 @@ def valid(producer: dict[str, Any]) -> bool: and isinstance(producer.get("head_sha"), str) and re.fullmatch(r"[0-9a-f]{40}", producer["head_sha"]) is not None and type(producer.get("run_attempt")) is int - and producer["run_attempt"] == int(attempt) + and producer["run_attempt"] > 0 and producer.get("status") in {"queued", "in_progress", "completed", "waiting", "pending", "requested"} ) @@ -500,6 +505,19 @@ def valid(producer: dict[str, Any]) -> bool: if not valid(run): raise ValueError("Source callback producer identity does not match") sha = run["head_sha"] + if run["run_attempt"] < int(attempt): + raise ValueError("Source callback names a future producer attempt") + if run["run_attempt"] > int(attempt): + original = api(f"repos/{REPOSITORY}/actions/runs/{run_id}/attempts/{attempt}") + if ( + not valid(original) + or original["run_attempt"] != int(attempt) + or original["head_sha"] != sha + ): + raise ValueError( + "Older source callback does not match its immutable producer" + ) + raise ObsoleteSourceNotification("Producer has a newer validated attempt") if activity == "completed": wait_for_source_run(run_id) run = api(f"repos/{REPOSITORY}/actions/runs/{run_id}") @@ -507,9 +525,28 @@ def valid(producer: dict[str, Any]) -> bool: raise ValueError( "Source callback producer changed while awaiting completion" ) - ref = api(f"repos/{REPOSITORY}/git/ref/heads/{quote(branch, safe='/')}") - if ref.get("object", {}).get("sha") != sha: - raise ValueError("Source callback branch advanced beyond its producer") + if run["run_attempt"] > int(attempt): + raise ObsoleteSourceNotification( + "Producer attempt advanced during completion wait" + ) + 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 ( + not isinstance(current_sha, str) + or re.fullmatch(r"[0-9a-f]{40}", current_sha) is None + ): + raise ValueError("Source producer branch has invalid current metadata") + if current_sha != sha: + raise ObsoleteSourceNotification("Verified source producer branch advanced") # A delayed start callback must not invalidate an already completed attempt. return sha, "completed" if run["status"] == "completed" else "in_progress" @@ -919,11 +956,15 @@ def main() -> None: SOURCE_CACHE.clear() SOURCE_EVIDENCE.clear() run_id = os.environ.get("SOURCE_RUN_ID", "") - notification = source_notification( - run_id, - os.environ.get("SOURCE_RUN_ATTEMPT", ""), - os.environ.get("SOURCE_NOTIFICATION_ACTIVITY", ""), - ) + try: + notification = source_notification( + run_id, + os.environ.get("SOURCE_RUN_ATTEMPT", ""), + os.environ.get("SOURCE_NOTIFICATION_ACTIVITY", ""), + ) + except ObsoleteSourceNotification as error: + print(f"::warning::Ignoring obsolete verified source callback: {error}") + return if notification is None: wait_for_source_run(run_id) else: @@ -988,7 +1029,9 @@ 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): front = front_queue_entry() - head = front.get("headCommit") if front else None + if front is None: + return + head = front.get("headCommit") if head and head.get("oid") not in published: refs = api( f"repos/{REPOSITORY}/git/matching-refs/{QUEUE_PREFIX.removeprefix('refs/')}" diff --git a/tests/test_reconcile_queue.py b/tests/test_reconcile_queue.py index 3d3da76..5923add 100644 --- a/tests/test_reconcile_queue.py +++ b/tests/test_reconcile_queue.py @@ -3,7 +3,9 @@ from __future__ import annotations import base64 +import contextlib import importlib.util +import io import os import subprocess import unittest @@ -1533,7 +1535,7 @@ def rerun(_run_id: str) -> None: with ( patch.object(queue, "wait_for_source_run", side_effect=rerun), - self.assertRaises(ValueError), + self.assertRaises(queue.ObsoleteSourceNotification), ): self.notification("completed") self.assertEqual(self.fixture.writes, []) @@ -1547,7 +1549,6 @@ def test_tampered_producer_fields_never_bind(self) -> None: {"event": "push"}, {"head_sha": "not-a-sha"}, {"head_branch": "automation/package-updates-../other"}, - {"run_attempt": 2}, {"workflow_id": True}, ] for change in updates: @@ -1568,11 +1569,108 @@ def test_missing_phase_or_attempt_is_not_arbitrary_source_dispatch(self) -> None ): self.notification(activity, attempt) - def test_current_branch_must_still_name_producer_head(self) -> None: + def test_current_branch_advance_is_obsolete_after_validation(self) -> None: self.fixture.head = "d" * 40 + with self.assertRaises(queue.ObsoleteSourceNotification): + self.notification("in_progress") + + def test_future_attempt_is_a_real_metadata_error(self) -> None: + with self.assertRaises(ValueError): + self.notification("in_progress", "2") + + def test_head_mutation_during_wait_is_a_real_error(self) -> None: + def mutate(_run_id: str) -> None: + self.run["head_sha"] = "e" * 40 + + with ( + patch.object(queue, "wait_for_source_run", side_effect=mutate), + self.assertRaises(ValueError), + ): + self.notification("completed") + + def test_obsolete_callbacks_main_is_noop_before_any_admission(self) -> None: + original = self.run.copy() + for reason in ("old_attempt", "advanced", "deleted"): + with self.subTest(reason=reason): + self.run.clear() + self.run.update(original) + self.fixture.head = self.fixture.sha + if reason == "old_attempt": + self.run["run_attempt"] = 2 + if reason == "advanced": + self.fixture.head = "d" * 40 + + def api( + path: str, + method: str = "GET", + data: object = None, + *, + obsolete_reason: str = reason, + ) -> Any: + if "/attempts/" in path: + return original + if ( + obsolete_reason == "deleted" + and self.fixture.branch in path + and "/git/ref/" in path + ): + raise subprocess.CalledProcessError( + 1, ["gh", "api", path], stderr="HTTP 404: Not Found" + ) + return self.fixture._api(path, method, data) + + output = io.StringIO() + with ( + patch.dict( + os.environ, + { + "SOURCE_RUN_ID": self.run_id, + "SOURCE_RUN_ATTEMPT": "1", + "SOURCE_NOTIFICATION_ACTIVITY": "in_progress", + }, + ), + 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_EVIDENCE", {}), + patch.object(queue, "admit") as admit, + patch.object(queue, "dispatch_entry") as dispatch, + patch.object(queue, "graphql") as graphql, + contextlib.redirect_stdout(output), + ): + queue.main() + self.assertIn( + "Ignoring obsolete verified source callback", output.getvalue() + ) + self.assertEqual(self.fixture.writes, []) + admit.assert_not_called() + dispatch.assert_not_called() + graphql.assert_not_called() + + def test_tampered_old_producer_is_not_obsolete(self) -> None: + self.run.update(run_attempt=2, repository={"full_name": "other/repository"}) with self.assertRaises(ValueError): self.notification("in_progress") + def test_invalid_immutable_old_attempt_is_not_obsolete(self) -> None: + self.run["run_attempt"] = 2 + old = {**self.run, "run_attempt": 1, "head_sha": "e" * 40} + + def api(path: str, method: str = "GET", data: object = None) -> Any: + return ( + old if "/attempts/" in path else self.fixture._api(path, method, data) + ) + + 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", {}), + self.assertRaises(ValueError), + ): + queue.source_notification(self.run_id, "1", "in_progress") + def test_legacy_queue_callback_keeps_original_contract(self) -> None: self.run["head_branch"] = "gh-readonly-queue/main/pr-7-example" with patch.object(queue, "wait_for_source_run") as wait: @@ -1700,6 +1798,22 @@ def test_front_changes_fourteen_seconds_after_publication(self) -> None: dispatch.assert_called_once_with(self.next) api.assert_called_once() + def test_empty_queue_returns_without_sleep_or_warning(self) -> None: + output = io.StringIO() + with ( + patch.object(queue, "front_queue_entry", return_value=None) as front, + patch.object(queue, "api") as api, + patch.object(queue, "dispatch_entry") as dispatch, + patch.object(queue.time, "sleep") as sleep, + contextlib.redirect_stdout(output), + ): + queue.advance_after_publication({"passed"}) + front.assert_called_once() + sleep.assert_not_called() + api.assert_not_called() + dispatch.assert_not_called() + self.assertEqual(output.getvalue(), "") + def test_reused_old_head_never_dispatches_again(self) -> None: with ( patch.object( From bb50f8e1dea9ec6e4ebd2b32b559b0746130338a Mon Sep 17 00:00:00 2001 From: IanHollow <72767437+IanHollow@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:26:36 -0700 Subject: [PATCH 3/5] Document v2.12 source callbacks and queue recovery Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com> --- docs/releases/v2.12.0.md | 39 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) create mode 100644 docs/releases/v2.12.0.md diff --git a/docs/releases/v2.12.0.md b/docs/releases/v2.12.0.md new file mode 100644 index 0000000..7b70e9f --- /dev/null +++ b/docs/releases/v2.12.0.md @@ -0,0 +1,39 @@ +# Nix Forge CI v2.12.0 + +## Changelog + +- Add explicit package-source start and completion notifications using the + existing repository token and narrowly scoped callback jobs. Validate actual + producer identity and run attempt before deriving its source head or activity. +- Invalidate all previous required source results and admission when a rerun + starts. Completion waits for its producer, revalidates current attempts and + preserves actual failed-job links. Verified obsolete callbacks make no writes; + tampered producer metadata and future attempts remain errors. +- After publishing complete queue success, wait at most 55 seconds for GitHub's + asynchronous merge to expose the next verified front ref, then dispatch its + missing validation once. Empty queues return quietly. Running, failed or + incomplete candidates do not start this additional poll. + +Existing queue callbacks retain their original payload. Opted-in package +callers declare optional `source_run_attempt` and `source_activity` reconciler +dispatch inputs and forward them to `source-run-attempt` and `source-activity`. +Use `source-activity: in_progress` in a source-only start callback, then +`source-activity: completed` after all source validation jobs. Retain the queue +completion callback name and restrict callbacks to their approved ref namespaces. +Update source branches only after the new caller workflow contract is merged. + +No token permissions, required job gates, signing trust or release-builder pins +change. Full expected-head PR and merge-group validation remains required. +The 161 repository regression tests and offline workflow/security checks passed +locally; required hosted validation must pass before publishing this release. +Coverage includes actual callback payloads, immutable producer identity, +completion races, obsolete notifications, 16-job source invalidation, request +budgets and asynchronous next-candidate dispatch. + +The source implementation was reviewed at +`8cc10aa7a626997ef82bb46b6c3e91e0544869ef`. The signed tag and verified release +manifest identify the protected source commit including these notes. Publication +uses the separately pinned SLSA source builder and verifies archive checksums, +all provenance subjects and protected-main ancestry. Restore the v2.11.0 pin and +remove the optional source notification inputs to roll back. Compatible v2.x +consumers have no scheduled support end date. From 56b021f21a975e552cebca27e7596964057c5792 Mon Sep 17 00:00:00 2001 From: IanHollow <72767437+IanHollow@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:31:54 -0700 Subject: [PATCH 4/5] docs: register v2.12 release notes in strict site navigation 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 59324ff..6cdcf29 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.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 - v2.9.0 release notes: releases/v2.9.0.md From 128a7e6fed565147424c30227f4d72bd4406d00f Mon Sep 17 00:00:00 2001 From: IanHollow <72767437+IanHollow@users.noreply.github.com> Date: Mon, 12 Oct 2026 19:34:41 -0700 Subject: [PATCH 5/5] test: bind callback fixture to the sandbox Python interpreter Signed-off-by: IanHollow <72767437+IanHollow@users.noreply.github.com> --- tests/test_source_callback_action.py | 25 ++++++++++++++----------- 1 file changed, 14 insertions(+), 11 deletions(-) diff --git a/tests/test_source_callback_action.py b/tests/test_source_callback_action.py index d06ece7..78c2137 100644 --- a/tests/test_source_callback_action.py +++ b/tests/test_source_callback_action.py @@ -3,6 +3,7 @@ import json import os import subprocess +import sys import tempfile import unittest from pathlib import Path @@ -13,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]]: + def invoke(self, activity: str) -> tuple[int, list[str], str]: action = yaml.load( ( Path(__file__).parents[1] / "actions/queue-completion/action.yml" @@ -25,7 +26,7 @@ def invoke(self, activity: str) -> tuple[int, list[str]]: receipt = root / "arguments.json" gh = root / "gh" gh.write_text( - "#!/usr/bin/env python3\nimport json, os, sys\n" + 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" ) @@ -45,13 +46,15 @@ def invoke(self, activity: str) -> tuple[int, list[str]]: text=True, check=False, ) - return result.returncode, json.loads( - receipt.read_text() - ) if receipt.exists() else [] + return ( + result.returncode, + json.loads(receipt.read_text()) if receipt.exists() else [], + result.stderr, + ) def test_legacy_queue_payload_is_unchanged(self) -> None: - status, args = self.invoke("") - self.assertEqual(status, 0) + status, args, stderr = self.invoke("") + self.assertEqual(status, 0, stderr) self.assertEqual( args, [ @@ -68,13 +71,13 @@ def test_legacy_queue_payload_is_unchanged(self) -> None: def test_source_start_and_completion_bind_actual_attempt(self) -> None: for activity in ("in_progress", "completed"): with self.subTest(activity=activity): - status, args = self.invoke(activity) - self.assertEqual(status, 0) + 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) def test_invalid_phase_cannot_dispatch(self) -> None: - status, args = self.invoke("completed --ref unreviewed") - self.assertNotEqual(status, 0) + status, args, stderr = self.invoke("completed --ref unreviewed") + self.assertNotEqual(status, 0, stderr) self.assertEqual(args, [])