diff --git a/docker/runners/runner-image.lock.json b/docker/runners/runner-image.lock.json index 5c9bce7597..282bbe00c4 100644 --- a/docker/runners/runner-image.lock.json +++ b/docker/runners/runner-image.lock.json @@ -1,12 +1,12 @@ { "base_image_digest": "sha256:3ba65aa20f86a0fad9df2b2c259c613df006b2e6d0bfcc8a146afb8c525a9751", "gh_version": "2.67.0", - "identity_digest": "ed01484a7f156587d67854428f1f1161", + "identity_digest": "00e1e1b5419d2c70d66e048c4cd235de", "image_version": 7, "kubectl_version": "1.32.1", "python_version": "3.12", "runner_version": "2.336.0", - "shared_env_digest": "5bf234dcdf3a0ec86c927d09", + "shared_env_digest": "ab7726c96d67995001339ef7", "shared_env_install_args": "--frozen --all-extras --all-groups --no-install-project", "uv_version": "0.6.14" } diff --git a/docs/runbooks/bulk-pr-operations.md b/docs/runbooks/bulk-pr-operations.md new file mode 100644 index 0000000000..20df190e39 --- /dev/null +++ b/docs/runbooks/bulk-pr-operations.md @@ -0,0 +1,129 @@ +# Bulk PR operations — mandatory throttled path (OMN-16284) + +## Why this exists + +At ~01:45Z on 2026-08-20 a merge-sweep lane armed/update-branched ~108 PRs +in one unthrottled burst. Update-branching triggers a full fresh +check-suite per PR; `onex_change_control` PRs alone carry ~63 checks each +(~19 needing `[self-hosted, omnibase-ci]` runners). Result: OCC's queued-run +count grew to ~1065, the shared org-level runner pool (88 runners, +`visibility=all`, no per-repo fair-share) sat 77-88/88 busy for ~4 hours, and +every landing chain org-wide (core, OCC, infra, omnimarket) starved. The +burst then settled ~95-100% RED from incident-window transients, producing +exactly **1 merge out of ~108**. + +This is distinct from OMN-16171 (self-amplified bot-edit re-triggering — +a debounce fix on the *reactive* side). This runbook is the *dispatch* +side: nothing mechanically capped how many check-suites a bulk operation +could enqueue at once. The rule ("throttle, serialize heavy work") existed +only in prose and did not bind — same failure class documented in memory +`feedback_a_rule_is_not_a_mechanism`. + +## The mandatory path + +**Every supported bulk PR operation — update-branch, arming/auto-merge sweeps, +and mass reruns — routes through +`scripts/ci/bulk_pr_throttle.py`.** Do not hand-roll a `for pr in $prs; do gh +... ; done` loop for more than a small handful of PRs. That loop is exactly +the failure mode this ticket exists to close. + +```bash +# Plan only — no gh calls, no mutation: +uv run scripts/ci/bulk_pr_throttle.py \ + --owner OmniNode-ai --repo onex_change_control \ + --prs 6751,6752,6753,6754,6755 \ + --operation rerun-failed \ + --dry-run + +# Real run — processes in wave-size-10 batches by default, blocking before +# each wave while repos///actions/runs?status=queued exceeds 150: +uv run scripts/ci/bulk_pr_throttle.py \ + --owner OmniNode-ai --repo onex_change_control \ + --prs 6751,6752,6753,6754,6755 \ + --operation rerun-failed +``` + +### Operations supported + +| `--operation` | What it does per PR | +| --- | --- | +| `update-branch` | `PUT /repos/{owner}/{repo}/pulls/{pr}/update-branch` — triggers a fresh check-suite. The single highest-cost operation; this is what caused the incident. | +| `arm-automerge` | `gh pr merge --squash --auto` | +| `rerun-failed` | Resolves the PR's head SHA, lists completed workflow runs, and reruns failed jobs for runs with `conclusion` equal to `failure` or `cancelled`. | +| `noop-dry-run` | No-op outcome, always succeeds — for exercising wave/threshold logic without touching `gh` at all. | + +### Flags + +| Flag | Default | Notes | +| --- | --- | --- | +| `--owner` / `--repo` | **required, no default** | Fail-fast: never silently assume an org/repo. | +| `--prs` | **required** | Comma-separated PR numbers. | +| `--operation` | **required** | One of the four above. | +| `--wave-size` | 10 | Flag-overridable up to a **hard ceiling of 25** (`MAX_WAVE_SIZE` in the script) — not overridable past that point by any flag. | +| `--queue-depth-threshold` | 150 | Queued-run count above which the tool blocks before dispatching the next wave. | +| `--max-total-prs` | 50 | Batches larger than this are **refused** unless you pass `--max-total-prs ` explicitly, raising the cap for that invocation. There is no flag that skips this check entirely — only one that raises the number, so oversized batches are always a stated, visible choice. | +| `--poll-seconds` | 30 | How often to re-poll queue depth while blocked. | +| `--max-wait-seconds` | 1800 | If the queue never drops below threshold within this window, the tool **raises and stops** — it does not silently proceed into a saturated fleet. | +| `--dry-run` | off | Prints the wave plan; makes zero `gh` calls. | +| `--receipt` | `.onex_state/bulk-pr-throttle/--.json` | Where the JSON wave receipt is written. | + +## The mechanical guard + +Per CLAUDE.md rule 5 ("detection without enforcement gets ignored"), the +throttle is not advisory prose — it is load-bearing in the tool itself: + +- The queue-depth gate (`wait_for_queue_depth` in `bulk_pr_throttle.py`) has + **no bypass parameter** anywhere in the module or its CLI — no `--force`, + no `--skip-throttle`, no `--ignore-threshold`. A caller cannot opt out of + throttling short of not using this tool at all. +- `--wave-size` is capped at `MAX_WAVE_SIZE` (25) in code, not just by + default — passing a larger value is a hard refusal + (`ValueError`/exit code 1), not a warning. +- Processing more than `--max-total-prs` PRs without explicitly raising the + cap is a hard refusal before any `gh` call is made (`TotalPrLimitExceededError`). +- A queue that never drains within `--max-wait-seconds` is a hard refusal + (`QueueDepthTimeoutError`), not a silently-skipped wave. + +**What is NOT yet wired (controller follow-up, out of scope for this PR):** +a CLAUDE.md pointer to this runbook, and/or a pre-flight lint in the sweep +skills that rejects a raw multi-PR `gh` loop outside this tool. This PR +cannot edit `omni_home/CLAUDE.md` (it lives outside any product-repo +worktree; see CLAUDE.md rule 9's `omni_home`-itself exception) — that +doctrine-wiring step is the controller's follow-up, tracked against this +runbook. + +## Wave receipts + +Every non-dry-run invocation writes a JSON receipt (default under +`.onex_state/bulk-pr-throttle/`) recording, per wave: wave index, the exact +PR numbers, queue depth before and after, start/completion timestamps, and +the per-PR outcome (`success`, `detail`). Wave progress is also logged to +stdout as it happens: + +```text +[bulk-pr-throttle] wave 1/1: depth_before=5 count=3 prs=[6751, 6752, 6753] operation=rerun-failed +[bulk-pr-throttle] pr=6751 success=True detail=reran=[98765432] +[bulk-pr-throttle] pr=6752 success=True detail=no completed failed or cancelled runs to rerun +[bulk-pr-throttle] pr=6753 success=True detail=reran=[98765440] +[bulk-pr-throttle] wave 1/1: depth_after=6 +``` + +## Coordination with other lanes + +Bulk operations touch shared fleet capacity. Before running a batch, +claim the specific PR numbers you intend to touch in +`docs/tracking/ROLLING_WORK_LEDGER.md` (§1a) so a concurrent lane's +residual-remediation batch doesn't collide with yours on the same PRs. + +## Tests + +`scripts/ci/tests/test_bulk_pr_throttle.py` covers wave partitioning +(including the hard ceiling), threshold blocking (mocked queue-depth +callable, including a mid-batch block on a later wave), dry-run plan output +(zero `gh` calls), all refusal paths (total-PR cap, unknown operation, empty +input, missing owner/repo), the `gh` CLI integration seam (mocked +`_run_gh`, never the real API), and the CLI entrypoint end to end. + +```bash +uv run pytest scripts/ci/tests/test_bulk_pr_throttle.py -v +``` diff --git a/pyproject.toml b/pyproject.toml index 703b9238a1..780eecf050 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "omnibase_infra" -version = "0.38.9" +version = "0.38.10" description = "ONEX Infrastructure - Service integration and database infrastructure tools" authors = [{ name = "OmniNode.ai", email = "contact@omninode.ai" }] license = {text = "MIT"} diff --git a/scripts/ci/bulk_pr_throttle.py b/scripts/ci/bulk_pr_throttle.py new file mode 100644 index 0000000000..7ab141dec9 --- /dev/null +++ b/scripts/ci/bulk_pr_throttle.py @@ -0,0 +1,736 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Mechanical dispatch throttle for bulk PR operations (OMN-16284). + +Why this exists +---------------- +At ~01:45Z on 2026-08-20 a merge-sweep lane armed/update-branched ~108 PRs +in one unthrottled burst. Update-branching triggers a full fresh check-suite +per PR; onex_change_control PRs alone carry ~63 checks each (~19 needing +self-hosted ``omnibase-ci`` runners). The queued-run count grew to ~1065, +the shared org-level runner pool (88 runners, no per-repo fair-share) sat +77-88/88 busy for ~4 hours, and every landing chain org-wide starved. The +burst settled ~95-100% RED from incident-window transients: 1 merge out of +~108. The rule ("throttle, serialize heavy work") existed only in prose and +did not bind — same failure class as ``feedback_a_rule_is_not_a_mechanism``. + +This script is the mechanism. ALL bulk PR operations (update-branch, +arm-automerge, mass reruns, mass body edits) should route through it rather +than a hand-rolled loop over ``gh``. + +Mechanical guard (no bypass) +----------------------------- +The queue-depth gate (:func:`wait_for_queue_depth`) has no force/skip/bypass +parameter anywhere in this module or its CLI. A caller cannot opt out of +throttling short of not using this tool at all — see +``docs/runbooks/bulk-pr-operations.md`` for the doctrine-wiring follow-up +that makes *not using it* visible. + +Usage +----- + bulk_pr_throttle.py --owner OmniNode-ai --repo onex_change_control \\ + --prs 6751,6752,6753 --operation rerun-failed --dry-run + + bulk_pr_throttle.py --owner OmniNode-ai --repo onex_change_control \\ + --prs 6751,6752,6753 --operation rerun-failed \\ + --wave-size 5 --queue-depth-threshold 150 + +Exit codes: 0 = all waves completed with all PR operations succeeding +(or a dry-run plan was printed), 1 = refused (bad input / cap exceeded / +queue-depth timeout) or at least one PR operation failed. +""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +import time +from collections.abc import Callable, Sequence +from dataclasses import asdict, dataclass +from datetime import UTC, datetime +from pathlib import Path + +# --------------------------------------------------------------------------- +# Tunables. Every numeric default here is deliberately conservative — the +# 2026-08-20 incident was caused by the ABSENCE of these defaults, not by +# them being set wrong. +# --------------------------------------------------------------------------- + +DEFAULT_WAVE_SIZE = 10 +#: Hard ceiling on --wave-size. Not flag-overridable past this point — the +#: mechanism must not be defeatable by simply passing a huge wave size. +MAX_WAVE_SIZE = 25 +DEFAULT_QUEUE_DEPTH_THRESHOLD = 150 +#: Default cap on total PRs processed in one invocation. Exceeding this +#: without explicitly passing --max-total-prs is refused (rule: "no silent +#: defaults", applied to batch size rather than owner/repo). +DEFAULT_MAX_TOTAL_PRS = 50 +DEFAULT_BLOCK_POLL_SECONDS = 30.0 +#: Maximum total time this tool will block waiting for queue depth to drop +#: before a single wave. A gate that can block forever is indistinguishable +#: from a hang; this raises QueueDepthTimeoutError instead of proceeding. +DEFAULT_MAX_BLOCK_SECONDS = 1800.0 +DEFAULT_GH_TIMEOUT_SECONDS = 120.0 + +VALID_OPERATIONS = ( + "update-branch", + "arm-automerge", + "rerun-failed", + "noop-dry-run", +) + + +class BulkPrThrottleError(RuntimeError): + """Base class for refusal / fatal errors raised by this tool.""" + + +class TotalPrLimitExceededError(BulkPrThrottleError): + """Raised when the PR batch exceeds the cap without an explicit override.""" + + +class QueueDepthTimeoutError(BulkPrThrottleError): + """Raised when queue depth stays above threshold past max_wait_seconds.""" + + +@dataclass(frozen=True) +class PrOutcome: + pr_number: int + success: bool + detail: str + + +@dataclass(frozen=True) +class WaveReceipt: + wave_index: int + pr_numbers: tuple[int, ...] + operation: str + dry_run: bool + queue_depth_before: int + queue_depth_after: int | None + started_at: str + completed_at: str + outcomes: tuple[PrOutcome, ...] + + +@dataclass(frozen=True) +class BulkRunReport: + owner: str + repo: str + operation: str + wave_size: int + queue_depth_threshold: int + dry_run: bool + waves: tuple[WaveReceipt, ...] + + +class PartialBulkRunError(BulkPrThrottleError): + """Raised when a bulk run aborts after one or more receiptable waves.""" + + def __init__(self, message: str, report: BulkRunReport) -> None: + super().__init__(message) + self.report = report + + +# --------------------------------------------------------------------------- +# Wave partitioning +# --------------------------------------------------------------------------- + + +def partition_into_waves( + pr_numbers: Sequence[int], wave_size: int +) -> list[tuple[int, ...]]: + """Split ``pr_numbers`` into consecutive waves of at most ``wave_size``.""" + if wave_size < 1: + raise ValueError(f"wave_size must be >= 1, got {wave_size}") + if wave_size > MAX_WAVE_SIZE: + raise ValueError( + f"wave_size {wave_size} exceeds the hard ceiling of {MAX_WAVE_SIZE} " + "— the ceiling exists so a bulk operation can never dispatch an " + "unthrottled burst regardless of flags" + ) + items = list(pr_numbers) + return [tuple(items[i : i + wave_size]) for i in range(0, len(items), wave_size)] + + +# --------------------------------------------------------------------------- +# Refusal: total PR cap +# --------------------------------------------------------------------------- + + +def validate_total_prs( + pr_numbers: Sequence[int], + *, + max_total_prs: int, + explicit_max_total_prs: bool, +) -> None: + """Refuse to process more than ``max_total_prs`` PRs. + + ``explicit_max_total_prs`` changes the refusal wording only. It is not a + bypass; a caller can raise the cap explicitly, but the raised cap remains a + real ceiling. + """ + if len(pr_numbers) <= max_total_prs: + return + if explicit_max_total_prs: + raise TotalPrLimitExceededError( + f"refusing to process {len(pr_numbers)} PRs against the explicit " + f"cap of {max_total_prs}. Pass --max-total-prs {len(pr_numbers)} " + "(or higher) explicitly if this larger batch is intentional." + ) + raise TotalPrLimitExceededError( + f"refusing to process {len(pr_numbers)} PRs against the default " + f"cap of {max_total_prs}. Pass --max-total-prs {len(pr_numbers)} " + "(or higher) explicitly to raise the cap — this is a fail-fast " + "guard against an accidental unthrottled burst, not a hard limit." + ) + + +# --------------------------------------------------------------------------- +# Threshold blocking (the queue-depth gate — no bypass parameter exists) +# --------------------------------------------------------------------------- + + +def wait_for_queue_depth( + *, + get_queue_depth: Callable[[], int], + threshold: int, + poll_seconds: float, + max_wait_seconds: float, + sleep_fn: Callable[[float], None] = time.sleep, + log: Callable[[str], None] = lambda msg: None, +) -> int: + """Block, polling ``get_queue_depth``, until depth <= threshold. + + Raises :class:`QueueDepthTimeoutError` rather than proceeding if the + queue never drops within ``max_wait_seconds`` — a persistently saturated + fleet is exactly the condition this tool must refuse to dispatch into. + """ + if poll_seconds <= 0: + raise ValueError(f"poll_seconds must be > 0, got {poll_seconds}") + waited = 0.0 + depth = get_queue_depth() + while depth > threshold: + if waited >= max_wait_seconds: + raise QueueDepthTimeoutError( + f"queue depth {depth} still above threshold {threshold} after " + f"waiting {waited:.0f}s (max_wait_seconds={max_wait_seconds}) " + "— aborting rather than dispatching into a saturated fleet" + ) + log( + f"queue depth {depth} > threshold {threshold}; " + f"blocking {poll_seconds:.0f}s before re-poll" + ) + sleep_fn(poll_seconds) + waited += poll_seconds + depth = get_queue_depth() + return depth + + +# --------------------------------------------------------------------------- +# Core wave-gated run +# --------------------------------------------------------------------------- + + +def run_bulk_operation( + *, + owner: str, + repo: str, + pr_numbers: Sequence[int], + operation: str, + wave_size: int = DEFAULT_WAVE_SIZE, + queue_depth_threshold: int = DEFAULT_QUEUE_DEPTH_THRESHOLD, + max_total_prs: int = DEFAULT_MAX_TOTAL_PRS, + explicit_max_total_prs: bool = False, + dry_run: bool = False, + get_queue_depth: Callable[[], int] | None = None, + apply_pr_operation: Callable[[str, str, int, str], PrOutcome] | None = None, + poll_seconds: float = DEFAULT_BLOCK_POLL_SECONDS, + max_wait_seconds: float = DEFAULT_MAX_BLOCK_SECONDS, + sleep_fn: Callable[[float], None] = time.sleep, + now_fn: Callable[[], datetime] = lambda: datetime.now(UTC), + log: Callable[[str], None] = print, +) -> BulkRunReport: + """Process ``pr_numbers`` through ``operation`` in bounded, queue-gated waves.""" + if not owner: + raise BulkPrThrottleError( + "owner must be explicitly provided (no silent default)" + ) + if not repo: + raise BulkPrThrottleError( + "repo must be explicitly provided (no silent default)" + ) + if operation not in VALID_OPERATIONS: + raise BulkPrThrottleError( + f"unknown operation {operation!r}; must be one of {VALID_OPERATIONS}" + ) + if not pr_numbers: + raise BulkPrThrottleError("pr_numbers must be non-empty") + + validate_total_prs( + pr_numbers, + max_total_prs=max_total_prs, + explicit_max_total_prs=explicit_max_total_prs, + ) + waves = partition_into_waves(list(pr_numbers), wave_size) + + if dry_run: + log( + f"[bulk-pr-throttle] DRY-RUN plan: {len(pr_numbers)} PR(s) across " + f"{len(waves)} wave(s) of <= {wave_size}, operation={operation}, " + f"owner={owner}, repo={repo}, queue_depth_threshold={queue_depth_threshold}" + ) + for idx, wave in enumerate(waves, start=1): + log(f"[bulk-pr-throttle] wave {idx}: {list(wave)}") + ts = now_fn().isoformat() + return BulkRunReport( + owner=owner, + repo=repo, + operation=operation, + wave_size=wave_size, + queue_depth_threshold=queue_depth_threshold, + dry_run=True, + waves=tuple( + WaveReceipt( + wave_index=idx, + pr_numbers=wave, + operation=operation, + dry_run=True, + queue_depth_before=-1, + queue_depth_after=-1, + started_at=ts, + completed_at=ts, + outcomes=(), + ) + for idx, wave in enumerate(waves, start=1) + ), + ) + + if get_queue_depth is None or apply_pr_operation is None: + raise BulkPrThrottleError( + "get_queue_depth and apply_pr_operation are required outside dry-run mode" + ) + + wave_receipts: list[WaveReceipt] = [] + + def partial_report() -> BulkRunReport: + return BulkRunReport( + owner=owner, + repo=repo, + operation=operation, + wave_size=wave_size, + queue_depth_threshold=queue_depth_threshold, + dry_run=False, + waves=tuple(wave_receipts), + ) + + for idx, wave in enumerate(waves, start=1): + started_at = now_fn().isoformat() + try: + depth_before = wait_for_queue_depth( + get_queue_depth=get_queue_depth, + threshold=queue_depth_threshold, + poll_seconds=poll_seconds, + max_wait_seconds=max_wait_seconds, + sleep_fn=sleep_fn, + log=log, + ) + except BulkPrThrottleError as exc: + if wave_receipts: + raise PartialBulkRunError(str(exc), partial_report()) from exc + raise + log( + f"[bulk-pr-throttle] wave {idx}/{len(waves)}: depth_before={depth_before} " + f"count={len(wave)} prs={list(wave)} operation={operation}" + ) + outcomes = tuple(apply_pr_operation(owner, repo, pr, operation) for pr in wave) + for outcome in outcomes: + log( + f"[bulk-pr-throttle] pr={outcome.pr_number} " + f"success={outcome.success} detail={outcome.detail}" + ) + completed_at = now_fn().isoformat() + try: + depth_after = get_queue_depth() + except BulkPrThrottleError as exc: + wave_receipts.append( + WaveReceipt( + wave_index=idx, + pr_numbers=wave, + operation=operation, + dry_run=False, + queue_depth_before=depth_before, + queue_depth_after=None, + started_at=started_at, + completed_at=completed_at, + outcomes=outcomes, + ) + ) + raise PartialBulkRunError(str(exc), partial_report()) from exc + log(f"[bulk-pr-throttle] wave {idx}/{len(waves)}: depth_after={depth_after}") + wave_receipts.append( + WaveReceipt( + wave_index=idx, + pr_numbers=wave, + operation=operation, + dry_run=False, + queue_depth_before=depth_before, + queue_depth_after=depth_after, + started_at=started_at, + completed_at=completed_at, + outcomes=outcomes, + ) + ) + + return BulkRunReport( + owner=owner, + repo=repo, + operation=operation, + wave_size=wave_size, + queue_depth_threshold=queue_depth_threshold, + dry_run=False, + waves=tuple(wave_receipts), + ) + + +def write_receipt(report: BulkRunReport, path: Path) -> None: + """Serialize ``report`` as JSON to ``path`` (parent dirs created as needed).""" + payload = { + "owner": report.owner, + "repo": report.repo, + "operation": report.operation, + "wave_size": report.wave_size, + "queue_depth_threshold": report.queue_depth_threshold, + "dry_run": report.dry_run, + "waves": [ + { + "wave_index": wave.wave_index, + "pr_numbers": list(wave.pr_numbers), + "pr_count": len(wave.pr_numbers), + "operation": wave.operation, + "dry_run": wave.dry_run, + "queue_depth_before": wave.queue_depth_before, + "queue_depth_after": wave.queue_depth_after, + "started_at": wave.started_at, + "completed_at": wave.completed_at, + "outcomes": [asdict(o) for o in wave.outcomes], + } + for wave in report.waves + ], + } + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + + +# --------------------------------------------------------------------------- +# gh CLI integration (production seam — tests monkeypatch _run_gh, never hit +# the real GitHub API) +# --------------------------------------------------------------------------- + + +def _run_gh(args: list[str]) -> subprocess.CompletedProcess[str]: + try: + return subprocess.run( # nosec B603 - fixed argv, no shell, trusted gh binary + ["gh", *args], + capture_output=True, + text=True, + check=False, + timeout=DEFAULT_GH_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired as exc: + return subprocess.CompletedProcess( + args=["gh", *args], + returncode=124, + stdout=exc.stdout or "", + stderr=f"gh command timed out after {DEFAULT_GH_TIMEOUT_SECONDS:.0f}s", + ) + + +def gh_queue_depth(owner: str, repo: str) -> int: + """Live queued-run count for ``owner/repo`` via the gh CLI.""" + result = _run_gh( + [ + "api", + f"repos/{owner}/{repo}/actions/runs?status=queued", + "--jq", + ".total_count", + ] + ) + if result.returncode != 0: + raise BulkPrThrottleError( + f"gh api queued-run count failed: {result.stderr.strip()}" + ) + return int(result.stdout.strip()) + + +def gh_apply_pr_operation( + owner: str, repo: str, pr_number: int, operation: str +) -> PrOutcome: + """Dispatch one PR-scoped operation via the gh CLI.""" + if operation == "update-branch": + result = _run_gh( + [ + "api", + "-X", + "PUT", + f"repos/{owner}/{repo}/pulls/{pr_number}/update-branch", + ] + ) + return PrOutcome( + pr_number=pr_number, + success=result.returncode == 0, + detail=(result.stdout or result.stderr).strip(), + ) + if operation == "arm-automerge": + result = _run_gh( + [ + "pr", + "merge", + str(pr_number), + "--repo", + f"{owner}/{repo}", + "--squash", + "--auto", + ] + ) + return PrOutcome( + pr_number=pr_number, + success=result.returncode == 0, + detail=(result.stdout or result.stderr).strip(), + ) + if operation == "rerun-failed": + return _gh_rerun_failed(owner, repo, pr_number) + if operation == "noop-dry-run": + return PrOutcome(pr_number=pr_number, success=True, detail="noop") + raise BulkPrThrottleError(f"unknown operation {operation!r}") + + +def _gh_rerun_failed(owner: str, repo: str, pr_number: int) -> PrOutcome: + """Find the PR's head-SHA workflow runs and rerun failed terminal runs. + + GitHub Actions exposes runner-loss / timeout incident-window reds as both + ``failure`` and ``cancelled`` workflow-run conclusions. The operation name + intentionally stays ``rerun-failed`` because GitHub's endpoint is + ``rerun-failed-jobs``; the selector includes terminal cancelled runs so the + throttle can clear the exact cancellation-heavy incident class it was built + to control. + """ + head_result = _run_gh( + [ + "pr", + "view", + str(pr_number), + "--repo", + f"{owner}/{repo}", + "--json", + "headRefOid", + ] + ) + if head_result.returncode != 0: + return PrOutcome( + pr_number=pr_number, + success=False, + detail=f"pr view failed: {head_result.stderr.strip()}", + ) + try: + head_sha = json.loads(head_result.stdout)["headRefOid"] + except (json.JSONDecodeError, KeyError) as exc: + return PrOutcome( + pr_number=pr_number, + success=False, + detail=f"could not parse head SHA: {exc}", + ) + + runs: list[dict[str, object]] = [] + page = 1 + total_count: int | None = None + while total_count is None or len(runs) < total_count: + runs_result = _run_gh( + [ + "api", + f"repos/{owner}/{repo}/actions/runs?head_sha={head_sha}&per_page=100&page={page}", + ] + ) + if runs_result.returncode != 0: + return PrOutcome( + pr_number=pr_number, + success=False, + detail=f"run list failed: {runs_result.stderr.strip()}", + ) + try: + payload = json.loads(runs_result.stdout) + except json.JSONDecodeError as exc: + return PrOutcome( + pr_number=pr_number, + success=False, + detail=f"could not parse run list: {exc}", + ) + page_runs = payload.get("workflow_runs", []) + if not isinstance(page_runs, list): + return PrOutcome( + pr_number=pr_number, + success=False, + detail="could not parse run list: workflow_runs is not a list", + ) + if total_count is None: + raw_total_count = payload.get("total_count", len(page_runs)) + total_count = ( + raw_total_count if isinstance(raw_total_count, int) else len(page_runs) + ) + if not page_runs: + break + runs.extend(page_runs) + page += 1 + + rerunnable_conclusions = {"failure", "cancelled"} + failed_run_ids = [ + r["id"] + for r in runs + if r.get("status") == "completed" + and r.get("conclusion") in rerunnable_conclusions + ] + if not failed_run_ids: + return PrOutcome( + pr_number=pr_number, + success=True, + detail="no completed failed or cancelled runs to rerun", + ) + + reran: list[int] = [] + errors: list[str] = [] + for run_id in failed_run_ids: + rerun_result = _run_gh( + [ + "api", + "-X", + "POST", + f"repos/{owner}/{repo}/actions/runs/{run_id}/rerun-failed-jobs", + ] + ) + if rerun_result.returncode == 0: + reran.append(run_id) + else: + errors.append(f"run {run_id}: {rerun_result.stderr.strip()}") + + success = not errors + detail = f"reran={reran}" + (f" errors={errors}" if errors else "") + return PrOutcome(pr_number=pr_number, success=success, detail=detail) + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def _parse_pr_numbers(raw: str) -> list[int]: + numbers: list[int] = [] + for chunk in raw.split(","): + chunk = chunk.strip() + if not chunk: + continue + try: + numbers.append(int(chunk)) + except ValueError as exc: + raise BulkPrThrottleError(f"invalid PR number {chunk!r} in --prs") from exc + if not numbers: + raise BulkPrThrottleError("--prs produced zero PR numbers") + return numbers + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + description="Mechanical dispatch throttle for bulk PR operations (OMN-16284)." + ) + parser.add_argument( + "--owner", required=True, help="GitHub org/owner. No default — fail-fast." + ) + parser.add_argument( + "--repo", required=True, help="GitHub repo name. No default — fail-fast." + ) + parser.add_argument("--prs", required=True, help="Comma-separated PR numbers.") + parser.add_argument("--operation", required=True, choices=VALID_OPERATIONS) + parser.add_argument("--wave-size", type=int, default=DEFAULT_WAVE_SIZE) + parser.add_argument( + "--queue-depth-threshold", type=int, default=DEFAULT_QUEUE_DEPTH_THRESHOLD + ) + parser.add_argument( + "--max-total-prs", + type=int, + default=None, + help=f"Explicitly raise the default cap ({DEFAULT_MAX_TOTAL_PRS}).", + ) + parser.add_argument( + "--poll-seconds", type=float, default=DEFAULT_BLOCK_POLL_SECONDS + ) + parser.add_argument( + "--max-wait-seconds", type=float, default=DEFAULT_MAX_BLOCK_SECONDS + ) + parser.add_argument("--dry-run", action="store_true") + parser.add_argument( + "--receipt", + type=Path, + default=None, + help="Path to write the JSON wave receipt " + "(default: .onex_state/bulk-pr-throttle/--.json)", + ) + args = parser.parse_args(argv) + + try: + pr_numbers = _parse_pr_numbers(args.prs) + except BulkPrThrottleError as exc: + print(f"[bulk-pr-throttle] REFUSED: {exc}", file=sys.stderr) + return 1 + + explicit_max_total_prs = args.max_total_prs is not None + max_total_prs = ( + args.max_total_prs if explicit_max_total_prs else DEFAULT_MAX_TOTAL_PRS + ) + receipt_path = args.receipt + if receipt_path is None: + ts = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ") + receipt_path = ( + Path(".onex_state") + / "bulk-pr-throttle" + / f"{args.owner}-{args.repo}-{ts}.json" + ) + + try: + report = run_bulk_operation( + owner=args.owner, + repo=args.repo, + pr_numbers=pr_numbers, + operation=args.operation, + wave_size=args.wave_size, + queue_depth_threshold=args.queue_depth_threshold, + max_total_prs=max_total_prs, + explicit_max_total_prs=explicit_max_total_prs, + dry_run=args.dry_run, + get_queue_depth=( + None if args.dry_run else lambda: gh_queue_depth(args.owner, args.repo) + ), + apply_pr_operation=(None if args.dry_run else gh_apply_pr_operation), + poll_seconds=args.poll_seconds, + max_wait_seconds=args.max_wait_seconds, + ) + except PartialBulkRunError as exc: + write_receipt(exc.report, receipt_path) + print(f"[bulk-pr-throttle] receipt written to {receipt_path}") + print(f"[bulk-pr-throttle] REFUSED: {exc}", file=sys.stderr) + return 1 + except (BulkPrThrottleError, ValueError) as exc: + print(f"[bulk-pr-throttle] REFUSED: {exc}", file=sys.stderr) + return 1 + + write_receipt(report, receipt_path) + print(f"[bulk-pr-throttle] receipt written to {receipt_path}") + + failures = [o for wave in report.waves for o in wave.outcomes if not o.success] + return 1 if failures else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/ci/tests/test_bulk_pr_throttle.py b/scripts/ci/tests/test_bulk_pr_throttle.py new file mode 100644 index 0000000000..b04c65fc2c --- /dev/null +++ b/scripts/ci/tests/test_bulk_pr_throttle.py @@ -0,0 +1,1120 @@ +# SPDX-FileCopyrightText: 2025 OmniNode.ai Inc. +# SPDX-License-Identifier: MIT +"""Unit tests for bulk_pr_throttle.py (OMN-16284). + +Covers the four explicit test categories from the ticket: wave partitioning, +threshold blocking (mocked gh calls), dry-run plan output, and refusal +paths — plus the gh CLI integration seam and the CLI entrypoint. +""" + +from __future__ import annotations + +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +SCRIPTS_CI = Path(__file__).parent.parent +sys.path.insert(0, str(SCRIPTS_CI)) + + +# --------------------------------------------------------------------------- +# Wave partitioning +# --------------------------------------------------------------------------- + + +class TestPartitionIntoWaves: + def test_default_size_evenly_divides(self): + from bulk_pr_throttle import partition_into_waves + + waves = partition_into_waves(list(range(1, 21)), 10) + assert waves == [tuple(range(1, 11)), tuple(range(11, 21))] + + def test_custom_size(self): + from bulk_pr_throttle import partition_into_waves + + waves = partition_into_waves([1, 2, 3, 4, 5], 2) + assert waves == [(1, 2), (3, 4), (5,)] + + def test_uneven_last_wave(self): + from bulk_pr_throttle import partition_into_waves + + waves = partition_into_waves(list(range(1, 12)), 5) + assert waves == [(1, 2, 3, 4, 5), (6, 7, 8, 9, 10), (11,)] + + def test_single_wave_when_fewer_than_wave_size(self): + from bulk_pr_throttle import partition_into_waves + + waves = partition_into_waves([1, 2, 3], 10) + assert waves == [(1, 2, 3)] + + def test_zero_wave_size_raises(self): + from bulk_pr_throttle import partition_into_waves + + with pytest.raises(ValueError, match="wave_size must be >= 1"): + partition_into_waves([1, 2, 3], 0) + + def test_negative_wave_size_raises(self): + from bulk_pr_throttle import partition_into_waves + + with pytest.raises(ValueError, match="wave_size must be >= 1"): + partition_into_waves([1, 2, 3], -5) + + def test_exceeds_hard_ceiling_raises(self): + from bulk_pr_throttle import MAX_WAVE_SIZE, partition_into_waves + + with pytest.raises(ValueError, match="hard ceiling"): + partition_into_waves(list(range(1, 100)), MAX_WAVE_SIZE + 1) + + def test_exactly_at_ceiling_is_allowed(self): + from bulk_pr_throttle import MAX_WAVE_SIZE, partition_into_waves + + waves = partition_into_waves(list(range(1, MAX_WAVE_SIZE + 1)), MAX_WAVE_SIZE) + assert len(waves) == 1 + assert len(waves[0]) == MAX_WAVE_SIZE + + +# --------------------------------------------------------------------------- +# Refusal paths: total PR cap, unknown operation, empty inputs, missing owner/repo +# --------------------------------------------------------------------------- + + +class TestRefusalPaths: + def test_within_default_cap_does_not_raise(self): + from bulk_pr_throttle import DEFAULT_MAX_TOTAL_PRS, validate_total_prs + + validate_total_prs( + list(range(1, DEFAULT_MAX_TOTAL_PRS + 1)), + max_total_prs=DEFAULT_MAX_TOTAL_PRS, + explicit_max_total_prs=False, + ) # no raise + + def test_exceeds_default_cap_without_explicit_flag_raises(self): + from bulk_pr_throttle import ( + DEFAULT_MAX_TOTAL_PRS, + TotalPrLimitExceededError, + validate_total_prs, + ) + + with pytest.raises(TotalPrLimitExceededError, match="explicit"): + validate_total_prs( + list(range(1, DEFAULT_MAX_TOTAL_PRS + 2)), + max_total_prs=DEFAULT_MAX_TOTAL_PRS, + explicit_max_total_prs=False, + ) + + def test_exceeds_cap_with_explicit_flag_does_not_raise(self): + from bulk_pr_throttle import validate_total_prs + + validate_total_prs( + list(range(1, 200)), + max_total_prs=250, + explicit_max_total_prs=True, + ) # no raise + + def test_exceeds_explicit_cap_with_explicit_flag_raises(self): + from bulk_pr_throttle import TotalPrLimitExceededError, validate_total_prs + + with pytest.raises(TotalPrLimitExceededError, match="explicit cap"): + validate_total_prs( + list(range(1, 201)), + max_total_prs=100, + explicit_max_total_prs=True, + ) + + def test_run_bulk_operation_rejects_empty_owner(self): + from bulk_pr_throttle import BulkPrThrottleError, run_bulk_operation + + with pytest.raises(BulkPrThrottleError, match="owner"): + run_bulk_operation( + owner="", + repo="omnibase_infra", + pr_numbers=[1, 2], + operation="noop-dry-run", + dry_run=True, + ) + + def test_run_bulk_operation_rejects_empty_repo(self): + from bulk_pr_throttle import BulkPrThrottleError, run_bulk_operation + + with pytest.raises(BulkPrThrottleError, match="repo"): + run_bulk_operation( + owner="OmniNode-ai", + repo="", + pr_numbers=[1, 2], + operation="noop-dry-run", + dry_run=True, + ) + + def test_run_bulk_operation_rejects_unknown_operation(self): + from bulk_pr_throttle import BulkPrThrottleError, run_bulk_operation + + with pytest.raises(BulkPrThrottleError, match="unknown operation"): + run_bulk_operation( + owner="OmniNode-ai", + repo="omnibase_infra", + pr_numbers=[1], + operation="delete-everything", + dry_run=True, + ) + + def test_run_bulk_operation_rejects_empty_pr_list(self): + from bulk_pr_throttle import BulkPrThrottleError, run_bulk_operation + + with pytest.raises(BulkPrThrottleError, match="non-empty"): + run_bulk_operation( + owner="OmniNode-ai", + repo="omnibase_infra", + pr_numbers=[], + operation="noop-dry-run", + dry_run=True, + ) + + def test_run_bulk_operation_requires_callables_outside_dry_run(self): + from bulk_pr_throttle import BulkPrThrottleError, run_bulk_operation + + with pytest.raises(BulkPrThrottleError, match="required"): + run_bulk_operation( + owner="OmniNode-ai", + repo="omnibase_infra", + pr_numbers=[1, 2], + operation="update-branch", + dry_run=False, + get_queue_depth=None, + apply_pr_operation=None, + ) + + def test_run_bulk_operation_exceeding_total_cap_raises_before_any_gh_call(self): + from bulk_pr_throttle import ( + DEFAULT_MAX_TOTAL_PRS, + TotalPrLimitExceededError, + run_bulk_operation, + ) + + calls = {"queue_depth": 0, "apply": 0} + + def get_queue_depth() -> int: + calls["queue_depth"] += 1 + return 0 + + def apply_pr_operation(owner: str, repo: str, pr: int, operation: str): + calls["apply"] += 1 + raise AssertionError("should never be called") + + with pytest.raises(TotalPrLimitExceededError): + run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=list(range(1, DEFAULT_MAX_TOTAL_PRS + 5)), + operation="update-branch", + dry_run=False, + get_queue_depth=get_queue_depth, + apply_pr_operation=apply_pr_operation, + ) + assert calls == {"queue_depth": 0, "apply": 0} + + +# --------------------------------------------------------------------------- +# Threshold blocking (mocked queue-depth callable — the "gh call" seam) +# --------------------------------------------------------------------------- + + +class TestWaitForQueueDepth: + def test_immediate_pass_when_already_below_threshold(self): + from bulk_pr_throttle import wait_for_queue_depth + + sleeps: list[float] = [] + depth = wait_for_queue_depth( + get_queue_depth=lambda: 42, + threshold=150, + poll_seconds=5.0, + max_wait_seconds=60.0, + sleep_fn=sleeps.append, + ) + assert depth == 42 + assert sleeps == [] + + def test_blocks_and_polls_until_below_threshold(self): + from bulk_pr_throttle import wait_for_queue_depth + + depths = iter([200, 180, 160, 100]) + sleeps: list[float] = [] + depth = wait_for_queue_depth( + get_queue_depth=lambda: next(depths), + threshold=150, + poll_seconds=10.0, + max_wait_seconds=1000.0, + sleep_fn=sleeps.append, + ) + assert depth == 100 + assert sleeps == [10.0, 10.0, 10.0] + + def test_boundary_depth_equal_to_threshold_does_not_block(self): + from bulk_pr_throttle import wait_for_queue_depth + + sleeps: list[float] = [] + depth = wait_for_queue_depth( + get_queue_depth=lambda: 150, + threshold=150, + poll_seconds=5.0, + max_wait_seconds=60.0, + sleep_fn=sleeps.append, + ) + assert depth == 150 + assert sleeps == [] + + def test_persistently_high_depth_times_out(self): + from bulk_pr_throttle import QueueDepthTimeoutError, wait_for_queue_depth + + sleeps: list[float] = [] + with pytest.raises(QueueDepthTimeoutError, match="still above threshold"): + wait_for_queue_depth( + get_queue_depth=lambda: 999, + threshold=150, + poll_seconds=10.0, + max_wait_seconds=25.0, + sleep_fn=sleeps.append, + ) + # polls until waited >= max_wait_seconds, never silently proceeds + assert sum(sleeps) >= 25.0 + + def test_rejects_non_positive_poll_seconds(self): + from bulk_pr_throttle import wait_for_queue_depth + + with pytest.raises(ValueError, match="poll_seconds"): + wait_for_queue_depth( + get_queue_depth=lambda: 999, + threshold=150, + poll_seconds=0.0, + max_wait_seconds=25.0, + ) + + def test_no_bypass_flag_exists_for_the_gate(self): + """The gate has no force/skip parameter — this is the mechanical guard.""" + import inspect + + from bulk_pr_throttle import wait_for_queue_depth + + params = set(inspect.signature(wait_for_queue_depth).parameters) + assert not params & {"force", "skip", "bypass", "ignore_threshold"} + + +# --------------------------------------------------------------------------- +# Full run_bulk_operation flow (waves + logging + receipts) +# --------------------------------------------------------------------------- + + +class TestRunBulkOperationFlow: + def test_dry_run_never_touches_gh_and_prints_plan(self, capsys): + from bulk_pr_throttle import run_bulk_operation + + report = run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[1, 2, 3, 4, 5], + operation="rerun-failed", + wave_size=2, + dry_run=True, + get_queue_depth=None, + apply_pr_operation=None, + ) + assert report.dry_run is True + assert len(report.waves) == 3 + assert report.waves[0].pr_numbers == (1, 2) + assert report.waves[1].pr_numbers == (3, 4) + assert report.waves[2].pr_numbers == (5,) + for wave in report.waves: + assert wave.outcomes == () + + out = capsys.readouterr().out + assert "DRY-RUN plan" in out + assert "wave 1:" in out + assert "wave 3:" in out + + def test_multi_wave_flow_calls_apply_per_pr_and_polls_depth_per_wave(self): + from bulk_pr_throttle import PrOutcome, run_bulk_operation + + depth_calls = 0 + + def get_queue_depth() -> int: + nonlocal depth_calls + depth_calls += 1 + return 10 # always under threshold + + applied: list[tuple[str, str, int, str]] = [] + + def apply_pr_operation( + owner: str, repo: str, pr: int, operation: str + ) -> PrOutcome: + applied.append((owner, repo, pr, operation)) + return PrOutcome(pr_number=pr, success=True, detail="ok") + + report = run_bulk_operation( + owner="OmniNode-ai", + repo="omnibase_infra", + pr_numbers=[101, 102, 103, 104, 105], + operation="update-branch", + wave_size=2, + queue_depth_threshold=150, + dry_run=False, + get_queue_depth=get_queue_depth, + apply_pr_operation=apply_pr_operation, + ) + + assert len(report.waves) == 3 + assert [pr for _, _, pr, _ in applied] == [101, 102, 103, 104, 105] + assert all(op == "update-branch" for _, _, _, op in applied) + # one queue-depth poll before each wave, plus one after each wave + assert depth_calls == 6 + for wave in report.waves: + assert wave.queue_depth_before == 10 + assert wave.queue_depth_after == 10 + assert all(o.success for o in wave.outcomes) + + def test_flow_blocks_mid_batch_when_a_later_wave_sees_high_depth(self): + """Threshold blocking (mocked gh call) applies per-wave, not just once.""" + from bulk_pr_throttle import PrOutcome, run_bulk_operation + + # wave 1 poll: 20 (ok). wave 1 after-poll: 20. + # wave 2 poll: 300 (blocks), then 300 (blocks), then 50 (ok). wave 2 after: 50. + depth_sequence = iter([20, 20, 300, 300, 50, 50]) + sleeps: list[float] = [] + + def get_queue_depth() -> int: + return next(depth_sequence) + + def apply_pr_operation( + owner: str, repo: str, pr: int, operation: str + ) -> PrOutcome: + return PrOutcome(pr_number=pr, success=True, detail="ok") + + report = run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[1, 2, 3, 4], + operation="rerun-failed", + wave_size=2, + queue_depth_threshold=150, + dry_run=False, + get_queue_depth=get_queue_depth, + apply_pr_operation=apply_pr_operation, + poll_seconds=1.0, + max_wait_seconds=100.0, + sleep_fn=sleeps.append, + ) + assert len(report.waves) == 2 + assert report.waves[1].queue_depth_before == 50 + assert sleeps == [1.0, 1.0] + + def test_logs_timestamp_count_and_depth_before_after_to_stdout(self, capsys): + from bulk_pr_throttle import PrOutcome, run_bulk_operation + + run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[6751, 6752, 6753], + operation="rerun-failed", + wave_size=10, + dry_run=False, + get_queue_depth=lambda: 5, + apply_pr_operation=lambda o, r, pr, op: PrOutcome( + pr_number=pr, success=True, detail="ok" + ), + ) + out = capsys.readouterr().out + assert "wave 1/1" in out + assert "depth_before=5" in out + assert "depth_after=5" in out + assert "6751" in out and "6752" in out and "6753" in out + + def test_later_wave_timeout_preserves_completed_wave_receipt(self): + from bulk_pr_throttle import PartialBulkRunError, PrOutcome, run_bulk_operation + + depth_sequence = iter([10, 10, 999, 999]) + + with pytest.raises(PartialBulkRunError) as exc_info: + run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[1, 2, 3, 4], + operation="rerun-failed", + wave_size=2, + queue_depth_threshold=150, + dry_run=False, + get_queue_depth=lambda: next(depth_sequence), + apply_pr_operation=lambda o, r, pr, op: PrOutcome( + pr_number=pr, success=True, detail="ok" + ), + poll_seconds=1.0, + max_wait_seconds=1.0, + sleep_fn=lambda seconds: None, + ) + + report = exc_info.value.report + assert len(report.waves) == 1 + assert report.waves[0].pr_numbers == (1, 2) + assert report.waves[0].queue_depth_after == 10 + + +# --------------------------------------------------------------------------- +# Receipt file +# --------------------------------------------------------------------------- + + +class TestWriteReceipt: + def test_receipt_json_shape(self, tmp_path): + from bulk_pr_throttle import PrOutcome, run_bulk_operation, write_receipt + + report = run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[1, 2], + operation="rerun-failed", + wave_size=10, + dry_run=False, + get_queue_depth=lambda: 3, + apply_pr_operation=lambda o, r, pr, op: PrOutcome( + pr_number=pr, success=True, detail="ok" + ), + ) + receipt_path = tmp_path / "receipt.json" + write_receipt(report, receipt_path) + + data = json.loads(receipt_path.read_text()) + assert data["owner"] == "OmniNode-ai" + assert data["repo"] == "onex_change_control" + assert data["operation"] == "rerun-failed" + assert len(data["waves"]) == 1 + wave = data["waves"][0] + assert wave["queue_depth_before"] == 3 + assert wave["queue_depth_after"] == 3 + assert wave["pr_numbers"] == [1, 2] + assert len(wave["outcomes"]) == 2 + assert wave["outcomes"][0]["pr_number"] == 1 + assert "started_at" in wave + assert "completed_at" in wave + + def test_receipt_creates_parent_dirs(self, tmp_path): + from bulk_pr_throttle import run_bulk_operation, write_receipt + + report = run_bulk_operation( + owner="OmniNode-ai", + repo="onex_change_control", + pr_numbers=[1], + operation="noop-dry-run", + dry_run=True, + ) + nested = tmp_path / "a" / "b" / "c" / "receipt.json" + write_receipt(report, nested) + assert nested.exists() + + +# --------------------------------------------------------------------------- +# gh CLI integration seam (mock _run_gh — never call the real GitHub API) +# --------------------------------------------------------------------------- + + +class TestGhIntegration: + def test_run_gh_times_out_as_failed_completed_process(self, monkeypatch): + import bulk_pr_throttle + + def fake_subprocess_run(*args, **kwargs): + raise subprocess.TimeoutExpired(cmd=args[0], timeout=kwargs["timeout"]) + + monkeypatch.setattr(subprocess, "run", fake_subprocess_run) + result = bulk_pr_throttle._run_gh(["api", "repos/owner/repo/actions/runs"]) + assert result.returncode == 124 + assert "timed out" in result.stderr + + def test_gh_queue_depth_parses_jq_output(self, monkeypatch): + import bulk_pr_throttle + + def fake_run_gh(args): + assert args[0] == "api" + assert "actions/runs?status=queued" in args[1] + assert args[-2:] == ["--jq", ".total_count"] + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="42\n", stderr="" + ) + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + assert ( + bulk_pr_throttle.gh_queue_depth("OmniNode-ai", "onex_change_control") == 42 + ) + + def test_gh_queue_depth_raises_on_nonzero_exit(self, monkeypatch): + import bulk_pr_throttle + + def fake_run_gh(args): + return subprocess.CompletedProcess( + args=args, returncode=1, stdout="", stderr="HTTP 502" + ) + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + with pytest.raises(bulk_pr_throttle.BulkPrThrottleError, match="HTTP 502"): + bulk_pr_throttle.gh_queue_depth("OmniNode-ai", "onex_change_control") + + def test_gh_apply_pr_operation_update_branch(self, monkeypatch): + import bulk_pr_throttle + + seen = {} + + def fake_run_gh(args): + seen["args"] = args + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="ok", stderr="" + ) + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "omnibase_infra", 2805, "update-branch" + ) + assert outcome.success is True + assert outcome.pr_number == 2805 + assert seen["args"] == [ + "api", + "-X", + "PUT", + "repos/OmniNode-ai/omnibase_infra/pulls/2805/update-branch", + ] + + def test_gh_apply_pr_operation_arm_automerge(self, monkeypatch): + import bulk_pr_throttle + + def fake_run_gh(args): + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="", stderr="" + ) + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "omnibase_infra", 2805, "arm-automerge" + ) + assert outcome.success is True + + def test_gh_rerun_failed_no_failed_runs(self, monkeypatch): + import bulk_pr_throttle + + def fake_run_gh(args): + if args[:2] == ["pr", "view"]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps({"headRefOid": "abc123"}), + stderr="", + ) + if args[0] == "api" and "actions/runs?head_sha=" in args[1]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps({"total_count": 0, "workflow_runs": []}), + stderr="", + ) + raise AssertionError(f"unexpected gh call: {args}") + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "onex_change_control", 6751, "rerun-failed" + ) + assert outcome.success is True + assert "no completed failed or cancelled runs" in outcome.detail + + def test_gh_rerun_failed_reruns_matching_runs(self, monkeypatch): + import bulk_pr_throttle + + rerun_calls = [] + + def fake_run_gh(args): + if args[:2] == ["pr", "view"]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps({"headRefOid": "deadbeef"}), + stderr="", + ) + if args[0] == "api" and "actions/runs?head_sha=" in args[1]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps( + { + "total_count": 4, + "workflow_runs": [ + { + "id": 111, + "status": "completed", + "conclusion": "failure", + }, + { + "id": 222, + "status": "completed", + "conclusion": "success", + }, + { + "id": 333, + "status": "completed", + "conclusion": "failure", + }, + { + "id": 444, + "status": "in_progress", + "conclusion": "failure", + }, + ], + } + ), + stderr="", + ) + if args[:2] == ["api", "-X"] and "rerun-failed-jobs" in args[-1]: + rerun_calls.append(args[-1]) + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="", stderr="" + ) + raise AssertionError(f"unexpected gh call: {args}") + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "onex_change_control", 6751, "rerun-failed" + ) + assert outcome.success is True + assert len(rerun_calls) == 2 + assert any("111" in c for c in rerun_calls) + assert any("333" in c for c in rerun_calls) + assert not any("222" in c for c in rerun_calls) + assert not any("444" in c for c in rerun_calls) + + def test_gh_rerun_failed_reruns_terminal_cancelled_runs(self, monkeypatch): + import bulk_pr_throttle + + rerun_calls = [] + + def fake_run_gh(args): + if args[:2] == ["pr", "view"]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps({"headRefOid": "deadbeef"}), + stderr="", + ) + if args[0] == "api" and "actions/runs?head_sha=" in args[1]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps( + { + "total_count": 4, + "workflow_runs": [ + { + "id": 111, + "status": "completed", + "conclusion": "cancelled", + }, + { + "id": 222, + "status": "completed", + "conclusion": "success", + }, + { + "id": 333, + "status": "completed", + "conclusion": None, + }, + { + "id": 444, + "status": "in_progress", + "conclusion": "cancelled", + }, + ], + } + ), + stderr="", + ) + if args[:2] == ["api", "-X"] and "rerun-failed-jobs" in args[-1]: + rerun_calls.append(args[-1]) + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="", stderr="" + ) + raise AssertionError(f"unexpected gh call: {args}") + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "onex_change_control", 6751, "rerun-failed" + ) + assert outcome.success is True + assert len(rerun_calls) == 1 + assert any("111" in c for c in rerun_calls) + assert not any("222" in c for c in rerun_calls) + assert not any("333" in c for c in rerun_calls) + assert not any("444" in c for c in rerun_calls) + + def test_gh_rerun_failed_paginates_workflow_runs(self, monkeypatch): + import bulk_pr_throttle + + rerun_calls = [] + + def fake_run_gh(args): + if args[:2] == ["pr", "view"]: + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps({"headRefOid": "deadbeef"}), + stderr="", + ) + if args[0] == "api" and "actions/runs?head_sha=" in args[1]: + if "&page=1" in args[1]: + payload = { + "total_count": 2, + "workflow_runs": [ + { + "id": 111, + "status": "completed", + "conclusion": "failure", + } + ], + } + elif "&page=2" in args[1]: + payload = { + "total_count": 2, + "workflow_runs": [ + { + "id": 222, + "status": "completed", + "conclusion": "cancelled", + } + ], + } + else: + raise AssertionError(f"unexpected page: {args}") + return subprocess.CompletedProcess( + args=args, + returncode=0, + stdout=json.dumps(payload), + stderr="", + ) + if args[:2] == ["api", "-X"] and "rerun-failed-jobs" in args[-1]: + rerun_calls.append(args[-1]) + return subprocess.CompletedProcess( + args=args, returncode=0, stdout="", stderr="" + ) + raise AssertionError(f"unexpected gh call: {args}") + + monkeypatch.setattr(bulk_pr_throttle, "_run_gh", fake_run_gh) + outcome = bulk_pr_throttle.gh_apply_pr_operation( + "OmniNode-ai", "onex_change_control", 6751, "rerun-failed" + ) + assert outcome.success is True + assert len(rerun_calls) == 2 + assert any("111" in c for c in rerun_calls) + assert any("222" in c for c in rerun_calls) + + +# --------------------------------------------------------------------------- +# PR-number parsing +# --------------------------------------------------------------------------- + + +class TestParsePrNumbers: + def test_parses_comma_separated(self): + from bulk_pr_throttle import _parse_pr_numbers + + assert _parse_pr_numbers("1,2,3") == [1, 2, 3] + + def test_strips_whitespace(self): + from bulk_pr_throttle import _parse_pr_numbers + + assert _parse_pr_numbers(" 1, 2 ,3 ") == [1, 2, 3] + + def test_ignores_empty_chunks(self): + from bulk_pr_throttle import _parse_pr_numbers + + assert _parse_pr_numbers("1,,2,") == [1, 2] + + def test_invalid_number_raises(self): + from bulk_pr_throttle import BulkPrThrottleError, _parse_pr_numbers + + with pytest.raises(BulkPrThrottleError, match="invalid PR number"): + _parse_pr_numbers("1,abc,3") + + def test_all_empty_raises(self): + from bulk_pr_throttle import BulkPrThrottleError, _parse_pr_numbers + + with pytest.raises(BulkPrThrottleError, match="zero PR numbers"): + _parse_pr_numbers(" , , ") + + +# --------------------------------------------------------------------------- +# CLI entrypoint (main) +# --------------------------------------------------------------------------- + + +class TestMainCli: + def test_dry_run_end_to_end(self, capsys, tmp_path): + from bulk_pr_throttle import main + + receipt = tmp_path / "receipt.json" + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "6751,6752,6753", + "--operation", + "rerun-failed", + "--wave-size", + "2", + "--dry-run", + "--receipt", + str(receipt), + ] + ) + assert result == 0 + out = capsys.readouterr().out + assert "DRY-RUN plan" in out + assert receipt.exists() + data = json.loads(receipt.read_text()) + assert data["dry_run"] is True + + def test_missing_owner_is_a_hard_argparse_error(self): + from bulk_pr_throttle import main + + with pytest.raises(SystemExit): + main( + [ + "--repo", + "onex_change_control", + "--prs", + "1", + "--operation", + "noop-dry-run", + "--dry-run", + ] + ) + + def test_missing_repo_is_a_hard_argparse_error(self): + from bulk_pr_throttle import main + + with pytest.raises(SystemExit): + main( + [ + "--owner", + "OmniNode-ai", + "--prs", + "1", + "--operation", + "noop-dry-run", + "--dry-run", + ] + ) + + def test_refuses_batch_larger_than_default_cap_without_flag(self, capsys): + from bulk_pr_throttle import DEFAULT_MAX_TOTAL_PRS, main + + prs = ",".join(str(n) for n in range(1, DEFAULT_MAX_TOTAL_PRS + 5)) + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + prs, + "--operation", + "noop-dry-run", + "--dry-run", + ] + ) + assert result == 1 + err = capsys.readouterr().err + assert "REFUSED" in err + + def test_partial_run_error_writes_completed_wave_receipt( + self, monkeypatch, tmp_path + ): + import bulk_pr_throttle + from bulk_pr_throttle import ( + BulkRunReport, + PartialBulkRunError, + PrOutcome, + WaveReceipt, + main, + ) + + partial = BulkRunReport( + owner="OmniNode-ai", + repo="onex_change_control", + operation="rerun-failed", + wave_size=10, + queue_depth_threshold=150, + dry_run=False, + waves=( + WaveReceipt( + wave_index=1, + pr_numbers=(1, 2), + operation="rerun-failed", + dry_run=False, + queue_depth_before=5, + queue_depth_after=None, + started_at="2026-08-22T00:00:00+00:00", + completed_at="2026-08-22T00:00:01+00:00", + outcomes=( + PrOutcome(pr_number=1, success=True, detail="ok"), + PrOutcome(pr_number=2, success=True, detail="ok"), + ), + ), + ), + ) + + def fake_run_bulk_operation(**kwargs): + raise PartialBulkRunError("queue depth probe failed", partial) + + monkeypatch.setattr( + bulk_pr_throttle, "run_bulk_operation", fake_run_bulk_operation + ) + receipt = tmp_path / "partial.json" + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "1,2,3", + "--operation", + "rerun-failed", + "--receipt", + str(receipt), + ] + ) + assert result == 1 + data = json.loads(receipt.read_text()) + assert data["waves"][0]["pr_numbers"] == [1, 2] + assert data["waves"][0]["queue_depth_after"] is None + + def test_explicit_max_total_prs_flag_allows_large_batch(self, capsys): + from bulk_pr_throttle import DEFAULT_MAX_TOTAL_PRS, main + + prs = ",".join(str(n) for n in range(1, DEFAULT_MAX_TOTAL_PRS + 5)) + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + prs, + "--operation", + "noop-dry-run", + "--dry-run", + "--max-total-prs", + str(DEFAULT_MAX_TOTAL_PRS + 10), + ] + ) + assert result == 0 + + def test_invalid_operation_is_a_hard_argparse_error(self): + from bulk_pr_throttle import main + + with pytest.raises(SystemExit): + main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "1", + "--operation", + "delete-everything", + "--dry-run", + ] + ) + + def test_wave_size_over_hard_ceiling_refuses(self, capsys): + from bulk_pr_throttle import MAX_WAVE_SIZE, main + + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "1,2,3", + "--operation", + "noop-dry-run", + "--dry-run", + "--wave-size", + str(MAX_WAVE_SIZE + 1), + ] + ) + assert result == 1 + err = capsys.readouterr().err + assert "REFUSED" in err + + def test_non_dry_run_uses_real_gh_seam(self, monkeypatch, tmp_path): + """Non-dry-run wires get_queue_depth/apply_pr_operation to the gh functions.""" + import bulk_pr_throttle + from bulk_pr_throttle import PrOutcome, main + + monkeypatch.setattr(bulk_pr_throttle, "gh_queue_depth", lambda owner, repo: 5) + monkeypatch.setattr( + bulk_pr_throttle, + "gh_apply_pr_operation", + lambda owner, repo, pr, op: PrOutcome( + pr_number=pr, success=True, detail="ok" + ), + ) + receipt = tmp_path / "receipt.json" + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "1,2", + "--operation", + "rerun-failed", + "--receipt", + str(receipt), + ] + ) + assert result == 0 + data = json.loads(receipt.read_text()) + assert data["dry_run"] is False + assert data["waves"][0]["queue_depth_before"] == 5 + + def test_non_dry_run_reports_failure_exit_code_on_pr_failure( + self, monkeypatch, tmp_path + ): + import bulk_pr_throttle + from bulk_pr_throttle import PrOutcome, main + + monkeypatch.setattr(bulk_pr_throttle, "gh_queue_depth", lambda owner, repo: 5) + monkeypatch.setattr( + bulk_pr_throttle, + "gh_apply_pr_operation", + lambda owner, repo, pr, op: PrOutcome( + pr_number=pr, success=False, detail="boom" + ), + ) + receipt = tmp_path / "receipt.json" + result = main( + [ + "--owner", + "OmniNode-ai", + "--repo", + "onex_change_control", + "--prs", + "1", + "--operation", + "rerun-failed", + "--receipt", + str(receipt), + ] + ) + assert result == 1 diff --git a/scripts/generate_application_database_table_grants.py b/scripts/generate_application_database_table_grants.py index b818f9fcc3..09f1a5aa1b 100644 --- a/scripts/generate_application_database_table_grants.py +++ b/scripts/generate_application_database_table_grants.py @@ -51,6 +51,7 @@ TOPOLOGY_PROFILE_INSTANCE_MAP, ) from omnibase_infra.topology.table_grant_derivation import ( + LEGACY_MIGRATION_TABLE_DECLARATIONS, STATE_IO_TABLE_DECLARATIONS, ContractTableDeclaration, TopologyTableGrants, @@ -299,13 +300,17 @@ def main(argv: Sequence[str] | None = None) -> int: if args.write and args.check: parser.error("--write and --check are mutually exclusive") + supplemental_declarations = ( + STATE_IO_TABLE_DECLARATIONS + LEGACY_MIGRATION_TABLE_DECLARATIONS + ) declarations = ( - load_contract_declarations(args.contracts_root) + STATE_IO_TABLE_DECLARATIONS + load_contract_declarations(args.contracts_root) + supplemental_declarations ) print( - f"loaded {len(declarations) - len(STATE_IO_TABLE_DECLARATIONS)} " + f"loaded {len(declarations) - len(supplemental_declarations)} " f"db_io.db_tables declaration(s) from {args.contracts_root} plus " - f"{len(STATE_IO_TABLE_DECLARATIONS)} state_io declaration(s)" + f"{len(STATE_IO_TABLE_DECLARATIONS)} state_io declaration(s) and " + f"{len(LEGACY_MIGRATION_TABLE_DECLARATIONS)} legacy migration declaration(s)" ) exit_code = 0 diff --git a/src/omnibase_infra/topology/table_grant_derivation.py b/src/omnibase_infra/topology/table_grant_derivation.py index 2ac4d10fce..f8d696d01c 100644 --- a/src/omnibase_infra/topology/table_grant_derivation.py +++ b/src/omnibase_infra/topology/table_grant_derivation.py @@ -59,6 +59,7 @@ "WRITE_PRIVILEGES", "DOMAIN_PROJECTION_BINDINGS", "STATE_IO_TABLE_DECLARATIONS", + "LEGACY_MIGRATION_TABLE_DECLARATIONS", "TENANT_TABLES_PHYSICALLY_IN_PUBLIC_UNTIL_OMN15359", "INTERNAL_TABLES_PHYSICALLY_IN_PUBLIC_UNTIL_OMN15359", "physical_grant_schema_for_table", @@ -128,6 +129,35 @@ class ContractTableDeclaration: ), ) +# Some migration-owned projection relations landed before their producing node's +# ``db_io.db_tables`` contract is available in the omnimarket contracts root +# consumed by this repo's CI. They still participate in the same grant +# derivation path because runtime wiring validates the topology grant before the +# physical-schema bridge can resolve the insert target (OMN-16316). Keeping this +# as an explicit supplemental manifest is preferable to hand-maintaining the +# generated YAML: ``--check`` continues to prove every shipped instance and +# rendered catalog against one auditable derivation source. +LEGACY_MIGRATION_TABLE_DECLARATIONS: tuple[ContractTableDeclaration, ...] = ( + ContractTableDeclaration( + node="legacy_migration:tenant_inference_credentials", + contract_path=Path( + "docker/migrations/forward/nodes/node_projection_tenant_credentials/" + "0000_create_tenant_inference_credentials.sql" + ), + table=ModelDbTableDeclaration( + name="tenant_inference_credentials", + database_ref="application", + schema="tenant", + migration=( + "docker/migrations/forward/nodes/node_projection_tenant_credentials/" + "0000_create_tenant_inference_credentials.sql" + ), + access="write", + role="credentials", + ), + ), +) + @dataclass(frozen=True, slots=True) class UnmappableDeclaration: diff --git a/uv.lock b/uv.lock index cde27b31de..ed6fa79984 100644 --- a/uv.lock +++ b/uv.lock @@ -2132,7 +2132,7 @@ wheels = [ [[package]] name = "omnibase-infra" -version = "0.38.9" +version = "0.38.10" source = { editable = "." } dependencies = [ { name = "a2a-sdk" },