diff --git a/.jules/bolt.md b/.jules/bolt.md index a86b7aafd..78673a8eb 100644 --- a/.jules/bolt.md +++ b/.jules/bolt.md @@ -43,3 +43,6 @@ ## 2026-07-09 - Avoid N+1 API blocking in SBOM aggregator **Learning:** The `collect_inventories` function in `scripts/ci/sbom_inventory_aggregator.py` was fetching SBOMs from the GitHub dependency graph synchronously for every repository in the organization. For large organizations (up to 500 repos), this N+1 network/CLI bottleneck significantly stalled the aggregation workflow. **Action:** Use `concurrent.futures.ThreadPoolExecutor` to fetch SBOMs concurrently when multiple repositories are provided, bounded by a `max_workers` limit (e.g., 10) to avoid overwhelming the CLI/API, while preserving the fast serial path for single-item inputs. +## 2026-08-10 - Avoid N+1 API blocking in agent mention sweep +**Learning:** In `scripts/ci/agent_mention_sweep.py`, synchronously fetching recent pull requests for every active repository using `list_recent_pull_requests` causes an N+1 API bottleneck. For organizations with many repositories, this delays sweep operations. +**Action:** Use `concurrent.futures.ThreadPoolExecutor` with bounded `max_workers` to fetch multiple repositories concurrently, significantly speeding up the collection of candidate pull requests. diff --git a/scripts/ci/agent_mention_sweep.py b/scripts/ci/agent_mention_sweep.py index 9b64909a0..712e1b8a7 100644 --- 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 @@ -78,9 +80,7 @@ def flatten_pages( if not isinstance(collection, list): raise ValueError("paginated GitHub response is not a list") if not all(isinstance(record, dict) for record in collection): - raise ValueError( - "paginated GitHub response contains a non-object record" - ) + raise ValueError("paginated GitHub response contains a non-object record") records.extend(collection) return records @@ -155,10 +155,24 @@ def list_recent_pull_requests( organization=organization, repository_source=repository_source, ) - for repository in repositories: + + def fetch_repo_pulls( + repository: str, + stop_event: threading.Event | None = None, + ) -> tuple[list[dict[str, Any]], Exception | None]: + """Return one repository's bounded PR candidates and isolated failure. + + Collection stops before the next page when ``stop_event`` is set or + when the first item older than ``cutoff`` is reached. A non-positive + or non-integer pull-request number is returned as a ``ValueError`` for + serialized handling by the caller. + """ + repo_pulls: list[dict[str, Any]] = [] try: page = 1 while True: + if stop_event is not None and stop_event.is_set(): + break response = client.request( [ f"repos/{repository}/pulls", @@ -182,9 +196,7 @@ def list_recent_pull_requests( reached_cutoff = False for pull_request in pull_requests: if ( - parse_timestamp( - str(pull_request.get("updated_at") or "") - ) + parse_timestamp(str(pull_request.get("updated_at") or "")) < cutoff ): reached_cutoff = True @@ -194,23 +206,60 @@ def list_recent_pull_requests( raise ValueError( "GitHub returned an invalid pull request number" ) - yield { - "number": number, - "repository": repository, - "pull_request": { - "url": ( - "https://api.github.com/repos/" - f"{repository}/pulls/{number}" - ) - }, - } + repo_pulls.append( + { + "number": number, + "repository": repository, + "pull_request": { + "url": ( + "https://api.github.com/repos/" + f"{repository}/pulls/{number}" + ) + }, + } + ) if reached_cutoff or len(pull_requests) < 100: break page += 1 except Exception as exc: # noqa: BLE001 - repository isolation boundary + return repo_pulls, exc + return repo_pulls, None + + def emit_repo_result( + repository: str, + result: tuple[list[dict[str, Any]], Exception | None], + ) -> Iterator[dict[str, Any]]: + """Yield worker results and report failures serially on the caller thread.""" + repo_pulls, error = result + if error is not None: if on_error is None: - raise - on_error(repository, exc) + raise error + on_error(repository, error) + yield from repo_pulls + + if len(repositories) <= 1: + for repository in repositories: + yield from emit_repo_result(repository, fetch_repo_pulls(repository)) + else: + max_workers = min(10, len(repositories)) + stop_event = threading.Event() + executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) + futures = [ + executor.submit(fetch_repo_pulls, repository, stop_event) + for repository in repositories + ] + completed = False + try: + for repository, future in zip(repositories, futures, strict=True): + result = future.result() + yield from emit_repo_result(repository, result) + completed = True + finally: + if completed: + executor.shutdown(wait=True) + else: + stop_event.set() + executor.shutdown(wait=False, cancel_futures=True) def list_recent_comments( @@ -305,50 +354,54 @@ def record_failure(scope: str, error: Exception) -> None: counters.failures += 1 message = " ".join(str(error).split()) or error.__class__.__name__ - print( - f"::warning::Agent mention sweep skipped {scope}: {message[:1000]}" - ) + print(f"::warning::Agent mention sweep skipped {scope}: {message[:1000]}") - for issue in list_recent_pull_requests( + issues = list_recent_pull_requests( target_client, organization=organization, repository_source=repository_source, since=since, on_error=record_failure, - ): - issue_scope = f"{issue.get('repository')}#{issue.get('number')}" - try: - requests = build_requests_for_pull_request( - target_client, - issue=issue, - since=since, - ) - except Exception as exc: # noqa: BLE001 - pull-request isolation boundary - record_failure(issue_scope, exc) - continue - for request in requests: - request_scope = f"{issue_scope}/comment-{request.comment_id}" + ) + try: + for issue in issues: + issue_scope = f"{issue.get('repository')}#{issue.get('number')}" try: - queued_agents = dispatch_request( - request, - target_client=target_client, - dispatch_client=dispatch_client, - opencode_allowlist=opencode_allowlist, - dry_run=dry_run, - ledger_artifact_cache=ledger_artifact_cache, + requests = build_requests_for_pull_request( + target_client, + issue=issue, + since=since, ) - except Exception as exc: # noqa: BLE001 - request isolation boundary - record_failure(request_scope, exc) + except Exception as exc: # noqa: BLE001 - pull-request isolation boundary + record_failure(issue_scope, exc) continue - if not queued_agents: - continue - dispatched += 1 - if dispatched >= max_dispatches: - print( - "Agent mention sweep reached dispatch limit " - f"{max_dispatches}; isolated failures={counters.failures}." - ) - return dispatched + for request in requests: + request_scope = f"{issue_scope}/comment-{request.comment_id}" + try: + queued_agents = dispatch_request( + request, + target_client=target_client, + dispatch_client=dispatch_client, + opencode_allowlist=opencode_allowlist, + dry_run=dry_run, + ledger_artifact_cache=ledger_artifact_cache, + ) + except Exception as exc: # noqa: BLE001 - request isolation boundary + record_failure(request_scope, exc) + continue + if not queued_agents: + continue + dispatched += 1 + if dispatched >= max_dispatches: + print( + "Agent mention sweep reached dispatch limit " + f"{max_dispatches}; isolated failures={counters.failures}." + ) + return dispatched + finally: + close = getattr(issues, "close", None) + if close is not None: + close() print( "Agent mention sweep completed with " f"{dispatched} dispatch(es) and {counters.failures} isolated failure(s)." @@ -375,9 +428,7 @@ def main(argv: Sequence[str] | None = None) -> int: ) metrics = SweepMetrics() sweep( - target_client=GitHubClient( - os.environ.get("TARGET_REPOSITORY_TOKEN", "") - ), + target_client=GitHubClient(os.environ.get("TARGET_REPOSITORY_TOKEN", "")), dispatch_client=GitHubClient(os.environ.get("AGENT_DISPATCH_TOKEN", "")), organization=args.organization, repository_source=args.repository_source, diff --git a/tests/test_agent_mention_sweep.py b/tests/test_agent_mention_sweep.py index 0747bb02b..bc7ce1010 100644 --- a/tests/test_agent_mention_sweep.py +++ b/tests/test_agent_mention_sweep.py @@ -2,8 +2,10 @@ from __future__ import annotations +import concurrent.futures import importlib import sys +import threading from datetime import datetime, timezone from pathlib import Path @@ -128,9 +130,9 @@ def test_timestamp_cutoff_and_page_validation() -> None: {"a": 1}, {"b": 2}, ] - assert sweep.flatten_pages( - [{"items": [{"a": 1}]}], collection_key="items" - ) == [{"a": 1}] + assert sweep.flatten_pages([{"items": [{"a": 1}]}], collection_key="items") == [ + {"a": 1} + ] with pytest.raises(ValueError, match="empty"): sweep.flatten_pages(None) with pytest.raises(ValueError, match="page is not an object"): @@ -145,12 +147,14 @@ def test_accessible_repository_sources_filter_and_validate() -> None: """PAT and installation-token repository inventories are both supported.""" sweep = module() - organization_response = [[ - repository(), - repository("archived", archived=True), - repository("disabled", disabled=True), - repository("outside", owner="outside"), - ]] + organization_response = [ + [ + repository(), + repository("archived", archived=True), + repository("disabled", disabled=True), + repository("outside", owner="outside"), + ] + ] organization_client = FakeClient( {"orgs/ContextualWisdomLab/repos": organization_response} ) @@ -185,9 +189,9 @@ def test_accessible_repository_sources_filter_and_validate() -> None: ) invalid_client = FakeClient( { - "orgs/ContextualWisdomLab/repos": [[ - {**repository(), "full_name": "bad/name"} - ]] + "orgs/ContextualWisdomLab/repos": [ + [{**repository(), "full_name": "bad/name"}] + ] } ) with pytest.raises(ValueError, match="full_name"): @@ -205,10 +209,12 @@ def test_recent_pull_request_filtering() -> None: client = FakeClient( { "orgs/ContextualWisdomLab/repos": [[repository()]], - "repos/ContextualWisdomLab/example/pulls": [[ - pull_list_item(7, "2026-08-05T11:00:00Z"), - pull_list_item(8, "2026-08-04T11:59:59Z"), - ]], + "repos/ContextualWisdomLab/example/pulls": [ + [ + pull_list_item(7, "2026-08-05T11:00:00Z"), + pull_list_item(8, "2026-08-04T11:59:59Z"), + ] + ], } ) assert list( @@ -222,9 +228,9 @@ def test_recent_pull_request_filtering() -> None: bad_number_client = FakeClient( { "orgs/ContextualWisdomLab/repos": [[repository()]], - "repos/ContextualWisdomLab/example/pulls": [[ - {"number": 0, "updated_at": "2026-08-05T11:00:00Z"} - ]], + "repos/ContextualWisdomLab/example/pulls": [ + [{"number": 0, "updated_at": "2026-08-05T11:00:00Z"}] + ], } ) with pytest.raises(ValueError, match="pull request number"): @@ -368,6 +374,137 @@ def dispatch_new_work(request, **kwargs): ) +def test_sweep_closes_pull_iterator_at_dispatch_limit(monkeypatch) -> None: + """Reaching the work limit explicitly closes repository collection.""" + + sweep = module() + + class PullIterator: + """Expose whether the sweep closes its candidate source.""" + + def __init__(self) -> None: + self.closed = False + self.sent = False + + def __iter__(self): + return self + + def __next__(self): + if self.sent: + raise StopIteration + self.sent = True + return candidate() + + def close(self) -> None: + self.closed = True + + source = PullIterator() + request = mention_request(7, 10, "opencode-agent") + monkeypatch.setattr( + sweep, "list_recent_pull_requests", lambda *args, **kwargs: source + ) + monkeypatch.setattr( + sweep, + "build_requests_for_pull_request", + lambda *args, **kwargs: (request,), + ) + monkeypatch.setattr( + sweep, "dispatch_request", lambda *args, **kwargs: ("@opencode-agent",) + ) + + assert ( + sweep.sweep( + target_client=FakeClient(), + dispatch_client=FakeClient(), + organization="ContextualWisdomLab", + repository_source="organization", + lookback_hours=24, + max_dispatches=1, + opencode_allowlist=frozenset(), + now=datetime(2026, 8, 5, tzinfo=timezone.utc), + ) + == 1 + ) + assert source.closed is True + + +def test_closing_concurrent_collection_cancels_pending_repository_work( + monkeypatch, +) -> None: + """Closing a partial result stops running work and cancels queued futures.""" + + sweep = module() + + class FakeFuture: + """Run one submitted call on demand for deterministic scheduling.""" + + def __init__(self, function, arguments) -> None: + self.function = function + self.arguments = arguments + self.cancelled = False + self.executed = False + + def result(self): + self.executed = True + return self.function(*self.arguments) + + class FakeExecutor: + """Record the shutdown contract and model one running worker.""" + + instance = None + + def __init__(self, *, max_workers) -> None: + self.max_workers = max_workers + self.futures = [] + self.shutdown_call = None + FakeExecutor.instance = self + + def submit(self, function, *arguments): + future = FakeFuture(function, arguments) + self.futures.append(future) + return future + + def shutdown(self, *, wait, cancel_futures=False) -> None: + self.shutdown_call = (wait, cancel_futures) + if not wait: + # A task that was already running observes the shared stop + # signal before issuing its first page request. + self.futures[1].result() + for future in self.futures[2:]: + future.cancelled = True + + monkeypatch.setattr(concurrent.futures, "ThreadPoolExecutor", FakeExecutor) + client = FakeClient( + { + "orgs/ContextualWisdomLab/repos": [ + [repository("first"), repository("running"), repository("queued")] + ], + "repos/ContextualWisdomLab/first/pulls": [ + pull_list_item(1, "2026-08-05T11:00:00Z") + ], + } + ) + candidates = sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + ) + + assert next(candidates)["number"] == 1 + candidates.close() + + executor = FakeExecutor.instance + assert executor.max_workers == 3 + assert executor.shutdown_call == (False, True) + assert executor.futures[1].executed is True + assert executor.futures[2].cancelled is True + assert [call[0][0] for call in client.calls] == [ + "orgs/ContextualWisdomLab/repos", + "repos/ContextualWisdomLab/first/pulls", + ] + + def test_sweep_noops_do_not_starve_new_mentions_across_repeated_runs( monkeypatch, ) -> None: @@ -494,3 +631,256 @@ def test_main_constructs_clients_and_forwards_options(monkeypatch) -> None: assert captured[0]["lookback_hours"] == 48 assert captured[0]["max_dispatches"] == 3 assert captured[0]["dry_run"] is True + + +def test_list_recent_pull_requests_concurrent_multiple_repositories(): + """Testing N+1 API block solution branch where multiple repositories are checked concurrently.""" + sweep = module() + + class SmartFakeClient(FakeClient): + def __init__(self, responses, explode_repos=False): + super().__init__(responses) + self.explode_repos = explode_repos + + def request(self, args, *, input_payload=None): + if self.explode_repos and args[0].startswith("repos/"): + raise ValueError("Simulated Exception") + return super().request(args, input_payload=input_payload) + + client = SmartFakeClient( + { + "orgs/ContextualWisdomLab/repos": [ + [ + repository(name="example1"), + repository(name="example2"), + ] + ], + "repos/ContextualWisdomLab/example1/pulls": [ + [pull_list_item(number=1, updated_at="2026-08-05T11:00:00Z")] + ], + "repos/ContextualWisdomLab/example2/pulls": [ + [pull_list_item(number=2, updated_at="2026-08-05T11:00:00Z")] + ], + } + ) + pulls = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + ) + ) + assert len(pulls) == 2 + assert {p["number"] for p in pulls} == {1, 2} + + # Test error handling logic inside fetch_repo_pulls + error_client = SmartFakeClient( + { + "orgs/ContextualWisdomLab/repos": [ + [ + repository(name="example1"), + repository(name="example2"), + ] + ], + }, + explode_repos=True, + ) + + metrics = sweep.SweepMetrics() + dispatched = sweep.sweep( + target_client=error_client, + dispatch_client=error_client, + organization="ContextualWisdomLab", + repository_source="organization", + lookback_hours=1, + max_dispatches=10, + opencode_allowlist=frozenset(), + dry_run=True, + now=datetime(2026, 8, 5, 11, 0, tzinfo=timezone.utc), + metrics=metrics, + ) + assert dispatched == 0 + assert metrics.failures == 2 # One failure per repository + + +def test_concurrent_repository_errors_are_reported_serially_in_source_order() -> None: + """Worker failures reach the caller callback in repository order on its thread.""" + sweep = module() + caller_thread = threading.get_ident() + barrier = threading.Barrier(2) + + class ConcurrentFailureClient(FakeClient): + def request(self, args, *, input_payload=None): + if args[0] == "orgs/ContextualWisdomLab/repos": + return [[repository("first"), repository("second")]] + barrier.wait(timeout=2) + raise ValueError(args[0]) + + failures = [] + assert list( + sweep.list_recent_pull_requests( + ConcurrentFailureClient(), + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + on_error=lambda name, error: failures.append( + (name, str(error), threading.get_ident()) + ), + ) + ) == [] + + assert [name for name, _error, _thread in failures] == [ + "ContextualWisdomLab/first", + "ContextualWisdomLab/second", + ] + assert {thread for _name, _error, thread in failures} == {caller_thread} + + +def test_sweep_failures_during_processing(monkeypatch, capsys) -> None: + """Sweep gracefully isolates exceptions during request building and dispatching.""" + sweep = module() + client = FakeClient( + { + "orgs/ContextualWisdomLab/repos": [[repository()]], + "repos/ContextualWisdomLab/example/pulls": [ + [pull_list_item(number=1), pull_list_item(number=2)] + ], + } + ) + + def fail_build(*args, **kwargs): + if kwargs.get("issue", {}).get("number") == 1: + raise ValueError("Build Error") + return [mention_request(2, 20, "opencode-agent")] + + monkeypatch.setattr(sweep, "build_requests_for_pull_request", fail_build) + + def fail_dispatch(*args, **kwargs): + raise ValueError("Dispatch Error") + + monkeypatch.setattr(sweep, "dispatch_request", fail_dispatch) + + metrics = sweep.SweepMetrics() + sweep.sweep( + target_client=client, + dispatch_client=client, + organization="ContextualWisdomLab", + repository_source="organization", + lookback_hours=1, + max_dispatches=10, + opencode_allowlist=frozenset(), + now=datetime(2026, 8, 5, 11, 0, tzinfo=timezone.utc), + metrics=metrics, + ) + assert metrics.failures == 2 # One for PR 1 build fail, one for PR 2 dispatch fail + + +def test_pagination_break_without_cutoff() -> None: + """Pagination terminates gracefully when less than 100 items are returned.""" + sweep = module() + client = FakeClient( + { + "orgs/ContextualWisdomLab/repos": [[repository()]], + "repos/ContextualWisdomLab/example/pulls": [ + [pull_list_item(number=1, updated_at="2026-08-05T11:00:00Z")], + [], # Page 2 returns empty list + ], + } + ) + pulls = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + ) + ) + assert len(pulls) == 1 + + +def test_pagination_empty_page_loop_break() -> None: + """Pagination terminates when an empty list of pull requests is returned without exceptions.""" + sweep = module() + client = FakeClient( + { + "orgs/ContextualWisdomLab/repos": [[repository()]], + "repos/ContextualWisdomLab/example/pulls": [[]], + } + ) + pulls = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + ) + ) + assert len(pulls) == 0 + + +def test_accessible_repository_invalid_repo_name() -> None: + """Accessible repository invalid name check.""" + sweep = module() + organization_response = [ + [ + { + "full_name": "ContextualWisdomLab/invalid name with spaces", + "owner": {"login": "ContextualWisdomLab"}, + "archived": False, + "disabled": False, + } + ] + ] + client = FakeClient({"orgs/ContextualWisdomLab/repos": organization_response}) + import pytest + + with pytest.raises(ValueError, match="invalid repository full_name"): + sweep.list_accessible_repositories( + client, organization="ContextualWisdomLab", repository_source="organization" + ) + + +def test_flatten_pages_direct_list_return(): + """Flatten pages handles direct list of dicts properly.""" + sweep = module() + direct_list = [{"foo": "bar"}, {"baz": "qux"}] + assert sweep.flatten_pages(direct_list) == direct_list + + +def test_list_recent_pull_requests_multiple_pages(): + """Pagination fetches multiple pages correctly without errors.""" + sweep = module() + client = FakeClient( + { + "orgs/ContextualWisdomLab/repos": [[repository()]], + "repos/ContextualWisdomLab/example/pulls": [ + [ + pull_list_item(number=i, updated_at="2026-08-05T11:00:00Z") + for i in range(1, 101) + ], + [pull_list_item(number=101, updated_at="2026-08-05T11:00:00Z")], + [], + ], + } + ) + + def request(args, *, input_payload=None): + if "repos/ContextualWisdomLab/example/pulls" in args[0]: + page = int( + next(arg for arg in args if arg.startswith("page=")).split("=")[1] + ) + return client.responses["repos/ContextualWisdomLab/example/pulls"][page - 1] + return client.responses.get(args[0]) + + client.request = request + + pulls = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T10:00:00Z", + ) + ) + assert len(pulls) == 101