diff --git a/README.md b/README.md index 9b2541b..f939246 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 @@ -155,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 ``` @@ -168,6 +175,9 @@ can pause further merges until `deploybot unpause`. [pipeline] 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 @@ -184,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/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/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 ed813ed..77bbb17 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -12,10 +12,12 @@ 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 from typing import Any, Callable, Iterable +from urllib.parse import quote from . import __version__ from .config import ConfigError, QueueConfig, initialize_config, load_config @@ -137,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"): @@ -282,11 +287,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 +312,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 +604,8 @@ class QueueEntry: dependencies: list[int] state: str = "waiting" reasons: list[str] | None = None + priority_at: str | None = None + integration_batch_id: str | None = None def classify( self, @@ -668,6 +683,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 @@ -708,8 +724,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 +752,40 @@ 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 + # 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", @@ -768,6 +817,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" @@ -781,7 +840,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, @@ -805,6 +864,34 @@ 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 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", + "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", @@ -848,8 +935,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( @@ -884,7 +976,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 @@ -962,6 +1054,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" ) @@ -971,13 +1066,17 @@ 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. + canonical = min(int(value["number"]) for value in matches) + self._registry_issue_cache = canonical + return canonical if not create: return None - self.ensure_labels() + self.ensure_labels_exist() created = self._json( "api", "--method", @@ -990,7 +1089,11 @@ 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. + 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( @@ -1101,22 +1204,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, @@ -1183,6 +1301,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", @@ -1215,6 +1358,11 @@ 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, ) -> QueueEntry: fields = ",".join( ( @@ -1244,8 +1392,10 @@ 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 @@ -1253,30 +1403,26 @@ 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 []) - 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, - ) - else source_paths - ) - target.append(path) + 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 = ( + 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( @@ -1306,12 +1452,20 @@ def snapshot( dependencies=structured_dependencies( 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, 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]: @@ -1329,7 +1483,12 @@ 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.number if entry.priority_at else 0, + entry.queued_at or "9999", + entry.number, + ), ) def add_label(self, number: int, label: str) -> None: @@ -1400,35 +1559,82 @@ 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( + def externally_integrated_merge( + self, number: int, expected_head: str + ) -> str | None: + value = self._json( "pr", "view", str(number), "--repo", self.repository, "--json", - "headRefOid,isDraft,labels,state", + "headRefOid,mergeCommit,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") + 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, + 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), @@ -1666,8 +1872,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,54 +2068,160 @@ def record_repair( return payload -def command_promote(client: GitHub, *, emit: bool = True) -> dict[str, Any]: +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, + *, + 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]] = [] 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 - entry = client.snapshot(number, require_marker=False, allow_blocked_label=True) + return "skip", None, None + entry = client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + known_comments=comments, + defer_paths_until_ready=True, + ) 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, + ) + if client.config.blocked_label in entry.labels: + 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": - queue_from_intent(client, entry, intent) - promoted.append(number) - elif entry.state == "blocked": + changed = queue_from_intent( + client, + entry, + intent, + comments=comments, + labels=set(entry.labels), + ) + if changed: + 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)) 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.open_pull_request_numbers(): + for number in client.integration_pull_request_numbers(): comments = client.comments(number) integration = latest_payload( comments, @@ -1915,7 +2233,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 "") @@ -1947,6 +2270,116 @@ 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() + ): + 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( + { + "branch": branch, + "dispatched_ci": dispatched, + "checks": exact_checks, + "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) @@ -2029,6 +2462,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 +2489,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 +2512,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 +2532,55 @@ 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 +2591,7 @@ def queue_from_intent( "intent_id": intent_id, }, ) + return True @dataclass @@ -2140,8 +2612,65 @@ 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, + held_numbers: set[int] | None = None, +) -> FreezeResult: + held_numbers = held_numbers or set() + if known_entries is None: + all_entries = [ + entry for entry in client.queue() if entry.number not in held_numbers + ] + else: + queued_numbers = set(client.queued_numbers()) - held_numbers + 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: ( + entry.priority_at or entry.queued_at or "9999", + entry.number if entry.priority_at else 0, + entry.queued_at or "9999", + entry.number, + ), + ) entries, blocked_entries = split_blocked_entries( all_entries, client.config.blocked_label ) @@ -2259,6 +2788,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 +2818,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 +2847,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 +2868,39 @@ 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, + 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: + 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 +2915,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,20 +2938,43 @@ def command_merge( + ", ".join(f"#{value}" for value in missing_dependencies) ) - fresh = client.snapshot(number) - 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) + 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( @@ -2430,6 +3002,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: @@ -2490,7 +3063,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 +3119,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 +3133,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"]) @@ -2711,6 +3297,100 @@ 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 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 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, *, @@ -2723,30 +3403,122 @@ def command_react( result = {"state": "paused", "reason": control.get("reason")} 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) - integrations: list[dict[str, Any]] = [] - if ( - frozen.batch is not None - and client.config.integration.mode == "all" - and not promoted_integrations + + reconciled_merges = reconcile_externally_merged_threads(client) + + 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, + 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, + 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 + + 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"] = [ + { + "number": number, + "overlapping_waiting": waiting, + "reason": "waiting for overlapping deploy intents to finish gates", + } + for number, waiting in overlap_holds.items() + ] + 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": 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": [], "waiting": [], "integration_required": frozen.overlap_groups, "next_batch": [], } + drained = ( + combine_drain_results(drained, deferred_drain) + if drained + else deferred_drain + ) else: - drained = command_drain(client, json_output=False, emit=False) if ( frozen.batch is not None and client.config.integration.mode == "overlap" @@ -2766,6 +3538,47 @@ def command_react( entries=overlap_entries, ) ) + 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, + 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) + ) + + if not drained: + # Establish durable ownership for every overlapping source before any + # independent member of the frozen batch is allowed to merge. A setup + # 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: @@ -2799,6 +3612,8 @@ def command_react( "drain": drained, "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/src/agent_merge_queue/config.py b/src/agent_merge_queue/config.py index 99d7571..9e76e51 100644 --- a/src/agent_merge_queue/config.py +++ b/src/agent_merge_queue/config.py @@ -44,6 +44,9 @@ class PipelineConfig: thread_active_hours: int ci_workflows: tuple[str, ...] 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 @@ -116,6 +119,9 @@ class QueueConfig: thread_active_hours = 72 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 @@ -158,6 +164,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 +429,21 @@ 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, + ), + ci_failure_grace_seconds=_non_negative_int( + pipeline.get("ci_failure_grace_seconds"), + "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/src/agent_merge_queue/doctor.py b/src/agent_merge_queue/doctor.py index 4b2e314..46968c2 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 ) @@ -134,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/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_cli.py b/tests/test_cli.py index 2318db4..bd7e515 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -30,16 +30,24 @@ completed_batch_ids, entries_in_batch, effective_queue_marker, + freeze_queue, generated_only_change, latest_batch_marker, latest_marker, marker_queued_at, + marker_priority_at, main, + near_ready_overlap_holds, new_batch, overlap_groups, + promote_integrations, queue_state_body, + queue_from_intent, queue_timestamp, + reconcile_externally_merged_threads, reusable_batch, + settle_integration_checks, + should_settle_batch, structured_dependencies, ) from agent_merge_queue.config import parse_config @@ -84,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 ( @@ -107,6 +139,278 @@ 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_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_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_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( + { + "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) + 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 + 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=[], + ) + client.merge.assert_called_once_with( + 1, + value.head_sha, + 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( @@ -141,6 +445,100 @@ 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_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( [ @@ -183,6 +581,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( [ @@ -463,6 +877,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( @@ -504,7 +945,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 +1014,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) @@ -774,6 +1231,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 @@ -985,6 +1461,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"] @@ -1135,6 +1642,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( { @@ -1154,17 +1701,293 @@ 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.queued_numbers.return_value = [] + client.integration_pull_request_numbers.return_value = [] + client.create_integration_pull_request.return_value = { + "number": 99, + "conflict": None, + } + with ( + 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", + 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_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_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( + { + "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, "conflict": None} + ) + + def drain(*_args, **_kwargs): + events.append("drain") + return {"merged": [{"number": 3, "merge_sha": "a" * 40}]} + + with ( + 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), + redirect_stdout(io.StringIO()), + ): + command_react(client, follow=False, timeout_seconds=10) + + self.assertEqual(events, ["integration", "settle", "drain", "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"), ): - result = command_react(client, follow=False, timeout_seconds=10) + command_react(client, follow=False, timeout_seconds=10) + 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() @@ -1236,7 +2059,9 @@ 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": { @@ -1256,7 +2081,12 @@ 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.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"}, + } client.base_sha.return_value = sha client.workflow_runs.return_value = [ { @@ -1294,7 +2124,122 @@ 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"}] + 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 + 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"], + ) + 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_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() diff --git a/tests/test_config.py b/tests/test_config.py index ffab40b..e031d43 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -183,6 +183,9 @@ def test_parses_pipeline_integration_and_health_policy(self) -> None: "pipeline": { "ci_workflows": ["Main CI"], "deploy_workflows": ["Production"], + "batch_settle_seconds": 30, + "ci_failure_grace_seconds": 45, + "promotion_workers": 3, "auto_promote": False, "verifications": [ { @@ -196,6 +199,9 @@ 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") @@ -231,6 +237,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__": 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_doctor.py b/tests/test_doctor.py index aa94a89..055cd6e 100644 --- a/tests/test_doctor.py +++ b/tests/test_doctor.py @@ -118,6 +118,108 @@ 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"]) + + 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() 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 diff --git a/tests/test_skill.py b/tests/test_skill.py index 32dba4a..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) @@ -87,6 +85,12 @@ 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",