AirPlan-V2/lib/air_runtime/sdb_backends.py

"""AirSDB backends — 5 static analyzer backends + AnalysisDiff.

Backends:
  CppcheckBackend   — C/C++ via cppcheck
  ClangTidyBackend  — C/C++ via clang-tidy
  RustClippyBackend — Rust via cargo clippy
  GoVetBackend      — Go via go vet + staticcheck
  TypeScriptBackend — TypeScript via tsc --noEmit
"""

from __future__ import annotations

import json
import shutil
import subprocess
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path


# ---------------------------------------------------------------------------
# Unified result type
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class AnalysisResult:
    tool: str
    file: str
    line: int | None
    column: int | None
    severity: str  # error | warning | info
    message: str
    rule_id: str | None = None


# ---------------------------------------------------------------------------
# Abstract base
# ---------------------------------------------------------------------------

class StaticAnalyzerBackend(ABC):
    """Abstract base for every static-analysis backend."""

    @abstractmethod
    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        """Run the analyzer and return structured findings."""

    @property
    @abstractmethod
    def name(self) -> str:
        """Short identifier for this backend (e.g. 'cppcheck')."""

    @property
    @abstractmethod
    def install_hint(self) -> str:
        """Human-readable hint shown when the tool is not installed."""

    # -- helpers available to all backends ---------------------------------

    def _check_tool(self, tool_cmd: str) -> None:
        """Raise RuntimeError if *tool_cmd* is not on PATH."""
        if not shutil.which(tool_cmd):
            raise RuntimeError(self.install_hint)

    @staticmethod
    def _run(cmd: list[str], *, cwd: Path | None = None) -> subprocess.CompletedProcess:
        """Run *cmd* and capture stdout/stderr. Returns CompletedProcess."""
        return subprocess.run(
            cmd,
            cwd=cwd,
            capture_output=True,
            text=True,
            timeout=300,
        )


# ---------------------------------------------------------------------------
# CppcheckBackend
# ---------------------------------------------------------------------------

class CppcheckBackend(StaticAnalyzerBackend):
    """C/C++ static analysis via cppcheck."""

    name = "cppcheck"
    install_hint = (
        "cppcheck is not installed. "
        "Install it with: sudo apt install cppcheck  (Debian/Ubuntu)  "
        "or: brew install cppcheck  (macOS)"
    )

    # Template: file:line:column:severity:id:message
    _TEMPLATE = "{file}:{line}:{column}:{severity}:{id}:{message}"

    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        self._check_tool("cppcheck")

        src = str(target) if target else str(project_root)
        cmd = [
            "cppcheck",
            "--quiet",
            f"--template={self._TEMPLATE}",
            "--force",
            src,
        ]
        proc = self._run(cmd, cwd=project_root)

        results: list[AnalysisResult] = []
        for line in proc.stderr.splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split(":", 5)
            if len(parts) < 6:
                continue
            try:
                ln = int(parts[1]) if parts[1].strip() else None
            except ValueError:
                ln = None
            try:
                col = int(parts[2]) if parts[2].strip() else None
            except ValueError:
                col = None

            severity = parts[3].strip()
            # Map cppcheck severities to our unified set
            if severity not in ("error", "warning", "info"):
                if severity in ("performance", "portability", "style"):
                    severity = "warning"
                else:
                    severity = "info"

            results.append(AnalysisResult(
                tool=self.name,
                file=parts[0].strip(),
                line=ln,
                column=col,
                severity=severity,
                message=parts[5].strip(),
                rule_id=parts[4].strip() or None,
            ))
        return results


# ---------------------------------------------------------------------------
# ClangTidyBackend
# ---------------------------------------------------------------------------

class ClangTidyBackend(StaticAnalyzerBackend):
    """C/C++ static analysis via clang-tidy."""

    name = "clang-tidy"
    install_hint = (
        "clang-tidy is not installed. "
        "Install it with: sudo apt install clang-tidy  (Debian/Ubuntu)  "
        "or: brew install llvm  (macOS, then use llvm/bin/clang-tidy)"
    )

    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        self._check_tool("clang-tidy")

        src = str(target) if target else str(project_root)
        cmd = [
            "clang-tidy",
            "--quiet",
            src,
        ]
        # Use compile_commands.json if present
        comp_db = project_root / "compile_commands.json"
        if comp_db.exists():
            cmd.append(f"-p={comp_db.parent}")

        proc = self._run(cmd, cwd=project_root)

        results: list[AnalysisResult] = []
        # clang-tidy output format: <file>:<line>:<col>: warning: <message> [check-name]
        for line in proc.stderr.splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split(":", 3)
            if len(parts) < 4:
                continue
            try:
                ln = int(parts[1].strip()) if parts[1].strip() else None
            except ValueError:
                ln = None
            try:
                col = int(parts[2].strip()) if parts[2].strip() else None
            except ValueError:
                col = None

            msg_part = parts[3].strip()
            severity = "warning"
            # Detect "error:" prefix
            if msg_part.startswith("error:"):
                severity = "error"
                msg_part = msg_part[len("error:"):].strip()
            elif msg_part.startswith("warning:"):
                msg_part = msg_part[len("warning:"):].strip()
            elif msg_part.startswith("note:"):
                severity = "info"
                msg_part = msg_part[len("note:"):].strip()

            # Extract [check-name] at the end
            rule_id = None
            if msg_part.endswith("]"):
                bracket = msg_part.rfind("[")
                if bracket != -1:
                    rule_id = msg_part[bracket + 1:-1].strip()
                    msg_part = msg_part[:bracket].strip()

            results.append(AnalysisResult(
                tool=self.name,
                file=parts[0].strip(),
                line=ln,
                column=col,
                severity=severity,
                message=msg_part,
                rule_id=rule_id,
            ))
        return results


