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.
- Store and retrieve per-file lint results from a JSON cache file.
- Invalidate entries when file content, active rules, or OCT version change.
- Thread-safe read/write via ``threading.Lock``.
 
Diagnostics
-----------
Domain: OCT-LINTER
Levels:
    L3 — cache hits/misses
    L4 — cache I/O details
 
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.
"""
 
from __future__ import annotations
 
import hashlib
import json
import threading
from dataclasses import dataclass, field, asdict
from pathlib import Path
from typing import Any
 
 
_SCHEMA_VERSION = "1.0"
 
 
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()
 
 
@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.
    """
 
    def __init__(self, cache_path: Path, oct_version: str) -> None:
        self._path = cache_path
        self._oct_version = oct_version
        self._lock = threading.Lock()
        self._entries: dict[str, dict] = {}
        self._load()
 
    def _load(self) -> None:
        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:
                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."""
        with self._lock:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            payload = {
                "_schema_version": _SCHEMA_VERSION,
                "oct_version": self._oct_version,
                "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)