From 9f7f35153581aed1ea283d7ae34f1d766b98564f Mon Sep 17 00:00:00 2001 From: Matthew Berman <748450+mberman84@users.noreply.github.com> Date: Sun, 21 Jun 2026 09:34:11 -0700 Subject: [PATCH] Fix DeployBot delivery stalls and observability --- README.md | 26 +- .../claude-code/skills/deploybot/SKILL.md | 4 + .../skills/deploybot/SKILL.md | 4 + docs/reference.md | 13 +- examples/github-workflow.yml | 4 + skills/deploybot/SKILL.md | 4 + src/agent_merge_queue/cli.py | 507 +++++++++++++++--- tests/test_cli.py | 443 ++++++++++++++- tests/test_skill.py | 2 + 9 files changed, 914 insertions(+), 93 deletions(-) diff --git a/README.md b/README.md index c551078..4330060 100644 --- a/README.md +++ b/README.md @@ -87,11 +87,12 @@ the user does not repeat `deploy`. No polling timer is involved. 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. 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: +aligned with `pipeline.ci_workflows`. A five-minute scheduled reconciliation +rereads all durable state in case GitHub concurrency coalesces the last pending +event in a burst. The privileged worker never checks out or 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@b06346119a29d0be1c25d1643fdc06a03f7b94a9 @@ -150,14 +151,19 @@ they cannot create the original per-pull-request deploy intent. ## Delivery controller `deploybot status` reports active metadata-only agent threads, pending native -notifications, every PR stage, deploy requests, queue order, overlaps, -exact-`main` CI, deployment, and pipeline pause state. It never stores prompts, +notifications, every PR stage, deploy requests and their exact authorized heads, +queue order, queued and pre-queue intent overlaps, exact-`main` CI, deployment, +and pipeline pause state. It alerts when a deploy request exceeds the configured +ready-to-merge target and names the current gate. It never stores prompts, transcripts, source, or credentials. `deploybot react` promotes ready intent, skips blockers, drains independent -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: +work, and creates integration PRs when configured. Draft status and incomplete +checks or reviews remain waiting states; they do not create a repair latch. A +conflict, failed gate, unresolved review, manual block, or stale authorized head +produces a repair handoff containing the source thread, base/head SHAs, source +paths, and one return command. Old draft-only repair latches self-clear once the +controller recognizes them: 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, diff --git a/adapters/claude-code/skills/deploybot/SKILL.md b/adapters/claude-code/skills/deploybot/SKILL.md index b351676..caed755 100644 --- a/adapters/claude-code/skills/deploybot/SKILL.md +++ b/adapters/claude-code/skills/deploybot/SKILL.md @@ -27,6 +27,10 @@ exact heads, pending checks/reviews, blockers, dependencies, overlaps, exact is still waiting outside it. Never publish prompts, transcripts, source, or credentials to the thread registry. +Treat `active_intent_overlap_groups` as the pre-queue collision view. Treat a +`request-to-ready` alert as an ownership signal: report its exact active gate and +the source-thread action, rather than describing the empty queue as idle. + Use these state meanings: - `ready`: every configured merge gate passes. diff --git a/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md b/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md index b351676..caed755 100644 --- a/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md +++ b/adapters/codex/agent-merge-queue/skills/deploybot/SKILL.md @@ -27,6 +27,10 @@ exact heads, pending checks/reviews, blockers, dependencies, overlaps, exact is still waiting outside it. Never publish prompts, transcripts, source, or credentials to the thread registry. +Treat `active_intent_overlap_groups` as the pre-queue collision view. Treat a +`request-to-ready` alert as an ownership signal: report its exact active gate and +the source-thread action, rather than describing the empty queue as idle. + Use these state meanings: - `ready`: every configured merge gate passes. diff --git a/docs/reference.md b/docs/reference.md index 8b2411a..2daf74b 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -23,7 +23,7 @@ DeployBot resolves the pull request for the current branch. | `deploybot init [--force]` | Write a safe starter policy. Existing files are preserved unless `--force` is supplied. | | `deploybot ensure-labels` | Create or refresh the configured queue, blocked, intent, pause, and registry labels. | | `deploybot doctor [--json]` | Check authentication, policy, labels, actors, checks, workflows, and branch protection without changing repository state. | -| `deploybot status [--json]` | Read active thread metadata, pending native notifications, PR stages, deploy intent, queue state, exact-main CI, deployment, and pipeline control state. | +| `deploybot status [--json]` | Read active thread metadata, pending native notifications, PR stages, exact-head deploy intent, pre-queue intent overlaps, request-stage timing alerts, queue state, exact-main CI, deployment, and pipeline control state. | | `deploybot plan [--json]` | Read the ordered queue, dependencies, blockers, and source-overlap groups. | | `deploybot inspect [PR] [--json]` | Evaluate one exact PR head without granting merge authority. | | `deploybot metrics [--limit N] [--json]` | Summarize p50, p95, and maximum delivery timings for recent merged PRs. The default limit is 25. | @@ -164,7 +164,7 @@ Provider fields are: | `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. | +| `ready_to_merge_target_minutes` | Positive request-to-ready and queued-to-merge timing target; default 15. | | `merge_to_live_target_minutes` | Positive timing target; default 10. | | `auto_promote` | Default `true`. | | `intent_scope` | Currently must be `"head"`. | @@ -195,10 +195,11 @@ The composite Action runs `deploybot react` from the checked-out default branch. | `timeout` | `"1800"` | Release-follow timeout in seconds. | The workflow needs `contents: write`, `pull-requests: write`, `checks: read`, -`issues: write`, and `actions: write`. Use the event filters and concurrency -group in [`examples/github-workflow.yml`](../examples/github-workflow.yml), pin -the Action to a reviewed full commit SHA, and keep its `workflow_run.workflows` -list aligned with `pipeline.ci_workflows`. +`issues: write`, and `actions: write`. Use the event filters, concurrency group, +and scheduled full-state reconciliation in +[`examples/github-workflow.yml`](../examples/github-workflow.yml), pin the Action +to a reviewed full commit SHA, and keep its `workflow_run.workflows` list aligned +with `pipeline.ci_workflows`. ## Authentication and environment diff --git a/examples/github-workflow.yml b/examples/github-workflow.yml index f3d4668..0c73f34 100644 --- a/examples/github-workflow.yml +++ b/examples/github-workflow.yml @@ -11,6 +11,10 @@ on: types: [completed] check_suite: types: [completed] + # GitHub concurrency coalesces pending event bursts. Reconcile the complete + # durable state periodically so the last cancelled event cannot strand work. + schedule: + - cron: "*/5 * * * *" workflow_dispatch: permissions: diff --git a/skills/deploybot/SKILL.md b/skills/deploybot/SKILL.md index b351676..caed755 100644 --- a/skills/deploybot/SKILL.md +++ b/skills/deploybot/SKILL.md @@ -27,6 +27,10 @@ exact heads, pending checks/reviews, blockers, dependencies, overlaps, exact is still waiting outside it. Never publish prompts, transcripts, source, or credentials to the thread registry. +Treat `active_intent_overlap_groups` as the pre-queue collision view. Treat a +`request-to-ready` alert as an ownership signal: report its exact active gate and +the source-thread action, rather than describing the empty queue as idle. + Use these state meanings: - `ready`: every configured merge gate passes. diff --git a/src/agent_merge_queue/cli.py b/src/agent_merge_queue/cli.py index 1ccbe2e..d559a5a 100755 --- a/src/agent_merge_queue/cli.py +++ b/src/agent_merge_queue/cli.py @@ -586,6 +586,59 @@ def split_blocked_entries( return eligible, blocked +def reason_requires_repair(reason: str) -> bool: + """Return whether a gate needs source-owner action instead of more time.""" + if reason == "pull request is draft": + return False + if reason == "GitHub reports the pull request merge state as DRAFT": + return False + if reason == "GitHub is still computing mergeability": + return False + if reason.endswith(" is not complete"): + return False + if "score is missing for the current head" in reason: + return False + if reason.endswith(" has not reviewed the current head"): + return False + if re.fullmatch(r"\d+/\d+ exact-head approvals complete", reason): + return False + return True + + +def deployment_repair_required(entry: "QueueEntry") -> bool: + if any(verdict.state == "blocked" for verdict in entry.review_verdicts): + return True + waiting_reasons = { + reason + for verdict in entry.review_verdicts + if verdict.state == "waiting" + for reason in verdict.reasons + } + waiting_reasons.update( + f"{name} is not complete" + for name, status in entry.checks.items() + if status not in {"failed", "passed"} + ) + transitional = { + "pull request is draft", + "GitHub reports the pull request merge state as DRAFT", + "GitHub is still computing mergeability", + *waiting_reasons, + } + return any(reason not in transitional for reason in entry.reasons or []) + + +def repair_marker_is_transitional(marker: dict[str, Any] | None) -> bool: + if not marker: + return False + reasons = [ + value.strip() + for value in str(marker.get("reason") or "").split(";") + if value.strip() + ] + return bool(reasons) and not any(reason_requires_repair(value) for value in reasons) + + @dataclass class QueueEntry: number: int @@ -861,11 +914,18 @@ def base_sha(self) -> str: ) return str(value["sha"]) - def workflow_runs(self) -> list[dict[str, Any]]: - return self._paged_object_items( - f"repos/{self.repository}/actions/runs?branch={self.config.base_branch}&per_page=100", - "workflow_runs", + def workflow_runs(self, *, limit: int | None = None) -> list[dict[str, Any]]: + endpoint = ( + f"repos/{self.repository}/actions/runs?" + f"branch={self.config.base_branch}&per_page=100" ) + if limit is None: + return self._paged_object_items(endpoint, "workflow_runs") + data = self._json("api", endpoint) + values = data.get("workflow_runs") if isinstance(data, dict) else None + if not isinstance(values, list): + raise QueueError(f"unexpected GitHub response for {endpoint}") + return [value for value in values if isinstance(value, dict)][:limit] def workflow_runs_for_branch(self, branch: str) -> list[dict[str, Any]]: return self._paged_object_items( @@ -873,6 +933,107 @@ def workflow_runs_for_branch(self, branch: str) -> list[dict[str, Any]]: "workflow_runs", ) + def workflow_runs_for_workflows( + self, + names: Iterable[str], + *, + limit: int = 100, + status: str | None = None, + since: datetime | None = None, + ) -> list[dict[str, Any]]: + configured = tuple(dict.fromkeys(names)) + if not configured: + return [] + values = self._json( + "workflow", + "list", + "--repo", + self.repository, + "--all", + "--limit", + "1000", + "--json", + "id,name,state", + ) + workflows = values if isinstance(values, list) else [] + runs: list[dict[str, Any]] = [] + for name in configured: + matches = [ + value + for value in workflows + if str(value.get("name") or "") == name + and str(value.get("state") or "") == "active" + ] + if len(matches) != 1: + raise QueueError( + f"configured workflow {name!r} did not resolve to one active workflow" + ) + workflow_id = int(matches[0]["id"]) + status_filter = f"&status={quote(status, safe='')}" if status else "" + page = 1 + workflow_runs: list[dict[str, Any]] = [] + while True: + endpoint = ( + f"repos/{self.repository}/actions/workflows/{workflow_id}/runs?" + f"branch={quote(self.config.base_branch, safe='')}{status_filter}&" + f"per_page={min(max(limit, 1), 100)}&page={page}" + ) + data = self._json("api", endpoint) + values = data.get("workflow_runs") if isinstance(data, dict) else None + if not isinstance(values, list): + raise QueueError(f"unexpected GitHub response for {endpoint}") + workflow_runs.extend( + value for value in values if isinstance(value, dict) + ) + oldest = min( + ( + parse_time( + str( + value.get("created_at") or value.get("updated_at") or "" + ) + ) + for value in values + if isinstance(value, dict) + ), + default=None, + ) + if ( + len(values) < min(max(limit, 1), 100) + or (since is not None and oldest is not None and oldest <= since) + or (since is None and len(workflow_runs) >= limit) + ): + break + page += 1 + runs.extend(workflow_runs) + ordered = sorted( + runs, + key=lambda value: str( + value.get("created_at") or value.get("updated_at") or "" + ), + reverse=True, + ) + if since is not None: + return [ + value + for value in ordered + if (parse_time(str(value.get("created_at") or "")) or since) >= since + ] + return ordered[:limit] + + def successful_workflow_runs( + self, + names: Iterable[str], + *, + limit: int = 100, + since: datetime | None = None, + ) -> list[dict[str, Any]]: + return self.workflow_runs_for_workflows( + names, + limit=limit, + status="success", + since=since, + ) + 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", @@ -1044,11 +1205,26 @@ def dispatch_deploy_workflows( return dispatched def recent_merged_pull_requests(self, limit: int) -> list[dict[str, Any]]: - values = self._paged_api( - "repos/" - f"{self.repository}/pulls?state=closed&sort=updated&direction=desc&per_page=100" - ) - return [value for value in values if value.get("merged_at")][:limit] + merged: list[dict[str, Any]] = [] + page = 1 + while len(merged) < limit: + endpoint = ( + "repos/" + f"{self.repository}/pulls?state=closed&sort=updated&direction=desc" + f"&per_page=100&page={page}" + ) + values = self._json("api", endpoint) + if not isinstance(values, list): + raise QueueError(f"unexpected GitHub response for {endpoint}") + merged.extend( + value + for value in values + if isinstance(value, dict) and value.get("merged_at") + ) + if len(values) < 100: + break + page += 1 + return merged[:limit] def is_ancestor(self, ancestor: str, descendant: str) -> bool: comparison = self._json( @@ -1316,6 +1492,59 @@ def comments(self, number: int) -> list[dict[str, Any]]: f"repos/{self.repository}/issues/{number}/comments?per_page=100" ) + def comments_for_pull_requests( + self, numbers: Iterable[int] + ) -> dict[int, list[dict[str, Any]]]: + selected = sorted({int(number) for number in numbers}) + if not selected: + return {} + selections = "\n".join( + f"""pr_{number}: pullRequest(number: {number}) {{ + comments(last: 100) {{ + pageInfo {{ hasPreviousPage }} + nodes {{ databaseId body createdAt author {{ login }} }} + }} + }}""" + for number in selected + ) + query = f"""query($owner: String!, $name: String!) {{ + repository(owner: $owner, name: $name) {{ + {selections} + }} + }}""" + data = self._json( + "api", + "graphql", + "-f", + f"query={query}", + "-F", + f"owner={self.owner}", + "-F", + f"name={self.name}", + ) + try: + repository = data["data"]["repository"] + except (KeyError, TypeError) as error: + raise QueueError("could not read pull-request comments") from error + result: dict[int, list[dict[str, Any]]] = {} + for number in selected: + connection = repository.get(f"pr_{number}") or {} + comments = connection.get("comments") or {} + if (comments.get("pageInfo") or {}).get("hasPreviousPage"): + result[number] = self.comments(number) + continue + result[number] = [ + { + "id": value.get("databaseId"), + "body": value.get("body"), + "created_at": value.get("createdAt"), + "user": {"login": (value.get("author") or {}).get("login")}, + } + for value in comments.get("nodes") or [] + if isinstance(value, dict) + ] + return result + def reviews(self, number: int) -> list[dict[str, Any]]: return self._paged_api( f"repos/{self.repository}/pulls/{number}/reviews?per_page=100" @@ -1784,6 +2013,37 @@ def command_inspect( def pipeline_status(client: GitHub) -> dict[str, Any]: queued = client.queue() queued_by_number = {entry.number: entry for entry in queued} + open_numbers = client.open_pull_request_numbers() + inspect_numbers = [ + number for number in open_numbers if number not in queued_by_number + ] + + def inspect( + number: int, + ) -> tuple[int, list[dict[str, Any]], QueueEntry]: + comments = client.comments(number) + return ( + number, + comments, + client.snapshot( + number, + require_marker=False, + allow_blocked_label=True, + known_comments=comments, + ), + ) + + inspected: dict[int, tuple[list[dict[str, Any]], QueueEntry]] = {} + if inspect_numbers: + with ThreadPoolExecutor( + max_workers=min( + max(client.config.pipeline.promotion_workers, 8), + len(inspect_numbers), + ) + ) as executor: + for number, comments, entry in executor.map(inspect, inspect_numbers): + inspected[number] = (comments, entry) + active_intents: list[tuple[QueueEntry, dict[str, Any], dict[str, Any]]] = [] stages: dict[str, list[dict[str, Any]]] = { "draft": [], "reviewing": [], @@ -1792,19 +2052,29 @@ def pipeline_status(client: GitHub) -> dict[str, Any]: "queued": [], "blocked": [], } - for number in client.open_pull_request_numbers(): + for number in open_numbers: if number in queued_by_number: value = entry_dict(queued_by_number[number]) value["pipeline_stage"] = "queued" stages["queued"].append(value) continue - entry = client.snapshot( - number, - require_marker=False, - allow_blocked_label=True, - ) + comments, entry = inspected[number] labels = set(entry.labels) - if client.config.pipeline.intent_label in labels: + intent = latest_intent(comments, client.trusted_logins) + active_intent = bool(intent and intent.get("state") == "requested") + stale_intent = bool( + active_intent + and client.config.pipeline.intent_scope == "head" + and intent.get("requested_head") != entry.head_sha + ) + if client.config.pipeline.intent_label in labels and ( + not active_intent + or stale_intent + or client.config.blocked_label in labels + or deployment_repair_required(entry) + ): + stage = "blocked" + elif client.config.pipeline.intent_label in labels: stage = "deploy_requested" elif entry.is_draft: stage = "draft" @@ -1816,16 +2086,64 @@ def pipeline_status(client: GitHub) -> dict[str, Any]: stage = "blocked" value = entry_dict(entry) value["pipeline_stage"] = stage + if client.config.pipeline.intent_label in labels: + value["deploy_intent"] = ( + { + "intent_id": intent.get("intent_id"), + "requested_at": intent.get("requested_at"), + "requested_head": intent.get("requested_head"), + "head_matches": not stale_intent, + "state": intent.get("state"), + } + if intent + else None + ) + if ( + client.config.pipeline.intent_label in labels + and active_intent + and intent is not None + ): + active_intents.append((entry, intent, value)) stages[stage].append(value) main_sha = client.base_sha() delivery = release_state( main_sha=main_sha, - runs=client.workflow_runs(), + runs=client.workflow_runs_for_workflows( + ( + *client.config.pipeline.ci_workflows, + *client.config.pipeline.deploy_workflows, + ), + limit=100, + ), config=client.config.pipeline, ) now = datetime.now(timezone.utc) alerts: list[dict[str, Any]] = [] queue_target = client.config.pipeline.ready_to_merge_target_minutes * 60 + for entry, intent, value in active_intents: + timestamp = parse_time(str(intent.get("requested_at") or "")) + elapsed = (now - timestamp).total_seconds() if timestamp else None + if elapsed is None or elapsed <= queue_target: + continue + requested_head = str(intent.get("requested_head") or "") + if requested_head and requested_head != entry.head_sha: + active_gate = "deploy intent is bound to an older head" + elif client.config.blocked_label in entry.labels: + active_gate = "repair is blocked; source thread must resume" + else: + active_gate = "; ".join(entry.reasons or []) or "promotion worker" + alerts.append( + { + "stage": "request-to-ready", + "pull_request": entry.number, + "elapsed_seconds": elapsed, + "target_seconds": queue_target, + "active_gate": active_gate, + "requested_head": requested_head or None, + "current_head": entry.head_sha, + "pipeline_stage": value["pipeline_stage"], + } + ) for entry in queued: timestamp = parse_time(entry.queued_at) elapsed = (now - timestamp).total_seconds() if timestamp else None @@ -1862,6 +2180,9 @@ def pipeline_status(client: GitHub) -> dict[str, Any]: "pull_requests": stages, "queue": [entry_dict(entry) for entry in queued], "overlap_groups": overlap_groups(queued), + "active_intent_overlap_groups": overlap_groups( + [entry for entry, _, _ in active_intents] + ), "release": delivery, "alerts": alerts, } @@ -1876,7 +2197,8 @@ def print_pipeline_status(value: dict[str, Any], *, json_output: bool) -> None: "threads: " f"{len(value['threads'])} active; " f"notifications: {len(value.get('notifications') or [])} pending; " - f"deploy requests: {len(stages['deploy_requested'])}; " + "deploy requests: " + f"{sum(1 for entries in stages.values() for entry in entries if entry.get('deploy_intent'))}; " f"queue: {len(value['queue'])}; " f"release: {value['release']['state']}" ) @@ -2275,6 +2597,32 @@ def reconcile_externally_merged_threads(client: GitHub) -> list[dict[str, Any]]: except QueueError: continue if not merge_sha: + try: + pull = client.pull_head(number) + except QueueError: + continue + if pull.get("state") != "CLOSED": + continue + updated_at = utc_now() + client.record_thread( + ThreadRecord( + provider=str(record["provider"]), + thread_id=str(record["thread_id"]), + phase="abandoned", + 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 = {"pull_request": number, "state": "abandoned"} + reconciled.append(value) + notify( + client.config.pipeline, + "deploy-abandoned", + {"repository": client.repository, **value}, + ) continue updated_at = utc_now() record_deployment_notification_obligation( @@ -2348,29 +2696,41 @@ def evaluate( client.config.pipeline.intent_scope == "head" and intent.get("requested_head") != entry.head_sha ): - return ( - "waiting", - { - "number": number, - "reasons": [ - "deploy intent is bound to an older head; the trusted source " - "agent must run deploybot refresh-request" - ], - }, - entry, + reason = ( + "deploy intent is bound to an older head; the trusted source agent " + "must run deploybot refresh-request after fresh gates pass" ) + record_repair(client, entry, intent, reason) + return "blocked", {"number": number, "reason": reason}, 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, + repair = latest_payload( + comments, + REPAIR_MARKER, + coordinator_logins(client), ) + if repair_marker_is_transitional(repair): + if deployment_repair_required(entry): + reason = "; ".join(entry.reasons or ["blocked"]) + record_repair(client, entry, intent, reason) + return "blocked", {"number": number, "reason": reason}, entry + client.remove_label(number, client.config.blocked_label) + entry.labels = [ + label + for label in entry.labels + if label != client.config.blocked_label + ] + else: + return ( + "waiting", + { + "number": number, + "reasons": [ + "repair is blocked; the trusted source agent must run " + "deploybot resume after fresh gates pass" + ], + }, + entry, + ) if entry.state == "ready": changed = queue_from_intent( client, @@ -2382,7 +2742,7 @@ def evaluate( if changed: return "promoted", {"number": number}, entry return "ready", None, entry - if entry.state == "blocked": + if deployment_repair_required(entry): reason = "; ".join(entry.reasons or ["blocked"]) record_repair(client, entry, intent, reason) return "blocked", {"number": number, "reason": reason}, entry @@ -4218,44 +4578,65 @@ def command_control(client: GitHub, *, state: str, reason: str | None) -> None: def delivery_metrics(client: GitHub, *, limit: int) -> dict[str, Any]: - deploy_runs = [ - run - for run in client.workflow_runs() - if str(run.get("name") or "") in set(client.config.pipeline.deploy_workflows) - and run.get("conclusion") == "success" + pulls = client.recent_merged_pull_requests(limit) + merged_times = [ + value + for value in (parse_time(str(pull.get("merged_at") or "")) for pull in pulls) + if value is not None ] - samples: list[dict[str, Any]] = [] - for pull in client.recent_merged_pull_requests(limit): + earliest_merge = min(merged_times, default=None) + deploy_runs = sorted( + client.successful_workflow_runs( + client.config.pipeline.deploy_workflows, + limit=max(100, limit), + since=earliest_merge, + ), + key=lambda item: str(item.get("updated_at") or ""), + ) + comments_by_number = client.comments_for_pull_requests( + int(pull["number"]) for pull in pulls + ) + + def sample(pull: dict[str, Any]) -> dict[str, Any]: number = int(pull["number"]) - comments = client.comments(number) + comments = comments_by_number.get(number, []) intent = latest_intent(comments, client.trusted_logins) marker = queue_marker_for_client(client, comments) merged_at = str(pull.get("merged_at") or "") or None merge_sha = str(pull.get("merge_commit_sha") or "") live_at: str | None = None if merge_sha: - for run in sorted( - deploy_runs, key=lambda item: str(item.get("updated_at") or "") - ): + merged_time = parse_time(merged_at) + for run in deploy_runs: + created_time = parse_time(str(run.get("created_at") or "")) deployed_sha = str(run.get("head_sha") or "") + if merged_time and created_time: + if created_time < merged_time: + continue if deployed_sha and client.is_ancestor(merge_sha, deployed_sha): live_at = str(run.get("updated_at") or "") or None break requested_at = str((intent or {}).get("requested_at") or "") or None queued_at = str((marker or {}).get("queued_at") or "") or None - samples.append( - { - "pull_request": number, - "requested_at": requested_at, - "queued_at": queued_at, - "merged_at": merged_at, - "live_at": live_at, - "request_to_queue_seconds": seconds_between(requested_at, queued_at), - "queue_to_merge_seconds": seconds_between(queued_at, merged_at), - "merge_to_live_seconds": seconds_between(merged_at, live_at), - "request_to_live_seconds": seconds_between(requested_at, live_at), - } - ) + return { + "pull_request": number, + "requested_at": requested_at, + "queued_at": queued_at, + "merged_at": merged_at, + "live_at": live_at, + "request_to_queue_seconds": seconds_between(requested_at, queued_at), + "queue_to_merge_seconds": seconds_between(queued_at, merged_at), + "merge_to_live_seconds": seconds_between(merged_at, live_at), + "request_to_live_seconds": seconds_between(requested_at, live_at), + } + + if not pulls: + samples: list[dict[str, Any]] = [] + else: + with ThreadPoolExecutor( + max_workers=min(client.config.pipeline.promotion_workers, len(pulls)) + ) as executor: + samples = list(executor.map(sample, pulls)) return summarize_metrics(samples) diff --git a/tests/test_cli.py b/tests/test_cli.py index 87d856a..7a0962b 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -4,6 +4,7 @@ import json import unittest from contextlib import redirect_stderr, redirect_stdout +from datetime import datetime, timezone from unittest.mock import Mock, call, patch from agent_merge_queue.cli import ( @@ -30,6 +31,8 @@ command_thread_acknowledge, command_unblock, completed_batch_ids, + delivery_metrics, + deployment_repair_required, entries_in_batch, effective_queue_marker, freeze_queue, @@ -42,6 +45,7 @@ near_ready_overlap_holds, new_batch, overlap_groups, + pipeline_status, promote_integrations, pull_request_feature_summary, queue_state_body, @@ -56,7 +60,7 @@ thread_notification_id, ) from agent_merge_queue.config import parse_config -from agent_merge_queue.records import integration_body, intent_body +from agent_merge_queue.records import integration_body, intent_body, repair_body from agent_merge_queue.reviews import ReviewVerdict @@ -178,6 +182,48 @@ def test_ensure_labels_exist_is_race_safe(self) -> None: "--force", ) + def test_comment_batch_normalizes_graphql_and_falls_back_when_truncated( + self, + ) -> None: + client = object.__new__(GitHub) + client.owner = "example" + client.name = "repo" + client._json = Mock( + return_value={ + "data": { + "repository": { + "pr_1": { + "comments": { + "pageInfo": {"hasPreviousPage": False}, + "nodes": [ + { + "databaseId": 11, + "body": "one", + "createdAt": "2026-06-20T00:00:00Z", + "author": {"login": "trusted"}, + } + ], + } + }, + "pr_2": { + "comments": { + "pageInfo": {"hasPreviousPage": True}, + "nodes": [], + } + }, + } + } + } + ) + client.comments = Mock(return_value=[{"id": 22, "body": "fallback"}]) + + result = client.comments_for_pull_requests([2, 1]) + + self.assertEqual(result[1][0]["user"]["login"], "trusted") + self.assertEqual(result[1][0]["created_at"], "2026-06-20T00:00:00Z") + self.assertEqual(result[2][0]["body"], "fallback") + client.comments.assert_called_once_with(2) + def test_status_is_a_read_only_pipeline_view(self) -> None: status = {"repository": "example/repo"} with ( @@ -248,6 +294,116 @@ def test_settle_window_only_collects_a_mixed_ready_burst(self) -> None: waiting.labels.append("merge-queue-blocked") self.assertFalse(should_settle_batch(client, [ready, waiting])) + def test_draft_with_pending_gates_waits_without_repair(self) -> None: + value = entry(1, state="blocked") + value.is_draft = True + value.checks = {"CI": "pending", "Optional": "failed"} + value.reasons = [ + "pull request is draft", + "GitHub reports the pull request merge state as DRAFT", + "CI is not complete", + "Greptile score is missing for the current head", + "Greptile has not reviewed the current head", + "0/1 exact-head approvals complete", + ] + value.review_verdicts = ( + ReviewVerdict( + "Greptile", + "waiting", + ( + "Greptile score is missing for the current head", + "Greptile has not reviewed the current head", + ), + ), + ReviewVerdict( + "GitHub approvals", + "waiting", + ("0/1 exact-head approvals complete",), + ), + ) + + self.assertFalse(deployment_repair_required(value)) + + value.reasons.append("CI failed") + self.assertTrue(deployment_repair_required(value)) + + def test_status_exposes_intent_head_overlap_and_request_delay(self) -> None: + draft = entry(1, "shared.py", state="blocked") + draft.labels = ["deploy-requested"] + draft.is_draft = True + draft.checks = {"CI": "pending"} + draft.reasons = ["pull request is draft", "CI is not complete"] + conflict = entry(2, "shared.py", state="blocked") + conflict.labels = ["deploy-requested"] + conflict.reasons = ["pull request conflicts with main"] + intents = { + number: { + "id": number, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "trusted"}, + "body": intent_body( + intent_id=f"intent-{number}", + state="requested", + requested_at="2026-06-20T00:00:00Z", + requested_head=value.head_sha, + ), + } + for number, value in ((1, draft), (2, conflict)) + } + client = Mock() + client.config = CONFIG + client.trusted_logins = {"trusted"} + client.queue.return_value = [] + inactive = entry(3, "shared.py") + inactive.labels = [] + intents[3] = { + "id": 3, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "trusted"}, + "body": intent_body( + intent_id="intent-3", + state="requested", + requested_at="2026-06-20T00:00:00Z", + requested_head=inactive.head_sha, + ), + } + client.open_pull_request_numbers.return_value = [1, 2, 3] + client.comments.side_effect = lambda number: [intents[number]] + client.snapshot.side_effect = lambda number, **_kwargs: { + 1: draft, + 2: conflict, + 3: inactive, + }[number] + client.base_sha.return_value = "f" * 40 + client.workflow_runs.return_value = [] + client.workflow_runs_for_workflows.return_value = [] + client.pipeline_control.return_value = {"state": "running"} + client.thread_records.return_value = [] + client.deployment_notifications.return_value = [] + + result = pipeline_status(client) + + self.assertEqual( + [value["number"] for value in result["pull_requests"]["deploy_requested"]], + [1], + ) + self.assertEqual( + [value["number"] for value in result["pull_requests"]["blocked"]], + [2], + ) + self.assertTrue( + result["pull_requests"]["deploy_requested"][0]["deploy_intent"][ + "head_matches" + ] + ) + self.assertEqual( + result["active_intent_overlap_groups"][0]["pull_requests"], [1, 2] + ) + self.assertEqual({alert["pull_request"] for alert in result["alerts"]}, {1, 2}) + self.assertTrue( + all(alert["stage"] == "request-to-ready" for alert in result["alerts"]) + ) + def test_overlap_mode_holds_only_ready_members_of_near_ready_components( self, ) -> None: @@ -317,6 +473,46 @@ def test_reconciles_an_externally_merged_requested_thread(self) -> None: self.assertEqual(obligation.state, "awaiting-verification") self.assertEqual(obligation.merge_sha, merge_sha) + def test_reconcile_marks_closed_unmerged_requested_thread_abandoned(self) -> None: + head_sha = "a" * 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, + ), + } + ] + client.externally_integrated_merge.return_value = None + client.pull_head.return_value = {"state": "CLOSED", "head_sha": head_sha} + + with patch("agent_merge_queue.cli.notify") as notify: + result = reconcile_externally_merged_threads(client) + + self.assertEqual(result, [{"pull_request": 42, "state": "abandoned"}]) + self.assertEqual(client.record_thread.call_args.args[0].phase, "abandoned") + notify.assert_called_once_with( + CONFIG.pipeline, + "deploy-abandoned", + {"repository": "example/repo", "pull_request": 42, "state": "abandoned"}, + ) + def test_verified_release_creates_native_thread_notification(self) -> None: sha = "a" * 40 release = { @@ -656,7 +852,6 @@ def test_pending_thread_notification_retries_with_stable_identity(self) -> None: self.assertNotIn("/new", first["thread_notifications"][0]["message"]) client.record_thread.assert_not_called() client.record_deployment_notification.assert_not_called() - client.is_ancestor.assert_not_called() def test_delivered_notification_is_not_reopened_by_release_follow(self) -> None: sha = "a" * 40 @@ -1301,6 +1496,96 @@ def test_workflow_history_uses_every_paginated_page(self) -> None: "repos/example/repo/actions/runs?branch=main&per_page=100", ) + def test_limited_workflow_history_uses_one_page(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client.config = CONFIG + client._json = Mock( + return_value={"workflow_runs": [{"id": number} for number in range(100)]} + ) + + result = client.workflow_runs(limit=25) + + self.assertEqual([run["id"] for run in result], list(range(25))) + self.assertNotIn("--paginate", client._json.call_args.args) + + def test_successful_workflow_history_filters_by_workflow(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client.config = CONFIG + client._json = Mock( + side_effect=[ + [{"id": 7, "name": "Deploy", "state": "active"}], + {"workflow_runs": [{"id": 9, "name": "Deploy"}]}, + ] + ) + + result = client.successful_workflow_runs(["Deploy"], limit=25) + + self.assertEqual(result, [{"id": 9, "name": "Deploy"}]) + self.assertIn( + "actions/workflows/7/runs?branch=main&status=success&per_page=25&page=1", + client._json.call_args.args[-1], + ) + + def test_workflow_histories_are_combined_by_recency_before_limit(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client.config = CONFIG + client._json = Mock( + side_effect=[ + [ + {"id": 7, "name": "Deploy A", "state": "active"}, + {"id": 8, "name": "Deploy B", "state": "active"}, + ], + {"workflow_runs": [{"id": 70, "created_at": "2026-06-20T00:00:00Z"}]}, + {"workflow_runs": [{"id": 80, "created_at": "2026-06-20T01:00:00Z"}]}, + ] + ) + + result = client.successful_workflow_runs(["Deploy A", "Deploy B"], limit=1) + + self.assertEqual([run["id"] for run in result], [80]) + + def test_workflow_history_pages_until_the_requested_time_window(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client.config = CONFIG + recent = [ + {"id": number, "created_at": "2026-06-20T02:00:00Z"} + for number in range(100) + ] + client._json = Mock( + side_effect=[ + [{"id": 7, "name": "Deploy", "state": "active"}], + {"workflow_runs": recent}, + {"workflow_runs": [{"id": 101, "created_at": "2026-06-20T00:00:00Z"}]}, + ] + ) + + result = client.successful_workflow_runs( + ["Deploy"], + since=datetime(2026, 6, 20, 1, tzinfo=timezone.utc), + ) + + self.assertEqual(len(result), 100) + self.assertIn("page=2", client._json.call_args.args[-1]) + + def test_recent_merged_pulls_stop_after_enough_results(self) -> None: + client = object.__new__(GitHub) + client.repository = "example/repo" + client._json = Mock( + return_value=[ + {"number": number, "merged_at": "2026-06-20T00:00:00Z"} + for number in range(1, 31) + ] + ) + + result = client.recent_merged_pull_requests(25) + + self.assertEqual([pull["number"] for pull in result], list(range(1, 26))) + client._json.assert_called_once() + def test_integration_status_requires_every_source_authorization(self) -> None: client = object.__new__(GitHub) client.source_deploy_authorized = Mock(return_value=True) @@ -2056,18 +2341,21 @@ def test_promote_queues_ready_intent_but_leaves_waiting_visible(self) -> None: ready.labels = ["deploy-requested"] waiting = entry(2, state="waiting") waiting.labels = ["deploy-requested"] + waiting.checks = {"CI": "pending"} waiting.reasons = ["CI is not complete"] - intent = intent_body( - intent_id="intent-1", - state="requested", - requested_at="2026-06-20T00:00:00Z", - requested_head=ready.head_sha, - ) - intent_comment = { - "id": 1, - "created_at": "2026-06-20T00:00:00Z", - "user": {"login": "trusted"}, - "body": intent, + intent_comments = { + value.number: { + "id": value.number, + "created_at": "2026-06-20T00:00:00Z", + "user": {"login": "trusted"}, + "body": intent_body( + intent_id=f"intent-{value.number}", + state="requested", + requested_at="2026-06-20T00:00:00Z", + requested_head=value.head_sha, + ), + } + for value in (ready, waiting) } client = Mock() client.config = CONFIG @@ -2076,7 +2364,7 @@ def test_promote_queues_ready_intent_but_leaves_waiting_visible(self) -> None: client.coordinator_logins = {"coordinator"} client.intent_numbers.return_value = [1, 2] client.active_integration_sources.return_value = set() - client.comments.return_value = [intent_comment] + client.comments.side_effect = lambda number: [intent_comments[number]] client.snapshot.side_effect = [ready, waiting] client.labels.return_value = {"deploy-requested"} with redirect_stdout(io.StringIO()): @@ -2117,6 +2405,88 @@ def test_promote_never_auto_resumes_a_repair_block(self) -> None: self.assertIn("deploybot resume", result["waiting"][0]["reasons"][0]) client.add_label.assert_not_called() + def test_promote_clears_a_transitional_draft_block(self) -> None: + ready = entry(1) + ready.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=ready.head_sha, + ), + } + repair_comment = { + "id": 2, + "created_at": "2026-06-20T00:01:00Z", + "user": {"login": "coordinator"}, + "body": repair_body( + { + "head_sha": ready.head_sha, + "reason": ( + "pull request is draft; CI is not complete; " + "Greptile score is missing for the current head" + ), + } + ), + } + 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, repair_comment] + client.snapshot.return_value = ready + + with redirect_stdout(io.StringIO()): + result = command_promote(client) + + self.assertEqual(result["promoted"], [1]) + client.remove_label.assert_called_with(1, "merge-queue-blocked") + client.add_label.assert_called_with(1, "merge-queue") + + def test_promote_records_stale_intent_as_thread_repair(self) -> None: + value = entry(1) + value.labels = ["deploy-requested"] + 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="a" * 40, + provider="codex", + thread_id="thread-1", + ), + } + 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 = value + client.labels.return_value = {"deploy-requested"} + client.base_sha.return_value = "b" * 40 + + with redirect_stdout(io.StringIO()): + result = command_promote(client) + + self.assertEqual(result["promoted"], []) + self.assertEqual(result["blocked"][0]["number"], 1) + self.assertIn("older head", result["blocked"][0]["reason"]) + client.add_label.assert_called_with(1, "merge-queue-blocked") + self.assertEqual(client.record_thread.call_args.args[0].phase, "blocked") + def test_resume_atomically_requeues_repaired_exact_head(self) -> None: value = entry(1) value.labels = ["deploy-requested", "merge-queue-blocked"] @@ -2214,6 +2584,51 @@ def add_comment(_: int, body: str): ) ) + def test_metrics_skip_deployments_that_finished_before_the_merge(self) -> None: + merge_sha = "m" * 40 + deployed_sha = "d" * 40 + client = Mock() + client.config = CONFIG + client.trusted_logins = {"trusted"} + client.successful_workflow_runs.return_value = [ + { + "name": "Deploy", + "conclusion": "success", + "head_sha": "o" * 40, + "created_at": "2026-06-20T00:20:00Z", + "updated_at": "2026-06-20T00:30:00Z", + }, + { + "name": "Deploy", + "conclusion": "success", + "head_sha": deployed_sha, + "created_at": "2026-06-20T01:01:00Z", + "updated_at": "2026-06-20T01:05:00Z", + }, + ] + client.recent_merged_pull_requests.return_value = [ + { + "number": 42, + "merged_at": "2026-06-20T01:00:00Z", + "merge_commit_sha": merge_sha, + } + ] + client.comments_for_pull_requests.return_value = {42: []} + client.is_ancestor.return_value = True + + result = delivery_metrics(client, limit=25) + + self.assertEqual(result["sample_count"], 1) + self.assertEqual(result["samples"][0]["live_at"], "2026-06-20T01:05:00Z") + client.is_ancestor.assert_called_once_with(merge_sha, deployed_sha) + call_args = client.successful_workflow_runs.call_args + self.assertEqual(call_args.args, (CONFIG.pipeline.deploy_workflows,)) + self.assertEqual(call_args.kwargs["limit"], 100) + self.assertEqual( + call_args.kwargs["since"], + datetime(2026, 6, 20, 1, tzinfo=timezone.utc), + ) + def test_integration_pr_contains_every_frozen_head(self) -> None: first = entry(1, "shared.py") second = entry(2, "shared.py") diff --git a/tests/test_skill.py b/tests/test_skill.py index 5e882cc..4e373ca 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -63,6 +63,8 @@ def test_github_workflow_wakes_after_named_ci_finishes(self) -> None: encoding="utf-8" ) self.assertIn("workflow_run:", workflow) + self.assertIn("schedule:", workflow) + self.assertIn('cron: "*/5 * * * *"', workflow) self.assertIn("workflows: [CI]", workflow) self.assertIn("github.event.repository.default_branch", workflow) self.assertIn(