""" core/static_analysis.py — v4.6 Production Gate: Static Analysis Runs language-appropriate linters against the ACTUAL on-disk files that the last patch touched. This is a *defensive* production gate: any patch that introduces new lint errors is rejected before ``run_repo_tests`` is invoked, so AutoDebugLoop can request a corrected patch immediately instead of consuming a full test-run cycle. Supported linters ----------------- * Python → ``ruff check --output-format=concise`` (fallback: ``python -m ruff``) * JS/TS → ``eslint --no-eslintrc --format compact`` (fallback: ``npx --no-install eslint``) Design contract --------------- * NEVER modifies files (read-only static analysis). * NEVER assumes the linter is installed. If the binary is missing, the check is reported as ``skipped=True`` — this is treated as a *neutral* result by the pipeline (does not block the patch). * Only ``modified_files`` are linted. This keeps runtime bounded regardless of repo size and ensures we blame the current patch, not pre-existing tech-debt in the uploaded ZIP. * All commands run with a strict timeout and captured combined output. The output shape mirrors ``core.test_runner.TestRunResult`` so downstream UI/timeline code can format both consistently. """ from __future__ import annotations import os import shutil import subprocess import time from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _PY_EXTS = {".py", ".pyi"} _JS_EXTS = {".js", ".jsx", ".mjs", ".cjs", ".ts", ".tsx"} DEFAULT_TIMEOUT = 60 # seconds — linters are quick; anything longer is a bug. # --------------------------------------------------------------------------- # Result dataclass # --------------------------------------------------------------------------- @dataclass class LintResult: """Structured result for a single linter invocation.""" linter: str = "" language: str = "" passed: bool = True skipped: bool = False exit_code: Optional[int] = None files_checked: List[str] = field(default_factory=list) issue_count: int = 0 logs: str = "" reason: str = "" duration_s: float = 0.0 command: str = "" def to_dict(self) -> Dict[str, Any]: return asdict(self) @dataclass class StaticAnalysisReport: """Aggregate report across all linters that ran for this patch.""" passed: bool = True skipped: bool = False total_issues: int = 0 results: List[LintResult] = field(default_factory=list) summary: str = "" def to_dict(self) -> Dict[str, Any]: return { "passed": self.passed, "skipped": self.skipped, "total_issues": self.total_issues, "summary": self.summary, "results": [r.to_dict() for r in self.results], } def combined_logs(self) -> str: parts: List[str] = [] for r in self.results: header = ( f"── {r.linter} ({r.language}) — " f"{'PASS' if r.passed else ('SKIP' if r.skipped else 'FAIL')} " f"[{len(r.files_checked)} file(s), {r.issue_count} issue(s)] ──" ) parts.append(header) if r.reason: parts.append(f"reason: {r.reason}") if r.logs: parts.append(r.logs.strip()) parts.append("") return "\n".join(parts).strip() # --------------------------------------------------------------------------- # File classification # --------------------------------------------------------------------------- def _classify_files(project_root: str, files: Sequence[str]) -> Tuple[List[str], List[str]]: """Return (python_files, js_ts_files) as absolute paths, existing only.""" root = Path(project_root).resolve() py_files: List[str] = [] js_files: List[str] = [] for rel in files or []: if not rel: continue candidate = Path(rel) abs_path = candidate if candidate.is_absolute() else (root / rel) try: abs_path = abs_path.resolve() except OSError: continue # Defensive: must stay inside project_root. try: abs_path.relative_to(root) except ValueError: continue if not abs_path.is_file(): continue ext = abs_path.suffix.lower() if ext in _PY_EXTS: py_files.append(str(abs_path)) elif ext in _JS_EXTS: js_files.append(str(abs_path)) # De-duplicate while preserving order. def _dedup(seq: Iterable[str]) -> List[str]: seen: set = set() out: List[str] = [] for item in seq: if item not in seen: seen.add(item) out.append(item) return out return _dedup(py_files), _dedup(js_files) def _has_tool(binary: str) -> bool: return shutil.which(binary) is not None def _rel(paths: Sequence[str], root: str) -> List[str]: root_p = Path(root).resolve() out: List[str] = [] for p in paths: try: out.append(str(Path(p).resolve().relative_to(root_p)).replace(os.sep, "/")) except Exception: out.append(p) return out # --------------------------------------------------------------------------- # Ruff (Python) # --------------------------------------------------------------------------- def _ruff_command() -> Optional[List[str]]: if _has_tool("ruff"): return ["ruff", "check", "--output-format=concise", "--no-cache"] # ruff also ships as a Python module in some environments. try: probe = subprocess.run( ["python", "-c", "import ruff"], capture_output=True, timeout=6, ) if probe.returncode == 0: return ["python", "-m", "ruff", "check", "--output-format=concise", "--no-cache"] except Exception: pass return None def run_ruff(project_root: str, files: Sequence[str], timeout: int = DEFAULT_TIMEOUT) -> LintResult: result = LintResult(linter="ruff", language="python", files_checked=list(files)) if not files: result.skipped = True result.reason = "no python files in patch scope" return result argv = _ruff_command() if not argv: result.skipped = True result.reason = "ruff binary not on PATH" return result argv = argv + list(files) result.command = " ".join([*argv[:-len(files)], f"<{len(files)} file(s)>"]) start = time.time() try: proc = subprocess.run( argv, cwd=project_root, capture_output=True, text=True, timeout=timeout, ) result.exit_code = proc.returncode stdout = proc.stdout or "" stderr = proc.stderr or "" combined = stdout + (("\n[stderr]\n" + stderr) if stderr.strip() else "") result.logs = combined.strip() # ruff prints one issue per line in concise format. result.issue_count = sum( 1 for line in stdout.splitlines() if line.strip() and not line.startswith("Found ") ) result.passed = proc.returncode == 0 except subprocess.TimeoutExpired as exc: result.exit_code = -1 result.passed = False result.reason = "timeout" result.logs = ( f"[ruff] TIMEOUT after {timeout}s\n" f"partial stdout:\n{(exc.stdout or '')[:2000]}\n" f"partial stderr:\n{(exc.stderr or '')[:1000]}" ) except FileNotFoundError: result.skipped = True result.reason = "ruff binary disappeared mid-run" except Exception as exc: # noqa: BLE001 result.passed = False result.exit_code = -1 result.reason = str(exc) result.logs = f"[ruff] launch failed: {exc}" finally: result.duration_s = round(time.time() - start, 3) result.files_checked = _rel(files, project_root) return result # --------------------------------------------------------------------------- # ESLint (JS/TS) # --------------------------------------------------------------------------- def _eslint_command(project_root: str) -> Optional[List[str]]: # Prefer project-local install first. local = Path(project_root) / "node_modules" / ".bin" / "eslint" if local.exists(): return [str(local), "--format", "compact"] if _has_tool("eslint"): return ["eslint", "--format", "compact"] if _has_tool("npx"): # --no-install prevents npx from silently downloading eslint at runtime # (which would violate the "no network side-effects" contract). return ["npx", "--no-install", "eslint", "--format", "compact"] return None def run_eslint(project_root: str, files: Sequence[str], timeout: int = DEFAULT_TIMEOUT) -> LintResult: result = LintResult(linter="eslint", language="javascript/typescript", files_checked=list(files)) if not files: result.skipped = True result.reason = "no js/ts files in patch scope" return result argv = _eslint_command(project_root) if not argv: result.skipped = True result.reason = "eslint not available (no local install, PATH, or npx)" return result # If the project has no eslint config, ``--no-eslintrc`` avoids surprising # errors from user home configs; but if a config IS present we let it win. has_config = any( (Path(project_root) / name).exists() for name in ( ".eslintrc", ".eslintrc.js", ".eslintrc.cjs", ".eslintrc.json", ".eslintrc.yml", ".eslintrc.yaml", "eslint.config.js", "eslint.config.mjs", "eslint.config.cjs", ) ) if not has_config: argv = argv + ["--no-eslintrc", "--no-config-lookup"] argv = argv + list(files) result.command = " ".join([*argv[:-len(files)], f"<{len(files)} file(s)>"]) start = time.time() try: proc = subprocess.run( argv, cwd=project_root, capture_output=True, text=True, timeout=timeout, ) result.exit_code = proc.returncode stdout = proc.stdout or "" stderr = proc.stderr or "" combined = stdout + (("\n[stderr]\n" + stderr) if stderr.strip() else "") result.logs = combined.strip() # Compact format lines look like: # path/file.js: line 3, col 5, Error - message (rule-id) result.issue_count = sum( 1 for line in stdout.splitlines() if ": line " in line and ("Error" in line or "Warning" in line) ) # ESLint exits 0 on success, 1 on lint issues, 2 on fatal error. # We treat both 1 and 2 as failure so the patch is rejected. result.passed = proc.returncode == 0 if proc.returncode == 2 and not result.reason: result.reason = "eslint fatal configuration error" except subprocess.TimeoutExpired as exc: result.exit_code = -1 result.passed = False result.reason = "timeout" result.logs = ( f"[eslint] TIMEOUT after {timeout}s\n" f"partial stdout:\n{(exc.stdout or '')[:2000]}\n" f"partial stderr:\n{(exc.stderr or '')[:1000]}" ) except FileNotFoundError: result.skipped = True result.reason = "eslint binary disappeared mid-run" except Exception as exc: # noqa: BLE001 result.passed = False result.exit_code = -1 result.reason = str(exc) result.logs = f"[eslint] launch failed: {exc}" finally: result.duration_s = round(time.time() - start, 3) result.files_checked = _rel(files, project_root) return result # --------------------------------------------------------------------------- # Public entry point # --------------------------------------------------------------------------- def run_static_analysis( project_root: str, modified_files: Sequence[str], timeout: int = DEFAULT_TIMEOUT, strict: bool = True, ) -> StaticAnalysisReport: """Run every applicable linter against ``modified_files``. Parameters ---------- project_root Root of the extracted ZIP (absolute path). modified_files Relative or absolute paths that the last patch touched. Only the files with recognised extensions will be linted; the rest are ignored silently. timeout Per-linter timeout in seconds. strict When True (default) a linter that ran and reported issues fails the overall report. When False, all failures are downgraded to warnings (``passed=True`` but ``total_issues>0``) — useful for "advisory" mode where the user just wants to see the report. Returns ------- StaticAnalysisReport Aggregate result across all linters. """ report = StaticAnalysisReport() if not project_root or not os.path.isdir(project_root): report.skipped = True report.summary = "no project_root — static analysis skipped" return report if not modified_files: report.skipped = True report.summary = "no modified files — static analysis skipped" return report py_files, js_files = _classify_files(project_root, modified_files) if not py_files and not js_files: report.skipped = True report.summary = "modified files are not python/js/ts — nothing to lint" return report if py_files: report.results.append(run_ruff(project_root, py_files, timeout=timeout)) if js_files: report.results.append(run_eslint(project_root, js_files, timeout=timeout)) ran_any = any(not r.skipped for r in report.results) if not ran_any: report.skipped = True skipped_reasons = "; ".join( f"{r.linter}: {r.reason or 'skipped'}" for r in report.results ) report.summary = f"all linters skipped ({skipped_reasons})" return report report.total_issues = sum(r.issue_count for r in report.results if not r.skipped) any_failed = any(not r.passed and not r.skipped for r in report.results) report.passed = (not any_failed) if strict else True if report.passed: report.summary = ( f"OK — {len(py_files)} py + {len(js_files)} js/ts file(s) clean" if report.total_issues == 0 else f"advisory: {report.total_issues} issue(s) (non-blocking)" ) else: report.summary = ( f"FAIL — {report.total_issues} lint issue(s) across " f"{len(py_files) + len(js_files)} file(s)" ) return report __all__ = [ "LintResult", "StaticAnalysisReport", "run_ruff", "run_eslint", "run_static_analysis", ]