# ---------------------------------------------------------------------------
# RustClippyBackend
# ---------------------------------------------------------------------------

class RustClippyBackend(StaticAnalyzerBackend):
    """Rust static analysis via cargo clippy."""

    name = "clippy"
    install_hint = (
        "cargo clippy is not available. "
        "Install Rust toolchain: curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh  "
        "then: rustup component add clippy"
    )

    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        self._check_tool("cargo")

        cmd = [
            "cargo", "clippy",
            "--message-format=json",
        ]
        # If a specific target file/dir is given, we still run cargo clippy
        # on the whole crate (cargo does not support single-file analysis).
        proc = self._run(cmd, cwd=project_root)

        results: list[AnalysisResult] = []
        for line in proc.stdout.splitlines():
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
            except json.JSONDecodeError:
                continue
            if obj.get("reason") != "compiler-message":
                continue
            msg = obj.get("message", {})
            level = msg.get("level", "")
            if level == "error":
                severity = "error"
            elif level in ("warning",):
                severity = "warning"
            else:
                severity = "info"

            for span in msg.get("spans", []):
                results.append(AnalysisResult(
                    tool=self.name,
                    file=span.get("file_name", ""),
                    line=span.get("line_start"),
                    column=span.get("column_start"),
                    severity=severity,
                    message=msg.get("message", ""),
                    rule_id=msg.get("code", {}).get("code") or None,
                ))

        # If no JSON output (e.g. compile error), also parse stderr
        if not results and proc.stderr:
            for line in proc.stderr.splitlines():
                line = line.strip()
                if "error" in line.lower() and ":" in line:
                    results.append(AnalysisResult(
                        tool=self.name,
                        file=str(project_root),
                        line=None,
                        column=None,
                        severity="error",
                        message=line,
                        rule_id=None,
                    ))
        return results


# ---------------------------------------------------------------------------
# GoVetBackend
# ---------------------------------------------------------------------------

class GoVetBackend(StaticAnalyzerBackend):
    """Go static analysis via go vet + staticcheck."""

    name = "go-vet"
    install_hint = (
        "go is not installed. "
        "Install Go: https://go.dev/dl/  "
        "For staticcheck: go install honnef.co/go/tools/cmd/staticcheck@latest"
    )

    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        self._check_tool("go")

        results: list[AnalysisResult] = []

        # 1) go vet  — JSON output
        vet_cmd = ["go", "vet", "./..."]
        proc = self._run(vet_cmd, cwd=project_root)
        if proc.stderr:
            results.extend(self._parse_go_vet_output(proc.stderr))

        # 2) staticcheck (optional — don't fail if not installed)
        if shutil.which("staticcheck"):
            sc_cmd = ["staticcheck", "-f=json", "./..."]
            sc_proc = self._run(sc_cmd, cwd=project_root)
            for line in sc_proc.stdout.splitlines():
                line = line.strip()
                if not line:
                    continue
                try:
                    obj = json.loads(line)
                except json.JSONDecodeError:
                    continue
                severity = "warning"
                if obj.get("severity", "") == "error":
                    severity = "error"
                results.append(AnalysisResult(
                    tool="staticcheck",
                    file=obj.get("location", {}).get("file", ""),
                    line=obj.get("location", {}).get("line"),
                    column=obj.get("location", {}).get("column"),
                    severity=severity,
                    message=obj.get("message", ""),
                    rule_id=obj.get("code", ""),
                ))

        return results

    @staticmethod
    def _parse_go_vet_output(text: str) -> list[AnalysisResult]:
        """Parse go vet stderr output.

        go vet output format (non-JSON):
          <file>:<line>: <message>
        """
        results: list[AnalysisResult] = []
        for line in text.splitlines():
            line = line.strip()
            if not line:
                continue
            parts = line.split(":", 2)
            if len(parts) < 3:
                continue
            try:
                ln = int(parts[1].strip()) if parts[1].strip() else None
            except ValueError:
                ln = None
            results.append(AnalysisResult(
                tool="go vet",
                file=parts[0].strip(),
                line=ln,
                column=None,
                severity="warning",
                message=parts[2].strip(),
                rule_id=None,
            ))
        return results


