mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-04-25 00:51:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

feat(file_tools): harden read_file with size guard, dedup, and device blocking (#4315)

Browse source 
* feat(file_tools): harden read_file with size guard, dedup, and device blocking

Three improvements to read_file_tool to reduce wasted context tokens and
prevent process hangs:

1. Character-count guard: reads that produce more than 100K characters
   (≈25-35K tokens across tokenisers) are rejected with an error that
   tells the model to use offset+limit for a smaller range.  The
   effective cap is min(file_size, 100K) so small files that happen to
   have long lines aren't over-penalised.  Large truncated files also
   get a hint nudging toward targeted reads.

2. File-read deduplication: when the same (path, offset, limit) is read
   a second time and the file hasn't been modified (mtime unchanged),
   return a lightweight stub instead of re-sending the full content.
   Writes and patches naturally change mtime, so post-edit reads always
   return fresh content.  The dedup cache is cleared on context
   compression — after compression the original read content is
   summarised away, so the model needs the full content again.

3. Device path blocking: paths like /dev/zero, /dev/random, /dev/stdin
   etc. are rejected before any I/O to prevent process hangs from
   infinite-output or blocking-input devices.

Tests: 17 new tests covering all three features plus the dedup-reset-
on-compression integration.  All 52 file-read tests pass (35 existing +
17 new).  Full tool suite (2124 tests) passes with 0 failures.

* feat: make file_read_max_chars configurable, add docs

Add file_read_max_chars to DEFAULT_CONFIG (default 100K).  read_file_tool
reads this on first call and caches for the process lifetime.  Users on
large-context models can raise it; users on small local models can lower it.

Also adds a 'File Read Safety' section to the configuration docs
explaining the char limit, dedup behavior, and example values.
This commit is contained in:
Teknium 2026-03-31 12:53:19 -07:00 • committed by GitHub
parent d3f1987a05
commit e3f8347be3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
5 changed files with 605 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

5
hermes_cli/config.py
View file

@ -256,6 +256,11 @@ DEFAULT_CONFIG = {
"enabled": True,
"max_snapshots": 50, # Max checkpoints to keep per directory
},
# Maximum characters returned by a single read_file call. Reads that
# exceed this are rejected with guidance to use offset+limit.
# 100K chars ≈ 25–35K tokens across typical tokenisers.
"file_read_max_chars": 100_000,
"compression": {
"enabled": True,

9
run_agent.py
View file

@ -5361,6 +5361,15 @@ class AIAgent:
if _post_progress < 0.85:
self._context_pressure_warned = False
# Clear the file-read dedup cache. After compression the original
# read content is summarised away — if the model re-reads the same
# file it needs the full content, not a "file unchanged" stub.
try:
from tools.file_tools import reset_file_dedup
reset_file_dedup(task_id)
except Exception:
pass
return compressed, new_system_prompt
def _execute_tool_calls(self, assistant_message, messages: list, effective_task_id: str, api_call_count: int = 0) -> None:

378
tests/tools/test_file_read_guards.py Normal file
View file

@ -0,0 +1,378 @@
#!/usr/bin/env python3
"""
Tests for read_file_tool safety guards: device-path blocking,
character-count limits, file deduplication, and dedup reset on
context compression.
Run with: python -m pytest tests/tools/test_file_read_guards.py -v
"""
import json
import os
import tempfile
import time
import unittest
from unittest.mock import patch, MagicMock
from tools.file_tools import (
read_file_tool,
clear_read_tracker,
reset_file_dedup,
_is_blocked_device,
_get_max_read_chars,
_DEFAULT_MAX_READ_CHARS,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
class _FakeReadResult:
"""Minimal stand-in for FileOperations.read_file return value."""
def __init__(self, content="line1\nline2\n", total_lines=2, file_size=100):
self.content = content
self._total_lines = total_lines
self._file_size = file_size
def to_dict(self):
return {
"content": self.content,
"total_lines": self._total_lines,
"file_size": self._file_size,
}
def _make_fake_ops(content="hello\n", total_lines=1, file_size=6):
fake = MagicMock()
fake.read_file = lambda path, offset=1, limit=500: _FakeReadResult(
content=content, total_lines=total_lines, file_size=file_size,
)
return fake
# ---------------------------------------------------------------------------
# Device path blocking
# ---------------------------------------------------------------------------
class TestDevicePathBlocking(unittest.TestCase):
"""Paths like /dev/zero should be rejected before any I/O."""
def test_blocked_device_detection(self):
for dev in ("/dev/zero", "/dev/random", "/dev/urandom", "/dev/stdin",
"/dev/tty", "/dev/console", "/dev/stdout", "/dev/stderr",
"/dev/fd/0", "/dev/fd/1", "/dev/fd/2"):
self.assertTrue(_is_blocked_device(dev), f"{dev} should be blocked")
def test_safe_device_not_blocked(self):
self.assertFalse(_is_blocked_device("/dev/null"))
self.assertFalse(_is_blocked_device("/dev/sda1"))
def test_proc_fd_blocked(self):
self.assertTrue(_is_blocked_device("/proc/self/fd/0"))
self.assertTrue(_is_blocked_device("/proc/12345/fd/2"))
def test_proc_fd_other_not_blocked(self):
self.assertFalse(_is_blocked_device("/proc/self/fd/3"))
self.assertFalse(_is_blocked_device("/proc/self/maps"))
def test_normal_files_not_blocked(self):
self.assertFalse(_is_blocked_device("/tmp/test.py"))
self.assertFalse(_is_blocked_device("/home/user/.bashrc"))
def test_read_file_tool_rejects_device(self):
"""read_file_tool returns an error without any file I/O."""
result = json.loads(read_file_tool("/dev/zero", task_id="dev_test"))
self.assertIn("error", result)
self.assertIn("device file", result["error"])
# ---------------------------------------------------------------------------
# Character-count limits
# ---------------------------------------------------------------------------
class TestCharacterCountGuard(unittest.TestCase):
"""Large reads should be rejected with guidance to use offset/limit."""
def setUp(self):
clear_read_tracker()
def tearDown(self):
clear_read_tracker()
@patch("tools.file_tools._get_file_ops")
@patch("tools.file_tools._get_max_read_chars", return_value=_DEFAULT_MAX_READ_CHARS)
def test_oversized_read_rejected(self, _mock_limit, mock_ops):
"""A read that returns >max chars is rejected."""
big_content = "x" * (_DEFAULT_MAX_READ_CHARS + 1)
mock_ops.return_value = _make_fake_ops(
content=big_content,
total_lines=5000,
file_size=len(big_content) + 100, # bigger than content
)
result = json.loads(read_file_tool("/tmp/huge.txt", task_id="big"))
self.assertIn("error", result)
self.assertIn("safety limit", result["error"])
self.assertIn("offset and limit", result["error"])
self.assertIn("total_lines", result)
@patch("tools.file_tools._get_file_ops")
def test_small_read_not_rejected(self, mock_ops):
"""Normal-sized reads pass through fine."""
mock_ops.return_value = _make_fake_ops(content="short\n", file_size=6)
result = json.loads(read_file_tool("/tmp/small.txt", task_id="small"))
self.assertNotIn("error", result)
self.assertIn("content", result)
@patch("tools.file_tools._get_file_ops")
@patch("tools.file_tools._get_max_read_chars", return_value=_DEFAULT_MAX_READ_CHARS)
def test_content_under_limit_passes(self, _mock_limit, mock_ops):
"""Content just under the limit should pass through fine."""
mock_ops.return_value = _make_fake_ops(
content="y" * (_DEFAULT_MAX_READ_CHARS - 1),
file_size=_DEFAULT_MAX_READ_CHARS - 1,
)
result = json.loads(read_file_tool("/tmp/justunder.txt", task_id="under"))
self.assertNotIn("error", result)
self.assertIn("content", result)
# ---------------------------------------------------------------------------
# File deduplication
# ---------------------------------------------------------------------------
class TestFileDedup(unittest.TestCase):
"""Re-reading an unchanged file should return a lightweight stub."""
def setUp(self):
clear_read_tracker()
self._tmpdir = tempfile.mkdtemp()
self._tmpfile = os.path.join(self._tmpdir, "dedup_test.txt")
with open(self._tmpfile, "w") as f:
f.write("line one\nline two\n")
def tearDown(self):
clear_read_tracker()
try:
os.unlink(self._tmpfile)
os.rmdir(self._tmpdir)
except OSError:
pass
@patch("tools.file_tools._get_file_ops")
def test_second_read_returns_dedup_stub(self, mock_ops):
"""Second read of same file+range returns dedup stub."""
mock_ops.return_value = _make_fake_ops(
content="line one\nline two\n", file_size=20,
)
# First read — full content
r1 = json.loads(read_file_tool(self._tmpfile, task_id="dup"))
self.assertNotIn("dedup", r1)
# Second read — should get dedup stub
r2 = json.loads(read_file_tool(self._tmpfile, task_id="dup"))
self.assertTrue(r2.get("dedup"), "Second read should return dedup stub")
self.assertIn("unchanged", r2.get("content", ""))
@patch("tools.file_tools._get_file_ops")
def test_modified_file_not_deduped(self, mock_ops):
"""After the file is modified, dedup returns full content."""
mock_ops.return_value = _make_fake_ops(
content="line one\nline two\n", file_size=20,
)
read_file_tool(self._tmpfile, task_id="mod")
# Modify the file — ensure mtime changes
time.sleep(0.05)
with open(self._tmpfile, "w") as f:
f.write("changed content\n")
r2 = json.loads(read_file_tool(self._tmpfile, task_id="mod"))
self.assertNotEqual(r2.get("dedup"), True, "Modified file should not dedup")
@patch("tools.file_tools._get_file_ops")
def test_different_range_not_deduped(self, mock_ops):
"""Same file but different offset/limit should not dedup."""
mock_ops.return_value = _make_fake_ops(
content="line one\nline two\n", file_size=20,
)
read_file_tool(self._tmpfile, offset=1, limit=500, task_id="rng")
r2 = json.loads(read_file_tool(
self._tmpfile, offset=10, limit=500, task_id="rng",
))
self.assertNotEqual(r2.get("dedup"), True)
@patch("tools.file_tools._get_file_ops")
def test_different_task_not_deduped(self, mock_ops):
"""Different task_ids have separate dedup caches."""
mock_ops.return_value = _make_fake_ops(
content="line one\nline two\n", file_size=20,
)
read_file_tool(self._tmpfile, task_id="task_a")
r2 = json.loads(read_file_tool(self._tmpfile, task_id="task_b"))
self.assertNotEqual(r2.get("dedup"), True)
# ---------------------------------------------------------------------------
# Dedup reset on compression
# ---------------------------------------------------------------------------
class TestDedupResetOnCompression(unittest.TestCase):
"""reset_file_dedup should clear the dedup cache so post-compression
reads return full content."""
def setUp(self):
clear_read_tracker()
self._tmpdir = tempfile.mkdtemp()
self._tmpfile = os.path.join(self._tmpdir, "compress_test.txt")
with open(self._tmpfile, "w") as f:
f.write("original content\n")
def tearDown(self):
clear_read_tracker()
try:
os.unlink(self._tmpfile)
os.rmdir(self._tmpdir)
except OSError:
pass
@patch("tools.file_tools._get_file_ops")
def test_reset_clears_dedup(self, mock_ops):
"""After reset_file_dedup, the same read returns full content."""
mock_ops.return_value = _make_fake_ops(
content="original content\n", file_size=18,
)
# First read — populates dedup cache
read_file_tool(self._tmpfile, task_id="comp")
# Verify dedup works before reset
r_dedup = json.loads(read_file_tool(self._tmpfile, task_id="comp"))
self.assertTrue(r_dedup.get("dedup"), "Should dedup before reset")
# Simulate compression
reset_file_dedup("comp")
# Read again — should get full content
r_post = json.loads(read_file_tool(self._tmpfile, task_id="comp"))
self.assertNotEqual(r_post.get("dedup"), True,
"Post-compression read should return full content")
@patch("tools.file_tools._get_file_ops")
def test_reset_all_tasks(self, mock_ops):
"""reset_file_dedup(None) clears all tasks."""
mock_ops.return_value = _make_fake_ops(
content="original content\n", file_size=18,
)
read_file_tool(self._tmpfile, task_id="t1")
read_file_tool(self._tmpfile, task_id="t2")
reset_file_dedup() # no task_id — clear all
r1 = json.loads(read_file_tool(self._tmpfile, task_id="t1"))
r2 = json.loads(read_file_tool(self._tmpfile, task_id="t2"))
self.assertNotEqual(r1.get("dedup"), True)
self.assertNotEqual(r2.get("dedup"), True)
@patch("tools.file_tools._get_file_ops")
def test_reset_preserves_loop_detection(self, mock_ops):
"""reset_file_dedup does NOT affect the consecutive-read counter."""
mock_ops.return_value = _make_fake_ops(
content="original content\n", file_size=18,
)
# Build up consecutive count (read 1 and 2)
read_file_tool(self._tmpfile, task_id="loop")
# 2nd read is deduped — doesn't increment consecutive counter
read_file_tool(self._tmpfile, task_id="loop")
reset_file_dedup("loop")
# 3rd read — counter should still be at 2 from before reset
# (dedup was hit for read 2, but consecutive counter was 1 for that)
# After reset, this read goes through full path, incrementing to 2
r3 = json.loads(read_file_tool(self._tmpfile, task_id="loop"))
# Should NOT be blocked or warned — counter restarted since dedup
# intercepted reads before they reached the counter
self.assertNotIn("error", r3)
# ---------------------------------------------------------------------------
# Large-file hint
# ---------------------------------------------------------------------------
class TestLargeFileHint(unittest.TestCase):
"""Large truncated files should include a hint about targeted reads."""
def setUp(self):
clear_read_tracker()
def tearDown(self):
clear_read_tracker()
@patch("tools.file_tools._get_file_ops")
def test_large_truncated_file_gets_hint(self, mock_ops):
content = "line\n" * 400 # 2000 chars, small enough to pass char guard
fake = _make_fake_ops(content=content, total_lines=10000, file_size=600_000)
# Make to_dict return truncated=True
orig_read = fake.read_file
def patched_read(path, offset=1, limit=500):
r = orig_read(path, offset, limit)
orig_to_dict = r.to_dict
def new_to_dict():
d = orig_to_dict()
d["truncated"] = True
return d
r.to_dict = new_to_dict
return r
fake.read_file = patched_read
mock_ops.return_value = fake
result = json.loads(read_file_tool("/tmp/bigfile.log", task_id="hint"))
self.assertIn("_hint", result)
self.assertIn("section you need", result["_hint"])
# ---------------------------------------------------------------------------
# Config override
# ---------------------------------------------------------------------------
class TestConfigOverride(unittest.TestCase):
"""file_read_max_chars in config.yaml should control the char guard."""
def setUp(self):
clear_read_tracker()
# Reset the cached value so each test gets a fresh lookup
import tools.file_tools as _ft
_ft._max_read_chars_cached = None
def tearDown(self):
clear_read_tracker()
import tools.file_tools as _ft
_ft._max_read_chars_cached = None
@patch("tools.file_tools._get_file_ops")
@patch("hermes_cli.config.load_config", return_value={"file_read_max_chars": 50})
def test_custom_config_lowers_limit(self, _mock_cfg, mock_ops):
"""A config value of 50 should reject reads over 50 chars."""
mock_ops.return_value = _make_fake_ops(content="x" * 60, file_size=60)
result = json.loads(read_file_tool("/tmp/cfgtest.txt", task_id="cfg1"))
self.assertIn("error", result)
self.assertIn("safety limit", result["error"])
self.assertIn("50", result["error"]) # should show the configured limit
@patch("tools.file_tools._get_file_ops")
@patch("hermes_cli.config.load_config", return_value={"file_read_max_chars": 500_000})
def test_custom_config_raises_limit(self, _mock_cfg, mock_ops):
"""A config value of 500K should allow reads up to 500K chars."""
# 200K chars would be rejected at the default 100K but passes at 500K
mock_ops.return_value = _make_fake_ops(
content="y" * 200_000, file_size=200_000,
)
result = json.loads(read_file_tool("/tmp/cfgtest2.txt", task_id="cfg2"))
self.assertNotIn("error", result)
self.assertIn("content", result)
if __name__ == "__main__":
unittest.main()

203
tools/file_tools.py
View file

@ -15,6 +15,80 @@ logger = logging.getLogger(__name__)
_EXPECTED_WRITE_ERRNOS = {errno.EACCES, errno.EPERM, errno.EROFS}
# ---------------------------------------------------------------------------
# Read-size guard: cap the character count returned to the model.
# We're model-agnostic so we can't count tokens; characters are a safe proxy.
# 100K chars ≈ 25–35K tokens across typical tokenisers. Files larger than
# this in a single read are a context-window hazard — the model should use
# offset+limit to read the relevant section.
#
# Configurable via config.yaml: file_read_max_chars: 200000
# ---------------------------------------------------------------------------
_DEFAULT_MAX_READ_CHARS = 100_000
_max_read_chars_cached: int | None = None
def _get_max_read_chars() -> int:
"""Return the configured max characters per file read.
Reads ``file_read_max_chars`` from config.yaml on first call, caches
the result for the lifetime of the process. Falls back to the
built-in default if the config is missing or invalid.
"""
global _max_read_chars_cached
if _max_read_chars_cached is not None:
return _max_read_chars_cached
try:
from hermes_cli.config import load_config
cfg = load_config()
val = cfg.get("file_read_max_chars")
if isinstance(val, (int, float)) and val > 0:
_max_read_chars_cached = int(val)
return _max_read_chars_cached
except Exception:
pass
_max_read_chars_cached = _DEFAULT_MAX_READ_CHARS
return _max_read_chars_cached
# If the total file size exceeds this AND the caller didn't specify a narrow
# range (limit <= 200), we include a hint encouraging targeted reads.
_LARGE_FILE_HINT_BYTES = 512_000 # 512 KB
# ---------------------------------------------------------------------------
# Device path blocklist — reading these hangs the process (infinite output
# or blocking on input). Checked by path only (no I/O).
# ---------------------------------------------------------------------------
_BLOCKED_DEVICE_PATHS = frozenset({
# Infinite output — never reach EOF
"/dev/zero", "/dev/random", "/dev/urandom", "/dev/full",
# Blocks waiting for input
"/dev/stdin", "/dev/tty", "/dev/console",
# Nonsensical to read
"/dev/stdout", "/dev/stderr",
# fd aliases
"/dev/fd/0", "/dev/fd/1", "/dev/fd/2",
})
def _is_blocked_device(filepath: str) -> bool:
"""Return True if the path would hang the process (infinite output or blocking input).
Uses the *literal* path — no symlink resolution — because the model
specifies paths directly and realpath follows symlinks all the way
through (e.g. /dev/stdin → /proc/self/fd/0 → /dev/pts/0), defeating
the check.
"""
normalized = os.path.expanduser(filepath)
if normalized in _BLOCKED_DEVICE_PATHS:
return True
# /proc/self/fd/0-2 and /proc/<pid>/fd/0-2 are Linux aliases for stdio
if normalized.startswith("/proc/") and normalized.endswith(
("/fd/0", "/fd/1", "/fd/2")
):
return True
return False
# Paths that file tools should refuse to write to without going through the
# terminal tool's approval system. These match prefixes after os.path.realpath.
_SENSITIVE_PATH_PREFIXES = ("/etc/", "/boot/", "/usr/lib/systemd/")
@ -53,11 +127,15 @@ def _is_expected_write_exception(exc: Exception) -> bool:
_file_ops_lock = threading.Lock()
_file_ops_cache: dict = {}
# Track files read per task to detect re-read loops after context compression.
# Track files read per task to detect re-read loops and deduplicate reads.
# Per task_id we store:
# "last_key": the key of the most recent read/search call (or None)
# "consecutive": how many times that exact call has been repeated in a row
# "read_history": set of (path, offset, limit) tuples for get_read_files_summary
# "dedup": dict mapping (resolved_path, offset, limit) → mtime float
# Used to skip re-reads of unchanged files. Reset on
# context compression (the original content is summarised
# away so the model needs the full content again).
_read_tracker_lock = threading.Lock()
_read_tracker: dict = {}
@ -195,8 +273,19 @@ def clear_file_ops_cache(task_id: str = None):
def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str = "default") -> str:
"""Read a file with pagination and line numbers."""
try:
# Security: block direct reads of internal Hermes cache/index files
# to prevent prompt injection via catalog or hub metadata files.
# ── Device path guard ─────────────────────────────────────────
# Block paths that would hang the process (infinite output,
# blocking on input). Pure path check — no I/O.
if _is_blocked_device(path):
return json.dumps({
"error": (
f"Cannot read '{path}': this is a device file that would "
"block or produce infinite output."
),
})
# ── Hermes internal path guard ────────────────────────────────
# Prevent prompt injection via catalog or hub metadata files.
import pathlib as _pathlib
from hermes_constants import get_hermes_home as _get_hh
_resolved = _pathlib.Path(path).expanduser().resolve()
@ -217,20 +306,83 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str =
})
except ValueError:
pass
# ── Dedup check ───────────────────────────────────────────────
# If we already read this exact (path, offset, limit) and the
# file hasn't been modified since, return a lightweight stub
# instead of re-sending the same content. Saves context tokens.
resolved_str = str(_resolved)
dedup_key = (resolved_str, offset, limit)
with _read_tracker_lock:
task_data = _read_tracker.setdefault(task_id, {
"last_key": None, "consecutive": 0,
"read_history": set(), "dedup": {},
})
cached_mtime = task_data.get("dedup", {}).get(dedup_key)
if cached_mtime is not None:
try:
current_mtime = os.path.getmtime(resolved_str)
if current_mtime == cached_mtime:
return json.dumps({
"content": (
"File unchanged since last read. The content from "
"the earlier read_file result in this conversation is "
"still current — refer to that instead of re-reading."
),
"path": path,
"dedup": True,
}, ensure_ascii=False)
except OSError:
pass # stat failed — fall through to full read
# ── Perform the read ──────────────────────────────────────────
file_ops = _get_file_ops(task_id)
result = file_ops.read_file(path, offset, limit)
if result.content:
result.content = redact_sensitive_text(result.content)
result_dict = result.to_dict()
# Track reads to detect *consecutive* re-read loops.
# The counter resets whenever any other tool is called in between,
# so only truly back-to-back identical reads trigger warnings/blocks.
# ── Character-count guard ─────────────────────────────────────
# We're model-agnostic so we can't count tokens; characters are
# the best proxy we have. If the read produced an unreasonable
# amount of content, reject it and tell the model to narrow down.
# Note: we check the formatted content (with line-number prefixes),
# not the raw file size, because that's what actually enters context.
content_len = len(result.content or "")
file_size = result_dict.get("file_size", 0)
max_chars = _get_max_read_chars()
if content_len > max_chars:
total_lines = result_dict.get("total_lines", "unknown")
return json.dumps({
"error": (
f"Read produced {content_len:,} characters which exceeds "
f"the safety limit ({max_chars:,} chars). "
"Use offset and limit to read a smaller range. "
f"The file has {total_lines} lines total."
),
"path": path,
"total_lines": total_lines,
"file_size": file_size,
}, ensure_ascii=False)
# Large-file hint: if the file is big and the caller didn't ask
# for a narrow window, nudge toward targeted reads.
if (file_size and file_size > _LARGE_FILE_HINT_BYTES
and limit > 200
and result_dict.get("truncated")):
result_dict.setdefault("_hint", (
f"This file is large ({file_size:,} bytes). "
"Consider reading only the section you need with offset and limit "
"to keep context usage efficient."
))
# ── Track for consecutive-loop detection ──────────────────────
read_key = ("read", path, offset, limit)
with _read_tracker_lock:
task_data = _read_tracker.setdefault(task_id, {
"last_key": None, "consecutive": 0, "read_history": set(),
})
# Ensure "dedup" key exists (backward compat with old tracker state)
if "dedup" not in task_data:
task_data["dedup"] = {}
task_data["read_history"].add((path, offset, limit))
if task_data["last_key"] == read_key:
task_data["consecutive"] += 1
@ -239,6 +391,15 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 500, task_id: str =
task_data["consecutive"] = 1
count = task_data["consecutive"]
# Store dedup entry (mtime at read time).
# Writes/patches will naturally change mtime, so subsequent
# dedup checks after edits will see a different mtime and
# return the full content — no special handling needed.
try:
task_data["dedup"][dedup_key] = os.path.getmtime(resolved_str)
except OSError:
pass # Can't stat — skip dedup for this entry
if count >= 4:
# Hard block: stop returning content to break the loop
return json.dumps({
@ -296,6 +457,28 @@ def clear_read_tracker(task_id: str = None):
_read_tracker.clear()
def reset_file_dedup(task_id: str = None):
"""Clear the deduplication cache for file reads.
Called after context compression — the original read content has been
summarised away, so the model needs the full content if it reads the
same file again. Without this, reads after compression would return
a "file unchanged" stub pointing at content that no longer exists in
context.
Call with a task_id to clear just that task, or without to clear all.
"""
with _read_tracker_lock:
if task_id:
task_data = _read_tracker.get(task_id)
if task_data and "dedup" in task_data:
task_data["dedup"].clear()
else:
for task_data in _read_tracker.values():
if "dedup" in task_data:
task_data["dedup"].clear()
def notify_other_tool_call(task_id: str = "default"):
"""Reset consecutive read/search counter for a task.
@ -466,7 +649,7 @@ def _check_file_reqs():
READ_FILE_SCHEMA = {
"name": "read_file",
"description": "Read a text file with line numbers and pagination. Use this instead of cat/head/tail in terminal. Output format: 'LINE_NUM|CONTENT'. Suggests similar filenames if not found. Use offset and limit for large files. NOTE: Cannot read images or binary files — use vision_analyze for images.",
"description": "Read a text file with line numbers and pagination. Use this instead of cat/head/tail in terminal. Output format: 'LINE_NUM|CONTENT'. Suggests similar filenames if not found. Use offset and limit for large files. Reads exceeding ~100K characters are rejected; use offset and limit to read specific sections of large files. NOTE: Cannot read images or binary files — use vision_analyze for images.",
"parameters": {
"type": "object",
"properties": {

20
website/docs/user-guide/configuration.md
View file

@ -360,6 +360,26 @@ memory:
user_char_limit: 1375 # ~500 tokens
```
## File Read Safety
Controls how much content a single `read_file` call can return. Reads that exceed the limit are rejected with an error telling the agent to use `offset` and `limit` for a smaller range. This prevents a single read of a minified JS bundle or large data file from flooding the context window.
```yaml
file_read_max_chars: 100000 # default — ~25-35K tokens
```
Raise it if you're on a model with a large context window and frequently read big files. Lower it for small-context models to keep reads efficient:
```yaml
# Large context model (200K+)
file_read_max_chars: 200000
# Small local model (16K context)
file_read_max_chars: 30000
```
The agent also deduplicates file reads automatically — if the same file region is read twice and the file hasn't changed, a lightweight stub is returned instead of re-sending the content. This resets on context compression so the agent can re-read files after their content is summarized away.
## Git Worktree Isolation
Enable isolated git worktrees for running multiple agents in parallel on the same repo: