lint_cache.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/linter/lint_cache.py
 
"""
Purpose
-------
FS-510 - incremental lint cache. Content-hash keyed so unchanged files
are not re-linted on subsequent runs.
 
Responsibilities
----------------
- Compute deterministic content hashes for source files.
- Compute the linter's own code identity, so an edit to a module that
  decides a verdict invalidates every cached verdict (plan 265).
- Store and retrieve per-file lint results from a JSON cache file.
- Invalidate entries when file content, active rules, the OCT version or
  the linter's own code change.
- Thread-safe read/write via ``threading.Lock``.
 
Diagnostics
-----------
Domain: OCT-LINTER
Levels:
    L2 - (none) the cache save is reported at L3
    L3 - cache hits/misses, the entry count at save, the header field
         that caused a whole-store drop, and an unreadable linter package
    L4 - cache I/O details, including the path loaded from
 
Contracts
---------
- Cache file lives at ``.option_c/cache/lint-cache.json``.
- Corrupt or missing cache file is silently replaced (no crash).
- Cache is never authoritative - a ``--no-cache`` flag bypasses reads.
- The linter identity is a STORE-level property, not a per-entry one: a
  linter edit invalidates everything, so ``get``/``put`` keep their
  signatures and the per-entry key is unchanged.
- An unreadable linter package yields the stable string ``"unavailable"``
  rather than a guessed digest, so a frozen install still caches.
- Every user of the shared store (``oct lint`` and ``oct health``) builds
  it with :func:`current_linter_identity`. Two users with different
  identities discard each other's entries on every load.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
"""
 
from __future__ import annotations
 
import hashlib
import json
import threading
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any
from oct.core.diagnostics import _dbg
 
 
_SCHEMA_VERSION = "1.0"
 
#: Plan 265: the identity of a linter package that cannot be read - a
#: frozen or zipped install. Stable, so such an install still caches;
#: never a guessed digest.
IDENTITY_UNAVAILABLE = "unavailable"
 
#: Plan 265: verdict-deciding files that live outside the linter package,
#: as paths relative to the package's parent (``oct/oct/``). The three
#: ``deps/`` modules decide the ``dependencies_match`` verdict (2026-09-28).
_EXTRA_IDENTITY_FILES = (
    "tools/secret_scanner.py",
    "deps/dependency_check.py",
    "deps/import_graph.py",
    "deps/oct_deps.py",
)
 
 
def content_hash(text: str) -> str:
    """Return a SHA-256 hex digest of *text*."""
    return hashlib.sha256(text.encode("utf-8")).hexdigest()
 
 
def rules_hash(active_rules: dict[str, bool]) -> str:
    """Return a SHA-256 hex digest of the sorted active-rules dict."""
    canonical = json.dumps(active_rules, sort_keys=True)
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
 
 
def linter_identity_hash(package_dir: Path) -> str:
    """Return a SHA-256 hex digest of the linter's own code (plan 265).
 
    The digest covers every ``*.py`` directly under *package_dir* plus
    the verdict-deciding files named in ``_EXTRA_IDENTITY_FILES``, sorted
    by their path relative to *package_dir*'s parent. Each file's
    contribution is prefixed by that relative path, so a rename moves the
    digest even when no byte of any file changes.
 
    Returns ``IDENTITY_UNAVAILABLE`` when the package cannot be read
    (a frozen or zipped install), logging the reason once; a guessed
    digest would be worse than a stable sentinel.
    """
    func = "linter_identity_hash"
    anchor = package_dir.parent
    if not package_dir.is_dir():
        _dbg(
            "OCT-LINTER", func,
            f"linter identity unavailable: {package_dir} is not a directory", 3,
        )
        return IDENTITY_UNAVAILABLE
    try:
        members = [
            (f"{package_dir.name}/{path.name}", path)
            for path in package_dir.glob("*.py")
            if path.is_file()
        ]
        for relative in _EXTRA_IDENTITY_FILES:
            candidate = anchor / relative
            if candidate.is_file():
                members.append((relative, candidate))
        members.sort(key=lambda member: member[0])
        digest = hashlib.sha256()
        for relative, path in members:
            digest.update(relative.encode("utf-8"))
            digest.update(b"\0")
            digest.update(path.read_bytes())
            digest.update(b"\0")
    except OSError as exc:
        _dbg("OCT-LINTER", func, f"linter identity unavailable: {exc}", 3)
        return IDENTITY_UNAVAILABLE
    _dbg("OCT-LINTER", func, f"linter identity over {len(members)} file(s)", 4)
    return digest.hexdigest()
 
 
