manual_slop/src/summary_cache.py

import hashlib
import json

from pathlib import Path
from typing  import Optional, Dict

from src.result_types import Result, ErrorInfo, ErrorKind


def get_file_hash(content: str) -> str:
 """
 Returns SHA256 hash of the content.
 [C: tests/test_summary_cache.py:test_get_file_hash, tests/test_summary_cache.py:test_summary_cache]
 """
 return hashlib.sha256(content.encode("utf-8")).hexdigest()

class SummaryCache:
 """
 A hash-based cache for file summaries to avoid redundant processing.
 Invalidates when content hash changes.
 """
 def __init__(self, cache_file: Optional[str] = None, max_entries: int = 1000):
  if cache_file:
   self.cache_file = Path(cache_file)
  else:
   # Default relative to current working directory
   self.cache_file = Path(".slop_cache/summary_cache.json")
  self.max_entries = max_entries
  self.cache: Dict[str, Dict[str, str]] = {}
  self.load()

 def load(self) -> Result[bool]:
  """
  Loads cache from disk.
  [C: src/tool_presets.py:ToolPresetManager._read_raw, src/workspace_manager.py:WorkspaceManager._load_file, tests/test_gui_phase3.py:test_create_track, tests/test_history_management.py:test_save_separation, tests/test_session_logging.py:test_open_session_creates_subdir_and_registry]
  """
  if not self.cache_file.exists():
   return Result(data=False)
  try:
   with open(self.cache_file, "r", encoding="utf-8") as f:
    self.cache = json.load(f)
   return Result(data=True)
  except (OSError, json.JSONDecodeError) as e:
   self.cache = {}
   return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.load", original=e)])

 def save(self) -> Result[bool]:
  """Saves cache to disk."""
  try:
   self.cache_file.parent.mkdir(parents=True, exist_ok=True)
   with open(self.cache_file, "w", encoding="utf-8") as f:
    json.dump(self.cache, f, indent=1)
   return Result(data=True)
  except OSError as e:
   return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.save", original=e)])

 def get_summary(self, file_path: str, content_hash: str) -> str:
  """
  Returns cached summary if hash matches, otherwise "".
  [C: tests/test_summary_cache.py:test_summary_cache, tests/test_summary_cache.py:test_summary_cache_lru]
  """
  entry = self.cache.get(file_path)
  if entry and entry.get("hash") == content_hash:
   # LRU: move to end
   val = self.cache.pop(file_path)
   self.cache[file_path] = val
   return val.get("summary") or ""
  return ""

 def set_summary(self, file_path: str, content_hash: str, summary: str) -> None:
  """
  Stores summary in cache and saves to disk.
  [C: tests/test_summary_cache.py:test_summary_cache, tests/test_summary_cache.py:test_summary_cache_lru]
  """
  if file_path in self.cache:
   self.cache.pop(file_path)
  self.cache[file_path] = {
   "hash": content_hash,
   "summary": summary
  }
  # Enforce LRU size limit
  while len(self.cache) > self.max_entries:
   # pop first item (oldest)
   first_key = next(iter(self.cache))
   self.cache.pop(first_key)
  self.save()

 def clear(self) -> Result[bool]:
  """
  Clears the cache both in-memory and on disk.
  [C: tests/conftest.py:reset_ai_client]
  """
  self.cache.clear()
  if not self.cache_file.exists():
   return Result(data=True)
  try:
   self.cache_file.unlink()
   return Result(data=True)
  except OSError as e:
   return Result(data=False, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.clear", original=e)])

 def get_stats(self) -> Result[dict]:
  """Returns dictionary of cache statistics."""
  size_bytes = 0
  if self.cache_file.exists():
   try:
    size_bytes = self.cache_file.stat().st_size
   except OSError as e:
    return Result(data={"entries": len(self.cache), "size_bytes": 0}, errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="summary_cache.get_stats", original=e)])
  return Result(data={
   "entries": len(self.cache),
   "size_bytes": size_bytes
  })