fix(automation): yield completed mention repositories fairly

CHANGELOG.md

@@ -22,6 +22,10 @@ Semantic Versioning where the repository publishes a release.
 
 ### Changed
 
+- Emit completed repository pull-list requests as they finish in the five-minute
+  agent-mention sweep, while retaining the four-worker ceiling, rotation, and
+  exact-name dispatch ledger, so one slow repository cannot hide ready sibling
+  repositories.
 - Require the hourly repair worker to establish an exact-head root cause, enumerate the smallest remediation candidates, and prove writer authority, sealed-path scope, credentials, dependency order, verifiability, and causal effect before editing; infeasible or external blockers leave the tree unchanged while the broader loop continues with another eligible PR or buyer-visible product gap.
 - Run the bounded Quarantine Sandbox Runtime heartbeat at minute 14 without granting the caller model secrets, repository mutation permissions, approval, merge, release, artifact-execution, or final security-verdict authority.
 - Run the bounded Clearfolio PR review-feedback repair caller at minute 23 of every hour while keeping the shared scheduler free of product-specific timers and repository names for modular reuse by naruon, contextual-orchestrator, Inkspan, and other CWL services.

docs/doctoring/agent-mention-concurrency-isolation.md

@@ -1,13 +1,14 @@
 # Review-agent mention routing reliability
 
-Review date: **2026-08-19**
+Review date: **2026-08-20**
 
 ## Incident
 
 Trusted `@opencode-agent` comments could remain unacknowledged and fail to start the existing OpenCode review path. Two independent control-plane defects produced the same operator-visible symptom before model execution.
 
 1. The OpenCode `repository_dispatch.client_payload` exceeded GitHub's ten-property limit, so GitHub rejected the request with HTTP 422 before the trusted wrapper started.
 2. Interactive `issue_comment` routing and the five-minute organization sweep shared one workflow-level concurrency group. Under the default single-pending contract, a newly queued sweep could replace a pending interactive mention before exact-head resolution, durable claim creation, dispatch, or acknowledgement.
+3. The organization sweep submitted pull-list requests to four workers but consumed their futures in repository-list order. A slow earlier repository could therefore hide a completed later repository until the sweep approached its 15-minute job limit or its bounded dispatch frontier.
 
 Neither defect is evidence that the requesting maintainer, model, repository allowlist, or final review result is invalid.
 
@@ -17,6 +18,7 @@ The permanent regression contracts were committed before their corresponding pro
 
 - `tests/test_agent_mention_dispatch_payload_limit.py` requires both dispatch hops to stay at or below ten top-level payload properties and requires the router to reject an oversized payload before GitHub does.
 - `tests/test_agent_mention_queue_isolation.py` requires the interactive route and scheduled sweep to use different job-level concurrency groups, with `queue: max` on the interactive route and no cancellation of in-progress interactive work.
+- `tests/test_agent_mention_timeout_bounds.py` requires a completed later repository to yield before a deliberately blocked earlier repository, while retaining the fixed four-worker ceiling and bounded generator shutdown.
 
 ## Decision
 
@@ -54,6 +56,16 @@ GitHub documents that `queue: max` permits up to 100 pending jobs or workflow ru
 
 Concurrency is not the idempotency authority. Duplicate forwarding remains governed by the complete canonical invocation key, exact-key downstream concurrency, and the immutable exact-name Actions artifact ledger.
 
+### Fair repository completion
+
+The sweep now consumes the existing four-worker repository futures through
+`concurrent.futures.as_completed`. Repository rotation still changes the
+starting position every five minutes, and the four-worker ceiling remains in
+place. The only changed ordering is the local observation order: a repository
+whose pull-list request completes first can expose its recent PR comments
+before a slower sibling. Dispatch remains sequential through the existing
+ledger and exact-head validation boundaries.
+
 ## Preserved boundaries
 
 - No model provider, reviewer identity, repository allowlist, token name, credential scope, or branch-protection rule changes.