# ---------------------------------------------------------------------------
# TypeScriptBackend
# ---------------------------------------------------------------------------

class TypeScriptBackend(StaticAnalyzerBackend):
    """TypeScript static analysis via tsc --noEmit."""

    name = "tsc"
    install_hint = (
        "tsc (TypeScript compiler) is not installed. "
        "Install it with: npm install -g typescript  "
        "or add it to your project: npm install --save-dev typescript"
    )

    def analyze(self, project_root: Path, target: Path | None = None) -> list[AnalysisResult]:
        # tsc can be installed locally (npx) or globally
        tsc_cmd = self._find_tsc()
        if tsc_cmd is None:
            raise RuntimeError(self.install_hint)

        cmd = tsc_cmd + ["--noEmit", "--pretty", "false"]
        proc = self._run(cmd, cwd=project_root)

        results: list[AnalysisResult] = []
        # tsc output format: <file>(<line>,<col>): error TS<code>: <message>
        for line in proc.stdout.splitlines():
            line = line.strip()
            if not line:
                continue
            results.append(self._parse_tsc_line(line))
        return results

    def _find_tsc(self) -> list[str] | None:
        """Return the tsc command as a list, or None if not found."""
        if shutil.which("tsc"):
            return ["tsc"]
        if shutil.which("npx"):
            return ["npx", "tsc"]
        return None

    @staticmethod
    def _parse_tsc_line(line: str) -> AnalysisResult:
        """Parse a single tsc diagnostic line.

        Format: <file>(<line>,<col>): error TS1234: <message>
        """
        severity = "error"
        rule_id = None

        # Split on the first colon-space after the position paren
        # e.g. "src/foo.ts(10,5): error TS2322: Type 'string' ..."
        main_parts = line.split(": ", 1)
        location_part = main_parts[0] if main_parts else line
        message = main_parts[1].strip() if len(main_parts) > 1 else ""

        # Extract file, line, column from "file(line,col)"
        file_part = location_part
        ln = None
        col = None
        paren = location_part.rfind("(")
        if paren != -1 and location_part.endswith(")"):
            file_part = location_part[:paren]
            pos_str = location_part[paren + 1:-1]
            pos_parts = pos_str.split(",", 1)
            try:
                ln = int(pos_parts[0].strip()) if pos_parts[0].strip() else None
            except ValueError:
                pass
            if len(pos_parts) > 1:
                try:
                    col = int(pos_parts[1].strip()) if pos_parts[1].strip() else None
                except ValueError:
                    pass

        # Extract severity + rule from " error TS2322" in the remainder
        if len(main_parts) > 1:
            # The part between the first colon-space and the message
            # is in the original line — re-parse
            rest = line[len(location_part) + 2:]  # after ": "
            if rest.startswith("error "):
                severity = "error"
                rest = rest[len("error "):]
            elif rest.startswith("warning "):
                severity = "warning"
                rest = rest[len("warning "):]
            # rest now starts with "TS1234: message"
            ts_parts = rest.split(": ", 1)
            if ts_parts:
                rule_id = ts_parts[0].strip() or None
                if len(ts_parts) > 1:
                    message = ts_parts[1].strip()

        return AnalysisResult(
            tool="tsc",
            file=file_part,
            line=ln,
            column=col,
            severity=severity,
            message=message,
            rule_id=rule_id,
        )


# ---------------------------------------------------------------------------
# AnalysisDiff
# ---------------------------------------------------------------------------

class AnalysisDiff:
    """Compare two lists of AnalysisResult and classify findings as
    new, resolved, or unchanged."""

    @staticmethod
    def _key(r: AnalysisResult) -> tuple[str, int | None, str | None]:
        """Dedup key: (file, line, rule_id)."""
        return (r.file, r.line, r.rule_id)

    def diff(
        self,
        before: list[AnalysisResult],
        after: list[AnalysisResult],
    ) -> dict:
        before_keys = {self._key(r): r for r in before}
        after_keys = {self._key(r): r for r in after}

        before_set = set(before_keys.keys())
        after_set = set(after_keys.keys())

        new_keys = after_set - before_set
        resolved_keys = before_set - after_set
        unchanged_keys = before_set & after_set

        return {
            "new": [after_keys[k] for k in new_keys],
            "resolved": [before_keys[k] for k in resolved_keys],
            "unchanged": [before_keys[k] for k in unchanged_keys],
        }


# ---------------------------------------------------------------------------
# Convenience registry
# ---------------------------------------------------------------------------

BACKENDS: dict[str, StaticAnalyzerBackend] = {
    "cppcheck": CppcheckBackend(),
    "clang-tidy": ClangTidyBackend(),
    "clippy": RustClippyBackend(),
    "go-vet": GoVetBackend(),
    "tsc": TypeScriptBackend(),
}