feat: add tool-level file cache with LRU eviction and mtime invalidation

Introduces FileCache (OrderedDict LRU, st_mtime_ns validation) to avoid
redundant disk reads and duplicate content in conversation context.
Read tools return a short "[Cached]" message on cache hit instead of
resending unchanged file content, saving tokens. Write/edit/delete tools
invalidate affected paths; str_replace pre-warms the cache after edits.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-11 22:26:51 -05:00
parent 2c532adbbc
commit d829e6553c
8 changed files with 623 additions and 10 deletions

View File

@@ -73,11 +73,19 @@ class ShellToolConfig(BaseModel):
max_output_bytes: int = Field(default=65536, description="Max output capture size in bytes") max_output_bytes: int = Field(default=65536, description="Max output capture size in bytes")
class FileCacheConfig(BaseModel):
"""File cache configuration."""
enabled: bool = Field(default=True, description="Enable file content caching")
max_entries: int = Field(default=128, description="Maximum cached file entries (LRU eviction)")
class FilesystemToolConfig(BaseModel): class FilesystemToolConfig(BaseModel):
"""Filesystem tool limits.""" """Filesystem tool limits."""
max_file_size_bytes: int = Field(default=1_048_576, description="Max file size for read/write") max_file_size_bytes: int = Field(default=1_048_576, description="Max file size for read/write")
binary_detection: bool = Field(default=True, description="Detect and reject binary files") binary_detection: bool = Field(default=True, description="Detect and reject binary files")
cache: FileCacheConfig = Field(default_factory=FileCacheConfig, description="File cache settings")
class ToolsConfig(BaseModel): class ToolsConfig(BaseModel):

View File

@@ -1,13 +1,17 @@
"""Edit tools: str_replace and patch_apply.""" """Edit tools: str_replace and patch_apply."""
from __future__ import annotations
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from app.models.config import AppConfig
from app.models.tool_call import ToolResult, ToolResultStatus from app.models.tool_call import ToolResult, ToolResultStatus
from app.tools.base import BaseTool from app.tools.base import BaseTool
from app.utils.file_cache import FileCache, cached_read_file
from app.utils.file_helpers import ( from app.utils.file_helpers import (
FileSizeError, FileSizeError,
PathSecurityError, PathSecurityError,
@@ -37,6 +41,12 @@ class StrReplaceTool(BaseTool):
) )
params_model = StrReplaceParams params_model = StrReplaceParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute( def execute(
self, *, tool_call_id: str, file_path: str, old_str: str, new_str: str, **kwargs: Any self, *, tool_call_id: str, file_path: str, old_str: str, new_str: str, **kwargs: Any
) -> ToolResult: ) -> ToolResult:
@@ -44,11 +54,12 @@ class StrReplaceTool(BaseTool):
# Read the file # Read the file
try: try:
content = safe_read_file( content = cached_read_file(
file_path, file_path,
self.workspace_root, self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes, max_size_bytes=fs_config.max_file_size_bytes,
check_binary=fs_config.binary_detection, check_binary=fs_config.binary_detection,
cache=self._file_cache,
) )
except PathSecurityError as exc: except PathSecurityError as exc:
return ToolResult( return ToolResult(
@@ -117,8 +128,14 @@ class StrReplaceTool(BaseTool):
safe_path = resolve_safe_path(file_path, self.workspace_root) safe_path = resolve_safe_path(file_path, self.workspace_root)
rel_path = safe_path.relative_to(self.workspace_root) rel_path = safe_path.relative_to(self.workspace_root)
except (PathSecurityError, ValueError): except (PathSecurityError, ValueError):
safe_path = None
rel_path = Path(file_path) rel_path = Path(file_path)
# Pre-warm cache with the new content (we already have it in memory).
if self._file_cache is not None and safe_path is not None:
self._file_cache.invalidate(safe_path)
self._file_cache.put(safe_path, new_content)
return ToolResult( return ToolResult(
tool_call_id=tool_call_id, tool_call_id=tool_call_id,
tool_name=self.name, tool_name=self.name,
@@ -144,6 +161,12 @@ class PatchApplyTool(BaseTool):
) )
params_model = PatchApplyParams params_model = PatchApplyParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute(self, *, tool_call_id: str, file_path: str, patch: str, **kwargs: Any) -> ToolResult: def execute(self, *, tool_call_id: str, file_path: str, patch: str, **kwargs: Any) -> ToolResult:
try: try:
safe_path = resolve_safe_path(file_path, self.workspace_root) safe_path = resolve_safe_path(file_path, self.workspace_root)
@@ -195,6 +218,9 @@ class PatchApplyTool(BaseTool):
error=f"Patch failed (exit {result.returncode}): {result.stderr or result.stdout}", error=f"Patch failed (exit {result.returncode}): {result.stderr or result.stdout}",
) )
if self._file_cache is not None:
self._file_cache.invalidate(safe_path)
try: try:
rel_path = safe_path.relative_to(self.workspace_root) rel_path = safe_path.relative_to(self.workspace_root)
except ValueError: except ValueError:

