fix(automation): restore bounded OpenCode mention routing

docs/automation/review-agent-comment-invocation.md

@@ -1,6 +1,6 @@
 # Review-agent comment invocation
 
-Updated: 2026-08-06
+Updated: 2026-08-19
 
 ## Purpose
 
@@ -28,6 +28,8 @@ Wrapper workflows use the verified key in their non-cancelling concurrency group
 
 Target-repository acknowledgement comments and reactions are user-experience signals only. They are not dispatch authority because repository writers, bot identities, or credential rotation could otherwise forge or invalidate a marker. A failed acknowledgement cannot cause completed agent work to be redispatched.
 
+When a live claim exists without a visible receipt comment, the router republishes the acknowledgement without forwarding the request again; reaction failures are warnings and do not block the durable comment.
+
 A user or fine-grained token enumerates organization repositories. When the OpenCode GitHub App installation token is the available credential, the sweep instead uses GitHub's installation-repositories endpoint, which returns only repositories accessible to that installation. This avoids depending on an organization-issues endpoint whose documented fine-grained token support is user-token-oriented.
 
 This preserves the central MSA boundary without copying privileged workflow code into every product repository.

scripts/ci/agent_mention_router.py

@@ -34,6 +34,7 @@
 ACTOR_RE = re.compile(r"^[A-Za-z0-9-]+$")
 RECEIPT_RE = re.compile(r"<!-- cwl-agent-mention-receipt:(\d+) -->")
 REPOSITORY_DISPATCH_CLIENT_PAYLOAD_MAX_KEYS = 10
+GITHUB_API_TIMEOUT_SECONDS = 30
 
 
 @dataclass(frozen=True)
@@ -66,21 +67,29 @@ def request(
         *,
         input_payload: dict[str, Any] | None = None,
     ) -> Any:
-        """Execute ``gh api`` and decode its optional JSON response."""
+        """Execute one bounded ``gh api`` request and decode optional JSON."""
 
         command = ["gh", "api", *args]
         if input_payload is not None:
             command.extend(["--input", "-"])
         environment = os.environ.copy()
         environment["GH_TOKEN"] = self._token
