fix(context): dedupe subdirectory hints by content digest and skip backup/vendor dirs
SubdirectoryHintTracker re-injected identical context files whenever the same AGENTS.md was reachable through more than one path. Symlinked shared workspaces, hardlinks, and timestamped backup copies all alias a single file, so a normal session could ship the same 8KB of instructions two or three times. Nothing deduped it and nothing excluded directories that only ever hold copies. Two changes: * Track a sha256 of every injected hint body. Repeat content is skipped, and the working directory's own context file is seeded at construction so the copy prompt_builder already loaded at startup is never sent again. * Skip directories that hold copies rather than authoritative context (backups, node_modules, venv, site-packages, .git, .Trash, vendor, caches). Screening is relative to working_dir, so a project that legitimately lives under vendor/ keeps discovering its own subdirectory hints. Measured on a real session that touched a symlinked shared workspace: 3 injections / ~24,000 chars before, 1 injection / 8,112 chars after. 14 new tests cover symlink aliasing, byte-identical copies, working-dir seeding, distinct content still being injected, each excluded directory name, excluded ancestors, and the working-dir-inside-excluded-name case.
This commit is contained in:
@@ -13,6 +13,7 @@ the conversation without modifying the system prompt (preserving prompt caching)
|
||||
Inspired by Block/goose's SubdirectoryHintTracker.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import logging
|
||||
import os
|
||||
import shlex
|
||||
@@ -45,6 +46,18 @@ _COMMAND_TOOLS = {"terminal"}
|
||||
# Prevents scanning all the way to / for deeply nested paths.
|
||||
_MAX_ANCESTOR_WALK = 5
|
||||
|
||||
# Directory names that never contain authoritative project context.
|
||||
# Backups, vendored deps, VCS internals, and caches routinely hold *copies* of
|
||||
# AGENTS.md; loading those duplicates real context and inflates the prompt.
|
||||
_EXCLUDED_DIR_NAMES = frozenset({
|
||||
"node_modules", "venv", ".venv", "__pycache__",
|
||||
".git", ".hg", ".svn",
|
||||
".Trash", ".cache", ".tox", ".mypy_cache", ".pytest_cache",
|
||||
"site-packages", "dist-packages",
|
||||
"backups", "backup", ".backups",
|
||||
"vendor", "third_party",
|
||||
})
|
||||
|
||||
|
||||
def _is_ancestor_or_same(a: Path, b: Path) -> bool:
|
||||
"""Check if *a* is the same as or an ancestor of *b* (parent directory check)."""
|
||||
@@ -54,6 +67,7 @@ def _is_ancestor_or_same(a: Path, b: Path) -> bool:
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
class SubdirectoryHintTracker:
|
||||
"""Track which directories the agent visits and load hints on first access.
|
||||
|
||||
@@ -70,8 +84,34 @@ class SubdirectoryHintTracker:
|
||||
def __init__(self, working_dir: Optional[str] = None):
|
||||
self.working_dir = Path(working_dir or os.getcwd()).resolve()
|
||||
self._loaded_dirs: Set[Path] = set()
|
||||
# Content digests already injected — prevents re-sending the same file
|
||||
# reachable through symlinks, hardlinks, or duplicated copies.
|
||||
self._loaded_digests: Set[str] = set()
|
||||
# Pre-mark the working dir as loaded (startup context handles it)
|
||||
self._loaded_dirs.add(self.working_dir)
|
||||
self._seed_working_dir_digest()
|
||||
|
||||
def _seed_working_dir_digest(self) -> None:
|
||||
"""Record the CWD context file's digest so it is never re-injected.
|
||||
|
||||
``prompt_builder`` already loads the working directory's context file at
|
||||
startup. Seeding its digest here means the same content reached through
|
||||
a different path (a symlink farm, a shared workspace) is recognised as a
|
||||
duplicate instead of being sent a second time.
|
||||
"""
|
||||
for filename in _HINT_FILENAMES:
|
||||
candidate = self.working_dir / filename
|
||||
try:
|
||||
if not candidate.is_file():
|
||||
continue
|
||||
content = candidate.read_text(encoding="utf-8").strip()
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
if content:
|
||||
self._loaded_digests.add(
|
||||
hashlib.sha256(content.encode("utf-8")).hexdigest()
|
||||
)
|
||||
break # first match wins, mirroring startup loading
|
||||
|
||||
def check_tool_call(
|
||||
self,
|
||||
@@ -193,8 +233,25 @@ class SubdirectoryHintTracker:
|
||||
# check as a best-effort safeguard.
|
||||
if not _is_ancestor_or_same(self.working_dir, path):
|
||||
return False
|
||||
if self._is_excluded(path):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _is_excluded(self, path: Path) -> bool:
|
||||
"""True when the path sits inside a directory that holds copies, not context.
|
||||
|
||||
Directories the user is deliberately working inside are never excluded —
|
||||
if ``working_dir`` is itself under ``vendor/``, that segment is legitimate
|
||||
and only segments *below* the working dir are screened.
|
||||
"""
|
||||
try:
|
||||
rel_parts = path.relative_to(self.working_dir).parts
|
||||
except ValueError:
|
||||
# Paths outside the working dir are already rejected by
|
||||
# _is_valid_subdir before this runs; treat as excluded defensively.
|
||||
return True
|
||||
return any(part in _EXCLUDED_DIR_NAMES for part in rel_parts)
|
||||
|
||||
def _load_hints_for_directory(self, directory: Path) -> Optional[str]:
|
||||
"""Load hint files from a directory. Returns formatted text or None.
|
||||
|
||||
@@ -230,6 +287,19 @@ class SubdirectoryHintTracker:
|
||||
content = hint_path.read_text(encoding="utf-8").strip()
|
||||
if not content:
|
||||
continue
|
||||
# Skip content we've already injected. The same AGENTS.md is
|
||||
# routinely reachable through several paths (symlinked shared
|
||||
# workspaces, hardlinks, copied backups); re-sending it burns
|
||||
# context for zero new information.
|
||||
digest = hashlib.sha256(content.encode("utf-8")).hexdigest()
|
||||
if digest in self._loaded_digests:
|
||||
logger.debug(
|
||||
"Skipping duplicate hint content at %s (digest %s)",
|
||||
hint_path,
|
||||
digest[:12],
|
||||
)
|
||||
break
|
||||
self._loaded_digests.add(digest)
|
||||
# Same security scan as startup context loading
|
||||
content = _scan_context_content(content, filename)
|
||||
if len(content) > _MAX_HINT_CHARS:
|
||||
|
||||
@@ -176,3 +176,111 @@ class TestOutsideWorkspaceRejection:
|
||||
outside.mkdir(exist_ok=True)
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(project))
|
||||
assert tracker._is_valid_subdir(outside) is False
|
||||
|
||||
|
||||
class TestContentDeduplication:
|
||||
"""The same context content must never be injected twice (ref: symlinked
|
||||
shared workspaces, hardlinks, and copied backups all alias one file)."""
|
||||
|
||||
def test_symlinked_duplicate_not_reinjected(self, tmp_path):
|
||||
"""Two directories whose AGENTS.md is the same file yield one injection."""
|
||||
real = tmp_path / "real"
|
||||
real.mkdir()
|
||||
(real / "AGENTS.md").write_text("Shared workspace instructions")
|
||||
|
||||
mirror = tmp_path / "mirror"
|
||||
mirror.mkdir()
|
||||
(mirror / "AGENTS.md").symlink_to(real / "AGENTS.md")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
first = tracker.check_tool_call("read_file", {"path": str(real / "x.py")})
|
||||
second = tracker.check_tool_call("read_file", {"path": str(mirror / "y.py")})
|
||||
|
||||
assert first is not None
|
||||
assert "Shared workspace instructions" in first
|
||||
assert second is None
|
||||
|
||||
def test_identical_copy_not_reinjected(self, tmp_path):
|
||||
"""Byte-identical copies in unrelated directories dedupe by digest."""
|
||||
a = tmp_path / "a"
|
||||
b = tmp_path / "b"
|
||||
a.mkdir()
|
||||
b.mkdir()
|
||||
(a / "AGENTS.md").write_text("Same content")
|
||||
(b / "AGENTS.md").write_text("Same content")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
assert tracker.check_tool_call("read_file", {"path": str(a / "f.py")}) is not None
|
||||
assert tracker.check_tool_call("read_file", {"path": str(b / "f.py")}) is None
|
||||
|
||||
def test_differing_content_still_injected(self, tmp_path):
|
||||
"""Dedupe must not suppress genuinely different context."""
|
||||
a = tmp_path / "a"
|
||||
b = tmp_path / "b"
|
||||
a.mkdir()
|
||||
b.mkdir()
|
||||
(a / "AGENTS.md").write_text("Alpha rules")
|
||||
(b / "AGENTS.md").write_text("Beta rules")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
first = tracker.check_tool_call("read_file", {"path": str(a / "f.py")})
|
||||
second = tracker.check_tool_call("read_file", {"path": str(b / "f.py")})
|
||||
|
||||
assert first is not None and "Alpha rules" in first
|
||||
assert second is not None and "Beta rules" in second
|
||||
|
||||
def test_working_dir_content_seeded(self, tmp_path):
|
||||
"""A copy of the CWD's own context file is not re-injected."""
|
||||
(tmp_path / "AGENTS.md").write_text("Root instructions")
|
||||
elsewhere = tmp_path / "elsewhere"
|
||||
elsewhere.mkdir()
|
||||
(elsewhere / "AGENTS.md").write_text("Root instructions")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
assert tracker.check_tool_call("read_file", {"path": str(elsewhere / "f.py")}) is None
|
||||
|
||||
|
||||
class TestExcludedDirectories:
|
||||
"""Backups, vendored deps, and caches hold copies — never context."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"excluded",
|
||||
["backups", "node_modules", ".git", "venv", "site-packages", ".Trash", "vendor"],
|
||||
)
|
||||
def test_excluded_directory_skipped(self, tmp_path, excluded):
|
||||
target = tmp_path / excluded / "snapshot"
|
||||
target.mkdir(parents=True)
|
||||
(target / "AGENTS.md").write_text("Stale archived instructions")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
assert tracker.check_tool_call("read_file", {"path": str(target / "f.py")}) is None
|
||||
|
||||
def test_excluded_ancestor_blocks_descendant(self, tmp_path):
|
||||
"""A hint nested under an excluded ancestor is still skipped."""
|
||||
deep = tmp_path / "backups" / "2026" / "proj"
|
||||
deep.mkdir(parents=True)
|
||||
(deep / "AGENTS.md").write_text("Archived")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
assert tracker.check_tool_call("read_file", {"path": str(deep / "f.py")}) is None
|
||||
|
||||
def test_working_dir_inside_excluded_name_still_works(self, tmp_path):
|
||||
"""If the user works inside e.g. vendor/, its own subdirs stay eligible."""
|
||||
root = tmp_path / "vendor" / "myproject"
|
||||
root.mkdir(parents=True)
|
||||
sub = root / "pkg"
|
||||
sub.mkdir()
|
||||
(sub / "AGENTS.md").write_text("Package rules")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(root))
|
||||
result = tracker.check_tool_call("read_file", {"path": str(sub / "f.py")})
|
||||
assert result is not None and "Package rules" in result
|
||||
|
||||
def test_normal_directory_unaffected(self, tmp_path):
|
||||
normal = tmp_path / "backend"
|
||||
normal.mkdir()
|
||||
(normal / "AGENTS.md").write_text("Backend rules")
|
||||
|
||||
tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
|
||||
result = tracker.check_tool_call("read_file", {"path": str(normal / "f.py")})
|
||||
assert result is not None and "Backend rules" in result
|
||||
|
||||
Reference in New Issue
Block a user