View File

@@ -1,12 +1,16 @@
"""Filesystem tools: read_file, list_dir, write_file, make_dir, delete_file.""" """Filesystem tools: read_file, list_dir, write_file, make_dir, delete_file."""
from __future__ import annotations
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
from app.models.config import AppConfig
from app.models.tool_call import ToolResult, ToolResultStatus from app.models.tool_call import ToolResult, ToolResultStatus
from app.tools.base import BaseTool from app.tools.base import BaseTool
from app.utils.file_cache import FileCache, cached_read_file
from app.utils.file_helpers import ( from app.utils.file_helpers import (
BinaryFileError, BinaryFileError,
FileSizeError, FileSizeError,
@@ -36,14 +40,22 @@ class ReadFileTool(BaseTool):
description = "Read the full contents of a text file. Returns the file content as a string." description = "Read the full contents of a text file. Returns the file content as a string."
params_model = ReadFileParams params_model = ReadFileParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult: def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult:
fs_config = self.config.tools.filesystem fs_config = self.config.tools.filesystem
hits_before = self._file_cache.stats.hits if self._file_cache else 0
try: try:
content = safe_read_file( content = cached_read_file(
file_path, file_path,
self.workspace_root, self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes, max_size_bytes=fs_config.max_file_size_bytes,
check_binary=fs_config.binary_detection, check_binary=fs_config.binary_detection,
cache=self._file_cache,
) )
except PathSecurityError as exc: except PathSecurityError as exc:
return ToolResult( return ToolResult(
@@ -74,6 +86,23 @@ class ReadFileTool(BaseTool):
error=str(exc), error=str(exc),
) )
# On cache hit the file is unchanged — its content is already in
# conversation context from the earlier read, so avoid resending it.
was_cache_hit = (
self._file_cache is not None
and self._file_cache.stats.hits > hits_before
)
if was_cache_hit:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=(
f"[Cached] {file_path} is unchanged since last read "
f"({len(content):,} chars). Content is already in conversation context."
),
)
return ToolResult( return ToolResult(
tool_call_id=tool_call_id, tool_call_id=tool_call_id,
tool_name=self.name, tool_name=self.name,
@@ -92,6 +121,12 @@ class ReadManyFilesTool(BaseTool):
) )
params_model = ReadManyFilesParams params_model = ReadManyFilesParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute(self, *, tool_call_id: str, file_paths: list[str], **kwargs: Any) -> ToolResult: def execute(self, *, tool_call_id: str, file_paths: list[str], **kwargs: Any) -> ToolResult:
if not file_paths: if not file_paths:
return ToolResult( return ToolResult(
@@ -106,14 +141,26 @@ class ReadManyFilesTool(BaseTool):
success_count = 0 success_count = 0
for fp in file_paths: for fp in file_paths:
hits_before = self._file_cache.stats.hits if self._file_cache else 0
try: try:
content = safe_read_file( content = cached_read_file(
fp, fp,
self.workspace_root, self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes, max_size_bytes=fs_config.max_file_size_bytes,
check_binary=fs_config.binary_detection, check_binary=fs_config.binary_detection,
cache=self._file_cache,
) )
sections.append(f"=== {fp} ===\n{content}") was_hit = (
self._file_cache is not None
and self._file_cache.stats.hits > hits_before
)
if was_hit:
sections.append(
f"=== {fp} ===\n[Cached] Unchanged since last read "
f"({len(content):,} chars). Already in conversation context."
)
else:
sections.append(f"=== {fp} ===\n{content}")
success_count += 1 success_count += 1
except (PathSecurityError, FileNotFoundError, FileSizeError, BinaryFileError) as exc: except (PathSecurityError, FileNotFoundError, FileSizeError, BinaryFileError) as exc:
sections.append(f"=== {fp} ===\n[ERROR] {exc}") sections.append(f"=== {fp} ===\n[ERROR] {exc}")
@@ -225,6 +272,12 @@ class WriteFileTool(BaseTool):
) )
params_model = WriteFileParams params_model = WriteFileParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute(self, *, tool_call_id: str, file_path: str, content: str, **kwargs: Any) -> ToolResult: def execute(self, *, tool_call_id: str, file_path: str, content: str, **kwargs: Any) -> ToolResult:
fs_config = self.config.tools.filesystem fs_config = self.config.tools.filesystem
try: try:
@@ -249,6 +302,9 @@ class WriteFileTool(BaseTool):
error=str(exc), error=str(exc),
) )
if self._file_cache is not None:
self._file_cache.invalidate(safe_path)
try: try:
rel_path = safe_path.relative_to(self.workspace_root) rel_path = safe_path.relative_to(self.workspace_root)
except ValueError: except ValueError:
@@ -330,6 +386,12 @@ class DeleteFileTool(BaseTool):
description = "Delete a single file. Does not delete directories." description = "Delete a single file. Does not delete directories."
params_model = DeleteFileParams params_model = DeleteFileParams
def __init__(
self, workspace_root: Path, config: AppConfig, file_cache: FileCache | None = None
) -> None:
super().__init__(workspace_root, config)
self._file_cache = file_cache
def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult: def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult:
try: try:
safe_path = resolve_safe_path(file_path, self.workspace_root) safe_path = resolve_safe_path(file_path, self.workspace_root)
@@ -367,6 +429,9 @@ class DeleteFileTool(BaseTool):
error=f"Failed to delete file: {exc}", error=f"Failed to delete file: {exc}",
) )
if self._file_cache is not None:
self._file_cache.invalidate(safe_path)
try: try:
rel_path = safe_path.relative_to(self.workspace_root) rel_path = safe_path.relative_to(self.workspace_root)
except ValueError: except ValueError:

