diff --git a/docs/automation/review-agent-comment-invocation.md b/docs/automation/review-agent-comment-invocation.md index 3d2ca496d..cc8f8c58c 100644 --- a/docs/automation/review-agent-comment-invocation.md +++ b/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. diff --git a/scripts/ci/agent_mention_router.py b/scripts/ci/agent_mention_router.py old mode 100644 new mode 100755 index 2b5453139..77d19bb40 --- a/scripts/ci/agent_mention_router.py +++ b/scripts/ci/agent_mention_router.py @@ -34,6 +34,7 @@ ACTOR_RE = re.compile(r"^[A-Za-z0-9-]+$") RECEIPT_RE = re.compile(r"") 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 diff --git a/scripts/ci/agent_mention_sweep.py b/scripts/ci/agent_mention_sweep.py old mode 100644 new mode 100755 index 9b64909a0..315d0b519 --- a/scripts/ci/agent_mention_sweep.py +++ b/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,8 +149,9 @@ 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( @@ -155,46 +159,48 @@ def list_recent_pull_requests( 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: diff --git a/tests/test_agent_mention_acknowledgement_recovery.py b/tests/test_agent_mention_acknowledgement_recovery.py new file mode 100644 index 000000000..840e69a06 --- /dev/null +++ b/tests/test_agent_mention_acknowledgement_recovery.py @@ -0,0 +1,156 @@ +"""Regression tests for post-dispatch mention acknowledgement recovery.""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import ModuleType + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +MODULE_PATH = ROOT / "scripts" / "ci" / "agent_mention_router.py" + + +def load_module() -> ModuleType: + """Load the central mention router from its script path.""" + + module_name = "agent_mention_router_acknowledgement_recovery" + spec = importlib.util.spec_from_file_location(module_name, MODULE_PATH) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[module_name] = module + spec.loader.exec_module(module) + return module + + +def request(module: ModuleType): + """Build one exact trusted OpenCode mention request.""" + + return module.MentionRequest( + repository="ContextualWisdomLab/.github", + pull_request_number=1099, + pull_request_head_sha="a" * 40, + pull_request_base_branch="main", + comment_id=91, + actor="maintainer", + agents=("opencode-agent",), + pull_request_base_sha="b" * 40, + ) + + +class FakeClient: + """Capture API traffic while simulating ledger and UX failures.""" + + def __init__( + self, + *, + existing_claim: bool = False, + fail_reaction: bool = False, + fail_comment: bool = False, + ) -> None: + """Initialize deterministic response and failure controls.""" + + self.existing_claim = existing_claim + self.fail_reaction = fail_reaction + self.fail_comment = fail_comment + self.calls: list[tuple[list[str], dict | None]] = [] + + def request(self, args, *, input_payload=None): + """Record one request and return or raise the configured outcome.""" + + arguments = list(args) + self.calls.append((arguments, input_payload)) + endpoint = arguments[0] + if endpoint.endswith("/actions/artifacts"): + if not self.existing_claim: + return {"total_count": 0, "artifacts": []} + name = next( + value.removeprefix("name=") + for value in arguments + if value.startswith("name=") + ) + return { + "total_count": 1, + "artifacts": [{"id": 17, "name": name, "expired": False}], + } + if endpoint.endswith("/reactions") and self.fail_reaction: + raise RuntimeError("Resource not accessible by integration (HTTP 403)") + if endpoint.endswith("/issues/1099/comments") and self.fail_comment: + raise RuntimeError("comment publication failed") + return None + + +def dispatch_mutations(client: FakeClient) -> list[tuple[list[str], dict | None]]: + """Return only repository-dispatch mutation calls.""" + + return [call for call in client.calls if call[0][0].endswith("/dispatches")] + + +def acknowledgement_comments(client: FakeClient) -> list[dict]: + """Return published target-PR acknowledgement payloads.""" + + return [ + payload + for args, payload in client.calls + if args[0].endswith("/issues/1099/comments") and payload is not None + ] + + +def test_existing_durable_claim_heals_missing_acknowledgement() -> None: + """A ledgered invocation is acknowledged without a duplicate dispatch.""" + + module = load_module() + central = FakeClient(existing_claim=True) + target = FakeClient() + + assert module.dispatch_request( + request(module), + target_client=target, + dispatch_client=central, + opencode_allowlist=frozenset({"ContextualWisdomLab/.github"}), + ) == () + + assert dispatch_mutations(central) == [] + comments = acknowledgement_comments(target) + assert len(comments) == 1 + assert "Already queued @opencode-agent on this exact request" in comments[0]["body"] + assert "cwl-agent-mention-receipt:91" in comments[0]["body"] + + +def test_reaction_failure_does_not_hide_successful_dispatch(capsys) -> None: + """A cosmetic reaction 403 cannot suppress the durable acknowledgement.""" + + module = load_module() + central = FakeClient() + target = FakeClient(fail_reaction=True) + + assert module.dispatch_request( + request(module), + target_client=target, + dispatch_client=central, + opencode_allowlist=frozenset({"ContextualWisdomLab/.github"}), + ) == ("@opencode-agent",) + + assert len(dispatch_mutations(central)) == 1 + assert len(acknowledgement_comments(target)) == 1 + assert "::warning::" in capsys.readouterr().out + + +def test_acknowledgement_comment_failure_remains_visible() -> None: + """A missing durable receipt still fails so a later sweep can repair it.""" + + module = load_module() + central = FakeClient() + target = FakeClient(fail_comment=True) + + with pytest.raises(RuntimeError, match="comment publication failed"): + module.dispatch_request( + request(module), + target_client=target, + dispatch_client=central, + opencode_allowlist=frozenset({"ContextualWisdomLab/.github"}), + ) + + assert len(dispatch_mutations(central)) == 1 diff --git a/tests/test_agent_mention_downstream_idempotency.py b/tests/test_agent_mention_downstream_idempotency.py index 4fc40a782..23634f293 100644 --- a/tests/test_agent_mention_downstream_idempotency.py +++ b/tests/test_agent_mention_downstream_idempotency.py @@ -34,6 +34,7 @@ def test_downstream_workflows_claim_artifacts_and_bind_exact_key() -> None: assert "requested_agent" in text assert "cancel-in-progress: false" in text assert "queue: max" in text + assert "cancel-in-progress: true" not in text assert "^[0-9a-f]{64}$" in text assert "^[1-9][0-9]*$" in text assert "actions/artifacts" in text diff --git a/tests/test_agent_mention_idempotency.py b/tests/test_agent_mention_idempotency.py index 499730a22..fc112116d 100644 --- a/tests/test_agent_mention_idempotency.py +++ b/tests/test_agent_mention_idempotency.py @@ -320,13 +320,12 @@ def test_reaction_or_ack_failure_cannot_redispatch_completed_agents() -> None: mention_request = request(module) central = ArtifactAwareClient() failing_target = ArtifactAwareClient(fail_target_call=1) - with pytest.raises(RuntimeError, match="target call"): - module.dispatch_request( - mention_request, - target_client=failing_target, - dispatch_client=central, - opencode_allowlist=frozenset({mention_request.repository}), - ) + assert module.dispatch_request( + mention_request, + target_client=failing_target, + dispatch_client=central, + opencode_allowlist=frozenset({mention_request.repository}), + ) == ("@cwl-noema-review", "@opencode-agent") assert dispatch_events(central) == [ "agent-mention-noema", "agent-mention-opencode", @@ -348,4 +347,4 @@ def test_reaction_or_ack_failure_cannot_redispatch_completed_agents() -> None: opencode_allowlist=frozenset({mention_request.repository}), ) == () assert dispatch_events(retry) == [] - assert retry_target.calls == [] + assert len(retry_target.calls) == 2 diff --git a/tests/test_agent_mention_rejection_idempotency.py b/tests/test_agent_mention_rejection_idempotency.py index 843454f3d..2e12e3867 100644 --- a/tests/test_agent_mention_rejection_idempotency.py +++ b/tests/test_agent_mention_rejection_idempotency.py @@ -64,3 +64,29 @@ def test_rejected_only_request_is_mutation_free() -> None: ) == () assert target.calls == [] assert central.calls == [] + + +def test_empty_request_is_mutation_free() -> None: + """An already-filtered request does not emit a rejection or mutate GitHub.""" + + module = load_module() + request = module.MentionRequest( + "ContextualWisdomLab/example", + 17, + "a" * 40, + "main", + 91, + "maintainer", + (), + ) + target = FakeClient() + central = FakeClient() + + assert module.dispatch_request( + request, + target_client=target, + dispatch_client=central, + opencode_allowlist=frozenset(), + ) == () + assert target.calls == [] + assert central.calls == [] diff --git a/tests/test_agent_mention_sweep_regressions.py b/tests/test_agent_mention_sweep_regressions.py index d9c0c4f2a..643f562cc 100644 --- a/tests/test_agent_mention_sweep_regressions.py +++ b/tests/test_agent_mention_sweep_regressions.py @@ -94,6 +94,116 @@ def test_pull_pagination_stops_at_cutoff_without_loading_later_pages() -> None: assert sweep.flatten_pages([{"number": 1}]) == [{"number": 1}] +def test_recent_pull_requests_use_bounded_parallel_repository_fetches(monkeypatch) -> None: + """Repository fetches are parallel but results remain repository ordered.""" + + sweep = module() + client = PagingClient( + { + ("orgs/ContextualWisdomLab/repos", 1): [[ + repository("first"), + repository("second"), + ]], + ("repos/ContextualWisdomLab/first/pulls", 1): [pull(1)], + ("repos/ContextualWisdomLab/second/pulls", 1): [pull(2)], + } + ) + worker_limits = [] + real_executor = sweep.concurrent.futures.ThreadPoolExecutor + + def recording_executor(*, max_workers): + worker_limits.append(max_workers) + return real_executor(max_workers=max_workers) + + monkeypatch.setattr( + sweep.concurrent.futures, + "ThreadPoolExecutor", + recording_executor, + ) + results = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + ) + ) + assert [result["repository"] for result in results] == [ + "ContextualWisdomLab/first", + "ContextualWisdomLab/second", + ] + assert worker_limits == [2] + + +def test_repeated_sweeps_rotate_repository_dispatch_frontier(monkeypatch) -> None: + """Five-minute sweeps do not starve later repositories at the limit.""" + + sweep = module() + client = PagingClient( + { + ("orgs/ContextualWisdomLab/repos", 1): [[ + repository("first"), + repository("second"), + ]], + ("repos/ContextualWisdomLab/first/pulls", 1): [pull(1)], + ("repos/ContextualWisdomLab/second/pulls", 1): [pull(2)], + } + ) + processed = [] + monkeypatch.setattr( + sweep, + "build_requests_for_pull_request", + lambda *args, issue, **kwargs: processed.append(issue["repository"]) + or (mention_request(10),), + ) + monkeypatch.setattr( + sweep, + "dispatch_request", + lambda *args, **kwargs: ("@cwl-noema-review",), + ) + common = { + "target_client": client, + "dispatch_client": object(), + "organization": "ContextualWisdomLab", + "repository_source": "organization", + "lookback_hours": 24, + "max_dispatches": 1, + "opencode_allowlist": frozenset(), + } + assert ( + sweep.sweep( + **common, now=datetime(2026, 8, 6, 0, 0, tzinfo=timezone.utc) + ) + == 1 + ) + assert ( + sweep.sweep( + **common, now=datetime(2026, 8, 6, 0, 5, tzinfo=timezone.utc) + ) + == 1 + ) + assert len(processed) == 2 + assert {name.rsplit("/", 1)[-1] for name in processed} == { + "first", + "second", + } + + +def test_recent_pull_requests_skip_executor_when_no_repositories() -> None: + """An empty organization inventory does not create worker threads.""" + + sweep = module() + client = PagingClient({("orgs/ContextualWisdomLab/repos", 1): []}) + assert list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + ) + ) == [] + + def test_pull_pagination_stops_on_empty_followup_page() -> None: """A full page followed by an empty page terminates without page three.""" diff --git a/tests/test_agent_mention_timeout_bounds.py b/tests/test_agent_mention_timeout_bounds.py new file mode 100644 index 000000000..3035a177e --- /dev/null +++ b/tests/test_agent_mention_timeout_bounds.py @@ -0,0 +1,226 @@ +"""Bounded GitHub subprocess and repository-fanout regression tests.""" + +from __future__ import annotations + +import importlib +import subprocess +import sys +import threading +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +SCRIPTS = ROOT / "scripts" / "ci" +sys.path.insert(0, str(SCRIPTS)) + + +def router_module(): + """Reload the central mention router for isolated monkeypatching.""" + + return importlib.reload(importlib.import_module("agent_mention_router")) + + +def sweep_module(): + """Reload the organization sweep for isolated monkeypatching.""" + + router_module() + return importlib.reload(importlib.import_module("agent_mention_sweep")) + + +def repository(name: str) -> dict: + """Return one active repository record.""" + + return { + "full_name": f"ContextualWisdomLab/{name}", + "owner": {"login": "ContextualWisdomLab"}, + "archived": False, + "disabled": False, + } + + +def pull(number: int) -> dict: + """Return one recent pull-request list record.""" + + return {"number": number, "updated_at": "2026-08-20T00:00:00Z"} + + +class InventoryClient: + """Serve a deterministic repository inventory and one pull per repository.""" + + def __init__(self, names: tuple[str, ...]) -> None: + """Store the repository names exposed to the sweep.""" + + self.names = names + + def request(self, args, *, input_payload=None): + """Return the organization inventory or one repository pull list.""" + + del input_payload + endpoint = args[0] + if endpoint == "orgs/ContextualWisdomLab/repos": + return [repository(name) for name in self.names] + name = endpoint.split("/")[2] + return [pull(self.names.index(name) + 1)] + + +def test_github_client_applies_one_finite_timeout(monkeypatch) -> None: + """Every ``gh api`` subprocess receives the reviewed timeout bound.""" + + router = router_module() + observed = [] + + def fake_run(command, **kwargs): + observed.append((command, kwargs)) + return SimpleNamespace(stdout='{"ok": true}\n', returncode=0) + + monkeypatch.setattr(router.subprocess, "run", fake_run) + result = router.GitHubClient("token").request(["repos/x/y"]) + + assert result == {"ok": True} + assert observed[0][1]["timeout"] == router.GITHUB_API_TIMEOUT_SECONDS == 30 + + +def test_github_client_converts_timeout_to_bounded_diagnostic(monkeypatch) -> None: + """A hung CLI request fails visibly without leaking token or payload data.""" + + router = router_module() + + def timeout_run(command, **kwargs): + raise subprocess.TimeoutExpired(command, kwargs["timeout"]) + + monkeypatch.setattr(router.subprocess, "run", timeout_run) + with pytest.raises(RuntimeError, match="gh api timed out after 30 seconds"): + router.GitHubClient("secret-token").request( + ["repos/x/y"], + input_payload={"sensitive": "value"}, + ) + + +def test_repository_fanout_uses_exactly_four_workers_at_scale(monkeypatch) -> None: + """Five repositories exercise the fixed four-worker production ceiling.""" + + sweep = sweep_module() + names = ("alpha", "bravo", "charlie", "delta", "echo") + real_executor = sweep.concurrent.futures.ThreadPoolExecutor + worker_limits = [] + + def recording_executor(*, max_workers): + worker_limits.append(max_workers) + return real_executor(max_workers=max_workers) + + monkeypatch.setattr( + sweep.concurrent.futures, + "ThreadPoolExecutor", + recording_executor, + ) + results = list( + sweep.list_recent_pull_requests( + InventoryClient(names), + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-19T00:00:00Z", + ) + ) + + assert worker_limits == [4] + assert [item["repository"] for item in results] == [ + f"ContextualWisdomLab/{name}" for name in names + ] + + +def test_empty_inventory_does_not_construct_an_executor(monkeypatch) -> None: + """The zero-repository fast path never allocates worker threads.""" + + sweep = sweep_module() + + def forbidden_executor(*args, **kwargs): + raise AssertionError(f"executor called with {args!r} {kwargs!r}") + + monkeypatch.setattr( + sweep.concurrent.futures, + "ThreadPoolExecutor", + forbidden_executor, + ) + assert list( + sweep.list_recent_pull_requests( + InventoryClient(()), + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-19T00:00:00Z", + ) + ) == [] + + +def test_generator_close_stops_additional_pages_after_inflight_request( + monkeypatch, +) -> None: + """Closing after the dispatch frontier bounds a running repository fetch.""" + + sweep = sweep_module() + page_two_started = threading.Event() + release_page_two = threading.Event() + shutdown_started = threading.Event() + + class ClosingClient: + """Keep the second repository in one bounded in-flight request.""" + + def request(self, args, *, input_payload=None): + del input_payload + endpoint = args[0] + if endpoint == "orgs/ContextualWisdomLab/repos": + return [repository("alpha"), repository("bravo")] + page = 1 + for index, value in enumerate(args[:-1]): + if value == "-f" and args[index + 1].startswith("page="): + page = int(args[index + 1].split("=", 1)[1]) + if endpoint.endswith("alpha/pulls"): + return [pull(1)] + if page == 1: + return [pull(number) for number in range(100, 200)] + if page == 2: + page_two_started.set() + assert release_page_two.wait(2) + return [pull(number) for number in range(200, 300)] + raise AssertionError(f"unexpected third page request: {args!r}") + + real_executor = sweep.concurrent.futures.ThreadPoolExecutor + + class RecordingExecutor: + """Expose the moment shutdown begins while delegating real workers.""" + + def __init__(self, *, max_workers): + self._inner = real_executor(max_workers=max_workers) + + def submit(self, *args, **kwargs): + return self._inner.submit(*args, **kwargs) + + def shutdown(self, *, wait, cancel_futures): + shutdown_started.set() + return self._inner.shutdown( + wait=wait, + cancel_futures=cancel_futures, + ) + + monkeypatch.setattr( + sweep.concurrent.futures, + "ThreadPoolExecutor", + RecordingExecutor, + ) + generator = sweep.list_recent_pull_requests( + ClosingClient(), + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-19T00:00:00Z", + ) + assert next(generator)["repository"] == "ContextualWisdomLab/alpha" + assert page_two_started.wait(2) + + closer = threading.Thread(target=generator.close) + closer.start() + assert shutdown_started.wait(2) + release_page_two.set() + closer.join(2) + + assert not closer.is_alive()