diff --git a/Dockerfile b/Dockerfile index ea8fc70..7bfa1a1 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,7 +14,7 @@ COPY pyproject.toml README.md ./ COPY src/ src/ RUN pip install --no-cache-dir --prefix=/install \ - "agentfield>=0.1.84" \ + "agentfield==0.1.126" \ "hax-sdk>=0.2.4" \ "pydantic>=2.0" \ "httpx>=0.27" \ diff --git a/src/pr_af/app.py b/src/pr_af/app.py index 9c22bb5..87e736b 100644 --- a/src/pr_af/app.py +++ b/src/pr_af/app.py @@ -5,6 +5,7 @@ import hmac import json import os +import shutil import subprocess import threading import time @@ -114,6 +115,58 @@ def _checkout_pr_branch(target_dir: str, pr_number: int) -> None: raise ValueError(f"git checkout of PR #{pr_number} (pr-review) failed: {checkout.stderr.strip()}") +def _workspace_mtime(path: str) -> float: + """Last-touched time of a review workspace. + + The directory's own mtime doesn't move on re-review — but every fetch + rewrites ``.git/FETCH_HEAD`` — so take the freshest of the markers. + """ + times: list[float] = [] + for candidate in ( + path, + os.path.join(path, ".git"), + os.path.join(path, ".git", "FETCH_HEAD"), + ): + try: + times.append(os.path.getmtime(candidate)) + except OSError: + continue + return max(times) if times else 0.0 + + +def _reap_stale_workspaces(workdir: str, keep: str = "") -> None: + """Delete review workspaces idle for more than PR_AF_WORKSPACE_TTL_DAYS. + + Clones under PR_AF_WORKDIR were never removed, so the persistent volume + grew one checkout per reviewed PR forever (#65). Runs lazily whenever a + managed workspace is resolved — no daemon. The workspace being resolved + for the current review (``keep``) is never touched, and an idle TTL means + a concurrently active workspace has a fresh ``.git/FETCH_HEAD`` and is + skipped. Default 7 days; <= 0 disables. + """ + try: + ttl_days = float(os.getenv("PR_AF_WORKSPACE_TTL_DAYS", "7")) + except ValueError: + ttl_days = 7.0 + if ttl_days <= 0 or not os.path.isdir(workdir): + return + cutoff = time.time() - ttl_days * 86400 + keep_abs = os.path.abspath(keep) if keep else "" + for name in os.listdir(workdir): + path = os.path.join(workdir, name) + if not os.path.isdir(path) or os.path.islink(path): + continue + if keep_abs and os.path.abspath(path) == keep_abs: + continue + if _workspace_mtime(path) >= cutoff: + continue + print( + f"[PR-AF] Reaping stale workspace (idle > {ttl_days:g}d): {path}", + flush=True, + ) + shutil.rmtree(path, ignore_errors=True) + + def _resolve_repo(repo_path: str | None, pr_url: str | None) -> str: workdir = os.getenv("PR_AF_WORKDIR", "/workspaces") target = repo_path @@ -139,6 +192,7 @@ def _resolve_repo(repo_path: str | None, pr_url: str | None) -> str: workspace_name = f"{repo_name}-pr{pr_number}" if pr_number else repo_name target_dir = os.path.join(workdir, workspace_name) os.makedirs(workdir, exist_ok=True) + _reap_stale_workspaces(workdir, keep=target_dir) clone_url = target gh_token = os.getenv("GH_TOKEN", "") @@ -202,6 +256,7 @@ async def review( ignore_paths: list[str] | None = None, hints: list[str] | None = None, models: dict[str, str] | None = None, + max_concurrent_agents: int | None = None, max_concurrent_reviewers: int | None = None, max_coverage_iterations: int | None = None, max_review_depth: int = 2, @@ -232,6 +287,7 @@ async def review( ignore_paths=ignore_paths or [], hints=hints or [], models=models, + max_concurrent_agents=max_concurrent_agents, max_concurrent_reviewers=max_concurrent_reviewers, max_coverage_iterations=max_coverage_iterations, max_review_depth=min(max_review_depth, 3), @@ -319,6 +375,7 @@ def _webhook_review_limits() -> dict[str, object]: """ limits: dict[str, object] = {} for env_name, input_key, minimum in ( + ("PR_AF_MAX_CONCURRENT_AGENTS", "max_concurrent_agents", 1), ("PR_AF_MAX_CONCURRENT_REVIEWERS", "max_concurrent_reviewers", 1), ("PR_AF_MAX_REVIEW_DEPTH", "max_review_depth", 0), ("PR_AF_MAX_COVERAGE_ITERATIONS", "max_coverage_iterations", 1), diff --git a/src/pr_af/config.py b/src/pr_af/config.py index b923d45..90991f8 100644 --- a/src/pr_af/config.py +++ b/src/pr_af/config.py @@ -41,8 +41,23 @@ class BudgetConfig(BaseModel): ) # Concurrency + # Review-wide agent budget: the total number of leaf agent invocations + # (reviewers, obligation verifiers, adversary batches, gates, …) allowed + # in flight at once, across ALL phases — including phases that run + # concurrently (coverage loop ‖ consistency-verify). This is the knob that + # bounds peak opencode-subprocess count and therefore peak memory (#65). + max_concurrent_agents: int = Field( + default_factory=lambda: int(os.getenv("PR_AF_MAX_CONCURRENT_AGENTS", "8")) + ) + # Deprecated alias: historically capped only reviewer dimensions. Still + # honored — the effective budget is min(max_concurrent_agents, this). max_concurrent_reviewers: int = 8 + # Cap on consistency-verify obligations (one verifier agent each). + max_consistency_obligations: int = Field( + default_factory=lambda: int(os.getenv("PR_AF_MAX_CONSISTENCY_OBLIGATIONS", "12")) + ) + # Inner loop caps (per-reviewer) max_reference_follows_per_reviewer: int = 3 max_child_spawns_per_reviewer: int = 2 @@ -256,6 +271,8 @@ def from_input(cls, review_input: ReviewInput) -> ReviewConfig: config.budget.max_cost_usd = review_input.max_cost_usd config.budget.max_duration_seconds = review_input.max_duration_seconds + if review_input.max_concurrent_agents is not None: + config.budget.max_concurrent_agents = review_input.max_concurrent_agents if review_input.max_concurrent_reviewers is not None: config.budget.max_concurrent_reviewers = review_input.max_concurrent_reviewers if review_input.max_coverage_iterations is not None: diff --git a/src/pr_af/evidence.py b/src/pr_af/evidence.py index b8bceeb..61953a8 100644 --- a/src/pr_af/evidence.py +++ b/src/pr_af/evidence.py @@ -16,10 +16,20 @@ # reviewers, evidence-extract/verify, adversary, compound, consistency). Cache by # (abspath, mtime) so a re-checkout (new mtime) invalidates — zero quality cost, just # eliminates redundant disk reads within a review. +# The cache is process-lifetime and spans reviews/repos, so it is bounded by BYTES, +# not just entry count — 2000 large files could otherwise pin multiple GB (#65). _FILE_CACHE: dict[tuple[str, float], list[str]] = {} +_FILE_CACHE_BYTES = 0 +_FILE_CACHE_MAX_BYTES = 128 * 1024 * 1024 +_FILE_CACHE_MAX_ENTRIES = 2000 + +# Each identifier mentioned by a finding costs one repo-wide `grep` child; finding +# bodies can mention dozens, and extraction runs 10 findings at a time (#65). +_MAX_IDENTIFIERS_PER_FINDING = 8 def _read_file_lines(abspath: str) -> list[str]: + global _FILE_CACHE_BYTES try: mtime = os.path.getmtime(abspath) except OSError: @@ -33,9 +43,14 @@ def _read_file_lines(abspath: str) -> list[str]: lines = handle.read().splitlines(keepends=True) except OSError: return [] - if len(_FILE_CACHE) > 2000: # bound memory across many repos/files + size = sum(len(line) for line in lines) + if size > _FILE_CACHE_MAX_BYTES: # pathological single file: serve it uncached + return lines + if len(_FILE_CACHE) >= _FILE_CACHE_MAX_ENTRIES or _FILE_CACHE_BYTES + size > _FILE_CACHE_MAX_BYTES: _FILE_CACHE.clear() + _FILE_CACHE_BYTES = 0 _FILE_CACHE[key] = lines + _FILE_CACHE_BYTES += size return lines @@ -141,7 +156,7 @@ async def _extract_for_finding(finding: ReviewFinding) -> EvidencePackage: async with semaphore: normalized_file = _normalize_relative_path(repo_path, finding.file_path) text_blob = "\n".join([finding.title, finding.body, finding.evidence]) - identifiers = _extract_mentioned_identifiers(text_blob) + identifiers = _extract_mentioned_identifiers(text_blob)[:_MAX_IDENTIFIERS_PER_FINDING] primary_task = asyncio.to_thread( _read_code_snippet, diff --git a/src/pr_af/orchestrator.py b/src/pr_af/orchestrator.py index 021d016..1305f27 100644 --- a/src/pr_af/orchestrator.py +++ b/src/pr_af/orchestrator.py @@ -13,11 +13,15 @@ import os import subprocess import time -from typing import Any, cast +from fnmatch import fnmatch +from typing import TYPE_CHECKING, Any, cast from uuid import uuid4 import httpx +if TYPE_CHECKING: + from collections.abc import Awaitable + from .config import AUTO_DEPTH_THRESHOLDS, DEPTH_PROFILES, ReviewConfig from .diff_engine import parse_unified_diff from .evidence import EvidencePackage, build_dimension_pack, extract_evidence_for_findings @@ -81,6 +85,28 @@ def _unwrap(result: object) -> dict: return cast("dict", result) +def _matches_ignore_pattern(path: str, pattern: str) -> bool: + """Glob match for ignore_paths entries. + + fnmatch's ``*`` crosses ``/``, so ``.github/**`` and ``**/*.generated.*`` + work directly; the extra branches make ``**/name`` also match a top-level + ``name`` and let extension-only patterns (``*.md``) match by basename. + """ + path = path.lstrip("/") + if fnmatch(path, pattern): + return True + if pattern.startswith("**/") and fnmatch(path, pattern[3:]): + return True + prefix = pattern[:-3] + if pattern.endswith("/**") and (path == prefix or path.startswith(prefix + "/")): + return True + return "/" not in pattern and fnmatch(os.path.basename(path), pattern) + + +def _is_ignored_path(path: str, patterns: list[str]) -> bool: + return any(_matches_ignore_pattern(path, pattern) for pattern in patterns) + + class ReviewOrchestrator: """Orchestrates the 7-phase PR review pipeline. @@ -131,6 +157,28 @@ def __init__(self, app: Any, input: ReviewInput, config: ReviewConfig | None = N self.adversary_challenged_count = 0 self.effective_depth: str = "standard" + # Review-wide agent-concurrency budget (#65): ONE semaphore for every + # leaf agent invocation across all phases — including phases that run + # concurrently (coverage loop ‖ consistency-verify), which previously + # each brought their own limiter (or none) and oversubscribed the + # process. min() keeps the deprecated max_concurrent_reviewers knob + # binding for callers that still set it. + agent_cap = min( + self.config.budget.max_concurrent_agents, + self.config.budget.max_concurrent_reviewers, + ) + self._agent_semaphore = asyncio.Semaphore(max(1, agent_cap)) + + async def _agent_slot(self, coro: Awaitable[Any]) -> Any: + """Await one leaf agent call under the review-wide concurrency budget. + + Only leaf calls may be wrapped: a coroutine that itself acquires the + budget (directly or via a child) must never run under a held slot, or + low caps deadlock on hold-and-wait. + """ + async with self._agent_semaphore: + return await coro + async def run(self) -> ReviewResult: print("[PR-AF] Starting 7-phase pipeline", flush=True) @@ -268,7 +316,9 @@ async def _run_review_phases( kept = reviewer_findings if len(reviewer_findings) > 1: try: - pw = await post_worthiness_gate(findings=[f.model_dump() for f in reviewer_findings]) + pw = await self._agent_slot( + post_worthiness_gate(findings=[f.model_dump() for f in reviewer_findings]) + ) sel = [f for i, f in enumerate(reviewer_findings) if i in set(pw.get("keep_indices", range(len(reviewer_findings))))] if sel: @@ -425,26 +475,62 @@ async def _run_intake(self) -> IntakeResult: else: raise ValueError("One of pr_url, diff_text, or repo_path is required") - result_raw = await intake_phase( + self.pr_data = self._apply_ignore_paths(self.pr_data) + + result_raw = await self._agent_slot(intake_phase( + pr_data=self.pr_data.model_dump(), depth=self.input.depth, - ) + )) self.agent_invocations += 1 self._register_cost("intake", self._extract_cost(result_raw)) intake = IntakeResult.model_validate(result_raw) return intake + def _apply_ignore_paths(self, pr_data: GitHubPRData) -> GitHubPRData: + """Drop ignore_paths-matched files from the review input. + + Applied before intake so every downstream consumer — depth resolution, + anatomy, meta-selectors, reviewers, obligation extraction — sees only + reviewable files. Generated churn (a lockfile regen can be a 60k-line + diff) otherwise inflates every prompt, trips depth escalation, and + drives the fan-out that OOM-killed the node in #65. + """ + patterns = self.config.ignore_paths + if not patterns or not pr_data.changed_files: + return pr_data + kept: list[ChangedFile] = [] + dropped: list[ChangedFile] = [] + for cf in pr_data.changed_files: + (dropped if _is_ignored_path(cf.path, patterns) else kept).append(cf) + if not dropped: + return pr_data + dropped_lines = sum(len(cf.patch.splitlines()) for cf in dropped if cf.patch) + listed = ", ".join(cf.path for cf in dropped[:5]) + suffix = f" (+{len(dropped) - 5} more)" if len(dropped) > 5 else "" + print( + f"[PR-AF] Ignoring {len(dropped)} file(s) matching ignore_paths, " + f"{dropped_lines} diff lines dropped: {listed}{suffix}", + flush=True, + ) + filtered_diff = "\n".join( + f"diff --git a/{cf.path} b/{cf.path}\n--- a/{cf.path}\n+++ b/{cf.path}\n{cf.patch}" + for cf in kept + if cf.patch + ) + return pr_data.model_copy(update={"changed_files": kept, "diff": filtered_diff}) + async def _run_anatomy(self, intake: IntakeResult) -> AnatomyResult: if self._budget_or_timeout_exhausted("anatomy"): raise BudgetExhaustedError(self._budget_exhausted_message("anatomy")) if self.pr_data is None: raise RuntimeError("PR data not initialized") - result_raw = await anatomy_phase( + result_raw = await self._agent_slot(anatomy_phase( pr_data=self.pr_data.model_dump(), intake=intake.model_dump(), repo_path=self.input.repo_path or "", - ) + )) self.agent_invocations += 1 self._register_cost("anatomy", self._extract_cost(result_raw)) anatomy = AnatomyResult.model_validate(result_raw) @@ -489,14 +575,14 @@ async def _run_meta_selectors( async def run_lens(lens_name: str) -> MetaDimensionResult: fn = lens_map[lens_name] - result_raw = await fn( + result_raw = await self._agent_slot(fn( intake=intake.model_dump(), anatomy=anatomy.model_dump(), depth=review_depth, repo_path=self.input.repo_path or "", diff_patches=self._build_file_patches(), reviewer_feedback=reviewer_feedback, - ) + )) self.agent_invocations += 1 self._register_cost("meta_selectors", self._extract_cost(result_raw)) return MetaDimensionResult.model_validate(result_raw) @@ -568,12 +654,12 @@ async def _run_evidence_verification( ev_packages = {f.title: evidence_map[f.title].model_dump() for f in high_priority if f.title in evidence_map} - verifier_raw = await evidence_verifier( + verifier_raw = await self._agent_slot(evidence_verifier( findings=[f.model_dump() for f in high_priority], evidence_packages=ev_packages if ev_packages else None, pr_context=self._build_pr_context_string(), repo_path=self.input.repo_path or "", - ) + )) self.agent_invocations += 1 self._register_cost("adversary", self._extract_cost(verifier_raw)) @@ -665,13 +751,13 @@ async def run_batch(batch: list[ReviewFinding]) -> list[AdversaryResult]: if ev_entry: batch_evidence[f.title] = ev_entry - adversary_raw = await adversary_phase( + adversary_raw = await self._agent_slot(adversary_phase( findings=[f.model_dump() for f in batch], ai_generated_confidence=ai_confidence, pr_context=self._build_pr_context_string(), repo_path=self.input.repo_path or "", evidence_packages=batch_evidence if batch_evidence else None, - ) + )) self.agent_invocations += 1 self._register_cost("adversary", self._extract_cost(adversary_raw)) return self._extract_adversary_results(adversary_raw) @@ -692,7 +778,10 @@ async def _run_parallel_review( reviewer_feedback: str = "", ) -> None: max_depth = self.config.budget.max_review_depth - semaphore = asyncio.Semaphore(self.config.budget.max_concurrent_reviewers) + # The shared review-wide budget, NOT a fresh local semaphore: the + # coverage loop calls this again while consistency-verify is running, + # and independent limiters oversubscribed the process (#65). + semaphore = self._agent_semaphore async def run_dimension(dim: ReviewDimension, depth: int) -> None: if self._budget_or_timeout_exhausted("review"): @@ -706,7 +795,9 @@ async def run_dimension(dim: ReviewDimension, depth: int) -> None: primed = "" if self.config.budget.evidence_pack_reviewers and self.input.repo_path: try: - primed = build_dimension_pack(self.input.repo_path, dim.target_files, dim_patches) + primed = await asyncio.to_thread( + build_dimension_pack, self.input.repo_path, dim.target_files, dim_patches + ) except Exception: # noqa: BLE001 primed = "" @@ -846,11 +937,11 @@ async def _run_coverage_loop( reviewed_clusters = self._reviewed_clusters(anatomy, findings) dimension_names = [d.name for d in plan.dimensions] - gate_raw = await coverage_gate( + gate_raw = await self._agent_slot(coverage_gate( anatomy=anatomy.model_dump(), reviewed_clusters=reviewed_clusters, dimension_names_reviewed=dimension_names, - ) + )) self.agent_invocations += 1 self._register_cost("coverage", self._extract_cost(gate_raw)) gate = gate_raw if isinstance(gate_raw, dict) else {} @@ -919,15 +1010,16 @@ async def _run_consistency_verify(self, all_findings: list[ReviewFinding]) -> li return all_findings repo = self.input.repo_path or "" try: - ob_raw = await extract_obligations( + ob_raw = await self._agent_slot(extract_obligations( diff_patches=diff_patches, repo_path=repo, pr_context=self._build_pr_context_string() - ) + )) except Exception as exc: # noqa: BLE001 print(f"[PR-AF] Consistency-verify (extract) skipped: {exc}", flush=True) return all_findings self.agent_invocations += 1 self._register_cost("review", self._extract_cost(ob_raw)) - obligations = (ob_raw.get("obligations", []) if isinstance(ob_raw, dict) else [])[:12] + max_obligations = self.config.budget.max_consistency_obligations + obligations = (ob_raw.get("obligations", []) if isinstance(ob_raw, dict) else [])[:max_obligations] if not obligations: print("[PR-AF] Consistency-verify: 0 obligations", flush=True) return all_findings @@ -935,7 +1027,7 @@ async def _run_consistency_verify(self, all_findings: list[ReviewFinding]) -> li async def _verify(o: dict) -> dict: try: - return await verify_obligation(obligation=o, repo_path=repo) + return await self._agent_slot(verify_obligation(obligation=o, repo_path=repo)) except Exception: # noqa: BLE001 return {"holds": True} @@ -1416,10 +1508,10 @@ async def _dedup_compound_findings( ) -> list[ReviewFinding]: individual_summary = "\n".join(f"- [{f.severity}] {f.title} ({f.file_path})" for f in individual_findings[:20]) - dedup_raw = await compound_dedup_phase( + dedup_raw = await self._agent_slot(compound_dedup_phase( compound_findings=[f.model_dump() for f in compound_findings], individual_findings_summary=individual_summary, - ) + )) self.agent_invocations += 1 self._register_cost("cross_ref", self._extract_cost(dedup_raw)) @@ -1459,11 +1551,11 @@ async def _run_compound_analysis( cluster_evidence = { title: evidence_map[title].model_dump() for title in cluster_titles if title in evidence_map } - task = compound_finder_phase( + task = self._agent_slot(compound_finder_phase( cluster_findings=[finding.model_dump() for finding in cluster], repo_path=self.input.repo_path or "", evidence_map=cluster_evidence or None, - ) + )) compound_tasks.append(task) results = await asyncio.gather(*compound_tasks, return_exceptions=True) diff --git a/src/pr_af/schemas/input.py b/src/pr_af/schemas/input.py index 7912a93..874e143 100644 --- a/src/pr_af/schemas/input.py +++ b/src/pr_af/schemas/input.py @@ -37,6 +37,7 @@ class ReviewInput(BaseModel): models: dict[str, str] | None = None # Budget overrides + max_concurrent_agents: int | None = None # Review-wide leaf-agent budget max_concurrent_reviewers: int | None = None max_coverage_iterations: int | None = None max_review_depth: int = 2 # Max recursive sub-review depth (1=flat, 2=one sub-level, 3=max) diff --git a/tests/test_agent_budget.py b/tests/test_agent_budget.py new file mode 100644 index 0000000..7055623 --- /dev/null +++ b/tests/test_agent_budget.py @@ -0,0 +1,181 @@ +"""Tests for the review-wide agent-concurrency budget (#65). + +Validation contract: + +* concurrent leaf-agent invocations never exceed the cap at any instant, even + when the coverage/review path and consistency-verify run concurrently +* the deprecated ``max_concurrent_reviewers`` knob still binds (effective cap + is the min of both knobs) +* the consistency-verify obligation count is configurable, not a literal +* ``PR_AF_MAX_CONCURRENT_AGENTS`` / ``PR_AF_MAX_CONSISTENCY_OBLIGATIONS`` env + knobs and the ``max_concurrent_agents`` input field plumb through +""" + +from __future__ import annotations + +import asyncio + +import pytest + +import pr_af.orchestrator as orchestrator_module +from pr_af.app import _webhook_review_limits +from pr_af.config import BudgetConfig, ReviewConfig +from pr_af.orchestrator import ReviewOrchestrator +from pr_af.schemas.input import ChangedFile, GitHubPRData, ReviewInput +from pr_af.schemas.pipeline import ReviewDimension, ReviewPlan + + +class ConcurrencyProbe: + """Counts in-flight fake agent calls and records the peak.""" + + def __init__(self) -> None: + self.active = 0 + self.peak = 0 + self.calls = 0 + + async def run(self) -> None: + self.active += 1 + self.calls += 1 + self.peak = max(self.peak, self.active) + await asyncio.sleep(0.02) + self.active -= 1 + + +def _make_orchestrator(config: ReviewConfig) -> ReviewOrchestrator: + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="diff"), config=config) + orchestrator.pr_data = GitHubPRData( + owner="", + repo="", + number=0, + title="t", + description="", + diff="@@ -1 +1 @@\n+x", + changed_files=[ + ChangedFile(path="src/a.py", status="modified", additions=1, deletions=0, patch="@@ -1 +1 @@\n+x") + ], + ) + return orchestrator + + +def _plan(n: int) -> ReviewPlan: + return ReviewPlan( + dimensions=[ + ReviewDimension(id=f"d{i}", name=f"D{i}", review_prompt="p", target_files=["src/a.py"]) + for i in range(n) + ] + ) + + +async def _drain(queue: asyncio.Queue) -> None: + while await queue.get() is not None: + pass + + +async def test_shared_budget_caps_leaf_agents_across_concurrent_phases( + monkeypatch: pytest.MonkeyPatch, +) -> None: + config = ReviewConfig() + config.budget.max_concurrent_agents = 2 + orchestrator = _make_orchestrator(config) + probe = ConcurrencyProbe() + + async def fake_review_dimension(**kwargs: object) -> dict: + await probe.run() + return {"findings": [], "sub_reviews": []} + + async def fake_extract_obligations(**kwargs: object) -> dict: + await probe.run() + return {"obligations": [{"id": i} for i in range(4)]} + + async def fake_verify_obligation(**kwargs: object) -> dict: + await probe.run() + return {"holds": True} + + monkeypatch.setattr(orchestrator_module, "review_dimension", fake_review_dimension) + monkeypatch.setattr(orchestrator_module, "extract_obligations", fake_extract_obligations) + monkeypatch.setattr(orchestrator_module, "verify_obligation", fake_verify_obligation) + + queue: asyncio.Queue = asyncio.Queue() + await asyncio.wait_for( + asyncio.gather( + orchestrator._run_parallel_review(_plan(3), queue), + orchestrator._run_consistency_verify([]), + _drain(queue), + ), + timeout=5, + ) + + assert probe.calls == 3 + 1 + 4 + assert probe.peak <= 2 + + +async def test_deprecated_reviewer_knob_still_binds(monkeypatch: pytest.MonkeyPatch) -> None: + config = ReviewConfig() + config.budget.max_concurrent_agents = 8 + config.budget.max_concurrent_reviewers = 1 + orchestrator = _make_orchestrator(config) + probe = ConcurrencyProbe() + + async def fake_review_dimension(**kwargs: object) -> dict: + await probe.run() + return {"findings": [], "sub_reviews": []} + + monkeypatch.setattr(orchestrator_module, "review_dimension", fake_review_dimension) + + queue: asyncio.Queue = asyncio.Queue() + await asyncio.wait_for( + asyncio.gather(orchestrator._run_parallel_review(_plan(3), queue), _drain(queue)), + timeout=5, + ) + + assert probe.calls == 3 + assert probe.peak == 1 + + +async def test_consistency_obligation_cap_is_configurable(monkeypatch: pytest.MonkeyPatch) -> None: + config = ReviewConfig() + config.budget.max_consistency_obligations = 2 + orchestrator = _make_orchestrator(config) + verify_calls = 0 + + async def fake_extract_obligations(**kwargs: object) -> dict: + return {"obligations": [{"id": i} for i in range(5)]} + + async def fake_verify_obligation(**kwargs: object) -> dict: + nonlocal verify_calls + verify_calls += 1 + return {"holds": True} + + monkeypatch.setattr(orchestrator_module, "extract_obligations", fake_extract_obligations) + monkeypatch.setattr(orchestrator_module, "verify_obligation", fake_verify_obligation) + + await asyncio.wait_for(orchestrator._run_consistency_verify([]), timeout=5) + + assert verify_calls == 2 + + +def test_env_knobs_drive_budget_defaults(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("PR_AF_MAX_CONCURRENT_AGENTS", "4") + monkeypatch.setenv("PR_AF_MAX_CONSISTENCY_OBLIGATIONS", "6") + budget = BudgetConfig() + assert budget.max_concurrent_agents == 4 + assert budget.max_consistency_obligations == 6 + + monkeypatch.delenv("PR_AF_MAX_CONCURRENT_AGENTS") + monkeypatch.delenv("PR_AF_MAX_CONSISTENCY_OBLIGATIONS") + budget = BudgetConfig() + assert budget.max_concurrent_agents == 8 + assert budget.max_consistency_obligations == 12 + + +def test_input_override_plumbs_through() -> None: + config = ReviewConfig.from_input(ReviewInput(diff_text="d", max_concurrent_agents=3)) + assert config.budget.max_concurrent_agents == 3 + + +def test_webhook_limits_include_agent_budget(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("PR_AF_MAX_CONCURRENT_AGENTS", "2") + monkeypatch.delenv("PR_AF_MAX_CONCURRENT_REVIEWERS", raising=False) + monkeypatch.delenv("PR_AF_MAX_REVIEW_DEPTH", raising=False) + monkeypatch.delenv("PR_AF_MAX_COVERAGE_ITERATIONS", raising=False) + assert _webhook_review_limits() == {"max_concurrent_agents": 2} diff --git a/tests/test_evidence_caps.py b/tests/test_evidence_caps.py new file mode 100644 index 0000000..3e378f2 --- /dev/null +++ b/tests/test_evidence_caps.py @@ -0,0 +1,86 @@ +"""Tests for evidence-extraction resource caps (#65). + +Validation contract: + +* at most ``_MAX_IDENTIFIERS_PER_FINDING`` repo-wide grep searches are + dispatched per finding, no matter how many identifiers the body mentions +* the shared file cache is bounded by bytes, not just entry count, and a + single pathological file larger than the cap is never cached +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +import pr_af.evidence as evidence +from pr_af.evidence import extract_evidence_for_findings +from pr_af.schemas.pipeline import ReviewFinding + + +def _finding(body: str) -> ReviewFinding: + return ReviewFinding( + dimension_id="d", + dimension_name="D", + file_path="src/a.py", + line_start=1, + line_end=1, + severity="suggestion", + title="t", + body=body, + confidence=0.5, + ) + + +async def test_identifier_grep_fanout_is_capped( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + searched: list[str] = [] + + def fake_find_function_callers(repo_path: str, ident: str, exclude_file: str = "") -> list[str]: + searched.append(ident) + return [] + + monkeypatch.setattr(evidence, "_find_function_callers", fake_find_function_callers) + body = " ".join(f"`identifier_number_{i}`" for i in range(20)) + + await extract_evidence_for_findings( + findings=[_finding(body)], repo_path=str(tmp_path), diff_patches={} + ) + + assert 0 < len(searched) <= evidence._MAX_IDENTIFIERS_PER_FINDING + + +def _fresh_cache(monkeypatch: pytest.MonkeyPatch, max_bytes: int) -> None: + monkeypatch.setattr(evidence, "_FILE_CACHE", {}) + monkeypatch.setattr(evidence, "_FILE_CACHE_BYTES", 0) + monkeypatch.setattr(evidence, "_FILE_CACHE_MAX_BYTES", max_bytes) + + +def test_file_cache_is_byte_bounded(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + _fresh_cache(monkeypatch, max_bytes=100) + file_a = tmp_path / "a.py" + file_b = tmp_path / "b.py" + file_a.write_text("a" * 80) + file_b.write_text("b" * 80) + + evidence._read_file_lines(str(file_a)) + assert len(evidence._FILE_CACHE) == 1 + assert evidence._FILE_CACHE_BYTES == 80 + + evidence._read_file_lines(str(file_b)) + assert len(evidence._FILE_CACHE) == 1 + assert evidence._FILE_CACHE_BYTES == 80 + + +def test_oversized_file_served_uncached(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + _fresh_cache(monkeypatch, max_bytes=100) + big = tmp_path / "big.py" + big.write_text("x" * 500) + + lines = evidence._read_file_lines(str(big)) + + assert lines == ["x" * 500] + assert evidence._FILE_CACHE == {} + assert evidence._FILE_CACHE_BYTES == 0 diff --git a/tests/test_ignore_paths.py b/tests/test_ignore_paths.py new file mode 100644 index 0000000..947eb31 --- /dev/null +++ b/tests/test_ignore_paths.py @@ -0,0 +1,103 @@ +"""Tests for ignore_paths filtering at intake (#65). + +Validation contract: + +* generated/lockfile churn is dropped from the review input before any agent + sees it — changed_files and the diff both shrink +* a lockfile-regen PR no longer resolves to a deep review: auto-depth is + computed from the filtered diff +* pattern semantics: ``**/name`` matches top-level and nested, ``dir/**`` + matches the subtree, extension globs match by basename +""" + +from __future__ import annotations + +from pr_af.config import ReviewConfig +from pr_af.orchestrator import ReviewOrchestrator, _is_ignored_path +from pr_af.schemas.input import ChangedFile, GitHubPRData, ReviewInput +from pr_af.schemas.pipeline import IntakeResult + +DEFAULT_PATTERNS = ReviewConfig().ignore_paths + + +def _pr_data(files: list[ChangedFile]) -> GitHubPRData: + diff = "\n".join( + f"diff --git a/{f.path} b/{f.path}\n--- a/{f.path}\n+++ b/{f.path}\n{f.patch}" for f in files + ) + return GitHubPRData( + owner="o", repo="r", number=1, title="t", description="", diff=diff, changed_files=files + ) + + +def _changed(path: str, patch: str) -> ChangedFile: + return ChangedFile( + path=path, status="modified", additions=patch.count("\n+"), deletions=0, patch=patch + ) + + +def _intake(review_depth: str = "") -> IntakeResult: + return IntakeResult( + pr_type="feature", + complexity="standard", + languages=[], + areas_touched=[], + risk_signals=[], + ai_generated=0.0, + review_depth=review_depth, + pr_summary="", + ) + + +def test_pattern_semantics() -> None: + assert _is_ignored_path("package-lock.json", DEFAULT_PATTERNS) + assert _is_ignored_path("web/package-lock.json", DEFAULT_PATTERNS) + assert _is_ignored_path("yarn.lock", DEFAULT_PATTERNS) + assert _is_ignored_path(".github/workflows/ci.yml", DEFAULT_PATTERNS) + assert _is_ignored_path("docs/README.md", DEFAULT_PATTERNS) + assert _is_ignored_path("vendor/lib/x.go", DEFAULT_PATTERNS) + assert _is_ignored_path("dist/app.min.js", DEFAULT_PATTERNS) + assert not _is_ignored_path("src/app.py", DEFAULT_PATTERNS) + assert not _is_ignored_path("src/lockfile_parser.py", DEFAULT_PATTERNS) + + +def test_lockfile_regen_is_filtered_before_intake() -> None: + lock_patch = "@@ -1,3 +1,60000 @@\n" + "\n".join(f'+ "dep-{i}": "1.0.{i}"' for i in range(200)) + src_patch = "@@ -1,2 +1,3 @@\n+import os" + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="d")) + pr_data = _pr_data([_changed("package-lock.json", lock_patch), _changed("src/app.py", src_patch)]) + + filtered = orchestrator._apply_ignore_paths(pr_data) + + assert [f.path for f in filtered.changed_files] == ["src/app.py"] + assert "dep-0" not in filtered.diff + assert "import os" in filtered.diff + + orchestrator.pr_data = filtered + assert orchestrator._resolve_depth(_intake()) == "quick" + + +def test_lockfile_only_pr_filters_to_empty() -> None: + lock_patch = "@@ -1 +1,3 @@\n+a\n+b\n+c" + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="d")) + pr_data = _pr_data([_changed("package-lock.json", lock_patch)]) + + filtered = orchestrator._apply_ignore_paths(pr_data) + + assert filtered.changed_files == [] + assert filtered.diff == "" + + +def test_no_match_returns_input_unchanged() -> None: + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="d")) + pr_data = _pr_data([_changed("src/app.py", "@@ -1 +1 @@\n+x")]) + + assert orchestrator._apply_ignore_paths(pr_data) is pr_data + + +def test_empty_patterns_disable_filtering() -> None: + config = ReviewConfig() + config.ignore_paths = [] + orchestrator = ReviewOrchestrator(app=None, input=ReviewInput(diff_text="d"), config=config) + pr_data = _pr_data([_changed("package-lock.json", "@@ -1 +1 @@\n+x")]) + + assert orchestrator._apply_ignore_paths(pr_data) is pr_data diff --git a/tests/test_workspace_reaper.py b/tests/test_workspace_reaper.py new file mode 100644 index 0000000..e8ca24b --- /dev/null +++ b/tests/test_workspace_reaper.py @@ -0,0 +1,88 @@ +"""Tests for the stale-workspace reaper (#65). + +Validation contract: + +* workspaces idle past PR_AF_WORKSPACE_TTL_DAYS are deleted on resolution +* fresh workspaces, the active workspace, and non-directories are never touched +* a recent ``.git/FETCH_HEAD`` counts as activity even when the dir mtime is old +* TTL <= 0 disables reaping +""" + +from __future__ import annotations + +import os +import time +from pathlib import Path + +import pytest + +from pr_af.app import _reap_stale_workspaces + +WEEK = 8 * 86400 + + +def _backdate(path: Path, seconds: float) -> None: + stamp = time.time() - seconds + os.utime(path, (stamp, stamp)) + + +def _make_workspace(workdir: Path, name: str, idle_seconds: float) -> Path: + ws = workdir / name + (ws / ".git").mkdir(parents=True) + _backdate(ws / ".git", idle_seconds) + _backdate(ws, idle_seconds) + return ws + + +def test_stale_workspace_is_reaped(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PR_AF_WORKSPACE_TTL_DAYS", raising=False) + stale = _make_workspace(tmp_path, "old-repo-pr1", idle_seconds=WEEK) + fresh = _make_workspace(tmp_path, "new-repo-pr2", idle_seconds=60) + + _reap_stale_workspaces(str(tmp_path)) + + assert not stale.exists() + assert fresh.exists() + + +def test_active_workspace_is_kept_even_when_stale( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("PR_AF_WORKSPACE_TTL_DAYS", raising=False) + active = _make_workspace(tmp_path, "repo-pr3", idle_seconds=WEEK) + + _reap_stale_workspaces(str(tmp_path), keep=str(active)) + + assert active.exists() + + +def test_fresh_fetch_head_counts_as_activity( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.delenv("PR_AF_WORKSPACE_TTL_DAYS", raising=False) + ws = _make_workspace(tmp_path, "repo-pr4", idle_seconds=WEEK) + (ws / ".git" / "FETCH_HEAD").write_text("ref") + + _reap_stale_workspaces(str(tmp_path)) + + assert ws.exists() + + +def test_ttl_zero_disables_reaping(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("PR_AF_WORKSPACE_TTL_DAYS", "0") + stale = _make_workspace(tmp_path, "repo-pr5", idle_seconds=WEEK) + + _reap_stale_workspaces(str(tmp_path)) + + assert stale.exists() + + +def test_plain_files_are_never_touched(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("PR_AF_WORKSPACE_TTL_DAYS", raising=False) + stray = tmp_path / "notes.txt" + stray.write_text("keep me") + _backdate(stray, WEEK) + + _reap_stale_workspaces(str(tmp_path)) + + assert stray.exists()