diff --git a/.jules/bolt.md b/.jules/bolt.md index 420e6d7e2..b0dc63ad0 100644 --- a/.jules/bolt.md +++ b/.jules/bolt.md @@ -47,3 +47,6 @@ ## 2026-08-09 - [대용량 로그 스캔 시 정규표현식 실행 전 O(N) 서브스트링 검증 선행] **Learning:** `classify_testthat_failure`에서 테스트 실패 내역이 없는 2MB 로그 파일을 대상으로 정규표현식을 실행하면 약 20ms가 소요되지만, 단순 문자열 검색은 약 1ms만 소요됩니다. 문자열 존재 여부가 정규표현식 매칭의 전제 조건일 때, 콜드 패스(Cold Path)에서 순서 최적화는 매우 큰 성능 차이를 만듭니다. **Action:** 대용량 텍스트 입력(CI 로그 등)에서 복잡한 정규표현식을 파싱하기 전에 항상 빠른 O(N) 문자열 존재 여부 확인을 먼저 수행하십시오. +## 2026-08-18 - Concurrently yield items from an executor +**Learning:** `ThreadPoolExecutor.shutdown(wait=False, cancel_futures=True)` cancels pending futures but cannot interrupt a running external request. Generator shutdown latency therefore depends on every transport call having a finite request timeout. Worker callbacks must also avoid mutating caller-owned state concurrently. +**Action:** Keep the executor's non-waiting shutdown in a `finally` block, enforce and translate a finite `gh api` timeout in `GitHubClient.request`, and report repository failures from the generator thread rather than worker threads. diff --git a/CHANGELOG.md b/CHANGELOG.md index 872968b36..c30ca21db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,7 @@ Semantic Versioning where the repository publishes a release. ### Changed +- Bound concurrent agent-mention API calls with a finite GitHub CLI timeout, report worker failures from the generator thread, preserve independent mention and sweep queues, and enforce the repository-dispatch ten-key limit while keeping OpenCode review-only flags wrapper-owned. - 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 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. - Run the bounded DiskSage repair heartbeat at minute 37 of every hour, dispatch no more than one exact-head repair, and wait two hours before redispatching an unchanged head so legitimate OpenCode or NVIDIA NIM latency does not create duplicate writers. diff --git a/scripts/ci/agent_mention_router.py b/scripts/ci/agent_mention_router.py index 2b5453139..a475b295a 100644 --- a/scripts/ci/agent_mention_router.py +++ b/scripts/ci/agent_mention_router.py @@ -53,12 +53,13 @@ class MentionRequest: class GitHubClient: """Small token-bound wrapper around ``gh api`` for JSON requests.""" - def __init__(self, token: str) -> None: - """Initialize a client with one non-empty GitHub credential.""" + def __init__(self, token: str, *, timeout_seconds: int = 60) -> None: + """Initialize a client with one credential and request timeout.""" if not token: raise ValueError("GitHub token is required") self._token = token + self._timeout_seconds = timeout_seconds def request( self, @@ -73,14 +74,21 @@ def request( 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=self._timeout_seconds, + ) + except subprocess.TimeoutExpired as exc: + raise RuntimeError( + f"gh api timed out after {self._timeout_seconds} seconds" + ) from exc return_code = int(getattr(completed, "returncode", 0)) if return_code: diagnostic = " ".join( @@ -374,12 +382,7 @@ def repository_dispatch_body( event_type: str, client_payload: dict[str, Any], ) -> dict[str, Any]: - """Return a repository_dispatch body within GitHub's 10-key payload limit. - - GitHub's create-repository-dispatch endpoint accepts at most 10 top-level - ``client_payload`` properties. A larger object is rejected with HTTP 422, - so mention routing cannot enqueue a review. - """ + """Return a repository-dispatch body within GitHub's 10-key payload limit.""" if len(client_payload) > REPOSITORY_DISPATCH_CLIENT_PAYLOAD_MAX_KEYS: raise ValueError( @@ -387,10 +390,7 @@ def repository_dispatch_body( f"{len(client_payload)} keys; GitHub allows at most " f"{REPOSITORY_DISPATCH_CLIENT_PAYLOAD_MAX_KEYS}" ) - return { - "event_type": event_type, - "client_payload": client_payload, - } + return {"event_type": event_type, "client_payload": client_payload} def noema_payload(request: MentionRequest) -> dict[str, Any]: diff --git a/scripts/ci/agent_mention_sweep.py b/scripts/ci/agent_mention_sweep.py index 9b64909a0..924de4f95 100644 --- a/scripts/ci/agent_mention_sweep.py +++ b/scripts/ci/agent_mention_sweep.py @@ -4,6 +4,7 @@ from __future__ import annotations import argparse +import concurrent.futures import os import re from dataclasses import dataclass @@ -139,6 +140,65 @@ def list_accessible_repositories( return sorted(set(names)) +def _fetch_repo_recent_pull_requests( + client: GitHubClient, + repository: str, + cutoff: datetime, +) -> list[dict[str, Any]]: + """Fetch recent open pull requests for a single repository.""" + results: list[dict[str, Any]] = [] + 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: + 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" + ) + results.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 + return results + def list_recent_pull_requests( client: GitHubClient, *, @@ -155,62 +215,35 @@ 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: - 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": 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 - if on_error is None: - raise - on_error(repository, exc) + + if not repositories: # pragma: no cover + return + + executor = concurrent.futures.ThreadPoolExecutor( + max_workers=min(10, len(repositories)) + ) + try: + future_repositories = { + executor.submit( + _fetch_repo_recent_pull_requests, + client, + repository, + cutoff, + ): repository + for repository in repositories + } + for future in concurrent.futures.as_completed(future_repositories): + repository = future_repositories[future] + try: + pull_requests = future.result() + except Exception as exc: # noqa: BLE001 - repository isolation boundary + if on_error is None: + raise + on_error(repository, exc) + continue + yield from pull_requests + finally: + executor.shutdown(wait=False, cancel_futures=True) def list_recent_comments( diff --git a/tests/test_agent_mention_concurrency_regressions.py b/tests/test_agent_mention_concurrency_regressions.py new file mode 100644 index 000000000..10cd1481f --- /dev/null +++ b/tests/test_agent_mention_concurrency_regressions.py @@ -0,0 +1,244 @@ +"""Concurrency and transport regressions for the agent-mention sweep.""" + +from __future__ import annotations + +import importlib +import sys +import threading +import time +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 router module for isolated subprocess monkeypatching.""" + + return importlib.reload(importlib.import_module("agent_mention_router")) + + +def sweep_module(): + """Reload the sweep module for isolated executor tests.""" + + return importlib.reload(importlib.import_module("agent_mention_sweep")) + + +def repository(name: str) -> dict: + """Build one active organization repository record.""" + + return { + "full_name": f"ContextualWisdomLab/{name}", + "owner": {"login": "ContextualWisdomLab"}, + "archived": False, + "disabled": False, + } + + +def pull(number: int, updated_at: str = "2026-08-06T11:00:00Z") -> dict: + """Build one pull-list response record.""" + + return {"number": number, "updated_at": updated_at} + + +class PagingClient: + """Serve endpoint/page responses and record request order.""" + + def __init__(self, responses) -> None: + """Initialize an endpoint/page response map.""" + + self.responses = responses + self.calls: list[list[str]] = [] + + def request(self, args, *, input_payload=None): + """Return a configured page or raise its configured exception.""" + + del input_payload + args = list(args) + self.calls.append(args) + endpoint = args[0] + 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]) + response = self.responses[(endpoint, page)] + if isinstance(response, Exception): + raise response + return response + + +def test_github_client_request_is_timeout_bounded(monkeypatch) -> None: + """Every gh subprocess has a finite timeout with an explicit diagnostic.""" + + router = router_module() + + def time_out(command, **kwargs): + timeout = kwargs.get("timeout") + assert timeout == 60 + raise router.subprocess.TimeoutExpired(command, timeout) + + monkeypatch.setattr(router.subprocess, "run", time_out) + + with pytest.raises(RuntimeError, match="timed out after 60 seconds"): + router.GitHubClient("token").request(["repos/x/y"]) + + +def test_rate_limit_failure_is_one_bounded_request(monkeypatch) -> None: + """Rate-limit diagnostics remain isolated without automatic retry amplification.""" + + router = router_module() + calls = [] + + def rate_limited(command, **kwargs): + calls.append((command, kwargs)) + return SimpleNamespace( + stdout="", + stderr="API rate limit exceeded\n", + returncode=1, + ) + + monkeypatch.setattr(router.subprocess, "run", rate_limited) + + with pytest.raises(RuntimeError, match="API rate limit exceeded"): + router.GitHubClient("token").request(["repos/x/y"]) + + assert len(calls) == 1 + assert calls[0][1]["timeout"] == 60 + + +def test_repository_error_sink_runs_on_generator_thread() -> None: + """Repository failures mutate caller-owned metrics only on the consumer thread.""" + + sweep = sweep_module() + caller_thread = threading.get_ident() + failures = [] + client = PagingClient( + { + ("orgs/ContextualWisdomLab/repos", 1): [[ + repository("broken"), + repository("healthy"), + ]], + ("repos/ContextualWisdomLab/broken/pulls", 1): RuntimeError( + "forbidden" + ), + ("repos/ContextualWisdomLab/healthy/pulls", 1): [pull(7)], + } + ) + + results = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + on_error=lambda scope, error: failures.append( + (threading.get_ident(), scope, str(error)) + ), + ) + ) + + assert [result["repository"] for result in results] == [ + "ContextualWisdomLab/healthy" + ] + assert failures == [ + (caller_thread, "ContextualWisdomLab/broken", "forbidden") + ] + + +def test_pull_pagination_still_stops_at_cutoff() -> None: + """Concurrent repository fetching preserves cutoff-aware page traversal.""" + + sweep = sweep_module() + recent = [pull(number) for number in range(1, 101)] + client = PagingClient( + { + ("orgs/ContextualWisdomLab/repos", 1): [[repository("example")]], + ("repos/ContextualWisdomLab/example/pulls", 1): recent, + ("repos/ContextualWisdomLab/example/pulls", 2): [ + pull(101, "2026-08-01T00:00:00Z") + ], + } + ) + + results = list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + ) + ) + + assert len(results) == 100 + pull_calls = [args for args in client.calls if args[0].endswith("/pulls")] + assert len(pull_calls) == 2 + assert any("page=2" in args for args in pull_calls) + assert not any("page=3" in args for args in pull_calls) + + +def test_generator_close_does_not_wait_for_running_repository() -> None: + """Early consumer exit keeps non-waiting executor shutdown semantics.""" + + sweep = sweep_module() + slow_started = threading.Event() + release_slow = threading.Event() + + class EarlyCloseClient: + """Return one fast result while keeping another request in progress.""" + + def request(self, args, *, input_payload=None): + del input_payload + endpoint = args[0] + if endpoint == "orgs/ContextualWisdomLab/repos": + return [[repository("fast"), repository("slow")]] + if endpoint == "repos/ContextualWisdomLab/fast/pulls": + return [pull(1)] + if endpoint == "repos/ContextualWisdomLab/slow/pulls": + slow_started.set() + release_slow.wait(timeout=2) + return [pull(2)] + raise AssertionError(endpoint) + + generator = sweep.list_recent_pull_requests( + EarlyCloseClient(), + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + ) + assert next(generator)["repository"] == "ContextualWisdomLab/fast" + assert slow_started.wait(timeout=1) + + started = time.monotonic() + generator.close() + elapsed = time.monotonic() - started + release_slow.set() + + assert elapsed < 0.5 + + +def test_repository_failure_without_sink_still_fails_closed() -> None: + """Moving error handling does not convert unhandled failures into success.""" + + sweep = sweep_module() + client = PagingClient( + { + ("orgs/ContextualWisdomLab/repos", 1): [[repository("broken")]], + ("repos/ContextualWisdomLab/broken/pulls", 1): RuntimeError( + "forbidden" + ), + } + ) + + with pytest.raises(RuntimeError, match="forbidden"): + list( + sweep.list_recent_pull_requests( + client, + organization="ContextualWisdomLab", + repository_source="organization", + since="2026-08-05T00:00:00Z", + ) + ) diff --git a/tests/test_agent_mention_router.py b/tests/test_agent_mention_router.py index 874a79e4f..6993f5a59 100644 --- a/tests/test_agent_mention_router.py +++ b/tests/test_agent_mention_router.py @@ -333,6 +333,8 @@ def fake_run(command, **kwargs): assert "secret-token" not in command assert kwargs["env"]["GH_TOKEN"] == "secret-token" assert kwargs["input"] == '{"a": 1}' + assert kwargs["shell"] is False + assert kwargs["timeout"] == 60 monkeypatch.setattr( module.subprocess, "run", @@ -341,6 +343,22 @@ def fake_run(command, **kwargs): assert client.request(["repos/x/y"]) is None +def test_github_client_bounds_request_timeout(monkeypatch) -> None: + """A hung GitHub CLI request becomes an isolated routing failure.""" + + module = load_module() + timeout = module.subprocess.TimeoutExpired(["gh", "api"], 60) + + def timed_out_run(*args, **kwargs): + """Raise the bounded transport timeout fixture.""" + del args, kwargs + raise timeout + + monkeypatch.setattr(module.subprocess, "run", timed_out_run) + with pytest.raises(RuntimeError, match="timed out after 60 seconds"): + module.GitHubClient("secret-token").request(["repos/x/y"]) + + def test_load_event_and_main_paths(tmp_path: Path, monkeypatch, capsys) -> None: """CLI rejects malformed JSON, ignores irrelevant events, and dispatches input.""" diff --git a/tests/test_agent_mention_sweep_regressions.py b/tests/test_agent_mention_sweep_regressions.py index d9c0c4f2a..52745ce12 100644 --- a/tests/test_agent_mention_sweep_regressions.py +++ b/tests/test_agent_mention_sweep_regressions.py @@ -4,6 +4,7 @@ import importlib import sys +import threading from datetime import datetime, timezone from pathlib import Path @@ -161,21 +162,27 @@ def test_repository_failure_is_isolated_and_later_repository_runs() -> None: } ) failures = [] + callback_threads = [] + consumer_thread = threading.get_ident() + + def on_error(scope, error): + failures.append((scope, str(error))) + callback_threads.append(threading.get_ident()) + results = list( sweep.list_recent_pull_requests( client, organization="ContextualWisdomLab", repository_source="organization", since="2026-08-05T00:00:00Z", - on_error=lambda scope, error: failures.append( - (scope, str(error)) - ), + on_error=on_error, ) ) assert [result["repository"] for result in results] == [ "ContextualWisdomLab/healthy" ] assert failures == [("ContextualWisdomLab/broken", "forbidden")] + assert callback_threads == [consumer_thread] def mention_request(comment_id: int):