@@ -72,6 +84,8 @@ After protected integration:
 3. require the trusted OpenCode wrapper and review-only scheduler dispatch to start for the same repository, pull request, and exact head;
 4. verify that a scheduled sweep cannot cancel or replace the interactive route;
 5. distinguish downstream provider or review failure from routing failure rather than treating every missing verdict as the same incident.
+6. verify the slow-first/fast-later repository regression remains green so a
+   delayed repository cannot starve a completed sibling's comment inventory.
 
 A receipt proves routing and durable claim processing. It is not an approval and never substitutes for exact-head checks or branch protection.
 
@@ -82,6 +96,8 @@ Do not restore either defective boundary:
 - do not increase the first- or second-hop payload beyond GitHub's limit;
 - do not move local and scheduled work back into one workflow-level concurrency group;
 - do not replace `queue: max` with the default single-pending interactive queue unless another independently reviewed durable queue preserves every eligible request.
+- do not remove the four-worker ceiling or replace completion-order observation
+  with repository-list-order waiting without a new bounded fairness contract.
 
 A safe emergency degradation may suspend the scheduled sweep while retaining the isolated interactive route.
 
@@ -92,3 +108,5 @@ GitHub. (n.d.). *Control the concurrency of workflows and jobs*. GitHub Docs. Re
 GitHub. (n.d.). *REST API endpoints for repositories: Create a repository dispatch event*. GitHub Docs. Retrieved August 19, 2026, from https://docs.github.com/en/rest/repos/repos#create-a-repository-dispatch-event
 
 GitHub. (n.d.). *Store and share data with workflow artifacts*. GitHub Docs. Retrieved August 19, 2026, from https://docs.github.com/en/actions/tutorials/store-and-share-data
+
+Python Software Foundation. (n.d.). *concurrent.futures — Launching parallel tasks*. Python documentation. Retrieved August 20, 2026, from https://docs.python.org/3/library/concurrent.futures.html#concurrent.futures.as_completed

scripts/ci/agent_mention_sweep.py

@@ -219,12 +219,13 @@ def fetch(repository: str) -> list[dict[str, Any]]:
     executor = concurrent.futures.ThreadPoolExecutor(
         max_workers=min(4, len(repositories))
     )
-    futures = [
-        (repository, executor.submit(fetch, repository))
+    futures = {
+        executor.submit(fetch, repository): repository
         for repository in repositories
-    ]
+    }
     try:
-        for repository, future in futures:
+        for future in concurrent.futures.as_completed(futures):
+            repository = futures[future]
             try:
                 yield from future.result()
             except Exception as exc:  # noqa: BLE001 - repository isolation boundary
@@ -233,7 +234,7 @@ def fetch(repository: str) -> list[dict[str, Any]]:
                 on_error(repository, exc)
     finally:
         stop_event.set()
-        for _, future in futures:
+        for future in futures:
             future.cancel()
         executor.shutdown(wait=True, cancel_futures=True)
 

tests/test_agent_mention_timeout_bounds.py

@@ -72,6 +72,7 @@ def test_github_client_applies_one_finite_timeout(monkeypatch) -> None:
     observed = []
 
     def fake_run(command, **kwargs):
+        """Return a successful JSON response while recording subprocess options."""
         observed.append((command, kwargs))
         return SimpleNamespace(stdout='{"ok": true}\n', returncode=0)
 
@@ -88,6 +89,7 @@ def test_github_client_converts_timeout_to_bounded_diagnostic(monkeypatch) -> No
     router = router_module()
 
     def timeout_run(command, **kwargs):
+        """Raise the same timeout the production wrapper must translate."""
         raise subprocess.TimeoutExpired(command, kwargs["timeout"])
 
     monkeypatch.setattr(router.subprocess, "run", timeout_run)
@@ -107,6 +109,7 @@ def test_repository_fanout_uses_exactly_four_workers_at_scale(monkeypatch) -> No
     worker_limits = []
 
     def recording_executor(*, max_workers):
+        """Record the configured worker ceiling before creating real workers."""
         worker_limits.append(max_workers)
         return real_executor(max_workers=max_workers)
 
@@ -125,9 +128,66 @@ def recording_executor(*, max_workers):
     )
 
     assert worker_limits == [4]
-    assert [item["repository"] for item in results] == [
+    assert sorted(item["repository"] for item in results) == sorted(
         f"ContextualWisdomLab/{name}" for name in names
-    ]
+    )
+
+
+def test_repository_fanout_yields_fast_repository_before_slow_one() -> None:
+    """A slow first repository must not hide a completed later repository."""
+
+    sweep = sweep_module()
+    slow_started = threading.Event()
+    fast_finished = threading.Event()
+    release_slow = threading.Event()
+    first_yielded = threading.Event()
+
+    class FairnessClient:
+        """Block one repository while allowing the next one to complete."""
+
+        def request(self, args, *, input_payload=None):
+            """Return the inventory or one deliberately paced pull list."""
+
+            del input_payload
+            endpoint = args[0]
+            if endpoint == "orgs/ContextualWisdomLab/repos":
+                return [repository("alpha"), repository("bravo")]
+            if endpoint.endswith("alpha/pulls"):
+                slow_started.set()
+                assert release_slow.wait(2)
+                return [pull(1)]
+            if endpoint.endswith("bravo/pulls"):
+                fast_finished.set()
+                return [pull(2)]
+            raise AssertionError(f"unexpected endpoint: {endpoint}")
+
+    generator = sweep.list_recent_pull_requests(
+        FairnessClient(),
+        organization="ContextualWisdomLab",
+        repository_source="organization",
+        since="2026-08-19T00:00:00Z",
+    )
+    first_result = []
+
+    def read_first_result() -> None:
+        """Read one result without blocking the test's release signal."""
+
+        first_result.append(next(generator))
+        first_yielded.set()
+
+    reader = threading.Thread(target=read_first_result)
+    reader.start()
+    assert slow_started.wait(2)
+    assert fast_finished.wait(2)
+    yielded_before_slow_release = first_yielded.wait(0.5)
+    release_slow.set()
+    reader.join(2)
+    assert not reader.is_alive()
+    assert yielded_before_slow_release
+    assert first_result[0]["repository"] == "ContextualWisdomLab/bravo"
+    assert {item["repository"] for item in generator} == {
+        "ContextualWisdomLab/alpha"
+    }
 
 
 def test_empty_inventory_does_not_construct_an_executor(monkeypatch) -> None:
@@ -136,6 +196,7 @@ def test_empty_inventory_does_not_construct_an_executor(monkeypatch) -> None:
     sweep = sweep_module()
 
     def forbidden_executor(*args, **kwargs):
+        """Fail the test if the empty inventory allocates an executor."""
         raise AssertionError(f"executor called with {args!r} {kwargs!r}")
 
     monkeypatch.setattr(
@@ -167,6 +228,7 @@ class ClosingClient:
         """Keep the second repository in one bounded in-flight request."""
 
         def request(self, args, *, input_payload=None):
+            """Return page one or pause page two until the closer releases it."""
             del input_payload
             endpoint = args[0]
             if endpoint == "orgs/ContextualWisdomLab/repos":
@@ -191,12 +253,15 @@ class RecordingExecutor:
         """Expose the moment shutdown begins while delegating real workers."""
 
         def __init__(self, *, max_workers):
+            """Create the real bounded executor used by this test double."""
             self._inner = real_executor(max_workers=max_workers)
 
         def submit(self, *args, **kwargs):
+            """Forward one repository fetch to the delegated executor."""
             return self._inner.submit(*args, **kwargs)
 
         def shutdown(self, *, wait, cancel_futures):
+            """Record shutdown before forwarding the bounded cancellation call."""
             shutdown_started.set()
             return self._inner.shutdown(
                 wait=wait,