fix(sweep): bound process exit with cooperative worker cancellation

scripts/ci/agent_mention_sweep.py

@@ -31,17 +31,13 @@
 # log tail and metrics. Stop dispatching new work with margin to spare so
 # the sweep exits cleanly and reports what it completed.
 #
-# Returning early only stops NEW work: list_recent_pull_requests' generator
-# cleanup still blocks (executor.shutdown(wait=True)) until every currently
-# RUNNING repository fetch finishes on its own. GitHubClient's rate-limit
-# retry costs up to ~255s worst case for one repository (six attempts, each
-# up to the 30s subprocess timeout, plus ~75s of backoff between them), and
-# up to max_workers of those can be running concurrently at the moment the
-# deadline trips (bounded by that ceiling, not multiplied by it, since they
-# run in parallel). Budget = 900s job timeout - ~60s setup/checkout
-# overhead - ~255s worst-case cleanup wait, with a further margin still
-# unspent.
-DEFAULT_TIME_BUDGET_SECONDS = 480.0
+# Returning early stops NEW work and immediately abandons running fetches:
+# list_recent_pull_requests' generator cleanup no longer blocks
+# (executor.shutdown(wait=False)) on currently RUNNING repository fetches.
+# We no longer need to budget ~255s for a worst-case GitHubClient rate-limit
+# retry cleanup wait. Budget = 900s job timeout - ~60s setup/checkout
+# overhead, leaving a generous margin.
+DEFAULT_TIME_BUDGET_SECONDS = 780.0
 
 
 @dataclass
@@ -254,7 +250,7 @@ def fetch(repository: str) -> list[dict[str, Any]]:
         stop_event.set()
         for future in futures:
             future.cancel()
-        executor.shutdown(wait=True, cancel_futures=True)
+        executor.shutdown(wait=False, cancel_futures=True)
 
 
 def list_recent_comments(

tests/test_agent_mention_router.py

@@ -765,6 +765,36 @@ def test_dispatched_agents_fetches_multiple_candidates_concurrently() -> None:
     assert len(artifact_calls) == 2
 
 
+def test_dispatched_agents_fetches_multiple_candidates_concurrently_shutdown(monkeypatch) -> None:
+    """The thread pool shutdown uses wait=False to ensure fast cleanup."""
+
+    module = load_module()
+    request = module.parse_event(
+        event("@cwl-noema-review @opencode-agent")
+    )
+    assert request is not None
+    client = FakeClient()
+
+    import concurrent.futures
+    import threading
+
+    shutdown_called_with_no_wait = False
+
+    class MockExecutor(concurrent.futures.ThreadPoolExecutor):
+        def shutdown(self, wait=True, cancel_futures=False):
+            nonlocal shutdown_called_with_no_wait
+            if not wait and cancel_futures:
+                shutdown_called_with_no_wait = True
+            super().shutdown(wait=wait, cancel_futures=cancel_futures)
+
+    monkeypatch.setattr(concurrent.futures, "ThreadPoolExecutor", MockExecutor)
+
+    observed = module.dispatched_agents(request, client)
+
+    assert observed == frozenset()
+    assert shutdown_called_with_no_wait
+
+
 def test_dispatched_agents_single_candidate_skips_thread_pool() -> None:
     """Exactly one uncached agent stays on the plain sequential path."""
 

tests/test_agent_mention_sweep.py

@@ -511,6 +511,75 @@ def recording_build_requests(client, *, issue, since):
     assert "time budget" in capsys.readouterr().out
 
 
+def test_list_recent_pull_requests_shutdown_behavior(monkeypatch) -> None:
+    """The generator correctly invokes executor.shutdown(wait=False, cancel_futures=True) on cleanup."""
+
+    sweep = module()
+    client = FakeClient()
+    monkeypatch.setattr(
+        sweep, "list_accessible_repositories", lambda *args, **kwargs: ["ContextualWisdomLab/repo"]
+    )
+
+    import concurrent.futures
+    import threading
+    shutdown_called_with_no_wait = False
+
+    # We need a latch to ensure the worker starts running before we close.
+    worker_started = threading.Event()
+    worker_can_finish = threading.Event()
+
+    def fake_request(*args, **kwargs):
+        worker_started.set()
+        worker_can_finish.wait(timeout=5)
+        return [{"number": 1, "created_at": "2026-08-05T00:00:00Z", "updated_at": "2026-08-05T00:00:00Z"}]
+
+    monkeypatch.setattr(client, "request", fake_request)
+
+    class MockExecutor(concurrent.futures.ThreadPoolExecutor):
+        def shutdown(self, wait=True, cancel_futures=False):
+            nonlocal shutdown_called_with_no_wait
+            if not wait and cancel_futures:
+                shutdown_called_with_no_wait = True
+            super().shutdown(wait=wait, cancel_futures=cancel_futures)
+
+    monkeypatch.setattr(concurrent.futures, "ThreadPoolExecutor", MockExecutor)
+
+    gen = sweep.list_recent_pull_requests(
+        client,
+        organization="ContextualWisdomLab",
+        repository_source="organization",
+        since="2026-08-05T00:00:00Z",
+    )
+
+    # Prime the generator to start the executor and workers.
+    try:
+        next(gen)
+    except StopIteration:
+        pass
+
+    worker_started.wait(timeout=2)
+    # Now close the generator, which will trigger the finally block.
+    # The worker is still running and blocked on worker_can_finish,
+    # so if wait=True, close() would hang. Since wait=False, close()
+    # will return immediately.
+    import time
+    start = time.monotonic()
+    gen.close()
+    elapsed = time.monotonic() - start
+
+    # We must explicitly advance the generator (or let it close) properly
+    # to measure latency.
+
+    # Release the worker so the test suite can clean up.
+    worker_can_finish.set()
+
+    # We must explicitly advance the generator (or let it close) properly
+    # to measure latency.
+
+    assert shutdown_called_with_no_wait
+    assert elapsed < 1.0
+
+
 def test_sweep_time_budget_can_be_disabled(monkeypatch) -> None:
     """Passing None for the time budget preserves unbounded iteration."""