cache: add expiration support and stats/evict methods

ResultCache entries now expire after max_age_seconds (default 7 days).
Expired entries are automatically pruned on lookup and can also be
bulk-evicted with evict_expired(). New stats() method returns entry
count, total bytes, and oldest timestamp for monitoring cache health.

Tests cover time-based expiry, eviction, and statistics.

Author: frankentini

src/ai_sec_scan/cache.py

@@ -12,6 +12,7 @@
 
 CACHE_VERSION = 1
 DEFAULT_CACHE_DIR = ".ai-sec-scan-cache"
+DEFAULT_MAX_AGE = 7 * 24 * 3600  # 7 days
 
 
 def file_hash(path: Path) -> str:
@@ -39,12 +40,18 @@ def _cache_key(file_path: str, provider: str, model: str) -> str:
 class ResultCache:
     """Disk-backed cache that maps (file, provider, model) to findings.
 
-    Cache entries are invalidated when the file content hash changes.
+    Cache entries are invalidated when the file content hash changes or
+    when they exceed ``max_age_seconds`` (default 7 days).
     """
 
-    def __init__(self, cache_dir: Path | None = None) -> None:
+    def __init__(
+        self,
+        cache_dir: Path | None = None,
+        max_age_seconds: int | None = None,
+    ) -> None:
         self._dir = cache_dir or Path(DEFAULT_CACHE_DIR)
         self._dir.mkdir(parents=True, exist_ok=True)
+        self._max_age = max_age_seconds if max_age_seconds is not None else DEFAULT_MAX_AGE
 
     @property
     def cache_dir(self) -> Path:
@@ -76,6 +83,13 @@ def get(
         if data.get("content_hash") != content_hash:
             return None
 
+        # Check expiration
+        ts = data.get("timestamp")
+        if self._max_age > 0 and isinstance(ts, (int, float)):
+            if time.time() - ts > self._max_age:
+                entry_path.unlink(missing_ok=True)
+                return None
+
         try:
             return [Finding.model_validate(f) for f in data.get("findings", [])]
         except Exception:
@@ -111,3 +125,60 @@ def clear(self) -> int:
             entry.unlink()
             count += 1
         return count
+
+    def evict_expired(self) -> int:
+        """Remove cache entries older than max_age_seconds.
+
+        Returns:
+            Number of entries evicted.
+        """
+        if self._max_age <= 0:
+            return 0
+
+        now = time.time()
+        evicted = 0
+        for entry_path in self._dir.glob("*.json"):
+            try:
+                data = json.loads(entry_path.read_text(encoding="utf-8"))
+            except (json.JSONDecodeError, OSError):
+                entry_path.unlink(missing_ok=True)
+                evicted += 1
+                continue
+
+            ts = data.get("timestamp")
+            if isinstance(ts, (int, float)) and now - ts > self._max_age:
+                entry_path.unlink(missing_ok=True)
+                evicted += 1
+
+        return evicted
+
+    def stats(self) -> dict[str, Any]:
+        """Return cache statistics.
+
+        Returns:
+            Dict with ``total_entries``, ``total_bytes``, and ``oldest_timestamp``.
+        """
+        total_entries = 0
+        total_bytes = 0
+        oldest: float | None = None
+
+        for entry_path in self._dir.glob("*.json"):
+            total_entries += 1
+            try:
+                total_bytes += entry_path.stat().st_size
+            except OSError:
+                continue
+            try:
+                data = json.loads(entry_path.read_text(encoding="utf-8"))
+                ts = data.get("timestamp")
+                if isinstance(ts, (int, float)):
+                    if oldest is None or ts < oldest:
+                        oldest = ts
+            except (json.JSONDecodeError, OSError):
+                continue
+
+        return {
+            "total_entries": total_entries,
+            "total_bytes": total_bytes,
+            "oldest_timestamp": oldest,
+        }

tests/test_cache.py

@@ -2,7 +2,10 @@
 
 from __future__ import annotations
 
+import json
+import time
 from pathlib import Path
+from unittest.mock import patch
 
 import pytest
 
@@ -104,3 +107,51 @@ def test_cache_dir_created(self, tmp_path: Path) -> None:
         cache_dir = tmp_path / "deep" / "nested" / "cache"
         cache = ResultCache(cache_dir=cache_dir)
         assert cache_dir.exists()
+
+    def test_expired_entry_returns_none(
+        self, tmp_path: Path, sample_finding: Finding
+    ) -> None:
+        cache = ResultCache(cache_dir=tmp_path / "cache", max_age_seconds=60)
+        cache.put("app.py", "hash1", "anthropic", "claude-3", [sample_finding])
+
+        # Simulate passage of time by patching time.time
+        with patch("ai_sec_scan.cache.time") as mock_time:
+            mock_time.time.return_value = time.time() + 120
+            result = cache.get("app.py", "hash1", "anthropic", "claude-3")
+        assert result is None
+
+    def test_non_expired_entry_returned(
+        self, tmp_path: Path, sample_finding: Finding
+    ) -> None:
+        cache = ResultCache(cache_dir=tmp_path / "cache", max_age_seconds=300)
+        cache.put("app.py", "hash1", "anthropic", "claude-3", [sample_finding])
+        result = cache.get("app.py", "hash1", "anthropic", "claude-3")
+        assert result is not None
+        assert len(result) == 1
+
+    def test_evict_expired(self, tmp_path: Path, sample_finding: Finding) -> None:
+        cache = ResultCache(cache_dir=tmp_path / "cache", max_age_seconds=60)
+        cache.put("old.py", "h1", "anthropic", "claude-3", [sample_finding])
+
+        # Manually backdate the timestamp
+        for entry in cache.cache_dir.glob("*.json"):
+            data = json.loads(entry.read_text())
+            data["timestamp"] = time.time() - 120
+            entry.write_text(json.dumps(data))
+
+        cache.put("new.py", "h2", "anthropic", "claude-3", [])
+        evicted = cache.evict_expired()
+        assert evicted == 1
+        # The fresh entry should still exist
+        assert cache.get("new.py", "h2", "anthropic", "claude-3") is not None
+
+    def test_stats(self, tmp_path: Path, sample_finding: Finding) -> None:
+        cache = ResultCache(cache_dir=tmp_path / "cache")
+        assert cache.stats()["total_entries"] == 0
+
+        cache.put("a.py", "h1", "anthropic", "claude-3", [sample_finding])
+        cache.put("b.py", "h2", "anthropic", "claude-3", [])
+        stats = cache.stats()
+        assert stats["total_entries"] == 2
+        assert stats["total_bytes"] > 0
+        assert stats["oldest_timestamp"] is not None