-        completed = subprocess.run(
-            command,
-            input=None if input_payload is None else json.dumps(input_payload),
-            text=True,
-            capture_output=True,
-            check=False,
-            env=environment,
-        )
+        try:
+            completed = subprocess.run(
+                command,
+                input=None if input_payload is None else json.dumps(input_payload),
+                text=True,
+                capture_output=True,
+                shell=False,
+                check=False,
+                env=environment,
+                timeout=GITHUB_API_TIMEOUT_SECONDS,
+            )
+        except subprocess.TimeoutExpired as exc:
+            raise RuntimeError(
+                "gh api timed out after "
+                f"{GITHUB_API_TIMEOUT_SECONDS} seconds"
+            ) from exc
         return_code = int(getattr(completed, "returncode", 0))
         if return_code:
             diagnostic = " ".join(
@@ -464,6 +473,16 @@ def dispatch_request(
         )
         return handles
 
+    acknowledgement_cache_key = (
+        f"acknowledgement:{request.repository}:{request.pull_request_number}:"
+        f"{request.pull_request_head_sha}:{request.comment_id}"
+    )
+    if (
+        ledger_artifact_cache is not None
+        and ledger_artifact_cache.get(acknowledgement_cache_key)
+    ):
+        return ()
+
     existing = dispatched_agents(
         request,
         dispatch_client,
@@ -472,7 +491,10 @@ def dispatch_request(
     )
     missing = tuple(agent for agent in dispatchable if agent not in existing)
     handles = tuple(f"@{agent}" for agent in missing)
-    if not missing:
+    existing_handles = tuple(
+        f"@{agent}" for agent in dispatchable if agent in existing
+    )
+    if not missing and not existing:
         if rejected:
             print(
                 "Rejected agent mention without target mutation "
@@ -501,18 +523,24 @@ def dispatch_request(
             ledger_artifact_cache[agent_ledger_artifact_name(request, agent)] = True
 
     target_api = f"repos/{request.repository}"
-    target_client.request(
-        [
-            f"{target_api}/issues/comments/{request.comment_id}/reactions",
-            "-X",
-            "POST",
-        ],
-        input_payload={"content": "eyes"},
-    )
-    status_parts = [f"Queued {' and '.join(handles)}"]
-    existing_handles = tuple(
-        f"@{agent}" for agent in dispatchable if agent in existing
-    )
+    try:
+        target_client.request(
+            [
+                f"{target_api}/issues/comments/{request.comment_id}/reactions",
+                "-X",
+                "POST",
+            ],
+            input_payload={"content": "eyes"},
+        )
+    except Exception as exc:  # noqa: BLE001 - acknowledgement is cosmetic
+        message = " ".join(str(exc).split()) or exc.__class__.__name__
+        print(
+            "::warning::Agent mention acknowledgement reaction failed; "
+            f"durable dispatch state is preserved: {message[:1000]}"
+        )
+    status_parts: list[str] = []
+    if handles:
+        status_parts.append(f"Queued {' and '.join(handles)}")
     if existing_handles:
         status_parts.append(
             f"Already queued {' and '.join(existing_handles)} on this exact request"
@@ -538,6 +566,8 @@ def dispatch_request(
         ],
         input_payload={"body": acknowledgement},
     )
+    if ledger_artifact_cache is not None:
+        ledger_artifact_cache[acknowledgement_cache_key] = True
     return handles
 
 

scripts/ci/agent_mention_sweep.py

@@ -4,8 +4,10 @@
 from __future__ import annotations
 
 import argparse
+import concurrent.futures
 import os
 import re
+import threading
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
 from typing import Any, Callable, Iterator, Sequence
@@ -21,6 +23,7 @@
 ORG_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]+$")
 REPOSITORY_RE = re.compile(r"^ContextualWisdomLab/[A-Za-z0-9_.-]+$")
 REPOSITORY_SOURCES = frozenset({"organization", "installation"})
+REPOSITORY_ROTATION_SECONDS = 5 * 60
 
 
 @dataclass
@@ -146,55 +149,58 @@ def list_recent_pull_requests(
     repository_source: str,
     since: str,
     on_error: Callable[[str, Exception], None] | None = None,
+    rotation_offset: int = 0,
 ) -> Iterator[dict[str, Any]]:
-    """Yield recent open pull requests with lazy cutoff-aware pagination."""
+    """Yield recent open pull requests with bounded fair repository rotation."""
 
     cutoff = parse_timestamp(since)
     repositories = list_accessible_repositories(
         client,
         organization=organization,
         repository_source=repository_source,
     )
-    for repository in repositories:
-        try:
-            page = 1
-            while True:
-                response = client.request(
-                    [
-                        f"repos/{repository}/pulls",
-                        "-X",
-                        "GET",
-                        "-f",
-                        "state=open",
-                        "-f",
-                        "sort=updated",
-                        "-f",
-                        "direction=desc",
-                        "-f",
-                        "per_page=100",
-                        "-f",
-                        f"page={page}",
-                    ]
-                )
-                pull_requests = flatten_pages(response)
-                if not pull_requests:
+    if not repositories:
+        return
+    rotation_offset %= len(repositories)
+    repositories = repositories[rotation_offset:] + repositories[:rotation_offset]
+    stop_event = threading.Event()
+
+    def fetch(repository: str) -> list[dict[str, Any]]:
+        """Fetch one repository's recent open pull requests."""
+
+        results: list[dict[str, Any]] = []
+        page = 1
+        while not stop_event.is_set():
+            response = client.request(
+                [
+                    f"repos/{repository}/pulls",
+                    "-X",
+                    "GET",
+                    "-f",
+                    "state=open",
+                    "-f",
+                    "sort=updated",
+                    "-f",
+                    "direction=desc",
+                    "-f",
+                    "per_page=100",
+                    "-f",
+                    f"page={page}",
+                ]
+            )
+            pull_requests = flatten_pages(response)
+            if not pull_requests:
+                break
+            reached_cutoff = False
+            for pull_request in pull_requests:
+                if parse_timestamp(str(pull_request.get("updated_at") or "")) < cutoff:
+                    reached_cutoff = True
                     break
-                reached_cutoff = False
-                for pull_request in pull_requests:
-                    if (
-                        parse_timestamp(
-                            str(pull_request.get("updated_at") or "")
-                        )
-                        < cutoff
-                    ):
-                        reached_cutoff = True
-                        break
-                    number = pull_request.get("number")
-                    if not isinstance(number, int) or number < 1:
-                        raise ValueError(
-                            "GitHub returned an invalid pull request number"
-                        )
-                    yield {
+                number = pull_request.get("number")
+                if not isinstance(number, int) or number < 1:
+                    raise ValueError("GitHub returned an invalid pull request number")
+                results.append(
+                    {
                         "number": number,
                         "repository": repository,
                         "pull_request": {
@@ -204,13 +210,32 @@ def list_recent_pull_requests(
                             )
                         },
                     }
-                if reached_cutoff or len(pull_requests) < 100:
-                    break
-                page += 1
-        except Exception as exc:  # noqa: BLE001 - repository isolation boundary
-            if on_error is None:
-                raise
-            on_error(repository, exc)
+                )
+            if reached_cutoff or len(pull_requests) < 100:
+                break
+            page += 1
+        return results
+
+    executor = concurrent.futures.ThreadPoolExecutor(
+        max_workers=min(4, len(repositories))
+    )
+    futures = [
+        (repository, executor.submit(fetch, repository))
+        for repository in repositories
+    ]
+    try:
+        for repository, future in futures:
+            try:
+                yield from future.result()
+            except Exception as exc:  # noqa: BLE001 - repository isolation boundary
+                if on_error is None:
+                    raise
+                on_error(repository, exc)
+    finally:
+        stop_event.set()
+        for _, future in futures:
+            future.cancel()
+        executor.shutdown(wait=True, cancel_futures=True)
 
 
 def list_recent_comments(
@@ -295,7 +320,9 @@ def sweep(
 
     if max_dispatches < 1 or max_dispatches > 100:
         raise ValueError("max dispatches must be between 1 and 100")
-    since = cutoff_timestamp(lookback_hours, now=now)
+    current = now or datetime.now(timezone.utc)
+    since = cutoff_timestamp(lookback_hours, now=current)
+    rotation_offset = int(current.timestamp() // REPOSITORY_ROTATION_SECONDS)
     counters = metrics if metrics is not None else SweepMetrics()
     ledger_artifact_cache: dict[str, bool] = {}
     dispatched = 0
@@ -315,6 +342,7 @@ def record_failure(scope: str, error: Exception) -> None:
         repository_source=repository_source,
         since=since,
         on_error=record_failure,
+        rotation_offset=rotation_offset,
     ):
         issue_scope = f"{issue.get('repository')}#{issue.get('number')}"
         try: