hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity

fix: preserve known file baselines across partial rereads

Browse Source
This commit is contained in:
teknium1
2026-09-21 15:34:23 -07:00
committed by Teknium
parent eeb220d40c
commit e636aedebf
6 changed files with 205 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

76
tests/tools/test_known_file_write_baseline.py Normal file
View File

@@ -0,0 +1,76 @@
"""A narrower read must not discard unchanged whole-file knowledge."""
import json
import os
from tools.file_tools import clear_file_ops_cache
from tools.registry import registry
def _call(name, path, task_id, **arguments):
result = registry.dispatch(name, {"path": str(path), **arguments}, task_id=task_id)
assert isinstance(result, str)
return json.loads(result)
def test_partial_reread_keeps_an_unchanged_full_read_or_write(tmp_path):
original = "first\nsecond\nthird\n"
for source in ("read", "write", "pages"):
task = f"known-{source}"
path = tmp_path / f"{source}.txt"
try:
if source == "write":
assert "error" not in _call("write_file", path, task, content=original)
else:
path.write_text(original, encoding="utf-8")
if source == "pages":
for offset in (1, 2, 3):
assert "error" not in _call("read_file", path, task, offset=offset, limit=1)
else:
assert "error" not in _call("read_file", path, task)
assert "error" not in _call("read_file", path, task, offset=2, limit=2)
written = _call("write_file", path, task, content="replacement\n")
assert "error" not in written, (source, written)
assert path.read_text(encoding="utf-8") == "replacement\n"
finally:
clear_file_ops_cache(task)
def test_partial_read_cannot_refresh_a_changed_full_baseline(tmp_path):
original = "first\nsecond\nthird\n"
for source in ("write", "pages"):
path = tmp_path / f"changed-{source}.txt"
task = f"snapshot-{source}"
try:
if source == "write":
assert "error" not in _call("write_file", path, task, content=original)
else:
path.write_text(original, encoding="utf-8")
assert "error" not in _call("read_file", path, task, offset=1, limit=1)
stamp = path.stat()
path.write_text("other\nsecond\nthird\n", encoding="utf-8")
# mtime alone cannot identify bytes: editors/copy tools can preserve it.
os.utime(path, ns=(stamp.st_atime_ns, stamp.st_mtime_ns))
assert "error" not in _call("read_file", path, task, offset=2, limit=2)
refused = _call("write_file", path, task, content="replacement\n")
assert refused.get("stale_write_blocked"), (source, refused)
assert path.read_text(encoding="utf-8") == "other\nsecond\nthird\n"
# Reading all of the new version is recovery, not a bypass.
assert "error" not in _call("read_file", path, task)
assert "error" not in _call("write_file", path, task, content="merged\n")
assert path.read_text(encoding="utf-8") == "merged\n"
finally:
clear_file_ops_cache(task)
# A page that hides part of a line never supplies whole-file knowledge.
from tools.tool_output_limits import get_max_line_length
path = tmp_path / "clamped.txt"
path.write_text("x" * (get_max_line_length() + 10) + "\nlast\n", encoding="utf-8")
try:
assert "error" not in _call("read_file", path, "clamped")
refused = _call("write_file", path, "clamped", content="replacement\n")
assert refused.get("stale_write_blocked"), refused
assert path.read_text(encoding="utf-8").startswith("x" * (get_max_line_length() + 10))
finally:
clear_file_ops_cache("clamped")

19
tools/file_operations.py
View File

