diff --git a/agent/subdirectory_hints.py b/agent/subdirectory_hints.py index ca96c664cb..4e9f7f5ed3 100644 --- a/agent/subdirectory_hints.py +++ b/agent/subdirectory_hints.py @@ -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: diff --git a/tests/agent/test_subdirectory_hints.py b/tests/agent/test_subdirectory_hints.py index 16c2b5b725..01de13d4ea 100644 --- a/tests/agent/test_subdirectory_hints.py +++ b/tests/agent/test_subdirectory_hints.py @@ -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