From 91ac9f4ca4cebf70a167a63d327c9133232b3a06 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:14:15 -0700 Subject: [PATCH 01/14] Accelerate and harden burst deliveries --- README.md | 5 +- action.yml | 2 +- src/agent_merge_queue/cli.py | 322 +++++++++++++++++++++++++------- src/agent_merge_queue/doctor.py | 34 +++- tests/test_cli.py | 134 ++++++++++++- tests/test_doctor.py | 45 +++++ tests/test_skill.py | 8 + 7 files changed, 478 insertions(+), 72 deletions(-) diff --git a/README.md b/README.md index 9b2541b..c8e2873 100644 --- a/README.md +++ b/README.md @@ -88,7 +88,10 @@ Install `examples/github-workflow.yml` on the default branch. It reacts to deploy labels, ready/synchronize events, reviews, named CI `workflow_run` completions, and completed external check suites. Keep its `workflows` list aligned with `pipeline.ci_workflows`. The privileged worker never checks out or -executes pull-request code. Pin the Action to the full reviewed release commit: +executes pull-request code. The Action follows releases by default so the same +serialized worker can dispatch deployment when GitHub suppresses the +`workflow_run` event for token-dispatched CI. Pin the Action to the full +reviewed release commit: ```yaml - uses: Forward-Future/DeployBot@13d7293b181581d2e4d59d8a605df76f7feb88a6 diff --git a/action.yml b/action.yml index dc3bff6..9574a17 100644 --- a/action.yml +++ b/action.yml @@ -8,7 +8,7 @@ inputs: follow: description: Follow cumulative main through CI and deployment required: false - default: "false" + default: "true" dispatch_ci: description: Dispatch configured CI after a merge made with github.token required: false diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index ed813ed..77323cf 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -282,11 +282,17 @@ def marker_queued_at(marker: dict[str, Any] | None) -> str | None: return str(value) if value else None +def marker_priority_at(marker: dict[str, Any] | None) -> str | None: + value = marker.get("priority_at") if marker else None + return str(value) if value else None + + def queue_state_body( state: str, head_sha: str, *, queued_at: str | None, + priority_at: str | None = None, reason: str | None = None, intent_id: str | None = None, integration_batch_id: str | None = None, @@ -301,6 +307,8 @@ def queue_state_body( } if queued_at: marker["queued_at"] = queued_at + if priority_at: + marker["priority_at"] = priority_at if reason: marker["reason"] = reason if intent_id: @@ -591,6 +599,7 @@ class QueueEntry: dependencies: list[int] state: str = "waiting" reasons: list[str] | None = None + priority_at: str | None = None def classify( self, @@ -708,8 +717,8 @@ def resolve_pr(self, selector: str | None) -> int: arguments.extend(["--repo", self.repository, "--json", "number"]) return int(self._json(*arguments)["number"]) - def ensure_labels(self) -> None: - labels = ( + def label_specs(self) -> tuple[tuple[str, str, str], ...]: + return ( ( self.config.queue_label, "0E8A16", @@ -736,7 +745,37 @@ def ensure_labels(self) -> None: "DeployBot metadata registry (never transcript contents)", ), ) - for name, color, description in labels: + + def ensure_labels(self) -> None: + for name, color, description in self.label_specs(): + self._run( + "label", + "create", + name, + "--repo", + self.repository, + "--color", + color, + "--description", + description, + "--force", + ) + + def ensure_labels_exist(self) -> None: + values = self._json( + "label", + "list", + "--repo", + self.repository, + "--limit", + "1000", + "--json", + "name", + ) + existing = {str(value.get("name") or "") for value in values} + for name, color, description in self.label_specs(): + if name in existing: + continue self._run( "label", "create", @@ -971,13 +1010,15 @@ def registry_issue_number(self, *, create: bool) -> int | None: if "pull_request" not in value and str(value.get("title") or "") == self.config.pipeline.registry_title ] - if len(matches) > 1: - raise QueueError("more than one DeployBot registry issue exists") if matches: - return int(matches[0]["number"]) + # GitHub Issues has no uniqueness constraint. Concurrent first-use + # workers can both create a registry after observing none. The + # lowest issue number is deterministic and comments remain trusted + # by authenticated author, so all workers safely converge there. + return min(int(value["number"]) for value in matches) if not create: return None - self.ensure_labels() + self.ensure_labels_exist() created = self._json( "api", "--method", @@ -990,7 +1031,9 @@ def registry_issue_number(self, *, create: bool) -> int | None: "-f", f"labels[]={self.config.pipeline.registry_label}", ) - return int(created["number"]) + # Re-read after creation so a concurrent lower-numbered creator wins + # before this worker writes any metadata to its candidate issue. + return self.registry_issue_number(create=False) or int(created["number"]) def issue_comment(self, number: int, body: str) -> None: self._json( @@ -1215,6 +1258,9 @@ def snapshot( *, require_marker: bool = True, allow_blocked_label: bool = False, + known_comments: list[dict[str, Any]] | None = None, + known_source_paths: list[str] | None = None, + known_generated_paths: list[str] | None = None, ) -> QueueEntry: fields = ",".join( ( @@ -1244,8 +1290,8 @@ def snapshot( if pull.get("state") != "OPEN": raise QueueError(f"PR #{number} is not open") - comments = self.comments(number) - reviews = self.reviews(number) + comments = known_comments if known_comments is not None else self.comments(number) + reviews = self.reviews(number) if self.config.review_providers else [] needs_threads = any( provider.kind == "bot" and provider.require_resolved_threads for provider in self.config.review_providers @@ -1254,29 +1300,31 @@ def snapshot( marker = queue_marker_for_client(self, comments) head_sha = str(pull["headRefOid"]) checks = check_states(pull.get("statusCheckRollup") or []) - changed_files = self.files(number) - source_paths: list[str] = [] - generated_paths: list[str] = [] - for value in changed_files: - path = str(value.get("filename") or "") - if not path: - continue - if path in self.config.generated_paths: - generated_paths.append(path) - source_paths.append(path) - continue - target = ( - generated_paths - if generated_only_change( - path, - value.get("patch"), - generated_paths=self.config.generated_paths, - generated_version_paths=self.config.generated_version_paths, - asset_version_pattern=self.config.asset_version_pattern, + source_paths = list(known_source_paths or []) + generated_paths = list(known_generated_paths or []) + if known_source_paths is None or known_generated_paths is None: + source_paths = [] + generated_paths = [] + for value in self.files(number): + path = str(value.get("filename") or "") + if not path: + continue + if path in self.config.generated_paths: + generated_paths.append(path) + source_paths.append(path) + continue + target = ( + generated_paths + if generated_only_change( + path, + value.get("patch"), + generated_paths=self.config.generated_paths, + generated_version_paths=self.config.generated_version_paths, + asset_version_pattern=self.config.asset_version_pattern, + ) + else source_paths ) - else source_paths - ) - target.append(path) + target.append(path) body = str(pull.get("body") or "") entry = QueueEntry( @@ -1306,6 +1354,7 @@ def snapshot( dependencies=structured_dependencies( body, self.config.dependency_directive ), + priority_at=marker_priority_at(marker), ) entry.classify( self.config, @@ -1329,7 +1378,11 @@ def queue(self) -> list[QueueEntry]: entries = [self.snapshot(number) for number in self.queued_numbers()] return sorted( entries, - key=lambda entry: (entry.queued_at or "9999", entry.number), + key=lambda entry: ( + entry.priority_at or entry.queued_at or "9999", + entry.queued_at or "9999", + entry.number, + ), ) def add_label(self, number: int, label: str) -> None: @@ -1666,8 +1719,14 @@ def command_request( ) -> dict[str, Any]: number = client.resolve_pr(selector) actor = client.require_actor(client.trusted_logins, "request deployment") - client.ensure_labels() - entry = client.snapshot(number, require_marker=False, allow_blocked_label=True) + client.ensure_labels_exist() + entry = client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + known_source_paths=[], + known_generated_paths=[], + ) requested_at = utc_now() intent_id = hashlib.sha256( f"{client.repository}:{number}:{actor}:{requested_at}".encode() @@ -1856,7 +1915,12 @@ def record_repair( return payload -def command_promote(client: GitHub, *, emit: bool = True) -> dict[str, Any]: +def command_promote( + client: GitHub, + *, + emit: bool = True, + captured_entries: list[QueueEntry] | None = None, +) -> dict[str, Any]: promoted: list[int] = [] waiting: list[dict[str, Any]] = [] blocked: list[dict[str, Any]] = [] @@ -1871,7 +1935,14 @@ def command_promote(client: GitHub, *, emit: bool = True) -> dict[str, Any]: intent = latest_intent(comments, client.trusted_logins) if not intent or intent.get("state") != "requested": continue - entry = client.snapshot(number, require_marker=False, allow_blocked_label=True) + entry = client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + known_comments=comments, + ) + if captured_entries is not None: + captured_entries.append(entry) if ( client.config.pipeline.intent_scope == "head" and intent.get("requested_head") != entry.head_sha @@ -1886,9 +1957,27 @@ def command_promote(client: GitHub, *, emit: bool = True) -> dict[str, Any]: } ) continue + if client.config.blocked_label in entry.labels: + waiting.append( + { + "number": number, + "reasons": [ + "repair is blocked; the trusted source agent must run " + "deploybot resume after fresh gates pass" + ], + } + ) + continue if entry.state == "ready": - queue_from_intent(client, entry, intent) - promoted.append(number) + changed = queue_from_intent( + client, + entry, + intent, + comments=comments, + labels=set(entry.labels), + ) + if changed: + promoted.append(number) elif entry.state == "blocked": reason = "; ".join(entry.reasons or ["blocked"]) record_repair(client, entry, intent, reason) @@ -2029,6 +2118,7 @@ def command_enqueue(client: GitHub, selector: str | None) -> None: previous = ( { "head_sha": entry.queued_head_sha, + "priority_at": entry.priority_at, "queued_at": entry.queued_at, "state": entry.queue_state, } @@ -2055,7 +2145,14 @@ def command_enqueue(client: GitHub, selector: str | None) -> None: already_queued=client.config.queue_label in entry.labels, now=utc_now(), ) - body = queue_state_body("queued", entry.head_sha, queued_at=queued_at) + body = queue_state_body( + "queued", + entry.head_sha, + queued_at=queued_at, + priority_at=( + str(previous.get("priority_at") or "") or None if previous else None + ), + ) client.comment(number, body) if client.config.queue_label in entry.labels: # Re-adding the label emits a fresh `labeled` event for GitHub-hosted @@ -2071,7 +2168,10 @@ def queue_from_intent( client: GitHub, entry: QueueEntry, intent: dict[str, Any], -) -> None: + *, + comments: list[dict[str, Any]] | None = None, + labels: set[str] | None = None, +) -> bool: if entry.state != "ready": raise QueueError( f"PR #{entry.number} is not ready to queue: " @@ -2088,28 +2188,53 @@ def queue_from_intent( f"PR #{entry.number} deploy intent is bound to an older head; " "run deploybot refresh-request after fresh review" ) - comments = client.comments(entry.number) - previous = queue_marker_for_client(client, comments) + current_comments = comments if comments is not None else client.comments(entry.number) + previous = queue_marker_for_client(client, current_comments) + current_labels = labels if labels is not None else client.labels(entry.number) + if ( + client.config.queue_label in current_labels + and previous + and previous.get("state") == "queued" + and previous.get("head_sha") == entry.head_sha + and previous.get("intent_id") == intent_id + ): + entry.queued_head_sha = entry.head_sha + entry.queued_at = marker_queued_at(previous) + entry.priority_at = marker_priority_at(previous) + entry.queue_state = "queued" + return False queued_at = queue_timestamp( previous, already_queued=client.config.queue_label in entry.labels, now=utc_now(), ) + priority_at = ( + marker_priority_at(previous) + or str(intent.get("requested_at") or "") + or queued_at + ) client.comment( entry.number, queue_state_body( "queued", entry.head_sha, queued_at=queued_at, + priority_at=priority_at, intent_id=intent_id, ), ) - labels = client.labels(entry.number) - if client.config.queue_label in labels: + if client.config.queue_label in current_labels: client.remove_label(entry.number, client.config.queue_label) client.add_label(entry.number, client.config.queue_label) - if client.config.blocked_label in labels: + if client.config.blocked_label in current_labels: client.remove_label(entry.number, client.config.blocked_label) + entry.labels = sorted( + (current_labels - {client.config.blocked_label}) | {client.config.queue_label} + ) + entry.queued_head_sha = entry.head_sha + entry.queued_at = queued_at + entry.priority_at = priority_at + entry.queue_state = "queued" notify( client.config.pipeline, "queued", @@ -2120,6 +2245,7 @@ def queue_from_intent( "intent_id": intent_id, }, ) + return True @dataclass @@ -2140,8 +2266,28 @@ def as_dict(self) -> dict[str, Any]: } -def freeze_queue(client: GitHub) -> FreezeResult: - all_entries = client.queue() +def freeze_queue( + client: GitHub, *, known_entries: list[QueueEntry] | None = None +) -> FreezeResult: + if known_entries is None: + all_entries = client.queue() + else: + queued_numbers = set(client.queued_numbers()) + entries_by_number = { + entry.number: entry + for entry in known_entries + if entry.number in queued_numbers + } + for number in sorted(queued_numbers - entries_by_number.keys()): + entries_by_number[number] = client.snapshot(number) + all_entries = sorted( + entries_by_number.values(), + key=lambda entry: ( + entry.priority_at or entry.queued_at or "9999", + entry.queued_at or "9999", + entry.number, + ), + ) entries, blocked_entries = split_blocked_entries( all_entries, client.config.blocked_label ) @@ -2259,6 +2405,7 @@ def command_block(client: GitHub, selector: str | None, reason: str) -> None: "blocked", entry.head_sha, queued_at=str(previous.get("queued_at") or "") or None, + priority_at=str(previous.get("priority_at") or "") or None, reason=reason, intent_id=str(previous.get("intent_id") or "") or None, ), @@ -2288,6 +2435,7 @@ def command_unblock(client: GitHub, selector: str | None) -> None: "queued", entry.head_sha, queued_at=str(previous.get("queued_at") or "") or None, + priority_at=str(previous.get("priority_at") or "") or None, intent_id=str(previous.get("intent_id") or "") or None, ), ) @@ -2316,6 +2464,9 @@ def command_dequeue(client: GitHub, selector: str | None, reason: str) -> None: queued_at=( str(previous.get("queued_at") or "") or None if previous else None ), + priority_at=( + str(previous.get("priority_at") or "") or None if previous else None + ), reason=reason, intent_id=( str(previous.get("intent_id") or "") or None if previous else None @@ -2334,22 +2485,34 @@ def command_merge( batch_id: str, *, emit: bool = True, + frozen_entry: QueueEntry | None = None, + frozen_batch: dict[str, Any] | None = None, + active_numbers: set[int] | None = None, ) -> str: require_running_pipeline(client) - number = client.resolve_pr(selector) - batch = latest_batch_marker( - client.comments(number), - coordinator_logins(client), - batch_id=batch_id, - ) + number = frozen_entry.number if frozen_entry is not None else client.resolve_pr(selector) + batch = frozen_batch + if batch is not None and str(batch.get("batch_id") or "") != batch_id: + raise QueueError(f"frozen context does not match batch {batch_id}") + if batch is None: + batch = latest_batch_marker( + client.comments(number), + coordinator_logins(client), + batch_id=batch_id, + ) if batch is None: raise QueueError(f"PR #{number} has no trusted marker for batch {batch_id}") frozen = {int(value) for value in batch.get("pull_requests") or []} if number not in frozen: raise QueueError(f"PR #{number} is not a member of batch {batch_id}") - entries = client.queue() - frozen_entries = [value for value in entries if value.number in frozen] - entry = next((value for value in frozen_entries if value.number == number), None) + if frozen_entry is None: + entries = client.queue() + frozen_entries = [value for value in entries if value.number in frozen] + entry = next((value for value in frozen_entries if value.number == number), None) + current_numbers = {value.number for value in frozen_entries} + else: + entry = frozen_entry + current_numbers = active_numbers or {number} if entry is None: raise QueueError(f"PR #{number} is not in the merge queue") if entry.state != "ready": @@ -2364,9 +2527,7 @@ def command_merge( if entry.queued_at and str(batch.get("frozen_at") or "") < entry.queued_at: raise QueueError(f"batch {batch_id} predates the current queue authorization") - peers = batch_overlap_peers( - batch, number, {value.number for value in frozen_entries} - ) + peers = batch_overlap_peers(batch, number, current_numbers) if peers: peers = ", ".join(f"#{value}" for value in peers) raise QueueError( @@ -2389,7 +2550,11 @@ def command_merge( + ", ".join(f"#{value}" for value in missing_dependencies) ) - fresh = client.snapshot(number) + fresh = client.snapshot( + number, + known_source_paths=entry.source_paths, + known_generated_paths=entry.generated_paths, + ) if fresh.state != "ready": raise QueueError( f"PR #{number} changed before merge: " @@ -2490,7 +2655,15 @@ def drain_frozen_batch( if any(dependency in pending for dependency in entry.dependencies): continue try: - merge_sha = command_merge(client, str(number), batch_id, emit=False) + merge_sha = command_merge( + client, + str(number), + batch_id, + emit=False, + frozen_entry=entry, + frozen_batch=frozen.batch, + active_numbers=set(pending), + ) except QueueError as error: if "GitHub is still computing mergeability" in str(error): retryable = True @@ -2538,7 +2711,11 @@ def drain_frozen_batch( def command_drain( - client: GitHub, *, json_output: bool, emit: bool = True + client: GitHub, + *, + json_output: bool, + emit: bool = True, + initial_frozen: FreezeResult | None = None, ) -> dict[str, Any]: require_running_pipeline(client) batch_ids: list[str] = [] @@ -2548,7 +2725,8 @@ def command_drain( next_batch: list[int] = [] while True: - frozen = freeze_queue(client) + frozen = initial_frozen or freeze_queue(client) + initial_frozen = None if frozen.batch is None: break batch_id = str(frozen.batch["batch_id"]) @@ -2724,8 +2902,13 @@ def command_react( print(json.dumps(result, indent=2, sort_keys=True)) return result promoted_integrations = promote_integrations(client) - promoted = command_promote(client, emit=False) - frozen = freeze_queue(client) + captured_entries: list[QueueEntry] = [] + promoted = command_promote( + client, + emit=False, + captured_entries=captured_entries, + ) + frozen = freeze_queue(client, known_entries=captured_entries) integrations: list[dict[str, Any]] = [] if ( frozen.batch is not None @@ -2746,7 +2929,12 @@ def command_react( "next_batch": [], } else: - drained = command_drain(client, json_output=False, emit=False) + drained = command_drain( + client, + json_output=False, + emit=False, + initial_frozen=frozen, + ) if ( frozen.batch is not None and client.config.integration.mode == "overlap" diff --git a/src/agent_merge_queue/doctor.py b/src/agent_merge_queue/doctor.py index 4b2e314..21931c0 100644 --- a/src/agent_merge_queue/doctor.py +++ b/src/agent_merge_queue/doctor.py @@ -101,7 +101,14 @@ def diagnose( ) return rows repo = str(value.get("nameWithOwner") or "") - code, _, detail = _json("repo", "view", repo, "--json", "nameWithOwner", cwd=root) + code, repository, detail = _json( + "repo", + "view", + repo, + "--json", + "nameWithOwner,hasIssuesEnabled", + cwd=root, + ) if code: rows.append(row("repository", "fail", detail, "Check repository access.")) return rows @@ -109,6 +116,31 @@ def diagnose( if config is None: return rows + issues_enabled = (repository or {}).get("hasIssuesEnabled") + issues_status = "ok" if issues_enabled is True else "fail" + if issues_enabled is None: + issues_status = "warn" + rows.append( + row( + "issue-registry", + issues_status, + ( + "GitHub Issues is enabled for durable DeployBot metadata" + if issues_enabled is True + else ( + "GitHub Issues is disabled" + if issues_enabled is False + else "Could not confirm whether GitHub Issues is enabled" + ) + ), + ( + None + if issues_enabled is True + else "Enable Issues before recording deploy intent or pipeline state." + ), + ) + ) + code, workflows, detail = _json( "workflow", "list", "--repo", repo, "--json", "name,path,state", cwd=root ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 2318db4..df1ee97 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -34,10 +34,12 @@ latest_batch_marker, latest_marker, marker_queued_at, + marker_priority_at, main, new_batch, overlap_groups, queue_state_body, + queue_from_intent, queue_timestamp, reusable_batch, structured_dependencies, @@ -107,6 +109,87 @@ def test_marker_without_queue_time_stays_missing(self) -> None: marker_queued_at({"queued_at": "2026-06-20T00:00:00Z"}), "2026-06-20T00:00:00Z", ) + self.assertEqual( + marker_priority_at({"priority_at": "2026-06-19T23:59:00Z"}), + "2026-06-19T23:59:00Z", + ) + + def test_intent_request_time_is_the_stable_queue_priority(self) -> None: + value = entry(1) + value.labels = ["deploy-requested"] + intent = { + "intent_id": "intent-1", + "requested_at": "2026-06-20T00:00:00Z", + "requested_head": value.head_sha, + "state": "requested", + } + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"trusted"} + + changed = queue_from_intent( + client, + value, + intent, + comments=[], + labels={"deploy-requested"}, + ) + + self.assertTrue(changed) + self.assertEqual(value.priority_at, "2026-06-20T00:00:00Z") + self.assertIn( + '"priority_at": "2026-06-20T00:00:00Z"', + client.comment.call_args.args[1], + ) + + def test_registry_race_converges_on_the_lowest_issue_number(self) -> None: + client = object.__new__(GitHub) + client.config = CONFIG + client.repository = "example/repo" + client._paged_api = Mock( + side_effect=[ + [], + [ + {"number": 10, "title": "DeployBot delivery registry"}, + {"number": 9, "title": "DeployBot delivery registry"}, + ], + ] + ) + client._json = Mock(side_effect=[[], {"number": 10}]) + client._run = Mock(return_value="") + + self.assertEqual(client.registry_issue_number(create=True), 9) + + def test_frozen_merge_fast_path_never_rescans_the_whole_queue(self) -> None: + value = entry(1, "a.py") + batch = new_batch([value], frozen_at="2026-06-20T00:01:00Z") + client = Mock() + client.config = CONFIG + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"trusted"} + client.pipeline_control.return_value = {"state": "running"} + client.snapshot.return_value = value + client.merge.return_value = "m" * 40 + client.comments.return_value = [] + + merged = command_merge( + client, + "1", + str(batch["batch_id"]), + frozen_entry=value, + frozen_batch=batch, + active_numbers={1}, + ) + + self.assertEqual(merged, "m" * 40) + client.queue.assert_not_called() + client.snapshot.assert_called_once_with( + 1, + known_source_paths=["a.py"], + known_generated_paths=[], + ) def test_generated_paths_are_configurable(self) -> None: self.assertTrue( @@ -504,7 +587,15 @@ def test_drain_merges_independent_ready_entries_and_skips_overlap(self) -> None: ): result = command_drain(client, json_output=True) - merge.assert_called_once_with(client, "1", "batch", emit=False) + merge.assert_called_once_with( + client, + "1", + "batch", + emit=False, + frozen_entry=first, + frozen_batch=frozen.batch, + active_numbers={1, 2, 3, 4}, + ) self.assertEqual(result["merged"][0]["number"], 1) self.assertEqual(result["integration_required"][0]["pull_requests"], [2, 3]) self.assertEqual(result["waiting"][0]["number"], 4) @@ -565,7 +656,15 @@ def test_drain_completes_waiting_pass_then_promotes_next_batch(self) -> None: ): result = command_drain(client, json_output=True) - merge.assert_called_once_with(client, "2", "second", emit=False) + merge.assert_called_once_with( + client, + "2", + "second", + emit=False, + frozen_entry=ready, + frozen_batch=second.batch, + active_numbers={1, 2}, + ) self.assertEqual(result["batch_ids"], ["first", "second"]) self.assertEqual(result["merged"][0]["number"], 2) self.assertEqual(result["waiting"][0]["number"], 1) @@ -985,6 +1084,37 @@ def test_promote_queues_ready_intent_but_leaves_waiting_visible(self) -> None: self.assertIn("intent-1", client.comment.call_args.args[1]) client.add_label.assert_called_with(1, "merge-queue") + def test_promote_never_auto_resumes_a_repair_block(self) -> None: + blocked = entry(1) + blocked.labels = ["deploy-requested", "merge-queue-blocked"] + intent_comment = { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "trusted"}, + "body": intent_body( + intent_id="intent-1", + state="requested", + requested_at="2026-06-20T00:00:00Z", + requested_head=blocked.head_sha, + ), + } + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"coordinator"} + client.intent_numbers.return_value = [1] + client.active_integration_sources.return_value = set() + client.comments.return_value = [intent_comment] + client.snapshot.return_value = blocked + + with redirect_stdout(io.StringIO()): + result = command_promote(client) + + self.assertEqual(result["promoted"], []) + self.assertIn("deploybot resume", result["waiting"][0]["reasons"][0]) + client.add_label.assert_not_called() + def test_resume_atomically_requeues_repaired_exact_head(self) -> None: value = entry(1) value.labels = ["deploy-requested", "merge-queue-blocked"] diff --git a/tests/test_doctor.py b/tests/test_doctor.py index aa94a89..d7afacf 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -118,6 +118,51 @@ def fake_json(*arguments: str, cwd: Path): self.assertEqual(labels["status"], "warn") self.assertEqual(checks["status"], "warn") + def test_disabled_issues_is_a_registry_failure(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + (root / ".mergequeue.toml").write_text( + '[queue]\nrequired_checks = ["CI"]\ntrusted_actors = ["trusted"]\n', + encoding="utf-8", + ) + + def fake_json(*arguments: str, cwd: Path): + joined = " ".join(arguments) + if "repo view" in joined: + return ( + 0, + {"nameWithOwner": "owner/repo", "hasIssuesEnabled": False}, + "", + ) + if "users/trusted" in joined: + return 0, {"login": "trusted"}, "" + if "workflow list" in joined or "label list" in joined: + return 0, [], "" + if "pr list" in joined: + return 0, [], "" + if "protection" in joined: + return 1, None, "not available" + return 0, {"login": "owner"}, "" + + with ( + patch( + "agent_merge_queue.doctor.shutil.which", return_value="/usr/bin/gh" + ), + patch( + "agent_merge_queue.doctor._gh", return_value=(0, "authenticated") + ), + patch("agent_merge_queue.doctor._json", side_effect=fake_json), + ): + rows = diagnose( + config_path=None, + repository="owner/repo", + cwd=root, + ) + + registry = next(value for value in rows if value["check"] == "issue-registry") + self.assertEqual(registry["status"], "fail") + self.assertIn("disabled", registry["detail"]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_skill.py b/tests/test_skill.py index 32dba4a..01e83a6 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -87,6 +87,14 @@ def test_action_dispatches_ci_after_builtin_token_merge(self) -> None: self.assertIn("actions: write", example) self.assertIn("workflow_dispatch:", workflow) + def test_action_follows_release_when_workflow_run_is_suppressed(self) -> None: + action = (ROOT / "action.yml").read_text(encoding="utf-8") + follow_input = action.split(" follow:\n", 1)[1].split( + " dispatch_ci:\n", 1 + )[0] + self.assertIn('default: "true"', follow_input) + self.assertIn('args+=(--follow)', action) + def test_clients_pin_the_immutable_status_release(self) -> None: paths = [ ROOT / "adapters" / "codex" / "agent-merge-queue" / ".mcp.json", From 9129bcc7fb845394714e4ab528c9e447a6e22813 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:20:48 -0700 Subject: [PATCH 02/14] Coalesce near-ready delivery bursts --- README.md | 1 + src/agent_merge_queue/cli.py | 121 +++++++++++++++++++++++++------- src/agent_merge_queue/config.py | 15 ++++ tests/test_cli.py | 49 +++++++++++++ tests/test_config.py | 12 ++++ 5 files changed, 173 insertions(+), 25 deletions(-) diff --git a/README.md b/README.md index c8e2873..bc1b040 100644 --- a/README.md +++ b/README.md @@ -171,6 +171,7 @@ can pause further merges until `deploybot unpause`. [pipeline] ci_workflows = ["CI"] deploy_workflows = ["Deploy"] +batch_settle_seconds = 15 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 77323cf..9722efa 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -600,6 +600,7 @@ class QueueEntry: state: str = "waiting" reasons: list[str] | None = None priority_at: str | None = None + integration_batch_id: str | None = None def classify( self, @@ -677,6 +678,7 @@ def __init__( if self.repository.count("/") != 1: raise QueueError(f"invalid repository name: {self.repository}") self.owner, self.name = self.repository.split("/", 1) + self._registry_issue_cache: int | None = None self.trusted_logins = { self.owner if value == "@repository-owner" else value for value in self.config.trusted_actors @@ -1001,6 +1003,9 @@ def is_ancestor(self, ancestor: str, descendant: str) -> bool: return comparison.get("status") in {"ahead", "identical"} def registry_issue_number(self, *, create: bool) -> int | None: + cached = getattr(self, "_registry_issue_cache", None) + if cached is not None: + return cached values = self._paged_api( f"repos/{self.repository}/issues?state=all&labels={self.config.pipeline.registry_label}&per_page=100" ) @@ -1015,7 +1020,9 @@ def registry_issue_number(self, *, create: bool) -> int | None: # workers can both create a registry after observing none. The # lowest issue number is deterministic and comments remain trusted # by authenticated author, so all workers safely converge there. - return min(int(value["number"]) for value in matches) + canonical = min(int(value["number"]) for value in matches) + self._registry_issue_cache = canonical + return canonical if not create: return None self.ensure_labels_exist() @@ -1033,7 +1040,9 @@ def registry_issue_number(self, *, create: bool) -> int | None: ) # Re-read after creation so a concurrent lower-numbered creator wins # before this worker writes any metadata to its candidate issue. - return self.registry_issue_number(create=False) or int(created["number"]) + canonical = self.registry_issue_number(create=False) or int(created["number"]) + self._registry_issue_cache = canonical + return canonical def issue_comment(self, number: int, body: str) -> None: self._json( @@ -1355,6 +1364,11 @@ def snapshot( body, self.config.dependency_directive ), priority_at=marker_priority_at(marker), + integration_batch_id=( + str(marker.get("integration_batch_id") or "") or None + if marker + else None + ), ) entry.classify( self.config, @@ -1380,6 +1394,7 @@ def queue(self) -> list[QueueEntry]: entries, key=lambda entry: ( entry.priority_at or entry.queued_at or "9999", + entry.number if entry.priority_at else 0, entry.queued_at or "9999", entry.number, ), @@ -1453,35 +1468,54 @@ def dependency_is_merged(self, number: int) -> bool: ) return comparison.get("status") in {"ahead", "identical"} - def merge(self, number: int, head_sha: str) -> str: - authorization = self._json( - "pr", - "view", - str(number), - "--repo", - self.repository, - "--json", - "headRefOid,isDraft,labels,state", - ) - labels = { - str(label.get("name") or "") for label in authorization.get("labels") or [] - } - if authorization.get("state") != "OPEN": - raise QueueError(f"PR #{number} is no longer open") - if authorization.get("isDraft"): - raise QueueError(f"PR #{number} returned to draft before merge") - if authorization.get("headRefOid") != head_sha: - raise QueueError(f"PR #{number} changed immediately before merge") + def merge( + self, + number: int, + head_sha: str, + *, + authorization_entry: QueueEntry | None = None, + ) -> str: + if authorization_entry is None: + authorization = self._json( + "pr", + "view", + str(number), + "--repo", + self.repository, + "--json", + "headRefOid,isDraft,labels,state", + ) + labels = { + str(label.get("name") or "") + for label in authorization.get("labels") or [] + } + if authorization.get("state") != "OPEN": + raise QueueError(f"PR #{number} is no longer open") + if authorization.get("isDraft"): + raise QueueError(f"PR #{number} returned to draft before merge") + if authorization.get("headRefOid") != head_sha: + raise QueueError(f"PR #{number} changed immediately before merge") + marker = queue_marker_for_client(self, self.comments(number)) + integration_batch_id = str(marker.get("integration_batch_id") or "") if marker else "" + else: + labels = set(authorization_entry.labels) + if authorization_entry.is_draft: + raise QueueError(f"PR #{number} returned to draft before merge") + if authorization_entry.head_sha != head_sha: + raise QueueError(f"PR #{number} changed immediately before merge") + marker = { + "state": authorization_entry.queue_state, + "head_sha": authorization_entry.queued_head_sha, + } + integration_batch_id = authorization_entry.integration_batch_id or "" if self.config.queue_label not in labels or self.config.blocked_label in labels: raise QueueError(f"PR #{number} queue authorization was revoked") - marker = queue_marker_for_client(self, self.comments(number)) if ( not marker or marker.get("state") != "queued" or marker.get("head_sha") != head_sha ): raise QueueError(f"PR #{number} durable queue authorization was revoked") - integration_batch_id = str(marker.get("integration_batch_id") or "") if integration_batch_id: integration = latest_payload( self.comments(number), @@ -2284,6 +2318,7 @@ def freeze_queue( entries_by_number.values(), key=lambda entry: ( entry.priority_at or entry.queued_at or "9999", + entry.number if entry.priority_at else 0, entry.queued_at or "9999", entry.number, ), @@ -2489,7 +2524,8 @@ def command_merge( frozen_batch: dict[str, Any] | None = None, active_numbers: set[int] | None = None, ) -> str: - require_running_pipeline(client) + if frozen_entry is None or frozen_batch is None: + require_running_pipeline(client) number = frozen_entry.number if frozen_entry is not None else client.resolve_pr(selector) batch = frozen_batch if batch is not None and str(batch.get("batch_id") or "") != batch_id: @@ -2567,7 +2603,11 @@ def command_merge( if fresh.queued_at and str(batch.get("frozen_at") or "") < fresh.queued_at: raise QueueError(f"batch {batch_id} predates the current queue authorization") - merge_sha = client.merge(number, fresh.head_sha) + merge_sha = client.merge( + number, + fresh.head_sha, + authorization_entry=fresh, + ) merged_comments = client.comments(number) intent = latest_intent(merged_comments, client.trusted_logins) integration = latest_payload( @@ -2889,6 +2929,20 @@ def release_follow_needed(client: GitHub) -> bool: return True +def should_settle_batch(client: GitHub, entries: list[QueueEntry]) -> bool: + has_ready = any( + client.config.queue_label in entry.labels and entry.state == "ready" + for entry in entries + ) + has_near_ready = any( + client.config.queue_label not in entry.labels + and client.config.blocked_label not in entry.labels + and entry.state == "waiting" + for entry in entries + ) + return has_ready and has_near_ready + + def command_react( client: GitHub, *, @@ -2908,6 +2962,23 @@ def command_react( emit=False, captured_entries=captured_entries, ) + if ( + client.config.pipeline.batch_settle_seconds + and should_settle_batch(client, captured_entries) + ): + time.sleep(client.config.pipeline.batch_settle_seconds) + settled_entries: list[QueueEntry] = [] + settled = command_promote( + client, + emit=False, + captured_entries=settled_entries, + ) + promoted["promoted"] = list( + dict.fromkeys(promoted["promoted"] + settled["promoted"]) + ) + promoted["blocked"] = promoted["blocked"] + settled["blocked"] + promoted["waiting"] = settled["waiting"] + captured_entries = settled_entries frozen = freeze_queue(client, known_entries=captured_entries) integrations: list[dict[str, Any]] = [] if ( diff --git a/src/agent_merge_queue/config.py b/src/agent_merge_queue/config.py index 99d7571..e9439b6 100644 --- a/src/agent_merge_queue/config.py +++ b/src/agent_merge_queue/config.py @@ -44,6 +44,7 @@ class PipelineConfig: thread_active_hours: int ci_workflows: tuple[str, ...] deploy_workflows: tuple[str, ...] + batch_settle_seconds: int ready_to_merge_target_minutes: int merge_to_live_target_minutes: int auto_promote: bool @@ -116,6 +117,7 @@ class QueueConfig: thread_active_hours = 72 ci_workflows = ["CI"] deploy_workflows = ["Deploy"] +batch_settle_seconds = 15 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true @@ -158,6 +160,14 @@ def _positive_int(value: Any, field: str, default: int) -> int: return value +def _non_negative_int(value: Any, field: str, default: int) -> int: + if value is None: + return default + if not isinstance(value, int) or isinstance(value, bool) or value < 0: + raise ConfigError(f"{field} must be a non-negative integer") + return value + + def _boolean(value: Any, field: str, default: bool) -> bool: if value is None: return default @@ -415,6 +425,11 @@ def parse_config(payload: dict[str, Any]) -> QueueConfig: pipeline.get("deploy_workflows", ["Deploy"]), "pipeline.deploy_workflows", ), + batch_settle_seconds=_non_negative_int( + pipeline.get("batch_settle_seconds"), + "pipeline.batch_settle_seconds", + 15, + ), ready_to_merge_target_minutes=_positive_int( pipeline.get("ready_to_merge_target_minutes"), "pipeline.ready_to_merge_target_minutes", diff --git a/tests/test_cli.py b/tests/test_cli.py index df1ee97..386ff42 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -42,6 +42,7 @@ queue_from_intent, queue_timestamp, reusable_batch, + should_settle_batch, structured_dependencies, ) from agent_merge_queue.config import parse_config @@ -144,6 +145,30 @@ def test_intent_request_time_is_the_stable_queue_priority(self) -> None: client.comment.call_args.args[1], ) + def test_settle_window_only_collects_a_mixed_ready_burst(self) -> None: + ready = entry(1) + waiting = entry(2, state="waiting") + waiting.labels = ["deploy-requested"] + client = Mock() + client.config = CONFIG + + self.assertTrue(should_settle_batch(client, [ready, waiting])) + self.assertFalse(should_settle_batch(client, [ready])) + waiting.labels.append("merge-queue-blocked") + self.assertFalse(should_settle_batch(client, [ready, waiting])) + + def test_simultaneous_intents_use_pull_request_number_as_fifo_tiebreaker(self) -> None: + first = entry(1) + second = entry(2) + first.priority_at = second.priority_at = "2026-06-20T00:00:00Z" + first.queued_at = "2026-06-20T00:00:02Z" + second.queued_at = "2026-06-20T00:00:01Z" + client = object.__new__(GitHub) + client.queued_numbers = Mock(return_value=[2, 1]) + client.snapshot = Mock(side_effect=lambda number: {1: first, 2: second}[number]) + + self.assertEqual([value.number for value in client.queue()], [1, 2]) + def test_registry_race_converges_on_the_lowest_issue_number(self) -> None: client = object.__new__(GitHub) client.config = CONFIG @@ -190,6 +215,11 @@ def test_frozen_merge_fast_path_never_rescans_the_whole_queue(self) -> None: known_source_paths=["a.py"], known_generated_paths=[], ) + client.merge.assert_called_once_with( + 1, + value.head_sha, + authorization_entry=value, + ) def test_generated_paths_are_configurable(self) -> None: self.assertTrue( @@ -873,6 +903,25 @@ def test_final_merge_rechecks_durable_authorization(self) -> None: client.merge(1, value.head_sha) self.assertEqual(client._json.call_count, 1) + def test_prevalidated_merge_avoids_duplicate_authorization_reads(self) -> None: + value = entry(1) + client = object.__new__(GitHub) + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client._json = Mock(return_value={"merged": True, "sha": "m" * 40}) + client.comments = Mock() + + result = client.merge( + 1, + value.head_sha, + authorization_entry=value, + ) + + self.assertEqual(result, "m" * 40) + client.comments.assert_not_called() + self.assertEqual(client._json.call_count, 1) + def test_final_integration_merge_requires_current_exact_head_intent(self) -> None: integration_head = "f" * 40 source_head = "a" * 40 diff --git a/tests/test_config.py b/tests/test_config.py index ffab40b..787f82b 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -183,6 +183,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: "pipeline": { "ci_workflows": ["Main CI"], "deploy_workflows": ["Production"], + "batch_settle_seconds": 30, "auto_promote": False, "verifications": [ { @@ -196,6 +197,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: } ) self.assertEqual(config.pipeline.ci_workflows, ("Main CI",)) + self.assertEqual(config.pipeline.batch_settle_seconds, 30) self.assertFalse(config.pipeline.auto_promote) self.assertEqual(config.pipeline.verifications[0].expected_status, 200) self.assertEqual(config.integration.mode, "all") @@ -231,6 +233,16 @@ def test_rejects_invalid_integration_mode_and_boolean(self) -> None: "pipeline": {"intent_scope": "pull-request"}, } ) + with self.assertRaisesRegex(ConfigError, "batch_settle_seconds"): + parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "pipeline": {"batch_settle_seconds": -1}, + } + ) if __name__ == "__main__": From 606b440263c66ff71156ca0bc331c69a34419b5f Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:26:21 -0700 Subject: [PATCH 03/14] Keep release ownership through CI retries --- README.md | 1 + src/agent_merge_queue/cli.py | 14 +++++++++-- src/agent_merge_queue/config.py | 7 ++++++ src/agent_merge_queue/pipeline.py | 25 ++++++++++++++++++-- tests/test_config.py | 2 ++ tests/test_pipeline.py | 39 +++++++++++++++++++++++++++++++ 6 files changed, 84 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index bc1b040..7c06e40 100644 --- a/README.md +++ b/README.md @@ -172,6 +172,7 @@ can pause further merges until `deploybot unpause`. ci_workflows = ["CI"] deploy_workflows = ["Deploy"] batch_settle_seconds = 15 +ci_failure_grace_seconds = 90 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 9722efa..d7f6d1d 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -809,6 +809,16 @@ def open_pull_request_numbers(self) -> list[int]: ) ] + def integration_pull_request_numbers(self) -> list[int]: + prefix = self.config.integration.branch_prefix.rstrip("/") + "/" + return [ + int(value["number"]) + for value in self._paged_api( + f"repos/{self.repository}/pulls?state=open&per_page=100" + ) + if str((value.get("head") or {}).get("ref") or "").startswith(prefix) + ] + def intent_numbers(self) -> list[int]: values = self._paged_api( f"repos/{self.repository}/pulls?state=open&per_page=100" @@ -822,7 +832,7 @@ def intent_numbers(self) -> list[int]: def active_integration_sources(self) -> set[int]: sources: set[int] = set() - for number in self.open_pull_request_numbers(): + for number in self.integration_pull_request_numbers(): marker = latest_payload( self.comments(number), INTEGRATION_MARKER, @@ -2026,7 +2036,7 @@ def command_promote( def promote_integrations(client: GitHub) -> list[int]: promoted: list[int] = [] - for number in client.open_pull_request_numbers(): + for number in client.integration_pull_request_numbers(): comments = client.comments(number) integration = latest_payload( comments, diff --git a/src/agent_merge_queue/config.py b/src/agent_merge_queue/config.py index e9439b6..9d5c032 100644 --- a/src/agent_merge_queue/config.py +++ b/src/agent_merge_queue/config.py @@ -45,6 +45,7 @@ class PipelineConfig: ci_workflows: tuple[str, ...] deploy_workflows: tuple[str, ...] batch_settle_seconds: int + ci_failure_grace_seconds: int ready_to_merge_target_minutes: int merge_to_live_target_minutes: int auto_promote: bool @@ -118,6 +119,7 @@ class QueueConfig: ci_workflows = ["CI"] deploy_workflows = ["Deploy"] batch_settle_seconds = 15 +ci_failure_grace_seconds = 90 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true @@ -430,6 +432,11 @@ def parse_config(payload: dict[str, Any]) -> QueueConfig: "pipeline.batch_settle_seconds", 15, ), + ci_failure_grace_seconds=_non_negative_int( + pipeline.get("ci_failure_grace_seconds"), + "pipeline.ci_failure_grace_seconds", + 90, + ), ready_to_merge_target_minutes=_positive_int( pipeline.get("ready_to_merge_target_minutes"), "pipeline.ready_to_merge_target_minutes", diff --git a/src/agent_merge_queue/pipeline.py b/src/agent_merge_queue/pipeline.py index 1239116..88b0e0d 100644 --- a/src/agent_merge_queue/pipeline.py +++ b/src/agent_merge_queue/pipeline.py @@ -215,6 +215,8 @@ def follow_release( last_verifications: list[dict[str, Any]] = [] dispatched_deployments: list[dict[str, Any]] = [] dispatched_for: set[tuple[str, int]] = set() + failed_ci_key: tuple[str, int] | None = None + failed_ci_deadline = 0.0 while True: main_sha = client.base_sha() runs = client.workflow_runs() @@ -222,7 +224,26 @@ def follow_release( main_sha=main_sha, runs=runs, config=client.config.pipeline ) observed_sha = main_sha - if value["state"] in {"ci-failed", "deploy-failed"}: + clock: float | None = None + if value["state"] == "ci-failed": + ci = value.get("latest_ci") or {} + key = (main_sha, int(ci.get("id") or 0)) + clock = time.monotonic() + if key != failed_ci_key: + failed_ci_key = key + failed_ci_deadline = ( + clock + client.config.pipeline.ci_failure_grace_seconds + ) + if ( + not client.config.pipeline.ci_failure_grace_seconds + or clock >= failed_ci_deadline + ): + return { + **value, + "dispatched_deployments": dispatched_deployments, + "verifications": [], + } + elif value["state"] == "deploy-failed": return { **value, "dispatched_deployments": dispatched_deployments, @@ -250,7 +271,7 @@ def follow_release( "dispatched_deployments": dispatched_deployments, "verifications": checks, } - if time.monotonic() >= deadline: + if (clock if clock is not None else time.monotonic()) >= deadline: state = ( "verify-failed" if value["state"] == "verified" and last_verifications diff --git a/tests/test_config.py b/tests/test_config.py index 787f82b..75a2244 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -184,6 +184,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: "ci_workflows": ["Main CI"], "deploy_workflows": ["Production"], "batch_settle_seconds": 30, + "ci_failure_grace_seconds": 45, "auto_promote": False, "verifications": [ { @@ -198,6 +199,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: ) self.assertEqual(config.pipeline.ci_workflows, ("Main CI",)) self.assertEqual(config.pipeline.batch_settle_seconds, 30) + self.assertEqual(config.pipeline.ci_failure_grace_seconds, 45) self.assertFalse(config.pipeline.auto_promote) self.assertEqual(config.pipeline.verifications[0].expected_status, 200) self.assertEqual(config.integration.mode, "all") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 2d3da9f..c062175 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -172,6 +172,45 @@ def test_follow_dispatches_deploy_after_token_dispatched_ci(self) -> None: self.assertEqual(result["state"], "verified") self.assertEqual(result["dispatched_deployments"][0]["id"], 9) + def test_follow_absorbs_a_ci_rerun_during_failure_grace(self) -> None: + sha = "a" * 40 + failed = { + "id": 1, + "name": "CI", + "head_sha": sha, + "status": "completed", + "conclusion": "failure", + "event": "workflow_dispatch", + } + retrying = {**failed, "status": "in_progress", "conclusion": None} + passed = {**failed, "conclusion": "success"} + deploy = { + "id": 2, + "name": "Deploy", + "head_sha": sha, + "status": "completed", + "conclusion": "success", + } + client = Mock() + client.config = CONFIG + client.base_sha.return_value = sha + client.workflow_runs.side_effect = [ + [failed], + [retrying], + [passed, deploy], + ] + + with ( + patch("agent_merge_queue.pipeline.time.sleep"), + patch( + "agent_merge_queue.pipeline.time.monotonic", + side_effect=[0, 1, 2], + ), + ): + result = follow_release(client, timeout_seconds=10, poll_seconds=1) + + self.assertEqual(result["state"], "verified") + def test_follow_switches_to_newer_cumulative_main(self) -> None: old = "a" * 40 new = "b" * 40 From 989007d09562929b4be9022e0ab6e17cb41f6714 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:35:50 -0700 Subject: [PATCH 04/14] Parallelize burst promotion safely --- README.md | 1 + src/agent_merge_queue/cli.py | 118 +++++++++++++++++++++----------- src/agent_merge_queue/config.py | 7 ++ tests/test_cli.py | 35 ++++++++++ tests/test_config.py | 2 + 5 files changed, 122 insertions(+), 41 deletions(-) diff --git a/README.md b/README.md index 7c06e40..7850990 100644 --- a/README.md +++ b/README.md @@ -173,6 +173,7 @@ ci_workflows = ["CI"] deploy_workflows = ["Deploy"] batch_settle_seconds = 15 ci_failure_grace_seconds = 90 +promotion_workers = 4 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index d7f6d1d..6281be6 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -12,6 +12,7 @@ import subprocess import sys import time +from concurrent.futures import ThreadPoolExecutor from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path @@ -1245,6 +1246,31 @@ def files(self, number: int) -> list[dict[str, Any]]: f"repos/{self.repository}/pulls/{number}/files?per_page=100" ) + def changed_paths(self, number: int) -> tuple[list[str], list[str]]: + source_paths: list[str] = [] + generated_paths: list[str] = [] + for value in self.files(number): + path = str(value.get("filename") or "") + if not path: + continue + if path in self.config.generated_paths: + generated_paths.append(path) + source_paths.append(path) + continue + target = ( + generated_paths + if generated_only_change( + path, + value.get("patch"), + generated_paths=self.config.generated_paths, + generated_version_paths=self.config.generated_version_paths, + asset_version_pattern=self.config.asset_version_pattern, + ) + else source_paths + ) + target.append(path) + return source_paths, generated_paths + def review_threads(self, number: int) -> list[dict[str, Any]]: data = self._json( "api", @@ -1280,6 +1306,7 @@ def snapshot( known_comments: list[dict[str, Any]] | None = None, known_source_paths: list[str] | None = None, known_generated_paths: list[str] | None = None, + defer_paths_until_ready: bool = False, ) -> QueueEntry: fields = ",".join( ( @@ -1321,29 +1348,11 @@ def snapshot( checks = check_states(pull.get("statusCheckRollup") or []) source_paths = list(known_source_paths or []) generated_paths = list(known_generated_paths or []) - if known_source_paths is None or known_generated_paths is None: - source_paths = [] - generated_paths = [] - for value in self.files(number): - path = str(value.get("filename") or "") - if not path: - continue - if path in self.config.generated_paths: - generated_paths.append(path) - source_paths.append(path) - continue - target = ( - generated_paths - if generated_only_change( - path, - value.get("patch"), - generated_paths=self.config.generated_paths, - generated_version_paths=self.config.generated_version_paths, - asset_version_pattern=self.config.asset_version_pattern, - ) - else source_paths - ) - target.append(path) + paths_are_known = ( + known_source_paths is not None and known_generated_paths is not None + ) + if not paths_are_known and not defer_paths_until_ready: + source_paths, generated_paths = self.changed_paths(number) body = str(pull.get("body") or "") entry = QueueEntry( @@ -1385,6 +1394,8 @@ def snapshot( require_marker=require_marker, allow_blocked_label=allow_blocked_label, ) + if not paths_are_known and defer_paths_until_ready and entry.state == "ready": + entry.source_paths, entry.generated_paths = self.changed_paths(number) return entry def queued_numbers(self) -> list[int]: @@ -1969,49 +1980,54 @@ def command_promote( waiting: list[dict[str, Any]] = [] blocked: list[dict[str, Any]] = [] integration_sources = client.active_integration_sources() - for number in client.intent_numbers(): + + def evaluate( + number: int, + ) -> tuple[str, dict[str, Any] | None, QueueEntry | None]: if number in integration_sources: - waiting.append( - {"number": number, "reasons": ["cumulative integration PR is active"]} + return ( + "waiting", + {"number": number, "reasons": ["cumulative integration PR is active"]}, + None, ) - continue comments = client.comments(number) intent = latest_intent(comments, client.trusted_logins) if not intent or intent.get("state") != "requested": - continue + return "skip", None, None entry = client.snapshot( number, require_marker=False, allow_blocked_label=True, known_comments=comments, + defer_paths_until_ready=True, ) - if captured_entries is not None: - captured_entries.append(entry) if ( client.config.pipeline.intent_scope == "head" and intent.get("requested_head") != entry.head_sha ): - waiting.append( + return ( + "waiting", { "number": number, "reasons": [ "deploy intent is bound to an older head; the trusted source " "agent must run deploybot refresh-request" ], - } + }, + entry, ) - continue if client.config.blocked_label in entry.labels: - waiting.append( + return ( + "waiting", { "number": number, "reasons": [ "repair is blocked; the trusted source agent must run " "deploybot resume after fresh gates pass" ], - } + }, + entry, ) - continue if entry.state == "ready": changed = queue_from_intent( client, @@ -2021,13 +2037,33 @@ def command_promote( labels=set(entry.labels), ) if changed: - promoted.append(number) - elif entry.state == "blocked": + return "promoted", {"number": number}, entry + return "ready", None, entry + if entry.state == "blocked": reason = "; ".join(entry.reasons or ["blocked"]) record_repair(client, entry, intent, reason) - blocked.append({"number": number, "reason": reason}) - else: - waiting.append({"number": number, "reasons": entry.reasons or []}) + return "blocked", {"number": number, "reason": reason}, entry + return ( + "waiting", + {"number": number, "reasons": entry.reasons or []}, + entry, + ) + + numbers = sorted(client.intent_numbers()) + if numbers: + with ThreadPoolExecutor( + max_workers=min(client.config.pipeline.promotion_workers, len(numbers)) + ) as executor: + outcomes = executor.map(evaluate, numbers) + for state, payload, entry in outcomes: + if captured_entries is not None and entry is not None: + captured_entries.append(entry) + if state == "promoted" and payload is not None: + promoted.append(int(payload["number"])) + elif state == "blocked" and payload is not None: + blocked.append(payload) + elif state == "waiting" and payload is not None: + waiting.append(payload) result = {"promoted": promoted, "waiting": waiting, "blocked": blocked} if emit: print(json.dumps(result, indent=2, sort_keys=True)) diff --git a/src/agent_merge_queue/config.py b/src/agent_merge_queue/config.py index 9d5c032..9e76e51 100644 --- a/src/agent_merge_queue/config.py +++ b/src/agent_merge_queue/config.py @@ -46,6 +46,7 @@ class PipelineConfig: deploy_workflows: tuple[str, ...] batch_settle_seconds: int ci_failure_grace_seconds: int + promotion_workers: int ready_to_merge_target_minutes: int merge_to_live_target_minutes: int auto_promote: bool @@ -120,6 +121,7 @@ class QueueConfig: deploy_workflows = ["Deploy"] batch_settle_seconds = 15 ci_failure_grace_seconds = 90 +promotion_workers = 4 ready_to_merge_target_minutes = 15 merge_to_live_target_minutes = 10 auto_promote = true @@ -437,6 +439,11 @@ def parse_config(payload: dict[str, Any]) -> QueueConfig: "pipeline.ci_failure_grace_seconds", 90, ), + promotion_workers=_positive_int( + pipeline.get("promotion_workers"), + "pipeline.promotion_workers", + 4, + ), ready_to_merge_target_minutes=_positive_int( pipeline.get("ready_to_merge_target_minutes"), "pipeline.ready_to_merge_target_minutes", diff --git a/tests/test_cli.py b/tests/test_cli.py index 386ff42..b091d5d 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -254,6 +254,41 @@ def test_generated_paths_are_configurable(self) -> None: ) ) + def test_waiting_promotion_defers_changed_file_fetch(self) -> None: + client = object.__new__(GitHub) + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"trusted"} + client.comments = Mock(return_value=[]) + client.changed_paths = Mock(return_value=(["a.py"], [])) + client._json = Mock( + return_value={ + "baseRefName": "main", + "body": "", + "headRefOid": "a" * 40, + "isDraft": False, + "labels": [{"name": "deploy-requested"}], + "mergeStateStatus": "CLEAN", + "mergeable": "MERGEABLE", + "number": 1, + "state": "OPEN", + "statusCheckRollup": [], + "title": "Waiting", + "url": "https://example.test/1", + } + ) + + value = client.snapshot( + 1, + require_marker=False, + allow_blocked_label=True, + defer_paths_until_ready=True, + ) + + self.assertEqual(value.state, "waiting") + client.changed_paths.assert_not_called() + def test_required_checks_do_not_accept_skipped(self) -> None: states = check_states( [ diff --git a/tests/test_config.py b/tests/test_config.py index 75a2244..e031d43 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -185,6 +185,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: "deploy_workflows": ["Production"], "batch_settle_seconds": 30, "ci_failure_grace_seconds": 45, + "promotion_workers": 3, "auto_promote": False, "verifications": [ { @@ -200,6 +201,7 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: self.assertEqual(config.pipeline.ci_workflows, ("Main CI",)) self.assertEqual(config.pipeline.batch_settle_seconds, 30) self.assertEqual(config.pipeline.ci_failure_grace_seconds, 45) + self.assertEqual(config.pipeline.promotion_workers, 3) self.assertFalse(config.pipeline.auto_promote) self.assertEqual(config.pipeline.verifications[0].expected_status, 200) self.assertEqual(config.integration.mode, "all") From 719d5810ff33575b8ebe4d553fd38c487f78ce81 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:49:48 -0700 Subject: [PATCH 05/14] Keep near-ready overlaps in one batch --- README.md | 4 ++ src/agent_merge_queue/cli.py | 80 ++++++++++++++++++++-- tests/test_cli.py | 128 +++++++++++++++++++++++++++++++++++ 3 files changed, 208 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 7850990..60389e3 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,10 @@ work, and creates integration PRs when configured. A conflict produces a repair handoff containing the source thread, base/head SHAs, source paths, and one return command: +In `overlap` mode, a ready source waits when another active, near-ready intent +belongs to the same source-overlap component. Unrelated ready work still drains, +and the held component freezes together once its remaining gates pass. + ```bash deploybot resume ``` diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 6281be6..352339d 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -2347,12 +2347,18 @@ def as_dict(self) -> dict[str, Any]: def freeze_queue( - client: GitHub, *, known_entries: list[QueueEntry] | None = None + client: GitHub, + *, + known_entries: list[QueueEntry] | None = None, + held_numbers: set[int] | None = None, ) -> FreezeResult: + held_numbers = held_numbers or set() if known_entries is None: - all_entries = client.queue() + all_entries = [ + entry for entry in client.queue() if entry.number not in held_numbers + ] else: - queued_numbers = set(client.queued_numbers()) + queued_numbers = set(client.queued_numbers()) - held_numbers entries_by_number = { entry.number: entry for entry in known_entries @@ -2989,6 +2995,59 @@ def should_settle_batch(client: GitHub, entries: list[QueueEntry]) -> bool: return has_ready and has_near_ready +def near_ready_overlap_holds( + client: GitHub, entries: list[QueueEntry] +) -> dict[int, list[int]]: + """Keep ready overlap peers together while another active intent settles.""" + if client.config.integration.mode != "overlap": + return {} + ready = { + entry.number: entry + for entry in entries + if entry.state == "ready" + and client.config.queue_label in entry.labels + and client.config.blocked_label not in entry.labels + } + waiting = { + entry.number: entry + for entry in entries + if entry.state == "waiting" + and client.config.queue_label not in entry.labels + and client.config.blocked_label not in entry.labels + } + if not ready or not waiting: + return {} + + def load_paths(entry: QueueEntry) -> tuple[int, list[str], list[str]]: + source_paths, generated_paths = client.changed_paths(entry.number) + return entry.number, source_paths, generated_paths + + values = list(waiting.values()) + with ThreadPoolExecutor( + max_workers=min(client.config.pipeline.promotion_workers, len(values)) + ) as executor: + for number, source_paths, generated_paths in executor.map(load_paths, values): + waiting[number].source_paths = sorted(set(source_paths)) + waiting[number].generated_paths = sorted(set(generated_paths)) + + waiting_numbers = set(waiting) + holds: dict[int, list[int]] = {} + for group in overlap_groups([*ready.values(), *waiting.values()]): + members = {int(value) for value in group["pull_requests"]} + waiting_peers = sorted(members & waiting_numbers) + if not waiting_peers: + continue + for number in sorted(members & ready.keys()): + entry = ready[number] + comments = client.comments(number) + marker = latest_batch_marker(comments, coordinator_logins(client)) + completed = completed_batch_ids(comments, coordinator_logins(client)) + if active_batch([entry], {number: marker}, completed) is not None: + continue + holds[number] = waiting_peers + return holds + + def command_react( client: GitHub, *, @@ -3025,7 +3084,20 @@ def command_react( promoted["blocked"] = promoted["blocked"] + settled["blocked"] promoted["waiting"] = settled["waiting"] captured_entries = settled_entries - frozen = freeze_queue(client, known_entries=captured_entries) + overlap_holds = near_ready_overlap_holds(client, captured_entries) + promoted["held"] = [ + { + "number": number, + "overlapping_waiting": waiting, + "reason": "waiting for overlapping deploy intents to finish gates", + } + for number, waiting in overlap_holds.items() + ] + frozen = freeze_queue( + client, + known_entries=captured_entries, + held_numbers=set(overlap_holds), + ) integrations: list[dict[str, Any]] = [] if ( frozen.batch is not None diff --git a/tests/test_cli.py b/tests/test_cli.py index b091d5d..00d81ef 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -36,6 +36,7 @@ marker_queued_at, marker_priority_at, main, + near_ready_overlap_holds, new_batch, overlap_groups, queue_state_body, @@ -157,6 +158,78 @@ def test_settle_window_only_collects_a_mixed_ready_burst(self) -> None: waiting.labels.append("merge-queue-blocked") self.assertFalse(should_settle_batch(client, [ready, waiting])) + def test_overlap_mode_holds_only_ready_members_of_near_ready_components( + self, + ) -> None: + config = parse_config( + { + "queue": {"required_checks": ["CI"], "trusted_actors": ["trusted"]}, + "integration": {"mode": "overlap"}, + } + ) + overlapping = entry(1, "shared.py") + independent = entry(2, "other.py") + waiting = entry(3, state="waiting") + waiting.labels = ["deploy-requested"] + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.changed_paths.return_value = (["shared.py"], []) + client.comments.return_value = [] + + holds = near_ready_overlap_holds( + client, [overlapping, independent, waiting] + ) + + self.assertEqual(holds, {1: [3]}) + client.changed_paths.assert_called_once_with(3) + + def test_overlap_hold_includes_transitive_ready_members(self) -> None: + config = parse_config( + { + "queue": {"required_checks": ["CI"], "trusted_actors": ["trusted"]}, + "integration": {"mode": "overlap"}, + } + ) + first = entry(1, "shared.py", "bridge.py") + second = entry(2, "bridge.py") + waiting = entry(3, state="waiting") + waiting.labels = ["deploy-requested"] + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.changed_paths.return_value = (["shared.py"], []) + client.comments.return_value = [] + + self.assertEqual( + near_ready_overlap_holds(client, [first, second, waiting]), + {1: [3], 2: [3]}, + ) + + def test_overlap_hold_does_not_rewrite_an_existing_frozen_batch(self) -> None: + config = parse_config( + { + "queue": {"required_checks": ["CI"], "trusted_actors": ["trusted"]}, + "integration": {"mode": "overlap"}, + } + ) + ready = entry(1, "shared.py") + waiting = entry(2, state="waiting") + waiting.labels = ["deploy-requested"] + batch = new_batch([ready], frozen_at="2026-06-20T00:01:00Z") + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.changed_paths.return_value = (["shared.py"], []) + client.comments.return_value = [] + with patch( + "agent_merge_queue.cli.latest_batch_marker", return_value=batch + ): + self.assertEqual( + near_ready_overlap_holds(client, [ready, waiting]), + {}, + ) + def test_simultaneous_intents_use_pull_request_number_as_fifo_tiebreaker(self) -> None: first = entry(1) second = entry(2) @@ -1380,6 +1453,61 @@ def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: drain.assert_not_called() self.assertEqual(result["integrations"], [{"number": 99}]) + def test_reactor_holds_overlap_peer_but_drains_independent_work(self) -> None: + config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + "coordinator_actors": ["coordinator"], + }, + "integration": {"mode": "overlap"}, + } + ) + overlapping = entry(1, "shared.py") + waiting = entry(2, state="waiting") + waiting.labels = ["deploy-requested"] + independent = entry(3, "other.py") + entries = [overlapping, waiting, independent] + batch = new_batch([independent], frozen_at="2026-06-20T00:01:00Z") + frozen = FreezeResult(batch, [independent], [], [], []) + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.pipeline_control.return_value = {"state": "running"} + client.changed_paths.return_value = (["shared.py"], []) + client.comments.return_value = [] + + def promote(*_args, **kwargs): + kwargs["captured_entries"].extend(entries) + return { + "promoted": [1, 3], + "waiting": [{"number": 2, "reasons": ["CI is not complete"]}], + "blocked": [], + } + + with ( + patch("agent_merge_queue.cli.promote_integrations", return_value=[]), + patch("agent_merge_queue.cli.command_promote", side_effect=promote), + patch( + "agent_merge_queue.cli.freeze_queue", return_value=frozen + ) as freeze, + patch( + "agent_merge_queue.cli.command_drain", + return_value={"merged": [{"number": 3, "merge_sha": "a" * 40}]}, + ), + redirect_stdout(io.StringIO()), + ): + result = command_react(client, follow=False, timeout_seconds=10) + + freeze.assert_called_once_with( + client, + known_entries=entries, + held_numbers={1}, + ) + self.assertEqual(result["promoted"]["held"][0]["number"], 1) + self.assertEqual(result["drain"]["merged"][0]["number"], 3) + def test_reactor_dispatches_ci_once_after_a_merged_batch(self) -> None: client = Mock() client.config = CONFIG From 545deed94021ac3ab994e326586d14e81a326f5f Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 20:58:41 -0700 Subject: [PATCH 06/14] Fail closed before partial integration drains --- README.md | 4 ++ src/agent_merge_queue/cli.py | 16 +++--- src/agent_merge_queue/doctor.py | 38 ++++++++++++++ tests/test_cli.py | 90 +++++++++++++++++++++++++++++++++ tests/test_doctor.py | 57 +++++++++++++++++++++ 5 files changed, 199 insertions(+), 6 deletions(-) diff --git a/README.md b/README.md index 60389e3..f939246 100644 --- a/README.md +++ b/README.md @@ -194,6 +194,10 @@ expected_status = 200 mode = "overlap" ``` +For `overlap` or `all` mode with the hosted coordinator, enable **Allow GitHub +Actions to create and approve pull requests** under the repository's Actions +workflow-permission settings. `deploybot doctor` reports this prerequisite. + ## Review providers Required checks are always exact-head gates. Optional providers use normalized diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 352339d..5b7dd27 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -3118,12 +3118,6 @@ def command_react( "next_batch": [], } else: - drained = command_drain( - client, - json_output=False, - emit=False, - initial_frozen=frozen, - ) if ( frozen.batch is not None and client.config.integration.mode == "overlap" @@ -3143,6 +3137,16 @@ def command_react( entries=overlap_entries, ) ) + # Establish durable ownership for every overlapping source before any + # independent member of the frozen batch is allowed to merge. A setup + # failure while creating the integration PR must leave the batch intact + # and must never strand an already-merged partial batch before CI. + drained = command_drain( + client, + json_output=False, + emit=False, + initial_frozen=frozen, + ) dispatched_ci: list[dict[str, Any]] = [] if dispatch_ci and drained.get("merged"): try: diff --git a/src/agent_merge_queue/doctor.py b/src/agent_merge_queue/doctor.py index 21931c0..46968c2 100644 --- a/src/agent_merge_queue/doctor.py +++ b/src/agent_merge_queue/doctor.py @@ -166,6 +166,44 @@ def diagnose( ) ) + if ( + config.integration.mode in {"overlap", "all"} + and "github-actions[bot]" in config.coordinator_actors + ): + code, workflow_permissions, detail = _json( + "api", f"repos/{repo}/actions/permissions/workflow", cwd=root + ) + if code: + rows.append( + row( + "actions-integration-prs", + "warn", + detail or "Could not read GitHub Actions workflow permissions", + "Confirm Actions may create integration pull requests.", + ) + ) + else: + allowed = bool( + (workflow_permissions or {}).get("can_approve_pull_request_reviews") + ) + rows.append( + row( + "actions-integration-prs", + "ok" if allowed else "fail", + ( + "GitHub Actions may create integration pull requests" + if allowed + else "GitHub Actions cannot create integration pull requests" + ), + ( + None + if allowed + else "Enable Settings > Actions > General > Allow GitHub " + "Actions to create and approve pull requests." + ), + ) + ) + code, labels, detail = _json( "label", "list", "--repo", repo, "--limit", "1000", "--json", "name", cwd=root ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 00d81ef..405046c 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1453,6 +1453,96 @@ def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: drain.assert_not_called() self.assertEqual(result["integrations"], [{"number": 99}]) + def test_reactor_establishes_overlap_integration_before_draining(self) -> None: + config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + "coordinator_actors": ["coordinator"], + }, + "integration": {"mode": "overlap"}, + } + ) + first = entry(1, "shared.py") + second = entry(2, "shared.py") + independent = entry(3, "other.py") + batch = new_batch( + [first, second, independent], frozen_at="2026-06-20T00:00:00Z" + ) + frozen = FreezeResult( + batch, + [first, second, independent], + [], + [], + [{"pull_requests": [1, 2], "source_paths": ["shared.py"]}], + ) + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.pipeline_control.return_value = {"state": "running"} + events: list[str] = [] + client.create_integration_pull_request.side_effect = lambda **_kwargs: ( + events.append("integration") or {"number": 99} + ) + + def drain(*_args, **_kwargs): + events.append("drain") + return {"merged": [{"number": 3, "merge_sha": "a" * 40}]} + + with ( + patch("agent_merge_queue.cli.promote_integrations", return_value=[]), + patch("agent_merge_queue.cli.command_promote", return_value={}), + patch("agent_merge_queue.cli.freeze_queue", return_value=frozen), + patch("agent_merge_queue.cli.command_drain", side_effect=drain), + redirect_stdout(io.StringIO()), + ): + command_react(client, follow=False, timeout_seconds=10) + + self.assertEqual(events, ["integration", "drain"]) + + def test_reactor_does_not_drain_when_overlap_integration_creation_fails( + self, + ) -> None: + config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + "coordinator_actors": ["coordinator"], + }, + "integration": {"mode": "overlap"}, + } + ) + first = entry(1, "shared.py") + second = entry(2, "shared.py") + batch = new_batch([first, second], frozen_at="2026-06-20T00:00:00Z") + frozen = FreezeResult( + batch, + [first, second], + [], + [], + [{"pull_requests": [1, 2], "source_paths": ["shared.py"]}], + ) + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.pipeline_control.return_value = {"state": "running"} + client.create_integration_pull_request.side_effect = QueueError( + "GitHub Actions may not create pull requests" + ) + with ( + patch("agent_merge_queue.cli.promote_integrations", return_value=[]), + patch("agent_merge_queue.cli.command_promote", return_value={}), + patch("agent_merge_queue.cli.freeze_queue", return_value=frozen), + patch("agent_merge_queue.cli.command_drain") as drain, + redirect_stdout(io.StringIO()), + self.assertRaisesRegex(QueueError, "may not create"), + ): + command_react(client, follow=False, timeout_seconds=10) + + drain.assert_not_called() + def test_reactor_holds_overlap_peer_but_drains_independent_work(self) -> None: config = parse_config( { diff --git a/tests/test_doctor.py b/tests/test_doctor.py index d7afacf..055cd6e 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -163,6 +163,63 @@ def fake_json(*arguments: str, cwd: Path): self.assertEqual(registry["status"], "fail") self.assertIn("disabled", registry["detail"]) + def test_overlap_mode_requires_actions_pull_request_creation(self) -> None: + with tempfile.TemporaryDirectory() as temp: + root = Path(temp) + (root / ".mergequeue.toml").write_text( + """ +[queue] +required_checks = ["CI"] +trusted_actors = ["trusted"] +coordinator_actors = ["github-actions[bot]"] + +[integration] +mode = "overlap" +""", + encoding="utf-8", + ) + + def fake_json(*arguments: str, cwd: Path): + joined = " ".join(arguments) + if "repo view" in joined: + return ( + 0, + {"nameWithOwner": "owner/repo", "hasIssuesEnabled": True}, + "", + ) + if "actions/permissions/workflow" in joined: + return 0, {"can_approve_pull_request_reviews": False}, "" + if "users/trusted" in joined: + return 0, {"login": "trusted"}, "" + if "workflow list" in joined or "label list" in joined: + return 0, [], "" + if "pr list" in joined: + return 0, [], "" + if "protection" in joined: + return 1, None, "not available" + return 0, {"login": "owner"}, "" + + with ( + patch( + "agent_merge_queue.doctor.shutil.which", return_value="/usr/bin/gh" + ), + patch( + "agent_merge_queue.doctor._gh", return_value=(0, "authenticated") + ), + patch("agent_merge_queue.doctor._json", side_effect=fake_json), + ): + rows = diagnose( + config_path=None, + repository="owner/repo", + cwd=root, + ) + + permission = next( + value for value in rows if value["check"] == "actions-integration-prs" + ) + self.assertEqual(permission["status"], "fail") + self.assertIn("cannot create", permission["detail"]) + if __name__ == "__main__": unittest.main() From c8a0c2a134b5b2d013a2713e12ddc60a8cd77472 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:01:59 -0700 Subject: [PATCH 07/14] Freeze authoritative overlap paths --- src/agent_merge_queue/cli.py | 79 ++++++++++++++++++++++++++++-------- tests/test_cli.py | 68 +++++++++++++++++++++++++++++++ 2 files changed, 130 insertions(+), 17 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 5b7dd27..466fee2 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -1164,22 +1164,37 @@ def create_integration_pull_request( "If a merge conflict remains, an agent must resolve it without " "dropping either side before marking this PR ready." ) - pull = self._json( - "api", - "--method", - "POST", - f"repos/{self.repository}/pulls", - "-f", - f"title={self.config.integration.title_prefix}: {safe_id}", - "-f", - f"head={branch}", - "-f", - f"base={self.config.base_branch}", - "-f", - f"body={body}", - "-F", - f"draft={'true' if conflict else 'false'}", - ) + try: + pull = self._json( + "api", + "--method", + "POST", + f"repos/{self.repository}/pulls", + "-f", + f"title={self.config.integration.title_prefix}: {safe_id}", + "-f", + f"head={branch}", + "-f", + f"base={self.config.base_branch}", + "-f", + f"body={body}", + "-F", + f"draft={'true' if conflict else 'false'}", + ) + except QueueError: + # No durable integration marker can exist without its PR. The + # batch remains frozen for retry, so remove the private branch + # instead of leaving an orphan that can mask a clean replay. + try: + self._run( + "api", + "--method", + "DELETE", + f"repos/{self.repository}/git/refs/heads/{branch}", + ) + except QueueError: + pass + raise number = int(pull["number"]) marker = { "base_sha": base_sha, @@ -2359,13 +2374,43 @@ def freeze_queue( ] else: queued_numbers = set(client.queued_numbers()) - held_numbers - entries_by_number = { + known_by_number = { entry.number: entry for entry in known_entries if entry.number in queued_numbers } + entries_by_number = dict(known_by_number) for number in sorted(queued_numbers - entries_by_number.keys()): entries_by_number[number] = client.snapshot(number) + + # Deferred path reads are safe while an intent is only waiting, but a + # frozen batch must persist a complete overlap graph. GitHub can report + # transient UNKNOWN mergeability for an already-queued PR; hydrate its + # paths here so a later retry cannot merge overlapping sources as if + # they were independent. + needs_paths = [ + entry + for entry in known_by_number.values() + if not entry.source_paths and not entry.generated_paths + ] + + def load_paths(entry: QueueEntry) -> tuple[int, list[str], list[str]]: + source_paths, generated_paths = client.changed_paths(entry.number) + return entry.number, source_paths, generated_paths + + if needs_paths: + with ThreadPoolExecutor( + max_workers=min( + client.config.pipeline.promotion_workers, len(needs_paths) + ) + ) as executor: + for number, source_paths, generated_paths in executor.map( + load_paths, needs_paths + ): + entries_by_number[number].source_paths = sorted(set(source_paths)) + entries_by_number[number].generated_paths = sorted( + set(generated_paths) + ) all_entries = sorted( entries_by_number.values(), key=lambda entry: ( diff --git a/tests/test_cli.py b/tests/test_cli.py index 405046c..d3d17aa 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -30,6 +30,7 @@ completed_batch_ids, entries_in_batch, effective_queue_marker, + freeze_queue, generated_only_change, latest_batch_marker, latest_marker, @@ -684,6 +685,33 @@ def test_completed_batch_is_not_reused(self) -> None: self.assertEqual(completed, {batch["batch_id"]}) self.assertIsNone(active_batch([first], {1: batch}, completed)) + def test_freeze_hydrates_paths_for_transiently_waiting_queued_entries( + self, + ) -> None: + first = entry(1, state="waiting") + second = entry(2, state="waiting") + client = Mock() + client.config = CONFIG + client.coordinator_logins = {"coordinator"} + client.queued_numbers.return_value = [1, 2] + client.changed_paths.side_effect = lambda _number: (["shared.py"], []) + client.comments.return_value = [] + + with patch( + "agent_merge_queue.cli.utc_now", return_value="2026-06-20T00:01:00Z" + ): + frozen = freeze_queue(client, known_entries=[first, second]) + + self.assertEqual(client.changed_paths.call_count, 2) + self.assertEqual( + frozen.overlap_groups, + [{"pull_requests": [1, 2], "source_paths": ["shared.py"]}], + ) + self.assertEqual( + frozen.batch["source_paths"], + {"1": ["shared.py"], "2": ["shared.py"]}, + ) + def test_queue_timestamp_preserves_refresh_but_resets_reenqueue(self) -> None: previous = {"queued_at": "2026-06-20T00:00:00Z"} self.assertEqual( @@ -1422,6 +1450,46 @@ def test_integration_pr_contains_every_frozen_head(self) -> None: ], ) + def test_failed_integration_pr_creation_removes_its_orphan_branch(self) -> None: + first = entry(1, "shared.py") + second = entry(2, "shared.py") + batch = new_batch([first, second], frozen_at="2026-06-20T00:00:00Z") + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "overlap"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value="b" * 40) + client._json = Mock( + side_effect=[ + {}, + {"sha": "1" * 40}, + {"sha": "2" * 40}, + [], + QueueError("GitHub Actions may not create pull requests"), + ] + ) + client._run = Mock(return_value="") + + with self.assertRaisesRegex(QueueError, "may not create"): + client.create_integration_pull_request(batch=batch, entries=[first, second]) + + branch = f"deploybot/integration/{batch['batch_id']}" + client._run.assert_called_once_with( + "api", + "--method", + "DELETE", + f"repos/example/repo/git/refs/heads/{branch}", + ) + def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: config = parse_config( { From 0c62ff4dc067d78e6bce56c9c825f72dee32466d Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:09:25 -0700 Subject: [PATCH 08/14] Own cumulative integration CI end to end --- src/agent_merge_queue/cli.py | 220 ++++++++++++++++++++++++++++++++++- tests/test_cli.py | 122 +++++++++++++++++-- 2 files changed, 326 insertions(+), 16 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 466fee2..7d05367 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any, Callable, Iterable +from urllib.parse import quote from . import __version__ from .config import ConfigError, QueueConfig, initialize_config, load_config @@ -857,6 +858,28 @@ def workflow_runs(self) -> list[dict[str, Any]]: "workflow_runs", ) + def workflow_runs_for_branch(self, branch: str) -> list[dict[str, Any]]: + return self._paged_object_items( + f"repos/{self.repository}/actions/runs?branch={quote(branch, safe='')}&per_page=100", + "workflow_runs", + ) + + def pull_head(self, number: int) -> dict[str, str]: + value = self._json( + "pr", + "view", + str(number), + "--repo", + self.repository, + "--json", + "headRefName,headRefOid,state", + ) + return { + "branch": str(value.get("headRefName") or ""), + "head_sha": str(value.get("headRefOid") or ""), + "state": str(value.get("state") or ""), + } + def source_deploy_authorized(self, number: int, expected_head: str) -> bool: source = self._json( "pr", @@ -900,8 +923,13 @@ def integration_sources_authorized(self, integration: dict[str, Any]) -> bool: return False return True - def dispatch_ci_workflows(self) -> list[dict[str, Any]]: - configured = self.config.pipeline.ci_workflows + def dispatch_ci_workflows( + self, + *, + ref: str | None = None, + names: Iterable[str] | None = None, + ) -> list[dict[str, Any]]: + configured = tuple(names or self.config.pipeline.ci_workflows) if not configured: raise QueueError("no CI workflow is configured for post-merge dispatch") values = self._json( @@ -936,7 +964,7 @@ def dispatch_ci_workflows(self) -> list[dict[str, Any]]: "--repo", self.repository, "--ref", - self.config.base_branch, + ref or self.config.base_branch, ) dispatched.append({"id": workflow_id, "name": name}) return dispatched @@ -2131,6 +2159,113 @@ def promote_integrations(client: GitHub) -> list[int]: return promoted +def settle_integration_checks( + client: GitHub, + *, + timeout_seconds: int, + poll_seconds: int, + numbers: Iterable[int] | None = None, +) -> list[dict[str, Any]]: + """Own token-created integration CI until each clean PR is merge-ready.""" + if numbers is None: + values = client.integration_pull_request_numbers() + if not isinstance(values, list): + return [] + selected = [int(value) for value in values] + else: + selected = [int(value) for value in numbers] + results: list[dict[str, Any]] = [] + configured = tuple(client.config.pipeline.ci_workflows) + for number in selected: + comments = client.comments(number) + integration = latest_payload( + comments, + INTEGRATION_MARKER, + coordinator_logins(client), + ) + if not integration or integration.get("conflict"): + continue + pull = client.pull_head(number) + branch = str(pull.get("branch") or "") + head_sha = str(pull.get("head_sha") or "") + if pull.get("state") != "OPEN" or not branch or not head_sha: + raise QueueError(f"integration PR #{number} is no longer open") + deadline = time.monotonic() + timeout_seconds + dispatched: list[dict[str, Any]] = [] + while True: + current = client.pull_head(number) + if current.get("state") != "OPEN": + raise QueueError(f"integration PR #{number} is no longer open") + if current.get("head_sha") != head_sha: + raise QueueError( + f"integration PR #{number} changed while CI ownership was active" + ) + runs = client.workflow_runs_for_branch(branch) + latest: dict[str, dict[str, Any]] = {} + for run in runs: + name = str(run.get("name") or "") + if ( + name not in configured + or str(run.get("head_sha") or "") != head_sha + or str(run.get("event") or "") != "workflow_dispatch" + ): + continue + previous = latest.get(name) + key = (str(run.get("created_at") or ""), int(run.get("id") or 0)) + previous_key = ( + str((previous or {}).get("created_at") or ""), + int((previous or {}).get("id") or 0), + ) + if previous is None or key > previous_key: + latest[name] = run + missing = [name for name in configured if name not in latest] + if missing: + dispatched.extend( + client.dispatch_ci_workflows(ref=branch, names=missing) + ) + else: + failed = [ + name + for name, run in latest.items() + if str(run.get("status") or "") == "completed" + and str(run.get("conclusion") or "") != "success" + ] + if failed: + raise QueueError( + f"integration PR #{number} CI failed: " + ", ".join(failed) + ) + if all( + str(run.get("status") or "") == "completed" + and str(run.get("conclusion") or "") == "success" + for run in latest.values() + ): + entry = client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + ) + if entry.state == "ready": + results.append( + { + "branch": branch, + "dispatched_ci": dispatched, + "head_sha": head_sha, + "pull_request": number, + "state": "ready", + } + ) + break + if entry.state == "blocked": + raise QueueError( + f"integration PR #{number} is blocked: " + + "; ".join(entry.reasons or ["unknown gate"]) + ) + if time.monotonic() >= deadline: + raise QueueError(f"integration PR #{number} CI timed out") + time.sleep(poll_seconds) + return results + + def command_resume(client: GitHub, selector: str | None) -> None: number = client.resolve_pr(selector) comments = client.comments(number) @@ -3093,6 +3228,34 @@ def load_paths(entry: QueueEntry) -> tuple[int, list[str], list[str]]: return holds +def combine_drain_results( + first: dict[str, Any], second: dict[str, Any] +) -> dict[str, Any]: + batch_ids: list[str] = [] + for value in (first, second): + candidates = list(value.get("batch_ids") or []) + if not candidates and value.get("batch_id"): + candidates = [str(value["batch_id"])] + batch_ids.extend( + candidate for candidate in candidates if candidate not in batch_ids + ) + waiting = { + int(value["number"]): value + for result in (first, second) + for value in result.get("waiting") or [] + } + return { + "batch_id": batch_ids[-1] if batch_ids else None, + "batch_ids": batch_ids, + "integration_required": list(first.get("integration_required") or []) + + list(second.get("integration_required") or []), + "merged": list(first.get("merged") or []) + + list(second.get("merged") or []), + "next_batch": list(second.get("next_batch") or []), + "waiting": list(waiting.values()), + } + + def command_react( client: GitHub, *, @@ -3105,6 +3268,25 @@ def command_react( result = {"state": "paused", "reason": control.get("reason")} print(json.dumps(result, indent=2, sort_keys=True)) return result + + def own_integration_checks( + numbers: Iterable[int] | None = None, + ) -> list[dict[str, Any]]: + try: + return settle_integration_checks( + client, + timeout_seconds=timeout_seconds, + poll_seconds=10, + numbers=numbers, + ) + except QueueError as error: + if client.config.pipeline.pause_on_failure: + client.set_pipeline_control( + "paused", "integration CI ownership failed: " + str(error) + ) + raise + + integration_checks = own_integration_checks() promoted_integrations = promote_integrations(client) captured_entries: list[QueueEntry] = [] promoted = command_promote( @@ -3157,6 +3339,7 @@ def command_react( ) drained: dict[str, Any] = { "batch_id": frozen.batch["batch_id"], + "batch_ids": [frozen.batch["batch_id"]], "merged": [], "waiting": [], "integration_required": frozen.overlap_groups, @@ -3182,16 +3365,42 @@ def command_react( entries=overlap_entries, ) ) + drained = {} + + created_clean = [ + int(value["number"]) + for value in integrations + if value.get("number") and not value.get("conflict") + ] + newly_promoted_integrations: list[int] = [] + if created_clean: + # GITHUB_TOKEN-created pull requests do not emit a dependable recursive + # pull_request/workflow_run chain. Dispatch and own their exact-head CI + # under this coordinator before any source batch is allowed to drain. + integration_checks.extend(own_integration_checks(created_clean)) + newly_promoted_integrations = promote_integrations(client) + promoted_integrations = list( + dict.fromkeys(promoted_integrations + newly_promoted_integrations) + ) + + if not drained: # Establish durable ownership for every overlapping source before any # independent member of the frozen batch is allowed to merge. A setup - # failure while creating the integration PR must leave the batch intact - # and must never strand an already-merged partial batch before CI. + # or integration-CI failure leaves the batch intact and cannot strand + # an already-merged partial batch before exact-main CI. drained = command_drain( client, json_output=False, emit=False, initial_frozen=frozen, ) + if newly_promoted_integrations: + integration_drain = command_drain( + client, + json_output=False, + emit=False, + ) + drained = combine_drain_results(drained, integration_drain) dispatched_ci: list[dict[str, Any]] = [] if dispatch_ci and drained.get("merged"): try: @@ -3225,6 +3434,7 @@ def command_react( "drain": drained, "dispatched_ci": dispatched_ci, "integrations": integrations, + "integration_checks": integration_checks, "release": release, } print(json.dumps(result, indent=2, sort_keys=True)) diff --git a/tests/test_cli.py b/tests/test_cli.py index d3d17aa..f6d8499 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -44,6 +44,7 @@ queue_from_intent, queue_timestamp, reusable_batch, + settle_integration_checks, should_settle_batch, structured_dependencies, ) @@ -1509,17 +1510,38 @@ def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: client.config = config client.coordinator_logins = {"coordinator"} client.pipeline_control.return_value = {"state": "running"} - client.create_integration_pull_request.return_value = {"number": 99} + client.create_integration_pull_request.return_value = { + "number": 99, + "conflict": None, + } with ( - patch("agent_merge_queue.cli.promote_integrations", return_value=[]), + patch( + "agent_merge_queue.cli.settle_integration_checks", + side_effect=[[], [{"pull_request": 99, "state": "ready"}]], + ), + patch( + "agent_merge_queue.cli.promote_integrations", + side_effect=[[], [99]], + ), patch("agent_merge_queue.cli.command_promote", return_value={}), patch("agent_merge_queue.cli.freeze_queue", return_value=frozen), - patch("agent_merge_queue.cli.command_drain") as drain, + patch( + "agent_merge_queue.cli.command_drain", + return_value={ + "batch_id": "integration", + "batch_ids": ["integration"], + "integration_required": [], + "merged": [{"number": 99, "merge_sha": "a" * 40}], + "next_batch": [], + "waiting": [], + }, + ) as drain, redirect_stdout(io.StringIO()), ): result = command_react(client, follow=False, timeout_seconds=10) - drain.assert_not_called() - self.assertEqual(result["integrations"], [{"number": 99}]) + drain.assert_called_once_with(client, json_output=False, emit=False) + self.assertEqual(result["integrations"][0]["number"], 99) + self.assertEqual(result["drain"]["merged"][0]["number"], 99) def test_reactor_establishes_overlap_integration_before_draining(self) -> None: config = parse_config( @@ -1551,7 +1573,7 @@ def test_reactor_establishes_overlap_integration_before_draining(self) -> None: client.pipeline_control.return_value = {"state": "running"} events: list[str] = [] client.create_integration_pull_request.side_effect = lambda **_kwargs: ( - events.append("integration") or {"number": 99} + events.append("integration") or {"number": 99, "conflict": None} ) def drain(*_args, **_kwargs): @@ -1559,7 +1581,19 @@ def drain(*_args, **_kwargs): return {"merged": [{"number": 3, "merge_sha": "a" * 40}]} with ( - patch("agent_merge_queue.cli.promote_integrations", return_value=[]), + patch( + "agent_merge_queue.cli.settle_integration_checks", + side_effect=lambda *_args, **kwargs: ( + [] + if kwargs.get("numbers") is None + else events.append("settle") + or [{"pull_request": 99, "state": "ready"}] + ), + ), + patch( + "agent_merge_queue.cli.promote_integrations", + side_effect=[[], [99]], + ), patch("agent_merge_queue.cli.command_promote", return_value={}), patch("agent_merge_queue.cli.freeze_queue", return_value=frozen), patch("agent_merge_queue.cli.command_drain", side_effect=drain), @@ -1567,7 +1601,7 @@ def drain(*_args, **_kwargs): ): command_react(client, follow=False, timeout_seconds=10) - self.assertEqual(events, ["integration", "drain"]) + self.assertEqual(events, ["integration", "settle", "drain", "drain"]) def test_reactor_does_not_drain_when_overlap_integration_creation_fails( self, @@ -1736,7 +1770,7 @@ def test_reactor_follows_release_without_a_new_merge(self) -> None: ) self.assertEqual(result["release"], release) - def test_reactor_does_not_follow_idle_all_mode_integration_batch(self) -> None: + def test_reactor_does_not_follow_conflicted_all_mode_integration_batch(self) -> None: config = parse_config( { "queue": { @@ -1756,7 +1790,10 @@ def test_reactor_does_not_follow_idle_all_mode_integration_batch(self) -> None: client.config = config client.coordinator_logins = {"coordinator"} client.pipeline_control.return_value = {"state": "running"} - client.create_integration_pull_request.return_value = {"number": 99} + client.create_integration_pull_request.return_value = { + "number": 99, + "conflict": {"number": 2, "reason": "merge conflict"}, + } client.base_sha.return_value = sha client.workflow_runs.return_value = [ { @@ -1794,7 +1831,70 @@ def test_reactor_does_not_follow_idle_all_mode_integration_batch(self) -> None: drain.assert_not_called() follow_release.assert_not_called() self.assertIsNone(result["release"]) - self.assertEqual(result["integrations"], [{"number": 99}]) + self.assertEqual(result["integrations"][0]["number"], 99) + + def test_integration_ci_dispatch_is_owned_until_the_pr_is_ready(self) -> None: + number = 38 + head_sha = "a" * 40 + branch = "deploybot/integration/batch" + marker = { + "batch_id": "batch", + "conflict": None, + "heads": {"1": "1" * 40, "2": "2" * 40}, + "pull_requests": [1, 2], + } + client = Mock() + client.config = CONFIG + client.coordinator_logins = {"coordinator"} + client.comments.return_value = [ + { + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body(marker), + } + ] + client.pull_head.return_value = { + "branch": branch, + "head_sha": head_sha, + "state": "OPEN", + } + active = { + "id": 7, + "name": "CI", + "head_sha": head_sha, + "event": "workflow_dispatch", + "status": "in_progress", + "conclusion": None, + "created_at": "2026-06-20T00:01:00Z", + } + successful = { + **active, + "status": "completed", + "conclusion": "success", + } + client.workflow_runs_for_branch.side_effect = [[], [active], [successful]] + client.dispatch_ci_workflows.return_value = [{"id": 7, "name": "CI"}] + ready = entry(8, "combined.py") + ready.number = number + ready.head_sha = head_sha + client.snapshot.return_value = ready + + with ( + patch("agent_merge_queue.cli.time.monotonic", return_value=0), + patch("agent_merge_queue.cli.time.sleep"), + ): + result = settle_integration_checks( + client, + timeout_seconds=10, + poll_seconds=0, + numbers=[number], + ) + + client.dispatch_ci_workflows.assert_called_once_with( + ref=branch, + names=["CI"], + ) + self.assertEqual(result[0]["state"], "ready") def test_reactor_pauses_when_post_merge_ci_dispatch_fails(self) -> None: client = Mock() From 641813592c8f3e99979cee2a4dbf7ee7b467c735 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:13:38 -0700 Subject: [PATCH 09/14] Use exact commit checks for integration gates --- src/agent_merge_queue/cli.py | 13 +++++++++++++ tests/test_cli.py | 29 +++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 7d05367..602c060 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -139,8 +139,11 @@ def check_states(checks: Iterable[dict[str, Any]]) -> dict[str, str]: continue timestamp = str( check.get("startedAt") + or check.get("started_at") or check.get("createdAt") + or check.get("created_at") or check.get("completedAt") + or check.get("completed_at") or "" ) if timestamp.startswith("0001-01-01"): @@ -864,6 +867,12 @@ def workflow_runs_for_branch(self, branch: str) -> list[dict[str, Any]]: "workflow_runs", ) + def commit_check_runs(self, head_sha: str) -> list[dict[str, Any]]: + return self._paged_object_items( + f"repos/{self.repository}/commits/{head_sha}/check-runs?per_page=100", + "check_runs", + ) + def pull_head(self, number: int) -> dict[str, str]: value = self._json( "pr", @@ -1347,6 +1356,7 @@ def snapshot( require_marker: bool = True, allow_blocked_label: bool = False, known_comments: list[dict[str, Any]] | None = None, + known_checks: dict[str, str] | None = None, known_source_paths: list[str] | None = None, known_generated_paths: list[str] | None = None, defer_paths_until_ready: bool = False, @@ -1389,6 +1399,7 @@ def snapshot( marker = queue_marker_for_client(self, comments) head_sha = str(pull["headRefOid"]) checks = check_states(pull.get("statusCheckRollup") or []) + checks.update(known_checks or {}) source_paths = list(known_source_paths or []) generated_paths = list(known_generated_paths or []) paths_are_known = ( @@ -2239,10 +2250,12 @@ def settle_integration_checks( and str(run.get("conclusion") or "") == "success" for run in latest.values() ): + exact_checks = check_states(client.commit_check_runs(head_sha)) entry = client.snapshot( number, require_marker=False, allow_blocked_label=True, + known_checks=exact_checks, ) if entry.state == "ready": results.append( diff --git a/tests/test_cli.py b/tests/test_cli.py index f6d8499..9396097 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -406,6 +406,22 @@ def test_latest_check_rerun_wins(self) -> None: ) self.assertEqual(states["CI"], "failed") + states = check_states( + [ + { + "name": "CI", + "conclusion": "ACTION_REQUIRED", + "started_at": "2026-06-20T00:00:00Z", + }, + { + "name": "CI", + "conclusion": "SUCCESS", + "started_at": "2026-06-20T00:01:00Z", + }, + ] + ) + self.assertEqual(states["CI"], "passed") + def test_undated_queued_rerun_hides_older_success(self) -> None: states = check_states( [ @@ -1874,6 +1890,13 @@ def test_integration_ci_dispatch_is_owned_until_the_pr_is_ready(self) -> None: } client.workflow_runs_for_branch.side_effect = [[], [active], [successful]] client.dispatch_ci_workflows.return_value = [{"id": 7, "name": "CI"}] + client.commit_check_runs.return_value = [ + { + "name": "CI", + "conclusion": "success", + "started_at": "2026-06-20T00:01:00Z", + } + ] ready = entry(8, "combined.py") ready.number = number ready.head_sha = head_sha @@ -1894,6 +1917,12 @@ def test_integration_ci_dispatch_is_owned_until_the_pr_is_ready(self) -> None: ref=branch, names=["CI"], ) + client.snapshot.assert_called_once_with( + number, + require_marker=False, + allow_blocked_label=True, + known_checks={"CI": "passed"}, + ) self.assertEqual(result[0]["state"], "ready") def test_reactor_pauses_when_post_merge_ci_dispatch_fails(self) -> None: From 1383a8a0abad159a5fdf1e4d1003c4a02a9f3755 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:16:31 -0700 Subject: [PATCH 10/14] Preserve integration CI evidence through merge --- src/agent_merge_queue/cli.py | 44 ++++++++++++++-- tests/test_cli.py | 99 ++++++++++++++++++++++++++++++++++++ 2 files changed, 138 insertions(+), 5 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 602c060..7d1a77d 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -1398,8 +1398,19 @@ def snapshot( threads = self.review_threads(number) if needs_threads else [] marker = queue_marker_for_client(self, comments) head_sha = str(pull["headRefOid"]) - checks = check_states(pull.get("statusCheckRollup") or []) + check_rollup = list(pull.get("statusCheckRollup") or []) + checks = check_states(check_rollup) checks.update(known_checks or {}) + integration = latest_payload( + comments, + INTEGRATION_MARKER, + coordinator_logins(self), + ) + if integration and any( + checks.get(name) != "passed" for name in self.config.required_checks + ): + checks = check_states(check_rollup + self.commit_check_runs(head_sha)) + checks.update(known_checks or {}) source_paths = list(known_source_paths or []) generated_paths = list(known_generated_paths or []) paths_are_known = ( @@ -2124,7 +2135,11 @@ def evaluate( return result -def promote_integrations(client: GitHub) -> list[int]: +def promote_integrations( + client: GitHub, + *, + known_checks_by_number: dict[int, dict[str, str]] | None = None, +) -> list[int]: promoted: list[int] = [] for number in client.integration_pull_request_numbers(): comments = client.comments(number) @@ -2138,7 +2153,12 @@ def promote_integrations(client: GitHub) -> list[int]: labels = client.labels(number) if client.config.queue_label in labels: continue - entry = client.snapshot(number, require_marker=False, allow_blocked_label=True) + entry = client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + known_checks=(known_checks_by_number or {}).get(number), + ) if entry.state != "ready": continue batch_id = str(integration.get("batch_id") or "") @@ -2262,6 +2282,7 @@ def settle_integration_checks( { "branch": branch, "dispatched_ci": dispatched, + "checks": exact_checks, "head_sha": head_sha, "pull_request": number, "state": "ready", @@ -3300,7 +3321,13 @@ def own_integration_checks( raise integration_checks = own_integration_checks() - promoted_integrations = promote_integrations(client) + promoted_integrations = promote_integrations( + client, + known_checks_by_number={ + int(value["pull_request"]): dict(value.get("checks") or {}) + for value in integration_checks + }, + ) captured_entries: list[QueueEntry] = [] promoted = command_promote( client, @@ -3391,7 +3418,14 @@ def own_integration_checks( # pull_request/workflow_run chain. Dispatch and own their exact-head CI # under this coordinator before any source batch is allowed to drain. integration_checks.extend(own_integration_checks(created_clean)) - newly_promoted_integrations = promote_integrations(client) + newly_promoted_integrations = promote_integrations( + client, + known_checks_by_number={ + int(value["pull_request"]): dict(value.get("checks") or {}) + for value in integration_checks + if int(value["pull_request"]) in created_clean + }, + ) promoted_integrations = list( dict.fromkeys(promoted_integrations + newly_promoted_integrations) ) diff --git a/tests/test_cli.py b/tests/test_cli.py index 9396097..127d6f4 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -40,6 +40,7 @@ near_ready_overlap_holds, new_batch, overlap_groups, + promote_integrations, queue_state_body, queue_from_intent, queue_timestamp, @@ -364,6 +365,65 @@ def test_waiting_promotion_defers_changed_file_fetch(self) -> None: self.assertEqual(value.state, "waiting") client.changed_paths.assert_not_called() + def test_integration_snapshot_uses_exact_commit_check_fallback(self) -> None: + head_sha = "a" * 40 + marker = { + "batch_id": "batch", + "conflict": None, + "heads": {"1": "1" * 40, "2": "2" * 40}, + "pull_requests": [1, 2], + } + client = object.__new__(GitHub) + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"coordinator"} + client.comments = Mock( + return_value=[ + { + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body(marker), + } + ] + ) + client.changed_paths = Mock(return_value=(["combined.py"], [])) + client.commit_check_runs = Mock( + return_value=[ + { + "name": "CI", + "conclusion": "success", + "started_at": "2026-06-20T00:01:00Z", + } + ] + ) + client._json = Mock( + return_value={ + "baseRefName": "main", + "body": "", + "headRefOid": head_sha, + "isDraft": False, + "labels": [], + "mergeStateStatus": "UNSTABLE", + "mergeable": "MERGEABLE", + "number": 38, + "state": "OPEN", + "statusCheckRollup": [], + "title": "Integration", + "url": "https://example.test/38", + } + ) + + value = client.snapshot( + 38, + require_marker=False, + allow_blocked_label=True, + ) + + self.assertEqual(value.state, "ready") + self.assertEqual(value.checks["CI"], "passed") + client.commit_check_runs.assert_called_once_with(head_sha) + def test_required_checks_do_not_accept_skipped(self) -> None: states = check_states( [ @@ -1925,6 +1985,45 @@ def test_integration_ci_dispatch_is_owned_until_the_pr_is_ready(self) -> None: ) self.assertEqual(result[0]["state"], "ready") + def test_integration_promotion_reuses_owned_exact_checks(self) -> None: + number = 38 + head_sha = "a" * 40 + marker = { + "batch_id": "batch", + "conflict": None, + "heads": {"1": "1" * 40, "2": "2" * 40}, + "pull_requests": [1, 2], + } + client = Mock() + client.config = CONFIG + client.coordinator_logins = {"coordinator"} + client.integration_pull_request_numbers.return_value = [number] + client.comments.return_value = [ + { + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body(marker), + } + ] + client.labels.return_value = set() + ready = entry(8, "combined.py") + ready.number = number + ready.head_sha = head_sha + client.snapshot.return_value = ready + + promoted = promote_integrations( + client, + known_checks_by_number={number: {"CI": "passed"}}, + ) + + self.assertEqual(promoted, [number]) + client.snapshot.assert_called_once_with( + number, + require_marker=False, + allow_blocked_label=True, + known_checks={"CI": "passed"}, + ) + def test_reactor_pauses_when_post_merge_ci_dispatch_fails(self) -> None: client = Mock() client.config = CONFIG From 7f6f6e82aa032cdd59238ad0128e957a8e71c788 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:30:22 -0700 Subject: [PATCH 11/14] Reconcile exact external batch merges --- src/agent_merge_queue/cli.py | 134 +++++++++++++++++++++++++++++------ tests/test_cli.py | 93 ++++++++++++++++++++++++ 2 files changed, 207 insertions(+), 20 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 7d1a77d..f924c11 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -1554,6 +1554,32 @@ def dependency_is_merged(self, number: int) -> bool: ) return comparison.get("status") in {"ahead", "identical"} + def externally_integrated_merge( + self, number: int, expected_head: str + ) -> str | None: + value = self._json( + "pr", + "view", + str(number), + "--repo", + self.repository, + "--json", + "headRefOid,mergeCommit,state", + ) + merge_sha = str((value.get("mergeCommit") or {}).get("oid") or "") + if ( + value.get("state") != "MERGED" + or value.get("headRefOid") != expected_head + or not merge_sha + ): + return None + base_sha = self.base_sha() + if not self.is_ancestor(expected_head, base_sha): + return None + if not self.is_ancestor(merge_sha, base_sha): + return None + return merge_sha + def merge( self, number: int, @@ -2035,6 +2061,55 @@ def record_repair( return payload +def reconcile_externally_merged_threads(client: GitHub) -> list[dict[str, Any]]: + reconciled: list[dict[str, Any]] = [] + records = client.thread_records() + if not isinstance(records, list): + return reconciled + for record in records: + if record.get("phase") != "deploy-requested" or not record.get( + "pull_request" + ): + continue + number = int(record["pull_request"]) + comments = client.comments(number) + intent = latest_intent(comments, client.trusted_logins) + expected_head = str((intent or {}).get("requested_head") or "") + if not expected_head: + continue + try: + merge_sha = client.externally_integrated_merge(number, expected_head) + except QueueError: + continue + if not merge_sha: + continue + updated_at = utc_now() + client.record_thread( + ThreadRecord( + provider=str(record["provider"]), + thread_id=str(record["thread_id"]), + phase="merged", + updated_at=updated_at, + title=str(record.get("title") or "") or None, + branch=str(record.get("branch") or "") or None, + pull_request=number, + url=str(record.get("url") or "") or None, + ) + ) + value = { + "head_sha": expected_head, + "merge_sha": merge_sha, + "pull_request": number, + } + reconciled.append(value) + notify( + client.config.pipeline, + "externally-merged", + {"repository": client.repository, **value}, + ) + return reconciled + + def command_promote( client: GitHub, *, @@ -2852,28 +2927,43 @@ def command_merge( + ", ".join(f"#{value}" for value in missing_dependencies) ) - fresh = client.snapshot( - number, - known_source_paths=entry.source_paths, - known_generated_paths=entry.generated_paths, - ) - if fresh.state != "ready": - raise QueueError( - f"PR #{number} changed before merge: " - + "; ".join(fresh.reasons or ["unknown gate"]) + externally_integrated = False + try: + fresh = client.snapshot( + number, + known_source_paths=entry.source_paths, + known_generated_paths=entry.generated_paths, ) - if str((batch.get("heads") or {}).get(str(number)) or "") != fresh.head_sha: - raise QueueError(f"PR #{number} changed after batch {batch_id} was frozen") - if fresh.dependencies != expected_dependencies: - raise QueueError(f"PR #{number} dependencies changed before merge") - if fresh.queued_at and str(batch.get("frozen_at") or "") < fresh.queued_at: - raise QueueError(f"batch {batch_id} predates the current queue authorization") + except QueueError as error: + merge_sha = ( + client.externally_integrated_merge(number, expected_head) + if "is not open" in str(error) + else None + ) + if not merge_sha: + raise + fresh = entry + externally_integrated = True + if not externally_integrated: + if fresh.state != "ready": + raise QueueError( + f"PR #{number} changed before merge: " + + "; ".join(fresh.reasons or ["unknown gate"]) + ) + if str((batch.get("heads") or {}).get(str(number)) or "") != fresh.head_sha: + raise QueueError(f"PR #{number} changed after batch {batch_id} was frozen") + if fresh.dependencies != expected_dependencies: + raise QueueError(f"PR #{number} dependencies changed before merge") + if fresh.queued_at and str(batch.get("frozen_at") or "") < fresh.queued_at: + raise QueueError( + f"batch {batch_id} predates the current queue authorization" + ) - merge_sha = client.merge( - number, - fresh.head_sha, - authorization_entry=fresh, - ) + merge_sha = client.merge( + number, + fresh.head_sha, + authorization_entry=fresh, + ) merged_comments = client.comments(number) intent = latest_intent(merged_comments, client.trusted_logins) integration = latest_payload( @@ -2901,6 +2991,7 @@ def command_merge( "head_sha": fresh.head_sha, "merge_sha": merge_sha, "intent_id": (intent or {}).get("intent_id"), + "ownership": "external" if externally_integrated else "deploybot", }, ) if integration: @@ -3303,6 +3394,8 @@ def command_react( print(json.dumps(result, indent=2, sort_keys=True)) return result + reconciled_merges = reconcile_externally_merged_threads(client) + def own_integration_checks( numbers: Iterable[int] | None = None, ) -> list[dict[str, Any]]: @@ -3482,6 +3575,7 @@ def own_integration_checks( "dispatched_ci": dispatched_ci, "integrations": integrations, "integration_checks": integration_checks, + "reconciled_merges": reconciled_merges, "release": release, } print(json.dumps(result, indent=2, sort_keys=True)) diff --git a/tests/test_cli.py b/tests/test_cli.py index 127d6f4..63da6a8 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -44,6 +44,7 @@ queue_state_body, queue_from_intent, queue_timestamp, + reconcile_externally_merged_threads, reusable_batch, settle_integration_checks, should_settle_batch, @@ -187,6 +188,47 @@ def test_overlap_mode_holds_only_ready_members_of_near_ready_components( self.assertEqual(holds, {1: [3]}) client.changed_paths.assert_called_once_with(3) + def test_reconciles_an_externally_merged_requested_thread(self) -> None: + head_sha = "a" * 40 + merge_sha = "m" * 40 + client = Mock() + client.config = CONFIG + client.repository = "example/repo" + client.trusted_logins = {"trusted"} + client.thread_records.return_value = [ + { + "phase": "deploy-requested", + "provider": "codex", + "pull_request": 42, + "thread_id": "thread-42", + } + ] + client.comments.return_value = [ + { + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "trusted"}, + "body": intent_body( + intent_id="intent-42", + state="requested", + requested_at="2026-06-20T00:00:00Z", + requested_head=head_sha, + provider="codex", + thread_id="thread-42", + ), + } + ] + client.externally_integrated_merge.return_value = merge_sha + + result = reconcile_externally_merged_threads(client) + + self.assertEqual( + result, + [{"head_sha": head_sha, "merge_sha": merge_sha, "pull_request": 42}], + ) + record = client.record_thread.call_args.args[0] + self.assertEqual(record.phase, "merged") + self.assertEqual(record.pull_request, 42) + def test_overlap_hold_includes_transitive_ready_members(self) -> None: config = parse_config( { @@ -297,6 +339,57 @@ def test_frozen_merge_fast_path_never_rescans_the_whole_queue(self) -> None: authorization_entry=value, ) + def test_frozen_merge_accepts_the_same_head_already_integrated_externally( + self, + ) -> None: + value = entry(1, "a.py") + batch = new_batch([value], frozen_at="2026-06-20T00:01:00Z") + client = Mock() + client.config = CONFIG + client.trusted_logins = {"trusted"} + client.coordinator_logins = {"coordinator"} + client.snapshot.side_effect = QueueError("PR #1 is not open") + client.externally_integrated_merge.return_value = "e" * 40 + client.comments.return_value = [] + + merged = command_merge( + client, + "1", + str(batch["batch_id"]), + frozen_entry=value, + frozen_batch=batch, + active_numbers={1}, + ) + + self.assertEqual(merged, "e" * 40) + client.externally_integrated_merge.assert_called_once_with(1, value.head_sha) + client.merge.assert_not_called() + + def test_external_merge_must_be_exact_and_ancestral_to_main(self) -> None: + head_sha = "a" * 40 + merge_sha = "m" * 40 + base_sha = "b" * 40 + client = object.__new__(GitHub) + client.repository = "example/repo" + client._json = Mock( + return_value={ + "headRefOid": head_sha, + "mergeCommit": {"oid": merge_sha}, + "state": "MERGED", + } + ) + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + + self.assertEqual( + client.externally_integrated_merge(1, head_sha), + merge_sha, + ) + self.assertEqual( + [call.args for call in client.is_ancestor.call_args_list], + [(head_sha, base_sha), (merge_sha, base_sha)], + ) + def test_generated_paths_are_configurable(self) -> None: self.assertTrue( generated_only_change( From 5d1cc2c9b431de0968ef0f54aa2313c3f91767f7 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:51:05 -0700 Subject: [PATCH 12/14] Integrate policy reference and formatting --- docs/reference.md | 3 +++ src/agent_merge_queue/cli.py | 32 +++++++++++++++++++------------- tests/test_cli.py | 20 +++++++++----------- tests/test_docs.py | 3 +++ tests/test_skill.py | 10 +++------- 5 files changed, 37 insertions(+), 31 deletions(-) diff --git a/docs/reference.md b/docs/reference.md index b5872f0..e88a8d9 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -159,6 +159,9 @@ Provider fields are: | `thread_active_hours` | Positive integer; default 72. | | `ci_workflows` | Workflow names followed as exact-main CI. Default: `["CI"]`. | | `deploy_workflows` | Deployment workflow names. Default: `["Deploy"]`. | +| `batch_settle_seconds` | Non-negative window for coalescing near-ready deploy requests before freezing a batch. Default: 15. | +| `ci_failure_grace_seconds` | Non-negative window for an exact-main CI retry to replace a failed attempt before the release fails. Default: 90. | +| `promotion_workers` | Positive maximum number of deploy requests promoted concurrently. Default: 4. | | `ready_to_merge_target_minutes` | Positive timing target; default 15. | | `merge_to_live_target_minutes` | Positive timing target; default 10. | | `auto_promote` | Default `true`. | diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index f924c11..5cd96e7 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -1389,7 +1389,9 @@ def snapshot( if pull.get("state") != "OPEN": raise QueueError(f"PR #{number} is not open") - comments = known_comments if known_comments is not None else self.comments(number) + comments = ( + known_comments if known_comments is not None else self.comments(number) + ) reviews = self.reviews(number) if self.config.review_providers else [] needs_threads = any( provider.kind == "bot" and provider.require_resolved_threads @@ -1608,7 +1610,9 @@ def merge( if authorization.get("headRefOid") != head_sha: raise QueueError(f"PR #{number} changed immediately before merge") marker = queue_marker_for_client(self, self.comments(number)) - integration_batch_id = str(marker.get("integration_batch_id") or "") if marker else "" + integration_batch_id = ( + str(marker.get("integration_batch_id") or "") if marker else "" + ) else: labels = set(authorization_entry.labels) if authorization_entry.is_draft: @@ -2067,9 +2071,7 @@ def reconcile_externally_merged_threads(client: GitHub) -> list[dict[str, Any]]: if not isinstance(records, list): return reconciled for record in records: - if record.get("phase") != "deploy-requested" or not record.get( - "pull_request" - ): + if record.get("phase") != "deploy-requested" or not record.get("pull_request"): continue number = int(record["pull_request"]) comments = client.comments(number) @@ -2527,7 +2529,9 @@ def queue_from_intent( f"PR #{entry.number} deploy intent is bound to an older head; " "run deploybot refresh-request after fresh review" ) - current_comments = comments if comments is not None else client.comments(entry.number) + current_comments = ( + comments if comments is not None else client.comments(entry.number) + ) previous = queue_marker_for_client(client, current_comments) current_labels = labels if labels is not None else client.labels(entry.number) if ( @@ -2867,7 +2871,9 @@ def command_merge( ) -> str: if frozen_entry is None or frozen_batch is None: require_running_pipeline(client) - number = frozen_entry.number if frozen_entry is not None else client.resolve_pr(selector) + number = ( + frozen_entry.number if frozen_entry is not None else client.resolve_pr(selector) + ) batch = frozen_batch if batch is not None and str(batch.get("batch_id") or "") != batch_id: raise QueueError(f"frozen context does not match batch {batch_id}") @@ -2885,7 +2891,9 @@ def command_merge( if frozen_entry is None: entries = client.queue() frozen_entries = [value for value in entries if value.number in frozen] - entry = next((value for value in frozen_entries if value.number == number), None) + entry = next( + (value for value in frozen_entries if value.number == number), None + ) current_numbers = {value.number for value in frozen_entries} else: entry = frozen_entry @@ -3374,8 +3382,7 @@ def combine_drain_results( "batch_ids": batch_ids, "integration_required": list(first.get("integration_required") or []) + list(second.get("integration_required") or []), - "merged": list(first.get("merged") or []) - + list(second.get("merged") or []), + "merged": list(first.get("merged") or []) + list(second.get("merged") or []), "next_batch": list(second.get("next_batch") or []), "waiting": list(waiting.values()), } @@ -3427,9 +3434,8 @@ def own_integration_checks( emit=False, captured_entries=captured_entries, ) - if ( - client.config.pipeline.batch_settle_seconds - and should_settle_batch(client, captured_entries) + if client.config.pipeline.batch_settle_seconds and should_settle_batch( + client, captured_entries ): time.sleep(client.config.pipeline.batch_settle_seconds) settled_entries: list[QueueEntry] = [] diff --git a/tests/test_cli.py b/tests/test_cli.py index 63da6a8..6d8d653 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -181,9 +181,7 @@ def test_overlap_mode_holds_only_ready_members_of_near_ready_components( client.changed_paths.return_value = (["shared.py"], []) client.comments.return_value = [] - holds = near_ready_overlap_holds( - client, [overlapping, independent, waiting] - ) + holds = near_ready_overlap_holds(client, [overlapping, independent, waiting]) self.assertEqual(holds, {1: [3]}) client.changed_paths.assert_called_once_with(3) @@ -267,15 +265,15 @@ def test_overlap_hold_does_not_rewrite_an_existing_frozen_batch(self) -> None: client.coordinator_logins = {"coordinator"} client.changed_paths.return_value = (["shared.py"], []) client.comments.return_value = [] - with patch( - "agent_merge_queue.cli.latest_batch_marker", return_value=batch - ): + with patch("agent_merge_queue.cli.latest_batch_marker", return_value=batch): self.assertEqual( near_ready_overlap_holds(client, [ready, waiting]), {}, ) - def test_simultaneous_intents_use_pull_request_number_as_fifo_tiebreaker(self) -> None: + def test_simultaneous_intents_use_pull_request_number_as_fifo_tiebreaker( + self, + ) -> None: first = entry(1) second = entry(2) first.priority_at = second.priority_at = "2026-06-20T00:00:00Z" @@ -1850,9 +1848,7 @@ def promote(*_args, **kwargs): with ( patch("agent_merge_queue.cli.promote_integrations", return_value=[]), patch("agent_merge_queue.cli.command_promote", side_effect=promote), - patch( - "agent_merge_queue.cli.freeze_queue", return_value=frozen - ) as freeze, + patch("agent_merge_queue.cli.freeze_queue", return_value=frozen) as freeze, patch( "agent_merge_queue.cli.command_drain", return_value={"merged": [{"number": 3, "merge_sha": "a" * 40}]}, @@ -1939,7 +1935,9 @@ def test_reactor_follows_release_without_a_new_merge(self) -> None: ) self.assertEqual(result["release"], release) - def test_reactor_does_not_follow_conflicted_all_mode_integration_batch(self) -> None: + def test_reactor_does_not_follow_conflicted_all_mode_integration_batch( + self, + ) -> None: config = parse_config( { "queue": { diff --git a/tests/test_docs.py b/tests/test_docs.py index ba8b105..a9df7a0 100644 --- a/tests/test_docs.py +++ b/tests/test_docs.py @@ -127,6 +127,9 @@ def test_reference_names_every_policy_field(self) -> None: "thread_active_hours", "ci_workflows", "deploy_workflows", + "batch_settle_seconds", + "ci_failure_grace_seconds", + "promotion_workers", "ready_to_merge_target_minutes", "merge_to_live_target_minutes", "auto_promote", diff --git a/tests/test_skill.py b/tests/test_skill.py index 01e83a6..9651ba9 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -59,9 +59,7 @@ def test_github_workflow_wakes_after_named_ci_finishes(self) -> None: "github.event.pull_request.head.repo.full_name == github.repository", workflow, ) - self.assertIn( - "github.event.check_suite.app.slug != 'github-actions'", workflow - ) + self.assertIn("github.event.check_suite.app.slug != 'github-actions'", workflow) self.assertIn("github.event.check_suite.pull_requests[0].base.ref", workflow) self.assertIn("persist-credentials: false", workflow) @@ -89,11 +87,9 @@ def test_action_dispatches_ci_after_builtin_token_merge(self) -> None: def test_action_follows_release_when_workflow_run_is_suppressed(self) -> None: action = (ROOT / "action.yml").read_text(encoding="utf-8") - follow_input = action.split(" follow:\n", 1)[1].split( - " dispatch_ci:\n", 1 - )[0] + follow_input = action.split(" follow:\n", 1)[1].split(" dispatch_ci:\n", 1)[0] self.assertIn('default: "true"', follow_input) - self.assertIn('args+=(--follow)', action) + self.assertIn("args+=(--follow)", action) def test_clients_pin_the_immutable_status_release(self) -> None: paths = [ From 185d1678d4676e86d8ca79a4fc8d56ccb67c8db6 Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:54:28 -0700 Subject: [PATCH 13/14] Serialize mixed all-mode integration batches --- src/agent_merge_queue/cli.py | 55 ++++++++++++++----- tests/test_cli.py | 102 +++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 13 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 5cd96e7..8d25fca 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -3450,6 +3450,30 @@ def own_integration_checks( promoted["blocked"] = promoted["blocked"] + settled["blocked"] promoted["waiting"] = settled["waiting"] captured_entries = settled_entries + + drained: dict[str, Any] = {} + queued_integrations: set[int] = set() + if client.config.integration.mode == "all": + queued_numbers = set(client.queued_numbers()) + queued_integrations = queued_numbers.intersection( + client.integration_pull_request_numbers() + ) + if queued_integrations: + # A ready integration PR owns an earlier frozen source batch. Drain + # those integration PRs first, while holding newly queued sources, + # so the older integration cannot suppress all-mode integration of + # the new work or let that work merge directly. + integration_frozen = freeze_queue( + client, + held_numbers=queued_numbers - queued_integrations, + ) + drained = command_drain( + client, + json_output=False, + emit=False, + initial_frozen=integration_frozen, + ) + overlap_holds = near_ready_overlap_holds(client, captured_entries) promoted["held"] = [ { @@ -3459,24 +3483,26 @@ def own_integration_checks( } for number, waiting in overlap_holds.items() ] - frozen = freeze_queue( - client, - known_entries=captured_entries, - held_numbers=set(overlap_holds), - ) + merged_integrations = { + int(value["number"]) for value in drained.get("merged") or [] + } + if queued_integrations - merged_integrations: + frozen = FreezeResult(None, [], [], [], []) + else: + frozen = freeze_queue( + client, + known_entries=captured_entries, + held_numbers=set(overlap_holds), + ) integrations: list[dict[str, Any]] = [] - if ( - frozen.batch is not None - and client.config.integration.mode == "all" - and not promoted_integrations - ): + if frozen.batch is not None and client.config.integration.mode == "all": integrations.append( client.create_integration_pull_request( batch=frozen.batch, entries=frozen.queue, ) ) - drained: dict[str, Any] = { + deferred_drain: dict[str, Any] = { "batch_id": frozen.batch["batch_id"], "batch_ids": [frozen.batch["batch_id"]], "merged": [], @@ -3484,6 +3510,11 @@ def own_integration_checks( "integration_required": frozen.overlap_groups, "next_batch": [], } + drained = ( + combine_drain_results(drained, deferred_drain) + if drained + else deferred_drain + ) else: if ( frozen.batch is not None @@ -3504,8 +3535,6 @@ def own_integration_checks( entries=overlap_entries, ) ) - drained = {} - created_clean = [ int(value["number"]) for value in integrations diff --git a/tests/test_cli.py b/tests/test_cli.py index 6d8d653..1552d85 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1677,6 +1677,8 @@ def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: client.config = config client.coordinator_logins = {"coordinator"} client.pipeline_control.return_value = {"state": "running"} + client.queued_numbers.return_value = [] + client.integration_pull_request_numbers.return_value = [] client.create_integration_pull_request.return_value = { "number": 99, "conflict": None, @@ -1710,6 +1712,104 @@ def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: self.assertEqual(result["integrations"][0]["number"], 99) self.assertEqual(result["drain"]["merged"][0]["number"], 99) + def test_reactor_drains_existing_integration_before_new_all_mode_batch( + self, + ) -> None: + config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + "coordinator_actors": ["coordinator"], + }, + "integration": {"mode": "all"}, + } + ) + existing = entry(99, "old.py") + source = entry(1, "new.py") + existing_batch = new_batch([existing], frozen_at="2026-06-20T00:00:00Z") + source_batch = new_batch([source], frozen_at="2026-06-20T00:01:00Z") + existing_frozen = FreezeResult(existing_batch, [existing], [], [], []) + source_frozen = FreezeResult(source_batch, [source], [], [], []) + old_drain = { + "batch_id": existing_batch["batch_id"], + "batch_ids": [existing_batch["batch_id"]], + "integration_required": [], + "merged": [{"number": 99, "merge_sha": "a" * 40}], + "next_batch": [], + "waiting": [], + } + new_drain = { + "batch_id": "new-integration", + "batch_ids": ["new-integration"], + "integration_required": [], + "merged": [{"number": 100, "merge_sha": "b" * 40}], + "next_batch": [], + "waiting": [], + } + client = Mock() + client.config = config + client.coordinator_logins = {"coordinator"} + client.pipeline_control.return_value = {"state": "running"} + client.queued_numbers.return_value = [1, 99] + client.integration_pull_request_numbers.return_value = [99] + client.create_integration_pull_request.return_value = { + "number": 100, + "conflict": None, + } + with ( + patch( + "agent_merge_queue.cli.settle_integration_checks", + side_effect=[[], [{"pull_request": 100, "state": "ready"}]], + ), + patch( + "agent_merge_queue.cli.promote_integrations", + side_effect=[[], [100]], + ), + patch( + "agent_merge_queue.cli.command_promote", + return_value={"promoted": [], "waiting": [], "blocked": []}, + ), + patch( + "agent_merge_queue.cli.freeze_queue", + side_effect=[existing_frozen, source_frozen], + ) as freeze, + patch( + "agent_merge_queue.cli.command_drain", + side_effect=[old_drain, new_drain], + ) as drain, + redirect_stdout(io.StringIO()), + ): + result = command_react(client, follow=False, timeout_seconds=10) + + self.assertEqual( + freeze.call_args_list, + [ + call(client, held_numbers={1}), + call(client, known_entries=[], held_numbers=set()), + ], + ) + self.assertEqual( + drain.call_args_list, + [ + call( + client, + json_output=False, + emit=False, + initial_frozen=existing_frozen, + ), + call(client, json_output=False, emit=False), + ], + ) + client.create_integration_pull_request.assert_called_once_with( + batch=source_batch, + entries=[source], + ) + self.assertEqual( + [value["number"] for value in result["drain"]["merged"]], + [99, 100], + ) + def test_reactor_establishes_overlap_integration_before_draining(self) -> None: config = parse_config( { @@ -1957,6 +2057,8 @@ def test_reactor_does_not_follow_conflicted_all_mode_integration_batch( client.config = config client.coordinator_logins = {"coordinator"} client.pipeline_control.return_value = {"state": "running"} + client.queued_numbers.return_value = [] + client.integration_pull_request_numbers.return_value = [] client.create_integration_pull_request.return_value = { "number": 99, "conflict": {"number": 2, "reason": "merge conflict"}, From 0d1140ca8e1ebc387d94cede032d7e0728f21b2d Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:56:29 -0700 Subject: [PATCH 14/14] Document race-safe label creation --- src/agent_merge_queue/cli.py | 3 +++ tests/test_cli.py | 24 ++++++++++++++++++++++++ 2 files changed, 27 insertions(+) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 8d25fca..77bbb17 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -783,6 +783,9 @@ def ensure_labels_exist(self) -> None: for name, color, description in self.label_specs(): if name in existing: continue + # The pre-read preserves existing label metadata in the normal + # path; --force makes the remaining list/create race idempotent if + # another first-use worker creates this label concurrently. self._run( "label", "create", diff --git a/tests/test_cli.py b/tests/test_cli.py index 1552d85..bd7e515 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -92,6 +92,30 @@ def entry(number: int, *paths: str, state: str = "ready") -> QueueEntry: class QueueCoreTest(unittest.TestCase): + def test_ensure_labels_exist_is_race_safe(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client._json = Mock(return_value=[]) + client._run = Mock() + client.label_specs = Mock( + return_value=(("merge-queue", "0E8A16", "Queue label"),) + ) + + client.ensure_labels_exist() + + client._run.assert_called_once_with( + "label", + "create", + "merge-queue", + "--repo", + "example/repo", + "--color", + "0E8A16", + "--description", + "Queue label", + "--force", + ) + def test_status_is_a_read_only_pipeline_view(self) -> None: status = {"repository": "example/repo"} with (