Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion agent/prompt_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -1299,11 +1299,24 @@ def _status_line(feature) -> str:
# =========================================================================

def _truncate_content(content: str, filename: str, max_chars: int = CONTEXT_FILE_MAX_CHARS) -> str:
"""Head/tail truncation with a marker in the middle."""
"""Head/tail truncation with a marker in the middle.

Emits a warning when content is actually dropped. The in-band marker only
reaches the model — an author who lets AGENTS.md grow past the cap gets no
signal that the middle of their file stopped being loaded, so the omission
has to surface in the log too.
"""
if len(content) <= max_chars:
return content
head_chars = int(max_chars * CONTEXT_TRUNCATE_HEAD_RATIO)
tail_chars = int(max_chars * CONTEXT_TRUNCATE_TAIL_RATIO)
dropped = len(content) - head_chars - tail_chars
logger.warning(
"Context file %s is %d chars (limit %d) — dropping %d chars (%.0f%%) "
"from the middle. Split the overflow into separate docs and reference "
"them, or the omitted sections will never reach the model.",
filename, len(content), max_chars, dropped, 100 * dropped / len(content),
)
head = content[:head_chars]
tail = content[-tail_chars:]
marker = f"\n\n[...truncated {filename}: kept {head_chars}+{tail_chars} of {len(content)} chars. Use file tools to read the full file.]\n\n"
Expand Down
75 changes: 75 additions & 0 deletions agent/subdirectory_hints.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
Inspired by Block/goose's SubdirectoryHintTracker.
"""

import hashlib
import logging
import os
import shlex
Expand Down Expand Up @@ -45,6 +46,18 @@
# 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",
})

class SubdirectoryHintTracker:
"""Track which directories the agent visits and load hints on first access.

Expand All @@ -61,8 +74,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,
Expand Down Expand Up @@ -166,8 +205,24 @@ def _is_valid_subdir(self, path: Path) -> bool:
return False
if path in self._loaded_dirs:
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:
# Outside the working dir — screen every segment.
rel_parts = path.parts
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."""
self._loaded_dirs.add(directory)
Expand All @@ -184,9 +239,29 @@ def _load_hints_for_directory(self, directory: Path) -> Optional[str]:
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:
logger.warning(
"Subdirectory hint %s is %d chars (limit %d) — dropping "
"%d chars from the end. Split the overflow into separate "
"docs and reference them from the hint file.",
hint_path, len(content), _MAX_HINT_CHARS,
len(content) - _MAX_HINT_CHARS,
)
content = (
content[:_MAX_HINT_CHARS]
+ f"\n\n[...truncated {filename}: {len(content):,} chars total]"
Expand Down
30 changes: 30 additions & 0 deletions tests/agent/test_prompt_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -1195,3 +1195,33 @@ def test_guidance_is_string(self):





class TestContextFileTruncationIsAudible:
"""A context file that outgrows the cap loses its middle. The in-band marker
only reaches the model, so the author needs a log line too (ref: AGENTS.md
silently shedding 66% of its content)."""

def test_oversized_context_file_warns(self, caplog):
import logging
from agent.prompt_builder import _truncate_content, CONTEXT_FILE_MAX_CHARS

oversized = "y" * (CONTEXT_FILE_MAX_CHARS * 2)
with caplog.at_level(logging.WARNING, logger="agent.prompt_builder"):
out = _truncate_content(oversized, "AGENTS.md")

assert len(out) < len(oversized)
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
assert warnings, caplog.text
assert "AGENTS.md" in warnings[0].getMessage()

def test_within_limit_returns_unmodified_and_quiet(self, caplog):
import logging
from agent.prompt_builder import _truncate_content

content = "z" * 100
with caplog.at_level(logging.WARNING, logger="agent.prompt_builder"):
out = _truncate_content(content, "AGENTS.md")

assert out == content
assert not [r for r in caplog.records if r.levelno >= logging.WARNING]
145 changes: 145 additions & 0 deletions tests/agent/test_subdirectory_hints.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,3 +232,148 @@ def patched_is_dir(self):
)
# Result may be None (backend skipped) — the key point is no crash
assert result is None or isinstance(result, str)


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


class TestTruncationIsAudible:
"""Silent truncation is how AGENTS.md stopped being fully loaded for months.
Dropping content must always leave a trace in the log."""

def test_oversized_hint_warns(self, tmp_path, caplog):
import logging
from agent.subdirectory_hints import _MAX_HINT_CHARS

sub = tmp_path / "big"
sub.mkdir()
(sub / "AGENTS.md").write_text("x" * (_MAX_HINT_CHARS + 5_000))

tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
with caplog.at_level(logging.WARNING, logger="agent.subdirectory_hints"):
result = tracker.check_tool_call("read_file", {"path": str(sub / "f.py")})

assert result is not None
assert any(
"dropping" in r.message.lower() and r.levelno == logging.WARNING
for r in caplog.records
), caplog.text

def test_within_limit_stays_quiet(self, tmp_path, caplog):
import logging

sub = tmp_path / "small"
sub.mkdir()
(sub / "AGENTS.md").write_text("short and fine")

tracker = SubdirectoryHintTracker(working_dir=str(tmp_path))
with caplog.at_level(logging.WARNING, logger="agent.subdirectory_hints"):
result = tracker.check_tool_call("read_file", {"path": str(sub / "f.py")})

assert result is not None
assert not [r for r in caplog.records if r.levelno >= logging.WARNING]