core/static_analysis.py

"""
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",
]