refactor(tools): hand-compact docstrings across file_tools group (keep every WHY)

This commit is contained in:
Teknium
2026-09-02 23:17:52 -07:00
parent 48de33b77f
commit e4436ec9e5
6 changed files with 91 additions and 226 deletions

View File

@@ -38,11 +38,8 @@ _NOT_FOUND_TTL_SECONDS = 60.0 # a path that didn't exist may be created soon
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.
"""
"""Get-or-create the tracker entry for *task_id*, back-filling missing containers
(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"):
@@ -96,9 +93,8 @@ def _pop_not_found(op: str, resolved_str: str, task_id: 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.
Skips the subprocess + similar-name walk when the model retries the same
missing path. *op* is "read" or "search" (different error JSON shapes).
Evicted by TTL, by write_file/patch on the path, or by any other tool call.
*op* is "read" or "search" (different error JSON shapes). Evicted by TTL,
by write_file/patch on the path, or by any other tool call.
"""
with _read_tracker_lock:
task_data = _read_tracker.get(task_id)
@@ -109,10 +105,9 @@ def _check_not_found_cache(op: str, resolved_str: str, task_id: str) -> str | No
if time.monotonic() - ts > _NOT_FOUND_TTL_SECONDS:
_pop_not_found(op, resolved_str, task_id)
return None
# The path may have been created since the miss was cached (terminal,
# 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.
# "check → create → read" is common, so never serve a stale miss for a path
# that now exists. The stat runs OUTSIDE the 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:
_pop_not_found(op, resolved_str, task_id)
@@ -139,11 +134,8 @@ def _bump_consecutive(task_data: dict, key: tuple) -> int:
def reset_file_dedup(task_id: str = None):
"""Clear the read-dedup cache (one task, or all when ``task_id`` is None).
Called after context compression: the original read content was summarised
away, so a "file unchanged" stub would point at content no longer in context.
"""
"""Clear the read-dedup cache (one task, or all when ``task_id`` is None). Called
after context compression: a "file unchanged" stub would point at summarised-away content."""
with _read_tracker_lock:
if task_id:
targets = [_read_tracker[task_id]] if _read_tracker.get(task_id) else []
@@ -158,11 +150,9 @@ def reset_file_dedup(task_id: str = None):
def notify_other_tool_call(task_id: str = "default"):
"""Reset the consecutive read/search counter for a task.
Called by the dispatcher for every tool OTHER than read_file/search_files,
so loop detection only fires on truly consecutive repeats. Also clears the
stub-hit counters and the not-found cache: any other tool may have created
a previously-missing path (the serve-side stat covers most cases; clearing
covers the rest, e.g. permission flips).
Called by the dispatcher for every tool OTHER than read_file/search_files.
Also clears stub-hit counters and the not-found cache: any other tool may
have created a previously-missing path (or flipped its permissions).
"""
with _read_tracker_lock:
task_data = _read_tracker.get(task_id)
@@ -175,11 +165,8 @@ def notify_other_tool_call(task_id: str = "default"):
def _invalidate_dedup_for_path(filepath: str, task_id: str) -> None:
"""Evict every dedup entry (all offset/limit ranges) and not-found entry for *filepath*.
Called after write_file/patch so the next read returns fresh content
instead of a stale "unchanged" stub. Acquires ``_read_tracker_lock`` itself.
"""
"""Evict every dedup entry (all offset/limit ranges) and not-found entry for *filepath*
after a write, so the next read returns fresh content. Acquires the lock itself."""
try:
resolved = str(_resolve_path_for_task(filepath, task_id))
except (OSError, ValueError):
@@ -214,10 +201,7 @@ def _update_read_timestamp(filepath: str, task_id: str) -> None:
def _check_file_staleness(filepath: str, task_id: str) -> str | None:
"""Warn (don't block) when the file's mtime changed since this task last read it.
``None`` when never read, fresh, or unstattable (a deleted file is the
write's problem to report).
"""
``None`` when never read, fresh, or unstattable (a deleted file is the write's problem)."""
try:
resolved = str(_resolve_path_for_task(filepath, task_id))
except (OSError, ValueError):
@@ -241,11 +225,8 @@ def _check_file_staleness(filepath: str, task_id: str) -> str | 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
recognised, else the task's workspace root, else the first path's parent.
"""
"""Best-effort note that successful edits made prior verification stale. cwd: the
first edited path's recognised project root, else the workspace root, else the first parent."""
from pathlib import Path
paths = [p for p in resolved_paths if p]