@@ -581,9 +581,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
hint_parts.append(
f"Use offset={end_line + 1} to continue reading "
f"(showing {offset}-{end_line} of {total_lines} lines)")
from tools.tool_output_limits import get_max_line_length
max_line_length = get_max_line_length()
truncated_lines = any(len(line) > max_line_length for line in content.split('\n'))
return ReadResult(
content=self._add_line_numbers(content, offset), total_lines=total_lines,
file_size=file_size, truncated=truncated, hint=" ".join(hint_parts))
file_size=file_size, truncated=truncated, hint=" ".join(hint_parts),
truncated_lines=True if truncated_lines else None)
def read_file(self, path: str, offset: int = 1, limit: int = 2000) -> ReadResult:
"""Read a file with pagination, binary detection, and line numbers.
@@ -724,6 +728,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
kept = bytearray() # first ``clamp`` bytes of that line
have_partial = False # that line has bytes but no newline yet
last_byte = b""
digest = hashlib.sha256()
try:
with open(full, "rb") as fh:
sample = fh.read(1000)
@@ -735,6 +740,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
chunk = fh.read(1 << 20)
if not chunk:
break
digest.update(chunk)
last_byte = chunk[-1:]
if lineno > end_line:
# Past the window: only the line count and trailing byte
@@ -767,10 +773,12 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
page.append(bytes(kept) + b"\n")
read_output = _strip_terminal_fence_leaks(b"".join(page).decode("utf-8", errors="replace"))
return self._assemble_read_result(
result = self._assemble_read_result(
read_output, offset=offset, end_line=end_line, total_lines=total_lines,
file_size=file_size,
file_ends_with_newline=(last_byte == b"\n") if file_size else None)
result._snapshot = (st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns, digest.digest())
return result
@staticmethod
def _image_redirect_result(file_size: int) -> ReadResult:
@@ -924,9 +932,13 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
f"Note: offset {offset} is beyond the end of the file "
f"({total_lines} lines total). Retry with offset <= "
f"{total_lines}."))
from tools.tool_output_limits import get_max_line_length
max_line_length = get_max_line_length()
truncated_lines = any(len(line) > max_line_length for line in read_output.split('\n'))
return ReadResult(
content=self._add_line_numbers(read_output, offset), total_lines=total_lines,
file_size=file_size, truncated=truncated, hint=hint)
file_size=file_size, truncated=truncated, hint=hint,
truncated_lines=True if truncated_lines else None)
# Confusable characters seen in real filenames, collapsed after NFC.
_CONFUSABLES = (
@@ -1295,6 +1307,7 @@ class ShellFileOperations(LintMixin, SearchMixin, FileOperations):
lsp_diagnostics = self._maybe_lsp_diagnostics(path, pre_content=pre_content, post_content=content) or None
return WriteResult(
bytes_written=len(content_bytes), dirs_created=dirs_created, verified=content_verified,
_content_sha256=hashlib.sha256(content_bytes).hexdigest(),
lint=lint_result.to_dict() if lint_result else None, lsp_diagnostics=lsp_diagnostics)
# --- PATCH (replace mode) -----------------------------------------------

7
tools/file_operations_common.py
View File

@@ -17,6 +17,7 @@ class ReadResult:
total_lines: int = 0
file_size: int = 0
truncated: bool = False
truncated_lines: Optional[bool] = None
hint: Optional[str] = None
is_binary: bool = False
is_image: bool = False
@@ -25,9 +26,10 @@ class ReadResult:
dimensions: Optional[str] = None # For images: "WIDTHxHEIGHT"
error: Optional[str] = None
similar_files: List[str] = field(default_factory=list)
_snapshot: Optional[tuple] = None
def to_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items() if v is not None and v != []}
return {k: v for k, v in self.__dict__.items() if not k.startswith("_") and v is not None and v != []}
@dataclass
@@ -38,6 +40,7 @@ class WriteResult:
# True when the on-disk sha256 matched the intended content; None when the
# backend couldn't verify (no sha256sum). A mismatch is a hard error, never a flag.
verified: Optional[bool] = None
_content_sha256: Optional[str] = None
lint: Optional[Dict[str, Any]] = None
# LSP semantic diagnostics, kept separate from ``lint`` (syntax) so the model
# reads the two as independent signals. None when LSP is off/inapplicable.
@@ -46,7 +49,7 @@ class WriteResult:
warning: Optional[str] = None
def to_dict(self) -> dict:
return {k: v for k, v in self.__dict__.items() if v is not None}
return {k: v for k, v in self.__dict__.items() if not k.startswith("_") and v is not None}
@dataclass

69
tools/file_tools.py
View File

@@ -35,6 +35,7 @@ from tools.file_tools_write_guards import (
_is_internal_file_tool_content, _stale_overwrite_blocker, _stale_write_refusal)
from tools.file_tools_read_tracking import (
_bump_consecutive, _cap_read_tracker_data, _check_file_staleness, _check_not_found_cache,
_file_metadata, _file_version,
_mark_full_write_baseline, _mark_verification_stale, _note_read_coverage, _patch_failure_lock,
_patch_failure_tracker, _read_tracker, _read_tracker_lock, _record_not_found,
_record_patch_failure, _reset_patch_failures, _task_data, _update_read_timestamp)
@@ -104,6 +105,7 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
f"{lines_kept} line(s) (showing lines {offset}-{next_offset - 1} of "
f"{total_lines}). Use offset={next_offset} to continue.")
if len(trimmed.split("\n", 1)[0]) >= max_chars:
result_dict["truncated_lines"] = True
result_dict["hint"] += (
" Note: the first line alone exceeded the budget and was "
"clamped mid-line; its remainder is not retrievable via offset.")
@@ -455,6 +457,9 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
total_lines = len(lines)
end_line = offset + limit - 1
page_text = "\n".join(lines[offset - 1:end_line])
from tools.tool_output_limits import get_max_line_length
max_line_length = get_max_line_length()
truncated_lines = any(len(line) > max_line_length for line in page_text.split('\n'))
result_dict = {
"content": file_ops._add_line_numbers(page_text, offset) if page_text else "",
"total_lines": total_lines,
@@ -476,7 +481,8 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
redacted = result_dict["content"] != rendered
else:
redacted = False
if offset == 1 and not result_dict["truncated"] and not redacted:
if (offset == 1 and not result_dict["truncated"] and not redacted
and not truncated_lines and not result_dict.get("truncated_lines")):
# The whole document was shown, so a text-authorable format (.ipynb)
# may later be overwritten by write_file; the binary-container guard
# keeps refusing .docx/.xlsx/.pdf regardless of this baseline.
@@ -522,7 +528,7 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_str: str,
offset: int, limit: int, dedup_key: tuple, *, partial: bool,
redacted: bool = False, end_line: int | None = None,
total_lines=None) -> int:
total_lines=None, version_before=None, snapshot=None) -> int:
"""Bookkeeping after a real (non-stub) read; returns the consecutive-read count.
Per-task tracker under the lock (stub counter, history, consecutive count,
@@ -535,7 +541,9 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
background-review read-mark (a FULL read of a skill file counts like
skill_view so a follow-up skill_manage(patch) is accepted).
"""
complete = not partial
version = (snapshot or _file_version(resolved_str)) if version_before is not None else None
stable = version is not None and version[:-1] == version_before == _file_metadata(resolved_str)
complete = False
with _read_tracker_lock:
task_data["dedup_hits"].pop(dedup_key, None)
task_data["dedup_generation_reads"].add(dedup_key)
@@ -543,15 +551,28 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
count = _bump_consecutive(task_data, ("read", path, offset, limit))
try:
_mtime_now = os.path.getmtime(resolved_str)
task_data["dedup"][dedup_key] = _mtime_now
task_data.setdefault("read_timestamps", {})[resolved_str] = _mtime_now
if partial and end_line is not None:
complete, redacted = _note_read_coverage(
task_data, resolved_str, _mtime_now, offset, end_line, total_lines, redacted)
except OSError:
pass
if complete and not redacted:
task_data.setdefault("full_write_baselines", set()).add(resolved_str)
baselines = task_data["full_write_baselines"]
if stable and version is not None and count < 4:
task_data["dedup"][dedup_key] = version_before
# A narrower view does not undo knowledge of these same bytes. Do
# not revive a baseline after a partial read of a different version.
complete = baselines.get(resolved_str) == version
if not complete:
complete = not partial
if partial and end_line is not None:
complete, redacted = _note_read_coverage(
task_data, resolved_str, version, offset, end_line, total_lines, redacted)
complete = complete and not redacted
if complete:
baselines[resolved_str] = version
if not complete:
baselines.pop(resolved_str, None)
if not stable or count >= 4:
task_data["dedup"].pop(dedup_key, None)
task_data["dedup_generation_reads"].discard(dedup_key)
_cap_read_tracker_data(task_data)
try:
@@ -642,27 +663,27 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
dedup_key = (resolved_str, offset, limit)
with _read_tracker_lock:
task_data = _task_data(task_id)
cached_mtime = task_data["dedup"].get(dedup_key)
cached_version = task_data["dedup"].get(dedup_key)
# First unchanged read after a compaction boundary serves full content
# (the summary may have dropped exact bytes); later ones get the stub.
content_served_in_generation = dedup_key in task_data["dedup_generation_reads"]
# Same rule as skill_view: the review fork shares the parent's task_id and its
# read-before-write guard needs a real read, which the stub path never records (#95976).
if cached_mtime is not None and not is_background_review():
try:
if os.path.getmtime(resolved_str) == cached_mtime and content_served_in_generation:
return _dedup_stub_or_block(task_data, dedup_key, path)
except OSError:
pass # stat failed — fall through to full read
file_ops = _get_file_ops(task_id)
version_before = _file_metadata(resolved_str) if _file_ops_uses_host_paths(file_ops) else None
if (cached_version is not None and not is_background_review()
and version_before == cached_version and content_served_in_generation):
return _dedup_stub_or_block(task_data, dedup_key, path)
result = _get_file_ops(task_id).read_file(path, offset, limit)
result = file_ops.read_file(path, offset, limit)
result_dict = result.to_dict()
# Cache a not-found result for retries. Deliberately NO early return:
# error results still flow through the tracking below unchanged.
# Failed reads cannot establish whole-file knowledge.
_err = result_dict.get("error") or ""
if isinstance(_err, str) and _err.startswith("File not found:"):
_record_not_found("read", resolved_str, task_id, json.dumps(result_dict, ensure_ascii=False))
if _err or result_dict.get("is_binary"):
return json.dumps(result_dict, ensure_ascii=False)
# Char budget on the FORMATTED content (what enters context), BEFORE
# redaction (skip the regex pass on huge content); truncate gracefully
@@ -705,7 +726,10 @@ def read_file_tool(path: str, offset: int = 1, limit: int = DEFAULT_READ_LIMIT,
end_line = min(end_line, total_lines)
count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")),
redacted=redacted, end_line=end_line, total_lines=total_lines)
redacted=redacted or bool(result_dict.get("truncated_lines")),
end_line=end_line, total_lines=total_lines,
version_before=version_before,
snapshot=getattr(result, "_snapshot", None))
if count >= 4:
return tool_error(
f"BLOCKED: You have read this exact file region {count} times in a row. "
@@ -865,7 +889,8 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
return json.dumps(_stale_write_refusal(path, blocker, _resolved), ensure_ascii=False)
warnings = _edit_warnings([path], path_to_resolved, task_id)
rewrite_hint = _whole_file_rewrite_hint(task_id, _resolved, content)
result_dict = _get_file_ops(task_id).write_file(_resolved or path, content).to_dict()
result = _get_file_ops(task_id).write_file(_resolved or path, content)
result_dict = result.to_dict()
if warnings:
result_dict["_warning"] = warnings[0]
if rewrite_hint and not result_dict.get("error"):
@@ -881,7 +906,7 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
result_dict["files_modified"] = [_resolved]
# Own write = current whole-file content: consecutive
# same-task writes stay unblocked. patch never does this.
_mark_full_write_baseline(_resolved, task_id)
_mark_full_write_baseline(_resolved, task_id, getattr(result, "_content_sha256", None))
_note_edited(task_id, [path], path_to_resolved, session_id)
return json.dumps(result_dict, ensure_ascii=False)
except Exception as e:

81
tools/file_tools_read_tracking.py
View File

@@ -3,12 +3,12 @@
Process-lifetime state behind read_file/search_files/write_file/patch.
Per task_id ``_read_tracker``
stores: ``last_key``/``consecutive`` (loop detection; reset by any OTHER tool
call), ``read_history`` (diagnostics), ``dedup`` (key -> mtime; survives context
call), ``read_history`` (diagnostics), ``dedup`` (key -> file metadata; survives context
compression), ``dedup_generation_reads`` (keys whose full content was served since
the last compaction boundary; cleared on compression so one recovery read returns
full content), ``dedup_hits`` (stub-loop breaker), ``read_timestamps``
(staleness warnings), ``read_coverage`` (per resolved path: the line ranges the
task has paged through at one mtime — contiguous pages that reach the last line
task has paged through at one file version — contiguous pages that reach the last line
count as a whole-file read), ``full_write_baselines`` (resolved paths whose
whole-file content this task saw via unredacted read_file page(s) or wrote via
write_file; required before write_file may overwrite an existing file — patch
@@ -16,12 +16,14 @@ never qualifies) and ``not_found`` (short-TTL negative cache). Every
container is hard-capped (``_cap_read_tracker_data``) so long sessions stay small.
"""
import hashlib
import logging
import os
import stat
import threading
import time
from tools.file_state import _evict_oldest, _mtime_or_none
from tools.file_state import _evict_oldest
from tools.file_tools_paths import _authoritative_workspace_root, _resolve_path_for_task
logger = logging.getLogger("tools.file_tools")
@@ -50,10 +52,9 @@ def _task_data(task_id: str) -> dict:
(search_tool / tests create partial entries). Lock must be held."""
task_data = _read_tracker.setdefault(task_id, {
"last_key": None, "consecutive": 0, "read_history": set()})
for key in ("dedup", "dedup_hits", "read_timestamps", "read_coverage"):
for key in ("dedup", "dedup_hits", "read_timestamps", "read_coverage", "full_write_baselines"):
task_data.setdefault(key, {})
for key in ("dedup_generation_reads", "full_write_baselines"):
task_data.setdefault(key, set())
task_data.setdefault("dedup_generation_reads", set())
return task_data
@@ -155,13 +156,14 @@ def _bump_consecutive(task_data: dict, key: tuple) -> int:
def reset_file_dedup(task_id: str = None):
"""Advance the read-dedup generation after context compression (one task, or all
when ``task_id`` is None). The per-key ``dedup`` mtime map is PRESERVED so unchanged
when ``task_id`` is None). The per-key ``dedup`` metadata map is PRESERVED so unchanged
files keep returning stubs instead of re-bloating the reclaimed context; the
generation-read set is cleared so the FIRST unchanged read of each key after
compaction returns full content the summary may have dropped. Stub-hit counters
are cleared so the hard block restarts fresh. write_file baselines survive
exactly like the dedup map does — for files whose mtime still matches the
stamp this task recorded; a baseline whose file changed underneath is dropped
exactly like the dedup map does — while the file metadata still matches the
stamp this task recorded; byte identity is checked before overwriting. A baseline
whose file changed underneath is dropped
(the stat runs outside the lock so a hung mount cannot stall other tasks)."""
with _read_tracker_lock:
if task_id:
@@ -172,13 +174,15 @@ def reset_file_dedup(task_id: str = None):
if "dedup_hits" in task_data:
task_data["dedup_hits"].clear()
task_data.setdefault("dedup_generation_reads", set()).clear()
candidates = [(task_data, list(task_data.get("full_write_baselines", ())),
dict(task_data.get("read_timestamps", {}))) for task_data in targets]
for task_data, baselines, stamps in candidates:
changed = {p for p in baselines if _mtime_or_none(p) != stamps.get(p)}
candidates = [(task_data, dict(task_data.get("full_write_baselines", {})))
for task_data in targets]
for task_data, baselines in candidates:
changed = {p for p, version in baselines.items() if _file_metadata(p) != version[:-1]}
if changed:
with _read_tracker_lock:
task_data.get("full_write_baselines", set()).difference_update(changed)
for p in changed:
if task_data["full_write_baselines"].get(p) == baselines[p]:
task_data["full_write_baselines"].pop(p, None)
def notify_other_tool_call(task_id: str = "default"):
@@ -238,37 +242,70 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
_cap_read_tracker_data(task_data)
def _mark_full_write_baseline(resolved: str, task_id: str) -> None:
def _file_metadata(resolved: str) -> tuple | None:
try:
st = os.stat(resolved)
return st.st_dev, st.st_ino, st.st_size, st.st_mtime_ns, st.st_ctime_ns
except OSError:
return None
def _file_version(resolved: str) -> tuple | None:
"""A byte snapshot, not just mtime (editors/copy tools can preserve that)."""
try:
if not stat.S_ISREG(os.stat(resolved).st_mode):
return None
fd = os.open(resolved, os.O_RDONLY | getattr(os, "O_NONBLOCK", 0))
with os.fdopen(fd, "rb") as stream:
before = os.fstat(stream.fileno())
if not stat.S_ISREG(before.st_mode):
return None
digest = hashlib.file_digest(stream, "sha256").digest()
after = os.stat(resolved)
fields = ("st_dev", "st_ino", "st_size", "st_mtime_ns", "st_ctime_ns")
version = tuple(getattr(before, name) for name in fields)
if version == tuple(getattr(after, name) for name in fields):
return (*version, digest)
return None
except OSError:
return None
def _mark_full_write_baseline(resolved: str, task_id: str, expected_sha256: str | None = None) -> None:
"""Record that *task_id* saw the whole current content of *resolved* (full
unredacted read_file, or its own successful write_file), so a later
write_file may replace the file. Acquires the lock itself."""
version = _file_version(resolved)
if version is None or (expected_sha256 is not None and version[-1].hex() != expected_sha256):
return
with _read_tracker_lock:
task_data = _task_data(task_id)
task_data["full_write_baselines"].add(str(resolved))
task_data["full_write_baselines"][str(resolved)] = version
_cap_read_tracker_data(task_data)
def _has_full_write_baseline(resolved: str, task_id: str) -> bool:
with _read_tracker_lock:
task_data = _read_tracker.get(task_id) or {}
return str(resolved) in task_data.get("full_write_baselines", set())
baseline = task_data.get("full_write_baselines", {}).get(str(resolved))
return baseline is not None and _file_version(resolved) == baseline
_READ_COVERAGE_RANGES_CAP = 256
def _note_read_coverage(task_data: dict, resolved: str, mtime: float, start: int, end: int,
def _note_read_coverage(task_data: dict, resolved: str, version: tuple, start: int, end: int,
total_lines, redacted: bool) -> tuple[bool, bool]:
"""Merge the page ``start..end`` into this task's coverage of *resolved* and return
``(complete, redacted_any)``: whether pages taken at this same *mtime* now reach from
``(complete, redacted_any)``: whether pages taken at this same *version* now reach from
line 1 to *total_lines*, and whether any of them came back redacted. A file too large
for one read_file page (>2000 lines / the char budget) can only ever be seen this
way, so paging through it must count as a whole-file read. A new mtime restarts the
way, so paging through it must count as a whole-file read. A new version restarts the
coverage (the earlier pages describe a file that no longer exists). Lock must be held."""
coverage = task_data.setdefault("read_coverage", {})
entry = coverage.get(resolved)
if entry is None or entry["mtime"] != mtime or len(entry["ranges"]) > _READ_COVERAGE_RANGES_CAP:
entry = coverage[resolved] = {"mtime": mtime, "ranges": [], "redacted": False}
if entry is None or entry["version"] != version or len(entry["ranges"]) > _READ_COVERAGE_RANGES_CAP:
entry = coverage[resolved] = {"version": version, "ranges": [], "redacted": False}
entry["redacted"] = entry["redacted"] or redacted
merged: list[tuple[int, int]] = []
for s, e in sorted(entry["ranges"] + [(start, end)]):

2
website/docs/reference/tools-reference.md
View File

@@ -117,6 +117,8 @@ Scoped to the Feishu document-comment handler. Drives comment read/write operati
| `search_files` | Search file contents or find files by name. Use this instead of grep/rg/find/ls in terminal. Ripgrep-backed, faster than shell equivalents. Content search (target='content'): Regex search inside files. Output modes: full matches with line… | — |
| `write_file` | Write content to a file, completely replacing existing content. Use this instead of echo/cat heredoc in terminal. Creates parent directories automatically. OVERWRITES the entire file — use 'patch' for targeted edits. For an existing file, call read_file first: write_file refuses (file untouched) when the task has no current full read/write of the file or the file changed on disk since — on refusal, read_file, merge, retry. Auto-runs syntax checks on .py/.json/.yaml/.toml and other linted languages; only NEW errors introduced by the write are surfaced. | — |
For local files, a full unredacted read (including all pages of the same file version) or a successful `write_file` supplies a whole-file baseline. Reading a smaller region afterward does not discard that baseline while the bytes remain unchanged. A changed file, an unread file, or a view with hidden/redacted or clamped content still needs a full current read before replacement; `patch` remains available for targeted edits. Writes made through terminal commands or `execute_code` do not establish a `write_file` baseline.
## `homeassistant` toolset
| Tool | Description | Requires environment |