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

Merge branch 'simp/r3-32-A' into simp/integration3

Browse Source
This commit is contained in:
Teknium
2026-09-02 23:12:01 -07:00
parent 65fce48e0f 48de33b77f
commit dffbe435f7
6 changed files with 361 additions and 755 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/file_state.py
+64 -108
View File
@@ -1,24 +1,11 @@
"""Cross-agent file state coordination.
Prevents mangled edits when concurrent subagents (same process, same
filesystem) touch the same file: subagent B writes a file that subagent A
already read, so A's next write would clobber B's changes with stale content.
Complements the single-agent path-overlap check in
``run_agent._should_parallelize_tool_batch``.
A process-wide ``FileStateRegistry`` tracks, per resolved path:
* per-agent read stamps {task_id: {path: (mtime, read_ts, partial)}}
* last writer globally {path: (task_id, write_ts)}
* a per-path ``threading.Lock`` for read->modify->write sections
Hooks used by the file tools: ``record_read`` (read_file), ``note_write``
(after write_file/patch), ``check_stale`` (BEFORE write_file/patch),
``lock_path`` (wrap the whole read->modify->write block) and ``writes_since``
(delegate_tool's subagent-completion reminder).
All methods are no-ops when ``HERMES_DISABLE_FILE_STATE_GUARD=1``. This is
separate from ``file_tools._read_tracker``, which handles per-task
consecutive-read loop detection.
filesystem) touch the same file: B writes a file A already read, so A's next
write would clobber B's changes. Complements the single-agent path-overlap
check in ``run_agent._should_parallelize_tool_batch``. A process-wide
``FileStateRegistry`` tracks per-agent read stamps, the global last writer and
a per-path lock; every method is a no-op under ``HERMES_DISABLE_FILE_STATE_GUARD=1``.
"""
from __future__ import annotations
@@ -35,12 +22,45 @@ from typing import Dict, Iterable, List, Optional, Tuple
# so the model re-reads in full.
ReadStamp = Tuple[float, float, bool]
# Bounded so long sessions don't accumulate unbounded state; oldest by
# insertion order are dropped on overflow.
# Bounded so long sessions don't accumulate unbounded state.
_MAX_PATHS_PER_AGENT = 4096
_MAX_GLOBAL_WRITERS = 4096
def _disabled() -> bool:
# Re-read each call so tests can toggle via monkeypatch.setenv.
return os.environ.get("HERMES_DISABLE_FILE_STATE_GUARD", "").strip() == "1"
def _mtime_or_none(resolved: str) -> Optional[float]:
try:
return os.path.getmtime(resolved)
except OSError:
return None
def _fmt_ts(ts: float) -> str:
# Short wall-clock for warnings; avoids datetime formatting on the hot path.
return time.strftime("%H:%M:%S", time.localtime(ts))
def _evict_oldest(container, cap: int) -> None:
"""Pop entries until *container* is within *cap*.
Sets pop arbitrary entries (they only feed diagnostic summaries); dicts pop
oldest by insertion order. An evicted entry costs one redundant re-send
(dedup) or one non-mtime staleness check — graceful degradation, not a bug.
"""
for _ in range(len(container) - cap):
try:
if isinstance(container, set):
container.pop()
else:
container.pop(next(iter(container)))
except (StopIteration, KeyError):
break
class FileStateRegistry:
"""Process-wide coordinator for cross-agent file edits."""
@@ -51,50 +71,31 @@ class FileStateRegistry:
self._meta_lock = threading.Lock() # guards _path_locks
self._state_lock = threading.Lock() # guards _reads + _last_writer
def _lock_for(self, resolved: str) -> threading.Lock:
with self._meta_lock:
lock = self._path_locks.get(resolved)
if lock is None:
lock = threading.Lock()
self._path_locks[resolved] = lock
return lock
@contextmanager
def lock_path(self, resolved: str):
"""Per-path lock: threads on the same path serialize, different paths proceed."""
lock = self._lock_for(resolved)
lock.acquire()
try:
with self._meta_lock:
lock = self._path_locks.setdefault(resolved, threading.Lock())
with lock:
yield
finally:
lock.release()
def record_read(
self,
task_id: str,
resolved: str,
*,
partial: bool = False,
mtime: Optional[float] = None,
) -> None:
def _stamp(self, task_id: str, resolved: str, mtime: float, now: float, partial: bool) -> None:
"""Caller holds ``_state_lock``."""
agent_reads = self._reads[task_id]
agent_reads[resolved] = (float(mtime), now, bool(partial))
_evict_oldest(agent_reads, _MAX_PATHS_PER_AGENT)
def record_read(self, task_id: str, resolved: str, *, partial: bool = False,
mtime: Optional[float] = None) -> None:
if _disabled():
return
mtime = _mtime_or_none(resolved) if mtime is None else mtime
if mtime is None:
return
now = time.time()
with self._state_lock:
agent_reads = self._reads[task_id]
agent_reads[resolved] = (float(mtime), now, bool(partial))
_cap_dict(agent_reads, _MAX_PATHS_PER_AGENT)
self._stamp(task_id, resolved, mtime, time.time(), partial)
def note_write(
self,
task_id: str,
resolved: str,
*,
mtime: Optional[float] = None,
) -> None:
def note_write(self, task_id: str, resolved: str, *, mtime: Optional[float] = None) -> None:
"""Record a successful write: global last-writer AND this agent's own
read stamp (a write is an implicit read of the current content)."""
if _disabled():
@@ -105,9 +106,8 @@ class FileStateRegistry:
now = time.time()
with self._state_lock:
self._last_writer[resolved] = (task_id, now)
_cap_dict(self._last_writer, _MAX_GLOBAL_WRITERS)
self._reads[task_id][resolved] = (float(mtime), now, False)
_cap_dict(self._reads[task_id], _MAX_PATHS_PER_AGENT)
_evict_oldest(self._last_writer, _MAX_GLOBAL_WRITERS)
self._stamp(task_id, resolved, mtime, now, False)
def check_stale(self, task_id: str, resolved: str) -> Optional[str]:
"""Model-facing warning if this write would be stale, else ``None``.
@@ -139,16 +139,14 @@ class FileStateRegistry:
f"{resolved} was modified by sibling subagent "
f"{writer_tid!r} but this agent never read it. "
"Read the file before writing to avoid overwriting "
"the sibling's changes."
)
"the sibling's changes.")
read_ts = stamp[1]
if writer_ts > read_ts:
return (
f"{resolved} was modified by sibling subagent "
f"{writer_tid!r} at {_fmt_ts(writer_ts)} — after "
f"this agent's last read at {_fmt_ts(read_ts)}. "
"Re-read the file before writing."
)
"Re-read the file before writing.")
if stamp is not None:
read_mtime, _read_ts, partial = stamp
@@ -156,27 +154,20 @@ class FileStateRegistry:
return (
f"{resolved} was modified since you last read it "
"on disk (external edit or unrecorded writer). "
"Re-read the file before writing."
)
"Re-read the file before writing.")
if partial:
return (
f"{resolved} was last read with offset/limit pagination "
"(partial view). Re-read the whole file before "
"overwriting it."
)
"overwriting it.")
return None
return (
f"{resolved} was not read by this agent. "
"Read the file first so you can write an informed edit."
)
"Read the file first so you can write an informed edit.")
def writes_since(
self,
exclude_task_id: str,
since_ts: float,
paths: Iterable[str],
) -> Dict[str, List[str]]:
def writes_since(self, exclude_task_id: str, since_ts: float,
paths: Iterable[str]) -> Dict[str, List[str]]:
"""``{writer_task_id: [paths]}`` for writes after ``since_ts`` by agents
other than ``exclude_task_id`` (delegate_task's "subagent modified files
you previously read" reminder)."""
@@ -213,36 +204,6 @@ def get_registry() -> FileStateRegistry:
return _registry
def _disabled() -> bool:
# Re-read each call so tests can toggle via monkeypatch.setenv.
return os.environ.get("HERMES_DISABLE_FILE_STATE_GUARD", "").strip() == "1"
def _mtime_or_none(resolved: str) -> Optional[float]:
try:
return os.path.getmtime(resolved)
except OSError:
return None
def _fmt_ts(ts: float) -> str:
# Short wall-clock for warnings; avoids datetime formatting on the hot path.
return time.strftime("%H:%M:%S", time.localtime(ts))
def _cap_dict(d: dict, limit: int) -> None:
"""Trim ``d`` to ``limit`` entries by dropping the insertion-order oldest."""
over = len(d) - limit
if over <= 0:
return
it = iter(d)
for _ in range(over):
try:
d.pop(next(it))
except (StopIteration, KeyError):
break
# Convenience wrappers (short names used at call sites).
def record_read(task_id: str, resolved_or_path: str | Path, *, partial: bool = False) -> None:
_registry.record_read(task_id, str(resolved_or_path), partial=partial)
@@ -260,11 +221,7 @@ def lock_path(resolved_or_path: str | Path):
return _registry.lock_path(str(resolved_or_path))
def writes_since(
exclude_task_id: str,
since_ts: float,
paths: Iterable[str | Path],
) -> Dict[str, List[str]]:
def writes_since(exclude_task_id: str, since_ts: float, paths: Iterable[str | Path]) -> Dict[str, List[str]]:
return _registry.writes_since(exclude_task_id, since_ts, [str(p) for p in paths])
@@ -280,5 +237,4 @@ __all__ = [
"check_stale",
"lock_path",
"writes_since",
"known_reads",
]
"known_reads"]
tools/file_tools.py
+118 -251
View File
@@ -1,12 +1,10 @@
#!/usr/bin/env python3
"""File Tools Module - LLM agent file manipulation tools.
Companion modules (every name is re-imported here so ``tools.file_tools.X``
keeps working for callers and test patches):
* ``file_tools_paths`` — task-aware path resolution / ``~`` expansion.
* ``file_tools_write_guards`` — sensitive-path, protected-instruction,
approval, mirror and binary-document guards.
* ``file_tools_read_tracking`` — per-task dedup / loop-detection / staleness state.
Companions (every name re-imported here so ``tools.file_tools.X`` keeps working
for callers and test patches): ``file_tools_paths`` (task-aware resolution),
``file_tools_write_guards`` (write-side guards), ``file_tools_read_tracking``
(per-task dedup / loop-detection / staleness state).
"""
import base64
@@ -24,72 +22,30 @@ from pathlib import Path
from agent.file_safety import get_read_block_error
from tools.binary_extensions import has_binary_extension
from tools.file_operations import (
ShellFileOperations,
normalize_read_pagination,
normalize_search_pagination,
)
ShellFileOperations, normalize_read_pagination, normalize_search_pagination)
from tools import file_state
from agent.redact import redact_sensitive_text
from tools.file_tools_paths import ( # noqa: F401 (re-exported)
_CONTAINER_PATH_BACKENDS_FALLBACK,
_TERMINAL_CWD_SENTINELS,
_authoritative_workspace_root,
_configured_terminal_cwd,
_expand_tilde,
_normalize_without_host_deref,
_path_resolution_warning,
_registered_task_cwd_override,
_resolve_base_dir,
_resolve_path,
_resolve_path_for_task,
_sentinel_free_abs_cwd,
_terminal_env_type_for_task,
_uses_container_paths,
)
_CONTAINER_PATH_BACKENDS_FALLBACK, _TERMINAL_CWD_SENTINELS, _authoritative_workspace_root,
_configured_terminal_cwd, _expand_tilde, _normalize_without_host_deref,
_path_resolution_warning, _registered_task_cwd_override, _resolve_base_dir, _resolve_path,
_resolve_path_for_task, _sentinel_free_abs_cwd, _terminal_env_type_for_task,
_uses_container_paths)
from tools.file_tools_write_guards import ( # noqa: F401 (re-exported)
_PROTECTED_INSTRUCTION_BASENAMES,
_READ_DEDUP_STATUS_MESSAGE,
_SENSITIVE_EXACT_PATHS,
_SENSITIVE_PATH_PREFIXES,
_check_approval_required_write,
_check_binary_document_write,
_check_cross_profile_path,
_check_protected_instruction_write,
_check_sensitive_path,
_get_container_mirror_prefix_for_task,
_get_hermes_config_resolved,
_get_real_hermes_home,
_is_internal_file_status_text,
_is_internal_file_tool_content,
_looks_like_read_file_line_numbered_content,
_protected_instruction_config,
_protected_instruction_reason,
_request_protected_instruction_approval,
)
_PROTECTED_INSTRUCTION_BASENAMES, _READ_DEDUP_STATUS_MESSAGE, _SENSITIVE_EXACT_PATHS,
_SENSITIVE_PATH_PREFIXES, _check_approval_required_write, _check_binary_document_write,
_check_cross_profile_path, _check_protected_instruction_write, _check_sensitive_path,
_get_container_mirror_prefix_for_task, _get_hermes_config_resolved, _get_real_hermes_home,
_is_internal_file_status_text, _is_internal_file_tool_content,
_looks_like_read_file_line_numbered_content, _protected_instruction_config,
_protected_instruction_reason, _request_protected_instruction_approval)
from tools.file_tools_read_tracking import ( # noqa: F401 (re-exported)
_DEDUP_CAP,
_NOT_FOUND_CAP,
_NOT_FOUND_TTL_SECONDS,
_READ_HISTORY_CAP,
_READ_TIMESTAMPS_CAP,
_bump_consecutive,
_cap_read_tracker_data,
_check_file_staleness,
_check_not_found_cache,
_invalidate_dedup_for_path,
_mark_verification_stale,
_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,
notify_other_tool_call,
reset_file_dedup,
)
_DEDUP_CAP, _NOT_FOUND_CAP, _NOT_FOUND_TTL_SECONDS, _READ_HISTORY_CAP, _READ_TIMESTAMPS_CAP,
_bump_consecutive, _cap_read_tracker_data, _check_file_staleness, _check_not_found_cache,
_invalidate_dedup_for_path, _mark_verification_stale, _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,
notify_other_tool_call, reset_file_dedup)
logger = logging.getLogger(__name__)
@@ -120,11 +76,8 @@ def _truncate_to_char_budget(content: str, max_chars: int) -> tuple[str, int, bo
"""Trim line-numbered ``read_file`` content to the last COMPLETE line within *max_chars*.
Returns ``(kept_text, lines_kept, truncated)`` so the caller can offer a
``next_offset`` instead of rejecting the read. Lines are already clamped to
``get_max_line_length()`` upstream; the overflow handled here is many lines
under the line-count limit (logs, wide CSV). If not even the first line
fits it is clamped mid-line (slicing never splits a code point) so the read
is never empty and the cursor advances.
``next_offset`` instead of rejecting the read. If not even the first line
fits it is clamped mid-line so the read is never empty and the cursor advances.
"""
if len(content) <= max_chars:
return content, (content.count("\n") + 1 if content else 0), False
@@ -155,22 +108,19 @@ def _apply_char_budget(result_dict: dict, content: str, offset: int, total_lines
result_dict["hint"] = (
f"Output truncated at the {max_chars:,}-char read budget after "
f"{lines_kept} line(s) (showing lines {offset}-{next_offset - 1} of "
f"{total_lines}). Use offset={next_offset} to continue."
)
f"{total_lines}). Use offset={next_offset} to continue.")
if len(trimmed.split("\n", 1)[0]) >= max_chars:
result_dict["hint"] += (
" Note: the first line alone exceeded the budget and was "
"clamped mid-line; its remainder is not retrievable via "
"offset."
)
"offset.")
return trimmed
# Above this size, a wide read (limit > 200) gets a hint toward targeted reads.
_LARGE_FILE_HINT_BYTES = 512_000
# Device/fd paths whose reads hang the process (infinite output or blocking on
# input). Checked by path only — no I/O.
# Device/fd paths whose reads hang the process. Checked by path only — no I/O.
_BLOCKED_DEVICE_PATHS = frozenset({
"/dev/zero", "/dev/random", "/dev/urandom", "/dev/full", # never reach EOF
"/dev/stdin", "/dev/tty", "/dev/console", # block on input
@@ -182,16 +132,12 @@ _BLOCKED_DEVICE_PATHS = frozenset({
_BLOCKED_PROC_SUFFIXES = (
"/fd/0", "/fd/1", "/fd/2", # stdio aliases
"/environ", "/cmdline", "/maps", "/smaps", "/smaps_rollup", "/numa_maps",
"/mem", "/auxv", "/pagemap",
)
"/mem", "/auxv", "/pagemap")
def _file_ops_uses_host_paths(file_ops) -> bool:
"""True when *file_ops* targets the same host filesystem as Hermes.
Only then may we stat paths or rewrite V4A headers to host-absolute paths;
a container/remote backend has its own filesystem namespace.
"""
"""True when *file_ops* targets the host filesystem (only then may we stat paths
or rewrite V4A headers to host-absolute paths; sandboxes have their own namespace)."""
env = getattr(file_ops, "env", None)
if env is None:
return True
@@ -210,12 +156,11 @@ _V4A_MOVE_HEADER_RE = re.compile(r'^(\*\*\*\s*Move\s+File:\s*)(.+?)\s*->\s*(.+)$
def _rewrite_v4a_patch_paths_for_host(patch: str, path_to_resolved: dict, file_ops) -> str:
"""Rewrite V4A ``*** Update/Add/Delete/Move File:`` headers to the resolved host paths.
"""Rewrite V4A file headers to the resolved host paths (host backends only).
``patch_tool`` resolves every header against the task's workspace for
locking/staleness/reporting; the shell layer must patch those SAME files
rather than re-resolving a relative header against its own cwd (which can
differ — the git-worktree cwd bug). Only applied for host-filesystem backends.
The shell layer must patch the SAME files ``patch_tool`` resolved for
locking/staleness, not re-resolve a relative header against its own cwd
(which can differ — the git-worktree cwd bug).
"""
if not _file_ops_uses_host_paths(file_ops):
return patch
@@ -239,9 +184,8 @@ def _is_blocked_device_path(path: str) -> bool:
def _is_blocked_device(filepath: str, base_dir: str | Path | None = None) -> bool:
"""True if the path (literal, any symlink hop, or final realpath) is a blocked device.
The literal path is checked first so aliases like /dev/stdin are caught
before they resolve to terminal-specific paths; each symlink hop is checked
so an alias to a device cannot bypass the guard.
Literal first so /dev/stdin is caught before resolving to a terminal path;
every symlink hop is checked so an alias cannot bypass the guard.
"""
expanded = _expand_tilde(filepath)
if base_dir is not None and not os.path.isabs(expanded):
@@ -308,30 +252,17 @@ def _is_expected_write_exception(exc: Exception) -> bool:
isinstance(exc, OSError) and exc.errno in _EXPECTED_WRITE_ERRNOS)
# ---------------------------------------------------------------------------
# ShellFileOperations per terminal environment
# ---------------------------------------------------------------------------
# ── ShellFileOperations per terminal environment ─────────────────────────
_file_ops_lock = threading.Lock()
_file_ops_cache: dict = {}
def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
"""Build the terminal environment for *task_id* from config + per-task overrides.
Mirrors terminal_tool's own environment construction (shared
``_create_configured_env``) so a file tool that runs before any terminal
command uses the configured backend (docker, modal, ...) rather than local.
"""
"""Build the terminal environment for *task_id* via the shared ``_create_configured_env``,
so a file tool that runs before any terminal command still gets the configured backend."""
from tools.terminal_tool import (
_CONTAINER_BACKENDS,
_create_configured_env,
_get_env_config,
_is_unusable_container_cwd,
_resolve_task_host_cwd,
_select_image,
get_session_cwd,
resolve_task_overrides,
)
_CONTAINER_BACKENDS, _create_configured_env, _get_env_config, _is_unusable_container_cwd,
_resolve_task_host_cwd, _select_image, get_session_cwd, resolve_task_overrides)
config = _get_env_config()
env_type = config["env_type"]
@@ -341,18 +272,15 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
except Exception:
recorded_cwd = None
cwd = overrides.get("cwd") or recorded_cwd or config["cwd"]
# Re-apply the container cwd guard _get_env_config() already ran on
# config["cwd"]: a gateway/TUI/ACP cwd override is a raw HOST path, and
# ``docker run -w <host-path>`` starts the container in a directory that
# doesn't exist there, so search_files & co silently return nothing.
# Valid in-container overrides (/workspace, /root, ...) pass untouched.
# Re-apply the container cwd guard: a gateway/TUI/ACP override is a raw HOST
# path and ``docker run -w <host-path>`` makes search_files & co silently
# return nothing. Valid in-container overrides (/workspace, /root) pass.
if env_type in _CONTAINER_BACKENDS and _is_unusable_container_cwd(cwd):
if cwd != config["cwd"]:
logger.info(
"Ignoring host/relative cwd override %r for %s backend "
"(won't exist in sandbox). Using %r instead.",
cwd, env_type, config["cwd"],
)
cwd, env_type, config["cwd"])
cwd = config["cwd"]
logger.info("Creating new %s environment for task %s...", env_type, task_id[:8])
terminal_env = _create_configured_env(
@@ -367,23 +295,20 @@ def _create_terminal_env_for_file_ops(raw_task_id: str, task_id: str):
def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
"""Get or create ShellFileOperations for the task's terminal environment.
Thread-safe via terminal_tool's per-task creation locks, so concurrent tool
calls never build duplicate sandboxes. Subagent task_ids collapse to
"default" (``_resolve_container_task_id``) so delegate_task children share
the parent's container and cached file_ops; RL/benchmark task_ids with a
registered env override keep their isolation.
Uses terminal_tool's per-task creation locks (no duplicate sandboxes).
Subagent task_ids collapse to "default" (``_resolve_container_task_id``) so
delegate_task children share the parent's container; RL/benchmark task_ids
with a registered env override keep their isolation.
"""
from tools.terminal_tool import (
_active_environments, _env_lock, _last_activity, _start_cleanup_thread,
_creation_locks, _creation_locks_lock, _resolve_container_task_id,
get_session_cwd, record_session_cwd,
)
get_session_cwd, record_session_cwd)
raw_task_id = task_id or "default"
task_id = _resolve_container_task_id(raw_task_id)
# Fast path: cached AND the environment is still alive (the cleanup thread
# may have killed it).
# Fast path: cached AND the environment is still alive (cleanup thread may have killed it).
with _file_ops_lock:
cached = _file_ops_cache.get(task_id)
if cached is not None:
@@ -391,10 +316,8 @@ def _get_file_ops(task_id: str = "default") -> ShellFileOperations:
if task_id in _active_environments:
_last_activity[task_id] = time.time()
return cached
# Environment was cleaned up: rescue the old cwd into the session
# record before dropping the stale entry, FILL-ONLY — ``cached.cwd``
# is a snapshot of the SHARED env's cwd, not attributable to this
# session, so never overwrite a record the session wrote for itself.
# Env was cleaned up: rescue its cwd into the session record FILL-ONLY
# (``cached.cwd`` is the SHARED env's cwd, not this session's own).
old_cwd = getattr(cached, "cwd", None)
if old_cwd:
try:
@@ -441,18 +364,15 @@ _SPECIAL_FILE_KINDS = (
(stat.S_ISFIFO, "a FIFO (named pipe)"),
(stat.S_ISSOCK, "a socket"),
(stat.S_ISCHR, "a character device"),
(stat.S_ISBLK, "a block device"),
)
(stat.S_ISBLK, "a block device"))
def _special_file_kind(path) -> str | None:
"""Human name for a non-regular file type that would hang a read, else None.
Stat-based sibling of the name-based ``_is_blocked_device`` guard: a FIFO
or socket inside a workspace hangs ``read_file`` as hard as ``/dev/zero``
but carries no recognizable name. Host filesystems only (see
``_file_ops_uses_host_paths``). Missing/unstattable paths return None and
flow to the normal read path's own error handling.
Stat-based sibling of ``_is_blocked_device``: a FIFO/socket in a workspace
hangs like ``/dev/zero`` but has no recognizable name. Host filesystems
only; unstattable paths return None and flow to the normal read path.
"""
try:
mode = os.stat(os.fspath(path)).st_mode # follows symlinks, matching a real read
@@ -467,19 +387,13 @@ def _special_file_kind(path) -> str | None:
def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task_id: str) -> str | None:
"""Render an extractable document (.docx/.xlsx/.pdf/...) as paginated text.
Returns the JSON result, a tool_error for a binary document whose
extraction failed for an actionable reason (size cap, encrypted,
malformed), or ``None`` to fall through to the normal read path. Runs
BEFORE the binary-extension guard so Office files can render as text.
Returns the JSON result, a tool_error for an actionable extraction failure
(size cap, encrypted, malformed), or ``None`` to fall through to the normal
read path. Runs BEFORE the binary-extension guard.
"""
from tools.read_extract import (
ANYDOC_EXTENSIONS,
EXTRACTABLE_EXTENSIONS,
MAX_DOCUMENT_BYTES,
ExtractionError,
extract_document_bytes,
is_extractable_document,
)
ANYDOC_EXTENSIONS, EXTRACTABLE_EXTENSIONS, MAX_DOCUMENT_BYTES, ExtractionError,
extract_document_bytes, is_extractable_document)
if not is_extractable_document(str(_resolved)):
return None
@@ -492,10 +406,8 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
extracted_text = extract_document_bytes(document_bytes, str(_resolved))
except (ExtractionError, ValueError, base64.binascii.Error) as exc:
logger.debug("document extraction failed for %s", path, exc_info=True)
# Binary document formats surface the specific failure: the fallthrough
# can only yield a generic binary-file error or garbage bytes. .ipynb
# (plain JSON) and byte-transport errors (ValueError/binascii) fall
# through — only a specific ExtractionError carries an actionable reason.
# Binary formats surface the specific failure (fallthrough would only
# give a generic binary error); .ipynb and byte-transport errors fall through.
_doc_ext = _resolved.suffix.lower()
_binary_doc = _doc_ext in ANYDOC_EXTENSIONS or (
_doc_ext in EXTRACTABLE_EXTENSIONS and _doc_ext != ".ipynb")
@@ -504,8 +416,7 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
return tool_error(
f"Cannot read '{path}' ({_doc_ext}): document "
f"extraction failed — {exc}. Use terminal utilities "
"to inspect or convert the file."
)
"to inspect or convert the file.")
return None
lines = extracted_text.splitlines()
@@ -517,13 +428,11 @@ def _read_extracted_document(path: str, _resolved, offset: int, limit: int, task
"total_lines": total_lines,
"file_size": binary.file_size,
"truncated": total_lines > end_line,
"extracted_document": True,
}
"extracted_document": True}
if result_dict["truncated"]:
result_dict["hint"] = (
f"Use offset={end_line + 1} to continue reading "
f"(showing {offset}-{min(end_line, total_lines)} of {total_lines} lines)"
)
f"(showing {offset}-{min(end_line, total_lines)} of {total_lines} lines)")
max_chars = _get_max_read_chars()
if len(result_dict["content"]) > max_chars:
_apply_char_budget(result_dict, result_dict["content"], offset, total_lines, max_chars)
@@ -550,8 +459,7 @@ def _dedup_stub_or_block(task_data: dict, dedup_key: tuple, path: str) -> str:
"still current. Proceed with your task using "
"the information you already have.",
path=path,
already_read=hits + 1,
)
already_read=hits + 1)
return json.dumps({
"status": "unchanged",
@@ -566,13 +474,10 @@ def _record_successful_read(task_data: dict, task_id: str, path: str, resolved_s
offset: int, limit: int, dedup_key: tuple, *, partial: bool) -> int:
"""Bookkeeping after a real (non-stub) read; returns the consecutive-read count.
Per-task tracker (under the lock): clear this key's stub-loop counter, add to
history, bump the consecutive counter, and store the mtime (feeds dedup and
the write/patch staleness warning). Then, OUTSIDE our lock so the registry's
own locking isn't nested under it: the cross-agent registry (lets write/patch
detect sibling-subagent writes) and the background-review read-mark (a FULL
read of a skill file counts like skill_view so a follow-up
skill_manage(patch) is accepted; no-op outside review forks).
Per-task tracker under the lock (stub counter, history, consecutive count,
mtime for dedup + staleness). Then OUTSIDE our lock (no nested locking): the
cross-agent registry, and the background-review read-mark (a FULL read of a
skill file counts like skill_view so a follow-up skill_manage(patch) is accepted).
"""
with _read_tracker_lock:
task_data["dedup_hits"].pop(dedup_key, None)
@@ -614,8 +519,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
if _is_blocked_device(path, base_dir=device_base):
return tool_error(
f"Cannot read '{path}': this is a device file that would "
"block or produce infinite output."
)
"block or produce infinite output.")
_resolved = _resolve_path_for_task(path, task_id)
@@ -629,9 +533,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
f"'{path}' is {kind}, not a regular file — reading "
"it would block indefinitely, so no read was "
"attempted. Use terminal utilities if you need to "
"interact with it."
),
})
"interact with it.")})
extracted = _read_extracted_document(path, _resolved, offset, limit, task_id)
if extracted is not None:
@@ -642,14 +544,11 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
if has_binary_extension(str(_resolved)):
return tool_error(
f"Cannot read binary file '{path}' ({_resolved.suffix.lower()}). "
"Use vision_analyze for images, or terminal to inspect binary files."
)
"Use vision_analyze for images, or terminal to inspect binary files.")
# Hermes internal denylist: blocks prompt injection via catalog/hub
# metadata and credential stores under HERMES_HOME. Pass the resolved
# path: get_read_block_error's own resolve() runs against the process
# cwd, so a relative "auth.json" read with TERMINAL_CWD == HERMES_HOME
# would otherwise miss the denylist.
# Hermes internal denylist (prompt injection via catalog metadata,
# credential stores). Pass the RESOLVED path: the denylist's own
# resolve() uses the process cwd and would miss a relative "auth.json".
block_error = get_read_block_error(str(_resolved))
if block_error:
return tool_error(block_error)
@@ -676,24 +575,20 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
result_dict = result.to_dict()
# Cache a not-found result for retries. Deliberately NO early return:
# error results still flow through the tracking block below
# (consecutive-loop detection, dedup bookkeeping) — serving from the
# cache is the optimization, recording must stay side-effect-identical.
# error results still flow through the tracking below unchanged.
_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))
# Char budget is checked on the FORMATTED content (that is what enters
# context) and BEFORE redaction, to skip the regex pass on huge content.
# Graceful truncation instead of rejection: the model gets what fits plus
# a next_offset, rather than guessing a smaller limit and burning a turn.
# Char budget on the FORMATTED content (what enters context), BEFORE
# redaction (skip the regex pass on huge content); truncate gracefully
# with a next_offset instead of rejecting.
file_size = result_dict.get("file_size", 0)
max_chars = _get_max_read_chars()
if len(result.content or "") > max_chars:
result.content = _apply_char_budget(
result_dict, result.content or "", offset,
result_dict.get("total_lines", "unknown"), max_chars,
)
result_dict.get("total_lines", "unknown"), max_chars)
if result.content:
result.content = redact_sensitive_text(result.content, file_read=True)
result_dict["content"] = result.content
@@ -703,8 +598,7 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
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."
))
"to keep context usage efficient."))
count = _record_successful_read(task_data, task_id, path, resolved_str, offset, limit,
dedup_key, partial=(offset > 1) or bool(result_dict.get("truncated")))
@@ -714,22 +608,18 @@ def read_file_tool(path: str, offset: int = 1, limit: int = 2000, task_id: str =
"The content has NOT changed. You already have this information. "
"STOP re-reading and proceed with your task.",
path=path,
already_read=count,
)
already_read=count)
if count >= 3:
result_dict["_warning"] = (
f"You have read this exact file region {count} times consecutively. "
"The content has not changed since your last read. Use the information you already have. "
"If you are stuck in a loop, stop reading and proceed with writing or responding."
)
"If you are stuck in a loop, stop reading and proceed with writing or responding.")
return json.dumps(result_dict, ensure_ascii=False)
except Exception as e:
return tool_error(str(e))
# ---------------------------------------------------------------------------
# Shared write/patch plumbing
# ---------------------------------------------------------------------------
# ── Shared write/patch plumbing ──────────────────────────────────────────
def _resolve_or_none(filepath: str, task_id: str) -> str | None:
"""Task-resolved path string, or None when resolution fails for any reason."""
@@ -739,15 +629,6 @@ def _resolve_or_none(filepath: str, task_id: str) -> str | None:
return None
def _first_error(*checks) -> str | None:
"""Run zero-arg guard thunks in order; return the first non-empty error string.
Lazy on purpose: later guards may prompt the user, and a hard deny earlier
in the chain must short-circuit them.
"""
return next(filter(None, (check() for check in checks)), None)
def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: str,
cross_profile: bool) -> str | None:
"""Run the shared write/patch guards in order; return the first error string.
@@ -761,11 +642,12 @@ def _write_precheck_error(paths: list[str], content_paths: list[str], task_id: s
None if cross_profile else _check_cross_profile_path(p, task_id))
if err:
return err
return _first_error(
*(lambda p=p: _check_binary_document_write(p, task_id) for p in content_paths),
lambda: _check_protected_instruction_write(paths, task_id),
lambda: _check_approval_required_write(paths, task_id),
)
for p in content_paths:
err = _check_binary_document_write(p, task_id)
if err:
return err
return (_check_protected_instruction_write(paths, task_id)
or _check_approval_required_write(paths, task_id))
def _edit_warnings(paths: list[str], path_to_resolved: dict, task_id: str) -> list[str]:
@@ -803,18 +685,15 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
cross-PROFILE guard it was named for no longer exists).
"""
# write_file checks the binary-document guard before the mirror guard.
err = _first_error(
lambda: _check_sensitive_path(path, task_id),
lambda: _check_binary_document_write(path, task_id),
lambda: _check_protected_instruction_write([path], task_id),
lambda: _check_approval_required_write([path], task_id),
lambda: None if cross_profile else _check_cross_profile_path(path, task_id),
lambda: (
"Refusing to write internal read_file display text as file content. "
"Strip read_file line-number prefixes or reconstruct the intended "
"file contents before writing."
) if _is_internal_file_tool_content(content) else None,
)
err = (_check_sensitive_path(path, task_id)
or _check_binary_document_write(path, task_id)
or _check_protected_instruction_write([path], task_id)
or _check_approval_required_write([path], task_id)
or (None if cross_profile else _check_cross_profile_path(path, task_id)))
if not err and _is_internal_file_tool_content(content):
err = ("Refusing to write internal read_file display text as file content. "
"Strip read_file line-number prefixes or reconstruct the intended "
"file contents before writing.")
if err:
return tool_error(err)
try:
@@ -853,12 +732,10 @@ def write_file_tool(path: str, content: str, task_id: str = "default",
def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
"""Extract every path named in V4A headers, rejecting ``..`` traversal.
Returns ``(all_paths, content_write_paths)`` or a tool_error string.
Header paths come from patch CONTENT (skill text, web extract, prompt
injection) so they are more attacker-influenceable than the explicit
``path=`` arg, which keeps its legitimate ``..`` use. Move headers check
BOTH endpoints (a Move onto /etc/crontab must hit the sensitive-path check).
Delete/Move don't write text, so only Update/Add feed the binary-document guard.
Returns ``(all_paths, content_write_paths)`` or a tool_error string. Header
paths come from patch CONTENT (more attacker-influenceable than ``path=``,
which keeps its legitimate ``..`` use). Move headers check BOTH endpoints;
only Update/Add write text and feed the binary-document guard.
"""
from tools.path_security import has_traversal_component
@@ -873,8 +750,7 @@ def _collect_v4a_header_paths(patch: str) -> tuple[list[str], list[str]] | str:
f"V4A patch header contains '..' traversal: {v4a_path!r}. "
"Use the agent's cwd-relative path (no '..') or an absolute "
"path in '*** Update File:' / '*** Add File:' / "
"'*** Delete File:' / '*** Move File:' headers."
)
"'*** Delete File:' / '*** Move File:' headers.")
paths.append(v4a_path)
if writes_text:
content_paths.append(v4a_path)
@@ -941,11 +817,9 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
_note_edited(task_id, _paths_to_check, _path_to_resolved, session_id)
# Clear failure counters so a future miss starts a fresh count.
_reset_patch_failures(task_id, [_r for _r in _path_to_resolved.values() if _r])
# old_string-not-found hint. Per-file failure escalation is tracked for
# replace mode only (V4A failures are rare; the generic hint suffices).
# The generic hint is suppressed when patch_replace already attached a
# richer "Did you mean?" snippet. The escalating hint after 3 failures
# exists so the model recognises the loop and changes approach.
# old_string-not-found hint. Failure escalation is tracked for replace
# mode only (V4A misses are rare); the generic hint is suppressed when
# patch_replace already attached a richer "Did you mean?" snippet.
if result_dict.get("error") and "Could not find" in str(result_dict["error"]):
failure_count = 0
if mode == "replace" and path:
@@ -958,13 +832,11 @@ def patch_tool(mode: str = "replace", path: str = None, old_string: str = None,
"content, (2) use a longer / more unique old_string with "
"surrounding context lines, or (3) use write_file to "
"replace the entire file if the targeted region is hard "
"to anchor."
)
"to anchor.")
elif "Did you mean one of these sections?" not in str(result_dict["error"]):
result_dict["_hint"] = (
"old_string not found. Use read_file to verify the current "
"content, or search_files to locate the text."
)
"content, or search_files to locate the text.")
return json.dumps(result_dict, ensure_ascii=False)
except Exception as e:
return tool_error(str(e))
@@ -983,8 +855,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
search_key = ("search", pattern, target, str(path), file_glob or "", limit, offset)
with _read_tracker_lock:
task_data = _read_tracker.setdefault(task_id, {
"last_key": None, "consecutive": 0, "read_history": set(),
})
"last_key": None, "consecutive": 0, "read_history": set()})
count = _bump_consecutive(task_data, search_key)
if count >= 4:
@@ -993,8 +864,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
"The results have NOT changed. You already have this information. "
"STOP re-searching and proceed with your task.",
pattern=pattern,
already_searched=count,
)
already_searched=count)
try:
resolved_search_path = str(_resolve_path_for_task(path, task_id))
@@ -1016,8 +886,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
result = _get_file_ops(task_id).search(
pattern=pattern, path=path, target=target, file_glob=file_glob,
limit=limit, offset=offset, output_mode=output_mode, context=context
)
limit=limit, offset=offset, output_mode=output_mode, context=context)
omitted = _filter_read_blocked_search_results(result, task_id)
for m in getattr(result, "matches", None) or ():
if getattr(m, "content", None):
@@ -1027,8 +896,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
if omitted:
result_dict["_omitted"] = (
f"{omitted} result(s) omitted because they target credential, "
"token, cache, or secret-bearing environment files."
)
"token, cache, or secret-bearing environment files.")
# No early return on a cached miss — same rationale as the read path.
_search_err = result_dict.get("error") or ""
@@ -1038,8 +906,7 @@ def search_tool(pattern: str, target: str = "content", path: str = ".",
if count >= 3:
result_dict["_warning"] = (
f"You have run this exact search {count} times consecutively. "
"The results have not changed. Use the information you already have."
)
"The results have not changed. Use the information you already have.")
result_json = json.dumps(result_dict, ensure_ascii=False)
if result_dict.get("truncated"):
tools/file_tools_paths.py
+15 -35
View File
@@ -47,11 +47,7 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
"""Best-effort terminal backend type for path-resolution decisions."""
try:
from tools.terminal_tool import (
_active_environments,
_env_lock,
_get_env_config,
_resolve_container_task_id,
)
_active_environments, _env_lock, _get_env_config, _resolve_container_task_id)
try:
container_key = _resolve_container_task_id(task_id)
@@ -61,14 +57,11 @@ def _terminal_env_type_for_task(task_id: str = "default") -> str:
env = _active_environments.get(container_key) or _active_environments.get(task_id)
if env is not None:
name = env.__class__.__name__.lower()
for hint in _ENV_CLASS_NAME_HINTS:
if hint in name:
return hint
hint = next((h for h in _ENV_CLASS_NAME_HINTS if h in name), None)
stamped = getattr(env, "_hermes_backend_name", None)
if isinstance(stamped, str) and stamped:
return stamped
cfg = _get_env_config()
return str(cfg.get("env_type") or os.getenv("TERMINAL_ENV") or "local").lower()
if hint or (isinstance(stamped, str) and stamped):
return hint or stamped
return str(_get_env_config().get("env_type") or os.getenv("TERMINAL_ENV") or "local").lower()
except Exception:
return str(os.getenv("TERMINAL_ENV") or "local").lower()
@@ -103,9 +96,7 @@ def _sentinel_free_abs_cwd(raw: str | None) -> str | None:
if raw.lower() in _TERMINAL_CWD_SENTINELS:
return None
expanded = _expand_tilde(raw)
if not os.path.isabs(expanded):
return None
return expanded
return expanded if os.path.isabs(expanded) else None
def _configured_terminal_cwd() -> str | None:
@@ -151,19 +142,11 @@ def _authoritative_workspace_root(task_id: str = "default") -> str | None:
recorded = get_session_cwd(task_id)
except Exception:
recorded = None
if recorded:
return recorded
registered = _registered_task_cwd_override(task_id)
if registered:
return registered
return _configured_terminal_cwd()
return recorded or _registered_task_cwd_override(task_id) or _configured_terminal_cwd()
def _resolve_base_dir(
task_id: str = "default",
*,
container_paths: bool | None = None,
) -> Path | PurePosixPath:
task_id: str = "default", *, container_paths: bool | None = None) -> Path | PurePosixPath:
"""Return the ABSOLUTE base directory for resolving relative paths.
Uses ``_authoritative_workspace_root`` (live cwd → registered override →
@@ -255,16 +238,13 @@ def _path_resolution_warning(filepath: str, resolved: Path, task_id: str = "defa
root = _normalize_without_host_deref(Path(_expand_tilde(workspace_root)))
else:
root = Path(_expand_tilde(workspace_root)).resolve()
try:
resolved.relative_to(root)
if resolved.is_relative_to(root):
return None
except ValueError:
return (
f"Relative path {filepath!r} resolved to {str(resolved)!r}, which is "
f"OUTSIDE the active workspace ({str(root)!r}). The edit will land in "
f"a different directory than the terminal's cwd. If this is not "
f"intended (e.g. a git-worktree session writing into the main "
f"checkout), pass an absolute path under the workspace instead."
)
return (
f"Relative path {filepath!r} resolved to {str(resolved)!r}, which is "
f"OUTSIDE the active workspace ({str(root)!r}). The edit will land in "
f"a different directory than the terminal's cwd. If this is not "
f"intended (e.g. a git-worktree session writing into the main "
f"checkout), pass an absolute path under the workspace instead.")
except Exception:
return None
tools/file_tools_read_tracking.py
+44 -110
View File
@@ -1,23 +1,12 @@
"""Per-task read/search bookkeeping for the file tools.
Process-lifetime state behind ``read_file`` / ``search_files`` / ``write_file`` /
``patch``; ``tools.file_tools`` re-imports every name here. Per task_id
``_read_tracker`` stores:
last_key / consecutive most recent read-or-search key and its repeat count
(loop detection; reset by any OTHER tool call).
read_history set of (path, offset, limit) — diagnostic summaries only.
dedup (resolved_path, offset, limit) -> mtime; skip identical
re-reads of unchanged files. Cleared on context
compression (the original content was summarised away).
dedup_hits per-key count of stub returns, to break stub loops.
read_timestamps resolved_path -> mtime at last read/write by this task;
write/patch warn when the file changed underneath.
not_found (op, resolved_path) -> (monotonic, cached error JSON);
short-TTL negative cache for retried missing paths.
Every container is hard-capped (``_cap_read_tracker_data``) so a long CLI
session accretes a few hundred KB at most instead of ~1.5MB per 10k reads.
Process-lifetime state behind read_file/search_files/write_file/patch;
``tools.file_tools`` re-imports every name here. Per task_id ``_read_tracker``
stores: ``last_key``/``consecutive`` (loop detection; reset by any OTHER tool
call), ``read_history`` (diagnostics), ``dedup`` (key -> mtime; cleared on
context compression), ``dedup_hits`` (stub-loop breaker), ``read_timestamps``
(staleness warnings) and ``not_found`` (short-TTL negative cache). Every
container is hard-capped (``_cap_read_tracker_data``) so long sessions stay small.
"""
import logging
@@ -25,6 +14,7 @@ import os
import threading
import time
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")
@@ -33,8 +23,7 @@ _read_tracker_lock = threading.Lock()
_read_tracker: dict = {}
# Consecutive patch failures per (task_id, resolved_path); escalates the hint
# when the model keeps failing the same file (stale view, ambiguous old_string).
# Reset on a successful patch to that path.
# when the model keeps failing the same file. Reset on a successful patch.
_patch_failure_lock = threading.Lock()
_patch_failure_tracker: dict = {} # {task_id: {resolved_path: count}}
_PATCH_FAILURE_PATHS_CAP = 64
@@ -48,24 +37,16 @@ _NOT_FOUND_CAP = 500
_NOT_FOUND_TTL_SECONDS = 60.0 # a path that didn't exist may be created soon
def _new_task_data() -> dict:
return {
"last_key": None, "consecutive": 0,
"read_history": set(), "dedup": {},
"dedup_hits": {}, "read_timestamps": {},
}
def _task_data(task_id: str) -> dict:
"""Get-or-create the tracker entry for *task_id*, back-filling any missing keys.
Must be called with ``_read_tracker_lock`` held. Entries created by older
code paths (or injected by tests) may lack the newer containers.
"""
task_data = _read_tracker.setdefault(task_id, _new_task_data())
for key, factory in (("dedup", dict), ("dedup_hits", dict), ("read_timestamps", dict)):
if key not in task_data:
task_data[key] = factory()
task_data = _read_tracker.setdefault(task_id, {
"last_key": None, "consecutive": 0, "read_history": set()})
for key in ("dedup", "dedup_hits", "read_timestamps"):
task_data.setdefault(key, {})
return task_data
@@ -74,11 +55,8 @@ def _record_patch_failure(task_id: str, resolved_path: str) -> int:
with _patch_failure_lock:
task_failures = _patch_failure_tracker.setdefault(task_id, {})
# Evict the oldest entry once a task has failed on many distinct files.
if len(task_failures) >= _PATCH_FAILURE_PATHS_CAP and resolved_path not in task_failures:
try:
del task_failures[next(iter(task_failures))]
except StopIteration:
pass
if resolved_path not in task_failures:
_evict_oldest(task_failures, _PATCH_FAILURE_PATHS_CAP - 1)
task_failures[resolved_path] = task_failures.get(resolved_path, 0) + 1
return task_failures[resolved_path]
@@ -89,29 +67,10 @@ def _reset_patch_failures(task_id: str, resolved_paths: list) -> None:
return
with _patch_failure_lock:
task_failures = _patch_failure_tracker.get(task_id)
if not task_failures:
return
for rp in resolved_paths:
for rp in resolved_paths if task_failures else ():
task_failures.pop(rp, None)
def _evict_oldest(container, cap: int) -> None:
"""Pop entries until *container* is within *cap*.
Sets pop arbitrary entries (they only feed diagnostic summaries); dicts pop
oldest by insertion order. An evicted entry costs one redundant re-send
(dedup) or one non-mtime staleness check — graceful degradation, not a bug.
"""
for _ in range(len(container) - cap):
try:
if isinstance(container, set):
container.pop()
else:
container.pop(next(iter(container)))
except (StopIteration, KeyError):
break
def _cap_read_tracker_data(task_data: dict) -> None:
"""Enforce size caps on the per-task sub-containers. Call with ``_read_tracker_lock`` held."""
# Caps are read at call time so tests can monkeypatch the module constants.
@@ -120,13 +79,20 @@ def _cap_read_tracker_data(task_data: dict) -> None:
("dedup", _DEDUP_CAP),
("dedup_hits", _DEDUP_CAP),
("read_timestamps", _READ_TIMESTAMPS_CAP),
("not_found", _NOT_FOUND_CAP),
):
("not_found", _NOT_FOUND_CAP)):
container = task_data.get(key)
if container is not None and len(container) > cap:
_evict_oldest(container, cap)
def _pop_not_found(op: str, resolved_str: str, task_id: str) -> None:
"""Drop the negative-cache entry for *(op, resolved_str)*. Lock must be held."""
task_data = _read_tracker.get(task_id)
nf = task_data.get("not_found") if task_data else None
if nf:
nf.pop((op, resolved_str), None)
def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | None:
"""Return cached not-found JSON for *(op, resolved_str)* if still fresh.
@@ -136,28 +102,20 @@ def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | No
"""
with _read_tracker_lock:
task_data = _read_tracker.get(task_id)
if not task_data:
return None
nf = task_data.get("not_found")
if not nf:
return None
entry = nf.get((op, resolved_str))
entry = (task_data.get("not_found") or {}).get((op, resolved_str)) if task_data else None
if entry is None:
return None
ts, cached_json = entry
if time.monotonic() - ts > _NOT_FOUND_TTL_SECONDS:
nf.pop((op, resolved_str), None)
_pop_not_found(op, resolved_str, task_id)
return None
# The path may have been created since the miss was cached (terminal,
# another agent, ...) — the "check → create → read" pattern is common, so
# serving a stale miss breaks it. The stat runs OUTSIDE the global tracker
# lock: a hung stat on a dead network mount must not stall every task.
# another agent, ...) — "check → create → read" is common, so serving a
# stale miss breaks it. The stat runs OUTSIDE the global tracker lock: a
# hung stat on a dead network mount must not stall every task.
if os.path.exists(resolved_str):
with _read_tracker_lock:
task_data = _read_tracker.get(task_id)
nf = task_data.get("not_found") if task_data else None
if nf:
nf.pop((op, resolved_str), None)
_pop_not_found(op, resolved_str, task_id)
return None
return cached_json
@@ -166,8 +124,7 @@ def _record_not_found(op: str, resolved_str: str, task_id: str, error_json: str)
"""Cache a not-found error so the next *op* call for *resolved_str* skips I/O."""
with _read_tracker_lock:
task_data = _task_data(task_id)
nf = task_data.setdefault("not_found", {})
nf[(op, resolved_str)] = (time.monotonic(), error_json)
task_data.setdefault("not_found", {})[(op, resolved_str)] = (time.monotonic(), error_json)
_cap_read_tracker_data(task_data)
@@ -212,11 +169,9 @@ def notify_other_tool_call(task_id: str = "default"):
if task_data:
task_data["last_key"] = None
task_data["consecutive"] = 0
if "dedup_hits" in task_data:
task_data["dedup_hits"].clear()
nf = task_data.get("not_found")
if nf:
nf.clear()
for key in ("dedup_hits", "not_found"):
if task_data.get(key):
task_data[key].clear()
def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
@@ -237,10 +192,8 @@ def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
if dedup:
for k in [k for k in dedup if k[0] == resolved]:
del dedup[k]
nf = task_data.get("not_found")
if nf:
nf.pop(("read", resolved), None)
nf.pop(("search", resolved), None)
_pop_not_found("read", resolved, task_id)
_pop_not_found("search", resolved, task_id)
def _update_read_timestamp(filepath: str, task_id: str) -> None:
@@ -271,9 +224,7 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | None:
return None
with _read_tracker_lock:
task_data = _read_tracker.get(task_id)
if not task_data:
return None
read_mtime = task_data.get("read_timestamps", {}).get(resolved)
read_mtime = task_data.get("read_timestamps", {}).get(resolved) if task_data else None
if read_mtime is None:
return None
try:
@@ -284,16 +235,12 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | None:
return (
f"Warning: {filepath} was modified since you last read it "
"(external edit or concurrent agent). The content you read may be "
"stale. Consider re-reading the file to verify before writing."
)
"stale. Consider re-reading the file to verify before writing.")
return None
def _mark_verification_stale(
task_id: str,
resolved_paths: list[str],
session_id: str | None = None,
) -> None:
def _mark_verification_stale(task_id: str, resolved_paths: list[str],
session_id: str | None = None) -> None:
"""Best-effort note that successful edits made prior verification stale.
The workspace cwd is the first edited path's project root when one is
@@ -308,22 +255,9 @@ def _mark_verification_stale(
from agent.coding_context import project_facts_for
from agent.verification_evidence import mark_workspace_edited
cwd = None
for path in paths:
try:
candidate = str(Path(path).parent)
except Exception:
continue
if project_facts_for(candidate):
cwd = candidate
break
if cwd is None:
cwd = _authoritative_workspace_root(task_id)
if cwd is None:
try:
cwd = str(Path(paths[0]).parent)
except Exception:
cwd = None
parents = [str(Path(p).parent) for p in paths]
cwd = (next((c for c in parents if project_facts_for(c)), None)
or _authoritative_workspace_root(task_id) or parents[0])
mark_workspace_edited(session_id=session_id or task_id, cwd=cwd, paths=paths)
except Exception:
logger.debug("verification stale marker failed", exc_info=True)
tools/file_tools_write_guards.py
+101 -207
View File
@@ -2,14 +2,10 @@
Every guard returns ``None`` when the write may proceed, else an error string
the tool returns verbatim. ``tools.file_tools`` re-imports every name here.
Guards (in the order the tools apply them):
* ``_check_sensitive_path`` — system paths + the Hermes config.yaml (hard deny).
* ``_check_binary_document_write`` — text write would corrupt an Office/PDF container.
* ``_check_protected_instruction_write`` — AGENTS.md-style files: ALWAYS ask, no yolo bypass.
* ``_check_approval_required_write`` — ~/.ssh/config-style files: normal approval gate.
* ``_check_cross_profile_path`` — sandbox-mirror writes the host never reads (lost-work guard).
* ``_is_internal_file_tool_content`` — refuse to persist read_file display text as a file.
Guards, in the order the tools apply them: ``_check_sensitive_path`` (hard
deny), ``_check_binary_document_write``, ``_check_protected_instruction_write``
(ALWAYS ask), ``_check_approval_required_write`` (normal gate),
``_check_cross_profile_path`` (sandbox-mirror lost-work), ``_is_internal_file_tool_content``.
"""
import fnmatch
@@ -25,31 +21,47 @@ from tools.file_tools_paths import _expand_tilde, _resolve_path_for_task
_SENSITIVE_PATH_PREFIXES = (
"/etc/", "/boot/", "/usr/lib/systemd/",
"/private/etc/",
"/private/var/db/", "/private/var/root/",
)
"/private/var/db/", "/private/var/root/")
_SENSITIVE_EXACT_PATHS = {"/var/run/docker.sock", "/run/docker.sock"}
_hermes_config_resolved: str | None = None
_hermes_config_resolved_loaded = False
_real_hermes_home_cached: str | None = None
_real_hermes_home_loaded = False
def _get_hermes_config_resolved() -> str | None:
"""Return the resolved absolute path of the Hermes config file (cached)."""
global _hermes_config_resolved, _hermes_config_resolved_loaded
if _hermes_config_resolved_loaded:
return _hermes_config_resolved
_hermes_config_resolved_loaded = True
try:
from hermes_cli.config import get_config_path
_hermes_config_resolved = str(get_config_path().resolve())
except Exception:
if not _hermes_config_resolved_loaded:
_hermes_config_resolved_loaded = True
try:
_hermes_config_resolved = str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve())
from hermes_cli.config import get_config_path
_hermes_config_resolved = str(get_config_path().resolve())
except Exception:
_hermes_config_resolved = None
try:
_hermes_config_resolved = str(Path(_expand_tilde("~/.hermes/config.yaml")).resolve())
except Exception:
_hermes_config_resolved = None
return _hermes_config_resolved
def _get_real_hermes_home() -> str | None:
"""Return the realpath of the authoritative Hermes home (cached)."""
global _real_hermes_home_cached, _real_hermes_home_loaded
if not _real_hermes_home_loaded:
_real_hermes_home_loaded = True
try:
from hermes_constants import get_hermes_home
_real_hermes_home_cached = os.path.realpath(str(get_hermes_home()))
except Exception:
try:
_real_hermes_home_cached = os.path.realpath(_expand_tilde("~/.hermes"))
except Exception:
_real_hermes_home_cached = None
return _real_hermes_home_cached
def _resolved_or_raw(filepath: str, task_id: str) -> str:
"""Task-resolved path string, falling back to the raw input on resolution failure."""
try:
@@ -60,64 +72,29 @@ def _resolved_or_raw(filepath: str, task_id: str) -> str:
def _check_sensitive_path(filepath: str, task_id: str = "default") -> str | None:
"""Return an error message if the path targets a sensitive system location."""
resolved = _resolved_or_raw(filepath, task_id)
normalized = os.path.normpath(_expand_tilde(filepath))
_err = (
f"Refusing to write to sensitive system path: {filepath}\n"
"Use the terminal tool with sudo if you need to modify system files."
)
for prefix in _SENSITIVE_PATH_PREFIXES:
if resolved.startswith(prefix) or normalized.startswith(prefix):
return _err
if resolved in _SENSITIVE_EXACT_PATHS or normalized in _SENSITIVE_EXACT_PATHS:
return _err
candidates = (_resolved_or_raw(filepath, task_id), os.path.normpath(_expand_tilde(filepath)))
if any(c.startswith(_SENSITIVE_PATH_PREFIXES) or c in _SENSITIVE_EXACT_PATHS for c in candidates):
return (
f"Refusing to write to sensitive system path: {filepath}\n"
"Use the terminal tool with sudo if you need to modify system files.")
# approvals.mode and other security settings live in config.yaml; a
# prompt-injected agent could silently disable exec approval by editing it.
hermes_config = _get_hermes_config_resolved()
if hermes_config and (resolved == hermes_config or normalized == hermes_config):
if hermes_config and hermes_config in candidates:
return (
f"Refusing to write to Hermes config file: {filepath}\n"
"Agent cannot modify security-sensitive configuration. "
"Edit ~/.hermes/config.yaml directly or use 'hermes config' instead."
)
"Edit ~/.hermes/config.yaml directly or use 'hermes config' instead.")
return None
# ---------------------------------------------------------------------------
# Protected agent-instruction files (always-ask approval gate)
# ---------------------------------------------------------------------------
# ── Protected agent-instruction files (always-ask approval gate) ─────────
# Files that steer FUTURE agent behavior are a prompt-injection persistence
# vector: an injected edit to AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules
# (or a project-local .hermes tree) outlives the turn and poisons every later
# session. Writes ALWAYS require human approval — even under --yolo — and fail
# closed when no human channel exists. (Ported from Roo-Code's
# RooProtectedController; the terminal-tool vector is gated separately.)
#
# Basenames match in ANY directory (instruction files load from cwd trees) and
# case-insensitively (case-insensitive filesystems; loaders probe variants).
# vector (AGENTS.md / CLAUDE.md / SOUL.md / .cursorrules / project .hermes tree).
# Writes ALWAYS require human approval — even under --yolo — and fail closed
# without a human channel. Basenames match in ANY directory, case-insensitively.
_PROTECTED_INSTRUCTION_BASENAMES = frozenset({
"agents.md", "claude.md", "soul.md", ".cursorrules",
})
_real_hermes_home_cached: str | None = None
_real_hermes_home_loaded = False
def _get_real_hermes_home() -> str | None:
"""Return the realpath of the authoritative Hermes home (cached)."""
global _real_hermes_home_cached, _real_hermes_home_loaded
if _real_hermes_home_loaded:
return _real_hermes_home_cached
_real_hermes_home_loaded = True
try:
from hermes_constants import get_hermes_home
_real_hermes_home_cached = os.path.realpath(str(get_hermes_home()))
except Exception:
try:
_real_hermes_home_cached = os.path.realpath(_expand_tilde("~/.hermes"))
except Exception:
_real_hermes_home_cached = None
return _real_hermes_home_cached
"agents.md", "claude.md", "soul.md", ".cursorrules"})
def _protected_instruction_config() -> tuple[bool, list[str]]:
@@ -129,10 +106,8 @@ def _protected_instruction_config() -> tuple[bool, list[str]]:
try:
from hermes_cli.config import load_config, cfg_get
cfg = load_config()
enabled = cfg_get(cfg, "security", "protected_instruction_files",
default=True)
extra = cfg_get(cfg, "security", "protected_instruction_extra_patterns",
default=[])
enabled = cfg_get(cfg, "security", "protected_instruction_files", default=True)
extra = cfg_get(cfg, "security", "protected_instruction_extra_patterns", default=[])
except Exception:
return True, []
if not isinstance(enabled, bool):
@@ -166,18 +141,15 @@ def _protected_instruction_reason(filepath: str, task_id: str = "default",
# mirror guard, write_approval); this gate targets PROJECT-LOCAL files only.
# Must run before the ``.hermes`` component rule, which would match the home.
real_home = _get_real_hermes_home()
if real_home and (resolved == real_home
or resolved.startswith(real_home + os.sep)):
if real_home and (resolved == real_home or resolved.startswith(real_home + os.sep)):
return None
for candidate in (normalized, resolved):
base = os.path.basename(candidate)
base_lower = base.lower()
if base_lower in _PROTECTED_INSTRUCTION_BASENAMES:
if base_lower in _PROTECTED_INSTRUCTION_BASENAMES or any(
fnmatch.fnmatch(base_lower, pattern.lower()) for pattern in extra_patterns):
return base
for pattern in extra_patterns:
if fnmatch.fnmatch(base_lower, pattern.lower()):
return base
# Project-local .hermes config dirs (<repo>/.hermes/config.yaml) steer
# behavior too. Only the IMMEDIATE parent counts — matching any ancestor
# would gate every write inside a checkout living under ~/.hermes.
@@ -187,8 +159,11 @@ def _protected_instruction_reason(filepath: str, task_id: str = "default",
return None
def _request_protected_instruction_approval(
reasons: list[str], task_id: str = "default") -> str | None:
_APPROVAL_UNAVAILABLE = "requires approval but the approval subsystem is unavailable."
_NO_HUMAN = "requires approval but no interactive user or gateway is present to approve it."
def _request_protected_instruction_approval(reasons: list[str], task_id: str = "default") -> str | None:
"""Ask the human to approve a write to protected instruction file(s).
Returns ``None`` when approved, else a BLOCKED error string. Deliberately
@@ -200,30 +175,24 @@ def _request_protected_instruction_approval(
description = (
f"Write to protected agent-instruction file(s): {targets}. "
"These files steer future agent behavior; approval is always "
"required (not bypassed by auto-approve)."
)
"required (not bypassed by auto-approve).")
display = f"<write to {targets}>"
blocked = (
f"BLOCKED: write to protected agent-instruction file(s) ({targets}) "
"{why} The user has NOT consented to this write. Do NOT retry it or "
"attempt the same edit via another path (terminal, execute_code, "
"etc.)."
)
timed_out = blocked.format(
why="approval prompt timed out without a user response. "
"Silence is not consent.")
"etc.).")
timed_out = blocked.format(why="approval prompt timed out without a user response. Silence is not consent.")
denied = blocked.format(why="was denied by the user.")
try:
import tools.approval as _approval
except Exception:
return blocked.format(why="requires approval but the approval "
"subsystem is unavailable.")
return blocked.format(why=_APPROVAL_UNAVAILABLE)
# Gateway surface: block on the button round-trip when a notify callback
# is registered for this session. One-operation only — no scope buttons.
session_key = _approval.get_current_session_key()
notify_cb = None
try:
with _approval._lock:
notify_cb = _approval._gateway_notify_cbs.get(session_key)
@@ -237,53 +206,32 @@ def _request_protected_instruction_approval(
"pattern_keys": ["protected_instruction_file"],
"description": description,
"allow_permanent": False,
"allow_session": False,
}
decision = _approval._await_gateway_decision(
session_key, notify_cb, approval_data, surface="gateway",
)
"allow_session": False}
decision = _approval._await_gateway_decision(session_key, notify_cb, approval_data, surface="gateway")
if decision.get("notify_failed"):
return blocked.format(
why="requires approval but the approval request could not "
"be delivered.")
choice = decision.get("choice")
# Any tapped scope is a one-operation grant; nothing is persisted.
if decision.get("resolved") and choice in {"once", "session", "always"}:
return None
if not decision.get("resolved"):
return timed_out
return denied
# CLI surface: per-thread approval callback (prompt_toolkit panel).
callback = None
try:
from tools.terminal_tool import _get_approval_callback
callback = _get_approval_callback()
except Exception:
callback = None
if callback is not None:
return blocked.format(why="requires approval but the approval request could not be delivered.")
choice, timed = decision.get("choice"), not decision.get("resolved")
else:
# CLI surface: per-thread approval callback (prompt_toolkit panel).
try:
from tools.terminal_tool import _get_approval_callback
callback = _get_approval_callback()
except Exception:
callback = None
if callback is None:
# No human channel (script, cron, background thread): fail closed —
# auto-approving here would recreate the persistence vector.
return blocked.format(why=_NO_HUMAN)
choice = _approval.prompt_dangerous_approval(
display, description,
allow_permanent=False,
allow_session=False,
approval_callback=callback,
)
if choice in {"once", "session", "always"}:
return None
if choice == "timeout":
return timed_out
return denied
# No human channel (script, cron, background thread): fail closed —
# auto-approving here would recreate the persistence vector.
return blocked.format(
why="requires approval but no interactive user or gateway is "
"present to approve it.")
display, description, allow_permanent=False, allow_session=False, approval_callback=callback)
timed = choice == "timeout"
# Any tapped scope is a one-operation grant; nothing is persisted.
if not timed and choice in {"once", "session", "always"}:
return None
return timed_out if timed else denied
def _check_protected_instruction_write(paths: list[str],
task_id: str = "default") -> str | None:
def _check_protected_instruction_write(paths: list[str], task_id: str = "default") -> str | None:
"""Gate a write/patch touching protected instruction files.
ONE protected file gates the ENTIRE multi-file patch: a single prompt lists
@@ -293,19 +241,14 @@ def _check_protected_instruction_write(paths: list[str],
enabled, extra = _protected_instruction_config()
if not enabled:
return None
reasons = [
r for r in (
_protected_instruction_reason(p, task_id, enabled=enabled, extra_patterns=extra)
for p in paths
) if r
]
reasons = [r for r in (_protected_instruction_reason(p, task_id, enabled=enabled, extra_patterns=extra)
for p in paths) if r]
if not reasons:
return None
return _request_protected_instruction_approval(reasons, task_id)
def _check_approval_required_write(paths: list[str],
task_id: str = "default") -> str | None:
def _check_approval_required_write(paths: list[str], task_id: str = "default") -> str | None:
"""Gate a write/patch touching an approval-required path (``~/.ssh/config``).
Not credentials and not hard-denied, but they can steer process execution
@@ -326,26 +269,22 @@ def _check_approval_required_write(paths: list[str],
description = (
f"Write to SSH client config file(s): {display_targets}. "
"The SSH config can carry ProxyCommand / Match exec directives that "
"run commands, so writes require your approval."
)
"run commands, so writes require your approval.")
blocked = (
f"BLOCKED: write to SSH config file(s) ({display_targets}) "
"{why} Do NOT retry it via another path (terminal, execute_code) "
"without the user's explicit consent."
)
"without the user's explicit consent.")
try:
import tools.approval as _approval
except Exception:
return blocked.format(why="requires approval but the approval "
"subsystem is unavailable.")
return blocked.format(why=_APPROVAL_UNAVAILABLE)
result = _approval._run_approval_gate(
pattern_key="ssh_config_write",
description=description,
display_target=f"<write to {display_targets}>",
cron_deny_message=blocked.format(
why="requires approval but this cron session denies it."),
cron_deny_message=blocked.format(why="requires approval but this cron session denies it."),
single_query_deny_message=blocked.format(
why="requires approval but single-query (-q) sessions run "
"without a user present to approve it. To allow flagged "
@@ -353,10 +292,7 @@ def _check_approval_required_write(paths: list[str],
"approve in config.yaml."),
autoapprove_log_prefix="ssh_config_write",
fail_closed_when_no_human=True,
no_human_block_message=blocked.format(
why="requires approval but no interactive user or gateway is "
"present to approve it."),
)
no_human_block_message=blocked.format(why=_NO_HUMAN))
if result.get("approved"):
return None
return result.get("message") or blocked.format(why="was denied.")
@@ -366,31 +302,17 @@ def _get_container_mirror_prefix_for_task(task_id: str = "default") -> str | Non
"""Return the container-side Hermes mirror prefix for persistent Docker file tools."""
try:
from tools.terminal_tool import (
_active_environments,
_env_lock,
_get_env_config,
_resolve_container_task_id,
)
_active_environments, _env_lock, _get_env_config, _resolve_container_task_id)
container_key = _resolve_container_task_id(task_id)
except Exception:
return None
try:
with _env_lock:
env = _active_environments.get(container_key) or _active_environments.get(task_id)
if env is not None:
if env.__class__.__name__ == "DockerEnvironment" and bool(
getattr(env, "_persistent", False)
):
return "/root/.hermes"
return None
persistent_docker = (env.__class__.__name__ == "DockerEnvironment"
and bool(getattr(env, "_persistent", False)))
return "/root/.hermes" if persistent_docker else None
config = _get_env_config()
except Exception:
return None
if config.get("env_type") == "docker" and config.get("container_persistent", True):
return "/root/.hermes"
return None
@@ -406,23 +328,14 @@ def _check_cross_profile_path(filepath: str, task_id: str = "default") -> str |
Fails open on import error — the sensitive-path guard and denylist still apply.
"""
try:
from agent.file_safety import (
get_container_mirror_warning,
get_sandbox_mirror_warning,
)
from agent.file_safety import get_container_mirror_warning, get_sandbox_mirror_warning
except Exception:
return None
resolved = _resolved_or_raw(filepath, task_id)
warning = get_sandbox_mirror_warning(resolved)
if warning is not None:
return warning
return get_container_mirror_warning(
resolved,
mirror_prefix=_get_container_mirror_prefix_for_task(task_id),
)
return get_container_mirror_warning(resolved, mirror_prefix=_get_container_mirror_prefix_for_task(task_id))
def _check_binary_document_write(filepath: str, task_id: str = "default") -> str | None:
@@ -442,8 +355,7 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
"corrupt the file (read_file showed you EXTRACTED text, not the real "
"bytes). Use the docx/xlsx/powerpoint skills or a library like "
"python-docx/openpyxl/python-pptx via the terminal to create or edit "
"this document."
)
"this document.")
if is_pdf_path(filepath):
try:
resolved = Path(_resolve_path_for_task(filepath, task_id))
@@ -456,21 +368,17 @@ def _check_binary_document_write(filepath: str, task_id: str = "default") -> str
"read_file showed you EXTRACTED text, not the real bytes — writing "
"text back would destroy the document. Use the pdf skill or a PDF "
"library via the terminal to modify it. (Creating a NEW .pdf file "
"is allowed.)"
)
"is allowed.)")
except OSError:
pass
return None
# ---------------------------------------------------------------------------
# Internal display text must never be persisted as file content
# ---------------------------------------------------------------------------
# ── Internal display text must never be persisted as file content ────────
_READ_DEDUP_STATUS_MESSAGE = (
"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."
)
"still current — refer to that instead of re-reading.")
def _is_internal_file_status_text(content: str) -> bool:
@@ -484,12 +392,8 @@ def _is_internal_file_status_text(content: str) -> bool:
if not isinstance(content, str):
return False
stripped = content.strip()
if not stripped:
return False
if stripped == _READ_DEDUP_STATUS_MESSAGE:
return True
return (_READ_DEDUP_STATUS_MESSAGE in stripped
and len(stripped) <= 2 * len(_READ_DEDUP_STATUS_MESSAGE))
return bool(stripped) and _READ_DEDUP_STATUS_MESSAGE in stripped and (
len(stripped) <= 2 * len(_READ_DEDUP_STATUS_MESSAGE))
def _looks_like_read_file_line_numbered_content(content: str) -> bool:
@@ -501,30 +405,20 @@ def _looks_like_read_file_line_numbered_content(content: str) -> bool:
"""
if not isinstance(content, str):
return False
lines = [line for line in content.splitlines() if line.strip()]
if len(lines) < 2:
return False
numbered: list[int] = []
for line in lines:
prefix, sep, _rest = line.lstrip().partition("|")
if sep and prefix.isdigit():
numbered.append(int(prefix))
if len(numbered) < 2 or len(numbered) / len(lines) < 0.6:
return False
consecutive_pairs = sum(
1 for prev, current in zip(numbered, numbered[1:])
if current == prev + 1
)
consecutive_pairs = sum(1 for prev, current in zip(numbered, numbered[1:]) if current == prev + 1)
return consecutive_pairs >= len(numbered) - 1
def _is_internal_file_tool_content(content: str) -> bool:
"""Return True when content is file-tool display text, not intended file bytes."""
return (
_is_internal_file_status_text(content)
or _looks_like_read_file_line_numbered_content(content)
)
return _is_internal_file_status_text(content) or _looks_like_read_file_line_numbered_content(content)
tools/fuzzy_match.py
+19 -44
View File
@@ -18,8 +18,7 @@ Span = tuple[int, int]
IDENTICAL_STRINGS_ERROR = (
"No edit was applied because old_string and new_string are identical. "
"Provide the existing text to replace in old_string and the changed "
"replacement text in new_string."
)
"replacement text in new_string.")
UNICODE_MAP = {
"\u201c": '"', "\u201d": '"', # smart double quotes
@@ -32,8 +31,7 @@ UNICODE_MAP = {
"\u2000": " ", "\u2001": " ", "\u2002": " ", "\u2003": " ",
"\u2004": " ", "\u2005": " ", "\u2006": " ", "\u2007": " ",
"\u2008": " ", "\u2009": " ", "\u200a": " ", "\u202f": " ",
"\u205f": " ", "\u3000": " ",
}
"\u205f": " ", "\u3000": " "}
def _unicode_normalize(text: str) -> str:
@@ -58,8 +56,7 @@ def _window_spans(content: str, content_lines: list[str], n: int,
"""Spans of every ``n``-line window starting at ``i`` for which ``accept(i)``."""
return [
_calculate_line_positions(content_lines, i, i + n, len(content))
for i in range(len(content_lines) - n + 1) if accept(i)
]
for i in range(len(content_lines) - n + 1) if accept(i)]
def _match_transformed_lines(content: str, pattern: str,
@@ -82,12 +79,9 @@ def _strip_boundary(lines: list[str]) -> list[str]:
def _build_orig_to_norm_map(original: str) -> list[int]:
"""Map each original index to its index in ``_unicode_normalize(original)``.
UNICODE_MAP replacements can expand one char into several, so the map is
needed to translate normalised spans back. Length is ``len(original)+1``;
the last entry is a sentinel one past the final character.
"""
"""Map each original index to its index in ``_unicode_normalize(original)``
(replacements can expand one char into several). Length ``len(original)+1``;
the last entry is a sentinel one past the final character."""
result: list[int] = []
norm_pos = 0
for char in original:
@@ -180,18 +174,9 @@ def _map_normalized_positions(original: str, normalized: str,
# ORIGINAL content.
def _strategy_exact(content: str, pattern: str) -> list[Span]:
"""Strategy 1: exact, non-overlapping occurrences (str.replace semantics)."""
matches = []
start = 0
while True:
pos = content.find(pattern, start)
if pos == -1:
break
matches.append((pos, pos + len(pattern)))
# Advance past the whole match: overlapping spans would corrupt the
# file under replace_all (reverse-order apply on stale offsets).
start = pos + len(pattern)
return matches
"""Strategy 1: exact, non-overlapping occurrences (str.replace semantics —
overlapping spans would corrupt the file under replace_all)."""
return [m.span() for m in re.finditer(re.escape(pattern), content)]
def _strategy_line_trimmed(content: str, pattern: str) -> list[Span]:
@@ -257,8 +242,7 @@ def _strategy_block_anchor(content: str, pattern: str) -> list[Span]:
potential_matches = {
i for i in range(len(norm_content_lines) - n + 1)
if norm_content_lines[i].strip() == first_line
and norm_content_lines[i + n - 1].strip() == last_line
}
and norm_content_lines[i + n - 1].strip() == last_line}
# Looser thresholds (0.10/0.30) matched unrelated blocks; these are the safe floor.
threshold = 0.50 if len(potential_matches) == 1 else 0.70
pattern_middle = '\n'.join(pattern_lines[1:-1])
@@ -300,8 +284,7 @@ def _strategy_context_aware(content: str, pattern: str) -> list[Span]:
return False
return all(
not p_line.strip() or _sim(p_line.strip(), c_line.strip()) >= 0.80
for p_line, c_line in zip(pattern_lines, block_lines)
)
for p_line, c_line in zip(pattern_lines, block_lines))
return _window_spans(content, content_lines, n, accept)
@@ -316,8 +299,7 @@ STRATEGIES: list[tuple[str, Callable[[str, str], list[Span]]]] = [
("trimmed_boundary", _strategy_trimmed_boundary),
("unicode_normalized", _strategy_unicode_normalized),
("block_anchor", _strategy_block_anchor),
("context_aware", _strategy_context_aware),
]
("context_aware", _strategy_context_aware)]
# Matches from these only *approximately* resemble old_string — fine for one
# unique replacement, never safe under replace_all.
@@ -386,15 +368,13 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
return content, 0, None, (
f"Found {len(matches)} matches for old_string. "
f"Provide more context to make it unique, or use replace_all=True. "
f"Matches:\n{locations}"
)
f"Matches:\n{locations}")
if replace_all and len(matches) > 1 and strategy_name in SIMILARITY_STRATEGIES:
return content, 0, None, (
f"Found {len(matches)} approximate matches via the "
f"'{strategy_name}' strategy; replace_all only applies to exact "
f"matches. Provide the precise text (whitespace included) so an "
f"exact/line-trimmed match can be made."
)
f"exact/line-trimmed match can be made.")
# Non-exact matches came through some normalization, so new_string may
# carry serialization drift the file doesn't have.
@@ -408,8 +388,7 @@ def fuzzy_find_and_replace(content: str, old_string: str, new_string: str,
effective_new = _preserve_unicode_in_replacement(content, matches, old_string, effective_new)
new_content = _apply_replacements(
content, matches, effective_new,
old_string=old_string if strategy_name != "exact" else None,
)
old_string=old_string if strategy_name != "exact" else None)
return new_content, len(matches), strategy_name, None
return content, 0, None, "Could not find a match for old_string in the file"
@@ -441,8 +420,7 @@ def _detect_escape_drift(content: str, matches: list[Span],
f"serialization artifact where an apostrophe or quote got "
f"prefixed with a spurious backslash. Re-read the file with "
f"read_file and pass old_string/new_string without "
f"backslash-escaping {plain!r} characters."
)
f"backslash-escaping {plain!r} characters.")
return _detect_backslash_doubling(matched_regions, old_string, new_string)
@@ -476,8 +454,7 @@ def _detect_backslash_doubling(matched_regions: str, old_string: str,
"were JSON-escaped one extra time; applying new_string verbatim would "
"double every backslash in the file. Re-read the file with read_file "
"and resend old_string/new_string with the backslash counts exactly "
"as they appear in the file."
)
"as they appear in the file.")
def _maybe_unescape_new_string(new_string: str, content: str, matches: list[Span]) -> str:
@@ -626,8 +603,7 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
continue
seen_ranges.add((start, end))
parts.append("\n".join(
f"{start + j + 1:4d}| {content_lines[start + j]}" for j in range(end - start)
))
f"{start + j + 1:4d}| {content_lines[start + j]}" for j in range(end - start)))
if not parts:
return ""
result = "\n---\n".join(parts)
@@ -640,8 +616,7 @@ def find_closest_lines(old_string: str, content: str, context_lines: int = 2, ma
"\n\nWhitespace difference detected (→ = tab, · = space):\n"
f" file has: {_visualize_whitespace(best_line)}\n"
f" you sent: {_visualize_whitespace(old_lines[0])}\n"
"Use the exact whitespace shown in 'file has'."
)
"Use the exact whitespace shown in 'file has'.")
return result