def current_linter_identity() -> str:
    """Return the identity of the linter package this module belongs to.
 
    The one place that names the package :func:`linter_identity_hash`
    covers, so every user of the shared store writes the same header.
    Before 2026-09-28 ``oct health`` built its store without an identity,
    so the header flipped between the digest and ``"unavailable"``, and
    ``oct lint`` and ``oct health`` discarded each other's entries on
    every alternation.
    """
    return linter_identity_hash(Path(__file__).resolve().parent)
 
 
@dataclass
class LintCacheEntry:
    """One cached lint result for a single file."""
 
    content_hash: str
    rule_results: dict[str, bool]
    json_detail: dict[str, Any]
    oct_version: str
    timestamp: str
 
 
class LintCache:
    """Thread-safe, content-addressed lint result cache.
 
    Loads from *cache_path* on init; call :meth:`save` to persist.
 
    *linter_identity* is the digest of the linter's own code (plan 265),
    computed once per construction by :func:`linter_identity_hash`. It is
    a whole-store property: a linter edit invalidates every entry, so it
    lives in the header beside *oct_version* rather than in the key.
    """
 
    def __init__(
        self,
        cache_path: Path,
        oct_version: str,
        linter_identity: str = IDENTITY_UNAVAILABLE,
    ) -> None:
        self._path = cache_path
        self._oct_version = oct_version
        self._linter_identity = linter_identity
        self._lock = threading.Lock()
        self._entries: dict[str, dict] = {}
        self._load()
 
    def _load(self) -> None:
        func = "_load"
        _dbg("OCT-LINTER", func, f"loading lint cache from {self._path}", 4)
        if not self._path.is_file():
            return
        try:
            raw = json.loads(self._path.read_text(encoding="utf-8"))
            if not isinstance(raw, dict):
                return
            if raw.get("_schema_version") != _SCHEMA_VERSION:
                return
            if raw.get("oct_version") != self._oct_version:
                _dbg(
                    "OCT-LINTER", func,
                    "dropping the whole lint cache: oct_version differs", 3,
                )
                return
            if raw.get("linter_identity") != self._linter_identity:
                _dbg(
                    "OCT-LINTER", func,
                    "dropping the whole lint cache: linter_identity differs", 3,
                )
                return
            entries = raw.get("entries")
            if isinstance(entries, dict):
                self._entries = entries
        except (json.JSONDecodeError, OSError):
            self._entries = {}
 
    def get(self, file_hash: str, active_rules_hash: str) -> LintCacheEntry | None:
        """Return cached entry if content + rules match, else ``None``."""
        with self._lock:
            key = f"{file_hash}:{active_rules_hash}"
            entry = self._entries.get(key)
            if entry is None:
                return None
            if entry.get("content_hash") != file_hash:
                return None
            return LintCacheEntry(
                content_hash=entry["content_hash"],
                rule_results=entry["rule_results"],
                json_detail=entry["json_detail"],
                oct_version=entry["oct_version"],
                timestamp=entry["timestamp"],
            )
 
    def put(
        self,
        file_hash: str,
        active_rules_hash: str,
        entry: LintCacheEntry,
    ) -> None:
        """Store a lint result in the cache."""
        with self._lock:
            key = f"{file_hash}:{active_rules_hash}"
            self._entries[key] = asdict(entry)
 
    def save(self) -> None:
        """Write the cache to disk. Creates parent directories if needed."""
        func = "save"
        _dbg("OCT-LINTER", func, f"saving lint cache: {self.entry_count} entries", 3)
        with self._lock:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            payload = {
                "_schema_version": _SCHEMA_VERSION,
                "oct_version": self._oct_version,
                "linter_identity": self._linter_identity,
                "entries": self._entries,
            }
            self._path.write_text(
                json.dumps(payload, indent=2), encoding="utf-8",
            )
 
    @property
    def entry_count(self) -> int:
        """Number of cached entries (for diagnostics)."""
        with self._lock:
            return len(self._entries)