From e83c765879b12b9f8995188ff706b62ac2148647 Mon Sep 17 00:00:00 2001 From: Coy Geek <65363919+coygeek@users.noreply.github.com> Date: Sat, 20 Jun 2026 21:39:28 -0700 Subject: [PATCH] fix: prove reused integration branches Read existing integration refs before replaying merges, reset stale safe branches back to the computed base, and record already-contained heads only after ancestry proof. Add regression coverage for stale bases, mismatched markers, idempotent reruns, and draft recovery. Co-authored-by: factory-droid[bot] <138933559+factory-droid[bot]@users.noreply.github.com> --- src/agent_merge_queue/cli.py | 503 ++++++++++++++----- tests/test_cli.py | 946 ++++++++++++++++++++++++++++++++++- 2 files changed, 1321 insertions(+), 128 deletions(-) diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 2041997..ea18093 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -90,6 +90,7 @@ ) PULL_REQUEST_NUMBER = re.compile(r"#(\d+)\b") RELEASE_REPAIR_LEASE_PREFIX = "DeployBot release repair lease v1 " +INTEGRATION_SEAL_PREFIX = "DeployBot integration seal v1 " PR_OPENING_PHASES = {"pr-draft", "pr-review", "ready"} FAILED_CHECK_STATES = { "ACTION_REQUIRED", @@ -1366,6 +1367,88 @@ def is_ancestor(self, ancestor: str, descendant: str) -> bool: ) return comparison.get("status") in {"ahead", "identical"} + def branch_ref_sha(self, branch: str) -> str: + value = self._json("api", f"repos/{self.repository}/git/ref/heads/{branch}") + sha = (value.get("object") or {}).get("sha") if isinstance(value, dict) else "" + if not sha: + raise QueueError(f"could not read integration branch {branch}") + return str(sha) + + def integration_pulls( + self, branch: str, *, state: str = "open" + ) -> list[dict[str, Any]]: + values = self._json( + "api", + f"repos/{self.repository}/pulls?state={state}&head={self.owner}:{branch}", + ) + return values if isinstance(values, list) else [] + + def open_integration_pulls(self, branch: str) -> list[dict[str, Any]]: + return self.integration_pulls(branch, state="open") + + def latest_integration_marker(self, pull_number: int) -> dict[str, Any] | None: + return latest_payload( + self.comments(pull_number), + INTEGRATION_MARKER, + self.coordinator_logins, + ) + + @staticmethod + def integration_marker_matches( + marker: dict[str, Any], + *, + batch_id: str, + heads: dict[str, str], + pull_requests: list[int], + ) -> bool: + return ( + marker.get("batch_id") == batch_id + and marker.get("heads") == heads + and marker.get("pull_requests") == pull_requests + ) + + def create_integration_seal( + self, + *, + parent_sha: str, + payload: dict[str, Any], + ) -> str: + parent = self._json( + "api", f"repos/{self.repository}/git/commits/{parent_sha}" + ) + tree_sha = str((parent.get("tree") or {}).get("sha") or "") + if not tree_sha: + raise QueueError("GitHub did not return the integration candidate tree") + value = self._json( + "api", + "--method", + "POST", + f"repos/{self.repository}/git/commits", + "-f", + f"message={INTEGRATION_SEAL_PREFIX}{json.dumps(payload, sort_keys=True)}", + "-f", + f"tree={tree_sha}", + "-f", + f"parents[]={parent_sha}", + ) + seal_sha = str(value.get("sha") or "") + if not seal_sha: + raise QueueError("GitHub did not create the integration seal commit") + return seal_sha + + def integration_seal(self, sha: str) -> dict[str, Any]: + value = self._json("api", f"repos/{self.repository}/git/commits/{sha}") + message = str(value.get("message") or "") + if not message.startswith(INTEGRATION_SEAL_PREFIX): + raise QueueError("existing integration branch is not immutably sealed") + try: + payload = json.loads(message[len(INTEGRATION_SEAL_PREFIX) :]) + except json.JSONDecodeError as error: + raise QueueError("existing integration branch has an invalid seal") from error + if not isinstance(payload, dict): + raise QueueError("existing integration branch has an invalid seal") + return payload + def registry_issue_number(self, *, create: bool) -> int | None: cached = getattr(self, "_registry_issue_cache", None) if cached is not None: @@ -1727,7 +1810,33 @@ def create_integration_pull_request( batch_id = str(batch["batch_id"]) safe_id = re.sub(r"[^a-zA-Z0-9._-]+", "-", batch_id)[-80:] branch = f"{self.config.integration.branch_prefix}/{safe_id}" + staging_branch = ( + f"{self.config.integration.branch_prefix}-staging/" + f"{safe_id}-{secrets.token_hex(8)}" + ) base_sha = self.base_sha() + entry_heads = {str(entry.number): entry.head_sha for entry in entries} + frozen_heads = batch.get("heads") + frozen_pull_requests = batch.get("pull_requests") + if isinstance(frozen_heads, dict) and isinstance( + frozen_pull_requests, list + ): + heads = {str(number): str(sha) for number, sha in frozen_heads.items()} + frozen_numbers = [int(number) for number in frozen_pull_requests] + if set(heads) != {str(number) for number in frozen_numbers}: + raise QueueError("frozen integration batch has inconsistent members") + if any(heads.get(number) != sha for number, sha in entry_heads.items()): + raise QueueError("current integration entries changed after freeze") + else: + heads = entry_heads + frozen_numbers = [entry.number for entry in entries] + batch_heads = heads + batch_numbers = frozen_numbers + heads = entry_heads + pull_requests = [entry.number for entry in entries] + frozen_sources = [(number, heads[str(number)]) for number in pull_requests] + current_tip = base_sha + staging_created = False try: self._json( "api", @@ -1735,148 +1844,308 @@ def create_integration_pull_request( "POST", f"repos/{self.repository}/git/refs", "-f", - f"ref=refs/heads/{branch}", + f"ref=refs/heads/{staging_branch}", "-f", f"sha={base_sha}", ) - except QueueError as error: - if "Reference already exists" not in str(error): - raise + staging_created = True + merged_heads: list[str] = [] + conflict: dict[str, Any] | None = None + for number, head_sha in frozen_sources: + try: + if self.is_ancestor(head_sha, current_tip): + if head_sha not in merged_heads: + merged_heads.append(head_sha) + continue + merge = self._json( + "api", + "--method", + "POST", + f"repos/{self.repository}/merges", + "-f", + f"base={staging_branch}", + "-f", + f"head={head_sha}", + "-f", + f"commit_message=DeployBot batch {batch_id}: PR #{number}", + ) + if isinstance(merge, dict) and merge.get("sha"): + current_tip = str(merge["sha"]) + else: + current_tip = self.branch_ref_sha(staging_branch) + if head_sha not in merged_heads: + merged_heads.append(head_sha) + except QueueError as error: + conflict = { + "number": number, + "head_sha": head_sha, + "reason": str(error), + } + break - merged_heads: list[str] = [] - conflict: dict[str, Any] | None = None - for entry in entries: + seal = { + "base_sha": base_sha, + "batch_id": batch_id, + "conflict": conflict, + "heads": heads, + "merged_heads": merged_heads, + "pull_requests": pull_requests, + } + seal_sha = self.create_integration_seal( + parent_sha=current_tip, + payload=seal, + ) + integration_sha = seal_sha try: self._json( "api", "--method", "POST", - f"repos/{self.repository}/merges", - "-f", - f"base={branch}", + f"repos/{self.repository}/git/refs", "-f", - f"head={entry.head_sha}", + f"ref=refs/heads/{branch}", "-f", - f"commit_message=DeployBot batch {batch_id}: PR #{entry.number}", + f"sha={seal_sha}", ) - merged_heads.append(entry.head_sha) except QueueError as error: - conflict = { - "number": entry.number, - "head_sha": entry.head_sha, - "reason": str(error), + if "Reference already exists" not in str(error): + raise + integration_sha = self.branch_ref_sha(branch) + existing_seal = self.integration_seal(integration_sha) + sealed_head_map_value = existing_seal.get("heads") + sealed_numbers_value = existing_seal.get("pull_requests") + if ( + existing_seal.get("batch_id") != batch_id + or not isinstance(sealed_head_map_value, dict) + or not isinstance(sealed_numbers_value, list) + ): + raise QueueError( + "existing sealed integration branch does not match " + "the active frozen batch" + ) + sealed_head_map = { + str(number): str(sha) + for number, sha in sealed_head_map_value.items() } - break - - existing = self._json( - "api", - f"repos/{self.repository}/pulls?state=open&head={self.owner}:{branch}", - ) - if isinstance(existing, list) and existing: - pull = existing[0] - else: - members = ", ".join(f"#{entry.number}" for entry in entries) - body = ( - f"DeployBot cumulative batch `{batch_id}` for {members}.\n\n" - "Every source head was frozen and independently authorized. " - "If a merge conflict remains, an agent must resolve it without " - "dropping either side before marking this PR ready." - ) - 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'}", + sealed_numbers = [int(number) for number in sealed_numbers_value] + if set(sealed_head_map) != {str(number) for number in sealed_numbers}: + raise QueueError( + "existing sealed integration branch has inconsistent members" + ) + if any( + number not in batch_numbers + or batch_heads.get(str(number)) != sealed_head_map[str(number)] + for number in sealed_numbers + ) or any( + sealed_head_map.get(number) != sha + for number, sha in entry_heads.items() + ): + raise QueueError( + "existing sealed integration branch does not match " + "the active frozen subset" + ) + heads = sealed_head_map + pull_requests = sealed_numbers + sealed_base = str(existing_seal.get("base_sha") or "") + if ( + not re.fullmatch(r"[0-9a-f]{40}", sealed_base) + or not self.is_ancestor(sealed_base, integration_sha) + ): + raise QueueError( + "existing sealed integration branch does not contain " + "its recorded base" + ) + sealed_heads = [ + str(value) for value in existing_seal.get("merged_heads") or [] + ] + if len(sealed_heads) != len(set(sealed_heads)) or not set( + sealed_heads + ).issubset(set(heads.values())): + raise QueueError( + "existing sealed integration branch has invalid merged heads" + ) + for sealed_head in sealed_heads: + if not self.is_ancestor(sealed_head, integration_sha): + raise QueueError( + "existing sealed integration branch does not contain " + "a recorded source head" + ) + sealed_conflict = existing_seal.get("conflict") + if sealed_conflict is not None and not isinstance( + sealed_conflict, dict + ): + raise QueueError( + "existing sealed integration branch has invalid conflict data" + ) + if isinstance(sealed_conflict, dict): + try: + conflict_number = int(sealed_conflict.get("number") or 0) + except (TypeError, ValueError) as error: + raise QueueError( + "existing sealed integration branch has invalid " + "conflict data" + ) from error + if ( + conflict_number not in pull_requests + or str(sealed_conflict.get("head_sha") or "") + != heads[str(conflict_number)] + or not str(sealed_conflict.get("reason") or "") + ): + raise QueueError( + "existing sealed integration branch has invalid " + "conflict data" + ) + if sealed_conflict is None and set(sealed_heads) != set(heads.values()): + raise QueueError( + "existing sealed integration branch is incomplete without " + "a recorded conflict" + ) + seal = existing_seal + conflict = sealed_conflict + merged_heads = [str(value) for value in seal.get("merged_heads") or []] + + existing = self.open_integration_pulls(branch) + reused_existing_pull = bool(existing) + if existing: + pull = existing[0] + else: + members = ", ".join(f"#{number}" for number in pull_requests) + body = ( + f"DeployBot cumulative batch `{batch_id}` for {members}.\n\n" + "Every source head was frozen and independently authorized. " + "If a merge conflict remains, an agent must resolve it without " + "dropping either side before marking this PR ready." ) - 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( + 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 as error: + # Another coordinator may have created the immutable PR + # after our read. Re-read instead of deleting its sealed ref. + existing = self.open_integration_pulls(branch) + if existing: + pull = existing[0] + else: + closed = [ + value + for value in self.integration_pulls( + branch, state="closed" + ) + if not value.get("merged_at") + ] + if not closed: + raise error + previous = closed[0] + reopened = self._json( + "api", + "--method", + "PATCH", + f"repos/{self.repository}/pulls/{previous['number']}", + "-f", + "state=open", + ) + pull = reopened if isinstance(reopened, dict) else previous + reused_existing_pull = True + + author = str((pull.get("user") or {}).get("login") or "") + identity_error: str | None = None + if self.config.integration.require_non_actions_author: + if author.lower() == "github-actions[bot]": + identity_error = ( + "integration PRs require a GitHub App installation token; " + "pass the action's token input and do not use github.token" + ) + elif author.lower() not in { + login.lower() for login in self.coordinator_logins + }: + identity_error = ( + f"integration PR author {author or ''} is not trusted; " + "add the GitHub App bot login to queue.coordinator_actors" + ) + if identity_error: + cleanup_calls = ( + ( + "api", + "--method", + "PATCH", + f"repos/{self.repository}/pulls/{pull['number']}", + "-f", + "state=closed", + ), + ( "api", "--method", "DELETE", f"repos/{self.repository}/git/refs/heads/{branch}", - ) - except QueueError: - pass - raise - author = str((pull.get("user") or {}).get("login") or "") - identity_error: str | None = None - if self.config.integration.require_non_actions_author: - if author.lower() == "github-actions[bot]": - identity_error = ( - "integration PRs require a GitHub App installation token; " - "pass the action's token input and do not use github.token" - ) - elif author.lower() not in { - login.lower() for login in self.coordinator_logins - }: - identity_error = ( - f"integration PR author {author or ''} is not trusted; " - "add the GitHub App bot login to queue.coordinator_actors" + ), ) - if identity_error: - # Installation-token identity is available on the PR response, - # unlike GET /user. Remove the unusable workflow-token PR before it - # can own the frozen source batch. - try: + for cleanup_call in cleanup_calls: + try: + self._run(*cleanup_call) + except QueueError: + pass + raise QueueError(identity_error) + integration_number = int(pull["number"]) + if reused_existing_pull and conflict is not None and not pull.get("draft"): self._run( - "api", - "--method", - "PATCH", - f"repos/{self.repository}/pulls/{pull['number']}", - "-f", - "state=closed", + "pr", + "ready", + str(integration_number), + "--undo", + "--repo", + self.repository, ) + if reused_existing_pull and conflict is None and pull.get("draft"): self._run( - "api", - "--method", - "DELETE", - f"repos/{self.repository}/git/refs/heads/{branch}", + "pr", "ready", str(integration_number), "--repo", self.repository ) - except QueueError: - pass - raise QueueError(identity_error) - number = int(pull["number"]) - marker = { - "author": author, - "base_sha": base_sha, - "batch_id": batch_id, - "conflict": conflict, - "created_at": utc_now(), - "heads": {str(entry.number): entry.head_sha for entry in entries}, - "merged_heads": merged_heads, - "pull_requests": [entry.number for entry in entries], - } - self.comment(number, integration_body(marker)) - # The cumulative PR now owns the queue position, but the source intent - # remains discoverable until delivery succeeds. This lets promotion - # recover the exact-head request if the integration closes, conflicts, - # or rejects a source that changed after the batch was frozen. - for entry in entries: - source_labels = self.labels(entry.number) - if self.config.queue_label in source_labels: - self.remove_label(entry.number, self.config.queue_label) - return { - "number": number, - "url": pull.get("html_url"), - "branch": branch, - "conflict": conflict, - "batch_id": batch_id, - } + marker = { + **seal, + "author": author, + "created_at": utc_now(), + "integration_sha": integration_sha, + } + self.comment(integration_number, integration_body(marker)) + # The cumulative PR now owns the queue position, but the source intent + # remains discoverable until delivery succeeds. This lets promotion + # recover the exact-head request if the integration closes, conflicts, + # or rejects a source that changed after the batch was frozen. + for source_number in pull_requests: + source_labels = self.labels(source_number) + if self.config.queue_label in source_labels: + self.remove_label(source_number, self.config.queue_label) + return { + "number": integration_number, + "url": pull.get("html_url"), + "branch": branch, + "conflict": conflict, + "batch_id": batch_id, + } + finally: + if staging_created: + try: + self._run( + "api", + "--method", + "DELETE", + f"repos/{self.repository}/git/refs/heads/{staging_branch}", + ) + except QueueError: + pass def _paged_api(self, endpoint: str) -> list[dict[str, Any]]: data = self._json("api", "--paginate", "--slurp", endpoint) diff --git a/tests/test_cli.py b/tests/test_cli.py index 491756a..dc727f6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -11,6 +11,7 @@ from agent_merge_queue.cli import ( FreezeResult, GitHub, + INTEGRATION_SEAL_PREFIX, QueueEntry, QueueError, RELEASE_REPAIR_LEASE_PREFIX, @@ -73,11 +74,13 @@ ) from agent_merge_queue.config import parse_config from agent_merge_queue.records import ( + INTEGRATION_MARKER, PullRequestThreadOwnerRecord, ThreadRecord, control_body, integration_body, intent_body, + latest_payload, release_repair_body, repair_body, thread_record_body, @@ -121,6 +124,26 @@ def entry(number: int, *paths: str, state: str = "ready") -> QueueEntry: return value +def integration_seal_payload( + batch: dict[str, object], + entries: list[QueueEntry], + *, + base_sha: str, + merged_heads: list[str] | None = None, + conflict: dict[str, object] | None = None, +) -> dict[str, object]: + return { + "base_sha": base_sha, + "batch_id": batch["batch_id"], + "conflict": conflict, + "heads": {str(value.number): value.head_sha for value in entries}, + "merged_heads": merged_heads + if merged_heads is not None + else list(dict.fromkeys(value.head_sha for value in entries)), + "pull_requests": [value.number for value in entries], + } + + def deployment_notification( *, main_sha: str, state: str = "pending", merge_sha: str = "m" * 40 ) -> dict[str, object]: @@ -2541,12 +2564,16 @@ def test_integration_can_require_non_actions_author(self) -> None: client._json = Mock( side_effect=[ {}, - {}, + {"status": "behind"}, + {"sha": "1" * 40}, + {"status": "behind"}, + {"sha": "2" * 40}, {}, [], {"number": 99, "user": {"login": "github-actions[bot]"}}, ] ) + client.create_integration_seal = Mock(return_value="s" * 40) client._run = Mock(return_value="") with self.assertRaisesRegex(QueueError, "GitHub App installation token"): @@ -2554,7 +2581,7 @@ def test_integration_can_require_non_actions_author(self) -> None: batch={"batch_id": "batch"}, entries=[entry(1), entry(2)], ) - self.assertEqual(client._run.call_count, 2) + self.assertEqual(client._run.call_count, 3) def test_integration_requires_app_author_to_be_a_coordinator(self) -> None: config = parse_config( @@ -2575,12 +2602,16 @@ def test_integration_requires_app_author_to_be_a_coordinator(self) -> None: client._json = Mock( side_effect=[ {}, - {}, + {"status": "behind"}, + {"sha": "1" * 40}, + {"status": "behind"}, + {"sha": "2" * 40}, {}, [], {"number": 99, "user": {"login": "deploybot-app[bot]"}}, ] ) + client.create_integration_seal = Mock(return_value="s" * 40) client._run = Mock(return_value="") with self.assertRaisesRegex(QueueError, "queue.coordinator_actors"): @@ -2588,7 +2619,7 @@ def test_integration_requires_app_author_to_be_a_coordinator(self) -> None: batch={"batch_id": "batch"}, entries=[entry(1), entry(2)], ) - self.assertEqual(client._run.call_count, 2) + self.assertEqual(client._run.call_count, 3) def test_queue_marker_rejects_forgery_and_free_text_injection(self) -> None: sha = "a" * 40 @@ -3806,6 +3837,32 @@ def test_metrics_skip_deployments_that_finished_before_the_merge(self) -> None: datetime(2026, 6, 20, 1, tzinfo=timezone.utc), ) + def test_integration_seal_round_trips_through_git_commit(self) -> None: + parent_sha = "p" * 40 + seal_sha = "s" * 40 + payload = {"batch_id": "batch-1", "heads": {"1": "1" * 40}} + client = object.__new__(GitHub) + client.repository = "example/repo" + client._json = Mock( + side_effect=[ + {"tree": {"sha": "t" * 40}}, + {"sha": seal_sha}, + { + "message": INTEGRATION_SEAL_PREFIX + + json.dumps(payload, sort_keys=True) + }, + ] + ) + + created = client.create_integration_seal( + parent_sha=parent_sha, + payload=payload, + ) + restored = client.integration_seal(created) + + self.assertEqual(created, seal_sha) + self.assertEqual(restored, payload) + def test_integration_pr_contains_every_frozen_head(self) -> None: first = entry(1, "shared.py") second = entry(2, "shared.py") @@ -3827,12 +3884,17 @@ def test_integration_pr_contains_every_frozen_head(self) -> None: client._json = Mock( side_effect=[ {}, + {"status": "behind"}, {"sha": "1" * 40}, + {"status": "behind"}, {"sha": "2" * 40}, + {}, [], {"number": 99, "html_url": "https://example.test/99"}, ] ) + client.create_integration_seal = Mock(return_value="s" * 40) + client._run = Mock(return_value="") client.comment = Mock() client.labels = Mock(return_value={"merge-queue", "deploy-requested"}) client.remove_label = Mock() @@ -3857,6 +3919,322 @@ def test_integration_pr_contains_every_frozen_head(self) -> None: ], ) + def test_concurrent_integration_pr_creation_reuses_the_winner(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value="b" * 40) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + {}, + [], + QueueError("pull request already exists"), + [{"number": 99, "html_url": "https://example.test/99"}], + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client._run = Mock(return_value="") + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + client.comment.assert_called_once() + + def test_sealed_recovery_allows_prs_with_the_same_head(self) -> None: + first = entry(1, "shared.py") + second = entry(2, "shared.py") + second.head_sha = first.head_sha + entries = [first, second] + batch = new_batch(entries, frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + sealed_sha = "f" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(side_effect=[False, True, True, True]) + client._json = Mock( + side_effect=[ + {}, + {"sha": "m" * 40}, + QueueError("Reference already exists"), + {"object": {"sha": sealed_sha}}, + [{"number": 99, "html_url": "https://example.test/99"}], + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + entries, + base_sha=base_sha, + ) + ) + client._run = Mock(return_value="") + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=entries) + + self.assertEqual(result["number"], 99) + marker = latest_payload( + [ + { + "id": 1, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": client.comment.call_args.args[1], + } + ], + INTEGRATION_MARKER, + {"coordinator"}, + ) + self.assertEqual(marker["merged_heads"], [first.head_sha]) + + def test_closed_unmerged_integration_pr_is_reopened(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + closed_pull = { + "number": 99, + "html_url": "https://example.test/99", + "state": "closed", + "merged_at": None, + } + reopened_pull = {**closed_pull, "state": "open"} + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value="b" * 40) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + {}, + [], + QueueError("pull request already exists"), + [], + [closed_pull], + reopened_pull, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client._run = Mock(return_value="") + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + self.assertTrue( + any( + "pulls?state=closed" in str(call.args) + for call in client._json.call_args_list + ) + ) + self.assertTrue( + any( + "repos/example/repo/pulls/99" in call.args + and "state=open" in call.args + for call in client._json.call_args_list + ) + ) + + def test_sealed_recovery_uses_full_batch_after_partial_label_cleanup(self) -> None: + first = entry(1, "shared.py") + second = entry(2, "other.py") + batch = new_batch([first, second], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + sealed_sha = "f" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": sealed_sha}}, + [{"number": 99, "html_url": "https://example.test/99"}], + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first, second], + base_sha=base_sha, + ) + ) + client._run = Mock(return_value="") + client.comment = Mock() + client.labels = Mock( + side_effect=lambda number: ( + {"merge-queue"} if number == second.number else set() + ) + ) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=[second]) + + self.assertEqual(result["number"], 99) + marker = latest_payload( + [ + { + "id": 1, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": client.comment.call_args.args[1], + } + ], + INTEGRATION_MARKER, + {"coordinator"}, + ) + self.assertEqual(marker["pull_requests"], [1, 2]) + self.assertEqual(marker["heads"], batch["heads"]) + client.remove_label.assert_called_once_with(2, "merge-queue") + + def test_overlap_integration_preserves_the_selected_subset(self) -> None: + first = entry(1, "shared.py") + second = entry(2, "shared.py") + independent = entry(3, "independent.py") + batch = new_batch( + [first, second, independent], 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.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + {}, + [], + {"number": 99, "html_url": "https://example.test/99"}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client._run = Mock(return_value="") + client.comment = Mock() + client.labels = Mock(return_value={"merge-queue"}) + client.remove_label = Mock() + + result = client.create_integration_pull_request( + batch=batch, + entries=[first, second], + ) + + self.assertEqual(result["number"], 99) + seal = client.create_integration_seal.call_args.kwargs["payload"] + self.assertEqual(seal["pull_requests"], [1, 2]) + self.assertEqual( + seal["heads"], + {"1": first.head_sha, "2": second.head_sha}, + ) + self.assertEqual( + [call.args[0] for call in client.remove_label.call_args_list], + [1, 2], + ) + + def test_recovered_seal_rejects_non_object_conflict(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + malformed = integration_seal_payload( + batch, + [first], + base_sha="b" * 40, + merged_heads=[], + ) + malformed["conflict"] = "merge conflict" + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.base_sha = Mock(return_value="b" * 40) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": "f" * 40}}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock(return_value=malformed) + client._run = Mock(return_value="") + + with self.assertRaisesRegex(QueueError, "invalid conflict data"): + client.create_integration_pull_request(batch=batch, entries=[first]) + def test_conflicted_integration_keeps_source_intents_discoverable(self) -> None: first = entry(1, "shared.py") second = entry(2, "shared.py") @@ -3878,12 +4256,17 @@ def test_conflicted_integration_keeps_source_intents_discoverable(self) -> None: client._json = Mock( side_effect=[ {}, + {"status": "behind"}, {"sha": "1" * 40}, + {"status": "behind"}, QueueError("gh: Merge conflict (HTTP 409)"), + {}, [], {"number": 99, "html_url": "https://example.test/99"}, ] ) + client.create_integration_seal = Mock(return_value="s" * 40) + client._run = Mock(return_value="") client.comment = Mock() client.labels = Mock(return_value={"merge-queue", "deploy-requested"}) client.remove_label = Mock() @@ -4058,7 +4441,7 @@ def test_resumed_integration_finishes_source_label_delegation(self) -> None: ], ) - def test_failed_integration_pr_creation_removes_its_orphan_branch(self) -> None: + def test_failed_integration_pr_creation_preserves_sealed_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") @@ -4079,24 +4462,565 @@ def test_failed_integration_pr_creation_removes_its_orphan_branch(self) -> None: client._json = Mock( side_effect=[ {}, + {"status": "behind"}, {"sha": "1" * 40}, + {"status": "behind"}, {"sha": "2" * 40}, + {}, [], QueueError("GitHub Actions may not create pull requests"), + [], + [], ] ) + client.create_integration_seal = Mock(return_value="s" * 40) 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}", + client._run.assert_called_once() + cleanup = client._run.call_args.args + self.assertEqual(cleanup[:3], ("api", "--method", "DELETE")) + self.assertIn("git/refs/heads/deploybot/integration-staging/", cleanup[3]) + + def test_existing_sealed_integration_on_older_base_is_recovered(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + old_base_sha = "a" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": old_base_sha}}, + [], + {"number": 99, "html_url": "https://example.test/99"}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first], + base_sha=old_base_sha, + ) + ) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": old_base_sha, + "batch_id": batch["batch_id"], + "heads": {"1": first.head_sha}, + "integration_sha": old_base_sha, + "pull_requests": [1], + } + ), + } + ] + ) + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + client._run = Mock(return_value="") + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + self.assertFalse(any("PATCH" in call.args for call in client._json.call_args_list)) + self.assertFalse( + any( + "repos/example/repo/merges" in call.args + for call in client._json.call_args_list + ) + ) + client.comment.assert_called_once() + + def test_existing_integration_branch_with_unmarked_tip_fails_closed(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + marked_tip = "f" * 40 + untrusted_tip = "x" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": untrusted_tip}}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + side_effect=QueueError("existing integration branch is not immutably sealed") + ) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": base_sha, + "batch_id": batch["batch_id"], + "heads": {"1": first.head_sha}, + "integration_sha": marked_tip, + "pull_requests": [1], + } + ), + } + ] + ) + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + client._run = Mock(return_value="") + + with self.assertRaisesRegex(QueueError, "not immutably sealed"): + client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertFalse(any("PATCH" in call.args for call in client._json.call_args_list)) + self.assertFalse( + any( + "repos/example/repo/merges" in call.args + for call in client._json.call_args_list + ) + ) + client.comment.assert_not_called() + + def test_existing_sealed_integration_without_pr_is_recovered(self) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": base_sha}}, + [], + {"number": 99, "html_url": "https://example.test/99"}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first], + base_sha=base_sha, + ) + ) + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + client._run = Mock(return_value="") + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + client.comment.assert_called_once() + + def test_existing_integration_branch_with_mismatched_marker_is_rejected( + self, + ) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": "a" * 40}}, + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + mismatched_seal = integration_seal_payload( + batch, + [first], + base_sha=base_sha, + ) + mismatched_seal["batch_id"] = "different-batch" + client.integration_seal = Mock(return_value=mismatched_seal) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": base_sha, + "batch_id": "different-batch", + "heads": {"1": first.head_sha}, + "pull_requests": [1], + } + ), + } + ] + ) + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + client._run = Mock(return_value="") + + with self.assertRaisesRegex(QueueError, "does not match"): + client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertFalse( + any( + "repos/example/repo/merges" in call.args + for call in client._json.call_args_list + ) + ) + self.assertFalse( + any("PATCH" in call.args for call in client._json.call_args_list) + ) + client.comment.assert_not_called() + + def test_existing_integration_branch_rerun_is_idempotent_and_truthful( + self, + ) -> None: + first = entry(1, "shared.py") + second = entry(2, "shared.py") + batch = new_batch([first, second], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + tip_sha = "f" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": tip_sha}}, + [{"number": 99, "html_url": "https://example.test/99"}], + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first, second], + base_sha=base_sha, + ) + ) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": base_sha, + "batch_id": batch["batch_id"], + "heads": { + "1": first.head_sha, + "2": second.head_sha, + }, + "conflict": None, + "integration_sha": tip_sha, + "merged_heads": [first.head_sha, second.head_sha], + "pull_requests": [1, 2], + } + ), + } + ] + ) + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + client._run = Mock(return_value="") + + result = client.create_integration_pull_request( + batch=batch, entries=[first, second] + ) + + self.assertEqual(result["number"], 99) + self.assertFalse( + any( + "repos/example/repo/merges" in call.args + for call in client._json.call_args_list + ) + ) + marker = latest_payload( + [ + { + "id": 2, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": client.comment.call_args.args[1], + } + ], + INTEGRATION_MARKER, + {"coordinator"}, + ) + self.assertEqual(marker["base_sha"], base_sha) + self.assertEqual(marker["heads"], {"1": first.head_sha, "2": second.head_sha}) + self.assertEqual(marker["integration_sha"], tip_sha) + self.assertEqual(marker["merged_heads"], [first.head_sha, second.head_sha]) + + def test_existing_draft_integration_pr_is_marked_ready_after_clean_rerun( + self, + ) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + tip_sha = "f" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(return_value=True) + client._json = Mock( + side_effect=[ + {}, + QueueError("Reference already exists"), + {"object": {"sha": tip_sha}}, + [{"number": 99, "html_url": "https://example.test/99", "draft": True}], + ] + ) + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first], + base_sha=base_sha, + ) + ) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": base_sha, + "batch_id": batch["batch_id"], + "conflict": None, + "heads": {"1": first.head_sha}, + "integration_sha": tip_sha, + "merged_heads": [first.head_sha], + "pull_requests": [1], + } + ), + } + ] + ) + client._run = Mock() + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + client._run.assert_any_call( + "pr", "ready", "99", "--repo", "example/repo" + ) + marker = latest_payload( + [ + { + "id": 2, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": client.comment.call_args.args[1], + } + ], + INTEGRATION_MARKER, + {"coordinator"}, + ) + self.assertIsNone(marker["conflict"]) + + def test_existing_ready_integration_pr_is_redrafted_after_sealed_conflict( + self, + ) -> None: + first = entry(1, "shared.py") + batch = new_batch([first], frozen_at="2026-06-20T00:00:00Z") + base_sha = "b" * 40 + tip_sha = "f" * 40 + client = object.__new__(GitHub) + client.config = parse_config( + { + "queue": { + "required_checks": ["CI"], + "trusted_actors": ["trusted"], + }, + "integration": {"mode": "all"}, + } + ) + client.repository = "example/repo" + client.owner = "example" + client.name = "repo" + client.coordinator_logins = {"coordinator"} + client.base_sha = Mock(return_value=base_sha) + client.is_ancestor = Mock(side_effect=[False, True]) + client._json = Mock( + side_effect=[ + {}, + QueueError("merge conflict"), + QueueError("Reference already exists"), + {"object": {"sha": tip_sha}}, + [{"number": 99, "html_url": "https://example.test/99", "draft": False}], + ] + ) + conflict = { + "number": 1, + "head_sha": first.head_sha, + "reason": "merge conflict", + } + client.create_integration_seal = Mock(return_value="s" * 40) + client.integration_seal = Mock( + return_value=integration_seal_payload( + batch, + [first], + base_sha=base_sha, + merged_heads=[], + conflict=conflict, + ) + ) + client.comments = Mock( + return_value=[ + { + "id": 1, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "coordinator"}, + "body": integration_body( + { + "base_sha": base_sha, + "batch_id": batch["batch_id"], + "conflict": { + "number": 1, + "head_sha": first.head_sha, + "reason": "merge conflict", + }, + "heads": {"1": first.head_sha}, + "integration_sha": tip_sha, + "merged_heads": [], + "pull_requests": [1], + } + ), + } + ] + ) + client._run = Mock() + client.comment = Mock() + client.labels = Mock(return_value=set()) + client.remove_label = Mock() + + result = client.create_integration_pull_request(batch=batch, entries=[first]) + + self.assertEqual(result["number"], 99) + client._run.assert_any_call( + "pr", "ready", "99", "--undo", "--repo", "example/repo" + ) + marker = latest_payload( + [ + { + "id": 2, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": client.comment.call_args.args[1], + } + ], + INTEGRATION_MARKER, + {"coordinator"}, ) + self.assertEqual(marker["conflict"]["number"], 1) + self.assertEqual(marker["merged_heads"], []) def test_reactor_uses_cumulative_pr_in_all_mode(self) -> None: config = parse_config(