View File

@@ -8,6 +8,7 @@ from typing import TYPE_CHECKING, Any
from app.models.config import AppConfig from app.models.config import AppConfig
from app.tools.base import BaseTool from app.tools.base import BaseTool
from app.utils.file_cache import FileCache
if TYPE_CHECKING: if TYPE_CHECKING:
from app.services.skills import SkillsManager from app.services.skills import SkillsManager
@@ -89,6 +90,7 @@ def create_default_registry(
config: AppConfig, config: AppConfig,
skills_manager: SkillsManager | None = None, skills_manager: SkillsManager | None = None,
skill_runner: object | None = None, skill_runner: object | None = None,
file_cache: FileCache | None = None,
) -> ToolRegistry: ) -> ToolRegistry:
"""Create a ToolRegistry populated with all built-in tools. """Create a ToolRegistry populated with all built-in tools.
@@ -97,6 +99,7 @@ def create_default_registry(
config: Application configuration. config: Application configuration.
skills_manager: Optional skills manager for skill tools. skills_manager: Optional skills manager for skill tools.
skill_runner: Optional SkillRunner for package skill activation. skill_runner: Optional SkillRunner for package skill activation.
file_cache: Optional file cache shared across file-reading tools.
""" """
# Read tools # Read tools
from app.tools.filesystem import ListDirTool, ReadFileTool, ReadManyFilesTool from app.tools.filesystem import ListDirTool, ReadFileTool, ReadManyFilesTool
@@ -119,8 +122,8 @@ def create_default_registry(
registry = ToolRegistry() registry = ToolRegistry()
# Read # Read
registry.register(ReadFileTool(workspace_root, config)) registry.register(ReadFileTool(workspace_root, config, file_cache=file_cache))
registry.register(ReadManyFilesTool(workspace_root, config)) registry.register(ReadManyFilesTool(workspace_root, config, file_cache=file_cache))
registry.register(ListDirTool(workspace_root, config)) registry.register(ListDirTool(workspace_root, config))
# Search # Search
@@ -128,13 +131,13 @@ def create_default_registry(
registry.register(FindFilesTool(workspace_root, config)) registry.register(FindFilesTool(workspace_root, config))
# Write # Write
registry.register(WriteFileTool(workspace_root, config)) registry.register(WriteFileTool(workspace_root, config, file_cache=file_cache))
registry.register(MakeDirTool(workspace_root, config)) registry.register(MakeDirTool(workspace_root, config))
registry.register(DeleteFileTool(workspace_root, config)) registry.register(DeleteFileTool(workspace_root, config, file_cache=file_cache))
# Edit # Edit
registry.register(StrReplaceTool(workspace_root, config)) registry.register(StrReplaceTool(workspace_root, config, file_cache=file_cache))
registry.register(PatchApplyTool(workspace_root, config)) registry.register(PatchApplyTool(workspace_root, config, file_cache=file_cache))
# Shell # Shell
registry.register(RunCommandTool(workspace_root, config)) registry.register(RunCommandTool(workspace_root, config))

View File

@@ -97,11 +97,20 @@ class SneakyCodeApp(App):
self._config.skills, self._config.agent.workspace_root self._config.skills, self._config.agent.workspace_root
) )
# Create file cache if enabled
self._file_cache = None
fs_cache_cfg = self._config.tools.filesystem.cache
if fs_cache_cfg.enabled:
from app.utils.file_cache import FileCache
self._file_cache = FileCache(max_entries=fs_cache_cfg.max_entries)
# Create tool registry (SkillRunner wired after registry exists) # Create tool registry (SkillRunner wired after registry exists)
self._tool_registry = create_default_registry( self._tool_registry = create_default_registry(
self._config.agent.workspace_root, self._config.agent.workspace_root,
self._config, self._config,
skills_manager=self._skills_manager, skills_manager=self._skills_manager,
file_cache=self._file_cache,
) )
# Create SkillRunner and late-bind it to skill tools # Create SkillRunner and late-bind it to skill tools

185
app/utils/file_cache.py Normal file
View File

@@ -0,0 +1,185 @@
"""File cache with LRU eviction and mtime-based invalidation."""
from __future__ import annotations
from collections import OrderedDict
from dataclasses import dataclass, field
from pathlib import Path
from app.utils.file_helpers import (
BinaryFileError,
FileSizeError,
PathSecurityError,
check_file_size,
is_binary_file,
resolve_safe_path,
)
@dataclass(slots=True)
class CacheEntry:
"""A cached file's content and modification timestamp."""
content: str
mtime_ns: int
@dataclass
class CacheStats:
"""Running statistics for a FileCache instance."""
hits: int = 0
misses: int = 0
invalidations: int = 0
evictions: int = 0
@property
def hit_rate(self) -> float:
"""Return cache hit rate as a float between 0.0 and 1.0."""
total = self.hits + self.misses
if total == 0:
return 0.0
return self.hits / total
class FileCache:
"""LRU file-content cache with mtime-based invalidation.
Keyed by resolved absolute ``Path``. Each lookup performs a cheap
``stat()`` syscall to verify the file hasn't changed on disk — if the
nanosecond mtime differs the entry is evicted and the caller gets a
cache miss.
Not thread-safe (single-threaded agent loop).
"""
def __init__(self, max_entries: int = 128) -> None:
self._max_entries = max_entries
self._entries: OrderedDict[Path, CacheEntry] = OrderedDict()
self._stats = CacheStats()
# -- public API --------------------------------------------------
def get(self, path: Path) -> str | None:
"""Return cached content if *path* hasn't changed, else ``None``.
A ``stat()`` call checks ``st_mtime_ns``; on mismatch the stale
entry is silently removed.
"""
entry = self._entries.get(path)
if entry is None:
self._stats.misses += 1
return None
try:
current_mtime_ns = path.stat().st_mtime_ns
except OSError:
# File gone — evict and miss.
self._remove(path)
self._stats.misses += 1
return None
if current_mtime_ns != entry.mtime_ns:
self._remove(path)
self._stats.invalidations += 1
self._stats.misses += 1
return None
# Cache hit — move to end (most-recently used).
self._entries.move_to_end(path)
self._stats.hits += 1
return entry.content
def put(self, path: Path, content: str) -> None:
"""Store *content* for *path* with its current ``st_mtime_ns``.
Evicts the least-recently-used entry when over capacity.
"""
try:
mtime_ns = path.stat().st_mtime_ns
except OSError:
# Can't stat — don't cache.
return
if path in self._entries:
# Update existing; move to end.
self._entries[path] = CacheEntry(content=content, mtime_ns=mtime_ns)
self._entries.move_to_end(path)
else:
self._entries[path] = CacheEntry(content=content, mtime_ns=mtime_ns)
# Evict LRU if over capacity.
while len(self._entries) > self._max_entries:
self._entries.popitem(last=False)
self._stats.evictions += 1
def invalidate(self, path: Path) -> None:
"""Remove *path* from the cache if present."""
if path in self._entries:
del self._entries[path]
self._stats.invalidations += 1
def clear(self) -> None:
"""Remove all entries."""
self._entries.clear()
@property
def stats(self) -> CacheStats:
"""Return the running cache statistics."""
return self._stats
def __len__(self) -> int:
return len(self._entries)
# -- internals ---------------------------------------------------
def _remove(self, path: Path) -> None:
"""Delete an entry without bumping invalidation stats."""
self._entries.pop(path, None)
def cached_read_file(
path: str | Path,
workspace_root: Path,
max_size_bytes: int = 1_048_576,
check_binary: bool = True,
cache: FileCache | None = None,
) -> str:
"""Read a file with full security checks, using *cache* when available.
Security checks (path sandboxing, size limit, binary detection) run on
**every** call — only the ``Path.read_text()`` I/O is skipped on a cache
hit.
When *cache* is ``None`` this behaves identically to
:func:`~app.utils.file_helpers.safe_read_file`.
Raises:
PathSecurityError: If the path escapes the workspace.
FileSizeError: If the file is too large.
BinaryFileError: If the file is binary and *check_binary* is True.
FileNotFoundError: If the file does not exist.
"""
safe_path = resolve_safe_path(path, workspace_root)
if not safe_path.exists():
raise FileNotFoundError(f"File not found: {safe_path}")
check_file_size(safe_path, max_size_bytes)
if check_binary and is_binary_file(safe_path):
raise BinaryFileError(f"File appears to be binary: {safe_path}")
# Try cache.
if cache is not None:
cached = cache.get(safe_path)
if cached is not None:
return cached
# Cache miss (or no cache) — read from disk.
content = safe_path.read_text(encoding="utf-8")
if cache is not None:
cache.put(safe_path, content)
return content

View File

@@ -68,6 +68,9 @@ tools:
filesystem: filesystem:
max_file_size_bytes: 1048576 # 1 MB max_file_size_bytes: 1048576 # 1 MB
binary_detection: true binary_detection: true
cache:
enabled: true
max_entries: 128
session: session:
session_dir: ".sneakycode/sessions" session_dir: ".sneakycode/sessions"

View File

@@ -0,0 +1,314 @@
"""Tests for the file cache with LRU eviction and mtime invalidation."""
import os
import time
from pathlib import Path
import pytest
from app.models.config import AppConfig, load_config
from app.models.tool_call import ToolResultStatus
from app.tools.filesystem import ReadFileTool, ReadManyFilesTool
from app.utils.file_cache import CacheStats, FileCache, cached_read_file
from app.utils.file_helpers import BinaryFileError, FileSizeError, PathSecurityError
# ---------------------------------------------------------------------------
# FileCache unit tests
# ---------------------------------------------------------------------------
class TestFileCache:
def test_put_and_get_roundtrip(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "hello.txt"
f.write_text("hello world")
cache.put(f, "hello world")
assert cache.get(f) == "hello world"
def test_get_returns_none_for_missing_key(self, tmp_path: Path) -> None:
cache = FileCache()
assert cache.get(tmp_path / "nope.txt") is None
def test_mtime_change_causes_miss(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "data.txt"
f.write_text("v1")
cache.put(f, "v1")
# Mutate the file so mtime changes
time.sleep(0.05) # ensure mtime differs
f.write_text("v2")
assert cache.get(f) is None # stale → miss
assert cache.stats.invalidations == 1
def test_lru_eviction_at_capacity(self, tmp_path: Path) -> None:
cache = FileCache(max_entries=3)
files = []
for i in range(4):
f = tmp_path / f"f{i}.txt"
f.write_text(f"content-{i}")
files.append(f)
# Fill cache to capacity
for f in files[:3]:
cache.put(f, f.read_text())
assert len(cache) == 3
# Adding a 4th evicts the LRU (files[0])
cache.put(files[3], files[3].read_text())
assert len(cache) == 3
assert cache.get(files[0]) is None # evicted
assert cache.stats.evictions == 1
# files[1..3] still present
for f in files[1:]:
assert cache.get(f) is not None
def test_invalidate_removes_entry(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "rm.txt"
f.write_text("bye")
cache.put(f, "bye")
assert len(cache) == 1
cache.invalidate(f)
assert len(cache) == 0
assert cache.get(f) is None
assert cache.stats.invalidations == 1
def test_invalidate_noop_for_missing(self) -> None:
cache = FileCache()
cache.invalidate(Path("/nonexistent"))
assert cache.stats.invalidations == 0
def test_clear_empties_cache(self, tmp_path: Path) -> None:
cache = FileCache()
for i in range(5):
f = tmp_path / f"c{i}.txt"
f.write_text(str(i))
cache.put(f, str(i))
assert len(cache) == 5
cache.clear()
assert len(cache) == 0
def test_stats_accuracy(self, tmp_path: Path) -> None:
cache = FileCache(max_entries=2)
a = tmp_path / "a.txt"
b = tmp_path / "b.txt"
c = tmp_path / "c.txt"
a.write_text("a")
b.write_text("b")
c.write_text("c")
# Miss
cache.get(a)
assert cache.stats.misses == 1
assert cache.stats.hits == 0
# Put + hit
cache.put(a, "a")
cache.get(a)
assert cache.stats.hits == 1
# Fill + evict
cache.put(b, "b")
cache.put(c, "c") # evicts a
assert cache.stats.evictions == 1
def test_hit_rate(self) -> None:
stats = CacheStats(hits=3, misses=1)
assert stats.hit_rate == pytest.approx(0.75)
def test_hit_rate_zero_total(self) -> None:
stats = CacheStats()
assert stats.hit_rate == 0.0
def test_file_deleted_after_caching(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "gone.txt"
f.write_text("here")
cache.put(f, "here")
f.unlink()
assert cache.get(f) is None # stat fails → miss
def test_put_skips_when_stat_fails(self) -> None:
cache = FileCache()
cache.put(Path("/totally/nonexistent"), "data")
assert len(cache) == 0
def test_get_moves_to_end(self, tmp_path: Path) -> None:
"""Accessing an entry makes it most-recently-used, protecting from eviction."""
cache = FileCache(max_entries=3)
files = []
for i in range(3):
f = tmp_path / f"lru{i}.txt"
f.write_text(f"c{i}")
files.append(f)
cache.put(f, f"c{i}")
# Touch files[0] to make it MRU
cache.get(files[0])
# Add a new entry — files[1] (LRU) should be evicted, not files[0]
extra = tmp_path / "extra.txt"
extra.write_text("x")
cache.put(extra, "x")
assert cache.get(files[0]) is not None # protected by access
assert cache.get(files[1]) is None # evicted
# ---------------------------------------------------------------------------
# cached_read_file tests
# ---------------------------------------------------------------------------
class TestCachedReadFile:
def test_without_cache_matches_safe_read(self, tmp_path: Path) -> None:
f = tmp_path / "plain.txt"
f.write_text("hello")
content = cached_read_file(f, tmp_path, cache=None)
assert content == "hello"
def test_populates_on_miss_returns_on_hit(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "cached.txt"
f.write_text("data")
# First call: miss → read from disk → populate cache
content1 = cached_read_file(f, tmp_path, cache=cache)
assert content1 == "data"
assert cache.stats.misses == 1
assert cache.stats.hits == 0
# Second call: hit → from cache
content2 = cached_read_file(f, tmp_path, cache=cache)
assert content2 == "data"
assert cache.stats.hits == 1
def test_security_checks_run_on_cached_path(self, tmp_path: Path) -> None:
cache = FileCache()
with pytest.raises(PathSecurityError):
cached_read_file("/etc/passwd", tmp_path, cache=cache)
def test_binary_check_runs_on_cached_path(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "bin.dat"
f.write_bytes(b"\x00binary\x00")
with pytest.raises(BinaryFileError):
cached_read_file(f, tmp_path, cache=cache)
def test_size_check_runs_on_cached_path(self, tmp_path: Path) -> None:
cache = FileCache()
f = tmp_path / "big.txt"
f.write_text("x" * 200)
# First read populates cache
cached_read_file(f, tmp_path, max_size_bytes=1000, cache=cache)
# Now make file too big on disk — security check should catch it
# even though content is cached
f.write_text("x" * 2000)
with pytest.raises(FileSizeError):
cached_read_file(f, tmp_path, max_size_bytes=1000, cache=cache)
def test_file_not_found(self, tmp_path: Path) -> None:
cache = FileCache()
with pytest.raises(FileNotFoundError):
cached_read_file(tmp_path / "nope.txt", tmp_path, cache=cache)
# ---------------------------------------------------------------------------
# Tool-level cache-hit dedup tests
# ---------------------------------------------------------------------------
@pytest.fixture
def config() -> AppConfig:
return load_config()
@pytest.fixture
def tmp_workspace(tmp_path: Path, config: AppConfig) -> tuple[Path, AppConfig]:
config.agent.workspace_root = tmp_path
return tmp_path, config
class TestReadFileToolCacheHit:
def test_first_read_returns_full_content(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
ws, cfg = tmp_workspace
cache = FileCache()
(ws / "hello.txt").write_text("hello world")
tool = ReadFileTool(ws, cfg, file_cache=cache)
result = tool.run("tc-1", {"file_path": "hello.txt"})
assert result.status == ToolResultStatus.SUCCESS
assert result.output == "hello world"
def test_second_read_returns_cached_message(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
ws, cfg = tmp_workspace
cache = FileCache()
(ws / "hello.txt").write_text("hello world")
tool = ReadFileTool(ws, cfg, file_cache=cache)
tool.run("tc-1", {"file_path": "hello.txt"})
result2 = tool.run("tc-2", {"file_path": "hello.txt"})
assert result2.status == ToolResultStatus.SUCCESS
assert "[Cached]" in result2.output
assert "hello.txt" in result2.output
assert "hello world" not in result2.output
def test_changed_file_returns_full_content_again(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
ws, cfg = tmp_workspace
cache = FileCache()
f = ws / "data.txt"
f.write_text("v1")
tool = ReadFileTool(ws, cfg, file_cache=cache)
tool.run("tc-1", {"file_path": "data.txt"})
# Mutate file so mtime changes
time.sleep(0.05)
f.write_text("v2")
result2 = tool.run("tc-2", {"file_path": "data.txt"})
assert result2.status == ToolResultStatus.SUCCESS
assert result2.output == "v2"
assert "[Cached]" not in result2.output
def test_no_cache_always_returns_content(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
ws, cfg = tmp_workspace
(ws / "hello.txt").write_text("hello")
tool = ReadFileTool(ws, cfg, file_cache=None)
r1 = tool.run("tc-1", {"file_path": "hello.txt"})
r2 = tool.run("tc-2", {"file_path": "hello.txt"})
assert r1.output == "hello"
assert r2.output == "hello"
class TestReadManyFilesToolCacheHit:
def test_cached_files_get_short_message(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
ws, cfg = tmp_workspace
cache = FileCache()
(ws / "a.txt").write_text("alpha")
(ws / "b.txt").write_text("bravo")
tool = ReadManyFilesTool(ws, cfg, file_cache=cache)
# First read — full content
r1 = tool.run("tc-1", {"file_paths": ["a.txt", "b.txt"]})
assert "alpha" in r1.output
assert "bravo" in r1.output
# Second read — cached messages
r2 = tool.run("tc-2", {"file_paths": ["a.txt", "b.txt"]})
assert "[Cached]" in r2.output
assert "alpha" not in r2.output
assert "bravo" not in r2.output